arrayfifo.c

/**
 * Simple circular event queue using ints with a asynchronous, user
 * provided handler (consumer). All locked by it self. Using pthread.
 *
 * Is intended to work only with positive integers
 *
 * This can be used to delay work on places where you need to return fast.
 *
 * compiling: gcc -o arayfifo arrayfifo.c -lpthread
 */

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

#define EVQ_MAX 10
#define EVQ_SIZ (EVQ_MAX - 1)

struct evqueue {
        int ev[EVQ_MAX];
        int start;
        int end;
        pthread_mutex_t lock;
        pthread_cond_t cond;
        pthread_t task;
        void (*user_handler)(int);
};

#define evq_diff(q) (((q)->end + 1) % EVQ_MAX)
#define evq_full(q) ((q)->start == evq_diff(q))
#define evq_empty(q) ((q)->start == (q)->end)

/* Signal user's callback that queue has events
 * to be processed
 *
 * q => The queue
 */
void evq_signal(struct evqueue *q)
{
        pthread_cond_signal(&q->cond);
}

/* Unqueue event and return to user
 *
 * Returns the event on success and -1 on error.
 */
int evq_get(struct evqueue *q)
{
        int ev;

        pthread_mutex_lock(&q->lock);
        if (evq_empty(q)) {
                ev = -1;
        } else {
                ev = q->ev[q->start];
                q->start = (q->start + 1) % EVQ_MAX;
        }
        pthread_mutex_unlock(&q->lock);
        return ev;
}

/* this is not part if API :-P */
static void *evq_handler(void *evq)
{
        struct evqueue *evqp = evq;

        printf("Starting event handler\n");
        for (;;) {
                pthread_mutex_lock(&evqp->lock);
                while (evq_empty(evqp))
                        pthread_cond_wait(&evqp->cond, &evqp->lock);
                pthread_mutex_unlock(&evqp->lock);

                evqp->user_handler(evq_get(evqp));
        }
}

/* Initialize queue, it has to be allocated first
 * Second argument is the user callback handler.
 *
 * Is called asynchronously every time use calls
 * evq_signal() and queue is not empty.
 *
 * The prototype is void handler(int), the event to be
 * processed is passed as only parameter.
 */
void evq_init(struct evqueue *q, void (*handler)(int))
{
        assert(q);
        assert(handler);

        q->start = q->end = 0;
        q->user_handler = handler;
        pthread_mutex_init(&q->lock, NULL);
        pthread_cond_init(&q->cond, NULL);
        pthread_create(&q->task, NULL, evq_handler, q);
}

/* Enqueue and event to the queue
 *
 * Returns 0 on success and -1 on error
 */
int evq_put(struct evqueue *q, int ev)
{
        int rc = 0;

        pthread_mutex_lock(&q->lock);
        if (evq_full(q)) {
                rc = -1;
        } else {
                q->ev[q->end] = ev;
                q->end = (q->end + 1) % EVQ_MAX;
        }
        pthread_mutex_unlock(&q->lock);
        return rc;
}


/* --------- User stuff ------------- */

/* Declare a queue */
static struct evqueue evq;

/* Declare a hanlder */
void my_handler(int ev)
{
        printf("Handling event on user handler %d\n", ev);
}


int main(void)
{
        int i;

        /* Initialize the queue */
        evq_init(&evq, my_handler);

        for (i = 0; i < 2000000 ; i++) {
                if (evq_put(&evq, i)) { /* Enqueue events */
                        puts("Full");
                        usleep(20 * 1000);
                }
                evq_signal(&evq); /* Ask for their processing */
        }

        return 0;
}