diff --git a/backend/app/backtest/worker.py b/backend/app/backtest/worker.py index f4fb102..be1763f 100644 --- a/backend/app/backtest/worker.py +++ b/backend/app/backtest/worker.py @@ -2,6 +2,7 @@ from __future__ import annotations import json +import logging import multiprocessing as mp import os import queue @@ -17,6 +18,8 @@ from typing import Any import psutil +logger = logging.getLogger(__name__) + class BacktestWorkerError(RuntimeError): """Raised when a spawned worker fails before returning a task result.""" @@ -254,6 +257,13 @@ def _worker_entry(task: dict[str, Any], event_queue, cancel_event) -> None: if store is not None: with suppress(Exception): store.db.close() + # 终态消息已入队: 显式冲刷队列后立即退出。大数据量任务跳过解释器 + # teardown (GC、DuckDB 线程 join、DLL 卸载), 否则收尾可达数十秒, + # 会撞上父进程 10s 退出预算。close+join_thread 保证消息完整落管。 + with suppress(Exception): + event_queue.close() + event_queue.join_thread() + os._exit(0) def run_worker_task( @@ -311,10 +321,19 @@ def run_worker_task( failure = message process.join(timeout=10.0) + worker_exit_forcibly = False if process.is_alive(): + # 终态消息 (result/error) 已完整送达, 子进程只是退出收尾慢: + # 强制结束并继续走结果/错误处理, 不把已送达的成功结果当失败丢弃。 process.terminate() process.join(timeout=5.0) - raise BacktestWorkerError("backtest worker returned but did not exit within 10 seconds") + worker_exit_forcibly = True + logger.warning( + "%s worker delivered its terminal message but did not exit within " + "10s; terminated forcibly (exitcode=%s)", + task["kind"], + process.exitcode, + ) if failure is not None: raise BacktestWorkerError( f"{failure.get('message', 'worker failed')}\n{failure.get('traceback', '')}".rstrip() @@ -329,6 +348,7 @@ def run_worker_task( "parent_rss_before_bytes": parent_rss_before, "parent_rss_after_worker_exit_bytes": _rss_bytes(), "worker_exitcode": process.exitcode, + "worker_exit_forcibly": worker_exit_forcibly, } kind = task["kind"] if kind == "backtest": diff --git a/backend/tests/backtest/test_worker_process.py b/backend/tests/backtest/test_worker_process.py index d45e32b..e76193f 100644 --- a/backend/tests/backtest/test_worker_process.py +++ b/backend/tests/backtest/test_worker_process.py @@ -448,6 +448,62 @@ def test_worker_terminates_child_after_cancel_grace(monkeypatch, tmp_path): assert process.exitcode == -15 +def test_worker_accepts_delivered_result_when_child_exit_is_slow(monkeypatch, tmp_path): + """终态消息已送达但子进程退出收尾超时: 应强杀后采纳结果, 而非丢弃报错。""" + + class FakeQueue: + def __init__(self): + self._messages = [{"type": "result", "payload": {"status": "ok"}}] + + def get(self, timeout): + if self._messages: + return self._messages.pop(0) + raise queue.Empty + + def close(self): + pass + + def join_thread(self): + pass + + class FakeEvent: + def set(self): + pass + + class FakeProcess: + def __init__(self): + self.alive = True + self.exitcode = None + + def start(self): + pass + + def is_alive(self): + return self.alive + + def join(self, timeout=None): + pass + + def terminate(self): + self.alive = False + self.exitcode = -15 + + process = FakeProcess() + context = SimpleNamespace( + Queue=FakeQueue, + Event=FakeEvent, + Process=lambda **_kwargs: process, + ) + monkeypatch.setattr(worker_module.mp, "get_context", lambda _method: context) + + result = run_worker_task({"kind": "mining", "data_dir": str(tmp_path), "config": {}}) + + assert result["status"] == "ok" + assert result["worker"]["worker_exit_forcibly"] is True + assert result["worker"]["worker_exitcode"] == -15 + assert process.exitcode == -15 + + def test_spawn_walkforward_skips_folds_before_available_matrix_data(tmp_path): configured_start = date(2024, 1, 1) market_start = configured_start + timedelta(days=4)