Add bounded thread_queue implementation

src/SConscript

@@ -6,6 +6,7 @@ ucore_env = env.Clone()
 #substitute @version_xx@ strings
 subst_version_info = dict(('@' + key + '@', version_info[key]) for key in version_info)
 ucore_env.Substfile('ucore_version.c.in', SUBST_DICT = subst_version_info)
+ucore_env.Append(CFLAGS = ['-pthread'])
 
 sources = ucore_env.Glob('*.c')
 headers = ucore_env.Glob('*.h')

src/ucore_threadqueue.c

@@ -0,0 +1,202 @@
+#include <stdlib.h>
+#include <string.h>
+#include <errno.h>
+#include <pthread.h>
+#include <sys/time.h>
+#include "ucore_threadqueue.h"
+
+
+int uc_thread_queue_init(struct uc_threadqueue *queue, long max_elements)
+{
+    int rc = 0;
+    if (queue == NULL || max_elements <= 0) {
+        return EINVAL;
+    }
+
+    memset(queue, 0, sizeof(struct uc_threadqueue));
+
+    rc = pthread_cond_init(&queue->read_cond, NULL);
+    if (rc != 0) {
+        return rc;
+    }
+
+    rc = pthread_cond_init(&queue->write_cond, NULL);
+    if (rc != 0) {
+        pthread_cond_destroy(&queue->read_cond);
+        return rc;
+    }
+
+    rc = pthread_mutex_init(&queue->mutex, NULL);
+    if (rc != 0) {
+        pthread_cond_destroy(&queue->read_cond);
+        pthread_cond_destroy(&queue->write_cond);
+        return rc;
+    }
+
+    queue->max_elements = max_elements;
+    queue->last = &queue->first;
+
+    return 0;
+
+}
+
+int uc_thread_queue_add(struct uc_threadqueue *queue, struct uc_threadmsg *msg)
+{
+    pthread_mutex_lock(&queue->mutex);
+
+    //Wait if the queue is full
+    while(queue->num_elements >= queue->max_elements) {
+        queue->num_write_waiters++;
+        pthread_cond_wait(&queue->write_cond, &queue->mutex);
+        queue->num_write_waiters--;
+    }
+    
+    //insert the new element
+    msg->next = NULL;
+    if(msg->msgtype >= 0) { //add at the tail
+        *(queue->last) = msg;
+        queue->last = &msg->next;
+    } else { //add at the head ("priority message")
+        msg->next = queue->first;
+        if(queue->first == NULL) {
+            queue->last = &msg->next;
+        }
+        queue->first = msg;
+    }
+
+    //signal blocked readers
+	if(queue->num_read_waiters == 1) {
+        //if there's just one thread waiting to pop elements
+        //use _cond_signal. _cond_broadcast can be more expensive (os dependent)
+		pthread_cond_signal(&queue->read_cond);
+	} else if(queue->num_read_waiters > 1) {
+        //signall all threads that there's items available.
+		pthread_cond_broadcast(&queue->read_cond);
+    }
+
+    queue->num_elements++;
+
+    pthread_mutex_unlock(&queue->mutex);
+
+    return 0;
+
+}
+
+int uc_thread_queue_get(struct uc_threadqueue *queue, const struct timespec *timeout, struct uc_threadmsg **msg)
+{
+    int rc = 0;
+    struct timespec abstimeout;
+    struct uc_threadmsg *first_msg;
+
+    if (queue == NULL || msg == NULL) {
+        return EINVAL;
+    }
+    if (timeout) {
+        struct timeval now;
+
+        gettimeofday(&now, NULL);
+        abstimeout.tv_sec = now.tv_sec + timeout->tv_sec;
+        abstimeout.tv_nsec = (now.tv_usec * 1000) + timeout->tv_nsec;
+        if (abstimeout.tv_nsec >= 1000000000) {
+            abstimeout.tv_sec++;
+            abstimeout.tv_nsec -= 1000000000;
+        }
+    }
+
+    pthread_mutex_lock(&queue->mutex);
+
+    /* Will wait until awakened by a signal or broadcast */
+    while (queue->first == NULL && rc != ETIMEDOUT) {  
+        //Need to loop to handle spurious wakeups, or the case
+        //another thread popped the remaining elment before we did
+        queue->num_read_waiters++;
+        if (timeout) {
+            rc = pthread_cond_timedwait(&queue->read_cond, &queue->mutex, &abstimeout);
+        } else {
+            pthread_cond_wait(&queue->read_cond, &queue->mutex);
+
+        }
+        queue->num_read_waiters--;
+    }
+
+    if (rc == ETIMEDOUT) {
+        pthread_mutex_unlock(&queue->mutex);
+        return rc;
+    }
+
+    //remove the first element
+    first_msg = queue->first;
+    queue->first = first_msg->next;
+    if(queue->first == NULL)
+        queue->last = &queue->first;
+    first_msg->next = NULL;
+
+    queue->num_elements--;
+
+    //signal blocked writers
+	if(queue->num_write_waiters == 1) {
+        //if there's just one thread waiting to push elements
+        //use _cond_signal. _cond_broadcast can be more expensive (os dependent)
+		pthread_cond_signal(&queue->write_cond);
+	} else if(queue->num_write_waiters > 1) {
+        //signall all threads that there's items available.
+		pthread_cond_broadcast(&queue->write_cond);
+    }
+
+    pthread_mutex_unlock(&queue->mutex);
+    *msg = first_msg;
+
+    return 0;
+}
+
+//maybe caller should supply a callback for cleaning the elements ?
+int uc_thread_queue_cleanup(struct uc_threadqueue *queue, int freedata)
+{
+    int rc;
+
+    if (queue == NULL) {
+        return EINVAL;
+    }
+
+    pthread_mutex_lock(&queue->mutex);
+
+     /* We don't always know if there's threads still using the queue,
+      * e.g. blocking on the mutex.
+     * It's up to the user to ensure consistency when destroying the queue.
+     *
+     * But if we *know* there are some threads still using the queue,
+     * we can't destroy it.
+     */
+    if(queue->num_read_waiters != 0 || queue->num_write_waiters != 0) {
+        pthread_mutex_unlock(&queue->mutex);
+        return EBUSY;
+    }
+
+    if(freedata) {
+        struct uc_threadmsg *p;
+        struct uc_threadmsg *next;
+        for(p = queue->first; p; p = next) {
+            next = p->next;
+            free(p);
+        }
+    }
+
+    pthread_mutex_unlock(&queue->mutex);
+    rc = pthread_mutex_destroy(&queue->mutex);
+    pthread_cond_destroy(&queue->read_cond);
+    pthread_cond_destroy(&queue->write_cond);
+
+    return rc;
+
+}
+
+long thread_queue_length(struct uc_threadqueue *queue)
+{
+    long length;
+    // get the length properly
+    pthread_mutex_lock(&queue->mutex);
+    length = queue->num_elements;
+    pthread_mutex_unlock(&queue->mutex);
+    return length;
+
+}

src/ucore_threadqueue.h

@@ -0,0 +1,221 @@
+#ifndef _THREADQUEUE_H_
+#define _THREADQUEUE_H_ 1
+
+#include <pthread.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+/**
+ * @defgroup ThreadQueue ThreadQueue
+ *
+ * Little API for waitable queues, typically used for passing messages
+ * between threads.
+ * 
+ * @author Nils O. Selåsdal <NOS@Utel.no>
+ */
+
+/**
+ * @mainpage
+ *  @htmlonly
+ *  <pre>
+ *  Copyright (c) 2002-2003 Nils O. Selåsdal <NOS@Utel.no>.
+ *  All rights reserved, all wrongs reversed.
+ *  
+ *  Redistribution and use in source and binary forms, with or without
+ *  modification, are permitted provided that the following conditions
+ *  are met:
+ *
+ *  1. Redistributions of source code must retain the above copyright
+ *     notice, this list of conditions and the following disclaimer.
+ *  2. Redistributions in binary form must reproduce the above copyright
+ *     notice, this list of conditions and the following disclaimer in the
+ *     documentation and/or other materials provided with the distribution.
+ *
+ *  THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
+ *  IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
+ *  OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
+ *  IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
+ *  INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
+ *  NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ *  DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ *  THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ *  (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
+ *  THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *  </pre>
+ *  @endhtmlonly
+ */
+ 
+/**
+ * A thread message.
+ *
+ * @ingroup ThreadQueue
+ *
+ * This is used for passing to #thread_queue_get for retreive messages.
+ * the date is stored in the data member, the message type in the  #msgtype.
+ * 
+ * Typical:
+ * @code
+ * struct threadmsg;
+ * struct myfoo *foo;
+ * while(1)
+ * 	ret = thread_queue_get(&queue,NULL,&message);
+ * 	..
+ * 	foo = msg.data;
+ * 	switch(msg.msgtype){
+ * 		...
+ * 	}
+ * }
+ * @endcode
+ *
+ */
+struct uc_threadmsg{
+	/**
+	 * Holds the messagetype
+	 */
+	long msgtype;
+
+    struct uc_threadmsg *next;
+};
+
+
+/**
+ * A ThreadQueue
+ * 
+ * @ingroup ThreadQueue
+ * 
+ * You should threat this struct as opaque, never ever set/get any
+ * of the variables. You have been warned.
+ */
+struct uc_threadqueue {
+/**
+ * Number of elements in the queue, never set this, never read this.
+ * Use #threadqueue_length to read it.
+ */
+	long num_elements;
+
+/** Max number of elements this queue will hold */
+    long max_elements;
+/**
+ * Mutex for the queue, never touch.
+ */
+	pthread_mutex_t mutex;
+/**
+ * Condition variable for readers on the queue, never touch.
+ */
+	pthread_cond_t read_cond;
+/**
+ * Condition variable for writers on the queue (if the queue is full) never touch.
+ */
+	pthread_cond_t write_cond;
+/** 
+ * Number of threads blocking on writing to the queue
+ */
+    long num_read_waiters;
+/** 
+ * Number of threads blocking on reading from  the queue
+ */
+    long num_write_waiters;
+
+/**
+ * Internal pointers for the queue, never touch.
+ */
+	struct uc_threadmsg *first,**last;
+};
+
+/**
+ * Initializes a queue.
+ * 
+ * @ingroup ThreadQueue
+ * 
+ * thread_queue_init initializes a new threadqueue. A new queue must always
+ * be initialized before it is used.
+ *
+ * @param queue Pointer to the queue that should be initialized
+ * @param max_elements Max number of elements this queue can hold.
+ *
+ * @return 0 on success see pthread_mutex_init
+ */
+int uc_thread_queue_init(struct uc_threadqueue *queue, long max_elements);
+
+/**
+ * Adds a message to a queue
+ *
+ * @ingroup ThreadQueue
+ * 
+ * thread_queue_add adds a "message" to the specified queue, a message
+ * is just a pointer to a anything of the users choice. Nothing is copied
+ * so the user must keep track on (de)allocation of the data.
+ * A message type is also specified, it is not used for anything else than
+ * given back when a message is retreived from the queue.
+ *
+ * @param queue Pointer to the queue on where the message should be added.
+ * @param data the "message".
+ * @param msgtype a long specifying the message type, choice of the user.
+ * @return 0 on succes ENOMEM if out of memory EINVAL if queue is NULL
+ */
+int uc_thread_queue_add(struct uc_threadqueue *queue, struct uc_threadmsg *msg);
+
+/**
+ * Gets a message from a queue
+ * 
+ * @ingroup ThreadQueue
+ *
+ * thread_queue_get gets a message from the specified queue, it will block
+ * the caling thread untill a message arrives, or the (optional) timeout occurs.
+ * If timeout is NULL, there will be no timeout, and thread_queue_get will wait
+ * untill a message arrives.
+ *
+ * struct timespec is defined as:
+ * @code
+ *	struct timespec {
+ *                 long    tv_sec;         // seconds 
+ *                 long    tv_nsec;        // nanoseconds 
+ *             };
+ * @endcode
+ * 
+ * @param queue Pointer to the queue to wait on for a message.
+ * @param timeout timeout on how long to wait on a message, or NULL for no timeout
+ * @param msg pointer where the uc_threadmsg* is stored
+ *
+ * @return 0 on success EINVAL if queue is NULL ETIMEDOUT if timeout occurs
+ */
+int uc_thread_queue_get(struct uc_threadqueue *queue, const struct timespec *timeout, struct uc_threadmsg **msg);
+
+
+/**
+ * Gets the length of a queue
+ * 
+ * @ingroup ThreadQueue
+ *
+ * threadqueue_length returns the number of messages waiting in the queue
+ * 
+ * @param queue Pointer to the queue for which to get the length
+ * @return the length(number of pending messages) in the queue
+ */
+long uc_thread_queue_length( struct uc_threadqueue *queue );
+
+/**
+ * @ingroup ThreadQueue
+ * Cleans up the queue.
+ * 
+ * threadqueue_cleanup cleans up and destroys the queue.
+ * This will remove all messages from a queue, and reset it. If
+ * freedata is != 0 free(3) will be called on all pending messages in the queue
+ * You cannot call this if there are someone currently adding or getting messages
+ * from the queue.
+ * After a queue have been cleaned, it cannot be used again untill #thread_queue_init
+ * has been called on the queue.
+ * 
+ * @param queue Pointer to the queue that should be cleaned
+ * @param freedata set to nonzero if free(3) should be called on remaining
+ * messages
+ * @return 0 on success EINVAL if queue is NULL EBUSY if someone is holding any locks on the queue
+ */
+int uc_thread_queue_cleanup(struct uc_threadqueue *queue, int freedata);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

test/threadqueue.c

@@ -0,0 +1,70 @@
+#include <stdio.h>
+#include <stdlib.h>
+#include "ucore_threadqueue.h"
+
+struct uc_threadqueue queue;
+
+#define CNT 10000000
+
+struct my_msg {
+    struct uc_threadmsg uc_msg;
+    int counter;
+};
+
+void *reader(void *arg)
+{
+    int i;
+    int last;
+    for(i = 0; i < CNT/2; i++) {
+        struct uc_threadmsg *msg;
+        struct my_msg *my;
+        if(uc_thread_queue_get(&queue, NULL, &msg) != 0) {
+            printf("uc_threadqueue_get failed\n");
+            return NULL;
+        }
+        my = (struct my_msg*)msg;
+        last = my->counter;
+        free(my);
+    }
+
+    printf("reader finished, last = %d\n", last);
+
+    return NULL;
+}
+
+int main(int argc, char *argv[])
+{
+    pthread_t tid1,tid2;
+    int i;
+
+    uc_thread_queue_init(&queue, 100);
+
+    if(pthread_create(&tid1, NULL, reader, NULL) != 0) {
+        perror("pthread_create");
+        return 1;
+    }
+
+    if(pthread_create(&tid2, NULL, reader, NULL) != 0) {
+        perror("pthread_create");
+        return 1;
+    }
+
+    for(i = 0; i < CNT; i++) {
+        struct my_msg *msg = malloc(sizeof *msg);
+        if(msg == NULL) {
+            perror("malloc");
+            return 1;
+        }
+        msg->uc_msg.msgtype = 11;
+        msg->counter = i;
+        if(uc_thread_queue_add(&queue, &msg->uc_msg) != 0) {
+            printf("uc_threadqueue_add failed\n");
+            return 2;
+        }
+    }
+    printf("writer finished\n");
+    pthread_join(tid1, NULL);
+    pthread_join(tid2, NULL);
+
+    return 0;
+}