Enhance queue performance with CPU relax backoff and fast path for reserve(1); Increase version number

Author: kzhdev

CHANGELOG

@@ -1,8 +1,11 @@
-# v1.2.3 - unreleased
+# v1.2.3 - 2026-01-29
 - 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
+- Align hot atomics to cache lines and add light spin backoff under contention
+- Add reserve(1) fast path using fetch_add
+- Add SLICK_QUEUE_ENABLE_CPU_RELAX to control pause/yield backoff
 - Guard against zero-length reservations
 - Add limits header
 

CMakeLists.txt

@@ -1,7 +1,7 @@
 cmake_minimum_required(VERSION 3.10)
 
 project(slick-queue
-        VERSION 1.2.2
+        VERSION 1.2.3
         DESCRIPTION "A C++ Lock-Free MPMC queue"
         LANGUAGES CXX)
 

README.md

@@ -225,6 +225,8 @@ SlickQueue(const char* shm_name);                  // Reader/Attacher
 
 **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.
 
+**CPU Relax Backoff**: Define `SLICK_QUEUE_ENABLE_CPU_RELAX=0` to disable the pause/yield backoff used on contended CAS loops (default is enabled). Disabling may reduce latency in very short contention bursts but can increase CPU usage under load.
+
 **⚠️ 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

@@ -12,13 +12,15 @@
 #pragma once
 
 #include <cstdint>
+#include <cstddef>
 #include <atomic>
 #include <stdexcept>
 #include <string>
 #include <cassert>
 #include <thread>
 #include <chrono>
 #include <limits>
+#include <new>
 
 #if defined(_MSC_VER)
 #ifndef NOMINMAX
@@ -37,6 +39,9 @@
 #include <cerrno>
 #endif
 
+#if defined(_MSC_VER) && (defined(_M_IX86) || defined(_M_X64))
+#include <immintrin.h>
+#endif
 
 #ifndef SLICK_QUEUE_ENABLE_LOSS_DETECTION
 #if !defined(NDEBUG)
@@ -46,6 +51,10 @@
 #endif
 #endif
 
+#ifndef SLICK_QUEUE_ENABLE_CPU_RELAX
+#define SLICK_QUEUE_ENABLE_CPU_RELAX 1
+#endif
+
 namespace slick {
 
 /**
@@ -66,14 +75,20 @@ class SlickQueue {
 
     using reserved_info = uint64_t;
 
+#if defined(__cpp_lib_hardware_interference_size)
+    static constexpr std::size_t cacheline_size = std::hardware_destructive_interference_size;
+#else
+    static constexpr std::size_t cacheline_size = 64;
+#endif
+
     uint32_t size_;
     uint32_t mask_;
     T* data_ = nullptr;
     slot* control_ = nullptr;
     std::atomic<reserved_info>* reserved_ = nullptr;
-    std::atomic<reserved_info> reserved_local_{0};
+    alignas(cacheline_size) std::atomic<reserved_info> reserved_local_{0};
 #if SLICK_QUEUE_ENABLE_LOSS_DETECTION
-    std::atomic<uint64_t> loss_count_{0};
+    alignas(cacheline_size) std::atomic<uint64_t> loss_count_{0};
 #endif
     bool own_ = false;
     bool use_shm_ = false;
@@ -223,11 +238,23 @@ class SlickQueue {
         if (n > size_) [[unlikely]] {
             throw std::runtime_error("required size " + std::to_string(n) + " > queue size " + std::to_string(size_));
         }
+        if (n == 1) {
+            constexpr reserved_info step = (1ULL << 16);
+            auto prev = reserved_->fetch_add(step, std::memory_order_release);
+            auto index = get_index(prev);
+            auto prev_size = get_size(prev);
+            if (prev_size != 1) {
+                auto expected = make_reserved_info(index + 1, prev_size);
+                reserved_->compare_exchange_strong(expected, make_reserved_info(index + 1, 1),
+                    std::memory_order_release, std::memory_order_relaxed);
+            }
+            return index;
+        }
         auto reserved = reserved_->load(std::memory_order_relaxed);
-        uint64_t next;
-        uint64_t index;
+        uint64_t next = 0;
+        uint64_t index = 0;
         bool buffer_wrapped = false;
-        do {
+        for (;;) {
             buffer_wrapped = false;
             index = get_index(reserved);
             auto idx = index & mask_;
@@ -240,7 +267,11 @@ class SlickQueue {
             else {
                 next = make_reserved_info(index + n, n);
             }
-        } while(!reserved_->compare_exchange_weak(reserved, next, std::memory_order_release, std::memory_order_relaxed));
+            if (reserved_->compare_exchange_weak(reserved, next, std::memory_order_release, std::memory_order_relaxed)) {
+                break;
+            }
+            cpu_relax();
+        }
         if (buffer_wrapped) {
             // queue wrapped, set current slock.data_index to the reserved index to let the reader
             // know the next available data is in different slot.
@@ -370,6 +401,7 @@ class SlickQueue {
                 // Successfully claimed the item
                 return std::make_pair(&data_[current_index & mask_], current_slot->size);
             }
+            cpu_relax();
             // CAS failed, another consumer claimed it, retry
         }
     }
@@ -420,6 +452,22 @@ class SlickQueue {
         return static_cast<uint32_t>(reserved & 0xFFFF);
     }
 
+    static inline void cpu_relax() noexcept {
+#if SLICK_QUEUE_ENABLE_CPU_RELAX
+#if defined(_MSC_VER) && (defined(_M_IX86) || defined(_M_X64))
+        _mm_pause();
+#elif defined(__i386__) || defined(__x86_64__)
+        __builtin_ia32_pause();
+#elif defined(__aarch64__) || defined(__arm__)
+        __asm__ __volatile__("yield" ::: "memory");
+#else
+        std::this_thread::yield();
+#endif
+#else
+        (void)0;
+#endif
+    }
+
 #if defined(_MSC_VER)
     void allocate_shm_data(const char* const shm_name, bool open_only) {
         DWORD BF_SZ;

tests/tests.cpp

@@ -125,6 +125,23 @@ TEST(SlickQueueTests, BufferWrap) {
   EXPECT_EQ(strncmp(read.first, "789", 3), 0);
 }
 
+TEST(SlickQueueTests, ReadLastUsesLatestReserveSize) {
+  SlickQueue<int> queue(8);
+
+  auto first = queue.reserve(2);
+  *queue[first] = 1;
+  *queue[first + 1] = 2;
+  queue.publish(first, 2);
+
+  auto last = queue.reserve(1);
+  *queue[last] = 3;
+  queue.publish(last, 1);
+
+  auto latest = queue.read_last();
+  ASSERT_NE(latest, nullptr);
+  EXPECT_EQ(*latest, 3);
+}
+
 TEST(SlickQueueTests, LossyOverwriteSkipsOldData) {
   SlickQueue<int> queue(2);
   uint64_t read_cursor = 0;