diff options
Diffstat (limited to 'library/cpp/threading/bounded_queue')
9 files changed, 287 insertions, 0 deletions
diff --git a/library/cpp/threading/bounded_queue/CMakeLists.darwin-x86_64.txt b/library/cpp/threading/bounded_queue/CMakeLists.darwin-x86_64.txt new file mode 100644 index 00000000000..bd9b68ab62b --- /dev/null +++ b/library/cpp/threading/bounded_queue/CMakeLists.darwin-x86_64.txt @@ -0,0 +1,14 @@ + +# This file was generated by the build system used internally in the Yandex monorepo. +# Only simple modifications are allowed (adding source-files to targets, adding simple properties +# like target_include_directories). These modifications will be ported to original +# ya.make files by maintainers. Any complex modifications which can't be ported back to the +# original buildsystem will not be accepted. + + + +add_library(cpp-threading-bounded_queue INTERFACE) +target_link_libraries(cpp-threading-bounded_queue INTERFACE + contrib-libs-cxxsupp + yutil +) diff --git a/library/cpp/threading/bounded_queue/CMakeLists.linux-aarch64.txt b/library/cpp/threading/bounded_queue/CMakeLists.linux-aarch64.txt new file mode 100644 index 00000000000..613a18db103 --- /dev/null +++ b/library/cpp/threading/bounded_queue/CMakeLists.linux-aarch64.txt @@ -0,0 +1,15 @@ + +# This file was generated by the build system used internally in the Yandex monorepo. +# Only simple modifications are allowed (adding source-files to targets, adding simple properties +# like target_include_directories). These modifications will be ported to original +# ya.make files by maintainers. Any complex modifications which can't be ported back to the +# original buildsystem will not be accepted. + + + +add_library(cpp-threading-bounded_queue INTERFACE) +target_link_libraries(cpp-threading-bounded_queue INTERFACE + contrib-libs-linux-headers + contrib-libs-cxxsupp + yutil +) diff --git a/library/cpp/threading/bounded_queue/CMakeLists.linux-x86_64.txt b/library/cpp/threading/bounded_queue/CMakeLists.linux-x86_64.txt new file mode 100644 index 00000000000..613a18db103 --- /dev/null +++ b/library/cpp/threading/bounded_queue/CMakeLists.linux-x86_64.txt @@ -0,0 +1,15 @@ + +# This file was generated by the build system used internally in the Yandex monorepo. +# Only simple modifications are allowed (adding source-files to targets, adding simple properties +# like target_include_directories). These modifications will be ported to original +# ya.make files by maintainers. Any complex modifications which can't be ported back to the +# original buildsystem will not be accepted. + + + +add_library(cpp-threading-bounded_queue INTERFACE) +target_link_libraries(cpp-threading-bounded_queue INTERFACE + contrib-libs-linux-headers + contrib-libs-cxxsupp + yutil +) diff --git a/library/cpp/threading/bounded_queue/CMakeLists.txt b/library/cpp/threading/bounded_queue/CMakeLists.txt new file mode 100644 index 00000000000..f8b31df0c11 --- /dev/null +++ b/library/cpp/threading/bounded_queue/CMakeLists.txt @@ -0,0 +1,17 @@ + +# This file was generated by the build system used internally in the Yandex monorepo. +# Only simple modifications are allowed (adding source-files to targets, adding simple properties +# like target_include_directories). These modifications will be ported to original +# ya.make files by maintainers. Any complex modifications which can't be ported back to the +# original buildsystem will not be accepted. + + +if (CMAKE_SYSTEM_NAME STREQUAL "Linux" AND CMAKE_SYSTEM_PROCESSOR STREQUAL "aarch64" AND NOT HAVE_CUDA) + include(CMakeLists.linux-aarch64.txt) +elseif (CMAKE_SYSTEM_NAME STREQUAL "Darwin" AND CMAKE_SYSTEM_PROCESSOR STREQUAL "x86_64") + include(CMakeLists.darwin-x86_64.txt) +elseif (WIN32 AND CMAKE_SYSTEM_PROCESSOR STREQUAL "AMD64" AND NOT HAVE_CUDA) + include(CMakeLists.windows-x86_64.txt) +elseif (CMAKE_SYSTEM_NAME STREQUAL "Linux" AND CMAKE_SYSTEM_PROCESSOR STREQUAL "x86_64" AND NOT HAVE_CUDA) + include(CMakeLists.linux-x86_64.txt) +endif() diff --git a/library/cpp/threading/bounded_queue/CMakeLists.windows-x86_64.txt b/library/cpp/threading/bounded_queue/CMakeLists.windows-x86_64.txt new file mode 100644 index 00000000000..bd9b68ab62b --- /dev/null +++ b/library/cpp/threading/bounded_queue/CMakeLists.windows-x86_64.txt @@ -0,0 +1,14 @@ + +# This file was generated by the build system used internally in the Yandex monorepo. +# Only simple modifications are allowed (adding source-files to targets, adding simple properties +# like target_include_directories). These modifications will be ported to original +# ya.make files by maintainers. Any complex modifications which can't be ported back to the +# original buildsystem will not be accepted. + + + +add_library(cpp-threading-bounded_queue INTERFACE) +target_link_libraries(cpp-threading-bounded_queue INTERFACE + contrib-libs-cxxsupp + yutil +) diff --git a/library/cpp/threading/bounded_queue/bounded_queue.h b/library/cpp/threading/bounded_queue/bounded_queue.h new file mode 100644 index 00000000000..c5c67140865 --- /dev/null +++ b/library/cpp/threading/bounded_queue/bounded_queue.h @@ -0,0 +1,89 @@ +#pragma once + +#include <util/generic/yexception.h> + +//https://www.1024cores.net/home/lock-free-algorithms/queues/bounded-mpmc-queue + +namespace NThreading { + template<typename T> + class TBoundedQueue { + public: + explicit TBoundedQueue(size_t size) + : Buffer_(new TCell[size]) + , Mask_(size - 1) + { + Y_ENSURE(size >= 2 && (size & (size - 1)) == 0); + + for (size_t i = 0; i < size; ++i) { + Buffer_[i].Sequence.store(i, std::memory_order_relaxed); + } + } + + template <typename T_> + [[nodiscard]] bool Enqueue(T_&& data) noexcept { + TCell* cell; + size_t pos = EnqueuePos_.load(std::memory_order_relaxed); + + for (;;) { + cell = &Buffer_[pos & Mask_]; + size_t seq = cell->Sequence.load(std::memory_order_acquire); + intptr_t dif = (intptr_t)seq - (intptr_t)pos; + + if (dif == 0) { + if (EnqueuePos_.compare_exchange_weak(pos, pos + 1, std::memory_order_relaxed)) { + break; + } + } else if (dif < 0) { + return false; + } else { + pos = EnqueuePos_.load(std::memory_order_relaxed); + } + } + + static_assert(noexcept(cell->Data = std::forward<T_>(data))); + cell->Data = std::forward<T_>(data); + cell->Sequence.store(pos + 1, std::memory_order_release); + + return true; + } + + [[nodiscard]] bool Dequeue(T& data) noexcept { + TCell* cell; + size_t pos = DequeuePos_.load(std::memory_order_relaxed); + + for (;;) { + cell = &Buffer_[pos & Mask_]; + size_t seq = cell->Sequence.load(std::memory_order_acquire); + intptr_t dif = (intptr_t)seq - (intptr_t)(pos + 1); + + if (dif == 0) { + if (DequeuePos_.compare_exchange_weak(pos, pos + 1, std::memory_order_relaxed)) { + break; + } + } else if (dif < 0) { + return false; + } else { + pos = DequeuePos_.load(std::memory_order_relaxed); + } + } + + static_assert(noexcept(data = std::move(cell->Data))); + data = std::move(cell->Data); + + cell->Sequence.store(pos + Mask_ + 1, std::memory_order_release); + return true; + } + private: + struct TCell { + std::atomic<size_t> Sequence; + T Data; + }; + + std::unique_ptr<TCell[]> Buffer_; + const size_t Mask_; + + alignas(64) std::atomic<size_t> EnqueuePos_ = 0; + alignas(64) std::atomic<size_t> DequeuePos_ = 0; + }; +} + diff --git a/library/cpp/threading/bounded_queue/bounded_queue_ut.cpp b/library/cpp/threading/bounded_queue/bounded_queue_ut.cpp new file mode 100644 index 00000000000..bb5b6eb787b --- /dev/null +++ b/library/cpp/threading/bounded_queue/bounded_queue_ut.cpp @@ -0,0 +1,106 @@ +#include "bounded_queue.h" + +#include <library/cpp/testing/unittest/registar.h> +#include <util/thread/factory.h> + +using namespace NThreading; + +Y_UNIT_TEST_SUITE(TBoundedQueueTest) { + Y_UNIT_TEST(QueueSize) { + const size_t queueSize = 16; + TBoundedQueue<size_t> boundedQueue(queueSize); + + for (size_t i = 0; i < queueSize; ++i) { + UNIT_ASSERT(boundedQueue.Enqueue(i)); + } + UNIT_ASSERT(!boundedQueue.Enqueue(0)); + size_t tmp = 0; + UNIT_ASSERT(boundedQueue.Dequeue(tmp)); + UNIT_ASSERT(boundedQueue.Enqueue(0)); + UNIT_ASSERT(!boundedQueue.Enqueue(0)); + } + + Y_UNIT_TEST(Move) { + const size_t queueSize = 16; + TBoundedQueue<TString> boundedQueue(queueSize); + + for (size_t i = 0; i < queueSize; ++i) { + TString v = "xxx"; + UNIT_ASSERT(boundedQueue.Enqueue(std::move(v))); + UNIT_ASSERT(v.empty()); + } + + { + TString v = "xxx"; + UNIT_ASSERT(!boundedQueue.Enqueue(std::move(v))); + UNIT_ASSERT(v == "xxx"); + } + + TString v; + UNIT_ASSERT(boundedQueue.Dequeue(v)); + UNIT_ASSERT(v == "xxx"); + } + + Y_UNIT_TEST(MPMC) { + size_t queueSize = 16; + size_t producers = 10; + size_t consumers = 10; + size_t itemsCount = 10000; + + TVector<THolder<IThreadFactory::IThread>> threads; + TBoundedQueue<std::pair<size_t, size_t>> boundedQueue(queueSize); + + std::atomic<size_t> itemCounter = 0; + std::atomic<size_t> consumed = 0; + + for (size_t i = 0; i < consumers; ++i) { + threads.push_back(SystemThreadFactory()->Run( + [&]() { + TVector<size_t> prevItems(producers); + for (;;) { + std::pair<size_t, size_t> item; + while (!boundedQueue.Dequeue(item)) { + ; + } + + if (item.first >= producers) { + break; + } + + UNIT_ASSERT(item.second > prevItems[item.first]); + prevItems[item.first] = item.second; + ++consumed; + } + }) + ); + } + + for (size_t i = 0; i < producers ; ++i) { + threads.push_back(SystemThreadFactory()->Run( + [&, producerNum = i]() { + for (;;) { + size_t item = ++itemCounter; + if (item > itemsCount) { + break; + } + + while (!boundedQueue.Enqueue(std::make_pair(producerNum, item))) { + ; + } + } + + while (!boundedQueue.Enqueue(std::make_pair(producers, size_t(0)))) { + ; + } + }) + ); + } + + + for (auto& t : threads) { + t->Join(); + } + + UNIT_ASSERT_VALUES_EQUAL(consumed.load(), itemsCount); + } +} diff --git a/library/cpp/threading/bounded_queue/ut/ya.make b/library/cpp/threading/bounded_queue/ut/ya.make new file mode 100644 index 00000000000..5db0a227131 --- /dev/null +++ b/library/cpp/threading/bounded_queue/ut/ya.make @@ -0,0 +1,8 @@ +UNITTEST_FOR(library/cpp/threading/bounded_queue) + +SRCS( + bounded_queue_ut.cpp +) + +END() + diff --git a/library/cpp/threading/bounded_queue/ya.make b/library/cpp/threading/bounded_queue/ya.make new file mode 100644 index 00000000000..4c5336f0311 --- /dev/null +++ b/library/cpp/threading/bounded_queue/ya.make @@ -0,0 +1,9 @@ +LIBRARY() + +SRCS( + bounded_queue.h +) + +END() + +RECURSE_FOR_TESTS(ut) |