[ICRDMA] Event based cq implementation. EXT-1228 (#29463)

Add blocking mode (ibv_get_cq_event) to reduce CPU consumption.

https://www.rdmamojo.com/2013/03/09/ibv_get_cq_event/
 Conflicts:
	ydb/library/actors/interconnect/ut/lib/ic_test_cluster.h
	ydb/library/actors/interconnect/ut/lib/node.h

Author: dcherednik

ydb/core/driver_lib/run/kikimr_services_initializers.cpp

@@ -633,9 +633,23 @@ void TBasicServicesInitializer::InitializeServices(NActors::TActorSystemSetup* s
             TIntrusivePtr<TInterconnectProxyCommon> icCommon;
             icCommon.Reset(new TInterconnectProxyCommon);
 
+            NMonitoring::TDynamicCounterPtr interconectCounters = GetServiceCounters(counters, "interconnect");
+
             if (icConfig.GetUseRdma()) {
+                NInterconnect::NRdma::ECqMode rdmaCqMode = NInterconnect::NRdma::ECqMode::EVENT;
+                if (icConfig.HasRdmaCqMode()) {
+                    switch (icConfig.GetRdmaCqMode()) {
+                        case NKikimrConfig::TInterconnectConfig::CQ_EVENT:
+                            rdmaCqMode = NInterconnect::NRdma::ECqMode::EVENT;
+                            break;
+                        case NKikimrConfig::TInterconnectConfig::CQ_POLLING:
+                            rdmaCqMode = NInterconnect::NRdma::ECqMode::POLLING;
+                            break;
+                    }
+                }
                 setup->LocalServices.emplace_back(NInterconnect::NRdma::MakeCqActorId(),
-                    TActorSetupCmd(NInterconnect::NRdma::CreateCqActor(-1, icConfig.GetRdmaMaxWr()), TMailboxType::ReadAsFilled, interconnectPoolId));
+                    TActorSetupCmd(NInterconnect::NRdma::CreateCqActor(-1, icConfig.GetRdmaMaxWr(), rdmaCqMode, interconectCounters.Get()),
+                        TMailboxType::ReadAsFilled, interconnectPoolId));
 
                 // Interconnect uses rdma mem pool directly
                 const auto counters = GetServiceCounters(appData->Counters, "utils");
@@ -644,7 +658,7 @@ void TBasicServicesInitializer::InitializeServices(NActors::TActorSystemSetup* s
                 setup->RcBufAllocator = std::make_shared<TRdmaAllocatorWithFallback>(icCommon->RdmaMemPool);
             }
             icCommon->NameserviceId = nameserviceId;
-            icCommon->MonCounters = GetServiceCounters(counters, "interconnect");
+            icCommon->MonCounters = interconectCounters;
             icCommon->ChannelsConfig = channels;
             icCommon->Settings = settings;
             icCommon->DestructorId = GetDestructActorID();

ydb/core/protos/config.proto

@@ -442,6 +442,11 @@ message TInterconnectConfig {
         IC_SO_MSG_ZEROCOPY = 1;
     };
 
+    enum ERdmaCqMode {
+        CQ_POLLING = 0;
+        CQ_EVENT = 1;
+    };
+
     repeated TChannel Channel = 1;
     optional bool FirstTryBeforePoll = 2; // DEPRECATED
     optional bool StartTcp = 3 [default = false];
@@ -479,6 +484,7 @@ message TInterconnectConfig {
     optional bool UseRdma = 52;
     optional uint32 RdmaMaxWr = 53 [default = 4096];
     optional bool RdmaChecksum = 54 [default = true];
+    optional ERdmaCqMode RdmaCqMode = 55;
 
     // ballast is added to IC handshake frames to ensure correctness of jumbo frames transmission over network
     optional uint32 HandshakeBallastSize = 14;

ydb/library/actors/interconnect/rdma/cq_actor/cq_actor.cpp

@@ -226,10 +226,18 @@ class TCqActor: public TInterconnectLoggingBase, public TActorBootstrapped<TCqAc
     TCqFactory CqFactory;
 };
 
-NActors::IActor* CreateCqActor(int maxCqe, int maxWr, NMonitoring::TDynamicCounters* counters) {
-    return new TCqActor([maxCqe, maxWr, counters](const TRdmaCtx* ctx) {
-        return CreateSimpleCq(ctx, TlsActivationContext->AsActorContext().ActorSystem(), maxCqe, maxWr, counters);
-    });
+NActors::IActor* CreateCqActor(int maxCqe, int maxWr, ECqMode mode, NMonitoring::TDynamicCounters* counters) {
+    switch (mode) {
+        case NInterconnect::NRdma::ECqMode::POLLING:
+            return new TCqActor([maxCqe, maxWr, counters](const TRdmaCtx* ctx) {
+                return CreateSimpleCq(ctx, TlsActivationContext->AsActorContext().ActorSystem(), maxCqe, maxWr, counters);
+            });
+
+        case NInterconnect::NRdma::ECqMode::EVENT:
+            return new TCqActor([maxCqe, maxWr, counters](const TRdmaCtx* ctx) {
+                return CreateSimpleEventDrivenCq(ctx, TlsActivationContext->AsActorContext().ActorSystem(), maxCqe, maxWr, counters);
+            });
+    }
 }
 
 NActors::TActorId MakeCqActorId() {

ydb/library/actors/interconnect/rdma/cq_actor/cq_actor.h

@@ -12,12 +12,17 @@ namespace NMonitoring {
 
 namespace NInterconnect::NRdma {
 
+enum class ECqMode : ui8 {
+    POLLING = 0,
+    EVENT = 1
+};
+
 /*
  * Creates CQ actor - abstraction to commuticate with CQ from actor system.
  * creates at least one CQ per rdma context
  * maxCqe - max capacity of single queue under CQ actor abstruction. -1 - use limit from rdma context
  */
-NActors::IActor* CreateCqActor(int maxCqe, int maxWr, NMonitoring::TDynamicCounters* counters = nullptr);
+NActors::IActor* CreateCqActor(int maxCqe, int maxWr, ECqMode mode, NMonitoring::TDynamicCounters* counters);
 NActors::TActorId MakeCqActorId();
 
 }

ydb/library/actors/interconnect/rdma/rdma.cpp

@@ -11,6 +11,7 @@
 #include <library/cpp/monlib/dynamic_counters/counters.h>
 
 #include <util/system/thread.h>
+#include <util/system/yield.h>
 
 namespace NInterconnect::NRdma {
 
@@ -22,25 +23,135 @@ NMonitoring::TDynamicCounterPtr MakeCounters(NMonitoring::TDynamicCounters* coun
     return counters;
 }
 
-TCqCommon::~TCqCommon() {
-    if (Cq) {
-            ibv_destroy_cq(Cq);
-    }
+static void SigHandler(int) noexcept {
+    // Empty handler. We just need to interrupt read syscall
+}
+
+void SetSigHandler() noexcept {
+    struct sigaction sigActionData;
+    sigemptyset(&sigActionData.sa_mask);
+    sigActionData.sa_handler = &SigHandler;
+    sigActionData.sa_flags = SA_INTERRUPT;
+    sigaction(SIGUSR1, &sigActionData, nullptr);
 }
 
 class TSimpleCq: public TSimpleCqBase {
 public:
-    using TSimpleCqBase::TSimpleCqBase;
+    TSimpleCq(NActors::TActorSystem* as, size_t sz, NMonitoring::TDynamicCounters* c) noexcept
+        : TSimpleCqBase(as, sz, c, true)
+    {}
+
+    int Init(const TRdmaCtx* ctx, int maxCqe) noexcept {
+        return TSimpleCqBase::Init(ctx, maxCqe, nullptr);
+    }
 
     virtual ~TSimpleCq() {
-        Stop();
+        // For simple polling mode CQ, we just can destroy ibv CQ without any issues just aftre joining to the thread
+        Cont.store(false, std::memory_order_relaxed);
+        if (Thread.Running())
+            Thread.Join();
+
+        DestroyCq();
+    }
+};
+
+class TSimpleEventDrivenCq: public TSimpleCqBase {
+public:
+    TSimpleEventDrivenCq(NActors::TActorSystem* as, size_t sz, NMonitoring::TDynamicCounters* c) noexcept
+        : TSimpleCqBase(as, sz, c, false)
+    {}
+
+    int Init(const TRdmaCtx* ctx, int maxCqe) noexcept {
+        CompChannel = ibv_create_comp_channel(ctx->GetContext());
+        if (!CompChannel) {
+            return errno;
+        }
+
+        int err = TSimpleCqBase::Init(ctx, maxCqe, CompChannel);
+        if (err) {
+            return err;
+        }
+
+        err = ibv_req_notify_cq(Cq, 0);
+        if (err) {
+            return errno;
+        }
+
+        return 0;
+    }
+
+    void Idle() noexcept override final {
+        struct ibv_cq *evCq = nullptr;
+        void *evCtx = nullptr;
+
+        int err = ibv_get_cq_event(CompChannel, &evCq, &evCtx);
+        if (err) {
+            if (errno != EINTR) {
+                NotifyErr();
+                return;
+            }
+        }
+
+        if (!evCq) {
+            return;
+        }
+        // TODO: batch ack
+        ibv_ack_cq_events(evCq, 1);
+        err = ibv_req_notify_cq(evCq, 0);
+        if (err) {
+            Cerr << "Couldn't request CQ notification\n" << Endl;
+            NotifyErr();
+            Y_DEBUG_ABORT_UNLESS(false);
+        }
     }
+
+    virtual ~TSimpleEventDrivenCq() {
+        // For event driven CQ stopping is a bit complicated
+
+        // 1. Lock the verbs builder. This prevents possibility to add new WR. Not nessesearly but just to be sure
+        // No deadlock here - the builder routine is protected by TryLock semantic.
+        VerbsBuildingState.Lock.Acquire();
+
+        // 2. Set flag to exit from loop
+        Cont.store(false, std::memory_order_relaxed);
+
+        // 3. Send signal to the thread to interrupt waiting on the read syscall ()
+        // NOTE: There is a tiny chanse the signal was send before thread blocked on the read syscall
+        // so in this case repeat send signal until cq thread finished
+        while (!Finished.load(std::memory_order_relaxed)) {
+            Awake();
+            if (Finished.load(std::memory_order_relaxed)) {
+                break;
+            }
+            ThreadYield();
+        }
+
+        // 4. As usual, join and destroy CQ
+        if (Thread.Running())
+            Thread.Join();
+
+        DestroyCq();
+
+        // 5. Destroy completion event channel
+        if (ibv_destroy_comp_channel(CompChannel)) {
+            // https://www.rdmamojo.com/2012/10/26/ibv_destroy_comp_channel
+            Cerr << "Unable to destroy completion event channel, errno: " << errno << Endl;
+            // it should not happen, but if it happens it is not a fatal error for production
+            Y_DEBUG_ABORT_UNLESS(false);
+        }
+    }
+private:
+    ibv_comp_channel* CompChannel;
 };
 
 ICq::TPtr CreateSimpleCq(const TRdmaCtx* ctx, NActors::TActorSystem* as, int maxCqe, int maxWr, NMonitoring::TDynamicCounters* counter) noexcept {
     return CreateCq<TSimpleCq>(ctx, as, maxCqe, maxWr, counter);
 }
 
+ICq::TPtr CreateSimpleEventDrivenCq(const TRdmaCtx* ctx, NActors::TActorSystem* as, int maxCqe, int maxWr, NMonitoring::TDynamicCounters* counter) noexcept {
+    return CreateCq<TSimpleEventDrivenCq>(ctx, as, maxCqe, maxWr, counter);
+}
+
 const int TQueuePair::UnknownQpState = IBV_QPS_UNKNOWN;
 
 TQueuePair::~TQueuePair() {

ydb/library/actors/interconnect/rdma/rdma.h

@@ -77,6 +77,7 @@ class ICq {
 };
 
 ICq::TPtr CreateSimpleCq(const TRdmaCtx* ctx, NActors::TActorSystem* as, int maxCqe, int maxWr, NMonitoring::TDynamicCounters* counter) noexcept;
+ICq::TPtr CreateSimpleEventDrivenCq(const TRdmaCtx* ctx, NActors::TActorSystem* as, int maxCqe, int maxWr, NMonitoring::TDynamicCounters* counter) noexcept;
 
 struct THandshakeData {
     ui32 QpNum;