Restoring authorship annotation for Anton Samokhvalov <pg83@yandex.ru>. Commit 2 of 2.

1 files changed, 140 insertions, 140 deletions

diff --git a/library/cpp/actors/interconnect/interconnect_tcp_session.h b/library/cpp/actors/interconnect/interconnect_tcp_session.h
index 1ecd829132..7fc00dbcc5 100644
--- a/library/cpp/actors/interconnect/interconnect_tcp_session.h
+++ b/library/cpp/actors/interconnect/interconnect_tcp_session.h
@@ -30,61 +30,61 @@
 #include <unordered_map>
 
 namespace NActors {
-    class TSlowPathChecker { 
-        using TTraceCallback = std::function<void(double)>; 
-        TTraceCallback Callback; 
-        const NHPTimer::STime Start; 
-
-    public: 
-        TSlowPathChecker(TTraceCallback&& callback) 
-            : Callback(std::move(callback)) 
+    class TSlowPathChecker {
+        using TTraceCallback = std::function<void(double)>;
+        TTraceCallback Callback;
+        const NHPTimer::STime Start;
+
+    public:
+        TSlowPathChecker(TTraceCallback&& callback)
+            : Callback(std::move(callback))
             , Start(GetCycleCountFast())
-        { 
-        } 
+        {
+        }
 
-        ~TSlowPathChecker() { 
+        ~TSlowPathChecker() {
             const NHPTimer::STime end = GetCycleCountFast();
-            const NHPTimer::STime elapsed = end - Start; 
-            if (elapsed > 1000000) { 
-                Callback(NHPTimer::GetSeconds(elapsed) * 1000); 
-            } 
-        } 
-
-        operator bool() const { 
-            return false; 
+            const NHPTimer::STime elapsed = end - Start;
+            if (elapsed > 1000000) {
+                Callback(NHPTimer::GetSeconds(elapsed) * 1000);
+            }
         }
-    }; 
 
-#define LWPROBE_IF_TOO_LONG(...)                                               \ 
-    if (auto __x = TSlowPathChecker{[&](double ms) { LWPROBE(__VA_ARGS__); }}) \ 
-        ;                                                                      \ 
-    else 
+        operator bool() const {
+            return false;
+        }
+    };
 
-    class TTimeLimit { 
-    public: 
-        TTimeLimit(ui64 limitInCycles) 
+#define LWPROBE_IF_TOO_LONG(...)                                               \
+    if (auto __x = TSlowPathChecker{[&](double ms) { LWPROBE(__VA_ARGS__); }}) \
+        ;                                                                      \
+    else
+
+    class TTimeLimit {
+    public:
+        TTimeLimit(ui64 limitInCycles)
             : UpperLimit(limitInCycles == 0 ? 0 : GetCycleCountFast() + limitInCycles)
-        { 
-        } 
+        {
+        }
 
-        TTimeLimit(ui64 startTS, ui64 limitInCycles) 
-            : UpperLimit(limitInCycles == 0 ? 0 : startTS + limitInCycles) 
-        { 
-        } 
+        TTimeLimit(ui64 startTS, ui64 limitInCycles)
+            : UpperLimit(limitInCycles == 0 ? 0 : startTS + limitInCycles)
+        {
+        }
 
-        bool CheckExceeded() { 
+        bool CheckExceeded() {
             return UpperLimit != 0 && GetCycleCountFast() > UpperLimit;
-        } 
+        }
 
-        const ui64 UpperLimit; 
-    }; 
+        const ui64 UpperLimit;
+    };
 
-    static constexpr TDuration DEFAULT_DEADPEER_TIMEOUT = TDuration::Seconds(10); 
+    static constexpr TDuration DEFAULT_DEADPEER_TIMEOUT = TDuration::Seconds(10);
     static constexpr TDuration DEFAULT_LOST_CONNECTION_TIMEOUT = TDuration::Seconds(10);
-    static constexpr ui32 DEFAULT_MAX_INFLIGHT_DATA = 10240 * 1024; 
-    static constexpr ui32 DEFAULT_TOTAL_INFLIGHT_DATA = 4 * 10240 * 1024; 
+    static constexpr ui32 DEFAULT_MAX_INFLIGHT_DATA = 10240 * 1024;
+    static constexpr ui32 DEFAULT_TOTAL_INFLIGHT_DATA = 4 * 10240 * 1024;
 
-    class TInterconnectProxyTCP; 
+    class TInterconnectProxyTCP;
 
     enum class EUpdateState : ui8 {
         NONE,                 // no updates generated by input session yet
@@ -93,22 +93,22 @@ namespace NActors {
         CONFIRMING,           // confirmation inflight
     };
 
-    struct TReceiveContext: public TAtomicRefCount<TReceiveContext> { 
-        /* All invokations to these fields should be thread-safe */ 
+    struct TReceiveContext: public TAtomicRefCount<TReceiveContext> {
+        /* All invokations to these fields should be thread-safe */
 
-        ui64 ControlPacketSendTimer = 0; 
-        ui64 ControlPacketId = 0; 
+        ui64 ControlPacketSendTimer = 0;
+        ui64 ControlPacketId = 0;
 
-        // number of packets received by input session 
-        TAtomic PacketsReadFromSocket = 0; 
-        TAtomic DataPacketsReadFromSocket = 0; 
+        // number of packets received by input session
+        TAtomic PacketsReadFromSocket = 0;
+        TAtomic DataPacketsReadFromSocket = 0;
 
         // last processed packet by input session
         std::atomic_uint64_t LastProcessedPacketSerial = 0;
         static constexpr uint64_t LastProcessedPacketSerialLockBit = uint64_t(1) << 63;
 
-        // for hardened checks 
-        TAtomic NumInputSessions = 0; 
+        // for hardened checks
+        TAtomic NumInputSessions = 0;
 
         NHPTimer::STime StartTime;
 
@@ -160,9 +160,9 @@ namespace NActors {
         ui64 GetLastProcessedPacketSerial() {
             return LastProcessedPacketSerial.load() & ~LastProcessedPacketSerialLockBit;
         }
-    }; 
+    };
 
-    class TInputSessionTCP 
+    class TInputSessionTCP
        : public TActorBootstrapped<TInputSessionTCP>
        , public TInterconnectLoggingBase
     {
@@ -174,14 +174,14 @@ namespace NActors {
         struct TEvCheckDeadPeer : TEventLocal<TEvCheckDeadPeer, EvCheckDeadPeer> {};
         struct TEvResumeReceiveData : TEventLocal<TEvResumeReceiveData, EvResumeReceiveData> {};
 
-    public: 
-        static constexpr EActivityType ActorActivityType() { 
+    public:
+        static constexpr EActivityType ActorActivityType() {
             return INTERCONNECT_SESSION_TCP;
-        } 
+        }
 
         TInputSessionTCP(const TActorId& sessionId,
-                         TIntrusivePtr<NInterconnect::TStreamSocket> socket, 
-                         TIntrusivePtr<TReceiveContext> context, 
+                         TIntrusivePtr<NInterconnect::TStreamSocket> socket,
+                         TIntrusivePtr<TReceiveContext> context,
                          TInterconnectProxyCommon::TPtr common,
                          std::shared_ptr<IInterconnectMetrics> metrics,
                          ui32 nodeId,
@@ -189,8 +189,8 @@ namespace NActors {
                          TDuration deadPeerTimeout,
                          TSessionParams params);
 
-    private: 
-        friend class TActorBootstrapped<TInputSessionTCP>; 
+    private:
+        friend class TActorBootstrapped<TInputSessionTCP>;
 
         void Bootstrap();
 
@@ -204,13 +204,13 @@ namespace NActors {
             cFunc(TEvConfirmUpdate::EventType, HandleConfirmUpdate)
         )
 
-    private: 
+    private:
         TRope IncomingData;
 
         const TActorId SessionId;
-        TIntrusivePtr<NInterconnect::TStreamSocket> Socket; 
+        TIntrusivePtr<NInterconnect::TStreamSocket> Socket;
         TPollerToken::TPtr PollerToken;
-        TIntrusivePtr<TReceiveContext> Context; 
+        TIntrusivePtr<TReceiveContext> Context;
         TInterconnectProxyCommon::TPtr Common;
         const ui32 NodeId;
         const TSessionParams Params;
@@ -235,11 +235,11 @@ namespace NActors {
 
         THolder<TEvUpdateFromInputSession> UpdateFromInputSession;
 
-        ui64 ConfirmedByInput; 
+        ui64 ConfirmedByInput;
 
         std::shared_ptr<IInterconnectMetrics> Metrics;
 
-        bool CloseInputSessionRequested = false; 
+        bool CloseInputSessionRequested = false;
 
         void CloseInputSession();
 
@@ -258,12 +258,12 @@ namespace NActors {
 
         TDeque<TIntrusivePtr<TRopeAlignedBuffer>> Buffers;
 
-        static constexpr size_t NumPreallocatedBuffers = 16; 
-        void PreallocateBuffers(); 
+        static constexpr size_t NumPreallocatedBuffers = 16;
+        void PreallocateBuffers();
 
-        inline ui64 GetMaxCyclesPerEvent() const { 
+        inline ui64 GetMaxCyclesPerEvent() const {
             return DurationToCycles(TDuration::MicroSeconds(500));
-        } 
+        }
 
         const TDuration DeadPeerTimeout;
         TInstant LastReceiveTimestamp;
@@ -278,21 +278,21 @@ namespace NActors {
 
         void HandlePingResponse(TDuration passed);
         void HandleClock(TInstant clock);
-    }; 
+    };
 
-    class TInterconnectSessionTCP 
+    class TInterconnectSessionTCP
        : public TActor<TInterconnectSessionTCP>
        , public TInterconnectLoggingBase
     {
-        enum { 
+        enum {
             EvCheckCloseOnIdle = EventSpaceBegin(TEvents::ES_PRIVATE),
             EvCheckLostConnection,
             EvRam,
             EvTerminate,
             EvFreeItems,
-        }; 
+        };
 
-        struct TEvCheckCloseOnIdle : TEventLocal<TEvCheckCloseOnIdle, EvCheckCloseOnIdle> {}; 
+        struct TEvCheckCloseOnIdle : TEventLocal<TEvCheckCloseOnIdle, EvCheckCloseOnIdle> {};
         struct TEvCheckLostConnection : TEventLocal<TEvCheckLostConnection, EvCheckLostConnection> {};
 
         struct TEvRam : TEventLocal<TEvRam, EvRam> {
@@ -308,18 +308,18 @@ namespace NActors {
             {}
         };
 
-        const TInstant Created; 
-        TInstant NewConnectionSet; 
-        ui64 MessagesGot = 0; 
-        ui64 MessagesWrittenToBuffer = 0; 
-        ui64 PacketsGenerated = 0; 
-        ui64 PacketsWrittenToSocket = 0; 
-        ui64 PacketsConfirmed = 0; 
+        const TInstant Created;
+        TInstant NewConnectionSet;
+        ui64 MessagesGot = 0;
+        ui64 MessagesWrittenToBuffer = 0;
+        ui64 PacketsGenerated = 0;
+        ui64 PacketsWrittenToSocket = 0;
+        ui64 PacketsConfirmed = 0;
 
-    public: 
-        static constexpr EActivityType ActorActivityType() { 
-            return INTERCONNECT_SESSION_TCP; 
-        } 
+    public:
+        static constexpr EActivityType ActorActivityType() {
+            return INTERCONNECT_SESSION_TCP;
+        }
 
         TInterconnectSessionTCP(TInterconnectProxyTCP* const proxy, TSessionParams params);
         ~TInterconnectSessionTCP();
@@ -339,8 +339,8 @@ namespace NActors {
             return ReceiveContext->ClockSkew_us;
         }
 
-    private: 
-        friend class TInterconnectProxyTCP; 
+    private:
+        friend class TInterconnectProxyTCP;
 
         void Handle(TEvTerminate::TPtr& ev);
         void HandlePoison();
@@ -400,7 +400,7 @@ namespace NActors {
         void ReestablishConnectionWithHandshake(TDisconnectReason reason);
         void ReestablishConnectionExecute();
 
-        TInterconnectProxyTCP* const Proxy; 
+        TInterconnectProxyTCP* const Proxy;
 
         // various connection settings access
         TDuration GetDeadPeerTimeout() const;
@@ -418,11 +418,11 @@ namespace NActors {
         void IssuePingRequest();
         void Handle(TEvProcessPingRequest::TPtr ev);
 
-        //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// 
+        ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 
-        TInstant LastInputActivityTimestamp; 
-        TInstant LastPayloadActivityTimestamp; 
-        TWatchdogTimer<TEvCheckCloseOnIdle> CloseOnIdleWatchdog; 
+        TInstant LastInputActivityTimestamp;
+        TInstant LastPayloadActivityTimestamp;
+        TWatchdogTimer<TEvCheckCloseOnIdle> CloseOnIdleWatchdog;
         TWatchdogTimer<TEvCheckLostConnection> LostConnectionWatchdog;
 
         void OnCloseOnIdleTimerHit() {
@@ -435,26 +435,26 @@ namespace NActors {
             Terminate(TDisconnectReason::LostConnection());
         }
 
-        //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// 
+        ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 
         const TSessionParams Params;
         TMaybe<TEventHolderPool> Pool;
         TMaybe<TChannelScheduler> ChannelScheduler;
-        ui64 TotalOutputQueueSize; 
-        bool OutputStuckFlag; 
-        TRecentWnd<std::pair<ui64, ui64>> OutputQueueUtilization; 
+        ui64 TotalOutputQueueSize;
+        bool OutputStuckFlag;
+        TRecentWnd<std::pair<ui64, ui64>> OutputQueueUtilization;
         size_t NumEventsInReadyChannels = 0;
 
-        void SetOutputStuckFlag(bool state); 
-        void SwitchStuckPeriod(); 
+        void SetOutputStuckFlag(bool state);
+        void SwitchStuckPeriod();
 
-        using TSendQueue = TList<TTcpPacketOutTask>; 
-        TSendQueue SendQueue; 
-        TSendQueue SendQueueCache; 
-        TSendQueue::iterator SendQueuePos; 
+        using TSendQueue = TList<TTcpPacketOutTask>;
+        TSendQueue SendQueue;
+        TSendQueue SendQueueCache;
+        TSendQueue::iterator SendQueuePos;
         ui64 WriteBlockedCycles = 0; // start of current block period
         TDuration WriteBlockedTotal; // total incremental duration that session has been blocked
-        ui64 BytesUnwritten = 0; 
+        ui64 BytesUnwritten = 0;
 
         void TrimSendQueueCache();
 
@@ -467,21 +467,21 @@ namespace NActors {
             }
         }
 
-        ui64 OutputCounter; 
+        ui64 OutputCounter;
         ui64 LastSentSerial = 0;
 
-        TInstant LastHandshakeDone; 
+        TInstant LastHandshakeDone;
 
-        TIntrusivePtr<NInterconnect::TStreamSocket> Socket; 
+        TIntrusivePtr<NInterconnect::TStreamSocket> Socket;
         TPollerToken::TPtr PollerToken;
         ui32 SendBufferSize;
         ui64 InflightDataAmount = 0;
 
         std::unordered_map<TActorId, ui64, TActorId::THash> Subscribers;
 
-        // time at which we want to send confirmation packet even if there was no outgoing data 
-        ui64 UnconfirmedBytes = 0; 
-        TInstant ForcePacketTimestamp = TInstant::Max(); 
+        // time at which we want to send confirmation packet even if there was no outgoing data
+        ui64 UnconfirmedBytes = 0;
+        TInstant ForcePacketTimestamp = TInstant::Max();
         TPriorityQueue<TInstant, TVector<TInstant>, std::greater<TInstant>> FlushSchedule;
         size_t MaxFlushSchedule = 0;
         ui64 FlushEventsScheduled = 0;
@@ -494,71 +494,71 @@ namespace NActors {
 
         void GenerateHttpInfo(TStringStream& str);
 
-        TIntrusivePtr<TReceiveContext> ReceiveContext; 
+        TIntrusivePtr<TReceiveContext> ReceiveContext;
         TActorId ReceiverId;
         TDuration Ping;
 
-        ui64 ConfirmPacketsForcedBySize = 0; 
-        ui64 ConfirmPacketsForcedByTimeout = 0; 
+        ui64 ConfirmPacketsForcedBySize = 0;
+        ui64 ConfirmPacketsForcedByTimeout = 0;
 
-        ui64 LastConfirmed = 0; 
+        ui64 LastConfirmed = 0;
 
         TEvHandshakeDone::TPtr PendingHandshakeDoneEvent;
         bool StartHandshakeOnSessionClose = false;
 
         ui64 EqualizeCounter = 0;
-    }; 
+    };
 
-    class TInterconnectSessionKiller 
-       : public TActorBootstrapped<TInterconnectSessionKiller> { 
-        ui32 RepliesReceived = 0; 
-        ui32 RepliesNumber = 0; 
+    class TInterconnectSessionKiller
+       : public TActorBootstrapped<TInterconnectSessionKiller> {
+        ui32 RepliesReceived = 0;
+        ui32 RepliesNumber = 0;
         TActorId LargestSession = TActorId();
-        ui64 MaxBufferSize = 0; 
+        ui64 MaxBufferSize = 0;
         TInterconnectProxyCommon::TPtr Common;
 
-    public: 
+    public:
         static constexpr EActivityType ActorActivityType() {
             return INTERCONNECT_SESSION_KILLER;
         }
 
         TInterconnectSessionKiller(TInterconnectProxyCommon::TPtr common)
-            : Common(common) 
-        { 
-        } 
+            : Common(common)
+        {
+        }
 
         void Bootstrap() {
-            auto sender = SelfId(); 
+            auto sender = SelfId();
             const auto eventFabric = [&sender](const TActorId& recp) -> IEventHandle* {
-                auto ev = new TEvSessionBufferSizeRequest(); 
-                return new IEventHandle(recp, sender, ev, IEventHandle::FlagTrackDelivery); 
-            }; 
+                auto ev = new TEvSessionBufferSizeRequest();
+                return new IEventHandle(recp, sender, ev, IEventHandle::FlagTrackDelivery);
+            };
             RepliesNumber = TlsActivationContext->ExecutorThread.ActorSystem->BroadcastToProxies(eventFabric);
-            Become(&TInterconnectSessionKiller::StateFunc); 
-        } 
+            Become(&TInterconnectSessionKiller::StateFunc);
+        }
 
         STRICT_STFUNC(StateFunc,
             hFunc(TEvSessionBufferSizeResponse, ProcessResponse)
             cFunc(TEvents::TEvUndelivered::EventType, ProcessUndelivered)
         )
- 
+
         void ProcessResponse(TEvSessionBufferSizeResponse::TPtr& ev) {
-            RepliesReceived++; 
-            if (MaxBufferSize < ev->Get()->BufferSize) { 
-                MaxBufferSize = ev->Get()->BufferSize; 
-                LargestSession = ev->Get()->SessionID; 
-            } 
-            if (RepliesReceived == RepliesNumber) { 
+            RepliesReceived++;
+            if (MaxBufferSize < ev->Get()->BufferSize) {
+                MaxBufferSize = ev->Get()->BufferSize;
+                LargestSession = ev->Get()->SessionID;
+            }
+            if (RepliesReceived == RepliesNumber) {
                 Send(LargestSession, new TEvents::TEvPoisonPill);
-                AtomicUnlock(&Common->StartedSessionKiller); 
+                AtomicUnlock(&Common->StartedSessionKiller);
                 PassAway();
-            } 
+            }
         }
- 
+
         void ProcessUndelivered() {
-            RepliesReceived++; 
+            RepliesReceived++;
         }
-    }; 
+    };
 
     void CreateSessionKillingActor(TInterconnectProxyCommon::TPtr common);