How to make message queue with Condition variable - Purpose & Design

1. Purpose

This blog will explain how to make the queue for message transfer between threads in Linux. We assume that pthread is available in your environment.

Generally speaking, some OS may provide the message queue mechanism. But, pthread library supports the basic primitives such as mutex (lock mechanism) and condition variable (signal mechanism). This will show how to make the queue with these primitives.

2. Design

We will design the queue to transfer the message between 2 threads. One thread is called as producer thread and the other is called consumer thread. Producer thread will generate the message and put it on the queue and consumer thread will take the message from the queue and handle it. We assume that each thread's role is fixed and message will be transferred uni-directionally just like below diagram.

We will call the activity which consumer thread takes the message from queue as "get". If the queue is empty, the "get" method will wait until message comes in the queue. We will call the activity which producer thread stores the message into queue as "put". If the queue is full, the "put" method will wait until queue status is changed. As you know, waiting for some condition and signalling this waiting thread is implemented via condition variable. So this blog will show how to use the condition variable.

We need the mutex to prevent the data corruption by concurrent access to queue's internal data structure. The "put" and "get" methods can update the queue's internal data, so we will define and use the mutex to protect the queue's internal data. Both "put" and "get" methods require the wait activity, so we will use two condition variables for them. The "put" method will use put_cond and and "get" method will use get_cond.
And the wait operation for condition variables requires the mutex also. So we will share the same mutex which is already defined to protect queue's internal data.
(The relationship between condition variable and mutex will be covered later in details.)

Message type will be defined as "void *" and both producer and consumer threads will type-cast this as their usage. As you know, "void *" type is so general, so you can use this for any purpose without source modification. Just like the return type of malloc is defined as "void *" in C language. We will use the array data structure for the container of message queue. Because the queue's size is fixed at initialization, array is proper than linked list. We will access the array structure with index variable and this array will be filled and retrieved as circular format.

We defined the two index to connect put and get method with index. Write index is related with put method and get read index is related with get method. Write index is increased when message is stored in the queue though put activity and read index is increased when message is retrieved from the queue through get activity.

Above diagram shows three states of queue. Left one shows initial state of queue which has the length as 9. Write index and read index points at '0'. Middle state shows that write index is increased as 4 after producer thread stored 4 messages in the queue. Right state shows that consumer thread fetch the 2 messages from the queue. So read index is increased as 2.

We will implement that the length of queue is configured when queue is initialized. Generally speaking, queue's length is configurable and depends on user's usage. However, if the system is designed well, 3~5 will be suitable.