chronaxis.fast.session
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""一个fast会话的多文件调度和最终累计。23worker只求当前帧贡献；主进程控制统一预算、交付顺序及成功发布前的完整性。4concatenate可暂停可选预读；time_order必要暂存完成后稳定归并。临时目录和5进程均属于本次调用，异常不返回部分结果。内联与batch不经传输协议。6"""7from hashlib import sha2568import heapq9import json10import multiprocessing11from multiprocessing.connection import wait12import os13from pathlib import Path14from tempfile import mkdtemp15from time import perf_counter1617from ..results.budget import ResourceBudget, _reserve_owned_ram, _release_owned_ram, _cleanup_owner18from ..results.errors import BudgetExceededError19from ..dbc_loading import load_dbc_file20from .._control_wire import ControlChannel21from .blocks import RAM, LIMIT, encode, decode, batches22from .compute import _Accumulator, _check_program_spec23from .stream import FileContributions242526def effective_spec(spec, process_count=None):27 """取得不修改原声明的执行视图。参数: spec字典，可选正整数覆盖。返回: 新dict。28 异常: 非正原生整数；不接受bool、字符串或浮点数。29 """30 count = spec.get('process_count', 1) if process_count is None else process_count31 if type(count) is not int or count < 1:32 raise ValueError('process_count需要大于等于1的整数，不接受bool')33 return dict(spec, process_count=count)343536def _worker_process_limit(process_count: int, file_count: int) -> int:37 """从已验证的进程上限和文件数取得文件工作进程数，不包含主进程。3839 参数为正整数。返回至多文件数的工作进程上限；只能并发处理一份时返回0，40 表示在当前进程执行。时间归并的内联暂存不改变这个判定；无输入或输出操作。41 """42 count = min(process_count, file_count)43 return count if count > 1 else 0444546def file_specs(spec):47 """展开有限文件声明。参数: 会话spec。返回: 每文件字典列表。异常: 无输入IO。"""48 inputs = spec.get('inputs', [dict(path=spec['path'], offset_ns=spec.get('offset_ns', 0),49 asc_options=spec.get('asc_options', {}))])50 common = {key: value for key, value in spec.items() if key not in ('inputs', '_database')}51 return [dict(common, **item) for item in inputs]525354def _checked_file_specs(spec, programs_checked):55 """取得实际文件声明，并复用本会话已经检查过的同一组公式。5657 参数: 本次独立的spec字典、公共编译器是否已检查顶层公式。58 返回: 文件声明列表；低层inputs覆盖的公式仍检查原文本和指令。59 异常: 非法公式或声明错误；不因顶层已编译就放行另一组覆盖值。60 """61 original = spec['metrics']62 prepared = original if programs_checked else [_check_program_spec(item) for item in original]63 # 缓存只活到准备结束，保留源对象以避免身份复用，不跨会话或按文本猜等价。64 checked = {id(original): (original, prepared), id(prepared): (prepared, prepared)}65 spec['metrics'] = prepared66 specs = file_specs(spec)67 for item in specs:68 actual = item['metrics']69 cached = checked.get(id(actual))70 if cached is None or cached[0] is not actual:71 cached = (actual, [_check_program_spec(metric) for metric in actual])72 checked[id(actual)] = cached73 item['metrics'] = cached[1]74 return specs757677class SessionAccumulator:78 """只有主会话持有的标量累计器。计数、目标缺失和帧身份按计算项分别核验。"""7980 def __init__(self, spec):81 """准备纯累计状态。参数: 已核验声明。返回: 新对象。异常: 名称或单位字段缺失。"""82 self.metrics = [_Accumulator(item) for item in spec['metrics']]83 self.raw = 084 self.targets = [0] * len(self.metrics)85 self.valid = [0] * len(self.metrics)86 self.selected = [0] * len(self.metrics)87 self.kinds = [0] * len(self.metrics)88 self.compatible = set()89 self.profile = spec.get('_profile', False)90 self.accumulate_seconds = 0.09192 def consume(self, record):93 """按输入顺序逐项相加。参数: 一帧贡献。返回: None。异常: 累计溢出。"""94 started = perf_counter() if self.profile else 095 for index, value, keep in record[2]:96 self.metrics[index].add(value, keep)97 if self.profile:98 self.accumulate_seconds += perf_counter() - started99100 def finish_file(self, facts):101 """合并已关闭文件的事实。参数: 文件summary。返回: None。异常: 计数或身份非法。"""102 if type(facts['raw']) is not int or facts['raw'] < 0:103 raise ValueError('文件原始记录计数非法')104 self.raw += facts['raw']105 for field in ('targets', 'valid', 'selected', 'kinds'):106 values = facts[field]107 if len(values) != len(self.metrics) or any(type(v) is not int or v < 0 for v in values):108 raise ValueError('文件计算项计数非法')109 for i in range(len(self.metrics)):110 if not facts['selected'][i] <= facts['valid'][i] <= facts['targets'][i] <= facts['raw']:111 raise ValueError('文件目标/有效/入选计数矛盾')112 self.targets[i] += facts['targets'][i]113 self.valid[i] += facts['valid'][i]114 self.selected[i] += facts['selected'][i]115 self.kinds[i] |= facts['kinds'][i]116 if self.kinds[i].bit_count() > 1:117 raise ValueError(self.metrics[i].name + '跨文件匹配多种实际帧型，请明确选择')118 self.compatible.update(facts['compatible'])119120 def result(self):121 """核完整消费后生成业务行。参数: 无。返回: (行list, 提示list)。异常: 贡献计数不符。"""122 rows = []123 for i, metric in enumerate(self.metrics):124 if (metric.matched, metric.selected) != (self.valid[i], self.selected[i]):125 raise ValueError(metric.name + '贡献未完整且唯一消费')126 row = metric.result()127 if row['value'] is None:128 row['reason'] = ('empty_file' if not self.raw else129 'no_target_in_domain' if self.targets[i] else 'no_target_frames')130 rows.append(row)131 warnings = [f'{identifier:#x} 按经典 CAN 实录解码 DBC 的 FD 声明' for identifier in sorted(self.compatible)]132 return rows, warnings133134135class FilePool:136 """管理有限工作槽及每文件连续块前缀；不在内存保存所有暂存块描述。"""137138 def __init__(self, specs, budget, directory, cancel, statistics, *, _programs_checked=False):139 """初始化但不启动。参数: 文件spec列表、共享账本、暂存目录、外部取消、统计dict。140 返回: 新调度器。异常: 无；start安装进程资源的关闭责任。141 _programs_checked只来自父会话的实际准备结果，不读取配置中的标志。142 """143 self.specs, self.budget, self.directory = specs, budget, Path(directory)144 self._budget, self._held_cost = budget, 0145 self._started = self._closing = self._closed = False146 self._programs_checked = _programs_checked147 self.cancel, self.stats = cancel, statistics148 self.context = multiprocessing.get_context('spawn')149 self.stop = self.context.Event()150 self.slots = []151 self.files = [dict(produced=0, consumed=0, spool=0, eof=None, packet=None,152 last_sequence=-1, last_time=None) for _ in specs]153 self.current = 0154 self.next_file = 0155 self.ordered = specs[0].get('file_merge', 'concatenate') == 'time_order'156 self.worker_cost = 0157 self.digest = sha256(Path(specs[0]['dbc']).read_bytes()).hexdigest()158 if self.digest != specs[0].get('_dbc_digest', self.digest):159 raise ValueError('编译后DBC内容变化')160161 def check_cancel(self):162 """传递外部取消。参数: 无。返回: None。异常: InterruptedError。"""163 if self.cancel is not None and self.cancel.is_set():164 if self.stop is not None:165 self.stop.set()166 raise InterruptedError('fast会话取消')167168169 def start(self, count):170 """预付真实工作槽后启动；任何部分构造失败仍由本池关闭已登记资源。171172 参数: count实际工作数。返回: None。异常: 原预算/启动/归属错误；173 持续清理失败在原异常的fast_pool_cleanup_owner保留内部恢复入口。174 """175 from .session_worker import worker_main176 if self._started or self._closing or self._closed:177 raise RuntimeError('fast工作池已经启动或关闭，不能重复启动')178 self._started = True179 try:180 per_worker = 3 * RAM + Path(self.specs[0]['dbc']).stat().st_size * 64181 _reserve_owned_ram(self, per_worker * count, 'worker_cost')182 for _ in range(count):183 self.check_cancel()184 # 先安装槽，再创建管道；第二根管道或启动失败时，第一根仍有关闭责任。185 slot = dict(control=None, data=None, child_control=None, child_data=None,186 process=None, tree=None, startup=None, phase='idle', file=None, stop_requested=False)187 self.slots.append(slot)188 slot['control'], slot['child_control'] = self.context.Pipe()189 slot['control'] = ControlChannel(slot['control'],'fast',parent=True,max_bytes=self.budget.memory_budget_bytes)190 slot['child_control'] = ControlChannel(slot['child_control'],'fast',parent=False,max_bytes=self.budget.memory_budget_bytes)191 slot['data'], slot['child_data'] = self.context.Pipe(duplex=False)192 from .._process_start import spawn_entry193 try:194 target,args,slot['startup']=spawn_entry(self.context,worker_main,195 (slot['child_control'],slot['child_data'],self.stop))196 except BaseException as error:197 slot['startup']=getattr(error,'process_start_cleanup_owner',None)198 raise199 process = self.context.Process(target=target,args=args)200 slot['process'] = process201 process.start()202 from ..batch.process_tree import ProcessTree203 try:204 slot['tree'] = (ProcessTree(process.pid) if slot['startup'] is None else205 ProcessTree(process.pid,registration=slot['startup'],process=process))206 except BaseException as error:207 slot['tree'] = getattr(error,'process_tree_cleanup_owner',None)208 raise209 if slot['startup'] is not None:210 slot['startup'].close();slot['startup']=None211 self.stats['worker_pids'].append(process.pid)212 # 子进程已取得自己的端点，父进程关闭副本；失败端点留在同一槽供重试。213 for key in ('child_control', 'child_data'):214 connection = slot[key]215 try:216 connection.close()217 finally:218 if connection.closed:219 slot[key] = None220 self.assign(slot)221 except BaseException as error:222 _cleanup_owner(self, error, 'fast_pool_cleanup_owner', 'fast工作池启动')223 raise224225 def assign(self, slot):226 """按文件声明顺序派发下一文件。参数: 空闲槽。返回: None。异常: 断管。"""227 slot['phase'] = 'idle'228 if self.next_file < len(self.specs):229 index = self.next_file230 self.next_file += 1231 slot.update(file=index, phase='computing')232 operation = 'program_file' if self._programs_checked else 'file'233 slot['control'].send((operation, index, self.specs[index], str(self.directory), self.digest))234235 def grant(self, slot):236 """授权一个完整块的唯一交付路径。参数: 已报价工作槽。返回: None。异常: 必要暂存超额。"""237 if slot['phase'] != 'offered':238 return239 if not self.ordered and slot['file'] == self.current:240 slot.update(phase='direct', delivered=False, received=None)241 slot['control'].send(('direct',))242 return243 size = slot['size']244 quota = self.budget.spool_quota_bytes245 if quota is not None and self.budget.spool_bytes + size > quota:246 if self.ordered:247 raise BudgetExceededError('time_order必要暂存超过spool_quota_bytes')248 if not slot.get('paused', False):249 self.stats['prefetch_pauses'] += 1250 slot['paused'] = True251 return252 self.budget.reserve_spool(size)253 slot['phase'] = 'spooling'254 slot['control'].send(('spool',))255256 def pump(self, blocking=True):257 """推进有限控制和数据消息，优先响应所有worker错误。参数: 是否最多等50ms。258 返回: None。异常: 协议、worker退出、取消或IO错误。259 """260 self.check_cancel()261 for slot in self.slots:262 self.grant(slot)263 connections = [s['control'] for s in self.slots]264 connections += [s['data'] for s in self.slots if s['phase'] == 'direct' and s['received'] is None]265 started = perf_counter()266 ready = wait(connections, timeout=0.05 if blocking else 0)267 self.stats['main_wait_seconds'] += perf_counter() - started268 for slot in self.slots:269 process = slot['process']270 message = None271 pipe_error = None272 if slot['control'] in ready:273 try:274 message = slot['control'].recv()275 except (EOFError, BrokenPipeError) as error:276 pipe_error = error277 # 子进程可在发送error后正常退出，完整业务错误必须先于退出状态报告。278 if message is not None and message[0] == 'error':279 self.message(slot, message)280 if process.exitcode is not None:281 raise RuntimeError(f'fast工作进程意外退出：{process.pid}，{process.exitcode}') from pipe_error282 if pipe_error is not None:283 raise pipe_error284 if message is not None:285 self.message(slot, message)286 if slot['phase'] == 'direct' and slot['received'] is None and slot['data'] in ready:287 packet = slot['data'].recv_bytes(LIMIT)288 if len(packet) != slot['size']:289 raise ValueError('直传贡献长度不符')290 slot['received'] = packet291 self.stats['received_bytes'] += len(packet)292 if slot['phase'] == 'direct' and slot['received'] is not None and slot['delivered']:293 state = self.files[slot['file']]294 state['packet'] = slot['received']295 state['produced'] += 1296 slot.update(phase='consuming', received=None)297 self.grant(slot)298299 def message(self, slot, message):300 """核验工作槽状态及块序号。参数: 槽/控制tuple。返回: None。异常: 错误或协议不符。"""301 tag = message[0]302 if tag == 'error':303 raise RuntimeError(f"{self.specs[slot['file']]['path']}：{message[1]}：{message[2]}")304 if len(message) < 3 or message[1] != slot['file']:305 raise ValueError('工作进程文件身份不符')306 state = self.files[slot['file']]307 number = message[2]308 if type(number) is not int or number != state['produced']:309 raise ValueError('贡献块缺失、重复或顺序错误')310 if tag == 'offer' and slot['phase'] == 'computing':311 size = message[3]312 if type(size) is not int or not 0 < size <= LIMIT:313 raise ValueError('贡献块报价长度非法')314 slot.update(phase='offered', size=size, paused=False)315 elif tag == 'stored' and slot['phase'] == 'spooling':316 if message[3] != slot['size'] or state['spool'] != number:317 raise ValueError('暂存提交长度或前缀不符')318 state['spool'] += 1319 state['produced'] += 1320 self.stats['spool_written_bytes'] += message[3]321 self.stats['spool_live_bytes'] += message[3]322 self.stats['spool_peak_bytes'] = max(self.stats['spool_peak_bytes'], self.stats['spool_live_bytes'])323 slot['phase'] = 'computing'324 slot['control'].send(('next',))325 elif tag == 'delivered' and slot['phase'] == 'direct' and not slot['delivered']:326 if message[3] != slot['size']:327 raise ValueError('直传提交长度不符')328 slot['delivered'] = True329 elif tag == 'eof' and slot['phase'] == 'computing':330 if state['eof'] is not None:331 raise ValueError('重复文件EOF')332 state['eof'] = message[3]333 self.stats['files'].append(dict(file_index=slot['file'], pid=message[5], **message[4]))334 self.assign(slot)335 else:336 raise ValueError('贡献交付状态与消息不符')337338 def packet_records(self, index, number, packet):339 """验证块间顺序并展开一块。参数: 文件/块号/bytes。返回: 贡献生成器。异常: 顺序或协议错误。"""340 state = self.files[index]341 records = decode(packet, index, number, len(self.specs[index]['metrics']))342 if self.specs[index].get('_profile', False):343 started = perf_counter()344 records = list(records)345 self.stats['main_decode_seconds'] += perf_counter() - started346 for record in records:347 timestamp, sequence, _ = record348 if sequence <= state['last_sequence']:349 raise ValueError('贡献记录跨块重复或回退')350 if self.ordered and state['last_time'] is not None and timestamp < state['last_time']:351 raise ValueError('贡献时间跨块回退')352 state['last_sequence'], state['last_time'] = sequence, timestamp353 yield record354355 def read_spool(self, index, number):356 """读取并删除一个已提交块。参数: 文件/块号。返回: bytes。异常: 缺失、超长、IO。357 删除仅减少当前占用，不返还累计预约额度。358 """359 path = self.directory / f'{index}-{number}.block'360 started = perf_counter()361 with path.open('rb') as stream:362 packet = stream.read(LIMIT + 1)363 if len(packet) > LIMIT:364 raise ValueError('暂存块过大')365 path.unlink()366 self.stats['spool_live_bytes'] -= len(packet)367 self.stats['main_spool_read_seconds'] += perf_counter() - started368 return packet369370 def concatenate(self, accumulator):371 """消费当前文件，后续文件同时预读。参数: 会话累计器。返回: None。异常: 运行或累计错误。"""372 for index, state in enumerate(self.files):373 self.current = index374 while True:375 self.pump(False)376 number = state['consumed']377 direct = False378 if number < state['spool']:379 packet = self.read_spool(index, number)380 elif state['packet'] is not None:381 packet, state['packet'] = state['packet'], None382 direct = True383 elif state['eof'] is not None:384 if number != state['produced']:385 raise ValueError('文件EOF前贡献未完整消费')386 accumulator.finish_file(state['eof'])387 break388 else:389 self.pump()390 continue391 for record in self.packet_records(index, number, packet):392 accumulator.consume(record)393 packet = None394 state['consumed'] += 1395 if direct:396 slot = next(s for s in self.slots if s['file'] == index)397 slot['phase'] = 'computing'398 slot['control'].send(('next',))399400 def file_records(self, index):401 """逐块读取一个已完整暂存文件。参数: 文件号。返回: 贡献生成器。异常: 缺块/完整性错。"""402 state = self.files[index]403 while state['consumed'] < state['produced']:404 self.check_cancel()405 number = state['consumed']406 packet = self.read_spool(index, number)407 yield from self.packet_records(index, number, packet)408 state['consumed'] += 1409410 def time_order(self, accumulator):411 """完成必要暂存后k路稳定归并。参数: 累计器。返回: None。异常: 输入/资源/计算错误。"""412 while any(state['eof'] is None for state in self.files):413 self.pump()414 self.close(True)415 started = perf_counter()416 streams = [iter(self.file_records(i)) for i in range(len(self.files))]417 heap = []418 for index, stream in enumerate(streams):419 record = next(stream, None)420 if record is not None:421 heapq.heappush(heap, (record[0], index, record[1], record))422 while heap:423 _, index, _, record = heapq.heappop(heap)424 accumulator.consume(record)425 following = next(streams[index], None)426 if following is not None:427 heapq.heappush(heap, (following[0], index, following[1], following))428 for state in self.files:429 if state['consumed'] != state['produced']:430 raise ValueError('归并结束后仍缺少贡献块')431 accumulator.finish_file(state['eof'])432 self.stats['main_merge_and_consume_seconds'] += perf_counter() - started433434435 def close(self, success=False):436 """先结束实际写入者，再关闭句柄和自有费用；失败槽保留到后续重试。437438 参数: success要求正常退出。返回: None。异常: 原取消或汇总关闭错误；439 不删除任何暂存目录，未关闭资源仍由本池持有。440 """441 if self._closed:442 return443 self._closing = True444 errors = []445 if not success:446 try:447 self.stop.set()448 except BaseException as error:449 errors.append(error)450 for slot in tuple(self.slots):451 process = slot['process']452 try:453 if process is not None and not getattr(process, '_closed', False) and process.pid is not None:454 if success and not slot['stop_requested']:455 slot['control'].send(('close',))456 slot['stop_requested'] = True457 process.join(5)458 if process.is_alive():459 process.terminate()460 process.join(5)461 if success:462 raise RuntimeError('fast工作进程未正常关闭')463 if process.is_alive():464 raise RuntimeError('fast工作进程仍在运行')465 if success and process.exitcode != 0:466 raise RuntimeError('fast工作进程关闭失败')467 except BaseException as error:468 errors.append(error)469 tree = slot['tree']470 if tree is not None:471 try:472 tree.close()473 except BaseException as error:474 errors.append(error)475 finally:476 # 构造失败可能只剩OpenProcess句柄，不能只检查作业_handle。477 if tree._closed:478 slot['tree'] = None479 if process is not None:480 try:481 if not getattr(process, '_closed', False):482 if process.pid is not None:483 process.join(0)484 if process.is_alive():485 raise RuntimeError('fast写入进程未结束，保留资源责任')486 process.close()487 except BaseException as error:488 errors.append(error)489 finally:490 if getattr(process, '_closed', False):491 slot['process'] = None492 for key in ('control', 'data', 'child_control', 'child_data', 'startup'):493 connection = slot.get(key)494 if connection is not None:495 try:496 connection.close()497 except BaseException as error:498 errors.append(error)499 finally:500 if connection.closed:501 slot[key] = None502 if all(slot.get(key) is None for key in ('process', 'tree', 'control', 'data', 'child_control', 'child_data', 'startup')):503 self.slots.remove(slot)504 if not self.slots:505 try:506 _release_owned_ram(self, self.worker_cost, 'worker_cost')507 except BaseException as error:508 errors.append(error)509 self._closed = not self.slots and self._held_cost == 0510 if self._closed:511 # 所有子进程已结束后解除本池对自有同步事件的引用；不改调用方的cancel。512 self.stop = None513 if errors:514 first = errors[0]515 for other in errors[1:]:516 first.add_note('fast工作池另有清理失败: ' + type(other).__name__)517 if isinstance(first, (KeyboardInterrupt, SystemExit)):518 raise first519 raise RuntimeError('fast进程资源关闭失败') from first520521522523class _SessionResources:524 """仅管理一个fast会话的固定费用、读取器、工作池和自有目录。"""525526 def __init__(self, budget):527 """建立空责任槽。参数: 本次账本。返回: 新对象。异常: 无资源创建。"""528 self._budget, self._held_cost, self._fixed_cost = budget, 0, 0529 self.reader = self.pool = self.directory = self.directory_identity = None530 self._closed = False531532 def open_reader(self, spec, database, cancel):533 """创建并登记本会话读取器，收编构造中未能关闭的真实子所有者。534535 参数: 文件声明、DBC、取消。返回: FileContributions。异常: 原构造错误。536 ASC/BLF构造失败均收编实际未关闭的来源owner，直到成功关闭才清槽。537 """538 if self.reader is not None:539 raise RuntimeError('fast前一读取器尚未关闭')540 try:541 self.reader = FileContributions(542 spec, database, cancel, self._budget, _programs_checked=True)543 return self.reader544 except BaseException as error:545 pending = getattr(error, 'fast_reader_cleanup_owner', None)546 if pending is None:547 pending = getattr(error, 'asc_read_cleanup_owner', None)548 if pending is None:549 pending = getattr(error, 'blf_read_cleanup_owner', None)550 self.reader = pending551 raise552553 def close_reader(self):554 """关闭实际读取器；仅在真实关闭完成后清槽。参数: 无。返回: None。异常: 原关闭错误。"""555 reader = self.reader556 if reader is None:557 return558 try:559 reader.close()560 if not getattr(reader, 'closed', False):561 raise RuntimeError('fast读取器close返回但未完成关闭')562 finally:563 if getattr(reader, 'closed', False):564 self.reader = None565566 def create_directory(self, work_dir):567 """独占创建并登记目录，避免自动终结器在写入者未结束时删除目录。568569 参数: 可选工作父目录。返回: Path。异常: 创建或身份登记错误，已有目录不删除。570 """571 from ..results.owned_directory import directory_identity572 if work_dir is not None:573 Path(work_dir).mkdir(parents=True, exist_ok=True)574 self.directory = Path(mkdtemp(prefix='fast-session-', dir=work_dir))575 self.directory_identity = directory_identity(self.directory)576 return self.directory577578 def close(self):579 """依次停止工作池、读取器、删除自有目录并退款，任一步失败保留重试责任。580581 参数: 无。返回: None。异常: 原清理错误；未确认停止写入者时不删除目录。582 """583 if self._closed:584 return585 try:586 pool = self.pool587 if pool is not None:588 try:589 pool.close()590 if not pool._closed:591 raise RuntimeError('fast工作池close返回但未完成关闭')592 finally:593 if pool._closed:594 self.pool = None595 self.close_reader()596 if self.directory is not None:597 from ..results.owned_directory import remove_owned_directory598 remove_owned_directory(self.directory, self.directory_identity)599 self.directory = self.directory_identity = None600 _release_owned_ram(self, self._fixed_cost, '_fixed_cost')601 finally:602 self._closed = (self.pool is None and self.reader is None603 and self.directory is None and self._held_cost == 0)604605606def run_session(spec, database=None, *, work_dir=None, cancel_event=None, statistics=None,607 _programs_checked=False):608 """运行一个连续fast任务。参数: 编译spec、可复用DBC、临时目录、取消事件、可选统计dict。609 返回: (业务行list, 提示list)。异常: 配置/读取/公式/资源/协议/取消；不返回部分结果。610 _programs_checked由公共编译后的内部入口传入；直接调用默认仍检查文本和指令。611 """612 import cantools613 spec = effective_spec(spec)614 # 原生低层入口仍完整检查文本；同一会话后续文件只绑定已经核过的指令。615 # 公共编译器调用链显式传入准备状态，不信任 spec 中的同名字段。616 specs = _checked_file_specs(spec, _programs_checked)617 count = _worker_process_limit(spec['process_count'], len(specs))618 budget = ResourceBudget(spec.get('memory_budget_bytes', 268435456), spec.get('spool_quota_bytes'))619 stats = statistics if statistics is not None else {}620 stats.update(worker_pids=[], files=[], received_bytes=0, spool_written_bytes=0,621 spool_live_bytes=0, spool_peak_bytes=0, prefetch_pauses=0,622 main_wait_seconds=0.0, main_spool_read_seconds=0.0,623 main_decode_seconds=0.0, main_merge_and_consume_seconds=0.0,624 detailed_profile=spec.get('_profile', False))625 # 覆盖计划/累计器、内联读取空间与归并各流当前块；不是整个进程RSS承诺。626 cost = RAM * (2 + (len(specs) if spec.get('file_merge') == 'time_order' else 1))627 cost += len(json.dumps({k:v for k,v in spec.items() if k != '_database'}).encode()) * 32628 if count == 0 and all(Path(item['path']).suffix.lower() == '.blf' for item in specs):629 # 内联读取逐文件开关，最多一个BLF owner；从原固定费移交64KiB，避免重复计费。630 # 混合来源/工作进程的既有分摊不变，来源owner仍预约自身实际费用。631 cost -= 65536632 resources = _SessionResources(budget)633 accumulator = primary = None634 try:635 # 首次申请和累计器构造均在责任保护区内，申请完成后取消也有真实费用槽。636 _reserve_owned_ram(resources, cost, '_fixed_cost')637 accumulator = SessionAccumulator(spec)638 if cancel_event is not None and cancel_event.is_set():639 raise InterruptedError('fast会话启动前取消')640 if '_dbc_digest' in spec and sha256(Path(spec['dbc']).read_bytes()).hexdigest() != spec['_dbc_digest']:641 raise ValueError('编译后DBC内容变化')642 if database is None:643 database = load_dbc_file(spec['dbc'], cantools)644 if count == 0 and (len(specs) == 1 or spec.get('file_merge', 'concatenate') == 'concatenate'):645 for item in specs:646 reader = resources.open_reader(item, database, cancel_event)647 for record in reader:648 accumulator.consume(record)649 accumulator.finish_file(reader.summary())650 if spec.get('_profile', False):651 stats['files'].append(dict(path=item['path'], **reader.timings))652 resources.close_reader()653 else:654 directory = resources.create_directory(work_dir)655 pool = FilePool(specs, budget, directory, cancel_event, stats, _programs_checked=True)656 resources.pool = pool657 if count == 0:658 # 内联时间归并仍有必要暂存；读取器和目录均归当前资源所有者。659 for index, item in enumerate(specs):660 reader = resources.open_reader(item, database, cancel_event)661 number = 0662 for records in batches(reader):663 packet = encode(index, number, records)664 budget.reserve_spool(len(packet))665 (directory / f'{index}-{number}.block').write_bytes(packet)666 stats['spool_written_bytes'] += len(packet)667 stats['spool_live_bytes'] += len(packet)668 stats['spool_peak_bytes'] = max(stats['spool_peak_bytes'], stats['spool_live_bytes'])669 number += 1670 pool.files[index].update(produced=number, spool=number, eof=reader.summary())671 resources.close_reader()672 else:673 pool.start(count)674 if pool.ordered:675 pool.time_order(accumulator)676 else:677 pool.concatenate(accumulator)678 pool.close(True)679 result = accumulator.result()680 stats['raw_record_count'] = accumulator.raw681 stats['calculation_counts'] = [dict(name=metric.name, targets=accumulator.targets[i],682 valid=accumulator.valid[i], selected=accumulator.selected[i]) for i, metric in enumerate(accumulator.metrics)]683 resources.close()684 return result685 except BaseException as error:686 primary = error687 _cleanup_owner(resources, error, 'fast_session_cleanup_owner', 'fast会话')688 raise689 finally:690 try:691 stats.update(budget.snapshot())692 if accumulator is not None and accumulator.profile:693 stats['main_accumulate_seconds'] = accumulator.accumulate_seconds694 except BaseException as diagnostic:695 if primary is None:696 raise697 primary.add_note('fast资源统计附加失败: ' + type(diagnostic).__name__)698