chronaxis.batch.runner
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""在外层多进程执行独立计划，复用公共AnalysisRunner完成分析、发布和关闭。23每个任务拥有自己的RunSession、预算和目录；不拆分一个会话的时间轴，不共用4活动、迟滞或配对状态。最多process_count个future在途，未开始的任务不读计划文件。5取消或fail_fast停止派发，已经开始的任务正常完成关闭；不向工作进程注入异常。6只承诺每任务原有预算及同时运行数，不承诺全进程RSS或固定进程数带来的性能收益。7"""89from dataclasses import replace10import time11import os12from pathlib import Path13from threading import Event14from time import perf_counter1516from ..planning import load_analysis_plan17from ..planning.plan import AnalysisPlan18from ..runner import AnalysisRunner19from ..results.spool import _cleanup_owner20from .._batch_worker import _diagnostics21from .models import BatchJob, BatchJobResult, BatchReport, CleanupHandle222324def _new_root(value) -> Path:25 """检查可复用批量目录。参数: value为str/Path。返回: 绝对Path。异常: 类型/已有/只读路径错误。"""26 from ..results.publication import output_root27 return output_root(value)282930def _facts(job, plan=None) -> dict:31 """区分配置和输入事实。参数: job原声明；plan可选已加载计划。返回: 字段dict。异常: 无。"""32 loaded = plan if plan is not None else job.plan if isinstance(job.plan, AnalysisPlan) else None33 result = dict(job_id=job.job_id, plan_path=str(job.plan.absolute()) if isinstance(job.plan, Path) else None)34 if loaded is not None:35 result.update(plan_identity=loaded.identity, dataset_id=loaded.dataset.dataset_id,36 entity_id=loaded.dataset.entity, session_id=loaded.dataset.session_id,37 memory_budget_bytes=loaded.execution.memory_budget_bytes)38 return result394041def _cancelled(job, reason) -> BatchJobResult:42 """记录未启动取消。参数: job原声明；reason原因。返回: 取消结果。异常: 无，不读取计划。"""43 return BatchJobResult(**_facts(job), status='cancelled', error_type='not_started', error_message=reason)444546def _report(root, jobs, results, process_count, fail_fast, cancelled, label, reason) -> BatchReport:47 """补齐未启动项。参数: 原声明/结果及调度选项。返回: 完整报告。异常: 模型错误。"""48 outcomes = tuple(_cancelled(job, reason) if value is None else value49 for job, value in zip(jobs, results))50 # worker中的真实KeyboardInterrupt也属于取消请求，不只记录主线程的Event。51 requested = cancelled or any(j.error_type in ('KeyboardInterrupt', 'SystemExit') for j in outcomes)52 return BatchReport(root, outcomes, process_count, fail_fast, requested, label)535455def _run_job(job, directory, backend, stop, cancel) -> BatchJobResult:56 """执行一个独立任务并关闭所有资源。5758 参数: job为BatchJob；directory为独占新子目录；backend为覆盖；stop/cancel为Event。59 返回: 含实际发布和身份的最终结果；恢复失败保最高owner，不把异常栈留在结果中。60 异常: 普通分析/发布/取消转换成该任务结果；无法构造结果的内部错误交由调度器收尾。61 """62 if stop.is_set() or cancel.is_set():63 return _cancelled(job, '任务启动前已停止派发')64 started = perf_counter()65 plan = result = manifest = recovery = None66 metadata = {'worker_pid': os.getpid()}67 try:68 # 路径计划到本任务实际开始才加载，未派发的坏文件不会影响其他任务。69 plan = job.plan if isinstance(job.plan, AnalysisPlan) else load_analysis_plan(job.plan)70 directory.mkdir(parents=True, exist_ok=False)71 result = AnalysisRunner(backend).run(plan, work_dir=directory / 'work')72 metadata.update(input_identity=result.metadata.input_identity, run_id=result.metadata.run_id)73 manifest = result.write(directory / 'output')74 metadata['output_dir'] = manifest.output_dir75 metadata['manifest_path'] = manifest.manifest_path76 result.close()77 return BatchJobResult(**_facts(job, plan), **metadata, status='completed',78 elapsed_seconds=perf_counter() - started)79 except BaseException as error:80 completed = getattr(error, 'completed_manifest', None)81 if completed is not None:82 metadata.update(output_dir=completed.output_dir, manifest_path=completed.manifest_path)83 # 结果已经返回后由本层负责close；未返回的运行恢复owner继续沿AnalysisRunner原协议。84 if result is not None:85 _cleanup_owner(result, error, 'batch_cleanup_owner', '批量任务结果')86 owner = getattr(error, 'batch_cleanup_owner', None)87 if owner is None:88 owner = getattr(error, 'runner_cleanup_owner', None)89 # AnalysisRunner可能保留已经成功关闭的恢复属性；只有仍实际持有资源才转交报告。90 if owner is not None and not (getattr(owner, 'closed', False) or getattr(owner, '_closed', False)):91 recovery = CleanupHandle(owner)92 cancelled = isinstance(error, (KeyboardInterrupt, SystemExit))93 # 资源耗尽或真实取消应停止新派发；普通业务失败由fail_fast决定是否停止。94 if cancelled or isinstance(error, MemoryError):95 stop.set()96 return BatchJobResult(**_facts(job, plan), **metadata,97 status='cancelled' if cancelled else 'failed',98 error_type=type(error).__name__, error_message=str(error),99 error_chain=_diagnostics(error), recovery=recovery,100 elapsed_seconds=perf_counter() - started)101102103def _pause():104 """让出调度CPU等待下一次结果。参数: 无。返回: None。异常: 主进程取消。"""105 time.sleep(0.01)106107108class BatchRunner:109 """通过有限spawn工作进程执行独立任务；同一任务的连续文件保持原顺序。"""110111 def __init__(self, process_count=1, *, fail_fast=False, backend=None):112 """配置调度。参数: process_count正整数；fail_fast布尔；backend可选计算实现。113114 返回: None。异常: 参数不合法；构造时不启动进程或读取业务输入。115 """116 if type(process_count) is not int or process_count < 1:117 raise ValueError('process_count需要正原生int')118 if type(fail_fast) is not bool:119 raise TypeError('fail_fast需要bool')120 if backend is not None:121 AnalysisRunner(backend)122 self.process_count, self.fail_fast, self.backend = process_count, fail_fast, backend123124 def run(self, jobs, output_dir, *, cancel_event=None, device_label='pending') -> BatchReport:125 """运行独立任务，原声明顺序返回报告。126127 参数: jobs为BatchJob tuple；output_dir可复用目录；cancel_event可选threading.Event；128 device_label为报告来源标签。返回: BatchReport，close可重试子进程内失败清理。129 异常: 声明和路径错误在启动前拒绝；调度基础设施错误携带batch_report。130 副作用: 启动最多process_count个工作进程，每任务独立发布；取消只阻止未派发任务。131 """132 return self._run(jobs, output_dir, cancel_event=cancel_event, device_label=device_label)133134 def _run(self, jobs, output_dir, *, cancel_event, device_label, fast_specs=None, fast_rows=None):135 """共享派发/清理循环；fast_specs 仅由 fast 校验入口构造，可选 fast_rows 接收汇总所需小行。"""136 from .process_transport import ProcessWorker137 if type(jobs) is not tuple or any(not isinstance(j, BatchJob) for j in jobs):138 raise TypeError('批量任务需要BatchJob tuple')139 if len({j.job_id.casefold() for j in jobs}) != len(jobs):140 raise ValueError('批量job_id不能重复，大小写冲突也拒绝')141 if cancel_event is not None and not isinstance(cancel_event, Event):142 raise TypeError('cancel_event需要threading.Event')143 if type(device_label) is not str or not device_label.strip():144 raise ValueError('device_label需要非空文本')145 # 有待派发任务的独立batch即使上限为1也会创建受管工作进程。146 # 在创建输出前提示；不修改既有派发、失败报告或退出码。147 from ..process_capabilities import warn_managed_worker148 warn_managed_worker(bool(jobs) and not (cancel_event is not None and cancel_event.is_set()),149 '本独立批量任务')150 root = _new_root(output_dir)151 from ..results.publication import PublicationNames152 publication = PublicationNames(root, (*[j.job_id for j in jobs], 'batch_report.json'))153 cancel = Event() if cancel_event is None else cancel_event154 workers, free, active, last_job = [], [], {}, {}155 results = [None] * len(jobs)156 next_job, stopped, interrupted, infrastructure = 0, False, False, None157 while next_job < len(jobs) or active:158 try:159 if cancel.is_set():160 stopped = True161 # 先收齐本轮已完成结果再派发，避免观察到失败后继续启动下一任务。162 for worker, index in list(active.items()):163 try:164 if worker.ready():165 outcome = worker.receive()166 if fast_rows is not None and outcome.status == 'completed':167 fast_rows[jobs[index].job_id] = worker.fast_rows168 elif (fast_specs is not None and (cancel.is_set() or interrupted or infrastructure is not None)169 and worker.cancel_fast()):170 outcome = replace(_cancelled(jobs[index], '取消后工作进程未及时退出，已回收'),171 worker_pid=worker.pid,error_type='cancel_timeout')172 else:173 continue174 except Exception as error:175 outcome = BatchJobResult(**_facts(jobs[index]), status='failed',176 worker_pid=worker.pid, error_type=type(error).__name__,177 error_message=str(error), error_chain=_diagnostics(error))178 infrastructure = infrastructure or error179 stopped = True180 results[index] = outcome181 del active[worker]182 if self.fail_fast and outcome.status == 'failed':183 stopped = True184 if outcome.error_type in ('MemoryError', 'KeyboardInterrupt', 'SystemExit'):185 stopped = True186 if not worker.pending:187 free.append(worker)188 while not stopped and next_job < len(jobs):189 if cancel.is_set():190 stopped = True191 break192 if free:193 worker = free.pop()194 elif len(workers) < min(self.process_count, len(jobs)):195 worker = ProcessWorker() if fast_specs is None else ProcessWorker(fast=True)196 workers.append(worker)197 else:198 break199 index = next_job200 try:201 directory = root / publication.names[jobs[index].job_id]202 if fast_specs is None:203 worker.submit(jobs[index], directory, self.backend)204 else:205 worker.submit_fast(jobs[index], directory, fast_specs[jobs[index].job_id], fast_rows is not None)206 except Exception as error:207 results[index] = BatchJobResult(**_facts(jobs[index]), status='failed',208 worker_pid=worker.pid, error_type=type(error).__name__, error_message=str(error))209 infrastructure = infrastructure or error210 stopped = True211 break212 active[worker], last_job[worker] = index, index213 next_job += 1214 if not active:215 # 所有工作进程若都因未完成清理而暂停，剩余任务不再派发。216 break217 _pause()218 except KeyboardInterrupt as error:219 interrupted, stopped = True, True220 # 构造取消且清理持续失败时，原异常拥有尚未关闭的真实worker；221 # 不能仅返回取消报告而丢失该恢复入口。已清理的Ctrl+C仍按原流程返回。222 if getattr(error, 'batch_worker_cleanup_owner', None) is not None:223 infrastructure = infrastructure or error224 except Exception as error:225 infrastructure, stopped = infrastructure or error, True226 # 在途任务已结束；持续关闭失败留远程owner，不因结束调度而丢掉恢复入口。227 for worker in workers:228 if worker.pending:229 continue230 try:231 worker.close()232 except BaseException as error:233 index = last_job.get(worker)234 if index is not None:235 results[index] = replace(results[index], status='failed',236 error_type=type(error).__name__, error_message=str(error),237 recovery=CleanupHandle(worker))238 infrastructure = infrastructure or error239 reason = '用户取消，尚未启动' if cancel.is_set() or interrupted else '失败或未完成清理后停止派发'240 report = _report(root, jobs, results, self.process_count, self.fail_fast,241 cancel.is_set() or interrupted, device_label, reason)242 report = replace(report, worker_process_count=len(workers), _publication=publication)243 if infrastructure is not None:244 report = replace(report, infrastructure_error=type(infrastructure).__name__ + ': ' + str(infrastructure))245 infrastructure.batch_report = report246 raise infrastructure247 return report248