diff --git a/CHANGELOG.md b/CHANGELOG.md index 0972b45..005e902 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -95,6 +95,12 @@ different event from one that moved because it was wrong. timeout, 408, 409, 425, 429 and 5xx), `Retry-After` is honoured up to 60 seconds, the SDKs' own retries are off, and a failed case records the attempts it actually made. +- Two workers answering cases with the same text wrote the same cache entry + through one shared temporary file; on Windows the loser raised + `PermissionError` and aborted the run, losing every call already paid for + (#27). Each write now has a temporary file of its own, a write that still + fails is counted in the cache stats (`write_errors`) instead of raised, and + cases sharing a key are answered once per run, the rest from the cache. - Every validation message on the site's calculator and planner read "[object Object]" (#25). Messages now say what is wrong, mark the field invalid, and are tied to it for screen readers; a result is announced as one diff --git a/src/plumbline/runner/cache.py b/src/plumbline/runner/cache.py index f0a0518..4e857bb 100644 --- a/src/plumbline/runner/cache.py +++ b/src/plumbline/runner/cache.py @@ -12,13 +12,21 @@ A hit is recorded as a hit. It does not count toward cost or latency, because it measures disk rather than the model. + +Cases that share a key within one run are answered once: the runner holds a lock +per key across look up, call, and store, so the second waits and then hits. A +write that fails is counted and dropped rather than raised, because the answer +it was storing is already paid for. """ from __future__ import annotations +import contextlib import dataclasses import hashlib import json +import os +import tempfile from collections.abc import Sequence from pathlib import Path from typing import Any @@ -94,6 +102,7 @@ def __init__(self, directory: Path | str, *, enabled: bool = True) -> None: self.hits = 0 self.misses = 0 self.writes = 0 + self.write_errors = 0 def path_for(self, key: str) -> Path: return self.directory / key[:2] / f"{key}.json" @@ -117,19 +126,48 @@ def get(self, key: str) -> Prediction | None: return prediction def put(self, key: str, prediction: Prediction) -> None: + """Store an answer, or count the failure to; never raise. + + Each write goes through a temporary file of its own, so two writers of + one key cannot collide on a shared name. A write that still fails (a + virus scanner holding the file on Windows, a full disk) is counted in + ``stats`` and dropped: the answer is already in hand and paid for, and a + missing entry costs one more call on a later run, not this run. + """ if not self.enabled: return path = self.path_for(key) - path.parent.mkdir(parents=True, exist_ok=True) payload = {"version": CACHE_FORMAT_VERSION, "prediction": to_jsonable(prediction)} - temporary = path.with_suffix(".tmp") - temporary.write_text(json.dumps(payload, sort_keys=True, default=str), encoding="utf-8") - temporary.replace(path) + temporary: Path | None = None + try: + path.parent.mkdir(parents=True, exist_ok=True) + with tempfile.NamedTemporaryFile( + "w", + encoding="utf-8", + dir=path.parent, + prefix=f"{key}.", + suffix=".tmp", + delete=False, + ) as handle: + temporary = Path(handle.name) + handle.write(json.dumps(payload, sort_keys=True, default=str)) + os.replace(temporary, path) + except OSError: + self.write_errors += 1 + if temporary is not None: + with contextlib.suppress(OSError): + temporary.unlink() + return self.writes += 1 @property def stats(self) -> dict[str, int]: - return {"hits": self.hits, "misses": self.misses, "writes": self.writes} + return { + "hits": self.hits, + "misses": self.misses, + "writes": self.writes, + "write_errors": self.write_errors, + } def to_jsonable(prediction: Prediction) -> dict[str, Any]: diff --git a/src/plumbline/runner/execute.py b/src/plumbline/runner/execute.py index 521d883..61fc654 100644 --- a/src/plumbline/runner/execute.py +++ b/src/plumbline/runner/execute.py @@ -24,6 +24,7 @@ import hashlib import json +import threading import time from collections.abc import Callable, Mapping, Sequence from concurrent.futures import ThreadPoolExecutor @@ -524,9 +525,10 @@ def run( estimate = check_guard(cases, guard, pricing) records: list[CaseRecord | None] = [None] * len(cases) + locks = _KeyLocks() if cache is not None and cache.enabled else None def handle(index: int) -> None: - records[index] = _one_case(cases[index], adapter, cache, retry, table) + records[index] = _one_case(cases[index], adapter, cache, retry, table, locks) if workers == 1: for index in range(len(cases)): @@ -585,16 +587,51 @@ def handle(index: int) -> None: ) +class _KeyLocks: + """One lock per cache key, handed out on demand to the worker threads. + + Two cases with the same text share a key. Without a lock both miss the + cache, both call the adapter, and the same answer is billed twice. Holding + the key's lock across look up, call, and store makes the second one wait + and then find the first one's answer. Different keys never wait on each + other. + """ + + def __init__(self) -> None: + self._guard = threading.Lock() + self._locks: dict[str, threading.Lock] = {} + + def for_key(self, key: str) -> threading.Lock: + with self._guard: + return self._locks.setdefault(key, threading.Lock()) + + def _one_case( case: Case, adapter: Adapter, cache: Cache | None, retry: RetryPolicy, table: PricingTable, + locks: _KeyLocks | None = None, ) -> CaseRecord: labels = list(case.labels) key = cache_key(adapter, case.text, labels, case.question_type) if cache is not None else None + if key is not None and locks is not None: + with locks.for_key(key): + return _answer(case, labels, key, adapter, cache, retry, table) + return _answer(case, labels, key, adapter, cache, retry, table) + +def _answer( + case: Case, + labels: list[str], + key: str | None, + adapter: Adapter, + cache: Cache | None, + retry: RetryPolicy, + table: PricingTable, +) -> CaseRecord: + """From the cache if it holds the answer, otherwise from the adapter, with retries.""" if cache is not None and key is not None: hit = cache.get(key) if hit is not None: diff --git a/tests/test_runner.py b/tests/test_runner.py index a89fa71..ae92f7e 100644 --- a/tests/test_runner.py +++ b/tests/test_runner.py @@ -172,11 +172,74 @@ def test_reordering_labels_does_not_invalidate_the_cache() -> None: assert forward == backward +class CountingAdapter(Adapter): + """Answers through the mock, counting calls across threads, a little slowly.""" + + def __init__(self, inner: Adapter) -> None: + import threading + + self.inner = inner + self.name = inner.name + self.model_requested = inner.model_requested + self.revision = inner.revision + self.probability_semantics = inner.probability_semantics + self.calls = 0 + self._lock = threading.Lock() + + @property + def call_params(self): # type: ignore[override] + return self.inner.call_params + + def classify(self, text: str, labels: list[str], **asked: object) -> Prediction: + import time + + with self._lock: + self.calls += 1 + time.sleep(0.02) # long enough for every worker to arrive at once + return self.inner.classify(text, labels, **asked) # type: ignore[arg-type] + + +def test_identical_cases_under_many_workers_are_answered_once(tmp_path: Path) -> None: + """One key is one billed call, and duplicates writing it at once do not crash (#27).""" + import dataclasses + + base = make_cases(1, labels=LABELS)[0] + cases = [dataclasses.replace(base, id=f"dup-{i}") for i in range(16)] + adapter = CountingAdapter(an_adapter(cases)) + cache = Cache(tmp_path) + + result = execute.run(adapter, cases, cache=cache, workers=16) + + assert adapter.calls == 1 + assert all(record.ok for record in result.records) + assert sum(record.from_cache for record in result.records) == 15 + + +def test_a_cache_write_that_fails_does_not_end_the_run( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The answer is already paid for; losing the cache entry must not lose the run.""" + from plumbline.runner import cache as cache_module + + def refuse(*_: object) -> None: + raise PermissionError(32, "The process cannot access the file") + + monkeypatch.setattr(cache_module.os, "replace", refuse) + cases = make_cases(5, labels=LABELS) + cache = Cache(tmp_path) + + result = execute.run(an_adapter(cases), cases, cache=cache, workers=4) + + assert all(record.ok for record in result.records) + assert cache.stats["write_errors"] == 5 + assert not list(tmp_path.rglob("*.tmp")) + + def test_a_disabled_cache_never_reads_or_writes(tmp_path: Path) -> None: cases = make_cases(20, labels=LABELS) cache = Cache(tmp_path / "cache", enabled=False) execute.run(an_adapter(cases), cases, cache=cache, workers=2) - assert cache.stats == {"hits": 0, "misses": 0, "writes": 0} + assert cache.stats == {"hits": 0, "misses": 0, "writes": 0, "write_errors": 0} assert not (tmp_path / "cache").exists()