Click here to Skip to main content

Article

A simple thread pool class

digger_chen

Rate me:

2.95/5 (15 votes)

31 Dec 2003

86.9K

1.9K

14

12

A simple thread pool class

Download source - 4.13 Kb

Introduction

ThreadPool is a very important concept to improve the performance of multi-threaded programs. Generally the threadpool includes an asynchronous queue to keep the tasks that a thread can pick a task from and the caller can put a task into.

In the threadpool code, we define an Asynchronous queue, a Job interface and a ThreadPool class to manage the threads. The synchronization is the core of the threadpool.

I have implemented some classes that will make it easier for you.

Details

It is very easy to use the threadpool class. First you should implement your job class by inheriting the "Job" interface. That means you need to implement two interface functions.

class Job {
public:
    virtual void Process() = 0;
    virtual ~Job() {};
};

For example, you define your job class as

class MyJob1 : public Job {
public:
    MyJob1(int index) : data(index) {};
    virtual void Process() {
        cout << data << endl;
    }
    ~MyJob1() {};

private:
    int data;
};

The "Process" is called by the thread. Then you can use the threadpool as following.

ThreadPool tp(4, TRUE);
Job *job = new MyJob1(0);
tp.Assignment(job);

Of course I intentionally made it simple for you to understand how to create and use threadpool. You can freely modify it and add other features.

License

This article has no explicit license attached to it but may contain usage terms in the article text or the download files themselves. If in doubt please contact the author via the discussion board below.

A list of licenses authors might use can be found here

Written By

digger_chen

Web Developer

United States

hello

Comments and Discussions

paolobia4-Dec-14 0:50

4-Dec-14 0:50

Hi,
For have all work done at end...

threadpool.h

C++

#include <list>
#include <windows.h>

class MSCriticalSection
{
public:
    MSCriticalSection()  { InitializeCriticalSection(&cs); }
    ~MSCriticalSection() { DeleteCriticalSection(&cs); }
 
    void EnterCriticalSection() { ::EnterCriticalSection(&cs); }
    void LeaveCriticalSection() { ::LeaveCriticalSection(&cs); }

private:
    CRITICAL_SECTION cs;
}; 


/* Asynchronized Queue */
template <class T>
class AsynchQueue {
public:
	AsynchQueue();
	~AsynchQueue();

	void Enqueue(T* data);
	T* Dequeue();

private:
	std::list<T*> queue;
	MSCriticalSection cs;

	AsynchQueue(const AsynchQueue<T>& q);
	AsynchQueue &operator=(const AsynchQueue<T>&);
};

/* Job that passed to the thread */
class Job {
public:
	virtual void Process() = 0;
	virtual ~Job() {};
};

/* Thread Pool */
class ThreadPool 
{
    friend DWORD WINAPI ThreadFunc(LPVOID lpParam);
public:
	
    ThreadPool(int num, BOOL releaseJob);
	~ThreadPool();

	void Assignment(Job *job);

private:
	bool Work();

	ThreadPool(const ThreadPool &);
	ThreadPool &operator=(const ThreadPool &);

	int m_iThread;
	AsynchQueue<Job> queue;
	std::list<HANDLE> threads;

	BOOL m_releaseJob;
};

threadpool.cpp

C++

#include <list>
#include "threadpool.h"


template <class T>
AsynchQueue<T>::AsynchQueue() 
{
}

template <class T>
AsynchQueue<T>::~AsynchQueue() 
{
}

template <class T>
void AsynchQueue<T>::Enqueue(T* data) 
{
	queue.push_back(data);
}

template <class T>
T *AsynchQueue<T>::Dequeue()
{
  cs.EnterCriticalSection();  

  T* data = NULL;
  if (queue.size() > 0) //If no this line,The data maybe is null
  {
    data = queue.front();
    queue.pop_front();
  }
   
  cs.LeaveCriticalSection();
  return data;
}
 
DWORD WINAPI ThreadFunc(LPVOID lpParam) 
{
	ThreadPool *pool = static_cast<ThreadPool *>(lpParam);

    while( pool->Work() )
        ;

	return 0;
}

ThreadPool::ThreadPool(int num, BOOL releaseJob) : 
	m_iThread(num), m_releaseJob(releaseJob) 
{

}

ThreadPool::~ThreadPool() 
{
	for (int i=0; i<m_iThread; i++) 
    {
		DWORD dwThreadId;

		HANDLE hThread = CreateThread( 
			NULL,                        // default security attributes 
			0,                           // use default stack size  
			ThreadFunc,                  // thread function 
			this,						 // argument to thread function 
			CREATE_SUSPENDED,            // use default creation flags 
			&dwThreadId);                // returns the thread identifier 
		threads.push_back(hThread);
	}

	for (std::list<HANDLE>::iterator iter = threads.begin(); 
		 iter != threads.end();
		 iter ++) 
    {
		ResumeThread(*iter);
	}

	HANDLE *handles = new HANDLE[m_iThread];

	int i=0;
	for (std::list<HANDLE>::iterator iter = threads.begin(); 
		 iter != threads.end();
		 iter ++) 
    {
		handles[i++] = *iter;
	}

	WaitForMultipleObjects(m_iThread, handles, TRUE, INFINITE);
	delete [] handles;

	for (std::list<HANDLE>::iterator iter = threads.begin(); 
		 iter != threads.end();
		 iter ++) 
    {
		CloseHandle(*iter);
	}
}

void ThreadPool::Assignment(Job *job) 
{
	queue.Enqueue(job);
}

bool ThreadPool::Work()
{
    Job *job = queue.Dequeue();

    if(!job)
        return false;
   
    job->Process();

    if (m_releaseJob)
        delete job;

    return true;
};

test.cpp

C++

#include <iostream>
#include <tchar.h>
#include "threadpool.h"

using namespace std;

int isprime(int x)
{
    int prime = 0;
    int i = 0;

    for (i = (x - 1); i > 1; --i)
    {
        if ((x % i) == 0)
        {
            prime = 0;
            return prime;
        }
        else
        {
            prime = 1;
        }
    }

    return 1;
}


class MyJob : public Job {
public:
	MyJob(int i, unsigned long d) : idx(i), data(d) {} 

	virtual void Process() 
    {
		cout << "#" << idx << ":" << data << (isprime(data) ? " PRIME" : " NO") << endl;
	}

	~MyJob() {}                                      
private:
    int idx;
	unsigned long data;
};


void RandomTest(int i) 
{
	ThreadPool tp(i, TRUE);

	for (int i=0; i<100; i++) 
    {
		Job *job = new MyJob(i, (unsigned long)(rand() * RAND_MAX * i / 10) );
		tp.Assignment(job);
	}

	Job *job = new MyJob(9999, RAND_MAX ); 
	tp.Assignment(job);                   
}

int main(int argc, char **argv) 
{
    int n=8;
    if( argc==2) 
        n = atoi(argv[1]);

	RandomTest(n);
	return 0;
}

Mycro6-Jul-10 18:46

6-Jul-10 18:46

template <class T>
T *AsynchQueue<T>::Dequeue()
{
WaitForSingleObject(hEvent, INFINITE);
WaitForSingleObject(hMutex, INFINITE);
T* data = NULL;

if (queue.size() > 0) //If no this line,The data maybe is null
{
data = queue.front();
queue.pop_front();
if (queue.size() > 0)
SetEvent(hEvent); //SetEvent times > "WaitForSingleObject(hEvent, INFINITE)" 's times ,so queue.front() maybe return NULL

}

ReleaseMutex(hMutex);
return data;
}

void ThreadPool::Work()
{
Job *job = queue.Dequeue();
if(!job)
return;

job->Process();
if (m_releaseJob)
{
delete job;
}
};

Runtime problems with this sample code.

mark.steddom@gmail.com17-Feb-05 13:35

mark.steddom@gmail.com

17-Feb-05 13:35

First off… I have not modified this example at all! Smile | :)

Smile | :)

I compiled this sample code using Visual Studio .Net 2003. With no compile-time errors or warnings. I also tried compiling this project on another machine running Visual Studio 6 (latest service packs and hotfixes as of 2-17-05) with the same results.

When running the code (in release or debug) I encounter an error inside of the Dequeue() method. This is what I see happening inside Dequeue():

1. The queue is poped multiple times until the queue has no remaining elements.
2. SetEvent() is then skipped (as it should be when the queue.size < 1)
3. The next thread that runs the Dequeue() method makes a call to WaitForSingleObject(hEvent, INFINITE) and holds (as it should…)

4. BUT…. The next thread that runs the Dequeue() method makes the same call to WaitForSingleObject(hEvent, INFINITE) and receives back ‘WAIT_OBJECT_0’ so the queue ends up being poped yet again and the Dequeue() method ends up returning garbage data (job) to the work() method. Which in turn causes an Unhandled access violation while reading the garbage job.

It seems that the hEvent is being reset, mangled, or something causing the WaitForSingleObject() call to not block.

Any ideas on what may be going wrong. According to the MS documentation the code itself is perfect

Please help!

--Mark
mark.steddom@gmail.com

Re: Runtime problems with this sample code.

monkbiao27-Sep-05 22:17

27-Sep-05 22:17

The problem should happen when ThreadPool object destructed!

Letting the threads close nicely.

Nigel Atkinson14-Nov-04 14:19

14-Nov-04 14:19

I noticed that if a thread was stopped waiting for the event that signifies data to process, and the thread pool class destructed, they would be stopped dead. As each thread did not allocate any data etc, this didn't cause any problems. Anyhow I found a simple way to have each thread terminate normally. Rather than wait INFINITEly for the event, in the
T *AsynchQueue<T>::Dequeue() function, I changed this to wait 1 sec (or any amount of time you like). Then added an if statement that gets the data (as before) if WaitForSingleObject returns WAIT_OBJECT_0 otherwise returns from the function with NULL.

I changed void ThreadPool::Work() to only process if job <> NULL and do nothing otherwise.

So what happens? If the event waiting times out, null is returned to the Work function. This in turn does nothing and returns to the ThreadFunc function. This then tests for the thread pool stopping, and stops if need be. Yay! Wink | ;)

Wink | ;)

The extra work of checking the stop flag only happens once a second, if there is no work to do.

template <class T><br />
T *AsynchQueue<T>::Dequeue() {<br />
	<br />
	if( WaitForSingleObject(hEvent, 1000) == WAIT_OBJECT_0 )<br />
	{<br />
		WaitForSingleObject(hMutex, INFINITE);<br />
		T* data = queue.front();<br />
		queue.pop_front();<br />
		if (queue.size() > 0) SetEvent(hEvent);<br />
		ReleaseMutex(hMutex);<br />
		return data;<br />
	}<br />
	else <br />
		return NULL;<br />
}<br />
<br />
void ThreadPool::Work() {<br />
	Job *job = queue.Dequeue();		// Returns null if timed out.<br />
	if( job )<br />
	{<br />
		job->Process();<br />
		if (m_releaseJob) {<br />
			delete job;<br />
		}<br />
	}<br />
};

"Land a'hoy!" * CRASH * "I should av said that sooner eh?" - Eckles, The Goon Show

Re: Letting the threads close nicely.

Nigel Atkinson14-Nov-04 14:22

14-Nov-04 14:22

PS I really like your code. Been looking for some simple (yet working) thread pooling stuff to learn from for a while.

"Land a'hoy!" * CRASH * "I should av said that sooner eh?" - Eckles, The Goon Show

Mathieu DEGROUX16-Apr-04 5:01

Mathieu DEGROUX

16-Apr-04 5:01

D'Oh! | :doh:

There is a little bug in the source code.
In the header file, there is no preprocessor macros to ensure that the file is included one time.

What's inside --- the 100lines code (roughly) contains STL based queue, using some event/mutext to protect race condition

Anonymous2-Jan-04 5:02

2-Jan-04 5:02

And that's all?

Re: What's inside --- the 100lines code (roughly) contains STL based queue, using some event/mutext to protect race condition

Anonymous14-Jan-04 3:25

14-Jan-04 3:25

If it does the job, then so what?

WinAPI on polling

Anonymous2-Jan-04 4:37

2-Jan-04 4:37

QueueUserWorkItem CreateTimerQueueTimer WaitOrTimerCallbackFunc BindIoCompletionCallback API are more control-able...

Re: WinAPI on polling

Anonymous14-Jan-04 3:28

14-Jan-04 3:28

But they are also only available on Win2K onwards and don't offer a lot of extensibility.

Re: WinAPI on polling

Blake Miller25-Aug-05 6:30

25-Aug-05 6:30

Some potential Win32 thread-pooling API limitations from MSDN:
1. QueueUserWorkItem - There is no way to cancel a work item after it has been queued.
2. In general, you should avoid using thread local storage and queuing asynchronous calls that require a persistent thread - Does that toss out using MFC, which is dependent upon TLS?
3. Note that thread pooling is not compatible with the single-threaded apartment (STA) model - Could limit use of some ActieX.

General News Suggestion Question Bug Answer Joke Praise Rant Admin

Use Ctrl+Left/Right to switch messages, Ctrl+Up/Down to switch threads, Ctrl+Shift+Left/Right to switch pages.

Go to top