Implement queue_finish

diff --git a/queue.c b/queue.c
index 37dce732c8732ae5ad4e0ff0e09bfbb4a7d8bc00..290fd76085268c11e6812ff9fc25ba582ab97b54 100644 (file)
--- a/queue.c
+++ b/queue.c
@@ -5,6 +5,7 @@
 Queue *queue_create()
 {
        Queue *queue = malloc(sizeof(Queue));
+       queue->finish = false;
        queue->cancel = false;
        queue->list = list_create(NULL);
        pthread_cond_init(&queue->cv, NULL);
@@ -23,12 +24,14 @@ void queue_delete(Queue *queue)
 void queue_enqueue(Queue *queue, void *elem)
 {
        pthread_mutex_lock(&queue->mtx);
-       list_put(&queue->list, elem, NULL);
-       pthread_cond_signal(&queue->cv);
+       if (! queue->finish) {
+               list_put(&queue->list, elem, NULL);
+               pthread_cond_signal(&queue->cv);
+       }
        pthread_mutex_unlock(&queue->mtx);
 }
 
-void *dequeue(Queue *queue)
+void *queue_dequeue(Queue *queue)
 {
        return queue_dequeue_callback(queue, NULL);
 }
@@ -67,3 +70,21 @@ void queue_cancel(Queue *queue)
        pthread_cond_broadcast(&queue->cv);
        pthread_mutex_unlock(&queue->mtx);
 }
+
+void queue_finish(Queue *queue)
+{
+       pthread_mutex_lock(&queue->mtx);
+       queue->finish = true;
+       pthread_mutex_unlock(&queue->mtx);
+
+       while (true) {
+               pthread_mutex_lock(&queue->mtx);
+               ListPair *first = queue->list.first;
+               pthread_mutex_unlock(&queue->mtx);
+
+               if (first)
+                       sched_yield();
+               else
+                       break;
+       }
+}

diff --git a/queue.h b/queue.h
index 80b1e5410cbd522e79762d2009fe22fd6b74c9e5..cfa72b8d9f06e9bb8188abaa30e6e49aae26d995 100644 (file)
--- a/queue.h
+++ b/queue.h
@@ -9,6 +9,7 @@
 typedef struct
 {
        atomic_bool cancel;
+       bool finish;
        List list;
        pthread_cond_t cv;
        pthread_mutex_t mtx;
@@ -19,6 +20,7 @@ void queue_delete(Queue *queue);
 void queue_enqueue(Queue *queue, void *elem);
 void *queue_dequeue(Queue *queue);
 void *queue_dequeue_callback(Queue *queue, void (*callback)(void *elem));
-void queue_cancel(Queue *queue);
+void queue_cancel(Queue *queue); // disallow dequeing
+void queue_finish(Queue *queue); // disallow enqueing, wait until consumption finished
 
 #endif