From 5ec6a9953ad5d39120298bbfc26e95a6fa4b1dc3 Mon Sep 17 00:00:00 2001 From: FujitsuPolycom <87842395+FujitsuPolycom@users.noreply.github.com> Date: Sun, 30 Aug 2026 00:15:28 -0500 Subject: [PATCH] Read page-delta chunks concurrently --- deploy/deepseek_v4/tp4_profile.json | 2 +- deploy/glm52_35bpw/profile.json | 2 +- .../cache_manifest.py | 11 +++--- .../test_cache_manifest.py | 35 +++++++++++++++++++ 4 files changed, 44 insertions(+), 6 deletions(-) diff --git a/deploy/deepseek_v4/tp4_profile.json b/deploy/deepseek_v4/tp4_profile.json index 35f820f..7377823 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": "bc7cae86732c869ee8b2205d48ac5be6f580ee8b77a3e4ffd4c69dcd4f1bfae5" + "source_sha256": "bc238f96e550c7ec27d4081dd1f2e741d404aaf5c8572d89ccc5e76812be4d63" }, "model": { "repository": "deepseek-ai/DeepSeek-V4-Flash-0731", diff --git a/deploy/glm52_35bpw/profile.json b/deploy/glm52_35bpw/profile.json index 312be95..14dc752 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": "bc7cae86732c869ee8b2205d48ac5be6f580ee8b77a3e4ffd4c69dcd4f1bfae5" + "source_sha256": "bc238f96e550c7ec27d4081dd1f2e741d404aaf5c8572d89ccc5e76812be4d63" }, "model": { "repository": "brandonmusic/GLM-5.2-EXL3-TR3v4-3.5bpw-MTP78", diff --git a/sparkcache/persistent_context_cache/cache_manifest.py b/sparkcache/persistent_context_cache/cache_manifest.py index fcf8d85..764aa63 100644 --- a/sparkcache/persistent_context_cache/cache_manifest.py +++ b/sparkcache/persistent_context_cache/cache_manifest.py @@ -2263,8 +2263,7 @@ def _read_context_chunks( descriptors: Sequence[Mapping[str, Any]], required: frozenset[StateRecord], ) -> tuple[ContextChunk, ...]: - result = [] - for descriptor in descriptors: + def _read_one(descriptor: Mapping[str, Any]) -> ContextChunk: encoded = ( self.root / "chunks" / f"{descriptor['sha256']}.spcc" ).read_bytes() @@ -2280,8 +2279,12 @@ def _read_context_chunks( or chunk.logical_end != descriptor["logical_end"] ): raise CacheFormatError("chunk range disagrees with descriptor") - result.append(chunk) - return tuple(result) + return chunk + + if not descriptors: + return () + with ThreadPoolExecutor(max_workers=min(8, len(descriptors))) as pool: + return tuple(pool.map(_read_one, descriptors)) def publish_prefix_aliases( self, diff --git a/sparkcache/persistent_context_cache/test_cache_manifest.py b/sparkcache/persistent_context_cache/test_cache_manifest.py index a028828..cf7dfac 100644 --- a/sparkcache/persistent_context_cache/test_cache_manifest.py +++ b/sparkcache/persistent_context_cache/test_cache_manifest.py @@ -98,6 +98,41 @@ def _clear_once_in_subprocess( class ManifestStoreTests(unittest.TestCase): + def test_page_delta_chunk_reads_overlap_and_preserve_descriptor_order( + self, + ) -> None: + identity = _identity() + chunks = (_chunk(0, 256), _chunk(256, 512)) + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + store = ManifestStore(root) + digest = hashlib.sha256(b"parallel-page-delta-reads").hexdigest() + store.commit( + identity=identity, + context_digest=digest, + chunks=chunks, + span_tokens=512, + ) + manifest_path = ( + root / "manifests" / identity.storage_key / f"{digest}.json" + ) + descriptors = json.loads(manifest_path.read_bytes())["chunks"] + overlap = threading.Barrier(len(descriptors), timeout=2.0) + original_read_bytes = Path.read_bytes + + def read_with_overlap(path: Path) -> bytes: + if path.suffix == ".spcc": + overlap.wait() + return original_read_bytes(path) + + with mock.patch.object(Path, "read_bytes", read_with_overlap): + restored = store._read_context_chunks( + descriptors, + identity.required_records, + ) + + self.assertEqual(restored, chunks) + def test_page_extension_materializes_full_snapshot_after_base_root_removal( self, ) -> None: