blob: 2a5860f84cf505294f454953577a708b318b8b70 (
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
|
#pragma once
namespace NActors {
template<typename TEvent>
class TWatchdogTimer {
using TCallback = std::function<void()>;
const TDuration Timeout;
const TCallback Callback;
TMonotonic TriggerTimestamp = TMonotonic::Max();
bool EventScheduled = false;
ui32 Iteration;
static constexpr ui32 NumIterationsBeforeFiring = 2;
public:
TWatchdogTimer(TDuration timeout, TCallback callback)
: Timeout(timeout)
, Callback(std::move(callback))
{}
void Rearm(const TActorIdentity& actor) {
if (Timeout != TDuration::Zero() && Timeout != TDuration::Max()) {
TriggerTimestamp = TActivationContext::Monotonic() + Timeout;
Iteration = 0;
Schedule(actor);
}
}
void Disarm() {
TriggerTimestamp = TMonotonic::Max();
}
bool Armed() const {
return TriggerTimestamp != TMonotonic::Max();
}
void operator()(typename TEvent::TPtr& ev) {
Y_DEBUG_ABORT_UNLESS(EventScheduled);
EventScheduled = false;
if (!Armed()) {
// just do nothing
} else if (TActivationContext::Monotonic() < TriggerTimestamp) {
// the time hasn't come yet
Schedule(TActorIdentity(ev->Recipient));
} else if (Iteration < NumIterationsBeforeFiring) {
// time has come, but we will still give actor a chance to process some messages and rearm timer
++Iteration;
TActivationContext::Send(ev.Release()); // send this event into queue once more
EventScheduled = true;
} else {
// no chance to disarm, fire callback
Disarm();
Callback();
}
}
private:
void Schedule(const TActorIdentity& actor) {
Y_DEBUG_ABORT_UNLESS(Armed());
if (!EventScheduled) {
actor.Schedule(TriggerTimestamp, new TEvent);
EventScheduled = true;
}
}
};
}
|
