Bounded buffer synchronization problem

buffer.c

// bounded buffer problem

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <semaphore.h>
#include <pthread.h>

#include "buffer.h"

unsigned int SEED = 12345;                  // seed for RNG
unsigned int WAIT_TIME = 100000;            // wait time modulus
buffer_item buffer[BUFFER_SIZE];            // shared buffer
int size = 0, head = -1, tail = -1;         // buffer pointers
int produced_items = 0, consumed_items = 0; // producer/consumer counters
long total_items = 0;                       // number of items to process
pthread_mutex_t mutex;                      // binary semaphore for mutual exclusion
sem_t empty, full;                          // counting semaphores for buffer slots

int is_empty()
{
    return size == 0;
}

int is_full()
{
    return size == BUFFER_SIZE;
}

int insert_item(buffer_item item)
{
    if (is_full())
    {
        return -1;
    }

    if (size == 0)
    {
        head = 0;
        tail = 0;
    }
    else
    {
        tail = (tail + 1) % BUFFER_SIZE;
    }
    buffer[tail] = item;
    size++;
    return 0;
}

int remove_item(buffer_item * item)
{
    if (is_empty())
    {
        return -1;
    }

    *item = buffer[head];
    head = (head + 1) % BUFFER_SIZE;
    size--;
    if (size == 0)
    {
        head = -1;
        tail = -1;
    }
    return 0;
}

void print_buffer()
{
    for (int i = 0; i < BUFFER_SIZE; i++)
    {
        printf("%3d ", buffer[i]);
    }
    printf("\t(size %d)\n", size);
}

void * producer(void * param)
{
    int id = *(int *) param;
    buffer_item item;
    while (1)
    {
        // wait for random time
        usleep(rand_r(&SEED) % WAIT_TIME);

        // check if we're done
        if (produced_items == total_items)
        {
            pthread_exit(0);
        }

        // wait for an empty spot
        // sem_trywait returns -1 if it is blocked
        if (sem_trywait(&empty) == 0)
        {
            // acquire lock
            pthread_mutex_lock(&mutex);

            // insert item
            item = rand_r(&SEED) % 256;
            if (insert_item(item))
            {
                fprintf(stderr, "producer %3d error\n", id);
            }
            else
            {
                printf("producer %3d produced %3d\n", id, item);
                produced_items++;
            }

            // release
            pthread_mutex_unlock(&mutex);
            sem_post(&full);
        }
    }
}

void * consumer(void * param)
{
    int id = *(int *) param;
    buffer_item item;
    while (1)
    {
        // wait for random time
        usleep(rand_r(&SEED) % WAIT_TIME);

        // check if we're done
        if (consumed_items == total_items)
        {
            pthread_exit(0);
        }

        // wait for a filled spot
        // sem_trywait returns -1 if it is blocked
        if (sem_trywait(&full) == 0)
        {
            // acquire lock
            pthread_mutex_lock(&mutex);

            // consume item
            if (remove_item(&item))
            {
                fprintf(stderr, "consumer %3d error\n", id);
            }
            else
            {
                printf("consumer %3d consumed %3d\n", id, item);
                consumed_items++;
            }

            // release
            pthread_mutex_unlock(&mutex);
            sem_post(&empty);
        }
    }
}

int main(int argc, char ** argv)
{
    // get args
    if (argc < 4)
    {
        fprintf(stderr, "Not enough arguments.\n");
        fprintf(stderr, "Usage: %s pcount ccount nitems\n", argv[0]);
        exit(1);
    }
    long p_count = strtol(argv[1], NULL, 10);
    long c_count = strtol(argv[2], NULL, 10);
    total_items = strtol(argv[3], NULL, 10);

    // initialize semaphores
    pthread_mutex_init(&mutex, NULL);
    sem_init(&empty, 0, BUFFER_SIZE);
    sem_init(&full, 0, 0);

    // setup threads
    printf("Setting up %ld producer thread(s)\n", p_count);
    pthread_attr_t attr;
    pthread_attr_init(&attr);
    pthread_t producers[p_count];
    int producers_id[p_count];
    for (int i = 0; i < p_count; i++)
    {
        producers_id[i] = i;
        pthread_create(&producers[i], &attr, producer, producers_id + i);
    }
    printf("Setting up %ld consumer thread(s)\n", c_count);
    pthread_t consumers[c_count];
    int consumers_id[c_count];
    for (int i = 0; i < c_count; i++)
    {
        consumers_id[i] = i;
        pthread_create(&consumers[i], &attr, consumer, consumers_id + i);
    }

    // wait for threads to exit
    printf("Joining threads\n");
    for (int i = 0; i < p_count; i++)
    {
        pthread_join(producers[i], NULL);
    }
    for (int i = 0; i < c_count; i++)
    {
        pthread_join(consumers[i], NULL);
    }
    printf("completed %ld items\n", total_items);

    // clean up
    pthread_mutex_destroy(&mutex);
    sem_destroy(&full);
    sem_destroy(&empty);
}

buffer.h

// buffer header file

typedef int buffer_item;
#define BUFFER_SIZE 5

Makefile

all: buffer

buffer: buffer.h buffer.c
	clang buffer.c -o buffer -lpthread

test: buffer
	./buffer 1 1 5

test2: buffer
	./buffer 20 20 20

clean:
	rm buffer

./buffer 1 1 5
Setting up 1 producer thread(s)
Setting up 1 consumer thread(s)
Joining threads
producer   0 produced  27
consumer   0 consumed  27
producer   0 produced  96
consumer   0 consumed  96
producer   0 produced 242
consumer   0 consumed 242
producer   0 produced  57
producer   0 produced 204
consumer   0 consumed  57
consumer   0 consumed 204
completed 5 items

./buffer 20 20 100
Setting up 20 producer thread(s)
Setting up 20 consumer thread(s)
Joining threads
producer   5 produced 142
producer   1 produced 243
producer   9 produced 221
producer  14 produced  36
consumer   9 consumed 142
consumer  12 consumed 243
consumer   8 consumed 221
consumer   8 consumed  36
producer   4 produced 109
producer   0 produced 147
producer   4 produced 206
producer   2 produced 215
producer   1 produced 153
consumer  11 consumed 109
consumer   6 consumed 147
consumer   3 consumed 206
consumer  16 consumed 215
producer  14 produced   2
consumer   0 consumed 153
producer  13 produced  21
producer   4 produced 124
producer   9 produced 154
producer  19 produced 211
consumer   5 consumed   2
producer  12 produced 112
consumer   3 consumed  21
consumer   1 consumed 124
consumer  19 consumed 154
consumer  13 consumed 211
consumer   9 consumed 112
producer  14 produced 231
producer   3 produced 140
consumer   6 consumed 231
producer   0 produced  95
producer  15 produced  14
producer   8 produced 106
consumer   2 consumed 140
producer  12 produced 220
consumer  16 consumed  95
producer   3 produced 192
consumer  12 consumed  14
consumer  18 consumed 106
producer  12 produced  36
consumer  19 consumed 220
producer  16 produced  32
producer  15 produced 144
producer   6 produced 245
consumer  10 consumed 192
producer  17 produced 140
consumer  14 consumed  36
consumer   7 consumed  32
producer   1 produced  91
consumer  12 consumed 144
producer  10 produced 210
producer   6 produced  15
consumer  10 consumed 245
consumer  17 consumed 140
consumer  16 consumed  91
producer   4 produced 183
producer   1 produced 201
producer   5 produced  74
consumer   8 consumed 210
consumer   4 consumed  15
producer   3 produced 152
consumer  11 consumed 183
producer  19 produced 173
consumer  19 consumed 201
consumer  14 consumed  74
producer   2 produced 245
producer   6 produced 171
producer  15 produced  69
consumer   7 consumed 152
consumer   5 consumed 173
producer   3 produced 199
consumer   4 consumed 245
producer   7 produced 139
consumer  18 consumed 171
producer  17 produced  93
consumer  19 consumed  69
consumer   5 consumed 199
consumer   3 consumed 139
producer   9 produced 131
consumer   9 consumed  93
consumer  14 consumed 131
producer  10 produced 137
producer  11 produced 186
consumer  10 consumed 137
consumer   0 consumed 186
producer   5 produced  57
consumer   6 consumed  57
producer   4 produced 211
consumer   7 consumed 211
producer  18 produced 196
producer   0 produced 162
producer  16 produced 184
consumer  10 consumed 196
consumer   2 consumed 162
consumer  10 consumed 184
producer   2 produced  77
producer  12 produced 124
producer   7 produced 245
producer  14 produced 230
consumer  11 consumed  77
producer   4 produced  54
producer  15 produced 186
consumer  17 consumed 124
producer   1 produced  77
consumer   1 consumed 245
producer  12 produced 243
consumer  14 consumed 230
producer  11 produced  93
consumer  16 consumed  54
producer   9 produced 158
consumer  19 consumed 186
producer  14 produced 172
consumer   0 consumed  77
consumer  12 consumed 243
consumer   9 consumed  93
consumer  10 consumed 158
consumer  18 consumed 172
producer  13 produced  49
producer  14 produced 120
producer   9 produced 133
producer   7 produced 220
producer   3 produced  20
consumer   4 consumed  49
producer  13 produced 230
consumer   3 consumed 120
consumer  15 consumed 133
producer   7 produced  47
producer   8 produced  17
consumer   6 consumed 220
consumer  14 consumed  20
producer  10 produced 161
producer   5 produced 198
consumer   2 consumed 230
consumer   4 consumed  47
consumer  10 consumed  17
consumer   7 consumed 161
producer  19 produced  84
consumer   5 consumed 198
producer   8 produced 137
consumer   1 consumed  84
consumer  11 consumed 137
producer  12 produced 188
producer  18 produced 204
consumer  16 consumed 188
producer   0 produced  60
consumer  10 consumed 204
producer   1 produced 246
consumer   8 consumed  60
producer   0 produced 238
producer  16 produced 182
producer  11 produced 128
producer  14 produced 189
consumer  12 consumed 246
consumer  13 consumed 238
producer  11 produced 168
producer   2 produced 209
consumer  17 consumed 182
producer   8 produced 220
consumer   9 consumed 128
consumer   6 consumed 189
producer  13 produced  35
producer   4 produced  61
consumer  15 consumed 168
producer  17 produced   5
consumer  19 consumed 209
consumer   3 consumed 220
producer   2 produced 149
consumer  15 consumed  35
consumer   0 consumed  61
consumer  17 consumed   5
producer  16 produced 179
producer   6 produced 169
consumer   5 consumed 149
producer  12 produced 253
consumer   6 consumed 179
producer  12 produced 254
producer   9 produced  83
consumer  18 consumed 169
consumer   4 consumed 253
producer   5 produced 119
consumer   3 consumed 254
producer  14 produced 103
consumer   5 consumed  83
producer  19 produced  31
consumer  14 consumed 119
consumer  11 consumed 103
producer  14 produced  93
producer   3 produced 171
producer  15 produced 217
consumer   8 consumed  31
consumer  13 consumed  93
producer  11 produced 158
consumer  16 consumed 171
consumer   2 consumed 217
consumer  17 consumed 158
producer   1 produced 196
consumer   7 consumed 196
completed 100 items