add option to disable NotifyTransfer

Author: maning00

include/mori/io/backend.hpp

@@ -45,22 +45,25 @@ struct BackendConfig {
 struct RdmaBackendConfig : public BackendConfig {
   RdmaBackendConfig() : BackendConfig(BackendType::RDMA) {}
   RdmaBackendConfig(int qpPerTransfer_, int postBatchSize_, int numWorkerThreads_,
-                    PollCqMode pollCqMode_)
+                    PollCqMode pollCqMode_, bool enableNotification_)
       : BackendConfig(BackendType::RDMA),
         qpPerTransfer(qpPerTransfer_),
         postBatchSize(postBatchSize_),
         numWorkerThreads(numWorkerThreads_),
-        pollCqMode(pollCqMode_) {}
+        pollCqMode(pollCqMode_),
+        enableNotification(enableNotification_) {}
 
   int qpPerTransfer{1};
   int postBatchSize{-1};
   int numWorkerThreads{1};
   PollCqMode pollCqMode{PollCqMode::POLLING};
+  bool enableNotification{true};  // Enable/disable notification mechanism for transfer completion
 };
 
 inline std::ostream& operator<<(std::ostream& os, const RdmaBackendConfig& c) {
   return os << "qpPerTransfer[" << c.qpPerTransfer << "] postBatchSize[" << c.postBatchSize
-            << "] numWorkerThreads[" << c.numWorkerThreads << "]";
+            << "] numWorkerThreads[" << c.numWorkerThreads << "] enableNotification["
+            << c.enableNotification << "]";
 }
 
 /* ---------------------------------------------------------------------------------------------- */

src/application/transport/rdma/providers/ibverbs/ibverbs.cpp

@@ -171,9 +171,7 @@ void IBVerbsDeviceContext::ConnectEndpoint(const RdmaEndpointHandle& local,
   attr.dest_qp_num = remote.qpn;
   attr.rq_psn = 0;
   attr.max_dest_rd_atomic = devAttr->orig_attr.max_qp_rd_atom;
-  // Set min_rnr_timer to 20 (approx 4.19 ms) to allow receiver more time to replenish buffers
-  // Default was 12 (0.64 ms) which causes RNR exhaustion too quickly under burst
-  attr.min_rnr_timer = 20;
+  attr.min_rnr_timer = 12;
   attr.ah_attr.sl = 0;
   attr.ah_attr.src_path_bits = 0;
   attr.ah_attr.port_num = local.portId;
@@ -195,11 +193,10 @@ void IBVerbsDeviceContext::ConnectEndpoint(const RdmaEndpointHandle& local,
   // RTS
   attr.qp_state = IBV_QPS_RTS;
   attr.sq_psn = 0;
-  attr.timeout = 20;
+  attr.timeout = 14;
   attr.retry_cnt = 7;
-  attr.rnr_retry = 6;
+  attr.rnr_retry = 7;
   attr.max_rd_atomic = devAttr->orig_attr.max_qp_init_rd_atom;
-
   flags = IBV_QP_STATE | IBV_QP_SQ_PSN | IBV_QP_TIMEOUT | IBV_QP_RETRY_CNT | IBV_QP_RNR_RETRY |
           IBV_QP_MAX_QP_RD_ATOMIC;
   SYSCALL_RETURN_ZERO(ibv_modify_qp(qp, &attr, flags));

src/io/rdma/backend_impl.cpp

@@ -214,6 +214,11 @@ void NotifManager::RegisterEndpointByQpn(uint32_t qpn) {
     SYSCALL_RETURN_ZERO(epoll_ctl(epfd, EPOLL_CTL_ADD, ep->local.ibvHandle.compCh->fd, &ev));
   }
 
+  // Skip notification setup if disabled
+  if (!config.enableNotification) {
+    return;
+  }
+
   std::lock_guard<std::mutex> lock(mu);
   if (qpNotifCtx.find(qpn) != qpNotifCtx.end()) return;
 
@@ -260,6 +265,12 @@ void NotifManager::ProcessOneCqe(int qpn, const EpPair& ep) {
   while ((n = ibv_poll_cq(cq, batchSize, wc)) > 0) {
     for (int i = 0; i < n; ++i) {
       if (wc[i].opcode == IBV_WC_RECV) {
+        // Skip RECV processing if notification is disabled
+        if (!config.enableNotification) {
+          MORI_IO_WARN("Received unexpected RECV completion when notification is disabled");
+          continue;
+        }
+
         std::lock_guard<std::mutex> lock(mu);
 
         assert(qpNotifCtx.find(qpn) != qpNotifCtx.end());
@@ -311,6 +322,9 @@ void NotifManager::ProcessOneCqe(int qpn, const EpPair& ep) {
           } else {
             statusPtr->SetMessage(ibv_wc_status_str(wc[i].status));
             statusPtr->SetCode(StatusCode::ERR_RDMA_OP);
+            MORI_IO_ERROR("NotifManager receive cqe failed for task {} code {} status {}",
+                          msg->meta->id, static_cast<uint32_t>(wc[i].status),
+                          ibv_wc_status_str(wc[i].status));
             // set status to nullptr indicate that transfer failed
             msg->meta->status = nullptr;
           }
@@ -597,7 +611,7 @@ void RdmaBackendSession::ReadWrite(size_t localOffset, size_t remoteOffset, size
     status->SetMessage(ret.message);
     status->SetCode(ret.code);
   }
-  if (!ret.Failed()) {
+  if (!ret.Failed() && config.enableNotification) {
     RdmaOpRet notifRet = RdmaNotifyTransfer(eps, status, id);
     if (notifRet.Failed()) {
       status->SetMessage(notifRet.message);
@@ -626,7 +640,7 @@ void RdmaBackendSession::BatchReadWrite(const SizeVec& localOffsets, const SizeV
     status->SetMessage(ret.message);
     status->SetCode(ret.code);
   }
-  if (!ret.Failed()) {
+  if (!ret.Failed() && config.enableNotification) {
     RdmaOpRet notifRet = RdmaNotifyTransfer(eps, status, id);
     if (notifRet.Failed()) {
       status->SetMessage(notifRet.message);

src/pybind/mori.cpp

@@ -311,13 +311,15 @@ void RegisterMoriIo(pybind11::module_& m) {
   py::class_<mori::io::BackendConfig>(m, "BackendConfig");
 
   py::class_<mori::io::RdmaBackendConfig, mori::io::BackendConfig>(m, "RdmaBackendConfig")
-      .def(py::init<int, int, int, mori::io::PollCqMode>(), py::arg("qp_per_transfer") = 1,
+      .def(py::init<int, int, int, mori::io::PollCqMode, bool>(), py::arg("qp_per_transfer") = 1,
            py::arg("post_batch_size") = -1, py::arg("num_worker_threads") = -1,
-           py::arg("poll_cq_mode") = mori::io::PollCqMode::POLLING)
+           py::arg("poll_cq_mode") = mori::io::PollCqMode::POLLING,
+           py::arg("enable_notification") = true)
       .def_readwrite("qp_per_transfer", &mori::io::RdmaBackendConfig::qpPerTransfer)
       .def_readwrite("post_batch_size", &mori::io::RdmaBackendConfig::postBatchSize)
       .def_readwrite("num_worker_threads", &mori::io::RdmaBackendConfig::numWorkerThreads)
-      .def_readwrite("poll_cq_mode", &mori::io::RdmaBackendConfig::pollCqMode);
+      .def_readwrite("poll_cq_mode", &mori::io::RdmaBackendConfig::pollCqMode)
+      .def_readwrite("enable_notification", &mori::io::RdmaBackendConfig::enableNotification);
 
   py::class_<mori::io::IOEngineConfig>(m, "IOEngineConfig")
       .def(py::init<std::string, uint16_t>(), py::arg("host") = "", py::arg("port") = 0)