1 files changed, 29 insertions, 2 deletions

diff --git a/src/threadpool.c b/src/threadpool.c
index 66c0d06..31c300c 100644
--- a/src/threadpool.c
+++ b/src/threadpool.c
@@ -310,7 +310,7 @@ int cqueue_trypop(cqueue * const cq, task **ret) {
 }
 
 int cqueue_waitpop(cqueue * const cq, task **ret) {
-    if(!cq || !ret || !*ret)
+    if(!cq || !ret)
         RETURNWERR(EINVAL, -1);
 
     mtx_lock(cq->mutex);
@@ -348,6 +348,23 @@ int cqueue_cancel(cqueue * const cq) {
     return retval;
 }
 
+int cqueue_consumer(void *passed) {
+    if(!passed)
+        thrd_exit(thrd_error);
+    // Not setting errno because then I'd have to make a mutex for it
+
+    cqueue *cq = (cqueue *)passed;
+    
+    for(task *current_task;;) {
+        cqueue_waitpop(cq, &current_task);
+        if(!current_task)
+            thrd_exit(thrd_error);
+
+        current_task->cb(current_task->arg);
+    }
+
+    thrd_exit(thrd_success);
+}
 
 static void ___ucleanup_cqfree(void *cq) {
     if(!cq)
@@ -384,6 +401,10 @@ threadpool * threadpool_init(int threads) {
             for(int j = 0; j < i; j++)
                 free(tp->threads[j]);
         }
+
+        if(!cleanup_ERRORFLAGGED)
+            thrd_create(tp->threads[i], cqueue_consumer, tp->taskqueue);
+        // TODO: Error Checking ^
     }
 
     if(cleanup_ERRORFLAGGED)
@@ -404,8 +425,14 @@ void threadpool_free(threadpool *tp) {
         free(tp->threads[i]);
     }
     free(tp->threads);
-    cqueue_free(tp->taskqueue);
     free(tp);
     
     return;
+}
+
+int threadpool_addtask(threadpool * const tp, task * const task) {
+    if(!tp || !task)
+        RETURNWERR(EINVAL, -1);
+
+    return cqueue_append(tp->taskqueue, task);
 }
\ No newline at end of file