diff --git a/sanguo_orchestrator/pool.py b/sanguo_orchestrator/pool.py index e23d283..ac0bc27 100644 --- a/sanguo_orchestrator/pool.py +++ b/sanguo_orchestrator/pool.py @@ -4,6 +4,7 @@ Provides task storage and status tracking (not actual multiprocessing) """ import logging from concurrent.futures import ProcessPoolExecutor, Future +from concurrent.futures.process import BrokenProcessPool from multiprocessing import get_context from .task import Task, TaskState @@ -28,9 +29,21 @@ class TaskPool: return task def submit_work(self, task_id: str, func, *args) -> Future: - """Submit work to the process pool executor""" + """Submit work to the process pool executor. + + worker 被 OOM 击杀后 ProcessPoolExecutor 永久破损(BrokenProcessPool), + 重建池重试一次,避免一次异常让任务中心瘫痪到容器重启。 + """ logger.debug("submit_work task_id=%s", task_id) - return self.executor.submit(func, *args) + try: + return self.executor.submit(func, *args) + except BrokenProcessPool: + logger.warning("pool broken (worker died?), rebuilding for %s", task_id) + self.executor.shutdown(wait=False) + self.executor = ProcessPoolExecutor( + max_workers=self.max_workers, mp_context=get_context("spawn") + ) + return self.executor.submit(func, *args) def update_stage(self, task_id: str, stage: str): """Update the stage of a task"""