1 files changed, 45 insertions, 18 deletions

diff --git a/src/threadpool.c b/src/threadpool.c
index c4d8a5c..bd52c75 100644
--- a/src/threadpool.c
+++ b/src/threadpool.c
@@ -5,9 +5,37 @@
 #include <errno.h>
 #include <error.h>
 
+typedef struct task {
+    gcallback callback;
+    fcallback freecb;
+    void *data;
+} task;
+
+typedef struct tqnode {
+    struct tqnode *next;
+    struct tqnode *prev;
+    task *task;
+} tqnode;
+
+typedef struct taskqueue {
+    tqnode *start;
+    tqnode *end;
+    unsigned int size;
+} taskqueue;
+
+typedef struct ctqueue {
+    mtx_t mutex;
+    cnd_t cond;
+    unsigned char canceled;
+
+    taskqueue *tq;
+    thrd_t *thrdarr;
+    int talen;
+} ctqueue;
+
 
 task * task_init(gcallback callback, fcallback freecb, void *data) {
-    if(callback == NULL) {errno = EINVAL; return NULL;}
+    if(callback == NULL) ERRRET(EINVAL, NULL);
 
     task *tsk = calloc(1, sizeof(*tsk));
     if(!tsk)
@@ -33,7 +61,7 @@ void task_free(void *tsk) {
 }
 
 int task_fire(task *tsk) {
-    if(!tsk) {errno = EINVAL; return -1;}
+    if(!tsk) ERRRET(EINVAL, -1);
     return tsk->callback(tsk->data);
 }
 
@@ -46,7 +74,7 @@ int task_fired(task *tsk) {
 
 
 tqnode * tqnode_init(tqnode *next, tqnode *prev, task *tsk) {
-    if(!tsk) {errno = EINVAL; return NULL;}
+    if(!tsk) ERRRET(EINVAL, NULL);
     tqnode *node = calloc(1, sizeof(*node));
     if(!node)
         return NULL;
@@ -98,7 +126,7 @@ void taskqueue_free(void *tq) {
 }
 
 int taskqueue_handlefirst(taskqueue *tq, task *tsk) {
-    if(!tq || !tsk) {errno = EINVAL; return -1;}
+    if(!tq || !tsk) ERRRET(EINVAL, -1);
     if(tq->size) {return 0;}
 
     tqnode *first = tqnode_init(NULL, NULL, tsk);
@@ -113,7 +141,7 @@ int taskqueue_handlefirst(taskqueue *tq, task *tsk) {
 }
 
 int taskqueue_push(taskqueue *tq, task *tsk) {
-    if(!tq || !tsk) {errno = EINVAL; return -1;}
+    if(!tq || !tsk) ERRRET(EINVAL, -1);
 
     int hf;
     if((hf = taskqueue_handlefirst(tq, tsk)))
@@ -130,8 +158,8 @@ int taskqueue_push(taskqueue *tq, task *tsk) {
 }
 
 task * taskqueue_pop(taskqueue *tq) {
-    if(!tq) {errno = EINVAL; return NULL;}
-    if(tq->size <= 0) {errno = ENODATA; return NULL;}
+    if(!tq) ERRRET(EINVAL, NULL);
+    if(tq->size <= 0) ERRRET(ENODATA, NULL);
 
     tqnode *end = tq->end;
     task *ret = end->task; 
@@ -150,7 +178,7 @@ task * taskqueue_pop(taskqueue *tq) {
 }
 
 int taskqueue_pushfront(taskqueue *tq, task *tsk) {
-    if(!tq || !tsk) {errno = EINVAL; return -1;}
+    if(!tq || !tsk) ERRRET(EINVAL, -1);
 
     int hf;
     if((hf = taskqueue_handlefirst(tq, tsk)))
@@ -167,8 +195,8 @@ int taskqueue_pushfront(taskqueue *tq, task *tsk) {
 }
 
 task * taskqueue_popback(taskqueue *tq) {
-    if(!tq) {errno = EINVAL; return NULL;}
-    if(tq->size <= 0) {errno = ENODATA; return NULL;}
+    if(!tq) ERRRET(EINVAL, NULL);
+    if(tq->size <= 0) ERRRET(ENODATA, NULL);
 
     tqnode *start = tq->start;
     task *ret = start->task;
@@ -187,7 +215,7 @@ task * taskqueue_popback(taskqueue *tq) {
 }
 
 int taskqueue_size(taskqueue *tq) {
-    if(!tq) {errno = EINVAL; return -1;}
+    if(!tq) ERRRET(EINVAL, -1);
     return tq->size;
 }
 
@@ -219,7 +247,7 @@ static void ___ucl_cnddestroy(void *cond) {
 }
 
 ctqueue * ctqueue_init(int nthreads) {
-    if(nthreads <= 0) {errno = EINVAL; return NULL;}
+    if(nthreads <= 0) ERRRET(EINVAL, NULL);
     cleanup_CREATE(6);
 
     ctqueue *ctq = calloc(1, sizeof(*ctq));
@@ -264,7 +292,7 @@ ctqueue * ctqueue_init(int nthreads) {
 }
 
 int ctqueue_cancel(ctqueue *ctq) {
-    if(!ctq) {errno = EINVAL; return -1;}
+    if(!ctq) ERRRET(EINVAL, -1);
 
     __CTQ_INLOCK(ctq, 1, 
         ctq->canceled = 1;
@@ -297,7 +325,7 @@ void ctqueue_free(void *ctq) {
 }
 
 int ctqueue_waitpush(ctqueue *ctq, task *tsk) {
-    if(!ctq || !tsk) {errno = EINVAL; return -1;}
+    if(!ctq || !tsk) ERRRET(EINVAL, -1);
     int retval = 0;
 
     __CTQ_INLOCK(ctq, -1, 
@@ -310,7 +338,7 @@ int ctqueue_waitpush(ctqueue *ctq, task *tsk) {
 }
 
 task * ctqueue_waitpop(ctqueue *ctq) {
-    if(!ctq) {errno = EINVAL; return NULL;}
+    if(!ctq) ERRRET(EINVAL, NULL);
     task *retval = NULL;
 
     __CTQ_INLOCK(ctq, NULL, 
@@ -318,9 +346,8 @@ task * ctqueue_waitpop(ctqueue *ctq) {
             cnd_wait(&ctq->cond, &ctq->mutex);
 
         if(ctq->canceled) {
-            errno = ECANCELED;
             mtx_unlock(&ctq->mutex);
-            return NULL;
+            ERRRET(ECANCELED, NULL);
         }
 
         retval = taskqueue_pop(ctq->tq);
@@ -346,7 +373,7 @@ static int __CTQ_CONSUMER(void *ctq) {
 }
 
 int ctqueue_start(ctqueue *ctq) {
-    if(!ctq) {errno = EINVAL; return -1;}
+    if(!ctq) ERRRET(EINVAL, -1);
 
     ctq->canceled = 0;