[ICRDMA] Add RDMA configuration into YDB config. EXT-1506 (#28843)

Author: dcherednik

ydb/core/client/server/msgbus_server_ic_debug.cpp

@@ -67,6 +67,7 @@ class TInterconnectDebugActor : public TMessageBusSecureRequest<TMessageBusServe
             params.SoftLoad = record.GetSoftLoad();
             params.Duration = TDuration::MicroSeconds(record.GetDuration());
             params.UseProtobufWithPayload = record.GetUseProtobufWithPayload();
+            params.RdmaMode = record.GetRdmaMode();
             Callback = [=](const TActorContext& ctx) {
                 ui32 poolId = 0;
                 if (record.HasServicePool()) {

ydb/core/driver_lib/cli_utils/cli_cmds_debug.cpp

@@ -18,6 +18,7 @@ class TInterconnectLoad : public TClientCommand {
     TDuration Duration;
     bool UseProtobufWithPayload = false;
     TString ServicePool;
+    ui32 RdmaMode = 0;
 
 public:
     TInterconnectLoad()
@@ -83,6 +84,10 @@ class TInterconnectLoad : public TClientCommand {
         config.Opts->AddLongOption("service-pool", "service pool name")
             .RequiredArgument()
             .StoreResult(&ServicePool);
+
+        config.Opts->AddLongOption("rdma-mode", "rdma mode for data transfer (0 - disabled; 1 - use rdma)")
+            .RequiredArgument()
+            .StoreResult(&RdmaMode);
     }
 
     int Run(TConfig& config) override {
@@ -110,6 +115,10 @@ class TInterconnectLoad : public TClientCommand {
             request.SetServicePool(ServicePool);
         }
 
+        if (RdmaMode) {
+            request.SetRdmaMode(RdmaMode);
+        }
+
         auto callback = [](const NMsgBusProxy::TBusResponse& response) {
             return response.Record.GetStatus() == NMsgBusProxy::MSTATUS_OK ? 0 : 1;
         };

ydb/core/driver_lib/run/kikimr_services_initializers.cpp

@@ -235,6 +235,8 @@
 #include <ydb/library/actors/interconnect/load.h>
 #include <ydb/library/actors/interconnect/poller/poller_actor.h>
 #include <ydb/library/actors/interconnect/poller/poller_tcp.h>
+#include <ydb/library/actors/interconnect/rdma/cq_actor/cq_actor.h>
+#include <ydb/library/actors/interconnect/rdma/mem_pool.h>
 #include <ydb/library/actors/util/affinity.h>
 #include <ydb/library/actors/wilson/wilson_uploader.h>
 #include <ydb/library/slide_limiter/service/service.h>
@@ -503,6 +505,10 @@ static TInterconnectSettings GetInterconnectSettings(const NKikimrConfig::TInter
         }
     }
 
+    if (config.HasRdmaChecksum()) {
+        result.RdmaChecksum = config.GetRdmaChecksum();
+    }
+
     return result;
 }
 
@@ -626,6 +632,17 @@ void TBasicServicesInitializer::InitializeServices(NActors::TActorSystemSetup* s
 
             TIntrusivePtr<TInterconnectProxyCommon> icCommon;
             icCommon.Reset(new TInterconnectProxyCommon);
+
+            if (icConfig.GetUseRdma()) {
+                setup->LocalServices.emplace_back(NInterconnect::NRdma::MakeCqActorId(),
+                    TActorSetupCmd(NInterconnect::NRdma::CreateCqActor(-1, icConfig.GetRdmaMaxWr()), TMailboxType::ReadAsFilled, interconnectPoolId));
+
+                // Interconnect uses rdma mem pool directly
+                const auto counters = GetServiceCounters(appData->Counters, "utils");
+                icCommon->RdmaMemPool = NInterconnect::NRdma::CreateSlotMemPool(counters.Get());
+                // Clients via wrapper to handle allocation fail
+                setup->RcBufAllocator = std::make_shared<TRdmaAllocatorWithFallback>(icCommon->RdmaMemPool);
+            }
             icCommon->NameserviceId = nameserviceId;
             icCommon->MonCounters = GetServiceCounters(counters, "interconnect");
             icCommon->ChannelsConfig = channels;

ydb/core/protos/config.proto

@@ -476,6 +476,9 @@ message TInterconnectConfig {
     optional bool ValidateIncomingPeerViaDirectLookup = 44;
     optional uint32 SocketBacklogSize = 45; // SOMAXCONN if not set or zero
     optional ESocketSendOptimization SocketSendOptimization = 51 [default = IC_SO_DISABLED];
+    optional bool UseRdma = 52;
+    optional uint32 RdmaMaxWr = 53 [default = 4096];
+    optional bool RdmaChecksum = 54 [default = true];
 
     // ballast is added to IC handshake frames to ensure correctness of jumbo frames transmission over network
     optional uint32 HandshakeBallastSize = 14;

ydb/core/protos/msgbus.proto

@@ -648,6 +648,7 @@ message TInterconnectDebug {
     optional uint64 RescheduleMax = 205;
 
     optional string SecurityToken = 13 [(Ydb.sensitive) = true];
+    optional uint32 RdmaMode = 14;
 }
 
 message TConsoleRequest {

ydb/library/actors/interconnect/load.cpp

@@ -197,9 +197,11 @@ namespace NInterconnect {
                 // create message and send it to the first hop
                 THolder<TEvLoadMessage> ev;
                 if (Params.UseProtobufWithPayload && size) {
-                    auto buffer = TRopeAlignedBuffer::Allocate(size);
-                    memset(buffer->GetBuffer(), '*', size);
-                    ev.Reset(new TEvLoadMessage(Hops, id, TRope(buffer)));
+                    TRcBuf buffer = Params.RdmaMode
+                        ? ctx.ActorSystem()->GetRcBufAllocator()->AllocRcBuf(size, 0, 0)
+                        : TRcBuf(TRopeAlignedBuffer::Allocate(size));
+                    memset(buffer.GetDataMut(), '*', size);
+                    ev.Reset(new TEvLoadMessage(Hops, id, TRope(std::move(buffer))));
                 } else {
                     TString payload;
                     if (size) {

ydb/library/actors/interconnect/load.h

@@ -18,6 +18,7 @@ namespace NInterconnect {
         bool SoftLoad;                      // is the load soft?
         TDuration Duration;                 // test duration
         bool UseProtobufWithPayload;        // store payload separately
+        ui32 RdmaMode;                      // rdma params bitmap, 0 - not used
     };
     NActors::IActor* CreateLoadActor(const TLoadParams& params);