chronaxis.execution.run_session
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""执行当前分析计划并管理一次运行的资源。23公开AnalysisRunner选择实际计算后端后交给本会话。构造只检查声明、申请有限预算4并创建独占工作目录；execute调用记录流执行树，关闭输入，再把结果及导出文件5移交给AnalysisResult。默认依据真实相邻记录的时间间隔计算，逐消息累计由同一6执行树按原消息语义处理；不根据旧配置版本选择另一套流程。78错误保留原异常和明确的清理所有者。只有各层实际关闭后才退还对应费用；成功移交9后的结果由调用方关闭，本会话不重复关闭它。现场留存仅由diagnostics.retain_work10决定；目录删除始终核对本次创建身份，不触碰已有目录或其他任务的文件。11"""1213# 协调层只调用已有生产协议；UUID/时区时间是运行身份，不参与稳定输入摘要。14from datetime import datetime, timezone15from pathlib import Path16import tempfile17from uuid import uuid418from ..planning.plan import AnalysisPlan19from ..planning.legacy_execution import require_can_decode_scope20from ..planning.input_requirements import collect_input_requirements21from ..results.budget import ResourceBudget22from ..results.errors import AnalysisError, ResultClosedError23from ..results.manifest import RunMetadata24from ..results.result import AnalysisResult25from ..results.spool import ResultSpool, _cleanup_owner, _reserve_owned_ram, _release_owned_ram26from .history import _metadata_bytes27from .static_metadata import static_metadata_bytes282930class RunSession:31 """拥有一次执行的资源树和协调费用，原计划仅借用且不重新解析。3233 _constructing标识当前未返回工厂的最高槽。记录执行树负责其输入与计算子资源，34 本会话负责执行树、最终暂存及结果的唯一移交。35 _result在最后费用释放成功前仍自有；_completed后移交且_closed为True。36 """3738 def __init__(self, plan, work_dir=None):39 """申请有限固定状态，拒绝已有或受保护目录，不读取来源。4041 参数: plan为AnalysisPlan，auto在资源申请前解析；work_dir为全新路径，None创建42 唯一临时根。返回: 单次执行owner。异常: 类型、路径、预算或取消原错；43 持续构造清理失败附runner_cleanup_owner，已有目录不会被写入诊断。44 """45 # 类型拒绝不接管调用方计划；其来源与raw_config从未复制或修改。46 if not isinstance(plan, AnalysisPlan) or work_dir is not None and not isinstance(work_dir, (str, Path)):47 raise TypeError('RunSession需AnalysisPlan及可选新工作路径')48 from ..planning.backend_selection import execution_plan49 self._automatic_backend = plan.execution.backend == 'auto'50 plan = execution_plan(plan)51 # typed计划也可直接构造；先检查有限策略值，实际需求在支付协调费用后统一计算。52 require_can_decode_scope(plan)53 from ..planning.export_options import require_export_options54 require_export_options(plan)55 self._plan = plan56 self._budget = ResourceBudget(plan.execution.memory_budget_bytes, plan.execution.spool_quota_bytes)57 self._requirements = None58 self._fixed_cost = self._metadata_cost = self._held_cost = 059 self._record_calculations = None60 self._spool = self._result = self._metadata = None61 self._constructing = None62 self._artifact_exports = None63 self._path = self._run_id = self._started_at = None64 self._path_identity = None65 self._owns_path = False66 self._retain_work = plan.diagnostics.retain_work67 self._started = self._completed = self._closed = self._closing = self._failed = False68 try:69 # 先付协调数组、路径和运行标识的有限槽，再构造这些容器。70 cost = (65536 + 8192 * len(plan.sources) + 4096 * len(plan.signals)71 + 16 * _metadata_bytes((plan.identity, plan.dataset.dataset_id,72 plan.dataset.entity, plan.dataset.session_id, plan.dataset.domain, plan.dataset.timebase))73 + (0 if work_dir is None else 32 * _metadata_bytes(str(work_dir))))74 _reserve_owned_ram(self, cost, '_fixed_cost')75 # 原计划共享同一套绑定；按对象身份收一次声明费用，不按计算项重复展开。76 # 新需求集合/定位索引另按声明个数付费，不用共享树大小乘放大系数。77 declarations = (plan.bindings, plan.rules, plan.windows, plan.calculations,78 plan.message_calculations, plan.intervals, plan.derived)79 index_count = (len(plan.signals) + len(plan.sources) + len(plan.rules)80 + len(plan.windows) + len(plan.calculations) + len(plan.message_calculations))81 _reserve_owned_ram(self, 65536 + static_metadata_bytes(declarations)82 + 4096 * index_count, '_fixed_cost')83 self._require_static_derived()84 self._requirements = collect_input_requirements(plan)85 # 先解析实际工作路径，再独占创建本次目录并保存其身份。86 candidate = Path(tempfile.gettempdir()) if work_dir is None else Path(work_dir)87 candidate = candidate.resolve()88 if work_dir is None:89 self._path = Path(tempfile.mkdtemp(prefix='chronaxis-run-', dir=candidate))90 else:91 candidate.mkdir(parents=True, exist_ok=False)92 self._path = candidate93 from ..results.owned_directory import directory_identity94 self._path_identity = directory_identity(self._path)95 self._owns_path = True96 self._run_id = uuid4().hex97 self._started_at = datetime.now(timezone.utc).isoformat()98 except BaseException as exc:99 # 本构造只有自有有限费用与新根；关闭不删除文件，也不向未取得的目录写诊断。100 self._failed = True101 exc.runner_cleanup_owner = self102 _cleanup_owner(self, exc, 'runner_cleanup_owner', '运行会话构造')103 raise104105 def _require_static_derived(self):106 """在需求收集和输入动作前复核全部派生声明，不把派生项变成运行根。107108 参数: 无，借用本会话原计划及声明绑定。109 返回: None；原静态对象、声明顺序和完整身份均须一致，空任务也不省略校验。110 异常: TypeError/ValueError为typed声明或静态对象被替换；预算和取消错误原样传播。111 所有解析表、遍历容器和展开树先付到本owner的_fixed_cost槽，构造失败由112 原RunSession清理链回收；本方法不创建目录、不打开来源、不生成普通样本。113 """114 from ..core.expression import ExprNode, StaticExpression, validate_expression115 plan = self._plan116 definitions = definition = static = checked = stack = sizes = node = None117 try:118 if (type(plan.derived) is not tuple or type(plan.derived_expressions) is not tuple119 or len(plan.derived) != len(plan.derived_expressions)120 or any(type(item) is not StaticExpression for item in plan.derived_expressions)):121 raise ValueError('计划派生与静态表达式数量或类型不一致')122 if not plan.derived and not plan.bindings.derived_names:123 return124 # 使用协调owner既有费用槽；解析表、遍历栈和展开树都在创建前付款。125 _reserve_owned_ram(self, 65536 + 32 * _metadata_bytes(plan.derived), '_fixed_cost')126 definitions = tuple(item.definition for item in plan.derived)127 if tuple(item.name for item in definitions) != plan.bindings.derived_names:128 raise ValueError('计划派生与原绑定名称空间不一致')129 total = 0130 for static in plan.derived_expressions:131 # 与历史阶段同样计共享DAG的展开边，不能按唯一节点低估repr空间。132 _reserve_owned_ram(self, 4096, '_fixed_cost')133 stack, sizes = [(static.root, False)], {}134 while stack:135 node, expanded = stack.pop()136 if not isinstance(node, ExprNode):137 raise TypeError('计划静态表达式需ExprNode树')138 if id(node) in sizes:139 continue140 if not expanded:141 _reserve_owned_ram(self, 4096 + 512 * len(node.args), '_fixed_cost')142 stack.append((node, True))143 stack.extend((child, False) for child in node.args)144 continue145 own = _metadata_bytes((node.operator, node.dtype, node.unit, node.unit_status,146 node.value, node.name, node.labels, node.aliases, node.constant,147 node.allow_float_literal))148 sizes[id(node)] = 4096 + 16 * own + sum(sizes[id(child)] for child in node.args)149 total += sizes[id(static.root)] + 16 * _metadata_bytes((150 static.text, static.normalized_ast, static.dependencies))151 stack = sizes = node = None152 _reserve_owned_ram(self, (1 + len(definitions)) * total, '_fixed_cost')153 for definition, static in zip(definitions, plan.derived_expressions):154 if static.text != definition.name:155 raise ValueError('计划派生静态表达式需对应原名称')156 checked = validate_expression(definition.name, plan.bindings,157 derived=definitions, unit=definition.unit)158 if checked != static:159 raise ValueError('计划派生与原完整静态表达式不一致')160 checked = None161 finally:162 # 构造失败后最高owner会退款，异常栈先撤掉新树和遍历容器的引用。163 plan = definitions = definition = static = checked = stack = sizes = node = None164165 @property166 def closed(self):167 """读取协调owner关闭状态。参数: 无。返回: bool。异常: 无。168 """169 return self._closed170171 @property172 def failed(self):173 """读取执行或本owner清理失败事实。参数: 无。返回: bool。异常: 无。174 """175 return self._failed176177 @property178 def work_dir(self):179 """读取本次已取得的新根。参数: 无。返回: Path或构造未取得时None。异常: 无。180 """181 return self._path182183 def _construct(self, slot, factory, *args, **kwargs):184 """调用真实工厂并立即登记返回owner，失败只接其明确恢复属性。185186 参数: slot是本会话有限责任槽；factory/args/kwargs为既有生产入口。187 返回: 实际owner。异常: 原工厂错误传播，不对旧半初始化对象调用close。188 """189 self._constructing = slot190 try:191 owner = factory(*args, **kwargs)192 setattr(self, slot, owner)193 return owner194 except BaseException as exc:195 # 在标识仍为当前工厂时接管；其他阶段内部spool不能误当最终spool。196 self._adopt_cleanup(exc)197 raise198 finally:199 self._constructing = None200201202 def execute(self):203 """读取一次并返回独立AnalysisResult，重复调用或失败后禁止重新扫描。204205 参数: 无。返回: 完成且尚未关闭的AnalysisResult。异常: 原执行/预算/取消206 错误保留，普通非AnalysisError由公开AnalysisRunner包装；本方法不反复清理，原错207 附runner_cleanup_owner供公开层统一清理或恢复后close。208 """209 if self._started or self._closed or self._closing or self._failed:210 raise ResultClosedError('RunSession只能执行一次，且不能在关闭或失败后推进')211 self._started = True212 try:213 self._run_record_calculations()214 self._close_inputs()215 # 这里只核一致性，不用差额替旧工厂抵账；残留费用不能被成功结果掩盖。216 self._assert_owned_balance()217 self._spool.seal()218 result_options = ({'owned_work': (self._path, self._path_identity)})219 self._construct('_result', AnalysisResult, self._spool, self._metadata,220 table_format=self._plan.outputs.tables, artifact_exports=self._artifact_exports,221 **result_options)222 self._owns_path = False223 # 构造成功时实际result已拥有同一导出仓；在父槽可达后才撤会话责任。224 self._artifact_exports = None225 self._spool = None226 # 结果暂留最高槽直到本owner全部费用成功释放；最后取消仍能从会话关真实结果。227 self._clear_contracts()228 self._release_costs()229 self._assert_owned_balance()230 result, self._result = self._result, None231 self._completed = self._closed = True232 return result233 except BaseException as exc:234 self._failed = True235 self._adopt_cleanup(exc)236 self._diagnose(exc)237 exc.runner_cleanup_owner = self238 raise239240 def _run_record_calculations(self):241 """在原会话资源树内执行按需周期统计，保持一个公开AnalysisRunner。242243 参数: 无。返回: None，完成结果仍由execute统一封存和移交。244 异常: 静态未支持组合在I/O前拒绝，实际子owner统一登记及清理。245 """246 from ..planning.record_execution import record_execution_errors247 from .record_session_calculations import RecordSessionCalculations248 plan = self._plan249 errors = record_execution_errors(plan)250 if errors:251 raise AnalysisError('; '.join(errors))252 self._construct('_spool', ResultSpool, self._budget, self._path / 'result',253 result_buffer_rows=plan.execution.result_buffer_rows,254 requested_tables=(('events',) if plan.rules else ())+(('calculations',) if any(m.operation!='select' for m in plan.calculations) or plan.message_calculations else ())255 +(('sequences',) if plan.sequences else ())+ (('segments',) if plan.windows else ())256 +(('contributions',) if plan.outputs.contributions=='full' and plan.message_calculations else ())257 +(('input_trace',) if plan.diagnostics.trace else ())258 +(('selections',) if any(m.operation=='select' for m in plan.calculations) else ())259 +(('aggregations',) if plan.aggregations else ()),260 enable_sequences=bool(plan.sequences),261 enable_contributions=plan.outputs.contributions=='full' and bool(plan.message_calculations),262 enable_trace=plan.diagnostics.trace,263 enable_selections=any(m.operation=='select' for m in plan.calculations),264 enable_aggregations=bool(plan.aggregations),265 failure_diagnostics=plan.diagnostics.retain_work)266 self._construct('_record_calculations', RecordSessionCalculations, plan, self._budget, self._path,267 requirements=self._requirements, declaration_owner=self,268 direct_integrals=self._automatic_backend)269 self._record_calculations.execute_into(self._spool)270 if self._record_calculations._exports is not None:271 self._record_calculations._exports.take_exports(self)272 diagnostic_inputs=(() if self._record_calculations._diagnostics is None else273 self._record_calculations._diagnostics.summaries(self._record_calculations._cursors))274 frame_notices = tuple((cursor.source_name, notice) for cursor in self._record_calculations._cursors275 for notice in cursor.frame_format_notices)276 analysis_outcome=self._record_calculations.analysis_outcome()277 _reserve_owned_ram(self, 65536 + 32 * _metadata_bytes((plan.metadata,plan.presets,diagnostic_inputs,frame_notices,analysis_outcome,plan.aggregations)), '_metadata_cost')278 self._metadata = RunMetadata(run_id=self._run_id, input_identity=None,279 plan_hash=plan.identity, backend=plan.execution.backend, config_version=plan.config_version,280 source_time_interpretation='original-record-times',281 session_boundaries=(), session_ids=(plan.dataset.session_id,),282 started_at=self._started_at, parsing_policy='requested-fields-and-driver-times',283 device_label=plan.metadata.get('device_label', 'pending'), retain_work=plan.diagnostics.retain_work,284 threshold_durations=self._record_calculations.threshold_summaries(),input_diagnostics=diagnostic_inputs,285 frame_format_notices=frame_notices,analysis_outcome=analysis_outcome,aggregation_definitions=plan.aggregations,286 limitations=tuple(f'{source}.{notice.alias}: DBC声明CAN FD，按显式经典CAN定位读取；原帧未转换'287 for source, notice in frame_notices),288 trace_requested=plan.diagnostics.trace,integrity_diagnostics=plan.diagnostics.integrity)289290 def _adopt_cleanup(self, exc):291 """接管当前工厂明确移交的最高资源所有者，保留失败后可重试关闭的责任。292293 参数: exc为原构造错误。返回: None。异常: 原属性访问错误；不凭字段猜测内层资源。294 """295 if self._constructing == '_record_calculations':296 owner = getattr(exc, 'record_session_cleanup_owner', None)297 if owner is not None and self._record_calculations is None:298 self._record_calculations = owner299 elif self._constructing in ('_result', '_spool'):300 result = getattr(exc, 'analysis_result_cleanup_owner', None)301 if result is not None:302 self._result, self._spool = result, None303 else:304 spool = getattr(exc, 'result_spool_cleanup_owner', None)305 if spool is not None:306 self._spool = spool307308 def _diagnose(self, exc):309 """在本次已取得的最终暂存标失败，诊断清理副错只附note。310311 参数: exc为实际原错误。返回: None。异常: 不替换主异常，不接触未取得或已有根。312 """313 spool = self._spool if self._result is None else self._result.spool314 if spool is not None and spool.path is not None:315 try:316 if self._result is not None:317 self._result.status = 'cancelled' if isinstance(exc, (KeyboardInterrupt, SystemExit)) else 'failed'318 spool._fail(exc)319 except BaseException as diagnostic:320 exc.add_note(f'运行会话失败诊断附加失败: {type(diagnostic).__name__}')321322 @staticmethod323 def _require_closed(owner):324 """检查真实close完成及已知费用槽，不让旧提前closed掩盖尚持有资源。325326 参数: owner为刚关闭的实际对象。返回: None。异常: AnalysisError为未完成关闭；327 旧深层缺口只据真实槽拒绝，不推测、转移或自行释放未知余额。328 """329 if not (getattr(owner, 'closed', False) or getattr(owner, '_closed', False)):330 raise AnalysisError('内部资源close返回但没有完成关闭')331 for slot in ('_held_cost', '_input_cost', '_state_cost', '_pending_cost', '_cost'):332 if getattr(owner, slot, 0):333 raise AnalysisError(f'内部资源关闭仍持有{slot}，需恢复后由实际owner处理')334335 def _close_slot(self, slot):336 """关闭一个最高owner并在成功后才清槽。参数: slot责任名。返回: None。异常: 原关闭错误。337 """338 owner = getattr(self, slot)339 if owner is not None:340 owner.close()341 self._require_closed(owner)342 setattr(self, slot, None)343344345 def _close_inputs(self):346 """关闭记录执行树，失败时保留其最高所有者供调用方重试。347348 参数: 无。返回: None。异常: 原关闭错误；不越过尚未关闭的依赖资源。349 """350 self._close_slot('_record_calculations')351352 def _clear_contracts(self):353 """释放本会话借用的有限声明，随后才退还覆盖它们的预算。354355 参数: 无。返回: None。异常: 无；不修改调用方计划或已经移交的结果。356 """357 self._requirements = None358 self._plan = self._metadata = self._run_id = self._started_at = None359360 def _release_costs(self):361 """逐笔释放本owner真实槽，覆盖释放前后取消。参数: 无。返回: None。异常: 原预算错误。362 """363 for slot in ('_metadata_cost', '_fixed_cost'):364 _release_owned_ram(self, getattr(self, slot), slot)365366 def _assert_owned_balance(self):367 """检查执行树已关闭且账本只含本会话及待移交结果费用。368369 参数: 无。返回: None。异常: AnalysisError表示真实资源或费用残留，不靠差额退费。370 """371 if self._record_calculations is not None:372 raise AnalysisError('成功前仍有未关闭内部资源')373 expected = self._held_cost374 if self._artifact_exports is not None:375 expected += self._artifact_exports.owned_ram376 if self._result is not None:377 expected += self._result._held_cost + self._result.spool._held_cost378 if self._result._artifact_exports is not None and self._result._artifact_exports is not self._artifact_exports:379 expected += self._result._artifact_exports.owned_ram380 elif self._spool is not None:381 expected += self._spool._held_cost382 if self._budget.ram_bytes != expected:383 raise AnalysisError('会话实际账本与已登记所有者费用不一致')384385 def close(self):386 """可重试关闭本次最高资源和协调费用，保留文件及已移交结果。387388 参数: 无。返回: None，成功后幂等。异常: 原句柄、预算或取消错误附389 runner_cleanup_owner；不会用共享余额抵销旧工厂遗失的费用。390 """391 if self._closed:392 return393 self._closing = True394 try:395 # 可选固定编码借用生产者和spool，先解除其引用和真实费用；若失败，396 # 下层保持可达，恢复后从同一sink槽继续关闭，不越级退费。397 # 未移交result拥有最终spool，始终从最高父关闭；失败时不重复关闭内层。398 self._close_slot('_result')399 self._close_slot('_spool')400 self._close_inputs()401 self._close_slot('_artifact_exports')402 self._clear_contracts()403 self._release_costs()404 if self._owns_path and not self._retain_work:405 from ..results.owned_directory import remove_owned_directory406 remove_owned_directory(self._path, self._path_identity)407 self._owns_path = False408 if self._held_cost == 0:409 self._closed = True410 except BaseException as exc:411 self._failed = True412 exc.runner_cleanup_owner = self413 raise414