Implementing Concurrency Abstractions in C
Introduction
Go is one of my favorite programming languages. Several things make the Go programming language unique. From flexible interfaces to robust concurrency abstractions. One that stands out to me is the channel in Go. Good abstractions hide the complexity of an implementation from the user of the programming interface. Channels are an excellent example. To implement something that looks like Go's channels, we shall evolve through several simple data structures, pointing out flaws and then fixing those flaws. First we shall start by implementing a queue.
Implementing a Basic Queue
A queue is a basic first in first out data structure that can be easily implemented using arrays or linked lists. We shall use the array implementation here. First we define a struct to represent the Queue as follows:
typedef struct {
int size; // how many items do we have in the queue at any given time
int capacity; // what is the maximum number of items the queue can hold
void **backing_array; // an array that can hold any data type;
int front; // this is where we dequeue from
} Queue;
Initializing the Queue
We define a simple method to help create a new queue as follows:
Queue *queue_init(int capacity){
Queue *q = malloc(sizeof(Queue));
q->backing_array = calloc(capacity, sizeof(void*));
q->front = 0;
q->size = 0;
q->capacity = capacity;
return q;
}
Performing Queue Operations
The most important operations are enqueueing and dequeuing. We shall define methods to help carry out those operations along with other helper methods to tell us if the queue is full or empty, and to tell us the size of the queue.
To enqueue we use the following function.
int queue_enqueue(Queue *q, void *data){
if (q->size >= q->capacity){ // we can only add items into the queue when it's not full
return -1; // we use a -1 to signal failure
}
int insert_at = (q->front + q->size) % q->capacity;
q->backing_array[insert_at] = data;
q->size++;
return 0; // 0 signal success
}
To dequeue we use the following function:
void *queue_dequeue(Queue *q){
if (q->size == 0){ // can only dequeue from a queue that is not empty
return NULL; // NULL means nothing was dequeued
}
void *dequeued_data = q->backing_array[q->front];
// move front of queue minding queue capacity
q->front = (q->front + 1) % q->capacity;
// decrease size of queue
q->size--;
return dequeued_data;
}
Then we define helper methods as shown below, and we can use them to make our queuing and dequeuing methods more descriptive, but I skip those here:
int queue_is_full(Queue *q) {
return q->size == q->capacity;
}
int queue_is_empty(Queue *q) {
return q->size == 0;
}
Avoiding Race Conditions
If we intend to use the queue for concurrent operations, we need to avoid race conditions. For example say we have a producer putting items into the queue and a consumer taking items out of the queue, what happens if both threads access or modify the queue's shared state concurrently? To prevent concurrent threads from modifying or observing the queue's shared state inconsistently, we protect the queue's critical sections with a mutex. We do the same for when removing items from the queue. We introduce a mutex to the queue's struct.
pthread_mutex_t mutex;
We also update the queue's init function so we can initialize the mutex
pthread_mutex_init(&q->mutex, NULL);
Now we can identify the critical paths of the enqueue and dequeue functions and use the mutex to protect them.
int queue_enqueue(Queue *q, void *data){
pthread_mutex_lock(&q->mutex); // we lock the mutex here
if (q->size >= q->capacity){
pthread_mutex_unlock(&q->mutex); // unlock if we can't queue up an item
return -1; // we use a -1 to signal failure
}
int insert_at = (q->front + q->size) % q->capacity;
q->backing_array[insert_at] = data;
q->size++;
pthread_mutex_unlock(&q->mutex); // unlock after we queue up an item
return 0; // 0 signal success
}
Then we do the same for the dequeue operation.
void *queue_dequeue(Queue *q){
pthread_mutex_lock(&q->mutex); // we lock the mutex here before we access the queue
if (q->size == 0){
pthread_mutex_unlock(&q->mutex); // unlock if we cant dequeue up an item
return NULL; // NULL means nothing was dequeued
}
void *dequeued_data = q->backing_array[q->front];
// move front of queue minding queue capacity
q->front = (q->front + 1) % q->capacity;
// decrease size of queue
q->size--;
pthread_mutex_unlock(&q->mutex); // unlock after queuing an item
return dequeued_data;
}
And finally the helper methods:
int queue_is_full(Queue *q) {
pthread_mutex_lock(&q->mutex);
int is_full = q->size == q->capacity;
pthread_mutex_unlock(&q->mutex);
return is_full;
}
int queue_is_empty(Queue *q) {
pthread_mutex_lock(&q->mutex);
int is_empty = q->size == 0;
pthread_mutex_unlock(&q->mutex);
return is_empty;
}
Although these helper functions are individually thread-safe, using them to check the queue and then performing an operation separately does not make the entire check-and-act sequence atomic. This will become important when we introduce the producer-consumer pattern.
The Producer-Consumer Pattern
Now we shall proceed to use the queue. In our main function we create two threads. One to act as a producer, generating items into the queue and the other as a consumer pulling items out of the queue. The producer and consumer are quite simple. We simply put and remove an integer into and from the queue. Let's start with the producer.
void *producer(void *data) {
int i = 1;
Queue *q = (Queue *)data;
while (1) {
if (!queue_is_full(q)){ // ensure queue is not full before enqueuing items
int *data_to_produce = malloc(sizeof(int));
*data_to_produce = i++;
queue_enqueue(q, data_to_produce);
printf("Producer %d is enqueued\n", *data_to_produce);
}
}
}
And then the consumer:
void *consumer(void *data) {
Queue *q = (Queue *) data;
while (1) {
if (!queue_is_empty(q)){ // pull items from queue if its not empty
const int *data_to_consume = queue_dequeue(q);
printf("Consumer %d is dequeued\n", *data_to_consume);
free(data_to_consume);
}
}
}
The Problem: Busy Waiting
Looking at the producer and the consumer, we can see that for each loop we are continuously asking: is the queue empty, is the queue full? This consumes CPU and a better approach would be to make the threads sleep when it cannot proceed. To solve this problem we shall introduce condition variables and proceed to make the queue a blocking queue.
First we update the queue's struct to include condition variables. One to tell if the queue is not empty and another to tell if it is not full. We add the following to the struct:
pthread_cond_t not_full;
pthread_cond_t not_empty;
Next, we proceed to update the enqueue and dequeue functions. While the queue is full we wait, effectively making the producer thread sleep. When we add an item to the queue, we signal not_empty, allowing a waiting consumer to wake and re-check the queue. We apply similar pattern to the consumer as well.
Note that pthread_cond_wait() releases the mutex and blocks the calling thread as part of a single atomic operation with respect to the condition-variable protocol: it releases the mutex and blocks the calling thread. When the thread is awakened, it reacquires the mutex before returning. Also, we use while rather than if because waking from a condition variable does not guarantee that the condition is still true. Another thread may have changed the queue before this thread reacquires the mutex. Therefore, after waking, the thread must re-check the condition.
void queue_enqueue(Queue *q, void *data) {
pthread_mutex_lock(&q->mutex);
while (q->size == q->capacity) { // queue is full
pthread_cond_wait(&q->not_full, &q->mutex); // block until we can proceed
}
const int back = (q->front + q->size) % q->capacity;
q->backing_array[back] = data;
q->size++;
pthread_cond_signal(&q->not_empty); // signal that the queue is not empty
pthread_mutex_unlock(&q->mutex);
}
void *queue_dequeue(Queue *q) {
pthread_mutex_lock(&q->mutex);
while (q->size == 0) { // queue is empty
pthread_cond_wait(&q->not_empty, &q->mutex); // block until we can proceed
}
void *data = q->backing_array[q->front];
q->front = (q->front + 1) % q->capacity;
q->size--;
pthread_cond_signal(&q->not_full); // signal that the queue is not full
pthread_mutex_unlock(&q->mutex);
// ownership of data is transferred to the caller
return data;
}
The Consumer and Producer without Busy Waiting
Now that we have used condition variables to eliminate busy waiting we update the producer and consumer.
void *producer(void *data) {
int i = 1;
Queue *q = (Queue *)data;
while (1) {
int *data_to_produce = malloc(sizeof(int));
*data_to_produce = i++;
queue_enqueue(q, data_to_produce);
printf("Producer %d is enqueued\n", *data_to_produce);
}
}
void *consumer(void *data) {
Queue *q = (Queue *)data;
while (1) {
const int *data_to_consume = queue_dequeue(q);
printf("Consumer %d is dequeued\n", *data_to_consume);
free(data_to_consume);
}
}
Introducing Channels
The queue exposes some of the mechanisms used to implement concurrent communication, such as its capacity, blocking behavior, and synchronization primitives. We can hide these implementation details behind a channel abstraction.
For now, we shall concern ourselves with two channel operations: putting items into the channel and retrieving items from the channel. We define a channel contract as follows:
typedef struct {
Queue *blocking_queue;
}Channel;
void *channel_get(Channel *channel);
void channel_put(Channel *channel, void *p);
Channel *channel_make(int capacity);
We use the blocking queue that we built earlier as the underlying data store for the channel. The channel provides a simpler interface for communicating values between concurrent threads without requiring the producer or consumer to interact directly with the queue. The implementation is deliberately simple:
Channel *channel_make(int capacity) {
Channel *channel = malloc(sizeof(Channel));
channel->blocking_queue = queue_init(capacity);
return channel;
}
void *channel_get(Channel *channel) {
return queue_dequeue(channel->blocking_queue);
}
void channel_put(Channel *channel, void *p) {
queue_enqueue(channel->blocking_queue, p);
}
This implementation doesn't introduce any new synchronization mechanism. Instead, it builds on the blocking queue we already implemented. A producer can now communicate a value without knowing how the underlying queue is implemented: channel_put(channel, data);. Likewise, a consumer can retrieve a value through: int *data = channel_get(channel);