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
87
88
|
#pragma once
#include <util/datetime/base.h>
#include <util/generic/ptr.h>
#include <util/system/event.h>
#include <util/system/spinlock.h>
#include <util/thread/factory.h>
#include <functional>
#include <queue>
#include <utility>
namespace NPersQueue {
class TPQLibPrivate;
class TScheduler {
struct TCallbackHandlersCompare;
public:
using TCallback = std::function<void()>;
class TCallbackHandler: public TAtomicRefCount<TCallbackHandler> {
friend class TScheduler;
friend struct TCallbackHandlersCompare;
template <class TFunc>
TCallbackHandler(TInstant time, TFunc&& func, TPQLibPrivate* pqLib, const void* queueTag)
: Time(time)
, Callback(std::forward<TFunc>(func))
, PQLib(pqLib)
, QueueTag(queueTag)
{
}
void Execute();
public:
// Cancels execution of callback.
// If callback is already canceled or executes (executed), does nothing.
// Posteffect: callback is guaranteed to be destroyed after this call.
void TryCancel();
private:
TInstant Time;
TCallback Callback;
TAdaptiveLock Lock;
TPQLibPrivate* PQLib;
const void* QueueTag;
};
public:
// Starts a scheduler thread.
explicit TScheduler(TPQLibPrivate* pqLib);
// Stops a scheduler thread.
~TScheduler();
// Schedules a new callback to be executed
template <class TFunc>
TIntrusivePtr<TCallbackHandler> Schedule(TInstant time, const void* queueTag, TFunc&& func) {
TIntrusivePtr<TCallbackHandler> handler(new TCallbackHandler(time, std::forward<TFunc>(func), PQLib, queueTag));
AddToSchedule(handler);
return handler;
}
template <class TFunc>
TIntrusivePtr<TCallbackHandler> Schedule(TDuration delta, const void* queueTag, TFunc&& func) {
return Schedule(TInstant::Now() + delta, queueTag, std::forward<TFunc>(func));
}
private:
void AddToSchedule(TIntrusivePtr<TCallbackHandler> handler);
void ShutdownAndWait();
void SchedulerThread();
private:
struct TCallbackHandlersCompare {
bool operator()(const TIntrusivePtr<TCallbackHandler>& h1, const TIntrusivePtr<TCallbackHandler>& h2) const;
};
using TCallbackQueue = std::priority_queue<TIntrusivePtr<TCallbackHandler>, std::vector<TIntrusivePtr<TCallbackHandler>>, TCallbackHandlersCompare>;
TAdaptiveLock Lock;
TAutoEvent Event;
TCallbackQueue Callbacks;
TAtomic Shutdown;
THolder<IThreadFactory::IThread> Thread;
TPQLibPrivate* PQLib;
};
} // namespace NPersQueue
|
