gngrwzrd.lists

// compiled as:
// gcc gg.c -Wall -pedantic -std=c99 -pthread -D_XOPEN_SOURCE=700

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

typedef struct
{
    pthread_rwlock_t rwlock;

    // mux protects 'write_owner' and 'recursive_wlocks' for reading,
    // for writing the write-lock must be acquired as well
    pthread_mutex_t mux;
    pthread_t write_owner;
    int recursive_wlocks;
} rrw_lock_t;

// initilize a rrw_lock_t
int rrw_init(rrw_lock_t *rrwl)
{
    int res = pthread_mutex_init(&rrwl->mux, NULL);
    if (res)
        return res;

    res = pthread_rwlock_init(&rrwl->rwlock, NULL);
    if (res)
    {
        pthread_mutex_destroy(&rrwl->mux);
        return res;
    }

    rrwl->recursive_wlocks = 0;
    return 0;
}

// destroy a rrw_lock_t
int rrw_destroy(rrw_lock_t *rrwl)
{
    int r1 = pthread_rwlock_destroy(&rrwl->rwlock);
    int r2 = pthread_mutex_destroy(&rrwl->mux);
    if (r1)
        return r1;
    if (r2)
        return r2;
    return 0;
}

// acquire rrwl->mux
void rrw_acquire_mux(rrw_lock_t *rrwl)
{
    int res = pthread_mutex_lock(&rrwl->mux);
    if (res)
    {
        perror("rrw_acquire_mux::pthread_mutex_lock");
        exit(res);
    }
}

// release rrwl->mux
void rrw_release_mux(rrw_lock_t *rrwl)
{
    int res = pthread_mutex_unlock(&rrwl->mux);
    if (res)
    {
        perror("rrw_release_mux::pthread_mutex_unlock");
        exit(res);
    }
}

// acquire read lock
int rrw_read_lock(rrw_lock_t *rrwl)
{
    return pthread_rwlock_rdlock(&rrwl->rwlock);
}

// release read lock
int rrw_read_unlock(rrw_lock_t *rrwl)
{
    return pthread_rwlock_unlock(&rrwl->rwlock);
}

// acquire write lock
int rrw_write_lock(rrw_lock_t *rrwl)
{
    pthread_t self = pthread_self();
    bool call_wrlock = false;
    int res;

    rrw_acquire_mux(rrwl);
        // check if we already own write lock
        if (rrwl->recursive_wlocks &&
            pthread_equal(self, rrwl->write_owner))
        {
            ++(rrwl->recursive_wlocks);
        }
        else
        {
            call_wrlock = true;
        }
    rrw_release_mux(rrwl);

    if (!call_wrlock)
        return 0;

    // we don't own write lock, try to acquire it
    res = pthread_rwlock_wrlock(&rrwl->rwlock);
    if (res)
        return res;

    rrw_acquire_mux(rrwl);
        assert(rrwl->recursive_wlocks == 0);
        rrwl->recursive_wlocks = 1;
        rrwl->write_owner = self;
    rrw_release_mux(rrwl);

    return 0;
}

// release write lock
int rrw_write_unlock(rrw_lock_t *rrwl)
{
    int res = 0;

    rrw_acquire_mux(rrwl);
        // shouldn't be calling this if not the owner
        assert(rrwl->recursive_wlocks &&
               pthread_equal(pthread_self(), rrwl->write_owner));

        if (rrwl->recursive_wlocks > 1)
        {
            --(rrwl->recursive_wlocks);
        }
        else
        {
             res = pthread_rwlock_unlock(&rrwl->rwlock);
             if (!res)
                rrwl->recursive_wlocks = 0;
        }
    rrw_release_mux(rrwl);

    return res;
}


//-----------------------------------------------------------------------------
// Test code

typedef struct
{
    rrw_lock_t *rrwl;
    int thread_num;
} thread_param;

void* reader_thread(void *p)
{
    int counter = 0;
    thread_param tp = *((thread_param*)p);
    free(p);

    for (;;)
    {
        rrw_read_lock(tp.rrwl);
        rrw_read_lock(tp.rrwl);
        rrw_read_unlock(tp.rrwl);
        rrw_read_unlock(tp.rrwl);
        if (++counter % 10000 == 0)
        {
            counter = 0;
            fprintf(stderr, "Reader %d alive\n", tp.thread_num);
        }
    }

    return 0;
}

void* writer_thread(void *p)
{
    int counter = 0;
    thread_param tp = *((thread_param*)p);
    free(p);

    for (;;)
    {

        rrw_write_lock(tp.rrwl);
        rrw_write_lock(tp.rrwl);
        rrw_write_unlock(tp.rrwl);
        rrw_write_unlock(tp.rrwl);
        fprintf(stderr,"writer\n");
        fprintf(stderr,"writer: count: %i\n",counter);
        if (++counter % 500000 == 0)
        {

            counter = 0;
            fprintf(stderr, "Writer %d alive\n", tp.thread_num);
        }
    }

    return 0;
}

#define NUM_READERS 4
#define NUM_WRITERS 2

int main()
{
    int n, res;
    rrw_lock_t l;
    pthread_t readers[NUM_READERS];
    pthread_t writers[NUM_WRITERS];

    res = rrw_init(&l);
    if (res)
    {
        perror("rrw_init");
        exit(res);
    }

    fprintf(stderr, "Starting %d readers and %d writers\n",
            NUM_READERS, NUM_WRITERS);

    for (n = 0; n < NUM_READERS; ++n)
    {
        thread_param *tp = malloc(sizeof(thread_param));
        tp->rrwl = &l;
        tp->thread_num = n + 1;

        res = pthread_create(readers + n, NULL, &reader_thread, tp);
        if (res)
        {
            perror("reader-pthread_create");
            exit(res);
        }
    }

    for (n = 0; n < NUM_WRITERS; ++n)
    {
        thread_param *tp = malloc(sizeof(thread_param));
        tp->rrwl = &l;
        tp->thread_num = n + 1;

        res = pthread_create(writers + n, NULL, &writer_thread, tp);
        if (res)
        {
            perror("writer-pthread_create");
            exit(res);
        }
    }

    // currently runs for ever, make sure all threads continue to ouput to stderr

    for (n = 0; n < NUM_READERS; ++n)
        pthread_join(readers[n], NULL);

    for (n = 0; n < NUM_WRITERS; ++n)
        pthread_join(writers[n], NULL);

    res = rrw_destroy(&l);
    if (res)
        perror("rrw_destroy");
    return res;
}