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;
}