blob: c190105a592e2baf6220e308bb98e8cbe234f03d (
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
|
#pragma once
namespace NActors {
template <typename TEvent>
class TWatchdogTimer {
using TCallback = std::function<void()>;
const TDuration Timeout;
const TCallback Callback;
TInstant LastResetTimestamp;
TEvent* ExpectedEvent = nullptr;
ui32 Iteration = 0;
static constexpr ui32 NumIterationsBeforeFiring = 2;
public:
TWatchdogTimer(TDuration timeout, TCallback callback)
: Timeout(timeout)
, Callback(std::move(callback))
{
}
void Arm(const TActorIdentity& actor) {
if (Timeout != TDuration::Zero() && Timeout != TDuration::Max()) {
Schedule(Timeout, actor);
Reset();
}
}
void Reset() {
LastResetTimestamp = TActivationContext::Now();
}
void Disarm() {
ExpectedEvent = nullptr;
}
void operator()(typename TEvent::TPtr& ev) {
if (ev->Get() == ExpectedEvent) {
const TInstant now = TActivationContext::Now();
const TInstant barrier = LastResetTimestamp + Timeout;
if (now < barrier) {
// the time hasn't come yet
Schedule(barrier - now, 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
} else {
// no chance to disarm, fire callback
Callback();
ExpectedEvent = nullptr;
Iteration = 0;
}
}
}
private:
void Schedule(TDuration timeout, const TActorIdentity& actor) {
auto ev = MakeHolder<TEvent>();
ExpectedEvent = ev.Get();
Iteration = 0;
actor.Schedule(timeout, ev.Release());
}
};
}
|
