Incomplete buffer errored and sink queued

Signed-off-by: Andrew Opalach <andrew@akon.city>

6 files changed, 48 insertions, 282 deletions

diff --git a/src/util/blocking_ring_buffer.c b/src/util/blocking_ring_buffer.c
deleted file mode 100644
index ec6cbb8..0000000
--- a/src/util/blocking_ring_buffer.c
+++ /dev/null
@@ -1,193 +0,0 @@
-#include <al/lib.h>
-
-#include "blocking_ring_buffer.h"
-
-#define LOCK(n) \
-    buffer->req_start = n; \
-    nn_cond_wait(&buffer->cond, &buffer->mutex);
-
-#define SIGNAL() \
-    nn_cond_signal(&buffer->cond);
-
-void camu_ring_buffer_init(struct camu_ring_buffer *buffer, s32 size)
-{
-    buffer->size = size;
-    buffer->data = (u8 *)al_malloc(buffer->size);
-    nn_mutex_init(&buffer->mutex);
-    nn_cond_init(&buffer->cond);
-    camu_ring_buffer_reset(buffer);
-    buffer->reset_unlock = false;
-    buffer->enabled = true;
-}
-
-// TODO: does locking here actually prevent this from messing something up.
-void camu_ring_buffer_reset(struct camu_ring_buffer *buffer)
-{
-    nn_mutex_lock(&buffer->mutex);
-    buffer->valid_size = 0;
-    buffer->start = 0;
-    buffer->end = 0;
-    // If get_chunk was waiting it doesn't matter where req_start
-    // was, there should be room now.
-    if (nn_cond_is_waiting(&buffer->cond)) {
-        buffer->reset_unlock = true;
-        SIGNAL();
-    }
-    buffer->wrap = false;
-    nn_mutex_unlock(&buffer->mutex);
-}
-
-void camu_ring_buffer_disable(struct camu_ring_buffer *buffer)
-{
-    nn_mutex_lock(&buffer->mutex);
-    buffer->enabled = false;
-    // No state on the buffer will be edited by the time a get_chunk
-    // returns because of this. So, the next call to get_chunk after
-    // the buffer is re-enabled should be correct.
-    if (nn_cond_is_waiting(&buffer->cond)) {
-        buffer->reset_unlock = true;
-        SIGNAL();
-    }
-    nn_mutex_unlock(&buffer->mutex);
-}
-
-void camu_ring_buffer_enable(struct camu_ring_buffer *buffer)
-{
-    nn_mutex_lock(&buffer->mutex);
-    buffer->enabled = true;
-    nn_mutex_unlock(&buffer->mutex);
-}
-
-s32 camu_ring_buffer_space(struct camu_ring_buffer *buffer)
-{
-    s32 available = 0;
-    nn_mutex_lock(&buffer->mutex);
-    if (buffer->end < buffer->start) {
-        available = buffer->start - buffer->end;
-    } else {
-        available = buffer->size - buffer->end;
-    }
-    nn_mutex_unlock(&buffer->mutex);
-    return available;
-}
-
-s32 camu_ring_buffer_occupied(struct camu_ring_buffer *buffer)
-{
-    s32 filled = 0;
-    nn_mutex_lock(&buffer->mutex);
-    if (buffer->end < buffer->start) {
-        filled = (buffer->valid_size - buffer->start) + buffer->end;
-    } else {
-        // If the buffer is empty this will equate to 0.
-        filled = buffer->end - buffer->start;
-    }
-    nn_mutex_unlock(&buffer->mutex);
-    return filled;
-}
-
-u8 *camu_ring_buffer_get_chunk(struct camu_ring_buffer *buffer, s32 n)
-{
-    nn_mutex_lock(&buffer->mutex);
-
-    if (!buffer->enabled) return NULL;
-
-    if (buffer->end < buffer->start) {
-        if (buffer->end + n > buffer->size) {
-            // Unsatisfiable, need to increase buffer size.
-            al_assert(n < buffer->end);
-            // Wait until read to the end of the buffer, wrap around, and read
-            // past the amount of data requested in this call (n).
-            buffer->wrap = true;
-            LOCK(n);
-        } else if (buffer->end + n >= buffer->start) {
-            // Wait until we read past our current write position plus the amount
-            // requested in this call (n).
-            LOCK(buffer->end + n);
-        }
-    } else if (buffer->end + n > buffer->size) {
-        // valid_size now marks the end of valid data in the buffer, this may be
-        // less than the actual buffer size.
-        buffer->valid_size = buffer->end;
-        if (n >= buffer->start) {
-            // Unsatisfiable, need to increase buffer size.
-            al_assert(n < buffer->valid_size);
-            // Wait until we read past at least the amount of data requested in this call (n).
-            LOCK(n);
-        }
-        // Set end to zero after waiting because it's possible start is also 0 at this point
-        // and we would end up with start == end while the buffer is full with data.
-        buffer->end = 0;
-    }
-
-    // If a reset happened while waiting, the behavior for the structure is to assume
-    // you didn't want watever data you were waiting to write.
-    if (buffer->reset_unlock) {
-        buffer->reset_unlock = false;
-        return NULL;
-    }
-
-    // Return a pointer to the start of the writable data and advance end by
-    // exactly the amount requested (n).
-    u8 *data = &buffer->data[buffer->end];
-    buffer->end += n;
-
-    return data;
-}
-
-s32 camu_ring_buffer_read(struct camu_ring_buffer *buffer, u8 **data, s32 n)
-{
-    nn_mutex_lock(&buffer->mutex);
-
-    // Set data even if we are going to read 0 bytes to avoid possible issues
-    // with data being NULL after a call to read.
-    *data = &buffer->data[buffer->start];
-
-    // If start == end, the buffer is empty.
-    if (buffer->start == buffer->end) return 0;
-
-    if (buffer->end < buffer->start) {
-        // If we would read past the valid data in the buffer, instead
-        // read to the end of the valid data.
-        if (buffer->start + n > buffer->valid_size) {
-            n = buffer->valid_size - buffer->start;
-        }
-    } else if (buffer->start + n > buffer->end) {
-        // If we would read to or past end, instead read to the end.
-        n = buffer->end - buffer->start;
-    }
-
-    buffer->start += n;
-
-    // If we read to the end of the valid data, wrap around to the begining.
-    //if (buffer->start == buffer->valid_size && buffer->start != buffer->end) {
-    if (buffer->start == buffer->valid_size) {
-        buffer->start = 0;
-        buffer->valid_size = 0;
-        // We wrapped and aren't specifically waiting for a position after the wrap, so unlock.
-        if (nn_cond_is_waiting(&buffer->cond) && !buffer->wrap) {
-            SIGNAL();
-        }
-        buffer->wrap = false;
-    } else if (nn_cond_is_waiting(&buffer->cond) && (!buffer->wrap && buffer->start > buffer->req_start)) {
-        // We read past the position we were waiting for, so unlock.
-        SIGNAL();
-    }
-
-    // After SIGNAL(), LOCK() won't proceed until this thread unlocks.
-
-    return n;
-}
-
-// Unlock the mutex, this needs to be called after a call to either get_chunk() or read() but
-// only after the pointer has been used.
-void camu_ring_buffer_unlock(struct camu_ring_buffer *buffer)
-{
-    nn_mutex_unlock(&buffer->mutex);
-}
-
-void camu_ring_buffer_free(struct camu_ring_buffer *buffer)
-{
-    if (buffer->data) al_free(buffer->data);
-    nn_mutex_destroy(&buffer->mutex);
-    nn_cond_destroy(&buffer->cond);
-}
diff --git a/src/util/blocking_ring_buffer.h b/src/util/blocking_ring_buffer.h
deleted file mode 100644
index a5288d0..0000000
--- a/src/util/blocking_ring_buffer.h
+++ /dev/null
@@ -1,42 +0,0 @@
-#pragma once
-
-#include <al/types.h>
-#include <nnwt/thread.h>
-
-// Ring buffer with blocking write and non-blocking read.
-//  - Write can block and always returns the exact amount requested.
-//  - Read will return as many bytes as possible, which can be less than requested or 0.
-
-struct camu_ring_buffer {
-    u8 *data;
-    s32 size;
-    // The position of the last valid byte at the end of the buffer.
-    // This can be less than the total size of the buffer.
-    s32 valid_size;
-    s32 start;
-    s32 end;
-    // The last byte in the chunk that we are waiting to be fully available.
-    s32 req_start;
-    // True if we need to wait for the start to wrap around
-    // to the begining before considering signaling.
-    bool wrap;
-    struct nn_mutex mutex;
-    struct nn_cond cond;
-    // Signal get_chunk to return null and not modifiy the buffer
-    // if it was blocking and reset was called.
-    bool reset_unlock;
-    // If the buffer is disabled, all calls to get_chunk will return
-    // early and not block.
-    bool enabled;
-};
-
-void camu_ring_buffer_init(struct camu_ring_buffer *buffer, s32 size);
-void camu_ring_buffer_reset(struct camu_ring_buffer *buffer);
-void camu_ring_buffer_disable(struct camu_ring_buffer *buffer);
-void camu_ring_buffer_enable(struct camu_ring_buffer *buffer);
-s32 camu_ring_buffer_space(struct camu_ring_buffer *buffer);
-s32 camu_ring_buffer_occupied(struct camu_ring_buffer *buffer);
-u8 *camu_ring_buffer_get_chunk(struct camu_ring_buffer *buffer, s32 n);
-s32 camu_ring_buffer_read(struct camu_ring_buffer *buffer, u8 **data, s32 n);
-void camu_ring_buffer_unlock(struct camu_ring_buffer *buffer);
-void camu_ring_buffer_free(struct camu_ring_buffer *buffer);
diff --git a/src/util/color_palette.c b/src/util/color_palette.c
index b67f1bb..bb4da03 100644
--- a/src/util/color_palette.c
+++ b/src/util/color_palette.c
@@ -1,5 +1,7 @@
 #include "color_palette.h"
 
+struct camu_color_palette global_color_palette = { 0 };
+
 #ifdef NAUNET_HAS_JSON
 #include <nnwt/file.h>
 #include <jansson.h>
@@ -27,13 +29,9 @@ static char *normal_colors[16] = {
     "color14",
     "color15"
 };
-#endif
-
-struct camu_color_palette global_color_palette = { 0 };
 
 bool camu_color_palette_init(str *path)
 {
-#ifdef NAUNET_HAS_JSON
     struct nn_file file;
     if (!nn_file_open(&file, path, 0)) {
         return false;
@@ -62,8 +60,5 @@ bool camu_color_palette_init(str *path)
         global_color_palette.colors[i + 2] = (u32)al_str_to_long(al_str_w((char *)color, 1, 6), 16);
     }
     return true;
-#else
-    (void)path;
-    return false;
-#endif
 }
+#endif
diff --git a/src/util/color_palette.h b/src/util/color_palette.h
index a71360c..e3e4ed5 100644
--- a/src/util/color_palette.h
+++ b/src/util/color_palette.h
@@ -32,4 +32,6 @@ struct camu_color_palette {
 
 extern struct camu_color_palette global_color_palette;
 
+#ifdef NAUNET_HAS_JSON
 bool camu_color_palette_init(str *path);
+#endif
diff --git a/src/util/meson.build b/src/util/meson.build
index dd67515..a83a66a 100644
--- a/src/util/meson.build
+++ b/src/util/meson.build
@@ -1,4 +1,4 @@
-util_src = ['color_palette.c', 'blocking_ring_buffer.c']
+util_src = ['color_palette.c']
 blake3 = dependency('libblake3', required: false, allow_fallback: false)
 if not blake3.found()
   blake3_opts = cmake.subproject_options()
diff --git a/src/util/queue.h b/src/util/queue.h
index ac0dfec..1d788cb 100644
--- a/src/util/queue.h
+++ b/src/util/queue.h
@@ -6,48 +6,52 @@
 #define queue(type)             \
     struct {                    \
         array(type) a;          \
-        struct nn_mutex mutex; \
+        struct nn_mutex mutex;  \
     }
 
-#define camu_queue_size(q, r)         \
-    do {                              \
-        nn_mutex_lock(&(q).mutex);   \
-        r = (q).a.size;               \
-        nn_mutex_unlock(&(q).mutex); \
-    } while (0)
+#define camu_queue_size(q, r)     \
+AL_MACRO_WRAP                     \
+({                                \
+    nn_mutex_lock(&(q).mutex);    \
+    r = (q).a.size;               \
+    nn_mutex_unlock(&(q).mutex);  \
+})
 
-#define camu_queue_init(q)          \
-    do {                            \
-        al_array_init((q).a);       \
-        nn_mutex_init(&(q).mutex); \
-    } while (0)
+#define camu_queue_init(q)      \
+AL_MACRO_WRAP                   \
+({                              \
+    al_array_init((q).a);       \
+    nn_mutex_init(&(q).mutex);  \
+})
 
-#define camu_queue_push(q, item)      \
-    do {                              \
-        nn_mutex_lock(&(q).mutex);   \
-        al_array_push((q).a, item);   \
-        nn_mutex_unlock(&(q).mutex); \
-    } while (0)
+#define camu_queue_push(q, item)  \
+AL_MACRO_WRAP                     \
+({                                \
+    nn_mutex_lock(&(q).mutex);    \
+    al_array_push((q).a, item);   \
+    nn_mutex_unlock(&(q).mutex);  \
+})
 
-#define camu_queue_pop(q, r)          \
-    do {                              \
-        nn_mutex_lock(&(q).mutex);   \
-        al_array_pop_at((q).a, 0, r); \
-        nn_mutex_unlock(&(q).mutex); \
-    } while (0)
+#define camu_queue_pop(q, r)      \
+AL_MACRO_WRAP                     \
+({                                \
+    nn_mutex_lock(&(q).mutex);    \
+    al_array_pop_at((q).a, 0, r); \
+    nn_mutex_unlock(&(q).mutex);  \
+})
 
-#define camu_queue_try_pop(q, s, r)   \
-    do {                              \
-        nn_mutex_lock(&(q).mutex);   \
-        s = (q).a.size;               \
-        if (s > 0) {                  \
-            al_array_pop_at((q).a, 0, r); \
-        }                             \
-        nn_mutex_unlock(&(q).mutex); \
-    } while (0)
+#define camu_queue_try_pop(q, s, r) \
+AL_MACRO_WRAP                     \
+({                                \
+    nn_mutex_lock(&(q).mutex);    \
+    if ((s = (q).a.size) > 0)     \
+        al_array_pop_at((q).a, 0, r); \
+    nn_mutex_unlock(&(q).mutex);  \
+})
 
-#define camu_queue_free(q)             \
-    do {                               \
-        nn_mutex_destroy(&(q).mutex); \
-        al_array_free((q).a);          \
-    } while (0)
+#define camu_queue_free(q)         \
+AL_MACRO_WRAP                      \
+({                                 \
+    nn_mutex_destroy(&(q).mutex);  \
+    al_array_free((q).a);          \
+})