From 0f30ec889c7a70ab7f7b79836d1a34ddf8659a47 Mon Sep 17 00:00:00 2001
From: Elias Fleckenstein <eliasfleckenstein@web.de>
Date: Sun, 13 Feb 2022 21:12:08 +0100
Subject: [PATCH] Implement queue_finish

---
 queue.c | 27 ++++++++++++++++++++++++---
 queue.h |  4 +++-
 2 files changed, 27 insertions(+), 4 deletions(-)

diff --git a/queue.c b/queue.c
index 37dce73..290fd76 100644
--- 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 80b1e54..cfa72b8 100644
--- 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
-- 
2.44.0