From 33921fcc51b43f0f274e26b059178fe663dcbcb7 Mon Sep 17 00:00:00 2001 From: CJohn Date: Sun, 19 Jul 2026 14:47:12 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=B8=B2=E8=A1=8C=E5=8C=96=20Numba=20pa?= =?UTF-8?q?rallel=20=E5=86=85=E6=A0=B8=EF=BC=8C=E9=81=BF=E5=85=8D=E7=AD=96?= =?UTF-8?q?=E7=95=A5=E9=A1=B5=E5=B9=B6=E5=8F=91=E5=B4=A9=E6=BA=83?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 并发 run_all 会触发 workqueue Concurrent access，导致后端进程中断。 为 matrix parallel 内核加进程锁，前端对 run_all 做 pending 去重。 --- backend/app/backtest/matrix.py | 47 +++++++++------- backend/app/backtest/numba_runtime.py | 23 ++++++++ backend/tests/backtest/test_numba_runtime.py | 59 ++++++++++++++++++++ frontend/src/pages/Screener.tsx | 24 +++++++- 4 files changed, 130 insertions(+), 23 deletions(-) create mode 100644 backend/app/backtest/numba_runtime.py create mode 100644 backend/tests/backtest/test_numba_runtime.py diff --git a/backend/app/backtest/matrix.py b/backend/app/backtest/matrix.py index d343682..d488e42 100644 --- a/backend/app/backtest/matrix.py +++ b/backend/app/backtest/matrix.py @@ -27,6 +27,7 @@ import pyarrow.compute as pc import pyarrow.dataset as pads from app.backtest.minute_trigger import build_minute_exit_reference +from app.backtest.numba_runtime import run_numba_parallel from app.price_limits import ( MAIN_BOARD_ST_LIMIT_CHANGE_DATE, numpy_limit_pct_vectors, @@ -2925,12 +2926,14 @@ def valid_shift( "valid_shift", (source, valid, index.offsets, index.rows), {"periods": int(periods)}, - lambda: _valid_shift_kernel( - source, - valid, - index.offsets, - index.rows, - int(periods), + lambda: run_numba_parallel( + lambda: _valid_shift_kernel( + source, + valid, + index.offsets, + index.rows, + int(periods), + ) ), ) @@ -3072,14 +3075,16 @@ def _valid_rolling_reduce( if ddof < 0 or ddof >= window: raise ValueError("valid rolling ddof must be in [0, window)") index = _resolve_valid_bar_index(source, valid, bar_index) - return _valid_rolling_kernel( - source, - valid, - index.offsets, - index.rows, - int(window), - int(operation), - int(ddof), + return run_numba_parallel( + lambda: _valid_rolling_kernel( + source, + valid, + index.offsets, + index.rows, + int(window), + int(operation), + int(ddof), + ) ) @@ -3307,12 +3312,14 @@ def valid_ewm_adjust_false( "valid_ewm_adjust_false", (source, valid, index.offsets, index.rows), {"alpha": alpha_value}, - lambda: _valid_ewm_kernel( - source, - valid, - index.offsets, - index.rows, - alpha_value, + lambda: run_numba_parallel( + lambda: _valid_ewm_kernel( + source, + valid, + index.offsets, + index.rows, + alpha_value, + ) ), ) diff --git a/backend/app/backtest/numba_runtime.py b/backend/app/backtest/numba_runtime.py new file mode 100644 index 0000000..53b42e1 --- /dev/null +++ b/backend/app/backtest/numba_runtime.py @@ -0,0 +1,23 @@ +"""Serialize Numba ``@njit(parallel=True)`` kernels across request threads. + +Numba's default ``workqueue`` threading layer is not thread-safe. Concurrent +calls from FastAPI worker threads terminate the process +(``Concurrent access has been detected`` → socket hang up). + +A process-wide lock is enough: overlapping kernels queue instead of crashing. +""" +from __future__ import annotations + +import threading +from collections.abc import Callable +from typing import TypeVar + +_NUMBA_PARALLEL_LOCK = threading.RLock() + +T = TypeVar("T") + + +def run_numba_parallel(fn: Callable[[], T]) -> T: + """Run a Numba parallel kernel under the process-wide lock.""" + with _NUMBA_PARALLEL_LOCK: + return fn() diff --git a/backend/tests/backtest/test_numba_runtime.py b/backend/tests/backtest/test_numba_runtime.py new file mode 100644 index 0000000..99625f2 --- /dev/null +++ b/backend/tests/backtest/test_numba_runtime.py @@ -0,0 +1,59 @@ +"""Numba parallel kernels must stay safe under concurrent callers.""" +from __future__ import annotations + +import threading +from concurrent.futures import ThreadPoolExecutor + +import numpy as np +import pytest + +from app.backtest.numba_runtime import run_numba_parallel + + +def test_run_numba_parallel_serializes_concurrent_calls(): + """Two threads must not overlap inside the parallel critical section.""" + active = 0 + max_active = 0 + lock = threading.Lock() + + def work() -> int: + nonlocal active, max_active + with lock: + active += 1 + max_active = max(max_active, active) + try: + total = 0 + for i in range(20_000): + total += i + return total + finally: + with lock: + active -= 1 + + with ThreadPoolExecutor(max_workers=8) as pool: + list(pool.map(lambda _: run_numba_parallel(work), range(16))) + + assert max_active == 1 + + +@pytest.mark.skipif( + __import__("importlib").util.find_spec("numba") is None, + reason="numba not installed on this platform", +) +def test_valid_shift_kernel_survives_concurrent_threads(): + """Regression for workqueue 'Concurrent access has been detected' crashes.""" + from app.backtest.matrix import valid_shift + + rng = np.random.default_rng(0) + values = rng.normal(size=(64, 32)).astype(np.float32) + values[::7, ::3] = np.nan + + def once() -> np.ndarray: + return valid_shift(values, 3) + + with ThreadPoolExecutor(max_workers=4) as pool: + results = list(pool.map(lambda _: once(), range(8))) + + baseline = results[0] + for other in results[1:]: + np.testing.assert_allclose(baseline, other, equal_nan=True) diff --git a/frontend/src/pages/Screener.tsx b/frontend/src/pages/Screener.tsx index a0166a2..6274c6a 100644 --- a/frontend/src/pages/Screener.tsx +++ b/frontend/src/pages/Screener.tsx @@ -232,6 +232,24 @@ export function Screener() { ) const cacheCoversPool = visiblePool.length > 0 && missingStrategyIds.length === 0 + // 防止 reload / auto-run / StrictMode 叠出并发 run_all（后端 Numba 会崩溃） + // 用 ref 同步门闩，避免同一渲染周期内 isPending 尚未更新导致重复触发 + const runAllPendingRef = useRef(false) + const requestRunAll = useCallback(( + vars: { date?: string; strategyIds?: string[] } = {}, + options?: Parameters[1], + ) => { + if (runAllPendingRef.current || runAll.isPending) return + runAllPendingRef.current = true + runAll.mutate(vars, { + ...options, + onSettled: (...args) => { + runAllPendingRef.current = false + options?.onSettled?.(...args) + }, + }) + }, [runAll]) + // 摘要只同步当前日期的卡片数量，避免旧日期缓存短暂显示成当前结果。 useEffect(() => { if (!summaryQuery.data || !asOf) return @@ -433,8 +451,8 @@ export function Screener() { // 未覆盖: 受系统开关控制 if (!screenerAutoRun) return runAllDateRef.current = runKey - runAll.mutate({ date: asOf, strategyIds: missingStrategyIds }) - }, [asOf, strategyPresets.length, summaryQuery.isSuccess, visiblePool, cacheCoversPool, missingStrategyIds, screenerAutoRun, assetType, runAll.isPending]) + requestRunAll({ date: asOf, strategyIds: missingStrategyIds }) + }, [asOf, strategyPresets.length, summaryQuery.isSuccess, visiblePool, cacheCoversPool, missingStrategyIds, screenerAutoRun, assetType, runAll.isPending, requestRunAll]) const run = useMutation({ mutationFn: ({ id, date }: { id: string; date: string }) => @@ -501,7 +519,7 @@ export function Screener() { mutationFn: api.strategyReload, onSuccess: () => { qc.invalidateQueries({ queryKey: ['screener-strategies'] }) - if (asOf) runAll.mutate({ date: asOf }) + if (asOf) requestRunAll({ date: asOf }) }, })