[ICRDMA] Rdma send recv primitives (#45051) NBYDB-2320

Implement basic send/receive ibverb wrappers to use inside IC 

12 files changed, 1196 insertions, 98 deletions

diff --git a/ydb/core/driver_lib/run/kikimr_services_initializers.cpp b/ydb/core/driver_lib/run/kikimr_services_initializers.cpp
index 8857c03ce23..d28e82e6598 100644
--- a/ydb/core/driver_lib/run/kikimr_services_initializers.cpp
+++ b/ydb/core/driver_lib/run/kikimr_services_initializers.cpp
@@ -263,6 +263,7 @@
 #include <ydb/library/actors/interconnect/poller/uring_poller_actor.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/interconnect/rdma/rdma.h>
 #include <ydb/core/retro_tracing_impl/distributed_collector/distributed_retro_collector.h>
 #include <ydb/library/actors/retro_tracing/collector/retro_collector.h>
 #include <ydb/library/actors/util/affinity.h>
@@ -737,7 +738,10 @@ void TBasicServicesInitializer::InitializeServices(NActors::TActorSystemSetup* s
                     }
                 }
                 setup->LocalServices.emplace_back(NInterconnect::NRdma::MakeCqActorId(),
-                    TActorSetupCmd(NInterconnect::NRdma::CreateCqActor(-1, icConfig.GetRdmaMaxWr(), rdmaCqMode, interconectCounters.Get()),
+                    TActorSetupCmd(NInterconnect::NRdma::CreateCqActor(
+                        NInterconnect::NRdma::TRdmaRuntimeParams{-1, static_cast<int>(icConfig.GetRdmaMaxWr()), 0, 0},
+                        rdmaCqMode,
+                        interconectCounters.Get()),
                         TMailboxType::ReadAsFilled, interconnectPoolId));
 
                 // Interconnect uses rdma mem pool directly
diff --git a/ydb/library/actors/interconnect/rdma/cq_actor/cq_actor.cpp b/ydb/library/actors/interconnect/rdma/cq_actor/cq_actor.cpp
index f05627f75b6..2e9411d04f1 100644
--- a/ydb/library/actors/interconnect/rdma/cq_actor/cq_actor.cpp
+++ b/ydb/library/actors/interconnect/rdma/cq_actor/cq_actor.cpp
@@ -16,7 +16,7 @@ using namespace NActors;
 
 namespace NInterconnect::NRdma {
 
-using TCqFactory = std::function<ICq::TPtr(const TRdmaCtx*)>;
+using TCqFactory = std::function<ICq::TPtr(const TRdmaCtx*, std::shared_ptr<IMemPool>)>;
 
 static const TDuration PeriodicActionInterval = TDuration::Seconds(1);
 
@@ -109,7 +109,7 @@ public:
         LOG_DEBUG_IC("ICRDMA", "Got TEvPollerRegisterResult for fd: %d",
                 ev->Get()->Socket.Get()->GetDescriptor());
         auto rdmaCtx = static_cast<TAsyncEventDesctiptor*>(ev->Get()->Socket.Get())->GetContext();
-        auto cqPtr = CqFactory(rdmaCtx);
+        auto cqPtr = CqFactory(rdmaCtx, MemPool);
         auto it = CqMap.find(rdmaCtx);
         Y_ABORT_UNLESS(it != CqMap.end());
         Y_ABORT_UNLESS(it->second.index() == 1);
@@ -204,7 +204,7 @@ private:
 
     void ProcessCqErr(auto it) {
         TCtxData& c = std::get<0>(it->second);
-        LOG_ERROR_IC("ICRDMA", "Cq error issued on ctx %s, notify all pending cq callbacks",
+        LOG_ERROR_IC("ICRDMA", "CQ/SRQ error issued on ctx %s, notify all pending cq callbacks",
             it->first->ToString().data());
         c.Cq->NotifyErr();
     }
@@ -223,9 +223,21 @@ private:
 
         switch (async_event.event_type) {
             case IBV_EVENT_CQ_ERR:
+            case IBV_EVENT_SRQ_ERR:
+            /*  Docs say:
+                All async events that ibv_get_async_event() returns must be
+                acknowledged using ibv_ack_async_event().  To avoid races,
+                destroying an object (CQ, SRQ or QP) will wait for all affiliated
+                events for the object to be acknowledged; this avoids an
+                application retrieving an affiliated event after the corresponding
+                object has already been destroyed.
+
+                So, to avoid deadlock we need ack event before decrementing refcount
+            */
                 ProcessCqErr(it);
+                ibv_ack_async_event(&async_event);
                 CqMap.erase(it);
-            break;
+                return true;
             default:
                 std::get<0>(it->second).AsyncEventToken->Request(true, false);
             break;
@@ -250,16 +262,16 @@ private:
     std::shared_ptr<NInterconnect::NRdma::IMemPool> MemPool;
 };
 
-NActors::IActor* CreateCqActor(int maxCqe, int maxWr, ECqMode mode, NMonitoring::TDynamicCounters* counters) {
+NActors::IActor* CreateCqActor(const TRdmaRuntimeParams& runtimeParams, 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);
+            return new TCqActor([runtimeParams, counters](const TRdmaCtx* ctx, std::shared_ptr<IMemPool> memPool) {
+                return CreateSimpleCq(ctx, TlsActivationContext->AsActorContext().ActorSystem(), runtimeParams, std::move(memPool), counters);
             });
 
         case NInterconnect::NRdma::ECqMode::EVENT:
-            return new TCqActor([maxCqe, maxWr, counters](const TRdmaCtx* ctx) {
-                return CreateSimpleEventDrivenCq(ctx, TlsActivationContext->AsActorContext().ActorSystem(), maxCqe, maxWr, counters);
+            return new TCqActor([runtimeParams, counters](const TRdmaCtx* ctx, std::shared_ptr<IMemPool> memPool) {
+                return CreateSimpleEventDrivenCq(ctx, TlsActivationContext->AsActorContext().ActorSystem(), runtimeParams, std::move(memPool), counters);
             });
     }
 }
diff --git a/ydb/library/actors/interconnect/rdma/cq_actor/cq_actor.h b/ydb/library/actors/interconnect/rdma/cq_actor/cq_actor.h
index 6c31c4572ad..4328164e6a1 100644
--- a/ydb/library/actors/interconnect/rdma/cq_actor/cq_actor.h
+++ b/ydb/library/actors/interconnect/rdma/cq_actor/cq_actor.h
@@ -17,12 +17,14 @@ enum class ECqMode : ui8 {
     EVENT = 1
 };
 
+struct TRdmaRuntimeParams;
+
 /*
  * 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, ECqMode mode, NMonitoring::TDynamicCounters* counters);
+NActors::IActor* CreateCqActor(const TRdmaRuntimeParams& runtimeParams, ECqMode mode, NMonitoring::TDynamicCounters* counters);
 NActors::TActorId MakeCqActorId();
 
 }
diff --git a/ydb/library/actors/interconnect/rdma/cq_actor/cq_actor_dummy.cpp b/ydb/library/actors/interconnect/rdma/cq_actor/cq_actor_dummy.cpp
index f4ce730f270..4dedd9dd3bf 100644
--- a/ydb/library/actors/interconnect/rdma/cq_actor/cq_actor_dummy.cpp
+++ b/ydb/library/actors/interconnect/rdma/cq_actor/cq_actor_dummy.cpp
@@ -35,7 +35,7 @@ public:
     }
 };
 
-NActors::IActor* CreateCqActor(int /*maxCqe*/, int /*maxWr*/, ECqMode, NMonitoring::TDynamicCounters* /*counters*/) {
+NActors::IActor* CreateCqActor(const TRdmaRuntimeParams&, ECqMode, NMonitoring::TDynamicCounters* /*counters*/) {
     return new TCqActorDummy();
 }
 
diff --git a/ydb/library/actors/interconnect/rdma/events.h b/ydb/library/actors/interconnect/rdma/events.h
index b990154a170..210197e615f 100644
--- a/ydb/library/actors/interconnect/rdma/events.h
+++ b/ydb/library/actors/interconnect/rdma/events.h
@@ -3,6 +3,12 @@
 #include "rdma.h"
 #include <ydb/library/actors/core/events.h>
 #include <ydb/library/actors/core/event_local.h>
+#include <ydb/library/actors/util/rc_buf.h>
+
+#include <memory>
+#include <string_view>
+#include <utility>
+#include <variant>
 
 namespace NInterconnect::NRdma {
 
@@ -11,6 +17,7 @@ namespace NInterconnect::NRdma {
         EvGetCqHandle = Start,
         EvRdmaIoDone,
         EvRdmaReadDone,
+        EvRdmaIoReceiveDone,
     };
 
     struct TEvGetCqHandle: public NActors::TEventLocal<TEvGetCqHandle, ui32(ERdma::EvGetCqHandle)> {
@@ -21,9 +28,16 @@ namespace NInterconnect::NRdma {
         ICq::TPtr CqPtr;
     };
 
-    struct TEvRdmaIoDone : public NActors::TEventLocal<TEvRdmaIoDone, (ui32)ERdma::EvRdmaIoDone> {
-        struct TSuccess {
-        };
+    struct TSuccess {
+    };
+
+    struct TSuccessReceive {
+        TRcBuf Buf;
+    };
+
+    template <typename TDerived, typename TSuccessRecord, ERdma EventId>
+    struct TEvRdmaIoDoneCommon : public NActors::TEventLocal<TDerived, (ui32)EventId> {
+        using TSuccess = TSuccessRecord;
 
         // Error during work completion, i.e. rdma read timeout
         struct TWcErr {
@@ -34,42 +48,55 @@ namespace NInterconnect::NRdma {
         struct TCqErr {
         };
 
-        // Post wr error
+        // Post WR error
         struct TWrErr {
             int Code;
         };
 
-        static TEvRdmaIoDone* Success() {
-            return new TEvRdmaIoDone();
+        static TDerived* Success(TSuccess success) {
+            return new TDerived(std::move(success));
+        }
+
+        template <typename... TArgs>
+        static TDerived* Success(TArgs&&... args) {
+            return new TDerived(TSuccess{std::forward<TArgs>(args)...});
         }
 
-        static TEvRdmaIoDone* WcError(int code) {
-            return new TEvRdmaIoDone(code);
+        static TDerived* WcError(int code) {
+            return new TDerived(TWcErr{code});
         }
 
-        static TEvRdmaIoDone* CqError() {
-            return new TEvRdmaIoDone(TCqErr());
+        static TDerived* CqError() {
+            return new TDerived(TCqErr());
         }
 
-        static TEvRdmaIoDone* WrError(int code) {
-            return new TEvRdmaIoDone(TWrErr(code));
+        static TDerived* WrError(int code) {
+            return new TDerived(TWrErr{code});
         }
 
-        TEvRdmaIoDone()
+        TEvRdmaIoDoneCommon()
             : Record(TSuccess())
         {}
 
-        TEvRdmaIoDone(int errCode)
+        TEvRdmaIoDoneCommon(TSuccess success)
+            : Record(std::move(success))
+        {}
+
+        TEvRdmaIoDoneCommon(int errCode)
             : Record(TWcErr {
                 .Code = errCode,
             })
         {}
 
-        TEvRdmaIoDone(TCqErr err)
+        TEvRdmaIoDoneCommon(TWcErr err)
+            : Record(err)
+        {}
+
+        TEvRdmaIoDoneCommon(TCqErr err)
             : Record(err)
         {}
 
-        TEvRdmaIoDone(TWrErr err)
+        TEvRdmaIoDoneCommon(TWrErr err)
             : Record(err)
         {}
 
@@ -112,6 +139,16 @@ namespace NInterconnect::NRdma {
         std::variant<TSuccess, TWcErr, TCqErr, TWrErr> Record;
     };
 
+    struct TEvRdmaIoDone : public TEvRdmaIoDoneCommon<TEvRdmaIoDone, TSuccess, ERdma::EvRdmaIoDone> {
+        using TBase = TEvRdmaIoDoneCommon<TEvRdmaIoDone, TSuccess, ERdma::EvRdmaIoDone>;
+        using TBase::TBase;
+    };
+
+    struct TEvRdmaIoReceiveDone : public TEvRdmaIoDoneCommon<TEvRdmaIoReceiveDone, TSuccessReceive, ERdma::EvRdmaIoReceiveDone> {
+        using TBase = TEvRdmaIoDoneCommon<TEvRdmaIoReceiveDone, TSuccessReceive, ERdma::EvRdmaIoReceiveDone>;
+        using TBase::TBase;
+    };
+
     struct TEvRdmaReadDone : NActors::TEventLocal<TEvRdmaReadDone, (ui32)ERdma::EvRdmaReadDone> {
         std::unique_ptr<NInterconnect::NRdma::TEvRdmaIoDone> Event;
         const NActors::TMonotonic ReadScheduledTs;
diff --git a/ydb/library/actors/interconnect/rdma/rdma.cpp b/ydb/library/actors/interconnect/rdma/rdma.cpp
index 5847fbf9b32..ec6e490313c 100644
--- a/ydb/library/actors/interconnect/rdma/rdma.cpp
+++ b/ydb/library/actors/interconnect/rdma/rdma.cpp
@@ -41,17 +41,18 @@ public:
         : TSimpleCqBase(as, sz, c, true)
     {}
 
-    int Init(const TRdmaCtx* ctx, int maxCqe) noexcept {
-        return TSimpleCqBase::Init(ctx, maxCqe, nullptr);
+    int Init(const TRdmaCtx* ctx, const TRdmaRuntimeParams& params, std::shared_ptr<IMemPool> memPool) noexcept {
+        return TSimpleCqBase::Init(ctx, params, std::move(memPool), nullptr);
     }
 
     virtual ~TSimpleCq() {
         // 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())
+        if (Thread.Running()) {
             Thread.Join();
+        }
 
-        DestroyCq();
+        Y_UNUSED(DestroyCq());
     }
 };
 
@@ -61,13 +62,13 @@ public:
         : TSimpleCqBase(as, sz, c, false)
     {}
 
-    int Init(const TRdmaCtx* ctx, int maxCqe) noexcept {
+    int Init(const TRdmaCtx* ctx, const TRdmaRuntimeParams& params, std::shared_ptr<IMemPool> memPool) noexcept {
         CompChannel = ibv_create_comp_channel(ctx->GetContext());
         if (!CompChannel) {
             return errno;
         }
 
-        int err = TSimpleCqBase::Init(ctx, maxCqe, CompChannel);
+        int err = TSimpleCqBase::Init(ctx, params, std::move(memPool), CompChannel);
         if (err) {
             return err;
         }
@@ -80,6 +81,11 @@ public:
         return 0;
     }
 
+    void NotifyErr() noexcept override {
+        DoNotifyErr();
+        WakeUntilFinished();
+    }
+
     void Idle() noexcept override final {
         struct ibv_cq *evCq = nullptr;
         void *evCtx = nullptr;
@@ -87,7 +93,7 @@ public:
         int err = ibv_get_cq_event(CompChannel, &evCq, &evCtx);
         if (err) {
             if (errno != EINTR) {
-                NotifyErr();
+                DoNotifyErr();
                 return;
             }
         }
@@ -100,7 +106,7 @@ public:
         err = ibv_req_notify_cq(evCq, 0);
         if (err) {
             Cerr << "Couldn't request CQ notification\n" << Endl;
-            NotifyErr();
+            DoNotifyErr();
             Y_DEBUG_ABORT_UNLESS(false);
         }
     }
@@ -118,22 +124,16 @@ public:
         // 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())
+        if (Thread.Running()) {
+            WakeUntilFinished();
+            // 4. As usual, join and destroy CQ
             Thread.Join();
+        }
 
-        DestroyCq();
+        const int destroyErr = DestroyCq();
 
         // 5. Destroy completion event channel
-        if (ibv_destroy_comp_channel(CompChannel)) {
+        if (!destroyErr && CompChannel && 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
@@ -141,15 +141,15 @@ public:
         }
     }
 private:
-    ibv_comp_channel* CompChannel;
+    ibv_comp_channel* CompChannel = nullptr;
 };
 
-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 CreateSimpleCq(const TRdmaCtx* ctx, NActors::TActorSystem* as, TRdmaRuntimeParams runtimeParams, std::shared_ptr<IMemPool> memPool, NMonitoring::TDynamicCounters* counter) noexcept {
+    return CreateCq<TSimpleCq>(ctx, as, runtimeParams, std::move(memPool), 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);
+ICq::TPtr CreateSimpleEventDrivenCq(const TRdmaCtx* ctx, NActors::TActorSystem* as, TRdmaRuntimeParams runtimeParams, std::shared_ptr<IMemPool> memPool, NMonitoring::TDynamicCounters* counter) noexcept {
+    return CreateCq<TSimpleEventDrivenCq>(ctx, as, runtimeParams, std::move(memPool), counter);
 }
 
 const int TQueuePair::UnknownQpState = IBV_QPS_UNKNOWN;
@@ -174,10 +174,14 @@ int TQueuePair::Init(TRdmaCtx* ctx, ICq* icq, int maxWr) noexcept {
     bzero(&qpInitAttr, sizeof(qpInitAttr));
     qpInitAttr.send_cq = cq;
     qpInitAttr.recv_cq = cq;
+    qpInitAttr.srq = icq->GetSrq();
     qpInitAttr.cap.max_send_wr = static_cast<ui32>(maxWr);
-    qpInitAttr.cap.max_recv_wr = static_cast<ui32>(maxWr);
     qpInitAttr.cap.max_send_sge = static_cast<ui32>(attr.max_sge);
-    qpInitAttr.cap.max_recv_sge = static_cast<ui32>(attr.max_sge);
+    // With SRQ attached the QP has no private receive queue; receive capacity is owned by SRQ.
+    if (!qpInitAttr.srq) {
+        qpInitAttr.cap.max_recv_wr = static_cast<ui32>(maxWr);
+        qpInitAttr.cap.max_recv_sge = static_cast<ui32>(attr.max_sge);
+    }
     qpInitAttr.qp_type = IBV_QPT_RC;
 
     TStringStream ss;
@@ -369,19 +373,51 @@ void TIbVerbsBuilderImpl::AddReadVerb(void* mrAddr, ui32 mrlKey, void* dstAddr,
     );
 }
 
-ibv_send_wr* TIbVerbsBuilderImpl::BuildListOfVerbs(std::vector<TWr*>& wr) noexcept {
+void TIbVerbsBuilderImpl::AddSendVerb(TRcBuf packet,
+    std::function<void(NActors::TActorSystem* as, TEvRdmaIoDone*)> ioCb) noexcept
+{
+    Y_ABORT_UNLESS(packet);
+    Y_ABORT_UNLESS(packet.GetSize() <= Max<ui32>());
+
+    WorkBuf.emplace_back(
+        TWrVerbData {
+            .Sg = {
+                .addr = reinterpret_cast<ui64>(packet.GetData()),
+                .length = static_cast<ui32>(packet.GetSize()),
+            },
+            .Wr = {
+               .wr_id = 0/*wrId*/,
+               .sg_list = nullptr,
+               .num_sge = 1,
+               .opcode = IBV_WR_SEND,
+               .send_flags = IBV_SEND_SIGNALED,
+            },
+            .SendBuf = std::move(packet),
+            .IoCb = std::move(ioCb)
+        }
+    );
+}
+
+ibv_send_wr* TIbVerbsBuilderImpl::BuildListOfVerbs(std::vector<TWr*>& wr, size_t deviceIndex) noexcept {
     Y_ABORT_UNLESS(wr.size() == WorkBuf.size());
     Y_ABORT_UNLESS(wr.size());
 
-    WorkBuf[0].Wr.sg_list = &WorkBuf[0].Sg;
-    WorkBuf[0].Wr.wr_id = wr[0]->GetId();
-    wr[0]->AttachCb(std::move(WorkBuf[0]).IoCb);
+    auto attach = [&](size_t i) {
+        WorkBuf[i].Wr.sg_list = &WorkBuf[i].Sg;
+        WorkBuf[i].Wr.wr_id = wr[i]->GetId();
+        if (WorkBuf[i].SendBuf) {
+            auto memReg = TryExtractFromRcBuf(WorkBuf[i].SendBuf);
+            Y_ABORT_UNLESS(!memReg.Empty());
+            WorkBuf[i].Sg.lkey = memReg.GetLKey(deviceIndex);
+        }
+        wr[i]->AttachCb(std::move(WorkBuf[i].IoCb));
+    };
+
+    attach(0);
 
     for (size_t i = 1; i < WorkBuf.size(); i++) {
-        WorkBuf[i].Wr.sg_list = &WorkBuf[i].Sg;
         WorkBuf[i - 1].Wr.next = &WorkBuf[i].Wr;
-        WorkBuf[i].Wr.wr_id = wr[i]->GetId();
-        wr[i]->AttachCb(std::move(WorkBuf[i]).IoCb);
+        attach(i);
         wr[i]->ResetTimer();
     }
 
diff --git a/ydb/library/actors/interconnect/rdma/rdma.h b/ydb/library/actors/interconnect/rdma/rdma.h
index 2854a92e69b..398200a6ec9 100644
--- a/ydb/library/actors/interconnect/rdma/rdma.h
+++ b/ydb/library/actors/interconnect/rdma/rdma.h
@@ -3,11 +3,15 @@
 #include <util/generic/noncopyable.h>
 #include <util/generic/vector.h>
 #include <util/stream/output.h>
+#include <ydb/library/actors/core/actorid.h>
 
+#include <functional>
+#include <memory>
 #include <optional>
 
 struct ibv_qp;
 struct ibv_cq;
+struct ibv_srq;
 struct ibv_wc;
 union ibv_gid;
 struct ibv_send_wr;
@@ -22,17 +26,25 @@ namespace NMonitoring {
 }
 
 class IOutputStream;
+class TRcBuf;
 
 namespace NInterconnect::NRdma {
 
 class TRdmaCtx;
 class TCqCommon;
 class TCqActor;
+class IMemPool;
 struct TEvRdmaIoDone;
 
 class TQueuePair;
 class IIbVerbsBuilder;
 
+struct TRdmaRuntimeParams {
+    int MaxCqe;   // max capacity of single queue under CQ actor abstruction. -1 - use limit from rdma context
+    int MaxWr;    // max number of work request in the pool (for READ verbs)
+    int MaxSrqWr; // max number of work request for shared recieve queue
+    int RecieveBufSz; // Size of one wr buffer
+};
 
 // Wrapper for ibv Completion Queue
 // Hides logic to controll work request count
@@ -51,6 +63,7 @@ public:
     using TPtr = std::shared_ptr<ICq>;
     virtual ~ICq() = default;
     virtual ibv_cq* GetCq() noexcept = 0;
+    virtual ibv_srq* GetSrq() noexcept = 0;
 
     struct TBusy {}; // try later
     struct TErr {};  // fatal error, cq must be recreated. All associated qp failed.
@@ -65,6 +78,9 @@ public:
     virtual std::optional<TErr> DoWrBatchAsync(std::shared_ptr<TQueuePair> qp, std::unique_ptr<IIbVerbsBuilder> builder) noexcept = 0;
     virtual TWrStats GetWrStats() const noexcept = 0;
 
+    virtual bool RegisterQpAsync(ui32 qpNum, NActors::TActorId actorId) noexcept = 0;
+    virtual bool DeregisterQpAsync(ui32 qpNum) noexcept = 0;
+
     static bool IsWrSuccess(const TAllocResult& ar) {
         return std::holds_alternative<IWr*>(ar);
     }
@@ -78,8 +94,8 @@ private:
     virtual void NotifyErr() noexcept = 0;
 };
 
-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;
+ICq::TPtr CreateSimpleCq(const TRdmaCtx* ctx, NActors::TActorSystem* as, TRdmaRuntimeParams runtimeParams, std::shared_ptr<IMemPool> memPool, NMonitoring::TDynamicCounters* counter) noexcept;
+ICq::TPtr CreateSimpleEventDrivenCq(const TRdmaCtx* ctx, NActors::TActorSystem* as, TRdmaRuntimeParams runtimeParams, std::shared_ptr<IMemPool> memPool, NMonitoring::TDynamicCounters* counter) noexcept;
 
 struct THandshakeData {
     ui32 QpNum;
@@ -134,6 +150,8 @@ public:
     virtual ~IIbVerbsBuilder() = default;
     virtual void AddReadVerb(void* mrAddr, ui32 mrlKey, void* dstAddr, ui32 dstRkey, ui32 dstSize,
         std::function<void(NActors::TActorSystem* as, TEvRdmaIoDone*)> ioCb) noexcept = 0;
+    virtual void AddSendVerb(TRcBuf packet,
+        std::function<void(NActors::TActorSystem* as, TEvRdmaIoDone*)> ioCb) noexcept = 0;
 private:
     IIbVerbsBuilder() noexcept = default;
 };
diff --git a/ydb/library/actors/interconnect/rdma/rdma_impl.h b/ydb/library/actors/interconnect/rdma/rdma_impl.h
index 6de5e45de96..eda075ea491 100644
--- a/ydb/library/actors/interconnect/rdma/rdma_impl.h
+++ b/ydb/library/actors/interconnect/rdma/rdma_impl.h
@@ -3,18 +3,26 @@
 #include "ctx.h"
 #include "events.h"
 #include "rdma.h"
+#include "mem_pool.h"
 
 #include <contrib/libs/ibdrv/include/infiniband/verbs.h>
+#include <library/cpp/containers/absl/flat_hash_map.h>
 #include <ydb/library/actors/core/actorsystem.h>
 #include <library/cpp/monlib/metrics/metric_registry.h>
 #include <library/cpp/monlib/metrics/metric_sub_registry.h>
+#include <library/cpp/threading/queue/mpsc_read_as_filled.h>
+
+
 
 #include <util/thread/lfqueue.h>
+#include <util/system/guard.h>
 #include <util/system/spinlock.h>
 #include <util/system/thread.h>
+#include <util/system/yield.h>
 #include <util/system/sanitizers.h>
 #include <util/system/compiler.h>
 
+#include <cerrno>
 #include <span>
 
 namespace NInterconnect::NRdma {
@@ -49,13 +57,16 @@ public:
 
     void AddReadVerb(void* mrAddr, ui32 mrlKey, void* dstAddr, ui32 dstRkey, ui32 dstSize,
         std::function<void(NActors::TActorSystem* as, TEvRdmaIoDone*)> ioCb) noexcept;
+    void AddSendVerb(TRcBuf packet,
+        std::function<void(NActors::TActorSystem* as, TEvRdmaIoDone*)> ioCb) noexcept;
     size_t GetVerbsNum() const noexcept;
-    ibv_send_wr* BuildListOfVerbs(std::vector<TWr*>& preparedWr) noexcept;
+    ibv_send_wr* BuildListOfVerbs(std::vector<TWr*>& preparedWr, size_t deviceIndex) noexcept;
 
 private:
     struct TWrVerbData {
         ibv_sge Sg;
         ibv_send_wr Wr;
+        TRcBuf SendBuf;
         std::function<void(NActors::TActorSystem* as, TEvRdmaIoDone*)> IoCb;
     };
     std::vector<TWrVerbData> WorkBuf;
@@ -63,6 +74,312 @@ private:
 
 void SetSigHandler() noexcept;
 
+class TSrq {
+    struct TRecieveSlot {
+        TRcBuf Buffer;
+    };
+
+public:
+    static constexpr ui64 SRQ_WR_MASK = 1ull << 63;
+    static constexpr int CqTerminalError = -1;
+
+    struct TCmd {
+        ui32 QpNum;
+        enum ECmd {
+            RegQp,
+            DeregQp
+        } Cmd;
+        std::variant<NActors::TActorId> Target;
+    };
+
+    ibv_srq* Get() noexcept { return Srq; }
+
+    int Init(const TRdmaCtx* ctx, const TRdmaRuntimeParams& params, std::shared_ptr<IMemPool> memPool) noexcept {
+        if (!memPool || params.MaxSrqWr <= 0 || params.RecieveBufSz <= 0) {
+            return EINVAL;
+        }
+        MemPool = std::move(memPool);
+        RecieveBufSz = params.RecieveBufSz;
+        DeviceIndex = ctx->GetDeviceIndex();
+
+        struct ibv_srq_init_attr srqInitAttr;
+        memset(&srqInitAttr, 0, sizeof(srqInitAttr));
+
+        srqInitAttr.attr.max_wr = params.MaxSrqWr;
+        srqInitAttr.attr.max_sge = 1;
+
+        Srq = ibv_create_srq(ctx->GetProtDomain(), &srqInitAttr);
+        if (!Srq) {
+            return errno;
+        }
+
+        const ui64 maxWr = srqInitAttr.attr.max_wr;
+        if (maxWr == 0) {
+            if (const int destroyErr = Destroy()) {
+                return destroyErr;
+            }
+            return EINVAL;
+        }
+
+        Slots.reserve(maxWr);
+        PendingRefillSlots.reserve(maxWr);
+
+        for (ui64 i = 0; i < maxWr; i++) {
+            auto buffer = MemPool->AllocRcBuf(RecieveBufSz, IMemPool::EMPTY);
+            if (!buffer) {
+                if (const int destroyErr = Destroy()) {
+                    return destroyErr;
+                }
+                return ENOMEM;
+            }
+            Slots.emplace_back(TRecieveSlot{std::move(*buffer)});
+            auto region = TryExtractFromRcBuf(Slots.back().Buffer);
+            if (region.Empty()) {
+                if (const int destroyErr = Destroy()) {
+                    return destroyErr;
+                }
+                return EINVAL;
+            }
+        }
+
+        for (ui64 i = 0; i < maxWr; i++) {
+            if (const int err = PostSlot(i, Slots[i].Buffer)) {
+                if (const int destroyErr = Destroy()) {
+                    return destroyErr;
+                }
+                return err;
+            }
+        }
+
+        CommandsClosed = false;
+        return 0;
+    }
+
+    void ProcessCommands() noexcept {
+        for (;;) {
+            std::unique_ptr<TCmd> cmd(Queue.Pop());
+            if (!cmd) {
+                return;
+            }
+
+            switch (cmd->Cmd) {
+                case TCmd::RegQp:
+                    QpActors[cmd->QpNum] = std::get<NActors::TActorId>(cmd->Target);
+                    break;
+                case TCmd::DeregQp:
+                    QpActors.erase(cmd->QpNum);
+                    break;
+            }
+        }
+    }
+
+    void DrainCommands() noexcept {
+        for (;;) {
+            std::unique_ptr<TCmd> cmd(Queue.Pop());
+            if (!cmd) {
+                return;
+            }
+        }
+    }
+
+    bool HasPendingRefillSlots() const noexcept {
+        return !PendingRefillSlots.empty();
+    }
+
+    int RetryPendingRefillSlots(size_t maxSlots) noexcept {
+        for (size_t i = 0; i < maxSlots && !PendingRefillSlots.empty(); ++i) {
+            const ui64 id = PendingRefillSlots.back();
+            PendingRefillSlots.pop_back();
+
+            const auto refill = RefillSlot(id);
+            switch (refill.Status) {
+                case ERefillStatus::Ok:
+                    break;
+                case ERefillStatus::NoMemory:
+                    PendingRefillSlots.push_back(id);
+                    return 0;
+                case ERefillStatus::Fatal:
+                    return refill.Err;
+            }
+        }
+        return 0;
+    }
+
+    int HandleWc(NActors::TActorSystem* as, ibv_wc* wc) noexcept {
+        ui64 id = wc->wr_id & ~SRQ_WR_MASK;
+        Y_DEBUG_ABORT_UNLESS(id < Slots.size());
+        if (Y_UNLIKELY(id >= Slots.size())) {
+            return CqTerminalError;
+        }
+
+        auto it = QpActors.find(wc->qp_num);
+        const bool registered = it != QpActors.end();
+
+        TRecieveSlot& slot = Slots[id];
+        TRcBuf received;
+        int receiveErr = 0;
+        if (wc->status == IBV_WC_SUCCESS) {
+            if (Y_UNLIKELY(wc->byte_len > slot.Buffer.GetSize())) {
+                receiveErr = EMSGSIZE;
+            } else {
+                slot.Buffer.TrimBack(wc->byte_len);
+                received = std::move(slot.Buffer);
+            }
+        }
+
+        const auto refill = RefillSlot(id);
+        if (refill.Status == ERefillStatus::NoMemory) {
+            PendingRefillSlots.push_back(id);
+        }
+
+        if (!registered) {
+            if (refill.Status == ERefillStatus::Fatal) {
+                return refill.Err;
+            }
+            return receiveErr ? CqTerminalError : 0;
+        }
+
+        if (wc->status == IBV_WC_SUCCESS && !receiveErr) {
+            as->Send(it->second, TEvRdmaIoReceiveDone::Success(std::move(received)));
+        } else {
+            as->Send(it->second, TEvRdmaIoReceiveDone::WcError(receiveErr ? receiveErr : wc->status));
+        }
+
+        if (refill.Status == ERefillStatus::Fatal) {
+            return refill.Err;
+        }
+        return receiveErr ? CqTerminalError : 0;
+    }
+
+    void NotifyTerminalError(NActors::TActorSystem* as, int error) noexcept {
+        if (!error) {
+            return;
+        }
+        for (const auto& [qpNum, actorId] : QpActors) {
+            Y_UNUSED(qpNum);
+            if (error > 0) {
+                as->Send(actorId, TEvRdmaIoReceiveDone::WrError(error));
+            } else {
+                as->Send(actorId, TEvRdmaIoReceiveDone::CqError());
+            }
+        }
+        QpActors.clear();
+    }
+
+    bool EnqueueCmd(TCmd* cmd) noexcept {
+        std::unique_ptr<TCmd> holder(cmd);
+        TGuard<TSpinLock> guard(CommandsLock);
+        if (CommandsClosed) {
+            return false;
+        }
+        Queue.Push(holder.release());
+        return true;
+    }
+
+    void CloseCommands() noexcept {
+        TGuard<TSpinLock> guard(CommandsLock);
+        CommandsClosed = true;
+    }
+
+    int Destroy() noexcept {
+        CloseCommands();
+        DrainCommands();
+        if (Srq) {
+            if (const int err = ibv_destroy_srq(Srq)) {
+                Cerr << "Unable to destroy SRQ, err: " << err << ", errno: " << errno << Endl;
+                Y_DEBUG_ABORT_UNLESS(false);
+                return err;
+            }
+            Srq = nullptr;
+        }
+        Slots.clear();
+        PendingRefillSlots.clear();
+        QpActors.clear();
+        MemPool.reset();
+        RecieveBufSz = 0;
+        DeviceIndex = 0;
+        return 0;
+    }
+
+    ~TSrq() {
+        Y_UNUSED(Destroy());
+    }
+private:
+    enum class ERefillStatus {
+        Ok,
+        NoMemory,
+        Fatal,
+    };
+
+    struct TRefillResult {
+        ERefillStatus Status;
+        int Err = 0;
+    };
+
+    TRefillResult RefillSlot(ui64 id) noexcept {
+        if (Y_UNLIKELY(!MemPool || !Srq || id >= Slots.size())) {
+            return {ERefillStatus::Fatal, EINVAL};
+        }
+
+        auto buffer = MemPool->AllocRcBuf(RecieveBufSz, IMemPool::EMPTY);
+        if (!buffer) {
+            return {ERefillStatus::NoMemory, ENOMEM};
+        }
+
+        if (const int err = PostSlot(id, *buffer)) {
+            return {ERefillStatus::Fatal, err};
+        }
+
+        Slots[id].Buffer = std::move(*buffer);
+        return {ERefillStatus::Ok};
+    }
+
+    int PostSlot(ui64 id, TRcBuf& buffer) noexcept {
+        if (Y_UNLIKELY(!Srq || id >= Slots.size())) {
+            return EINVAL;
+        }
+
+        auto region = TryExtractFromRcBuf(buffer);
+        if (region.Empty()) {
+            return EINVAL;
+        }
+
+        ibv_sge sg = {
+            .addr = reinterpret_cast<ui64>(region.GetAddr()),
+            .length = region.GetSize(),
+            .lkey = region.GetLKey(DeviceIndex),
+        };
+        ibv_recv_wr wr = {
+            .wr_id = SRQ_WR_MASK | id,
+            .sg_list = &sg,
+            .num_sge = 1,
+        };
+        ibv_recv_wr* badWr = nullptr;
+        if (const int err = ibv_post_srq_recv(Srq, &wr, &badWr)) {
+            Y_DEBUG_ABORT_UNLESS(badWr == &wr, "unexpected bad wr for single SRQ recv post");
+            return err;
+        }
+        Y_DEBUG_ABORT_UNLESS(!badWr, "bad wr must not be set on successful SRQ recv post");
+        if (Y_UNLIKELY(badWr)) {
+            return EIO;
+        }
+
+        return 0;
+    }
+
+    ibv_srq* Srq = nullptr;
+    std::shared_ptr<IMemPool> MemPool;
+    int RecieveBufSz = 0;
+    size_t DeviceIndex = 0;
+    std::vector<TRecieveSlot> Slots;
+    std::vector<ui64> PendingRefillSlots;
+    NThreading::TReadAsFilledQueue<TCmd> Queue;
+    TSpinLock CommandsLock;
+    bool CommandsClosed = true;
+    // TODO: replace with a paged radix/direct map for faster qp_num -> actor lookup on the receive hot path.
+    absl::flat_hash_map<ui32, NActors::TActorId> QpActors;
+};
+
 class TCqCommon : public ICq {
 public:
     TCqCommon(NActors::TActorSystem* as)
@@ -78,11 +395,23 @@ public:
         return Cq;
     }
 
-    int Init(const TRdmaCtx* ctx, int maxCqe, struct ibv_comp_channel* ch) noexcept {
-        Cq = ibv_create_cq(ctx->GetContext(), maxCqe, nullptr, ch, 0);
+    ibv_srq* GetSrq() noexcept {
+        return Srq.Get();
+    }
+
+    int Init(const TRdmaCtx* ctx, const TRdmaRuntimeParams& params, std::shared_ptr<IMemPool> memPool, struct ibv_comp_channel* ch) noexcept {
+        Cq = ibv_create_cq(ctx->GetContext(), params.MaxCqe, nullptr, ch, 0);
         if (!Cq) {
             return errno;
         }
+        if (params.MaxSrqWr > 0) {
+            if (const int err = Srq.Init(ctx, params, std::move(memPool))) {
+                if (const int destroyErr = DestroyCq()) {
+                    return destroyErr;
+                }
+                return err;
+            }
+        }
         return 0;
     }
 
@@ -94,14 +423,25 @@ public:
         SpinLockPause();
     }
 
-    void DestroyCq () noexcept {
+    int DestroyCq() noexcept {
+        if (const int err = Srq.Destroy()) {
+            return err;
+        }
         if (Cq) {
-            ibv_destroy_cq(Cq);
+            if (const int err = ibv_destroy_cq(Cq)) {
+                Cerr << "Unable to destroy CQ, err: " << err << ", errno: " << errno << Endl;
+                Y_DEBUG_ABORT_UNLESS(false);
+                return err;
+            }
+            Cq = nullptr;
         }
+        return 0;
     }
+
 protected:
     NActors::TActorSystem* const As;
     ibv_cq* Cq;
+    TSrq Srq;
 };
 
 class TWr : public ICq::IWr {
@@ -179,16 +519,28 @@ private:
 };
 
 template<class TCq>
-static ICq::TPtr CreateCq(const TRdmaCtx* ctx, NActors::TActorSystem* as, int maxCqe, int maxWr, NMonitoring::TDynamicCounters* counter) noexcept {
-    if (maxCqe <= 0) {
-        const ibv_device_attr& attr = ctx->GetDevAttr();
-        maxCqe = attr.max_cqe;
+static ICq::TPtr CreateCq(const TRdmaCtx* ctx, NActors::TActorSystem* as, TRdmaRuntimeParams runtimeParams, std::shared_ptr<IMemPool> memPool, NMonitoring::TDynamicCounters* counter) noexcept {
+    const ibv_device_attr& attr = ctx->GetDevAttr();
+    if (runtimeParams.MaxCqe <= 0) {
+        runtimeParams.MaxCqe = attr.max_cqe;
+    }
+    if (runtimeParams.MaxSrqWr < 0) {
+        runtimeParams.MaxSrqWr = attr.max_srq_wr;
     }
-    if (maxWr <= 0) {
-        maxWr = maxCqe;
+    if (runtimeParams.MaxWr <= 0) {
+        runtimeParams.MaxWr = runtimeParams.MaxCqe;
+    }
+
+    if (runtimeParams.MaxSrqWr > 0) {
+        if (attr.max_srq <= 0 || attr.max_srq_wr <= 0 || attr.max_srq_sge < 1) {
+            return nullptr; // or fail CQ creation
+        }
+
+        runtimeParams.MaxSrqWr = Min(runtimeParams.MaxSrqWr, attr.max_srq_wr);
     }
-    auto p = std::make_shared<TCq>(as, maxWr, counter);
-    int err = p->Init(ctx, maxCqe);
+
+    auto p = std::make_shared<TCq>(as, runtimeParams.MaxWr, counter);
+    int err = p->Init(ctx, runtimeParams, std::move(memPool));
     if (err) {
         return nullptr;
     }
@@ -212,7 +564,7 @@ protected:
         }
 
         ibv_send_wr* BuildListOfVerbs(std::vector<TWr*>& preparedWr) noexcept {
-            return static_cast<TIbVerbsBuilderImpl*>(VerbsBuilder.get())->BuildListOfVerbs(preparedWr);
+            return static_cast<TIbVerbsBuilderImpl*>(VerbsBuilder.get())->BuildListOfVerbs(preparedWr, Qp->GetDeviceIndex());
         }
 
         std::shared_ptr<TQueuePair> Qp;
@@ -224,7 +576,6 @@ public:
         : TCqCommon(as)
         , Thread(ThreadFunc, this)
         , Finished(false)
-        , Err(false)
         , NonBlockingPolling(nonBlockingPolling)
     {
         auto counter = MakeCounters(c);
@@ -248,8 +599,20 @@ public:
         Queue.Enqueue(static_cast<TWr*>(wr));
     }
 
+protected:
+    // Internal terminal transition. It may be called from the CQ poller thread,
+    // so it must not try to wake or wait for the poller.
+    void DoNotifyErr() noexcept {
+        SetTerminalError(TSrq::CqTerminalError);
+    }
+
+public:
     void NotifyErr() noexcept override {
-        Err.store(true, std::memory_order_relaxed);
+        DoNotifyErr();
+    }
+
+    void SetTerminalError(int error) noexcept {
+        TerminalError.store(error ? error : TSrq::CqTerminalError, std::memory_order_relaxed);
     }
 
     TWrStats GetWrStats() const noexcept override {
@@ -262,7 +625,7 @@ public:
     }
 
     std::optional<TErr> DoWrBatchAsync(std::shared_ptr<TQueuePair> qp, std::unique_ptr<IIbVerbsBuilder> builder) noexcept override {
-        if (Err.load(std::memory_order_relaxed)) {
+        if (TerminalError.load(std::memory_order_relaxed)) {
             return TErr();
         }
         Waiters.Enqueue(new TWaiterCtx(std::move(qp), std::move(builder)));
@@ -283,8 +646,24 @@ public:
         return {};
     }
 
-    // Build RDMA verbs and post it
-    // Returns false if it safe to sleep to wait for cq event
+    bool RegisterQpAsync(ui32 qpNum, NActors::TActorId actorId) noexcept override {
+        if (TerminalError.load(std::memory_order_relaxed)) {
+            return false;
+        }
+        auto cmd = new TSrq::TCmd{qpNum, TSrq::TCmd::RegQp, actorId};
+        return Srq.EnqueueCmd(cmd);
+    }
+
+    bool DeregisterQpAsync(ui32 qpNum) noexcept override {
+        if (TerminalError.load(std::memory_order_relaxed)) {
+            return false;
+        }
+        auto cmd = new TSrq::TCmd{qpNum, TSrq::TCmd::DeregQp, NActors::TActorId()};
+        return Srq.EnqueueCmd(cmd);
+    }
+
+    // Builds and posts pending RDMA send WRs.
+    // Returns false when there is no pending send work and the CQ thread may idle.
     bool ProcessWr(std::unique_ptr<TWaiterCtx>& ctx, std::vector<TWr*>& preparedWr, bool tryBuildAtOnce) noexcept {
         while (true) {
             if (ctx) {
@@ -302,7 +681,7 @@ public:
 
                     ibv_send_wr* wrList = ctx->BuildListOfVerbs(preparedWr);
 
-                    if (Err.load(std::memory_order_relaxed)) {
+                    if (TerminalError.load(std::memory_order_relaxed)) {
                         for (auto x : preparedWr) {
                             x->ReplyCqErr(As);
                         }
@@ -351,14 +730,14 @@ public:
         while (Cont.load(std::memory_order_relaxed)) {
             const constexpr size_t wcBatchSize = 16;
             std::array<ibv_wc, wcBatchSize> wcs;
-            if (Err.load(std::memory_order_relaxed)) {
+            if (TerminalError.load(std::memory_order_relaxed)) {
                 HandleErr();
                 Cont.store(false, std::memory_order_relaxed);
             } else {
                 int rv = Do(wcs);
                 if (rv < 0) {
                     //TODO: Is it correct err handling?
-                    Err.store(true, std::memory_order_relaxed);
+                    DoNotifyErr();
                 } else if (rv == 0) {
                     bool idleAllowed = false;
                     MaybeIdle.store(true);
@@ -369,8 +748,27 @@ public:
                         }
                         VerbsBuildingState.Lock.Release();
                     }
+                    if (Srq.HasPendingRefillSlots()) {
+                        static constexpr ui64 SrqRefillRetryPeriod = 1024;
+                        if ((++SrqRefillRetryCounter & (SrqRefillRetryPeriod - 1)) == 0) {
+                            if (auto error = Srq.RetryPendingRefillSlots(1)) {
+                                SetTerminalError(error);
+                                idleAllowed = false;
+                                MaybeIdle.store(false);
+                            }
+                        }
+                        if (Srq.HasPendingRefillSlots()) {
+                            idleAllowed = false;
+                            MaybeIdle.store(false);
+                            SpinLockPause();
+                        }
+                    } else {
+                        SrqRefillRetryCounter = 0;
+                    }
                     if (idleAllowed) {
-                        Idle();
+                        if (!TerminalError.load(std::memory_order_relaxed)) {
+                            Idle();
+                        }
                         MaybeIdle.store(false);
                     }
                 } else {
@@ -383,6 +781,10 @@ public:
     }
 
     void HandleErr() noexcept {
+        Srq.CloseCommands();
+        Srq.ProcessCommands();
+        const int terminalError = TerminalError.load(std::memory_order_relaxed);
+        Srq.NotifyTerminalError(As, terminalError);
         for (size_t i = 0; i < WrBuf.size(); i++) {
             TWr* wr = &WrBuf[i];
             wr->ReplyCqErr(As);
@@ -391,13 +793,22 @@ public:
         }
     }
 
+
+
     void HandleWc(ibv_wc* wc, size_t sz) noexcept {
         for (size_t i = 0; i < sz; i++, wc++) {
-            TWr* wr = &WrBuf[wc->wr_id];
-            double passed = wr->GetTimePassed();
-            RdmaDeviceVerbTimeUs->Collect(passed * 1000000.0);
-            wr->Reply(As, wc);
-            ReturnWr(wr);
+            if (wc->wr_id & TSrq::SRQ_WR_MASK) {
+                Srq.ProcessCommands();
+                if (auto error = Srq.HandleWc(As, wc)) {
+                    SetTerminalError(error);
+                }
+            } else {
+                TWr* wr = &WrBuf[wc->wr_id];
+                double passed = wr->GetTimePassed();
+                RdmaDeviceVerbTimeUs->Collect(passed * 1000000.0);
+                wr->Reply(As, wc);
+                ReturnWr(wr);
+            }
         }
     }
 
@@ -413,10 +824,32 @@ public:
     }
 
     void Awake() noexcept {
-        pthread_kill(CqThreadId, SIGUSR1);
+        if (CqThreadId) {
+            pthread_kill(CqThreadId, SIGUSR1);
+        }
     }
 
 protected:
+    bool IsCqThread() const noexcept {
+        return CqThreadId && pthread_equal(pthread_self(), CqThreadId);
+    }
+
+    // External terminal notifications may arrive through async_fd while the poller
+    // sleeps in ibv_get_cq_event() on a different completion channel. Since a
+    // signal can be delivered just before the blocking syscall, repeat wakeups
+    // until the poller observes TerminalError and exits the loop.
+    void WakeUntilFinished() noexcept {
+        if (!Thread.Running() || IsCqThread()) {
+            return;
+        }
+        while (!Finished.load(std::memory_order_relaxed)) {
+            Awake();
+            if (Finished.load(std::memory_order_relaxed)) {
+                break;
+            }
+            ThreadYield();
+        }
+    }
     TThread Thread;
     std::atomic<bool> Finished;
     std::atomic<bool> Cont;
@@ -438,12 +871,13 @@ protected:
 
     std::atomic<bool> MaybeIdle = false;
 
-    std::atomic<bool> Err;
+    std::atomic<int> TerminalError = 0;
     const bool NonBlockingPolling;
     std::atomic<ui64> Allocated;
+    ui64 SrqRefillRetryCounter = 0;
     NMonitoring::THistogramPtr RdmaDeviceVerbTimeUs;
 private:
-    pthread_t CqThreadId;
+    pthread_t CqThreadId = {};
 };
 
 }
diff --git a/ydb/library/actors/interconnect/rdma/ut/rdma_low_ut.cpp b/ydb/library/actors/interconnect/rdma/ut/rdma_low_ut.cpp
index dc4d8a01a3d..d6065084fdf 100644
--- a/ydb/library/actors/interconnect/rdma/ut/rdma_low_ut.cpp
+++ b/ydb/library/actors/interconnect/rdma/ut/rdma_low_ut.cpp
@@ -1,11 +1,21 @@
 #include <util/thread/pool.h>
+#include <util/generic/hash_set.h>
+#include <util/system/thread.h>
+#include <util/system/yield.h>
 
+#include <algorithm>
+#include <array>
+#include <atomic>
 #include <string.h>
+#include <vector>
 
 #include <contrib/libs/ibdrv/include/infiniband/verbs.h>
+#include <ydb/library/actors/core/actor_bootstrapped.h>
+#include <ydb/library/actors/core/hfunc.h>
 #include <ydb/library/actors/interconnect/rdma/ctx.h>
 #include <ydb/library/actors/interconnect/rdma/events.h>
 #include <ydb/library/actors/interconnect/rdma/rdma.h>
+#include <ydb/library/actors/interconnect/rdma/rdma_impl.h>
 #include <ydb/library/actors/interconnect/rdma/mem_pool.h>
 #include <ydb/library/actors/interconnect/address/interconnect_address.h>
 #include <ydb/library/actors/interconnect/poller/poller_actor.h>
@@ -24,6 +34,204 @@ static const size_t MEM_REG_SZ = 4096;
 class TRdmaLow : public TSkipFixture {};
 class TCqMode : public TSkipFixtureWithParams<NInterconnect::NRdma::ECqMode> {};
 
+struct TRegistrationTestCq {
+    THolder<TTestActorRuntimeBase> ActorSystem;
+    TRdmaCtx* Ctx = nullptr;
+    ICq::TPtr Cq;
+};
+
+class TReceiveDoneCounterActor : public TActorBootstrapped<TReceiveDoneCounterActor> {
+public:
+    explicit TReceiveDoneCounterActor(std::shared_ptr<std::atomic<ui32>> counter)
+        : Counter(std::move(counter))
+    {}
+
+    void Bootstrap() {
+        Become(&TReceiveDoneCounterActor::StateFunc);
+    }
+
+    void Handle(TEvRdmaIoReceiveDone::TPtr&) {
+        Counter->fetch_add(1, std::memory_order_relaxed);
+    }
+
+    void Handle(TEvents::TEvPing::TPtr& ev) {
+        Send(ev->Sender, new TEvents::TEvPong());
+    }
+
+    STRICT_STFUNC(StateFunc,
+        hFunc(TEvRdmaIoReceiveDone, Handle);
+        hFunc(TEvents::TEvPing, Handle);
+    )
+
+private:
+    std::shared_ptr<std::atomic<ui32>> Counter;
+};
+
+struct TEvSendReceiveProbeResult : public TEventLocal<TEvSendReceiveProbeResult, EventSpaceBegin(TEvents::ES_PRIVATE) + 1> {
+    bool Success = false;
+    TString ErrSource;
+    TString Payload;
+    TDuration Latency;
+
+    TEvSendReceiveProbeResult(bool success, TString errSource, TString payload, TDuration latency)
+        : Success(success)
+        , ErrSource(std::move(errSource))
+        , Payload(std::move(payload))
+        , Latency(latency)
+    {}
+};
+
+struct TEvTriggerCqTerminalError : public TEventLocal<TEvTriggerCqTerminalError, EventSpaceBegin(TEvents::ES_PRIVATE) + 2> {
+    std::shared_ptr<std::atomic<bool>> Ready;
+    std::shared_ptr<std::atomic<bool>> Start;
+
+    TEvTriggerCqTerminalError() = default;
+
+    TEvTriggerCqTerminalError(std::shared_ptr<std::atomic<bool>> ready, std::shared_ptr<std::atomic<bool>> start)
+        : Ready(std::move(ready))
+        , Start(std::move(start))
+    {}
+};
+
+struct TEvCqTerminalErrorTriggered : public TEventLocal<TEvCqTerminalErrorTriggered, EventSpaceBegin(TEvents::ES_PRIVATE) + 3> {
+    bool Success = false;
+    TDuration NotifyErrDuration;
+
+    TEvCqTerminalErrorTriggered(bool success, TDuration notifyErrDuration)
+        : Success(success)
+        , NotifyErrDuration(notifyErrDuration)
+    {}
+};
+
+struct TEvReceiveTerminalProbeResult : public TEventLocal<TEvReceiveTerminalProbeResult, EventSpaceBegin(TEvents::ES_PRIVATE) + 4> {
+    ui32 QpNum = 0;
+    TString ErrSource;
+
+    TEvReceiveTerminalProbeResult(ui32 qpNum, TString errSource)
+        : QpNum(qpNum)
+        , ErrSource(std::move(errSource))
+    {}
+};
+
+class TReceiveDoneProbeActor : public TActorBootstrapped<TReceiveDoneProbeActor> {
+public:
+    TReceiveDoneProbeActor(TActorId edge, std::shared_ptr<std::atomic<ui64>> sendTsUs)
+        : Edge(edge)
+        , SendTsUs(std::move(sendTsUs))
+    {}
+
+    void Bootstrap() {
+        Become(&TReceiveDoneProbeActor::StateFunc);
+    }
+
+    void Handle(TEvRdmaIoReceiveDone::TPtr& ev) {
+        const ui64 sendTsUs = SendTsUs->load(std::memory_order_acquire);
+        const TDuration latency = TInstant::Now() - TInstant::MicroSeconds(sendTsUs);
+
+        TString payload;
+        if (ev->Get()->IsSuccess()) {
+            const auto& received = std::get<TEvRdmaIoReceiveDone::TSuccess>(ev->Get()->Record).Buf;
+            payload = TString(received.GetData(), received.GetSize());
+        }
+
+        Send(Edge, new TEvSendReceiveProbeResult(ev->Get()->IsSuccess(), TString(ev->Get()->GetErrSource()), std::move(payload), latency));
+    }
+
+    STRICT_STFUNC(StateFunc,
+        hFunc(TEvRdmaIoReceiveDone, Handle);
+    )
+
+private:
+    TActorId Edge;
+    std::shared_ptr<std::atomic<ui64>> SendTsUs;
+};
+
+class TReceiveTerminalProbeActor : public TActorBootstrapped<TReceiveTerminalProbeActor> {
+public:
+    TReceiveTerminalProbeActor(TActorId edge, ui32 qpNum)
+        : Edge(edge)
+        , QpNum(qpNum)
+    {}
+
+    void Bootstrap() {
+        Become(&TReceiveTerminalProbeActor::StateFunc);
+    }
+
+    void Handle(TEvRdmaIoReceiveDone::TPtr& ev) {
+        Send(Edge, new TEvReceiveTerminalProbeResult(QpNum, TString(ev->Get()->GetErrSource())));
+    }
+
+    STRICT_STFUNC(StateFunc,
+        hFunc(TEvRdmaIoReceiveDone, Handle);
+    )
+
+private:
+    TActorId Edge;
+    ui32 QpNum;
+};
+
+class TNotifyCqTerminalErrorActor : public TActorBootstrapped<TNotifyCqTerminalErrorActor> {
+public:
+    TNotifyCqTerminalErrorActor(ICq::TPtr cq, TActorId edge)
+        : Cq(std::move(cq))
+        , Edge(edge)
+    {}
+
+    void Bootstrap() {
+        Become(&TNotifyCqTerminalErrorActor::StateFunc);
+    }
+
+    void Handle(TEvTriggerCqTerminalError::TPtr& ev) {
+        if (ev->Get()->Ready && ev->Get()->Start) {
+            ev->Get()->Ready->store(true, std::memory_order_release);
+            while (!ev->Get()->Start->load(std::memory_order_acquire)) {
+                ThreadYield();
+            }
+        }
+        auto* cqImpl = dynamic_cast<TSimpleCqBase*>(Cq.get());
+        const TInstant start = TInstant::Now();
+        if (cqImpl) {
+            cqImpl->NotifyErr();
+        }
+        Send(Edge, new TEvCqTerminalErrorTriggered(cqImpl != nullptr, TInstant::Now() - start));
+    }
+
+    STRICT_STFUNC(StateFunc,
+        hFunc(TEvTriggerCqTerminalError, Handle);
+    )
+
+private:
+    ICq::TPtr Cq;
+    TActorId Edge;
+};
+
+static TRegistrationTestCq CreateRegistrationTestCq(TString bindTo, ECqMode mode, TRdmaRuntimeParams params) {
+    auto [actorSystem, ctx] = PrepareTestRuntime(bindTo);
+    auto memPool = CreateDummyMemPool();
+
+    ICq::TPtr cq;
+    switch (mode) {
+        case ECqMode::POLLING:
+            cq = CreateSimpleCq(ctx, actorSystem->GetActorSystem(0), params, std::move(memPool), nullptr);
+            break;
+        case ECqMode::EVENT:
+            cq = CreateSimpleEventDrivenCq(ctx, actorSystem->GetActorSystem(0), params, std::move(memPool), nullptr);
+            break;
+    }
+
+    return TRegistrationTestCq{
+        .ActorSystem = std::move(actorSystem),
+        .Ctx = ctx,
+        .Cq = std::move(cq),
+    };
+}
+
+static void NotifyCqTerminalError(ICq::TPtr cq) {
+    auto* cqImpl = dynamic_cast<TSimpleCqBase*>(cq.get());
+    ASSERT_TRUE(cqImpl);
+    cqImpl->NotifyErr();
+}
+
 static NInterconnect::NRdma::TMemRegionPtr AllocSourceRegion(std::shared_ptr<IMemPool> memPool) {
     auto reg = memPool->Alloc(MEM_REG_SZ, IMemPool::EMPTY);
     memset(reg->GetAddr(), 0, MEM_REG_SZ);
@@ -32,6 +240,22 @@ static NInterconnect::NRdma::TMemRegionPtr AllocSourceRegion(std::shared_ptr<IMe
     return reg;
 }
 
+static void ConnectQps(TRdmaCtx* ctx, TQueuePair& qp1, TQueuePair& qp2) {
+    ASSERT_EQ(qp1.ToRtsState(THandshakeData{
+        .QpNum = qp2.GetQpNum(),
+        .SubnetPrefix = ctx->GetGid().global.subnet_prefix,
+        .InterfaceId = ctx->GetGid().global.interface_id,
+        .MtuIndex = ctx->GetPortAttr().active_mtu,
+    }), 0);
+
+    ASSERT_EQ(qp2.ToRtsState(THandshakeData{
+        .QpNum = qp1.GetQpNum(),
+        .SubnetPrefix = ctx->GetGid().global.subnet_prefix,
+        .InterfaceId = ctx->GetGid().global.interface_id,
+        .MtuIndex = ctx->GetPortAttr().active_mtu,
+    }), 0);
+}
+
 void DoReadInOneProcess(TString bindTo, NInterconnect::NRdma::ECqMode mode) {
     auto rdma = InitLocalRdmaStuff(bindTo, mode);
 
@@ -43,6 +267,135 @@ void DoReadInOneProcess(TString bindTo, NInterconnect::NRdma::ECqMode mode) {
     ASSERT_TRUE(strncmp((char*)reg1->GetAddr(), (char*)reg2->GetAddr(), MEM_REG_SZ) == 0);
 }
 
+void DoSendReceiveInOneProcess(TString bindTo, NInterconnect::NRdma::ECqMode mode) {
+    static constexpr ui32 ReceiveBufSz = 1024;
+    auto rdma = CreateRegistrationTestCq(bindTo, mode, TRdmaRuntimeParams{
+        .MaxCqe = 16,
+        .MaxWr = 4,
+        .MaxSrqWr = 8,
+        .RecieveBufSz = ReceiveBufSz,
+    });
+    ASSERT_TRUE(rdma.Cq);
+    ASSERT_TRUE(rdma.Cq->GetSrq());
+
+    auto senderQp = std::make_shared<TQueuePair>();
+    auto receiverQp = std::make_shared<TQueuePair>();
+    ASSERT_EQ(senderQp->Init(rdma.Ctx, rdma.Cq.get(), 16), 0);
+    ASSERT_EQ(receiverQp->Init(rdma.Ctx, rdma.Cq.get(), 16), 0);
+    ConnectQps(rdma.Ctx, *senderQp, *receiverQp);
+
+    const TActorId edge = rdma.ActorSystem->AllocateEdgeActor(0);
+    auto sendTsUs = std::make_shared<std::atomic<ui64>>(0);
+    const TActorId receiverActor = rdma.ActorSystem->Register(new TReceiveDoneProbeActor(edge, sendTsUs));
+    ASSERT_TRUE(rdma.Cq->RegisterQpAsync(receiverQp->GetQpNum(), receiverActor));
+
+    const TString payload = "RDMA_SEND_RECEIVE_LOW_LEVEL_TEST";
+    ASSERT_LT(payload.size(), ReceiveBufSz);
+
+    auto sendMemPool = CreateDummyMemPool();
+    auto sendRegion = sendMemPool->Alloc(payload.size(), IMemPool::EMPTY);
+    ASSERT_TRUE(sendRegion);
+    memcpy(sendRegion->GetAddr(), payload.data(), payload.size());
+
+    struct TSendResult {
+        int Err = 0;
+        bool BadWr = false;
+    };
+
+    auto sendPromise = NThreading::NewPromise<TSendResult>();
+    auto sendFuture = sendPromise.GetFuture();
+    TThread senderThread([sendPromise, senderQp, sendRegion, sendTsUs, deviceIndex = rdma.Ctx->GetDeviceIndex(), payloadSize = payload.size()]() mutable {
+        ibv_sge sg = {
+            .addr = reinterpret_cast<ui64>(sendRegion->GetAddr()),
+            .length = static_cast<ui32>(payloadSize),
+            .lkey = sendRegion->GetLKey(deviceIndex),
+        };
+        ibv_send_wr wr = {
+            .sg_list = &sg,
+            .num_sge = 1,
+            .opcode = IBV_WR_SEND,
+        };
+        ibv_send_wr* badWr = nullptr;
+
+        TSendResult result;
+        sendTsUs->store(TInstant::Now().MicroSeconds(), std::memory_order_release);
+        result.Err = senderQp->PostSend(&wr, &badWr);
+        result.BadWr = badWr != nullptr;
+        sendPromise.SetValue(result);
+    });
+    senderThread.Start();
+
+    const TSendResult sendResult = sendFuture.GetValueSync();
+    senderThread.Join();
+    ASSERT_EQ(sendResult.Err, 0);
+    ASSERT_FALSE(sendResult.BadWr);
+
+    auto ev = rdma.ActorSystem->GrabEdgeEvent<TEvSendReceiveProbeResult>(edge, TDuration::Seconds(5));
+    ASSERT_TRUE(ev);
+    ASSERT_TRUE(ev->Get()->Success) << ev->Get()->ErrSource;
+
+    Cerr << "RDMA send/receive actor latency: " << ev->Get()->Latency.MicroSeconds() << " us" << Endl;
+    ::testing::Test::RecordProperty("SendReceiveActorLatencyUs", ev->Get()->Latency.MicroSeconds());
+    EXPECT_EQ(ev->Get()->Payload, payload);
+
+    EXPECT_TRUE(rdma.Cq->DeregisterQpAsync(receiverQp->GetQpNum()));
+}
+
+void DoSendReceiveViaBuilderInOneProcess(TString bindTo, NInterconnect::NRdma::ECqMode mode) {
+    static constexpr ui32 ReceiveBufSz = 1024;
+    auto rdma = CreateRegistrationTestCq(bindTo, mode, TRdmaRuntimeParams{
+        .MaxCqe = 16,
+        .MaxWr = 4,
+        .MaxSrqWr = 8,
+        .RecieveBufSz = ReceiveBufSz,
+    });
+    ASSERT_TRUE(rdma.Cq);
+    ASSERT_TRUE(rdma.Cq->GetSrq());
+
+    auto senderQp = std::make_shared<TQueuePair>();
+    auto receiverQp = std::make_shared<TQueuePair>();
+    ASSERT_EQ(senderQp->Init(rdma.Ctx, rdma.Cq.get(), 16), 0);
+    ASSERT_EQ(receiverQp->Init(rdma.Ctx, rdma.Cq.get(), 16), 0);
+    ConnectQps(rdma.Ctx, *senderQp, *receiverQp);
+
+    const TActorId edge = rdma.ActorSystem->AllocateEdgeActor(0);
+    auto sendTsUs = std::make_shared<std::atomic<ui64>>(0);
+    const TActorId receiverActor = rdma.ActorSystem->Register(new TReceiveDoneProbeActor(edge, sendTsUs));
+    ASSERT_TRUE(rdma.Cq->RegisterQpAsync(receiverQp->GetQpNum(), receiverActor));
+
+    const TString payload = "RDMA_SEND_RECEIVE_BUILDER_TEST";
+    ASSERT_LT(payload.size(), ReceiveBufSz);
+
+    auto sendMemPool = CreateDummyMemPool();
+    auto sendBuf = sendMemPool->AllocRcBuf(payload.size(), IMemPool::EMPTY);
+    ASSERT_TRUE(sendBuf);
+    memcpy(sendBuf->UnsafeGetDataMut(), payload.data(), payload.size());
+
+    auto sendPromise = NThreading::NewPromise<TEvRdmaIoDone*>();
+    auto sendFuture = sendPromise.GetFuture();
+
+    auto builder = CreateIbVerbsBuilder(1);
+    ASSERT_TRUE(builder);
+    builder->AddSendVerb(*sendBuf, [sendPromise](TActorSystem*, TEvRdmaIoDone* ev) mutable {
+        sendPromise.SetValue(ev);
+    });
+
+    sendTsUs->store(TInstant::Now().MicroSeconds(), std::memory_order_release);
+    auto submitErr = rdma.Cq->DoWrBatchAsync(senderQp, std::move(builder));
+    ASSERT_FALSE(submitErr);
+
+    auto receiveEv = rdma.ActorSystem->GrabEdgeEvent<TEvSendReceiveProbeResult>(edge, TDuration::Seconds(5));
+    ASSERT_TRUE(receiveEv);
+    ASSERT_TRUE(receiveEv->Get()->Success) << receiveEv->Get()->ErrSource;
+    EXPECT_EQ(receiveEv->Get()->Payload, payload);
+
+    ASSERT_TRUE(sendFuture.Wait(TDuration::Seconds(5)));
+    std::unique_ptr<TEvRdmaIoDone> sendDone(sendFuture.GetValueSync());
+    ASSERT_TRUE(sendDone->IsSuccess()) << sendDone->GetErrSource();
+
+    EXPECT_TRUE(rdma.Cq->DeregisterQpAsync(receiverQp->GetQpNum()));
+}
+
 TEST_P(TCqMode, ReadInOneProcessIpV4) {
     DoReadInOneProcess("127.0.0.1", GetParam());
 }
@@ -51,6 +404,207 @@ TEST_P(TCqMode, ReadInOneProcessIpV6) {
     DoReadInOneProcess("::1", GetParam());
 }
 
+TEST_P(TCqMode, SendReceiveInOneProcessIpV4) {
+    DoSendReceiveInOneProcess("127.0.0.1", GetParam());
+}
+
+TEST_P(TCqMode, SendReceiveInOneProcessIpV6) {
+    DoSendReceiveInOneProcess("::1", GetParam());
+}
+
+TEST_P(TCqMode, SendReceiveViaBuilderInOneProcessIpV4) {
+    DoSendReceiveViaBuilderInOneProcess("127.0.0.1", GetParam());
+}
+
+TEST_P(TCqMode, SendReceiveViaBuilderInOneProcessIpV6) {
+    DoSendReceiveViaBuilderInOneProcess("::1", GetParam());
+}
+
+TEST_P(TCqMode, RegisterQpWithoutSrqIsRejected) {
+    auto rdma = CreateRegistrationTestCq("127.0.0.1", GetParam(), TRdmaRuntimeParams{
+        .MaxCqe = 8,
+        .MaxWr = 4,
+        .MaxSrqWr = 0,
+        .RecieveBufSz = 0,
+    });
+    ASSERT_TRUE(rdma.Cq);
+
+    const TActorId edge = rdma.ActorSystem->AllocateEdgeActor(0);
+    EXPECT_FALSE(rdma.Cq->RegisterQpAsync(42, edge));
+    EXPECT_FALSE(rdma.Cq->DeregisterQpAsync(42));
+}
+
+TEST_P(TCqMode, QpCanBeCreatedWithSrq) {
+    auto rdma = CreateRegistrationTestCq("127.0.0.1", GetParam(), TRdmaRuntimeParams{
+        .MaxCqe = 8,
+        .MaxWr = 4,
+        .MaxSrqWr = 4,
+        .RecieveBufSz = 1024,
+    });
+    ASSERT_TRUE(rdma.Cq);
+    ASSERT_TRUE(rdma.Cq->GetSrq());
+
+    TQueuePair qp;
+    EXPECT_EQ(qp.Init(rdma.Ctx, rdma.Cq.get(), 16), 0);
+}
+
+TEST_P(TCqMode, RegisteredQpGetsTerminalReceiveError) {
+    auto rdma = CreateRegistrationTestCq("127.0.0.1", GetParam(), TRdmaRuntimeParams{
+        .MaxCqe = 8,
+        .MaxWr = 4,
+        .MaxSrqWr = 4,
+        .RecieveBufSz = 1024,
+    });
+    ASSERT_TRUE(rdma.Cq);
+
+    const TActorId edge = rdma.ActorSystem->AllocateEdgeActor(0);
+    ASSERT_TRUE(rdma.Cq->RegisterQpAsync(42, edge));
+
+    NotifyCqTerminalError(rdma.Cq);
+
+    auto ev = rdma.ActorSystem->GrabEdgeEvent<TEvRdmaIoReceiveDone>(edge, TDuration::Seconds(5));
+    ASSERT_TRUE(ev);
+    EXPECT_TRUE(ev->Get()->IsCqError());
+    EXPECT_FALSE(rdma.Cq->RegisterQpAsync(43, edge));
+}
+
+TEST_P(TCqMode, NotifyErrFromActorNotifiesAllRegisteredQpActors) {
+    auto rdma = CreateRegistrationTestCq("127.0.0.1", GetParam(), TRdmaRuntimeParams{
+        .MaxCqe = 16,
+        .MaxWr = 4,
+        .MaxSrqWr = 8,
+        .RecieveBufSz = 1024,
+    });
+    ASSERT_TRUE(rdma.Cq);
+
+    const TActorId edge = rdma.ActorSystem->AllocateEdgeActor(0);
+    const std::array<ui32, 4> qpNums = {42, 43, 44, 45};
+    for (ui32 qpNum : qpNums) {
+        const TActorId actor = rdma.ActorSystem->Register(new TReceiveTerminalProbeActor(edge, qpNum));
+        ASSERT_TRUE(rdma.Cq->RegisterQpAsync(qpNum, actor));
+    }
+
+    const TActorId notifier = rdma.ActorSystem->Register(new TNotifyCqTerminalErrorActor(rdma.Cq, edge));
+    rdma.ActorSystem->Send(new IEventHandle(notifier, edge, new TEvTriggerCqTerminalError()), 0);
+
+    auto triggerResult = rdma.ActorSystem->GrabEdgeEvent<TEvCqTerminalErrorTriggered>(edge, TDuration::Seconds(5));
+    ASSERT_TRUE(triggerResult);
+    ASSERT_TRUE(triggerResult->Get()->Success);
+
+    THashSet<ui32> notifiedQps;
+    for (size_t i = 0; i < qpNums.size(); ++i) {
+        auto ev = rdma.ActorSystem->GrabEdgeEvent<TEvReceiveTerminalProbeResult>(edge, TDuration::Seconds(5));
+        ASSERT_TRUE(ev);
+        EXPECT_EQ(ev->Get()->ErrSource, "TCqErr");
+        notifiedQps.insert(ev->Get()->QpNum);
+    }
+
+    EXPECT_EQ(notifiedQps.size(), qpNums.size());
+    for (ui32 qpNum : qpNums) {
+        EXPECT_TRUE(notifiedQps.contains(qpNum)) << "Missing terminal notification for qp " << qpNum;
+    }
+    EXPECT_FALSE(rdma.Cq->RegisterQpAsync(46, edge));
+}
+
+TEST_P(TCqMode, ConcurrentRegisterAndNotifyErrEitherRejectsOrNotifies) {
+    static constexpr ui32 Iterations = 1000;
+    ui32 notificationCount = 0;
+    ui32 registerRejectedCount = 0;
+    std::vector<TDuration> notifyErrDurations;
+    notifyErrDurations.reserve(Iterations);
+
+    for (ui32 i = 0; i < Iterations; ++i) {
+        auto rdma = CreateRegistrationTestCq("127.0.0.1", GetParam(), TRdmaRuntimeParams{
+            .MaxCqe = 16,
+            .MaxWr = 4,
+            .MaxSrqWr = 8,
+            .RecieveBufSz = 1024,
+        });
+        ASSERT_TRUE(rdma.Cq);
+
+        const TActorId edge = rdma.ActorSystem->AllocateEdgeActor(0);
+        const ui32 qpNum = 10000 + i;
+        const TActorId receiver = rdma.ActorSystem->Register(new TReceiveTerminalProbeActor(edge, qpNum));
+        const TActorId notifier = rdma.ActorSystem->Register(new TNotifyCqTerminalErrorActor(rdma.Cq, edge));
+
+        auto ready = std::make_shared<std::atomic<bool>>(false);
+        auto start = std::make_shared<std::atomic<bool>>(false);
+        auto registered = std::make_shared<std::atomic<bool>>(false);
+
+        TThread registerThread([cq = rdma.Cq, receiver, qpNum, ready, start, registered]() {
+            while (!ready->load(std::memory_order_acquire)) {
+                ThreadYield();
+            }
+            start->store(true, std::memory_order_release);
+            registered->store(cq->RegisterQpAsync(qpNum, receiver), std::memory_order_release);
+        });
+        registerThread.Start();
+
+        rdma.ActorSystem->Send(new IEventHandle(notifier, edge, new TEvTriggerCqTerminalError(ready, start)), 0);
+
+        auto triggerResult = rdma.ActorSystem->GrabEdgeEvent<TEvCqTerminalErrorTriggered>(edge, TDuration::Seconds(5));
+        registerThread.Join();
+        ASSERT_TRUE(triggerResult) << "iteration " << i;
+        ASSERT_TRUE(triggerResult->Get()->Success) << "iteration " << i;
+        notifyErrDurations.push_back(triggerResult->Get()->NotifyErrDuration);
+
+        if (registered->load(std::memory_order_acquire)) {
+            auto ev = rdma.ActorSystem->GrabEdgeEvent<TEvReceiveTerminalProbeResult>(edge, TDuration::Seconds(5));
+            ASSERT_TRUE(ev) << "iteration " << i;
+            EXPECT_EQ(ev->Get()->QpNum, qpNum) << "iteration " << i;
+            EXPECT_EQ(ev->Get()->ErrSource, "TCqErr") << "iteration " << i;
+            ++notificationCount;
+        } else {
+            ++registerRejectedCount;
+        }
+    }
+
+    ASSERT_EQ(notifyErrDurations.size(), Iterations);
+    std::sort(notifyErrDurations.begin(), notifyErrDurations.end());
+    const size_t p99Index = (notifyErrDurations.size() * 99 + 99) / 100 - 1;
+    const TDuration notifyErrP99 = notifyErrDurations[p99Index];
+
+    Cerr << "Concurrent Register/NotifyErr stats: notifications=" << notificationCount
+         << ", registerRejected=" << registerRejectedCount
+         << ", notifyErrP99Us=" << notifyErrP99.MicroSeconds()
+         << Endl;
+    ::testing::Test::RecordProperty("NotificationCount", notificationCount);
+    ::testing::Test::RecordProperty("RegisterRejectedCount", registerRejectedCount);
+    ::testing::Test::RecordProperty("NotifyErrP99Us", notifyErrP99.MicroSeconds());
+
+    EXPECT_EQ(notificationCount + registerRejectedCount, Iterations);
+}
+
+TEST_P(TCqMode, DeregisteredQpDoesNotGetTerminalReceiveError) {
+    auto rdma = CreateRegistrationTestCq("127.0.0.1", GetParam(), TRdmaRuntimeParams{
+        .MaxCqe = 8,
+        .MaxWr = 4,
+        .MaxSrqWr = 4,
+        .RecieveBufSz = 1024,
+    });
+    ASSERT_TRUE(rdma.Cq);
+
+    auto receiveCounter = std::make_shared<std::atomic<ui32>>(0);
+    const TActorId deregisteredActor = rdma.ActorSystem->Register(new TReceiveDoneCounterActor(receiveCounter));
+    const TActorId edge = rdma.ActorSystem->AllocateEdgeActor(0);
+
+    ASSERT_TRUE(rdma.Cq->RegisterQpAsync(42, deregisteredActor));
+    ASSERT_TRUE(rdma.Cq->DeregisterQpAsync(42));
+    ASSERT_TRUE(rdma.Cq->RegisterQpAsync(43, edge));
+
+    NotifyCqTerminalError(rdma.Cq);
+
+    auto ev = rdma.ActorSystem->GrabEdgeEvent<TEvRdmaIoReceiveDone>(edge, TDuration::Seconds(5));
+    ASSERT_TRUE(ev);
+    EXPECT_TRUE(ev->Get()->IsCqError());
+
+    rdma.ActorSystem->Send(new IEventHandle(deregisteredActor, edge, new TEvents::TEvPing()), 0);
+    auto pong = rdma.ActorSystem->GrabEdgeEvent<TEvents::TEvPong>(edge, TDuration::Seconds(5));
+    ASSERT_TRUE(pong);
+    EXPECT_EQ(receiveCounter->load(std::memory_order_relaxed), 0);
+    EXPECT_FALSE(rdma.Cq->DeregisterQpAsync(42));
+}
+
 /*
  * This test covers the situation when sender is going to reuse memory but has no
  * information about remote reading in progress.
diff --git a/ydb/library/actors/interconnect/rdma/ut/utils.cpp b/ydb/library/actors/interconnect/rdma/ut/utils.cpp
index 436a3d02321..85ea8039ba8 100644
--- a/ydb/library/actors/interconnect/rdma/ut/utils.cpp
+++ b/ydb/library/actors/interconnect/rdma/ut/utils.cpp
@@ -44,7 +44,6 @@ std::tuple<THolder<NActors::TTestActorRuntimeBase>, TRdmaCtx*> PrepareTestRuntim
     NInterconnect::TAddress address(ip, 7777);
     auto ctx = NInterconnect::NRdma::NLinkMgr::GetCtx(address);
     RDMA_UT_EXPECT_TRUE(ctx);
-    Cerr << "Using verbs context: " << *ctx << ", on addr: " << ip << Endl;
 
     return {std::move(actorSystem), ctx};
 }
@@ -63,7 +62,7 @@ std::shared_ptr<TLocalRdmaStuff> InitLocalRdmaStuff(TString bindTo, NInterconnec
     rdma->Qp1 = std::make_shared<TQueuePair>();
     rdma->Qp2 = std::make_shared<TQueuePair>();
 
-    rdma->CqActorId = rdma->ActorSystem->Register(CreateCqActor(1, 1, cqMode, nullptr));
+    rdma->CqActorId = rdma->ActorSystem->Register(CreateCqActor(TRdmaRuntimeParams{1, 1, 0, 0}, cqMode, nullptr));
     rdma->CqPtr = GetCqHandle(rdma->ActorSystem.get(), rdma->Ctx, rdma->CqActorId);
 
     {
diff --git a/ydb/library/actors/interconnect/rdma/ya.make b/ydb/library/actors/interconnect/rdma/ya.make
index a5eb4459af9..65cbefb1e7d 100644
--- a/ydb/library/actors/interconnect/rdma/ya.make
+++ b/ydb/library/actors/interconnect/rdma/ya.make
@@ -15,7 +15,9 @@ IF (OS_LINUX)
         ydb/library/actors/util
         contrib/libs/ibdrv
         contrib/libs/protobuf
+        library/cpp/containers/absl
         library/cpp/monlib/dynamic_counters
+        library/cpp/threading/queue
     )
 
 ELSE()
diff --git a/ydb/library/actors/interconnect/ut/lib/node.h b/ydb/library/actors/interconnect/ut/lib/node.h
index 781e09610b7..a06005e848d 100644
--- a/ydb/library/actors/interconnect/ut/lib/node.h
+++ b/ydb/library/actors/interconnect/ut/lib/node.h
@@ -115,7 +115,7 @@ public:
                 TMailboxType::ReadAsFilled, 0));
         }
         setup.LocalServices.emplace_back(NInterconnect::NRdma::MakeCqActorId(),
-            TActorSetupCmd(NInterconnect::NRdma::CreateCqActor(-1, 1024, rdmaCqMode, nullptr),
+            TActorSetupCmd(NInterconnect::NRdma::CreateCqActor(NInterconnect::NRdma::TRdmaRuntimeParams{-1, 1024, 0, 0}, rdmaCqMode, nullptr),
             TMailboxType::ReadAsFilled, 0));
 
         const TActorId loggerActorId = loggerSettings ? loggerSettings->LoggerActorId : TActorId(0, "logger");