Aryan

.github/workflows/tests.yaml

@@ -0,0 +1,100 @@
+name: Queue Tests
+
+on:
+  push:
+  pull_request:
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+
+    steps:
+    - name: Checkout repo
+      uses: actions/checkout@v3
+      with:
+        fetch-depth: 0
+
+    # 🔹 Detect changed files
+    - name: Detect changes
+      id: changes
+      run: |
+        if [ "${{ github.event_name }}" = "pull_request" ]; then
+          BASE="origin/main"
+          git fetch origin main
+        else
+          BASE="${{ github.event.before }}"
+        fi
+
+        echo "Base: $BASE"
+        echo "Head: ${{ github.sha }}"
+
+        CHANGED_FILES=$(git diff --name-only $BASE ${{ github.sha }})
+
+        echo "Changed files:"
+        echo "$CHANGED_FILES"
+
+        # SPSC (bounded + unbounded)
+        if echo "$CHANGED_FILES" | grep -E "lockfree_spsc"; then
+          echo "run_spsc=true" >> $GITHUB_OUTPUT
+        fi
+
+        # MPSC
+        if echo "$CHANGED_FILES" | grep -E "lockfree_mpsc"; then
+          echo "run_mpsc=true" >> $GITHUB_OUTPUT
+        fi
+
+        # MPMC
+        if echo "$CHANGED_FILES" | grep -E "lockfree_mpmc|blocking_mpmc"; then
+          echo "run_mpmc=true" >> $GITHUB_OUTPUT
+        fi
+
+        # If tests themselves change → run corresponding
+        if echo "$CHANGED_FILES" | grep -E "test_spsc.cpp"; then
+          echo "run_spsc=true" >> $GITHUB_OUTPUT
+        fi
+
+        if echo "$CHANGED_FILES" | grep -E "test_mpsc.cpp"; then
+          echo "run_mpsc=true" >> $GITHUB_OUTPUT
+        fi
+
+        if echo "$CHANGED_FILES" | grep -E "test_mpmc.cpp"; then
+          echo "run_mpmc=true" >> $GITHUB_OUTPUT
+        fi
+
+    # 🔹 Install deps
+    - name: Install dependencies
+      run: |
+        sudo apt-get update
+        sudo apt-get install -y cmake g++ libgtest-dev
+
+    # 🔹 Build GTest
+    - name: Build GTest
+      run: |
+        cd /usr/src/gtest
+        sudo cmake .
+        sudo make
+        sudo cp lib/*.a /usr/lib
+
+    # 🔹 Build project
+    - name: Build project
+      run: |
+        rm -rf build
+        cmake -B build
+        cmake --build build
+
+    # 🔹 Selective tests
+    - name: Run SPSC tests
+      if: steps.changes.outputs.run_spsc == 'true'
+      run: cd build && ./test_spsc
+
+    - name: Run MPSC tests
+      if: steps.changes.outputs.run_mpsc == 'true'
+      run: cd build && ./test_mpsc
+
+    - name: Run MPMC tests
+      if: steps.changes.outputs.run_mpmc == 'true'
+      run: cd build && ./test_mpmc
+
+    # 🔹 Safety net (always runs)
+    # - name: Run full test suite (safety)
+    #   run: cd build && ctest --output-on-failure

CMakeLists.txt

@@ -0,0 +1,25 @@
+cmake_minimum_required(VERSION 3.14)
+project(ThreadsafeQueueLib)
+
+set(CMAKE_CXX_STANDARD 17)
+enable_testing()
+
+find_package(GTest REQUIRED)
+include_directories(include)
+
+# SPSC tests
+add_executable(test_spsc tests/test_spsc.cpp)
+target_link_libraries(test_spsc GTest::GTest GTest::Main pthread)
+
+# MPSC tests
+add_executable(test_mpsc tests/test_mpsc.cpp)
+target_link_libraries(test_mpsc GTest::GTest GTest::Main pthread)
+
+# MPMC tests
+add_executable(test_mpmc tests/test_mpmc.cpp)
+target_link_libraries(test_mpmc GTest::GTest GTest::Main pthread)
+
+# Register tests
+add_test(NAME SPSC COMMAND test_spsc)
+add_test(NAME MPSC COMMAND test_mpsc)
+add_test(NAME MPMC COMMAND test_mpmc)

include/FAST_lockfree_spsc_unbounded/block.hpp

@@ -0,0 +1,61 @@
+#ifndef FAST_LOCKFREE_SPSC_UNBOUNDED_BLOCK
+#define FAST_LOCKFREE_SPSC_UNBOUNDED_BLOCK
+
+#include "../utils.hpp"
+#include "slots.hpp"
+#include <atomic>
+
+namespace tsfqueue::impl{
+    template <typename T> 
+    class Block_FAST{
+    private:
+        using Slot_FAST = tsfqueue::FAST::Slot_FAST;
+        std::size_t capacity;
+        std::size_t mask;
+        char* raw;
+        char* data;
+        std::unique_ptr<Slot_FAST> slots;
+        template<typename U>
+        void inner_enqueue(U&& data){
+
+        }
+    public:
+        Block_FAST(std::size_t cap){
+            // Initialize by bringing it to the closest power of 2
+            // And then make mask
+            slots(static_cast<int>(cap));
+        }
+        ~Block_FAST(){
+            // Make destructor
+        }
+        Block_FAST(const Block_FAST& other) = delete;
+        Block_FAST& operator=(const Block_FAST& other) = delete;
+        Block_FAST(Block_FAST&& other){
+            // Move constructor
+        }
+        template<typename U>
+        bool try_enqueue(U&& data){
+            // Try to check if can push
+            if(slots->try_get()){
+                inner_enqueue(std::forward<U>(data));
+                return true;
+            }
+            return false;
+        }
+        template <typename U>
+        void wait_enqueue(U&& data){
+            slots->wait_and_get();
+            inner_enqueue(std::forward<U>(data));
+        }
+        template <typename U>
+        bool wait_enqueue_timed(U&& data, std::int64_t time_usecs){
+            if(slots->timed_get(time_usecs)){
+                inner_enqueue(std::forward<U>(data));
+                return true;
+            }
+            return false;
+        }
+    };
+}
+
+#endif

include/FAST_lockfree_spsc_unbounded/slots.hpp

@@ -0,0 +1,106 @@
+#ifndef FAST_LOCKFREE_SPSC_UNBOUNDED_SLOT
+#define FAST_LOCKFREE_SPSC_UNBOUNDED_SLOT
+
+#include "../utils.hpp"
+#include <atomic>
+
+namespace tsfqueue::FAST{
+#if defined(__MACH__)
+#include <mach/mach.h>
+    class Semaphore_FAST{
+    private:
+        semaphore_t sema;
+        Semaphore_FAST(const Semaphore& other) = delete;
+        Semaphore_FAST(Semaphore&& other) = delete;
+    public:
+        Semaphore_FAST(int count = 0){
+            assert(count >= 0);
+            kern_return_t ret = semaphore_create(mach_task_self(), &sema, SYNC_POLICY_FIFO, count);
+            assert(ret == KERN_SUCCESS);
+        }
+        ~Semaphore_FAST(){
+            semaphore_destroy(mach_task_self(), sema);
+        }
+        bool try_get(){
+            return timed_get(0);
+        }
+        bool timed_get(std::uint64_t time_usecs){
+            mach_timespec_t time;
+            time.tv_sec = static_cast<unsigned int>(time_usecs / 1000000);
+			time.tv_nsec = static_cast<int>((time_usecs % 1000000) * 1000);
+            kern_return_t ret = semaphore_timedwait(sema, time);
+            return ret == KERN_SUCCESS;
+        }
+        void wait_and_get(){
+            semaphore_wait(sema);
+        }
+        void signal(int times = 1){
+            while(times--)
+                while(semaphore_signal(sema) != KERN_SUCCESS);
+        }
+    };
+#endif
+
+
+    class Slot_FAST{
+    private:
+        std::atomic<int> counter;
+        Semaphore_FAST sema;
+        static constexpr int SPIN_COUNT = 1024; // Change this for benchmakrs & set to best possible
+        inline bool hot_path(){
+            for (int i = 0; i < SPIN_COUNT; i++){
+                if(counter.load(std::memory_order_acquire) > 0){
+                    counter.fetch_sub(1);
+                    return true;
+                }
+            }
+            return false;
+        }
+        bool get_with_sleep(std::int64_t time_usecs = -1){
+            if(hot_path()) return true;
+            counter.fetch_sub(1);
+            if(time_usecs < 0){
+                sema.wait_and_get()
+                return true;
+            }
+            if(sema.timed_get(static_cast<std::uint64_t>(time_usecs))){
+                return true;
+            }
+
+            // Restore the semaphore
+            // Signal happened just after timeout expired
+            // Thus we add a while loop which keeps checking until all boundary conditions
+            // are gone. Super clever. 
+            while(true){
+                int old = counter.fetch_add(1);
+                if(old < 0) return false; // Restored successfully
+                old = counter.fetch_sub(1);
+                if(old > 0 && sema.try_get()){
+                    return true;
+                }
+            }
+        }
+    public:
+        Slot_FAST(int count) : counter(count), sema(0) {}
+        bool try_get(){
+            // Ok since SPSC so counter cannot be decremented between load and fetch_sub
+            if(counter.load(std::memory_order_acquire) > 0){
+                counter.fetch_sub(1);
+                return true;
+            }
+            return false;
+        }
+        bool timed_get(std::int64_t time_usecs){
+            return get_with_sleep(time_usecs);
+        }
+        void wait_and_get(){
+            return get_with_sleep();
+        }
+        void signal(int times = 1){
+            int old = counter.fetch_add(1);
+            if(old < 0) sema.signal();
+        }
+    }
+};
+
+#endif