Emptying of queue when quitting

Author: Griezn

include/queue.h

@@ -4,6 +4,7 @@
 #include <stdatomic.h>
 #include <stdbool.h>
 #include <stddef.h>
+#include <stdlib.h>
 
 #include "data.h"
 
@@ -24,4 +25,6 @@ bool spsc_dequeue(spsc_queue_t *q, data_t **item);
 
 bool spsc_is_empty(spsc_queue_t *q);
 
+void empty_queue(spsc_queue_t *q);
+
 #endif //SPSC_QUEUE_LIBRARY_H

src/query.c

@@ -107,6 +107,7 @@ void *execute_step(void *args)
                 if (spsc_dequeue(left_queue, &left_input)) break;
                 if (atomic_load(&left_step->quit) && spsc_is_empty(left_queue)) {
                     atomic_store(&step->quit, true);
+                    empty_queue(left_queue);
                     break;
                 }
             }
@@ -116,6 +117,7 @@ void *execute_step(void *args)
                 if (spsc_dequeue(right_queue, &right_input)) break;
                 if (atomic_load(&right_step->quit) && spsc_is_empty(right_queue)) {
                     atomic_store(&step->quit, true);
+                    empty_queue(right_queue);
                     break;
                 }
             }
@@ -162,24 +164,20 @@ void *execute_step(void *args)
     skip:
         if (left_step) {
             if (op->left->type != WINDOW && left_input) {
-                assert(left_input);
                 free(left_input->data);
                 left_input->data = NULL;
-                free(left_input);
-                left_input = NULL;
-            } else {
+            }
+            if (left_input) {
                 free(left_input);
                 left_input = NULL;
             }
         }
         if (right_step) {
             if (op->right->type != WINDOW && right_input) {
-                assert(right_input);
                 free(right_input->data);
                 right_input->data = NULL;
-                free(right_input);
-                right_input = NULL;
-            } else {
+            }
+            if (right_input) {
                 free(right_input);
                 right_input = NULL;
             }

src/queue.c

@@ -66,3 +66,13 @@ bool spsc_is_empty(spsc_queue_t *q)
     return head == tail;
 }
 
+
+void empty_queue(spsc_queue_t *q)
+{
+    data_t *output;
+    while (!spsc_is_empty(q)) {
+        spsc_dequeue(q, &output);
+        free(output->data);
+        free(output);
+    }
+}