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

@@ -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);
+
 
 /**
  * @}*/

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;
+}
+    
+

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