1
\$\begingroup\$

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
\$\endgroup\$
1
\$\begingroup\$

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());
    }
};
\$\endgroup\$
0
\$\begingroup\$

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()
\$\endgroup\$
  • \$\begingroup\$ That's not true: //_lk unlocked. _lk is unlocked while waiting, but it's guaranteed to be locked again once exiting the wait. There are no spurious wakes either, when using the predicate parameter for wait(). Using _queue_mutex and _wait_mutex is still wrong! As I said before, changes to _events made in propagate under _queue_mutex are not guaranteed to be visible when synchronizing the condition variable on _wait_mutex instead. \$\endgroup\$ – Ext3h Oct 28 '16 at 6:35
  • \$\begingroup\$ So with your changes, there is now effectively only a single worker active at a time, as _lk is locked once the predicate is fulfilled, and it remains locked until the scope is exited, which is only after invoke() returns. \$\endgroup\$ – Ext3h Oct 28 '16 at 6:40
  • \$\begingroup\$ Would you be able to provide me with an example of the correct way to implement what you are describing? I am having a tough time trying to follow your explanations \$\endgroup\$ – Alex Zywicki Oct 28 '16 at 6:43
  • \$\begingroup\$ From what I am gathering from your comments, I should use the same mutex for propagate and for the wait lock? \$\endgroup\$ – Alex Zywicki Oct 28 '16 at 6:45
  • \$\begingroup\$ Correct. Perhaps it becomes clearer when you just call it _events_mutex, as the role of the mutex is to avoid data races on a specific data structure, in this case _events. You need to obtain a lock on that mutex for all accesses to _events, regardless of whether you are writing or just reading. \$\endgroup\$ – Ext3h Oct 28 '16 at 6:59

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.