29471089948

Committed 15 Jul 2026 11:28PM UTC coverage: 89.402% (-2.3%) from 91.656%

Build # 29471089948

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #5724 from randombit/jack/filter-fixes

Fix various bugs in the Pipe/Filter library

Coverage Stats

113560 of 127022 relevant lines covered (89.4%)

10935343.63 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.23

/src/lib/filters/threaded_fork.cpp

/*
* Threaded Fork
* (C) 2013 Joel Low
*     2013 Jack Lloyd
*
* Botan is released under the Simplified BSD License (see license.txt)
*/

#include <botan/filters.h>

#if defined(BOTAN_HAS_THREAD_UTILS)

   #include <botan/internal/barrier.h>
   #include <botan/internal/semaphore.h>

namespace Botan {

struct Threaded_Fork_Data {
      /*
      * Semaphore for indicating that there is work to be done (or to
      * quit)
      */
      Semaphore m_input_ready_semaphore;

      /*
      * Synchronises all threads to complete processing data in lock-step.
      */
      Barrier m_input_complete_barrier;

      /*
      * The work that needs to be done. This should be only when the threads
      * are NOT running (i.e. before notifying the work condition, after
      * the input_complete_barrier has reset.)
      */
      const uint8_t* m_input = nullptr;

      /*
      * The length of the work that needs to be done.
      */
      size_t m_input_length = 0;
};

/*
* Threaded_Fork constructor
*/
Threaded_Fork::Threaded_Fork(Filter* f1, Filter* f2, Filter* f3, Filter* f4) :
      Fork(nullptr, static_cast<size_t>(0)), m_thread_data(new Threaded_Fork_Data) {
   Filter* filters[4] = {f1, f2, f3, f4};
   set_next(filters, 4);
}

/*
* Threaded_Fork constructor
*/
Threaded_Fork::Threaded_Fork(Filter* filters[], size_t count) :
      Fork(nullptr, static_cast<size_t>(0)), m_thread_data(new Threaded_Fork_Data) {
   set_next(filters, count);
}

Threaded_Fork::~Threaded_Fork() {
   m_thread_data->m_input = nullptr;
   m_thread_data->m_input_length = 0;

   m_thread_data->m_input_ready_semaphore.release(m_threads.size());

   for(auto& thread : m_threads) {
      thread->join();
   }
}

std::string Threaded_Fork::name() const {
   return "Threaded Fork";
}

void Threaded_Fork::set_next(Filter* f[], size_t n) {
   Fork::set_next(f, n);
   n = m_next.size();

   if(n < m_threads.size()) {
      m_threads.resize(n);
   } else {
      m_threads.reserve(n);
      for(size_t i = m_threads.size(); i != n; ++i) {
         m_threads.push_back(std::make_shared<std::thread>([this, next = m_next[i]] { thread_entry(next); }));
      }
   }
}

void Threaded_Fork::send(const uint8_t input[], size_t length) {
   /*
   * The workers treat a nullptr input pointer as a shutdown marker, and a (nullptr, 0)
   * call is otherwise valid. Just ignore an empty input here, matching Filter::send
   */
   if(length == 0) {
      return;
   }

   if(!m_write_queue.empty()) {
      thread_delegate_work(m_write_queue.data(), m_write_queue.size());
   }
   thread_delegate_work(input, length);

   bool nothing_attached = true;
   for(size_t j = 0; j != total_ports(); ++j) {
      if(m_next[j] != nullptr) {
         nothing_attached = false;
      }
   }

   if(nothing_attached) {
      m_write_queue += std::make_pair(input, length);
   } else {
      m_write_queue.clear();
   }
}

void Threaded_Fork::thread_delegate_work(const uint8_t input[], size_t length) {
   //Set the data to do.
   m_thread_data->m_input = input;
   m_thread_data->m_input_length = length;

   //Let the workers start processing.
   m_thread_data->m_input_complete_barrier.wait(total_ports() + 1);
   m_thread_data->m_input_ready_semaphore.release(total_ports());

   //Wait for all the filters to finish processing.
   m_thread_data->m_input_complete_barrier.sync();

   //Reset the thread data
   m_thread_data->m_input = nullptr;
   m_thread_data->m_input_length = 0;
}

void Threaded_Fork::thread_entry(Filter* filter) {
   while(true) {
      m_thread_data->m_input_ready_semaphore.acquire();

      if(m_thread_data->m_input == nullptr) {
         break;
      }

      // Plain Fork skips null ports the same way
      if(filter != nullptr) {
         filter->write(m_thread_data->m_input, m_thread_data->m_input_length);
      }
      m_thread_data->m_input_complete_barrier.sync();
   }
}

}  // namespace Botan

#endif

1	/*
2	* Threaded Fork
3	* (C) 2013 Joel Low
4	* 2013 Jack Lloyd
5	*
6	* Botan is released under the Simplified BSD License (see license.txt)
7	*/
8
9	#include <botan/filters.h>
10
11	#if defined(BOTAN_HAS_THREAD_UTILS)
12
13	#include <botan/internal/barrier.h>
14	#include <botan/internal/semaphore.h>
15
16	namespace Botan {
17
18	struct Threaded_Fork_Data {	2✔
19	/*
20	* Semaphore for indicating that there is work to be done (or to
21	* quit)
22	*/
23	Semaphore m_input_ready_semaphore;
24
25	/*
26	* Synchronises all threads to complete processing data in lock-step.
27	*/
28	Barrier m_input_complete_barrier;
29
30	/*
31	* The work that needs to be done. This should be only when the threads
32	* are NOT running (i.e. before notifying the work condition, after
33	* the input_complete_barrier has reset.)
34	*/
35	const uint8_t* m_input = nullptr;
36
37	/*
38	* The length of the work that needs to be done.
39	*/
40	size_t m_input_length = 0;
41	};
42
43	/*
44	* Threaded_Fork constructor
45	*/
46	Threaded_Fork::Threaded_Fork(Filter* f1, Filter* f2, Filter* f3, Filter* f4) :	1✔
47	Fork(nullptr, static_cast<size_t>(0)), m_thread_data(new Threaded_Fork_Data) {	1✔
48	Filter* filters[4] = {f1, f2, f3, f4};	1✔
49	set_next(filters, 4);	1✔
50	}	1✔
51
52	/*
53	* Threaded_Fork constructor
54	*/
55	Threaded_Fork::Threaded_Fork(Filter* filters[], size_t count) :	1✔
56	Fork(nullptr, static_cast<size_t>(0)), m_thread_data(new Threaded_Fork_Data) {	1✔
57	set_next(filters, count);	1✔
58	}	1✔
59
60	Threaded_Fork::~Threaded_Fork() {	4✔
61	m_thread_data->m_input = nullptr;	2✔
62	m_thread_data->m_input_length = 0;	2✔
63
64	m_thread_data->m_input_ready_semaphore.release(m_threads.size());	2✔
65
66	for(auto& thread : m_threads) {	9✔
67	thread->join();	7✔
68	}
69	}	6✔
70
71	std::string Threaded_Fork::name() const {	×
72	return "Threaded Fork";	×
73	}
74
75	void Threaded_Fork::set_next(Filter* f[], size_t n) {	2✔
76	Fork::set_next(f, n);	2✔
77	n = m_next.size();	2✔
78
79	if(n < m_threads.size()) {	2✔
80	m_threads.resize(n);	×
81	} else {
82	m_threads.reserve(n);	2✔
83	for(size_t i = m_threads.size(); i != n; ++i) {	9✔
84	m_threads.push_back(std::make_shared<std::thread>([this, next = m_next[i]] { thread_entry(next); }));	14✔
85	}
86	}
87	}	2✔
88
89	void Threaded_Fork::send(const uint8_t input[], size_t length) {	920✔
90	/*
91	* The workers treat a nullptr input pointer as a shutdown marker, and a (nullptr, 0)
92	* call is otherwise valid. Just ignore an empty input here, matching Filter::send
93	*/
94	if(length == 0) {	920✔
95	return;
96	}
97
98	if(!m_write_queue.empty()) {	920✔
99	thread_delegate_work(m_write_queue.data(), m_write_queue.size());	×
100	}
101	thread_delegate_work(input, length);	920✔
102
103	bool nothing_attached = true;	920✔
104	for(size_t j = 0; j != total_ports(); ++j) {	5,517✔
105	if(m_next[j] != nullptr) {	4,597✔
106	nothing_attached = false;	4,597✔
107	}
108	}
109
110	if(nothing_attached) {	920✔
111	m_write_queue += std::make_pair(input, length);	×
112	} else {
113	m_write_queue.clear();	920✔
114	}
115	}
116
117	void Threaded_Fork::thread_delegate_work(const uint8_t input[], size_t length) {	920✔
118	//Set the data to do.
119	m_thread_data->m_input = input;	920✔
120	m_thread_data->m_input_length = length;	920✔
121
122	//Let the workers start processing.
123	m_thread_data->m_input_complete_barrier.wait(total_ports() + 1);	920✔
124	m_thread_data->m_input_ready_semaphore.release(total_ports());	920✔
125
126	//Wait for all the filters to finish processing.
127	m_thread_data->m_input_complete_barrier.sync();	920✔
128
129	//Reset the thread data
130	m_thread_data->m_input = nullptr;	920✔
131	m_thread_data->m_input_length = 0;	920✔
132	}	920✔
133
134	void Threaded_Fork::thread_entry(Filter* filter) {	7✔
135	while(true) {	4,597✔
136	m_thread_data->m_input_ready_semaphore.acquire();	4,604✔
137
138	if(m_thread_data->m_input == nullptr) {	4,604✔
139	break;
140	}
141
142	// Plain Fork skips null ports the same way
143	if(filter != nullptr) {	4,597✔
144	filter->write(m_thread_data->m_input, m_thread_data->m_input_length);	4,597✔
145	}
146	m_thread_data->m_input_complete_barrier.sync();	4,597✔
147	}
148	}	7✔
149
150	} // namespace Botan
151
152	#endif

randombit / botan / 29471089948

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous