engine: add single producer single consumer queue

This allows the main and background thread to communicate with
each other.

Signed-off-by: Ronald Caesar <github43132@proton.me>

CMakeLists.txt

@@ -65,6 +65,7 @@ add_library(Ballistic STATIC
         src/bal_errors.c
         src/bal_logging.c
         src/bal_memory.c
+        src/bal_spsc_queue.c
         src/bal_tier2_worker.c
 )
 
@@ -119,7 +120,7 @@ if (CMAKE_SYSTEM_NAME STREQUAL "Linux" OR CMAKE_SYSTEM_NAME STREQUAL "Darwin")
     target_link_libraries(${CDOC_NAME} PRIVATE ${CLANG_LIBRARY} cmark)
     set(PROJECT_HEADERS include/bal_engine.h include/bal_decoder.h
             include/bal_memory.h include/bal_types.h include/bal_errors.h include/bal_logging.h
-            include/bal_assembler.h include/bal_tier2_worker.h)
+            include/bal_assembler.h include/bal_tier2_worker.h include/bal_spsc_queue.h)
 
     add_custom_target(doc ALL
             COMMAND ${CMAKE_CURRENT_BINARY_DIR}/cdoc docs/cdoc ${PROJECT_HEADERS}

include/bal_spsc_queue.h

@@ -0,0 +1,64 @@
+#ifndef BALLISTIC_BAL_SPSC_QUEUE_H
+#define BALLISTIC_BAL_SPSC_QUEUE_H
+
+#include "bal_attributes.h"
+#include "bal_types.h"
+#include <assert.h>
+#include <stdatomic.h>
+#include <stdbool.h>
+
+#define BAL_SPSC_QUEUE_CAPACITY 256
+#define BAL_SPSC_QUEUE_MASK     (BAL_SPSC_QUEUE_CAPACITY - 1)
+
+static_assert(!(BAL_SPSC_QUEUE_CAPACITY & BAL_SPSC_QUEUE_CAPACITY - 1),
+              "Must be a power of 2 for fast bit arithmatic");
+
+/// Lock free Single Producer, Single Consumer queue.
+typedef struct
+{
+    /// Written by main thread, read by background thread.
+    BAL_ALIGNED(64) atomic_size_t head;
+
+    /// Written by background thread, read by main thread.
+    BAL_ALIGNED(64) atomic_size_t tail;
+
+    /// Ring buffer holding the guest addresses.
+    bal_guest_address_t buffer[BAL_SPSC_QUEUE_CAPACITY];
+} bal_spsc_queue_t;
+
+/// Initializes the SPSC queue.
+void bal_spsc_queue_init(bal_spsc_queue_t *queue);
+
+/// Pushes `address` into the queue.
+///
+/// # Safety
+///
+/// This function must ONLY be called by the main thread.
+///
+/// # Errors
+///
+/// Returns `true` on success.
+///
+/// Returns `false` if `queue` is NULL.
+///
+/// Returns `false` the queue is full.
+BAL_HOT bool bal_spsc_queue_push(bal_spsc_queue_t *queue, bal_guest_address_t address);
+
+/// Pops a guest virtual address from `queue` into `out_address`.
+///
+/// # Safety
+///
+/// This function must ONLY be called by the background thread.
+///
+/// # Errors
+///
+/// Returns `true` on success.
+///
+/// Returns `false` if any function argument is NULL.
+///
+/// Returns `false` if `queue` is full.
+BAL_HOT bool bal_spsc_queue_pop(bal_spsc_queue_t *queue, bal_guest_address_t *out_address);
+
+#endif // BALLISTIC_BAL_SPSC_QUEUE_H
+
+/*** end of file ***/

src/bal_spsc_queue.c

@@ -0,0 +1,65 @@
+#include "bal_spsc_queue.h"
+#include <stddef.h>
+
+void
+bal_spsc_queue_init(bal_spsc_queue_t *queue)
+{
+    if (BAL_UNLIKELY(NULL == queue))
+    {
+        return;
+    }
+
+    atomic_init(&queue->head, 0);
+    atomic_init(&queue->tail, 0);
+
+    for (size_t i = 0; i < BAL_SPSC_QUEUE_CAPACITY; ++i)
+    {
+        queue->buffer[i] = 0;
+    }
+}
+
+bool
+bal_spsc_queue_push(bal_spsc_queue_t *queue, const bal_guest_address_t address)
+{
+    if (BAL_UNLIKELY(NULL == queue))
+    {
+        return false;
+    }
+
+    const size_t current_head = atomic_load_explicit(&queue->head, memory_order_relaxed);
+    const size_t next_head    = current_head + 1 & BAL_SPSC_QUEUE_MASK;
+    const size_t current_tail = atomic_load_explicit(&queue->tail, memory_order_acquire);
+
+    if (next_head == current_tail)
+    {
+        // Queue is full. Drop the request.
+        return false;
+    }
+
+    queue->buffer[current_head] = address;
+    atomic_store_explicit(&queue->head, next_head, memory_order_release);
+    return true;
+}
+
+bool
+bal_spsc_queue_pop(bal_spsc_queue_t *queue, bal_guest_address_t *out_address)
+{
+    if (BAL_UNLIKELY(NULL == queue || NULL == out_address))
+    {
+        return false;
+    }
+
+    const size_t current_tail = atomic_load_explicit(&queue->tail, memory_order_relaxed);
+    const size_t current_head = atomic_load_explicit(&queue->head, memory_order_acquire);
+
+    if (current_head == current_tail)
+    {
+        // Queue is empty.
+        return false;
+    }
+
+    *out_address           = queue->buffer[current_head];
+    const size_t next_tail = current_tail + 1 & BAL_SPSC_QUEUE_MASK;
+    atomic_store_explicit(&queue->tail, next_tail, memory_order_release);
+    return true;
+}