- Multithreaded Work Queue in C++
- Producer-Consumer Model
- Work Queue Class
- Worker Item Class
- Consumer Thread Class
- Test Application
- Build and Run
This project is all about building a multithreaded application to devise a means to distribute tasks between threads so they can be processed concurrently.
Queues are good devices for transferring work items from one thread to another. In this repository I’ll discuss the design of a work queue class implemented in C++ that can be used with Thread class objects to easily build a multithreaded application.
One approach to multithreading is the producer-consumer model where one thread – the producer – places work items in a queue and one or more consumer threads waits for and removes the items to process.
When a consumer thread runs it checks the number of items on the queue. If there are one ore more work items on the queue the consumer removes one and processes it. If none are available the consumer waits for the producer to add items to the queue. These steps are repreated continually for the lifetime of the application.
The work queue class wqueue will be defined in the file wqueue.h
. It is based on the list
class from the Standard C++ Library. Lists provide methods for adding work items to the tail of the queue and removing items from the head of the queue – first in first out (FIFO) – in constant time. To serialize access to the queue and enable the producer thread to signal the consumer threads that work items are available for processing the queue class will be instrumented with a Pthread
mutex and condition variable – defined by the m_mutex
and m_condv
member variables respectively in this case.
#include <pthread.h>
#include <list>
using namespace std;
template <typename T> class wqueue
{
list<T> m_queue;
pthread_mutex_t m_mutex;
pthread_cond_t m_condv;
public:
wqueue() {
pthread_mutex_init(&m_mutex, NULL);
pthread_cond_init(&m_condv, NULL);
}
~wqueue() {
pthread_mutex_destroy(&m_mutex);
pthread_cond_destroy(&m_condv);
}
void add(T item) {
pthread_mutex_lock(&m_mutex);
m_queue.push_back(item);
pthread_cond_signal(&m_condv);
pthread_mutex_unlock(&m_mutex);
}
T remove() {
pthread_mutex_lock(&m_mutex);
while (m_queue.size() == 0) {
pthread_cond_wait(&m_condv, &m_mutex);
}
T item = m_queue.front();
m_queue.pop_front();
pthread_mutex_unlock(&m_mutex);
return item;
}
int size() {
pthread_mutex_lock(&m_mutex);
int size = m_queue.size();
pthread_mutex_unlock(&m_mutex);
return size;
}
};
The wqueue
class is defined as a template class since it uses a list object to queue work items of arbitrary class. The work item classes used in the test application will be discussed later in the article.
The great advantage to creating a work queue class in C++ is it encpasulates the Pthread
mechanisms necessary to serialize access to the work items on the list and signal when work items are added to the list. Programs that use the work queue can add and remove items with single method calls add()
and remove()
without having to concern themselves with the intricacies of making Pthread
calls.
The constructor simply initializes the Pthread mutex and condition variable data members.
The destructor deletes the mutex and condition variables. The list object is destroyed automatically.
To add a work item to the queue the add()
method is called passing a copy of the work item object. Normally standard C++ collections keep references to the template class object.
But for the work queue example the typename T will be work item pointers, so when the add()
method is called it will be passed a pointer by value and a reference to the pointer is stored in the list. You are better off storing pointers to work items on a queue so that you can control when they are destroyed.
To serialize access to the list the mutex
is locked and when the lock is acquired a reference to an item pointer is pushed to the back of the list
. Then the condition variable is signaled with a call to pthread_cond_signal()
which wakes up one of the consumer threads waiting to remove an item. Calling pthread_cond_broadcast()
to signal the condition variable would also work but this would cause all the consumer threads to wake up.
Since only one of the consumers at any given time can get a work item, the others would have to go back to sleep waiting for additional work items to placed on the queue. By signalling the condition instead of broadcasting, we ensure that only one thread wakes up at a time for each item added.
The remove()
method locks the mutex then checks to see if any work items are available. If not, pthread_cond_wait()
is called which automatically unlocks the mutex and waits for the producer thread to add an item. When the condition is signaled after an item is added by the producer thread, a copy of a pointer to a work item is taken off the list and returned to the consumer thread.
Note that if items are added to the queue while all the consumer threads are busy, there will be no consumer threads to receive the condition variable signals. However this is not a problem since the consumers always check the queue size when they return from doing work before trying to remove any work items.
The size()
method is just a utility method we can use to externally check the number of items on to the queue. The mutex must be locked and unlocked during this operation to avoid a race condition with the producer thread trying to add or another consumer thread trying to remove an item.
Work items will simply be a string and a number that are set to arbitrary values in the producer thread. Both values can be retrieved with their corresponding get
methods.
#include <stdio.h>
#include <stdlib.h>
#include <string>
#include "thread.h"
#include "wqueue.h"
class WorkItem
{
string m_message;
int m_number;
public:
WorkItem(const char* message, int number)
: m_message(message), m_number(number) {}
~WorkItem() {}
const char* getMessage() { return m_message.c_str(); }
int getNumber() { return m_number; }
};
The Thread class from this repository is used to create the consumer threads. Each thread is passed a reference the the work queue so it can grab work items.
The run()
method continually waits for and removes items to be processed which in this case just means displaying the string message and number assigned in the producer thread. The ID
of each thread is also displayed to differentiate them in the printed output.
class ConsumerThread : public Thread
{
wqueue<WorkItem*>& m_queue;
public:
ConsumerThread(wqueue<WorkItem*>& queue) : m_queue(queue) {}
void* run() {
// Remove 1 item at a time and process it. Blocks if no items are
// available to process.
for (int i = 0;; i++) {
printf("thread %lu, loop %d - waiting for item...\n",
(long unsigned int)self(), i);
WorkItem* item = (WorkItem*)m_queue.remove();
printf("thread %lu, loop %d - got one item\n",
(long unsigned int)self(), i);
printf("thread %lu, loop %d - item: message - %s, number - %d\n",
(long unsigned int)self(), i, item->getMessage(),
item->getNumber());
delete item;
}
return NULL;
}
}
The producer thread is nothing more that the main()
routine of the test application which is defined in the file main.cpp
as is the remainder of the code in this article.
The number of iterations through the main loop is passed in the command line. Two consumer threads are created and a single work queue. After the threads are started they will wait for items to be placed on the queue.
int main(int argc, char** argv)
{
// Process command line arguments
if ( argc != 2 ) {
printf("usage: %s <iterations>\n", argv[0]);
exit(-1);
}
int iterations = atoi(argv[1]);
// Create the queue and consumer (worker) threads
wqueue<WorkItem*> queue;
ConsumerThread* thread1 = new ConsumerThread(queue);
ConsumerThread* thread2 = new ConsumerThread(queue);
thread1->start();
thread2->start();
// Add items to the queue
WorkItem* item;
for (int i = 0; i < iterations; i++) {
item = new WorkItem("abc", 123);
queue.add(item);
item = new WorkItem("def", 456);
queue.add(item);
item = new WorkItem("ghi", 789);
queue.add(item);
sleep(2);
}
// Ctrl-C to end program
sleep(1)
printf("Enter Ctrl-C to end the program...\n");
while (1);
exit(0);
}
Each time through the main loop, 3
items are placed in the queue. After the specified number of iterations the producer will wait for a Ctrl-C
to end the program.
If you run the test application with an argument of 3
this is what the output will look like:
$ ./wqueue 3
thread 4547428352, loop 0 - waiting for item...
thread 4549251072, loop 0 - waiting for item...
thread 4547428352, loop 0 - got one item
thread 4549251072, loop 0 - got one item
thread 4547428352, loop 0 - item: message - abc, number - 123
thread 4549251072, loop 0 - item: message - def, number - 456
thread 4547428352, loop 1 - waiting for item...
thread 4549251072, loop 1 - waiting for item...
thread 4547428352, loop 1 - got one item
thread 4547428352, loop 1 - item: message - ghi, number - 789
thread 4547428352, loop 2 - waiting for item...
thread 4549251072, loop 1 - got one item
thread 4547428352, loop 2 - got one item
thread 4549251072, loop 1 - item: message - abc, number - 123
thread 4547428352, loop 2 - item: message - def, number - 456
thread 4549251072, loop 2 - waiting for item...
thread 4547428352, loop 3 - waiting for item...
thread 4549251072, loop 2 - got one item
thread 4549251072, loop 2 - item: message - ghi, number - 789
thread 4549251072, loop 3 - waiting for item...
thread 4547428352, loop 3 - got one item
thread 4549251072, loop 3 - got one item
thread 4547428352, loop 3 - item: message - abc, number - 123
thread 4549251072, loop 3 - item: message - def, number - 456
thread 4547428352, loop 4 - waiting for item...
thread 4549251072, loop 4 - waiting for item...
thread 4547428352, loop 4 - got one item
thread 4547428352, loop 4 - item: message - ghi, number - 789
thread 4547428352, loop 5 - waiting for item...
done