chronaxis.batch.process_transport
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""传递独立计划快照和小型结果，保留工作进程中的失败清理责任。23管道仅连接本次创建的可信子进程，不读取外部序列化对象。计划重建后核对身份，4大型结果在子进程内写出并关闭。清理失败的进程暂停接活，由报告句柄远程重试。5"""67from dataclasses import replace8import multiprocessing9from pathlib import Path10from time import monotonic1112from ..planning.plan import AnalysisPlan13from ..planning.checks import json_value14from .._control_wire import ControlChannel15from ..results.budget import _cleanup_owner16from .models import BatchJob, BatchJobResult, CleanupHandle17from .process_tree import ProcessTree18from .._process_start import spawn_entry19from .._batch_worker import worker_main202122def job_packet(job):23 """提取无活动资源的计划。参数: BatchJob。返回: 原生dict。异常: 配置转换错误。"""24 if isinstance(job.plan, Path):25 return dict(job_id=job.job_id, path=str(job.plan.absolute()))26 return dict(job_id=job.job_id, config=json_value(job.plan.raw_config),27 base_dir=str(job.plan.base_dir), identity=job.plan.identity)282930def restore_job(packet):31 """恢复独立计划。参数: 本进程收到的声明。返回: BatchJob。异常: 重建或身份不一致。"""32 if 'path' in packet:33 return BatchJob(packet['job_id'], Path(packet['path']))34 plan = AnalysisPlan.from_dict(packet['config'], base_dir=packet['base_dir'])35 if plan.identity != packet['identity']:36 raise ValueError('跨进程重建后的计划身份不一致')37 return BatchJob(packet['job_id'], plan)383940class ProcessWorker:41 """保存一个可复用工作进程、私有管道及其后代的清理责任。"""4243 def __init__(self, *, fast=False):44 """启动等待命令的spawn进程。参数: 无。返回: 新对象。异常: 启动/树登记失败立即回收。"""45 context = multiprocessing.get_context('spawn')46 self.connection = self._child = self.process = self.tree = self._startup = None47 self.fast_cancel = None48 self.cancel_deadline = self.fast_rows = self.pid = None49 self.closed = self.pending = self._pipe_lost = self._stop_requested = False50 self._constructing = True51 try:52 self.connection, self._child = context.Pipe()53 self.connection = ControlChannel(self.connection,'batch',parent=True)54 self._child = ControlChannel(self._child,'batch',parent=False)55 self.fast_cancel = context.Event() if fast else None56 args = (self._child, self.fast_cancel) if fast else (self._child,)57 try:58 target,args,self._startup=spawn_entry(context,worker_main,args)59 except BaseException as error:60 self._startup=getattr(error,'process_start_cleanup_owner',None)61 raise62 self.process = context.Process(target=target, args=args, daemon=False)63 self.process.start()64 self.pid = self.process.pid65 try:66 self.tree = (ProcessTree(self.pid) if self._startup is None else67 ProcessTree(self.pid,registration=self._startup,process=self.process))68 except BaseException as error:69 self.tree = getattr(error, 'process_tree_cleanup_owner', None)70 raise71 if self._startup is not None:72 self._startup.close();self._startup=None73 self._child.close()74 self._child = None75 self._constructing = False76 except BaseException as error:77 # 未成功返回的工作进程不参与握手；树或管道持续关闭失败仍保留真实owner。78 self._pipe_lost = True79 _cleanup_owner(self, error, 'batch_worker_cleanup_owner', '批量工作进程构造')80 raise8182 def submit(self, job, directory, backend):83 """发送独立任务。参数: job/目录/后端。返回: None。异常: 转换或通信错误。"""84 self.connection.send(('run', job_packet(job), str(directory), backend))8586 def submit_fast(self, job, directory, spec, total):87 """只传准备完成的文件参数，工作进程不重新解析模板或建立下层进程。"""88 packet = dict(job_id=job.job_id, spec=spec, total=total,89 path=str(job.plan) if isinstance(job.plan, Path) else None)90 self.connection.send(('fast', packet, str(directory), None))9192 def cancel_fast(self):93 """通知在途文件，协作退出超过五秒才结束本次拥有的进程树。返回是否已强制回收。"""94 if self.cancel_deadline is None:95 self.fast_cancel.set()96 self.cancel_deadline = monotonic() + 597 elif monotonic() >= self.cancel_deadline:98 self._pipe_lost = True99 self.close()100 return True101 return False102103 def ready(self):104 """查询结果或异常退出。参数: 无。返回: bool。异常: 原管道错误。"""105 try:106 return self.connection.poll() or not self.process.is_alive()107 except (EOFError, OSError) as error:108 if isinstance(error, (EOFError, BrokenPipeError)):109 self._pipe_lost = True110 raise ChildProcessError('分析工作进程管道已断开，PID=' + str(self.pid)) from error111112 def receive(self):113 """取结果并挂远程恢复句柄。参数: 无。返回: BatchJobResult。异常: 进程死亡或协议错误。"""114 try:115 # Windows 同伴退出时，探测或随后的读取都可能先报告断管。116 if not self.connection.poll():117 raise ChildProcessError('分析工作进程异常退出，PID=' + str(self.pid))118 message = self.connection.recv()119 except (EOFError, BrokenPipeError) as error:120 self._pipe_lost = True121 raise ChildProcessError('分析工作进程断开，PID=' + str(self.pid)) from error122 if len(message) == 4 and getattr(self,'fast_cancel',None) is not None:123 kind, result, self.pending, self.fast_rows = message124 else:125 kind, result, self.pending = message126 if kind == 'fast_result' and getattr(self,'fast_cancel',None) is not None and isinstance(result,dict):127 result = BatchJobResult(**result)128 kind = 'result'129 if kind != 'result' or not isinstance(result, BatchJobResult):130 raise RuntimeError('批量工作进程返回无效结果')131 return replace(result, recovery=CleanupHandle(self) if self.pending else None)132133 def close(self):134 """重试远程清理后关闭实际进程树及管道，半构造失败也可恢复。135136 参数: 无。返回: None。异常: 清理失败保留尚未关闭对象供下一次close。137 已关闭进程不再次调用is_alive；构造后未启动的进程也关闭其本地句柄。138 """139 if self.closed:140 return141 process = self.process142 active = (process is not None and getattr(process, '_closed', False) is not True143 and process.pid is not None)144 if active:145 if not self._pipe_lost and not self._stop_requested and process.is_alive():146 try:147 if self.pending:148 self.connection.send(('cleanup',))149 if self.tree is not None and getattr(self.tree,'_linux',False) is True and not self.connection.poll(5):150 self._pipe_lost=True151 raise TimeoutError('等待工作进程清理回复超时；保留进程树关闭责任')152 kind, detail = self.connection.recv()153 if kind != 'cleaned':154 raise RuntimeError('工作进程资源尚未关闭：' + '; '.join(detail))155 self.pending = False156 self.connection.send(('close',))157 self._stop_requested = True158 except (EOFError, BrokenPipeError):159 self._pipe_lost = True160 if self._pipe_lost:161 if not self._stop_requested:162 if self.tree is not None:163 self.tree.close()164 self.tree = None165 # 仅构造失败时作业可能尚未登记该进程，不能由空作业推断进程已结束。166 if getattr(self, '_constructing', False) and process.is_alive():167 process.kill()168 elif process.is_alive():169 process.kill()170 self._stop_requested = True171 process.join(timeout=5)172 if process.is_alive():173 raise TimeoutError('等待断管工作进程退出超时，PID=' + str(self.pid))174 else:175 if self.tree is not None and getattr(self.tree,'_linux',False) is True:176 process.join(timeout=5)177 if process.is_alive():178 self.tree.close();self.tree=None179 raise TimeoutError('工作进程未按关闭协议退出')180 else:process.join()181 if self.tree is not None:182 self.tree.close()183 self.tree = None184 for name in ('connection', '_child', '_startup'):185 connection = getattr(self, name, None)186 if connection is not None:187 try:188 connection.close()189 finally:190 if connection.closed:191 setattr(self, name, None)192 if process is not None and getattr(process, '_closed', False) is not True:193 process.close()194 self.fast_cancel = None195 self.pending, self.closed = False, True196