diff --git a/deploy/deepseek_v4/tp4_profile.json b/deploy/deepseek_v4/tp4_profile.json index a6a7f07..a0cb9d7 100644 --- a/deploy/deepseek_v4/tp4_profile.json +++ b/deploy/deepseek_v4/tp4_profile.json @@ -4,7 +4,7 @@ "cache_model_profile": "deepseek-v4-fp8-hma", "published_runtime_base": "ghcr.io/fujitsupolycom/gb10-vllm-serving@sha256:6fc26fdad81a18f0fff67ce0a05f6d90165625ea2e1cac8a6f39bfb462017028", "sparkcache": { - "source_sha256": "72f33311d1ee5811c2b3e4eefda98c9d17b5afe8d9d015a2e3c26c893f25aabc" + "source_sha256": "7efffbec85cce754b7aad057cead39f4b1e1cb7f35833e8df8e487d3d4a46867" }, "model": { "repository": "deepseek-ai/DeepSeek-V4-Flash-0731", @@ -23,7 +23,7 @@ "kv_cache_bytes_per_rank": 34359738368, "max_model_len": 524288, "max_num_seqs": 32, - "gpu_memory_utilization": 0.70, + "gpu_memory_utilization": 0.7, "tokenizer_mode": "deepseek_v4", "speculation_method": "dspark", "speculation_tokens": 5, diff --git a/deploy/glm52_35bpw/profile.json b/deploy/glm52_35bpw/profile.json index 0bbbfb9..08a6250 100644 --- a/deploy/glm52_35bpw/profile.json +++ b/deploy/glm52_35bpw/profile.json @@ -5,7 +5,7 @@ "published_runtime_base": "ghcr.io/fujitsupolycom/gb10-vllm-serving@sha256:6fc26fdad81a18f0fff67ce0a05f6d90165625ea2e1cac8a6f39bfb462017028", "base_image_requirement": "exact GLM-5.2 3.5-bpw R7 image recorded by the source container inspection", "sparkcache": { - "source_sha256": "72f33311d1ee5811c2b3e4eefda98c9d17b5afe8d9d015a2e3c26c893f25aabc" + "source_sha256": "7efffbec85cce754b7aad057cead39f4b1e1cb7f35833e8df8e487d3d4a46867" }, "model": { "repository": "brandonmusic/GLM-5.2-EXL3-TR3v4-3.5bpw-MTP78", diff --git a/deploy/glm53_flash/MTP3_STATUS.md b/deploy/glm53_flash/MTP3_STATUS.md index 15840ad..1427ac6 100644 --- a/deploy/glm53_flash/MTP3_STATUS.md +++ b/deploy/glm53_flash/MTP3_STATUS.md @@ -29,21 +29,35 @@ identified above. It requires a rebuilt image and startup validation. ## What remains in the performance issues +Source builds include request-attribution accounting, same-pass maintenance +inventory reuse, and optional deletion pacing. The immutable image identified +above does not contain these additions. Attribution also requires the matching +instrumented SparkRing scheduler; a package update alone cannot supply its events. + | Issue | Implemented | Remaining work | |---|---|---| -| [#60: sustained publication and eviction slowdown](https://github.com/FujitsuPolycom/sparkcache/issues/60) | Fewer repeated restore reads and hashes, bounded restore arenas, tiled native placement, publication-dependency protection, publication-backlog gauges, optional periodic full captures, and capacity guidance. | Maintenance still scans and sorts the inventory and evicts toward the low watermark; it has no per-pass time or entry budget. Reproduce the reported slowdown with a near-full 40 GiB store and matched before/after probes. | -| [#61: growing conversations lose local prefix reuse](https://github.com/FujitsuPolycom/sparkcache/issues/61) | Runtime GPU-lease accounting, preference for longer local prefixes, recurrent checkpoint retention, and opt-in restore/lease traces. | Exact per-request local-hit, verified-restore, and recompute token counters are not implemented. Validate local retention on the reported long-context, multi-turn workload with occasional images. | +| [#60: sustained publication and eviction slowdown](https://github.com/FujitsuPolycom/sparkcache/issues/60) | Reduced restore work, publication-dependency protection, backlog gauges, capacity guidance, same-pass inventory reuse, and opt-in deletion-attempt limits and cooldown. | Full inventory scans and individual filesystem operations still have no hard time bound. Validate the combined implementation against near-full 40 GiB traffic and matched before/after probes. | +| [#61: growing conversations lose local prefix reuse](https://github.com/FujitsuPolycom/sparkcache/issues/61) | Runtime lease accounting, local-prefix preference, checkpoint retention, and opt-in request attribution for accepted target execution. | Deploy the matching scheduler instrumentation and validate attribution and local retention on the reported long-context conversations with occasional images. | One saver admission per rank bounds concurrent optional work; it does not bound -the duration of an inventory scan or eviction pass. See +the duration of an inventory scan. Deletion pacing is disabled by default. +Smaller deletion budgets can repeat scans and increase total work. See [capacity and cleanup](../../sparkcache/README.md#capacity-and-cleanup). Backlog reports distinguish pending publications, their oldest age, and active maintenance, but reports may not refresh while the connector is idle. -Reuse traces distinguish restore offers, verified worker completion, and GPU -lease attachment. Their scheduler prefix-token field is block-aligned input, -not an exact local hash-hit measurement. API `cached_tokens` alone is not enough -to attribute reuse to GPU retention rather than persistent restoration. +Matching traces distinguish restore offers, worker completion, and GPU lease +attachment. Their scheduler prefix-token input and API `cached_tokens` alone +cannot establish exact attribution. + +The [request-attribution record](../../sparkcache/README.md#request-reuse-attribution) +uses authoritative scheduler events to distinguish consumed local prefixes, +verified external reuse, and accepted prompt computation. Counts accumulate +across preemption attempts; incomplete observations are labeled explicitly. + +The [maintenance inventory benchmark](../../docs/maintenance-inventory-validation.md) +measures reduced metadata work on CPU fixtures. It is not a DGX4 serving result +or confirmation that the original slowdown is resolved. Both issues should remain open until their remaining implementation questions and workload-specific results are recorded explicitly. diff --git a/docs/maintenance-inventory-validation.json b/docs/maintenance-inventory-validation.json new file mode 100644 index 0000000..16e5a3d --- /dev/null +++ b/docs/maintenance-inventory-validation.json @@ -0,0 +1,137 @@ +{ + "schema": "sparkcache-maintenance-inventory-benchmark/v1", + "status": "research-only", + "conditions": { + "platform": "Windows-11-10.0.26200-SP0", + "filesystem": "NTFS", + "filesystem_verification": "Get-Volume -DriveLetter C", + "python_execution": "native Windows", + "paired_method": "Separate sequential processes constructed equivalent temporary caches and invoked SparkContextCacheConnector._maintain_capacity(force=True).", + "timing": "Median of seven complete maintenance/reconciliation passes after an untimed instrumented pass; garbage collection precedes each timer.", + "excluded": [ + "fixture construction", + "publication", + "Python imports", + "I/O counting instrumentation" + ], + "included": [ + "exclusive cache lock", + "complete reference inventory", + "offer reconciliation" + ], + "eviction": "disabled by a 1000000000-byte capacity ceiling", + "gpu_execution": false, + "serving_or_nvme_qualification": false, + "tool": "tools/benchmark_maintenance_inventory.py" + }, + "sources": { + "baseline": { + "revision": "7fc9f85509624a0f9c5bc6bd7c5f70f1a43d389e", + "cache_manifest_sha256_lf": "88caa5b104407a212b4344b82ab914d6c5c7e72d021976ba6d1c44d13f7f2f2e", + "connector_sha256_lf": "082f2820f45cdf951629d045b5d316461f86372fae70425be9b5903825395a97", + "revision_matches_recorded_source_hashes": true + }, + "survivor_inventory": { + "mechanism": "MaintenanceReport.surviving_entries supplies metadata-qualified offers from the exclusive inventory pass.", + "cache_manifest_sha256_lf": "ddb3e2e00808791ed585122b8ab1171ac7db391c19fc62151f0f39107dee256a", + "connector_sha256_lf": "f92b1fcaf3385e5d4487b246f06c1de3fb02e8f05b67c055ab760d91ccd86eb4" + } + }, + "flat_page_history": { + "layout": "Opaque 64-byte attention pages grouped by 256 tokens, plus a 32-byte recurrent-state page.", + "branches": 8, + "extensions_per_branch": 32, + "shared_initial_tokens": 512, + "extension_tokens": 256, + "roots": 257, + "unique_chunks": 257, + "chunk_references": 4481, + "baseline": { + "root_reads": 514, + "chunk_stats": 4995, + "elapsed_ms": [ + 185.0543999898946, + 183.6131000018213, + 173.09400001249742, + 178.17310000828002, + 174.59690000396222, + 174.0127999946708, + 175.22859999735374 + ], + "median_ms": 175.22859999735374 + }, + "survivor_inventory": { + "root_reads": 257, + "chunk_stats": 514, + "elapsed_ms": [ + 72.49789999332279, + 70.9155000076862, + 72.2200000018347, + 70.33850000880193, + 76.28729999123607, + 89.75840000493918, + 72.21110000682529 + ], + "median_ms": 72.2200000018347 + }, + "conclusion": "Complete pass latency decreased 58.8 percent in this CPU fixture. Each root was read once; shared-chunk metadata probes scaled with unique files rather than root references. This does not establish a serving-throughput improvement." + }, + "token_rows": { + "roots": 64, + "unique_chunks": 32, + "chunk_references": 2048, + "baseline": { + "root_reads": 128, + "chunk_stats": 2112, + "elapsed_ms": [ + 37.09349999553524, + 37.26160000951495, + 37.37550000369083, + 37.524400002439506, + 37.23980000359006, + 41.24779999256134, + 52.597900008549914 + ], + "median_ms": 37.37550000369083 + }, + "survivor_inventory": { + "root_reads": 64, + "chunk_stats": 64, + "elapsed_ms": [ + 10.044100010418333, + 10.23929999792017, + 10.25409999419935, + 10.973099997499958, + 10.138300000107847, + 9.93929999822285, + 9.973099993658252 + ], + "median_ms": 10.138300000107847 + } + }, + "combined_source_repeat": { + "revision": "11658f0a3b9f155af1d95ff78a807c9b374f6093", + "kind": "pages", + "roots": 257, + "unique_chunks": 257, + "references": 4481, + "platform": "Windows-11-10.0.26200-SP0", + "filesystem": "NTFS", + "manifest_source_sha256": "f2cc0bc7570b59ed068ad3273b234c0f8c9bd945a6a68dca0014012a118d1400", + "connector_source_sha256": "f92b1fcaf3385e5d4487b246f06c1de3fb02e8f05b67c055ab760d91ccd86eb4", + "io_counts": { + "root_reads": 257, + "chunk_stats": 514 + }, + "elapsed_ms": [ + 76.01800000702497, + 72.13889999547973, + 73.28540000889916, + 82.84220000496134, + 89.14029999868944, + 79.15170000342187, + 78.41010000265669 + ], + "median_ms": 78.41010000265669 + } +} diff --git a/docs/maintenance-inventory-validation.md b/docs/maintenance-inventory-validation.md new file mode 100644 index 0000000..7f6ee9a --- /dev/null +++ b/docs/maintenance-inventory-validation.md @@ -0,0 +1,54 @@ +# Maintenance inventory and offer reconciliation + +Status: **implemented**. Serving-performance qualification is **research-only**. + +`MaintenanceReport.surviving_entries` identifies metadata-qualified roots from +the exclusive inventory pass. Offer reconciliation reuses that result without +rereading roots or probing shared payload files once per reference. + +The [CPU benchmark record](maintenance-inventory-validation.json) measures the +complete maintenance and reconciliation call on Windows 11 with NTFS. It +compares equivalent fixtures against source revision `7fc9f85509624a0f9c5bc6bd7c5f70f1a43d389e`. + +| Fixture | Baseline median | Inventory median | Root reads | Chunk metadata probes | +|---|---:|---:|---:|---:| +| Eight branches, 32 flat page extensions each | 175.23 ms | 72.22 ms | 514 to 257 | 4,995 to 514 | +| 64 token-row roots sharing 32 chunks | 37.38 ms | 10.14 ms | 128 to 64 | 2,112 to 64 | + +A repeat with combined source `11658f0a3b9f155af1d95ff78a807c9b374f6093`, +including canonical string ordering, measured 78.41 ms for the flat-page fixture +and retained the same reduced I/O counts. The JSON records that source and its +seven observations separately. + +The page fixture includes attention and recurrent-state bytes. It exercises +257 roots and 4,481 chunk references with tiny payloads, not a loaded model. +The measurements do not qualify DGX serving throughput or NVMe behavior. + +Each normal pass still reads the complete inventory and traverses its reference +graph. Root parsing falls from two passes to one, and chunk metadata probes +scale with unique files instead of the total number of root references. + +Run the [benchmark tool](../tools/benchmark_maintenance_inventory.py) from the +checkout being measured. Use the same tool against each source checkout. +Install the repository's CPU test dependencies; the connector fixture stubs +its model-runtime interfaces. + +```powershell +$env:BENCH_FILESYSTEM = 'NTFS' +$env:BENCH_KIND = 'pages' +python tools/benchmark_maintenance_inventory.py 8 32 +$env:BENCH_KIND = 'rows' +python tools/benchmark_maintenance_inventory.py 64 32 +``` + +Set `BENCH_FILESYSTEM` to the independently verified filesystem. Fixture +construction and counting instrumentation are outside the seven timed passes. +The artifact records normalized source hashes for both measured implementations. + +Regression tests in `sparkcache/test_maintenance_survivors.py` cover shared +payload metadata, corrupt exact roots shadowing aliases, storage-mode +eligibility, concurrent publication, and repeated inventory freshness. + +The snapshot does not authorize cache restoration. Payload integrity remains +mandatory at restore, and same-sized corruption becomes a verified miss. +No cache identity, namespace, or persisted format changes. diff --git a/sparkcache/README.md b/sparkcache/README.md index 768f7de..d06b1e6 100644 --- a/sparkcache/README.md +++ b/sparkcache/README.md @@ -311,6 +311,96 @@ A larger gap amortizes maintenance across more writes, but each pass evicts more data and can increase future misses. Compare those costs with observed publication age and maintenance activity before changing the gap. +### Deletion pacing + +Status: **implemented**; serving performance with these controls is +**research-only**. Both settings default to `0`, preserving unrestricted passes: + +- `spark_cache_maintenance_max_deletions` caps filesystem unlink attempts per + pass, including failed attempts, across manifests, aliases, debris, + descriptor segments, and payload objects. Its environment fallback is + `SPARK_CONTEXT_CACHE_MAINTENANCE_MAX_DELETIONS`. +- `spark_cache_maintenance_interval_ms` sets a minimum cooldown after a pass + finishes or fails. Forced post-commit calls also respect it; skipped calls + do not extend it. Its environment fallback is + `SPARK_CONTEXT_CACHE_MAINTENANCE_INTERVAL_MS`. + +Explicit connector settings take precedence over environment values. A test +configuration can select `128` deletion attempts and `1000` milliseconds; +these values are not a qualified serving-performance recommendation. + +With a deletion budget, existing orphan payloads and debris are reclaimed +before additional roots are selected. + +A pass can stop above the low watermark +or the capacity maximum; optional store admission remains blocked while +capacity is unsatisfied. + +A process-local pressure target retains the low watermark across passes, +even after usage falls below the high watermark. Restarting the process +loses this target; a later high-watermark crossing establishes it again. + +Background retries complete deferred cleanup without +making serving wait. Root-directory durability barriers still precede object +removal, and protected publication roots retain their complete object graphs. + +This is **not a scan-size or wall-clock bound**. Each admitted pass authenticates +the complete reference inventory and reconciles survivors. + +Status: **implemented**. A completed pass also returns metadata-qualified +surviving roots while the exclusive filesystem guard is held. + +The connector +uses this inventory to reconcile offers without rereading every root or +restatting shared chunks once per reference. + +Qualification checks logical +file sizes, not allocated disk space. Exact roots shadow aliases even when +invalid; aliases are eligible only for token-row storage. + +Protected roots and +their referenced objects retain the same deletion rules. + +The inventory is not retained between passes. Failed or busy passes use +independent metadata probes. + +If the offered-digest inventory changes during a +pass or probe, reconciliation defers withdrawals to a stable pass, preserving +concurrent publication of the same digest. + +The connector counter +`capacity_stale_inventory_snapshots` and worker capacity-report field +`maintenance_stale_inventory_snapshots` count these deferrals. + +Restore still +authenticates payload bytes; metadata qualification cannot establish their +integrity or turn an invalid restore into a hit. + +One filesystem +operation or durability barrier can take arbitrarily long. Smaller deletion +budgets can increase total scan work. + +Cooldown trades reclamation throughput +for gaps between that work. A larger watermark gap does not remove this cost. + +`MaintenanceReport.deletion_attempts` counts admitted unlink attempts; +`work_pending` reports cleanup deferred by the budget or orphan-first policy; +`skipped_cooldown` reports a pass skipped before inventory work. + +Connector +counters `capacity_deletion_attempts`, `capacity_budget_exhausted`, and +`capacity_skipped_cooldown` expose the same activity. Capacity log records +include `deletion_attempts` and `work_pending`. + +Prometheus gauges +`vllm:sparkcache_maintenance_deletion_attempts`, +`vllm:sparkcache_maintenance_budget_exhausted`, and +`vllm:sparkcache_maintenance_skipped_cooldown` sum reported cumulative counts +across ranks. + +They reset with workers and reflect the last worker reports, +not independent scrape-time measurements. + `spark_cache_ttl_seconds` expires manifests by recency; zero disables TTL. Maintenance preserves shared objects referenced by surviving manifests. @@ -424,6 +514,39 @@ The counters describe host-side operations. They do not report filesystem allocation, NVMe Data Units Written, controller write amplification, or NAND writes. +### Request reuse attribution + +Status: **implemented** with an optional scheduler callback. A runtime without +that callback cannot produce exact request attribution from connector offers. + +Set `SPARK_CONTEXT_CACHE_TRACE_REUSE=1` before startup. An instrumented scheduler +emits one `request_cache_attribution` event in `sparkcache-reuse-trace/v1` at +request cleanup. No request IDs are added to Prometheus labels. + +| Field | Meaning | +|---|---| +| `local_tokens_reused` | GPU-resident prompt tokens consumed by accepted target execution, including resident shared leases. | +| `external_tokens_reused` | Prompt tokens consumed after a successful all-rank persistent restore and the scheduler's final-token adjustment. | +| `prompt_tokens_computed` | Prompt intervals completed by accepted target execution, accumulated across preemption attempts. | +| `preemptions` | Request preemption generation observed by the scheduler. | +| `attribution_complete` | True only for normal completion with observed prompt completion and no missing or invalid accounting boundary. | + +Counts cover accepted target-prompt work across attempts. They can exceed the +original prompt length after preemption. They exclude output tokens, draft +execution, replay inside kernels, and rejected worker output. + +An offered restore earns no credit. A verified restore aborted before target +execution earns no reused-token credit. A follower consuming a resident GPU +lease records local reuse, even if a different request restored that lease. + +The restored state span and external prompt tokens reused are distinct. A +restore can write an already-local prefix, and a full prompt hit still needs +the final prompt token recomputed for sampling logits. + +Incomplete observations remain labeled incomplete. Their token fields are not +added to the connector's `attribution_completed_*` aggregate counters. Request +cleanup releases the ledger even when logging fails. + Telemetry is observational. It cannot change publication, restore, cache identity, or serving decisions. diff --git a/sparkcache/persistent_context_cache/cache_manifest.py b/sparkcache/persistent_context_cache/cache_manifest.py index dc82631..0c669b7 100644 --- a/sparkcache/persistent_context_cache/cache_manifest.py +++ b/sparkcache/persistent_context_cache/cache_manifest.py @@ -444,9 +444,12 @@ class CapacityPolicy: max_bytes: int = 0 low_watermark_bytes: int = 0 ttl_seconds: int = 0 + maintenance_max_deletions: int = 0 + maintenance_interval_ms: int = 0 def __post_init__(self) -> None: - for field in ("max_bytes", "low_watermark_bytes", "ttl_seconds"): + for field in ("max_bytes", "low_watermark_bytes", "ttl_seconds", + "maintenance_max_deletions", "maintenance_interval_ms"): value = getattr(self, field) if isinstance(value, bool) or not isinstance(value, int) or value < 0: raise ValueError(f"{field} must be a non-negative integer") @@ -473,9 +476,15 @@ class MaintenanceReport: evicted_entries: tuple[EntryKey, ...] = () capacity_satisfied: bool = True skipped_busy: bool = False + skipped_cooldown: bool = False + deletion_attempts: int = 0 + work_pending: bool = False aliases_evicted: int = 0 segments_deleted: int = 0 orphan_segments_deleted: int = 0 + # Metadata-qualified offers at the end of the exclusive inventory pass. + # None means no authoritative inventory was completed, not an empty cache. + surviving_entries: tuple[EntryKey, ...] | None = None @dataclass(frozen=True) @@ -2410,16 +2419,31 @@ def maintain( guard.__enter__() except BlockingIOError: return MaintenanceReport(capacity_satisfied=False, skipped_busy=True) + if time.monotonic() < getattr(self, "_maintenance_not_before", 0.0): + guard.__exit__(None, None, None) + return MaintenanceReport(capacity_satisfied=False, skipped_cooldown=True) try: + deletion_attempts = 0 + work_pending = False + + def admit_deletion() -> bool: + nonlocal deletion_attempts, work_pending + if (policy.maintenance_max_deletions > 0 + and deletion_attempts >= policy.maintenance_max_deletions): + work_pending = True + return False + deletion_attempts += 1 + return True + manifests_root = self.root / "manifests" aliases_root = self.root / "prefix-aliases" manifest_paths = ( - tuple(sorted(manifests_root.glob("*/*.json"))) + tuple(sorted(manifests_root.glob("*/*.json"), key=os.fspath)) if manifests_root.is_dir() else () ) alias_paths = ( - tuple(sorted(aliases_root.glob("*/*.json"))) + tuple(sorted(aliases_root.glob("*/*.json"), key=os.fspath)) if aliases_root.is_dir() else () ) @@ -2440,11 +2464,12 @@ def root_files(root: Path) -> tuple[Path, ...]: return () return tuple( sorted( - path + (path for directory in root.iterdir() if directory.is_dir() for path in directory.iterdir() - if path.is_file() + if path.is_file()), + key=os.fspath, ) ) @@ -2479,20 +2504,24 @@ def root_files(root: Path) -> tuple[Path, ...]: chunk_directory = self.root / "chunks" chunk_paths = ( tuple( - sorted(path for path in chunk_directory.iterdir() if path.is_file()) + sorted((path for path in chunk_directory.iterdir() if path.is_file()), + key=os.fspath) ) if chunk_directory.is_dir() else () ) chunk_sizes: dict[Path, int] = {} + chunk_logical_sizes: dict[str, int] = {} canonical_chunks: dict[str, Path] = {} for path in chunk_paths: try: - chunk_sizes[path] = _allocated_bytes(path.stat()) + metadata = path.stat() + chunk_sizes[path] = _allocated_bytes(metadata) except FileNotFoundError: continue if path.suffix == ".spcc" and _DIGEST.fullmatch(path.stem): canonical_chunks[path.stem] = path + chunk_logical_sizes[path.stem] = metadata.st_size references: Counter[str] = Counter() segment_references: Counter[tuple[str, str]] = Counter() @@ -2507,8 +2536,18 @@ def root_files(root: Path) -> tuple[Path, ...]: + sum(segment_sizes.values()) + sum(chunk_sizes.values()) ) - pressure_triggered = ( - policy.max_bytes > 0 and bytes_before > policy.max_bytes + pressure_target = (policy.max_bytes, policy.low_watermark_bytes) + retained_target = getattr(self, "_maintenance_pressure_target", None) + pressure_triggered = policy.max_bytes > 0 and ( + bytes_before > policy.max_bytes + or (policy.maintenance_max_deletions > 0 + and retained_target == pressure_target + and bytes_before > policy.low_watermark_bytes) + ) + # Preserve high-to-low hysteresis across bounded passes. Falling + # below the high watermark alone does not complete reclamation. + self._maintenance_pressure_target = ( + pressure_target if pressure_triggered and policy.maintenance_max_deletions > 0 else None ) projected_bytes = ( bytes_before @@ -2526,17 +2565,36 @@ def root_files(root: Path) -> tuple[Path, ...]: or references.get(path.stem, 0) == 0 ) ) + # With a deletion budget, reclaim already-unreferenced payloads + # before choosing further live roots. Otherwise a small budget + # could repeatedly remove roots while their orphan payloads wait. + orphan_debt = bool(root_debris_sizes) or any( + canonical_segments.get((path.parent.name, path.stem)) != path + or segment_references.get((path.parent.name, path.stem), 0) == 0 + for path in segment_sizes + ) or any( + canonical_chunks.get(path.stem) != path + or references.get(path.stem, 0) == 0 + for path in chunk_sizes + ) projected_references = references.copy() projected_segment_references = segment_references.copy() selected: list[_CapacityEntry] = [] selected_paths: set[Path] = set() def select(entry: _CapacityEntry) -> None: - nonlocal projected_bytes + nonlocal projected_bytes, work_pending if entry.path in selected_paths or ( entry.valid and entry.key in protected ): return + if policy.maintenance_max_deletions > 0 and orphan_debt: + work_pending = True + return + if (policy.maintenance_max_deletions > 0 + and len(selected) >= policy.maintenance_max_deletions): + work_pending = True + return selected.append(entry) selected_paths.add(entry.path) projected_bytes -= entry.manifest_bytes @@ -2554,7 +2612,10 @@ def select(entry: _CapacityEntry) -> None: if path is not None: projected_bytes -= segment_sizes.get(path, 0) - ordered = sorted(entries, key=lambda entry: (entry.mtime_ns, entry.path)) + # Cache namespaces and digests have fixed-width canonical names. + # Compare each path's string once instead of normalizing Path + # components throughout every comparison in the inventory sort. + ordered = sorted(entries, key=lambda entry: (entry.mtime_ns, os.fspath(entry.path))) for entry in ordered: expired = policy.ttl_seconds > 0 and ( current_ns - entry.mtime_ns >= policy.ttl_seconds * 10**9 @@ -2571,6 +2632,8 @@ def select(entry: _CapacityEntry) -> None: affected_root_directories: set[Path] = set() root_debris_deleted = 0 for path, size in root_debris_sizes.items(): + if not admit_deletion(): + continue try: path.unlink() except FileNotFoundError: @@ -2582,6 +2645,8 @@ def select(entry: _CapacityEntry) -> None: affected_root_directories.add(path.parent) root_debris_deleted += size for entry in selected: + if not admit_deletion(): + continue try: entry.path.unlink() except FileNotFoundError: @@ -2595,7 +2660,7 @@ def select(entry: _CapacityEntry) -> None: # Root removals are durable before any object they authorized can # be collected. A failed root-directory barrier stops maintenance # with every shared segment and chunk still present. - for directory in sorted(affected_root_directories): + for directory in sorted(affected_root_directories, key=os.fspath): _fsync_directory(directory) remaining_references: Counter[str] = Counter() @@ -2627,6 +2692,8 @@ def select(entry: _CapacityEntry) -> None: and remaining_segment_references.get(reference, 0) > 0 ): continue + if not admit_deletion(): + continue try: path.unlink() except OSError: @@ -2636,7 +2703,7 @@ def select(entry: _CapacityEntry) -> None: affected_segment_directories.add(path.parent) if path in initial_orphan_segments: orphan_segments_deleted += 1 - for directory in sorted(affected_segment_directories): + for directory in sorted(affected_segment_directories, key=os.fspath): _fsync_directory(directory) chunks_deleted = 0 @@ -2654,6 +2721,8 @@ def select(entry: _CapacityEntry) -> None: and remaining_references.get(path.stem, 0) > 0 ): continue + if not admit_deletion(): + continue try: path.unlink() except OSError: @@ -2674,13 +2743,35 @@ def select(entry: _CapacityEntry) -> None: - segment_bytes_deleted - chunk_bytes_deleted, ) + if (pressure_triggered and policy.maintenance_max_deletions > 0 + and bytes_after > policy.low_watermark_bytes): + work_pending = True + else: + self._maintenance_pressure_target = None exact_removed = sum(entry.key.root_kind == "manifest" for entry in removed) aliases_removed = sum( entry.key.root_kind == "prefix_alias" for entry in removed ) + remaining_entries = [entry for entry in entries if entry.path not in removed_paths] + # Exact roots shadow aliases even when corrupt or when an unlink + # failed. Match lookup's exact-first policy without re-reading roots. + exact_keys = { + (entry.key.storage_key, entry.key.context_digest) + for entry in remaining_entries if entry.key.root_kind == "manifest" + } + surviving_entries = tuple( + entry.key for entry in remaining_entries + if entry.valid + and (entry.key.root_kind == "manifest" + or (entry.key.storage_key, entry.key.context_digest) not in exact_keys) + and all(chunk_logical_sizes.get(digest) == size + for digest, size in entry.chunks) + ) return MaintenanceReport( bytes_before=bytes_before, bytes_after=bytes_after, + deletion_attempts=deletion_attempts, + work_pending=work_pending, bytes_reclaimed=bytes_before - bytes_after, manifests_evicted=exact_removed, chunks_deleted=chunks_deleted, @@ -2692,8 +2783,12 @@ def select(entry: _CapacityEntry) -> None: aliases_evicted=aliases_removed, segments_deleted=segments_deleted, orphan_segments_deleted=orphan_segments_deleted, + surviving_entries=surviving_entries, ) finally: + self._maintenance_not_before = ( + time.monotonic() + policy.maintenance_interval_ms / 1000.0 + ) guard.__exit__(None, None, None) def _manifest_path(self, identity: CacheIdentity, context_digest: str) -> Path: diff --git a/sparkcache/persistent_context_cache/test_maintenance_budget.py b/sparkcache/persistent_context_cache/test_maintenance_budget.py new file mode 100644 index 0000000..53c8ff9 --- /dev/null +++ b/sparkcache/persistent_context_cache/test_maintenance_budget.py @@ -0,0 +1,210 @@ +"""Deletion pacing preserves cache references and publication barriers.""" + +from unittest import mock + +import pytest + +from sparkcache.persistent_context_cache import cache_manifest as manifest +from sparkcache.persistent_context_cache.cache_manifest import ( + CapacityPolicy, + EntryKey, + ManifestStore, +) +from sparkcache.persistent_context_cache.test_cache_manifest import ( + _identity, + _variant_chunk, +) + + +def populate(root, count=3): + store = ManifestStore(root) + identity = _identity() + digests = [f"{index + 1:064x}" for index in range(count)] + for digest in digests: + store.commit( + identity=identity, + context_digest=digest, + chunks=[_variant_chunk(digest.encode())], + ) + return store, identity, digests + + +def test_budget_drains_orphans_before_evicting_more_roots(tmp_path): + store, _, _ = populate(tmp_path) + policy = CapacityPolicy( + max_bytes=1, low_watermark_bytes=1, maintenance_max_deletions=1 + ) + first = store.maintain(policy) + assert first.manifests_evicted == first.deletion_attempts == 1 + assert first.work_pending + second = store.maintain(policy) + assert second.manifests_evicted == 0 + assert second.chunks_deleted == second.deletion_attempts == 1 + for _ in range(8): + result = store.maintain(policy) + assert result.deletion_attempts <= 1 + if result.capacity_satisfied and not result.work_pending: + break + else: + pytest.fail("bounded passes did not converge") + assert result.bytes_after == 0 + + +def test_budget_preserves_protected_shared_payload(tmp_path): + store, identity, digests = populate(tmp_path, 1) + sibling = "f" * 64 + store.commit( + identity=identity, + context_digest=sibling, + chunks=[_variant_chunk(digests[0].encode())], + ) + policy = CapacityPolicy( + max_bytes=1, low_watermark_bytes=1, maintenance_max_deletions=1 + ) + report = store.maintain( + policy, protected_entries=[EntryKey(identity.storage_key, sibling)] + ) + assert report.manifests_evicted == 1 + assert report.chunks_deleted == 0 + assert not report.capacity_satisfied + lookup = store.lookup(identity, sibling) + assert lookup.is_hit + assert store.restore(lookup) == (_variant_chunk(digests[0].encode()),) + assert report.bytes_after == sum( + manifest._allocated_bytes(p.stat()) + for folder in ("manifests", "chunks") + for p in (tmp_path / folder).rglob("*") + if p.is_file() + ) + + +def test_budget_counts_failed_unlink_attempts(tmp_path): + store, _, _ = populate(tmp_path) + policy = CapacityPolicy( + max_bytes=1, low_watermark_bytes=1, maintenance_max_deletions=1 + ) + with mock.patch("pathlib.Path.unlink", side_effect=OSError("busy")) as unlink: + report = store.maintain(policy) + assert unlink.call_count == report.deletion_attempts == 1 + assert report.manifests_evicted == report.chunks_deleted == 0 + assert report.bytes_before == report.bytes_after + assert report.work_pending + + +def test_cooldown_skips_inventory_and_does_not_extend_itself(tmp_path): + store, _, _ = populate(tmp_path) + policy = CapacityPolicy( + max_bytes=1, + low_watermark_bytes=1, + maintenance_max_deletions=1, + maintenance_interval_ms=1000, + ) + with mock.patch.object(manifest.time, "monotonic", return_value=10.0): + store.maintain(policy) + with ( + mock.patch.object(manifest.time, "monotonic", return_value=10.9), + mock.patch.object(store, "_capacity_entry", side_effect=AssertionError("scan")), + ): + report = store.maintain(policy) + assert report.skipped_cooldown and not report.skipped_busy + assert not report.capacity_satisfied + assert report.deletion_attempts == 0 + with mock.patch.object(manifest.time, "monotonic", return_value=11.0): + resumed = store.maintain(policy) + assert not resumed.skipped_cooldown + assert resumed.chunks_deleted == 1 + + +def test_root_barrier_failure_preserves_payload_and_starts_cooldown(tmp_path): + store, _, _ = populate(tmp_path, 1) + policy = CapacityPolicy( + max_bytes=1, + low_watermark_bytes=1, + maintenance_max_deletions=2, + maintenance_interval_ms=1000, + ) + with ( + mock.patch.object(manifest.time, "monotonic", return_value=10.0), + mock.patch.object(manifest, "_fsync_directory", side_effect=OSError("barrier")), + ): + with pytest.raises(OSError, match="barrier"): + store.maintain(policy) + assert len(list((tmp_path / "chunks").iterdir())) == 1 + with mock.patch.object(manifest.time, "monotonic", return_value=10.1): + assert store.maintain(policy).skipped_cooldown + + +@pytest.mark.parametrize( + "field", ["maintenance_max_deletions", "maintenance_interval_ms"] +) +@pytest.mark.parametrize("value", [-1, True, 1.5, "1"]) +def test_maintenance_controls_reject_invalid_values(field, value): + with pytest.raises(ValueError, match=field): + CapacityPolicy(**{field: value}) + + +def test_budgeted_alias_cleanup_keeps_protected_descriptor_chain(tmp_path): + from sparkcache.persistent_context_cache.test_prefix_aliases import _publish_source + from sparkcache.spark_context_cache_codec import context_prefix_digest + + store, identity, tokens, _, chunks = _publish_source( + tmp_path, chunk_count=20, prefix_tokens=(1024, 4096) + ) + digest = context_prefix_digest(tokens, identity.storage_key, token_count=4096) + protected = [EntryKey(identity.storage_key, digest, "prefix_alias")] + policy = CapacityPolicy( + max_bytes=1, low_watermark_bytes=1, maintenance_max_deletions=2 + ) + for _ in range(20): + report = store.maintain(policy, protected_entries=protected) + assert report.deletion_attempts <= 2 + lookup = store.lookup(identity, digest, storage_mode="per_token_rows") + assert lookup.is_hit, lookup.reason + assert store.restore(lookup) == chunks[:16] + if report.deletion_attempts == 0: + break + else: + pytest.fail("cleanup did not settle around the protected alias") + for _ in range(30): + report = store.maintain(policy) + assert report.deletion_attempts <= 2 + if report.capacity_satisfied and not report.work_pending: + break + else: + pytest.fail("descriptor and payload cleanup did not converge") + assert report.bytes_after == 0 + + +@pytest.mark.parametrize("budget", [1, 2, 4]) +def test_pressure_target_survives_crossing_high_watermark(tmp_path, budget): + import shutil + + seed = tmp_path / "seed" + store, _, _ = populate(seed, 20) + before = store.maintain( + CapacityPolicy(max_bytes=10**9, low_watermark_bytes=10**9) + ).bytes_after + high = before * 99 // 100 + low = before * 80 // 100 + baseline_path = tmp_path / "baseline" + shutil.copytree(seed, baseline_path) + baseline = ManifestStore(baseline_path).maintain( + CapacityPolicy(max_bytes=high, low_watermark_bytes=low) + ) + policy = CapacityPolicy( + max_bytes=high, low_watermark_bytes=low, maintenance_max_deletions=budget + ) + saw_pending_below_high = False + for _ in range(100): + report = store.maintain(policy) + saw_pending_below_high |= ( + low < report.bytes_after <= high and report.work_pending + ) + if report.capacity_satisfied and not report.work_pending: + break + else: + pytest.fail("maintenance did not reach its retained low watermark") + assert saw_pending_below_high + assert report.bytes_after == baseline.bytes_after + assert report.bytes_after <= low + assert store._maintenance_pressure_target is None diff --git a/sparkcache/request_attribution.py b/sparkcache/request_attribution.py new file mode 100644 index 0000000..3a9c2f8 --- /dev/null +++ b/sparkcache/request_attribution.py @@ -0,0 +1,122 @@ +"""Attribute accepted target-prompt work from authoritative scheduler events. + +Reuse is credited when accepted execution consumes a selected prefix. Counts +accumulate across preemption attempts and can exceed the original prompt size. +They exclude draft work, replay inside kernels, and rejected worker outputs. +""" + +from dataclasses import dataclass + + +def _count(value: object) -> int: + if type(value) is not int or value < 0: + raise ValueError("Attribution token counts and generations must be nonnegative integers") + return value + + +@dataclass +class _Attempt: + generation: int + local: int + offered_external: int + external: int = 0 + pending_restore: bool = False + consumed: bool = False + progress: int = 0 + can_readmit: bool = False + + +class RequestAttribution: + def __init__(self, prompt_tokens: int): + self.prompt_tokens = _count(prompt_tokens) + self.local_tokens_reused = 0 + self.external_tokens_reused = 0 + self.prompt_tokens_computed = 0 + self.preemptions = 0 + self.valid = True + self.prompt_completed = False + self.attempt: _Attempt | None = None + + def record(self, event: str, **fields) -> None: + generation = _count(fields.get("preemptions", 0)) + if generation < self.preemptions: + self.valid = False + return + if event == "preempted": + self.preemptions = generation + self.attempt = None + return + if event == "admitted": + local = min(_count(fields["local_tokens"]), self.prompt_tokens) + external = min(_count(fields["external_tokens"]), self.prompt_tokens - local) + if (self.attempt is not None and generation == self.attempt.generation + and not self.attempt.can_readmit): + # Resuming a parked restore reports no newly allocated prefix. + # Its original allocation and receive outcome remain authoritative. + return + if generation != self.preemptions: + self.valid = False + self.preemptions = generation + self.attempt = _Attempt(generation, local, external, + pending_restore=external > 0) + return + attempt = self.attempt + if attempt is None or generation != attempt.generation: + self.valid = False + return + if event == "restore_finalized": + prefix = min(_count(fields["valid_prefix_tokens"]), self.prompt_tokens) + if not attempt.pending_restore or attempt.consumed: + self.valid = False + return + attempt.pending_restore = False + attempt.can_readmit = fields.get("success") is False + attempt.local = min(attempt.local, prefix) + accepted = max(0, prefix - attempt.local) + if accepted > attempt.offered_external: + self.valid = False + if fields.get("success") is True: + attempt.external = min(accepted, attempt.offered_external) + elif accepted: + # A failed restore cannot prove that its residual external + # prefix passed every rank's integrity boundary. + self.valid = False + return + if event != "prompt_step_completed": + raise ValueError("Unknown request attribution event") + start = min(_count(fields["start_token"]), self.prompt_tokens) + end = min(_count(fields["end_token"]), self.prompt_tokens) + if end < start: + raise ValueError("Completed prompt interval ends before it starts") + if fields.get("stale"): + return + if attempt.pending_restore: + self.valid = False + prefix = attempt.local + attempt.external + if start > max(prefix, attempt.progress): + self.valid = False + if not attempt.consumed: + # Work that recomputes an admitted prefix must not also claim that + # same prefix as consumed reuse. + used_local = min(attempt.local, start) + used_external = min(attempt.external, max(0, start - used_local)) + self.local_tokens_reused += used_local + self.external_tokens_reused += used_external + attempt.consumed = True + self.prompt_tokens_computed += max(0, end - max(start, attempt.progress)) + attempt.progress = max(attempt.progress, end) + self.prompt_completed |= end == self.prompt_tokens + + def summary(self, status: str) -> dict: + return { + "prompt_tokens": self.prompt_tokens, + "local_tokens_reused": self.local_tokens_reused, + "external_tokens_reused": self.external_tokens_reused, + "prompt_tokens_computed": self.prompt_tokens_computed, + "preemptions": self.preemptions, + "attribution_complete": self.valid and self.prompt_completed and status in ( + "FINISHED_STOPPED", "FINISHED_LENGTH_CAPPED", + ), + "status": status, + "accounting_scope": "accepted_target_prompt_work_across_attempts", + } diff --git a/sparkcache/spark_context_cache_config.py b/sparkcache/spark_context_cache_config.py index 2d57eb4..e18889a 100644 --- a/sparkcache/spark_context_cache_config.py +++ b/sparkcache/spark_context_cache_config.py @@ -634,6 +634,16 @@ def parse_connector_config( max_bytes=max_bytes, low_watermark_bytes=low_watermark_bytes, ttl_seconds=ttl_seconds, + maintenance_max_deletions=_nonnegative_config_int( + extra("spark_cache_maintenance_max_deletions", os.environ.get( + "SPARK_CONTEXT_CACHE_MAINTENANCE_MAX_DELETIONS", "0")), + "spark_cache_maintenance_max_deletions", + ), + maintenance_interval_ms=_nonnegative_config_int( + extra("spark_cache_maintenance_interval_ms", os.environ.get( + "SPARK_CONTEXT_CACHE_MAINTENANCE_INTERVAL_MS", "0")), + "spark_cache_maintenance_interval_ms", + ), ) except ValueError as error: raise RuntimeError(f"spark-context-cache: {error}") from error diff --git a/sparkcache/spark_context_cache_connector.py b/sparkcache/spark_context_cache_connector.py index ec00a7f..7e92caa 100644 --- a/sparkcache/spark_context_cache_connector.py +++ b/sparkcache/spark_context_cache_connector.py @@ -94,6 +94,7 @@ split_snapshot, ) from sparkcache.spark_context_cache_restore_timing import RestoreTiming +from sparkcache.request_attribution import RequestAttribution from sparkcache.held_inventory import HeldInventory from sparkcache.page_base_read_flights import ( PageBaseReadEvidence, @@ -615,6 +616,10 @@ def reduce(self) -> dict[str, int | float]: ), } reduced.update({key: value for key, value in alerts.items() if value}) + for name in ("deletion_attempts", "budget_exhausted", "skipped_cooldown"): + reduced[f"sparkcache_maintenance_{name}"] = sum( + int(status.get(f"maintenance_{name}", 0)) for status in capacity + ) async_capture = [ report.get("async_capture") for report in reports @@ -793,6 +798,21 @@ class SparkCachePromMetrics(KVConnectorPromMetrics): "Physical ranks reporting an active capacity scan or reconciliation at their last worker reports.", 1.0, ), + "sparkcache_maintenance_deletion_attempts": ( + "vllm:sparkcache_maintenance_deletion_attempts", + "Reported cumulative unlink attempts summed across physical ranks; resets with workers.", + 1.0, + ), + "sparkcache_maintenance_budget_exhausted": ( + "vllm:sparkcache_maintenance_budget_exhausted", + "Reported cumulative passes deferring deletion work across physical ranks.", + 1.0, + ), + "sparkcache_maintenance_skipped_cooldown": ( + "vllm:sparkcache_maintenance_skipped_cooldown", + "Reported cumulative cooldown skips across physical ranks.", + 1.0, + ), "sparkcache_capture_delayed_requests": ( "vllm:sparkcache_capture_delayed_requests", "Maximum delayed SparkCache capture requests on any physical rank.", @@ -961,6 +981,10 @@ def __init__( ) self._config = config self._trace_reuse_enabled = os.environ.get("SPARK_CONTEXT_CACHE_TRACE_REUSE") == "1" + self.request_cache_events_enabled = ( + self._trace_reuse_enabled and role is KVConnectorRole.SCHEDULER + ) + self._request_attribution: dict[str, RequestAttribution] = {} self._block_size = config.block_size self._tp_degree = config.tp_degree self._dcp_degree = config.dcp_degree @@ -1977,6 +2001,52 @@ def _trace_reuse( json.dumps(record, sort_keys=True, separators=(",", ":")), ) + def record_request_cache_event(self, request: "Request", event: str, **fields) -> None: + """Consume optional scheduler observations without changing serving. + + Matching offers and worker-local completion traces never call this + accounting path. The runtime must report adopted prefixes, final + receive outcomes, and accepted target-model execution explicitly. + """ + if not getattr(self, "request_cache_events_enabled", False): + return + request_id = request.request_id + state = self._request_attribution.get(request_id) + try: + if event == "finished": + state = self._request_attribution.pop(request_id, None) + if state is None: + return + status = fields.get("status", getattr(request, "status", "unknown")) + status = str(getattr(status, "name", status)) + summary = state.summary(status) + category = "complete" if summary["attribution_complete"] else "incomplete" + key = f"attribution_requests_{category}" + self.counters[key] = self.counters.get(key, 0) + 1 + if summary["attribution_complete"]: + for field in ("local_tokens_reused", "external_tokens_reused", + "prompt_tokens_computed"): + key = "attribution_completed_" + field + self.counters[key] = self.counters.get(key, 0) + summary[field] + self._trace_reuse("request_cache_attribution", request_id, **summary) + return + if state is None: + if event != "admitted": + # Late receive drains for retired requests do not allocate + # another ledger or credit an aborted response. + return + prompt_tokens = getattr(request, "num_prompt_tokens", None) + if prompt_tokens is None: + prompt_tokens = len(request.prompt_token_ids or ()) + state = RequestAttribution(prompt_tokens) + self._request_attribution[request_id] = state + state.record(event, **fields) + except Exception: # Diagnostics cannot interrupt verified-or-recompute. + if state is not None: + state.valid = False + key = "attribution_invalid_events" + self.counters[key] = self.counters.get(key, 0) + 1 + def shared_prefix_lease_attached(self, request_id: str, lease_key: str) -> None: follower = self._restore_flight_followers.get(request_id) if follower is not None and follower.lease_digest == lease_key: @@ -3025,7 +3095,7 @@ def register_kv_caches(self, kv_caches: dict[str, torch.Tensor]) -> None: ): report = self._maintain_capacity(force=True) self._ensure_capacity_thread() - if (report is not None and report.skipped_busy) or not bool( + if (report is not None and (report.skipped_busy or report.skipped_cooldown)) or not bool( self._capacity_status["capacity_satisfied"] ): self._capacity_wakeup.set() @@ -3258,7 +3328,8 @@ def _maintain_capacity_locked( if not policy.enabled: return None if not force and ( - policy.max_bytes == 0 or self._capacity_estimated_bytes <= policy.max_bytes + (policy.max_bytes == 0 or self._capacity_estimated_bytes <= policy.max_bytes) + and not self._capacity_status.get("maintenance_work_pending", False) ): return None self._capacity_maintenance_depth = getattr(self, "_capacity_maintenance_depth", 0) + 1 @@ -3279,6 +3350,7 @@ def _perform_capacity_maintenance_locked( """Keep scans and survivor reconciliation within the maintenance activity gauge.""" try: with self._store_cv: + held_snapshot = (self._held, self._held.revision, set(self._held)) protected = tuple( key for result_digest, base in getattr(self, "_publication_base_pins", {}).items() @@ -3301,6 +3373,13 @@ def _perform_capacity_maintenance_locked( "spark-context-cache: capacity maintenance failed: %s", error ) return None + if report.skipped_cooldown: + self.counters["capacity_skipped_cooldown"] = ( + self.counters.get("capacity_skipped_cooldown", 0) + 1 + ) + if wake_worker_on_unsatisfied: + self._capacity_wakeup.set() + return report if report.skipped_busy: self.counters["capacity_skipped_busy"] += 1 self._capacity_status.update( @@ -3317,6 +3396,13 @@ def _perform_capacity_maintenance_locked( bytes=report.bytes_after, bytes_exact=True, capacity_satisfied=report.capacity_satisfied, + maintenance_work_pending=report.work_pending, + ) + self.counters["capacity_deletion_attempts"] = ( + self.counters.get("capacity_deletion_attempts", 0) + report.deletion_attempts + ) + self.counters["capacity_budget_exhausted"] = ( + self.counters.get("capacity_budget_exhausted", 0) + int(report.work_pending) ) self.counters["capacity_runs"] += 1 self.counters["capacity_manifests_evicted"] += report.manifests_evicted @@ -3328,59 +3414,24 @@ def _perform_capacity_maintenance_locked( self.counters["prefix_alias_segments_deleted"] += int( getattr(report, "segments_deleted", 0) ) - if report.evicted_entries: + if report.surviving_entries is None: + # A report without an inventory cannot authorize withdrawals. + self._reconcile_held_capacity() + elif held_snapshot[2]: identity = self._identity(self._worker_rank()) - withdrawn = set() - candidates: dict[str, set[str]] = {} - for entry in report.evicted_entries: - if entry.storage_key != identity.storage_key: - continue - candidates.setdefault(entry.context_digest, set()).add( - getattr(entry, "root_kind", "manifest") - ) - for digest, evicted_roots in candidates.items(): - exact_exists = ( - "manifest" not in evicted_roots - and ( - Path(self._root) - / "manifests" - / identity.storage_key - / f"{digest}.json" - ).exists() - ) - alias_exists = ( - self._storage_mode == "per_token_rows" - and "prefix_alias" not in evicted_roots - and ( - Path(self._root) - / "prefix-aliases" - / identity.storage_key - / f"{digest}.json" - ).exists() - ) - if not exact_exists and not alias_exists: - withdrawn.add(digest) - continue - lookup, _is_alias = self._lookup_reusable( - identity, - digest, - verify_chunks=False, - verify_chunk_metadata=True, - ) - if not lookup.is_hit: - withdrawn.add(digest) - with self._store_cv: - self._held.difference_update(withdrawn) - # One digest can name both an exact manifest and its source-boundary - # alias. The targeted checks above retain the offer when either root - # remains; this complete pass also catches entries removed as debris. - self._reconcile_held_capacity() - if not report.capacity_satisfied and wake_worker_on_unsatisfied: + surviving = { + entry.context_digest for entry in report.surviving_entries + if entry.storage_key == identity.storage_key + and (entry.root_kind == "manifest" or self._storage_mode == "per_token_rows") + } + self._apply_held_capacity_survivors(held_snapshot, surviving) + if (not report.capacity_satisfied or report.work_pending) and wake_worker_on_unsatisfied: self._capacity_wakeup.set() if force or report.bytes_reclaimed: logger.info( "spark-context-cache: capacity bytes=%d max=%d reclaimed=%d" - " manifests=%d chunks=%d orphans=%d satisfied=%s", + " manifests=%d chunks=%d orphans=%d satisfied=%s" + " deletion_attempts=%d work_pending=%s", report.bytes_after, policy.max_bytes, report.bytes_reclaimed, @@ -3388,6 +3439,8 @@ def _perform_capacity_maintenance_locked( report.chunks_deleted, report.orphan_chunks_deleted, report.capacity_satisfied, + report.deletion_attempts, + report.work_pending, ) return report @@ -3412,7 +3465,7 @@ def _post_commit_was_evicted_locked( force=force_maintenance, wake_worker_on_unsatisfied=True, ) - if report is not None and not report.skipped_busy: + if report is not None and not (report.skipped_busy or report.skipped_cooldown): exact_evicted = ( EntryKey( identity.storage_key, @@ -3640,7 +3693,7 @@ def _finalize_streaming_capacity_commits( report = self._maintain_capacity_locked(force=True) if ( report is None - or report.skipped_busy + or (report.skipped_busy or report.skipped_cooldown) or not report.capacity_satisfied ): return False @@ -3684,15 +3737,32 @@ def _finalize_streaming_capacity_commits( self.counters["streaming_store_evicted"] += len(evicted) return True + def _apply_held_capacity_survivors( + self, + snapshot: tuple[HeldInventory, int, set[str]], + surviving: set[str], + ) -> None: + inventory, revision, held = snapshot + with self._store_cv: + if self._held is not inventory or self._held.revision != revision: + # A digest can be withdrawn and republished during a probe. + # An earlier inventory cannot revoke that publication; a + # stable pass reconciles it, and restores still verify bytes. + self.counters["capacity_stale_inventory_snapshots"] = ( + self.counters.get("capacity_stale_inventory_snapshots", 0) + 1 + ) + return + self._held.difference_update(held - surviving) + def _reconcile_held_capacity(self) -> None: - if not self._held: + with self._store_cv: + snapshot = (self._held, self._held.revision, set(self._held)) + if not snapshot[2]: return rank = self._worker_rank() identity = self._identity(rank) - with self._store_cv: - held = set(self._held) surviving = set() - for digest in held: + for digest in snapshot[2]: lookup, _is_alias = self._lookup_reusable( identity, digest, @@ -3701,8 +3771,7 @@ def _reconcile_held_capacity(self) -> None: ) if lookup.is_hit: surviving.add(digest) - with self._store_cv: - self._held.intersection_update(surviving) + self._apply_held_capacity_survivors(snapshot, surviving) def _ensure_capacity_thread(self) -> None: if self._capacity_thread is not None: @@ -3752,7 +3821,9 @@ def _capacity_worker_main(self) -> None: with self._capacity_handoff_cv: self._streaming_capacity_pending.difference_update(resolved) self._capacity_handoff_cv.notify_all() - retry_unsatisfied = False + retry_unsatisfied = bool( + self._capacity_status.get("maintenance_work_pending", False) + ) else: self.counters["streaming_capacity_retries"] += 1 retry_unsatisfied = True @@ -3765,6 +3836,7 @@ def _capacity_worker_main(self) -> None: retry_unsatisfied = bool( report is None or not report.capacity_satisfied + or report.work_pending or not bool(self._capacity_status["capacity_satisfied"]) ) if retry_unsatisfied: @@ -4940,6 +5012,7 @@ def request_finished( request: "Request", block_ids: list[int], ) -> tuple[bool, dict[str, Any] | None]: + self.record_request_cache_event(request, "finished") # A finished request id never recurs, so its scheduler-side tracking # state is dropped here. This is what keeps _need_load, _admitted, # and _store_progress bounded without evicting live entries. @@ -6458,6 +6531,10 @@ def get_kv_connector_stats(self): "streaming_capacity_shutdown_dropped" ], maintenance_retries=self.counters["capacity_retries"], + maintenance_deletion_attempts=self.counters.get("capacity_deletion_attempts", 0), + maintenance_stale_inventory_snapshots=self.counters.get("capacity_stale_inventory_snapshots", 0), + maintenance_budget_exhausted=self.counters.get("capacity_budget_exhausted", 0), + maintenance_skipped_cooldown=self.counters.get("capacity_skipped_cooldown", 0), ) report["capacity"] = capacity runtime = self._async_page_capture_runtime diff --git a/sparkcache/test_maintenance_pacing.py b/sparkcache/test_maintenance_pacing.py new file mode 100644 index 0000000..3cca4b4 --- /dev/null +++ b/sparkcache/test_maintenance_pacing.py @@ -0,0 +1,81 @@ +"""Forced maintenance respects cooldown without repeating survivor scans.""" + +from unittest import mock + +from sparkcache import test_spark_context_cache_connector as fixtures +from sparkcache.persistent_context_cache.cache_manifest import ( + CapacityPolicy, + MaintenanceReport, +) + + +def test_forced_post_commit_cooldown_does_not_reconcile_inventory(tmp_path): + connector = fixtures.AsyncRestoreTests()._cohort_connector(tmp_path) + connector._capacity_policy = CapacityPolicy( + max_bytes=10, low_watermark_bytes=8, maintenance_interval_ms=1000 + ) + connector._capacity_estimated_bytes = 11 + connector._capacity_status["capacity_satisfied"] = False + try: + with ( + mock.patch.object( + connector._store, + "maintain", + return_value=MaintenanceReport( + capacity_satisfied=False, skipped_cooldown=True + ), + ), + mock.patch.object(connector, "_reconcile_held_capacity") as reconcile, + ): + report = connector._maintain_capacity(force=True) + assert report.skipped_cooldown + assert not connector._capacity_status["capacity_satisfied"] + assert connector._capacity_estimated_bytes == 11 + assert connector.counters["capacity_skipped_cooldown"] == 1 + reconcile.assert_not_called() + finally: + connector.shutdown() + + +def test_pacing_totals_aggregate_physical_ranks(): + from sparkcache.spark_context_cache_connector import SparkCacheStats + + stats = SparkCacheStats( + data={ + "reports": [ + { + "rank": rank, + "capacity": { + "maintenance_deletion_attempts": 128, + "maintenance_budget_exhausted": rank, + "maintenance_skipped_cooldown": 2, + }, + } + for rank in range(4) + ] + } + ) + reduced = stats.reduce() + assert reduced["sparkcache_maintenance_deletion_attempts"] == 512 + assert reduced["sparkcache_maintenance_budget_exhausted"] == 6 + assert reduced["sparkcache_maintenance_skipped_cooldown"] == 8 + + +def test_latched_pressure_runs_maintenance_below_high_watermark(tmp_path): + connector = fixtures.AsyncRestoreTests()._cohort_connector(tmp_path) + connector._capacity_policy = CapacityPolicy( + max_bytes=10, low_watermark_bytes=8, maintenance_max_deletions=1 + ) + connector._capacity_estimated_bytes = 9 + connector._capacity_status["maintenance_work_pending"] = True + try: + with mock.patch.object( + connector, + "_perform_capacity_maintenance_locked", + return_value=MaintenanceReport(work_pending=True), + ) as run: + report = connector._maintain_capacity(force=False) + run.assert_called_once() + assert report.work_pending + finally: + connector.shutdown() diff --git a/sparkcache/test_maintenance_survivors.py b/sparkcache/test_maintenance_survivors.py new file mode 100644 index 0000000..76cba39 --- /dev/null +++ b/sparkcache/test_maintenance_survivors.py @@ -0,0 +1,186 @@ +"""A maintenance inventory qualifies offers without a second filesystem walk.""" + +from collections import Counter +from pathlib import Path + +import pytest + +from sparkcache.persistent_context_cache.cache_manifest import ( + CapacityPolicy, EntryKey, LookupResult, ManifestStore, +) +from sparkcache.persistent_context_cache.test_cache_manifest import _chunk, _identity +from sparkcache.persistent_context_cache.test_prefix_aliases import _publish_source +from sparkcache.test_spark_context_cache_connector import _make_connector + + +POLICY = CapacityPolicy(max_bytes=10**9, low_watermark_bytes=10**9) + + +@pytest.fixture +def connector(tmp_path): + connector = _make_connector(tmp_path, 0) + identity = _identity() + connector._identity = lambda rank: identity + connector._capacity_policy = POLICY + yield connector + connector.shutdown() + + +def populate(connector, roots=12, chunks=8): + identity = connector._identity(0) + payloads = tuple(_chunk(index * 256, (index + 1) * 256) for index in range(chunks)) + digests = {f"{index + 1:064x}" for index in range(roots)} + for digest in digests: + connector._store.commit(identity=identity, context_digest=digest, chunks=payloads) + connector._held.update(digests) + return digests + + +def test_normal_maintenance_reads_each_root_once_and_stats_unique_chunks(connector, monkeypatch): + digests = populate(connector) + counts = Counter() + read_bytes, stat = Path.read_bytes, Path.stat + + def read(path): + if path.suffix == ".json": + counts["roots"] += 1 + return read_bytes(path) + + def metadata(path, *args, **kwargs): + if path.parent.name == "chunks" and path.suffix == ".spcc": + counts["chunks"] += 1 + return stat(path, *args, **kwargs) + + monkeypatch.setattr(Path, "read_bytes", read) + monkeypatch.setattr(Path, "stat", metadata) + monkeypatch.setattr(connector, "_lookup_reusable", + lambda *args, **kwargs: pytest.fail("inventory was probed twice")) + report = connector._maintain_capacity(force=True) + assert report.surviving_entries is not None + assert set(connector._held) == digests + assert counts["roots"] == 12 + # Enumeration tests file type and obtains allocated/logical size once. + assert counts["chunks"] <= 2 * 8 + + +@pytest.mark.parametrize("damage", ["missing", "truncated", "same_size"]) +def test_shared_payload_metadata_qualifies_every_offer(connector, damage): + digests = populate(connector, roots=3, chunks=1) + chunk = next((Path(connector._root) / "chunks").glob("*.spcc")) + encoded = chunk.read_bytes() + if damage == "missing": + chunk.unlink() + elif damage == "truncated": + chunk.write_bytes(encoded[:-1]) + else: + chunk.write_bytes(encoded[:-1] + bytes([encoded[-1] ^ 1])) + connector._maintain_capacity(force=True) + assert set(connector._held) == (digests if damage == "same_size" else set()) + # Metadata qualification never substitutes for restore integrity checks. + for digest in digests: + assert not connector._store.lookup(connector._identity(0), digest).is_hit + + +def test_corrupt_exact_root_shadows_valid_alias_until_removed(tmp_path, monkeypatch): + store, identity, _, digest, _ = _publish_source(tmp_path, chunk_count=4, + prefix_tokens=(1024,)) + exact = store._manifest_path(identity, digest) + exact.write_bytes(b"{}") + unlink = Path.unlink + + def refuse_exact(path, *args, **kwargs): + if path == exact: + raise PermissionError("root retained") + return unlink(path, *args, **kwargs) + + monkeypatch.setattr(Path, "unlink", refuse_exact) + report = store.maintain(POLICY) + assert report.surviving_entries == () + assert not store.lookup(identity, digest, verify_chunks=False, + verify_chunk_metadata=True, storage_mode="per_token_rows").is_hit + monkeypatch.setattr(Path, "unlink", unlink) + report = store.maintain(POLICY) + assert EntryKey(identity.storage_key, digest, "prefix_alias") in report.surviving_entries + assert store.lookup(identity, digest, storage_mode="per_token_rows").is_hit + + +def test_alias_offers_are_excluded_from_block_page_connector(connector, tmp_path): + store, identity, _, digest, _ = _publish_source(tmp_path, chunk_count=4, + prefix_tokens=(1024,)) + store._manifest_path(identity, digest).unlink() + connector._identity = lambda rank: identity + connector._held.add(digest) + connector._storage_mode = "block_pages" + connector._maintain_capacity(force=True) + assert digest not in connector._held + connector._storage_mode = "per_token_rows" + connector._held.add(digest) + connector._maintain_capacity(force=True) + assert digest in connector._held + + +@pytest.mark.parametrize("mutation", ["add", "replace_inventory", "republish_same_digest"]) +def test_inventory_mutations_are_not_withdrawn_from_a_stale_pass(connector, monkeypatch, mutation): + stale = "a" * 64 + fresh = "b" * 64 + connector._held.add(stale) + maintain = connector._store.maintain + + def race(*args, **kwargs): + report = maintain(*args, **kwargs) + with connector._store_cv: + if mutation == "add": + connector._held.add(fresh) + elif mutation == "replace_inventory": + connector._held = {fresh} + else: + connector._held.remove(stale) + connector._held.add(stale) + return report + + monkeypatch.setattr(connector._store, "maintain", race) + connector._maintain_capacity(force=True) + assert fresh in connector._held if mutation != "republish_same_digest" else stale in connector._held + assert connector.counters["capacity_stale_inventory_snapshots"] == 1 + monkeypatch.setattr(connector._store, "maintain", maintain) + connector._maintain_capacity(force=True) + assert not connector._held + + +def test_fallback_probe_preserves_same_digest_republication(connector, monkeypatch): + digest = "a" * 64 + connector._held.add(digest) + + def changed(*args, **kwargs): + with connector._store_cv: + connector._held.remove(digest) + connector._held.add(digest) + return LookupResult(False, "absent"), False + + monkeypatch.setattr(connector, "_lookup_reusable", changed) + connector._reconcile_held_capacity() + assert digest in connector._held + assert connector.counters["capacity_stale_inventory_snapshots"] == 1 + + +def test_other_store_publications_are_observed_on_each_inventory(connector): + root = Path(connector._root) + assert connector._maintain_capacity(force=True).surviving_entries == () + other = ManifestStore(root) + digest = "a" * 64 + identity = connector._identity(0) + other.commit(identity=identity, context_digest=digest, chunks=[_chunk()]) + connector._held.add(digest) + report = connector._maintain_capacity(force=True) + assert report.surviving_entries == (EntryKey(identity.storage_key, digest),) + assert digest in connector._held + other._manifest_path(identity, digest).write_bytes(b"{}") + connector._maintain_capacity(force=True) + assert digest not in connector._held + + +def test_empty_offer_inventory_does_not_require_initialized_model_identity(connector, monkeypatch): + monkeypatch.setattr(connector, "_identity", + lambda rank: pytest.fail("model identity is not initialized")) + assert connector._maintain_capacity(force=True).surviving_entries == () + connector._reconcile_held_capacity() diff --git a/sparkcache/test_request_attribution.py b/sparkcache/test_request_attribution.py new file mode 100644 index 0000000..b020045 --- /dev/null +++ b/sparkcache/test_request_attribution.py @@ -0,0 +1,141 @@ +"""Prompt attribution follows accepted execution, not matching offers.""" + +import pytest + +from sparkcache.request_attribution import RequestAttribution + + +def admit(state, local=0, external=0, generation=0): + state.record("admitted", local_tokens=local, external_tokens=external, + preemptions=generation, source="prefix_lookup") + + +def completed(state, start, end, generation=0): + state.record("prompt_step_completed", start_token=start, end_token=end, + preemptions=generation) + + +def test_local_partial_tail_uses_final_scheduler_count(): + state = RequestAttribution(1100) + admit(state, local=1031) + completed(state, 1031, 1100) + assert state.summary("FINISHED_STOPPED") == { + "prompt_tokens": 1100, "local_tokens_reused": 1031, + "external_tokens_reused": 0, "prompt_tokens_computed": 69, + "preemptions": 0, "attribution_complete": True, + "status": "FINISHED_STOPPED", + "accounting_scope": "accepted_target_prompt_work_across_attempts", + } + + +def test_external_offer_without_finalization_is_not_reuse(): + state = RequestAttribution(1100) + admit(state, local=256, external=768) + completed(state, 256, 1100) + result = state.summary("FINISHED_STOPPED") + assert result["external_tokens_reused"] == 0 + assert not result["attribution_complete"] + + +def test_verified_restore_counts_only_effective_prompt_reuse_once(): + state = RequestAttribution(1024) + admit(state, local=256, external=768) + state.record("restore_finalized", valid_prefix_tokens=1023, + success=True, preemptions=0) + # Re-admission of the parked request does not replace its source accounting. + admit(state) + completed(state, 1023, 1024) + completed(state, 1023, 1024) + result = state.summary("FINISHED_STOPPED") + assert (result["local_tokens_reused"], result["external_tokens_reused"], + result["prompt_tokens_computed"]) == (256, 767, 1) + assert result["attribution_complete"] + + +def test_failed_restore_recomputes_and_does_not_credit_external_offer(): + state = RequestAttribution(1100) + admit(state, local=256, external=768) + state.record("restore_finalized", valid_prefix_tokens=256, + success=False, preemptions=0) + completed(state, 256, 1100) + result = state.summary("FINISHED_STOPPED") + assert result["external_tokens_reused"] == 0 + assert result["prompt_tokens_computed"] == 844 + assert result["attribution_complete"] + + +def test_gpu_lease_is_local_reuse_without_external_credit(): + state = RequestAttribution(1024) + state.record("admitted", local_tokens=1023, external_tokens=0, + preemptions=0, source="gpu_lease") + completed(state, 1023, 1024) + result = state.summary("FINISHED_STOPPED") + assert result["local_tokens_reused"] == 1023 + assert result["external_tokens_reused"] == 0 + + +def test_failed_private_restore_allows_fresh_local_lookup_in_same_generation(): + state = RequestAttribution(1100) + admit(state, external=1024) + state.record("restore_finalized", valid_prefix_tokens=0, + success=False, preemptions=0) + admit(state, local=768) + completed(state, 768, 1100) + result = state.summary("FINISHED_STOPPED") + assert result["external_tokens_reused"] == 0 + assert result["local_tokens_reused"] == 768 + assert result["prompt_tokens_computed"] == 332 + assert result["attribution_complete"] + + +def test_preemption_keeps_completed_work_and_counts_recomputation(): + state = RequestAttribution(1000) + admit(state, local=256) + completed(state, 256, 512) + state.record("preempted", preemptions=1) + admit(state, generation=1) + completed(state, 0, 1000, generation=1) + result = state.summary("FINISHED_STOPPED") + assert result["local_tokens_reused"] == 256 + assert result["prompt_tokens_computed"] == 1256 + assert result["preemptions"] == 1 + assert result["attribution_complete"] + + +def test_abort_before_execution_does_not_credit_verified_cache_warming(): + state = RequestAttribution(1024) + admit(state, external=1024) + state.record("restore_finalized", valid_prefix_tokens=1023, + success=True, preemptions=0) + result = state.summary("FINISHED_ABORTED") + assert result["external_tokens_reused"] == 0 + assert result["prompt_tokens_computed"] == 0 + assert not result["attribution_complete"] + + +def test_decode_after_preemption_credits_resident_prompt_without_output_tokens(): + state = RequestAttribution(1000) + admit(state) + completed(state, 0, 1000) + state.record("preempted", preemptions=1) + admit(state, local=1000, generation=1) + completed(state, 1000, 1000, generation=1) + result = state.summary("FINISHED_STOPPED") + assert result["local_tokens_reused"] == 1000 + assert result["prompt_tokens_computed"] == 1000 + + +def test_missing_attempt_and_unobserved_prefix_do_not_claim_complete_attribution(): + state = RequestAttribution(1000) + completed(state, 200, 1000) + assert not state.summary("FINISHED_STOPPED")["attribution_complete"] + + +@pytest.mark.parametrize("field,value", [("local_tokens", -1), + ("external_tokens", True), + ("preemptions", -1)]) +def test_invalid_scheduler_observation_is_rejected(field, value): + fields = dict(local_tokens=0, external_tokens=0, preemptions=0) + fields[field] = value + with pytest.raises(ValueError): + RequestAttribution(1000).record("admitted", **fields) diff --git a/sparkcache/test_reuse_trace.py b/sparkcache/test_reuse_trace.py index e143df5..d3a16e5 100644 --- a/sparkcache/test_reuse_trace.py +++ b/sparkcache/test_reuse_trace.py @@ -122,3 +122,59 @@ def fail(*args, **kwargs): assert connector.counters["shared_prefix_leases_attached"] == 1 finally: connector.shutdown() + + +def test_request_attribution_credits_scheduler_events_and_cleans_up(tmp_path, monkeypatch): + monkeypatch.setenv("SPARK_CONTEXT_CACHE_TRACE_REUSE", "1") + records = _records(monkeypatch) + connector = fixtures._make_connector(tmp_path, 0, role=connector_module.KVConnectorRole.SCHEDULER) + request = SimpleNamespace(request_id="attributed", num_prompt_tokens=1100, + status=SimpleNamespace(name="FINISHED_STOPPED")) + try: + assert connector.request_cache_events_enabled + connector.record_request_cache_event(request, "admitted", local_tokens=1031, + external_tokens=0, preemptions=0) + connector.record_request_cache_event(request, "prompt_step_completed", + start_token=1031, end_token=1100, preemptions=0) + connector.record_request_cache_event(request, "finished") + connector.record_request_cache_event(request, "finished") + connector.record_request_cache_event(request, "restore_finalized", + success=True, valid_prefix_tokens=1024) + assert connector._request_attribution == {} + assert len(records) == 1 + assert records[0]["local_tokens_reused"] == 1031 + assert records[0]["prompt_tokens_computed"] == 69 + assert records[0]["external_tokens_reused"] == 0 + assert records[0]["attribution_complete"] + assert connector.counters["attribution_requests_complete"] == 1 + finally: + connector.shutdown() + + +def test_request_attribution_invalid_event_and_log_failure_do_not_affect_serving(tmp_path, monkeypatch): + monkeypatch.setenv("SPARK_CONTEXT_CACHE_TRACE_REUSE", "1") + connector = fixtures._make_connector(tmp_path, 0, role=connector_module.KVConnectorRole.SCHEDULER) + request = SimpleNamespace(request_id="invalid", num_prompt_tokens=100, + status=SimpleNamespace(name="FINISHED_ABORTED")) + try: + connector.record_request_cache_event(request, "admitted", local_tokens=-1, + external_tokens=0, preemptions=0) + assert connector.counters["attribution_invalid_events"] == 1 + monkeypatch.setattr(connector_module.logger, "info", + lambda *args, **kwargs: (_ for _ in ()).throw(OSError("sink failed"))) + connector.record_request_cache_event(request, "finished") + assert connector._request_attribution == {} + assert connector.counters["attribution_requests_incomplete"] == 1 + finally: + connector.shutdown() + + +def test_request_attribution_disabled_has_no_per_request_state(tmp_path, monkeypatch): + monkeypatch.delenv("SPARK_CONTEXT_CACHE_TRACE_REUSE", raising=False) + connector = fixtures._make_connector(tmp_path, 0, role=connector_module.KVConnectorRole.SCHEDULER) + try: + assert not connector.request_cache_events_enabled + connector.record_request_cache_event(SimpleNamespace(request_id="off"), "admitted") + assert connector._request_attribution == {} + finally: + connector.shutdown() diff --git a/sparkcache/test_spark_context_cache_config.py b/sparkcache/test_spark_context_cache_config.py index 1e3ce8f..86baf6f 100644 --- a/sparkcache/test_spark_context_cache_config.py +++ b/sparkcache/test_spark_context_cache_config.py @@ -297,6 +297,8 @@ def test_capacity_policy_from_env(self) -> None: "SPARK_CONTEXT_CACHE_MAX_BYTES": "1000000", "SPARK_CONTEXT_CACHE_LOW_WATERMARK_BYTES": "900000", "SPARK_CONTEXT_CACHE_TTL_SECONDS": "3600", + "SPARK_CONTEXT_CACHE_MAINTENANCE_MAX_DELETIONS": "128", + "SPARK_CONTEXT_CACHE_MAINTENANCE_INTERVAL_MS": "1000", "SPARK_CONTEXT_CACHE_MODEL_PROFILE": "glm52-nvfp4", "SPARK_CONTEXT_CACHE_TARGET_CHECKPOINT_SHA256": "1" * 64, "SPARK_CONTEXT_CACHE_DRAFT_CHECKPOINT_SHA256": "2" * 64, @@ -307,6 +309,8 @@ def test_capacity_policy_from_env(self) -> None: self.assertEqual(config.capacity_policy.max_bytes, 1000000) self.assertEqual(config.capacity_policy.low_watermark_bytes, 900000) self.assertEqual(config.capacity_policy.ttl_seconds, 3600) + self.assertEqual(config.capacity_policy.maintenance_max_deletions, 128) + self.assertEqual(config.capacity_policy.maintenance_interval_ms, 1000) def test_missing_model_profile_is_rejected_explicitly(self) -> None: vllm, _ = _make_vllm_config() diff --git a/sparkcache/test_spark_context_cache_connector.py b/sparkcache/test_spark_context_cache_connector.py index 6aaade9..6e91949 100644 --- a/sparkcache/test_spark_context_cache_connector.py +++ b/sparkcache/test_spark_context_cache_connector.py @@ -3548,6 +3548,7 @@ def test_eviction_withdraws_held_digest_and_updates_capacity_metrics(self) -> No manifests_evicted=1, chunks_deleted=2, evicted_entries=(EntryKey(storage_key, removed),), + surviving_entries=(EntryKey(storage_key, survivor),), ) ) connector._reconcile_held_capacity = mock.Mock() diff --git a/tools/benchmark_maintenance_inventory.py b/tools/benchmark_maintenance_inventory.py new file mode 100644 index 0000000..a1f8909 --- /dev/null +++ b/tools/benchmark_maintenance_inventory.py @@ -0,0 +1,108 @@ +"""Measure complete maintenance and offer reconciliation using GPU-free fixtures. + +Run from the checkout being measured. Positional arguments specify root/chunk +counts for BENCH_KIND=rows (default), or branch/extension counts for pages. +BENCH_FILESYSTEM records the independently verified filesystem type. Test +fixtures supply connector API stubs; CUDA and model execution are not used. +""" +from collections import Counter +import gc +import os +import platform +import hashlib +import json +from pathlib import Path +import statistics +import subprocess +import sys +import tempfile +import time +from unittest.mock import patch + +sys.path.insert(0, str(Path.cwd())) +from sparkcache.persistent_context_cache.cache_manifest import CapacityPolicy +from sparkcache.persistent_context_cache.test_cache_manifest import _chunk, _identity +from sparkcache.test_spark_context_cache_connector import _make_connector + +roots, chunks = map(int, sys.argv[1:3]) +with tempfile.TemporaryDirectory(prefix="sparkcache-inventory-bench-") as directory: + connector = _make_connector(Path(directory), 0) + identity = _identity() + connector._identity = lambda rank: identity + connector._capacity_policy = CapacityPolicy(max_bytes=10**9, low_watermark_bytes=10**9) + kind = os.environ.get("BENCH_KIND", "rows") + if kind == "pages": + from sparkcache.persistent_context_cache.cache_manifest import CacheIdentity + from sparkcache.spark_context_cache_codec import context_prefix_digest + from sparkcache.spark_context_cache_hybrid import PageGroup, PageLayer, PageLayout, encode_page_snapshot + identity = CacheIdentity(target_checkpoint="1" * 64, draft_checkpoint="2" * 64, + quantization_layout="benchmark-page", rope_layout="glm53-hybrid-v1", + tp_degree=4, dcp_degree=4, chunk_tokens=256, + record_schema=("target_ckv", "logical_positions"), + publication_schema="page-tail-cow-v2") + connector._identity = lambda rank: identity + connector._storage_mode = "block_pages" + layout = PageLayout((PageGroup(256, (PageLayer("attention", "u8", (64,), 64),)), + PageGroup(1, (PageLayer("recurrent", "u8", (32,), 32),)))) + tokens_base = tuple(range(512)) + salt = "shared-base-page-profile" + base_digest = context_prefix_digest(tokens_base, salt, token_count=512) + connector._store.commit_page_snapshot(identity=identity, context_digest=base_digest, + span_tokens=512, snapshot=encode_page_snapshot(layout, (2, 1), + {"attention": b"a" * 128, "recurrent": b"r" * 32})) + for branch in range(roots): + tokens = tokens_base + tuple(range(100000 * (branch + 1), 100000 * (branch + 1) + chunks * 256)) + prior = base_digest + payload = b"a" * 128 + for stage in range(chunks): + payload += bytes((stage + branch,)) * 64 + boundary = (stage + 3) * 256 + snapshot = encode_page_snapshot(layout, (stage + 3, 1), { + "attention": payload, "recurrent": bytes((stage + 64,)) * 32}) + connector._store.commit_page_extension(identity=identity, base_context_digest=prior, + token_ids=tokens, identity_salt=salt, layout=layout, + base_block_counts=(stage + 2, 1), result_block_counts=(stage + 3, 1), + base_boundary_tokens=(stage + 2) * 256, + result_boundary_tokens=boundary, result_snapshot=snapshot) + prior = context_prefix_digest(tokens, salt, token_count=boundary) + digests = {path.stem for path in (Path(directory) / "manifests" / identity.storage_key).glob("*.json")} + references = sum(len(connector._store._capacity_entry(path).chunks) + for path in (Path(directory) / "manifests" / identity.storage_key).glob("*.json")) + else: + payloads = tuple(_chunk(i * 256, (i + 1) * 256) for i in range(chunks)) + digests = {f"{i + 1:064x}" for i in range(roots)} + for digest in digests: + connector._store.commit(identity=identity, context_digest=digest, chunks=payloads) + references = roots * chunks + unique_chunks = len(list((Path(directory) / "chunks").glob("*.spcc"))) + connector._held.update(digests) + counts = Counter() + read_bytes, stat = Path.read_bytes, Path.stat + + def read(path): + if path.suffix == ".json": + counts["root_reads"] += 1 + return read_bytes(path) + + def metadata(path, *args, **kwargs): + if path.parent.name == "chunks" and path.suffix == ".spcc": + counts["chunk_stats"] += 1 + return stat(path, *args, **kwargs) + + with patch.object(Path, "read_bytes", read), patch.object(Path, "stat", metadata): + connector._maintain_capacity(force=True) + elapsed = [] + for _ in range(7): + gc.collect() + start = time.perf_counter() + report = connector._maintain_capacity(force=True) + elapsed.append((time.perf_counter() - start) * 1000) + assert set(connector._held) == digests + assert report.manifests_evicted == report.chunks_deleted == 0 + connector.shutdown() + print(json.dumps({"revision": subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip(), + "kind": kind, "roots": len(digests), "unique_chunks": unique_chunks, "references": references, + "platform": platform.platform(), "filesystem": os.environ.get("BENCH_FILESYSTEM", "unspecified"), + "manifest_source_sha256": hashlib.sha256(Path('sparkcache/persistent_context_cache/cache_manifest.py').read_bytes().replace(b'\r\n', b'\n')).hexdigest(), + "connector_source_sha256": hashlib.sha256(Path('sparkcache/spark_context_cache_connector.py').read_bytes().replace(b'\r\n', b'\n')).hexdigest(), + "io_counts": counts, "elapsed_ms": elapsed, "median_ms": statistics.median(elapsed)}))