Handle buffer wrap properly in SlickQueue

Refactor reserved_info struct for better buffer management and add buffer wrap test case. Bump version to 1.0.2.0.

Author: kzhdev

CHANGELOG

@@ -1,3 +1,8 @@
+# v1.0.2.0 - 2025-09-29
+- Handles buffer wrap properly
+- Refactor to use reserved_info struct for better buffer management 
+- Add buffer wrap test case
+
 # v1.0.1.2 - 2025-09-23
 - Fixed dist_slick_queue might cause third-party dependency installation error in release build
 

CMakeLists.txt

@@ -1,6 +1,6 @@
 cmake_minimum_required(VERSION 3.10)
 
-set(BUILD_VERSION 1.0.1.2)
+set(BUILD_VERSION 1.0.2.0)
 
 project(slick_queue
         VERSION ${BUILD_VERSION}

include/slick_queue/slick_queue.h

@@ -51,13 +51,18 @@ class SlickQueue {
         uint32_t size = 1;
     };
 
+    struct reserved_info {
+        uint_fast64_t index_ = 0;
+        uint_fast32_t size_ = 0;
+    };
+
     uint32_t size_;
-    uint32_t buffered_size_;
     uint32_t mask_;
+    uint32_t last_data_size_;
     T* data_ = nullptr;
     slot* control_ = nullptr;
-    std::atomic_uint_fast64_t* reserved_ = nullptr;
-    std::atomic_uint_fast64_t reserved_local_{ 0 };
+    std::atomic<reserved_info>* reserved_ = nullptr;
+    std::atomic<reserved_info> reserved_local_;
     bool own_ = false;
     bool use_shm_ = false;
 #if defined(_MSC_VER)
@@ -81,10 +86,9 @@ class SlickQueue {
      */
     SlickQueue(uint32_t size, const char* const shm_name = nullptr)
         : size_(size)
-        , buffered_size_(size + 1024)   // add some buffer at the end
         , mask_(size - 1)
-        , data_(shm_name ? nullptr : new T[buffered_size_])   
-        , control_(shm_name ? nullptr : new slot[buffered_size_])
+        , data_(shm_name ? nullptr : new T[size_])   
+        , control_(shm_name ? nullptr : new slot[size_])
         , reserved_(shm_name ? nullptr : &reserved_local_)
         , own_(shm_name == nullptr)
         , use_shm_(shm_name != nullptr)
@@ -122,7 +126,7 @@ class SlickQueue {
         }
 #else
         if (lpvMem_) {
-            auto BF_SZ = static_cast<size_t>(64 + sizeof(slot) * buffered_size_ + sizeof(T) * buffered_size_);
+            auto BF_SZ = static_cast<size_t>(64 + sizeof(slot) * size_ + sizeof(T) * size_);
             munmap(lpvMem_, BF_SZ);
             lpvMem_ = nullptr;
         }
@@ -167,7 +171,7 @@ class SlickQueue {
      * @return Initial reading index
      */
     uint64_t initial_reading_index() const noexcept {
-        return reserved_->load(std::memory_order_relaxed);
+        return reserved_->load(std::memory_order_relaxed).index_;
     }
 
     /**
@@ -176,7 +180,33 @@ class SlickQueue {
      * @return The starting index of the reserved space
      */
     uint64_t reserve(uint32_t n = 1) noexcept {
-        return reserved_->fetch_add(n, std::memory_order_acq_rel);
+        auto reserved = reserved_->load(std::memory_order_relaxed);
+        reserved_info next;
+        uint64_t index;
+        bool buffer_wrapped = false;
+        do {
+            next = reserved;
+            index = reserved.index_;
+            auto idx = index & mask_;
+            if ((idx + n) > size_) {
+                // if there is no enough buffer left, start from the beginning
+                index += size_ - idx;
+                next.index_ = index + n;
+                next.size_ = n;
+                buffer_wrapped = true;
+            }
+            else {
+                next.index_ += n;
+                next.size_ = n;
+            }
+        } while(!reserved_->compare_exchange_weak(reserved, next, std::memory_order_release, std::memory_order_relaxed));
+        if (buffer_wrapped) {
+            // 
+            auto& slot = control_[reserved.index_ & mask_];
+            slot.size = n;
+            slot.data_index.store(index, std::memory_order_release);
+        }
+        return index;
     }
 
     /**
@@ -205,6 +235,7 @@ class SlickQueue {
     void publish(uint64_t index, uint32_t n = 1) noexcept {
         auto& slot = control_[index & mask_];
         slot.size = n;
+        last_data_size_ = n;
         slot.data_index.store(index, std::memory_order_release);
     }
 
@@ -214,9 +245,10 @@ class SlickQueue {
      * @return Pair of pointer to the data and the size of the data, or nullptr and 0 if no data is available
      */
     std::pair<T*, uint32_t> read(uint64_t& read_index) noexcept {
-        auto& slot = control_[read_index & mask_];
+        auto idx = read_index & mask_;
+        auto& slot = control_[idx];
         auto index = slot.data_index.load(std::memory_order_relaxed);
-        if (index != std::numeric_limits<uint64_t>::max() && reserved_->load(std::memory_order_relaxed) < index) {
+        if (index != std::numeric_limits<uint64_t>::max() && reserved_->load(std::memory_order_relaxed).index_ < index) {
             // queue has been reset
             read_index = 0;
         }
@@ -225,9 +257,12 @@ class SlickQueue {
             // data not ready yet
             return std::make_pair(nullptr, 0);
         }
+        else if (index > read_index && ((index & mask_) != idx)) {
+            read_index = index;
+        }
 
         auto& data = data_[read_index & mask_];
-        read_index = slot.data_index + slot.size;
+        read_index = index + slot.size;
         return std::make_pair(&data, slot.size);
     }
 
@@ -237,22 +272,10 @@ class SlickQueue {
     */
     T* read_last() noexcept {
         auto reserved = reserved_->load(std::memory_order_relaxed);
-        if (reserved == 0) {
+        if (reserved.index_ == 0) {
             return nullptr;
         }
-        auto index = reserved - 1;
-
-        // find last published data
-        auto begin = index & mask_;
-        while (control_[index & mask_].data_index.load(std::memory_order_relaxed) != index)
-        {
-            --index;
-            if ((index & mask_) == begin    // looped entire queue 
-                || index >= reserved)       // passed 0
-            {
-                return nullptr;
-            }
-        }
+        auto index = reserved.index_ - reserved.size_;
         return &data_[index & mask_];
     }
 
@@ -263,12 +286,12 @@ class SlickQueue {
      */
     void reset() noexcept {
         if (use_shm_) {
-            control_ = new ((uint8_t*)lpvMem_ + 64) slot[buffered_size_];
+            control_ = new ((uint8_t*)lpvMem_ + 64) slot[size_];
         } else {
             delete [] control_;
-            control_ = new slot[buffered_size_];
+            control_ = new slot[size_];
         }
-        reserved_->store(0, std::memory_order_release);
+        reserved_->store(reserved_info(), std::memory_order_release);
     }
 
 private:
@@ -305,12 +328,11 @@ class SlickQueue {
             }
             size_ = *reinterpret_cast<uint32_t*>(reinterpret_cast<uint8_t*>(lpvMem) + sizeof(std::atomic_uint_fast64_t));
             mask_ = size_ - 1;
-            buffered_size_ = size_ + 1024;
-            BF_SZ = 64 + sizeof(slot) * buffered_size_ + sizeof(T) * buffered_size_;
+            BF_SZ = 64 + sizeof(slot) * size_ + sizeof(T) * size_;
             UnmapViewOfFile(lpvMem);
         }
         else {
-            BF_SZ = 64 + sizeof(slot) * buffered_size_ + sizeof(T) * buffered_size_;
+            BF_SZ = 64 + sizeof(slot) * size_ + sizeof(T) * size_;
 
             SECURITY_ATTRIBUTES sa;
             sa.nLength = sizeof(SECURITY_ATTRIBUTES);
@@ -348,20 +370,20 @@ class SlickQueue {
         }
 
         if (own_) {
-            reserved_ = new (lpvMem_) std::atomic_uint_fast64_t{ 0 };
-            *reinterpret_cast<uint32_t*>(reinterpret_cast<uint8_t*>(lpvMem_) + sizeof(std::atomic_uint_fast64_t)) = size_;
-            control_ = new ((uint8_t*)lpvMem_ + 64) slot[buffered_size_];
-            data_ = new ((uint8_t*)lpvMem_ + 64 + sizeof(slot) * buffered_size_) T[buffered_size_];
+            reserved_ = new (lpvMem_) std::atomic<reserved_info>();
+            *reinterpret_cast<uint32_t*>(reinterpret_cast<uint8_t*>(lpvMem_) + sizeof(std::atomic<reserved_info>)) = size_;
+            control_ = new ((uint8_t*)lpvMem_ + 64) slot[size_];
+            data_ = new ((uint8_t*)lpvMem_ + 64 + sizeof(slot) * size_) T[size_];
         }
         else {
-            reserved_ = reinterpret_cast<std::atomic_uint_fast64_t*>(lpvMem_);
+            reserved_ = reinterpret_cast<std::atomic<reserved_info>*>(lpvMem_);
             control_ = reinterpret_cast<slot*>((uint8_t*)lpvMem_ + 64);
-            data_ = reinterpret_cast<T*>((uint8_t*)lpvMem_ + 64 + sizeof(slot) * buffered_size_);
+            data_ = reinterpret_cast<T*>((uint8_t*)lpvMem_ + 64 + sizeof(slot) * size_);
         }
     }
 #else
     void allocate_shm_data(const char* const shm_name, bool open_only) {
-        size_t BF_SZ = 64 + sizeof(slot) * buffered_size_ + sizeof(T) * buffered_size_;
+        size_t BF_SZ = 64 + sizeof(slot) * size_ + sizeof(T) * size_;
         shm_name_ = shm_name;
         int flags = open_only ? O_RDWR : (O_RDWR | O_CREAT | O_EXCL);
         shm_fd_ = shm_open(shm_name, flags, 0666);
@@ -392,14 +414,14 @@ class SlickQueue {
         }
 
         if (own_) {
-            reserved_ = new (lpvMem_) std::atomic_uint_fast64_t{ 0 };
-            *reinterpret_cast<uint32_t*>(reinterpret_cast<uint8_t*>(lpvMem_) + sizeof(std::atomic_uint_fast64_t)) = mask_ + 1;
-            control_ = new ((uint8_t*)lpvMem_ + 64) slot[buffered_size_];
-            data_ = new ((uint8_t*)lpvMem_ + 64 + sizeof(slot) * buffered_size_) T[buffered_size_];
+            reserved_ = new (lpvMem_) std::atomic<reserved_info>{ 0, 0 };
+            *reinterpret_cast<uint32_t*>(reinterpret_cast<uint8_t*>(lpvMem_) + sizeof(std::atomic<reserved_info>)) = mask_ + 1;
+            control_ = new ((uint8_t*)lpvMem_ + 64) slot[size_];
+            data_ = new ((uint8_t*)lpvMem_ + 64 + sizeof(slot) * size_) T[size_];
         } else {
-            reserved_ = reinterpret_cast<std::atomic_uint_fast64_t*>(lpvMem_);
+            reserved_ = reinterpret_cast<std::atomic<reserved_info>*>(lpvMem_);
             control_ = reinterpret_cast<slot*>((uint8_t*)lpvMem_ + 64);
-            data_ = reinterpret_cast<T*>((uint8_t*)lpvMem_ + 64 + sizeof(slot) * buffered_size_);
+            data_ = reinterpret_cast<T*>((uint8_t*)lpvMem_ + 64 + sizeof(slot) * size_);
         }
     }
 #endif

tests/tests.cpp

@@ -12,14 +12,6 @@ TEST_CASE("Read empty queue") {
   REQUIRE(read.first == nullptr);
 }
 
-//TEST_CASE("RESET read cursor") {
-//  SlickQueue<int> queue(2);
-//  uint64_t read_cursor = std::numeric_limits<uint64_t>::max();
-//  auto read = queue.read(read_cursor);
-//  REQUIRE(read.first == nullptr);
-//  REQUIRE(read_cursor == 0);
-//}
-
 TEST_CASE( "Reserve") {
   SlickQueue<int> queue(2);
   auto reserved = queue.reserve();
@@ -81,3 +73,35 @@ TEST_CASE( "Publish and read multiple" ) {
   REQUIRE(*read.first == 23);
 }
 
+TEST_CASE( "buffer wrap" ) {
+  SlickQueue<char> queue(8);
+  uint64_t read_cursor = 0;
+  
+  auto reserved = queue.reserve(3);
+  REQUIRE( reserved == 0 );
+  memcpy(queue[reserved], "123", 3);
+  queue.publish(reserved, 3);
+  auto read = queue.read(read_cursor);
+  REQUIRE( read.first != nullptr );
+  REQUIRE( read_cursor == 3);
+  REQUIRE( strncmp(read.first, "123", 3) == 0);
+
+  reserved = queue.reserve(3);
+  REQUIRE( reserved == 3 );
+  memcpy(queue[reserved], "456", 3);
+  queue.publish(reserved, 3);
+  read = queue.read(read_cursor);
+  REQUIRE( read.first != nullptr );
+  REQUIRE( read_cursor == 6);
+  REQUIRE( strncmp(read.first, "456", 3) == 0);
+
+  reserved = queue.reserve(3);
+  REQUIRE( reserved == 8 );
+  memcpy(queue[reserved], "789", 3);
+  queue.publish(reserved, 3);
+  read = queue.read(read_cursor);
+  REQUIRE( read.first != nullptr );
+  REQUIRE( read_cursor == 11);
+  REQUIRE( strncmp(read.first, "789", 3) == 0);
+}
+