intermediate changes

ref:cde9a383711a11544ce7e107a78147fb96cc4029

1 files changed, 68 insertions, 0 deletions

diff --git a/library/cpp/actors/interconnect/watchdog_timer.h b/library/cpp/actors/interconnect/watchdog_timer.h
new file mode 100644
index 0000000000..c190105a59
--- /dev/null
+++ b/library/cpp/actors/interconnect/watchdog_timer.h
@@ -0,0 +1,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());
+        }
+    };
+
+}