From 3646dfa7dfc507e44c8cbd09b76cd20da9bfedfc Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 15 Oct 2026 07:27:32 +0000 Subject: [PATCH] V2: feature-gated NIXL slot, shard fencing, and G4 refill TCP stays the correctness path. The rdma-nixl feature does not link libnixl. A shard-local directory, a simulated registration pool, and the cold sink/refill contract are covered by CPU tests. Co-authored-by: RustFS --- .github/workflows/ci.yml | 34 +- AGENTS.md | 8 +- COMPATIBILITY.md | 8 +- README.md | 35 +- THIRD_PARTY_NOTICES.md | 2 +- crates/bench/src/lib.rs | 14 + crates/cold/Cargo.toml | 4 + crates/cold/src/bin/cold_demo.rs | 25 ++ crates/cold/src/demo.rs | 104 +++++ crates/cold/src/lib.rs | 9 +- crates/cold/src/pack.rs | 37 +- crates/cold/src/sink.rs | 348 +++++++++++++++ crates/transport/Cargo.toml | 11 +- crates/transport/src/bin/shard_demo.rs | 21 + crates/transport/src/directory.rs | 561 +++++++++++++++++++++++++ crates/transport/src/iface.rs | 208 +++++++++ crates/transport/src/lib.rs | 18 +- crates/transport/src/registered.rs | 421 +++++++++++++++++++ crates/transport/src/shard_demo.rs | 122 ++++++ crates/transport/src/stage.rs | 80 ++++ crates/transport/tests/v2_control.rs | 56 +++ docs/ARCHITECTURE.md | 22 +- docs/BENCHMARKS.md | 7 +- docs/OPERATIONS.md | 15 + docs/REUSE_AUDIT.md | 16 +- docs/RISKS.md | 5 +- docs/V2_DATA_PATHS.md | 51 +++ docs/V2_TEST_MATRIX.md | 25 ++ docs/adr/0002-no-erasure-coding.md | 3 +- docs/adr/0004-nixl-transport.md | 37 ++ docs/adr/0005-shard-fencing.md | 27 ++ 31 files changed, 2286 insertions(+), 48 deletions(-) create mode 100644 crates/cold/src/bin/cold_demo.rs create mode 100644 crates/cold/src/demo.rs create mode 100644 crates/cold/src/sink.rs create mode 100644 crates/transport/src/bin/shard_demo.rs create mode 100644 crates/transport/src/directory.rs create mode 100644 crates/transport/src/iface.rs create mode 100644 crates/transport/src/registered.rs create mode 100644 crates/transport/src/shard_demo.rs create mode 100644 crates/transport/src/stage.rs create mode 100644 crates/transport/tests/v2_control.rs create mode 100644 docs/V2_DATA_PATHS.md create mode 100644 docs/V2_TEST_MATRIX.md create mode 100644 docs/adr/0004-nixl-transport.md create mode 100644 docs/adr/0005-shard-fencing.md diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b960acf..b6c31b6 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -24,14 +24,36 @@ jobs: run: cargo run -p gpucache-transport --bin fault-demo --quiet - name: Paired baseline stub run: cargo run -p gpucache-bench --bin paired-baseline --quiet - - name: Reject accelerator crates in V1 + - name: Shard and cold demos run: | - if grep -RInE 'cuda|doca|nixl|rdma|spdk' crates/*/Cargo.toml; then - echo "V1 crates must not depend on GPU or DPU stacks" - exit 1 - fi + cargo run -p gpucache-transport --bin shard-demo --quiet + cargo run -p gpucache-cold --bin cold-demo --quiet + - name: Optional RDMA feature still skips hardware + run: cargo test -p gpucache-transport --all-targets --features rdma-nixl + - name: Reject accelerator linkage and erasure coding + run: | + python3 - <<'PY' + import pathlib, re, sys + bad = re.compile( + r"^(cuda|doca|nixl|rdma|spdk|libnixl)([-_a-z0-9]*)?\s*=", + re.I, + ) + failed = False + for path in pathlib.Path("crates").glob("*/Cargo.toml"): + in_deps = False + for lineno, line in enumerate(path.read_text().splitlines(), 1): + stripped = line.split("#", 1)[0].strip() + if stripped.startswith("[") and stripped.endswith("]"): + in_deps = "dependencies" in stripped + continue + if in_deps and bad.match(stripped): + print(f"{path}:{lineno}: accelerator dependency is not allowed") + failed = True + if failed: + sys.exit(1) + PY if grep -RInE 'reed_solomon|erasure_coding|parity_shard' crates --include '*.rs'; then - echo "V1 must not add erasure coding" + echo "erasure coding must not be added" exit 1 fi - name: Keep the cold tier off the hot path diff --git a/AGENTS.md b/AGENTS.md index 61d1b16..c942fb6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -4,11 +4,11 @@ GPUCache is the G3.5 shared KV tier. RustFS is the colder G4 backend. Read [READ ## Phase -The active phase is **V1 milestone C**. Do not implement V2 RDMA or V3 DPU work in a V1 C change. Do not publish crates, cut releases, deploy, or edit a website unless asked. +The active phase is the **V2 first slice** (explicitly authorized). V1 A/B/C stay in tree and must keep passing. Do not start V3 DPU, DOCA, SPDK, or GDS work. Do not publish crates, cut releases, deploy, or edit a website unless asked. ## Honesty rules -- Do not claim DPU, DOCA, RDMA, SPDK, or CUDA acceleration. V1 is CPU only. An aarch64 build is not a BlueField test. +- Do not claim DPU, DOCA, RDMA, SPDK, or CUDA acceleration. The default build is CPU and TCP. Feature `rdma-nixl` does not link a device. An aarch64 build is not a BlueField test. GDS and BlueField stay in development / `NOT_RUN_HARDWARE`. - Do not invent vLLM, LMCache, or NIXL results. A real engine KV export/import needs a GPU and stays `NOT_RUN_HARDWARE` until that run exists. - Do not treat a KV block as model-agnostic long-term memory. A miss falls back to recompute. A wrong-tenant or wrong-layout hit is a defect. - Do not add erasure coding, parity, or durability replicas. A lost or failed hot block is a clean miss, then recompute and repopulate. Cold durability belongs to RustFS. An extra hot-prefix copy, if it is ever added, is a latency hint and not a durability promise. @@ -21,11 +21,11 @@ The active phase is **V1 milestone C**. Do not implement V2 RDMA or V3 DPU work | `gpucache-core-types` | `sha2` only | | `gpucache-protocol` | `gpucache-core-types` | | `gpucache-storage` | the two crates above | -| `gpucache-transport` | the three crates above. TCP only | +| `gpucache-transport` | the three crates above. TCP correctness path. Optional feature `rdma-nixl` must add no crate | | `gpucache-cold` | `gpucache-core-types` and `sha2`. Not a dependency of storage or transport | | `gpucache-bench` | nothing. Prints `NOT_RUN_HARDWARE` baseline rows and must not invent timings | -No crate may add CUDA, DOCA, NIXL, RDMA, SPDK, erasure coding, or replica placement. `unsafe` is forbidden. Registered buffers whose completion is unknown stay quarantined until drain or a newer epoch; do not free them only because a lease timer fired. The cold crate stays off the hot path. +No crate may add a CUDA, DOCA, NIXL, RDMA, SPDK, erasure-coding, or replica-placement dependency. The feature name `rdma-nixl` is a slot only. `unsafe` is forbidden. Registered buffers whose completion is unknown stay quarantined until drain or a newer epoch; do not free them only because a lease timer fired. The cold crate stays off the hot path. ## Checks diff --git a/COMPATIBILITY.md b/COMPATIBILITY.md index b56f20a..e6a80b8 100644 --- a/COMPATIBILITY.md +++ b/COMPATIBILITY.md @@ -13,11 +13,11 @@ Recorded for V1 milestones A, B, and C. This file is not a performance acceptanc | GPU / driver | None in this environment | `NOT_RUN_HARDWARE` | | vLLM | Not pinned; no engine commit | `NOT_RUN_HARDWARE` | | LMCache | Not a dependency | `NOT_RUN_HARDWARE` (route undecided) | -| NIXL / UCX / RDMA | Not a dependency | Not in V1 | -| DOCA / BlueField BFB / firmware | Not present | `NOT_RUN_HARDWARE` | -| RustFS server / client | No endpoint authorized in this environment. Contract mock covers Put, Head, RangeGet, Delete, pack, and manifest | Live probe `NOT_RUN_HARDWARE`. Mock `PASS` | +| NIXL / UCX / RDMA | Not linked. Candidate recorded: `nixl-sys` 1.3.1 (crates.io, updated 2026-07-07; docs.rs also lists 1.5.0), Apache-2.0, repo `ai-dynamo/nixl`. Feature `rdma-nixl` has no crate dependency. `stub-api` aborts on call, so it is not used. Mooncake Transfer Engine is Apache-2.0 and was not linked (in-tree `unsafe` FFI, not a pinned crates.io crate) | Live RDMA `NOT_RUN_HARDWARE`. `NixlTransport::open` returns `rdma not linked` with and without the feature | +| DOCA / BlueField BFB / firmware / GDS | Not present. Wording stays in development / `NOT_RUN_HARDWARE` | `NOT_RUN_HARDWARE` | +| RustFS server / client | No endpoint authorized in this environment. Contract mock covers Put, Head, RangeGet, Delete, pack, manifest, async sink, and cost-bounded refill | Live probe `NOT_RUN_HARDWARE`. Mock `PASS` | | TCP / file store | Loopback TCP and a temp directory on the build host | `PASS` for `cargo test --workspace --all-targets` on x86_64. Not an NVMe format and not a DPU run | -| Direct Rust deps | `sha2` 0.10.8; dev `proptest` 1.4.0 | Apache-2.0 OR MIT; see `Cargo.lock`. V1 C adds no direct crate. `gpucache-bench` has an empty dependency list | +| Direct Rust deps | `sha2` 0.10.8; dev `proptest` 1.4.0 | Apache-2.0 OR MIT; see `Cargo.lock`. The V2 slice adds no direct crate. `gpucache-bench` has an empty dependency list. Optional feature `rdma-nixl` adds none | | License of this repo | Apache-2.0 | See `LICENSE` | CI (see `.github/workflows/ci.yml`) builds and runs the same tests on x86_64 and on aarch64 (GitHub ARM runner, plus qemu-user cross execution). An aarch64 result is a CPU result. It does not certify BlueField-3, DOCA, or NVMe-on-DPU. diff --git a/README.md b/README.md index d85c42f..3b806e8 100644 --- a/README.md +++ b/README.md @@ -8,9 +8,9 @@ GPUCache does not do erasure coding and does not keep durability replicas. A los ## Status -**V1 milestones A, B, and C are implemented. DPU acceleration is not.** +**V1 milestones A, B, and C are implemented. The V2 first slice is in progress. DPU acceleration is not.** -V1 runs the cache contract on CPU. The long-term deployment target is a storage-side BlueField-3, but this repository does not link DOCA, CUDA, RDMA, or SPDK, and it does not claim that any of those paths exist. An aarch64 build is a CPU build. It is not a BlueField functional test. +V1 runs the cache contract on CPU. V2 adds a transport trait, a simulated registered-buffer pool, a shard-local directory, and a stricter G4 cold contract. The long-term deployment target is a storage-side BlueField-3, but this repository does not link DOCA, CUDA, RDMA, or SPDK, and it does not claim that any of those paths exist. Feature `rdma-nixl` does not link libnixl. An aarch64 build is a CPU build. It is not a BlueField functional test. GDS and BlueField offload are in development / `NOT_RUN_HARDWARE`. | Milestone | Status | | --- | --- | @@ -19,7 +19,12 @@ V1 runs the cache contract on CPU. The long-term deployment target is a storage- | V1 B: TCP service, file backend, LRU admission, tenant quotas, two-client reuse | Implemented in this tree | | V1 B: RustFS live Put/Head/RangeGet/Delete | `NOT_RUN_HARDWARE` (no authorized endpoint). Contract mock passes | | V1 C: fuzz, fault injection, synthetic token checks, paired-baseline stubs | Implemented in this tree. Real engine output is `NOT_RUN_HARDWARE` | -| V2: NIC RDMA (NIXL or Mooncake) plus a real RustFS cold path | Not started; blocked on V1 | +| V2: `KvTransport`, TCP default, feature `rdma-nixl` | In progress. The feature compiles and `NixlTransport::open` refuses. Live RDMA is `NOT_RUN_HARDWARE` | +| V2: bounded registered buffer pool (simulated pin, reuse, quarantine, backpressure) | Implemented in this tree. Not a NIC registration | +| V2: shard-local directory, epoch fencing, single-flight, capacity credits | Implemented in-process. One process is one shard. No cross-shard quorum | +| V2: G4 pack/manifest, async sink, cost-bounded refill | Implemented against the mock. Live RustFS is `NOT_RUN_HARDWARE` | +| V2: P/D, G3.5, and G4 hop list | Documented in [docs/V2_DATA_PATHS.md](docs/V2_DATA_PATHS.md). Not a zero-copy claim | +| V2: paired TTFT/TPOT against native NIXL | `NOT_RUN_HARDWARE` | | V3: BlueField-3 data-path offload and production hardening | Not started; not purchased and not measured. Not an erasure-coding milestone | | Erasure coding and durability replicas | Will not be built. A lost or failed hot block is a clean miss, then recompute and repopulate. Cold durability is RustFS G4 | @@ -43,7 +48,7 @@ The batch API is Lookup, Reserve/Put, Commit, Get, Release/Cancel, and Invalidat Lookup can return `PrefixHit` when a committed block's hash chain is a strict prefix of the query and the layout, dtype, shard, and model identity match. The hit is the whole stored block. `Get` must use `identity_for_prefix_hit`. A shorter `valid_length` on the same chain is an incompatible miss. The service does not slice inside a block. -`gpucache-cold` packs blocks into an object plus a manifest and can Put, Head, RangeGet, and Delete through a trait. The TCP hot path does not call it. Without `GPUCACHE_RUSTFS_*` variables the live probe is `NOT_RUN_HARDWARE`. The mock still runs. +`gpucache-cold` packs blocks into an object plus a manifest and can Put, Head, RangeGet, and Delete through a trait. The TCP hot path does not call it. Without `GPUCACHE_RUSTFS_*` variables the live probe is `NOT_RUN_HARDWARE`. The mock still runs. V2 adds a bounded async sink and a refill that returns no bytes when the manifest or the payload range exceeds its budget. Lifecycle: @@ -87,6 +92,22 @@ Paired GPU baselines stay `NOT_RUN_HARDWARE` and print no timings: cargo run -p gpucache-bench --bin paired-baseline ``` +One shard, simulated registration, and the TCP staging path: + +```bash +cargo run -p gpucache-transport --bin shard-demo +``` + +Success is `status=SHARD_OK`, `registration=SIMULATED`, and `rdma=NOT_RUN_HARDWARE`. + +Mock G4 pack, sink, and budget-capped refill: + +```bash +cargo run -p gpucache-cold --bin cold-demo +``` + +Success is `status=COLD_CONTRACT_OK` and `gds=NOT_RUN_HARDWARE`. `rustfs_live=NOT_RUN_HARDWARE` when no endpoint is configured. + `kv-block-xfer import ` reads one export frame on stdin and imports it into a fresh `MemoryStore`. It does not talk to an engine. ## Layout @@ -95,14 +116,14 @@ cargo run -p gpucache-bench --bin paired-baseline crates/core-types identity, keys, lifecycle, errors crates/protocol batch messages, replies, version negotiation crates/storage MemoryStore, FileStore, kv-block-xfer -crates/transport TCP server, client, share-demo, fault-demo -crates/cold RustFS contract, pack/manifest, live probe +crates/transport TCP server, client, transport trait, shard directory, share-demo, fault-demo, shard-demo +crates/cold RustFS contract, pack/manifest, async sink, refill, cold-demo crates/bench paired-baseline stub (no timings) docs/adr decision records COMPATIBILITY.md environment manifest ``` -Design notes: [docs/ARCHITECTURE.md](docs/ARCHITECTURE.md), [docs/PROTOCOL.md](docs/PROTOCOL.md), [docs/SECURITY.md](docs/SECURITY.md), [docs/OPERATIONS.md](docs/OPERATIONS.md), [docs/BENCHMARKS.md](docs/BENCHMARKS.md), [docs/CORRECTNESS.md](docs/CORRECTNESS.md), [docs/RISKS.md](docs/RISKS.md), [docs/V1C_TEST_MATRIX.md](docs/V1C_TEST_MATRIX.md). +Design notes: [docs/ARCHITECTURE.md](docs/ARCHITECTURE.md), [docs/PROTOCOL.md](docs/PROTOCOL.md), [docs/SECURITY.md](docs/SECURITY.md), [docs/OPERATIONS.md](docs/OPERATIONS.md), [docs/BENCHMARKS.md](docs/BENCHMARKS.md), [docs/CORRECTNESS.md](docs/CORRECTNESS.md), [docs/RISKS.md](docs/RISKS.md), [docs/V1C_TEST_MATRIX.md](docs/V1C_TEST_MATRIX.md), [docs/V2_DATA_PATHS.md](docs/V2_DATA_PATHS.md), [docs/V2_TEST_MATRIX.md](docs/V2_TEST_MATRIX.md). ## License diff --git a/THIRD_PARTY_NOTICES.md b/THIRD_PARTY_NOTICES.md index 052e06d..aa820ba 100644 --- a/THIRD_PARTY_NOTICES.md +++ b/THIRD_PARTY_NOTICES.md @@ -1,6 +1,6 @@ # Third-party notices -GPUCache V1 A, V1 B, and V1 C are Apache-2.0. See [LICENSE](LICENSE). V1 C adds no direct dependency. +GPUCache is Apache-2.0. See [LICENSE](LICENSE). The V2 first slice adds no direct dependency. NIXL (`nixl-sys`, Apache-2.0) and Mooncake (Apache-2.0) were audited and not linked. See [docs/adr/0004-nixl-transport.md](docs/adr/0004-nixl-transport.md). Direct dependencies: diff --git a/crates/bench/src/lib.rs b/crates/bench/src/lib.rs index 4c500a0..7d623d7 100644 --- a/crates/bench/src/lib.rs +++ b/crates/bench/src/lib.rs @@ -20,6 +20,13 @@ pub fn paired_baseline_report() -> String { "scenario_hbm_within=NOT_RUN_HARDWARE", "scenario_hbm_cliff=NOT_RUN_HARDWARE", "write_cost=NOT_RUN_HARDWARE", + "nixl_pd=NOT_RUN_HARDWARE", + "shared_dram=NOT_RUN_HARDWARE", + "shared_nvme=NOT_RUN_HARDWARE", + "rustfs_refill=NOT_RUN_HARDWARE", + "ttft=NOT_RUN_HARDWARE", + "tpot=NOT_RUN_HARDWARE", + "goodput=NOT_RUN_HARDWARE", "samples=NONE", "numeric_tolerance=NOT_APPLICABLE", "bypass_when_no_reuse=MISS_RECOMPUTE", @@ -50,6 +57,13 @@ mod tests { "scenario_multi_turn=NOT_RUN_HARDWARE", "scenario_hbm_within=NOT_RUN_HARDWARE", "scenario_hbm_cliff=NOT_RUN_HARDWARE", + "nixl_pd=NOT_RUN_HARDWARE", + "shared_dram=NOT_RUN_HARDWARE", + "shared_nvme=NOT_RUN_HARDWARE", + "rustfs_refill=NOT_RUN_HARDWARE", + "ttft=NOT_RUN_HARDWARE", + "tpot=NOT_RUN_HARDWARE", + "goodput=NOT_RUN_HARDWARE", "samples=NONE", ] { assert!(report.contains(key), "missing {key}"); diff --git a/crates/cold/Cargo.toml b/crates/cold/Cargo.toml index 816774d..7dbd1d4 100644 --- a/crates/cold/Cargo.toml +++ b/crates/cold/Cargo.toml @@ -8,6 +8,10 @@ authors.workspace = true description = "Optional RustFS G4 cold-tier contract. Not on the GPUCache hot path." publish = false +[[bin]] +name = "cold-demo" +path = "src/bin/cold_demo.rs" + [dependencies] gpucache-core-types.workspace = true sha2.workspace = true diff --git a/crates/cold/src/bin/cold_demo.rs b/crates/cold/src/bin/cold_demo.rs new file mode 100644 index 0000000..2814470 --- /dev/null +++ b/crates/cold/src/bin/cold_demo.rs @@ -0,0 +1,25 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 +//! Mock G4 pack, async sink, and cost-bounded refill. +//! +//! Success is `status=COLD_CONTRACT_OK`. Live RustFS stays +//! `NOT_RUN_HARDWARE` unless `GPUCACHE_RUSTFS_*` is set and the probe passes. +//! This is not a GDS run. + +use std::process::ExitCode; + +use gpucache_cold::run_cold_demo; + +fn main() -> ExitCode { + match run_cold_demo() { + Ok(report) => { + print!("{report}"); + ExitCode::SUCCESS + } + Err(err) => { + eprintln!("status=COLD_FAILED"); + eprintln!("error={err}"); + ExitCode::from(2) + } + } +} diff --git a/crates/cold/src/demo.rs b/crates/cold/src/demo.rs new file mode 100644 index 0000000..a385e3a --- /dev/null +++ b/crates/cold/src/demo.rs @@ -0,0 +1,104 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 + +//! CPU contract for the G4 pack, sink, and refill path. +//! +//! A missing `GPUCACHE_RUSTFS_*` environment is `NOT_RUN_HARDWARE`. The mock +//! result is not a measured RustFS or GDS run. + +use core::fmt; + +use gpucache_core_types::{content_checksum, synthetic_identity, CacheError}; + +use crate::http::{live_probe, ProbeStatus}; +use crate::pack::{contract_smoke, pack_blocks, PackedBlock}; +use crate::sink::{refill_generation, AsyncSink, ColdJob, RefillBudget, RefillOutcome}; +use crate::MockObjectStore; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ColdReport { + pub live: &'static str, +} + +impl fmt::Display for ColdReport { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + writeln!(f, "status=COLD_CONTRACT_OK")?; + writeln!(f, "mock=PASS")?; + writeln!(f, "async_sink=DRAINED")?; + writeln!(f, "refill=CHECKSUM_OK")?; + writeln!(f, "over_budget=NO_BYTES")?; + writeln!(f, "rustfs_live={}", self.live)?; + writeln!(f, "gds=NOT_RUN_HARDWARE")?; + Ok(()) + } +} + +pub fn run_cold_demo() -> Result { + let mut store = MockObjectStore::default(); + contract_smoke(&mut store)?; + let blocks = demo_blocks()?; + let pack = pack_blocks("demo", &blocks)?; + let mut sink = AsyncSink::new(2, 1024 * 1024)?; + sink.enqueue(ColdJob::from_pack(&pack))?; + if sink.drain(&mut store, 1)? != 1 { + return Err(CacheError::Storage { + reason: "cold sink", + }); + } + let budget = RefillBudget::new(1024 * 1024, 16)?; + match refill_generation(&store, &pack.manifest_key, 2, &budget)? { + RefillOutcome::Loaded { payload } if payload == blocks[1].payload => {} + _ => { + return Err(CacheError::ChecksumMismatch); + } + } + let tight = RefillBudget::new(1, 16)?; + if refill_generation(&store, &pack.manifest_key, 2, &tight)? != RefillOutcome::OverBudget { + return Err(CacheError::Malformed { + reason: "refill budget", + }); + } + let live = match live_probe().status { + ProbeStatus::NotRunHardware { .. } => "NOT_RUN_HARDWARE", + ProbeStatus::Passed => "PASSED", + ProbeStatus::Failed { .. } => { + return Err(CacheError::Storage { + reason: "object store", + }) + } + }; + Ok(ColdReport { live }) +} + +fn demo_blocks() -> Result, CacheError> { + let mut out = Vec::new(); + for (generation, byte) in [(1u64, 9u8), (2, 8u8)] { + let payload = vec![byte; 16]; + let mut identity = synthetic_identity(byte, 1); + identity.prefix_chain[0].0[0] = byte; + out.push(PackedBlock { + generation, + checksum: content_checksum(&payload), + payload, + identity, + }); + } + Ok(out) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn demo_stays_a_mock_without_an_endpoint() { + let report = run_cold_demo().unwrap(); + let text = report.to_string(); + assert!(text.contains("status=COLD_CONTRACT_OK")); + assert!(text.contains("over_budget=NO_BYTES")); + assert!(text.contains("gds=NOT_RUN_HARDWARE")); + if std::env::var_os("GPUCACHE_RUSTFS_ENDPOINT").is_none() { + assert!(text.contains("rustfs_live=NOT_RUN_HARDWARE")); + } + } +} diff --git a/crates/cold/src/lib.rs b/crates/cold/src/lib.rs index eae6265..7548fc1 100644 --- a/crates/cold/src/lib.rs +++ b/crates/cold/src/lib.rs @@ -9,11 +9,18 @@ #![forbid(unsafe_code)] +mod demo; mod http; mod mock; mod pack; mod sigv4; +mod sink; +pub use demo::{run_cold_demo, ColdReport}; pub use http::{live_probe, ProbeReport, ProbeStatus}; pub use mock::MockObjectStore; -pub use pack::{contract_smoke, pack_blocks, ObjectMeta, ObjectStore, PackedBlock, PayloadRange}; +pub use pack::{ + contract_smoke, pack_blocks, read_manifest, Manifest, ObjectMeta, ObjectStore, PackedBlock, + PayloadRange, +}; +pub use sink::{refill_generation, AsyncSink, ColdJob, RefillBudget, RefillOutcome}; diff --git a/crates/cold/src/pack.rs b/crates/cold/src/pack.rs index e06198b..884722c 100644 --- a/crates/cold/src/pack.rs +++ b/crates/cold/src/pack.rs @@ -32,6 +32,13 @@ pub struct PayloadRange { pub checksum: Checksum, } +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Manifest { + pub object_checksum: Checksum, + pub object_key: String, + pub ranges: Vec, +} + #[derive(Clone, Debug, PartialEq, Eq)] pub struct Pack { pub object_key: String, @@ -128,13 +135,13 @@ pub fn contract_smoke(store: &mut dyn ObjectStore) -> Result<(), CacheError> { if manifest_meta.checksum != content_checksum(&pack.manifest_bytes) { return Err(CacheError::ChecksumMismatch); } - let decoded = decode_manifest(&pack.manifest_bytes)?; - if decoded.len() != pack.ranges.len() { + let decoded = read_manifest(&pack.manifest_bytes)?; + if decoded.object_key != pack.object_key || decoded.ranges.len() != pack.ranges.len() { return Err(CacheError::Malformed { reason: "cold manifest", }); } - let range = &decoded[1]; + let range = &decoded.ranges[1]; let got = store.range_get(&pack.object_key, range.offset, range.length)?; verify_checksum(&got, range.checksum)?; if got != blocks[1].payload { @@ -166,7 +173,7 @@ fn sample_blocks() -> Result, CacheError> { Ok(out) } -fn decode_manifest(bytes: &[u8]) -> Result, CacheError> { +pub fn read_manifest(bytes: &[u8]) -> Result { let mut r = ByteReader::new(bytes); let magic = r.raw(4)?; if magic != MANIFEST_MAGIC || r.u16()? != VERSION { @@ -174,8 +181,18 @@ fn decode_manifest(bytes: &[u8]) -> Result, CacheError> { reason: "store corrupt", }); } - let _object_checksum = r.array32()?; - let _object_key = r.bytes(1024)?; + let object_checksum = Checksum(r.array32()?); + let object_key_bytes = r.bytes(1024)?; + let object_key = std::str::from_utf8(object_key_bytes) + .map_err(|_| CacheError::Storage { + reason: "store corrupt", + })? + .to_string(); + if !key_ok(&object_key) { + return Err(CacheError::Storage { + reason: "store corrupt", + }); + } let n = r.u32()? as usize; if n == 0 || n > 1024 { return Err(CacheError::Malformed { @@ -192,10 +209,14 @@ fn decode_manifest(bytes: &[u8]) -> Result, CacheError> { }); } r.finish()?; - Ok(ranges) + Ok(Manifest { + object_checksum, + object_key, + ranges, + }) } -fn key_ok(key: &str) -> bool { +pub(crate) fn key_ok(key: &str) -> bool { !key.is_empty() && key.len() <= 256 && key diff --git a/crates/cold/src/sink.rs b/crates/cold/src/sink.rs new file mode 100644 index 0000000..d7dbe68 --- /dev/null +++ b/crates/cold/src/sink.rs @@ -0,0 +1,348 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 + +//! Decoupled cold writes and a budget-capped refill. +//! +//! `AsyncSink` is a bounded queue drained by the caller. It is not a Tokio +//! runtime and it does not run on the hot TCP path. `refill_generation` +//! returns payload bytes only after a checksum match that fits the budget. +//! Over budget is an empty outcome: the engine recomputes. + +use std::collections::VecDeque; + +use gpucache_core_types::{content_checksum, CacheError, QuotaKind}; + +use crate::pack::{key_ok, read_manifest, Manifest, ObjectStore, Pack}; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ColdJob { + pub object_key: String, + pub object_bytes: Vec, + pub manifest_key: String, + pub manifest_bytes: Vec, +} + +impl ColdJob { + pub fn from_pack(pack: &Pack) -> Self { + Self { + object_key: pack.object_key.clone(), + object_bytes: pack.object_bytes.clone(), + manifest_key: pack.manifest_key.clone(), + manifest_bytes: pack.manifest_bytes.clone(), + } + } + + fn weight(&self) -> u64 { + self.object_bytes.len() as u64 + self.manifest_bytes.len() as u64 + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct RefillBudget { + pub max_bytes: u64, + pub max_ranges: u32, +} + +impl RefillBudget { + pub fn new(max_bytes: u64, max_ranges: u32) -> Result { + if max_bytes == 0 || max_ranges == 0 { + return Err(CacheError::Malformed { + reason: "refill budget", + }); + } + Ok(Self { + max_bytes, + max_ranges, + }) + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum RefillOutcome { + Loaded { + payload: Vec, + }, + /// Manifest or payload exceeds the budget. No payload bytes. + OverBudget, + /// Clean miss. The caller recomputes. + Missing, +} + +pub struct AsyncSink { + queue: VecDeque, + max_jobs: usize, + max_bytes: u64, + queued_bytes: u64, + cancelled: bool, +} + +impl AsyncSink { + pub fn new(max_jobs: usize, max_bytes: u64) -> Result { + if max_jobs == 0 || max_bytes == 0 { + return Err(CacheError::Malformed { + reason: "cold sink", + }); + } + Ok(Self { + queue: VecDeque::new(), + max_jobs, + max_bytes, + queued_bytes: 0, + cancelled: false, + }) + } + + pub fn queued(&self) -> usize { + self.queue.len() + } + + pub fn enqueue(&mut self, job: ColdJob) -> Result<(), CacheError> { + if self.cancelled { + return Err(CacheError::Cancelled); + } + if !key_ok(&job.object_key) || !key_ok(&job.manifest_key) { + return Err(CacheError::Malformed { reason: "cold key" }); + } + let add = job.weight(); + if add > self.max_bytes { + return Err(CacheError::ObjectTooLarge { + limit: self.max_bytes, + actual: add, + }); + } + let next = self.queued_bytes.saturating_add(add); + if self.queue.len() >= self.max_jobs || next > self.max_bytes { + return Err(CacheError::QuotaExceeded { + kind: QuotaKind::InFlightBytes, + }); + } + self.queued_bytes = next; + self.queue.push_back(job); + Ok(()) + } + + /// Put at most `max_jobs` queued packs. The object is written first, then + /// the manifest, so a manifest is not visible without its object. A + /// failed manifest write deletes the object just written. + pub fn drain( + &mut self, + store: &mut dyn ObjectStore, + max_jobs: usize, + ) -> Result { + if self.cancelled { + return Err(CacheError::Cancelled); + } + let mut done = 0u32; + while (done as usize) < max_jobs { + let Some(job) = self.queue.pop_front() else { + break; + }; + self.queued_bytes = self.queued_bytes.saturating_sub(job.weight()); + store.put(&job.object_key, &job.object_bytes)?; + if let Err(err) = store.put(&job.manifest_key, &job.manifest_bytes) { + let _ = store.delete(&job.object_key); + return Err(err); + } + done = done.saturating_add(1); + } + Ok(done) + } + + /// Drop queued work. Later enqueue and drain fail with `Cancelled`. + pub fn cancel(&mut self) { + self.queue.clear(); + self.queued_bytes = 0; + self.cancelled = true; + } +} + +/// Fetch one generation from a packed object if it fits `budget`. +/// +/// The manifest is not downloaded when its length exceeds the budget. The +/// payload range is not downloaded when its length exceeds the budget. +pub fn refill_generation( + store: &dyn ObjectStore, + manifest_key: &str, + generation: u64, + budget: &RefillBudget, +) -> Result { + if !key_ok(manifest_key) { + return Err(CacheError::Malformed { reason: "cold key" }); + } + let meta = match store.head(manifest_key) { + Ok(meta) => meta, + Err(CacheError::Storage { + reason: "object missing", + }) => return Ok(RefillOutcome::Missing), + Err(err) => return Err(err), + }; + if meta.len > budget.max_bytes { + return Ok(RefillOutcome::OverBudget); + } + let manifest_bytes = store.range_get(manifest_key, 0, meta.len)?; + if manifest_bytes.len() as u64 != meta.len || content_checksum(&manifest_bytes) != meta.checksum + { + return Err(CacheError::ChecksumMismatch); + } + let manifest = read_manifest(&manifest_bytes)?; + if manifest.ranges.len() as u32 > budget.max_ranges { + return Ok(RefillOutcome::OverBudget); + } + let Some(range) = manifest + .ranges + .iter() + .find(|range| range.generation == generation) + else { + return Ok(RefillOutcome::Missing); + }; + if range.length == 0 || range.length > budget.max_bytes { + return Ok(RefillOutcome::OverBudget); + } + load_range(store, &manifest, range.offset, range.length, range.checksum) +} + +fn load_range( + store: &dyn ObjectStore, + manifest: &Manifest, + offset: u64, + length: u64, + checksum: gpucache_core_types::Checksum, +) -> Result { + let payload = match store.range_get(&manifest.object_key, offset, length) { + Ok(payload) => payload, + Err(CacheError::Storage { + reason: "object missing", + }) => return Ok(RefillOutcome::Missing), + Err(err) => return Err(err), + }; + if payload.len() as u64 != length || content_checksum(&payload) != checksum { + return Err(CacheError::ChecksumMismatch); + } + Ok(RefillOutcome::Loaded { payload }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::pack::{pack_blocks, PackedBlock}; + use crate::MockObjectStore; + use gpucache_core_types::{content_checksum, synthetic_identity}; + use std::cell::Cell; + + fn block(generation: u64, byte: u8, len: usize) -> PackedBlock { + let payload = vec![byte; len]; + let mut identity = synthetic_identity(byte, 1); + identity.prefix_chain[0].0[0] = byte; + PackedBlock { + generation, + identity, + checksum: content_checksum(&payload), + payload, + } + } + + struct CountStore { + inner: MockObjectStore, + manifest_ranges: Cell, + object_ranges: Cell, + } + + impl ObjectStore for CountStore { + fn put(&mut self, key: &str, body: &[u8]) -> Result<(), CacheError> { + self.inner.put(key, body) + } + + fn head(&self, key: &str) -> Result { + self.inner.head(key) + } + + fn range_get(&self, key: &str, offset: u64, length: u64) -> Result, CacheError> { + if key.ends_with("/manifest") { + self.manifest_ranges.set(self.manifest_ranges.get() + 1); + } + if key.ends_with("/object") { + self.object_ranges.set(self.object_ranges.get() + 1); + } + self.inner.range_get(key, offset, length) + } + + fn delete(&mut self, key: &str) -> Result<(), CacheError> { + self.inner.delete(key) + } + } + + #[test] + fn drain_then_refill_checks_the_budget_before_the_payload() { + let blocks = vec![block(1, 1, 32), block(2, 2, 4096)]; + let pack = pack_blocks("refill", &blocks).unwrap(); + let mut store = CountStore { + inner: MockObjectStore::default(), + manifest_ranges: Cell::new(0), + object_ranges: Cell::new(0), + }; + let mut sink = AsyncSink::new(2, 8 * 1024 * 1024).unwrap(); + sink.enqueue(ColdJob::from_pack(&pack)).unwrap(); + assert_eq!(sink.drain(&mut store, 1).unwrap(), 1); + let manifest_len = pack.manifest_bytes.len() as u64; + assert!(manifest_len < 4096); + let tight = RefillBudget::new(manifest_len, 16).unwrap(); + let skipped = refill_generation(&store, &pack.manifest_key, 2, &tight).unwrap(); + assert_eq!(skipped, RefillOutcome::OverBudget); + assert_eq!(store.manifest_ranges.get(), 1); + assert_eq!(store.object_ranges.get(), 0); + let wide = RefillBudget::new(8 * 1024 * 1024, 16).unwrap(); + let loaded = refill_generation(&store, &pack.manifest_key, 2, &wide).unwrap(); + assert_eq!( + loaded, + RefillOutcome::Loaded { + payload: blocks[1].payload.clone(), + } + ); + assert_eq!(store.object_ranges.get(), 1); + } + + #[test] + fn cancel_drops_the_queue_and_writes_nothing() { + let blocks = vec![block(1, 3, 8), block(2, 4, 8)]; + let pack = pack_blocks("cancel", &blocks).unwrap(); + let mut sink = AsyncSink::new(1, 1024 * 1024).unwrap(); + sink.enqueue(ColdJob::from_pack(&pack)).unwrap(); + let again = sink.enqueue(ColdJob::from_pack(&pack)).unwrap_err(); + assert!(matches!( + again, + CacheError::QuotaExceeded { + kind: QuotaKind::InFlightBytes + } + )); + sink.cancel(); + assert_eq!(sink.queued(), 0); + let mut store = MockObjectStore::default(); + assert_eq!( + sink.drain(&mut store, 4).unwrap_err(), + CacheError::Cancelled + ); + assert!(matches!( + store.head(&pack.object_key), + Err(CacheError::Storage { + reason: "object missing" + }) + )); + } + + #[test] + fn huge_manifest_is_not_downloaded() { + let blocks = vec![block(1, 5, 8), block(2, 6, 8)]; + let pack = pack_blocks("tiny", &blocks).unwrap(); + let mut store = CountStore { + inner: MockObjectStore::default(), + manifest_ranges: Cell::new(0), + object_ranges: Cell::new(0), + }; + store.put(&pack.manifest_key, &pack.manifest_bytes).unwrap(); + let budget = RefillBudget::new(1, 16).unwrap(); + let outcome = refill_generation(&store, &pack.manifest_key, 1, &budget).unwrap(); + assert_eq!(outcome, RefillOutcome::OverBudget); + assert_eq!(store.manifest_ranges.get(), 0); + assert_eq!(store.object_ranges.get(), 0); + } +} diff --git a/crates/transport/Cargo.toml b/crates/transport/Cargo.toml index bd9b455..3a9bf7f 100644 --- a/crates/transport/Cargo.toml +++ b/crates/transport/Cargo.toml @@ -5,9 +5,14 @@ edition.workspace = true license.workspace = true rust-version.workspace = true authors.workspace = true -description = "Localhost TCP transport for the GPUCache V1 B cache service. No RDMA or DPU dependencies." +description = "Localhost TCP transport for GPUCache. Optional rdma-nixl feature does not link a device." publish = false +[features] +default = [] +# Slot only. Does not pull a native library. See docs/adr/0004-nixl-transport.md. +rdma-nixl = [] + [dependencies] gpucache-core-types.workspace = true gpucache-protocol.workspace = true @@ -25,5 +30,9 @@ path = "src/bin/share_demo.rs" name = "fault-demo" path = "src/bin/fault_demo.rs" +[[bin]] +name = "shard-demo" +path = "src/bin/shard_demo.rs" + [lints] workspace = true diff --git a/crates/transport/src/bin/shard_demo.rs b/crates/transport/src/bin/shard_demo.rs new file mode 100644 index 0000000..c30ed77 --- /dev/null +++ b/crates/transport/src/bin/shard_demo.rs @@ -0,0 +1,21 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 +//! One shard, simulated registration, TCP staging. Not an RDMA run. + +use std::process::ExitCode; + +use gpucache_transport::run_shard_demo; + +fn main() -> ExitCode { + match run_shard_demo() { + Ok(report) => { + print!("{report}"); + ExitCode::SUCCESS + } + Err(err) => { + eprintln!("status=SHARD_FAILED"); + eprintln!("error={err}"); + ExitCode::from(2) + } + } +} diff --git a/crates/transport/src/directory.rs b/crates/transport/src/directory.rs new file mode 100644 index 0000000..e77ae21 --- /dev/null +++ b/crates/transport/src/directory.rs @@ -0,0 +1,561 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 + +//! Shard-local fill directory. +//! +//! One process is one shard. A single machine uses the same API with shard +//! id 0. The map is keyed by the SHA-256 of a physical byte key, one entry +//! per in-flight fill, not per token. Membership changes bump the epoch and +//! drop in-flight fills. There is no cross-shard quorum and no payload store. +//! Credits are an admission limit, not a replica set. + +use std::collections::{BTreeMap, BTreeSet}; + +use gpucache_core_types::{content_checksum, CacheError, QuotaKind, HARD_BYTES_CAP}; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct MemberView { + pub node_id: u64, + pub incarnation: u64, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ShardParams { + pub credit_bytes: u64, + pub max_bytes_per_tenant: u64, + pub max_flights: usize, + pub max_waiters: usize, + pub max_members: usize, +} + +impl ShardParams { + pub fn new( + credit_bytes: u64, + max_bytes_per_tenant: u64, + max_flights: usize, + max_waiters: usize, + max_members: usize, + ) -> Result { + if credit_bytes == 0 + || max_bytes_per_tenant == 0 + || max_bytes_per_tenant > credit_bytes + || max_flights == 0 + || max_waiters == 0 + || max_members == 0 + { + return Err(CacheError::Malformed { + reason: "shard config", + }); + } + Ok(Self { + credit_bytes, + max_bytes_per_tenant, + max_flights, + max_waiters, + max_members, + }) + } +} + +/// Caller-owned fill request. The directory keeps the tenant id and the +/// physical-key checksum, not the key bytes and not a KV payload. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct FillRequest { + pub physical_key: Vec, + pub tenant: Vec, + pub bytes: u64, + pub request: [u8; 16], + pub epoch: u64, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Admit { + Leader { flight_id: u64, epoch: u64 }, + Joined { flight_id: u64, epoch: u64 }, +} + +struct Flight { + id: u64, + epoch: u64, + bytes: u64, + tenant: Vec, + leader: [u8; 16], + waiters: BTreeSet<[u8; 16]>, +} + +pub struct ShardDirectory { + shard_id: u64, + epoch: u64, + members: Vec, + flights: BTreeMap<[u8; 32], Flight>, + credits_used: u64, + credit_bytes: u64, + tenant_used: BTreeMap, u64>, + max_bytes_per_tenant: u64, + max_flights: usize, + max_waiters: usize, + max_members: usize, + next_flight: u64, +} + +impl ShardDirectory { + /// Same methods as [`Self::open`]. Shard id is 0 and the only member is + /// node 1 at incarnation 1. + pub fn single_node(params: ShardParams) -> Self { + Self::open(0, params) + } + + pub fn open(shard_id: u64, params: ShardParams) -> Self { + Self { + shard_id, + epoch: 1, + members: vec![MemberView { + node_id: shard_id.wrapping_add(1), + incarnation: 1, + }], + flights: BTreeMap::new(), + credits_used: 0, + credit_bytes: params.credit_bytes, + tenant_used: BTreeMap::new(), + max_bytes_per_tenant: params.max_bytes_per_tenant, + max_flights: params.max_flights, + max_waiters: params.max_waiters, + max_members: params.max_members, + next_flight: 1, + } + } + + pub fn shard_id(&self) -> u64 { + self.shard_id + } + + pub fn epoch(&self) -> u64 { + self.epoch + } + + pub fn members(&self) -> &[MemberView] { + &self.members + } + + pub fn credits_used(&self) -> u64 { + self.credits_used + } + + pub fn inflight(&self) -> usize { + self.flights.len() + } + + pub fn check_epoch(&self, request_epoch: u64) -> Result<(), CacheError> { + if request_epoch != self.epoch { + return Err(CacheError::EpochMismatch { + store: self.epoch, + request: request_epoch, + }); + } + Ok(()) + } + + /// Record a membership change. An unchanged incarnation is a no-op. + /// Any real change bumps the epoch and fails in-flight fills. + pub fn note_member(&mut self, node_id: u64, incarnation: u64) -> Result { + if incarnation == 0 { + return Err(CacheError::Malformed { + reason: "membership", + }); + } + if let Some(pos) = self + .members + .iter() + .position(|member| member.node_id == node_id) + { + if self.members[pos].incarnation == incarnation { + return Ok(self.epoch); + } + self.members[pos].incarnation = incarnation; + return Ok(self.fence_inflight()); + } + if self.members.len() >= self.max_members { + return Err(CacheError::Malformed { + reason: "membership", + }); + } + self.members.push(MemberView { + node_id, + incarnation, + }); + Ok(self.fence_inflight()) + } + + pub fn remove_member(&mut self, node_id: u64) -> Result { + let Some(pos) = self + .members + .iter() + .position(|member| member.node_id == node_id) + else { + return Err(CacheError::Malformed { + reason: "membership", + }); + }; + if self.members.len() == 1 { + return Err(CacheError::FencingRejected { reason: "member" }); + } + self.members.remove(pos); + Ok(self.fence_inflight()) + } + + pub fn begin_or_join(&mut self, req: &FillRequest) -> Result { + self.check_epoch(req.epoch)?; + validate_fill(req)?; + let key = flight_key(&req.physical_key); + if let Some(existing) = self.key_for_request(req.request) { + if existing != key { + return Err(CacheError::IdempotencyConflict); + } + let flight = &self.flights[&existing]; + if flight.leader == req.request { + return Ok(Admit::Leader { + flight_id: flight.id, + epoch: flight.epoch, + }); + } + return Ok(Admit::Joined { + flight_id: flight.id, + epoch: flight.epoch, + }); + } + if let Some(flight) = self.flights.get_mut(&key) { + if flight.tenant.as_slice() != req.tenant.as_slice() { + return Err(CacheError::Unauthorized); + } + if flight.waiters.len() >= self.max_waiters { + return Err(CacheError::QuotaExceeded { + kind: QuotaKind::InFlightRequests, + }); + } + flight.waiters.insert(req.request); + return Ok(Admit::Joined { + flight_id: flight.id, + epoch: flight.epoch, + }); + } + if self.flights.len() >= self.max_flights || self.next_flight == u64::MAX { + return Err(CacheError::QuotaExceeded { + kind: QuotaKind::InFlightRequests, + }); + } + self.reserve(&req.tenant, req.bytes)?; + let id = self.next_flight; + self.next_flight += 1; + self.flights.insert( + key, + Flight { + id, + epoch: self.epoch, + bytes: req.bytes, + tenant: req.tenant.clone(), + leader: req.request, + waiters: BTreeSet::new(), + }, + ); + Ok(Admit::Leader { + flight_id: id, + epoch: self.epoch, + }) + } + + pub fn complete_flight( + &mut self, + flight_id: u64, + request_epoch: u64, + request: [u8; 16], + ) -> Result<(), CacheError> { + self.finish(flight_id, request_epoch, request) + } + + pub fn fail_flight( + &mut self, + flight_id: u64, + request_epoch: u64, + request: [u8; 16], + ) -> Result<(), CacheError> { + self.finish(flight_id, request_epoch, request) + } + + fn finish( + &mut self, + flight_id: u64, + request_epoch: u64, + request: [u8; 16], + ) -> Result<(), CacheError> { + self.check_epoch(request_epoch)?; + let Some(key) = self + .flights + .iter() + .find(|(_, flight)| flight.id == flight_id) + .map(|(key, _)| *key) + else { + return Err(CacheError::UnknownGeneration); + }; + let flight = &self.flights[&key]; + if flight.leader != request || flight.epoch != request_epoch { + return Err(CacheError::FencingRejected { reason: "epoch" }); + } + let tenant = flight.tenant.clone(); + let bytes = flight.bytes; + self.flights.remove(&key); + self.release(&tenant, bytes) + } + + fn fence_inflight(&mut self) -> u64 { + self.epoch = self.epoch.saturating_add(1); + let flights = std::mem::take(&mut self.flights); + for (_, flight) in flights { + let _ = self.release(&flight.tenant, flight.bytes); + } + self.epoch + } + + fn reserve(&mut self, tenant: &[u8], bytes: u64) -> Result<(), CacheError> { + let used = self.tenant_used.get(tenant).copied().unwrap_or(0); + let tenant_next = used.checked_add(bytes).ok_or(CacheError::QuotaExceeded { + kind: QuotaKind::TenantRetainedBytes, + })?; + if tenant_next > self.max_bytes_per_tenant { + return Err(CacheError::QuotaExceeded { + kind: QuotaKind::TenantRetainedBytes, + }); + } + let total_next = self + .credits_used + .checked_add(bytes) + .ok_or(CacheError::QuotaExceeded { + kind: QuotaKind::InFlightBytes, + })?; + if total_next > self.credit_bytes { + return Err(CacheError::QuotaExceeded { + kind: QuotaKind::InFlightBytes, + }); + } + self.tenant_used.insert(tenant.to_vec(), tenant_next); + self.credits_used = total_next; + Ok(()) + } + + fn release(&mut self, tenant: &[u8], bytes: u64) -> Result<(), CacheError> { + let used = self.tenant_used.get(tenant).copied().unwrap_or(0); + let Some(next) = used.checked_sub(bytes) else { + return Err(CacheError::Storage { reason: "credits" }); + }; + if next == 0 { + self.tenant_used.remove(tenant); + } else { + self.tenant_used.insert(tenant.to_vec(), next); + } + self.credits_used = self + .credits_used + .checked_sub(bytes) + .ok_or(CacheError::Storage { reason: "credits" })?; + Ok(()) + } + + fn key_for_request(&self, request: [u8; 16]) -> Option<[u8; 32]> { + self.flights.iter().find_map(|(key, flight)| { + if flight.leader == request || flight.waiters.contains(&request) { + Some(*key) + } else { + None + } + }) + } +} + +fn flight_key(physical_key: &[u8]) -> [u8; 32] { + content_checksum(physical_key).0 +} + +fn validate_fill(req: &FillRequest) -> Result<(), CacheError> { + if req.physical_key.is_empty() || req.tenant.is_empty() || req.bytes == 0 { + return Err(CacheError::Malformed { + reason: "shard fill", + }); + } + if req.physical_key.len() > HARD_BYTES_CAP { + return Err(CacheError::ObjectTooLarge { + limit: HARD_BYTES_CAP as u64, + actual: req.physical_key.len() as u64, + }); + } + if req.tenant.len() > 128 { + return Err(CacheError::Malformed { + reason: "shard fill", + }); + } + if req.request == [0u8; 16] { + return Err(CacheError::Malformed { + reason: "request_id must not be all zeros", + }); + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn params() -> ShardParams { + ShardParams::new(100, 80, 4, 2, 4).unwrap() + } + + fn fill(key: &[u8], tenant: &[u8], bytes: u64, request: u128, epoch: u64) -> FillRequest { + FillRequest { + physical_key: key.to_vec(), + tenant: tenant.to_vec(), + bytes, + request: request.to_le_bytes(), + epoch, + } + } + + #[test] + fn single_node_is_one_shard_and_join_charges_once() { + let mut dir = ShardDirectory::single_node(params()); + assert_eq!(dir.shard_id(), 0); + assert_eq!(dir.members().len(), 1); + assert_eq!(dir.epoch(), 1); + let leader = dir + .begin_or_join(&fill(b"phys-a", b"tenant-a", 40, 1, 1)) + .unwrap(); + let joined = dir + .begin_or_join(&fill(b"phys-a", b"tenant-a", 40, 2, 1)) + .unwrap(); + assert!(matches!(leader, Admit::Leader { epoch: 1, .. })); + assert!(matches!(joined, Admit::Joined { epoch: 1, .. })); + assert_eq!(dir.credits_used(), 40); + assert_eq!(dir.inflight(), 1); + let Admit::Leader { flight_id, .. } = leader else { + panic!("leader"); + }; + dir.complete_flight(flight_id, 1, 1u128.to_le_bytes()) + .unwrap(); + assert_eq!(dir.credits_used(), 0); + assert_eq!(dir.inflight(), 0); + } + + #[test] + fn two_shard_values_do_not_share_flights() { + let mut left = ShardDirectory::open(0, params()); + let right = ShardDirectory::open(1, params()); + left.begin_or_join(&fill(b"phys-a", b"tenant-a", 10, 1, 1)) + .unwrap(); + assert_eq!(left.inflight(), 1); + assert_eq!(right.inflight(), 0); + assert_eq!(right.shard_id(), 1); + assert_ne!(left.members()[0].node_id, right.members()[0].node_id); + } + + #[test] + fn stale_epoch_releases_credits_and_hides_the_fill() { + let mut dir = ShardDirectory::single_node(params()); + let admit = dir + .begin_or_join(&fill(b"phys-a", b"tenant-a", 40, 1, 1)) + .unwrap(); + let Admit::Leader { flight_id, epoch } = admit else { + panic!("leader"); + }; + let bumped = dir.note_member(1, 2).unwrap(); + assert_eq!(bumped, 2); + assert_eq!(dir.credits_used(), 0); + assert_eq!(dir.inflight(), 0); + let err = dir + .complete_flight(flight_id, epoch, 1u128.to_le_bytes()) + .unwrap_err(); + assert_eq!( + err, + CacheError::EpochMismatch { + store: 2, + request: 1, + } + ); + dir.begin_or_join(&fill(b"phys-a", b"tenant-a", 40, 3, 2)) + .unwrap(); + assert_eq!(dir.credits_used(), 40); + } + + #[test] + fn other_tenant_cannot_join_and_quota_does_not_drop_the_leader() { + let mut dir = ShardDirectory::single_node(params()); + dir.begin_or_join(&fill(b"phys-a", b"tenant-a", 40, 1, 1)) + .unwrap(); + let foreign = dir + .begin_or_join(&fill(b"phys-a", b"tenant-b", 40, 2, 1)) + .unwrap_err(); + assert_eq!(foreign, CacheError::Unauthorized); + let quota = dir + .begin_or_join(&fill(b"phys-b", b"tenant-a", 50, 3, 1)) + .unwrap_err(); + assert_eq!( + quota, + CacheError::QuotaExceeded { + kind: QuotaKind::TenantRetainedBytes, + } + ); + assert_eq!(dir.credits_used(), 40); + assert_eq!(dir.inflight(), 1); + } + + #[test] + fn repeated_leader_does_not_charge_twice_and_last_member_stays() { + let mut dir = ShardDirectory::single_node(params()); + dir.begin_or_join(&fill(b"phys-a", b"tenant-a", 10, 1, 1)) + .unwrap(); + dir.begin_or_join(&fill(b"phys-a", b"tenant-a", 10, 1, 1)) + .unwrap(); + assert_eq!(dir.credits_used(), 10); + let err = dir.remove_member(1).unwrap_err(); + assert_eq!(err, CacheError::FencingRejected { reason: "member" }); + assert_eq!(dir.epoch(), 1); + assert_eq!(dir.inflight(), 1); + } + + #[test] + fn joiner_cannot_finish_and_leader_fail_releases_credits() { + let mut dir = ShardDirectory::single_node(params()); + let leader = dir + .begin_or_join(&fill(b"phys-a", b"tenant-a", 10, 1, 1)) + .unwrap(); + dir.begin_or_join(&fill(b"phys-a", b"tenant-a", 10, 2, 1)) + .unwrap(); + let Admit::Leader { flight_id, .. } = leader else { + panic!("leader"); + }; + let err = dir + .complete_flight(flight_id, 1, 2u128.to_le_bytes()) + .unwrap_err(); + assert_eq!(err, CacheError::FencingRejected { reason: "epoch" }); + assert_eq!(dir.credits_used(), 10); + dir.fail_flight(flight_id, 1, 1u128.to_le_bytes()).unwrap(); + assert_eq!(dir.credits_used(), 0); + assert_eq!(dir.inflight(), 0); + let conflict = dir + .begin_or_join(&fill(b"phys-a", b"tenant-a", 10, 1, 1)) + .unwrap(); + let Admit::Leader { flight_id, .. } = conflict else { + panic!("leader"); + }; + let err = dir + .begin_or_join(&fill(b"phys-b", b"tenant-a", 10, 1, 1)) + .unwrap_err(); + assert_eq!(err, CacheError::IdempotencyConflict); + dir.fail_flight(flight_id, 1, 1u128.to_le_bytes()).unwrap(); + } + + #[test] + fn unchanged_incarnation_does_not_fence() { + let mut dir = ShardDirectory::single_node(params()); + dir.begin_or_join(&fill(b"phys-a", b"tenant-a", 10, 1, 1)) + .unwrap(); + assert_eq!(dir.note_member(1, 1).unwrap(), 1); + assert_eq!(dir.inflight(), 1); + assert_eq!(dir.credits_used(), 10); + } +} diff --git a/crates/transport/src/iface.rs b/crates/transport/src/iface.rs new file mode 100644 index 0000000..2cff1f0 --- /dev/null +++ b/crates/transport/src/iface.rs @@ -0,0 +1,208 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 + +use gpucache_core_types::{CacheError, QuotaKind, HARD_BYTES_CAP}; + +/// Which byte path a caller is on. `RdmaNixl` is a named slot. This process +/// does not post RDMA work requests. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum TransportKind { + Tcp, + RdmaNixl, +} + +/// Whether the receipt came from the CPU contract or from a device that was +/// not run. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum HardwarePath { + /// Userspace TCP staging. Not a NIC measurement. + CpuContract, + NotRunHardware, +} + +/// Bytes the caller already holds. Submitting them is a host copy on TCP. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct TransferDesc<'a> { + pub bytes: &'a [u8], +} + +/// What `submit` actually did. `staged_bytes` is the length copied into the +/// host staging buffer. It is not a wire-byte count and not a latency. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct TransferReceipt { + pub kind: TransportKind, + pub host_copies: u32, + pub registration_reused: bool, + pub dma_posted: bool, + pub hardware: HardwarePath, + pub staged_bytes: u64, +} + +pub trait KvTransport { + fn kind(&self) -> TransportKind; + fn submit(&mut self, desc: &TransferDesc<'_>) -> Result; +} + +/// Default correctness path. `submit` copies into a bounded host buffer and +/// drops it. The cache RPC path remains `TcpClient`. +#[derive(Debug)] +pub struct TcpTransport { + cap: usize, + staged_bytes: u64, + submits: u64, +} + +impl TcpTransport { + pub fn new(cap: usize) -> Result { + if cap == 0 { + return Err(CacheError::Malformed { reason: "transfer" }); + } + if cap > HARD_BYTES_CAP { + return Err(CacheError::ObjectTooLarge { + limit: HARD_BYTES_CAP as u64, + actual: cap as u64, + }); + } + Ok(Self { + cap, + staged_bytes: 0, + submits: 0, + }) + } + + pub fn submits(&self) -> u64 { + self.submits + } + + pub fn staged_bytes(&self) -> u64 { + self.staged_bytes + } +} + +impl KvTransport for TcpTransport { + fn kind(&self) -> TransportKind { + TransportKind::Tcp + } + + fn submit(&mut self, desc: &TransferDesc<'_>) -> Result { + if desc.bytes.is_empty() { + return Err(CacheError::Malformed { reason: "transfer" }); + } + if desc.bytes.len() > self.cap { + return Err(CacheError::QuotaExceeded { + kind: QuotaKind::InFlightBytes, + }); + } + let mut staged = Vec::new(); + staged.extend_from_slice(desc.bytes); + self.staged_bytes = self.staged_bytes.saturating_add(staged.len() as u64); + self.submits = self.submits.saturating_add(1); + drop(staged); + Ok(TransferReceipt { + kind: TransportKind::Tcp, + host_copies: 1, + registration_reused: false, + dma_posted: false, + hardware: HardwarePath::CpuContract, + staged_bytes: desc.bytes.len() as u64, + }) + } +} + +/// NIXL slot. Constructing this value does not open a device. +/// +/// `open` fails in every build, including `--features rdma-nixl`. libnixl is +/// not linked. See `docs/adr/0004-nixl-transport.md`. +#[derive(Debug)] +pub struct NixlTransport { + _unlinked: (), +} + +impl NixlTransport { + pub fn open() -> Result { + let _selected = rdma_feature_selected(); + Err(CacheError::Storage { + reason: "rdma not linked", + }) + } + + #[cfg(test)] + fn unlinked() -> Self { + Self { _unlinked: () } + } +} + +impl KvTransport for NixlTransport { + fn kind(&self) -> TransportKind { + TransportKind::RdmaNixl + } + + fn submit(&mut self, _desc: &TransferDesc<'_>) -> Result { + Err(CacheError::Storage { + reason: "rdma not linked", + }) + } +} + +/// `true` only when the crate is built with `--features rdma-nixl`. +/// The feature does not mean a device was opened. +pub fn rdma_feature_selected() -> bool { + cfg!(feature = "rdma-nixl") +} + +/// Stable label for logs and demos. Never a timing. +pub fn rdma_probe() -> &'static str { + "NOT_RUN_HARDWARE" +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn tcp_submit_is_one_host_copy() { + let mut tcp = TcpTransport::new(64).unwrap(); + let receipt = tcp.submit(&TransferDesc { bytes: b"kv-bytes" }).unwrap(); + assert_eq!(receipt.kind, TransportKind::Tcp); + assert_eq!(receipt.host_copies, 1); + assert!(!receipt.dma_posted); + assert!(!receipt.registration_reused); + assert_eq!(receipt.hardware, HardwarePath::CpuContract); + assert_eq!(receipt.staged_bytes, 8); + assert_eq!(tcp.submits(), 1); + assert_eq!(tcp.kind(), TransportKind::Tcp); + } + + #[test] + fn tcp_submit_backpressures_past_the_cap() { + let mut tcp = TcpTransport::new(4).unwrap(); + let err = tcp.submit(&TransferDesc { bytes: b"12345" }).unwrap_err(); + assert_eq!( + err, + CacheError::QuotaExceeded { + kind: QuotaKind::InFlightBytes, + } + ); + assert_eq!(tcp.submits(), 0); + assert_eq!(tcp.staged_bytes(), 0); + } + + #[test] + fn rdma_open_does_not_post() { + assert_eq!(rdma_probe(), "NOT_RUN_HARDWARE"); + assert!(matches!( + NixlTransport::open(), + Err(CacheError::Storage { + reason: "rdma not linked" + }) + )); + let mut nixl = NixlTransport::unlinked(); + assert_eq!(nixl.kind(), TransportKind::RdmaNixl); + assert!(matches!( + nixl.submit(&TransferDesc { bytes: b"x" }), + Err(CacheError::Storage { + reason: "rdma not linked" + }) + )); + } +} diff --git a/crates/transport/src/lib.rs b/crates/transport/src/lib.rs index a7c1ce3..104dc50 100644 --- a/crates/transport/src/lib.rs +++ b/crates/transport/src/lib.rs @@ -1,22 +1,36 @@ // Copyright 2024-2026 RustFS, Inc. // SPDX-License-Identifier: Apache-2.0 -//! TCP service for the V1 B shared KV tier. +//! TCP service for the shared KV tier, plus the V2 control-plane types. //! //! The listener defaults to loopback and rejects a start with no tenant policy. //! Frames are length-delimited. The GCK1 request bytes are unchanged. Replies -//! use GCKR. There is no RDMA, CUDA, or DPU dependency in this crate. +//! use GCKR. Feature `rdma-nixl` does not link libnixl. Live RDMA is +//! `NOT_RUN_HARDWARE`. #![forbid(unsafe_code)] mod client; mod clock; mod demo; +mod directory; mod fault_demo; mod frame; +mod iface; mod pool; +mod registered; mod server; +mod shard_demo; +mod stage; pub use client::TcpClient; pub use demo::{run_share_demo, ShareReport}; +pub use directory::{Admit, FillRequest, MemberView, ShardDirectory, ShardParams}; pub use fault_demo::{run_fault_demo, FaultReport}; +pub use iface::{ + rdma_feature_selected, rdma_probe, HardwarePath, KvTransport, NixlTransport, TcpTransport, + TransferDesc, TransferReceipt, TransportKind, +}; +pub use registered::{CheckedBatch, PoolStats, RegisteredPool, SimCompletion, SlotId}; pub use server::{serve, ServerConfig, ServerHandle}; +pub use shard_demo::{run_shard_demo, ShardReport}; +pub use stage::stage_through_tcp; diff --git a/crates/transport/src/registered.rs b/crates/transport/src/registered.rs new file mode 100644 index 0000000..a16646f --- /dev/null +++ b/crates/transport/src/registered.rs @@ -0,0 +1,421 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 + +//! Bounded buffer pool that *simulates* pin/registration. +//! +//! Slots are ordinary `Vec` allocations. Nothing here calls `ibv_reg_mr` +//! or a NIC. Unknown completion quarantines a slot. Lease expiry does not +//! free it. A new region id is assigned only when an allocation is bound +//! again; retired ids are not recycled. + +use gpucache_core_types::{CacheError, QuotaKind}; + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] +pub struct SlotId(u64); + +impl SlotId { + pub const fn get(self) -> u64 { + self.0 + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum SimCompletion { + Completed, + Unknown, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct CheckedBatch { + id: u64, + slots: Vec, +} + +impl CheckedBatch { + pub fn id(&self) -> u64 { + self.id + } + + pub fn slots(&self) -> &[SlotId] { + &self.slots + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct PoolStats { + pub idle_unbound: usize, + pub idle_registered: usize, + pub checked_out: usize, + pub quarantined: usize, + pub registration_ops: u64, + pub reused_registrations: u64, + pub backing_reused: u64, + pub epoch: u64, + pub held_bytes: u64, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum RegionState { + Unbound, + Idle, + CheckedOut { batch: u64 }, + Quarantined { fenced: bool }, +} + +struct Region { + id: Option, + bytes: Vec, + state: RegionState, + had_registration: bool, +} + +/// Fixed set of host buffers. Checkout fails when every slot is busy. +pub struct RegisteredPool { + slot_bytes: usize, + regions: Vec, + next_id: u64, + next_batch: u64, + registration_ops: u64, + reused_registrations: u64, + backing_reused: u64, + epoch: u64, +} + +impl RegisteredPool { + pub fn new(capacity: usize, slot_bytes: usize) -> Result { + if capacity == 0 || capacity > 4096 || slot_bytes == 0 { + return Err(CacheError::Malformed { + reason: "buffer pool", + }); + } + if slot_bytes > 16 * 1024 * 1024 { + return Err(CacheError::ObjectTooLarge { + limit: 16 * 1024 * 1024, + actual: slot_bytes as u64, + }); + } + let mut regions = Vec::with_capacity(capacity); + for _ in 0..capacity { + regions.push(Region { + id: None, + bytes: vec![0u8; slot_bytes], + state: RegionState::Unbound, + had_registration: false, + }); + } + Ok(Self { + slot_bytes, + regions, + next_id: 1, + next_batch: 1, + registration_ops: 0, + reused_registrations: 0, + backing_reused: 0, + epoch: 1, + }) + } + + pub fn stats(&self) -> PoolStats { + let mut idle_unbound = 0; + let mut idle_registered = 0; + let mut checked_out = 0; + let mut quarantined = 0; + for region in &self.regions { + match region.state { + RegionState::Unbound => idle_unbound += 1, + RegionState::Idle => idle_registered += 1, + RegionState::CheckedOut { .. } => checked_out += 1, + RegionState::Quarantined { .. } => quarantined += 1, + } + } + PoolStats { + idle_unbound, + idle_registered, + checked_out, + quarantined, + registration_ops: self.registration_ops, + reused_registrations: self.reused_registrations, + backing_reused: self.backing_reused, + epoch: self.epoch, + held_bytes: (self.regions.len() * self.slot_bytes) as u64, + } + } + + /// Copy `payload` into a checked-out slot and zero the tail. The slot is + /// a host `Vec`, not a NIC registration. + pub fn copy_in(&mut self, id: SlotId, payload: &[u8]) -> Result<(), CacheError> { + let slot_bytes = self.slot_bytes; + let region = self.region_mut(id)?; + if !matches!(region.state, RegionState::CheckedOut { .. }) { + return Err(CacheError::NotReclaimable { + from: gpucache_core_types::BlockState::Absent, + }); + } + if payload.len() > slot_bytes { + return Err(CacheError::ObjectTooLarge { + limit: slot_bytes as u64, + actual: payload.len() as u64, + }); + } + region.bytes[..payload.len()].copy_from_slice(payload); + region.bytes[payload.len()..].fill(0); + let copied = ®ion.bytes[..payload.len()] == payload; + let tail_clear = region.bytes[payload.len()..].iter().all(|byte| *byte == 0); + if !copied || !tail_clear { + return Err(CacheError::ChecksumMismatch); + } + Ok(()) + } + + /// Take `n` idle slots. Already-bound slots do not count as a new + /// registration. The call changes nothing when it returns an error. + pub fn checkout_batch(&mut self, n: usize) -> Result { + if n == 0 || n > self.regions.len() { + return Err(CacheError::Malformed { + reason: "buffer pool", + }); + } + let available = self + .regions + .iter() + .filter(|region| matches!(region.state, RegionState::Unbound | RegionState::Idle)) + .count(); + if available < n { + return Err(CacheError::QuotaExceeded { + kind: QuotaKind::Registrations, + }); + } + if self.next_batch == u64::MAX { + return Err(CacheError::QuotaExceeded { + kind: QuotaKind::Registrations, + }); + } + let batch = self.next_batch; + self.next_batch += 1; + let mut slots = Vec::with_capacity(n); + let mut index = 0; + while slots.len() < n && index < self.regions.len() { + let state = self.regions[index].state; + match state { + RegionState::Idle => { + self.reused_registrations = self.reused_registrations.saturating_add(1); + let id = self.regions[index].id.ok_or(CacheError::Storage { + reason: "buffer pool", + })?; + self.regions[index].state = RegionState::CheckedOut { batch }; + slots.push(id); + } + RegionState::Unbound => { + if self.regions[index].had_registration { + self.backing_reused = self.backing_reused.saturating_add(1); + } + let id = self.alloc_id()?; + self.regions[index].id = Some(id); + self.regions[index].had_registration = true; + self.regions[index].state = RegionState::CheckedOut { batch }; + self.registration_ops = self.registration_ops.saturating_add(1); + slots.push(id); + } + RegionState::CheckedOut { .. } | RegionState::Quarantined { .. } => {} + } + index += 1; + } + if slots.len() != n { + return Err(CacheError::Storage { + reason: "buffer pool", + }); + } + Ok(CheckedBatch { id: batch, slots }) + } + + pub fn complete_batch( + &mut self, + batch: &CheckedBatch, + completion: SimCompletion, + ) -> Result<(), CacheError> { + self.ensure_batch(batch)?; + for region in &mut self.regions { + let Some(id) = region.id else { + continue; + }; + if batch.slots.contains(&id) { + region.state = match completion { + SimCompletion::Completed => RegionState::Idle, + SimCompletion::Unknown => RegionState::Quarantined { fenced: false }, + }; + } + } + Ok(()) + } + + /// Lease expiry is not proof that a copy has stopped. + pub fn on_lease_expiry(&mut self, id: SlotId) -> Result<(), CacheError> { + let region = self.region_mut(id)?; + if let RegionState::CheckedOut { .. } = region.state { + region.state = RegionState::Quarantined { fenced: false }; + } + Ok(()) + } + + /// Caller attests the copy is done. Only a quarantined slot returns to + /// the unbound pool, under a future id. + pub fn attest_drain(&mut self, id: SlotId) -> Result<(), CacheError> { + let region = self.region_mut(id)?; + if !matches!(region.state, RegionState::Quarantined { .. }) { + return Err(CacheError::NotReclaimable { + from: gpucache_core_types::BlockState::Writing, + }); + } + region.id = None; + region.state = RegionState::Unbound; + Ok(()) + } + + /// Move in-flight and already-quarantined slots to fenced quarantine. + /// Does not free them. + pub fn fence_epoch(&mut self) -> u64 { + self.epoch = self.epoch.saturating_add(1); + for region in &mut self.regions { + if matches!( + region.state, + RegionState::CheckedOut { .. } | RegionState::Quarantined { .. } + ) { + region.state = RegionState::Quarantined { fenced: true }; + } + } + self.epoch + } + + /// Free slots that fencing already marked. Unfenced quarantine stays. + pub fn reclaim_fenced(&mut self) -> Result { + let mut freed = 0; + for region in &mut self.regions { + if matches!(region.state, RegionState::Quarantined { fenced: true }) { + region.id = None; + region.state = RegionState::Unbound; + freed += 1; + } + } + Ok(freed) + } + + fn alloc_id(&mut self) -> Result { + if self.next_id == u64::MAX { + return Err(CacheError::QuotaExceeded { + kind: QuotaKind::Registrations, + }); + } + let id = SlotId(self.next_id); + self.next_id += 1; + Ok(id) + } + + fn ensure_batch(&self, batch: &CheckedBatch) -> Result<(), CacheError> { + if batch.slots.is_empty() { + return Err(CacheError::Malformed { + reason: "buffer pool", + }); + } + for id in &batch.slots { + let region = self.region(*id)?; + match region.state { + RegionState::CheckedOut { batch: got } if got == batch.id => {} + _ => return Err(CacheError::FencingRejected { reason: "buffer" }), + } + } + Ok(()) + } + + fn region(&self, id: SlotId) -> Result<&Region, CacheError> { + self.regions + .iter() + .find(|region| region.id == Some(id)) + .ok_or(CacheError::UnknownGeneration) + } + + fn region_mut(&mut self, id: SlotId) -> Result<&mut Region, CacheError> { + self.regions + .iter_mut() + .find(|region| region.id == Some(id)) + .ok_or(CacheError::UnknownGeneration) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn registration_is_reused_until_the_slot_is_retired() { + let mut pool = RegisteredPool::new(1, 16).unwrap(); + let first = pool.checkout_batch(1).unwrap(); + assert_eq!(pool.stats().registration_ops, 1); + pool.complete_batch(&first, SimCompletion::Completed) + .unwrap(); + let second = pool.checkout_batch(1).unwrap(); + let stats = pool.stats(); + assert_eq!(stats.registration_ops, 1); + assert_eq!(stats.reused_registrations, 1); + assert_eq!(stats.backing_reused, 0); + assert_eq!(first.slots()[0], second.slots()[0]); + } + + #[test] + fn checkout_backpressures_without_growing() { + let mut pool = RegisteredPool::new(1, 8).unwrap(); + let held = pool.checkout_batch(1).unwrap(); + let err = pool.checkout_batch(1).unwrap_err(); + assert_eq!( + err, + CacheError::QuotaExceeded { + kind: QuotaKind::Registrations, + } + ); + assert_eq!(pool.stats().held_bytes, 8); + assert_eq!(pool.stats().checked_out, 1); + drop(held); + } + + #[test] + fn unknown_completion_and_lease_expiry_do_not_free() { + let mut pool = RegisteredPool::new(1, 8).unwrap(); + let batch = pool.checkout_batch(1).unwrap(); + let id = batch.slots()[0]; + pool.complete_batch(&batch, SimCompletion::Unknown).unwrap(); + pool.on_lease_expiry(id).unwrap(); + let stats = pool.stats(); + assert_eq!(stats.quarantined, 1); + assert_eq!(stats.idle_unbound, 0); + assert_eq!(pool.reclaim_fenced().unwrap(), 0); + assert_eq!(pool.stats().quarantined, 1); + pool.attest_drain(id).unwrap(); + assert_eq!(pool.stats().idle_unbound, 1); + let again = pool.checkout_batch(1).unwrap(); + assert_ne!(again.slots()[0], id); + assert_eq!(pool.stats().registration_ops, 2); + assert_eq!(pool.stats().backing_reused, 1); + } + + #[test] + fn fence_then_reclaim_frees_only_fenced_slots() { + let mut pool = RegisteredPool::new(2, 4).unwrap(); + let batch = pool.checkout_batch(1).unwrap(); + let id = batch.slots()[0]; + pool.on_lease_expiry(id).unwrap(); + assert_eq!(pool.reclaim_fenced().unwrap(), 0); + pool.fence_epoch(); + assert_eq!(pool.stats().quarantined, 1); + assert_eq!(pool.reclaim_fenced().unwrap(), 1); + assert_eq!(pool.stats().idle_unbound, 2); + assert_eq!(pool.stats().held_bytes, 8); + } + + #[test] + fn failed_checkout_leaves_the_pool_unchanged() { + let mut pool = RegisteredPool::new(2, 4).unwrap(); + assert!(pool.checkout_batch(3).is_err()); + let stats = pool.stats(); + assert_eq!(stats.idle_unbound, 2); + assert_eq!(stats.registration_ops, 0); + } +} diff --git a/crates/transport/src/shard_demo.rs b/crates/transport/src/shard_demo.rs new file mode 100644 index 0000000..bcacd40 --- /dev/null +++ b/crates/transport/src/shard_demo.rs @@ -0,0 +1,122 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 + +//! Single-shard fencing, simulated registration, and the TCP staging path. +//! +//! One process is shard 0. RDMA is not opened. No timing is printed. + +use core::fmt; + +use gpucache_core_types::CacheError; + +use crate::directory::{Admit, FillRequest, ShardDirectory, ShardParams}; +use crate::iface::{rdma_probe, NixlTransport, TcpTransport}; +use crate::registered::{RegisteredPool, SimCompletion}; +use crate::stage::stage_through_tcp; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ShardReport; + +impl fmt::Display for ShardReport { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + writeln!(f, "status=SHARD_OK")?; + writeln!(f, "topology=SINGLE_SHARD")?; + writeln!(f, "flight=JOINED")?; + writeln!(f, "credits=SINGLE_CHARGE")?; + writeln!(f, "stale_epoch=FENCED")?; + writeln!(f, "unknown_completion=QUARANTINED")?; + writeln!(f, "lease_expiry=HELD")?; + writeln!(f, "tcp_path=HOST_COPY")?; + writeln!(f, "dma_posted=NO")?; + writeln!(f, "rdma={rdma}", rdma = rdma_probe())?; + writeln!(f, "registration=SIMULATED")?; + Ok(()) + } +} + +pub fn run_shard_demo() -> Result { + let params = ShardParams::new(1024, 1024, 8, 4, 4)?; + let mut dir = ShardDirectory::single_node(params); + if dir.shard_id() != 0 || dir.members().len() != 1 { + return Err(CacheError::Malformed { + reason: "shard config", + }); + } + let leader_req = FillRequest { + physical_key: b"phys-demo".to_vec(), + tenant: b"tenant-a".to_vec(), + bytes: 32, + request: 1u128.to_le_bytes(), + epoch: dir.epoch(), + }; + let joined_req = FillRequest { + request: 2u128.to_le_bytes(), + ..leader_req.clone() + }; + let leader = dir.begin_or_join(&leader_req)?; + let joined = dir.begin_or_join(&joined_req)?; + if !matches!(leader, Admit::Leader { .. }) || !matches!(joined, Admit::Joined { .. }) { + return Err(CacheError::Malformed { + reason: "shard fill", + }); + } + if dir.credits_used() != 32 || dir.inflight() != 1 { + return Err(CacheError::Malformed { reason: "credits" }); + } + let Admit::Leader { flight_id, epoch } = leader else { + return Err(CacheError::Malformed { + reason: "shard fill", + }); + }; + let bumped = dir.note_member(dir.members()[0].node_id, 2)?; + if bumped <= epoch || dir.inflight() != 0 || dir.credits_used() != 0 { + return Err(CacheError::FencingRejected { reason: "epoch" }); + } + if dir + .complete_flight(flight_id, epoch, leader_req.request) + .is_ok() + { + return Err(CacheError::FencingRejected { reason: "epoch" }); + } + + let mut pool = RegisteredPool::new(1, 16)?; + let batch = pool.checkout_batch(1)?; + let id = batch.slots()[0]; + pool.complete_batch(&batch, SimCompletion::Unknown)?; + pool.on_lease_expiry(id)?; + if pool.stats().quarantined != 1 || pool.stats().idle_unbound != 0 { + return Err(CacheError::NotReclaimable { + from: gpucache_core_types::BlockState::Quarantined, + }); + } + + let mut tcp = TcpTransport::new(64)?; + let mut stage_pool = RegisteredPool::new(1, 16)?; + let receipt = stage_through_tcp(&mut stage_pool, &mut tcp, &[b"stage"])?; + if receipt.dma_posted || receipt.host_copies != 1 { + return Err(CacheError::Malformed { reason: "transfer" }); + } + if NixlTransport::open().is_ok() || rdma_probe() != "NOT_RUN_HARDWARE" { + return Err(CacheError::Storage { + reason: "rdma not linked", + }); + } + Ok(ShardReport) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn demo_report_has_no_device_claim() { + let text = run_shard_demo().unwrap().to_string(); + assert!(text.contains("status=SHARD_OK")); + assert!(text.contains("topology=SINGLE_SHARD")); + assert!(text.contains("rdma=NOT_RUN_HARDWARE")); + assert!(text.contains("registration=SIMULATED")); + assert!(text.contains("dma_posted=NO")); + assert!(!text.contains("GB/s")); + assert!(!text.contains("TTFT")); + } +} diff --git a/crates/transport/src/stage.rs b/crates/transport/src/stage.rs new file mode 100644 index 0000000..f0558f3 --- /dev/null +++ b/crates/transport/src/stage.rs @@ -0,0 +1,80 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 + +use gpucache_core_types::CacheError; + +use crate::iface::{KvTransport, TransferDesc, TransferReceipt}; +use crate::registered::{RegisteredPool, SimCompletion}; + +/// Checkout a simulated registration batch, copy the payloads through TCP +/// staging, then complete or quarantine the batch. +/// +/// A failed submit quarantines the slots (`SimCompletion::Unknown`). The +/// receipt's `registration_reused` flag reports whether this checkout reused +/// an existing binding. `dma_posted` stays false. +pub fn stage_through_tcp( + pool: &mut RegisteredPool, + tcp: &mut impl KvTransport, + payloads: &[&[u8]], +) -> Result { + if payloads.is_empty() || payloads.len() > 64 { + return Err(CacheError::Malformed { reason: "transfer" }); + } + if payloads.iter().any(|payload| payload.is_empty()) { + return Err(CacheError::Malformed { reason: "transfer" }); + } + let before = pool.stats().reused_registrations; + let batch = pool.checkout_batch(payloads.len())?; + for (id, payload) in batch.slots().iter().copied().zip(payloads.iter().copied()) { + if let Err(err) = pool.copy_in(id, payload) { + let _ = pool.complete_batch(&batch, SimCompletion::Unknown); + return Err(err); + } + } + let mut joined = Vec::new(); + for payload in payloads { + joined.extend_from_slice(payload); + } + let submitted = tcp.submit(&TransferDesc { bytes: &joined }); + let completion = if submitted.is_ok() { + SimCompletion::Completed + } else { + SimCompletion::Unknown + }; + pool.complete_batch(&batch, completion)?; + let mut receipt = submitted?; + receipt.registration_reused = pool.stats().reused_registrations > before; + Ok(receipt) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::iface::{HardwarePath, TcpTransport, TransportKind}; + + #[test] + fn stage_reuses_registration_on_the_second_batch() { + let mut pool = RegisteredPool::new(1, 32).unwrap(); + let mut tcp = TcpTransport::new(64).unwrap(); + let first = stage_through_tcp(&mut pool, &mut tcp, &[b"abcd"]).unwrap(); + assert!(!first.registration_reused); + assert!(!first.dma_posted); + assert_eq!(first.kind, TransportKind::Tcp); + assert_eq!(first.hardware, HardwarePath::CpuContract); + let second = stage_through_tcp(&mut pool, &mut tcp, &[b"efgh"]).unwrap(); + assert!(second.registration_reused); + assert_eq!(pool.stats().quarantined, 0); + assert_eq!(tcp.submits(), 2); + } + + #[test] + fn oversized_stage_quarantines_instead_of_freeing() { + let mut pool = RegisteredPool::new(1, 8).unwrap(); + let mut tcp = TcpTransport::new(2).unwrap(); + let err = stage_through_tcp(&mut pool, &mut tcp, &[b"abcd"]).unwrap_err(); + assert!(matches!(err, CacheError::QuotaExceeded { .. })); + assert_eq!(pool.stats().quarantined, 1); + assert_eq!(tcp.submits(), 0); + assert!(pool.checkout_batch(1).is_err()); + } +} diff --git a/crates/transport/tests/v2_control.rs b/crates/transport/tests/v2_control.rs new file mode 100644 index 0000000..b76b648 --- /dev/null +++ b/crates/transport/tests/v2_control.rs @@ -0,0 +1,56 @@ +// Copyright 2024-2026 RustFS, Inc. +// SPDX-License-Identifier: Apache-2.0 + +//! Integration coverage for the V2 control plane. No GPU, NIC, or RustFS. + +use gpucache_core_types::{CacheError, QuotaKind}; +use gpucache_transport::{ + rdma_probe, stage_through_tcp, Admit, FillRequest, NixlTransport, RegisteredPool, + ShardDirectory, ShardParams, TcpTransport, TransportKind, +}; + +#[test] +fn directory_pool_and_tcp_compose_without_a_device() { + let mut dir = ShardDirectory::single_node(ShardParams::new(64, 64, 4, 4, 2).unwrap()); + let epoch = dir.epoch(); + let leader = dir + .begin_or_join(&FillRequest { + physical_key: b"block-physical".to_vec(), + tenant: b"tenant-a".to_vec(), + bytes: 16, + request: 7u128.to_le_bytes(), + epoch, + }) + .unwrap(); + assert!(matches!(leader, Admit::Leader { .. })); + let mut pool = RegisteredPool::new(2, 8).unwrap(); + let mut tcp = TcpTransport::new(32).unwrap(); + let receipt = stage_through_tcp(&mut pool, &mut tcp, &[b"aa", b"bb"]).unwrap(); + assert_eq!(receipt.kind, TransportKind::Tcp); + assert!(!receipt.dma_posted); + assert_eq!(pool.stats().checked_out, 0); + assert_eq!(rdma_probe(), "NOT_RUN_HARDWARE"); + assert!(matches!( + NixlTransport::open(), + Err(CacheError::Storage { + reason: "rdma not linked" + }) + )); +} + +#[test] +fn feature_slot_is_independent_of_a_linked_library() { + let selected = gpucache_transport::rdma_feature_selected(); + if cfg!(feature = "rdma-nixl") { + assert!(selected); + } else { + assert!(!selected); + } + let err = NixlTransport::open().unwrap_err(); + assert_ne!( + err, + CacheError::QuotaExceeded { + kind: QuotaKind::Registrations, + } + ); +} diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 4721511..4136056 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -1,4 +1,4 @@ -# Architecture (V1 C) +# Architecture (V1 C, V2 first slice) GPUCache is a shared KV data tier. The hot tier is G3.5. RustFS is an optional colder G4 capacity tier. It is not on the prefill-to-decode path. The inference engine still owns tokenization, attention, and scheduling. This repo does not replace them. @@ -18,14 +18,20 @@ gpucache-storage::MemoryStore v gpucache-core-types (identity, keys, lifecycle, errors) -gpucache-cold optional G4 pack/manifest. Not linked by transport. +gpucache-cold optional G4 pack/manifest, async sink, budgeted refill. + Not linked by transport. + +V2 control plane (same process, not a new wire protocol) + KvTransport TCP submit is a host copy. NIXL open refuses. + RegisteredPool simulated pin, reuse, quarantine. Not a NIC. + ShardDirectory one shard per process. Epoch fence, single-flight, credits. ``` `MemoryStore` remains the in-process contract. `FileStore` is what the TCP service uses. `gpucache-bench` only names the paired baselines and prints `NOT_RUN_HARDWARE`. Planned crates still absent: a standalone metadata service, an engine adapter, a DPU agent, and metrics. Their contracts stay out of `core-types` so GPU and DPU types do not leak in. ## Durability -The hot tier does not repair lost blocks. There is no erasure coding and no replica set. Loss, corruption, a short write, a full disk, eviction, and a failed check are clean misses. The engine recomputes and may write the block again. If the directory write fails after the in-memory commit, that commit is abandoned and the request id is forgotten, so a replay cannot report success for bytes that were not stored. An extra copy of a hot shared prefix is a possible later latency optimization, not a durability feature, and it is not in this tree. Cold durability is RustFS's responsibility. See [adr/0002-no-erasure-coding.md](adr/0002-no-erasure-coding.md). +The hot tier does not repair lost blocks. There is no erasure coding and no replica set. Loss, corruption, a short write, a full disk, eviction, and a failed check are clean misses. The engine recomputes and may write the block again. If the on-disk write fails after the in-memory commit, that commit is abandoned and the request id is forgotten, so a replay cannot report success for bytes that were not stored. An extra copy of a hot shared prefix is a possible later latency optimization, not a durability feature, and it is not in this tree. Cold durability is RustFS's responsibility. See [adr/0002-no-erasure-coding.md](adr/0002-no-erasure-coding.md). ## Identity @@ -65,3 +71,13 @@ ABSENT -> RESERVED -> WRITING -> COMMITTED -> EVICTING -> ABSENT ## Cold tier `gpucache-cold` packs several blocks into one object and a manifest of payload offsets. Put, Head, RangeGet, and Delete are a trait. The mock implements them in memory. A live HTTP client runs only when the operator sets `GPUCACHE_RUSTFS_*`. Transport does not depend on this crate. + +## V2 control plane + +`KvTransport` is the byte-path trait. `TcpTransport::submit` copies into a bounded host buffer and reports `dma_posted=false`. `NixlTransport::open` fails closed. Feature `rdma-nixl` does not link libnixl. See [adr/0004-nixl-transport.md](adr/0004-nixl-transport.md) and [V2_DATA_PATHS.md](V2_DATA_PATHS.md). + +`RegisteredPool` is a fixed set of `Vec` slots. Checkout reuses a binding instead of allocating a new region id. Unknown completion quarantines the slot. Lease expiry does not free it. `fence_epoch` then `reclaim_fenced` does, and so does `attest_drain`. Retired ids are not recycled. + +`ShardDirectory` is local to one shard. Shard 0 is the single-node case and uses the same methods. A membership change bumps the epoch, drops in-flight fills, and releases credits. Single-flight joins do not charge twice and do not return payload bytes. There is no per-token central directory and no per-block quorum. See [adr/0005-shard-fencing.md](adr/0005-shard-fencing.md). + +The cold sink queues a pack and drains it with Put. Refill Heads the manifest first. Over budget returns `OverBudget` and no payload. That outcome is a recompute, not a hit. diff --git a/docs/BENCHMARKS.md b/docs/BENCHMARKS.md index 9794f86..e74af99 100644 --- a/docs/BENCHMARKS.md +++ b/docs/BENCHMARKS.md @@ -16,7 +16,12 @@ Paired baselines use the same prefix-cache idea on both sides when a GPU run exi | HBM within capacity | `NOT_RUN_HARDWARE` | Not run | | HBM cliff | `NOT_RUN_HARDWARE` | Not run | | Write cost | `NOT_RUN_HARDWARE` | Not recorded. A memcpy figure would not count | +| V2 native NIXL P/D | `NOT_RUN_HARDWARE` | libnixl is not linked. `NixlTransport::open` refuses. Not a DPU result | +| V2 Rust shared DRAM | `NOT_RUN_HARDWARE` | The simulated pool is a correctness stand-in, not a bandwidth run | +| V2 Rust shared NVMe | `NOT_RUN_HARDWARE` | `FileStore` uses ordinary files. No device was formatted | +| V2 RustFS refill | `NOT_RUN_HARDWARE` | Mock refill passes. Live endpoint was not configured | +| V2 TTFT / TPOT / ITL / goodput | `NOT_RUN_HARDWARE` | No paired GPU trace. `staged_bytes` is a copy length, not wire bytes | -There is no `FAIL_PERFORMANCE` figure because there is no paired measurement to fail. Do not add a benchmark that prints a success banner without raw samples, a manifest, and the hardware it ran on. +There is no `FAIL_PERFORMANCE` figure because there is no paired measurement to fail. Do not add a benchmark that prints a success banner without raw samples, a manifest, and the hardware it ran on. The V2 CPU tests do not fill this gap. Correctness criteria for the CPU path are in [CORRECTNESS.md](CORRECTNESS.md). diff --git a/docs/OPERATIONS.md b/docs/OPERATIONS.md index 11883dc..dc5b19a 100644 --- a/docs/OPERATIONS.md +++ b/docs/OPERATIONS.md @@ -36,6 +36,21 @@ cargo run -p gpucache-transport --bin fault-demo The command uses a fresh temp directory. It commits a synthetic token block, reads it back from a second client, and checks the token ids exactly. An unknown key, a corrupt file after restart, a past deadline, a cancel, and a quota refusal all fail closed. The caller recomputes with the same pure function of those token ids. `status=FALLBACK_OK` is printed only after those checks. `tolerance=NONE` means no numeric slack was applied. A wrong tenant, model, tokenizer, adapter, or layout is not a hit. +## V2 shard and cold demos + +```bash +cargo run -p gpucache-transport --bin shard-demo +cargo run -p gpucache-cold --bin cold-demo +``` + +`shard-demo` checks one shard, a joined fill, epoch fencing, quarantine across lease expiry, and a TCP host copy. It prints `rdma=NOT_RUN_HARDWARE`. `cold-demo` runs the mock pack, sink, and refill. It prints `gds=NOT_RUN_HARDWARE`. Without `GPUCACHE_RUSTFS_*` it also prints `rustfs_live=NOT_RUN_HARDWARE`. + +Optional feature check, still with no device: + +```bash +cargo test -p gpucache-transport --all-targets --features rdma-nixl +``` + The paired-baseline stub does not measure anything: ```bash diff --git a/docs/REUSE_AUDIT.md b/docs/REUSE_AUDIT.md index a78ae58..74b6c20 100644 --- a/docs/REUSE_AUDIT.md +++ b/docs/REUSE_AUDIT.md @@ -1,13 +1,13 @@ -# Reuse audit (V1 A) +# Reuse audit -The execution spec asks for a look at LMCache, Mooncake, Dynamo KVBM, NIXL, OpenLake, ContextStore, and Fluxon before adding dependencies. V1 A adds none of them. +The execution spec asks for a look at LMCache, Mooncake, Dynamo KVBM, NIXL, OpenLake, ContextStore, and Fluxon before adding dependencies. None of them is a crate dependency. The V2 slice names NIXL as the RDMA slot and still does not link it. See [adr/0004-nixl-transport.md](adr/0004-nixl-transport.md). -| Candidate | Why it is not in the tree | +| Candidate | Why it is not linked | | --- | --- | | LMCache | Plausible vLLM bridge. Route not selected; GPU spike `NOT_RUN_HARDWARE`. | -| Mooncake | RDMA / KV transport. Out of scope until V2, and only after V1 correctness. | -| Dynamo KVBM | Another KV manager. Not required to define the CPU contract. | -| NIXL | Candidate V2 transport. No Rust binding is vendored. Ordinary NIC RDMA is not a DPU result. | -| OpenLake, ContextStore, Fluxon | Not required for identity, protocol, or `MemoryStore`. No version is pinned because no code is imported. | +| Mooncake | Apache-2.0. Rust Transfer Engine is in-tree bindgen FFI (`unsafe`) and needs the C++ library at build time. Not a crates.io pin. Recorded as the fallback, not this slice's dependency. | +| Dynamo KVBM | Another KV manager. Not required for the CPU contract or the shard directory. | +| NIXL | Apache-2.0. `nixl-sys` 1.3.1 (crates.io, 2026-07-07; docs.rs also shows 1.5.0) is the chosen slot. `stub-api` aborts if called. Real linkage needs `libnixl` and FFI. Feature `rdma-nixl` compiles with an empty dependency list. Live RDMA is `NOT_RUN_HARDWARE`. Ordinary NIC RDMA is not a DPU result. | +| OpenLake, ContextStore, Fluxon | Not required for identity, protocol, the shard directory, or the cold contract. No version is pinned because no code is imported. | -This is a scope decision, not a line-by-line license audit of those upstream trees. Nothing from them was copied. Direct dependencies are only `sha2` (runtime) and `proptest` (tests). V1 C did not add a crate. See [THIRD_PARTY_NOTICES.md](../THIRD_PARTY_NOTICES.md). +This is a license-and-binding audit of the transport choice, not a line-by-line audit of those upstream trees. Nothing from them was copied. Direct dependencies are only `sha2` (runtime) and `proptest` (tests). See [THIRD_PARTY_NOTICES.md](../THIRD_PARTY_NOTICES.md). diff --git a/docs/RISKS.md b/docs/RISKS.md index 4173e51..51491d3 100644 --- a/docs/RISKS.md +++ b/docs/RISKS.md @@ -13,7 +13,10 @@ | LRU admission | Mitigated with a pin check | Unpinned committed blocks can be evicted. Pinned and quarantined blocks are not. After a missing `LRU` file, order falls back to ascending generation. | | Partial prefix | Documented in ADR 0003 | Whole-block prefix only. A client that ignores `matched_prefix_blocks` can mis-align. Intra-block slice is a miss. | | RustFS live probe | `NOT_RUN_HARDWARE` | Mock Put/Head/RangeGet/Delete passes. A mock is not a measured RustFS run. | -| No RDMA / DPU measurement | `NOT_RUN_HARDWARE` | Do not infer a speedup from this milestone. | +| No RDMA / DPU measurement | `NOT_RUN_HARDWARE` | Feature `rdma-nixl` does not link libnixl. Do not infer a speedup. GDS and BlueField stay in development. | +| Simulated registration treated as a NIC pin | Mitigated by labeling | `RegisteredPool` uses `Vec`. `shard-demo` prints `registration=SIMULATED`. | +| Shard directory treated as a replica | Mitigated by ADR 0005 | Membership fences fills. It does not copy payload bytes and it does not vote a quorum. | +| Cold refill over budget returned as a hit | Mitigated | `OverBudget` and `Missing` carry no payload. Checksum failure is an error. | | License of a future DOCA or model stack | `NOT_RUN_LICENSE` | No DOCA binary and no model weights are in the tree. | | GPU rental or DPU purchase | `NOT_RUN_BUDGET` | No paid API, GPU, or DPU was used. | | Hot block loss | Accepted | No erasure coding and no durability replicas. The outcome is a clean miss, then recompute and repopulate. Cold durability is RustFS G4. See ADR 0002. | diff --git a/docs/V2_DATA_PATHS.md b/docs/V2_DATA_PATHS.md new file mode 100644 index 0000000..87efa47 --- /dev/null +++ b/docs/V2_DATA_PATHS.md @@ -0,0 +1,51 @@ +# V2 data paths + +These are the copies this tree actually performs, and the copies it does not. No path is a full zero-copy chain from GPU memory to a NIC to NVMe. GPU, GDS, BlueField, and live RDMA are `NOT_RUN_HARDWARE` / in development. An aarch64 build is not a BlueField test. + +`staged_bytes` on a TCP receipt is the length of one host staging copy. It is not NIC wire bytes and not a latency. + +## Path 1 — prefill / decode handoff + +| Hop | What this tree does | Copy or registration | +| --- | --- | --- | +| Prefill GPU HBM | Not executed | `NOT_RUN_HARDWARE` | +| GPU → host | Not linked. No CUDA copy and no GPUDirect | `NOT_RUN_HARDWARE` | +| Simulated register | `RegisteredPool` binds a `Vec` slot. A second checkout of the same slot does not allocate a new region id. Unknown completion quarantines the slot. Lease expiry does not free it | Host allocation. Not `ibv_reg_mr` | +| Host staging | `TcpTransport::submit` and `stage_through_tcp` copy the payload into a bounded buffer, then drop it. Over the cap is backpressure (`QuotaExceeded`) and the slots are quarantined | One host copy. `dma_posted=false` | +| TCP RPC | Existing `TcpClient` / `serve` path. Request bytes are still GCK1. The read side uses the V1 buffer pool | Host copy into the socket and the read buffer | +| NIC | Kernel TCP. Feature `rdma-nixl` does not open NIXL | No RDMA work request. `NOT_RUN_HARDWARE` | +| Decode host → GPU | Not executed | `NOT_RUN_HARDWARE` | + +A future NIXL path, once `libnixl` is linked on a machine that has it, would add registration, a posted transfer, and a completion event between the two host or GPU buffers. That path is not in this binary. See [adr/0004-nixl-transport.md](adr/0004-nixl-transport.md). + +## Path 2 — G3.5 reuse + +| Hop | What this tree does | Copy or registration | +| --- | --- | --- | +| Lookup | Metadata only. Hit, whole-block prefix hit, or miss. No payload | No data copy | +| Directory | `ShardDirectory` records one in-flight fill per physical-key checksum. A joiner does not reserve a second credit. The directory does not return bytes | No data copy. No per-token central lookup. No quorum | +| Hit `Get` | TCP returns the committed block | Host copies, same as path 1 | +| Miss, stale epoch, wrong tenant, checksum failure | No bytes. The engine recomputes | No data copy | +| File backend | `FileStore` writes an ordinary file under the directory the operator passed | Host copy plus the kernel page cache. Not an NVMe format | +| Quarantine | Unknown completion holds the simulated slot until drain or a newer fenced epoch | The `Vec` stays allocated and counts against the pool | + +Single-node is shard 0 with the same directory methods. See [adr/0005-shard-fencing.md](adr/0005-shard-fencing.md). + +## Path 3 — G4 refill + +The hot TCP service does not depend on `gpucache-cold`. RustFS is not on the prefill-to-decode path. + +| Hop | What this tree does | Copy or registration | +| --- | --- | --- | +| Pack | `pack_blocks` writes one object and one manifest in host memory | Host copy into the pack buffers | +| Async sink | `AsyncSink` queues that pair up to a job and byte cap. `drain` puts the object, then the manifest. `cancel` drops the queue and writes nothing | Host buffers until drain. Not a runtime thread pool | +| Mock store | In-process map. Contract test covers Put, Head, RangeGet, Delete | Not RustFS. Not a measurement | +| Live HTTP | `live_probe` runs only when `GPUCACHE_RUSTFS_*` is set. Otherwise `NOT_RUN_HARDWARE` | Full HTTP body copy. Not GDS | +| Refill | Head the manifest. If its length exceeds the budget, return `OverBudget` and do not download it. Otherwise download the manifest, and RangeGet one generation only when that range fits. Checksum mismatch is an error, not a hit | Host copy of the manifest and, when allowed, one payload range | +| Promote into G3.5 | A later hot `Put` through the TCP API | A separate host copy. Not chained to the GET | + +GDS and BlueField storage DMA are in development and were not run. + +## What a caller must do on failure + +Miss, timeout, checksum failure, over-budget refill, fenced epoch, and transport error are recomputes. None of them is a successful hit. A wrong-tenant or wrong-layout lookup stays a miss. diff --git a/docs/V2_TEST_MATRIX.md b/docs/V2_TEST_MATRIX.md new file mode 100644 index 0000000..e651425 --- /dev/null +++ b/docs/V2_TEST_MATRIX.md @@ -0,0 +1,25 @@ +# V2 test matrix + +CPU contract for the first V2 slice. Live RDMA, live RustFS, GPU export, and token goodput stay `NOT_RUN_HARDWARE`. This file is not a performance report. + +| Check | Result | Where | +| --- | --- | --- | +| Default workspace tests, including V1 A/B/C | PASS | `cargo test --workspace --all-targets` | +| TCP transport trait: one host copy, backpressure, `dma_posted=false` | PASS | `iface::tests` | +| `NixlTransport::open` refuses without a device | PASS | `iface::tests::rdma_open_does_not_post`, `tests/v2_control.rs` | +| Feature `rdma-nixl` still refuses `open` | PASS | `cargo test -p gpucache-transport --all-targets --features rdma-nixl` | +| Registration reuse, batch backpressure, quarantine, lease hold, fence reclaim | PASS | `registered::tests` | +| Failed TCP stage quarantines the simulated slots | PASS | `stage::tests` | +| Single-node directory is one shard; join charges credits once | PASS | `directory::tests` | +| Two shard values do not share flights | PASS | `directory::tests::two_shard_values_do_not_share_flights` | +| Stale epoch fences the fill and releases credits | PASS | `directory::tests::stale_epoch_releases_credits_and_hides_the_fill` | +| Other tenant cannot join; quota does not drop the leader | PASS | `directory::tests` | +| `shard-demo` prints `SHARD_OK` and `rdma=NOT_RUN_HARDWARE` | PASS | `cargo run -p gpucache-transport --bin shard-demo` | +| Cold mock Put/Head/RangeGet/Delete, pack, manifest | PASS | `contract_smoke` | +| Async sink drain, cancel writes nothing, queue backpressure | PASS | `sink::tests` | +| Refill over budget does not download the payload (or the manifest, when the manifest itself is over budget) | PASS | `sink::tests` | +| `cold-demo` prints `COLD_CONTRACT_OK` and `gds=NOT_RUN_HARDWARE` | PASS | `cargo run -p gpucache-cold --bin cold-demo` | +| Live RustFS Put/Head/RangeGet/Delete | NOT_RUN_HARDWARE | No `GPUCACHE_RUSTFS_*` endpoint | +| Live NIXL / Soft-RoCE / hardware RDMA | NOT_RUN_HARDWARE | `libnixl` is not linked. See [adr/0004-nixl-transport.md](adr/0004-nixl-transport.md) | +| GPU KV export, NIXL P/D, TTFT, TPOT, goodput | NOT_RUN_HARDWARE | `cargo run -p gpucache-bench --bin paired-baseline` | +| BlueField, DOCA, GDS, SPDK | NOT_RUN_HARDWARE | Not started. Not an erasure-coding milestone | diff --git a/docs/adr/0002-no-erasure-coding.md b/docs/adr/0002-no-erasure-coding.md index b8fad61..a266f99 100644 --- a/docs/adr/0002-no-erasure-coding.md +++ b/docs/adr/0002-no-erasure-coding.md @@ -24,5 +24,6 @@ DOCA EC, SPDK RAID targets, and similar offloads are not GPUCache components. Th - The README and roadmap do not list an EC milestone. - The V1 crates have no EC types, no parity buffers, and no replica placement. The file backend deletes or skips a bad file; it does not rebuild one. V1 C did not add parity. A short write or a full disk is a miss. -- RustFS remains responsible for any durable cold object it is asked to store. The V1 B cold client is off the hot path and is a contract mock unless an authorized endpoint is configured. +- RustFS remains responsible for any durable cold object it is asked to store. The cold client is off the hot path and is a contract mock unless an authorized endpoint is configured. +- V2 shard membership fences in-flight fills. It does not copy a block onto another node and it is not a durability replica. - A cache miss after a node loss is correct behavior. The KV block is not the user's only copy. diff --git a/docs/adr/0004-nixl-transport.md b/docs/adr/0004-nixl-transport.md new file mode 100644 index 0000000..809ce93 --- /dev/null +++ b/docs/adr/0004-nixl-transport.md @@ -0,0 +1,37 @@ +# ADR 0004: NIXL is the RDMA slot; it is not linked + +- Status: **Accepted** +- Date: 2026-10-15 +- Milestone: V2 first slice + +## Context + +V2 needs a transport besides TCP for a future RoCE or InfiniBand path. The execution spec says to reuse NIXL or Mooncake, prefer an existing Rust binding, and not rewrite an RDMA protocol. CI and this development host have no Soft-RoCE device and no `libnixl`. The workspace forbids `unsafe`. TCP is already the correctness backend. + +Candidates audited on 2026-10-15: + +| Candidate | License | Rust binding | Why it is or is not linked | +| --- | --- | --- | --- | +| NIXL (`ai-dynamo/nixl`) | Apache-2.0 | `nixl-sys` on crates.io. The 1.3.1 publish was updated 2026-07-07. docs.rs also lists 1.5.0. The crate describes a higher-level `nixl` wrapper. | Not linked. `stub-api` compiles without the C++ library and aborts the process if a NIXL function is called. A real build needs `libnixl` (default prefix `/opt/nvidia/nvda_nixl`) and FFI. | +| Mooncake Transfer Engine (`kvcache-ai/Mooncake`) | Apache-2.0 | In-tree `mooncake-transfer-engine/rust` uses bindgen and `unsafe` to link the C++ library. It is not a crates.io crate we can pin without vendoring. | Not linked. Fallback candidate only. | + +No CUDA, DOCA, or SPDK crate was considered. + +## Decision + +The V2 RDMA slot is NIXL, behind the Cargo feature `rdma-nixl` on `gpucache-transport`. + +- The feature has an empty dependency list. `Cargo.lock` does not pin `nixl-sys`. +- `NixlTransport::open` returns `CacheError::Storage { reason: "rdma not linked" }` whether or not the feature is enabled. +- `rdma_feature_selected()` is the only behavioral difference. It does not post a work request. +- `TcpTransport` implements `KvTransport` and is the default. `submit` copies bytes into a bounded host buffer. `dma_posted` is false. `staged_bytes` is that copy's length, not NIC wire bytes and not a latency. +- The GCK1 / GCKR client API is unchanged. `TcpClient` remains the cache RPC path. + +Live RDMA is `NOT_RUN_HARDWARE`. This ADR does not claim a TTFT, TPOT, or bandwidth number. + +## Consequences + +- `cargo test --workspace --all-targets` does not fetch NIXL and does not need a NIC. +- `cargo test -p gpucache-transport --all-targets --features rdma-nixl` still refuses `open`. +- A later change may add an optional dependency only after a non-aborting adapter exists and a machine with `libnixl` is authorized. That dependency stays feature-gated and off the default build. +- Mooncake is not a second half-implemented path. diff --git a/docs/adr/0005-shard-fencing.md b/docs/adr/0005-shard-fencing.md new file mode 100644 index 0000000..8a4335c --- /dev/null +++ b/docs/adr/0005-shard-fencing.md @@ -0,0 +1,27 @@ +# ADR 0005: Shard-local directory and fencing + +- Status: **Accepted** +- Date: 2026-10-15 +- Milestone: V2 first slice + +## Context + +A distributed KV tier needs to know which process may fill a block, and it needs to reject work from a stale membership. A central directory touched once per token, or a quorum vote per block, would sit on the miss path of recomputable data. V1 already fences buffer reclaim with a store epoch. The client wire format is stable from one process to many. + +## Decision + +`ShardDirectory` is local to one shard. One process is one shard. A single machine is shard 0 using the same methods as any other shard id. + +- Membership is a list of `(node_id, incarnation)` on that shard. A new member, a removed member, or a changed incarnation bumps the shard epoch and fails in-flight fills. An unchanged incarnation does not. The last member cannot be removed. +- There is no cross-shard quorum and no per-token index. The map key is the SHA-256 of the physical byte key. One entry is one in-flight fill. +- Single-flight: the first same-epoch caller is the leader and reserves capacity credits once. A later same-tenant caller joins and does not reserve again. A different tenant is `Unauthorized`. The directory stores no payload. `Admit` is `Leader` or `Joined` only. +- Completing or failing a fill requires the leader request id and the current epoch. A stale epoch is `EpochMismatch`. After the epoch moves, the fill is gone, credits are released, and a new fill may start. That new fill is not a hit. +- Credits are an admission ceiling (`InFlightBytes`, `TenantRetainedBytes`). Exhaustion refuses the new leader and does not evict an existing one. This is not a replica and not erasure coding. See [0002-no-erasure-coding.md](0002-no-erasure-coding.md). + +The GCK1 request bytes and GCKR replies are unchanged. + +## Consequences + +- Tests construct two `ShardDirectory` values and show they do not share flights. That is the single-node and the multi-shard API. It is not a multi-host deployment. +- A lost shard is a miss. The engine recomputes and may fill again. +- Buffer quarantine stays in `RegisteredPool`: unknown completion and lease expiry do not free a slot. Fence plus `reclaim_fenced`, or `attest_drain`, does. Region ids are not recycled. The pool's bytes are a `Vec`. No NIC memory registration is performed.