chronaxis.execution.parallel_can_decode
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""按文件并行解码并恢复一个来源的连续记录流。23concatenate只预读每个工作进程的一块，按文件顺序消费；time_order先把已校验4的有序文件流写入本次受配额约束的暂存，再稳定归并。父层继续使用原同刻组及5状态机。所有工作进程、管道、暂存和费用归本owner，关闭不会删除输入或他人文件。6"""7from dataclasses import replace8import heapq9from hashlib import sha25610import multiprocessing11import os12from pathlib import Path13import struct14from time import perf_counter15from uuid import uuid41617from .._control_wire import ControlChannel18from .can_block_codec import decode_block, decode_metadata19from ..results.budget import _reserve_owned_ram, _release_owned_ram, _cleanup_owner20from ..sources.models import freeze_metadata21from ..sources.protocol import SourceReadError22from .history import _metadata_bytes23from .parallel_can_worker import BLOCK_RAM, WIRE_LIMIT, worker_main242526def _with_position(record, input_index, sequence):27 """给本入口刚从固定块核验完成的记录重编号，不再解析其不可变字段键。2829 参数: record只能来自本次decode_block；input_index和sequence是待写入的新非负编号。30 返回: 新冻结记录；原始帧、字段值、状态和重建事实继续共享已核不可变内容。31 异常: 新编号、类型、状态或字段对结构非法时沿用内部工厂检查。32 仅供本模块已核记录使用；外部记录构造和新块解码仍完整解析规范字段。33 临时键集合至多4096项，按512+192*4096字节小于1MiB；它在已有8MiB34 接收/归并空间中、6MiB模型上界以外的2MiB余量内，用完即释放，不跨记录缓存。35 """36 from ..sources.text_can_decode import DecodedCANRecord37 if type(record) is not DecodedCANRecord:38 raise TypeError('CAN重编号需要本次已核解码记录')39 keys = None40 try:41 keys = {pair[0] for pair in record.values}42 return DecodedCANRecord._from_bound_values(record.source, input_index, sequence,43 record.raw, record.values, record.status, record.frame_values, record.frame_layout, keys)44 finally:45 if keys is not None:46 keys.clear()47 keys = None484950class ParallelCANDecodeRead:51 """拥有有限文件工作槽及唯一有序输出；文件EOF不结束来源。5253 各槽最多一个在途块。父预付子账本上限及接收展开空间，记录的长期消费者54 自行计费；归并时工作进程已退出，再为每个打开的暂存块预付空间。55 """5657 def __init__(self,spec,budget,signals,*,process_count,work_dir,emit_frame_ids=None,**options):58 """保存冻结文件任务，不读输入。参数: 来源/共享预算/绑定及执行选项。返回: owner。异常: 类型/预算。59 """60 self.spec,self._budget,self.signals=spec,budget,signals61 self._process_count=min(process_count,len(spec.inputs))62 self._work_dir=Path(work_dir)63 self._options=dict(options,emit_frame_ids=emit_frame_ids)64 self._integrity=options.get('integrity',True)65 context=self._options.pop('database_context',None)66 self._frame_context=None if context is None else (context.rules,context._cycle_periods,context._integrity if hasattr(context,'_integrity') else options.get('integrity',True))67 self._held_cost=self._fixed_cost=self._workers_cost=self._merge_cost=068 self._metadata_extra_cost=069 self._metadata_credit=self._metadata_model_cost=self._ready_model_cost=070 self._slots=[];self._file_slots={};self._facts={}71 self._next_file=self._index=self._offset=self._sequence=072 self._records=();self._position=0;self._pending_eof=False73 self._last_time=self._last_order=None74 self._metadata=None;self._root=None;self._paths=[];self._handles=[];self._heap=[]75 self._disk_blocks=[];self._disk_eof=[]76 self._closed=self._exhausted=self._started=self._spooled=False77 self.count=self.decode_calls=self.retained_count=078 self.first_ns=self.last_ns=None79 self.timings=dict(worker_read_seconds=0.,worker_serialize_seconds=0.,worker_wait_seconds=0.,80 worker_send_seconds=0.,main_wait_seconds=0.,received_bytes=0,worker_pids=[],files={})81 try:82 declarations = _metadata_bytes((tuple(str(item.path) for item in spec.inputs),signals,self._options))83 # 原32D预约不变。其中8D只用于持久READY及其暂存，不再算作其它阶段余量。84 # 余24D+1MiB覆盖原声明D、两份固定控制容器至多2D，以及最坏JSON转义下85 # 当前文本/ASCII片/累计编码三份至多18D；小头、64KiB片和其它槽用固定余量。86 # send返回即释放编码树，READY不与本槽控制编码或前一数据块重叠。87 self._metadata_credit=8*declarations88 _reserve_owned_ram(self,1048576+32*declarations, '_fixed_cost')89 except BaseException as error:90 _cleanup_owner(self,error,'text_can_cleanup_owner','并行CAN构造')91 raise9293 @property94 def closed(self):95 """返回实际关闭事实。参数: 无。返回: bool。异常: 无。"""96 return self._closed9798 @property99 def exhausted(self):100 """返回完整来源结束事实。参数: 无。返回: bool。异常: 无。"""101 return self._exhausted102103 @property104 def metadata(self):105 """返回同次DBC元数据。参数: 无。返回: 冻结映射。异常: 未开始时拒绝。"""106 if self._metadata is None:107 raise SourceReadError('并行CAN尚未初始化元数据')108 return self._metadata109110 def _receive(self,slot):111 """接收独立控制消息并监测退出。参数: 工作槽。返回: tuple。异常: 子错误/崩溃/取消。112 """113 started=perf_counter()114 while slot.get('reply') is None:115 # 每个槽最多保留一个控制通知；后续文件的错误不必等前文件读完才暴露。116 for other in self._slots:117 if other.get('control') is not None and other.get('reply') is None and other['control'].poll():118 try:119 message=other['control'].recv()120 except EOFError as error:121 raise SourceReadError('CAN工作进程控制连接提前关闭') from error122 if message[0]=='error':123 raise SourceReadError(f"{self.spec.inputs[other['file']].path}: {message[1]}: {message[2]}")124 other['reply']=message125 if slot.get('reply') is not None:126 break127 if slot['process'].exitcode is not None:128 raise SourceReadError(f"CAN工作进程意外退出：{self.spec.inputs[slot['file']].path}，退出码{slot['process'].exitcode}")129 slot['control'].poll(0.05)130 self.timings['main_wait_seconds']+=perf_counter()-started131 try:132 reply=slot.pop('reply')133 except EOFError as error:134 raise SourceReadError('CAN工作进程控制连接提前关闭') from error135 if reply[0]=='error':136 raise SourceReadError(f"{self.spec.inputs[slot['file']].path}: {reply[1]}: {reply[2]}")137 return reply138139 def _bytes(self,slot,size):140 """接收已预付且限定长度的数据。参数: 槽/声明长度。返回: bytes。异常: 断连/协议错。141 """142 if type(size) is not int or not 0<size<=WIRE_LIMIT:143 raise SourceReadError('CAN数据块声明长度非法')144 while not slot['data'].poll(0.05):145 if slot['process'].exitcode is not None:146 raise SourceReadError('CAN工作进程在传输中退出')147 packet=slot['data'].recv_bytes(WIRE_LIMIT)148 if len(packet)!=size:149 raise SourceReadError('CAN数据块长度不符')150 self.timings['received_bytes']+=size151 return packet152153 def _assign(self,slot,index):154 """给空闲进程一个冻结文件并核元数据。参数: 槽/文件位置。返回: None。异常: 读取或声明不符。155 """156 slot.update(file=index,block=0)157 spec=replace(self.spec,inputs=(self.spec.inputs[index],),file_merge='concatenate')158 slot['control'].send(('file',spec,self.signals,self._options,self._worker_limit,self._frame_context))159 reply=self._receive(slot)160 if reply[0]!='ready' or reply[1]!=slot['process'].pid:161 raise SourceReadError('CAN工作进程初始化身份不符')162 self._accept_metadata(self._bytes(slot,reply[2]))163 self._file_slots[index]=slot164 # 每槽只许可一个块；许可之前已在父账本支付远端及接收空间。165 slot['control'].send(('next',))166167 def _accept_metadata(self,packet):168 """在原费用份额内接收READY，仅为实际超出的固定图预付差额。169170 参数: packet为已按512KiB上限接收的当前包。171 返回: None，首份冻结事实保留到close，后续只比较后释放。172 异常: 包错误、事实不同、预算或取消；退款失败附注保原错及可重试拥有者。173 """174 decoded=metadata=None175 self._ready_model_cost=0176 size=len(packet)177 def reserve_model(amount):178 """先给新图及冻结副本付费。参数: amount单图上界。返回: None。异常: 原预算错误。"""179 self._ready_model_cost=amount180 # 首份已保留的图先占8D；超过8D的持久差额另有真实_metadata_extra_cost。181 # 当前槽8MiB只在没有数据块时使用。未知DBC单位由包内文本费用覆盖。182 free_credit=max(0,self._metadata_credit-self._metadata_model_cost)183 persistent=max(0,self._metadata_model_cost-self._metadata_credit)184 required=persistent+max(0,2*amount+size-BLOCK_RAM-free_credit)185 if required>self._metadata_extra_cost:186 _reserve_owned_ram(self,required-self._metadata_extra_cost,'_metadata_extra_cost')187 try:188 decoded=decode_metadata(packet,on_expand=reserve_model)189 metadata=freeze_metadata(decoded)190 if self._metadata is None:191 extra=max(0,self._ready_model_cost-self._metadata_credit)192 if extra>self._metadata_extra_cost:193 # 同一费用槽从暂存转为持久事实，只补差额，不把同一冻结图付两次。194 # 数据槽仍持有当前图时完成预约，worker退出后仍有真实登记。195 _reserve_owned_ram(self,extra-self._metadata_extra_cost,'_metadata_extra_cost')196 self._metadata=metadata197 self._metadata_model_cost=self._ready_model_cost198 elif metadata!=self._metadata:199 raise SourceReadError('并行文件使用的DBC元数据不一致')200 except BaseException as error:201 decoded=metadata=packet=None202 try:203 persistent=max(0,self._metadata_model_cost-self._metadata_credit)204 _release_owned_ram(self,self._metadata_extra_cost-persistent,'_metadata_extra_cost')205 except BaseException as cleanup:206 error.add_note('READY暂存退款失败: '+type(cleanup).__name__)207 raise208 else:209 decoded=metadata=packet=None210 persistent=max(0,self._metadata_model_cost-self._metadata_credit)211 _release_owned_ram(self,self._metadata_extra_cost-persistent,'_metadata_extra_cost')212 finally:213 self._ready_model_cost=0214215 def prepare_metadata(self):216 """启动有限工作槽，只初始化DBC再授予首块。参数: 无。返回: 元数据。异常: 启动/预算/元数据错。217 """218 if self._closed:219 raise SourceReadError('并行CAN已关闭')220 if self._started:221 return self.metadata222 try:223 # DBC估算与现有文本读取器一致，另计绑定图、单块构造及父接收展开。224 self._worker_limit=BLOCK_RAM+4*1048576+self.spec.dbc_path.stat().st_size*64+64*_metadata_bytes((self.signals,self._options,self._frame_context))225 _reserve_owned_ram(self,self._process_count*(self._worker_limit+BLOCK_RAM),'_workers_cost')226 context=multiprocessing.get_context('spawn')227 for _ in range(self._process_count):228 slot=dict(process=None,control=None,data=None,child_control=None,child_data=None,229 tree=None,startup=None,file=self._next_file,started=False)230 self._slots.append(slot)231 slot['control'],slot['child_control']=context.Pipe()232 slot['control']=ControlChannel(slot['control'],'can',parent=True,max_bytes=self._worker_limit)233 slot['child_control']=ControlChannel(slot['child_control'],'can',parent=False,max_bytes=self._worker_limit)234 slot['data'],slot['child_data']=context.Pipe(duplex=False)235 from .._process_start import spawn_entry236 try:237 target,args,slot['startup']=spawn_entry(context,worker_main,(slot['child_control'],slot['child_data']))238 except BaseException as error:239 slot['startup']=getattr(error,'process_start_cleanup_owner',None)240 raise241 process=context.Process(target=target,args=args)242 slot['process']=process243 process.start();slot['started']=True244 from ..batch.process_tree import ProcessTree245 try:246 slot['tree']=(ProcessTree(process.pid) if slot['startup'] is None else247 ProcessTree(process.pid,registration=slot['startup'],process=process))248 except BaseException as error:249 slot['tree']=getattr(error,'process_tree_cleanup_owner',None)250 raise251 if slot['startup'] is not None:252 slot['startup'].close();slot['startup']=None253 for key in ('child_control','child_data'):254 connection=slot[key]255 try:256 connection.close()257 finally:258 if connection.closed:259 slot[key]=None260 self.timings['worker_pids'].append(process.pid)261 self._assign(slot,self._next_file)262 self._next_file+=1263 self._started=True264 return self.metadata265 except BaseException as error:266 _cleanup_owner(self,error,'text_can_cleanup_owner','并行CAN启动')267 raise268269 def _block(self,index):270 """收一块并记录已见事实，原始计数不以传输记录数替代。参数: 文件位置。返回: 块。异常: 协议/读取错误。271 """272 slot=self._file_slots[index]273 reply=self._receive(slot)274 if reply[0]!='block':275 raise SourceReadError('CAN数据块控制类型不符')276 packet=self._bytes(slot,reply[1])277 sent=self._receive(slot)278 if sent[0]!='sent':279 raise SourceReadError('CAN数据块缺少发送完成确认')280 number,records,eof,facts=decode_block(packet)281 if number!=slot['block'] or type(records) is not list or len(records)>64:282 raise SourceReadError('CAN文件数据块序号或长度不符')283 self._record_identity(index,records)284 slot['block']+=1285 self.timings['worker_read_seconds']+=reply[2]286 self.timings['worker_serialize_seconds']+=reply[3]287 self.timings['worker_wait_seconds']+=reply[4]288 self.timings['worker_send_seconds']+=sent[1]289 profile=self.timings['files'].setdefault(str(index),dict(pid=slot['process'].pid,read_seconds=0.,290 serialize_seconds=0.,credit_wait_seconds=0.,send_seconds=0.,bytes=0,max_block_bytes=0))291 for name,value in zip(('read_seconds','serialize_seconds','credit_wait_seconds','send_seconds'),(*reply[2:5],sent[1])):292 profile[name]+=value293 profile['bytes']+=len(packet);profile['max_block_bytes']=max(profile['max_block_bytes'],len(packet))294 self._facts[index]=facts295 self.count=sum(f['count'] for f in self._facts.values())296 self.decode_calls=sum(f['decode_calls'] for f in self._facts.values())297 starts=[f['first_ns'] for f in self._facts.values() if f['first_ns'] is not None]298 ends=[f['last_ns'] for f in self._facts.values() if f['last_ns'] is not None]299 self.first_ns=min(starts) if starts else None300 self.last_ns=max(ends) if ends else None301 return records,eof,facts,packet302303 def _record_identity(self,index,records):304 """核当前包的来源和文件身份，不将包内路径用作新文件打开目标。305306 参数: index为当前输入位置，records为本包已恢复记录。307 返回: None。308 异常: 来源、原文件或序号不同步时SourceReadError。309 """310 expected=self.spec.inputs[index].path311 if any(record.source!=self.spec.name or record.input_index!=0312 or record.raw.path!=expected or record.sequence!=record.raw.sequence for record in records):313 raise SourceReadError('CAN数据块来源或原文件身份不符')314315 def _next_task(self,index):316 """完成文件后复用同一工作槽；没有下一文件时保留槽待统一关闭。参数: 文件位置。返回: None。异常: 初始化错。317 """318 slot=self._file_slots.pop(index)319 if self._next_file<len(self.spec.inputs):320 self._assign(slot,self._next_file)321 self._next_file+=1322323 def _stop_workers(self,*,completed=False):324 """关闭全部本次进程和管道，不等待满管道自行排空。参数: 无。返回: None。异常: 真实关闭失败保留槽。325 """326 failure=None327 while self._slots:328 slot=self._slots[-1];process=slot['process']329 # 按实际PID识别start已经生效后取消，不能只依赖尚未来得及写入的started标志。330 if process is not None and not getattr(process,'_closed',False) and process.pid is not None:331 if process.is_alive():332 try:333 slot['control'].send(('close',))334 except (OSError,EOFError):335 pass336 process.join(2 if completed else 0.15)337 if process.is_alive():338 process.terminate();process.join(2)339 if completed:340 failure=SourceReadError('CAN工作进程未正常结束')341 if process.is_alive() and slot['tree'] is not None:342 slot['tree'].close();slot['tree']=None343 if process.is_alive():344 raise SourceReadError('CAN工作进程尚未退出')345 if completed and process.exitcode!=0:346 failure=SourceReadError(f'CAN工作进程退出码不是0：{process.exitcode}')347 if slot['tree'] is not None:348 slot['tree'].close();slot['tree']=None349 if process is not None:350 if not getattr(process,'_closed',False):351 process.close()352 slot['process']=None353 for key in ('control','data','child_control','child_data','startup'):354 connection=slot.get(key)355 if connection is not None:356 try:357 connection.close()358 finally:359 if connection.closed:360 slot[key]=None361 self._slots.pop()362 self._file_slots.clear()363 _release_owned_ram(self,self._workers_cost,'_workers_cost')364 if failure is not None:365 raise failure366367 def _prepare_sorted(self):368 """先流式暂存已校验文件，再释放进程额度开始归并。参数: 无。返回: None。异常: 配额/IO/原输入错误。369 """370 root=self._work_dir/('file-merge-'+uuid4().hex)371 root.mkdir(parents=True,exist_ok=False)372 self._root=root373 writers={}374 try:375 # 轮询已就绪槽，不让慢文件阻止其他文件继续解码和受配额约束地落盘。376 while self._file_slots:377 ready=[index for index,slot in self._file_slots.items() if slot.get('reply') is not None or slot['control'].poll() or slot['process'].exitcode is not None]378 if not ready:379 index=next(iter(self._file_slots))380 self._file_slots[index]['control'].poll(0.05)381 continue382 for index in ready:383 records,eof,facts,packet=self._block(index)384 if index not in writers:385 path=self._root/(str(index)+'.blocks')386 handle=path.open('xb');self._paths.append(path);writers[index]=handle387 # CND1本身的8字节头提供落盘分帧，不再重复保存一份包长。388 self._budget.reserve_spool(len(packet)+(32 if self._integrity else 0))389 writers[index].write(packet[:8])390 if self._integrity:391 writers[index].write(sha256(packet).digest())392 writers[index].write(memoryview(packet)[8:])393 # 释放接收对象后再许可下一块，禁止两块同时占用同一额度。394 records=packet=None395 if eof:396 writers.pop(index).close();self._next_task(index)397 else:398 self._file_slots[index]['control'].send(('next',))399 finally:400 for handle in writers.values():401 handle.close()402 self._stop_workers(completed=True)403 # 每个归并输入最多保留一个有界块；过多文件不足预算时明确失败。404 _reserve_owned_ram(self,len(self.spec.inputs)*BLOCK_RAM,'_merge_cost')405 for index in range(len(self.spec.inputs)):406 handle=(self._root/(str(index)+'.blocks')).open('rb');self._handles.append(handle)407 self._stream_records=[((),0) for _ in self._handles]408 self._disk_blocks=[0 for _ in self._handles]409 self._disk_eof=[False for _ in self._handles]410 for index in range(len(self._handles)):411 self._push(index)412 self._spooled=True413414 def _push(self,index):415 """为一个有序文件取下一记录，堆键固定时间/文件位置/局部序号。参数: index。返回: None。异常: 暂存截断。416 """417 records,position=self._stream_records[index]418 while position==len(records):419 header=self._handles[index].read(8)420 if not header:421 if not self._disk_eof[index]:422 raise SourceReadError('CAN归并暂存缺少文件EOF')423 self._stream_records[index]=((),0)424 return425 if self._disk_eof[index]:426 raise SourceReadError('CAN归并暂存在EOF后仍有数据')427 if len(header)!=8:428 raise SourceReadError('CAN归并暂存长度头截断')429 magic,size=struct.unpack('!4sI',header)430 if magic!=b'CND1' or not 0<size<=WIRE_LIMIT-8:431 raise SourceReadError('CAN归并暂存数据块版本或长度非法')432 expected=self._handles[index].read(32) if self._integrity else None433 body=self._handles[index].read(size)434 if len(body)!=size:435 raise SourceReadError('CAN归并暂存数据块截断')436 packet=header+body437 body=None438 if expected is not None and sha256(packet).digest()!=expected:439 raise SourceReadError('CAN归并暂存完整性校验失败')440 number,records,eof,_=decode_block(packet)441 if number!=self._disk_blocks[index]:442 raise SourceReadError('CAN归并暂存块序号不符')443 self._record_identity(index,records)444 self._disk_blocks[index]+=1445 self._disk_eof[index]=eof446 position=0447 record=records[position]448 self._stream_records[index]=(records,position+1)449 heapq.heappush(self._heap,(record.raw.time_ns,index,record.raw.sequence,record))450451 def __iter__(self):452 """返回本次游标。参数: 无。返回: self。异常: 无。"""453 return self454455 def __next__(self):456 """返回按明确归并政策排序的下一记录。参数: 无。返回: DecodedCANRecord。异常: EOF或读取/预算错。457 """458 if self._closed or self._exhausted:459 raise StopIteration460 try:461 self.prepare_metadata()462 if self.spec.file_merge=='time_order':463 if not self._spooled:464 self._prepare_sorted()465 if not self._heap:466 self._exhausted=True467 raise StopIteration468 _,index,_,record=heapq.heappop(self._heap)469 result=_with_position(record,index,self._sequence)470 self._sequence+=1471 self._push(index)472 else:473 while self._position==len(self._records):474 self._records=()475 if self._pending_eof:476 fact=self._facts[self._index]477 if fact['first_order_time'] is not None and self._last_order is not None and fact['first_order_time']<self._last_order:478 raise SourceReadError('CAN文件边界时间回退；请核对时间或显式选择file_merge: time_order')479 self._last_order=fact['last_order_time'] if fact['last_order_time'] is not None else self._last_order480 self._offset+=fact['count'];self._next_task(self._index);self._index+=1481 self._pending_eof=False482 if self._index==len(self.spec.inputs):483 self._stop_workers(completed=True);self._exhausted=True484 raise StopIteration485 records,eof,fact,packet=self._block(self._index)486 if fact['first_order_time'] is not None and self._last_order is not None and fact['first_order_time']<self._last_order:487 raise SourceReadError('CAN文件边界时间回退；请核对时间或显式选择file_merge: time_order')488 self._records=records;self._position=0;self._pending_eof=eof489 # 当前块已由父接收额度承担，子侧原额度可构造下一块；二者费用分别预付。490 packet=None491 if not eof:492 self._file_slots[self._index]['control'].send(('next',))493 record=self._records[self._position];self._position+=1494 result=_with_position(record,self._index,self._offset+record.raw.sequence)495 if result.raw.time_ns is not None and self._last_time is not None and result.raw.time_ns<self._last_time:496 raise SourceReadError('CAN归并输出时间回退')497 self._last_time=result.raw.time_ns;self.retained_count+=1498 return result499 except StopIteration:500 raise501 except BaseException as error:502 _cleanup_owner(self,error,'text_can_cleanup_owner','并行CAN读取')503 raise504505 def close(self):506 """释放本次进程、块和登记暂存。参数: 无。返回: None。异常: 清理失败可重试。507 """508 if self._closed:509 return510 self._records=();self._heap.clear()511 if hasattr(self,'_stream_records'):512 self._stream_records.clear()513 self._stop_workers()514 while self._handles:515 self._handles[-1].close();self._handles.pop()516 while self._paths:517 self._paths[-1].unlink(missing_ok=True);self._paths.pop()518 if self._root is not None:519 self._root.rmdir();self._root=None520 self._metadata=None;self._metadata_model_cost=0521 _release_owned_ram(self,self._metadata_extra_cost,'_metadata_extra_cost')522 _release_owned_ram(self,self._merge_cost,'_merge_cost')523 _release_owned_ram(self,self._fixed_cost,'_fixed_cost')524 self._closed=True525