vit-vit/ctpl

Modern and efficient C++ Thread Pool Library

CTPL

Modern and efficient C++ Thread Pool Library

A thread pool is a programming pattern for parallel execution of jobs, http://en.wikipedia.org/wiki/Thread_pool_pattern.

More specifically, there are some threads dedicated to the pool and a container of jobs. The jobs come to the pool dynamically. A job is fetched and deleted from the container when there is an idle thread. The job is then run on that thread.

A thread pool is helpful when you want to minimize time of loading and destroying threads and when you want to limit the number of parallel jobs that run simultanuasly. For example, time consuming event handlers may be processed in a thread pool to make UI more responsive.

Features:

• standard c++ language, tested to compile on MS Visual Studio 2013 (2012?), gcc 4.8.2 and mingw 4.8.1(with posix threads)
• simple but effiecient solution, one header only, no need to compile a binary library
• query the number of idle threads and resize the pool dynamically
• one API to push to the thread pool any collable object: lambdas, functors, functions, result of bind expression
• collable objects with variadic number of parameters plus index of the thread running the object
• automatic template argument deduction
• get returned value of any type with standard c++ futures
• get fired exceptions with standard c++ futures
• use for any purpose under Apache license
• two variants, one depends on Boost Lockfree Queue library, http://boost.org, which is a header only library

Sample usage

void first(int id) { std::cout << "hello from " << id << '\n'; }

struct Second { void operator()(int id) const { std::cout << "hello from " << id << '\n'; } } second;

void third(int id, const std::string & additional_param) {}

int main () {

ctpl::thread_pool p(2 /* two threads in the pool */);

p.push(first); // function

p.push(third, "additional_param");

p.push( [] (int id){ std::cout << "hello from " << id << '\n'; }); // lambda

p.push(std::ref(second)); // functor, reference

p.push(const_cast<const Second &>(second)); // functor, copy ctor

p.push(std::move(second)); // functor, move ctor

}

ctpl's Issues

Trivial: Collable?

The README.md uses "collable" with an O twice, and that seems like a typo for "callable". I had some reservation that this was a specific technical term with subtle meaning, but Googling seems to suggest all uses are typos.

std::unique_lock shadow bug

CTPL/ctpl_stl.h

Line 48 in 437e135

std::unique_lock<std::mutex> lock(this->mutex);

CTPL/ctpl_stl.h

Line 54 in 437e135

std::unique_lock<std::mutex> lock(this->mutex);

CTPL/ctpl.h

Line 137 in 437e135

std::unique_lock<std::mutex> lock(this->mutex);

Hi,
you have a shadow bug in the link above.
It is caused by the default constructor of std::unique_lock. You are not locking your code.

See this video on why it is a bug: https://youtu.be/lkgszkPnV8g?t=2068.

Solution to the bug: Just name it.

Best regards,
Jack

Problem in passing my function to "push" method

Hello there,

So, i have a function with this definition:
static void Invoke( int id, std::unique_ptr<BaseService> svc );

And tried to pass it to ctpl "push" method to be queued in thread-pool:
pThreadPool->push( std::ref(App::Invoke), std::move( svc ) );

But I get this error:

/home/hadi/CLionProjects/App/App.cpp:211:27: error: no matching member function for call to 'push'
    pThreadPool->push( std::ref(App::Invoke), std::move( svc ) );
    ~~~~~~~~~~~~~^~~~
/home/hadi/CLionProjects/App/include/cptl/ctpl.h:152:14: note: candidate template ignored: substitution failure [with F = std::__1::reference_wrapper<void (int, std::__1::unique_ptr<BaseService, std::__1::default_delete<BaseService> >)>, Rest = <std::__1::unique_ptr<BaseService, std::__1::default_delete<BaseService> >>]: call to implicitly-deleted copy constructor of 'std::__1::unique_ptr<BaseService, std::__1::default_delete<BaseService> >'
        auto push(F && f, Rest&&... rest) ->std::future<decltype(f(0, rest...))> {
             ^                                                        ~~~~
/home/hadi/CLionProjects/App/include/cptl/ctpl.h:171:14: note: candidate function template not viable: requires single argument 'f', but 2 arguments were provided
        auto push(F && f) ->std::future<decltype(f(0))> {
             ^
1 error generated.

How to pass function of classe

I have this class

 class MyTask
 {
 public:
 	// Function to be executed by thread function
 	void run()
 	{
 		std::cout << "Task Start" << std::endl;
 
 		// Check if thread is requested to stop ?
 		while (stopRequested() == false)
 		{
 			std::cout << "Doing Some Work" << std::endl;
 			std::this_thread::sleep_for(std::chrono::milliseconds(1000));
 
 		}
 		std::cout << "Task End" << std::endl;
 	}
 };

I do
MyTask task; ctpl::thread_pool testThreadPool(4);

How can I push task.run() to ctpl::thread_pool?
testThreadPool.push(task.run);, isn't correct

Why does the function with arguments have errors?

Error C2672: "CTPL ::thread_pool::push" : no matching overloaded function found
Error C2780: "STD ::future : CTPL ::thread_pool::push(F &&)" : one parameter should be input, but two are provided
Error C2893: function template "STD ::future CTPL ::thread_pool::push(F &&,Rest &&...) "specialty
Error C2672: "STD ::vector< STD ::future, STD ::allocator<_Ty> : >::emplace_back" : overloaded function not found

#include "ctpl_stl.h"
#include "iostream"
#include "string"

void f(int j)
{
std::cout << j << std::endl;

}

int main(int argc, char **argv) {

ctpl::thread_pool p(8 /* two threads in the pool */);

std::vector< std::future<int> > results;

for (int i = 0; i < 8; ++i) {
	results.emplace_back(
		p.push(f, i));
}

for (auto && result : results)
	std::cout << result.get() << ' ';
std::cout << std::endl;
getchar();

return 0;

}

How to get threads details and termination?

I have some questions, maybe someone could help.
How to find out how many threads are working at the moment?
How know if all threads are terminated?
How to terminate a thread from inside of thread and from outside (from another thread)?

Thanks in advance.

Pushing lambdas to thread pool

TimerTask.cc: In lambda function:
TimerTask.cc:93:38: error: no matching function for call to ‘ctpl::thread_pool::push(TimerTask::TimerTask()::__lambda8::__lambda10)’
pool.push( { sleep(10); });
^
TimerTask.cc:93:38: note: candidates are:
In file included from TimerTask.cc:16:0:
ctpl.h:152:14: note: template<class F, class ... Rest> std::future<decltype (f(0, ctpl::thread_pool::push::rest ...))> ctpl::thread_pool::push(F&&, Rest&& ...)
auto push(F && f, Rest&&... rest) ->std::future<decltype(f(0, rest...))> {
^
ctpl.h:152:14: note: template argument deduction/substitution failed:
ctpl.h: In substitution of ‘template<class F, class ... Rest> std::future<decltype (f(0, ctpl::thread_pool::push::rest ...))> ctpl::thread_pool::push(F&&, Rest&& ...) [with F = TimerTask::TimerTask()::__lambda8::__lambda10; Rest = {}]’:
TimerTask.cc:93:38: required from here
ctpl.h:152:78: error: no match for call to ‘(TimerTask::TimerTask()::__lambda8::__lambda10) (int)’
auto push(F && f, Rest&&... rest) ->std::future<decltype(f(0, rest...))> {
^
TimerTask.cc:93:20: note: candidates are:
pool.push( { sleep(10); });
^
In file included from TimerTask.cc:16:0:
ctpl.h:152:78: note: void ()()
auto push(F && f, Rest&&... rest) ->std::future<decltype(f(0, rest...))> {
^
ctpl.h:152:78: note: candidate expects 1 argument, 2 provided
TimerTask.cc:93:22: note: TimerTask::TimerTask()::__lambda8::__lambda10
pool.push( { sleep(10); });
^
TimerTask.cc:93:22: note: candidate expects 0 arguments, 1 provided
In file included from TimerTask.cc:16:0:
ctpl.h:171:14: note: template std::future<decltype (f(0))> ctpl::thread_pool::push(F&&)
auto push(F && f) ->std::future<decltype(f(0))> {
^
ctpl.h:171:14: note: template argument deduction/substitution failed:
ctpl.h: In substitution of ‘template std::future<decltype (f(0))> ctpl::thread_pool::push(F&&) [with F = TimerTask::TimerTask()::__lambda8::__lambda10]’:
TimerTask.cc:93:38: required from here
ctpl.h:171:53: error: no match for call to ‘(TimerTask::TimerTask()::__lambda8::__lambda10) (int)’
auto push(F && f) ->std::future<decltype(f(0))> {
^
TimerTask.cc:93:20: note: candidates are:
pool.push( { sleep(10); });
^
In file included from TimerTask.cc:16:0:
ctpl.h:171:53: note: void ()()
auto push(F && f) ->std::future<decltype(f(0))> {
^
ctpl.h:171:53: note: candidate expects 1 argument, 2 provided
TimerTask.cc:93:22: note: TimerTask::TimerTask()::__lambda8::__lambda10
pool.push( { sleep(10); });
^
TimerTask.cc:93:22: note: candidate expects 0 arguments, 1 provided

Function call that should not compile is able to be passed into push()

The following program demonstrates the issue. This is using the STL version.

struct A {};

void f(int, A&) {}

int main()
{
	ctpl::thread_pool pool(4);
	
	A a;
	
	// Does not compile
	//f(0,std::move(a));
	
	// So this shouldn't be expected to compile either, yet it does
	//pool.push(f, std::move(a));
}

Possible thread issue?

It appears the mutex is being locked after pushing the data onto the queue, rather than before. Is this behavior correct?

    template<typename F, typename... Rest>
    auto push(F && f, Rest&&... rest) ->std::future<decltype(f(0, rest...))> {
        auto pck = std::make_shared<std::packaged_task<decltype(f(0, rest...))(int)>>(
            std::bind(std::forward<F>(f), std::placeholders::_1, std::forward<Rest>(rest)...)
        );

        auto _f = new std::function<void(int id)>([pck](int id) {
            (*pck)(id);
        });
        this->q.push(_f); // <-- see here...

        std::unique_lock<std::mutex> lock(this->mutex); // <-- ...and here
        this->cv.notify_one();

        return pck->get_future();
    }

error: invalid use of non-static member function

Hello,
I am trying to call a function call_from_thread_description from another function inside a class and I am getting the following error. I can't find the right to add a function to the pool.

Error: /home/hani/workspace/transform/Transform.cpp:127:104: error: invalid use of non-static member functiondr esults.push_back(thread_pool.push(Transform::call_from_thread_description, data[i], data[j], res));

void Transform::call_from_thread_description(const vector<string> & e1, const <string> & e2, <string>  & res)
{ 
//DO SOMETHING HERE
}

void Transform::computer_intersection()
{
    vector<string> res;
    int Num_Threads =  thread::hardware_concurrency();
    ctpl::thread_pool thread_pool(Num_Threads);
    for(int i = 0; i < data.size();i++)
    {
        data_sets tmp;
        for(int j = 0; j < data.size();j++)
        {
            results.push_back(thread_pool.push(call_from_thread_description, data[i], data[j], res));
            if(j % Num_Threads == 0 || j == data.size() -1)
            {
               results.clear();
            }
       }
  }
}

Small memory leak

First of all: thank you so much for this great piece of software!

I found a small memory leak in the stl version of your library, it is in line 233:

this->threads[i].reset(new std::thread(f));

As you can see, a new thread is created with new but it is never deleted. Valgrind gives me the following message:

==5111== 1,216 bytes in 4 blocks are possibly lost in loss record 1,514 of 1,551
==5111== at 0x4C31B25: calloc (in /usr/lib/valgrind/vgpreload_memcheck-amd64-linux.so)
==5111== by 0x40134A6: allocate_dtv (dl-tls.c:286)
==5111== by 0x40134A6: _dl_allocate_tls (dl-tls.c:530)
==5111== by 0x5293227: allocate_stack (allocatestack.c:627)
==5111== by 0x5293227: pthread_create@@GLIBC_2.2.5 (pthread_create.c:644)
==5111== by 0x5AD4A18: std::thread::_M_start_thread(std::unique_ptr<std::thread::_State, std::default_deletestd::thread::_State >, void (*)()) (in /usr/lib/x86_64-linux-gnu/libstdc++.so.6.0.25)
==5111== by 0x255CB8: std::thread::threadctpl::thread_pool::set_thread(int)::{lambda()#1}&(ctpl::thread_pool::set_thread(int)::{lambda()#1}&) (thread:126)
==5111== by 0x253E17: ctpl::thread_pool::set_thread(int) (ctpl_stl.h:233)
==5111== by 0x2536C5: ctpl::thread_pool::resize(int) (ctpl_stl.h:102)
==5111== by 0x25349A: ctpl::thread_pool::thread_pool(int) (ctpl_stl.h:76)

I don't think that this error is very important and should be fixed ASAP, but I just wanted to let you know :-)

Pushing a function to the pool

Hello,
Thank you so much for this library. I was wondering why I keep getting an error to push to the pool:
all_errors.cpp:244:25: error: no matching function for call to ‘ctpl::thread_pool::push(int)’

I am following your example to the tee and I am still getting this error. Any ideas?

reuse thread pool after calling stop(true) ?

Hi,

If I try to reuse the thread pool (using push) after calling stop(true), it doesn't create any new threads ? Something like the following sequence:

auto thread_pool  = new ctpl::thread_pool(2);
for (int i = 0; i < 10; i++)
{
    thread_pool->push(func, i);
}  
thread_pool->stop(true);
thread_pool->resize(4);
/* Wait a while */
thread_pool->push(func, i); // ==> func is not executed

Is this behavior intentional ? I have gotten around this by destroying the old pool and creating a new pool but that does not seem efficient.

Result: Process terminated due to timeout

Hi,
I am trying to run a simple program and I keep getting this error anytime I try and for loop >50. I was wondering if anyone could explain to me why that happens and how to fix it?

Resize thread pool

Hi,

In my code I create a thread pool with only one thread, and then push all the functions that I want to execute. After that, while the functions are being executed by that thread, I use another thread (that doesn't belong to the thread pool) to increase the number of threads in the thread pool, using the resize method. However, the resizing has no effect since the new threads I added do not execute anything, only the first thread (thread 0) executes the functions that are still in the queue.

Does this happen because when I push the functions they are automatically associated with a thread? And in my case, since when I push the functions there is only thread 0 in the thread pool, all functions become associated with that thread?

Also, is there a problem in calling resize from a function outside the thread pool?

Thank you in advance for your help!

Documentation of CTPL different versions

Thanks for the great library, is there an advantage to the boost lockfree queue version? I would prefer to use the STL version but wanted to understand what the anticipated tradeoff is.

How to wait for all Tasks to finish?

Sorry for writing this as an issue.

I could not figure out how it is ment to wait for all Tasks to finish.
If i use stop - the Threads get deleted and the threadpool ends.

Isn't there something like p.WaitForEverythingDone()?
Currently I use this in the main thread:
while (p.n_idle() < THREAD_COUNT )
{
}
But thats not really a good idea to burn 1 cpu core just for waiting?

Problem occurs when using multiple/nested function pointers and arguments

Because my functions do not conform to the required format, I create a helper function to solve this problem, namely

template<typename Fn, typename... Args>
inline void FuncWrapper(int, Fn &&fn, Args&&... args) {
	fn(forward<Args>(args)...);
}

.

That is, if I have a function, e.g.

void print(int num){cout<<num<<endl;}

, I should be able to push it into the thread pool by calling

using task_type = decltype(&print);
auto ptr = &FuncWrapper<task_type,int>;
pool.push(ptr,&print,num);

.

But this simply does not work, for some parts of the code lack the process of coping these arguments and applying "std::forward" to them.

Provide example for using class functions

Hi,
can you provide example how to use threadpool to call member function?

Refactor Push() to remove need for int argument

From using the CTPL library I have noticed that the fact that push() requires the first argument of the function being pushed makes using the library difficult at times.

In a way it forces anyone using the library to force their interface to conform to the need for that int parameter. This can be achieved by either conforming your interface to work with the pool or by wrapping your target function with another function that adds the leading int parameter.

I believe that it would make the library more accessible, transparent and user friendly if an overload of push() was provided that allowed you to push a function onto the pool without the need for the leading int parameter.

I would be willing to work up the changes and add them as a pull request if desired, but I wanted to run the idea by first.

Library crash

The issue is described here : http://stackoverflow.com/questions/30778799/exception-error-using-c-thread-pool-library-ctpl