[boost] [thread] new thread_pool and dispatcher classes

6 May 2007

      Here are a few classes which I think are very useful:

1. A thread_pool class. I think there's not much I can tell you that you 
don't know about it. It's something that has threads, and can accept 
tasks to be pushed into a queue, and executed in turn when a thread is 
available. It can also be suspended (stop handling tasks, but keep the 
threads alive) and resumed.

2. Two 'dispatcher' classes, one for plain threads, and one for 
thread_pools. They accept a single task, or several tasks, to be 
executed, either by dedicated threads (the thread_dispatcher class), or 
by a thread_pool (the thread_pool_dispatcher class). These classes 
provide a wait() method that block until all tasks are done.

IMO, these classes fill a hole in the Boost.Thread library, and can be a 
basic for the future<> template (I have to admit I didn't follow the 
future<> discussion, so I hope there's nothing here to contradict any 
consensus reached in that discussion, if any).

A few notes:

1. The semahpore class is there because it was needed for the 
thread_pool_dispather, but I think it will be a good general purpose 
addition to Boost.Thread. If I'm the only one thinking so, it can be 
moved as implementation detail of thread_pool_dispather.

2. The thread_task typedef should really be removed. Instead, a 'task' 
typedef should be added to boost::thread. So all occurances of 
'thread_task' should be replaced with 'thread::task'.

3. I'm not sure about the name 'dispatcher', but couldn't find anything 
I'm happy with. Something that catches the essence of the classes. Any 
suggestions are welcome.

4. The thread_pool class can benefit from two additional features:

4.A. add_threads(size_t) and remove_threads(size_t). I think I can add 
those if found useful.

4.B. pre_loop and post_loop hooks to be executed by each of the worker 
threads in the thread_pool, immediately after creation, and before 
termination. I actually have a use case for those, but I don't want to 
distract the discussion from the main things.

5. The asserts in thread_pool::~thread_pool, thread_pool::suspend, 
thread_pool::resume, thread_dispatcher::~thread_dispatcher, 
thread_pool_dispatcher::~thread_pool_dispatcher are supposed to catch 
user errors. These asserts are questionable. I'm not sure whether those 
use cases are indeed errors, and if so, should assert be the right way 
to deal with them. I chose to be a hard-liner for now, with an option to 
soften.

6. The other asserts are there to catch errors in my code.

7. Some or maybe all of the asserts should be replaced by BOOST_ASSERT. 
I'm not sure of the Boost convention here.

I hope it will be found useful!

#ifndef BOOST_THREAD_SEMAPHORE_HPP
#define BOOST_THREAD_SEMAPHORE_HPP

#include <cstdlib>
#include <cassert>
#include <boost/thread/mutex.hpp>
#include <boost/thread/condition.hpp>

namespace boost {

class semaphore
{
public:
	semaphore(std::size_t count) : m_count(count), m_mutex(), m_condition() { }

	void lock()
	{
		mutex::scoped_lock sl(m_mutex);
		while (m_count == 0)
			m_condition.wait(sl);
		assert(m_count > 0);
		m_count--;
	}

	void unlock()
	{
		mutex::scoped_lock sl(m_mutex);
		m_count++;
		m_condition.notify_one();
	}

private:
	std::size_t m_count;
	mutex m_mutex;
	condition m_condition;
};

} // namespace boost

#endif

#ifndef BOOST_THREAD_DISPATCHER_HPP
#define BOOST_THREAD_DISPATCHER_HPP

#include <cassert>
#include <set>
#include <algorithm>
#include <boost/noncopyable.hpp>
#include <boost/bind.hpp>
#include <boost/thread/thread.hpp>
#include <boost/thread/thread_task.hpp>

namespace boost {

class thread_dispatcher : private noncopyable
{
public:
	thread_dispatcher(const thread_task &t) :
		m_threads()
	{
		create_thread(t);
	}

	template <typename FwdIter>
	thread_dispatcher(FwdIter task_begin, FwdIter task_end) :
		m_threads()
	{
		std::for_each(task_begin, task_end, bind(&thread_dispatcher::create_thread, this, _1));
	}

	~thread_dispatcher()
	{
		assert(m_threads.empty());
	}

	void wait()
	{
		std::for_each(m_threads.begin(), m_threads.end(), bind(&thread::join, _1));
		m_threads.clear();
	}

private:
	std::set< shared_ptr<thread> > m_threads;

	void create_thread(const thread_task &t)
	{
		m_threads.insert(shared_ptr<thread>(new thread(t)));
	}
};

} // namespace boost

#endif

#ifndef BOOST_THREAD_POOL_HPP
#define BOOST_THREAD_POOL_HPP

#include <cstdlib>
#include <cassert>
#include <set>
#include <deque>
#include <algorithm>
#include <boost/noncopyable.hpp>
#include <boost/bind.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/condition.hpp>
#include <boost/thread/thread.hpp>
#include <boost/thread/thread_task.hpp>

namespace boost {

class thread_pool : private noncopyable
{
	enum state { active_, suspended_, terminating_ };

public:
	typedef function<void ()> task;

	thread_pool(std::size_t thread_count) :
		m_mutex(),
		m_condition(),
		m_threads(),
		m_tasks(),
		m_state(active_)
	{
		thread_task f = bind(&thread_pool::thread_func, this);
		while (thread_count > 0)
		{
			m_threads.insert(shared_ptr<thread>(new thread(f)));
			thread_count--;
		}
	}

	~thread_pool()
	{
		{
			mutex::scoped_lock sl(m_mutex);
			assert(m_state == active_);
			m_state = terminating_;
			m_condition.notify_all();
		}
		std::for_each(m_threads.begin(), m_threads.end(), bind(&thread::join, _1));
	}

	bool suspended() const
	{
		mutex::scoped_lock sl(m_mutex);
		assert(m_state == active_ || m_state == suspended_);
		return m_state == suspended_;
	}

	void suspend()
	{
		mutex::scoped_lock sl(m_mutex);
		assert(m_state == active_);
		m_state = suspended_;
	}

	void resume()
	{
		mutex::scoped_lock sl(m_mutex);
		assert(m_state == suspended_);
		m_state = active_;
		m_condition.notify_all();
	}

	void enqueue(const task &t)
	{
		mutex::scoped_lock sl(m_mutex);
		assert(m_state == active_ || m_state == suspended_);
		m_tasks.push_back(t);
		m_condition.notify_one();
	}

	void discard_pending_tasks()
	{
		mutex::scoped_lock sl(m_mutex);
		assert(m_state == active_ || m_state == suspended_);
		m_tasks.clear();
	}

private:
	mutable mutex m_mutex;
	condition m_condition;
	std::set< shared_ptr<thread> > m_threads;
	std::deque< task > m_tasks;
	state m_state;

	void thread_func()
	{
		task t;
		while (t = get_task(), t)
		{
			t();
			t.clear();
		}
	}

	task get_task()
	{
		mutex::scoped_lock sl(m_mutex);
		while (m_state == suspended_ || m_state == active_ && m_tasks.empty())
			m_condition.wait(sl);
		assert(m_state != suspended_);
		if (!m_tasks.empty())
		{
			task t = m_tasks.front();
			m_tasks.pop_front();
			return t;
		}
		assert(m_state == terminating_ && m_tasks.empty());
		return task();
	}
};

} // namespace boost

#endif

#ifndef BOOST_THREAD_POOL_DISPATCHER_HPP
#define BOOST_THREAD_POOL_DISPATCHER_HPP

#include <cstdlib>
#include <cassert>
#include <algorithm>
#include <boost/noncopyable.hpp>
#include <boost/bind.hpp>
#include <boost/thread/semaphore.hpp>
#include <boost/thread/thread_pool.hpp>

namespace boost {

class thread_pool_dispatcher : private noncopyable
{
public:
	thread_pool_dispatcher(const thread_pool::task &t, thread_pool &pool) :
		m_task_count(0), m_sem(0)
	{
		enqueue(t, pool);
	}

	template <typename FwdIter>
	thread_pool_dispatcher(FwdIter task_begin, FwdIter task_end, thread_pool &pool) :
		m_task_count(0), m_sem(0)
	{
		std::for_each(task_begin, task_end, bind(&thread_pool_dispatcher::enqueue, this, _1, ref(pool)));
	}

	~thread_pool_dispatcher()
	{
		assert(m_task_count == 0);
	}

	void wait()
	{
		while (m_task_count > 0)
		{
			m_sem.lock();
			m_task_count--;
		}
	}

private:
	std::size_t m_task_count;
	semaphore m_sem;

	void enqueue(const thread_pool::task &t, thread_pool &pool)
	{
		m_task_count++;
		pool.enqueue(bind(&thread_pool_dispatcher::execute_and_notify, this, t));
	}

	void execute_and_notify(const thread_pool::task &t)
	{
		t();
		m_sem.unlock();
	}
};

} // namespace boost

#endif

#ifndef BOOST_THREAD_TASK_HPP
#define BOOST_THREAD_TASK_HPP

#include <boost/function.hpp>

namespace boost {

typedef function<void ()> thread_task;

} // namespace boost

#endif

[boost] [thread] new thread_pool and dispatcher classes

Yuval Ronen