1 files changed, 76 insertions, 21 deletions

diff --git a/src/main.c b/src/main.c
index 95b55ff..d1048f0 100644
--- a/src/main.c
+++ b/src/main.c
@@ -9,40 +9,95 @@
 #include <error.h>
 #include <threads.h>
 #include <unistd.h>
+#include <string.h>
 
-int testcb(void *data) {
-    if(!data)
-        return -1;
+typedef int (*thingcb)(void*);
+struct thing {
+    thingcb cb;
+    void *arg;
+};
 
-    printf("%s\n", (char*)data);
-    return 0;
-}
+typedef struct queue {
+    mtx_t *cqmutex;
+    cnd_t *cqcnd;
+    unsigned char cstatus;
+    
+    int len;
+    int used;
+    
+    struct thing **queue;
+} ccqueue;
 
-int consumer(void *cq) {
-    if(!cq)
-        return -1;
+#define QUEUE_LEN 10
+
+int consumer(void *queue) {
+    if(!queue)
+        thrd_exit(-1);
 
-    cqueue *rcq = (cqueue*)cq;
-    for(task *tsk = NULL;;) {
-        tsk = cqueue_waitpop(rcq);
-        if(!tsk)
+    ccqueue *real = (ccqueue *)queue;
+    for(struct thing *thing = NULL;;) {
+        cnd_wait(real->cqcnd, real->cqmutex);
+        if(real->cstatus)
             thrd_exit(-1);
 
-        task_fire(tsk);
+        if(real->used > 0) {
+            thing = real->queue[real->used - 1];
+            real->used--;
+        }
+
+        if(thing != NULL)
+            if(thing->arg != NULL)
+                thing->cb(thing->arg);
     }
 
+    thrd_exit(0);
+}
+
+int testcb(void *data) {
+    if(!data)
+        return -1;
+
+    printf("%s\n", (char*)data);
     return 0;
 }
 
 int main() {
     // error(1, ENOTSUP, "No main file lol");
 
-    thrd_t thread;
-    cqueue *cq = cqueue_init(mtx_plain);
-    thrd_create(&thread, consumer, cq);
-    cqueue_addtask(cq, task_init(testcb, (void*)"This is some data"));
-    sleep(10);
-    cqueue_free(cq);
+    // Manually implement a threadpool and consumer in a "dumb" way to make sure I'm doing the correct thing
+    ccqueue *cqueue = xcalloc(1, sizeof(*cqueue));
+    cqueue->cqmutex = xcalloc(1, sizeof(*cqueue->cqmutex)); mtx_init(cqueue->cqmutex, mtx_plain);
+    cqueue->cqcnd = xcalloc(1, sizeof(*cqueue->cqcnd)); cnd_init(cqueue->cqcnd);
+    cqueue->cstatus = 0;
+    cqueue->len = QUEUE_LEN;
+    cqueue->used = 0;
+    cqueue->queue = xcalloc(cqueue->len, sizeof(struct thing *));
+
+    thrd_t *thread = xcalloc(1, sizeof(*thread));
+    thrd_create(thread, consumer, (void *)cqueue);
+
+    mtx_lock(cqueue->cqmutex);
+    cqueue->queue[cqueue->used] = xcalloc(1, sizeof(struct thing));
+    cqueue->queue[cqueue->used]->cb = testcb;
+    cqueue->queue[cqueue->used]->arg = "This is some data";
+    cqueue->used++;
+    mtx_unlock(cqueue->cqmutex);
+    cnd_signal(cqueue->cqcnd);
+
+    sleep(1);
+    
+    mtx_lock(cqueue->cqmutex);
+    cqueue->cstatus++;
+    mtx_unlock(cqueue->cqmutex);
+    cnd_broadcast(cqueue->cqcnd);
+
+    // Destroy and free everything
+    mtx_destroy(cqueue->cqmutex); cnd_destroy(cqueue->cqcnd);
+    free(cqueue->cqmutex); free(cqueue->cqcnd);
+    free(cqueue->queue);
+    free(cqueue);
 
     return 0;
-}
\ No newline at end of file
+}
+
+// Ok well now I'm pissed. The canceling works fine but it's not actually consuming the thing
\ No newline at end of file