C++ 14 thread pool executor design

Question

With a few experience of python and golang, I tried to make (simple) thread pool executor. Tasks of the executor must be copy-constructible std::function<void()> and the result of tasks are required in the main thread.

My Header and Implementation File:

#include <iostream>
#include <queue>
#include <chrono>
#include <atomic>
#include <thread>
#include <mutex>
#include <shared_mutex>
#include <functional>

class PoolExecutor
{
public:
    explicit PoolExecutor(const unsigned int pool_size)
        : stop_(false)
        , working_(0)
    {
        for(unsigned int i = 0; i < pool_size; ++i)
        {
            this->threads_.emplace_back(
                std::thread(std::bind(&PoolExecutor::worker, this)));
        }

    }
    virtual ~PoolExecutor()
    {
        this->stop_ = true;
        for(auto& thread : this->threads_)
        {
            thread.join();
        }
    }

    void submit(std::function<void()> func)
    {
        {
            std::lock_guard<std::mutex> guard(this->mu2_);
            this->work_queue_.push(func);
        }
        this->cv_.notify_all();
    }

    bool finished()
    {
        std::lock_guard<std::mutex> guard(this->mu2_);
        return this->working_ == 0 && this->work_queue_.empty();
    }

    void stop()
    {
        this->stop_ = true;
    }

private:
    std::vector<std::thread> threads_;
    std::shared_timed_mutex mu_;
    std::mutex mu2_;
    std::atomic_bool stop_;
    std::atomic_int working_;
    std::condition_variable_any cv_;
    std::queue<std::function<void()> > work_queue_;

    void worker()
    {
        std::function<void()> func;
        while(!this->stop_)
        {
            using namespace std::literals::chrono_literals;
            std::shared_lock<std::shared_timed_mutex> lock(this->mu_, std::try_to_lock);
            this->cv_.wait_for(lock, 10ms);

            bool accquired = false;
            {
                std::lock_guard<std::mutex> guard(this->mu2_);
                if(!this->work_queue_.empty())
                {
                    func = this->work_queue_.front();
                    this->work_queue_.pop();
                    accquired = true;
                    this->working_++;
                }
            }
            if(accquired)
            {
                func();
                --this->working_;
            }
        }
    }
};

Sample Usage:

int main()
{
    PoolExecutor executor(8);
    std::mutex mu;
    int sum = 0;

    auto func = [&sum, &mu](const int i) {
        std::cout << "work " << i << " try lock\n";
        {
            //simulate long running function
            using namespace std::literals::chrono_literals;
            std::lock_guard<std::mutex> guard(mu);
            std::this_thread::sleep_for(10ms);
            std::cout << "work " << i << " acquire lock\n";
            sum += i;
        }
    };

    for(auto i = 0; i < 100; ++i)
    {
        executor.submit(std::bind(func, i));
    }

    int count = 0;
    while(!executor.finished())
    {
        using namespace std::literals::chrono_literals;
        std::this_thread::sleep_for(200ms);
        std::cout << "sleep count: " << ++count << "\n";
    }

    std::cout << "sum: " << sum << "\n";
    return 0;
}

I've tested with sample code using MSVC 15 (also tested with shared_mutex instead of shared_timed_mutex) and clang 3.8. The result is printed as expected, but I cannot sure what I did it.

Some questions.

Q1. Am I using mutex and lock correctly?

Q2. It is possible to use single mutex in PoolExecutor class?

Q3. Is there any ways to wait executor until finished?

Answer 1

this->threads_.emplace_back(
    std::thread(std::bind(&PoolExecutor::worker, this)));

Creative use of std::bind, but a lambda would have sufficed:

this->threads_.emplace_back(
    std::thread([this] () { this->worker() });

---

void submit(std::function<void()> func)
{
    this->cv_.notify_all();
}

Why bother waking all worker threads when only a single slice of work was added? std::condition_variable_any::notify_one() is more appropriate.

---

void stop()
{
    this->stop_ = true;
}

Here, on the other hand, std::condition_variable_any::notify_all() should have been used, wake the threads currently sleeping.

void stop()
{
    this->stop_ = true;
    this->cv_.notify_all()
}

---

std::shared_lock<std::shared_timed_mutex> lock(this->mu_, std::try_to_lock);
this->cv_.wait_for(lock, 10ms);

This part only works as expected for the first thread, as it actually gets the lock. The second thread fails to obtain the lock.

See the documentation for std::condition_variable_any::wait_for. The lock must be locked before entering wait_for.

That means you can't use std::try_to_lock, but actually need std::lock.

---

While at it, you should get used to using the optional 3rd parameter on wait to pass in a predicate in order to filter spurious wakes:

std::shared_lock<std::shared_timed_mutex> lock(this->mu_, std::lock);
this->cv_.wait_for(lock, 10ms, [this] () {return this->stop_ || !this->work_queue_.empty());

That predicate is executed before waiting for the first time, so if e.g. this->stop_ is already set by the time lock is aquired, wait() with the predicate returns immediately without requiring another wake up call.

---

At this point you can do away with the "timed" stuff as well. You no longer need it. You can now also replace your std::condition_variable_any by the simpler std::condition_variable.

In fact, neither do you need the 2nd mutex either:

std::mutex mu_;
std::condition_variable cv_;

...

bool finished()
{
    std::lock_guard<std::mutex> guard(this->mu_);
    return this->working_ == 0 && this->work_queue_.empty();
}

void worker()
{
    while(!this->stop_)
    {
        std::function<void()> func;
        {
            std::unique_lock<std::mutex> lock(this->mu_);
            this->cv_.wait(lock, [this] () { return this->stop_ || !this->work_queue_.empty() );
            if(this->stop_) break;
            if(!this->work_queue_.empty())
            {
                func = std::move(this->work_queue_.front());
                this->work_queue_.pop();
                this->working_++;
            }
        }
        if(func)
        {
            func();
            --this->working_;
        }
    }
}

---

Q1. Am I using mutex and lock correctly?

No. Even though it might have even worked, you violated the contract by attempting to wait on an unlocked mutex.

Q2. It is possible to use single mutex in PoolExecutor class?

As seen above, yes.

Q3. Is there any ways to wait executor until finished?

If by "finished" you mean the executor is ready to be destroyed? That is already the case.

Otherwise if you just mean to wait until the pool runs empty? Well, you do have std::atomic_int working_ in there.

All you really need now, is a second condition variable:

void wait() {
    std::unique_lock<std::mutex> lock(this->mu_);
    this->cv2_.wait(lock, [this] () {
        return ( this->working_ == 0 && this->work_queue_.empty() ) || this->stop_
    });
}

Now update the worker to invoke this->cv2_.notify_all() after decrementing this->working_ and you are done.

Answer 2

The condition variable will automatically release the lock passed in while it's waiting. This means that you don't need a second mutex:

void worker()
{
    std::function<void()> func;
    while(!this->stop_)
    {
        using namespace std::literals::chrono_literals;

        bool accquired = false;
        {
            std::lock_guard<std::mutex> guard(this->mu2_);
            this->cv_.wait_for(this->mu2_, 10ms, [&](){
               return this->stop_ || this->work_queue_.empty();});
            if(!this->work_queue_.empty())
            {
                func = std::move(this->work_queue_.front());
                this->work_queue_.pop();
                accquired = true;
                this->working_++;
            }
        }
        if(accquired)
        {
            func();
            --this->working_;
        }
    }
}

To wait for all worker to finish you can join the threads (this will wait until the thread is finished) like you do in the destructor. You can check the joinable property of std::thread to see if it's already been joined or add another boolean flag:

void stop_and_wait_for_finish(){
    stop_ = true;
    if(finished_)return; //don't join threads twice


    for(auto& thread : this->threads_)
    {
        thread.join();
    }
    finished_ = true;

}

and in the destructor you simply call stop_and_wait_for_finish();