From 7eee1e99abb3a5c59a1bebf3f1c38998e7b90fe7 Mon Sep 17 00:00:00 2001 From: Ketor Date: Fri, 28 Aug 2026 18:11:18 +0800 Subject: [PATCH] perf: scale RDMA capacity and bound watermark reclaim --- CHANGELOG.md | 23 ++++ README.md | 21 ++-- docs/ARCHITECTURE.md | 28 ++--- docs/CONNECTORS.md | 73 +++++++------ docs/DEPLOY.md | 37 +++---- docs/METRICS.md | 30 +++--- src/cache/dfkv_server_main.cc | 11 +- src/cache/disk_cache_group.cc | 11 ++ src/cache/disk_slab_store.cc | 75 +++++++++++-- src/cache/disk_slab_store.h | 16 ++- src/cache/kv_node_server.cc | 17 +++ src/cache/rdma_server.cc | 114 ++++++++++++-------- src/cache/rdma_server.h | 19 ++-- src/cache/slab_allocator.cc | 9 +- src/cache/slab_allocator.h | 5 + src/transport/rdma_recv_segment.cc | 97 +++++++++++++++++ src/transport/rdma_recv_segment.h | 65 +++++++++++ src/transport/rdma_transport.cc | 119 ++++++++++++++++----- src/transport/rdma_transport.h | 24 ++--- test/cache/disk_slab_store_test.cc | 38 +++++++ test/cache/slab_allocator_test.cc | 31 ++++++ test/python/test_observability_contract.py | 10 ++ test/transport/rdma_loopback_test.cc | 83 +++++++++++++- test/transport/rdma_recv_segment_test.cc | 42 ++++++++ 24 files changed, 801 insertions(+), 197 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a21081b5..1c7849da 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,29 @@ ## Unreleased +### Adaptive RDMA receive capacity and bounded high-water reclaim + +- Replaced the eagerly committed monolithic RDMA receive segment with a lazy + chunk pool. `DFKV_RDMA_RECV_SEGMENT_SIZE` remains the hard process budget; + `DFKV_RDMA_RECV_CHUNK_BYTES` (default 256 MiB) controls incremental commits. + Each connection publishes and registers only the chunk containing its lease. +- Decoupled `DFKV_RDMA_MAX_BLOCK_BYTES` from physical connection geometry. It + remains the logical safety ceiling, while data QPs now negotiate a + power-of-two class sized to the current operation, with + `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. +- 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. +- 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 + 64 GiB high-water PUT improved from 5.39 to 6.71 GB/s, p99 from 63.8 to + 24.1 ms, and max latency from 496.8 to 50.8 ms. + ### vLLM hybrid state and explicit TCP staging - Accepted align-mode hybrid cache groups whose Mamba block size differs from diff --git a/README.md b/README.md index 3302b6a4..ecc9559f 100644 --- a/README.md +++ b/README.md @@ -265,10 +265,10 @@ docs/ ARCHITECTURE.md (layers · storage engines · RAM hot tier · wire p IP. A v2 capability probe is mandatory; once RDMA is requested it never switches transports. - **One-sided zero-copy data plane**: control descriptors/status use small 4-KiB - SEND/RECV buffers. PUT payloads RDMA-WRITE into leases from one process-wide, - pre-registered receive segment; GET payloads RDMA-WRITE directly into the - caller's registered buffer. No connection-sized block buffers or payload copy - are used. + SEND/RECV buffers. PUT payloads RDMA-WRITE into a lease from a lazily committed + server receive chunk; GET payloads RDMA-WRITE directly into the caller's + registered buffer. Data QPs negotiate a power-of-two slot class from the + current operation rather than reserving the logical maximum block size. - **Optional pipelining** (`DFKV_RDMA_DEPTH=K`): K requests in flight per connection. A network-latency hider, **not a throughput knob** — GET and PUT are both depth-flat (the per-connection serve loop is in-order; benchmarked GET ~1.24 GB/s at @@ -280,8 +280,8 @@ docs/ ARCHITECTURE.md (layers · storage engines · RAM hot tier · wire p credits waits under the existing bounded backpressure contract; congestion never overflows traffic into Tier 1. Lower tiers are considered only when the peer-health intersection makes every higher tier unavailable. Server threads - follow their QP rail NUMA node; the shared receive segment is registered on - each selected rail but is not separately NUMA-allocated per rail. + follow their QP rail NUMA node; receive chunks register lazily on the rail + that first leases them rather than pinning the full budget on every rail. - **Generation-fenced endpoint recovery**: each public operation captures one peer-topology generation for both of its possible physical attempts. Connections record that generation. A stale-generation pool entry is never @@ -319,8 +319,9 @@ fabric selection and capacity explicit. |---|---|---| | `--rdma-dev` | leave unset for one local HCA; list the fabric explicitly for multi-rail | Unset resolves and initializes the first `ACTIVE` local HCA (peer names may differ). An explicit comma list defines a fixed topology in first-occurrence order: every listed device must be present/openable with complete provider metadata, including a successful GID query, and complete anchor/MR initialization; a present `DOWN` entry remains initialized and monitored rather than being rejected. Both hosts still require compatible names/fabric. | | `DFKV_DISK_HASH_WEIGHT` | `10` | Flattens the intra-server disk ring share from ±20 % to ±3 % so the hottest disk stops gating the whole node (+5–6 % cold read, ~2× lower p99). **Re-routes existing keys** (cache miss + refill) — flip together with a restart/upgrade window. | -| `--rdma-depth` | `4` (default) | Handshake window is `min(client, server)`. Production connector batches require both sides to expose the same bounded window. Pull-read connections lease receive and source arenas: `2 × depth × slot_size`; with a 4 MiB declaration this is 33,587,200 bytes per data QP. | -| `DFKV_RDMA_RECV_SEGMENT_SIZE` | `64 GiB` on xb01 (`16 GiB` code default) | The xb01 14×TP8 envelope is 896 retained data plus 896 smaller control QPs; including 25% churn requires 35.4 GiB. Keep 64 GiB; 128 GiB adds no capacity benefit for this topology. | +| `--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_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,6 +338,8 @@ fabric selection and capacity explicit. | `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_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. | | `DFKV_RDMA_RAIL_COOLDOWN_MS` | `5000` (default) | Quarantine duration. No admission occurs during cooldown; afterward exactly one real operation is admitted as the recovery probe. Probe success clears the rail state, local failure starts another cooldown, and endpoint failure proves neither recovery nor a new local fault. | | local-rail attempts | fixed at `2` (not configurable) | One initial attempt plus at most one fresh retry. GET retries only unfinished work. PUT crosses rails only before request post; ambiguous post-submit failure is returned without replay. Both attempts retain one captured peer generation. | @@ -345,7 +348,7 @@ fabric selection and capacity explicit. | `DFKV_RDMA_ENDPOINT_CACHE_MAX` | autoscaled (floor `256`, process-wide) | Hard cap on live client RDMA endpoints. Unset, the cap follows the adopted ring: `nodes × 2 pools × max(pool limit, configured rails) × 1.25`, raised but never shrunk. Setting any budget env pins the complete budget and disables autoscaling. | | `DFKV_RDMA_QP_BUDGET` | endpoint cap (default) | Process-wide QP reservation limit. Keep equal to the endpoint cap while each endpoint owns one QP. | | `DFKV_RDMA_WR_BUDGET` | endpoint cap × `DFKV_RDMA_DEPTH` | Process-wide negotiated WR-slot reservation for unified data and control endpoints. | -| `DFKV_RDMA_REGISTERED_BYTES_BUDGET` | endpoint cap × depth × 2 × negotiated receive-slot bytes | Bounds aggregate server receive + pull arena bytes represented by live client endpoints. | +| `DFKV_RDMA_REGISTERED_BYTES_BUDGET` | endpoint cap × depth × 2 × logical-max slot bytes | Admission ceiling only. Actual reservation is charged from each adaptive connection class, so small QPs consume small budget. | | `DFKV_RDMA_RESOURCE_ACQUIRE_MS` | `10000` | Bounded wait for all endpoint/QP/WR/receive-slot credits before opening a new QP. Timeout fails the shard without oversubscribing resources. | | `DFKV_FANOUT_THREADS` | unset (default 32) | Bounds TCP batch, RDMA write, and compatibility helpers; RDMA reads use the separate bounded scheduler below. | | `DFKV_RDMA_READ_WORKERS` | `7` (default) | Process-wide hard cap for active RDMA GET/GET-Auto/SG-GET/Exist shards across every client handle. Concurrent batches advance round-robin. Seven workers replace the legacy executor's seven helpers per rank without adding a thread; unlike the legacy caller-participates path, active read QPs cannot grow with concurrent callers. Raise only when an A/B proves seven active depth-four QPs cannot saturate the target fabric. | diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 5fd0aaf2..f399a229 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -161,18 +161,20 @@ Production discovery uses MDS. ### 4.1 RDMA v2 resource and failure invariants -- `RdmaServer::Start` allocates one aligned `RecvSegment` per process, then - registers it once on every anchored rail's shared PD. Every endpoint on that - rail reuses the same MR; endpoint close drops only its shared-device reference. -- DCP2 advertises the client's exact `DFKV_RDMA_MAX_BLOCK_BYTES`. The server - validates it against `--max-msg`, negotiates `qd=min(client depth, server depth)`, - and leases `qd` contiguous slots. The connection keeps that lease—including - while idle in the client pool—until QP teardown or idle reclaim. -- A data slot is `align4K(4096 + declared_block)`: 4 KiB of wire-prefix space - plus the declared maximum raw payload, with no value-envelope allowance. - Segment sizing must cover every live and pooled data/control QP across all - client ranks/processes, not only currently in-flight operations. Exhaustion - rejects the connection; it is never reinterpreted as another protocol. +- `RdmaServer::Start` commits one aligned receive chunk. Additional chunks are + committed on allocation misses up to `DFKV_RDMA_RECV_SEGMENT_SIZE`; only the + chunk leased by a connection is registered on that endpoint's rail. +- `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. +- 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. - 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 @@ -403,7 +405,7 @@ data plane or the connection fails. | RAM tier lock shards | `--ram-tier-shards` / `DFKV_RAM_TIER_SHARDS` | 8 | 1-64; auto-halved while a shard would hold <32 extents | | read convoy/direct promotion | `DFKV_READ_COALESCE=1` | off | collapses identical reads; with RAM + io_uring, whole-value disk misses read directly into a hidden registered arena reservation | | RDMA transport | build `-DDFKV_WITH_RDMA=ON`, `DFKV_RDMA=1` | TCP | active-HCA discovery; `DFKV_RDMA_DEV` is an optional whitelist | -| RDMA v2 | `DFKV_RDMA=1` | TCP when RDMA was not requested | bounded 32,786-byte control buffers (32-KiB Members data) + mandatory shared registered receive segment | +| RDMA v2 | `DFKV_RDMA=1` | TCP when RDMA was not requested | bounded control buffers + adaptive data-QP classes + lazily committed receive chunks | | io_uring async GET | build `-DDFKV_WITH_URING`; `DFKV_SERVER_URING=0` disables | on when built, unavailable otherwise | RDMA v2 disk-read path | | first-request absolute deadline | `DFKV_TCP_FIRST_REQ_MS` / `DFKV_MDS_FIRST_REQ_MS` / `DFKV_METRICS_FIRST_REQ_MS` | 30000 ms; `0` = off | anchored at accept: the first complete frame/request line is due within this window, so a drip feeder cannot pin a handler thread; `dfkv_server` also publishes its resolved value as the `dfkv_tcp_first_req_ms` gauge | | MDS legacy control-plane shim | `DFKV_MDS_ACCEPT_LEGACY=1` | off (strict epoch gate) | widens the MDS control-plane version gate by exactly one epoch so v1.x peers are served with legacy 42/10-byte framing during a mixed-generation migration; data-plane listeners stay strictly epoch 6/7; drop the env once `dfkv_mds_legacy_frames_total` drains to zero | diff --git a/docs/CONNECTORS.md b/docs/CONNECTORS.md index 1f2918df..9c4ff1c0 100644 --- a/docs/CONNECTORS.md +++ b/docs/CONNECTORS.md @@ -84,10 +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)` 作为安全窗口。每连接注册 `2 × depth × (18 B + 32 KiB)` 的有界 SEND/RECV control buffer；pull-read 连接还从共享 segment 各租 `depth` 个 receive slot 和 source slot。 | -| `DFKV_RDMA_MAX_BLOCK_BYTES` | 4 MiB 安全上限 | 按连接器块几何精确设置 | DCP2 声明本连接最大 PUT/GET block，决定共享 segment 的 slot 大小；超声明请求在客户端失败且不上 wire。声明越准，同一 segment 可容纳的 live/pooled v2 连接越多。 | -| `DFKV_RDMA_RECV_SEGMENT_SIZE` | 16 GiB | 按下文 live/pooled 连接公式设置；xb01 为 64 GiB | server 启动时申请，并在每个选中 rail 的共享 PD 上注册；失败会拒绝启动，segment 无可用 lease 时拒绝新连接。 | -| `DFKV_RDMA_NUMA` | `0` | 显式多轨的大机可设 `1` | 建连时按调用线程 NUMA 选本地 rail（无本地 rail→轮转白名单），server serve 线程跟随 QP rail。单块共享 receive segment 不做 per-rail NUMA 分配；仅保证选轨/线程亲和。 | +| `DFKV_RDMA_DEPTH` | `4` | 两侧可不同，按容量选 | 握手协商 `min(client,server)`。每条 pull data QP 租 `2 × depth × adaptive_slot`；control buffer 始终有界。 | +| `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_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 不超过取整后的预算。 | | `DFKV_CUDA_PINNED_SLOT_BYTES` | 4 MiB（4194304） | 保持能容纳常见单块；不必按最大 payload 预分配 | 固定 publication slot 大小，取 4 KiB–64 MiB 的正十进制字节数。非法值告警并回退默认 sizing。大 payload 按 slot 大小分块发布，不要求一个 payload 对应一个 slot。 | @@ -120,30 +122,36 @@ memory 留余量。pool acquisition、pin/register、CUDA copy 或 stream sync 归还后释放其余 pool slot 并记录 driver 释放错误；`fork()` 后的 child 拒绝使用 parent 继承的 CUDA pool 状态。 -#### 1.2.1 `DFKV_RDMA_MAX_BLOCK_BYTES` 怎么定（含 L2 / L2-bypass 两套公式） +#### 1.2.1 逻辑块上限、connection class 与 receive budget -这个值在 **v2** 决定共享 receive segment 的 slot 大小： -`align4K(4 KiB + 声明的最大 raw payload)`。每条 pull-read 数据连接分别租 -`depth` 个 receive slot 和 source slot；所有连接共享一块启动期注册的 -`DFKV_RDMA_RECV_SEGMENT_SIZE`，因此声明保持精确仍有价值。 - -**共享 segment 容量必须按连接寿命算，不是按同时在飞请求算。** 数据 QP 的 -slot 为 +`DFKV_RDMA_MAX_BLOCK_BYTES` 是逻辑安全上限。每次 data Acquire 取当前 +scalar/batch/SG 操作里的最大对象，计算： +```text +connection_class = + min(logical_max, + next_power_of_two(max(actual_bytes, + DFKV_RDMA_CONNECTION_MIN_BLOCK_BYTES))) +S_data = align4K(4096 + connection_class) +B_connection = 2 × depth × S_data ``` -S_data = align4K(4096 + DFKV_RDMA_MAX_BLOCK_BYTES) -S_control = align4K(4096 + (18 + 32768)) = 40960 -B_required >= 2 × (N_data × depth × S_data - + N_control × depth × S_control) + +同一 peer/rail 的 idle pool 选择最小可满足 class；没有才建新 QP。pool 满时， +较小返回连接可替换最大的 idle QP。偶发大对象因此只放大自己的连接，不放大所有 +rank/process 的常驻连接。 + +server receive pool 以 `DFKV_RDMA_RECV_CHUNK_BYTES` 惰性提交，累计不超过 +`DFKV_RDMA_RECV_SEGMENT_SIZE`。预算仍按所有 rank/process 的 peak live + +pooled QP 规划： + +```text +B_required >= Σ B_connection(actual class) ``` -`N_data` / `N_control` 是该 server 上所有 rank、进程的**峰值 live + client -pool 中空闲连接**；lease 一直保留到 QP 被销毁或 `DFKV_RDMA_IDLE_MS` 回收， -线程峰值留下的 pooled QP 也要计入。4 MiB 声明、depth=4 时每条 pull data -QP 占 33,587,200 B；16 GiB 最多约 511 条、64 GiB 约 2046 条（均未扣 -control lease）。上线同时观察 -`dfkv_rdma_recv_segment_free_bytes` 与 `dfkv_rdma_v2_ready`；free 接近 0 -即扩容或缩小声明/depth/pool，避免新连接被拒绝。 +上线观察 committed/max/chunks/used/free/growth failures。hard budget 耗尽才拒绝 +新连接；增大 bootstrap timeout 不会增加容量。 + +块大小公式仍用于确定 logical max，但无需把该最大值代入每条 QP： **块大小取决于走哪条路径**——两条路径的分块规则不同： @@ -568,20 +576,19 @@ device segment 分组，不写入 namespace 或 raw value。 #### 冷连接池 -server 启动期会注册 process-wide receive segment；新 data QP 只创建小 -control buffers 并租 `depth × slot` 的 offset，不会为每连接注册 -`qd × block`。若首轮连接失败，检查 `dfkv_rdma_v2_ready`、receive segment -free bytes、每轨注册状态和 server 协商日志；不要用增大握手超时掩盖共享 -segment 容量或注册失败。 +server 启动只注册首个 receive chunk；新 data QP 租当前 operation class 对应的 +`depth × slot`，需要时再提交 chunk。若首轮连接失败，检查 +`dfkv_rdma_recv_segment_{bytes,max_bytes,chunks,free_bytes,growth_failures_total}` +和 `dfkv_rdma_v2_ready`；不要用增大握手超时掩盖 hard budget 或 HCA 注册失败。 #### 服务端 -L2-bypass 不需要独立 server 协议，但两项决定 v2 容量： +L2-bypass 不需要独立 server 协议，但三项决定 v2 容量： -- `DFKV_RDMA_RECV_SEGMENT_SIZE`：按 §1.2.1 的 peak live/pooled QP 公式； - 观察 `dfkv_rdma_recv_segment_free_bytes` 和 `dfkv_rdma_v2_ready` -- `DFKV_RDMA_MAX_BLOCK_BYTES` + `DFKV_RDMA_DEPTH`：共同决定每 QP 的 lease； - 声明要覆盖原版 L2 的较大整页对象，且总 lease 必须纳入共享 segment 预算。 +- `DFKV_RDMA_RECV_SEGMENT_SIZE`：receive-pool hard budget； +- `DFKV_RDMA_RECV_CHUNK_BYTES`：启动/增量提交粒度； +- `DFKV_RDMA_MAX_BLOCK_BYTES`、`DFKV_RDMA_CONNECTION_MIN_BLOCK_BYTES` 与 + `DFKV_RDMA_DEPTH`：分别决定逻辑上限、最小 class 和每 QP lease。 #### 验证清单 diff --git a/docs/DEPLOY.md b/docs/DEPLOY.md index 4b004364..b53d5507 100644 --- a/docs/DEPLOY.md +++ b/docs/DEPLOY.md @@ -239,13 +239,10 @@ LimitMEMLOCK=infinity # RDMA 需要锁页内存 WantedBy=multi-user.target ``` -> **`--max-msg` 默认 32 MiB（`32ull << 20`），是 client 声明值上限；client 默认 -> `DFKV_RDMA_MAX_BLOCK_BYTES` 为 4 MiB，实际 slot 取 `min(client 声明, server --max-msg)`，默认无需显式设置。** -> 它是 RDMA receive-segment 单 slot 上限；client 在 DCP2 协商时声明自己的 -> `DFKV_RDMA_MAX_BLOCK_BYTES`（默认 4 MiB）。若 server `--max-msg` < client 声明值，server -> 拒绝连接（日志 `client declared max block ... above this server's cap ...`），所有 PUT 失败 -> 但 exist 不受影响（不走 max_block 协商）。**不要降低 `--max-msg` 除非同时降低所有 client 的 -> `DFKV_RDMA_MAX_BLOCK_BYTES`。** +> **`--max-msg` 默认 32 MiB，是 server 接受的逻辑 block 上限；client +> `DFKV_RDMA_MAX_BLOCK_BYTES` 不得超过它。** client 每次 data Acquire 按当前 +> operation 最大对象选择 power-of-two connection class；逻辑上限不再决定所有 +> QP 的物理 slot。server 小于 client 逻辑声明仍会拒绝连接，避免超限对象被误收。 > > Completion timeout：`DFKV_RDMA_OP_TIMEOUT_MS`（默认 5000）约束单操作； > `DFKV_RDMA_BATCH_OP_TIMEOUT_MS`（unset/0 = 跟随前者）覆盖 **所有** multi-item @@ -262,9 +259,9 @@ WantedBy=multi-user.target > - `--ram-flush-threads ` / `DFKV_RAM_FLUSH_THREADS`：请求值会提高到至少 shard 数；实际值由 `dfkv_ram_flush_threads` 报告（默认请求=4×盘数，上限 16）。 > - 微调项（env only）：`DFKV_SLAB_TABLE_SYNC_MS` 控制 table sync 节奏（默认 100 ms，0=关；dirty epoch 重启仍无条件冷重置）。 > - 微调项（env only）：`DFKV_SLAB_EVICT_HIGH_PCT` / `DFKV_SLAB_EVICT_LOW_PCT` -> （默认 92 / 88，占容量百分比；high=0 关闭）：水位线主动驱逐在 demand 之前 -> 保留 headroom，是满环时 cross-class extent 互抢（自噬）的第一道防线；对应 -> 计数器 `dfkv_slab_watermark_evictions_total` / `dfkv_slab_cold_steals_total`。 +> （默认 92 / 88；high=0 关）使用 persistent hysteresis：越过 high 后持续后台 +> drain 到 low。每 tick 由 `DFKV_SLAB_EVICT_MAX_EXTENTS_PER_TICK`（默认 1） +> 限流；整 extent 只做一次连续 slots.tbl clear，避免高水位小写风暴。 > - `--slab-reclaim-ms ` / `--ram-reclaim-ms `（默认 50 / 10，`0`=关）：后台预回收和类再平衡。allocator 按 useful bytes + decayed read heat 选择 donor，跳过 pinned extent；通常保留默认。 > - `--slab-granularity `（默认 1 MiB）：最小 slot 量子。现有 `slots.tbl` 的 format/geometry 不匹配会 fail closed；修改必须换空目录，服务不会原地冷重建或忽略旧数据。 > - `--put-inflight-limit `（默认 0=关）：并发盘写超过 n 的 PUT 以 kCacheFull 快速拒绝（客户端视为普通 put 失败、不进 cooldown）= 用受控 miss 换掉过载排队尾延迟。RDMA 与 TCP 两条数据路径同受此门约束；RAM 热层的异步 flusher 落盘**不受**此门限制（否则背压会放大为 flush 丢弃）。 @@ -365,14 +362,11 @@ journalctl -u dfkv -n 10 --no-pager > `kNoCompatibleRail` 均不增加 local rail error。cooldown 到期仍只准入一个真实 > recovery probe。 > -> **v2 segment 预算**：`slot=align4K(4096 + max_raw_payload)`； -> pull-read 连接按 receive + source 两个 arena 计量： -> `segment >= 2 × Σ((live + client-pool-idle) data/control QP × depth × slot)`。 -> lease 保留到 QP 销毁或 idle reclaim，不能只数在飞请求。4 MiB/depth=4 -> 每条 pull data QP 占 33,587,200 B；16 GiB 约容纳 511 条，64 GiB 约 -> 2046 条（未扣 control lease）。上线先看 `dfkv_rdma_recv_segment_used_bytes`、 -> `free_bytes`、`largest_free_range_bytes`、`allocation_failures_total` 和 -> `dfkv_rdma_v2_ready`；free 或最大连续 range 接近一条新 lease 时会拒绝连接。 +> **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 或注册失败才拒绝连接。 ### 3a. 每节点 tenant quota @@ -469,7 +463,9 @@ flag 为 env facade）；未列 flag 的全部 env 均从源码排查就不误 | env | 默认 | 说明 | |---|---|---| -| `DFKV_RDMA_RECV_SEGMENT_SIZE` | `16 GiB`（xb01 `64 GiB`） | server：v2 共享 segment 总大小；pull client 每 data connection 约占 `2 × depth × align4K(4096+max_raw_payload)`，必须为基础连接和 churn 留余量 | +| `DFKV_RDMA_RECV_SEGMENT_SIZE` | `16 GiB` | server receive-pool hard budget；不再启动期全量分配 | +| `DFKV_RDMA_RECV_CHUNK_BYTES` | `256 MiB` | server 启动与增量提交粒度 | +| `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：控制面帧读写超时 | | `DFKV_RDMA_BATCH_OP_TIMEOUT_MS` | 0=跟随 RDMA_OP | client：multi-item Cache/Range/Exist、SG 窗口总期限 | @@ -499,7 +495,8 @@ flag 为 env facade）；未列 flag 的全部 env 均从源码排查就不误 | env | 默认 | 说明 | |---|---|---| -| `DFKV_SLAB_EVICT_HIGH_PCT` / `DFKV_SLAB_EVICT_LOW_PCT` | `92` / `88`, high=0 关 | 水位线主动驱逐阈值；`dfkv_slab_watermark_evictions_total`/`dfkv_slab_cold_steals_total` 对应 | +| `DFKV_SLAB_EVICT_HIGH_PCT` / `DFKV_SLAB_EVICT_LOW_PCT` | `92` / `88`, high=0 关 | high crossing latch，后台持续 drain 到 low | +| `DFKV_SLAB_EVICT_MAX_EXTENTS_PER_TICK` | `1` | 每盘每 tick 最多整 extent 淘汰数；平滑前台尾延迟 | | `DFKV_SLAB_COLD_STEAL_WINDOW` | — | 小受害者选择窗口（跨 size class 强占） | | `DFKV_SLAB_URING_WRITE` | — | slab io_uring 异步 PUT（需要 `-DDFKV_WITH_URING`） | | `DFKV_SLAB_RECLAIM_MS` | `50` | 背景预回收 tick（见 §3） | diff --git a/docs/METRICS.md b/docs/METRICS.md index d588413d..d530fec0 100644 --- a/docs/METRICS.md +++ b/docs/METRICS.md @@ -158,14 +158,16 @@ RDMA-listener scrape inventory。 | `dfkv_rdma_active_conns` | gauge | 当前服务中的 RDMA 连接 | | `dfkv_rdma_v2_conns_opened_total` | counter | server 累计打开的 v2 连接 | | `dfkv_rdma_v2_put_writes_total` / `dfkv_rdma_v2_get_writes_total` | counter | server 实际收到的 `WRITE_WITH_IMM` PUT / 实际发出的 RDMA WRITE GET payload | -| `dfkv_rdma_recv_segment_bytes` / `used_bytes` / `free_bytes` | gauge | process-wide segment 总量、当前 lease 占用、未租字节 | -| `dfkv_rdma_recv_segment_largest_free_range_bytes` | gauge | 最大连续 free range；明显小于 free 总量表示碎片导致大 lease 失败 | -| `dfkv_rdma_recv_segment_allocation_failures_total` | counter | allocator 找不到满足大小/对齐的连续 range 次数（含 admission retry） | +| `dfkv_rdma_recv_segment_bytes` / `max_bytes` / `chunks` | gauge | receive-pool 当前提交量 / hard budget / 已提交 chunk 数 | +| `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_allocation_failures_total` | counter | grow 后仍无法满足的最终 allocation | | `dfkv_rdma_pull_connections` / `dfkv_rdma_legacy_connections` | gauge | 当前 pull-read / legacy responder-write connection 数 | -| `dfkv_rdma_connection_bytes{class=\"data|control\"}` | gauge | data/control connection 当前持有的 recv+pull lease 字节 | -| `dfkv_rdma_recv_segment_registered_rails` | gauge | 成功把共享 segment 注册到 shared PD 的 rail 数 | -| `dfkv_rdma_v2_ready` | gauge | shared receive segment 是否已建立 | -| `dfkv_rdma_segment_evictions_total` / `dfkv_rdma_idle_reclaims_total` | counter | segment 压力淘汰 / idle timeout 回收连接 | +| `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 是否就绪 | +| `dfkv_rdma_segment_evictions_total` / `dfkv_rdma_idle_reclaims_total` | counter | hard budget 压力淘汰 / idle timeout 回收连接 | | `dfkv_uring_reads_total` / `dfkv_uring_init_fallbacks_total` | counter | io_uring 路径实际读 / 初始化失败回退同步连接数 | | `dfkv_rdma_rail_active_conns{dev}` | gauge | 每个本地 HCA 上当前连接数 | | `dfkv_rdma_rail_completions_total{dev}` / `dfkv_rdma_rail_completion_errors_total{dev}` | counter | 每 rail 请求完成 / 错误完成 | @@ -176,10 +178,8 @@ RDMA-listener scrape inventory。 > `configured == initialized == dfkv_rdma_recv_segment_registered_rails`。 > 启动 `ACTIVE` 只是快照；运行期以 > `dfkv_server_rdma_rails_active` 和逐设备 healthy 为准。active≥1 时 -> `dfkv_server_ring_eligible=1`；`active < initialized` 是 `PARTIAL`，不是整节点 -> 故障。active=0 必须立即退环，恢复到非零在 sample streak 后重新 eligible。 -> 随后确认 client/server QP 与 PUT/GET write counter 增长、 -> `dfkv_rdma_v2_ready=1` 且 receive-segment free bytes 有余量。 +> `dfkv_server_ring_eligible=1`。同时要求 `dfkv_rdma_v2_ready=1`、 +> `committed <= max`、growth/allocation failures 不增长。 读侧 convoy 合并与直读晋升（`DFKV_READ_COALESCE=1` 时才有增量；恒零 = 开关没生效）： | `dfkv_read_coalesce_leaders_total` | counter | 经 coalescer 登记并完成的读（同步 leader + io_uring flight 各计一次；>0 = 合并路径确实在环内） | @@ -194,7 +194,11 @@ slab 引擎内部（**resolved engine=slab 时输出**——按运行时实际 | `dfkv_slab_table_sync_total` | counter | slots.tbl fdatasync 周期数（`DFKV_SLAB_TABLE_SYNC_MS`，默认 100ms；限定崩溃复活毒化窗口） | | `dfkv_slab_extent_steals_total` / `dfkv_slab_extent_returns_total` | counter | 跨 class extent 抢占（伴随驱逐，容量失衡信号）/ 全空 extent 主动回池（无损再平衡） | | `dfkv_slab_cold_steals_total` | counter | 命中"全局最冷 extent"护栏的跨 class 抢占（donor 全部内容都冷于 `DFKV_SLAB_COLD_STEAL_WINDOW` 窗口才走此路，满环自噬护栏；0=窗口关闭） | -| `dfkv_slab_watermark_evictions_total` | counter | 水位线主动驱逐（`DFKV_SLAB_EVICT_HIGH_PCT`/`_LOW_PCT`，默认 92/88）——赶在 demand 前腾 headroom，满环自噬的第一道防线 | +| `dfkv_slab_watermark_evictions_total` | counter | high crossing 后持续到 low 的主动整 extent 驱逐数 | +| `dfkv_slab_watermark_extent_clears_total` | counter | watermark 使用单次连续 slots.tbl clear 的 extent 数；应与 watermark eviction 同步 | +| `dfkv_slab_watermark_ticks_total` / `dfkv_slab_watermark_active` | counter / gauge | bounded eviction tick 数 / hysteresis 是否仍在 drain | +| `dfkv_slab_watermark_last_tick_us` / `max_tick_us` | gauge | 最近 / 进程最大 watermark 锁区间微秒数 | +| `dfkv_slab_watermark_max_extents_per_tick` | gauge | 每盘每 tick 配置的 extent 上限 | | `dfkv_slab_table_rebuilt_objects` / `dfkv_slab_rebuild_scanned_bytes` / `dfkv_slab_rebuild_scan_chunks` / `dfkv_slab_rebuild_sparse_ranges` / `dfkv_slab_rebuild_mmap_scans` | gauge | 上次启动冷升从 slots.tbl 恢复的对象数 / 扫描字节 / chunk / sparse range / mmap 的表数（冷升目录规模诊断） | | `dfkv_slab_rebuild_corrupt_records_total` / `dfkv_slab_rebuild_rejected_records_total` / `dfkv_slab_rebuild_sequential_fallbacks_total` | counter | 启动 rebuild 丢弃的损坏记录 / 因几何不安全被清除的合法记录 / sparse seek 退化为顺序扫描的次数（任一非零都值得看日志） | | `dfkv_slab_deferred_removes_total` | counter | 被在飞 I/O 延迟执行的 Remove | @@ -332,7 +336,7 @@ C 客户端快照还含传输级指标（RDMA 构建）： | `dfkv_rdma_client_adhoc_user_mr_total` / `dfkv_rdma_client_transient_user_mr_active` | counter / gauge | pool 外实际注册累计 / 当前仍存活的一次性 MR；公开调用返回后 active 必须回到调用前基线 | | `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` | gauge | 实际请求高水位 / DCP2 声明上限 | +| `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_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/src/cache/dfkv_server_main.cc b/src/cache/dfkv_server_main.cc index f45068a4..f053b6de 100644 --- a/src/cache/dfkv_server_main.cc +++ b/src/cache/dfkv_server_main.cc @@ -76,7 +76,8 @@ int main(int argc, char** argv) { " --slab-table-sync-ms slab table sync cadence ms (env DFKV_SLAB_TABLE_SYNC_MS)\n" " --slab-reclaim-ms slab background free-slot reclaimer cadence ms, 0 = off (env DFKV_SLAB_RECLAIM_MS)\n" " --ram-reclaim-ms RAM tier background reclaimer cadence ms, 0 = off (env DFKV_RAM_RECLAIM_MS)\n" - " --rdma-recv-segment-size RDMA v2 shared receive segment bytes (default 16 GiB; env DFKV_RDMA_RECV_SEGMENT_SIZE)\n" + " --rdma-recv-segment-size RDMA receive-pool hard budget (default 16 GiB; env DFKV_RDMA_RECV_SEGMENT_SIZE)\n" + " --rdma-recv-chunk-bytes lazy receive-pool chunk bytes (default 256 MiB; env DFKV_RDMA_RECV_CHUNK_BYTES)\n" " --disk-hash-weight per-disk vnode multiplier (default 10; env DFKV_DISK_HASH_WEIGHT)\n" " --read-coalesce <0|1> read-side convoy merge + RAM promotion (default off; env DFKV_READ_COALESCE)\n" " --log log level: INFO|DEBUG|WARN|ERROR (env DFKV_LOG)\n" @@ -102,7 +103,8 @@ int main(int argc, char** argv) { "--ram-tier-numa", "--ram-tier-shards", "--slab-table-sync-ms", "--slab-reclaim-ms", "--ram-reclaim-ms", "--log", "--mds-registration-timeout-ms", "--max-msg", - "--rdma-recv-segment-size", "--disk-hash-weight", "--read-coalesce"}); + "--rdma-recv-segment-size", "--rdma-recv-chunk-bytes", + "--disk-hash-weight", "--read-coalesce"}); std::string dir = args.Get("--dir", "/tmp/dfkv_node"); std::string rdma_dev = args.Get("--rdma-dev", ""); std::string mds = args.Get("--mds", ""); @@ -190,10 +192,13 @@ int main(int argc, char** argv) { std::string ram_tier_shards = args.Get("--ram-tier-shards", ""); if (!ram_tier_shards.empty()) ::setenv("DFKV_RAM_TIER_SHARDS", ram_tier_shards.c_str(), 1); - // RDMA v2 receive segment size (shared receive buffer pool; default 2 GiB). + // RDMA v2 receive-pool budget and lazy commit chunk. std::string recv_seg = args.Get("--rdma-recv-segment-size", ""); if (!recv_seg.empty()) ::setenv("DFKV_RDMA_RECV_SEGMENT_SIZE", recv_seg.c_str(), 1); + std::string recv_chunk = args.Get("--rdma-recv-chunk-bytes", ""); + if (!recv_chunk.empty()) + ::setenv("DFKV_RDMA_RECV_CHUNK_BYTES", recv_chunk.c_str(), 1); // Disk hash weight — per-disk vnode multiplier (default 10; was 1). std::string disk_hash_weight = args.Get("--disk-hash-weight", ""); if (!disk_hash_weight.empty()) diff --git a/src/cache/disk_cache_group.cc b/src/cache/disk_cache_group.cc index ef8aabcd..52473b67 100644 --- a/src/cache/disk_cache_group.cc +++ b/src/cache/disk_cache_group.cc @@ -545,6 +545,17 @@ DiskSlabStore::Stats DiskCacheGroup::SlabStats() const { sum.steals += st.steals; sum.cold_steals += st.cold_steals; sum.watermark_evictions += st.watermark_evictions; + sum.watermark_extent_clears += st.watermark_extent_clears; + sum.watermark_ticks += st.watermark_ticks; + sum.watermark_active = + std::max(sum.watermark_active, st.watermark_active); + sum.watermark_last_tick_us = + std::max(sum.watermark_last_tick_us, st.watermark_last_tick_us); + sum.watermark_max_tick_us = + std::max(sum.watermark_max_tick_us, st.watermark_max_tick_us); + sum.watermark_max_extents_per_tick = + std::max(sum.watermark_max_extents_per_tick, + st.watermark_max_extents_per_tick); sum.extent_returns += st.extent_returns; sum.deferred_removes += st.deferred_removes; sum.inflight += st.inflight; diff --git a/src/cache/disk_slab_store.cc b/src/cache/disk_slab_store.cc index 8f4b7c57..fbe5a05d 100644 --- a/src/cache/disk_slab_store.cc +++ b/src/cache/disk_slab_store.cc @@ -157,6 +157,21 @@ DiskSlabStore::DiskSlabStore(Options opt, bool* ok) : opt_(std::move(opt)) { bind_wipes_.fetch_add(1, std::memory_order_relaxed); return true; }; + 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(), + TableOffset(extent, 0))) + return FailMetadata(); + record_writes_.fetch_add(std::max(residents, 1), + std::memory_order_relaxed); + eviction_record_clears_.fetch_add(residents, + std::memory_order_relaxed); + watermark_extent_clears_.fetch_add(1, + std::memory_order_relaxed); + return true; + }; ao.on_slot_evict = [this](const SlabAllocator::SlotRef& ref) { const uint8_t zero[kRecBytes] = {0}; if (!PwriteAll(table_fd_, zero, sizeof(zero), @@ -184,6 +199,16 @@ DiskSlabStore::DiskSlabStore(Options opt, bool* ok) : opt_(std::move(opt)) { if (low_pct >= high_pct && high_pct > 0) low_pct = high_pct - 1; evict_high_bytes_ = high_pct ? opt_.capacity_bytes / 100 * high_pct : 0; evict_low_bytes_ = opt_.capacity_bytes / 100 * low_pct; + if (const char* value = + std::getenv("DFKV_SLAB_EVICT_MAX_EXTENTS_PER_TICK")) { + char* end = nullptr; + const unsigned long parsed = std::strtoul(value, &end, 10); + if (end != value && *end == '\0' && parsed > 0 && parsed <= 1024) + evict_max_extents_per_tick_ = static_cast(parsed); + } + config_dump::RecordResolved( + "DFKV_SLAB_EVICT_MAX_EXTENTS_PER_TICK", + std::to_string(evict_max_extents_per_tick_)); ok_ = OpenOrInit(); if (ok_ && unclean_start_) ok_ = ResetUncleanEpoch(); @@ -1531,18 +1556,38 @@ void DiskSlabStore::ReclaimTick( const std::function& after_watermark_evict) { if (!Healthy()) return; // Watermark eviction is a composite mutation: allocator occupancy and the - // payload-length commit map must change under the same store lock. Otherwise - // a concurrent rewrite can commit a reused key between those two mutations, - // only for this pass to erase the new commit state. - if (evict_high_bytes_ != 0 && alloc_->UsedBytes() > evict_high_bytes_) { - std::lock_guard lk(mu_); - if (alloc_->UsedBytes() > evict_high_bytes_) { + // payload-length commit map change under the same store lock. Crossing high + // latches the background drain until low is reached, but each tick removes + // only a small configured number of extents so GET/PUT never sits behind a + // multi-GiB eviction burst. + if (evict_high_bytes_ != 0) { + const uint64_t used = alloc_->UsedBytes(); + if (used > evict_high_bytes_) + watermark_active_.store(true, std::memory_order_relaxed); + if (watermark_active_.load(std::memory_order_relaxed)) { + const auto started = std::chrono::steady_clock::now(); + std::lock_guard lk(mu_); std::vector evicted; - alloc_->EvictColdToTarget(evict_low_bytes_, /*max_extents=*/64, - &evicted); + alloc_->EvictColdToTarget(evict_low_bytes_, + evict_max_extents_per_tick_, &evicted); if (!evicted.empty()) { if (after_watermark_evict) after_watermark_evict(); - for (const BlockKey& key : evicted) ErasePayloadLocked(key.Filename()); + for (const BlockKey& key : evicted) + ErasePayloadLocked(key.Filename()); + } + if (alloc_->UsedBytes() <= evict_low_bytes_) + watermark_active_.store(false, std::memory_order_relaxed); + const uint64_t elapsed = static_cast( + std::chrono::duration_cast( + std::chrono::steady_clock::now() - started) + .count()); + watermark_ticks_.fetch_add(1, std::memory_order_relaxed); + watermark_last_tick_us_.store(elapsed, std::memory_order_relaxed); + uint64_t observed = + watermark_max_tick_us_.load(std::memory_order_relaxed); + while (elapsed > observed && + !watermark_max_tick_us_.compare_exchange_weak( + observed, elapsed, std::memory_order_relaxed)) { } } } @@ -1652,6 +1697,18 @@ DiskSlabStore::Stats DiskSlabStore::GetStats() const { st.steals = alloc_->Steals(); st.cold_steals = alloc_->ColdSteals(); st.watermark_evictions = alloc_->WatermarkEvictions(); + st.watermark_extent_clears = + watermark_extent_clears_.load(std::memory_order_relaxed); + st.watermark_ticks = + watermark_ticks_.load(std::memory_order_relaxed); + st.watermark_active = + watermark_active_.load(std::memory_order_relaxed) ? 1 : 0; + st.watermark_last_tick_us = + watermark_last_tick_us_.load(std::memory_order_relaxed); + st.watermark_max_tick_us = + watermark_max_tick_us_.load(std::memory_order_relaxed); + st.watermark_max_extents_per_tick = + evict_max_extents_per_tick_; st.extent_returns = alloc_->ExtentReturns(); st.allocated_bytes = alloc_->UsedBytes(); st.allocator_objects = alloc_->Count(); diff --git a/src/cache/disk_slab_store.h b/src/cache/disk_slab_store.h index 25ebbf1e..28f2b922 100644 --- a/src/cache/disk_slab_store.h +++ b/src/cache/disk_slab_store.h @@ -81,6 +81,12 @@ class DiskSlabStore : public StoreEngine { uint64_t steals = 0; uint64_t cold_steals = 0; uint64_t watermark_evictions = 0; + uint64_t watermark_extent_clears = 0; + uint64_t watermark_ticks = 0; + uint64_t watermark_active = 0; + uint64_t watermark_last_tick_us = 0; + uint64_t watermark_max_tick_us = 0; + uint64_t watermark_max_extents_per_tick = 0; uint64_t extent_returns = 0; uint64_t deferred_removes = 0; uint64_t inflight = 0; @@ -280,6 +286,7 @@ class DiskSlabStore : public StoreEngine { std::function uring_reap_hook_for_test_; std::atomic unclean_resets_{0}; std::atomic eviction_record_clears_{0}; + std::atomic watermark_extent_clears_{0}; // slots.tbl sync thread (see Options::table_sync_ms): fdatasync only when uint64_t rebuild_corrupt_records_ = 0; uint64_t rebuild_rejected_records_ = 0; @@ -313,8 +320,13 @@ class DiskSlabStore : public StoreEngine { // The batched-write submission ring is thread_local (see disk_slab_store.cc): // one per flush worker, no shared lock across the blocking CQE wait. std::vector reclaim_last_puts_; // reclaim-thread-local puts snapshot - uint64_t evict_high_bytes_ = 0; // used > this -> proactive cold eviction (0=off) - uint64_t evict_low_bytes_ = 0; // ... down to this + uint64_t evict_high_bytes_ = 0; // high crossing activates eviction + uint64_t evict_low_bytes_ = 0; // active eviction drains to this target + size_t evict_max_extents_per_tick_ = 1; + std::atomic watermark_active_{false}; + std::atomic watermark_ticks_{0}; + std::atomic watermark_last_tick_us_{0}; + std::atomic watermark_max_tick_us_{0}; std::thread reclaim_thread_; std::condition_variable reclaim_cv_; std::mutex reclaim_mu_; diff --git a/src/cache/kv_node_server.cc b/src/cache/kv_node_server.cc index bd75362a..22eea6fb 100644 --- a/src/cache/kv_node_server.cc +++ b/src/cache/kv_node_server.cc @@ -476,6 +476,23 @@ std::string KvNodeServer::MetricsText() const { metric("dfkv_slab_watermark_evictions_total", "counter", "Proactive watermark extent evictions (headroom kept ahead of demand)", ss.watermark_evictions); + metric("dfkv_slab_watermark_extent_clears_total", "counter", + "Whole-extent metadata clears used by watermark eviction", + ss.watermark_extent_clears); + metric("dfkv_slab_watermark_ticks_total", "counter", + "Bounded watermark eviction ticks", ss.watermark_ticks); + metric("dfkv_slab_watermark_active", "gauge", + "Whether high-water eviction remains latched until low watermark", + ss.watermark_active); + metric("dfkv_slab_watermark_last_tick_us", "gauge", + "Duration of the latest watermark tick in microseconds", + ss.watermark_last_tick_us); + metric("dfkv_slab_watermark_max_tick_us", "gauge", + "Maximum watermark tick duration in microseconds", + ss.watermark_max_tick_us); + metric("dfkv_slab_watermark_max_extents_per_tick", "gauge", + "Configured per-disk extent eviction cap per tick", + ss.watermark_max_extents_per_tick); metric("dfkv_slab_extent_returns_total", "counter", "Fully-free extents returned to the shared pool", ss.extent_returns); metric("dfkv_slab_deferred_removes_total", "counter", diff --git a/src/cache/rdma_server.cc b/src/cache/rdma_server.cc index fe36828e..8230a7fa 100644 --- a/src/cache/rdma_server.cc +++ b/src/cache/rdma_server.cc @@ -91,6 +91,18 @@ size_t RecvSegmentBytes() { return bytes; } +size_t RecvChunkBytes(size_t max_bytes) { + constexpr size_t kDefault = 256ull << 20; + const size_t fallback = std::min(kDefault, max_bytes); + const size_t parsed = rdma::ResolveRecvSegmentBytes( + std::getenv("DFKV_RDMA_RECV_CHUNK_BYTES"), fallback, + rdma::kV2DataOffset); + const size_t bytes = parsed == 0 ? 0 : std::min(parsed, max_bytes); + config_dump::RecordResolved("DFKV_RDMA_RECV_CHUNK_BYTES", + std::to_string(bytes)); + return bytes; +} + const char* DiscoveryStatusName(rdma::RdmaDiscoveryStatus status) { switch (status) { case rdma::RdmaDiscoveryStatus::kOk: @@ -271,32 +283,33 @@ Status RdmaServer::Start(int port) { dev_name_ = anchor_devs_.front(); config_dump::RecordResolved("DFKV_RDMA_DEV", resolved_devices); - // Allocate the mandatory process-wide receive segment before opening anchors. - recv_segment_bytes_ = RecvSegmentBytes(); + // Commit one small receive chunk at startup; additional chunks are allocated + // only when live connection leases need them. The legacy segment setting is + // 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_); const size_t min_slot_bytes = rdma::V2SlotSize(max_msg_); - if (!rdma::V2RecvSegmentFitsOneSlot(recv_segment_bytes_, max_msg_)) { + if (min_slot_bytes == 0 || + recv_segment_max_bytes_ < 2 * min_slot_bytes || + recv_segment_chunk_bytes_ == 0 || + !recv_segments_.Init(recv_segment_chunk_bytes_, + recv_segment_max_bytes_, + rdma::kV2DataOffset)) { LogTopologySummary(requested_devices.size(), 0, discovery.devices); DFKV_LOG_ERROR( - "rdma: configured v2 receive segment (" + - std::to_string(recv_segment_bytes_) + - " bytes) cannot fit one complete slot (" + - std::to_string(min_slot_bytes) + " bytes) at server max block " + - std::to_string(max_msg_) + - " bytes; raise DFKV_RDMA_RECV_SEGMENT_SIZE or lower --max-msg"); - recv_segment_bytes_ = 0; - ::close(listen_fd_); - listen_fd_ = -1; - return Status::kIOError; - } - if (!recv_segment_.Init(recv_segment_bytes_, rdma::kV2DataOffset)) { - LogTopologySummary(requested_devices.size(), 0, discovery.devices); - DFKV_LOG_ERROR("rdma: unable to allocate required v2 receive segment (" + - std::to_string(recv_segment_bytes_) + " bytes)"); - recv_segment_bytes_ = 0; + "rdma: invalid receive-pool geometry max=" + + std::to_string(recv_segment_max_bytes_) + + " chunk=" + std::to_string(recv_segment_chunk_bytes_) + + " minimum-two-slot-bytes=" + std::to_string(2 * min_slot_bytes) + + "; raise DFKV_RDMA_RECV_SEGMENT_SIZE, adjust " + "DFKV_RDMA_RECV_CHUNK_BYTES, or lower --max-msg"); + recv_segment_max_bytes_ = 0; + recv_segment_chunk_bytes_ = 0; ::close(listen_fd_); listen_fd_ = -1; return Status::kIOError; } + rdma::RecvSegment* initial_segment = recv_segments_.initial_segment(); std::vector> initialized_anchors; initialized_anchors.reserve(anchor_devs_.size()); @@ -304,7 +317,8 @@ Status RdmaServer::Start(int port) { std::unique_ptr anchor; if (initialize_anchor_for_test_) { anchor = initialize_anchor_for_test_( - device, user_regions_, recv_segment_.data(), recv_segment_.size()); + device, user_regions_, initial_segment->data(), + initial_segment->size()); } else { anchor = std::make_unique(); if (!anchor->Open(device.c_str(), rdma::kV2ControlCap, 1)) { @@ -314,10 +328,10 @@ Status RdmaServer::Start(int port) { DFKV_LOG_ERROR("rdma: failed to register explicit user region on " + device); anchor.reset(); - } else if (!anchor->RegisterRemoteRegion(recv_segment_.data(), - recv_segment_.size())) { + } else if (!anchor->RegisterRemoteRegion(initial_segment->data(), + initial_segment->size())) { DFKV_LOG_ERROR( - "rdma: failed to register required v2 receive segment on " + + "rdma: failed to register initial v2 receive chunk on " + device); anchor.reset(); } @@ -605,8 +619,8 @@ void RdmaServer::Serve(int boot_fd) { ::close(boot_fd); return; } - rdma::RecvSegment::Lease recv_lease = - recv_segment_.Allocate(K * slot_size, rdma::kV2DataOffset); + rdma::RecvSegmentPool::Lease recv_lease = + recv_segments_.Allocate(K * slot_size, rdma::kV2DataOffset); 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 @@ -641,29 +655,33 @@ void RdmaServer::Serve(int boot_fd) { for (int i = 0; i < 100 && !recv_lease; ++i) { std::this_thread::sleep_for(std::chrono::milliseconds(10)); recv_lease = - recv_segment_.Allocate(K * slot_size, rdma::kV2DataOffset); + recv_segments_.Allocate(K * slot_size, rdma::kV2DataOffset); } } if (!recv_lease) { + const auto stats = recv_segments_.stats(); DFKV_LOG_ERROR( - "rdma v2: shared receive segment exhausted; refusing connection " - "(need=" + + "rdma v2: receive pool exhausted; refusing connection (need=" + std::to_string(K * slot_size) + - " free=" + std::to_string(recv_segment_.free_bytes()) + ")"); + " free=" + std::to_string(stats.free_bytes) + + " committed=" + std::to_string(stats.committed_bytes) + + " max=" + std::to_string(stats.max_bytes) + ")"); ::close(boot_fd); return; } } - rdma::RecvSegment::Lease pull_lease; + rdma::RecvSegmentPool::Lease pull_lease; if (pull_read_requested) { pull_lease = - recv_segment_.Allocate(K * slot_size, rdma::kV2DataOffset); + recv_segments_.Allocate(K * slot_size, rdma::kV2DataOffset); if (!pull_lease) { + const auto stats = recv_segments_.stats(); DFKV_LOG_ERROR( "rdma v2: pull-read arena unavailable; refusing negotiated " - "connection (need=" + - std::to_string(K * slot_size) + - " free=" + std::to_string(recv_segment_.free_bytes()) + ")"); + "connection (need=" + std::to_string(K * slot_size) + + " free=" + std::to_string(stats.free_bytes) + + " committed=" + std::to_string(stats.committed_bytes) + + " max=" + std::to_string(stats.max_bytes) + ")"); ::close(boot_fd); return; } @@ -751,8 +769,8 @@ void RdmaServer::Serve(int boot_fd) { ::close(boot_fd); return; } - ibv_mr* recv_segment_mr = - ep.RegisterRemoteRegion(recv_segment_.data(), recv_segment_.size()); + ibv_mr* recv_segment_mr = ep.RegisterRemoteRegion( + recv_lease.segment()->data(), recv_lease.segment()->size()); if (!recv_segment_mr) { DFKV_LOG_ERROR("rdma v2: receive-segment MR unavailable on device " + (dev.empty() ? std::string("(auto)") : dev)); @@ -2003,22 +2021,30 @@ std::string RdmaServer::MetricsText() const { m(s, "dfkv_rdma_v2_get_continuation_slot_changes_total", "counter", "Multi-window GET continuations received on a different WQE slot", V2GetContinuationSlotChanges()); - const rdma::RecvSegment::Stats segment = recv_segment_.stats(); + const rdma::RecvSegmentPool::Stats segment = recv_segments_.stats(); m(s, "dfkv_rdma_recv_segment_bytes", "gauge", - "Process-wide registered receive-segment bytes", segment.total_bytes); + "Receive-pool bytes currently committed", segment.committed_bytes); + m(s, "dfkv_rdma_recv_segment_max_bytes", "gauge", + "Hard receive-pool commit budget", segment.max_bytes); + m(s, "dfkv_rdma_recv_segment_chunks", "gauge", + "Receive-pool chunks currently committed", segment.chunks); m(s, "dfkv_rdma_recv_segment_used_bytes", "gauge", - "Bytes leased from the process-wide receive segment", segment.used_bytes); + "Bytes leased from committed receive chunks", segment.used_bytes); m(s, "dfkv_rdma_recv_segment_free_bytes", "gauge", - "Unleased bytes remaining in the process-wide receive segment", - segment.free_bytes); + "Unleased bytes in committed receive chunks", segment.free_bytes); m(s, "dfkv_rdma_recv_segment_largest_free_range_bytes", "gauge", - "Largest contiguous unleased receive-segment range", + "Largest contiguous unleased range in any receive chunk", 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_growth_failures_total", "counter", + "Receive-pool growth attempts rejected by budget or allocation", + segment.growth_failures); m(s, "dfkv_rdma_recv_segment_allocation_failures_total", "counter", - "Receive-segment allocation attempts without a suitable free range", + "Receive-pool allocations that remained unsatisfied after growth", segment.allocation_failures); m(s, "dfkv_rdma_recv_segment_registered_rails", "gauge", - "RDMA rails with the process-wide receive segment registered", + "RDMA rails with the initial receive chunk registered", recv_segment_registered_rails_); m(s, "dfkv_rdma_pull_connections", "gauge", "Connections currently holding negotiated pull-read arenas", diff --git a/src/cache/rdma_server.h b/src/cache/rdma_server.h index 134eee37..564622de 100644 --- a/src/cache/rdma_server.h +++ b/src/cache/rdma_server.h @@ -217,23 +217,22 @@ class RdmaServer { std::unordered_set live_eps_; std::mutex writer_mu_; std::unordered_map> writers_; - // One process-wide pinned receive segment replaces per-connection payload - // buffers. Each v2 connection leases depth * aligned_slot_size bytes, and - // each rail's shared PD registers the segment once. This makes connection - // admission an explicit bounded resource instead of multiplying max payload - // buffers by every connection. - rdma::RecvSegment recv_segment_; + // Receive memory is committed in fixed-size chunks on demand. Connections + // lease only their negotiated slot geometry from any chunk; the old + // DFKV_RDMA_RECV_SEGMENT_SIZE remains the hard process budget. + rdma::RecvSegmentPool recv_segments_; std::atomic segment_evictions_{0}; - size_t recv_segment_bytes_ = 0; + size_t recv_segment_max_bytes_ = 0; + size_t recv_segment_chunk_bytes_ = 0; size_t recv_segment_registered_rails_ = 0; std::atomic pull_connections_{0}; std::atomic legacy_connections_{0}; std::atomic data_connection_bytes_{0}; std::atomic control_connection_bytes_{0}; // One anchor per resolved rail holds a lifetime shared device reference and - // registers the receive segment and caller pools on that rail's PD. Auto - // mode anchors only the first ACTIVE local rail. An explicit list is fixed - // at startup and may include inactive ports so they can recover in place. + // registers the initial receive chunk and caller pools on that rail's PD. + // Later chunks register lazily on the rail of the connection that leases + // them. Auto mode anchors only the first ACTIVE local rail. std::vector> anchors_; std::vector anchor_devs_; // fixed resolved rail names std::vector> rail_stats_; // indexed with anchor_devs_ diff --git a/src/cache/slab_allocator.cc b/src/cache/slab_allocator.cc index beaaa8f4..9f38ae07 100644 --- a/src/cache/slab_allocator.cc +++ b/src/cache/slab_allocator.cc @@ -902,10 +902,17 @@ size_t SlabAllocator::EvictColdToTarget(uint64_t target_bytes, } if (best < 0) break; ExtentMeta& extent = extents_[static_cast(best)]; + const bool extent_cleared = + opt_.on_extent_evict && + opt_.on_extent_evict(static_cast(best), + extent.total_slots, extent.resident_count); + if (opt_.on_extent_evict && !extent_cleared) return freed; while (extent.resident_head != kNoSlot) { const uint32_t index = extent.resident_head; SlotMeta& meta = slots_[index]; - if (opt_.on_slot_evict && !opt_.on_slot_evict(meta.ref)) return freed; + if (!extent_cleared && opt_.on_slot_evict && + !opt_.on_slot_evict(meta.ref)) + return freed; BlockKey victim = meta.key; FreeSlotLocked(index); if (evicted) evicted->push_back(std::move(victim)); diff --git a/src/cache/slab_allocator.h b/src/cache/slab_allocator.h index fa62fbaa..29fd2ff2 100644 --- a/src/cache/slab_allocator.h +++ b/src/cache/slab_allocator.h @@ -61,6 +61,11 @@ class SlabAllocator { // (slot_size - aligned_len) stays below this fraction. double max_waste = 0.25; std::function on_extent_bind; + // Whole-extent eviction can clear persistent metadata in one contiguous + // write. When set, watermark eviction uses this instead of per-slot calls. + std::function + on_extent_evict; std::function on_slot_evict; }; diff --git a/src/transport/rdma_recv_segment.cc b/src/transport/rdma_recv_segment.cc index b47d7923..738eec4d 100644 --- a/src/transport/rdma_recv_segment.cc +++ b/src/transport/rdma_recv_segment.cc @@ -157,4 +157,101 @@ void RecvSegment::Release(size_t offset, size_t bytes) { free_.emplace(begin, end - begin); } +bool RecvSegmentPool::Init(size_t chunk_bytes, size_t max_bytes, + size_t alignment) { + if (!IsPowerOfTwo(alignment) || alignment < sizeof(void*) || + chunk_bytes == 0 || max_bytes == 0) { + return false; + } + const size_t aligned_chunk = AlignUpChecked(chunk_bytes, alignment); + const size_t aligned_max = AlignUpChecked(max_bytes, alignment); + if (aligned_chunk == std::numeric_limits::max() || + aligned_max == std::numeric_limits::max() || + aligned_chunk > aligned_max) { + return false; + } + std::lock_guard lock(mu_); + if (!chunks_.empty()) { + return chunk_bytes_ == aligned_chunk && max_bytes_ == aligned_max && + alignment_ == alignment; + } + chunk_bytes_ = aligned_chunk; + max_bytes_ = aligned_max; + alignment_ = alignment; + auto initial = std::make_unique(); + if (!initial->Init(chunk_bytes_, alignment_)) { + chunk_bytes_ = 0; + max_bytes_ = 0; + alignment_ = 0; + return false; + } + chunks_.push_back(std::move(initial)); + return true; +} + +std::unique_ptr RecvSegmentPool::NewChunk( + size_t minimum_bytes) { + 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(); + 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; + return chunk; +} + +RecvSegmentPool::Lease RecvSegmentPool::Allocate(size_t bytes, + size_t alignment) { + 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 chunk = NewChunk(bytes); + 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(); + chunks_.push_back(std::move(chunk)); + growths_.fetch_add(1, std::memory_order_relaxed); + auto lease = owner->Allocate(bytes, alignment); + if (!lease) { + allocation_failures_.fetch_add(1, std::memory_order_relaxed); + return {}; + } + return Lease(owner, std::move(lease)); +} + +RecvSegment* RecvSegmentPool::initial_segment() const { + std::lock_guard lock(mu_); + return chunks_.empty() ? nullptr : chunks_.front().get(); +} + +RecvSegmentPool::Stats RecvSegmentPool::stats() const { + std::lock_guard lock(mu_); + Stats out; + out.max_bytes = max_bytes_; + out.chunks = chunks_.size(); + for (const auto& chunk : chunks_) { + const RecvSegment::Stats current = chunk->stats(); + out.committed_bytes += current.total_bytes; + out.used_bytes += current.used_bytes; + out.free_bytes += current.free_bytes; + out.largest_free_range = + std::max(out.largest_free_range, current.largest_free_range); + } + out.growths = growths_.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); + return out; +} + } // namespace dfkv::rdma diff --git a/src/transport/rdma_recv_segment.h b/src/transport/rdma_recv_segment.h index 557465c2..e874d017 100644 --- a/src/transport/rdma_recv_segment.h +++ b/src/transport/rdma_recv_segment.h @@ -6,7 +6,10 @@ #include #include #include +#include #include +#include +#include namespace dfkv::rdma { // Parse and align the process-wide receive-segment size without applying the @@ -74,6 +77,68 @@ class RecvSegment { std::atomic allocation_failures_{0}; }; +// Lazily commits fixed-size registered-memory candidates up to one hard +// process budget. A lease retains its owning chunk, so callers can register +// only that chunk on the selected rail instead of one monolithic arena. +class RecvSegmentPool { + public: + class Lease { + public: + Lease() = default; + Lease(const Lease&) = delete; + Lease& operator=(const Lease&) = delete; + Lease(Lease&&) noexcept = default; + Lease& operator=(Lease&&) noexcept = default; + + 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; + } + + private: + friend class RecvSegmentPool; + Lease(RecvSegment* segment, RecvSegment::Lease lease) + : lease_(std::move(lease)), segment_(segment) {} + + RecvSegment::Lease lease_; + RecvSegment* segment_ = nullptr; + }; + + struct Stats { + size_t max_bytes = 0; + size_t committed_bytes = 0; + size_t used_bytes = 0; + size_t free_bytes = 0; + size_t largest_free_range = 0; + size_t chunks = 0; + uint64_t growths = 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); + RecvSegment* initial_segment() const; + Stats stats() const; + + private: + std::unique_ptr NewChunk(size_t minimum_bytes); + + size_t chunk_bytes_ = 0; + size_t max_bytes_ = 0; + size_t alignment_ = 0; + mutable std::mutex mu_; + std::vector> chunks_; + std::atomic growths_{0}; + std::atomic allocation_failures_{0}; + std::atomic growth_failures_{0}; +}; + } // namespace dfkv::rdma #endif // DFKV_TRANSPORT_RDMA_RECV_SEGMENT_H_ diff --git a/src/transport/rdma_transport.cc b/src/transport/rdma_transport.cc index d15bc768..ed207ceb 100644 --- a/src/transport/rdma_transport.cc +++ b/src/transport/rdma_transport.cc @@ -392,6 +392,7 @@ struct RdmaTransport::Conn { bool credit_held = false; rdma::ResourceRequest budget_request; bool budget_held = false; + size_t declared_bytes = 0; Lane lane = Lane::kData; bool active_counted = false; bool live_counted = false; @@ -433,6 +434,9 @@ RdmaTransport::RdmaTransport(size_t max_msg, const std::string& dev_name) declared_(std::min( EnvBytes("DFKV_RDMA_MAX_BLOCK_BYTES", 4ull << 20), ResolveMaxPayload(max_msg))), + connection_min_block_bytes_(std::min( + EnvBytes("DFKV_RDMA_CONNECTION_MIN_BLOCK_BYTES", 256ull << 10), + static_cast(declared_))), depth_(4) { std::string list = dev_name; if (list.empty()) { @@ -524,6 +528,9 @@ RdmaTransport::RdmaTransport(size_t max_msg, const std::string& dev_name) "DFKV_RDMA_PRIMARY_DEV", preferred_rail_ ? devs_[*preferred_rail_] : std::string("")); config_dump::RecordResolved("DFKV_RDMA_DEPTH", std::to_string(depth_)); + config_dump::RecordResolved( + "DFKV_RDMA_CONNECTION_MIN_BLOCK_BYTES", + std::to_string(connection_min_block_bytes_)); config_dump::RecordResolved("DFKV_RDMA_NUMA", numa_aware_ ? "1" : "0"); const size_t endpoint_limit = static_cast( EnvBoundedInt("DFKV_RDMA_ENDPOINT_CACHE_MAX", 256, 65536)); @@ -1087,6 +1094,10 @@ bool RdmaTransport::ProbeV2(const std::string& node) const { RdmaTransport::AcquireResult RdmaTransport::Acquire( const std::string& node, Lane lane, const AcquireOptions& options) { + const size_t required_bound = + lane == Lane::kControl + ? static_cast(rdma::kV2ControlCap) + : ConnectionBound(options.required_data_bytes); AcquireResult result; const auto peer_snapshot = options.peer ? options.peer : peer_topologies_->Snapshot(node); @@ -1199,19 +1210,28 @@ RdmaTransport::AcquireResult RdmaTransport::Acquire( pool_candidates.erase(pool_candidates.begin() + static_cast(i - 1)); } - continue; } + } + size_t best = pool_candidates.size(); + size_t best_bound = 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->lifecycle.Activate()) { - MarkActive(candidate, lane); - pooled = candidate; - pool_candidates.erase( - pool_candidates.begin() + static_cast(i - 1)); - break; + candidate->declared_bytes >= required_bound && + candidate->declared_bytes < best_bound) { + best = i; + best_bound = candidate->declared_bytes; } } + if (best < pool_candidates.size() && + pool_candidates[best]->lifecycle.Activate()) { + pooled = pool_candidates[best]; + MarkActive(pooled, lane); + pool_candidates.erase(pool_candidates.begin() + + static_cast(best)); + } } } } @@ -1246,14 +1266,12 @@ RdmaTransport::AcquireResult RdmaTransport::Acquire( } endpoint_cache_misses_.fetch_add(1, std::memory_order_relaxed); - // Open, announce, and budget at the depth this node actually granted last - // time: churned reconnects stop reserving WR/slot capacity the server's - // clamp will never let them use (first contact still probes at depth_). + // Open, announce, and budget only the size class this operation needs. const size_t conn_depth = LearnedDepth(node, lane); const uint64_t conn_declared = lane == Lane::kControl ? static_cast(rdma::kV2ControlCap) - : declared_; + : static_cast(required_bound); const uint64_t conn_slot_bytes = static_cast(rdma::V2SlotSize( static_cast( std::max(conn_declared, rdma::kV2DataOffset)))); @@ -1319,6 +1337,7 @@ RdmaTransport::AcquireResult RdmaTransport::Acquire( conn->remote_lease_generation = remote_lease ? remote_lease->generation : 0; conn->remote_lease_held = remote_lease.has_value(); + conn->declared_bytes = static_cast(conn_declared); conn->remote_recovery_probe = remote_lease && remote_lease->recovery_probe; conn->lease = *lease; @@ -1641,6 +1660,19 @@ bool RdmaTransport::RegisterMemory(void* base, size_t size) { return true; } +std::vector RdmaTransport::IdleDataBounds( + const std::string& node) const { + std::lock_guard lock(mu_); + std::vector bounds; + const auto found = pool_.find(node); + if (found == pool_.end()) return bounds; + bounds.reserve(found->second.size()); + for (const Conn* conn : found->second) + bounds.push_back(conn->declared_bytes); + std::sort(bounds.begin(), bounds.end()); + return bounds; +} + std::string RdmaTransport::MetricsText() const { std::string s; s += "# HELP dfkv_rdma_client_conns_opened_total RDMA client connections opened\n"; @@ -1897,10 +1929,14 @@ std::string RdmaTransport::MetricsText() const { s += "# TYPE dfkv_rdma_client_max_block_seen_bytes gauge\n"; s += "dfkv_rdma_client_max_block_seen_bytes " + std::to_string(max_block_seen_.load(std::memory_order_relaxed)) + "\n"; - s += "# HELP dfkv_rdma_client_declared_max_block_bytes Negotiated maximum block declaration\n"; + s += "# HELP dfkv_rdma_client_declared_max_block_bytes Logical RDMA block safety ceiling\n"; s += "# TYPE dfkv_rdma_client_declared_max_block_bytes gauge\n"; s += "dfkv_rdma_client_declared_max_block_bytes " + std::to_string(declared_) + "\n"; + s += "# HELP dfkv_rdma_client_connection_min_block_bytes Smallest adaptive data-connection declaration\n"; + s += "# TYPE dfkv_rdma_client_connection_min_block_bytes gauge\n"; + s += "dfkv_rdma_client_connection_min_block_bytes " + + std::to_string(connection_min_block_bytes_) + "\n"; s += "# HELP dfkv_rdma_client_oversize_rejects_total Operations rejected above the declared maximum block\n"; s += "# TYPE dfkv_rdma_client_oversize_rejects_total counter\n"; s += "dfkv_rdma_client_oversize_rejects_total " + @@ -2031,6 +2067,7 @@ void RdmaTransport::Release(const std::string& node, Lane lane, Conn* c, bool reusable = false; bool publication_current = false; + Conn* displaced = nullptr; { std::lock_guard lk(mu_); publication_current = peer_topologies_->IsCurrent( @@ -2043,9 +2080,27 @@ void RdmaTransport::Release(const std::string& node, Lane lane, Conn* c, c->lane = lane; v.push_back(c); reusable = true; + } 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; + }); + if (largest != v.end() && + (*largest)->declared_bytes > c->declared_bytes && + (*largest)->lifecycle.RequestRetire()) { + displaced = *largest; + if (c->lifecycle.MakeIdle()) { + c->lane = lane; + *largest = c; + reusable = true; + } else { + v.erase(largest); + } + } } } } + if (displaced) Destroy(displaced); if (!reusable && !publication_current) stale_publication_reaps_.fetch_add(1, std::memory_order_relaxed); if (!reusable) Destroy(c); @@ -2086,6 +2141,10 @@ Status RdmaTransport::RoundTrip(const std::string& node, WireOp op, std::string attempt_out; uint64_t attempt_value_len = 0; AcquireOptions options; + options.required_data_bytes = + lane == Lane::kControl + ? 0 + : static_cast(std::max(payload_len, length)); options.force_new = attempt != 0; options.excluded = excluded; options.peer = peer; @@ -2310,6 +2369,7 @@ Status RdmaTransport::Remove(const std::string& node, const BlockKey& key) { std::vector RdmaTransport::CacheMany( const std::string& node, const std::vector& items) { const size_t count = items.size(); + size_t required_bytes = 0; std::vector result(count, Status::kIOError); if (count == 0) return result; std::vector bad(count, 0); @@ -2321,6 +2381,7 @@ std::vector RdmaTransport::CacheMany( bad[i] = 1; result[i] = Status::kInvalid; } else { + required_bytes = std::max(required_bytes, items[i].len); ++valid_count; } } @@ -2336,6 +2397,7 @@ std::vector RdmaTransport::CacheMany( for (size_t i = 0; i < count; ++i) if (bad[i]) result[i] = Status::kInvalid; AcquireOptions options; + options.required_data_bytes = required_bytes; options.force_new = attempt != 0; options.requested_credits = std::min(valid_count, depth_); options.excluded = excluded; @@ -2486,6 +2548,7 @@ std::vector RdmaTransport::RangeMany( std::vector attempt_outputs(count); std::vector attempt_value_lens(count, 0); AcquireOptions options; + options.required_data_bytes = static_cast(length); options.force_new = attempt != 0; options.requested_credits = std::min(count, depth_); options.excluded = excluded; @@ -2754,6 +2817,7 @@ std::vector RdmaTransport::RangeInto( NoteBlock(destination.n)) continue; AcquireOptions options; + options.required_data_bytes = destination.n; options.requested_credits = 1; options.peer = peer_topologies_->Snapshot(node); AcquireResult acquired = Acquire(node, Lane::kData, options); @@ -2876,6 +2940,7 @@ std::vector RdmaTransport::CacheFromMulti( std::vector completed(count, 0); std::vector totals(count, 0); size_t valid_count = 0; + size_t required_bytes = 0; for (size_t i = 0; i < count; ++i) { bool invalid = false; size_t total = 0; @@ -2895,6 +2960,7 @@ std::vector RdmaTransport::CacheFromMulti( bad[i] = 1; result[i] = Status::kInvalid; } else { + required_bytes = std::max(required_bytes, total); ++valid_count; } } @@ -2910,6 +2976,7 @@ std::vector RdmaTransport::CacheFromMulti( // validation results survive while connection-local progress restarts. completion_fault.BeginAttempt(attempt); AcquireOptions options; + options.required_data_bytes = required_bytes; options.force_new = attempt != 0; options.requested_credits = 1; options.excluded = excluded; @@ -3078,21 +3145,20 @@ std::vector RdmaTransport::CacheFromMulti( return result; } +size_t RdmaTransport::ConnectionBound(size_t required) const { + const size_t logical_max = OpBound(); + size_t wanted = std::max(required, connection_min_block_bytes_); + if (wanted >= logical_max) return logical_max; + size_t size_class = 1; + while (size_class < wanted && size_class <= logical_max / 2) + size_class <<= 1; + return size_class < wanted ? logical_max + : std::min(size_class, logical_max); +} + // Records n as a high-water candidate and reports whether it exceeds the -// declaration. Two problems this closes, both observed on a B200 node: -// -// 1. The declaration determines receive-segment capacity reserved for every -// data connection: queue depth multiplied by the aligned slot size. An -// over-generous value consumes scarce pinned memory across the fleet. The -// average transfer size (437 KiB on the incident host) says nothing about -// the peak block that must fit, so the observed high-water mark is the -// actionable sizing signal. -// -// 2. An UNDER-sized declaration failed silently. Oversized blocks become -// kInvalid, which the client's health accounting deliberately ignores, so -// upstream they are indistinguishable from an ordinary cache miss: no error, -// no log, no counter -- just a hit rate quietly capped for large pages. -// Anyone tuning this down would have had no signal that they went too far. +// logical safety declaration. Physical connection slots use ConnectionBound, +// so a rare large object no longer inflates every data connection. bool RdmaTransport::NoteBlock(size_t n) const { uint64_t prev = max_block_seen_.load(std::memory_order_relaxed); while (n > prev && @@ -3137,6 +3203,7 @@ std::vector RdmaTransport::RangeIntoMulti( } if (!valid || NoteBlock(capacity)) continue; AcquireOptions options; + options.required_data_bytes = capacity; options.requested_credits = 1; options.peer = peer_topologies_->Snapshot(node); AcquireResult acquired = Acquire(node, Lane::kSgData, options); diff --git a/src/transport/rdma_transport.h b/src/transport/rdma_transport.h index 4251e5a8..e5202f68 100644 --- a/src/transport/rdma_transport.h +++ b/src/transport/rdma_transport.h @@ -151,6 +151,7 @@ class RdmaTransport : public Transport { struct AcquireOptions { bool force_new = false; size_t requested_credits = 1; + size_t required_data_bytes = 0; RailMask excluded; std::shared_ptr peer; }; @@ -214,6 +215,7 @@ class RdmaTransport : public Transport { // never concurrently reused, while operation framing remains self-describing. std::unordered_map> pool_; // Exist/Remove/Members remain isolated from payload transfers. + std::vector IdleDataBounds(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 @@ -226,24 +228,22 @@ class RdmaTransport : public Transport { // min over rails of (negotiated max_sge) - 1; set once in the ctor. size_t sg_payload_segs_ = 29; size_t max_payload_; - // Maximum block declared during mandatory v2 negotiation. It is both a - // protocol bound and receive-segment geometry: each data connection leases - // queue_depth * aligned_slot_size from one fixed pinned segment. Inflating - // this value reduces connection capacity; understating it deterministically - // rejects larger operations. It must therefore be exact and nonzero. - // Default 4 MiB (env DFKV_RDMA_MAX_BLOCK_BYTES); capped by max_payload_. + // Logical per-object safety bound. Connection receive geometry is selected + // independently from the current operation's actual largest object. + // DFKV_RDMA_MAX_BLOCK_BYTES remains the deterministic hard rejection limit. uint64_t declared_ = 0; - size_t OpBound() const { // per-op payload bound honoring the declaration + // Smallest data connection declaration. Larger requests round up to the next + // power-of-two class, capped by declared_. Default 256 KiB. + size_t connection_min_block_bytes_ = 0; + size_t OpBound() const { return declared_ ? static_cast(declared_) : max_payload_; } - // Largest block this client has actually handed to the transport. Operators - // use the high-water mark to choose a tight DCP2 declaration: smaller slots - // admit more concurrent v2 connections into the fixed shared segment, while - // oversize operations must remain a deterministic client-side rejection. + size_t ConnectionBound(size_t required) const; + // Largest block this client has actually handed to the transport. mutable std::atomic max_block_seen_{0}; mutable std::atomic oversize_rejects_{0}; // Records n as a candidate high-water mark and reports whether it exceeds the - // declaration. Returns true for an oversized block (caller marks it kInvalid). + // logical bound. Returns true for an oversized block. bool NoteBlock(size_t n) const; size_t depth_; int connect_ms_ = 3000; // bootstrap TCP connect timeout (DFKV_RDMA_CONNECT_MS) diff --git a/test/cache/disk_slab_store_test.cc b/test/cache/disk_slab_store_test.cc index 54fc1b8c..80de530d 100644 --- a/test/cache/disk_slab_store_test.cc +++ b/test/cache/disk_slab_store_test.cc @@ -1355,6 +1355,44 @@ TEST_F(DiskSlabTest, WatermarkEvictionsStayDeadAcrossCleanRestart) { for (uint64_t i = 0; i < 4; ++i) EXPECT_FALSE(reopened.IsCached(K(i))); } +TEST_F(DiskSlabTest, WatermarkDrainIsLatchedAndBoundedPerTick) { + ::setenv("DFKV_SLAB_EVICT_HIGH_PCT", "50", 1); + ::setenv("DFKV_SLAB_EVICT_LOW_PCT", "25", 1); + ::setenv("DFKV_SLAB_EVICT_MAX_EXTENTS_PER_TICK", "1", 1); + auto options = Opts(8 * 4096, 4096, 4096); + options.reclaim_interval_ms = 0; + options.table_sync_ms = 0; + bool ok = false; + DiskSlabStore store(options, &ok); + ::unsetenv("DFKV_SLAB_EVICT_HIGH_PCT"); + ::unsetenv("DFKV_SLAB_EVICT_LOW_PCT"); + ::unsetenv("DFKV_SLAB_EVICT_MAX_EXTENTS_PER_TICK"); + ASSERT_TRUE(ok); + + std::string value(4000, 'w'); + for (uint64_t i = 0; i < 8; ++i) + ASSERT_EQ(store.Cache(K(i), value.data(), value.size()), Status::kOk); + + dfkv::DiskSlabStoreTestPeer::ReclaimNow(&store, {}); + auto stats = store.GetStats(); + EXPECT_EQ(store.Count(), 6u) + << "one bounded watermark extent plus regular free-slot reclaim"; + EXPECT_EQ(stats.watermark_evictions, 1u); + EXPECT_EQ(stats.watermark_extent_clears, 1u); + EXPECT_EQ(stats.watermark_active, 1u); + EXPECT_EQ(stats.watermark_max_extents_per_tick, 1u); + + for (int i = 0; i < 5; ++i) + dfkv::DiskSlabStoreTestPeer::ReclaimNow(&store, {}); + stats = store.GetStats(); + EXPECT_EQ(store.Count(), 1u); + EXPECT_EQ(stats.watermark_evictions, 6u); + EXPECT_EQ(stats.watermark_extent_clears, 6u); + EXPECT_EQ(stats.eviction_record_clears, 7u); + EXPECT_EQ(stats.watermark_ticks, 6u); + EXPECT_EQ(stats.watermark_active, 0u); +} + // Class rebalance regression (the "new value size retains only a sliver of its // writes" failure): fill the store with class A, then write a burst of class B // larger than B's first extent. Stock behavior self-evicts B forever (B ends diff --git a/test/cache/slab_allocator_test.cc b/test/cache/slab_allocator_test.cc index c4953a77..9b317487 100644 --- a/test/cache/slab_allocator_test.cc +++ b/test/cache/slab_allocator_test.cc @@ -320,6 +320,37 @@ TEST(SlabAllocator, EvictColdToTargetFreesGloballyColdestFirst) { EXPECT_TRUE(a.Contains(Key("k31"))) << "newest data must survive proactive eviction"; } +TEST(SlabAllocator, WholeExtentCallbackReplacesPerSlotMetadataClears) { + auto options = Opts(4 * 4096, 2); + uint32_t extent_calls = 0; + uint32_t slot_calls = 0; + uint32_t cleared_residents = 0; + options.on_extent_evict = + [&](uint32_t, uint32_t total_slots, uint32_t residents) { + ++extent_calls; + EXPECT_EQ(total_slots, 4u); + cleared_residents += residents; + return true; + }; + options.on_slot_evict = [&](const SlotRef&) { + ++slot_calls; + return true; + }; + SlabAllocator allocator(options); + std::vector evicted; + SlotRef ref; + for (int i = 0; i < 8; ++i) + ASSERT_TRUE(allocator.Put(Key("k" + std::to_string(i)), 4096, + &ref, &evicted)); + + evicted.clear(); + EXPECT_EQ(allocator.EvictColdToTarget(0, 1, &evicted), 1u); + EXPECT_EQ(extent_calls, 1u); + EXPECT_EQ(slot_calls, 0u); + EXPECT_EQ(cleared_residents, 4u); + EXPECT_EQ(evicted.size(), 4u); +} + TEST(SlabAllocator, EvictColdToTargetRespectsPinsAndTarget) { SlabAllocator a(Opts(4 * 4096, 4)); // 4 extents, 16 slots std::vector ev; diff --git a/test/python/test_observability_contract.py b/test/python/test_observability_contract.py index e7d8f01a..da300f70 100644 --- a/test/python/test_observability_contract.py +++ b/test/python/test_observability_contract.py @@ -101,6 +101,10 @@ "dfkv_rdma_rail_get_bytes_total", "dfkv_rdma_rail_put_bytes_total", "dfkv_rdma_recv_segment_bytes", + "dfkv_rdma_recv_segment_max_bytes", + "dfkv_rdma_recv_segment_chunks", + "dfkv_rdma_recv_segment_growths_total", + "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", @@ -143,6 +147,12 @@ "dfkv_slab_rebuild_rejected_records_total", "dfkv_slab_unclean_resets_total", "dfkv_slab_watermark_evictions_total", + "dfkv_slab_watermark_extent_clears_total", + "dfkv_slab_watermark_ticks_total", + "dfkv_slab_watermark_active", + "dfkv_slab_watermark_last_tick_us", + "dfkv_slab_watermark_max_tick_us", + "dfkv_slab_watermark_max_extents_per_tick", "dfkv_storage_healthy", "dfkv_tcp_max_connections", "dfkv_tcp_rejected_connections_total", diff --git a/test/transport/rdma_loopback_test.cc b/test/transport/rdma_loopback_test.cc index 98d3f776..bf46b39e 100644 --- a/test/transport/rdma_loopback_test.cc +++ b/test/transport/rdma_loopback_test.cc @@ -133,6 +133,16 @@ class RdmaTransportTestPeer { return found == transport->pool_.end() ? 0 : found->second.size(); } + static size_t ConnectionBound(const RdmaTransport& transport, + size_t required) { + return transport.ConnectionBound(required); + } + + static std::vector IdleDataBounds( + RdmaTransport* transport, const std::string& node) { + return transport->IdleDataBounds(node); + } + static uint64_t PeerPublication(const RdmaTransport& transport, const std::string& node) { return transport.peer_topologies_->Snapshot(node)->publication; @@ -206,8 +216,8 @@ class ScopedEnv { std::string SelfHdr() { return "test/model"; } void ConfigureTestRecvSegment() { - // Keep Soft-RoCE CI below modest RLIMIT_MEMLOCK. Production defaults to - // 2 GiB, but these fixtures use 256-KiB blocks and need only a small segment. + // Keep Soft-RoCE CI below modest RLIMIT_MEMLOCK. Production commits a + // 256-MiB initial chunk under a 16-GiB budget; fixtures need only 32 MiB. ::setenv("DFKV_RDMA_RECV_SEGMENT_SIZE", "33554432", 0); } @@ -3345,6 +3355,75 @@ TEST(RdmaLoopback, DeclaredCapsRoundTripAndClientSideBound) { EXPECT_EQ(got, v); } +TEST(RdmaLoopback, DataConnectionsUseActualBlockSizeClasses) { + if (!HaveRdma()) GTEST_SKIP() << "no RDMA device"; + ScopedEnv max_block("DFKV_RDMA_MAX_BLOCK_BYTES", "262144"); + ScopedEnv min_block("DFKV_RDMA_CONNECTION_MIN_BLOCK_BYTES", "4096"); + RdmaNode node("adaptive-classes"); + RdmaTransport transport(kMaxMsg); + KVClient client({{"n", node.addr}}, SelfHdr(), &transport); + + std::string small(4000, 's'); + ASSERT_TRUE(client.Put("small", small.data(), small.size())); + EXPECT_EQ(RdmaTransportTestPeer::IdleDataBounds(&transport, node.addr), + (std::vector{4096})); + + std::string large(60 * 1024, 'l'); + ASSERT_TRUE(client.Put("large", large.data(), large.size())); + EXPECT_EQ(RdmaTransportTestPeer::IdleDataBounds(&transport, node.addr), + (std::vector{4096, 65536})); + + std::string got(small.size(), '\0'); + ASSERT_TRUE(client.Get("small", got.data(), got.size())); + EXPECT_EQ(got, small); + EXPECT_EQ(RdmaTransportTestPeer::IdleDataBounds(&transport, node.addr), + (std::vector{4096, 65536})); + EXPECT_EQ(RdmaTransportTestPeer::ConnectionBound(transport, 1), 4096u); + EXPECT_EQ(RdmaTransportTestPeer::ConnectionBound(transport, 5000), 8192u); + EXPECT_EQ(RdmaTransportTestPeer::ConnectionBound(transport, 60 * 1024), + 65536u); +} + +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"); + RdmaNode node("lazy-recv-pool"); + + std::vector> transports; + std::vector> clients; + std::string value(60 * 1024, 'p'); + for (int i = 0; i < 4; ++i) { + auto transport = std::make_unique(kMaxMsg); + auto client = std::make_unique( + std::vector>{ + {"n", node.addr}}, + SelfHdr(), transport.get()); + ASSERT_TRUE(client->Put("pool-" + std::to_string(i), + value.data(), value.size())); + transports.push_back(std::move(transport)); + clients.push_back(std::move(client)); + } + + const std::string metrics = node.rsrv->MetricsText(); + const long chunks = + CounterVal(metrics, "dfkv_rdma_recv_segment_chunks"); + EXPECT_GE(chunks, 3); + EXPECT_EQ(CounterVal(metrics, "dfkv_rdma_recv_segment_max_bytes"), + 4 * 1024 * 1024); + EXPECT_EQ(CounterVal(metrics, "dfkv_rdma_recv_segment_bytes"), + chunks * 1024 * 1024); + EXPECT_EQ(CounterVal(metrics, + "dfkv_rdma_recv_segment_growths_total"), + chunks - 1); + EXPECT_EQ(CounterVal( + metrics, + "dfkv_rdma_recv_segment_growth_failures_total"), + 0); +} + // With no explicit override, DCP2 declares the transport's safe global cap; // blocks right up to that bound still round-trip. TEST(RdmaLoopback, DefaultDeclarationKeepsGlobalCap) { diff --git a/test/transport/rdma_recv_segment_test.cc b/test/transport/rdma_recv_segment_test.cc index 313cb0c6..6e23e77a 100644 --- a/test/transport/rdma_recv_segment_test.cc +++ b/test/transport/rdma_recv_segment_test.cc @@ -106,4 +106,46 @@ TEST(RecvSegment, SizeParserIs64BitAlignedAndNotWireCapped) { EXPECT_EQ(ResolveRecvSegmentBytes("8192", fallback, 3000), 0u); } +TEST(RecvSegmentPool, GrowsLazilyWithinHardBudgetAndReusesLeases) { + RecvSegmentPool pool; + ASSERT_TRUE(pool.Init(16u << 10, 48u << 10)); + EXPECT_EQ(pool.stats().committed_bytes, 16u << 10); + EXPECT_EQ(pool.stats().chunks, 1u); + + auto first = pool.Allocate(12u << 10); + auto second = pool.Allocate(12u << 10); + auto third = pool.Allocate(12u << 10); + ASSERT_TRUE(first); + ASSERT_TRUE(second); + ASSERT_TRUE(third); + EXPECT_NE(first.segment(), second.segment()); + EXPECT_NE(second.segment(), third.segment()); + + auto stats = pool.stats(); + EXPECT_EQ(stats.max_bytes, 48u << 10); + EXPECT_EQ(stats.committed_bytes, 48u << 10); + EXPECT_EQ(stats.used_bytes, 36u << 10); + EXPECT_EQ(stats.chunks, 3u); + EXPECT_EQ(stats.growths, 2u); + EXPECT_FALSE(pool.Allocate(12u << 10)); + EXPECT_EQ(pool.stats().growth_failures, 1u); + + RecvSegment* released_segment = second.segment(); + second.Reset(); + auto replacement = pool.Allocate(12u << 10); + ASSERT_TRUE(replacement); + EXPECT_EQ(replacement.segment(), released_segment); + EXPECT_EQ(pool.stats().growths, 2u); +} + +TEST(RecvSegmentPool, RejectsInvalidOrOversizedGeometry) { + RecvSegmentPool pool; + EXPECT_FALSE(pool.Init(0, 4096)); + EXPECT_FALSE(pool.Init(8192, 4096)); + ASSERT_TRUE(pool.Init(8192, 16384)); + EXPECT_EQ(pool.initial_segment()->size(), 8192u); + EXPECT_FALSE(pool.Allocate(20u << 10)); + EXPECT_EQ(pool.stats().committed_bytes, 8192u); +} + } // namespace dfkv::rdma