Add support for delaying interconnect events (#5954)

ydb-platform · Jun 26, 2024 · afab5f3 · afab5f3
1 parent 68df614
commit afab5f3
Show file tree

Hide file tree

Showing 6 changed files with 56 additions and 6 deletions.
diff --git a/ydb/core/driver_lib/run/kikimr_services_initializers.cpp b/ydb/core/driver_lib/run/kikimr_services_initializers.cpp
@@ -552,6 +552,10 @@ static TInterconnectSettings GetInterconnectSettings(const NKikimrConfig::TInter
         result.ErrorSleepRetryMultiplier = config.GetErrorSleepRetryMultiplier();
     }
 
+    if (config.HasEventDelayMicrosec()) {
+        result.EventDelay = TDuration::MicroSeconds(config.GetEventDelayMicrosec());
+    }
+
     return result;
 }
 

diff --git a/ydb/core/protos/config.proto b/ydb/core/protos/config.proto
@@ -432,6 +432,8 @@ message TInterconnectConfig {
     optional double ErrorSleepRetryMultiplier = 48;
 
     optional uint32 OutgoingHandshakeInflightLimit = 43;
+
+    optional uint64 EventDelayMicrosec = 49;
 }
 
 message TChannelProfileConfig {

diff --git a/ydb/library/actors/core/interconnect.h b/ydb/library/actors/core/interconnect.h
@@ -138,6 +138,7 @@ namespace NActors {
             EvCloseInputSession,
             EvPoisonSession,
             EvTerminate,
+            EvForwardDelayed,
             EvEnd
         };
 
@@ -264,5 +265,7 @@ namespace NActors {
         struct TEvPoisonSession : TEventLocal<TEvPoisonSession, EvPoisonSession> {};
 
         struct TEvTerminate : TEventLocal<TEvTerminate, EvTerminate> {};
+
+        struct TEvForwardDelayed : TEventLocal<TEvForwardDelayed, EvForwardDelayed> {};
     };
 }
diff --git a/ydb/library/actors/interconnect/interconnect_common.h b/ydb/library/actors/interconnect/interconnect_common.h
@@ -54,6 +54,7 @@ namespace NActors {
         TDuration FirstErrorSleep = TDuration::MilliSeconds(10);
         TDuration MaxErrorSleep = TDuration::Seconds(1);
         double ErrorSleepRetryMultiplier = 4.0;
+        TDuration EventDelay = TDuration::Zero();
 
         ui32 GetSendBufferSize() const {
             ui32 res = 512 * 1024; // 512 kb is the default value for send buffer

diff --git a/ydb/library/actors/interconnect/interconnect_tcp_session.cpp b/ydb/library/actors/interconnect/interconnect_tcp_session.cpp
@@ -89,6 +89,12 @@ namespace NActors {
         Proxy->Metrics->SubSubscribersCount(Subscribers.size());
         Subscribers.clear();
 
+        for (auto& d : DelayedEvents) {
+            d.Span.EndError("nondelivery");
+            TActivationContext::Send(IEventHandle::ForwardOnNondelivery(d.Event, TEvents::TEvUndelivered::Disconnected));
+        }
+        DelayedEvents.clear();
+
         ChannelScheduler->ForEach([&](TEventOutputChannel& channel) {
             channel.NotifyUndelivered();
         });
@@ -115,16 +121,12 @@ namespace NActors {
         Y_ABORT("TInterconnectSessionTCP::PassAway() can't be called directly");
     }
 
-    void TInterconnectSessionTCP::Forward(STATEFN_SIG) {
-        Proxy->ValidateEvent(ev, "Forward");
+    void TInterconnectSessionTCP::Enqueue(STATEFN_SIG) {
+        Proxy->ValidateEvent(ev, "Enqueue");
 
         LOG_DEBUG_IC_SESSION("ICS02", "send event from: %s to: %s", ev->Sender.ToString().data(), ev->Recipient.ToString().data());
         ++MessagesGot;
 
-        if (ev->Flags & IEventHandle::FlagSubscribeOnSession) {
-            Subscribe(ev);
-        }
-
         ui16 evChannel = ev->GetChannel();
         auto& oChannel = ChannelScheduler->GetOutputChannel(evChannel);
         const bool wasWorking = oChannel.IsWorking();
@@ -164,6 +166,35 @@ namespace NActors {
         IssueRam(true);
     }
 
+    void TInterconnectSessionTCP::Forward(STATEFN_SIG) {
+        Proxy->ValidateEvent(ev, "Forward");
+
+        if (ev->Flags & IEventHandle::FlagSubscribeOnSession) {
+            Subscribe(ev);
+        }
+
+        if (Y_UNLIKELY(Proxy->Common->Settings.EventDelay)) {
+            auto& d = DelayedEvents.emplace_back();
+            d.Event = std::move(ev);
+            if (Y_UNLIKELY(d.Event->TraceId)) {
+                d.Span = NWilson::TSpan(15 /*max verbosity*/, std::move(d.Event->TraceId), "Interconnect.Delay");
+                // Reparent event to the delay span
+                d.Event->TraceId = d.Span.GetTraceId();
+            }
+            Schedule(Proxy->Common->Settings.EventDelay, new TEvInterconnect::TEvForwardDelayed);
+        } else {
+            Enqueue(ev);
+        }
+    }
+
+    void TInterconnectSessionTCP::ForwardDelayed() {
+        Y_ABORT_UNLESS(!DelayedEvents.empty());
+        auto d = std::move(DelayedEvents.front());
+        DelayedEvents.pop_front();
+        d.Span.End();
+        Enqueue(d.Event);
+    }
+
     void TInterconnectSessionTCP::Subscribe(STATEFN_SIG) {
         LOG_DEBUG_IC_SESSION("ICS04", "subscribe for session state for %s", ev->Sender.ToString().data());
         const auto [it, inserted] = Subscribers.emplace(ev->Sender, ev->Cookie);

diff --git a/ydb/library/actors/interconnect/interconnect_tcp_session.h b/ydb/library/actors/interconnect/interconnect_tcp_session.h
@@ -445,7 +445,9 @@ namespace NActors {
         void Terminate(TDisconnectReason reason);
         void PassAway() override;
 
+        void Enqueue(STATEFN_SIG);
         void Forward(STATEFN_SIG);
+        void ForwardDelayed();
         void Subscribe(STATEFN_SIG);
         void Unsubscribe(STATEFN_SIG);
 
@@ -456,6 +458,7 @@ namespace NActors {
             TimeLimit.emplace(GetMaxCyclesPerEvent());
             STRICT_STFUNC_BODY(
                 fFunc(TEvInterconnect::EvForward, Forward)
+                cFunc(TEvInterconnect::EvForwardDelayed, ForwardDelayed)
                 cFunc(TEvents::TEvPoisonPill::EventType, HandlePoison)
                 fFunc(TEvInterconnect::TEvConnectNode::EventType, Subscribe)
                 fFunc(TEvents::TEvSubscribe::EventType, Subscribe)
@@ -611,6 +614,12 @@ namespace NActors {
 
         std::unordered_map<TActorId, ui64, TActorId::THash> Subscribers;
 
+        struct TDelayedEvent {
+            TAutoPtr<IEventHandle> Event;
+            NWilson::TSpan Span;
+        };
+        std::deque<TDelayedEvent> DelayedEvents;
+
         // time at which we want to send confirmation packet even if there was no outgoing data
         ui64 UnconfirmedBytes = 0;
         TMonotonic ForcePacketTimestamp = TMonotonic::Max();