Start reimplementing some of the earlier concurrent queue code

1 files changed, 89 insertions, 70 deletions

diff --git a/src/threadpool.c b/src/threadpool.c
index e1c11aa..0baa024 100644
--- a/src/threadpool.c
+++ b/src/threadpool.c
@@ -131,95 +131,114 @@ typedef struct cq {
     cnd_t cnd;
     
     unsigned char canceled;
-
 } cqueue;
 
 
+static void ___ucleanup_mtxd(void *mtx) {
+    if(!mtx)
+        return;
 
-// static void ___ucleanup_dfree(void *dll) {
-//     if(!dll)
-//         return;
+    mtx_destroy((mtx_t*)mtx);
+    return;
+}
 
-//     dlinkedlist_free((dlinkedlist *)dll);
-//     return;
-// }
+static void ___ucleanup_cndd(void *cnd) {
+    if(!cnd)
+        return;
 
-// static void ___ucleanup_cndd(void *cnd) {
-//     if(!cnd)
-//         return;
+    cnd_destroy((cnd_t *)cnd);
+    return;
+}
 
-//     cnd_destroy((cnd_t *)cnd);
-//     return;
-// }
+cqueue * cqueue_init() {
+    cleanup_CREATE(10);
+    
+    // Create base object
+    cqueue *cq = VALLOC(1, sizeof(*cq));
+    if(!cq)
+        RETURNWERR(errno, NULL);
+    cleanup_REGISTER(free, cq);
+    cq->canceled = 0;
 
-// static void ___ucleanup_mtxd(void *mtx) {
-//     if(!mtx)
-//         return;
+    // Initialize the mutex
+    if(mtx_init(&cq->mtx, mtx_plain) != thrd_success)
+        cleanup_MARK();
+    cleanup_CNDREGISTER(___ucleanup_mtxd, &cq->mtx);
+    
+    // Initialize the conditional
+    if(!cleanup_ERRORFLAGGED)
+        if(cnd_init(&cq->cnd) != thrd_success)
+            cleanup_MARK();
+    cleanup_CNDREGISTER(___ucleanup_cndd, &cq->cnd);
+
+    // Create the taskqueue
+    if(!cleanup_ERRORFLAGGED)
+        if(!(cq->taskqueue = dlinkedlist_init()))
+            cleanup_MARK();
+    cleanup_CNDREGISTER(dlinkedlist_free, cq->taskqueue);
+
+    // Create the thread list
+    if(!cleanup_ERRORFLAGGED)
+        if(!(cq->rthreads = dlinkedlist_init()))
+            cleanup_MARK();
+    cleanup_CNDREGISTER(dlinkedlist_free, cq->rthreads);
+
+    if(cleanup_ERRORFLAGGED)
+        cleanup_FIRE();
+
+    return cq;
+
+    // Lambdas would make this a million times easier, as I could wrap this whole thing in a while loop then run a bunch of in-line
+    // callbacks that do these operations and I wouldn't need this badness. That or I could use a goto, but I also hate that idea
+}
 
-//     mtx_destroy((mtx_t*)mtx);
-//     return;
-// }
+void cqueue_cancel(cqueue * const cq) {
+    if(!cq)
+        return;
 
-// cqueue * cqueue_init(int mtx_type) {
-//     cleanup_CREATE(10);
-    
-//     cqueue *cq = VALLOC(1, sizeof(*cq));
-//     if(!cq)
-//         return NULL;
-//     cleanup_REGISTER(free, cq);
-
-//     cq->canceled = FALSE;
-//     cq->list = dlinkedlist_init();
-//     if(!cq->list)
-//         cleanup_MARK();
-//     cleanup_CNDREGISTER(___ucleanup_dfree, cq->list);
-    
-//     if(!cleanup_ERRORFLAGGED)
-//         if(cnd_init(&cq->cnd) == thrd_error)
-//             cleanup_MARK();
-//     cleanup_CNDREGISTER(___ucleanup_cndd, &cq->cnd);
-    
-//     if(!cleanup_ERRORFLAGGED)
-//         if(mtx_init(&cq->mtx, mtx_type) != thrd_success)
-//             cleanup_MARK();
-//     cleanup_CNDREGISTER(___ucleanup_mtxd, &cq->mtx);
+    mtx_lock(&cq->mtx);
     
-//     if(cleanup_ERRORFLAGGED)
-//         cleanup_FIRE();
+    if(cq->canceled) {
+        mtx_unlock(&cq->mtx);
+        return;
+    }
+    cq->canceled = 1;
 
-//     return cq;
-// }
+    mtx_unlock(&cq->mtx);
+    cnd_broadcast(&cq->cnd);
+    
+    return;
+}
 
-// void cqueue_cancel(cqueue *cq) {
-//     if(!cq)
-//         return;
+static int ___cqueue_join(void *t) {
+    if(!t)
+        return -1;
 
-//     mtx_lock(cq->mtx);
-//     if(cq->canceled) {
-//         mtx_unlock(cq->mtx);
-//         thrd_exit(-1);
-//     }
+    int retval = 0;
+    thrd_t thread = *((thrd_t*)t);
+    thrd_join(thread, &retval);
+    
+    return retval;
+}
 
-//     cq->canceled++;
-//     mtx_unlock(cq->mtx);
-//     cnd_broadcast(cq->cnd);
+void cqueue_free(void *cq) {
+    if(!cq)
+        return;
 
-//     return;
-// }
+    cqueue *real = (cqueue *)cq;
 
-// void cqueue_free(cqueue *cq) {
-//     if(!cq)
-//         return;
+    // Cancel threads and wait for them to exit
+    cqueue_cancel(real);
+    dlinkedlist_foreach(real->rthreads, ___cqueue_join);
 
-//     cqueue_cancel(cq);
-//     mtx_destroy(cq->mtx);
-//     cnd_destroy(cq->cnd);
-//     free(cq->mtx);
-//     free(cq->cnd);
-//     dlinkedlist_free(cq->list);
+    // Threads are dead, no need to worry about concurrency anymore
+    mtx_destroy(&real->mtx);
+    cnd_destroy(&real->cnd);
+    dlinkedlist_free(real->rthreads);
+    dlinkedlist_free(real->taskqueue);
 
-//     return;
-// }
+    return;
+}
 
 // int cqueue_addtask(cqueue * const cq, task * const tsk) {
 //     if(!cq || !tsk)