From 5706cb392271ea40eab053314e7c0f4d9d4547ba Mon Sep 17 00:00:00 2001 From: kulikov Date: Fri, 21 Jul 2023 13:59:33 +0300 Subject: try to get rid of locks and allocations for elastic queue thread pool In case of heavy load and high rps current thread pool implementation seems to have problems at least with contention on lock inside condvar (long futex wait calls from http server listener thread), so try to implement something more efficient: - replace condvar with TEventCounter implementation without internal lock (pthread condvar maintains waiters wakeup order, thread pool doesn't need it); - introduce well-known bounded mpmc queue over ring buffer; - get rid of TDecrementingWrapper; - add options to turn on new pool in library/cpp/http/server and search/daemons (will remove after adoption); - make elastic queue ut check both versions; - workaround problems with android/arm build targets. --- .../threading/equeue/CMakeLists.darwin-x86_64.txt | 2 + .../threading/equeue/CMakeLists.linux-aarch64.txt | 1 + .../threading/equeue/CMakeLists.linux-x86_64.txt | 2 + .../threading/equeue/CMakeLists.windows-x86_64.txt | 2 + library/cpp/threading/equeue/equeue_ut.cpp | 79 +++++++--- library/cpp/threading/equeue/fast.h | 167 +++++++++++++++++++++ library/cpp/threading/equeue/ya.make | 9 ++ 7 files changed, 242 insertions(+), 20 deletions(-) create mode 100644 library/cpp/threading/equeue/fast.h (limited to 'library/cpp/threading/equeue') diff --git a/library/cpp/threading/equeue/CMakeLists.darwin-x86_64.txt b/library/cpp/threading/equeue/CMakeLists.darwin-x86_64.txt index 902b6d7a9ac..43ff0adc1c0 100644 --- a/library/cpp/threading/equeue/CMakeLists.darwin-x86_64.txt +++ b/library/cpp/threading/equeue/CMakeLists.darwin-x86_64.txt @@ -12,6 +12,8 @@ target_link_libraries(cpp-threading-equeue PUBLIC contrib-libs-cxxsupp yutil cpp-deprecated-atomic + cpp-threading-bounded_queue + cpp-yt-threading ) target_sources(cpp-threading-equeue PRIVATE ${CMAKE_SOURCE_DIR}/library/cpp/threading/equeue/equeue.cpp diff --git a/library/cpp/threading/equeue/CMakeLists.linux-aarch64.txt b/library/cpp/threading/equeue/CMakeLists.linux-aarch64.txt index 5653d3d3f3f..805b2379439 100644 --- a/library/cpp/threading/equeue/CMakeLists.linux-aarch64.txt +++ b/library/cpp/threading/equeue/CMakeLists.linux-aarch64.txt @@ -13,6 +13,7 @@ target_link_libraries(cpp-threading-equeue PUBLIC contrib-libs-cxxsupp yutil cpp-deprecated-atomic + cpp-threading-bounded_queue ) target_sources(cpp-threading-equeue PRIVATE ${CMAKE_SOURCE_DIR}/library/cpp/threading/equeue/equeue.cpp diff --git a/library/cpp/threading/equeue/CMakeLists.linux-x86_64.txt b/library/cpp/threading/equeue/CMakeLists.linux-x86_64.txt index 5653d3d3f3f..36f8a2632e3 100644 --- a/library/cpp/threading/equeue/CMakeLists.linux-x86_64.txt +++ b/library/cpp/threading/equeue/CMakeLists.linux-x86_64.txt @@ -13,6 +13,8 @@ target_link_libraries(cpp-threading-equeue PUBLIC contrib-libs-cxxsupp yutil cpp-deprecated-atomic + cpp-threading-bounded_queue + cpp-yt-threading ) target_sources(cpp-threading-equeue PRIVATE ${CMAKE_SOURCE_DIR}/library/cpp/threading/equeue/equeue.cpp diff --git a/library/cpp/threading/equeue/CMakeLists.windows-x86_64.txt b/library/cpp/threading/equeue/CMakeLists.windows-x86_64.txt index 902b6d7a9ac..43ff0adc1c0 100644 --- a/library/cpp/threading/equeue/CMakeLists.windows-x86_64.txt +++ b/library/cpp/threading/equeue/CMakeLists.windows-x86_64.txt @@ -12,6 +12,8 @@ target_link_libraries(cpp-threading-equeue PUBLIC contrib-libs-cxxsupp yutil cpp-deprecated-atomic + cpp-threading-bounded_queue + cpp-yt-threading ) target_sources(cpp-threading-equeue PRIVATE ${CMAKE_SOURCE_DIR}/library/cpp/threading/equeue/equeue.cpp diff --git a/library/cpp/threading/equeue/equeue_ut.cpp b/library/cpp/threading/equeue/equeue_ut.cpp index 8557f63ac06..2c7d2c7b1e8 100644 --- a/library/cpp/threading/equeue/equeue_ut.cpp +++ b/library/cpp/threading/equeue/equeue_ut.cpp @@ -1,4 +1,5 @@ #include "equeue.h" +#include "fast.h" #include @@ -9,18 +10,33 @@ Y_UNIT_TEST_SUITE(TElasticQueueTest) { const size_t MaxQueueSize = 20; const size_t ThreadCount = 10; - const size_t N = 100000; - static THolder Queue; + template + THolder MakeQueue(); - struct TQueueSetup { - TQueueSetup() { - Queue.Reset(new TElasticQueue(MakeHolder())); - Queue->Start(ThreadCount, MaxQueueSize); - } - ~TQueueSetup() { - Queue->Stop(); - } + template <> + THolder MakeQueue() { + return MakeHolder(MakeHolder()); + } + + template <> + THolder MakeQueue() { + return MakeHolder(); + } + + template + struct TEnv { + static inline THolder Queue; + + struct TQueueSetup { + TQueueSetup() { + Queue.Reset(MakeQueue()); + Queue->Start(ThreadCount, MaxQueueSize); + } + ~TQueueSetup() { + Queue->Stop(); + } + }; }; struct TCounters { @@ -37,7 +53,9 @@ Y_UNIT_TEST_SUITE(TElasticQueueTest) { //fill test -- fill queue with "endless" jobs TSystemEvent WaitEvent; - Y_UNIT_TEST(FillTest) { + + template + void FillTest() { Counters.Reset(); struct TWaitJob: public IObjectInQueue { @@ -47,7 +65,10 @@ Y_UNIT_TEST_SUITE(TElasticQueueTest) { } } job; - struct TLocalSetup: TQueueSetup { + struct TLocalSetup: TEnv::TQueueSetup { + TLocalSetup() { + WaitEvent.Reset(); + } ~TLocalSetup() { WaitEvent.Signal(); } @@ -56,19 +77,26 @@ Y_UNIT_TEST_SUITE(TElasticQueueTest) { size_t enqueued = 0; { TLocalSetup setup; - while (Queue->Add(&job) && enqueued < MaxQueueSize + 100) { + while (TEnv::Queue->Add(&job) && enqueued < MaxQueueSize + 100) { ++enqueued; } UNIT_ASSERT_VALUES_EQUAL(enqueued, MaxQueueSize); - UNIT_ASSERT_VALUES_EQUAL(enqueued, Queue->ObjectCount()); + UNIT_ASSERT_VALUES_EQUAL(enqueued, TEnv::Queue->ObjectCount()); } - UNIT_ASSERT_VALUES_EQUAL(0u, Queue->ObjectCount()); - UNIT_ASSERT_VALUES_EQUAL(0u, Queue->Size()); + UNIT_ASSERT_VALUES_EQUAL(0u, TEnv::Queue->ObjectCount()); + UNIT_ASSERT_VALUES_EQUAL(0u, TEnv::Queue->Size()); UNIT_ASSERT_VALUES_EQUAL((size_t)Counters.Processed, enqueued); } + Y_UNIT_TEST(FillTest) { + FillTest(); + } + + Y_UNIT_TEST(FillTestFast) { + FillTest(); + } //concurrent test -- send many jobs from different threads struct TJob: public IObjectInQueue { @@ -78,9 +106,10 @@ Y_UNIT_TEST_SUITE(TElasticQueueTest) { }; static TJob Job; + template static bool TryAdd() { AtomicIncrement(Counters.Total); - if (Queue->Add(&Job)) { + if (TEnv::Queue->Add(&Job)) { AtomicIncrement(Counters.Scheduled); return true; } else { @@ -89,16 +118,18 @@ Y_UNIT_TEST_SUITE(TElasticQueueTest) { } } + const size_t N = 100000; static size_t TryCounter; - Y_UNIT_TEST(ConcurrentTest) { + template + void ConcurrentTest() { Counters.Reset(); TryCounter = 0; struct TSender: public IThreadFactory::IThreadAble { void DoExecute() override { while ((size_t)AtomicIncrement(TryCounter) <= N) { - if (!TryAdd()) { + if (!TryAdd()) { Sleep(TDuration::MicroSeconds(50)); } } @@ -106,7 +137,7 @@ Y_UNIT_TEST_SUITE(TElasticQueueTest) { } sender; { - TQueueSetup setup; + typename TEnv::TQueueSetup setup; TVector< TAutoPtr > senders; for (size_t i = 0; i < ThreadCount; ++i) { @@ -122,4 +153,12 @@ Y_UNIT_TEST_SUITE(TElasticQueueTest) { UNIT_ASSERT_VALUES_EQUAL(Counters.Processed, Counters.Scheduled); UNIT_ASSERT_VALUES_EQUAL(Counters.Total, Counters.Scheduled + Counters.Discarded); } + + Y_UNIT_TEST(ConcurrentTest) { + ConcurrentTest(); + } + + Y_UNIT_TEST(ConcurrentTestFast) { + ConcurrentTest(); + } } diff --git a/library/cpp/threading/equeue/fast.h b/library/cpp/threading/equeue/fast.h new file mode 100644 index 00000000000..3f96e279fc9 --- /dev/null +++ b/library/cpp/threading/equeue/fast.h @@ -0,0 +1,167 @@ +#pragma once + +#include + +#include +#include +#include +#include +#include +#include + +#include + +#if defined(_android_) || defined(_arm_) +//by now library/cpp/yt/threading doesn't compile in targets like default-android-armv7a, fallback to ordinal elastic queue +#include "equeue.h" + class TFastElasticQueue + : public TElasticQueue + { + public: + explicit TFastElasticQueue(const TParams& params = {}) + : TElasticQueue(MakeHolder(params)) + { + } + }; +#else + +#include + +class TFastElasticQueue + : public TThreadPoolBase + , private IThreadFactory::IThreadAble +{ +public: + explicit TFastElasticQueue(const TParams& params = {}) + : TThreadPoolBase(params) + { + Y_ENSURE(!params.Blocking_); + } + + ~TFastElasticQueue() { + Stop(); + } + + void Start(size_t threadCount, size_t maxQueueSize) override { + Y_ENSURE(Threads_.empty()); + Y_ENSURE(maxQueueSize > 0); + + Queue_.Reset(new NThreading::TBoundedQueue(FastClp2(maxQueueSize + threadCount))); //threadCount is for stop events + MaxQueueSize_ = maxQueueSize; + + try { + for (size_t i = 0; i < threadCount; ++i) { + Threads_.push_back(Pool()->Run(this)); + } + } catch (...) { + Stop(); + throw; + } + + Stopped_ = false; + } + + size_t ObjectCount() const { + //GuardCount_ can be temporary incremented above real object count in queue + return Min(GuardCount_.load(), MaxQueueSize_); + } + + bool Add(IObjectInQueue* obj) override Y_WARN_UNUSED_RESULT { + if (Stopped_ || !obj) { + return false; + } + + if (GuardCount_.fetch_add(1) >= MaxQueueSize_) { + GuardCount_.fetch_sub(1); + return false; + } + + QueueSize_.fetch_add(1); + + if (!Queue_->Enqueue(obj)) { + //Simultaneous Dequeue calls can return not in exact fifo order of items, + //so there can be GuardCount_ < MaxQueueSize_ but Enqueue will fail because of + //the oldest enqueued item is not actually dequeued and ring buffer can't proceed. + GuardCount_.fetch_sub(1); + QueueSize_.fetch_sub(1); + return false; + } + + + Event_.NotifyOne(); + + return true; + } + + size_t Size() const noexcept override { + return QueueSize_.load(); + } + + void Stop() noexcept override { + Stopped_ = true; + + for (size_t i = 0; i < Threads_.size(); ++i) { + while (!Queue_->Enqueue(nullptr)) { + Sleep(TDuration::MilliSeconds(1)); + } + + Event_.NotifyOne(); + } + + while (!Threads_.empty()) { + Threads_.back()->Join(); + Threads_.pop_back(); + } + + Queue_.Reset(); + } + + void DoExecute() override { + TThread::SetCurrentThreadName(Params.ThreadName_.c_str()); + + while (true) { + IObjectInQueue* job = nullptr; + + Event_.Await([&]() { + return Queue_->Dequeue(job); + }); + + if (!job) { + break; + } + + QueueSize_.fetch_sub(1); + + Y_DEFER { + GuardCount_.fetch_sub(1); + }; + + if (Params.Catching_) { + try { + try { + job->Process(nullptr); + } catch (...) { + Cdbg << "[mtp queue] " << CurrentExceptionMessage() << Endl; + } + } catch (...) { + ; + } + } else { + job->Process(nullptr); + } + } + } +private: + std::atomic Stopped_ = false; + size_t MaxQueueSize_ = 0; + + alignas(64) std::atomic GuardCount_ = 0; + alignas(64) std::atomic QueueSize_ = 0; + + TVector> Threads_; + + THolder> Queue_; + NYT::NThreading::TEventCount Event_; +}; + +#endif diff --git a/library/cpp/threading/equeue/ya.make b/library/cpp/threading/equeue/ya.make index 445797aa121..95677366c93 100644 --- a/library/cpp/threading/equeue/ya.make +++ b/library/cpp/threading/equeue/ya.make @@ -3,14 +3,23 @@ LIBRARY() SRCS( equeue.h equeue.cpp + fast.h ) PEERDIR( library/cpp/deprecated/atomic + library/cpp/threading/bounded_queue ) +IF (NOT OS_ANDROID AND NOT ARCH_ARM) + PEERDIR( + library/cpp/yt/threading + ) +ENDIF() + END() RECURSE_FOR_TESTS( ut ) + -- cgit v1.3