Enhance lossy queue semantics and testing

- Update changelog with new features and fixes
- Improve README documentation for lossy semantics and loss detection
- Implement loss detection in SlickQueue with debug support
- Add tests for lossy overwrite behavior and loss detection
- Modify CMake to enable loss detection in tests

Author: kzhdev

CHANGELOG

@@ -1,4 +1,9 @@
 # v1.2.3 - unreleased
+- Initialize reserved counters for local and shared-memory queues
+- Validate shared-memory size is power-of-two and element size matches when attaching
+- Add lossy semantics documentation and debug loss detection counter
+- Add lossy overwrite tests for in-process and shared-memory queues
+- Guard against zero-length reservations
 - Add limits header
 
 # v1.2.2 - 2026-01-10

README.md

@@ -207,19 +207,24 @@ SlickQueue(const char* shm_name);                  // Reader/Attacher
 
 ### Core Methods
 
-- `uint64_t reserve(uint32_t n = 1)` - Reserve `n` slots for writing (blocks if queue is full)
+- `uint64_t reserve(uint32_t n = 1)` - Reserve `n` slots for writing (non-blocking; may overwrite old data if consumers lag)
 - `T* operator[](uint64_t slot)` - Access reserved slot
 - `void publish(uint64_t slot, uint32_t n = 1)` - Publish `n` written items to consumers
 - `std::pair<T*, uint32_t> read(uint64_t& cursor)` - Read next available item (independent cursor)
 - `std::pair<T*, uint32_t> read(std::atomic<uint64_t>& cursor)` - Read next available item (shared atomic cursor for work-stealing)
 - `T* read_last()` - Read the most recently published item without a cursor
 - `uint32_t size()` - Get queue capacity
+- `uint64_t loss_count() const` - Get count of skipped items due to overwrite (debug-only if enabled)
 - `void reset()` - Reset the queue, invalidating all existing data
 
 ### Important Constraints
 
 **Lock-Free Atomics Implementation**: SlickQueue uses a packed 64-bit atomic internally to guarantee lock-free operations on all platforms. This packs both the write index (48 bits) and the reservation size (16 bits) into a single atomic value.
 
+**Lossy Semantics**: SlickQueue does not apply backpressure. If producers advance by at least the queue size before a consumer reads, older entries will be overwritten and the consumer will skip ahead to the latest value for a slot. Size the queue and read frequency to bound loss.
+
+**Debug Loss Detection**: Define `SLICK_QUEUE_ENABLE_LOSS_DETECTION=1` to enable a per-instance skipped-item counter (enabled by default in Debug builds). Use `loss_count()` to inspect how many items were skipped.
+
 **⚠️ Reserve Size Limitation**: When using `read_last()`, the number of slots in any `reserve(n)` call **must not exceed 65,535** (2^16 - 1). This is because the size is stored in 16 bits within the packed atomic.
 
 - For typical use cases with `reserve()` or `reserve(1)`, this limit is not a concern

include/slick/queue.h

@@ -18,6 +18,7 @@
 #include <cassert>
 #include <thread>
 #include <chrono>
+#include <limits>
 
 #if defined(_MSC_VER)
 #ifndef NOMINMAX
@@ -34,7 +35,15 @@
 #include <fcntl.h>
 #include <unistd.h>
 #include <cerrno>
-#include <limits>
+#endif
+
+
+#ifndef SLICK_QUEUE_ENABLE_LOSS_DETECTION
+#if !defined(NDEBUG)
+#define SLICK_QUEUE_ENABLE_LOSS_DETECTION 1
+#else
+#define SLICK_QUEUE_ENABLE_LOSS_DETECTION 0
+#endif
 #endif
 
 namespace slick {
@@ -44,6 +53,7 @@ namespace slick {
  * 
  * This queue allows a multiple producer thread to write data and a multiple consumer thread to read data concurrently without locks.
  * It can optionally use shared memory for inter-process communication.
+ * This queue is lossy: if producers outrun consumers, older data may be overwritten.
  * 
  * @tparam T The type of elements stored in the queue.
  */
@@ -61,7 +71,10 @@ class SlickQueue {
     T* data_ = nullptr;
     slot* control_ = nullptr;
     std::atomic<reserved_info>* reserved_ = nullptr;
-    std::atomic<reserved_info> reserved_local_;
+    std::atomic<reserved_info> reserved_local_{0};
+#if SLICK_QUEUE_ENABLE_LOSS_DETECTION
+    std::atomic<uint64_t> loss_count_{0};
+#endif
     bool own_ = false;
     bool use_shm_ = false;
 #if defined(_MSC_VER)
@@ -75,6 +88,10 @@ class SlickQueue {
 
     static constexpr uint32_t HEADER_SIZE = 64;
 
+    static constexpr bool is_power_of_two(uint32_t value) noexcept {
+        return value != 0 && ((value & (value - 1)) == 0);
+    }
+
 public:
     /**
      * @brief Construct a new SlickQueue object
@@ -87,16 +104,20 @@ class SlickQueue {
      */
     SlickQueue(uint32_t size, const char* const shm_name = nullptr)
         : size_(size)
-        , mask_(size - 1)
-        , data_(shm_name ? nullptr : new T[size_])   
-        , control_(shm_name ? nullptr : new slot[size_])
-        , reserved_(shm_name ? nullptr : &reserved_local_)
+        , mask_(size ? size - 1 : 0)
         , own_(shm_name == nullptr)
         , use_shm_(shm_name != nullptr)
     {
-        assert((size && !(size & (size - 1))) && "size must power of 2");
+        if (!is_power_of_two(size_)) {
+            throw std::invalid_argument("size must power of 2");
+        }
         if (shm_name) {
             allocate_shm_data(shm_name, false);
+        } else {
+            reserved_ = &reserved_local_;
+            reserved_->store(0, std::memory_order_relaxed);
+            data_ = new T[size_];
+            control_ = new slot[size_];
         }
     }
 
@@ -108,7 +129,9 @@ class SlickQueue {
      * @throws std::runtime_error if shared memory allocation fails or the segment does not exist.
      */
     SlickQueue(const char* const shm_name)
-        : own_(false)
+        : size_(0)
+        , mask_(0)
+        , own_(false)
         , use_shm_(true)
     {
         allocate_shm_data(shm_name, true);
@@ -167,6 +190,19 @@ class SlickQueue {
      */
     constexpr uint32_t size() const noexcept { return size_; }
 
+    
+    /**
+     * @brief Get the number of items skipped due to overwrite (debug-only if enabled).
+     * @return Count of skipped items observed by this queue instance.
+     */
+    uint64_t loss_count() const noexcept {
+#if SLICK_QUEUE_ENABLE_LOSS_DETECTION
+        return loss_count_.load(std::memory_order_relaxed);
+#else
+        return 0;
+#endif
+    }
+
     /**
      * @brief Get the initial reading index, which is 0 if the queue is newly created or the current writing index if opened existing 
      * @return Initial reading index
@@ -181,6 +217,9 @@ class SlickQueue {
      * @return The starting index of the reserved space
      */
     uint64_t reserve(uint32_t n = 1) {
+        if (n == 0) [[unlikely]] {
+            throw std::invalid_argument("required size must be > 0");
+        }
         if (n > size_) [[unlikely]] {
             throw std::runtime_error("required size " + std::to_string(n) + " > queue size " + std::to_string(size_));
         }
@@ -257,7 +296,13 @@ class SlickQueue {
                 // queue has been reset
                 read_index = 0;
             }
-    
+
+#if SLICK_QUEUE_ENABLE_LOSS_DETECTION
+            if (index != std::numeric_limits<uint64_t>::max() && index > read_index && ((index & mask_) == idx)) {
+                loss_count_.fetch_add(index - read_index, std::memory_order_relaxed);
+            }
+#endif
+
             if (index == std::numeric_limits<uint64_t>::max() || index < read_index) {
                 // data not ready yet
                 return std::make_pair(nullptr, 0);
@@ -300,7 +345,15 @@ class SlickQueue {
                 // data not ready yet
                 return std::make_pair(nullptr, 0);
             }
-            else if (index > current_index && ((index & mask_) != idx)) {
+
+#if SLICK_QUEUE_ENABLE_LOSS_DETECTION
+            uint64_t overrun = 0;
+            if (index > current_index && ((index & mask_) == idx)) {
+                overrun = index - current_index;
+            }
+#endif
+
+            if (index > current_index && ((index & mask_) != idx)) {
                 // queue wrapped, skip the unused slots
                 read_index.compare_exchange_weak(current_index, index, std::memory_order_release, std::memory_order_relaxed);
                 continue;
@@ -309,6 +362,11 @@ class SlickQueue {
             // Try to atomically claim this item
             uint64_t next_index = index + current_slot->size;
             if (read_index.compare_exchange_weak(current_index, next_index, std::memory_order_release, std::memory_order_relaxed)) {
+#if SLICK_QUEUE_ENABLE_LOSS_DETECTION
+                if (overrun != 0) {
+                    loss_count_.fetch_add(overrun, std::memory_order_relaxed);
+                }
+#endif
                 // Successfully claimed the item
                 return std::make_pair(&data_[current_index & mask_], current_slot->size);
             }
@@ -343,6 +401,9 @@ class SlickQueue {
             control_ = new slot[size_];
         }
         reserved_->store(0, std::memory_order_release);
+#if SLICK_QUEUE_ENABLE_LOSS_DETECTION
+        loss_count_.store(0, std::memory_order_relaxed);
+#endif
     }
 
 private:
@@ -395,6 +456,9 @@ class SlickQueue {
             if (element_size != sizeof(T)) {
                 throw std::runtime_error("Shared memory element size mismatch. Expected " + std::to_string(sizeof(T)) + " but got " + std::to_string(element_size));
             }
+            if (!is_power_of_two(size_)) {
+                throw std::runtime_error("Shared memory size must be power of 2. Got " + std::to_string(size_));
+            }
             mask_ = size_ - 1;
             BF_SZ = HEADER_SIZE + sizeof(slot) * size_ + sizeof(T) * size_;
             UnmapViewOfFile(lpvMem);
@@ -462,6 +526,7 @@ class SlickQueue {
 
         if (own_) {
             reserved_ = new (lpvMem_) std::atomic<reserved_info>();
+            reserved_->store(0, std::memory_order_relaxed);
             // save queue size
             *reinterpret_cast<uint32_t*>(reinterpret_cast<uint8_t*>(lpvMem_) + sizeof(std::atomic<reserved_info>)) = size_;
             // save element size
@@ -501,11 +566,16 @@ class SlickQueue {
                     throw std::runtime_error("Failed to map shm for size read. err=" + std::to_string(errno));
                 }
                 uint32_t shm_size = *reinterpret_cast<uint32_t*>(reinterpret_cast<uint8_t*>(temp_map) + sizeof(std::atomic<reserved_info>));
+                uint32_t element_size = *reinterpret_cast<uint32_t*>(reinterpret_cast<uint8_t*>(temp_map) + sizeof(std::atomic<reserved_info>) + sizeof(shm_size));
                 munmap(temp_map, HEADER_SIZE);
 
                 if (shm_size != size_) {
                     throw std::runtime_error("Shared memory size mismatch. Expected " + std::to_string(size_) + " but got " + std::to_string(shm_size));
                 }
+
+                if (element_size != sizeof(T)) {
+                    throw std::runtime_error("Shared memory element size mismatch. Expected " + std::to_string(sizeof(T)) + " but got " + std::to_string(element_size));
+                }
             } else {
                 throw std::runtime_error("Failed to open/create shm. err=" + std::to_string(errno));
             }
@@ -520,8 +590,18 @@ class SlickQueue {
                 throw std::runtime_error("Failed to map shm for size read. err=" + std::to_string(errno));
             }
             size_ = *reinterpret_cast<uint32_t*>(reinterpret_cast<uint8_t*>(temp_map) + sizeof(std::atomic<reserved_info>));
-            mask_ = size_ - 1;
+            uint32_t element_size = *reinterpret_cast<uint32_t*>(reinterpret_cast<uint8_t*>(temp_map) + sizeof(std::atomic<reserved_info>) + sizeof(size_));
             munmap(temp_map, HEADER_SIZE);
+
+            if (!is_power_of_two(size_)) {
+                throw std::runtime_error("Shared memory size must be power of 2. Got " + std::to_string(size_));
+            }
+
+            if (element_size != sizeof(T)) {
+                throw std::runtime_error("Shared memory element size mismatch. Expected " + std::to_string(sizeof(T)) + " but got " + std::to_string(element_size));
+            }
+
+            mask_ = size_ - 1;
         }
 
         BF_SZ = HEADER_SIZE + sizeof(slot) * size_ + sizeof(T) * size_;
@@ -539,6 +619,7 @@ class SlickQueue {
 
         if (own_) {
             reserved_ = new (lpvMem_) std::atomic<reserved_info>();
+            reserved_->store(0, std::memory_order_relaxed);
             *reinterpret_cast<uint32_t*>(reinterpret_cast<uint8_t*>(lpvMem_) + sizeof(std::atomic<reserved_info>)) = size_;
             *reinterpret_cast<uint32_t*>(reinterpret_cast<uint8_t*>(lpvMem_) + sizeof(std::atomic<reserved_info>) + sizeof(size_)) = sizeof(T);
             control_ = new ((uint8_t*)lpvMem_ + HEADER_SIZE) slot[size_];

tests/CMakeLists.txt

@@ -16,5 +16,8 @@ add_executable(slick-queue-tests tests.cpp shm_tests.cpp)
 
 target_link_libraries(slick-queue-tests PRIVATE slick::queue GTest::gtest_main)
 
+# Enable loss detection in tests
+target_compile_definitions(slick-queue-tests PRIVATE SLICK_QUEUE_ENABLE_LOSS_DETECTION=1)
+
 include(GoogleTest)
 gtest_discover_tests(slick-queue-tests)

tests/shm_tests.cpp

@@ -150,6 +150,43 @@ TEST(ShmTests, AtomicCursorWorkStealing) {
   EXPECT_EQ(shared_cursor.load(), 100);
 }
 
+TEST(ShmTests, LossyOverwriteSkipsOldData) {
+  SlickQueue<int> server(2, "sq_lossy_overwrite");
+  SlickQueue<int> client("sq_lossy_overwrite");
+
+  auto s0 = server.reserve();
+  *server[s0] = 10;
+  server.publish(s0);
+
+  auto s1 = server.reserve();
+  *server[s1] = 20;
+  server.publish(s1);
+
+  auto s2 = server.reserve();
+  *server[s2] = 30;
+  server.publish(s2);
+
+  uint64_t read_cursor = 0;
+  auto read = client.read(read_cursor);
+  EXPECT_NE(read.first, nullptr);
+  EXPECT_EQ(*read.first, 30);
+  EXPECT_EQ(read_cursor, 3);
+
+#if SLICK_QUEUE_ENABLE_LOSS_DETECTION
+  EXPECT_EQ(client.loss_count(), 2u);
+#endif
+
+  read = client.read(read_cursor);
+  EXPECT_EQ(read.first, nullptr);
+}
+
+TEST(ShmTests, ElementSizeMismatch) {
+  SlickQueue<int> server(4, "sq_element_mismatch");
+  EXPECT_THROW({
+    SlickQueue<double> client("sq_element_mismatch");
+  }, std::runtime_error);
+}
+
 TEST(ShmTests, SizeMismatch) {
   // Create a shared memory queue with size 4
   SlickQueue<int> server(4, "sq_size_mismatch");