C++14 Thread Pool

Question

I have written the following thread pool implementation in C++14.

It appears to work fine on my system, but I am looking for a second opinion on the thread safety of my implementation and any other possible implementation flaws or concerns.

#include <atomic>
#include <condition_variable>
#include <exception>
#include <functional>
#include <mutex>
#include <queue>
#include <thread>
#include <tuple>
#include <utility>
#include <vector>


template<typename... event_args>
class thread_pool{
public:

    using handler_type = std::function<void(event_args...)>;

    thread_pool(handler_type&& handler, std::size_t N = 4): _handler(std::forward<handler_type&&>(handler)),_workers(N),_running(true)
    {
        for(auto&& worker: _workers)
        {
            worker = std::thread([this]()
            {
                while (_running)
                {
                    std::unique_lock<std::mutex> _lk{_wait_mutex};

                    _cv.wait(_lk, [this]{
                        return !_events.empty();
                    });

                    //a copy must be done here
                    auto data = _events.front();
                    //becauce the pop will invalidate any reference that we keep instead of copying
                    _events.pop();

                    _lk.unlock();

                    _cv.notify_all();

                    //call the handler with the data
                    invoke(std::move(_handler), std::move(data));

                }
            });

            worker.detach();
        }
    }

    ~thread_pool()
    {
        _running=false;

        _cv.notify_all();

        for(auto&& _worker: _workers)
        {
            if(_worker.joinable())
            {
                _worker.join();
            }
        }
    }

    handler_type& handler()
    {
        return _handler;
    }

    void propagate(event_args&&... args)
    {
        //lock before push
        std::unique_lock<std::mutex> _lk(_push_mutex);
        {
            _events.emplace(std::make_tuple(args...));
        }
        _lk.unlock();//explicit unlock
        _cv.notify_all();//let worker know that data is available
    }

private:

    handler_type _handler;

    std::queue<std::tuple<event_args...>> _events;

    std::vector<std::thread> _workers;

    std::atomic_bool _running;

    std::condition_variable _cv;

    std::mutex _wait_mutex;

    std::mutex _push_mutex;


    //helpers used to unpack tuple into function call
    template<typename Func, typename Tuple, std::size_t... I>
    auto invoke_(Func&& func, Tuple&& t, std::index_sequence<I...>)
    {
        return func(std::get<I>(std::forward<Tuple&&>(t))...);
    }

    template<typename Func, typename Tuple, typename Indicies = std::make_index_sequence<std::tuple_size<Tuple>::value>>
    auto invoke(Func&& func, Tuple&& t)
    {
        return invoke_(std::forward<Func&&>(func), std::forward<Tuple&&>(t), Indicies());
    }
};

With an example main:

#include <iostream>

int main(int argc, const char * argv[]) {

    std::mutex writemtx;


    thread_pool<int> pool{
        [&](int i){
            std::unique_lock<std::mutex> lk{writemtx};
            std::cout<<i<<" : "<<std::this_thread::get_id()<<std::endl;
        }
    };

    for (int i=0; i<16; ++i) {
        pool.propagate(std::move(i));
    }


    std::this_thread::sleep_for(std::chrono::seconds(2));
    return 0;
}

Which on my system produces an output of:

0 : 0x70000c42d000
1 : 0x70000c533000
2 : 0x70000c5b6000
3 : 0x70000c4b0000
4 : 0x70000c42d000
5 : 0x70000c533000
6 : 0x70000c5b6000
7 : 0x70000c4b0000
8 : 0x70000c42d000
9 : 0x70000c533000
10 : 0x70000c5b6000
11 : 0x70000c4b0000
12 : 0x70000c42d000
13 : 0x70000c533000
14 : 0x70000c5b6000
15 : 0x70000c4b0000

Answer 1

The destructor simply won't work reliably:

~thread_pool()
{
    _running=false;

    _cv.notify_all();

    for(auto&& _worker: _workers)
    {
        if(_worker.joinable())
        {
            _worker.join();
        }
    }
}

Let's have a look at your worker threads to see why:

while (_running)
{
    std::unique_lock<std::mutex> _lk{_wait_mutex};

    _cv.wait(_lk, [this]{
        return !_events.empty();
    });
    ...
}

Even when you set _running=false, any worker thread waiting on _cv can't exit the wait as the wait condition won't be fulfilled.

Only threads which were already inside _handler at the time of invocation can even reach the loop condition and hence terminate.

Which means your destructor can get caught in a deadlock.

---

Well, at least it could if you hadn't called worker.detach(). Because of that, joinable() now returns false.

Instead of a deadlock you now have something far worse: Use after free!

As the thread pool is destroyed, all the mutexes, _cv, _events, _handler and _running are destroyed despite being still in use by the worker threads.

The next allocation reusing these memory sections, will result in a bloody mess.

---

std::unique_lock<std::mutex> _lk(_push_mutex);
_events.emplace(std::make_tuple(args...));

VS

std::unique_lock<std::mutex> _lk{_wait_mutex};
...
auto data = _events.front();
_events.pop();

You must use the same mutex to protect access to _events, right now it's not even guaranteed that changes to _event propagate at all.

---

void propagate(event_args&&... args)
{
    ....
    _cv.notify_all()
}

Why notify_all()?

Only one work item was added, so notify_onesuffices completely. Any additional worker threads waking on the notification will just have to go back to sleep as _events is already empty again once they obtain the mutex.

---

---

So in in summary with the proposed fixes included:

template<typename... event_args>
class thread_pool{
public:

    using handler_type = std::function<void(event_args...)>;

    thread_pool(handler_type&& handler, std::size_t N = 4) :
        _handler(std::forward<handler_type&&>(handler)),
        _workers(N),
        _running(true)
    {
        for(auto&& worker: _workers)
        {
            //worker function
            worker = std::thread([this]()
            {
                while (_running)
                {
                    std::unique_lock<std::mutex> _lk(_events_mutex);

                    //wait for work
                    _cv.wait(_lk, [this]{
                        return !_events.empty() || !_running;
                    });
                    //_lk is locked

                    //check to see why we woke up
                    if(!_running){//was it a signal to exit
                        break;
                    }

                    if (!_events.empty()) {//was it new work
                        auto data = _events.front();
                        _events.pop();

                        //Unlock _lk after changes to _events are done
                        _lk.unlock();

                        invoke(std::move(_handler), std::move(data));
                    }
                }
            });
            //end worker function
        }
    }

    ~thread_pool()
    {
        _running=false;

        //let all workers know to exit
        _cv.notify_all();

        //attempt to join all workers
        for(auto&& _worker: _workers)
        {
            if(_worker.joinable())
            {
                _worker.join();
            }
        }
    }

    handler_type& handler()
    {
        return _handler;
    }

    void propagate(event_args&&... args)
    {
        {
            //lock before push
            std::unique_lock<std::mutex> _lk(_events_mutex);
            _events.emplace(std::make_tuple(args...));
        }
        _cv.notify_one();//let worker know that data is available
    }

private:

    handler_type _handler;

    std::queue<std::tuple<event_args...>> _events;

    std::mutex _events_mutex;

    std::vector<std::thread> _workers;

    std::atomic_bool _running;

    std::condition_variable _cv;


    //helpers used to unpack tuple into function call
    template<typename Func, typename Tuple, std::size_t... I>
    auto invoke_(Func&& func, Tuple&& t, std::index_sequence<I...>)
    {
        return func(std::get<I>(std::forward<Tuple&&>(t))...);
    }

    template<typename Func, typename Tuple, typename Indicies = std::make_index_sequence<std::tuple_size<Tuple>::value>>
    auto invoke(Func&& func, Tuple&& t)
    {
        return invoke_(std::forward<Func&&>(func), std::forward<Tuple&&>(t), Indicies());
    }
};

Answer 2

Based on the answer from Ext3h I have modified my code to the following:

template<typename... event_args>
class thread_pool{
public:

    using handler_type = std::function<void(event_args...)>;

    thread_pool(handler_type&& handler, std::size_t N = 4): _handler(std::forward<handler_type&&>(handler)),_workers(N),_running(true)
    {
        for(auto&& worker: _workers)
        {
            //worker function
            worker = std::thread([this]()
            {
                while (_running)
                {
                    //wait for work
                    std::unique_lock<std::mutex> _lk{_wait_mutex};
                    _cv.wait(_lk, [this]{
                        return !_events.empty() || !_running;
                    });
                    //_lk unlocked

                    //check to see why we woke up
                    if (!_events.empty()) {//was it new work
                        std::unique_lock<std::mutex> _readlk(_queue_mutex);
                        auto data = _events.front();
                        _events.pop();
                        _readlk.unlock();

                        invoke(std::move(_handler), std::move(data));
                    }else if(!_running){//was it a signal to exit
                        break;
                    }
                    //or was it spurious and we should just ignore it
                }
            });
            //end worker function
        }
    }

    ~thread_pool()
    {

        _running=false;

        //let all workers know to exit
        _cv.notify_all();


        //attempt to join all workers
        for(auto&& _worker: _workers)
        {
            if(_worker.joinable())
            {
                _worker.join();
            }
        }
    }

    handler_type& handler()
    {
        return _handler;
    }

    void propagate(event_args&&... args)
    {
        //lock before push
        std::unique_lock<std::mutex> _lk(_queue_mutex);
        {
            _events.emplace(std::make_tuple(args...));
        }
        _lk.unlock();//explicit unlock
        _cv.notify_one();//let worker know that data is available
    }

private:

    handler_type _handler;

    std::queue<std::tuple<event_args...>> _events;

    std::vector<std::thread> _workers;

    std::atomic_bool _running;

    std::condition_variable _cv;

    std::mutex _wait_mutex;

    std::mutex _queue_mutex;


    //helpers used to unpack tuple into function call
    template<typename Func, typename Tuple, std::size_t... I>
    auto invoke_(Func&& func, Tuple&& t, std::index_sequence<I...>)
    {
        return func(std::get<I>(std::forward<Tuple&&>(t))...);
    }

    template<typename Func, typename Tuple, typename Indicies = std::make_index_sequence<std::tuple_size<Tuple>::value>>
    auto invoke(Func&& func, Tuple&& t)
    {
        return invoke_(std::forward<Func&&>(func), std::forward<Tuple&&>(t), Indicies());
    }
};

Changes:

• Removed call to detach()

• Updated wait condition to watch for changes in _running

• Added condition check to make sure new data is available before trying to do work.
• Changed notify_all() to notify_one() in propagate()