Cache and parallelism benchmark¶
This notebook demonstrates the efficiency and reliability improvements of the SQLite/WAL cache compared with the original JSONL snapshot design. It uses deterministic simulated work instead of paid LLM calls.
The demonstrations cover:
- incremental write scaling;
- multiprocess integrity;
- duplicate LLM-call avoidance;
- interrupted-worker recovery;
- configuration mismatch diagnostics; and
- provenance inspection.
Performance varies by filesystem and machine. Run each benchmark several times and report medians, call counts, and integrity results rather than relying on one timing.
In [1]:
Copied!
# Copyright (c) 2025 Microsoft Corporation.
from __future__ import annotations
import asyncio
import contextlib
import json
import multiprocessing as mp
import shutil
import statistics
import subprocess # ruff: ignore[suspicious-subprocess-import] - invokes the local benchmark-qed CLI only
import sys
import tempfile
import time
from concurrent.futures import ProcessPoolExecutor
from pathlib import Path
from typing import TYPE_CHECKING, Any, TypeVar
import matplotlib.pyplot as plt
import nest_asyncio
import pandas as pd
from IPython.display import display as show
from benchmark_qed.autoe.chunk_assertion.cache import (
ContentAddressedCache,
build_cache_metadata,
compute_cache_key,
compute_config_fingerprint,
compute_logical_key,
)
from benchmark_qed.cache import CacheStore, create_default_cache_config, inspect_cache
if TYPE_CHECKING:
from collections.abc import Coroutine
from graphrag_cache import CacheConfig
RUNS = 5
BATCHES = 30
BATCH_SIZE = 100
WORKERS = 4
SHARED_KEYS = 30
SIMULATED_CALL_SECONDS = 0.01
T = TypeVar("T")
def cache_config(path: Path | str) -> CacheConfig:
"""Create a SQLite cache configuration for a database path."""
database_path = Path(path)
return create_default_cache_config(
database_path.parent, database_name=database_path.name
)
def run_async(coroutine: Coroutine[Any, Any, T]) -> T:
"""Run an async cache operation from a notebook cell."""
return asyncio.get_event_loop().run_until_complete(coroutine)
nest_asyncio.apply()
workspace = Path(tempfile.mkdtemp(prefix="benchmark-qed-cache-demo-"))
mp_context = mp.get_context("fork") if "fork" in mp.get_all_start_methods() else None
print(f"Workspace: {workspace}")
print(f"Multiprocess demonstrations available: {mp_context is not None}")
# Copyright (c) 2025 Microsoft Corporation.
from __future__ import annotations
import asyncio
import contextlib
import json
import multiprocessing as mp
import shutil
import statistics
import subprocess # ruff: ignore[suspicious-subprocess-import] - invokes the local benchmark-qed CLI only
import sys
import tempfile
import time
from concurrent.futures import ProcessPoolExecutor
from pathlib import Path
from typing import TYPE_CHECKING, Any, TypeVar
import matplotlib.pyplot as plt
import nest_asyncio
import pandas as pd
from IPython.display import display as show
from benchmark_qed.autoe.chunk_assertion.cache import (
ContentAddressedCache,
build_cache_metadata,
compute_cache_key,
compute_config_fingerprint,
compute_logical_key,
)
from benchmark_qed.cache import CacheStore, create_default_cache_config, inspect_cache
if TYPE_CHECKING:
from collections.abc import Coroutine
from graphrag_cache import CacheConfig
RUNS = 5
BATCHES = 30
BATCH_SIZE = 100
WORKERS = 4
SHARED_KEYS = 30
SIMULATED_CALL_SECONDS = 0.01
T = TypeVar("T")
def cache_config(path: Path | str) -> CacheConfig:
"""Create a SQLite cache configuration for a database path."""
database_path = Path(path)
return create_default_cache_config(
database_path.parent, database_name=database_path.name
)
def run_async(coroutine: Coroutine[Any, Any, T]) -> T:
"""Run an async cache operation from a notebook cell."""
return asyncio.get_event_loop().run_until_complete(coroutine)
nest_asyncio.apply()
workspace = Path(tempfile.mkdtemp(prefix="benchmark-qed-cache-demo-"))
mp_context = mp.get_context("fork") if "fork" in mp.get_all_start_methods() else None
print(f"Workspace: {workspace}")
print(f"Multiprocess demonstrations available: {mp_context is not None}")
Workspace: /var/folders/dn/k4bgd8bs47g6ldk94svck9hw0000gn/T/benchmark-qed-cache-demo-zjdm1vqo Multiprocess demonstrations available: True
Original JSONL baseline¶
The class below reproduces the original chunk-cache persistence behavior: load the complete JSONL file, keep an in-memory snapshot, write the complete snapshot to one shared .tmp file, and replace the cache on every flush.
In [2]:
Copied!
class LegacyJSONLCache:
"""Reproduce the original full-snapshot JSONL cache behavior."""
def __init__(self, path: Path) -> None:
self.path = path
self.path.parent.mkdir(parents=True, exist_ok=True)
self.data: dict[str, str] = {}
if self.path.exists():
for line in self.path.read_text(encoding="utf-8").splitlines():
if not line:
continue
try:
record = json.loads(line)
except json.JSONDecodeError:
continue
if record.get("key") and record.get("grade"):
self.data[str(record["key"])] = str(record["grade"])
def put(self, key: str, value: str) -> None:
"""Store a value in the in-memory snapshot."""
self.data.setdefault(key, value)
def flush(self) -> None:
"""Rewrite and atomically replace the complete JSONL snapshot."""
temporary_path = self.path.with_suffix(self.path.suffix + ".tmp")
with temporary_path.open("w", encoding="utf-8") as file:
for key, value in self.data.items():
file.write(json.dumps({"key": key, "grade": value}) + "\n")
temporary_path.replace(self.path)
def reset_path(path: Path) -> None:
"""Remove one benchmark cache and its SQLite sidecars."""
if path.is_dir():
shutil.rmtree(path)
elif path.exists():
path.unlink()
for suffix in ("-wal", "-shm"):
sidecar = Path(f"{path}{suffix}")
if sidecar.exists():
sidecar.unlink()
class LegacyJSONLCache:
"""Reproduce the original full-snapshot JSONL cache behavior."""
def __init__(self, path: Path) -> None:
self.path = path
self.path.parent.mkdir(parents=True, exist_ok=True)
self.data: dict[str, str] = {}
if self.path.exists():
for line in self.path.read_text(encoding="utf-8").splitlines():
if not line:
continue
try:
record = json.loads(line)
except json.JSONDecodeError:
continue
if record.get("key") and record.get("grade"):
self.data[str(record["key"])] = str(record["grade"])
def put(self, key: str, value: str) -> None:
"""Store a value in the in-memory snapshot."""
self.data.setdefault(key, value)
def flush(self) -> None:
"""Rewrite and atomically replace the complete JSONL snapshot."""
temporary_path = self.path.with_suffix(self.path.suffix + ".tmp")
with temporary_path.open("w", encoding="utf-8") as file:
for key, value in self.data.items():
file.write(json.dumps({"key": key, "grade": value}) + "\n")
temporary_path.replace(self.path)
def reset_path(path: Path) -> None:
"""Remove one benchmark cache and its SQLite sidecars."""
if path.is_dir():
shutil.rmtree(path)
elif path.exists():
path.unlink()
for suffix in ("-wal", "-shm"):
sidecar = Path(f"{path}{suffix}")
if sidecar.exists():
sidecar.unlink()
1. Incremental write scaling¶
This workload adds fixed-size batches and persists after every batch. The original cache rewrites every preceding entry; SQLite inserts only the new rows.
In [3]:
Copied!
async def benchmark_incremental_writes(
kind: str, run: int
) -> tuple[float, list[float]]:
"""Measure total and per-batch persistence latency."""
path = (
workspace
/ f"incremental-{kind}-{run}.{'jsonl' if kind == 'legacy' else 'sqlite3'}"
)
reset_path(path)
batch_times: list[float] = []
started = time.perf_counter()
if kind == "legacy":
cache = LegacyJSONLCache(path)
for batch in range(BATCHES):
batch_started = time.perf_counter()
for offset in range(BATCH_SIZE):
index = batch * BATCH_SIZE + offset
cache.put(f"key-{index}", "full_support")
cache.flush()
batch_times.append(time.perf_counter() - batch_started)
else:
cache = CacheStore(cache_config(path), "benchmark")
for batch in range(BATCHES):
batch_started = time.perf_counter()
entries = [
(f"key-{batch * BATCH_SIZE + offset}", "full_support", {})
for offset in range(BATCH_SIZE)
]
await cache.put_many(entries)
batch_times.append(time.perf_counter() - batch_started)
return time.perf_counter() - started, batch_times
incremental_runs: list[dict[str, Any]] = []
incremental_curves: dict[str, list[float]] = {}
for implementation in ("legacy", "sqlite"):
totals = []
curves = []
for run in range(RUNS):
total, curve = run_async(benchmark_incremental_writes(implementation, run))
totals.append(total)
curves.append(curve)
incremental_runs.append({
"implementation": implementation,
"median_seconds": statistics.median(totals),
"p95_batch_ms": statistics.quantiles(
[value for curve in curves for value in curve], n=20
)[18]
* 1000,
})
incremental_curves[implementation] = [
statistics.median(curve[index] for curve in curves) for index in range(BATCHES)
]
incremental_df = pd.DataFrame(incremental_runs)
show(incremental_df)
async def benchmark_incremental_writes(
kind: str, run: int
) -> tuple[float, list[float]]:
"""Measure total and per-batch persistence latency."""
path = (
workspace
/ f"incremental-{kind}-{run}.{'jsonl' if kind == 'legacy' else 'sqlite3'}"
)
reset_path(path)
batch_times: list[float] = []
started = time.perf_counter()
if kind == "legacy":
cache = LegacyJSONLCache(path)
for batch in range(BATCHES):
batch_started = time.perf_counter()
for offset in range(BATCH_SIZE):
index = batch * BATCH_SIZE + offset
cache.put(f"key-{index}", "full_support")
cache.flush()
batch_times.append(time.perf_counter() - batch_started)
else:
cache = CacheStore(cache_config(path), "benchmark")
for batch in range(BATCHES):
batch_started = time.perf_counter()
entries = [
(f"key-{batch * BATCH_SIZE + offset}", "full_support", {})
for offset in range(BATCH_SIZE)
]
await cache.put_many(entries)
batch_times.append(time.perf_counter() - batch_started)
return time.perf_counter() - started, batch_times
incremental_runs: list[dict[str, Any]] = []
incremental_curves: dict[str, list[float]] = {}
for implementation in ("legacy", "sqlite"):
totals = []
curves = []
for run in range(RUNS):
total, curve = run_async(benchmark_incremental_writes(implementation, run))
totals.append(total)
curves.append(curve)
incremental_runs.append({
"implementation": implementation,
"median_seconds": statistics.median(totals),
"p95_batch_ms": statistics.quantiles(
[value for curve in curves for value in curve], n=20
)[18]
* 1000,
})
incremental_curves[implementation] = [
statistics.median(curve[index] for curve in curves) for index in range(BATCHES)
]
incremental_df = pd.DataFrame(incremental_runs)
show(incremental_df)
| implementation | median_seconds | p95_batch_ms | |
|---|---|---|---|
| 0 | legacy | 0.048413 | 2.816558 |
| 1 | sqlite | 0.069407 | 3.786323 |
In [4]:
Copied!
for implementation, curve in incremental_curves.items():
plt.plot(
range(1, BATCHES + 1), [value * 1000 for value in curve], label=implementation
)
plt.xlabel("Persisted batch")
plt.ylabel("Median batch latency (ms)")
plt.title("Incremental cache persistence")
plt.legend()
plt.grid(alpha=0.3)
plt.show()
for implementation, curve in incremental_curves.items():
plt.plot(
range(1, BATCHES + 1), [value * 1000 for value in curve], label=implementation
)
plt.xlabel("Persisted batch")
plt.ylabel("Median batch latency (ms)")
plt.title("Incremental cache persistence")
plt.legend()
plt.grid(alpha=0.3)
plt.show()
2. Multiprocess integrity¶
Several processes load an empty cache, wait for the same start time, and then write disjoint keys. The check reports process errors and the number of entries that survive.
In [5]:
Copied!
def legacy_parallel_writer(
path_text: str, worker: int, start_at: float, count: int
) -> str | None:
"""Write disjoint entries through the original snapshot cache."""
cache = LegacyJSONLCache(Path(path_text))
while time.time() < start_at:
time.sleep(0.001)
for index in range(count):
cache.put(f"worker-{worker}-key-{index}", "full_support")
try:
cache.flush()
except OSError as exc:
return str(exc)
return None
def sqlite_parallel_writer(
path_text: str, worker: int, start_at: float, count: int
) -> str | None:
"""Write disjoint entries through the SQLite cache."""
cache = CacheStore(cache_config(path_text), "parallel-benchmark")
while time.time() < start_at:
time.sleep(0.001)
asyncio.run(
cache.put_many([
(f"worker-{worker}-key-{index}", "full_support", {})
for index in range(count)
])
)
return None
def count_legacy_entries(path: Path) -> int:
"""Count valid distinct keys surviving in a JSONL cache."""
if not path.exists():
return 0
keys = set()
for line in path.read_text(encoding="utf-8").splitlines():
with contextlib.suppress(json.JSONDecodeError, KeyError):
keys.add(json.loads(line)["key"])
return len(keys)
parallel_rows = []
if mp_context is None:
print("Skipped: this demonstration requires a platform with the fork start method.")
else:
expected = WORKERS * BATCH_SIZE
for implementation, writer in (
("legacy", legacy_parallel_writer),
("sqlite", sqlite_parallel_writer),
):
path = (
workspace
/ f"parallel-{implementation}.{'jsonl' if implementation == 'legacy' else 'sqlite3'}"
)
reset_path(path)
start_at = time.time() + 1
started = time.perf_counter()
with ProcessPoolExecutor(
max_workers=WORKERS, mp_context=mp_context
) as executor:
futures = [
executor.submit(writer, str(path), worker, start_at, BATCH_SIZE)
for worker in range(WORKERS)
]
errors = [future.result(timeout=30) for future in futures]
elapsed = time.perf_counter() - started
surviving = (
count_legacy_entries(path)
if implementation == "legacy"
else sum(
int(item["entries"])
for item in inspect_cache(path)["namespaces"]
if item["namespace"] == "parallel-benchmark"
)
)
parallel_rows.append({
"implementation": implementation,
"workers": WORKERS,
"expected_entries": expected,
"surviving_entries": surviving,
"write_errors": sum(error is not None for error in errors),
"seconds": elapsed,
"integrity": surviving == expected and not any(errors),
})
show(pd.DataFrame(parallel_rows))
def legacy_parallel_writer(
path_text: str, worker: int, start_at: float, count: int
) -> str | None:
"""Write disjoint entries through the original snapshot cache."""
cache = LegacyJSONLCache(Path(path_text))
while time.time() < start_at:
time.sleep(0.001)
for index in range(count):
cache.put(f"worker-{worker}-key-{index}", "full_support")
try:
cache.flush()
except OSError as exc:
return str(exc)
return None
def sqlite_parallel_writer(
path_text: str, worker: int, start_at: float, count: int
) -> str | None:
"""Write disjoint entries through the SQLite cache."""
cache = CacheStore(cache_config(path_text), "parallel-benchmark")
while time.time() < start_at:
time.sleep(0.001)
asyncio.run(
cache.put_many([
(f"worker-{worker}-key-{index}", "full_support", {})
for index in range(count)
])
)
return None
def count_legacy_entries(path: Path) -> int:
"""Count valid distinct keys surviving in a JSONL cache."""
if not path.exists():
return 0
keys = set()
for line in path.read_text(encoding="utf-8").splitlines():
with contextlib.suppress(json.JSONDecodeError, KeyError):
keys.add(json.loads(line)["key"])
return len(keys)
parallel_rows = []
if mp_context is None:
print("Skipped: this demonstration requires a platform with the fork start method.")
else:
expected = WORKERS * BATCH_SIZE
for implementation, writer in (
("legacy", legacy_parallel_writer),
("sqlite", sqlite_parallel_writer),
):
path = (
workspace
/ f"parallel-{implementation}.{'jsonl' if implementation == 'legacy' else 'sqlite3'}"
)
reset_path(path)
start_at = time.time() + 1
started = time.perf_counter()
with ProcessPoolExecutor(
max_workers=WORKERS, mp_context=mp_context
) as executor:
futures = [
executor.submit(writer, str(path), worker, start_at, BATCH_SIZE)
for worker in range(WORKERS)
]
errors = [future.result(timeout=30) for future in futures]
elapsed = time.perf_counter() - started
surviving = (
count_legacy_entries(path)
if implementation == "legacy"
else sum(
int(item["entries"])
for item in inspect_cache(path)["namespaces"]
if item["namespace"] == "parallel-benchmark"
)
)
parallel_rows.append({
"implementation": implementation,
"workers": WORKERS,
"expected_entries": expected,
"surviving_entries": surviving,
"write_errors": sum(error is not None for error in errors),
"seconds": elapsed,
"integrity": surviving == expected and not any(errors),
})
show(pd.DataFrame(parallel_rows))
| implementation | workers | expected_entries | surviving_entries | write_errors | seconds | integrity | |
|---|---|---|---|---|---|---|---|
| 0 | legacy | 4 | 400 | 100 | 3 | 1.007491 | False |
| 1 | sqlite | 4 | 400 | 400 | 0 | 1.019867 | True |
3. Duplicate expensive-call avoidance¶
Every process requests the same keys. The legacy baseline performs every simulated call. The leased cache permits only the claim owner to perform each call; other processes reuse its published result.
In [6]:
Copied!
def legacy_duplicate_worker(key_count: int, delay: float) -> int:
"""Perform every simulated expensive call without coalescing."""
calls = 0
for _index in range(key_count):
time.sleep(delay)
calls += 1
return calls
def leased_duplicate_worker(
path_text: str, owner: str, key_count: int, delay: float
) -> int:
"""Perform only simulated calls transactionally claimed by this worker."""
async def run() -> int:
store = CacheStore(cache_config(path_text), "llm-call-benchmark")
calls = 0
for index in range(key_count):
key = f"shared-{index}"
while await store.get(key) is None:
if store.try_acquire(key, owner, ttl_seconds=5):
await asyncio.sleep(delay)
await store.publish(
key,
"result",
{"simulated": True},
owner_id=owner,
logical_key=key,
config_fingerprint="demo",
)
calls += 1
break
await asyncio.sleep(0.002)
return calls
return asyncio.run(run())
coalescing_rows = []
if mp_context is None:
print("Skipped: this demonstration requires a platform with the fork start method.")
else:
legacy_started = time.perf_counter()
with ProcessPoolExecutor(max_workers=WORKERS, mp_context=mp_context) as executor:
legacy_calls = sum(
executor.map(
legacy_duplicate_worker,
[SHARED_KEYS] * WORKERS,
[SIMULATED_CALL_SECONDS] * WORKERS,
)
)
legacy_seconds = time.perf_counter() - legacy_started
leased_path = workspace / "leased-duplicate.sqlite3"
reset_path(leased_path)
leased_started = time.perf_counter()
with ProcessPoolExecutor(max_workers=WORKERS, mp_context=mp_context) as executor:
leased_calls = sum(
executor.map(
leased_duplicate_worker,
[str(leased_path)] * WORKERS,
[f"worker-{index}" for index in range(WORKERS)],
[SHARED_KEYS] * WORKERS,
[SIMULATED_CALL_SECONDS] * WORKERS,
)
)
leased_seconds = time.perf_counter() - leased_started
logical_requests = WORKERS * SHARED_KEYS
coalescing_rows = [
{
"implementation": "legacy",
"logical_requests": logical_requests,
"simulated_llm_calls": legacy_calls,
"duplicate_calls_avoided": logical_requests - legacy_calls,
"seconds": legacy_seconds,
},
{
"implementation": "sqlite_leases",
"logical_requests": logical_requests,
"simulated_llm_calls": leased_calls,
"duplicate_calls_avoided": logical_requests - leased_calls,
"seconds": leased_seconds,
},
]
show(pd.DataFrame(coalescing_rows))
def legacy_duplicate_worker(key_count: int, delay: float) -> int:
"""Perform every simulated expensive call without coalescing."""
calls = 0
for _index in range(key_count):
time.sleep(delay)
calls += 1
return calls
def leased_duplicate_worker(
path_text: str, owner: str, key_count: int, delay: float
) -> int:
"""Perform only simulated calls transactionally claimed by this worker."""
async def run() -> int:
store = CacheStore(cache_config(path_text), "llm-call-benchmark")
calls = 0
for index in range(key_count):
key = f"shared-{index}"
while await store.get(key) is None:
if store.try_acquire(key, owner, ttl_seconds=5):
await asyncio.sleep(delay)
await store.publish(
key,
"result",
{"simulated": True},
owner_id=owner,
logical_key=key,
config_fingerprint="demo",
)
calls += 1
break
await asyncio.sleep(0.002)
return calls
return asyncio.run(run())
coalescing_rows = []
if mp_context is None:
print("Skipped: this demonstration requires a platform with the fork start method.")
else:
legacy_started = time.perf_counter()
with ProcessPoolExecutor(max_workers=WORKERS, mp_context=mp_context) as executor:
legacy_calls = sum(
executor.map(
legacy_duplicate_worker,
[SHARED_KEYS] * WORKERS,
[SIMULATED_CALL_SECONDS] * WORKERS,
)
)
legacy_seconds = time.perf_counter() - legacy_started
leased_path = workspace / "leased-duplicate.sqlite3"
reset_path(leased_path)
leased_started = time.perf_counter()
with ProcessPoolExecutor(max_workers=WORKERS, mp_context=mp_context) as executor:
leased_calls = sum(
executor.map(
leased_duplicate_worker,
[str(leased_path)] * WORKERS,
[f"worker-{index}" for index in range(WORKERS)],
[SHARED_KEYS] * WORKERS,
[SIMULATED_CALL_SECONDS] * WORKERS,
)
)
leased_seconds = time.perf_counter() - leased_started
logical_requests = WORKERS * SHARED_KEYS
coalescing_rows = [
{
"implementation": "legacy",
"logical_requests": logical_requests,
"simulated_llm_calls": legacy_calls,
"duplicate_calls_avoided": logical_requests - legacy_calls,
"seconds": legacy_seconds,
},
{
"implementation": "sqlite_leases",
"logical_requests": logical_requests,
"simulated_llm_calls": leased_calls,
"duplicate_calls_avoided": logical_requests - leased_calls,
"seconds": leased_seconds,
},
]
show(pd.DataFrame(coalescing_rows))
| implementation | logical_requests | simulated_llm_calls | duplicate_calls_avoided | seconds | |
|---|---|---|---|---|---|
| 0 | legacy | 120 | 120 | 0 | 0.363362 |
| 1 | sqlite_leases | 120 | 30 | 90 | 0.716646 |
4. Interrupted-worker recovery¶
A short-lived process claims work and exits without publishing. Another worker cannot steal an active lease, but can claim it after expiration—without deleting lock files manually.
In [7]:
Copied!
def claim_and_exit(path_text: str) -> bool:
"""Claim work and exit without publishing to simulate interruption."""
return CacheStore(cache_config(path_text), "recovery-demo").try_acquire(
"abandoned-key", "terminated-worker", ttl_seconds=0.5
)
recovery_path = workspace / "recovery.sqlite3"
reset_path(recovery_path)
if mp_context is None:
print("Skipped: this demonstration requires a platform with the fork start method.")
else:
with ProcessPoolExecutor(max_workers=1, mp_context=mp_context) as executor:
original_claim = executor.submit(claim_and_exit, str(recovery_path)).result(
timeout=30
)
recovery_store = CacheStore(cache_config(recovery_path), "recovery-demo")
blocked_before_expiry = not recovery_store.try_acquire(
"abandoned-key", "recovery-worker", ttl_seconds=1
)
time.sleep(0.55)
recovered_after_expiry = recovery_store.try_acquire(
"abandoned-key", "recovery-worker", ttl_seconds=1
)
show(
pd.DataFrame([
{
"original_worker_claimed": original_claim,
"protected_before_expiry": blocked_before_expiry,
"recovered_after_expiry": recovered_after_expiry,
"manual_cleanup_required": False,
}
])
)
def claim_and_exit(path_text: str) -> bool:
"""Claim work and exit without publishing to simulate interruption."""
return CacheStore(cache_config(path_text), "recovery-demo").try_acquire(
"abandoned-key", "terminated-worker", ttl_seconds=0.5
)
recovery_path = workspace / "recovery.sqlite3"
reset_path(recovery_path)
if mp_context is None:
print("Skipped: this demonstration requires a platform with the fork start method.")
else:
with ProcessPoolExecutor(max_workers=1, mp_context=mp_context) as executor:
original_claim = executor.submit(claim_and_exit, str(recovery_path)).result(
timeout=30
)
recovery_store = CacheStore(cache_config(recovery_path), "recovery-demo")
blocked_before_expiry = not recovery_store.try_acquire(
"abandoned-key", "recovery-worker", ttl_seconds=1
)
time.sleep(0.55)
recovered_after_expiry = recovery_store.try_acquire(
"abandoned-key", "recovery-worker", ttl_seconds=1
)
show(
pd.DataFrame([
{
"original_worker_claimed": original_claim,
"protected_before_expiry": blocked_before_expiry,
"recovered_after_expiry": recovered_after_expiry,
"manual_cleanup_required": False,
}
])
)
| original_worker_claimed | protected_before_expiry | recovered_after_expiry | manual_cleanup_required | |
|---|---|---|---|---|
| 0 | True | True | True | False |
5. Configuration mismatch and provenance¶
The logical input fingerprint is independent from the model configuration. This allows the cache to detect that a result exists for the same input while correctly refusing to reuse it under different settings.
In [8]:
Copied!
provenance_path = workspace / "provenance.sqlite3"
cache = ContentAddressedCache(cache_config(provenance_path))
assertion = "The answer is supported by the passage."
chunk = "A representative passage."
prompt_template = "{assertion}\n{chunk}" # ruff: ignore[missing-f-string-syntax] - literal format template
first_metadata = build_cache_metadata(
model="model-a",
call_args={"temperature": 0, "api_key": "not-persisted"},
system_prompt="Judge support.",
user_prompt=prompt_template,
)
first_key = compute_cache_key(
assertion,
chunk,
model="model-a",
call_args={"temperature": 0, "api_key": "not-persisted"},
system_prompt="Judge support.",
user_prompt=prompt_template,
)
logical_key = compute_logical_key(assertion, chunk)
first_config = compute_config_fingerprint(
model="model-a",
call_args={"temperature": 0},
system_prompt="Judge support.",
user_prompt=prompt_template,
)
run_async(cache.claim(first_key, "demo-owner"))
run_async(
cache.publish(
first_key,
"full_support",
first_metadata,
owner_id="demo-owner",
logical_key=logical_key,
config_fingerprint=first_config,
)
)
second_metadata = build_cache_metadata(
model="model-b",
call_args={"temperature": 0},
system_prompt="Judge support.",
user_prompt=prompt_template,
)
mismatch_count, changed_fields = run_async(
cache.find_configuration_mismatches(
[
(
logical_key,
compute_config_fingerprint(
model="model-b",
call_args={"temperature": 0},
system_prompt="Judge support.",
user_prompt=prompt_template,
),
)
],
second_metadata,
)
)
show(
pd.DataFrame([
{
"matching_input": True,
"configuration_mismatches": mismatch_count,
"changed_fields": ", ".join(changed_fields),
"credential_persisted": "not-persisted"
in json.dumps(inspect_cache(provenance_path)),
}
])
)
provenance_path = workspace / "provenance.sqlite3"
cache = ContentAddressedCache(cache_config(provenance_path))
assertion = "The answer is supported by the passage."
chunk = "A representative passage."
prompt_template = "{assertion}\n{chunk}" # ruff: ignore[missing-f-string-syntax] - literal format template
first_metadata = build_cache_metadata(
model="model-a",
call_args={"temperature": 0, "api_key": "not-persisted"},
system_prompt="Judge support.",
user_prompt=prompt_template,
)
first_key = compute_cache_key(
assertion,
chunk,
model="model-a",
call_args={"temperature": 0, "api_key": "not-persisted"},
system_prompt="Judge support.",
user_prompt=prompt_template,
)
logical_key = compute_logical_key(assertion, chunk)
first_config = compute_config_fingerprint(
model="model-a",
call_args={"temperature": 0},
system_prompt="Judge support.",
user_prompt=prompt_template,
)
run_async(cache.claim(first_key, "demo-owner"))
run_async(
cache.publish(
first_key,
"full_support",
first_metadata,
owner_id="demo-owner",
logical_key=logical_key,
config_fingerprint=first_config,
)
)
second_metadata = build_cache_metadata(
model="model-b",
call_args={"temperature": 0},
system_prompt="Judge support.",
user_prompt=prompt_template,
)
mismatch_count, changed_fields = run_async(
cache.find_configuration_mismatches(
[
(
logical_key,
compute_config_fingerprint(
model="model-b",
call_args={"temperature": 0},
system_prompt="Judge support.",
user_prompt=prompt_template,
),
)
],
second_metadata,
)
)
show(
pd.DataFrame([
{
"matching_input": True,
"configuration_mismatches": mismatch_count,
"changed_fields": ", ".join(changed_fields),
"credential_persisted": "not-persisted"
in json.dumps(inspect_cache(provenance_path)),
}
])
)
| matching_input | configuration_mismatches | changed_fields | credential_persisted | |
|---|---|---|---|---|
| 0 | True | 1 | call_args, model | False |
In [ ]:
Copied!
inspection = inspect_cache(provenance_path)
print(json.dumps(inspection, indent=2, sort_keys=True))
print("\nEquivalent CLI command:")
print(f"uv run benchmark-qed cache inspect {provenance_path} --json")
cli_result = subprocess.run( # ruff: ignore[subprocess-without-shell-equals-true] - fixed local module invocation
[
sys.executable,
"-m",
"benchmark_qed",
"cache",
"inspect",
str(provenance_path),
"--json",
],
check=True,
capture_output=True,
text=True,
)
if json.loads(cli_result.stdout)["backend"] != "graphrag-cache":
message = "Cache inspection CLI returned an unexpected backend"
raise RuntimeError(message)
inspection = inspect_cache(provenance_path)
print(json.dumps(inspection, indent=2, sort_keys=True))
print("\nEquivalent CLI command:")
print(f"uv run benchmark-qed cache inspect {provenance_path} --json")
cli_result = subprocess.run( # ruff: ignore[subprocess-without-shell-equals-true] - fixed local module invocation
[
sys.executable,
"-m",
"benchmark_qed",
"cache",
"inspect",
str(provenance_path),
"--json",
],
check=True,
capture_output=True,
text=True,
)
if json.loads(cli_result.stdout)["backend"] != "graphrag-cache":
message = "Cache inspection CLI returned an unexpected backend"
raise RuntimeError(message)
6. Reporting guidance¶
When sharing results, report the machine, filesystem, Python version, benchmark constants, median timings, p95 batch latency, surviving entry count, write errors, and actual simulated LLM-call count.
The defensible conclusion is:
The SQLite/WAL cache is more efficient and reliable for repeated incremental writes and concurrent workloads. It prevents lost updates, coalesces duplicate model requests across processes, supports concurrent readers, warns about incompatible configurations, and recovers automatically from interrupted workers.
Do not claim that SQLite wins every single-key microbenchmark; the primary gains are scaling, integrity, duplicate-call avoidance, and recoverability.
In [10]:
Copied!
summary_rows = []
summary_rows.extend(incremental_runs)
summary_rows.extend(parallel_rows)
summary_rows.extend(coalescing_rows)
show(pd.DataFrame(summary_rows))
summary_rows = []
summary_rows.extend(incremental_runs)
summary_rows.extend(parallel_rows)
summary_rows.extend(coalescing_rows)
show(pd.DataFrame(summary_rows))
| implementation | median_seconds | p95_batch_ms | workers | expected_entries | surviving_entries | write_errors | seconds | integrity | logical_requests | simulated_llm_calls | duplicate_calls_avoided | |
|---|---|---|---|---|---|---|---|---|---|---|---|---|
| 0 | legacy | 0.048413 | 2.816558 | NaN | NaN | NaN | NaN | NaN | NaN | NaN | NaN | NaN |
| 1 | sqlite | 0.069407 | 3.786323 | NaN | NaN | NaN | NaN | NaN | NaN | NaN | NaN | NaN |
| 2 | legacy | NaN | NaN | 4.0 | 400.0 | 100.0 | 3.0 | 1.007491 | False | NaN | NaN | NaN |
| 3 | sqlite | NaN | NaN | 4.0 | 400.0 | 400.0 | 0.0 | 1.019867 | True | NaN | NaN | NaN |
| 4 | legacy | NaN | NaN | NaN | NaN | NaN | NaN | 0.363362 | NaN | 120.0 | 120.0 | 0.0 |
| 5 | sqlite_leases | NaN | NaN | NaN | NaN | NaN | NaN | 0.716646 | NaN | 120.0 | 30.0 | 90.0 |
In [ ]:
Copied!
# Run this after exporting any results you want to keep.
shutil.rmtree(workspace)
print(f"Removed temporary workspace: {workspace}")
# Run this after exporting any results you want to keep.
shutil.rmtree(workspace)
print(f"Removed temporary workspace: {workspace}")