C++11 Thread Pool

Question

I've been using a thread pool that I wrote using Windows threads for a while and I've decided to update it to use std::threads. I'm looking for some feedback on how this could be improved in terms of (in order of importance) reliability, usability, and performance.

Header

#pragma once

#include <queue>
#include <thread>
#include <vector>
#include <functional>

namespace std
{
    class mutex;
    class condition_variable;
}//namespace std

namespace Utilities
{
    class Semaphore final
    {
    private:
        std::condition_variable* _condition;
        std::mutex*              _mutex;
        unsigned                 _count;

    public:
        Semaphore();
        ~Semaphore();

    public:
        void Wait();
        void Notify();
    };

    class Worker final
    {
    public:
        typedef std::function<void()> Task;

    private: 
        std::queue<Task> _tasks;
        std::thread*     _thread;
        std::mutex*      _mutex;
        Semaphore        _semaphore;
        bool             _alive;

    public:
        Worker();
        ~Worker();

    public:
        std::thread::id id        () const;
        size_t          task_count() const;

        void Join      (         ) const;
        void PushTask  (Task task);
        void ClearTasks(         );
    };

    class ThreadPool final
    {
    private:
        std::vector<Worker> _workers;
        unsigned            _index;

    public:
        ThreadPool(unsigned thread_count = 0);
        ~ThreadPool();

    public:
        unsigned thread_count();
        unsigned task_count  ();

        void Join      (                 );
        void PushTask  (Worker::Task task);
        void ClearTasks(                 );
    };
}//namespace Utilities

Source

#include "ThreadPool.h"

#include <mutex>
#include <algorithm>
#include <condition_variable>

using namespace Utilities;

Semaphore::Semaphore() :
    _condition(new std::condition_variable()),
    _mutex    (new std::mutex             ()),
    _count    (0                            )
{ }

Semaphore::~Semaphore()
{
    delete _condition;
    delete _mutex;
}

void Semaphore::Wait()
{
    std::unique_lock<std::mutex> lock(*_mutex);
    while (!_count)
        _condition->wait(lock);
    --_count;
}//Wait

void Semaphore::Notify()
{
    std::unique_lock<std::mutex> lock(*_mutex);
    _condition->notify_one();
    ++_count;
}//Notify

Worker::Worker() :
    _mutex(new std::mutex()),
    _alive(true            )
{
    _thread = new std::thread(
    [&]
    {
        while (_alive)
        {
            _semaphore.Wait();

            while (!_tasks.empty())
            {
                _mutex->lock();
                Task task = _tasks.front();
                _mutex->unlock();

                task();

                _mutex->lock();
                if (!_tasks.empty())
                    _tasks.pop();
                _mutex->unlock();
            }
        }
    });
}

Worker::~Worker()
{
    ClearTasks();
    _alive = false;
    _semaphore.Notify();

    _thread->join();

    delete _thread;
    delete _mutex;
}

std::thread::id Worker::id() const
{ return _thread->get_id(); }
//id

size_t Worker::task_count() const
{ return _tasks.size(); }
//task_count

void Worker::Join() const
{ while (_tasks.size()); }
//Join

void Worker::PushTask(Task task)
{
    _mutex->lock();
    _tasks.push(task);
    _mutex->unlock();

    _semaphore.Notify();
}//PushTask

void Worker::ClearTasks()
{
    _mutex->lock();
    _tasks = std::queue<Task>();
    _mutex->unlock();
}//ClearTasks

ThreadPool::ThreadPool(unsigned thread_count) :
    _index(0)
{
    if (!thread_count)
        thread_count = std::thread::hardware_concurrency();
    _workers = std::vector<Worker>(std::max(1u, thread_count));
}

ThreadPool::~ThreadPool()
{
    ClearTasks();
    Join      ();
}

unsigned ThreadPool::thread_count()
{ return _workers.size(); }
//thread_count

unsigned ThreadPool::task_count()
{
    unsigned task_count = 0;
    for (auto& worker : _workers)
        task_count += worker.task_count();
    return task_count;
}//task_count

void ThreadPool::Join()
{ while (task_count()); }
//Join

void ThreadPool::PushTask(Worker::Task task)
{
    _workers[_index++].PushTask(task);
    if (_index >= _workers.size())
        _index = 0;
}//PushTask

void ThreadPool::ClearTasks()
{
    for (auto& worker : _workers)
        worker.ClearTasks();
}//ClearTasks

Usage Example

#include "ThreadPool.h"

#include <iostream>

static void Method_0(              ) { std::cout << "Method_0()" << std::endl; }
static void Method_1(float my_float) { std::cout << my_float     << std::endl; }
static void Method_2(int&  my_int  ) { my_int += my_int;                       }

struct Foo
{
           void Bar() { std::cout << "Foo::Bar() : Instance" << std::endl; }
    static void Baz() { std::cout << "Foo::Baz() : Static"   << std::endl; }
};

int main(int argc, char* argv[])
{
    using namespace Utilities;

    Foo foo;
    int my_int = 16;

    ThreadPool thread_pool;
    thread_pool.PushTask(Method_0);
    thread_pool.PushTask(std::bind(Method_1, 3.14f));
    thread_pool.PushTask(std::bind(Method_2, std::ref(my_int)));
    thread_pool.PushTask(std::bind(&Foo::Bar, foo));
    thread_pool.PushTask(&Foo::Baz);
    thread_pool.PushTask([]{ std::cout << "Lambda" << std::endl; });
    thread_pool.Join();

    std::cout << my_int << std::endl;

    return 0;
}

Answer 1

Miscellaneous:

I imagine you did this:

namespace std
{
    class mutex;
    class condition_variable;
}//namespace std

To have a forward declaration of those types and avoid the #includes. I don't like this though. There is no guarantee that mutex or condition_variable are plain class types. They could be anything, structs, template classes, or even typedefs to anything else. So this is certainly not portable across compilers and might even break on a future update of your current compiler. Including the relevant files in the header is cleaner, safer and also simpler.

Also refer to this question.

---

The busy wait loop in Worker::Join() could take advantage of std::this_thread::yield(), which might perform better than just spinning.

void Worker::Join() const
{ 
    while (_tasks.size())
    {
        std::this_thread::yield();
    }
}

---

Those comments at the end of every other method are plain noise.

void ThreadPool::ClearTasks()
{
    for (auto& worker : _workers)
        worker.ClearTasks();
}//ClearTasks

Your methods are not that big, I can see where they begin and end very easily without that comment.

---

While not technically wrong in the context you've used, names starting with an underscore bare a few restrictions in C++. FYI, I suggest reading this SO question.

---

Copy and assignment:

As it stands, your classes can be copied using operator = or the implicit copy constructor. Doing so would wreak havoc in your program, since each copy would be a shallow copy and multiple attempts to delete the same pointers would be done by the destructors of different objects. You should at the very least delete the copy constructor and assignment operator of Semaphore and Worker, and perhaps also from Thread. E.g.:

class Semaphore final
{
    Semaphore(const Semaphore &) = delete;
    Semaphore & operator = (const Semaphore &) = delete;
    // copy and assignment are now disallowed.
};

This falls into what is known as Rule of Three, which means that if a construtor does some non-trivial work, such as memory allocation, then more than likely you'll need to define custom copy constructor and assignment as well (or disable them).

---

Memory Management:

In the Semaphore class, condition_variable and mutex are declared as pointers. I would declare them by value to avoid extra memory allocations and improve data locality. I don't see any obvious reason for them to be pointers. If you do keep the pointers and dynamic allocations, then consider using a std::unique_ptr rather than manual new and delete calls, to ensure exception safety and simplify the resource management logic. Same applies to thread and mutex in Worker. Those could be declared by value as well. Otherwise, would be better as smart pointers too.