chronaxis._batch_worker

源码基线：8904c2f744951857659abd40aaa3535d57faf53c

1"""批量工作进程控制消息；只在收到对应任务时加载计算实现。"""2from dataclasses import replace3import os4from pathlib import Path5import signal6from threading import Event7from ._control_wire import ControlChannel89def _diagnostics(error) -> tuple[str, ...]:10    """保留有限异常链而不持有栈。参数: error原异常。返回: 文本tuple。异常: 原异常格式错误。"""11    seen, details = set(), []12    while error is not None and id(error) not in seen:13        seen.add(id(error))14        details.append(type(error).__name__ + ': ' + str(error))15        details.extend(getattr(error, '__notes__', ()))16        error = error.__cause__17    return tuple(details)181920def worker_main(connection, fast_cancel=None):21    """处理run/cleanup/close命令。参数: 私有管道。返回: None。异常: 断管退出，父进程负责树清理。"""22    if type(connection) is not ControlChannel:23        raise TypeError('batch工作进程需要固定字节控制端点')24    if fast_cancel is None:25        from .batch.runner import _run_job26        from .planning import process_execution27        process_execution.IN_BATCH_WORKER = True28        from .batch.process_transport import restore_job29        from .batch.models import BatchJobResult30    # Ctrl+C只由调度主进程接收，正在写结果的工作进程继续完整收尾。31    signal.signal(signal.SIGINT, signal.SIG_IGN)32    recovery = None33    try:34        while True:35            command = connection.recv()36            if command[0] == 'close':37                if recovery is not None:38                    recovery.close()39                return40            if command[0] == 'cleanup':41                try:42                    if recovery is not None:43                        recovery.close()44                    recovery = None45                    connection.send(('cleaned', None))46                except BaseException as error:47                    connection.send(('cleanup_failed', _diagnostics(error)))48                continue49            if recovery is not None:50                raise RuntimeError('存在未关闭资源的工作进程不能接收新任务')51            _, packet, directory, backend = command52            if fast_cancel is not None:53                if command[0] != 'fast':54                    raise RuntimeError('fast 工作进程收到未知命令')55                from .fast.worker import file_response56                connection.send(file_response(packet, Path(directory), fast_cancel))57                continue58            try:59                job = restore_job(packet)60                outcome = _run_job(job, Path(directory), backend, Event(), Event())61            except BaseException as error:62                outcome = BatchJobResult(packet['job_id'], 'failed',63                    plan_path=packet.get('path'), error_type=type(error).__name__,64                    error_message=str(error), error_chain=_diagnostics(error))65            recovery = outcome.recovery66            pending = recovery is not None and not recovery.closed67            # 只发送冻结的纯结果字段；CleanupHandle的原资源必须留在拥有它的进程。68            response = ('result', replace(outcome, recovery=None, worker_pid=os.getpid()), pending)69            connection.send(response)70    except EOFError:71        if recovery is not None:72            recovery.close()73    finally:74        connection.close()75