From 3797d014ffb8acaa19bf7e5d2edaa4c593ba3ff8 Mon Sep 17 00:00:00 2001 From: Ronald Caesar Date: Sun, 3 May 2026 00:56:58 -0400 Subject: [PATCH] engine: add single producer single consumer queue This allows the main and background thread to communicate with each other. Signed-off-by: Ronald Caesar --- CMakeLists.txt | 3 +- include/bal_spsc_queue.h | 64 +++++++++++++++++++++++++++++++++++++++ src/bal_spsc_queue.c | 65 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 131 insertions(+), 1 deletion(-) create mode 100644 include/bal_spsc_queue.h create mode 100644 src/bal_spsc_queue.c diff --git a/CMakeLists.txt b/CMakeLists.txt index 3056f89d..47fd5b29 100644 --- a/CMakeLists.txt +++ b/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} diff --git a/include/bal_spsc_queue.h b/include/bal_spsc_queue.h new file mode 100644 index 00000000..ff77854f --- /dev/null +++ b/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 +#include +#include + +#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 ***/ diff --git a/src/bal_spsc_queue.c b/src/bal_spsc_queue.c new file mode 100644 index 00000000..6a1dc0b5 --- /dev/null +++ b/src/bal_spsc_queue.c @@ -0,0 +1,65 @@ +#include "bal_spsc_queue.h" +#include + +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; +} \ No newline at end of file