From 7cab838fd0cbfab60c42d6fc88b10859005f9112 Mon Sep 17 00:00:00 2001 From: Ketor Date: Fri, 28 Aug 2026 21:59:10 +0800 Subject: [PATCH] perf: adapt RDMA depth and recycle receive capacity --- CHANGELOG.md | 19 ++- CMakeLists.txt | 6 + README.md | 3 +- docs/ARCHITECTURE.md | 15 +- docs/CONNECTORS.md | 11 +- docs/DEPLOY.md | 13 +- docs/METRICS.md | 4 + integration/vllm/README.md | 6 +- .../vllm/src/dfkv_vllm/client_ranks.py | 25 ++++ integration/vllm/src/dfkv_vllm/worker.py | 32 +++-- integration/vllm/tests/test_client_ranks.py | 44 +++++- src/cache/disk_slab_store.cc | 12 +- src/cache/disk_slab_store.h | 1 + src/cache/rdma_server.cc | 114 ++++++++++++--- src/cache/rdma_server.h | 6 + src/transport/rdma_recv_segment.cc | 132 +++++++++++++++--- src/transport/rdma_recv_segment.h | 37 +++-- src/transport/rdma_transport.cc | 97 ++++++++++++- src/transport/rdma_transport.h | 11 ++ src/transport/rdma_verbs.cc | 48 +++++++ src/transport/rdma_verbs.h | 7 + test/python/test_observability_contract.py | 5 + test/transport/rdma_loopback_test.cc | 104 +++++++++++++- test/transport/rdma_recv_segment_test.cc | 37 +++++ 24 files changed, 698 insertions(+), 91 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1c7849da..10ba61da 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,11 +14,26 @@ `DFKV_RDMA_CONNECTION_MIN_BLOCK_BYTES` (default 256 KiB) as the floor. - Idle endpoint reuse selects the smallest sufficient class, and a full idle pool prefers a smaller returning QP over its largest retained QP. +- Extended adaptive geometry to QP depth: scalar operations open depth-1 QPs, + while batched operations select the smallest sufficient power-of-two depth. + Pool reuse and resource accounting now match both block and depth classes, + with bounded per-class opened/active/idle metrics. +- Replaced per-connection pull-read MR registration with type-2 Memory Windows + over shared per-PD chunk MRs. Hardware without Memory Window support retains + the exact-range MR fallback and exposes which path each connection used. +- Affinitized growth chunks to their first rail/NUMA node, and added idle + shrinking for empty non-initial chunks (`DFKV_RDMA_RECV_CHUNK_IDLE_MS`, + default 60 s). The initial shared chunk is never released. - Changed watermark eviction to a persistent high/low hysteresis drain bounded by `DFKV_SLAB_EVICT_MAX_EXTENTS_PER_TICK` (default one). Whole extents clear `slots.tbl` in one contiguous write instead of one 64-byte pwrite per slot. -- Added receive-pool growth/budget metrics and watermark active/tick/duration/ - extent-clear metrics. +- Added receive-pool growth/shrink/budget, pull Memory Window/fallback, + adaptive connection-class, and watermark active/tick/duration/clear metrics. +- Reused one preallocated zero-record buffer for extent metadata clears, + removing allocation and repeated zero-fill work from the watermark lock. +- Completed vLLM replicated-MLA rank convergence: converged client-rank mode + automatically enables native same-host GPU rendezvous, so one rank performs + each remote GET and CUDA IPC publishes identical bytes to TP followers. - On xb01-0064, 80 live 1 MiB data QPs under a 64 MiB logical ceiling used 674 MiB across three lazy chunks and completed 10,000/10,000 PUTs; v2.23.3's fixed geometry exhausted an 8 GiB segment at roughly 15 such QPs. Sustained diff --git a/CMakeLists.txt b/CMakeLists.txt index 50e59a0e..0598857f 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -237,6 +237,12 @@ if(DFKV_BUILD_TESTS) -p test_rail_affinity.py) set_tests_properties(python_vllm_rail_affinity PROPERTIES ENVIRONMENT "PYTHONPATH=${CMAKE_CURRENT_SOURCE_DIR}/integration/common/src:${CMAKE_CURRENT_SOURCE_DIR}/integration/vllm/src") + add_test(NAME python_vllm_client_ranks + COMMAND ${PYTHON3} -m unittest discover + -s ${CMAKE_CURRENT_SOURCE_DIR}/integration/vllm/tests + -p test_client_ranks.py) + set_tests_properties(python_vllm_client_ranks PROPERTIES + ENVIRONMENT "PYTHONPATH=${CMAKE_CURRENT_SOURCE_DIR}/integration/common/src:${CMAKE_CURRENT_SOURCE_DIR}/integration/vllm/src") # Shared telemetry layer: pure-python, no build/lib needed (OTel-push test # self-skips when the SDK is absent). Includes the vendored-copy drift guard. add_test(NAME python_telemetry diff --git a/README.md b/README.md index ecc9559f..7e096cf5 100644 --- a/README.md +++ b/README.md @@ -322,6 +322,7 @@ fabric selection and capacity explicit. | `--rdma-depth` | `4` (default) | Handshake window is `min(client, server)`. Pull-read connections lease `2 × depth × adaptive_slot_size`; a 1 MiB actual object uses about 8.4 MiB even when the logical ceiling is 64 MiB. | | `DFKV_RDMA_RECV_SEGMENT_SIZE` | `16 GiB` hard budget | Maximum process receive-pool commitment; it is no longer allocated eagerly. Size for worst-case live/churn connections, but startup pins only one chunk. | | `DFKV_RDMA_RECV_CHUNK_BYTES` | `256 MiB` | Initial and incremental receive-pool commitment. Increase only if measured connection classes are routinely larger; committed chunks never exceed the hard budget. | +| `DFKV_RDMA_RECV_CHUNK_IDLE_MS` | `60000` | Empty non-initial chunks are released after this hold; `0` disables shrinking. Growth chunks are rail-affinitized and NUMA-bound, while the initial shared chunk is retained. | | `DFKV_RDMA_IDLE_MS` | `30000` for frequently replaced clients; otherwise default `600000` | Server dead-client reaper. A short interval bounds leaked receive-segment leases, but live clients must set `DFKV_RDMA_KEEPALIVE_MS` below this value or their next GET pays stale-QP recovery. | | `DFKV_RDMA_HEALTH_RECOVERY_SAMPLES` | `3` | Any healthy initialized rail keeps placement eligible; a partial loss stays online. Loss of the final healthy rail removes the node immediately. Only recovery from zero to nonzero waits the consecutive sample gate, then rejoins as `PARTIAL` or `ACTIVE`. | | `DFKV_RDMA_HEALTH_FILE` | unset | Diagnostic-only health input (`device port_state phys_state`, one per initialized/resolved rail, including an auto-discovered rail) for controlled fault injection. Production MUST leave this unset so health comes from sysfs. | @@ -337,7 +338,7 @@ fabric selection and capacity explicit. | `DFKV_RDMA_DEV` | leave unset for one local HCA; set each host's actual stable local whitelist for multi-rail | The local list may differ in cardinality on GPU and CPU hosts. Peer-aware selection uses exact shared names from HLT1; configured tiers require an explicit list. | | `DFKV_RDMA_RAIL_TIERS` | unset for homogeneous hosts; e.g. `mlx5_0|mlx5_1;mlx5_2` for heterogeneous hosts | Every name must be present in `DFKV_RDMA_DEV`; leftmost is highest priority. GPU and CPU hosts may expose different subsets, but interoperating rail names must describe the same fabric. The client uses the highest tier in `local enabled ∩ peer healthy`; missing/incomplete peer HLT1 fails closed only when tiers are configured. Credits, NUMA, latency, and quarantine never cause tier overflow. | | `DFKV_RDMA_PRIMARY_DEV` | unset | Optional preferred rail within `DFKV_RDMA_DEV`. Admission uses other configured rails only when the primary cannot serve. SGLang `rail_affinity=true` sets a rank-local primary plus one neighboring fallback by default; `rail_affinity_fallbacks=0` disables the fallback. | -| `DFKV_RDMA_DEPTH` | `4` (default) | Keep client/server defaults aligned for connector batch correctness. Throughput scaling comes from multiple pooled connections, not raising one QP's depth; lowering depth reduces receive-segment consumption only after validating the real engine workload. | +| `DFKV_RDMA_DEPTH` | `4` ceiling (default) | Actual QPs use the smallest sufficient depth class: scalar=1, batches round up by requested window and server cap. Keep the ceiling aligned with the server; class metrics show the resulting mix. | | `DFKV_RDMA_MAX_BLOCK_BYTES` | exact logical object ceiling | Deterministic oversize guard only; it no longer sizes every data QP. | | `DFKV_RDMA_CONNECTION_MIN_BLOCK_BYTES` | `256 KiB` | Smallest data-QP class. Each operation rounds its actual largest object to a power-of-two class; idle reuse picks the smallest sufficient QP. | | `DFKV_RDMA_RAIL_ERROR_THRESHOLD` | `3` (default) | Consecutive client-local rail failures required to quarantine that rail; endpoint/peer failures do not contribute. | diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index f399a229..cc83fd62 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -167,14 +167,15 @@ Production discovery uses MDS. - `DFKV_RDMA_MAX_BLOCK_BYTES` is the logical object safety ceiling. Each data connection advertises `next_power_of_two(max(actual operation bytes, DFKV_RDMA_CONNECTION_MIN_BLOCK_BYTES))`, capped by that ceiling. -- The server validates each connection class against `--max-msg`, negotiates - `qd=min(client depth, server depth)`, and leases `qd` receive plus `qd` pull - slots from any committed chunk. Idle reuse chooses the smallest sufficient - class; lease ownership lasts until QP teardown or idle reclaim. +- The client selects a second power-of-two class from the operation's requested + window: scalar QPs open at depth 1, while batches reuse/open the smallest + sufficient depth up to the client/server ceiling. +- The server validates the block class, negotiates the depth class, and leases + that many receive plus pull slots from rail-affinitized chunks. Pull arenas + use type-2 Memory Windows over a shared chunk MR, with exact-MR fallback. - A data slot is `align4K(4096 + connection_class)`. The hard receive budget - covers worst-case live and pooled QPs, but resident/pinned memory follows the - observed connection high-water in `DFKV_RDMA_RECV_CHUNK_BYTES` increments. - Exhausting the hard budget rejects the connection without changing protocol. + covers worst-case live/pooled QPs; resident memory grows by chunk and empty + non-initial chunks return after `DFKV_RDMA_RECV_CHUNK_IDLE_MS`. - Client host/device pools are registered once per rail at declaration time. Re-declaring the same base with a larger size registers the larger extent; the registration call returns false/nonzero unless the full range is ready. Buffers diff --git a/docs/CONNECTORS.md b/docs/CONNECTORS.md index 9c4ff1c0..a5195811 100644 --- a/docs/CONNECTORS.md +++ b/docs/CONNECTORS.md @@ -84,11 +84,12 @@ tp_rank=..,ver=`（无 `role`——HiCache 是前缀 L3 缓存，无生产/ |-----|------|------|------| | `DFKV_RDMA` | 一般路径未设 = TCP；**vLLM 直连无默认，必须 `1`** | 按连接器选择 | `1` 显式选择 native-verbs RDMA v2；请求 RDMA 后设备或协议不可用会失败，不会自动选择 TCP。`DfkvStoreConnector` 只接收 GPU 设备指针，构造时会关闭并拒绝任何非 RDMA handle。 | | `DFKV_RDMA_DEV` | 首个 `ACTIVE` 本地 HCA | 留空让两端各自选本地首口；多轨才显式写同 fabric 白名单 | 留空时 bootstrap 不发送设备名，client/server 可使用不同本地命名。逗号列表显式开启多轨，新连接在健康轨间轮转；显式设备名会发给 peer，故两端必须存在同名且互通的 fabric。设备名上限 **18 字节**（v2 bootstrap dev frame 限制），超长即 fail-fast 拒绝启动/建连，不会静默截断（见 `src/transport/dev_frame.h`）。 | -| `DFKV_RDMA_DEPTH` | `4` | 两侧可不同，按容量选 | 握手协商 `min(client,server)`。每条 pull data QP 租 `2 × depth × adaptive_slot`；control buffer 始终有界。 | +| `DFKV_RDMA_DEPTH` | `4` ceiling | 两侧保持上限一致 | scalar QP=depth1；batch按实际window选择最小power-of-two depth class，再由server cap钳制。 | | `DFKV_RDMA_MAX_BLOCK_BYTES` | 4 MiB 逻辑上限 | 覆盖最大合法对象 | 只做 deterministic oversize guard；不再让所有连接按最大值预留。 | | `DFKV_RDMA_CONNECTION_MIN_BLOCK_BYTES` | 256 KiB | 覆盖常见小块，保持默认起步 | 当前操作最大对象按 power-of-two 向上取 connection class；idle pool 选最小可满足 QP。 | | `DFKV_RDMA_RECV_SEGMENT_SIZE` | 16 GiB | 按 peak live/pooled QP 设 hard budget | server receive-pool 最大提交量；不再启动期全量申请。 | | `DFKV_RDMA_RECV_CHUNK_BYTES` | 256 MiB | 保持默认，除非常见 connection class 更大 | server 启动只提交一个 chunk，后续 allocation miss 按需增长，不超过 hard budget。 | +| `DFKV_RDMA_RECV_CHUNK_IDLE_MS` | 60 s | 保持默认 | 空闲非初始chunk到期返还；`0`关闭。增长chunk绑定首次使用rail/NUMA。 | | `DFKV_RDMA_NUMA` | `0` | 显式多轨的大机可设 `1` | 建连按调用线程 NUMA 选 rail；动态 chunk 在实际租用它的 endpoint rail 上注册。 | | `DFKV_RDMA_MAX_PAYLOAD_BYTES` | 64 MiB（67108864） | — | 客户端单 value payload 上限（不得超过 server 侧同名上限） | | `DFKV_CUDA_PINNED_POOL_BYTES` | 64 MiB（67108864） | 按进程允许的 CUDA pinned host memory / memlock 设置 | vLLM CUDA GET publication 的进程级 bounce pool 预算。取正十进制字节数，最大 4 GiB；预算向下取整为完整 slot，且至少容纳一个 slot、最多 4096 个。非法或与 slot 不兼容的组合告警并回退整组默认值。slot 按需创建，因此实际 pinned high-water 不超过取整后的预算。 | @@ -132,8 +133,10 @@ connection_class = min(logical_max, next_power_of_two(max(actual_bytes, DFKV_RDMA_CONNECTION_MIN_BLOCK_BYTES))) +depth_class = min(server_depth, + next_power_of_two(max(actual_window, 1))) S_data = align4K(4096 + connection_class) -B_connection = 2 × depth × S_data +B_connection = 2 × depth_class × S_data ``` 同一 peer/rail 的 idle pool 选择最小可满足 class；没有才建新 QP。pool 满时， @@ -586,9 +589,9 @@ server 启动只注册首个 receive chunk；新 data QP 租当前 operation cla L2-bypass 不需要独立 server 协议，但三项决定 v2 容量： - `DFKV_RDMA_RECV_SEGMENT_SIZE`：receive-pool hard budget； -- `DFKV_RDMA_RECV_CHUNK_BYTES`：启动/增量提交粒度； +- `DFKV_RDMA_RECV_CHUNK_BYTES` / `_IDLE_MS`：启动/增量提交粒度与空chunk保留期； - `DFKV_RDMA_MAX_BLOCK_BYTES`、`DFKV_RDMA_CONNECTION_MIN_BLOCK_BYTES` 与 - `DFKV_RDMA_DEPTH`：分别决定逻辑上限、最小 class 和每 QP lease。 + `DFKV_RDMA_DEPTH`：决定逻辑上限、最小block class和adaptive depth ceiling。 #### 验证清单 diff --git a/docs/DEPLOY.md b/docs/DEPLOY.md index b53d5507..3ba8487b 100644 --- a/docs/DEPLOY.md +++ b/docs/DEPLOY.md @@ -362,11 +362,11 @@ journalctl -u dfkv -n 10 --no-pager > `kNoCompatibleRail` 均不增加 local rail error。cooldown 到期仍只准入一个真实 > recovery probe。 > -> **v2 receive-pool 预算**：每条 data QP 按实际 connection class 计算 -> `slot=align4K(4096 + class)`，pull-read lease 为 -> `2 × depth × slot`。`DFKV_RDMA_RECV_SEGMENT_SIZE` 是 hard budget， -> `DFKV_RDMA_RECV_CHUNK_BYTES` 是惰性提交粒度。上线看 committed/max/chunks、 -> used/free、growth/allocation failures；只有 hard budget 或注册失败才拒绝连接。 +> **v2 receive-pool 预算**：每条data QP按实际block/depth二维class计算 +> `slot=align4K(4096 + block_class)`，pull-read lease为 +> `2 × depth_class × slot`。`DFKV_RDMA_RECV_SEGMENT_SIZE`是hard budget， +> `_CHUNK_BYTES`是惰性提交粒度，`_CHUNK_IDLE_MS`控制空chunk返还。 +> 上线看committed/max/chunks、used/free、growth/shrink/failure和MW/fallback。 ### 3a. 每节点 tenant quota @@ -445,7 +445,7 @@ flag 为 env facade）；未列 flag 的全部 env 均从源码排查就不误 | `--put-inflight-limit` / `DFKV_PUT_INFLIGHT_LIMIT` | `0`=关 | 并发盘写上限，超出返回 kCacheFull 快速拒绝 | | `--tcp-max-conns` / `DFKV_TCP_MAX_CONNS` | `512`, 硬上限 4096 | cache TCP handler 上限；超限 accept 恒拒 | | `--tcp-io-timeout-s` / `DFKV_TCP_IO_TIMEOUT_S` | `60`, 硬上限 3600 | per-syscall RCVTIMEO（秒） | -| `--rdma-depth` / `DFKV_RDMA_DEPTH` | `4` | server 提交 QP post 深度；与 client 协商取 `min` | +| `--rdma-depth` / `DFKV_RDMA_DEPTH` | `4` | server ceiling；client scalar=1，batch按实际window选择depth class后再协商取min | | `--rdma-numa` / `DFKV_RDMA_NUMA` | `0` | NUMA-aware rail choice（off/1） | | `--rdma-idle-ms` / `DFKV_RDMA_IDLE_MS` | — | idle connection reaper tick | | `--rdma-op-timeout-ms` / `DFKV_RDMA_OP_TIMEOUT_MS` | `5000` | per-op RDMA deadline | @@ -465,6 +465,7 @@ flag 为 env facade）；未列 flag 的全部 env 均从源码排查就不误 |---|---|---| | `DFKV_RDMA_RECV_SEGMENT_SIZE` | `16 GiB` | server receive-pool hard budget；不再启动期全量分配 | | `DFKV_RDMA_RECV_CHUNK_BYTES` | `256 MiB` | server 启动与增量提交粒度 | +| `DFKV_RDMA_RECV_CHUNK_IDLE_MS` | `60000` | 空闲非初始chunk返还延迟；`0`关闭缩容 | | `DFKV_RDMA_CONNECTION_MIN_BLOCK_BYTES` | `256 KiB` | client adaptive data-QP 最小 class；实际对象向上取 power-of-two | | `DFKV_RDMA_CONNECT_MS` | — | client：IB QP 建连超时 | | `DFKV_RDMA_IO_MS` | — | client：控制面帧读写超时 | diff --git a/docs/METRICS.md b/docs/METRICS.md index d530fec0..a7f60901 100644 --- a/docs/METRICS.md +++ b/docs/METRICS.md @@ -162,8 +162,10 @@ RDMA-listener scrape inventory。 | `dfkv_rdma_recv_segment_used_bytes` / `free_bytes` | gauge | 已提交 chunk 中 lease 占用 / 空闲字节 | | `dfkv_rdma_recv_segment_largest_free_range_bytes` | gauge | 任一 chunk 最大连续 free range | | `dfkv_rdma_recv_segment_growths_total` / `growth_failures_total` | counter | 启动后 chunk 增长成功 / 因预算或分配失败 | +| `dfkv_rdma_recv_segment_shrinks_total` / `released_bytes_total` / `chunk_idle_ms` | counter / gauge | 空闲缩容次数 / 已返还字节 / 非初始chunk空闲保留期 | | `dfkv_rdma_recv_segment_allocation_failures_total` | counter | grow 后仍无法满足的最终 allocation | | `dfkv_rdma_pull_connections` / `dfkv_rdma_legacy_connections` | gauge | 当前 pull-read / legacy responder-write connection 数 | +| `dfkv_rdma_pull_memory_windows_total` / `dfkv_rdma_pull_mr_fallbacks_total` | counter | exact lease使用type-2 MW隔离 / 硬件不支持时回退per-connection MR | | `dfkv_rdma_connection_bytes{class=\"data|control\"}` | gauge | data/control connection 当前 lease 字节；应随 adaptive class 而非 logical max 增长 | | `dfkv_rdma_recv_segment_registered_rails` | gauge | 成功注册初始 receive chunk 的 rail 数；后续 chunk 按使用 rail 惰性注册 | | `dfkv_rdma_v2_ready` | gauge | 初始 receive chunk 与 rail anchor 是否就绪 | @@ -337,6 +339,8 @@ C 客户端快照还含传输级指标（RDMA 构建）： | `dfkv_rdma_client_pool_mr_registrations_total` / `dfkv_rdma_client_pool_mr_registration_failures_total` | counter | shared-PD 上真实 `ibv_reg_mr` 次数 / 单 rail 显式 pool 注册失败；包含最终回滚的尝试 | | `dfkv_rdma_client_pool_mr_active_registrations` | gauge | 进程内仍有 endpoint 引用的 shared-PD MR generations；扩容成功后 anchor/空闲 endpoint 立即释放旧代，在飞旧 endpoint 到下次 acquire/close 才释放，确保旧 range 不中断 | | `dfkv_rdma_client_max_block_seen_bytes` / `dfkv_rdma_client_declared_max_block_bytes` / `dfkv_rdma_client_connection_min_block_bytes` | gauge | 实际请求高水位 / 逻辑安全上限 / adaptive QP 最小 class | +| `dfkv_rdma_client_connection_class_opened_total{block_bytes,depth}` | counter | adaptive二维class累计新建data QP | +| `dfkv_rdma_client_connection_class_active{block_bytes,depth}` / `connection_class_idle{block_bytes,depth}` | gauge | 当前执行中 / idle pool内的二维class分布；label集合受power-of-two class限制 | | `dfkv_rdma_client_oversize_rejects_total` | counter | 分配、注册或发帖前因超过声明上限而拒绝的操作 | | `dfkv_rdma_client_v2_probe_attempts_total` / `dfkv_rdma_client_v2_probe_failures_total` | counter | 必选 v2 bootstrap probe 尝试 / 失败 | | `dfkv_rdma_client_stale_pool_retries_total` | counter | pooled QP 失败后改用 fresh connection 的重试 | diff --git a/integration/vllm/README.md b/integration/vllm/README.md index 9521a7b9..f531f4fc 100644 --- a/integration/vllm/README.md +++ b/integration/vllm/README.md @@ -58,8 +58,12 @@ traffic if either requirement is missing. | `DFKV_RDMA` | **required: `1`** | Selects the required GPUDirect RDMA transport. Unset/TCP is rejected during connector construction; there is no TCP fallback. | | `PYTHONHASHSEED` | **required: fixed value** | Stabilizes vLLM's root block hash across processes and restarts. Use the same value (for example `0`) on every producer and consumer sharing a store. | | `DFKV_RDMA_DEV` | first local `ACTIVE` HCA | Optional ordered device/fabric list. With `rail_affinity=true`, the connector narrows this full per-host list to the worker's world-group local-rank primary plus bounded fallbacks before native open. | -| `DFKV_RDMA_DEPTH` | `4` | Negotiated request window (`min(client, server)`). Keep the defaults aligned for production connector batches. Per-connection bandwidth is depth-flat after connection setup; scale read throughput with sharded pooled connections, not by increasing depth. | +| `DFKV_RDMA_DEPTH` | `4` ceiling | Scalar QPs open at depth1; batches select the smallest sufficient power-of-two depth up to this ceiling and the server cap. | | `DFKV_RDMA_NUMA` | `0` | `1` pins buffers/threads to the rail's NUMA node and picks a NUMA-local rail per connection. Optional. | +| `DFKV_RDMA_CONNECTION_MIN_BLOCK_BYTES` | `256 KiB` | Minimum adaptive block class; logical max remains `DFKV_RDMA_MAX_BLOCK_BYTES`. | +| `DFKV_CONNECTOR_CLIENT_RANKS` | unset=`TP` | `auto` or `N` converges fully TP-replicated MLA stores to N evenly spread ranks. Sharded DCP/PCP layouts clamp to TP. Converged loads automatically enable native GPU same-host dedup so one rank performs each remote GET and CUDA IPC publishes to followers. | +| `DFKV_CONNECTOR_CLIENT_ELIDE` | auto in converged mode | Producer non-participants skip eager native-client creation. Explicit `0` disables elision without disabling load convergence. | +| `DFKV_NODE_DEDUP_GPU_ARENA_MB` | `512` | Per-rank GPU rendezvous arena used by converged loads; reduce only after the largest concurrent logical object/window fits. | | `DFKV_READ_SHARD_KEYS` | `16` | Target keys per read shard: splits one node's batched GET into parallel shards. The real read-throughput lever on few-node rings / large batches landing on one node (single connection drains ~166 MB/s serially); no-op on wide rings. | | `DFKV_READ_MAX_CONNS` | `8` | Per-node cap on concurrent read-shard connections (pairs with the above; `1` disables sharding). | | `DFKV_FANOUT_THREADS` | `32` | Client batch-op fan-out pool cap (clamped [1,1024]). Raise when callers × node-groups ≫ 32, or batch calls degrade to caller-serial and per-call latency grows from max(group) to sum(group). | diff --git a/integration/vllm/src/dfkv_vllm/client_ranks.py b/integration/vllm/src/dfkv_vllm/client_ranks.py index d4c96e64..cf4ecc10 100644 --- a/integration/vllm/src/dfkv_vllm/client_ranks.py +++ b/integration/vllm/src/dfkv_vllm/client_ranks.py @@ -17,10 +17,13 @@ from __future__ import annotations import logging +from collections.abc import MutableMapping logger = logging.getLogger(__name__) ENV_NAME = "DFKV_CONNECTOR_CLIENT_RANKS" +NODE_DEDUP_ENV = "DFKV_CLIENT_NODE_DEDUP" +GPU_DEDUP_ENV = "DFKV_CLIENT_NODE_DEDUP_GPU" def resolve_client_ranks( @@ -59,6 +62,28 @@ def participant(tp_rank: int, tp_size: int, effective: int) -> int | None: idx = tp_rank // spacing return idx if idx < effective else None +def configure_load_convergence( + env: MutableMapping[str, str], + tp_size: int, + effective: int, + replicated: bool, +) -> tuple[bool, str]: + """Enable the native same-host GPU rendezvous for converged replicated KV. + + All TP ranks still execute vLLM's load callback, but only the native dedup + fetch leader opens a remote QP; followers receive the identical MLA bytes + through CUDA IPC. Explicit operator values win over these safe defaults. + """ + if not replicated or effective >= tp_size: + return False, "clamped(no-load-convergence)" + if env.get(NODE_DEDUP_ENV, "1") != "1" or env.get( + GPU_DEDUP_ENV, "1" + ) != "1": + return False, "disabled(native-gpu-dedup)" + env.setdefault(NODE_DEDUP_ENV, "1") + env.setdefault(GPU_DEDUP_ENV, "1") + return True, f"native-gpu-dedup({effective}/{tp_size})" + ELIDE_ENV = "DFKV_CONNECTOR_CLIENT_ELIDE" diff --git a/integration/vllm/src/dfkv_vllm/worker.py b/integration/vllm/src/dfkv_vllm/worker.py index 03632ed7..6dcb730b 100644 --- a/integration/vllm/src/dfkv_vllm/worker.py +++ b/integration/vllm/src/dfkv_vllm/worker.py @@ -57,8 +57,13 @@ # dfkv handles its own RDMA bootstrap so the transfer-engine helpers are dropped. from ._determinism import ensure_deterministic_block_hashing from .client_ranks import ENV_NAME as CLIENT_RANKS_ENV -from .client_ranks import (ELIDE_ENV, participant, resolve_client_ranks, - should_create_client) +from .client_ranks import ( + ELIDE_ENV, + configure_load_convergence, + participant, + resolve_client_ranks, + should_create_client, +) from .coordinator import DfkvStoreCoordinator from .data import ( VLLM_MULTIWR_V2, @@ -1414,6 +1419,10 @@ def __init__( cr_reason, " (store convergence active)" if self.client_ranks < self.tp_size else "", ) + self.load_convergence, load_reason = configure_load_convergence( + os.environ, self.tp_size, self.client_ranks, replicated + ) + logger.info("client_ranks load convergence: %s", load_reason) self.metadata = KeyMetadata( model_name=model_config.model.rstrip("/").split("/")[-1], @@ -1542,18 +1551,21 @@ def __init__( cache_role=str(self.kv_role), require_rdma=_tcfg.truthy(extra.get("require_rdma", True)), ) - # Phase 2a (issue #111): producer non-participants skip the client - # (opt-in via DFKV_CONNECTOR_CLIENT_ELIDE=1; layout-clamped). The saved - # kwargs let the rank UN-elide lazily if a load ever reaches it — a - # producer DOES load on cross-instance prefix reuse (get_finished has - # no role gate), and failing those loads into whole-span recomputes - # would cost more than the elided connections save. + # Converged replicated layouts enable native same-host GPU rendezvous: + # one TP rank performs each remote GET and CUDA IPC publishes identical + # MLA bytes to followers. Producer non-participants also skip eager + # client creation by default; an explicit CLIENT_ELIDE=0 overrides. self._lazy_client_kwargs: dict | None = None self._lazy_client_lock = threading.Lock() self._kv_pool_regions: list[tuple[int, int]] = [] + elide_default = "1" if self.load_convergence else "0" create_client, elide_reason = should_create_client( - self.kv_role, self.tp_rank, self.tp_size, self.client_ranks, - _tcfg.truthy(os.environ.get(ELIDE_ENV, "0"))) + self.kv_role, + self.tp_rank, + self.tp_size, + self.client_ranks, + _tcfg.truthy(os.environ.get(ELIDE_ENV, elide_default)), + ) if not create_client: logger.info("dfkv client elided: %s", elide_reason) self.client = None diff --git a/integration/vllm/tests/test_client_ranks.py b/integration/vllm/tests/test_client_ranks.py index 226bdf39..2f472881 100644 --- a/integration/vllm/tests/test_client_ranks.py +++ b/integration/vllm/tests/test_client_ranks.py @@ -3,11 +3,16 @@ the effective count.""" import sys +import unittest from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) -from dfkv_vllm.client_ranks import participant, resolve_client_ranks +from dfkv_vllm.client_ranks import ( + configure_load_convergence, + participant, + resolve_client_ranks, +) def test_default_is_full_participation(): @@ -40,6 +45,24 @@ def test_participant_spread_is_even_and_exact(): # N=1: only rank 0. assert [participant(r, 4, 1) for r in range(4)] == [0, None, None, None] +def test_load_convergence_enables_native_gpu_rendezvous(): + env: dict[str, str] = {} + active, reason = configure_load_convergence(env, 8, 1, True) + assert active and "native-gpu-dedup" in reason + assert env["DFKV_CLIENT_NODE_DEDUP"] == "1" + assert env["DFKV_CLIENT_NODE_DEDUP_GPU"] == "1" + + +def test_load_convergence_clamps_sharded_and_honors_explicit_disable(): + env: dict[str, str] = {} + assert not configure_load_convergence(env, 8, 1, False)[0] + assert env == {} + env = {"DFKV_CLIENT_NODE_DEDUP_GPU": "0"} + active, reason = configure_load_convergence(env, 8, 1, True) + assert not active and "disabled" in reason + assert "DFKV_CLIENT_NODE_DEDUP" not in env + assert env["DFKV_CLIENT_NODE_DEDUP_GPU"] == "0" + # ---- Phase 2a: producer-side client elision (should_create_client) ---- @@ -79,3 +102,22 @@ def test_elide_noop_without_convergence(): def test_elide_n1_keeps_rank0_only(): created = [should_create_client("kv_producer", r, 8, 1, True)[0] for r in range(8)] assert created == [True] + [False] * 7 + + +class ClientRanksContractTest(unittest.TestCase): + def test_all_client_rank_contracts(self): + test_default_is_full_participation() + test_auto_converges_only_replicated() + test_explicit_n_clamps_but_never_errors() + test_participant_spread_is_even_and_exact() + test_load_convergence_enables_native_gpu_rendezvous() + test_load_convergence_clamps_sharded_and_honors_explicit_disable() + test_elide_off_always_creates() + test_elide_only_producer_non_participants() + test_elide_never_touches_consumers_or_both() + test_elide_noop_without_convergence() + test_elide_n1_keeps_rank0_only() + + +if __name__ == "__main__": + unittest.main() diff --git a/src/cache/disk_slab_store.cc b/src/cache/disk_slab_store.cc index fbe5a05d..965fc54e 100644 --- a/src/cache/disk_slab_store.cc +++ b/src/cache/disk_slab_store.cc @@ -134,6 +134,8 @@ DiskSlabStore::DiskSlabStore(Options opt, bool* ok) : opt_(std::move(opt)) { if (ok) *ok = false; return; } + zero_extent_records_.assign( + static_cast(max_slots_per_extent_) * kRecBytes, 0); SlabAllocator::Options ao; ao.extent_bytes = opt_.extent_bytes; @@ -149,9 +151,8 @@ DiskSlabStore::DiskSlabStore(Options opt, bool* ok) : opt_(std::move(opt)) { // rebinds are rare (workload-mix shifts), so the ms-scale sync under the // allocator lock is acceptable. ao.on_extent_bind = [this](uint32_t e) { - const std::vector zeros( - static_cast(max_slots_per_extent_) * kRecBytes, 0); - if (!PwriteAll(table_fd_, zeros.data(), zeros.size(), TableOffset(e, 0))) + if (!PwriteAll(table_fd_, zero_extent_records_.data(), + zero_extent_records_.size(), TableOffset(e, 0))) return FailMetadata(); if (::fdatasync(table_fd_) != 0) return FailMetadata(); bind_wipes_.fetch_add(1, std::memory_order_relaxed); @@ -159,9 +160,8 @@ DiskSlabStore::DiskSlabStore(Options opt, bool* ok) : opt_(std::move(opt)) { }; ao.on_extent_evict = [this](uint32_t extent, uint32_t, uint32_t residents) { - const std::vector zeros( - static_cast(max_slots_per_extent_) * kRecBytes, 0); - if (!PwriteAll(table_fd_, zeros.data(), zeros.size(), + if (!PwriteAll(table_fd_, zero_extent_records_.data(), + zero_extent_records_.size(), TableOffset(extent, 0))) return FailMetadata(); record_writes_.fetch_add(std::max(residents, 1), diff --git a/src/cache/disk_slab_store.h b/src/cache/disk_slab_store.h index 28f2b922..42d55e48 100644 --- a/src/cache/disk_slab_store.h +++ b/src/cache/disk_slab_store.h @@ -232,6 +232,7 @@ class DiskSlabStore : public StoreEngine { Options opt_; uint32_t num_extents_ = 0; uint32_t max_slots_per_extent_ = 0; + std::vector zero_extent_records_; std::unique_ptr alloc_; std::vector extent_fds_; // resident, one per extent (buffered) std::vector extent_dio_fds_; // O_DIRECT twins (only when direct_writes) diff --git a/src/cache/rdma_server.cc b/src/cache/rdma_server.cc index 8230a7fa..25f61c9a 100644 --- a/src/cache/rdma_server.cc +++ b/src/cache/rdma_server.cc @@ -103,6 +103,22 @@ size_t RecvChunkBytes(size_t max_bytes) { return bytes; } +uint64_t RecvChunkIdleMs() { + constexpr uint64_t kDefault = 60000; + const char* value = std::getenv("DFKV_RDMA_RECV_CHUNK_IDLE_MS"); + uint64_t out = kDefault; + if (value && *value) { + errno = 0; + char* end = nullptr; + const unsigned long long parsed = std::strtoull(value, &end, 10); + if (errno == 0 && end != value && *end == '\0') + out = std::min(parsed, 86400000); + } + config_dump::RecordResolved("DFKV_RDMA_RECV_CHUNK_IDLE_MS", + std::to_string(out)); + return out; +} + const char* DiscoveryStatusName(rdma::RdmaDiscoveryStatus status) { switch (status) { case rdma::RdmaDiscoveryStatus::kOk: @@ -288,6 +304,7 @@ Status RdmaServer::Start(int port) { // now the hard process budget rather than an eager allocation size. recv_segment_max_bytes_ = RecvSegmentBytes(); recv_segment_chunk_bytes_ = RecvChunkBytes(recv_segment_max_bytes_); + recv_chunk_idle_ms_ = RecvChunkIdleMs(); const size_t min_slot_bytes = rdma::V2SlotSize(max_msg_); if (min_slot_bytes == 0 || recv_segment_max_bytes_ < 2 * min_slot_bytes || @@ -366,13 +383,30 @@ Status RdmaServer::Start(int port) { running_ = true; accept_thread_ = std::thread([this] { NameThisThread("rdma-accept"); AcceptLoop(); }); + if (recv_chunk_idle_ms_ != 0) { + recv_trim_thread_ = std::thread([this] { + NameThisThread("rdma-recv-trim"); + std::unique_lock lock(recv_trim_mu_); + while (running_.load(std::memory_order_relaxed)) { + if (recv_trim_cv_.wait_for(lock, std::chrono::seconds(1), [this] { + return !running_.load(std::memory_order_relaxed); + })) + break; + lock.unlock(); + recv_segments_.TrimIdle(recv_chunk_idle_ms_); + lock.lock(); + } + }); + } return Status::kOk; } void RdmaServer::Stop() { if (!running_.exchange(false)) return; + recv_trim_cv_.notify_all(); if (listen_fd_ >= 0) ::shutdown(listen_fd_, SHUT_RDWR); // wake accept() if (accept_thread_.joinable()) accept_thread_.join(); + if (recv_trim_thread_.joinable()) recv_trim_thread_.join(); if (listen_fd_ >= 0) { ::close(listen_fd_); listen_fd_ = -1; } // Wake every in-flight Serve thread out of WaitComp, then join them all so no // handler call can race the owner's destruction after Stop() returns. @@ -596,6 +630,7 @@ void RdmaServer::Serve(int boot_fd) { const size_t rail_index = static_cast(std::distance(anchor_devs_.begin(), rail_it)); RailStats& rail_stats = *rail_stats_[rail_index]; + const int rail_numa = anchors_[rail_index]->numa_node(); // The client sends QpInfo first. Read and validate its mandatory v2 depth // before allocating per-connection control slots or leasing shared receive @@ -619,8 +654,9 @@ void RdmaServer::Serve(int boot_fd) { ::close(boot_fd); return; } - rdma::RecvSegmentPool::Lease recv_lease = - recv_segments_.Allocate(K * slot_size, rdma::kV2DataOffset); + rdma::RecvSegmentPool::Lease recv_lease = recv_segments_.Allocate( + K * slot_size, rdma::kV2DataOffset, static_cast(rail_index), + rail_numa); if (!recv_lease) { // Segment exhausted: evict the stalest idle connection(s) to make room // before refusing. Client pooled connections re-dial via the stale-retry @@ -654,8 +690,9 @@ void RdmaServer::Serve(int boot_fd) { // down its endpoint and releases the lease asynchronously. for (int i = 0; i < 100 && !recv_lease; ++i) { std::this_thread::sleep_for(std::chrono::milliseconds(10)); - recv_lease = - recv_segments_.Allocate(K * slot_size, rdma::kV2DataOffset); + recv_lease = recv_segments_.Allocate( + K * slot_size, rdma::kV2DataOffset, + static_cast(rail_index), rail_numa); } } if (!recv_lease) { @@ -672,8 +709,9 @@ void RdmaServer::Serve(int boot_fd) { } rdma::RecvSegmentPool::Lease pull_lease; if (pull_read_requested) { - pull_lease = - recv_segments_.Allocate(K * slot_size, rdma::kV2DataOffset); + pull_lease = recv_segments_.Allocate( + K * slot_size, rdma::kV2DataOffset, + static_cast(rail_index), rail_numa); if (!pull_lease) { const auto stats = recv_segments_.stats(); DFKV_LOG_ERROR( @@ -777,16 +815,24 @@ void RdmaServer::Serve(int boot_fd) { ::close(boot_fd); return; } + ibv_mr* pull_pool_mr = nullptr; ibv_mr* pull_segment_mr = nullptr; + uint32_t pull_rkey = 0; if (pull_read_requested) { - pull_segment_mr = - ep.RegisterRemoteReadRegion(pull_lease.data(), pull_lease.size()); - if (!pull_segment_mr) { - DFKV_LOG_ERROR( - "rdma v2: connection-private pull-read MR unavailable on device " + - (dev.empty() ? std::string("(auto)") : dev)); - ::close(boot_fd); - return; + pull_pool_mr = ep.RegisterRemoteReadPool( + pull_lease.segment()->data(), pull_lease.segment()->size()); + if (!pull_pool_mr) { + pull_segment_mr = + ep.RegisterRemoteReadRegion(pull_lease.data(), pull_lease.size()); + if (!pull_segment_mr) { + DFKV_LOG_ERROR( + "rdma v2: pull-read MR unavailable on device " + + (dev.empty() ? std::string("(auto)") : dev)); + ::close(boot_fd); + return; + } + pull_rkey = pull_segment_mr->rkey; + pull_mr_fallbacks_.fetch_add(1, std::memory_order_relaxed); } } DFKV_LOG_INFO("rdma conn: protocol=v2 declared=" + @@ -809,6 +855,25 @@ void RdmaServer::Serve(int boot_fd) { ::close(boot_fd); return; } + if (pull_read_requested && pull_pool_mr) { + if (ep.BindRemoteReadWindow(pull_pool_mr, pull_lease.data(), + pull_lease.size(), &pull_rkey)) { + pull_memory_windows_.fetch_add(1, std::memory_order_relaxed); + } else { + pull_segment_mr = + ep.RegisterRemoteReadRegion(pull_lease.data(), pull_lease.size()); + if (!pull_segment_mr) { + DFKV_LOG_ERROR( + "rdma v2: pull-read Memory Window bind and exact MR fallback " + "both failed on device " + + (dev.empty() ? std::string("(auto)") : dev)); + ::close(boot_fd); + return; + } + pull_rkey = pull_segment_mr->rkey; + pull_mr_fallbacks_.fetch_add(1, std::memory_order_relaxed); + } + } if (!ep.EnsurePoolMrs(user_regions_)) { DFKV_LOG_ERROR("rdma: connection could not attach explicit user MRs"); ::close(boot_fd); @@ -860,9 +925,9 @@ void RdmaServer::Serve(int boot_fd) { if (pull_read_requested) { const rdma::PullArenaInfo pull_info{ reinterpret_cast(pull_lease.data()), pull_lease.size(), - writer_token, pull_segment_mr->rkey, static_cast(K)}; - readiness_bytes = rdma::EncodeV2PullReadiness( - info, writer_token, pull_info, readiness); + writer_token, pull_rkey, static_cast(K)}; + readiness_bytes = + rdma::EncodeV2PullReadiness(info, writer_token, pull_info, readiness); } else { readiness_bytes = rdma::EncodeV2Readiness(info, writer_token, readiness); @@ -2037,6 +2102,15 @@ std::string RdmaServer::MetricsText() const { segment.largest_free_range); m(s, "dfkv_rdma_recv_segment_growths_total", "counter", "Receive chunks committed after startup", segment.growths); + m(s, "dfkv_rdma_recv_segment_shrinks_total", "counter", + "Receive-pool trim passes that released empty non-initial chunks", + segment.shrinks); + m(s, "dfkv_rdma_recv_segment_released_bytes_total", "counter", + "Receive-pool bytes returned after the idle hold period", + segment.released_bytes); + m(s, "dfkv_rdma_recv_segment_chunk_idle_ms", "gauge", + "Idle hold before an empty non-initial receive chunk is released", + recv_chunk_idle_ms_); m(s, "dfkv_rdma_recv_segment_growth_failures_total", "counter", "Receive-pool growth attempts rejected by budget or allocation", segment.growth_failures); @@ -2049,6 +2123,12 @@ std::string RdmaServer::MetricsText() const { m(s, "dfkv_rdma_pull_connections", "gauge", "Connections currently holding negotiated pull-read arenas", pull_connections_.load(std::memory_order_relaxed)); + m(s, "dfkv_rdma_pull_memory_windows_total", "counter", + "Pull-read connections isolated with type-2 Memory Windows", + pull_memory_windows_.load(std::memory_order_relaxed)); + m(s, "dfkv_rdma_pull_mr_fallbacks_total", "counter", + "Pull-read connections using exact per-connection MR fallback", + pull_mr_fallbacks_.load(std::memory_order_relaxed)); m(s, "dfkv_rdma_legacy_connections", "gauge", "Connections currently holding only legacy receive arenas", legacy_connections_.load(std::memory_order_relaxed)); diff --git a/src/cache/rdma_server.h b/src/cache/rdma_server.h index 564622de..ef870876 100644 --- a/src/cache/rdma_server.h +++ b/src/cache/rdma_server.h @@ -209,6 +209,9 @@ class RdmaServer { int port_ = 0; std::atomic running_{false}; std::thread accept_thread_; + std::thread recv_trim_thread_; + std::mutex recv_trim_mu_; + std::condition_variable recv_trim_cv_; // Track per-connection Serve threads + their endpoints so Stop() can wake them // out of WaitComp and join them before the handler's owner is destroyed, and // so finished threads are reaped incrementally (see ReapDoneLocked). @@ -224,8 +227,11 @@ class RdmaServer { std::atomic segment_evictions_{0}; size_t recv_segment_max_bytes_ = 0; size_t recv_segment_chunk_bytes_ = 0; + uint64_t recv_chunk_idle_ms_ = 60000; size_t recv_segment_registered_rails_ = 0; std::atomic pull_connections_{0}; + std::atomic pull_memory_windows_{0}; + std::atomic pull_mr_fallbacks_{0}; std::atomic legacy_connections_{0}; std::atomic data_connection_bytes_{0}; std::atomic control_connection_bytes_{0}; diff --git a/src/transport/rdma_recv_segment.cc b/src/transport/rdma_recv_segment.cc index 738eec4d..16635a46 100644 --- a/src/transport/rdma_recv_segment.cc +++ b/src/transport/rdma_recv_segment.cc @@ -1,12 +1,15 @@ #include "transport/rdma_recv_segment.h" #include +#include #include #include #include #include #include +#include "utils/numa_util.h" + namespace dfkv::rdma { namespace { @@ -22,6 +25,13 @@ size_t AlignUpChecked(size_t value, size_t alignment) { return (value + alignment - 1) & ~(alignment - 1); } +uint64_t SteadyMs() { + return static_cast( + std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()) + .count()); +} + } // namespace size_t ResolveRecvSegmentBytes(const char* value, size_t fallback, @@ -76,7 +86,7 @@ void RecvSegment::Lease::Reset() { RecvSegment::~RecvSegment() { std::free(data_); } -bool RecvSegment::Init(size_t bytes, size_t alignment) { +bool RecvSegment::Init(size_t bytes, size_t alignment, int numa_node) { if (data_) return size_ == bytes && alignment_ == alignment; if (bytes == 0 || !IsPowerOfTwo(alignment) || bytes % alignment != 0 || alignment < sizeof(void*)) { @@ -85,6 +95,7 @@ bool RecvSegment::Init(size_t bytes, size_t alignment) { void* raw = nullptr; if (::posix_memalign(&raw, alignment, bytes) != 0) return false; data_ = static_cast(raw); + numa::BindMemory(raw, bytes, numa_node); size_ = bytes; alignment_ = alignment; free_.emplace(0, bytes); @@ -157,6 +168,37 @@ void RecvSegment::Release(size_t offset, size_t bytes) { free_.emplace(begin, end - begin); } +RecvSegmentPool::Lease::Lease(Lease&& other) noexcept + : pool_(other.pool_), + lease_(std::move(other.lease_)), + segment_(other.segment_) { + other.pool_ = nullptr; + other.segment_ = nullptr; +} + +RecvSegmentPool::Lease& RecvSegmentPool::Lease::operator=( + Lease&& other) noexcept { + if (this == &other) return *this; + Reset(); + pool_ = other.pool_; + lease_ = std::move(other.lease_); + segment_ = other.segment_; + other.pool_ = nullptr; + other.segment_ = nullptr; + return *this; +} + +RecvSegmentPool::Lease::~Lease() { Reset(); } + +void RecvSegmentPool::Lease::Reset() { + RecvSegmentPool* pool = pool_; + RecvSegment* segment = segment_; + lease_.Reset(); + pool_ = nullptr; + segment_ = nullptr; + if (pool && segment) pool->NoteRelease(segment); +} + bool RecvSegmentPool::Init(size_t chunk_bytes, size_t max_bytes, size_t alignment) { if (!IsPowerOfTwo(alignment) || alignment < sizeof(void*) || @@ -178,8 +220,9 @@ bool RecvSegmentPool::Init(size_t chunk_bytes, size_t max_bytes, chunk_bytes_ = aligned_chunk; max_bytes_ = aligned_max; alignment_ = alignment; - auto initial = std::make_unique(); - if (!initial->Init(chunk_bytes_, alignment_)) { + auto initial = std::make_unique(); + initial->segment = std::make_unique(); + if (!initial->segment->Init(chunk_bytes_, alignment_)) { chunk_bytes_ = 0; max_bytes_ = 0; alignment_ = 0; @@ -189,36 +232,53 @@ bool RecvSegmentPool::Init(size_t chunk_bytes, size_t max_bytes, return true; } -std::unique_ptr RecvSegmentPool::NewChunk( - size_t minimum_bytes) { +std::unique_ptr RecvSegmentPool::NewChunk( + size_t minimum_bytes, int affinity, int numa_node) { const size_t aligned_min = AlignUpChecked(minimum_bytes, alignment_); if (aligned_min == std::numeric_limits::max()) return nullptr; size_t committed = 0; - for (const auto& chunk : chunks_) committed += chunk->size(); + for (const auto& chunk : chunks_) committed += chunk->segment->size(); if (committed >= max_bytes_ || aligned_min > max_bytes_ - committed) return nullptr; const size_t bytes = std::min(std::max(chunk_bytes_, aligned_min), max_bytes_ - committed); - auto chunk = std::make_unique(); - if (!chunk->Init(bytes, alignment_)) return nullptr; + auto chunk = std::make_unique(); + chunk->segment = std::make_unique(); + chunk->affinity = affinity; + if (!chunk->segment->Init(bytes, alignment_, numa_node)) return nullptr; return chunk; } RecvSegmentPool::Lease RecvSegmentPool::Allocate(size_t bytes, - size_t alignment) { + size_t alignment, + int affinity, + int numa_node) { if (bytes == 0 || !IsPowerOfTwo(alignment)) return {}; std::lock_guard lock(mu_); - for (const auto& chunk : chunks_) { - auto lease = chunk->Allocate(bytes, alignment); - if (lease) return Lease(chunk.get(), std::move(lease)); + auto try_chunks = [&](int wanted_affinity) -> Lease { + for (const auto& chunk : chunks_) { + if (chunk->affinity != wanted_affinity) continue; + auto lease = chunk->segment->Allocate(bytes, alignment); + if (lease) { + chunk->empty_since_ms = 0; + return Lease(this, chunk->segment.get(), std::move(lease)); + } + } + return {}; + }; + if (affinity >= 0) { + auto lease = try_chunks(affinity); + if (lease) return lease; } - auto chunk = NewChunk(bytes); + auto shared = try_chunks(-1); + if (shared) return shared; + auto chunk = NewChunk(bytes, affinity, numa_node); if (!chunk) { allocation_failures_.fetch_add(1, std::memory_order_relaxed); growth_failures_.fetch_add(1, std::memory_order_relaxed); return {}; } - RecvSegment* owner = chunk.get(); + RecvSegment* owner = chunk->segment.get(); chunks_.push_back(std::move(chunk)); growths_.fetch_add(1, std::memory_order_relaxed); auto lease = owner->Allocate(bytes, alignment); @@ -226,12 +286,48 @@ RecvSegmentPool::Lease RecvSegmentPool::Allocate(size_t bytes, allocation_failures_.fetch_add(1, std::memory_order_relaxed); return {}; } - return Lease(owner, std::move(lease)); + return Lease(this, owner, std::move(lease)); +} + +void RecvSegmentPool::NoteRelease(RecvSegment* segment) { + std::lock_guard lock(mu_); + for (const auto& chunk : chunks_) { + if (chunk->segment.get() != segment) continue; + if (chunk->segment->stats().used_bytes == 0) + chunk->empty_since_ms = SteadyMs(); + return; + } +} + +size_t RecvSegmentPool::TrimIdle(uint64_t idle_ms) { + if (idle_ms == 0) return 0; + const uint64_t now = SteadyMs(); + size_t released = 0; + std::lock_guard lock(mu_); + for (size_t i = chunks_.size(); i > 1; --i) { + auto& chunk = chunks_[i - 1]; + if (chunk->segment->stats().used_bytes != 0) { + chunk->empty_since_ms = 0; + continue; + } + if (chunk->empty_since_ms == 0) { + chunk->empty_since_ms = now; + continue; + } + if (now - chunk->empty_since_ms < idle_ms) continue; + released += chunk->segment->size(); + chunks_.erase(chunks_.begin() + static_cast(i - 1)); + } + if (released != 0) { + shrinks_.fetch_add(1, std::memory_order_relaxed); + released_bytes_.fetch_add(released, std::memory_order_relaxed); + } + return released; } RecvSegment* RecvSegmentPool::initial_segment() const { std::lock_guard lock(mu_); - return chunks_.empty() ? nullptr : chunks_.front().get(); + return chunks_.empty() ? nullptr : chunks_.front()->segment.get(); } RecvSegmentPool::Stats RecvSegmentPool::stats() const { @@ -240,7 +336,7 @@ RecvSegmentPool::Stats RecvSegmentPool::stats() const { out.max_bytes = max_bytes_; out.chunks = chunks_.size(); for (const auto& chunk : chunks_) { - const RecvSegment::Stats current = chunk->stats(); + const RecvSegment::Stats current = chunk->segment->stats(); out.committed_bytes += current.total_bytes; out.used_bytes += current.used_bytes; out.free_bytes += current.free_bytes; @@ -248,6 +344,8 @@ RecvSegmentPool::Stats RecvSegmentPool::stats() const { std::max(out.largest_free_range, current.largest_free_range); } out.growths = growths_.load(std::memory_order_relaxed); + out.shrinks = shrinks_.load(std::memory_order_relaxed); + out.released_bytes = released_bytes_.load(std::memory_order_relaxed); out.allocation_failures = allocation_failures_.load(std::memory_order_relaxed); out.growth_failures = growth_failures_.load(std::memory_order_relaxed); diff --git a/src/transport/rdma_recv_segment.h b/src/transport/rdma_recv_segment.h index e874d017..f9528a01 100644 --- a/src/transport/rdma_recv_segment.h +++ b/src/transport/rdma_recv_segment.h @@ -58,7 +58,7 @@ class RecvSegment { uint64_t allocation_failures = 0; }; - bool Init(size_t bytes, size_t alignment = 4096); + bool Init(size_t bytes, size_t alignment = 4096, int numa_node = -1); Lease Allocate(size_t bytes, size_t alignment = 4096); char* data() const { return data_; } @@ -87,23 +87,23 @@ class RecvSegmentPool { Lease() = default; Lease(const Lease&) = delete; Lease& operator=(const Lease&) = delete; - Lease(Lease&&) noexcept = default; - Lease& operator=(Lease&&) noexcept = default; + Lease(Lease&& other) noexcept; + Lease& operator=(Lease&& other) noexcept; + ~Lease(); explicit operator bool() const { return static_cast(lease_); } char* data() const { return lease_.data(); } size_t size() const { return lease_.size(); } RecvSegment* segment() const { return segment_; } - void Reset() { - lease_.Reset(); - segment_ = nullptr; - } + void Reset(); private: friend class RecvSegmentPool; - Lease(RecvSegment* segment, RecvSegment::Lease lease) - : lease_(std::move(lease)), segment_(segment) {} + Lease(RecvSegmentPool* pool, RecvSegment* segment, + RecvSegment::Lease lease) + : pool_(pool), lease_(std::move(lease)), segment_(segment) {} + RecvSegmentPool* pool_ = nullptr; RecvSegment::Lease lease_; RecvSegment* segment_ = nullptr; }; @@ -116,25 +116,38 @@ class RecvSegmentPool { size_t largest_free_range = 0; size_t chunks = 0; uint64_t growths = 0; + uint64_t shrinks = 0; + uint64_t released_bytes = 0; uint64_t allocation_failures = 0; uint64_t growth_failures = 0; }; bool Init(size_t chunk_bytes, size_t max_bytes, size_t alignment = 4096); - Lease Allocate(size_t bytes, size_t alignment = 4096); + Lease Allocate(size_t bytes, size_t alignment = 4096, + int affinity = -1, int numa_node = -1); + size_t TrimIdle(uint64_t idle_ms); RecvSegment* initial_segment() const; Stats stats() const; private: - std::unique_ptr NewChunk(size_t minimum_bytes); + struct Chunk { + std::unique_ptr segment; + int affinity = -1; + uint64_t empty_since_ms = 0; + }; + std::unique_ptr NewChunk(size_t minimum_bytes, int affinity, + int numa_node); + void NoteRelease(RecvSegment* segment); size_t chunk_bytes_ = 0; size_t max_bytes_ = 0; size_t alignment_ = 0; mutable std::mutex mu_; - std::vector> chunks_; + std::vector> chunks_; std::atomic growths_{0}; + std::atomic shrinks_{0}; + std::atomic released_bytes_{0}; std::atomic allocation_failures_{0}; std::atomic growth_failures_{0}; }; diff --git a/src/transport/rdma_transport.cc b/src/transport/rdma_transport.cc index ed207ceb..7ba4159d 100644 --- a/src/transport/rdma_transport.cc +++ b/src/transport/rdma_transport.cc @@ -393,6 +393,7 @@ struct RdmaTransport::Conn { rdma::ResourceRequest budget_request; bool budget_held = false; size_t declared_bytes = 0; + size_t depth = 1; Lane lane = Lane::kData; bool active_counted = false; bool live_counted = false; @@ -859,6 +860,12 @@ void RdmaTransport::CompleteRemoteLease(Conn* c, outcome); } +void RdmaTransport::MarkClassOpened(Conn* c) { + if (!c || c->lane == Lane::kControl) return; + std::lock_guard lock(connection_class_mu_); + ++connection_class_stats_[{c->declared_bytes, c->depth}].opened; +} + void RdmaTransport::MarkActive(Conn* c, Lane lane) { if (!c || c->active_counted || c->rail_index >= devs_.size()) return; c->lane = lane; @@ -866,6 +873,10 @@ void RdmaTransport::MarkActive(Conn* c, Lane lane) { const size_t index = static_cast(lane) * devs_.size() + c->rail_index; active_lane_rail_[index].fetch_add(1, std::memory_order_relaxed); + if (lane != Lane::kControl) { + std::lock_guard lock(connection_class_mu_); + ++connection_class_stats_[{c->declared_bytes, c->depth}].active; + } } void RdmaTransport::MarkInactive(Conn* c) { @@ -874,6 +885,12 @@ void RdmaTransport::MarkInactive(Conn* c) { const size_t index = static_cast(c->lane) * devs_.size() + c->rail_index; active_lane_rail_[index].fetch_sub(1, std::memory_order_relaxed); + if (c->lane != Lane::kControl) { + std::lock_guard lock(connection_class_mu_); + auto& active = + connection_class_stats_[{c->declared_bytes, c->depth}].active; + if (active != 0) --active; + } } void RdmaTransport::MarkLive(Conn* c) { @@ -1182,6 +1199,8 @@ RdmaTransport::AcquireResult RdmaTransport::Acquire( const uint64_t now = rdma::RailPolicy::NowMicros(); rail_policy_->Complete(*lease, now - lease_started, completion, now); }; + const size_t required_depth = + ConnectionDepth(node, lane, options.requested_credits); std::vector> pools; std::vector stale; @@ -1214,15 +1233,20 @@ RdmaTransport::AcquireResult RdmaTransport::Acquire( } size_t best = pool_candidates.size(); size_t best_bound = std::numeric_limits::max(); + size_t best_depth = std::numeric_limits::max(); for (size_t i = 0; i < pool_candidates.size(); ++i) { Conn* candidate = pool_candidates[i]; if (candidate->peer_id == peer_snapshot->peer_id && candidate->peer_publication == peer_snapshot->publication && candidate->rail_index == ridx && candidate->declared_bytes >= required_bound && - candidate->declared_bytes < best_bound) { + candidate->depth >= required_depth && + (candidate->declared_bytes < best_bound || + (candidate->declared_bytes == best_bound && + candidate->depth < best_depth))) { best = i; best_bound = candidate->declared_bytes; + best_depth = candidate->depth; } } if (best < pool_candidates.size() && @@ -1267,7 +1291,7 @@ RdmaTransport::AcquireResult RdmaTransport::Acquire( endpoint_cache_misses_.fetch_add(1, std::memory_order_relaxed); // Open, announce, and budget only the size class this operation needs. - const size_t conn_depth = LearnedDepth(node, lane); + const size_t conn_depth = required_depth; const uint64_t conn_declared = lane == Lane::kControl ? static_cast(rdma::kV2ControlCap) @@ -1338,6 +1362,7 @@ RdmaTransport::AcquireResult RdmaTransport::Acquire( remote_lease ? remote_lease->generation : 0; conn->remote_lease_held = remote_lease.has_value(); conn->declared_bytes = static_cast(conn_declared); + conn->depth = conn_depth; conn->remote_recovery_probe = remote_lease && remote_lease->recovery_probe; conn->lease = *lease; @@ -1398,8 +1423,9 @@ RdmaTransport::AcquireResult RdmaTransport::Acquire( return result; } conn->ep.set_remote_depth(remote.depth); - NoteNegotiatedDepth(node, lane, remote.depth); + conn->depth = std::min(conn_depth, remote.depth); if (remote.depth < conn_depth) { + NoteNegotiatedDepth(node, lane, remote.depth); DFKV_LOG_INFO("rdma: server depth " + std::to_string(remote.depth) + " < client depth " + std::to_string(conn_depth) + ": batching window clamped to " + @@ -1464,6 +1490,7 @@ RdmaTransport::AcquireResult RdmaTransport::Acquire( return result; } conns_opened_.fetch_add(1, std::memory_order_relaxed); + MarkClassOpened(conn); rail_conns_[ridx].fetch_add(1, std::memory_order_relaxed); MarkLive(conn); MarkActive(conn, lane); @@ -1673,12 +1700,59 @@ std::vector RdmaTransport::IdleDataBounds( return bounds; } +std::vector RdmaTransport::IdleDataDepths( + const std::string& node) const { + std::lock_guard lock(mu_); + std::vector depths; + const auto found = pool_.find(node); + if (found == pool_.end()) return depths; + depths.reserve(found->second.size()); + for (const Conn* conn : found->second) depths.push_back(conn->depth); + std::sort(depths.begin(), depths.end()); + return depths; +} + std::string RdmaTransport::MetricsText() const { std::string s; s += "# HELP dfkv_rdma_client_conns_opened_total RDMA client connections opened\n"; s += "# TYPE dfkv_rdma_client_conns_opened_total counter\n"; s += "dfkv_rdma_client_conns_opened_total " + std::to_string(conns_opened_.load(std::memory_order_relaxed)) + "\n"; + std::map class_stats; + { + std::lock_guard lock(connection_class_mu_); + class_stats = connection_class_stats_; + } + std::map idle_classes; + { + std::lock_guard lock(mu_); + for (const auto& [node, connections] : pool_) { + (void)node; + for (const Conn* conn : connections) + ++idle_classes[{conn->declared_bytes, conn->depth}]; + } + } + s += "# HELP dfkv_rdma_client_connection_class_opened_total Data QPs opened by adaptive block/depth class\n"; + s += "# TYPE dfkv_rdma_client_connection_class_opened_total counter\n"; + s += "# HELP dfkv_rdma_client_connection_class_active Data QPs currently executing by adaptive block/depth class\n"; + s += "# TYPE dfkv_rdma_client_connection_class_active gauge\n"; + for (const auto& [geometry, stats] : class_stats) { + const std::string labels = + "{block_bytes=\"" + std::to_string(geometry.first) + + "\",depth=\"" + std::to_string(geometry.second) + "\"} "; + s += "dfkv_rdma_client_connection_class_opened_total" + labels + + std::to_string(stats.opened) + "\n"; + s += "dfkv_rdma_client_connection_class_active" + labels + + std::to_string(stats.active) + "\n"; + } + s += "# HELP dfkv_rdma_client_connection_class_idle Data QPs currently idle by adaptive block/depth class\n"; + s += "# TYPE dfkv_rdma_client_connection_class_idle gauge\n"; + for (const auto& [geometry, idle] : idle_classes) { + s += "dfkv_rdma_client_connection_class_idle{block_bytes=\"" + + std::to_string(geometry.first) + "\",depth=\"" + + std::to_string(geometry.second) + "\"} " + std::to_string(idle) + + "\n"; + } const rdma::ResourceRequest resource_used = resource_budget_->used(); const rdma::ResourceRequest resource_limit = resource_budget_->limit(); const auto resource_gauge = [&](const char* name, const char* help, @@ -2083,10 +2157,12 @@ void RdmaTransport::Release(const std::string& node, Lane lane, Conn* c, } else if (lane != Lane::kControl && !v.empty()) { auto largest = std::max_element( v.begin(), v.end(), [](const Conn* left, const Conn* right) { - return left->declared_bytes < right->declared_bytes; + return left->budget_request.registered_bytes < + right->budget_request.registered_bytes; }); if (largest != v.end() && - (*largest)->declared_bytes > c->declared_bytes && + (*largest)->budget_request.registered_bytes > + c->budget_request.registered_bytes && (*largest)->lifecycle.RequestRetire()) { displaced = *largest; if (c->lifecycle.MakeIdle()) { @@ -3156,6 +3232,17 @@ size_t RdmaTransport::ConnectionBound(size_t required) const { : std::min(size_class, logical_max); } +size_t RdmaTransport::ConnectionDepth(const std::string& node, Lane lane, + size_t requested_credits) { + const size_t learned = std::max(1, LearnedDepth(node, lane)); + const size_t wanted = + std::min(learned, std::max(1, requested_credits)); + size_t depth_class = 1; + while (depth_class < wanted && depth_class <= learned / 2) + depth_class <<= 1; + return depth_class < wanted ? learned : std::min(depth_class, learned); +} + // Records n as a high-water candidate and reports whether it exceeds the // logical safety declaration. Physical connection slots use ConnectionBound, // so a rare large object no longer inflates every data connection. diff --git a/src/transport/rdma_transport.h b/src/transport/rdma_transport.h index e5202f68..d9d6379b 100644 --- a/src/transport/rdma_transport.h +++ b/src/transport/rdma_transport.h @@ -216,6 +216,7 @@ class RdmaTransport : public Transport { std::unordered_map> pool_; // Exist/Remove/Members remain isolated from payload transfers. std::vector IdleDataBounds(const std::string& node) const; + std::vector IdleDataDepths(const std::string& node) const; std::unordered_map> control_pool_; // Last successfully published caller memory declarations. RegisterMemory // holds mu_ through per-rail stage/commit, so Acquire can observe either the @@ -239,6 +240,8 @@ class RdmaTransport : public Transport { return declared_ ? static_cast(declared_) : max_payload_; } size_t ConnectionBound(size_t required) const; + size_t ConnectionDepth(const std::string& node, Lane lane, + size_t requested_credits); // Largest block this client has actually handed to the transport. mutable std::atomic max_block_seen_{0}; mutable std::atomic oversize_rejects_{0}; @@ -298,6 +301,14 @@ class RdmaTransport : public Transport { // Connections whose clamped negotiation refunded WR/registered budget. std::atomic depth_refunds_{0}; std::unique_ptr topology_; + using ConnectionClass = std::pair; // block bytes, QP depth + struct ConnectionClassStats { + uint64_t opened = 0; + uint64_t active = 0; + }; + void MarkClassOpened(Conn* c); + mutable std::mutex connection_class_mu_; + std::map connection_class_stats_; std::vector devs_; // stable discovered ACTIVE rail order std::vector> rail_tiers_; std::optional preferred_rail_; diff --git a/src/transport/rdma_verbs.cc b/src/transport/rdma_verbs.cc index dc866b5f..114f9051 100644 --- a/src/transport/rdma_verbs.cc +++ b/src/transport/rdma_verbs.cc @@ -344,6 +344,8 @@ RcEndpoint::~RcEndpoint() { Close(); } void RcEndpoint::Close() { if (qp_) { ibv_destroy_qp(qp_); qp_ = nullptr; } + for (auto* mw : connection_mw_) if (mw) ibv_dealloc_mw(mw); + connection_mw_.clear(); for (auto* m : smr_) if (m) ibv_dereg_mr(m); for (auto* m : rmr_) if (m) ibv_dereg_mr(m); for (auto* m : dmr_) if (m) ibv_dereg_mr(m); @@ -747,6 +749,52 @@ ibv_mr* RcEndpoint::RegisterRemoteReadRegion(void* base, size_t size) { return mr; } +ibv_mr* RcEndpoint::RegisterRemoteReadPool(void* base, size_t size) { + if (!ctx_ || !pd_ || !base || size == 0) return nullptr; + const int access = IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_READ; + const auto b = reinterpret_cast(base); + for (const auto& pool : pool_mr_) { + if (pool.base == b && pool.size >= size && + (pool.access & access) == access) + return pool.mr; + } + ibv_mr* mr = SharedAddPoolMr(ctx_, pd_, base, size, access); + if (!mr) return nullptr; + pool_mr_.insert(pool_mr_.begin(), PoolMr{b, size, access, mr}); + return mr; +} + +bool RcEndpoint::BindRemoteReadWindow(ibv_mr* pool_mr, void* base, + size_t size, uint32_t* rkey) { + if (!qp_ || !pd_ || !pool_mr || !base || size == 0 || !rkey) + return false; + ibv_mw* mw = ibv_alloc_mw(pd_, IBV_MW_TYPE_2); + if (!mw) return false; + ibv_send_wr wr{}, *bad = nullptr; + wr.wr_id = std::numeric_limits::max(); + wr.opcode = IBV_WR_BIND_MW; + wr.send_flags = IBV_SEND_SIGNALED; + wr.bind_mw.mw = mw; + wr.bind_mw.rkey = ibv_inc_rkey(mw->rkey); + wr.bind_mw.bind_info.mr = pool_mr; + wr.bind_mw.bind_info.addr = reinterpret_cast(base); + wr.bind_mw.bind_info.length = size; + wr.bind_mw.bind_info.mw_access_flags = IBV_ACCESS_REMOTE_READ; + if (ibv_post_send(qp_, &wr, &bad) != 0) { + ibv_dealloc_mw(mw); + return false; + } + ibv_wc wc{}; + const int got = WaitComp(&wc, 1, 10000); + if (got != 1 || wc.status != IBV_WC_SUCCESS) { + ibv_dealloc_mw(mw); + return false; + } + connection_mw_.push_back(mw); + *rkey = wr.bind_mw.rkey; + return true; +} + ibv_mr* RcEndpoint::RegisterUser(void* addr, size_t len) { if (!addr || len == 0) return nullptr; const auto address = reinterpret_cast(addr); diff --git a/src/transport/rdma_verbs.h b/src/transport/rdma_verbs.h index 04d59c2a..f9318f0f 100644 --- a/src/transport/rdma_verbs.h +++ b/src/transport/rdma_verbs.h @@ -168,6 +168,12 @@ class RcEndpoint { // Register an exact connection-private source arena for initiator READ. // Unlike RegisterRemoteRegion this never widens to a shared segment MR. ibv_mr* RegisterRemoteReadRegion(void* base, size_t size); + // Register a whole receive-pool chunk once per shared PD with REMOTE_READ. + // The broad MR rkey is never published; BindRemoteReadWindow narrows peer + // access to one connection's exact lease with a type-2 Memory Window. + ibv_mr* RegisterRemoteReadPool(void* base, size_t size); + bool BindRemoteReadWindow(ibv_mr* pool_mr, void* base, size_t size, + uint32_t* rkey); // Cumulative one-shot user MRs registered outside explicit pool regions. // These are never cached; TransientUserMrActive is the lifetime invariant. @@ -331,6 +337,7 @@ class RcEndpoint { // DONE, or after explicit responder-retirement proof on failure. std::vector transient_mr_; std::vector connection_mr_; + std::vector connection_mw_; QpInfo local_; std::atomic responder_cancelled_{false}; size_t pending_responder_writes_ = 0; // responder owner thread only diff --git a/test/python/test_observability_contract.py b/test/python/test_observability_contract.py index da300f70..d30e072f 100644 --- a/test/python/test_observability_contract.py +++ b/test/python/test_observability_contract.py @@ -104,11 +104,16 @@ "dfkv_rdma_recv_segment_max_bytes", "dfkv_rdma_recv_segment_chunks", "dfkv_rdma_recv_segment_growths_total", + "dfkv_rdma_recv_segment_shrinks_total", + "dfkv_rdma_recv_segment_released_bytes_total", + "dfkv_rdma_recv_segment_chunk_idle_ms", "dfkv_rdma_recv_segment_growth_failures_total", "dfkv_rdma_recv_segment_used_bytes", "dfkv_rdma_recv_segment_largest_free_range_bytes", "dfkv_rdma_recv_segment_allocation_failures_total", "dfkv_rdma_pull_connections", + "dfkv_rdma_pull_memory_windows_total", + "dfkv_rdma_pull_mr_fallbacks_total", "dfkv_rdma_legacy_connections", "dfkv_rdma_connection_bytes", "dfkv_ram_arena_bytes", diff --git a/test/transport/rdma_loopback_test.cc b/test/transport/rdma_loopback_test.cc index bf46b39e..f4f57970 100644 --- a/test/transport/rdma_loopback_test.cc +++ b/test/transport/rdma_loopback_test.cc @@ -142,6 +142,16 @@ class RdmaTransportTestPeer { RdmaTransport* transport, const std::string& node) { return transport->IdleDataBounds(node); } + static std::vector IdleDataDepths( + RdmaTransport* transport, const std::string& node) { + return transport->IdleDataDepths(node); + } + + static size_t ConnectionDepth(RdmaTransport* transport, + const std::string& node, size_t requested) { + return transport->ConnectionDepth(node, RdmaTransport::Lane::kData, + requested); + } static uint64_t PeerPublication(const RdmaTransport& transport, const std::string& node) { @@ -1823,6 +1833,10 @@ TEST(RdmaLoopback, MetricsCountersTrackOps) { "dfkv_rdma_recv_segment_largest_free_range_bytes"), 0); EXPECT_GE(CounterVal(srv_text, "dfkv_rdma_pull_connections"), 1); + EXPECT_GE( + CounterVal(srv_text, "dfkv_rdma_pull_memory_windows_total") + + CounterVal(srv_text, "dfkv_rdma_pull_mr_fallbacks_total"), + 1); EXPECT_GT(CounterVal( srv_text, "dfkv_rdma_connection_bytes{class=\"data\"}"), @@ -3384,12 +3398,98 @@ TEST(RdmaLoopback, DataConnectionsUseActualBlockSizeClasses) { 65536u); } +TEST(RdmaLoopback, DataConnectionsUseActualDepthClasses) { + if (!HaveRdma()) GTEST_SKIP() << "no RDMA device"; + ScopedEnv depth("DFKV_RDMA_DEPTH", "4"); + ScopedEnv max_block("DFKV_RDMA_MAX_BLOCK_BYTES", "262144"); + ScopedEnv min_block("DFKV_RDMA_CONNECTION_MIN_BLOCK_BYTES", "4096"); + RdmaNode node("adaptive-depth"); + RdmaTransport transport(kMaxMsg); + KVClient client({{"n", node.addr}}, SelfHdr(), &transport); + + std::string scalar(4000, 's'); + ASSERT_TRUE(client.Put("scalar", scalar.data(), scalar.size())); + EXPECT_EQ(RdmaTransportTestPeer::IdleDataDepths(&transport, node.addr), + (std::vector{1})); + + std::vector values(4, std::string(4000, 'b')); + std::vector items; + for (size_t i = 0; i < values.size(); ++i) { + items.push_back(CacheItem{ + ToBlockKey(SelfHdr(), "batch-depth-" + std::to_string(i)), + values[i].data(), values[i].size()}); + } + const auto statuses = transport.CacheMany(node.addr, items); + ASSERT_EQ(statuses.size(), items.size()); + for (Status status : statuses) EXPECT_EQ(status, Status::kOk); + EXPECT_EQ(RdmaTransportTestPeer::IdleDataDepths(&transport, node.addr), + (std::vector{1, 4})); + + EXPECT_EQ(RdmaTransportTestPeer::ConnectionDepth(&transport, node.addr, 1), + 1u); + EXPECT_EQ(RdmaTransportTestPeer::ConnectionDepth(&transport, node.addr, 3), + 4u); + const std::string metrics = transport.MetricsText(); + EXPECT_NE(metrics.find( + "dfkv_rdma_client_connection_class_opened_total" + "{block_bytes=\"4096\",depth=\"1\"} 1"), + std::string::npos); + EXPECT_NE(metrics.find( + "dfkv_rdma_client_connection_class_opened_total" + "{block_bytes=\"4096\",depth=\"4\"} 1"), + std::string::npos); + EXPECT_NE(metrics.find( + "dfkv_rdma_client_connection_class_idle" + "{block_bytes=\"4096\",depth=\"1\"} 1"), + std::string::npos); +} + +TEST(RdmaLoopback, ManyAdaptiveDepthOneConnectionsStayBounded) { + if (!HaveRdma()) GTEST_SKIP() << "no RDMA device"; + int count = 32; + if (const char* value = std::getenv("DFKV_RDMA_TEST_CONNECTIONS")) + count = std::max(1, std::min(2000, std::atoi(value))); + ScopedEnv depth("DFKV_RDMA_DEPTH", "4"); + ScopedEnv max_block("DFKV_RDMA_MAX_BLOCK_BYTES", "4096"); + ScopedEnv min_block("DFKV_RDMA_CONNECTION_MIN_BLOCK_BYTES", "4096"); + ScopedEnv qp_budget("DFKV_RDMA_QP_BUDGET", "4096"); + ScopedEnv wr_budget("DFKV_RDMA_WR_BUDGET", "16384"); + ScopedEnv registered_budget("DFKV_RDMA_REGISTERED_BYTES_BUDGET", + "2147483648"); + ScopedEnv endpoint_budget("DFKV_RDMA_ENDPOINT_CACHE_MAX", "4096"); + ScopedEnv recv_max("DFKV_RDMA_RECV_SEGMENT_SIZE", "1073741824"); + ScopedEnv recv_chunk("DFKV_RDMA_RECV_CHUNK_BYTES", "67108864"); + RdmaNode node("many-adaptive-conns"); + + std::vector> transports; + std::vector> clients; + transports.reserve(static_cast(count)); + clients.reserve(static_cast(count)); + std::string value(4000, 'c'); + for (int i = 0; i < count; ++i) { + auto transport = std::make_unique(kMaxMsg); + auto client = std::make_unique( + std::vector>{{"n", node.addr}}, + SelfHdr(), transport.get()); + ASSERT_TRUE(client->Put("conn-" + std::to_string(i), value.data(), + value.size())) + << "connection " << i; + transports.push_back(std::move(transport)); + clients.push_back(std::move(client)); + } + const std::string metrics = node.rsrv->MetricsText(); + EXPECT_GE(CounterVal(metrics, "dfkv_rdma_active_conns"), count); + EXPECT_LE(CounterVal(metrics, + "dfkv_rdma_connection_bytes{class=\"data\"}"), + static_cast(count) * 32 * 1024); +} + TEST(RdmaLoopback, ServerReceivePoolCommitsChunksOnDemand) { if (!HaveRdma()) GTEST_SKIP() << "no RDMA device"; ScopedEnv recv_max("DFKV_RDMA_RECV_SEGMENT_SIZE", "4194304"); ScopedEnv recv_chunk("DFKV_RDMA_RECV_CHUNK_BYTES", "1048576"); - ScopedEnv max_block("DFKV_RDMA_MAX_BLOCK_BYTES", "65536"); - ScopedEnv min_block("DFKV_RDMA_CONNECTION_MIN_BLOCK_BYTES", "4096"); + ScopedEnv max_block("DFKV_RDMA_MAX_BLOCK_BYTES", "262144"); + ScopedEnv min_block("DFKV_RDMA_CONNECTION_MIN_BLOCK_BYTES", "262144"); RdmaNode node("lazy-recv-pool"); std::vector> transports; diff --git a/test/transport/rdma_recv_segment_test.cc b/test/transport/rdma_recv_segment_test.cc index 6e23e77a..8b04fe70 100644 --- a/test/transport/rdma_recv_segment_test.cc +++ b/test/transport/rdma_recv_segment_test.cc @@ -1,6 +1,8 @@ #include "transport/rdma_recv_segment.h" +#include #include +#include #include #include @@ -148,4 +150,39 @@ TEST(RecvSegmentPool, RejectsInvalidOrOversizedGeometry) { EXPECT_EQ(pool.stats().committed_bytes, 8192u); } +TEST(RecvSegmentPool, AffinityPreventsCrossRailChunkReuse) { + RecvSegmentPool pool; + ASSERT_TRUE(pool.Init(16u << 10, 48u << 10)); + auto initial = pool.Allocate(12u << 10, 4096, 0); + auto rail0 = pool.Allocate(12u << 10, 4096, 0); + auto rail1 = pool.Allocate(12u << 10, 4096, 1); + ASSERT_TRUE(initial); + ASSERT_TRUE(rail0); + ASSERT_TRUE(rail1); + EXPECT_NE(rail0.segment(), rail1.segment()); + + rail0.Reset(); + EXPECT_FALSE(pool.Allocate(12u << 10, 4096, 1)) + << "rail1 must not borrow an empty chunk bound to rail0"; +} + +TEST(RecvSegmentPool, TrimsEmptyNonInitialChunksAfterIdleHold) { + RecvSegmentPool pool; + ASSERT_TRUE(pool.Init(16u << 10, 48u << 10)); + auto first = pool.Allocate(12u << 10); + auto second = pool.Allocate(12u << 10); + auto third = pool.Allocate(12u << 10); + ASSERT_EQ(pool.stats().chunks, 3u); + first.Reset(); + second.Reset(); + third.Reset(); + std::this_thread::sleep_for(std::chrono::milliseconds(3)); + EXPECT_EQ(pool.TrimIdle(1), 32u << 10); + const auto stats = pool.stats(); + EXPECT_EQ(stats.chunks, 1u); + EXPECT_EQ(stats.committed_bytes, 16u << 10); + EXPECT_EQ(stats.shrinks, 1u); + EXPECT_EQ(stats.released_bytes, 32u << 10); +} + } // namespace dfkv::rdma