chronaxis.fast.batch

源码基线：8904c2f744951857659abd40aaa3535d57faf53c

1"""先检查整批 fast 配置，再交给文件级工作进程；结果按声明顺序发布。"""23from dataclasses import replace4from pathlib import Path56from ..batch.models import BatchPlan7from ..batch.publication import _write_batch_report8from ..batch.runner import BatchRunner9from ..configuration import ConfigurationError, load_configuration10from ..configuration.diagnostics import locate_error11from ..planning.checks import PlanValidationError, configuration_error12from ..planning.dbc_defaults import scoped_dbc_cache13from .config import compile_config14from .worker import _dbc_digest151617@scoped_dbc_cache()18def prepare_batch(plan_path):19    """返回批量计划、按输入顺序的可序列化文件参数及模板标识。此时不读报文。"""20    path = Path(plan_path).absolute()21    try:22        document = load_configuration(path)23    except ConfigurationError as exc:24        raise configuration_error(exc) from exc25    raw = document.data26    try:27        plan = BatchPlan.from_dict(raw,base_dir=path.parent)28        if 'backend' in raw:29            raise PlanValidationError('$.backend','fast 批量入口已确定执行方式，不能指定 backend')30    except PlanValidationError as exc:31        raise locate_error(exc,document)32    template = 'template' in raw33    specs, errors, blocked = {}, [], []34    for job in plan.jobs:35        try:36            if template:37                from .config import compile_template_plan38                specs[job.job_id] = compile_template_plan(job.plan,template_path=path.parent/raw['template'])39            else:40                specs[job.job_id] = compile_config(job.plan)41            if specs[job.job_id].get('process_count', 1) != 1:42                raise PlanValidationError('$.execution.process_count','独立fast批量任务内部不能请求多进程')43        except PlanValidationError as exc:44            errors.extend(exc.diagnostics)45            blocked.extend(exc.blocked_checks)46    if errors:47        raise PlanValidationError.from_diagnostics(errors,blocked)48    return plan, specs, template495051def run_batch(plan_path, output, *, process_count=None, cancel_event=None, total=False):52    """运行独立 fast jobs/template 配置，返回本次带时间的批量报告路径。5354    失败或取消任务的报告仍可打开。基础设施异常保留 batch_report 以供检查或重试清理。55    process_count 省略时使用批量计划配置，total 只接收显式布尔值。56    """57    if type(total) is not bool:58        raise TypeError('total 需要布尔值')59    plan, specs, template = prepare_batch(plan_path)60    accumulator = outcomes = None61    if total:62        from .totals import FastTotals63        accumulator = FastTotals.from_specs(specs,template=template)64        if specs:65            digest = _dbc_digest(next(iter(specs.values()))['dbc'])66            for spec in specs.values():67                spec['_dbc_digest'] = digest68        outcomes = dict.fromkeys(specs)69    count = plan.process_count if process_count is None else process_count70    runner = BatchRunner(count,fail_fast=plan.fail_fast)71    extra = dict(engine='base_hpc_v1',budget_scope='managed_streaming_per_job')72    try:73        report = runner._run(plan.jobs,output,cancel_event=cancel_event,device_label='pending',74                             fast_specs=specs,fast_rows=outcomes)75    except BaseException as exc:76        report = getattr(exc,'batch_report',None)77        if report is not None:78            try:79                path = _write_batch_report(report,extra)80                exc.batch_report = replace(report,report_path=path)81            except BaseException as publication_error:82                exc.add_note('无法完成批量失败报告：' + str(publication_error))83        raise84    try:85        if accumulator is not None:86            try:87                extra['totals'] = accumulator.finish(outcomes)88            except OverflowError as exc:89                report = replace(report,infrastructure_error='总量溢出：'+str(exc))90        return _write_batch_report(report,extra)91    finally:92        report.close()93