Manage a pool of worker threads. More...

Classes
class	worker_thread_t
	Class used to manage the worker and worker thread. More...

Public Types
typedef std::shared_ptr< pool< W, A... > >	pointer_t
	A shared pointer for your pools.

typedef W::work_load_type	work_load_type
	The type of the workload item.

typedef fifo< work_load_type >	worker_fifo_t
	This type represents the type of the fifo used by the pool.

Public Member Functions
	pool (std::string const &name, std::size_t pool_size, typename worker_fifo_t::pointer_t in, typename worker_fifo_t::pointer_t out, A... args)
	Initializes a pool of worker threads.

	~pool ()
	Make sure that the thread pool is cleaned up.

W &	get_worker (int i)
	Get worker at index `i`.

W const &	get_worker (int i) const
	Get worker at index `i` (constant version).

bool	pop_front (work_load_type &v, std::int64_t usecs)
	Retrieve one work load of processed data.

void	push_back (work_load_type const &v)
	Push one work load of data.

std::size_t	size () const
	Retrieve the number of workers.

void	stop (bool immediate)
	Stop the threads.

void	wait ()
	Wait on the threads to be done.

Private Types
typedef worker_thread_t::vector_t	workers_t
	Vector of workers.

Private Attributes
worker_fifo_t::pointer_t	f_in
	The input FIFO.

std::string const	f_name
	The name of this pool of threads.

worker_fifo_t::pointer_t	f_out
	The output FIFO.

workers_t	f_workers = workers_t()
	The vector of workers.

Detailed Description

template<class W, class ... A>
class cppthread::pool< W, A >

This function manages a pool of worker threads. It allocates the threads, accepts incoming data, and returns outgoing data. Everything else is managed for you.

To make use of this template, you need to overload the snap_worker template and implement your own snap_worker::do_work() function.

Something like this:

struct data
{
    std::string f_func = "counter";
    int         f_counter = 0;
};
 
class foo
    : public snap_worker<data>
{
    void do_work()
    {
        if(f_workload.f_func == "counter")
        {
            ++f_workload.f_counter;
        }
        else if(f_workload.f_func == "odd-even")
        {
            f_workload.f_counter ^= 1;
        }
        // ...etc...
    }
};
 
cppthread_pool<foo> pool("my pool", 10);
 
// generate input
data data_in;
...fill data_in...
pool.push_back(data_in);
 
// retrieve output
data data_out;
if(pool.pop_front(data_out))
{
    ...use data_out...
}

You can do all the push_data() you need before doing any pop_front(). You may also want to consider looping with both interleaved or even have two threads: one feeder which does the push_data() and one consumer which does the pop_front().

Note that the thread pool does not guarantee order of processing. You must make sure that each individual chunk of data you pass in the push_front() function can be processed in any order.

Definition at line 47 of file pool.h.

Member Typedef Documentation

◆ pointer_t

template<class W , class ... A>

cppthread::pool< W, A >::pointer_t

We expect people to use pools as is in their objects, but if you'd like to allocate a pool, we recommand that you use the std::make_shared<>() function and create a shared pointer.

Definition at line 50 of file pool.h.

◆ work_load_type

template<class W , class ... A>

cppthread::pool< W, A >::work_load_type

Items that we add to the input and output FIFOs must be of that type.

The push_back() and pop_front() functions make use of items of this type to add items to the input fifo and pop items from the output fifo.

Note that the output fifo can be set to a null pointer. In that case the pop function cannot be used.

Definition at line 51 of file pool.h.

◆ worker_fifo_t

template<class W , class ... A>

cppthread::pool< W, A >::worker_fifo_t

The pool needs to accept incoming and outgoing items. These are added to an input and an output fifo. The workload is compatible with those FIFOs.

Definition at line 52 of file pool.h.

◆ workers_t

template<class W , class ... A>

cppthread::pool< W, A >::workers_t

private

This type is used internally to hold all the worker threads.

Definition at line 185 of file pool.h.

Constructor & Destructor Documentation

◆ pool()

template<class W , class ... A>

cppthread::pool< W, A >::pool	(	std::string const &	name,
		std::size_t	pool_size,
		typename worker_fifo_t::pointer_t	in,
		typename worker_fifo_t::pointer_t	out,
		A...	args
	)

inline

This constructor is used to initialize the specified number of worker threads in this pool.

At this time, all the worker threads get allocated upfront.

Todo:: We may want to add more threads as the work load increases which could allow for much less unused threads created. To do so we want to (1) time the do_work() function to get an idea of how long a thread takes to perform its work; and (2) have a threshold so we know when the client wants to create new worker threads if the load increases instead of assuming it should happen as soon as data gets added to the input FIFO. The pool_size parameter would then become a max_pool_size.

Parameters

[in]	name	The name of the pool.
[in]	pool_size	The number of threads to create.
[in]	in	The input FIFO (where workers receive workload.)
[in]	out	The output FIFO (where finished work is sent.)
[in]	args	Extra arguments to initialize the workers.

Definition at line 93 of file pool.h.

References cppthread::pool< W, A >::f_in, cppthread::pool< W, A >::f_name, cppthread::pool< W, A >::f_out, and cppthread::pool< W, A >::f_workers.

◆ ~pool()

template<class W , class ... A>

cppthread::pool< W, A >::~pool ( )

inline

The destructor of the snap thread pool stops all the threads and then waits on them. Assuming that your code doesn't loop forever, the result is that all the threads are stopped, joined, and all the incoming and outgoing data was cleared.

If you need to get the output data, then make sure to call stop(), wait(), and pop_front() until it returns false. Only then can you call the destructor.

You are safe to run that stop process in your own destructor.

Definition at line 123 of file pool.h.

References cppthread::pool< W, A >::stop(), and cppthread::pool< W, A >::wait().

Here is the call graph for this function:

Member Function Documentation

◆ get_worker() [1/2]

template<class W , class ... A>

cppthread::pool< W, A >::get_worker ( int i )

inline

This function returns a reference to the worker at index i.

Exceptions

std::range_error If the index is out of range (negative or larger or equal to the number of workers) then this exception is raised.

Parameters

[in] i The index of the worker to retrieve.

Returns: A reference to worker i.

Definition at line 134 of file pool.h.

References cppthread::pool< W, A >::f_workers.

◆ get_worker() [2/2]

template<class W , class ... A>

cppthread::pool< W, A >::get_worker ( int i ) const

inline

This function returns a reference to the worker at index i.

Exceptions

std::range_error If the index is out of range (negative or larger or equal to the number of workers) then this exception is raised.

Parameters

[in] i The index of the worker to retrieve.

Returns: A reference to worker i.

Definition at line 143 of file pool.h.

References cppthread::pool< W, A >::f_workers.

◆ pop_front()

template<class W , class ... A>

cppthread::pool< W, A >::pop_front	(	work_load_type &	v,
		std::int64_t	usecs
	)

inline

This function retrieves one T object from the output FIFO.

The usecs parameter can be set to -1 to wait until output is received. Set it to 0 to avoid waiting (check for data, if nothing return immediately) and a positive number to wait up to that many microseconds.

Make sure to check the function return value. If false, the v parameter is unmodified.

Once you called the stop() function, the pop_front() function can still be called until the out queue is emptied. The proper sequence is to

Note: The output fifo pointer can be set to nullptr. In that case, this function always returns false.

Parameters

[in]	v	A reference where the object gets copied.
[in]	usecs	The number of microseconds to wait.

Returns: true if an object was retrieved.

See also: snap_mutex::pop_front()

Definition at line 157 of file pool.h.

References cppthread::pool< W, A >::f_in, and cppthread::pool< W, A >::f_out.

◆ push_back()

template<class W , class ... A>

cppthread::pool< W, A >::push_back ( work_load_type const & v )

inline

This function adds a work load of data to the input. One of the worker threads will automatically pick up that work and have its snap_worker::do_work() function called.

Parameters

[in] v The work load of data to add to this pool.

See also: pop_front()

Definition at line 152 of file pool.h.

References cppthread::pool< W, A >::f_in.

◆ size()

template<class W , class ... A>

cppthread::pool< W, A >::size ( ) const

inline

This function returns the number of workers this thread pool is handling. The number is at least 1 as a thread pool can't currently be empty.

Returns: The number of workers in this thread pool.

Definition at line 129 of file pool.h.

References cppthread::pool< W, A >::f_workers.

◆ stop()

template<class W , class ... A>

cppthread::pool< W, A >::stop ( bool immediate )

inline

This function is called to ask all the threads to stop.

When the immediate parameter is set to true, whatever is left in the queue gets removed. This means you are likely to get invalid or at least incomplete results in the output. It should only be used if the plan is to cancel the current work process and trash everything away.

After a call to the stop() function, you may want to retrieve the last bits of data with the pop_front() function until it returns false and then call the join() function to wait on all the threads and call pop_front() again to make sure that you got all the output.

You may also call the join() function right after stop() and then the pop_front().

Note: It is not necessary to call pop_front() if you are cancelling the processing anyway. You can just ignore that data and it will be deleted as soon as the thread pool gets deleted.; The function can be called any number of times. It really only has an effect on the first call, though.

Parameters

[in] immediate Whether to clear the remaining items on the queue.

Definition at line 170 of file pool.h.

References cppthread::pool< W, A >::f_in.

Referenced by cppthread::pool< W, A >::~pool().

Here is the caller graph for this function:

◆ wait()

template<class W , class ... A>

cppthread::pool< W, A >::wait ( )

inline

This function waits on all the worker threads until they all exited. This ensures that all the output (see the pop_front() function) was generated and you can therefore end your processing.

The order of processing is as follow:

cppthread_pool my_pool;
 
...initialization...
 
data_t in;
data_t out;
 
while(load_data(in))
{
    pool.push_back(in);
 
    while(pool.pop_front(out))
    {
        // handle out, maybe:
        save_data(out);
    }
}
 
// no more input
pool.stop();
 
// optionally, empty the output pipe before the wait()
while(pool.pop_front(out))
{
    // handle out...
}
 
// make sure the workers are done
pool.wait();
 
// make sure the output is all managed
while(pool.pop_front(out))
{
    // handle out...
}
 
// now we're done

Emptying the output queue between the stop() and wait() is not required. It may always be empty or you will anyway get only a few small chunks that can wait in the buffer.

Note: The function can be called any number of times. After the first time, though, the vector of worker threads is empty so really nothing happens.