diff --git a/backend/app/backtest/engine.py b/backend/app/backtest/engine.py index 1bb0679..cd3b241 100644 --- a/backend/app/backtest/engine.py +++ b/backend/app/backtest/engine.py @@ -121,6 +121,17 @@ class _CacheEntry: self.ts = ts +class _InFlight: + """同 key 正在计算的占位: leader 算完通过 done 唤醒所有跟随者复用结果。""" + + __slots__ = ("done", "df", "error") + + def __init__(self) -> None: + self.done = threading.Event() + self.df: pl.DataFrame | None = None + self.error: BaseException | None = None + + class PanelCache: """LRU + TTL 数据面板缓存。""" @@ -132,6 +143,9 @@ class PanelCache: # 无锁的 move_to_end/del/popitem check-then-act 会抛 "OrderedDict mutated"。 # 用实例锁守护所有 OrderedDict 变更; compute_fn (重扫盘) 放锁外避免串行化。 self._lock = threading.Lock() + # single-flight: 同 key 只让一个线程 compute, 其余等其结果复用。 + # 否则优化器等场景下 max_workers 个线程冷启动同时 miss, 会并行加载 N 份同一面板。 + self._inflight: dict[str, _InFlight] = {} def get_or_compute( self, @@ -146,19 +160,43 @@ class PanelCache: now = time.monotonic() with self._lock: - if key in self._cache: - entry = self._cache[key] + entry = self._cache.get(key) + if entry is not None: if now - entry.ts < self._ttl: self._cache.move_to_end(key) return entry.df - del self._cache[key] + del self._cache[key] # 过期, 丢弃后重算 + # single-flight: 同 key 若已有线程在算, 登记为跟随者; 否则本线程当 leader。 + flight = self._inflight.get(key) + leader = flight is None + if leader: + flight = _InFlight() + self._inflight[key] = flight - # 计算在锁外 (可能重扫 parquet, 耗时); 并发相同 key 至多重复算一次, 不会崩 - df = compute_fn(symbols, start, end, columns, asset_type) + if not leader: + # 跟随者: 等 leader 算完直接复用, 不重复 compute (消除缓存踩踏)。 + flight.done.wait() + if flight.error is not None: + raise flight.error + return flight.df + + # leader: compute 放锁外 (不同 key 仍可并发, 保留原设计优点)。 + try: + df = compute_fn(symbols, start, end, columns, asset_type) + except BaseException as e: + # 失败不缓存: 摘除 inflight 让后续线程重试, 并把异常透传给已在等的跟随者。 + with self._lock: + self._inflight.pop(key, None) + flight.error = e + flight.done.set() + raise with self._lock: self._cache[key] = _CacheEntry(df=df, ts=now) if len(self._cache) > self._max_size: self._cache.popitem(last=False) + self._inflight.pop(key, None) + flight.df = df + flight.done.set() return df def invalidate(self) -> None: diff --git a/backend/tests/test_backtest_etf.py b/backend/tests/test_backtest_etf.py index db2d691..ac8e866 100644 --- a/backend/tests/test_backtest_etf.py +++ b/backend/tests/test_backtest_etf.py @@ -1,3 +1,4 @@ +import time import types from datetime import date @@ -77,6 +78,78 @@ def test_engine_stock_uses_daily_enriched_dir(monkeypatch, tmp_path): assert "kline_daily_enriched" in captured["path"] +def test_panel_cache_single_flight_computes_once(): + """N 个线程并发同 key 冷启动: compute_fn 只应被调用一次, 其余复用结果 (无缓存踩踏)。""" + import threading + + cache = PanelCache() + calls = [] + barrier = threading.Barrier(8) + df = pl.DataFrame({"symbol": ["510300"]}) + + def slow_compute(symbols, start, end, columns, asset_type): + calls.append(1) + time.sleep(0.05) # 拉长窗口, 逼出并发 miss + return df + + args = (["510300"], date(2026, 1, 1), date(2026, 1, 2), None) + results = [] + rlock = threading.Lock() + + def worker(): + barrier.wait() # 所有线程同时起跑, 制造冷启动踩踏 + r = cache.get_or_compute(*args, slow_compute, "stock") + with rlock: + results.append(r) + + threads = [threading.Thread(target=worker) for _ in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert sum(calls) == 1, f"面板被重复加载 {sum(calls)} 次, single-flight 失效" + assert len(results) == 8 and all(r is df for r in results) + + +def test_panel_cache_single_flight_error_propagates_and_retries(): + """leader compute 抛错: 不缓存失败, 异常透传给所有等待者, 后续调用可重试成功。""" + import threading + + cache = PanelCache() + barrier = threading.Barrier(4) + boom = RuntimeError("scan failed") + + def failing_compute(symbols, start, end, columns, asset_type): + time.sleep(0.03) + raise boom + + args = (["510300"], date(2026, 1, 1), date(2026, 1, 2), None) + errors = [] + elock = threading.Lock() + + def worker(): + barrier.wait() + try: + cache.get_or_compute(*args, failing_compute, "stock") + except RuntimeError as e: + with elock: + errors.append(e) + + threads = [threading.Thread(target=worker) for _ in range(4)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert len(errors) == 4 and all(e is boom for e in errors), "失败未透传给全部跟随者" + + # 失败未被缓存 —— 重试应重新 compute 并成功 + df = pl.DataFrame({"symbol": ["510300"]}) + got = cache.get_or_compute(*args, lambda *a: df, "stock") + assert got is df + + def test_job_key_includes_asset_type_and_is_consistent(): """stream 与 cancel 必须用同一 job_key: asset_type 进 key 且相同入参产出相同 key。""" from app.api.backtest import _make_job_key