aboutsummaryrefslogtreecommitdiffstats
path: root/library/cpp/threading/task_scheduler
diff options
context:
space:
mode:
authorkulikov <kulikov@yandex-team.ru>2022-02-10 16:49:34 +0300
committerDaniil Cherednik <dcherednik@yandex-team.ru>2022-02-10 16:49:34 +0300
commitc707901605d7b7c6cba0998cd52e1ae619c97762 (patch)
tree5d5cb817648f650d76cf1076100726fd9b8448e8 /library/cpp/threading/task_scheduler
parent65e5266709e7ff94b14ae128309e229de714b0df (diff)
downloadydb-c707901605d7b7c6cba0998cd52e1ae619c97762.tar.gz
Restoring authorship annotation for <kulikov@yandex-team.ru>. Commit 2 of 2.
Diffstat (limited to 'library/cpp/threading/task_scheduler')
-rw-r--r--library/cpp/threading/task_scheduler/task_scheduler.cpp458
-rw-r--r--library/cpp/threading/task_scheduler/task_scheduler.h162
-rw-r--r--library/cpp/threading/task_scheduler/task_scheduler_ut.cpp30
3 files changed, 325 insertions, 325 deletions
diff --git a/library/cpp/threading/task_scheduler/task_scheduler.cpp b/library/cpp/threading/task_scheduler/task_scheduler.cpp
index 407761eea7..174dde4bf7 100644
--- a/library/cpp/threading/task_scheduler/task_scheduler.cpp
+++ b/library/cpp/threading/task_scheduler/task_scheduler.cpp
@@ -1,246 +1,246 @@
#include "task_scheduler.h"
-
-#include <util/system/thread.h>
-#include <util/string/cast.h>
+
+#include <util/system/thread.h>
+#include <util/string/cast.h>
#include <util/stream/output.h>
-
-TTaskScheduler::ITask::~ITask() {}
-TTaskScheduler::IRepeatedTask::~IRepeatedTask() {}
-
-
-
-class TTaskScheduler::TWorkerThread
+
+TTaskScheduler::ITask::~ITask() {}
+TTaskScheduler::IRepeatedTask::~IRepeatedTask() {}
+
+
+
+class TTaskScheduler::TWorkerThread
: public ISimpleThread
-{
-public:
- TWorkerThread(TTaskScheduler& state)
- : Scheduler_(state)
- {
- }
-
+{
+public:
+ TWorkerThread(TTaskScheduler& state)
+ : Scheduler_(state)
+ {
+ }
+
TString DebugState = "?";
TString DebugId = "";
-private:
+private:
void* ThreadProc() noexcept override {
- Scheduler_.WorkerFunc(this);
- return nullptr;
- }
-private:
- TTaskScheduler& Scheduler_;
-};
-
-
-
-TTaskScheduler::TTaskScheduler(size_t threadCount, size_t maxTaskCount)
- : MaxTaskCount_(maxTaskCount)
-{
- for (size_t i = 0; i < threadCount; ++i) {
- Workers_.push_back(new TWorkerThread(*this));
- Workers_.back()->DebugId = ToString(i);
- }
-}
-
-TTaskScheduler::~TTaskScheduler() {
- try {
- Stop();
- } catch (...) {
+ Scheduler_.WorkerFunc(this);
+ return nullptr;
+ }
+private:
+ TTaskScheduler& Scheduler_;
+};
+
+
+
+TTaskScheduler::TTaskScheduler(size_t threadCount, size_t maxTaskCount)
+ : MaxTaskCount_(maxTaskCount)
+{
+ for (size_t i = 0; i < threadCount; ++i) {
+ Workers_.push_back(new TWorkerThread(*this));
+ Workers_.back()->DebugId = ToString(i);
+ }
+}
+
+TTaskScheduler::~TTaskScheduler() {
+ try {
+ Stop();
+ } catch (...) {
Cdbg << "task scheduled destruction error: " << CurrentExceptionMessage();
- }
-}
-
-void TTaskScheduler::Start() {
- for (auto& w : Workers_) {
- w->Start();
- }
-}
-
-void TTaskScheduler::Stop() {
- with_lock (Lock_) {
- IsStopped_ = true;
- CondVar_.BroadCast();
- }
-
- for (auto& w: Workers_) {
- w->Join();
- }
-
- Workers_.clear();
- Queue_.clear();
-}
-
-size_t TTaskScheduler::GetTaskCount() const {
- return static_cast<size_t>(AtomicGet(TaskCounter_));
-}
-
-namespace {
- class TTaskWrapper
- : public TTaskScheduler::ITask
- , TNonCopyable
- {
- public:
- TTaskWrapper(TTaskScheduler::ITaskRef task, TAtomic& counter)
- : Task_(task)
- , Counter_(counter)
- {
- AtomicIncrement(Counter_);
- }
-
+ }
+}
+
+void TTaskScheduler::Start() {
+ for (auto& w : Workers_) {
+ w->Start();
+ }
+}
+
+void TTaskScheduler::Stop() {
+ with_lock (Lock_) {
+ IsStopped_ = true;
+ CondVar_.BroadCast();
+ }
+
+ for (auto& w: Workers_) {
+ w->Join();
+ }
+
+ Workers_.clear();
+ Queue_.clear();
+}
+
+size_t TTaskScheduler::GetTaskCount() const {
+ return static_cast<size_t>(AtomicGet(TaskCounter_));
+}
+
+namespace {
+ class TTaskWrapper
+ : public TTaskScheduler::ITask
+ , TNonCopyable
+ {
+ public:
+ TTaskWrapper(TTaskScheduler::ITaskRef task, TAtomic& counter)
+ : Task_(task)
+ , Counter_(counter)
+ {
+ AtomicIncrement(Counter_);
+ }
+
~TTaskWrapper() override {
- AtomicDecrement(Counter_);
- }
- private:
- TInstant Process() override {
- return Task_->Process();
- }
- private:
- TTaskScheduler::ITaskRef Task_;
- TAtomic& Counter_;
- };
-}
-
-bool TTaskScheduler::Add(ITaskRef task, TInstant expire) {
- with_lock (Lock_) {
- if (!IsStopped_ && Workers_.size() > 0 && GetTaskCount() + 1 <= MaxTaskCount_) {
- ITaskRef newTask = new TTaskWrapper(task, TaskCounter_);
+ AtomicDecrement(Counter_);
+ }
+ private:
+ TInstant Process() override {
+ return Task_->Process();
+ }
+ private:
+ TTaskScheduler::ITaskRef Task_;
+ TAtomic& Counter_;
+ };
+}
+
+bool TTaskScheduler::Add(ITaskRef task, TInstant expire) {
+ with_lock (Lock_) {
+ if (!IsStopped_ && Workers_.size() > 0 && GetTaskCount() + 1 <= MaxTaskCount_) {
+ ITaskRef newTask = new TTaskWrapper(task, TaskCounter_);
Queue_.insert(std::make_pair(expire, TTaskHolder(newTask)));
-
- if (!Queue_.begin()->second.WaitingWorker) {
- CondVar_.Signal();
- }
- return true;
- }
- }
-
- return false;
-}
-
-namespace {
- class TRepeatedTask
- : public TTaskScheduler::ITask
- {
- public:
- TRepeatedTask(TTaskScheduler::IRepeatedTaskRef task, TDuration period, TInstant deadline)
- : Task_(task)
- , Period_(period)
- , Deadline_(deadline)
- {
- }
- private:
- TInstant Process() final {
- Deadline_ += Period_;
- if (Task_->Process()) {
- return Deadline_;
- } else {
- return TInstant::Max();
- }
- }
- private:
- TTaskScheduler::IRepeatedTaskRef Task_;
- TDuration Period_;
- TInstant Deadline_;
- };
-}
-
-bool TTaskScheduler::Add(IRepeatedTaskRef task, TDuration period) {
- const TInstant deadline = Now() + period;
- ITaskRef t = new TRepeatedTask(task, period, deadline);
- return Add(t, deadline);
-}
-
-
-const bool debugOutput = false;
-
+
+ if (!Queue_.begin()->second.WaitingWorker) {
+ CondVar_.Signal();
+ }
+ return true;
+ }
+ }
+
+ return false;
+}
+
+namespace {
+ class TRepeatedTask
+ : public TTaskScheduler::ITask
+ {
+ public:
+ TRepeatedTask(TTaskScheduler::IRepeatedTaskRef task, TDuration period, TInstant deadline)
+ : Task_(task)
+ , Period_(period)
+ , Deadline_(deadline)
+ {
+ }
+ private:
+ TInstant Process() final {
+ Deadline_ += Period_;
+ if (Task_->Process()) {
+ return Deadline_;
+ } else {
+ return TInstant::Max();
+ }
+ }
+ private:
+ TTaskScheduler::IRepeatedTaskRef Task_;
+ TDuration Period_;
+ TInstant Deadline_;
+ };
+}
+
+bool TTaskScheduler::Add(IRepeatedTaskRef task, TDuration period) {
+ const TInstant deadline = Now() + period;
+ ITaskRef t = new TRepeatedTask(task, period, deadline);
+ return Add(t, deadline);
+}
+
+
+const bool debugOutput = false;
+
void TTaskScheduler::ChangeDebugState(TWorkerThread* thread, const TString& state) {
- if (!debugOutput) {
+ if (!debugOutput) {
Y_UNUSED(thread);
Y_UNUSED(state);
- return;
- }
-
- thread->DebugState = state;
-
- TStringStream ss;
- ss << Now() << " " << thread->DebugId << ":\t";
- for (auto& w : Workers_) {
- ss << w->DebugState << " ";
- }
- ss << " [" << Queue_.size() << "] [" << TaskCounter_ << "]" << Endl;
+ return;
+ }
+
+ thread->DebugState = state;
+
+ TStringStream ss;
+ ss << Now() << " " << thread->DebugId << ":\t";
+ for (auto& w : Workers_) {
+ ss << w->DebugState << " ";
+ }
+ ss << " [" << Queue_.size() << "] [" << TaskCounter_ << "]" << Endl;
Cerr << ss.Str();
-}
-
-bool TTaskScheduler::Wait(TWorkerThread* thread, TQueueIterator& toWait) {
- ChangeDebugState(thread, "w");
- toWait->second.WaitingWorker = thread;
- return !CondVar_.WaitD(Lock_, toWait->first);
-}
-
-void TTaskScheduler::ChooseFromQueue(TQueueIterator& toWait) {
- for (TQueueIterator it = Queue_.begin(); it != Queue_.end(); ++it) {
- if (!it->second.WaitingWorker) {
- if (toWait == Queue_.end()) {
- toWait = it;
- } else if (it->first < toWait->first) {
- toWait->second.WaitingWorker = nullptr;
- toWait = it;
- }
- break;
- }
- }
-}
-
-void TTaskScheduler::WorkerFunc(TWorkerThread* thread) {
+}
+
+bool TTaskScheduler::Wait(TWorkerThread* thread, TQueueIterator& toWait) {
+ ChangeDebugState(thread, "w");
+ toWait->second.WaitingWorker = thread;
+ return !CondVar_.WaitD(Lock_, toWait->first);
+}
+
+void TTaskScheduler::ChooseFromQueue(TQueueIterator& toWait) {
+ for (TQueueIterator it = Queue_.begin(); it != Queue_.end(); ++it) {
+ if (!it->second.WaitingWorker) {
+ if (toWait == Queue_.end()) {
+ toWait = it;
+ } else if (it->first < toWait->first) {
+ toWait->second.WaitingWorker = nullptr;
+ toWait = it;
+ }
+ break;
+ }
+ }
+}
+
+void TTaskScheduler::WorkerFunc(TWorkerThread* thread) {
TThread::SetCurrentThreadName("TaskSchedWorker");
- TQueueIterator toWait = Queue_.end();
- ITaskRef toDo;
-
- for (;;) {
- TInstant repeat = TInstant::Max();
-
- if (!!toDo) {
- try {
- repeat = toDo->Process();
- } catch (...) {
+ TQueueIterator toWait = Queue_.end();
+ ITaskRef toDo;
+
+ for (;;) {
+ TInstant repeat = TInstant::Max();
+
+ if (!!toDo) {
+ try {
+ repeat = toDo->Process();
+ } catch (...) {
Cdbg << "task scheduler error: " << CurrentExceptionMessage();
- }
- }
-
-
- with_lock (Lock_) {
- ChangeDebugState(thread, "f");
-
- if (IsStopped_) {
- ChangeDebugState(thread, "s");
- return ;
- }
-
- if (!!toDo) {
- if (repeat < TInstant::Max()) {
+ }
+ }
+
+
+ with_lock (Lock_) {
+ ChangeDebugState(thread, "f");
+
+ if (IsStopped_) {
+ ChangeDebugState(thread, "s");
+ return ;
+ }
+
+ if (!!toDo) {
+ if (repeat < TInstant::Max()) {
Queue_.insert(std::make_pair(repeat, TTaskHolder(toDo)));
- }
- }
-
- toDo = nullptr;
-
- ChooseFromQueue(toWait);
-
- if (toWait != Queue_.end()) {
- if (toWait->first <= Now() || Wait(thread, toWait)) {
-
- toDo = toWait->second.Task;
- Queue_.erase(toWait);
- toWait = Queue_.end();
-
- if (!Queue_.empty() && !Queue_.begin()->second.WaitingWorker && Workers_.size() > 1) {
- CondVar_.Signal();
- }
-
- ChangeDebugState(thread, "p");
- }
- } else {
- ChangeDebugState(thread, "e");
- CondVar_.WaitI(Lock_);
- }
- }
- }
-}
+ }
+ }
+
+ toDo = nullptr;
+
+ ChooseFromQueue(toWait);
+
+ if (toWait != Queue_.end()) {
+ if (toWait->first <= Now() || Wait(thread, toWait)) {
+
+ toDo = toWait->second.Task;
+ Queue_.erase(toWait);
+ toWait = Queue_.end();
+
+ if (!Queue_.empty() && !Queue_.begin()->second.WaitingWorker && Workers_.size() > 1) {
+ CondVar_.Signal();
+ }
+
+ ChangeDebugState(thread, "p");
+ }
+ } else {
+ ChangeDebugState(thread, "e");
+ CondVar_.WaitI(Lock_);
+ }
+ }
+ }
+}
diff --git a/library/cpp/threading/task_scheduler/task_scheduler.h b/library/cpp/threading/task_scheduler/task_scheduler.h
index 166946a2aa..df4da941a8 100644
--- a/library/cpp/threading/task_scheduler/task_scheduler.h
+++ b/library/cpp/threading/task_scheduler/task_scheduler.h
@@ -1,86 +1,86 @@
-#pragma once
-
-#include <util/generic/vector.h>
-#include <util/generic/ptr.h>
-#include <util/generic/map.h>
-
-#include <util/datetime/base.h>
-
-#include <util/system/condvar.h>
-#include <util/system/mutex.h>
-
-class TTaskScheduler {
-public:
- class ITask;
- using ITaskRef = TIntrusivePtr<ITask>;
-
- class IRepeatedTask;
- using IRepeatedTaskRef = TIntrusivePtr<IRepeatedTask>;
-public:
- explicit TTaskScheduler(size_t threadCount = 1, size_t maxTaskCount = Max<size_t>());
- ~TTaskScheduler();
-
- void Start();
- void Stop();
-
- bool Add(ITaskRef task, TInstant expire);
- bool Add(IRepeatedTaskRef task, TDuration period);
-
- size_t GetTaskCount() const;
-private:
- class TWorkerThread;
-
- struct TTaskHolder {
- explicit TTaskHolder(ITaskRef& task)
- : Task(task)
- {
- }
- public:
- ITaskRef Task;
- TWorkerThread* WaitingWorker = nullptr;
- };
-
+#pragma once
+
+#include <util/generic/vector.h>
+#include <util/generic/ptr.h>
+#include <util/generic/map.h>
+
+#include <util/datetime/base.h>
+
+#include <util/system/condvar.h>
+#include <util/system/mutex.h>
+
+class TTaskScheduler {
+public:
+ class ITask;
+ using ITaskRef = TIntrusivePtr<ITask>;
+
+ class IRepeatedTask;
+ using IRepeatedTaskRef = TIntrusivePtr<IRepeatedTask>;
+public:
+ explicit TTaskScheduler(size_t threadCount = 1, size_t maxTaskCount = Max<size_t>());
+ ~TTaskScheduler();
+
+ void Start();
+ void Stop();
+
+ bool Add(ITaskRef task, TInstant expire);
+ bool Add(IRepeatedTaskRef task, TDuration period);
+
+ size_t GetTaskCount() const;
+private:
+ class TWorkerThread;
+
+ struct TTaskHolder {
+ explicit TTaskHolder(ITaskRef& task)
+ : Task(task)
+ {
+ }
+ public:
+ ITaskRef Task;
+ TWorkerThread* WaitingWorker = nullptr;
+ };
+
using TQueueType = TMultiMap<TInstant, TTaskHolder>;
using TQueueIterator = TQueueType::iterator;
-private:
+private:
void ChangeDebugState(TWorkerThread* thread, const TString& state);
- void ChooseFromQueue(TQueueIterator& toWait);
- bool Wait(TWorkerThread* thread, TQueueIterator& toWait);
-
- void WorkerFunc(TWorkerThread* thread);
-private:
- bool IsStopped_ = false;
-
- TAtomic TaskCounter_ = 0;
+ void ChooseFromQueue(TQueueIterator& toWait);
+ bool Wait(TWorkerThread* thread, TQueueIterator& toWait);
+
+ void WorkerFunc(TWorkerThread* thread);
+private:
+ bool IsStopped_ = false;
+
+ TAtomic TaskCounter_ = 0;
TQueueType Queue_;
-
- TCondVar CondVar_;
- TMutex Lock_;
-
+
+ TCondVar CondVar_;
+ TMutex Lock_;
+
TVector<TAutoPtr<TWorkerThread>> Workers_;
-
- const size_t MaxTaskCount_;
-};
-
-class TTaskScheduler::ITask
- : public TAtomicRefCount<ITask>
-{
-public:
- virtual ~ITask();
-
- virtual TInstant Process() {//returns time to repeat this task
- return TInstant::Max();
- }
-};
-
-class TTaskScheduler::IRepeatedTask
- : public TAtomicRefCount<IRepeatedTask>
-{
-public:
- virtual ~IRepeatedTask();
-
- virtual bool Process() {//returns if to repeat task again
- return false;
- }
-};
-
+
+ const size_t MaxTaskCount_;
+};
+
+class TTaskScheduler::ITask
+ : public TAtomicRefCount<ITask>
+{
+public:
+ virtual ~ITask();
+
+ virtual TInstant Process() {//returns time to repeat this task
+ return TInstant::Max();
+ }
+};
+
+class TTaskScheduler::IRepeatedTask
+ : public TAtomicRefCount<IRepeatedTask>
+{
+public:
+ virtual ~IRepeatedTask();
+
+ virtual bool Process() {//returns if to repeat task again
+ return false;
+ }
+};
+
diff --git a/library/cpp/threading/task_scheduler/task_scheduler_ut.cpp b/library/cpp/threading/task_scheduler/task_scheduler_ut.cpp
index cb2daa718b..3b5203194a 100644
--- a/library/cpp/threading/task_scheduler/task_scheduler_ut.cpp
+++ b/library/cpp/threading/task_scheduler/task_scheduler_ut.cpp
@@ -2,7 +2,7 @@
#include <library/cpp/testing/unittest/registar.h>
#include <util/stream/output.h>
-#include <util/system/atomic.h>
+#include <util/system/atomic.h>
#include <util/generic/vector.h>
#include "task_scheduler.h"
@@ -12,13 +12,13 @@ class TTaskSchedulerTest: public TTestBase {
UNIT_TEST(Test);
UNIT_TEST_SUITE_END();
- class TCheckTask: public TTaskScheduler::IRepeatedTask {
+ class TCheckTask: public TTaskScheduler::IRepeatedTask {
public:
TCheckTask(const TDuration& delay)
: Start_(Now())
, Delay_(delay)
{
- AtomicIncrement(ScheduledTaskCounter_);
+ AtomicIncrement(ScheduledTaskCounter_);
}
~TCheckTask() override {
@@ -28,28 +28,28 @@ class TTaskSchedulerTest: public TTestBase {
const TDuration delay = Now() - Start_;
if (delay < Delay_) {
- AtomicIncrement(BadTimeoutCounter_);
+ AtomicIncrement(BadTimeoutCounter_);
}
- AtomicIncrement(ExecutedTaskCounter_);
+ AtomicIncrement(ExecutedTaskCounter_);
return false;
}
static bool AllTaskExecuted() {
- return AtomicGet(ScheduledTaskCounter_) == AtomicGet(ExecutedTaskCounter_);
+ return AtomicGet(ScheduledTaskCounter_) == AtomicGet(ExecutedTaskCounter_);
}
static size_t BadTimeoutCount() {
- return AtomicGet(BadTimeoutCounter_);
+ return AtomicGet(BadTimeoutCounter_);
}
private:
TInstant Start_;
TDuration Delay_;
- static TAtomic BadTimeoutCounter_;
- static TAtomic ScheduledTaskCounter_;
- static TAtomic ExecutedTaskCounter_;
+ static TAtomic BadTimeoutCounter_;
+ static TAtomic ScheduledTaskCounter_;
+ static TAtomic ExecutedTaskCounter_;
};
public:
@@ -72,15 +72,15 @@ class TTaskSchedulerTest: public TTestBase {
void ScheduleCheckTask(size_t delay) {
TDuration d = TDuration::MicroSeconds(delay);
- Scheduler_.Add(new TCheckTask(d), d);
+ Scheduler_.Add(new TCheckTask(d), d);
}
private:
- TTaskScheduler Scheduler_;
+ TTaskScheduler Scheduler_;
};
-TAtomic TTaskSchedulerTest::TCheckTask::BadTimeoutCounter_ = 0;
-TAtomic TTaskSchedulerTest::TCheckTask::ScheduledTaskCounter_ = 0;
-TAtomic TTaskSchedulerTest::TCheckTask::ExecutedTaskCounter_ = 0;
+TAtomic TTaskSchedulerTest::TCheckTask::BadTimeoutCounter_ = 0;
+TAtomic TTaskSchedulerTest::TCheckTask::ScheduledTaskCounter_ = 0;
+TAtomic TTaskSchedulerTest::TCheckTask::ExecutedTaskCounter_ = 0;
UNIT_TEST_SUITE_REGISTRATION(TTaskSchedulerTest);
generated by cgit v1.2.3 (git 2.25.1) at 2025-03-22 22:44:59 +0000