diff --git a/src/SConscript b/src/SConscript index d5dff4d..c28fcde 100644 --- a/src/SConscript +++ b/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') diff --git a/src/ucore_threadqueue.c b/src/ucore_threadqueue.c new file mode 100644 index 0000000..d5a10b0 --- /dev/null +++ b/src/ucore_threadqueue.c @@ -0,0 +1,202 @@ +#include +#include +#include +#include +#include +#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; + +} diff --git a/src/ucore_threadqueue.h b/src/ucore_threadqueue.h new file mode 100644 index 0000000..a2f9c6d --- /dev/null +++ b/src/ucore_threadqueue.h @@ -0,0 +1,221 @@ +#ifndef _THREADQUEUE_H_ +#define _THREADQUEUE_H_ 1 + +#include + +#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 + */ + +/** + * @mainpage + * @htmlonly + *
+ *  Copyright (c) 2002-2003 Nils O. Selåsdal .
+ *  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.
+ *  
+ * @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 diff --git a/test/threadqueue.c b/test/threadqueue.c new file mode 100644 index 0000000..f2b758f --- /dev/null +++ b/test/threadqueue.c @@ -0,0 +1,70 @@ +#include +#include +#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; +}