diff options
| author | komels <[email protected]> | 2022-04-15 16:53:39 +0300 |
|---|---|---|
| committer | komels <[email protected]> | 2022-04-15 16:53:39 +0300 |
| commit | 703a2fb6e100d202d1c7fcd052d73bd5affef408 (patch) | |
| tree | 22b7320c06bb04d86dbf7b9af9ae44281331cd15 /kikimr/persqueue/sdk/deprecated/cpp/v2/impl/scheduler.cpp | |
| parent | 3375bbfda1e2afb03aa2072bf5f2f2c3a26026e8 (diff) | |
Move 'kikimr/yndx'-depending tests out of ydb/core
ref:0a380e13308d579e0545a76924330d1ca5129c43
Diffstat (limited to 'kikimr/persqueue/sdk/deprecated/cpp/v2/impl/scheduler.cpp')
| -rw-r--r-- | kikimr/persqueue/sdk/deprecated/cpp/v2/impl/scheduler.cpp | 96 |
1 files changed, 0 insertions, 96 deletions
diff --git a/kikimr/persqueue/sdk/deprecated/cpp/v2/impl/scheduler.cpp b/kikimr/persqueue/sdk/deprecated/cpp/v2/impl/scheduler.cpp deleted file mode 100644 index ba6c4ebb405..00000000000 --- a/kikimr/persqueue/sdk/deprecated/cpp/v2/impl/scheduler.cpp +++ /dev/null @@ -1,96 +0,0 @@ -#include "persqueue_p.h" -#include "scheduler.h" - -#include <util/system/thread.h> - -namespace NPersQueue { -void TScheduler::TCallbackHandler::Execute() { - auto guard = Guard(Lock); - if (Callback) { - Y_VERIFY(PQLib->GetQueuePool().GetQueue(QueueTag).AddFunc(std::move(Callback))); - Callback = nullptr; - } -} - -void TScheduler::TCallbackHandler::TryCancel() { - auto guard = Guard(Lock); - if (Callback) { - Callback = nullptr; - } -} - -bool TScheduler::TCallbackHandlersCompare::operator()(const TIntrusivePtr<TScheduler::TCallbackHandler>& h1, const TIntrusivePtr<TScheduler::TCallbackHandler>& h2) const { - return h1->Time > h2->Time; -} - -TScheduler::TScheduler(TPQLibPrivate* pqLib) - : Shutdown(false) - , PQLib(pqLib) -{ - Thread = SystemThreadFactory()->Run([this] { - this->SchedulerThread(); - }); -} - -TScheduler::~TScheduler() { - ShutdownAndWait(); -} - -void TScheduler::AddToSchedule(TIntrusivePtr<TCallbackHandler> handler) { - { - auto guard = Guard(Lock); - Callbacks.push(std::move(handler)); - } - Event.Signal(); -} - -void TScheduler::ShutdownAndWait() { - AtomicSet(Shutdown, true); - Event.Signal(); - Thread->Join(); -} - -void TScheduler::SchedulerThread() { - TThread::SetCurrentThreadName("pqlib_scheduler"); - while (!AtomicGet(Shutdown)) { - TInstant deadline = TInstant::Max(); - std::vector<TIntrusivePtr<TCallbackHandler>> callbacks; - { - // define next deadline and get expired callbacks - auto guard = Guard(Lock); - const TInstant now = TInstant::Now(); - while (!Callbacks.empty()) { - const auto& top = Callbacks.top(); - if (top->Time <= now) { - callbacks.push_back(top); - Callbacks.pop(); - } else { - deadline = top->Time; - break; - } - } - } - - // execute callbacks - bool shutdown = false; - for (auto& callback : callbacks) { - if (shutdown) { - callback->TryCancel(); - } else { - callback->Execute(); - } - shutdown = shutdown || AtomicGet(Shutdown); - } - - if (!shutdown) { - Event.WaitD(deadline); - } - } - - // cancel all callbacks and clear data - while (!Callbacks.empty()) { - Callbacks.top()->TryCancel(); - Callbacks.pop(); - } -} -} // namespace NPersQueue |
