From d9517ad868a10d7fbf84db180f6af41a59aeaf2b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nils=20O=2E=20Sel=C3=A5sdal?= Date: Thu, 3 Oct 2013 22:04:33 +0200 Subject: [PATCH] Make the thread queue walk function take a cookie to the callback and implement a _clear() function to clear out the queue --- include/ucore/threadqueue.h | 25 +++++++++++++++--- src/threadqueue.c | 52 ++++++++++++++++++++++++++++++++----- test/test_threadqueue.c | 19 +++++++------- 3 files changed, 77 insertions(+), 19 deletions(-) diff --git a/include/ucore/threadqueue.h b/include/ucore/threadqueue.h index 383621d..61cfbb4 100644 --- a/include/ucore/threadqueue.h +++ b/include/ucore/threadqueue.h @@ -44,7 +44,7 @@ struct uc_threadmsg{ /** * Callback function for the walk and destroy functions. */ -typedef void (*uc_thread_queue_walk_func) (struct uc_threadmsg *msg); +typedef void (*uc_thread_queue_walk_func) (struct uc_threadmsg *msg, void *cookie); /** @@ -240,23 +240,40 @@ long uc_thread_queue_length( struct uc_threadqueue *queue ); * @param queue Pointer to the queue that should be cleaned * @param free_func pointer to function that will be called for each item. * The function must not in anyway interact with the queue. + * @param cookie pointer passed back to the uc_thread_queue_walk_func * @return 0 on success EINVAL if queue is NULL EBUSY if someone is holding any locks on the queue */ -int uc_thread_queue_destroy(struct uc_threadqueue *queue, uc_thread_queue_walk_func free_func); +int uc_thread_queue_destroy(struct uc_threadqueue *queue, uc_thread_queue_walk_func free_func, void *cookie); /** * Walk the elements of the queue, intended for debugging * * The supplied function will be called for each item in the queue, internal queue * locks are held while the function is called, so the supplied function must not - * interact with the queue, else deadlock occors. + * interact with the queue, else deadlock occurs. * * @param queue Pointer to the queue to walk * @param free_func pointer to function that will be called for each item. + * @param cookie pointer passed back to the uc_thread_queue_walk_func * The function must not in anyway interact with the queue. * @return 0 on success */ -int uc_thread_queue_walk(struct uc_threadqueue *queue, uc_thread_queue_walk_func walk_func); +int uc_thread_queue_walk(struct uc_threadqueue *queue, uc_thread_queue_walk_func walk_func, void *cookie); + + +/** + * Clear the queue. + * The free_func unless NULL, will be called for each element + * locks are held while the function is called, so the supplied function must not + * interact with the queue, else deadlock occurs. + * + * @param queue Pointer to the queue to clear + * @param free_func pointer to function that will be called for each item, intended to e.g. release memory + * @param cookie pointer passed back to the uc_thread_queue_walk_func + * @return 0 on success +*/ +int uc_thread_queue_clear(struct uc_threadqueue *queue, uc_thread_queue_walk_func free_func, void *cookie); + /** * @}*/ diff --git a/src/threadqueue.c b/src/threadqueue.c index 47308df..c5da95f 100644 --- a/src/threadqueue.c +++ b/src/threadqueue.c @@ -199,19 +199,21 @@ int uc_thread_queue_tryget(struct uc_threadqueue *queue, } static void uc_thread_queue_walk_unlocked(struct uc_threadqueue *queue, - uc_thread_queue_walk_func walk_func) + uc_thread_queue_walk_func walk_func, + void *cookie) { struct uc_threadmsg *p; struct uc_threadmsg *next; for(p = queue->first; p; p = next) { next = p->next; - walk_func(p); + walk_func(p, cookie); } } int uc_thread_queue_walk(struct uc_threadqueue *queue, - uc_thread_queue_walk_func walk_func) + uc_thread_queue_walk_func walk_func, + void *cookie) { if (queue == NULL || !walk_func) { return EINVAL; @@ -219,7 +221,7 @@ int uc_thread_queue_walk(struct uc_threadqueue *queue, pthread_mutex_lock(&queue->mutex); - uc_thread_queue_walk_unlocked(queue, walk_func); + uc_thread_queue_walk_unlocked(queue, walk_func, cookie); pthread_mutex_unlock(&queue->mutex); @@ -227,7 +229,8 @@ int uc_thread_queue_walk(struct uc_threadqueue *queue, } int uc_thread_queue_destroy(struct uc_threadqueue *queue, - uc_thread_queue_walk_func free_func) + uc_thread_queue_walk_func free_func, + void *cookie) { int rc; @@ -250,7 +253,7 @@ int uc_thread_queue_destroy(struct uc_threadqueue *queue, } if (free_func) { - uc_thread_queue_walk_unlocked(queue, free_func); + uc_thread_queue_walk_unlocked(queue, free_func, cookie); } pthread_mutex_unlock(&queue->mutex); @@ -276,3 +279,40 @@ long uc_thread_queue_length(struct uc_threadqueue *queue) return length; } + +int uc_thread_queue_clear(struct uc_threadqueue *queue, + uc_thread_queue_walk_func free_func, + void *cookie) +{ + if (queue == NULL) { + return -EINVAL; + } + + pthread_mutex_lock(&queue->mutex); + + //Let the caller free the data + if (free_func != NULL) { + uc_thread_queue_walk_unlocked(queue, free_func, cookie); + } + + //reset the elements + queue->first = NULL; + queue->num_elements = 0; + queue->last = &queue->first; + + //notify writes, there should be space now + 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); + + return 0; +} + + diff --git a/test/test_threadqueue.c b/test/test_threadqueue.c index 933645a..c0f0cb3 100644 --- a/test/test_threadqueue.c +++ b/test/test_threadqueue.c @@ -50,7 +50,7 @@ START_TEST (test_thread_queue_one_thread) fail_if(my_msg->a != i); free(my_msg); } - fail_if(uc_thread_queue_destroy(&queue, NULL) != 0); + fail_if(uc_thread_queue_destroy(&queue, NULL, NULL) != 0); END_TEST @@ -59,8 +59,9 @@ END_TEST //pretty ugly.. static int test_thread_queue_walk_global; -static void test_thread_queue_walk_func(struct uc_threadmsg *msg) +static void test_thread_queue_walk_func(struct uc_threadmsg *msg, void *cookie) { + (void)cookie; struct thr_msg *my_msg = (struct thr_msg *)msg; fail_if(my_msg->a != test_thread_queue_walk_global); test_thread_queue_walk_global++; @@ -82,7 +83,7 @@ START_TEST (test_thread_queue_walk) test_thread_queue_walk_global = 0; - uc_thread_queue_walk(&queue, test_thread_queue_walk_func); + uc_thread_queue_walk(&queue, test_thread_queue_walk_func, NULL); END_TEST @@ -125,7 +126,7 @@ START_TEST (test_thread_queue_reader_writer) pthread_join(tid, NULL); - fail_if(uc_thread_queue_destroy(&queue, NULL) != 0); + fail_if(uc_thread_queue_destroy(&queue, NULL, NULL) != 0); END_TEST @@ -216,7 +217,7 @@ START_TEST (test_thread_queue_timeout1) fail_if(uc_thread_queue_init(&queue, 100) != 0); fail_if(uc_thread_queue_tryget(&queue, &timeout, &msg) != ETIMEDOUT); - uc_thread_queue_destroy(&queue, NULL); + uc_thread_queue_destroy(&queue, NULL, NULL); END_TEST @@ -251,7 +252,7 @@ START_TEST (test_thread_queue_timeout2) pthread_join(tid1, NULL); - uc_thread_queue_destroy(&queue, NULL); + uc_thread_queue_destroy(&queue, NULL, NULL); END_TEST @@ -270,7 +271,7 @@ START_TEST (test_thread_queue_tryadd_timeout_0_0) fail_if(uc_thread_queue_tryadd(&queue, &timeout, &msg[2].msg) != ETIMEDOUT); - fail_if(uc_thread_queue_destroy(&queue, NULL) != 0); + fail_if(uc_thread_queue_destroy(&queue, NULL, NULL) != 0); END_TEST @@ -289,7 +290,7 @@ START_TEST (test_thread_queue_tryadd_timeout_0_100) fail_if(uc_thread_queue_tryadd(&queue, &timeout, &msg[2].msg) != ETIMEDOUT); - fail_if(uc_thread_queue_destroy(&queue, NULL) != 0); + fail_if(uc_thread_queue_destroy(&queue, NULL, NULL) != 0); END_TEST @@ -308,7 +309,7 @@ START_TEST (test_thread_queue_tryadd_no_timeout_0_100) fail_if(uc_thread_queue_tryadd(&queue, &timeout, &msg[2].msg) != 0); - fail_if(uc_thread_queue_destroy(&queue, NULL) != 0); + fail_if(uc_thread_queue_destroy(&queue, NULL, NULL) != 0); END_TEST