1.1 --- /dev/null	Thu Jan 01 00:00:00 1970 +0000
     1.2 +++ b/src/tqueue.c	Fri Mar 02 23:49:10 2012 +1000
     1.3 @@ -0,0 +1,137 @@
     1.4 +/**
     1.5 + * $Id$
     1.6 + *
     1.7 + * Bounded, blocking queue for inter-thread communication.
     1.8 + *
     1.9 + * Copyright (c) 2012 Nathan Keynes.
    1.10 + *
    1.11 + * This program is free software; you can redistribute it and/or modify
    1.12 + * it under the terms of the GNU General Public License as published by
    1.13 + * the Free Software Foundation; either version 2 of the License, or
    1.14 + * (at your option) any later version.
    1.15 + *
    1.16 + * This program is distributed in the hope that it will be useful,
    1.17 + * but WITHOUT ANY WARRANTY; without even the implied warranty of
    1.18 + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    1.19 + * GNU General Public License for more details.
    1.20 + */
    1.21 +
    1.22 +#include <assert.h>
    1.23 +#include <pthread.h>
    1.24 +#include "tqueue.h"
    1.25 +
    1.26 +#define TQUEUE_LENGTH 64
    1.27 +
    1.28 +typedef struct {
    1.29 +    tqueue_callback callback;
    1.30 +    void *data;
    1.31 +    gboolean synchronous;
    1.32 +} tqueue_entry;
    1.33 +
    1.34 +struct {
    1.35 +    pthread_mutex_t mutex;
    1.36 +    pthread_cond_t consumer_wait;
    1.37 +    pthread_cond_t producer_sync_wait;
    1.38 +    pthread_cond_t producer_full_wait;
    1.39 +    int head;  /* next item returned by dequeue */
    1.40 +    int tail;  /* next item filled in by enqueue */
    1.41 +    int last_result; /* Result value of last dequeued callback */
    1.42 +    tqueue_entry tqueue[TQUEUE_LENGTH];
    1.43 +} tqueue = {PTHREAD_MUTEX_INITIALIZER, PTHREAD_COND_INITIALIZER, PTHREAD_COND_INITIALIZER, PTHREAD_COND_INITIALIZER, 0, 0, -1};
    1.44 +
    1.45 +/************** Producer thread **************/
    1.46 +#define TQUEUE_EMPTY() (tqueue.head == tqueue.tail)
    1.47 +#define TQUEUE_FULL() ((tqueue.head == tqueue.tail+1) || (tqueue.head == 0 && tqueue.tail == TQUEUE_LENGTH))
    1.48 +
    1.49 +static void tqueue_enqueue( tqueue_callback callback, void *data, gboolean sync )
    1.50 +{
    1.51 +    assert( !TQUEUE_FULL() );
    1.52 +    tqueue.tqueue[tqueue.tail].callback = callback;
    1.53 +    tqueue.tqueue[tqueue.tail].data = data;
    1.54 +    tqueue.tqueue[tqueue.tail].synchronous = sync;
    1.55 +    tqueue.tail++;
    1.56 +}
    1.57 +
    1.58 +/**
    1.59 + * Add a message to the UI queue and return immediately.
    1.60 + */
    1.61 +void tqueue_post_message( tqueue_callback callback, void *data )
    1.62 +{
    1.63 +    pthread_mutex_lock(&tqueue.mutex);
    1.64 +    if( TQUEUE_FULL() ) {
    1.65 +        /* Wait for the queue to clear */
    1.66 +        pthread_cond_wait(&tqueue.producer_full_wait, &tqueue.mutex);
    1.67 +    }
    1.68 +    tqueue_enqueue( callback, data, FALSE );
    1.69 +    pthread_cond_signal(&tqueue.consumer_wait);
    1.70 +    pthread_mutex_unlock(&tqueue.mutex);
    1.71 +}
    1.72 +
    1.73 +/**
    1.74 + * Add a message to the UI queue and wait for it to be handled.
    1.75 + * @return the result from the handler function.
    1.76 + */
    1.77 +int tqueue_send_message( tqueue_callback callback, void *data )
    1.78 +{
    1.79 +    pthread_mutex_lock(&tqueue.mutex);
    1.80 +    if( TQUEUE_FULL() ) {
    1.81 +        /* Wait for the queue to clear */
    1.82 +        pthread_cond_wait(&tqueue.producer_full_wait, &tqueue.mutex);
    1.83 +    }
    1.84 +    tqueue_enqueue( callback, data, TRUE );
    1.85 +    pthread_cond_signal(&tqueue.consumer_wait);
    1.86 +    pthread_cond_wait(&tqueue.producer_sync_wait, &tqueue.mutex);
    1.87 +    return tqueue.last_result;
    1.88 +    pthread_mutex_unlock(&tqueue.mutex);
    1.89 +}
    1.90 +
    1.91 +/************** Consumer thread **************/
    1.92 +
    1.93 +/* Note: must be called with mutex locked */
    1.94 +static void tqueue_process_loop() {
    1.95 +    while( !TQUEUE_EMPTY() ) {
    1.96 +        gboolean wasFull = TQUEUE_FULL();
    1.97 +        tqueue_callback callback = tqueue.tqueue[tqueue.head].callback;
    1.98 +        void *data = tqueue.tqueue[tqueue.head].data;
    1.99 +        gboolean sync = tqueue.tqueue[tqueue.head].synchronous;
   1.100 +        tqueue.head++;
   1.101 +
   1.102 +        if( wasFull ) {
   1.103 +            pthread_cond_signal( &tqueue.producer_full_wait );
   1.104 +        }
   1.105 +
   1.106 +        pthread_mutex_unlock(&tqueue.mutex);
   1.107 +        int result = callback(data);
   1.108 +        pthread_mutex_lock(&tqueue.mutex);
   1.109 +        if( sync ) {
   1.110 +            tqueue.last_result = result;
   1.111 +            pthread_cond_signal( &tqueue.producer_sync_wait );
   1.112 +        }
   1.113 +    }
   1.114 +}
   1.115 +
   1.116 +/**
   1.117 + * Process all messages in the queue, if any.
   1.118 + */
   1.119 +void tqueue_process_all()
   1.120 +{
   1.121 +    pthread_mutex_lock(&tqueue.mutex);
   1.122 +    if( !TQUEUE_EMPTY() ) {
   1.123 +        tqueue_process_loop();
   1.124 +    }
   1.125 +    pthread_mutex_unlock(&tqueue.mutex);
   1.126 +}
   1.127 +
   1.128 +/**
   1.129 + * Process the first message in the queue. If no messages are on the
   1.130 + * queue, waits for the next one to be queued and then processes it.
   1.131 + */
   1.132 +void tqueue_process_wait()
   1.133 +{
   1.134 +    pthread_mutex_lock(&tqueue.mutex);
   1.135 +    if( TQUEUE_EMPTY() ) {
   1.136 +        pthread_cond_wait( &tqueue.consumer_wait, &tqueue.mutex );
   1.137 +    }
   1.138 +    tqueue_process_loop();
   1.139 +    pthread_mutex_unlock(&tqueue.mutex);
   1.140 +}