diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml index a066607..c45c5f8 100644 --- a/.github/workflows/tests.yaml +++ b/.github/workflows/tests.yaml @@ -38,11 +38,6 @@ jobs: 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 @@ -53,11 +48,7 @@ jobs: 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 + if echo "$CHANGED_FILES" | grep -E "test_mpmc_unbounded_blocking.cpp|test_mpmc_bounded_lockfree.cpp"; then echo "run_mpmc=true" >> $GITHUB_OUTPUT fi @@ -87,14 +78,13 @@ jobs: 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 + run: | + cd build + ./test_mpmc_unbounded_blocking + ./test_mpmc_bounded_lockfree # 🔹 Safety net (always runs) - name: Run full test suite (safety) - run: cd build && ctest --output-on-failure \ No newline at end of file + run: cd build && ctest --output-on-failure diff --git a/CMakeLists.txt b/CMakeLists.txt index e78e8c7..3b35b71 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -2,32 +2,69 @@ cmake_minimum_required(VERSION 3.14) project(ThreadsafeQueueLib) set(CMAKE_CXX_STANDARD 17) +set(CMAKE_CXX_STANDARD_REQUIRED ON) enable_testing() find_package(Threads REQUIRED) find_package(GTest REQUIRED) include_directories(include) -option(ENABLE_TSAN "Enable ThreadSanitizer" OFF) +# --------------------------------------------------------------------------- +# Sanitizer options. Only one of TSAN / ASAN can be active per build. +# cmake -DENABLE_TSAN=ON .. -> ThreadSanitizer (data races) +# cmake -DENABLE_ASAN=ON .. -> AddressSanitizer (UAF, leaks, OOB) +# cmake -DENABLE_UBSAN=ON .. -> UndefinedBehaviorSanitizer +# Sanitizer builds use -O1 -g -fno-omit-frame-pointer for usable stacks. +# --------------------------------------------------------------------------- +option(ENABLE_TSAN "Enable ThreadSanitizer" OFF) +option(ENABLE_ASAN "Enable AddressSanitizer" OFF) +option(ENABLE_UBSAN "Enable UndefinedBehaviorSanitizer" OFF) + +if(ENABLE_TSAN AND ENABLE_ASAN) + message(FATAL_ERROR "TSAN and ASAN cannot be enabled together; pick one") +endif() if(ENABLE_TSAN) - add_compile_options(-fsanitize=thread -g -O1) + message(STATUS "ThreadSanitizer enabled") + add_compile_options(-fsanitize=thread -O1 -g -fno-omit-frame-pointer) add_link_options(-fsanitize=thread) + add_compile_definitions(TSFQUEUE_TSAN_BUILD=1) +endif() + +if(ENABLE_ASAN) + message(STATUS "AddressSanitizer enabled") + add_compile_options(-fsanitize=address -O1 -g -fno-omit-frame-pointer) + add_link_options(-fsanitize=address) +endif() + +if(ENABLE_UBSAN) + message(STATUS "UndefinedBehaviorSanitizer enabled") + add_compile_options(-fsanitize=undefined -O1 -g -fno-omit-frame-pointer) + add_link_options(-fsanitize=undefined) +endif() + +# --------------------------------------------------------------------------- +# 128-bit atomics (used by lockfree_mpmc_bounded packed entries). +# On x86_64, -mcx16 lets the compiler emit `cmpxchg16b` inline instead of +# generating __sync_*_16 libcalls that require linking libatomic. +# --------------------------------------------------------------------------- +if(CMAKE_SYSTEM_PROCESSOR MATCHES "(x86_64|AMD64)") + add_compile_options(-mcx16) endif() # SPSC tests add_executable(test_spsc tests/test_spsc.cpp) target_link_libraries(test_spsc GTest::GTest GTest::Main Threads::Threads) -# MPSC tests -add_executable(test_mpsc tests/test_mpsc.cpp) -target_link_libraries(test_mpsc GTest::GTest GTest::Main Threads::Threads) +# MPMC unbounded blocking tests +add_executable(test_mpmc_unbounded_blocking tests/test_mpmc_unbounded_blocking.cpp) +target_link_libraries(test_mpmc_unbounded_blocking GTest::GTest GTest::Main Threads::Threads) -# MPMC tests -add_executable(test_mpmc tests/test_mpmc.cpp) -target_link_libraries(test_mpmc GTest::GTest GTest::Main Threads::Threads) +# MPMC bounded lockfree tests +add_executable(test_mpmc_bounded_lockfree tests/test_mpmc_bounded_lockfree.cpp) +target_link_libraries(test_mpmc_bounded_lockfree GTest::GTest GTest::Main Threads::Threads) # Register tests add_test(NAME SPSC COMMAND test_spsc) -add_test(NAME MPSC COMMAND test_mpsc) -add_test(NAME MPMC COMMAND test_mpmc) \ No newline at end of file +add_test(NAME MPMC_UNBOUNDED_BLOCKING COMMAND test_mpmc_unbounded_blocking) +add_test(NAME MPMC_BOUNDED_LOCKFREE COMMAND test_mpmc_bounded_lockfree) diff --git a/include/lockfree_mpmc_bounded/defs.hpp b/include/lockfree_mpmc_bounded/defs.hpp new file mode 100644 index 0000000..6d9b585 --- /dev/null +++ b/include/lockfree_mpmc_bounded/defs.hpp @@ -0,0 +1,429 @@ +#ifndef LOCKFREE_MPMC_BOUNDED_DEFS +#define LOCKFREE_MPMC_BOUNDED_DEFS + +#include "../utils.hpp" +#include +#include +#include +#include +#include +#include +#include +#include + +namespace tsfqueue::impl{ + + template + class unit_value{ + static_assert(N == 8 || N == 16, + "unit_value only supports 8 or 16 byte entries; " + "choose value_type/index_type so sizeof({data,seq}) is 8 or 16"); + }; + template <> + class unit_value<8>{public: using type = __int64_t;}; + template <> + class unit_value<16>{public: using type = __int128_t;}; + // --------------------------------------------------- + + // For creating aligned custom types + template + class alignas(A) aligned_type : public T{}; + + // --------------------------------------------------- + + // Helper classes for Runtime and CompileTime Arrays + // Compile-time array initialization helper + template + class array_compileTime{ + std::array _data; + static_assert(N > 0 && ((N & (N-1)) == 0), "The size of compile-time array for queue must be positive and power of 2 !"); + + public: + explicit array_compileTime(size_t = N) noexcept : _data() {} // initialize elements as '0' + ~array_compileTime() noexcept = default; + T& operator[](size_t index) noexcept { return _data[index & (N - 1)]; } + const T& operator[](size_t index) const noexcept { return _data[index & (N - 1)]; } + [[nodiscard]] constexpr size_t size() const noexcept {return N;} + [[nodiscard]] constexpr size_t index_mask() const noexcept {return N-1;} + [[nodiscard]] constexpr size_t capacity() const noexcept {return N;} + }; + + // Runtime array initialization helper + template + class array_runtime{ + std::unique_ptr _datap alignas(2 * cache_line_size); + const size_t _n; + const size_t _index_mask; + static_assert(N == 0, "Runtime array is formed only when N == 0 !"); + public: + explicit array_runtime(size_t n = 0) noexcept : _datap(nullptr), _n(n), _index_mask(n-1){ + if (n > 0){ + assert(((n & (n-1)) == 0) && "Array size should be power of 2."); + _datap = std::unique_ptr{new T[n]}; + } + } + + ~array_runtime() noexcept = default; + + T& operator[](size_t _index) noexcept {return _datap[_index & _index_mask];} + const T& operator[](size_t _index) const noexcept {return _datap[_index & _index_mask];} + [[nodiscard]] size_t size() const noexcept {return _n;} + [[nodiscard]] size_t index_mask() const noexcept {return _index_mask;} + [[nodiscard]] size_t capacity() const noexcept {return _n;} + }; + // ------------------------------------------------------------------------------------------------- + + template + // Used alignas(...) so that members like write and read index (which are 128bit aligned) works well as whole class is aligned. + class alignas(2*cache_line_size) lockfree_mpmc_bounded{ + /* + For the implementation we start with a bounded array of size N of atomic<__int128_t> integers. + Here, the 128 bits of each cell is divided into three sections: + + ** [64bits(DATA), 64bits(sequence Number with the least significan bit for full/empty flag)] ** + + This bounded array, say 'queue' is our Main queue data structure, its a circular buffer of queue entries. + We have two atomic integers 'read_index' and 'write_index' + 'read_index' -> 1: read_index & (N-1) is the index from which we read. + 2: We read the value only when the seq. no. == read_index and cell not empty. + 'write_index' -> 1: write_index & (N-1) is the index where which we write. + 2: We write only when seq. no. == write_index and cell is empty. + + When is: + [Empty queue] -> When read_index == write_index. + [Full queue] -> When write_index == N + read_index. + + For this queue, to store any dataType of large size, we need CAS16, i.e Hardware Atomic Compare and Swap for 16 bytes (128 bit) integers. + This allow us to atomically add full 128 bits to the queue cell, we add sequence number, Flag and DATA at the same time. + -> We can store any size data by storing the 64bit pointer to the data. (Created a wrapper which make this queue maintain ownership at terminals using unique_ptr). + + PUSH: (see the code, all these points can occur at very different time, as queue is MPMC supporting) + - A: Got the current position at which write_index is pointing. If all goes good we might push here only, in this iteration. + - B: Got the seq part of the cell we saved in wr_index. [at this time it might not be empty, or emptied and again occupied, anything can occur] + - If seq confirms A and B occurred at same time: try to compare_exchange the cell with our new entry. On success, try to increment 'wr_index' (someone else might have already 'helped us'). + - Else if someone else pushed in the current cell (with or without a subsequent pop), or someone pushed but is still at position X: help by incrementing 'wr_index'. We might still find an empty cell after wr_index. + - Else if queue is full: return false. + + POP: (see the code, all these points can occur at very different time, as queue is MPMC supporting) + - A: chose a 'rd_index' where we try to read from in this iteration. + - B: read the data (whole cell) at that position after some time. + - If seq says no one popped between A and B: try to compare_exchange the cell with an empty entry. On success, return the data and try to increment 'rd_index'. + - Else if someone else popped and it got filled again, OR someone else popped (incremented read_index? - not sure, so we try to do so): help by incrementing 'rd_index' and retry from a higher index. + - Else if queue is empty: return false (no higher index will have data either). + + */ + + public: // Descriptive names for data and index types. + using value_type = dataT; + using index_type = indexT; + + static constexpr unsigned bits_in_index() noexcept { return sizeof(index_type) * CHAR_BIT; } + static constexpr unsigned bits_for_value(unsigned n) noexcept { + unsigned b{0}; + while (n != 0) { ++b; n >>= 1U; } + return b; + } + + static_assert(std::is_trivial_v, + "value_type must be trivial (queue stores entries via raw memory copy)"); + static_assert(std::is_unsigned_v, + "index_type must be unsigned (algorithm relies on defined wrap-around)"); + static_assert(sizeof(index_type) >= 4, + "index_type should be 4 bytes or wider; 1/2 byte indices are for experiments only"); + static_assert(sizeof(index_type) == 1 || sizeof(index_type) == 2 || + sizeof(index_type) == 4 || sizeof(index_type) == 8, + "index_type size must be one of: 1, 2, 4, 8"); + static_assert(N == 0 || (bits_in_index() > bits_for_value(N)), + "index_type must be wide enough to address N slots without wrap-around"); + + private: + + // For measuring size, so that we can align our main entry class with its size. + struct alignas(8) helper_entry{ + value_type _data; + index_type _index; + }; + + using entry_as_value = typename unit_value::type; + + constexpr static inline bool is_always_lock_free = std::atomic::is_always_lock_free; + + static constexpr bool USE_BUILTIN_16B{ + #if defined(__GNUC__) && defined(__clang__) + false // clang: std::atomic<__int128> is already hardware lock-free + #elif defined(__GNUC__) + true // gcc: std::atomic<__int128> falls back to libatomic (locked) -> use __sync_* builtin + #else + false // unknown compiler: trust std::atomic + #endif + }; + + // main 'entry' class (making class because by default members are private) + class alignas(sizeof(helper_entry)) entry{ + // Made this as Union, so that the same memory location stores both members and we can change any one based on need. + union entry_union{ + mutable entry_as_value _value; + struct entry_struct{ + value_type _data; + index_type _index; + } _x; + // Data, Sequence and Flag all '0' at the start. + entry_union(){_value = 0;} + } _u; + + public: + // We are clearing in each constructor because + entry() noexcept {clear();} + + explicit entry(index_type s) noexcept { + clear(); + _u._x._index = s; + } + + explicit entry(index_type s, value_type d) noexcept { + clear(); + _u._x._index = s; + _u._x._data = d; + } + + // Here no clearing because, it already changing the full entry. + explicit entry(entry_as_value ev) noexcept { + _u._value = ev; + } + + void clear(){_u._value = 0;} + + ~entry() noexcept = default; + + // For setting values later + void set_seq(index_type s){ + _u._x._index = s; + } + + void set(index_type s, value_type v){ + clear(); + _u._x._index = s; + _u._x._data = v; + } + + void set_full(entry_as_value ev){_u._value = ev;} + + index_type get_seq(){ + return _u._x._index; + } + + value_type get_data(){ + return _u._x._data; + } + + bool is_empty() const { return !(_u._x._index & 1U); } + bool is_full() const {return !is_empty();} + + // Needed because we might do, entry e = _array[index].load() -> this return a raw __int128_t or __int64_t... + // so need to overload '=' so that convert raw int to entry object. + entry & operator=(entry_as_value ev) noexcept { + _u._value = ev; + return *(this); + } + + [[using gnu: hot]] entry_as_value load() noexcept { + if constexpr (sizeof(entry_as_value)==16 && USE_BUILTIN_16B){ + return __sync_val_compare_and_swap(&this->_u._value, 0, 0); + }else{ + return reinterpret_cast*>(this)->load(); + } + } + + [[using gnu: hot]] entry_as_value load() const noexcept { + if constexpr (sizeof(entry_as_value)==16 && USE_BUILTIN_16B){ + return __sync_val_compare_and_swap(&this->_u._value, 0, 0); + }else{ + return reinterpret_cast*>(this)->load(); + } + } + + [[using gnu: hot]] bool compare_exchange(entry expected, entry new_to_put) noexcept { + if constexpr (sizeof(entry_as_value)==16 && USE_BUILTIN_16B){ + return __sync_bool_compare_and_swap(&this->_u._value, expected._u._value, new_to_put._u._value); + }else{ + return reinterpret_cast*>(this)->compare_exchange_strong(expected._u._value, new_to_put._u._value); + } + } + }; + + static_assert(sizeof(entry) == 2 || sizeof(entry) == 4 || sizeof(entry) == 8 || sizeof(entry) == 16, + "entry size not supported (must be 2, 4, 8 or 16 bytes for atomic CAS)"); + static_assert(sizeof(entry) == sizeof(helper_entry), + "entry and helper_entry must be of the same size"); + static_assert(sizeof(entry) == sizeof(entry_as_value), + "entry and entry_as_value must be of the same size"); + + // Type of _array, Based on condition whether size of array is given before or not, it assigns its type. + using array_t = typename std::conditional< + N == 0, array_runtime, 0>, + array_compileTime, N> + >::type; + + public: + lockfree_mpmc_bounded(uint64_t n = N): _write_index(0), _read_index(0), _array(n) // Here the constructor of _array got 'n' and Memory got allocated + { + if (N > 0){ + if (n != N){ + throw(std::invalid_argument{"The Compile time size given as a template argument should be same as constructor argument to mpmc_queue when deciding size at compile-time size (N > 0) !"}); + } + }else{ + if ((n & (n-1)) != 0){ + throw(std::invalid_argument{ + std::string{"The Runtime size provided to mpmc_queue::constructor should be power of 2 !"} + }); + }else if (bits_in_index() <= bits_for_value(n)){ + throw(std::invalid_argument{ + std::string{"The size given ["}+std::to_string(n)+std::string{"] is too large for index_type !"} + }); + } + } + // Initializing the _array with 0, 2, 4, ... -> (here LSB = 0) => Empty. + // and Index in incremental order so (seq of _array[i]) >> 1 == INDEX in array. + for (index_type i = 0;i < _array.size();i++){ + _array[i].set_seq(i << 1); + } + } + + // Best practice to remove all our data from the queue. + ~lockfree_mpmc_bounded(){ + value_type v; + while (pop(v)); + } + + // Removing Copy Constructor and Copy Assignment operator. + lockfree_mpmc_bounded(const lockfree_mpmc_bounded &) = delete; + lockfree_mpmc_bounded& operator=(const lockfree_mpmc_bounded &) = delete; + + // Removing Move Constructor and Move Assignment operator. + lockfree_mpmc_bounded(lockfree_mpmc_bounded &&) = delete; + lockfree_mpmc_bounded& operator=(lockfree_mpmc_bounded &&) = delete; + + [[using gnu: hot, flatten]] bool push(value_type) noexcept; + [[using gnu: hot, flatten]] bool enqueue(value_type) noexcept; + + [[using gnu: hot, flatten]] bool pop(value_type&) noexcept; + [[using gnu: hot, flatten]] bool dequeue(value_type&) noexcept; + + // If we want the index of the cell where we put it. + // (NOT QUEUE INDEX, BUT THE SEQUENCE NUMBER OF CELL BEFORE WE PUT THE VALUE) + [[using gnu: hot, flatten]] bool enqueue(value_type, index_type&) noexcept; + [[using gnu: hot, flatten]] bool push(value_type, index_type&) noexcept; + [[using gnu: hot, flatten]] bool dequeue(value_type&, index_type&) noexcept; + [[using gnu: hot, flatten]] bool pop(value_type&, index_type&) noexcept; + + // Takes a callable which returns a boolean and takes two argumets (value_type, index_type) + template + [[using gnu: hot, flatten]] bool pop_if(F&, value_type&) noexcept; + template + [[using gnu: hot, flatten]] bool pop_if(F&, value_type&, index_type&) noexcept; + + // Keep evicting the oldest value from the queue, until you are able to push successfully. + bool evict_until_push(value_type v){ + while (true){ + if (push(v)){return true;} + value_type being_lost; + pop(being_lost); + } + } + + bool evict_until_push(value_type v, index_type & i){ + while (true){ + if (push(v, i)){return true;} + value_type being_lost; + pop(being_lost); + } + } + + [[using gnu: hot, flatten]] bool exchange(index_type, value_type, value_type) noexcept; + + [[using gnu: hot, flatten]] bool empty() noexcept; + + // Using [[nodiscard]] because calling them and not using the return value is certainly a bug ! + [[using gnu: hot, flatten]] [[nodiscard]] bool empty() const noexcept; + + [[using gnu: hot, flatten]] [[nodiscard]] size_t size() const noexcept; + [[nodiscard]] size_t capacity() const noexcept; + + [[nodiscard]] constexpr size_t entry_size() const noexcept; + [[nodiscard]] static constexpr size_t size_n() noexcept; // 'const' does not makes sense in 'static' method, as relation with '*this'. + + private: + alignas(2 * cache_line_size) std::atomic _write_index; + alignas(2 * cache_line_size) std::atomic _read_index; + array_t _array; + }; + + template + class alignas(2*cache_line_size) lockfree_mpmc_bounded_unique_ptr{ + /* + This Class is a wrapper which uses 'lockfree_mpmc_bounded' queue and forms most of its function + but with a slight modification to make it work for 'unique_ptr'. + + PUSH - For pushing we take unique_ptr (ownership) as argument -> Retrive its underlying pointer + and RELEASE the unique_ptr [Now the ownership is GONE] -> Reinterpreat_Cast the pointer to 'uint64_t' + -> push it to the queue. + + POP - For popping we first pop the data from queue and get 'uint64_t' -> reinterpreat_cast it to Normal Pointer (T*) + -> Create a unique_ptr with that pointer [Ownership regained]. + */ + public: + using value_type = std::unique_ptr; + using underlying_type = lockfree_mpmc_bounded; + using index_type = typename underlying_type::index_type; + + static_assert(sizeof(T*) <= sizeof(uint64_t), "T* must fit in uint64_t"); + + lockfree_mpmc_bounded_unique_ptr() noexcept = default; + + lockfree_mpmc_bounded_unique_ptr(const lockfree_mpmc_bounded_unique_ptr&) = delete; + lockfree_mpmc_bounded_unique_ptr& operator=(const lockfree_mpmc_bounded_unique_ptr&) = delete; + lockfree_mpmc_bounded_unique_ptr(lockfree_mpmc_bounded_unique_ptr&&) = delete; + lockfree_mpmc_bounded_unique_ptr& operator=(lockfree_mpmc_bounded_unique_ptr&&) = delete; + + // Destructor: removes remaining entries so owned objects are freed. + // Not thread-safe: must not race with push/pop on other threads. + ~lockfree_mpmc_bounded_unique_ptr() noexcept { + value_type tmp; + while (pop(tmp)) {} + } + + [[using gnu: hot, flatten]] bool push(value_type &) noexcept; + [[using gnu: hot, flatten]] bool push(value_type &, index_type &) noexcept; + + [[using gnu: hot, flatten]] bool enqueue(value_type &) noexcept; + [[using gnu: hot, flatten]] bool enqueue(value_type &, index_type &) noexcept; + + [[using gnu: hot, flatten]] bool pop(value_type &) noexcept; + [[using gnu: hot, flatten]] bool pop(value_type &, index_type &) noexcept; + + [[using gnu: hot, flatten]] bool dequeue(value_type &) noexcept; + [[using gnu: hot, flatten]] bool dequeue(value_type &, index_type &) noexcept; + + // Keep evicting (and destroying) the oldest owned pointer until push succeeds. + bool evict_until_push(value_type &) noexcept; + bool evict_until_push(value_type &, index_type &) noexcept; + + [[using gnu: hot, flatten]] bool empty() noexcept; + [[using gnu: hot, flatten]] [[nodiscard]] bool empty() const noexcept; + + [[using gnu: hot, flatten]] [[nodiscard]] size_t size() const noexcept; + [[nodiscard]] size_t capacity() const noexcept; + + [[nodiscard]] constexpr size_t entry_size() const noexcept; + [[nodiscard]] static constexpr size_t size_n() noexcept; + + private: + underlying_type _q; + }; + +} + +#endif \ No newline at end of file diff --git a/include/lockfree_mpmc_bounded/impl.hpp b/include/lockfree_mpmc_bounded/impl.hpp new file mode 100644 index 0000000..867e136 --- /dev/null +++ b/include/lockfree_mpmc_bounded/impl.hpp @@ -0,0 +1,431 @@ +#ifndef LOCKFREE_MPMC_BOUNDED_IMPL +#define LOCKFREE_MPMC_BOUNDED_IMPL + +#include "defs.hpp" + + +namespace tsfqueue::impl{ + + template + [[using gnu: hot, flatten]] bool lockfree_mpmc_bounded::push(value_type data_to_push) noexcept { + while (true){ + // A: Got the current position at which write_index is pointing. + // I all goes good we might push here only, in this iteration. + index_type wr_index = _write_index.load(); + + // B: Got the seq part of the cell we saved in wr_index. + // [at this time it might not be empty, or emptied and again occupied anything can occur] + index_type seq = _array[wr_index].get_seq(); + + if (seq == static_cast((wr_index << 1))){ + // Confirms that A and B occurred at same time ! + + // Expected entry at cell wr_index + entry e{static_cast(wr_index << 1)}; + + // Entry to be placed at cell wr_index if everything goes right. + entry new_e{static_cast((wr_index << 1) | 1U), data_to_push}; + + // Try to push to wr_index if everything is as expected + if (_array[wr_index].compare_exchange(e, new_e)){ + + // ***[POSITION - X]*** + + // Till now its possible that after we added our data, someone else already 'helped us' and incremented 'wr_index' + // If NOT, we try to increment 'wr_index'. + _write_index.compare_exchange_strong(wr_index, wr_index + 1); + + return true; + } + }else if ((seq == static_cast((wr_index << 1) | 1U)) || + (seq == static_cast((wr_index + _array.size()) << 1))){ + // If at 'B', someone else pushed in the current cell, but no one else poped. + // OR, If at 'B', someone else pushed and someone else poped but not pushed after that. + // OR, someone pushed but is currently at position 'X', in that case we can help them by incrementing 'wr_index' + // In the first two cases also, we might have empty cell where we can push after wr_index. + _write_index.compare_exchange_strong(wr_index, wr_index + 1); + + }else if (static_cast(seq+(_array.size() << 1)) == static_cast((wr_index << 1) | 1U)){ + // Queue is full ! + return false; + } + } + } + + template + [[using gnu: hot, flatten]] bool lockfree_mpmc_bounded::enqueue(value_type data_to_enqueue) noexcept { + return push(data_to_enqueue); + } + + template + [[using gnu: hot, flatten]] bool lockfree_mpmc_bounded::pop(value_type & popped_data) noexcept { + + while (1){ + // A: choosed a 'rd_index' where we try to read from in this iteration. + index_type rd_index = _read_index.load(); + + // B: Here, we read the data at that position after some time. + entry e{_array[rd_index].load()}; + + if (e.get_seq() == static_cast((rd_index << 1) | 1U)){ + // No one poped between 'A' and 'B' + + // Creating an empty entry to place there + entry empty_entry{static_cast((rd_index + _array.size()) << 1)}; + + if (_array[rd_index].compare_exchange(e, empty_entry)){ + // Successfully popped ! + popped_data = e.get_data(); + + index_type tmp_index = rd_index; + _read_index.compare_exchange_strong(tmp_index, rd_index + 1); + + return true; + } + }else if (static_cast(e.get_seq() | 1U) == static_cast(((rd_index + _array.size())<<1) | 1U)){ + // someone else poped and it again got filled. OR someone else popoed (incremented read index ? -> Not sure, so we try to do so). + // Try to read from some higher index... + _read_index.compare_exchange_strong(rd_index, rd_index + 1); + }else if (e.get_seq() == static_cast(rd_index << 1)){ + // Queue is empty ! -> return false...sure that we will not find place to pop no higher index + return false; + } + } + } + + template + [[using gnu: hot, flatten]] bool lockfree_mpmc_bounded::dequeue(value_type & popped_data) noexcept { + return pop(popped_data); + } + + // Now Methods where what the exact index where we pushed or popped from. + + template + [[using gnu: hot, flatten]] bool lockfree_mpmc_bounded::push(value_type data_to_push, index_type & pushed_where) noexcept { + while (true){ + // A: Got the current position at which write_index is pointing. + // I all goes good we might push here only, in this iteration. + index_type wr_index = _write_index.load(); + + // B: Got the seq part of the cell we saved in wr_index. + // [at this time it might not be empty, or emptied and again occupied anything can occur] + index_type seq = _array[wr_index].get_seq(); + + if (seq == static_cast((wr_index << 1))){ + // Confirms that A and B occurred at same time ! + + // Expected entry at cell wr_index + entry e{static_cast(wr_index << 1)}; + + // Entry to be placed at cell wr_index if everything goes right. + entry new_e{static_cast((wr_index << 1) | 1U), data_to_push}; + + // Try to push to wr_index if everything is as expected + if (_array[wr_index].compare_exchange(e, new_e)){ + + pushed_where = wr_index; + + // ***[POSITION - X]*** + + // Till now its possible that after we added our data, someone else already 'helped us' and incremented 'wr_index' + // If NOT, we try to increment 'wr_index'. + _write_index.compare_exchange_strong(wr_index, wr_index + 1); + + return true; + } + }else if ((seq == static_cast((wr_index << 1) | 1U)) || + (seq == static_cast((wr_index + _array.size()) << 1))){ + // If at 'B', someone else pushed in the current cell, but no one else poped. + // OR, If at 'B', someone else pushed and someone else poped but not pushed after that. + // OR, someone pushed but is currently at position 'X', in that case we can help them by incrementing 'wr_index' + // In the first two cases also, we might have empty cell where we can push after wr_index. + + _write_index.compare_exchange_strong(wr_index, wr_index + 1); + }else if (static_cast(seq+(_array.size() << 1)) == static_cast((wr_index << 1) | 1U)){ + // Queue is full ! + return false; + } + } + } + + template + [[using gnu: hot, flatten]] bool lockfree_mpmc_bounded::enqueue(value_type data_to_enqueue, index_type & enqueued_where) noexcept { + return push(data_to_enqueue, enqueued_where); + } + + template + [[using gnu: hot, flatten]] bool lockfree_mpmc_bounded::pop(value_type & popped_data, index_type & popped_from) noexcept { + + while (1){ + // A: choosed a 'rd_index' where we try to read from in this iteration. + index_type rd_index = _read_index.load(); + + // B: Here, we read the data at that position after some time. + entry e{_array[rd_index].load()}; + + if (e.get_seq() == static_cast((rd_index << 1) | 1U)){ + // No one poped between 'A' and 'B' + + // Creating an empty entry to place there + entry empty_entry{static_cast((rd_index + _array.size()) << 1)}; + + + if (_array[rd_index].compare_exchange(e, empty_entry)){ + // Successfully popped ! + popped_data = e.get_data(); + popped_from = rd_index; + + index_type tmp_index = rd_index; + _read_index.compare_exchange_strong(tmp_index, rd_index + 1); + + return true; + } + }else if (static_cast(e.get_seq() | 1U) == static_cast(((rd_index + _array.size())<<1) | 1U)){ + // someone else poped and it again got filled. OR someone else popoed (incremented read index ? -> Not sure, so we try to do so). + // Try to read from some higher index... + _read_index.compare_exchange_strong(rd_index, rd_index + 1); + }else if (e.get_seq() == static_cast(rd_index << 1)){ + // Queue is empty ! -> return false...sure that we will not find place to pop no higher index + return false; + } + } + } + + template + [[using gnu: hot, flatten]] bool lockfree_mpmc_bounded::dequeue(value_type & popped_data, index_type & dequeued_from) noexcept { + return pop(popped_data, dequeued_from); + } + + template + template + [[using gnu: hot, flatten]] bool lockfree_mpmc_bounded::pop_if(F& f, value_type & popped_data) noexcept { + + while (1){ + // A: choosed a 'rd_index' where we try to read from in this iteration. + index_type rd_index = _read_index.load(); + + // B: Here, we read the data at that position after some time. + entry e{_array[rd_index].load()}; + + if (e.get_seq() == static_cast((rd_index << 1) | 1U)){ + // No one poped between 'A' and 'B' + + // Checking the condition + if (!f(e.get_data(), e.get_seq())){return false;} + + // Creating an empty entry to place there + entry empty_entry{static_cast((rd_index + _array.size()) << 1)}; + + if (_array[rd_index].compare_exchange(e, empty_entry)){ + // Successfully popped ! + popped_data = e.get_data(); + + index_type tmp_index = rd_index; + _read_index.compare_exchange_strong(tmp_index, rd_index + 1); + + return true; + } + }else if (static_cast(e.get_seq() | 1U) == static_cast(((rd_index + _array.size())<<1) | 1U)){ + // someone else poped and it again got filled. OR someone else popoed (incremented read index ? -> Not sure, so we try to do so). + // Try to read from some higher index... + _read_index.compare_exchange_strong(rd_index, rd_index + 1); + }else if (e.get_seq() == static_cast(rd_index << 1)){ + // Queue is empty ! -> return false...sure that we will not find place to pop no higher index + return false; + } + } + } + + + template + template + [[using gnu: hot, flatten]] bool lockfree_mpmc_bounded::pop_if(F& f, value_type & popped_data, index_type & popped_from) noexcept { + + while (1){ + // A: choosed a 'rd_index' where we try to read from in this iteration. + index_type rd_index = _read_index.load(); + + // B: Here, we read the data at that position after some time. + entry e{_array[rd_index].load()}; + + if (e.get_seq() == static_cast((rd_index << 1) | 1U)){ + // No one poped between 'A' and 'B' + + // Checking the condition + if (!f(e.get_data(), e.get_seq())){return false;} + + // Creating an empty entry to place there + entry empty_entry{static_cast((rd_index + _array.size()) << 1)}; + + if (_array[rd_index].compare_exchange(e, empty_entry)){ + // Successfully popped ! + popped_data = e.get_data(); + popped_from = rd_index; + + index_type tmp_index = rd_index; + _read_index.compare_exchange_strong(tmp_index, rd_index + 1); + + return true; + } + }else if (static_cast(e.get_seq() | 1U) == static_cast(((rd_index + _array.size())<<1) | 1U)){ + // someone else poped and it again got filled. OR someone else popoed (incremented read index ? -> Not sure, so we try to do so). + // Try to read from some higher index... + _read_index.compare_exchange_strong(rd_index, rd_index + 1); + }else if (e.get_seq() == static_cast(rd_index << 1)){ + // Queue is empty ! -> return false...sure that we will not find place to pop no higher index + return false; + } + } + } + + template + [[using gnu: hot, flatten]] bool lockfree_mpmc_bounded::exchange(index_type i, value_type expected_value, value_type new_value) noexcept { + entry e_old{static_cast((i << 1) | 1U), expected_value}; + entry e_new{static_cast((i << 1) | 1U), new_value}; + + return _array[i].compare_exchange(e_old, e_new); + } + + template + [[using gnu: hot, flatten]] bool lockfree_mpmc_bounded::empty() noexcept { + index_type rd_index = _read_index.load(); + entry e{_array[rd_index].load()}; + if (e.get_seq() == static_cast(rd_index << 1)) return true; + else return false; + } + + template + [[using gnu: hot, flatten]] bool lockfree_mpmc_bounded::empty() const noexcept { + index_type rd_index = _read_index.load(); + entry e{_array[rd_index].load()}; + if (e.get_seq() == static_cast(rd_index << 1)) return true; + else return false; + } + + // Improved this function's logic, i think this is better that the Erez Strauss's queue function. + template + [[using gnu: hot, flatten]] [[nodiscard]] size_t lockfree_mpmc_bounded::size() const noexcept { + index_type wr_index = _write_index.load(); + index_type rd_index = _read_index.load(); + if (wr_index >= rd_index) return (wr_index - rd_index); + return (_array.size() - ((rd_index - wr_index)&(_array.size()-1))); + } + + template + [[using gnu: hot, flatten]] [[nodiscard]] size_t lockfree_mpmc_bounded::capacity() const noexcept { + return _array.size(); + } + + template + [[nodiscard]] constexpr size_t lockfree_mpmc_bounded::entry_size() const noexcept { + return sizeof(entry); + } + + template + [[nodiscard]] constexpr size_t lockfree_mpmc_bounded::size_n() noexcept { + return N; + } + + // ------------------------------------------------------------------------ + // Unique pointer wrapper for lockfree_mpmc_bounded + // ------------------------------------------------------------------------ + + template + [[using gnu: hot, flatten]] bool lockfree_mpmc_bounded_unique_ptr::push(value_type & p) noexcept { + uint64_t raw = reinterpret_cast(p.get()); + if (_q.push(raw)) {(void)p.release(); return true;} + return false; + } + + template + [[using gnu: hot, flatten]] bool lockfree_mpmc_bounded_unique_ptr::push(value_type & p, index_type & i) noexcept { + uint64_t raw = reinterpret_cast(p.get()); + if (_q.push(raw, i)) {(void)p.release(); return true;} + return false; + } + + template + [[using gnu: hot, flatten]] bool lockfree_mpmc_bounded_unique_ptr::enqueue(value_type & p) noexcept { + return push(p); + } + + template + [[using gnu: hot, flatten]] bool lockfree_mpmc_bounded_unique_ptr::enqueue(value_type & p, index_type & i) noexcept { + return push(p, i); + } + + template + [[using gnu: hot, flatten]] bool lockfree_mpmc_bounded_unique_ptr::pop(value_type & out) noexcept { + uint64_t raw; + if (_q.pop(raw)) {out.reset(reinterpret_cast(raw)); return true;} + return false; + } + + template + [[using gnu: hot, flatten]] bool lockfree_mpmc_bounded_unique_ptr::pop(value_type & out, index_type & i) noexcept { + uint64_t raw; + if (_q.pop(raw, i)) {out.reset(reinterpret_cast(raw)); return true;} + return false; + } + + template + [[using gnu: hot, flatten]] bool lockfree_mpmc_bounded_unique_ptr::dequeue(value_type & out) noexcept { + return pop(out); + } + + template + [[using gnu: hot, flatten]] bool lockfree_mpmc_bounded_unique_ptr::dequeue(value_type & out, index_type & i) noexcept { + return pop(out, i); + } + + template + bool lockfree_mpmc_bounded_unique_ptr::evict_until_push(value_type & p) noexcept { + while (true){ + if (push(p)) return true; + value_type evicted; + pop(evicted); + } + } + + template + bool lockfree_mpmc_bounded_unique_ptr::evict_until_push(value_type & p, index_type & i) noexcept { + while (true){ + if (push(p, i)) return true; + value_type evicted; + pop(evicted); + } + } + + template + [[using gnu: hot, flatten]] bool lockfree_mpmc_bounded_unique_ptr::empty() noexcept { + return _q.empty(); + } + + template + [[using gnu: hot, flatten]] [[nodiscard]] bool lockfree_mpmc_bounded_unique_ptr::empty() const noexcept { + return _q.empty(); + } + + template + [[using gnu: hot, flatten]] [[nodiscard]] size_t lockfree_mpmc_bounded_unique_ptr::size() const noexcept { + return _q.size(); + } + + template + [[nodiscard]] size_t lockfree_mpmc_bounded_unique_ptr::capacity() const noexcept { + return _q.capacity(); + } + + template + [[nodiscard]] constexpr size_t lockfree_mpmc_bounded_unique_ptr::entry_size() const noexcept { + return _q.entry_size(); + } + + template + [[nodiscard]] constexpr size_t lockfree_mpmc_bounded_unique_ptr::size_n() noexcept { + return underlying_type::size_n(); + } + +} // namespace tsfqueue::impl + +#endif diff --git a/include/lockfree_mpmc_bounded/queue.hpp b/include/lockfree_mpmc_bounded/queue.hpp index e69de29..1b44573 100644 --- a/include/lockfree_mpmc_bounded/queue.hpp +++ b/include/lockfree_mpmc_bounded/queue.hpp @@ -0,0 +1,5 @@ +#ifndef LOCKFREE_MPMC_BOUNDED +#define LOCKFREE_MPMC_BOUNDED +#include "impl.hpp" + +#endif \ No newline at end of file diff --git a/tests/guide_to_run_tests/TEST_MPMC_BOUNDED_LOCKFREE.md b/tests/guide_to_run_tests/TEST_MPMC_BOUNDED_LOCKFREE.md new file mode 100644 index 0000000..5db56d2 --- /dev/null +++ b/tests/guide_to_run_tests/TEST_MPMC_BOUNDED_LOCKFREE.md @@ -0,0 +1,73 @@ +# Running `test_mpmc_bounded_lockfree` + +All commands run from the repo root. + +## 1. First-time setup + +```bash +rm -rf build +mkdir build && cd build +cmake .. +``` + +## 2. Normal build & run (everything) + +```bash +cmake --build . --target test_mpmc_bounded_lockfree +./test_mpmc_bounded_lockfree +``` + +## 3. Run a single test + +```bash +./test_mpmc_bounded_lockfree --gtest_filter=LockfreeMPMCBoundedQueue.BasicFIFO_SingleProducerSingleConsumer +./test_mpmc_bounded_lockfree --gtest_filter=*EdgeCases* +./test_mpmc_bounded_lockfree --gtest_list_tests +``` + +## 4. Run via ctest + +```bash +ctest -R MPMC_BOUNDED_LOCKFREE --output-on-failure +``` + +## 5. ThreadSanitizer build (data races) + +```bash +cd build && rm -rf * +cmake -DENABLE_TSAN=ON .. +cmake --build . --target test_mpmc_bounded_lockfree +./test_mpmc_bounded_lockfree +``` + +## 6. AddressSanitizer build (UAF, leaks, OOB) + +```bash +cd build && rm -rf * +cmake -DENABLE_ASAN=ON .. +cmake --build . --target test_mpmc_bounded_lockfree +./test_mpmc_bounded_lockfree +``` + +## 7. UndefinedBehaviorSanitizer build + +```bash +cd build && rm -rf * +cmake -DENABLE_UBSAN=ON .. +cmake --build . --target test_mpmc_bounded_lockfree +./test_mpmc_bounded_lockfree +``` + +## Test list + +| # | Test | Category | +|---|---|---| +| 1 | `LockfreeMPMCBoundedQueue.BasicFIFO_SingleProducerSingleConsumer` | Basic SPSC FIFO | +| 2 | `LockfreeMPMCBoundedQueue.EdgeCases_EmptyPop_FullPush_WrapAround` | Edge cases | +| 3 | `LockfreeMPMCBoundedQueue.MultiProducerMultiConsumer_NoLossNoDuplicates`| MPMC correctness | +| 4 | `LockfreeMPMCBoundedQueue.Linearizability_PerProducerMonotonicOrder` | Linearizability | +| 5 | `LockfreeMPMCBoundedUniquePtr.MemorySafety_DrainOnDestroy` | Memory safety | +| 6 | `LockfreeMPMCBoundedUniquePtr.MemorySafety_PushPopRoundtrip` | Memory safety | +| 7 | `LockfreeMPMCBoundedUniquePtr.MemorySafety_MPMC_NoLeak` | Memory safety (concurrent) | +| 8 | `LockfreeMPMCBoundedQueue.Lifecycle_RepeatedConstructDestruct` | Lifecycle | +| 9 | `LockfreeMPMCBoundedQueue.Lifecycle_NonCopyableNonMovable` | Lifecycle (type traits) | diff --git a/tests/test_mpmc_bounded_lockfree.cpp b/tests/test_mpmc_bounded_lockfree.cpp new file mode 100644 index 0000000..fc7107c --- /dev/null +++ b/tests/test_mpmc_bounded_lockfree.cpp @@ -0,0 +1,439 @@ +#include +#include +#include +#include +#include +#include +#include "tsfqueue.hpp" + + +template +using LockfreeMPMCBounded = tsfqueue::impl::lockfree_mpmc_bounded; + +template +using LockfreeMPMCBoundedUniquePtr = tsfqueue::impl::lockfree_mpmc_bounded_unique_ptr; + + +// --------------------------------------------------------------------------- +// 1. Basic FIFO behaviour with a single producer and single consumer. +// Pushes ascending values and verifies pops come out in the same order. +// --------------------------------------------------------------------------- +TEST(LockfreeMPMCBoundedQueue, BasicFIFO_SingleProducerSingleConsumer) { + constexpr size_t N = 16; + LockfreeMPMCBounded q; + + EXPECT_TRUE(q.empty()); + EXPECT_EQ(q.size(), 0u); + EXPECT_EQ(q.capacity(), N); + + for (uint64_t i = 1; i <= 8; ++i) { + EXPECT_TRUE(q.push(i)); + } + EXPECT_FALSE(q.empty()); + EXPECT_EQ(q.size(), 8u); + + for (uint64_t i = 1; i <= 8; ++i) { + uint64_t v = 0; + EXPECT_TRUE(q.pop(v)); + EXPECT_EQ(v, i); + } + + EXPECT_TRUE(q.empty()); + EXPECT_EQ(q.size(), 0u); +} + + +// --------------------------------------------------------------------------- +// 2. Edge cases: pop on empty, push on full, and fill-then-drain cycles +// to exercise the circular buffer's seq-number wrap. +// --------------------------------------------------------------------------- +TEST(LockfreeMPMCBoundedQueue, EdgeCases_EmptyPop_FullPush_WrapAround) { + constexpr size_t N = 4; + LockfreeMPMCBounded q; + + // Pop on empty queue should return false and not modify the out parameter. + uint64_t v = 0xDEADBEEF; + EXPECT_FALSE(q.pop(v)); + EXPECT_EQ(v, 0xDEADBEEFu); + + // Fill the queue to capacity. + for (uint64_t i = 0; i < N; ++i) { + EXPECT_TRUE(q.push(i)); + } + EXPECT_EQ(q.size(), N); + + // Push on full queue should return false. + EXPECT_FALSE(q.push(999)); + EXPECT_EQ(q.size(), N); + + // Drain and check ordering. + for (uint64_t i = 0; i < N; ++i) { + uint64_t out = 0; + EXPECT_TRUE(q.pop(out)); + EXPECT_EQ(out, i); + } + EXPECT_TRUE(q.empty()); + + // Repeat fill/drain several times to wrap the sequence numbers past N. + for (int cycle = 0; cycle < 100; ++cycle) { + for (uint64_t i = 0; i < N; ++i) { + EXPECT_TRUE(q.push(cycle * 1000 + i)); + } + EXPECT_FALSE(q.push(0)); + for (uint64_t i = 0; i < N; ++i) { + uint64_t out = 0; + EXPECT_TRUE(q.pop(out)); + EXPECT_EQ(out, cycle * 1000 + i); + } + EXPECT_FALSE(q.pop(v)); + } +} + + +// --------------------------------------------------------------------------- +// 3. MPMC correctness under contention: every pushed value is popped +// exactly once (no losses, no duplicates), verified via per-value counts. +// Under ThreadSanitizer the op count is reduced (TSAN is ~10x slower); +// TSAN still gets plenty of interleavings to find races. +// --------------------------------------------------------------------------- +TEST(LockfreeMPMCBoundedQueue, MultiProducerMultiConsumer_NoLossNoDuplicates) { + constexpr size_t N = 1024; + constexpr int num_producers = 4; + constexpr int num_consumers = 4; +#if defined(TSFQUEUE_TSAN_BUILD) + constexpr int ops_per_producer = 2000; +#else + constexpr int ops_per_producer = 5000; +#endif + constexpr int total_items = num_producers * ops_per_producer; + + LockfreeMPMCBounded q; + + std::vector> seen(total_items); + for (auto& c : seen) c.store(0); + + std::atomic popped_count{0}; + std::atomic producers_done{false}; + + + std::vector producers; + producers.reserve(num_producers); + for (int p = 0; p < num_producers; ++p) { + producers.emplace_back([&, p]() { + for (int j = 0; j < ops_per_producer; ++j) { + uint64_t val = static_cast(p * ops_per_producer + j); + while (!q.push(val)) { + std::this_thread::yield(); + } + } + }); + } + + std::vector consumers; + consumers.reserve(num_consumers); + for (int c = 0; c < num_consumers; ++c) { + consumers.emplace_back([&]() { + uint64_t val; + while (true) { + if (q.pop(val)) { + ASSERT_LT(val, static_cast(total_items)); + int prev = seen[val].fetch_add(1); + ASSERT_EQ(prev, 0) << "value " << val << " popped more than once"; + popped_count.fetch_add(1); + } else if (producers_done.load() && q.empty()) { + break; + } else { + std::this_thread::yield(); + } + } + }); + } + + for (auto& t : producers) t.join(); + producers_done.store(true); + for (auto& t : consumers) t.join(); + + EXPECT_EQ(popped_count.load(), total_items); + for (int i = 0; i < total_items; ++i) { + EXPECT_EQ(seen[i].load(), 1) << "value " << i << " was not popped exactly once"; + } + EXPECT_TRUE(q.empty()); +} + + +// --------------------------------------------------------------------------- +// 4. Linearizability sniff test: per-producer monotonic ordering. +// Each producer P pushes values tagged (P, seq=0,1,2,...). After the run, +// for every producer the pop-order subsequence of its values must be +// strictly increasing. This catches reorderings within a single producer +// that any valid linearization (FIFO per producer) forbids. +// --------------------------------------------------------------------------- +TEST(LockfreeMPMCBoundedQueue, Linearizability_PerProducerMonotonicOrder) { + constexpr size_t N = 256; + constexpr int num_producers = 4; + constexpr int num_consumers = 4; +#if defined(TSFQUEUE_TSAN_BUILD) + constexpr int ops_per_producer = 2000; +#else + constexpr int ops_per_producer = 5000; +#endif + + // Encode (producer_id, seq) into a 64-bit value: high 16 bits = producer, + // low 48 bits = sequence. + auto encode = [](uint64_t pid, uint64_t seq) { + return (pid << 48) | seq; + }; + auto decode_pid = [](uint64_t v) { return v >> 48; }; + auto decode_seq = [](uint64_t v) { return v & ((1ULL << 48) - 1); }; + + LockfreeMPMCBounded q; + std::atomic producers_done{false}; + + // Per-consumer pop log to avoid cross-consumer contention on a shared vector. + std::vector> consumer_logs(num_consumers); + + std::vector producers; + producers.reserve(num_producers); + for (int p = 0; p < num_producers; ++p) { + producers.emplace_back([&, p]() { + for (int j = 0; j < ops_per_producer; ++j) { + uint64_t v = encode(p, j); + while (!q.push(v)) std::this_thread::yield(); + } + }); + } + + std::vector consumers; + consumers.reserve(num_consumers); + for (int c = 0; c < num_consumers; ++c) { + consumers.emplace_back([&, c]() { + auto& log = consumer_logs[c]; + log.reserve(num_producers * ops_per_producer / num_consumers + 16); + uint64_t v; + while (true) { + if (q.pop(v)) { + log.push_back(v); + } else if (producers_done.load() && q.empty()) { + break; + } else { + std::this_thread::yield(); + } + } + }); + } + + for (auto& t : producers) t.join(); + producers_done.store(true); + for (auto& t : consumers) t.join(); + + // For each producer, gather the order in which its values were popped + // *within a single consumer*. Across consumers a producer's items can + // appear in any interleaving, but within one consumer they must be + // strictly increasing (since the consumer popped them in the order the + // queue handed them out, and the queue is FIFO per producer). + for (int c = 0; c < num_consumers; ++c) { + std::vector last_seq_by_producer(num_producers, -1); + for (uint64_t v : consumer_logs[c]) { + uint64_t pid = decode_pid(v); + uint64_t seq = decode_seq(v); + ASSERT_LT(pid, static_cast(num_producers)); + int64_t prev = last_seq_by_producer[pid]; + ASSERT_GT(static_cast(seq), prev) + << "consumer " << c << " popped producer " << pid + << " seq " << seq << " after seq " << prev + << " - FIFO ordering violated"; + last_seq_by_producer[pid] = static_cast(seq); + } + } +} + + +// --------------------------------------------------------------------------- +// 5. Memory safety: unique_ptr wrapper drain-on-destroy. +// Constructs N values, pushes them, destroys the queue without popping. +// All N destructors must run exactly once. Run this under ASAN to also +// catch UAF / double-free at the C heap level. +// --------------------------------------------------------------------------- +namespace { + std::atomic g_counted_alive{0}; + std::atomic g_counted_constructed{0}; + std::atomic g_counted_destroyed{0}; + + struct Counted { + int value; + explicit Counted(int v) : value(v) { + g_counted_constructed.fetch_add(1); + g_counted_alive.fetch_add(1); + } + ~Counted() { + g_counted_destroyed.fetch_add(1); + g_counted_alive.fetch_sub(1); + } + Counted(const Counted&) = delete; + Counted& operator=(const Counted&) = delete; + }; + + void reset_counters() { + g_counted_alive.store(0); + g_counted_constructed.store(0); + g_counted_destroyed.store(0); + } +} + +TEST(LockfreeMPMCBoundedUniquePtr, MemorySafety_DrainOnDestroy) { + constexpr size_t N = 16; + reset_counters(); + { + LockfreeMPMCBoundedUniquePtr q; + for (int i = 0; i < static_cast(N); ++i) { + std::unique_ptr p(new Counted(i)); + ASSERT_TRUE(q.push(p)); + EXPECT_EQ(p.get(), nullptr) << "push must release ownership"; + } + EXPECT_EQ(g_counted_alive.load(), static_cast(N)); + // Destructor runs here and must drain the remaining N items. + } + EXPECT_EQ(g_counted_alive.load(), 0) + << "destructor leaked owned objects"; + EXPECT_EQ(g_counted_constructed.load(), static_cast(N)); + EXPECT_EQ(g_counted_destroyed.load(), static_cast(N)); +} + + +// --------------------------------------------------------------------------- +// 6. Memory safety: unique_ptr wrapper roundtrip - construct, push, pop, +// let the popped unique_ptr go out of scope. Every constructed object +// must be destroyed exactly once. +// --------------------------------------------------------------------------- +TEST(LockfreeMPMCBoundedUniquePtr, MemorySafety_PushPopRoundtrip) { + constexpr size_t N = 32; + constexpr int kItems = N; + reset_counters(); + { + LockfreeMPMCBoundedUniquePtr q; + for (int i = 0; i < kItems; ++i) { + std::unique_ptr p(new Counted(i)); + ASSERT_TRUE(q.push(p)); + } + for (int i = 0; i < kItems; ++i) { + std::unique_ptr out; + ASSERT_TRUE(q.pop(out)); + ASSERT_NE(out.get(), nullptr); + EXPECT_EQ(out->value, i); + // out goes out of scope at next iteration -> destructor runs. + } + EXPECT_EQ(g_counted_alive.load(), 0); + } + EXPECT_EQ(g_counted_constructed.load(), kItems); + EXPECT_EQ(g_counted_destroyed.load(), kItems); +} + + +// --------------------------------------------------------------------------- +// 7. Memory safety under contention: unique_ptr wrapper with MPMC traffic. +// All produced objects must eventually be destroyed (either by consumer +// pop or by destructor drain). Best run under TSAN + (separately) ASAN. +// --------------------------------------------------------------------------- +TEST(LockfreeMPMCBoundedUniquePtr, MemorySafety_MPMC_NoLeak) { + constexpr size_t N = 256; + constexpr int num_producers = 4; + constexpr int num_consumers = 4; +#if defined(TSFQUEUE_TSAN_BUILD) + constexpr int ops_per_producer = 1000; +#else + constexpr int ops_per_producer = 3000; +#endif + constexpr int total_items = num_producers * ops_per_producer; + reset_counters(); + { + LockfreeMPMCBoundedUniquePtr q; + std::atomic producers_done{false}; + std::atomic popped{0}; + + std::vector producers; + for (int p = 0; p < num_producers; ++p) { + producers.emplace_back([&]() { + for (int j = 0; j < ops_per_producer; ++j) { + std::unique_ptr ptr(new Counted(j)); + while (!q.push(ptr)) std::this_thread::yield(); + } + }); + } + + std::vector consumers; + for (int c = 0; c < num_consumers; ++c) { + consumers.emplace_back([&]() { + std::unique_ptr out; + while (true) { + if (q.pop(out)) { + popped.fetch_add(1); + out.reset(); + } else if (producers_done.load() && q.empty()) { + break; + } else { + std::this_thread::yield(); + } + } + }); + } + + for (auto& t : producers) t.join(); + producers_done.store(true); + for (auto& t : consumers) t.join(); + + EXPECT_EQ(popped.load(), total_items); + // No outstanding items -> destructor drain has nothing to do. + } + EXPECT_EQ(g_counted_constructed.load(), total_items); + EXPECT_EQ(g_counted_destroyed.load(), total_items) + << "objects leaked: alive=" << g_counted_alive.load(); +} + + +// --------------------------------------------------------------------------- +// 8. Lifecycle: repeated construct/destruct cycles must not leak or corrupt +// state across instances. Run under ASAN to verify no heap mismanagement. +// --------------------------------------------------------------------------- +TEST(LockfreeMPMCBoundedQueue, Lifecycle_RepeatedConstructDestruct) { + constexpr size_t N = 64; +#if defined(TSFQUEUE_TSAN_BUILD) + constexpr int cycles = 200; +#else + constexpr int cycles = 2000; +#endif + + for (int cycle = 0; cycle < cycles; ++cycle) { + LockfreeMPMCBounded q; + EXPECT_TRUE(q.empty()); + for (uint64_t i = 0; i < N; ++i) { + ASSERT_TRUE(q.push(i + cycle)); + } + for (uint64_t i = 0; i < N; ++i) { + uint64_t v; + ASSERT_TRUE(q.pop(v)); + ASSERT_EQ(v, i + cycle); + } + EXPECT_TRUE(q.empty()); + } +} + + +// --------------------------------------------------------------------------- +// 9. Lifecycle: type traits / move/copy disabled at compile time. +// These are compile-time guarantees expressed as runtime EXPECTs so the +// test record makes the contract visible. +// --------------------------------------------------------------------------- +TEST(LockfreeMPMCBoundedQueue, Lifecycle_NonCopyableNonMovable) { + using Q = LockfreeMPMCBounded; + using UQ = LockfreeMPMCBoundedUniquePtr; + + // The base queue holds atomic members and must not be copied or moved. + EXPECT_FALSE(std::is_copy_constructible_v); + EXPECT_FALSE(std::is_copy_assignable_v); + + // The unique_ptr wrapper explicitly deletes copy/move. + EXPECT_FALSE(std::is_copy_constructible_v); + EXPECT_FALSE(std::is_copy_assignable_v); + EXPECT_FALSE(std::is_move_constructible_v); + EXPECT_FALSE(std::is_move_assignable_v); +} diff --git a/tests/test_mpmc_unbounded_blocking.cpp b/tests/test_mpmc_unbounded_blocking.cpp new file mode 100644 index 0000000..6fa29ff --- /dev/null +++ b/tests/test_mpmc_unbounded_blocking.cpp @@ -0,0 +1,216 @@ +#include +#include +#include +#include +#include +#include +#include "tsfqueue.hpp" +#include + + +// Basic sanity +TEST(MPMCQueue, BasicPushPop_Unbounded) { + tsfqueue::BlockingMPMCUnbounded q; + + EXPECT_TRUE(q.empty()); + + q.push(1); + q.push(2); + + int x; + EXPECT_TRUE(q.try_pop(x)); + EXPECT_EQ(x, 1); + + EXPECT_TRUE(q.try_pop(x)); + EXPECT_EQ(x, 2); + + EXPECT_TRUE(q.empty()); +} + +TEST(MPMCQueue, WaitFor_isTimeExact) { + tsfqueue::BlockingMPMCUnbounded q; + + EXPECT_TRUE(q.empty()); + + q.push(1); + q.push(2); + + int x; + EXPECT_TRUE(q.try_pop(x)); + EXPECT_EQ(x, 1); + + EXPECT_TRUE(q.try_pop(x)); + EXPECT_EQ(x, 2); + + EXPECT_TRUE(q.empty()); + + int wtime = 200; + int error = 10; + // auto func = [&](){ + // std::this_thread::sleep_for(std::chrono::seconds(10)); + // q.push(12); + // q.push(31); + // }; + // std::thread t(func); + auto start = std::chrono::high_resolution_clock::now(); + int res = q.wait_for_and_pop(x, std::chrono::milliseconds(wtime)); + auto end = std::chrono::high_resolution_clock::now(); + auto duration = std::chrono::duration_cast(end - start).count(); + EXPECT_NEAR((int)duration, (int)wtime, error); +} + +TEST(MPMCQueue, WaitForTester) { + tsfqueue::BlockingMPMCUnbounded q; + + EXPECT_TRUE(q.empty()); + + q.push(1); + q.push(2); + + int x; + EXPECT_TRUE(q.try_pop(x)); + EXPECT_EQ(x, 1); + + EXPECT_TRUE(q.try_pop(x)); + EXPECT_EQ(x, 2); + + EXPECT_TRUE(q.empty()); + + int wtime = 4000; + int pushTime = 2000; + int error = 50; + auto func = [&](){ + std::this_thread::sleep_for(std::chrono::milliseconds(pushTime)); + q.push(12); + }; + std::thread t(func); + auto start = std::chrono::high_resolution_clock::now(); + EXPECT_TRUE(q.wait_for_and_pop(x, std::chrono::milliseconds(wtime))); + auto end = std::chrono::high_resolution_clock::now(); + auto duration = std::chrono::duration_cast(end - start).count(); + EXPECT_NEAR((int)duration, (int)pushTime, error); + + EXPECT_EQ(x, 12); + + EXPECT_TRUE(q.empty()); + t.join(); +} + +TEST(MPMCQueue, DataRaceStressTest) { + tsfqueue::BlockingMPMCUnbounded q; + const int num_threads = 8; + const int ops_per_thread = 1000; + std::atomic total_popped{0}; + std::vector producers; + std::vector consumers; + + // Launch Producers + for (int i = 0; i < num_threads; ++i) { + producers.emplace_back([&q, ops_per_thread]() { + for (int j = 0; j < ops_per_thread; ++j) { + q.push(j); + } + }); + } + + // Launch Consumers + for (int i = 0; i < num_threads; ++i) { + consumers.emplace_back([&q, ops_per_thread, &total_popped]() { + for (int j = 0; j < ops_per_thread; ++j) { + int val; + // Using wait_for to ensure we don't block forever if a push is missed + if (q.wait_for_and_pop(val, std::chrono::milliseconds(100))) { + total_popped++; + } + } + }); + } + + for (auto& t : producers) t.join(); + for (auto& t : consumers) t.join(); + + // Verify all items were accounted for + EXPECT_EQ(total_popped.load(), num_threads * ops_per_thread); + EXPECT_TRUE(q.empty()); +} + +// Testing Emplace Back working and Static Assert working, i.e it does not create a copy of the object. +struct MockObject { + static int copies; + static int moves; + int x; + MockObject(int val) : x(val) {} + MockObject(const MockObject& other) { copies++; } + MockObject(MockObject&& other) noexcept { moves++; } + MockObject& operator=(const MockObject& other) { + x = other.x; + copies++; + return *this; + } + + MockObject& operator=(MockObject&& other) noexcept { + x = other.x; + moves++; + return *this; + } +}; +int MockObject::copies = 0; +int MockObject::moves = 0; + +TEST(MPMCQueue, EmplaceBackEfficiency_and_StaticAsserts) { + tsfqueue::BlockingMPMCUnbounded q; + MockObject::copies = 0; + MockObject::moves = 0; + // Emplace should construct the object directly in the internal storage + q.emplace_back(42); + + EXPECT_EQ(MockObject::copies, 0); + // Depending on implementation, moves should ideally be 0 or 1 + EXPECT_LE(MockObject::moves, 1); + + MockObject out(0); + + EXPECT_TRUE(q.try_pop(out)); + EXPECT_EQ(out.x, 42); +} + +// --------------------------------------------------------- +// Helper: A type that is NEITHER copyable NOR movable +// --------------------------------------------------------- +struct LockedType { + LockedType() = default; + LockedType(const LockedType&) = delete; + LockedType(LockedType&&) = delete; + LockedType& operator=(const LockedType&) = delete; + LockedType& operator=(LockedType&&) = delete; +}; + +// --------------------------------------------------------- +// Helper: A type that is ONLY movable +// --------------------------------------------------------- +struct MoveOnly { + MoveOnly() = default; + MoveOnly(const MoveOnly&) = delete; + MoveOnly(MoveOnly&&) = default; + MoveOnly& operator=(const MoveOnly&) = delete; + MoveOnly& operator=(MoveOnly&&) = default; +}; + +// --------------------------------------------------------- +// The Test Case +// --------------------------------------------------------- +TEST(MPMCQueue, StaticRequirementsValidation) { + // 1. Test Copy/Move Constructibility (Required for Push/Emplace) + bool is_pushable = std::is_copy_constructible_v || std::is_move_constructible_v; + EXPECT_TRUE(is_pushable) << "Queue must support copy or move for push operations."; + + // 2. Test Assignability (Required for try_pop(T& value)) + // This mirrors your failing static_assert: + // static_assert(std::is_copy_assignable_v || std::is_move_assignable_v) + + bool move_only_assignable = std::is_copy_assignable_v || std::is_move_assignable_v; + EXPECT_TRUE(move_only_assignable) << "Move-only types should be allowed if they are move-assignable."; + + bool locked_type_assignable = std::is_copy_assignable_v || std::is_move_assignable_v; + EXPECT_FALSE(locked_type_assignable) << "LockedType should fail the assignability requirement."; +} \ No newline at end of file