TP6 - SE - INSA Lyon

bag.c

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

#include "bag.h"

bag_t *bb_create(int size)
{
    assert(size > 0);

    bag_t *bag = malloc(sizeof(bag_t));
    assert(bag != NULL);

    bag->elem = malloc(size * sizeof(void *));
    assert(bag->elem);

    bag->size = size;
    bag->count = 0;
    bag->is_closed = 0;
    sem_init(&bag->available, 0, (unsigned int) size);
    sem_init(&bag->current, 0, 0);

    return bag;
}

void bb_add(bag_t *bag, void *element)
{
    assert(bag != NULL);                 // sanity check

    assert(bag->is_closed == 0);

    sem_wait(&bag->available); // Wait for availability

    assert(bag->is_closed == 0);   // adding to a closed bag is an error
    assert(bag->count < bag->size); // sanity check

    bag->elem[bag->count] = element;
    bag->count += 1;

    sem_post(&bag->current); // Bag is not empty anymore
}

void *bb_take(bag_t *bag)
{
    assert(bag != NULL); // sanity check

    sem_wait(&bag->current); // Wait for some elements

    if (bag->is_closed && bag->count == 0) return NULL;

    assert(bag->count > 0); // sanity check

    bag->count -= 1;
    void *r = bag->elem[bag->count];

    sem_post(&bag->available); // Bag is not empty anymore

    usleep(10);// artificial delay to increase the occurence of race conditions
    return r;
}

void bb_close(bag_t *bag, int N)
{
    assert(bag->is_closed == 0);

    bag->is_closed = 1;

    for (int i = 0; i < N; ++i)
    {
        sem_post(&bag->current);
    }
}

bag.h

#ifndef _BAG_H
#define _BAG_H
#include <pthread.h>
#include <semaphore.h>

// bounded bag with blocking add and take operations

typedef struct {
    int     size;  // maximum number of elements (== length of 'elem' array)
    void ** elem;  // array of pointer to elements (each pointer has type 'void*' )
    int     count; // number of elements currently in the buffer
    sem_t available; // size - count
    sem_t current; // count

    int is_closed; // boolean. cf last exercice
} bag_t ;

// Create a new bag with room for 'size' elements
bag_t * bb_create(int size);

// Insert an element in the bag.
// - if bag is full, block until there is room
void bb_add(bag_t *b, void * element);

// Retrieve an element from the bag
// - if bag is non-empty: remove one element and return it
// - if bag is empty and open, block until something is added
// - if bag is empty and closed return NULL
void* bb_take(bag_t *b);

// Close the bag. When a bag is closed:
// - closing it again is an error
// - calling bb_add is an error
// - bb_take doesn't block anymore on an empty bag, but returns NULL instead
// - all threads previously blocked in bb_take are unblocked (up to N of them)
void bb_close(bag_t *bag, int N);

#endif

main.c

#include <stdio.h>
#include <stdlib.h>
#include <assert.h>
#include <unistd.h>
#include <pthread.h>

#include "bag.h"

typedef struct
{
    int value;
} boxed_int_t;


typedef struct
{
    int tnum; // thread number in 0 .. N-1
    bag_t *bag;
    boxed_int_t *sum;
    pthread_mutex_t *mutex;
    // TODO: add other fields here
} consumer_arg_t;


// each consumer thread runs this function
void *consumer(void *arg)
{
    consumer_arg_t *carg = (consumer_arg_t *) arg;
    printf("consumer %d: start\n", carg->tnum);

    while (1)
    {
        pthread_mutex_lock(carg->mutex);

        boxed_int_t *box = bb_take(carg->bag);
        if (box != NULL)
        {
            carg->sum->value += box->value;
        } else
        {
            pthread_mutex_unlock(carg->mutex);
            pthread_exit(NULL);
        }

        pthread_mutex_unlock(carg->mutex);
    }

    printf("consumer %d: end\n", carg->tnum);
    return NULL; // dummy return to comply with required signature
}


typedef struct
{
    int tnum; // thread number in 0 .. N-1
    bag_t *bag;
    // TODO: add other fields here
} producer_arg_t;


// each producer thread runs this function
void *producer(void *arg)
{
    producer_arg_t *parg = (producer_arg_t *) arg;

    printf("producer %d:start \n", parg->tnum);
    int count;
    for (count = 0; count < parg->tnum + 1; count++)
    {
        boxed_int_t *box = malloc(sizeof(boxed_int_t));
        assert(box != NULL);
        box->value = 1;

        bb_add(parg->bag, box);
    }

    printf("producer %d:end\n", parg->tnum);

    return NULL; // dummy return to comply with required signature
}

int main(int argc, char **argv)
{
    assert(argc == 3);

    int N = atoi(argv[1]);
    assert(N > 0);

    pthread_t prod[N];
    pthread_t cons[N];

    int S = atoi(argv[2]);
    assert(S > 0);


    // shared container
    bag_t *bag = bb_create(S);
    assert(bag);

    int tnum;
    for (tnum = 0; tnum < N; tnum++)
    {
        producer_arg_t *parg = malloc(sizeof(producer_arg_t));
        assert(parg != NULL);

        parg->tnum = tnum;
        parg->bag = bag;

        pthread_create(&prod[tnum], NULL, producer, parg);
    }

    // shared result
    boxed_int_t *sum = malloc(sizeof(boxed_int_t));
    pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

    assert(sum != NULL);

    for (tnum = 0; tnum < N; tnum++)
    {
        consumer_arg_t *carg = malloc(sizeof(consumer_arg_t));
        assert(carg != NULL);

        carg->tnum = tnum;
        carg->bag = bag;
        carg->sum = sum;
        carg->mutex = &mutex;

        pthread_create(&cons[tnum], NULL, consumer, carg);
    }

    for (tnum = 0; tnum < N; tnum++)
    {
        pthread_join(prod[tnum], NULL);
    }
    bb_close(bag, N);

    for (tnum = 0; tnum < N; tnum++)
    {
        pthread_join(cons[tnum], NULL);
    }

    printf("expected sum=%d\n", N * (N + 1) / 2);
    printf("computed sum=%d\n", sum->value);

    return 0;
}

Makefile

main: main.c bag.c bag.h
	gcc -O3 -pthread -Wall bag.c main.c -o $@


clean:
	rm -f main