blob: 166946a2aa6cfd2f197f99560f30b21159e0293d (
plain) (
blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
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;
};
using TQueueType = TMultiMap<TInstant, TTaskHolder>;
using TQueueIterator = TQueueType::iterator;
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;
TQueueType Queue_;
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;
}
};
|
