(See also the follow-up question.)
I was in the mood for pthread.h and decided to write a concurrent stack data structure. My requirements:
- the stack has maximum capacity,
- pushing an element,
- removing the top element without returning it,
- peeking the top element,
- client is blocked whenever popping from an empty stack,
- client is blocked whenever pushing to a full stack.
My codez looks like this:
concurrent_stack.h
#ifndef CONCURRENT_STACK_H
#define CONCURRENT_STACK_H
#include <stdlib.h>
typedef struct concurrent_stack
{
/*******************************************
The number of elements stored in this stack.
*******************************************/
size_t size;
/**************************************************
The maximum number of elements this stack can hold.
**************************************************/
size_t capacity;
/**********************************************
The actual array holding the data of the stack.
**********************************************/
void** storage;
/************************************************
The mutual exclusion lock for updating the stack.
************************************************/
pthread_mutex_t mutex;
/*****************************
Guards against an empty stack.
*****************************/
pthread_cond_t empty_condition_variable;
/***************************
Guards against a full stack.
***************************/
pthread_cond_t full_condition_variable;
}
concurrent_stack;
/*****************************************
Initializes a new, empty concurrent stack.
*****************************************/
void concurrent_stack_init(concurrent_stack* stack, size_t capacity);
/****************************************
Pushes a datum onto the top of the stack.
****************************************/
void concurrent_stack_push(concurrent_stack* stack, void* datum);
/******************************************
Returns, but does not remove the top datum.
******************************************/
void* concurrent_stack_top(concurrent_stack* stack);
/****************************************
Removes the topmost datum from the stack.
****************************************/
void concurrent_stack_pop(concurrent_stack* stack);
/*******************************************
Returns the number of elements in the stack.
*******************************************/
size_t concurrent_stack_size(concurrent_stack* stack);
/*********************************
Returns the capacity of the stack.
*********************************/
size_t concurrent_stack_capacity(concurrent_stack* stack);
/***********************************
Releases all resources of the stack.
***********************************/
void concurrent_stack_free(concurrent_stack* stack);
#endif /* CONCURRENT_STACK_H */
concurrent_stack.c
#include "concurrent_stack.h"
#include <pthread.h>
#define MAX(A,B) (((A) > (B)) ? (A) : (B))
static const size_t MINIMUM_CAPACITY = 10;
void concurrent_stack_init(concurrent_stack* stack, size_t capacity)
{
stack->size = 0;
stack->capacity = MAX(capacity, MINIMUM_CAPACITY);
stack->storage = malloc(sizeof(void*) * stack->capacity);
pthread_mutex_init(&stack->mutex, NULL);
pthread_cond_init(&stack->empty_condition_variable, NULL);
pthread_cond_init(&stack->full_condition_variable, NULL);
}
void concurrent_stack_push(concurrent_stack* stack, void* datum)
{
pthread_mutex_lock(&stack->mutex);
while (stack->size == stack->capacity)
{
pthread_cond_wait(&stack->full_condition_variable, &stack->mutex);
}
stack->storage[stack->size++] = datum;
pthread_cond_signal(&stack->empty_condition_variable);
pthread_mutex_unlock(&stack->mutex);
}
void concurrent_stack_pop(concurrent_stack* stack)
{
pthread_mutex_lock(&stack->mutex);
while (stack->size == 0)
{
pthread_cond_wait(&stack->empty_condition_variable, &stack->mutex);
}
stack->size--;
pthread_cond_signal(&stack->full_condition_variable);
pthread_mutex_unlock(&stack->mutex);
}
void* concurrent_stack_top(concurrent_stack* stack)
{
void* ret;
pthread_mutex_lock(&stack->mutex);
while (stack->size == 0)
{
pthread_cond_wait(&stack->empty_condition_variable, &stack->mutex);
}
ret = stack->storage[stack->size - 1];
pthread_cond_signal(&stack->full_condition_variable);
pthread_mutex_unlock(&stack->mutex);
return ret;
}
size_t concurrent_stack_size(concurrent_stack* stack)
{
size_t size;
pthread_mutex_lock(&stack->mutex);
size = stack->size;
pthread_mutex_unlock(&stack->mutex);
return size;
}
size_t concurrent_stack_capacity(concurrent_stack* stack)
{
return stack->capacity;
}
void concurrent_stack_free(concurrent_stack* stack)
{
free(stack->storage);
pthread_mutex_destroy(&stack->mutex);
pthread_cond_destroy(&stack->empty_condition_variable);
pthread_cond_destroy(&stack->full_condition_variable);
}
main.c
#include "concurrent_stack.h"
#include <pthread.h>
#include <stdio.h>
typedef struct thread_config
{
size_t element_count;
concurrent_stack* stack;
}
thread_config;
void* producer_code(void* data)
{
thread_config* cfg = (thread_config*) data;
size_t limit = cfg->element_count;
concurrent_stack* stack = cfg->stack;
for (size_t i = 0; i != limit; ++i)
{
concurrent_stack_push(stack, (void*) i);
}
return NULL;
}
void* inspector_code(void* data)
{
thread_config* cfg = (thread_config*) data;
size_t limit = cfg->element_count;
concurrent_stack* stack = cfg->stack;
for (size_t i = 0; i != limit; ++i)
{
concurrent_stack_top(stack);
concurrent_stack_size(stack);
}
return NULL;
}
void* consumer_code(void* data)
{
thread_config* cfg = (thread_config*) data;
size_t limit = cfg->element_count;
concurrent_stack* stack = cfg->stack;
for (size_t i = 0; i != limit; ++i)
{
concurrent_stack_pop(stack);
}
return NULL;
}
static const size_t PRODUCERS = 3;
static const size_t CONSUMERS = 3;
static const size_t INSPECTORS = 1;
static const size_t PRODUCER_ELEMENTS = 91 * 1000;
static const size_t CONSUMER_ELEMENTS = 90 * 1000;
static const size_t INSPECTOR_ELEMENTS = 50 * 1000;
/**
In order to make sure that the program exits, you must guarantee that:
PRODUCER_ELEMENTS * PRODUCERS - CONSUMER_ELEMENTS * CONSUMERS <= STACK_CAPACITY.
Otherwise, after all consumers have done their job, the producers will fill
it again and finally block on it.
*/
static const size_t STACK_CAPACITY = 5000;
int main()
{
concurrent_stack st;
concurrent_stack_init(&st, STACK_CAPACITY);
pthread_t threads[PRODUCERS + CONSUMERS + INSPECTORS];
size_t next_thread_slot = 0;
thread_config producer_thread_config = { PRODUCER_ELEMENTS, &st };
thread_config consumer_thread_config = { CONSUMER_ELEMENTS, &st };
thread_config inspector_thread_config = { INSPECTOR_ELEMENTS, &st };
for (size_t i = 0; i != CONSUMERS; ++i)
{
pthread_create(&threads[next_thread_slot++],
NULL,
consumer_code,
(void*) &consumer_thread_config);
}
for (size_t i = 0; i != INSPECTORS; ++i)
{
pthread_create(&threads[next_thread_slot++],
NULL,
inspector_code,
(void*) &inspector_thread_config);
}
for (size_t i = 0; i != PRODUCERS; ++i)
{
pthread_create(&threads[next_thread_slot++],
NULL,
producer_code,
(void*) &producer_thread_config);
}
/* All threads created. Now join them. */
for (size_t i = 0; i != INSPECTORS + CONSUMERS + PRODUCERS; ++i)
{
pthread_join(threads[i], NULL);
}
size_t expected_stack_size =
PRODUCERS * PRODUCER_ELEMENTS - CONSUMERS * CONSUMER_ELEMENTS;
printf("Expected final stack size: %zu\n", expected_stack_size);
printf("Actual final stack size: %zu\n", concurrent_stack_size(&st));
concurrent_stack_free(&st);
}
Critique request
Please tell me anything that comes to mind, yet especially I am interested in comments relating to:
- code layout,
- correctness of synchronization.
