queue.c

#include <stdlib.h>
#include <pthread.h>
#include "logger.h"
#include "queue.h"

static void lock(queue_t *q)
{
	pthread_mutex_lock(&(q->mutex));
}

static void unlock(queue_t *q)
{
	pthread_mutex_unlock(&(q->mutex));
}

queue_t * queue_create(void)
{
    queue_t *q;
	if (!(q = (queue_t *)malloc(sizeof(queue_t))))
	{
		log_fatal("Allocation error");
		return NULL;
	}

	q->size = 0;
    q->head = NULL;
    q->tail = NULL;

	pthread_mutexattr_init(&(q->mutex_attr));
	pthread_mutexattr_settype(&(q->mutex_attr), PTHREAD_MUTEX_RECURSIVE);
	pthread_mutex_init(&(q->mutex), &(q->mutex_attr));
	pthread_cond_init(&(q->condvar), NULL);

    return q;
}

void queue_destroy(queue_t *q)
{
	pthread_cond_destroy(&(q->condvar));
    pthread_mutex_destroy(&(q->mutex));
	pthread_mutexattr_destroy(&(q->mutex_attr));
    free(q);
    q = NULL;
}

int queue_is_empty(queue_t *q)
{
    lock(q);
    int result = 0;
    if (q->head == NULL)
        result = 1;
    unlock(q);
    return result;
}

int queue_size(queue_t *q)
{
    lock(q);
    int result = q->size;
    unlock(q);
    return result;
}

void queue_enqueue(queue_t *q, void *el)
{
    queue_node_t *node = malloc(sizeof(*node));
    node->value = el;
    node->next = NULL;

    lock(q);

    if (q->head == NULL)
    {
        // List empty
        q->head = node;
        q->tail = node;
    }
    else
    {
        // Rewrite
        queue_node_t *tail = q->tail;
        tail->next = node;
        q->tail = node;
    }

	q->size++;

	pthread_cond_signal(&(q->condvar)); // signal that data has arrived on the queue
    unlock(q);
}

void *queue_dequeue(queue_t *q)
{
    lock(q);

	while (queue_is_empty(q) > 0)
		pthread_cond_wait(&(q->condvar), &(q->mutex));

    queue_node_t *node = q->head;

    if (node == NULL)
    {
        unlock(q);
        return NULL;
    }
    else
    {
        q->head = node->next;

        void *value = node->value;
        free(node);

		q->size--;
        unlock(q);
        return value;
    }
}

queue.h

#ifndef QUEUE_H
#define QUEUE_H

#include <pthread.h>

typedef struct queue_node_s {
    void *next;
    void *value;
} queue_node_t;

typedef struct queue_s {
    queue_node_t *head;
    queue_node_t *tail;
    pthread_mutex_t mutex;
	pthread_mutexattr_t mutex_attr;
	pthread_cond_t condvar;
	size_t size;
} queue_t;

queue_t *queue_create(void);
void queue_destroy(queue_t *q);
int queue_is_empty(queue_t *q);
int queue_size(queue_t *q);
void queue_enqueue(queue_t *q, void *el);
void *queue_dequeue(queue_t *q);

#endif