perf(umbp): optimize PoolClient BatchPut/BatchGet (#403)

- Per-pair RDMA transfer merge (get + put): collapse pages sharing a (localMR, remoteMR) pair into one BatchRead/BatchWrite with a single TransferStatus/uid; per-pair failures map back to per-key. Also fixes an unlocked dram_memories read race and exposes RDMA knobs via MORI_IO_* env.
- BatchGet: split into PartitionBatchGetTargets + ExecuteBatchGetPlan on one Submit/Wait path; the zero-copy path submits all peers and runs local DRAM/SSD + remote SSD inside the in-flight window, then waits all.
- BatchPut: mirror the same submit/wait split (SubmitRemoteBatchPut / WaitRemoteBatchPut); multi-peer writes posted before waiting, deferred local puts run in the in-flight window, staging stays per-peer serial, slot commit/abort finalized at wait.
- Cleanup: regroup the Put/Get hot paths into clear sections; best-effort logging on BatchAbortSlots RPC failure.

Author: isytwu

docs/MORI-IO-GUIDE.md

@@ -166,9 +166,11 @@ See `examples/io/example.py` for more complete examples including batch transfer
 | Setting | Description |
 |---------|-------------|
 | `set_log_level(level)` | Set MORI-IO log verbosity level |
-| `RdmaBackendConfig.qp_per_transfer` | Queue pairs per transfer (default 1, increase for higher bandwidth; with multi-NIC these QPs are spread across NICs, so aim for ≥2 QP per NIC) |
-| `RdmaBackendConfig.poll_cq_mode` | CQ polling mode: `POLLING` (busy-wait, lower latency) or `EVENT` (interrupt-driven) |
-| `RdmaBackendConfig.enable_notification` | Enable target-side completion notifications (default `True`) |
+| `RdmaBackendConfig.qp_per_transfer` | Queue pairs per transfer (default 1, increase for higher bandwidth; with multi-NIC these QPs are spread across NICs, so aim for ≥2 QP per NIC). Env: `MORI_IO_QP_PER_TRANSFER` |
+| `RdmaBackendConfig.poll_cq_mode` | CQ polling mode: `POLLING` (busy-wait, lower latency) or `EVENT` (interrupt-driven). Env: `MORI_IO_POLL_CQ_MODE` (`0`/`polling`, `1`/`event`) |
+| `RdmaBackendConfig.post_batch_size` | WRs per `ibv_post_send` chain (default `-1` = auto). Env: `MORI_IO_POST_BATCH_SIZE` |
+| `RdmaBackendConfig.num_worker_threads` | Worker threads for batch posting (default `1`; ignored when chunking / multi-NIC forces inline posting). Env: `MORI_IO_NUM_WORKER_THREADS` |
+| `RdmaBackendConfig.enable_notification` | Enable target-side completion notifications (default `True`). Env: `MORI_IO_ENABLE_NOTIFICATION` |
 | `RdmaBackendConfig.enable_transfer_chunking` | Split a large single transfer into `chunk_bytes` chunks pipelined across QPs (default `False`). Lifts single-transfer bandwidth from single-outstanding (~28 GB/s) to NIC line rate. Forces single-thread inline posting (ignores `num_worker_threads`). Env: `MORI_IO_ENABLE_CHUNKING` |
 | `RdmaBackendConfig.chunk_bytes` | Chunk size when chunking is on (default `65536` = 64 KB). Messages ≤ this are unchanged. Env: `MORI_IO_CHUNK_BYTES` |
 | `RdmaBackendConfig.max_chunks_per_transfer` | Cap on chunks per transfer to bound WR/SQ usage (default `64`). Env: `MORI_IO_MAX_CHUNKS` |

src/io/rdma/backend_impl.cpp

@@ -26,8 +26,10 @@
 
 #include <algorithm>
 #include <cstdlib>
+#include <cstring>
 #include <limits>
 #include <memory>
+#include <optional>
 #include <shared_mutex>
 #include <stdexcept>
 #include <string>
@@ -73,6 +75,27 @@ bool UsesInlineOnly(const RdmaBackendConfig& config) {
   return config.enableTransferChunking || config.numNicsPerTransfer > 1;
 }
 
+// Parse MORI_IO_POLL_CQ_MODE: "0"/"polling" or "1"/"event" (case-insensitive).
+// Returns nullopt for anything else so env::Override warns and keeps the
+// config value (no silent fallback).
+std::optional<PollCqMode> ParsePollCqMode(const char* raw) {
+  if (std::strcmp(raw, "0") == 0 || mori::env::detail::EqualsIgnoreCase(raw, "polling")) {
+    return PollCqMode::POLLING;
+  }
+  if (std::strcmp(raw, "1") == 0 || mori::env::detail::EqualsIgnoreCase(raw, "event")) {
+    return PollCqMode::EVENT;
+  }
+  return std::nullopt;
+}
+
+// Parse MORI_IO_POST_BATCH_SIZE: only -1 (auto) or a positive int are valid.
+// Rejects 0 / negatives (which the post path would silently clamp to 1) so
+// env::Override warns and keeps the config value.
+std::optional<int> ParsePostBatchSize(const char* raw) {
+  if (std::strcmp(raw, "-1") == 0) return -1;
+  return mori::env::detail::ParsePositiveInt(raw);
+}
+
 int ResolveRequestedNics(const RdmaBackendConfig& config, const TopoKey& local,
                          const TopoKey& remote) {
   if (local.loc == MemoryLocationType::GPU || remote.loc == MemoryLocationType::GPU) {
@@ -1235,6 +1258,14 @@ RdmaBackend::RdmaBackend(EngineKey k, const IOEngineConfig& engConfig,
                 mori::env::detail::ParsePositiveInt);
   env::Override("MORI_IO_NUM_NICS_PER_TRANSFER", config.numNicsPerTransfer,
                 mori::env::detail::ParsePositiveInt);
+  // Perf-sweep / ops overrides for knobs that otherwise only have a config
+  // field.  postBatchSize accepts any int (-1 = auto), the rest are positive.
+  env::Override("MORI_IO_QP_PER_TRANSFER", config.qpPerTransfer,
+                mori::env::detail::ParsePositiveInt);
+  env::Override("MORI_IO_POST_BATCH_SIZE", config.postBatchSize, ParsePostBatchSize);
+  env::Override("MORI_IO_NUM_WORKER_THREADS", config.numWorkerThreads,
+                mori::env::detail::ParsePositiveInt);
+  env::Override("MORI_IO_POLL_CQ_MODE", config.pollCqMode, ParsePollCqMode);
   ValidateRdmaNotificationConfig(config);
   ValidateRdmaTransferConfig(config);