From e6a3c9ffb7ddf61c091c65f06224585feaa16965 Mon Sep 17 00:00:00 2001 From: NanPan <111261006+poijygfdyy@users.noreply.github.com> Date: Sun, 18 Oct 2026 08:13:27 +0800 Subject: [PATCH] fix(checkpoints): serialize shared memory history cache operations (#6708) * fix(checkpoints): serialize shared memory history cache operations * fix(checkpoints): align memory cache contract with locking * test(checkpoints): make cache race regression deterministic * docs: remove overlapping cache note from memory-cache PR --- .../harness/deerflow/runtime/AGENTS.md | 7 + .../runtime/checkpoint_cache/memory.py | 65 +++++---- ...est_checkpoint_cache_memory_concurrency.py | 125 ++++++++++++++++++ 3 files changed, 168 insertions(+), 29 deletions(-) create mode 100644 backend/tests/test_checkpoint_cache_memory_concurrency.py diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md index 6da4f8f9a9..c1460f4b52 100644 --- a/backend/packages/harness/deerflow/runtime/AGENTS.md +++ b/backend/packages/harness/deerflow/runtime/AGENTS.md @@ -1,3 +1,10 @@ +### Checkpoint Cache Concurrency + +The memory history cache is shared by sync saver wrappers. Its threading lock +serializes lookup/LRU promotion, population/eviction, thread purge, close, and +statistics snapshots. Keep cache critical sections free of awaits and saver I/O; +immutable histories do not make the mutable LRU registry thread-safe. + ### Redis Checkpoint Cache Purge Thread purge escapes Redis glob metacharacters in the complete literal key stem diff --git a/backend/packages/harness/deerflow/runtime/checkpoint_cache/memory.py b/backend/packages/harness/deerflow/runtime/checkpoint_cache/memory.py index 00683dd2b1..def37d3c5d 100644 --- a/backend/packages/harness/deerflow/runtime/checkpoint_cache/memory.py +++ b/backend/packages/harness/deerflow/runtime/checkpoint_cache/memory.py @@ -1,7 +1,8 @@ -"""Process-local LRU backend. Zero serialization on the hit path.""" +"""Process-local LRU backend with serialized LRU and statistics access.""" from __future__ import annotations +import threading from collections import OrderedDict from typing import Any @@ -20,6 +21,7 @@ class MemoryCheckpointHistoryCache: def __init__(self, max_entries: int = 128) -> None: if max_entries < 0: raise ValueError("max_entries must be >= 0") + self._lock = threading.Lock() self._max_entries = max_entries self._data: OrderedDict[str, dict[str, Any]] = OrderedDict() self._hits = 0 @@ -31,26 +33,28 @@ class MemoryCheckpointHistoryCache: return self._max_entries > 0 def get_many(self, keys: list[str]) -> dict[str, dict[str, Any]]: - found: dict[str, dict[str, Any]] = {} - for key in keys: - entry = self._data.get(key) - if entry is None: - self._misses += 1 - continue - self._data.move_to_end(key) - self._hits += 1 - found[key] = _copy_entry(entry) - return found + with self._lock: + found: dict[str, dict[str, Any]] = {} + for key in keys: + entry = self._data.get(key) + if entry is None: + self._misses += 1 + continue + self._data.move_to_end(key) + self._hits += 1 + found[key] = _copy_entry(entry) + return found def set_many(self, entries: dict[str, dict[str, Any]]) -> None: - if not self.enabled: - return - for key, entry in entries.items(): - self._data[key] = _copy_entry(entry) - self._data.move_to_end(key) - while len(self._data) > self._max_entries: - self._data.popitem(last=False) - self._evictions += 1 + with self._lock: + if not self.enabled: + return + for key, entry in entries.items(): + self._data[key] = _copy_entry(entry) + self._data.move_to_end(key) + while len(self._data) > self._max_entries: + self._data.popitem(last=False) + self._evictions += 1 async def aget_many(self, keys: list[str]) -> dict[str, dict[str, Any]]: return self.get_many(keys) @@ -60,20 +64,23 @@ class MemoryCheckpointHistoryCache: def delete_thread(self, key_prefix: str, thread_id: str) -> None: """Purge every entry of one thread (lifecycle, not invalidation).""" - stem = thread_key_stem(key_prefix, thread_id) - for key in [k for k in self._data if k.startswith(stem)]: - del self._data[key] + with self._lock: + stem = thread_key_stem(key_prefix, thread_id) + for key in [k for k in self._data if k.startswith(stem)]: + del self._data[key] async def adelete_thread(self, key_prefix: str, thread_id: str) -> None: self.delete_thread(key_prefix, thread_id) def stats(self) -> CheckpointCacheStats: - return CheckpointCacheStats( - hits=self._hits, - misses=self._misses, - evictions=self._evictions, - entries=len(self._data), - ) + with self._lock: + return CheckpointCacheStats( + hits=self._hits, + misses=self._misses, + evictions=self._evictions, + entries=len(self._data), + ) async def aclose(self) -> None: - self._data.clear() + with self._lock: + self._data.clear() diff --git a/backend/tests/test_checkpoint_cache_memory_concurrency.py b/backend/tests/test_checkpoint_cache_memory_concurrency.py new file mode 100644 index 0000000000..8b31228578 --- /dev/null +++ b/backend/tests/test_checkpoint_cache_memory_concurrency.py @@ -0,0 +1,125 @@ +"""Concurrent sync readers share the provider's process-local history cache.""" + +import asyncio +import threading +from collections import OrderedDict +from concurrent.futures import ThreadPoolExecutor +from queue import Queue + +import pytest + +from deerflow.runtime.checkpoint_cache.base import make_history_key +from deerflow.runtime.checkpoint_cache.memory import MemoryCheckpointHistoryCache + + +@pytest.mark.parametrize("mutation", ["evict", "purge", "close"]) +def test_lookup_survives_concurrent_removal(mutation): + cache = MemoryCheckpointHistoryCache(max_entries=1) + key = make_history_key("p", "t1", "", "c1", "messages") + entry = {"writes": [("task", "messages", "original")], "seed": "seed"} + cache.set_many({key: entry}) + looked_up = threading.Event() + release_reader = threading.Event() + writer_started = threading.Event() + writer_finished = threading.Event() + coordination = Queue() + removal_reported = threading.Event() + + def report_removal(): + if not removal_reported.is_set(): + removal_reported.set() + coordination.put("removed") + + class PausedLookup(OrderedDict): + def get(self, lookup_key, default=None): + value = super().get(lookup_key, default) + if lookup_key == key: + looked_up.set() + assert release_reader.wait(5), "reader was not released" + return value + + def __delitem__(self, deleted_key): + super().__delitem__(deleted_key) + if deleted_key == key: + report_removal() + + def popitem(self, last=True): + removed_key, value = super().popitem(last=last) + if removed_key == key: + report_removal() + return removed_key, value + + def clear(self): + contained_target = key in self + super().clear() + if contained_target: + report_removal() + + class ObservedLock: + """Report a contender before it blocks behind the paused reader.""" + + def __init__(self): + self._lock = threading.Lock() + + def __enter__(self): + if self._lock.locked(): + coordination.put("writer_waiting") + self._lock.acquire() + return self + + def __exit__(self, exc_type, exc_value, traceback): + self._lock.release() + + # Pause at the actual lookup -> LRU promotion window, not before get_many. + cache._data = PausedLookup(cache._data) + cache._lock = ObservedLock() + + def remove(): + writer_started.set() + try: + if mutation == "evict": + cache.set_many({"other": {"writes": []}}) + elif mutation == "purge": + cache.delete_thread("p", "t1") + else: + asyncio.run(cache.aclose()) + finally: + writer_finished.set() + + with ThreadPoolExecutor(max_workers=2) as pool: + reader = pool.submit(cache.get_many, [key]) + try: + assert looked_up.wait(5) + writer = pool.submit(remove) + assert writer_started.wait(5) + # A serialized writer reports that it is waiting on the reader's + # lock. On unfixed main it instead removes the key and reports that, + # deterministically reproducing the get -> move_to_end race. + assert coordination.get(timeout=5) == "writer_waiting" + finally: + release_reader.set() + assert reader.result(timeout=5) == {key: entry} + writer.result(timeout=5) + assert cache.get_many([key]) == {} + assert cache.stats().entries == int(mutation == "evict") + + +def test_parallel_read_write_keeps_capacity_and_accounting(): + cache = MemoryCheckpointHistoryCache(max_entries=4) + start = threading.Barrier(8) + + def exercise(worker): + start.wait(timeout=5) + for n in range(200): + key = f"{worker}:{n}" + cache.set_many({key: {"writes": [("task", "messages", key)]}}) + result = cache.get_many([key]) + if key in result: + assert result[key]["writes"] == [("task", "messages", key)] + + with ThreadPoolExecutor(max_workers=8) as pool: + list(pool.map(exercise, range(8))) + stats = cache.stats() + assert stats.entries == 4 + assert stats.evictions == 8 * 200 - 4 + assert stats.hits + stats.misses == 8 * 200