From 3e7fd8a34b6ad44f8899ab79218e7961e924ccd4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nils=20O=2E=20Sel=C3=A5sdal?= Date: Wed, 2 Jan 2013 16:54:20 +0100 Subject: [PATCH] Implement tryadd, adding itmes with a timeout --- include/ucore/ucore_threadqueue.h | 12 +++++- src/ucore_threadqueue.c | 31 ++++++++++++++-- test/test_threadqueue.c | 62 +++++++++++++++++++++++++++++++ 3 files changed, 99 insertions(+), 6 deletions(-) diff --git a/include/ucore/ucore_threadqueue.h b/include/ucore/ucore_threadqueue.h index 8bc3c17..78b8496 100644 --- a/include/ucore/ucore_threadqueue.h +++ b/include/ucore/ucore_threadqueue.h @@ -166,11 +166,19 @@ int uc_thread_queue_init(struct uc_threadqueue *queue, long max_elements); * the front of the queue, i.e. it can be considered a "priority" message. * * @param queue Pointer to the queue on where the message should be added. + * @param timeout timeout on how long to wait on a message, or NULL for no timeout * @param data the "message". - * @return 0 on succes ENOMEM if out of memory EINVAL if queue is NULL, or other + * @return 0 on succes ENOMEM if out of memory EINVAL if queue is NULL, ETIMEDOUT if timeout occured or other * errno values if pthread functions failed. */ -int uc_thread_queue_add(struct uc_threadqueue *queue, struct uc_threadmsg *msg); +int uc_thread_queue_tryadd(struct uc_threadqueue *queue, const struct timespec *timeout, struct uc_threadmsg *msg); + +/** + * Like uc_thread_queue_add but will wait indefintly. + * @see uc_thread_queue_add + */ +#define uc_thread_queue_add(queue, msg)\ +uc_thread_queue_tryadd(queue, NULL, msg) /** * Gets a message from a queue diff --git a/src/ucore_threadqueue.c b/src/ucore_threadqueue.c index 9866cfc..3fcfe42 100644 --- a/src/ucore_threadqueue.c +++ b/src/ucore_threadqueue.c @@ -57,21 +57,44 @@ int uc_thread_queue_init(struct uc_threadqueue *queue, long max_elements) return rc; } -int uc_thread_queue_add(struct uc_threadqueue *queue, struct uc_threadmsg *msg) +int uc_thread_queue_tryadd(struct uc_threadqueue *queue, const struct timespec *timeout, struct uc_threadmsg *msg) { + int rc = 0; + struct timespec abstimeout; + if (queue == NULL || msg == NULL) { return EINVAL; } + if (timeout) { + uc_get_absolute_time(&abstimeout, timeout); + } + pthread_mutex_lock(&queue->mutex); //Wait if the queue is full - while(queue->num_elements >= queue->max_elements) { + while(queue->num_elements >= queue->max_elements && rc != ETIMEDOUT) { + queue->num_write_waiters++; - pthread_cond_wait(&queue->write_cond, &queue->mutex); + if(timeout) { + //fastpath, check for a poll + if(timeout->tv_sec == 0 && timeout->tv_nsec == 0) { + rc = ETIMEDOUT; + } else { + rc = pthread_cond_timedwait(&queue->write_cond, &queue->mutex, &abstimeout); + } + + } else { + pthread_cond_wait(&queue->write_cond, &queue->mutex); + } queue->num_write_waiters--; } - + + if (rc == ETIMEDOUT) { + pthread_mutex_unlock(&queue->mutex); + return rc; + } + //insert the new element msg->next = NULL; if (msg->msgtype >= 0) { //add at the tail diff --git a/test/test_threadqueue.c b/test/test_threadqueue.c index f57d579..4096377 100644 --- a/test/test_threadqueue.c +++ b/test/test_threadqueue.c @@ -55,6 +55,8 @@ START_TEST (test_thread_queue_one_thread) END_TEST + + //pretty ugly.. static int test_thread_queue_walk_global; static void test_thread_queue_walk_func(struct uc_threadmsg *msg) @@ -253,6 +255,63 @@ START_TEST (test_thread_queue_timeout2) END_TEST +START_TEST (test_thread_queue_tryadd_timeout_0_0) + + struct uc_threadqueue queue; + int i; + struct thr_msg msg[3]; + memset(msg, 0, sizeof msg); + struct timespec timeout = {0,0}; + + fail_if(uc_thread_queue_init(&queue, 2) != 0); + for (i = 0; i < 2; i++) { + fail_if(uc_thread_queue_add(&queue, &msg[i].msg) != 0); + } + + fail_if(uc_thread_queue_tryadd(&queue, &timeout, &msg[2].msg) != ETIMEDOUT); + + fail_if(uc_thread_queue_destroy(&queue, NULL) != 0); + +END_TEST + +START_TEST (test_thread_queue_tryadd_timeout_0_100) + + struct uc_threadqueue queue; + int i; + struct thr_msg msg[3]; + memset(msg, 0, sizeof msg); + struct timespec timeout = {0,100}; + + fail_if(uc_thread_queue_init(&queue, 2) != 0); + for (i = 0; i < 2; i++) { + fail_if(uc_thread_queue_add(&queue, &msg[i].msg) != 0); + } + + fail_if(uc_thread_queue_tryadd(&queue, &timeout, &msg[2].msg) != ETIMEDOUT); + + fail_if(uc_thread_queue_destroy(&queue, NULL) != 0); + +END_TEST + +START_TEST (test_thread_queue_tryadd_no_timeout_0_100) + + struct uc_threadqueue queue; + int i; + struct thr_msg msg[3]; + memset(msg, 0, sizeof msg); + struct timespec timeout = {0,100}; + + fail_if(uc_thread_queue_init(&queue, 3) != 0); + for (i = 0; i < 2; i++) { + fail_if(uc_thread_queue_add(&queue, &msg[i].msg) != 0); + } + + fail_if(uc_thread_queue_tryadd(&queue, &timeout, &msg[2].msg) != 0); + + fail_if(uc_thread_queue_destroy(&queue, NULL) != 0); + +END_TEST + Suite *threadqueue_suite(void) { @@ -268,6 +327,9 @@ Suite *threadqueue_suite(void) tcase_add_test(tc1, test_thread_queue_multiple_writers); tcase_add_test(tc1, test_thread_queue_timeout1); tcase_add_test(tc1, test_thread_queue_timeout2); + tcase_add_test(tc1, test_thread_queue_tryadd_timeout_0_0); + tcase_add_test(tc1, test_thread_queue_tryadd_timeout_0_100); + tcase_add_test(tc1, test_thread_queue_tryadd_no_timeout_0_100); //lets start with this: tcase_set_timeout(tc1, 15);