Initial commit

src/main.c

@@ -0,0 +1,203 @@
+#include <stdio.h>
+#include <time.h>
+#include <stdint.h>
+#include <stdlib.h>
+#include <sys/time.h>
+#include <unistd.h>
+
+#include <stdatomic.h>
+#include <pthread.h>
+
+
+#include <dlfcn.h>
+
+#include <string.h>
+
+#include "ptQueue.h"
+
+
+void *threadMain(void *param);
+
+
+
+void microsecond_sleep(int microseconds) {
+    struct timespec ts;
+    ts.tv_sec = microseconds / 1000000;
+    ts.tv_nsec = (microseconds % 1000000) * 1000;
+    nanosleep(&ts, NULL);
+}
+
+int main(int argc, char** argv)
+{
+
+    pthread_t tid;
+    pthread_attr_t attr;
+
+    char errbuf[1024];
+
+
+    ptQueue* qOut = ptQueueCreate(errbuf);
+    if(qOut == NULL)
+    {
+        printf("Failed to create out q: %s\n", errbuf);
+        abort();
+    }
+    ptQueue* qIn = ptQueueCreate(errbuf);
+    if(qIn == NULL)
+    {
+        printf("Failed to create out q: %s\n", errbuf);
+        abort();
+    }
+
+    ptQueue** ptArgs = malloc(sizeof(ptQueue*) * 2);
+    if(ptArgs == NULL)
+    {
+        abort();
+    }
+    ptArgs[0] = qOut;
+    ptArgs[1] = qIn;
+
+    pthread_attr_init(&attr);
+    pthread_create(&tid, &attr, threadMain, ptArgs);
+
+
+    for(int i = 1; i < 4096; i++)
+    {
+        int* pass = malloc(sizeof(int));
+        if(pass == NULL)
+        {
+            abort();
+        }
+        *pass = i;
+
+        int excode = ptQueuePush(qOut, pass, errbuf);
+        if(excode)
+        {
+            printf("Unable to push: %s\n", errbuf);
+            abort();
+        }
+    }
+
+    int* ii = malloc(sizeof(int));
+    if(ii == NULL)
+    {
+        abort();
+    }
+    *ii = 0;
+
+    int excode = ptQueuePush(qOut, ii, errbuf);
+    if(excode)
+    {
+        printf("Unable to push 0: %s\n", errbuf);
+        abort();
+    }
+
+    ptQueueElem* tailIn = qIn->tail;
+
+    while (1) 
+    {
+        void* payload = atomic_load(&(tailIn->payload));
+        if(payload == NULL)
+        {
+            microsecond_sleep(100);
+            continue;
+        }
+        int* code = payload;
+
+        printf("<Done reading %d\n", *code);
+
+
+        int* pass = malloc(sizeof(int));
+        if(pass != NULL)
+        {
+            *pass = *code;
+            int excode = ptQueuePush(qOut, pass, errbuf);
+            if(excode)
+            {
+                abort();
+            }
+        }
+
+        if(*code == 0)
+        {
+            free(payload);
+            break;
+        }
+        free(payload);
+        ptQueueElem* newTail = atomic_load(&(tailIn->nextEl));
+        if (newTail == NULL)
+        {
+            abort();
+        }
+        free(tailIn);
+        tailIn = newTail;
+    }
+
+    pthread_join(tid, NULL);
+    return 0;
+}
+
+
+
+void *threadMain(void *param) 
+{
+    ptQueue** params = param;
+    ptQueue* qIn = params[0];
+    ptQueue* qOut = params[1];
+    ptQueueElem* tailIn = atomic_load(&(qIn->head));
+    char errbuf[1024];
+
+    while (1) 
+    {
+        void* payload = atomic_load(&(tailIn->payload));
+        if(payload == NULL)
+        {
+            microsecond_sleep(100);
+            continue;
+        }
+        int* code = payload;
+
+        printf(">Done reading %d\n", *code);
+
+
+        int* pass = malloc(sizeof(int));
+        if(pass != NULL)
+        {
+            *pass = *code;
+            ptQueuePush(qOut, pass, errbuf);
+        }
+
+        if(*code == 0)
+        {
+            free(payload);
+            break;
+        }
+        free(payload);
+        ptQueueElem* newTail = atomic_load(&(tailIn->nextEl));
+        if (newTail == NULL)
+        {
+            abort();
+        }
+        free(tailIn);
+        tailIn = newTail;
+    
+    }
+
+    for(int i = 4096; i >= 0; i--)
+    {
+        int* pass = malloc(sizeof(int));
+        if(pass == NULL)
+        {
+            continue;
+        }
+        *pass = i;
+        int excode = ptQueuePush(qOut, pass, errbuf);
+        if(excode != 0)
+        {
+            printf("failed to push: %s\n", errbuf);
+            abort();
+        }
+    }
+
+  pthread_exit(0);
+}

src/ptQueue.c

@@ -0,0 +1,83 @@
+#include <unistd.h>
+#include <stdatomic.h>
+#include <stdlib.h>
+#include <stdio.h>
+
+#include "ptQueue.h"
+
+
+
+int ptQueuePush(ptQueue* q, void* payload, char* errbuf)
+{
+    _Atomic(ptQueueElem*) tail = atomic_load(&q->tail);
+    
+    
+    ptQueueElem* newNode = (ptQueueElem*)malloc(sizeof(ptQueue));
+    if(newNode == NULL)
+    {
+        sprintf(errbuf, "unable to allocate new node");
+        return 1;
+    }
+    newNode->payload = NULL;
+    newNode->nextEl = NULL;
+
+    q->tail = newNode;
+    
+    ptQueueElem* oldNext = atomic_load(&tail->nextEl);
+
+    do {
+        tail->nextEl = newNode;
+    } while (!atomic_compare_exchange_weak(&tail->nextEl, &oldNext, newNode));
+
+
+    void* oldPayload = atomic_load(&tail->payload);
+
+    do {
+        tail->payload = payload;
+    } while (!atomic_compare_exchange_weak(&tail->payload, &oldPayload, payload));
+    return 0;
+}
+
+
+ptQueue* ptQueueCreate(char* errbuf)
+{
+    ptQueue* q = malloc(sizeof(ptQueue));
+    if(q == NULL)
+    {
+        sprintf(errbuf, "unable to allocate queue struct");
+        return NULL;
+    }
+    ptQueueElem* baseElm = malloc(sizeof(ptQueueElem));
+    if(baseElm == NULL)
+    {
+        sprintf(errbuf, "unable to allocate first queue elem");
+        free(q);
+        return NULL;
+    }
+    baseElm->nextEl = NULL;
+    baseElm->payload = NULL;
+    q->head = baseElm;
+    q->tail = baseElm;
+
+    return q;
+}
+
+void ptQueueFree(ptQueue* q)
+{
+    if(q == NULL)
+    {
+        return;
+    }
+    ptQueueElem* head = q->head;
+    while(head != NULL)
+    {
+        ptQueueElem* newHead = head->nextEl;
+        if(head->payload != NULL)
+        {
+            free(head->payload);
+        }
+        free(head);
+        head = newHead;
+    }
+    free(q);
+}