chronaxis.execution.parallel_can_worker
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""单个可复用CAN文件工作进程：读取、解码并按许可交付有限数据块。23每个工作进程只解析一个当前文件，不启动子进程，不运行事件或积分状态机。4父进程预付全部工作空间后授予一次next许可；控制错误独立于数据块发送。5文件EOF保留观测事实，等待下一文件任务；来源EOF由父进程统一判定。6"""7from collections.abc import Mapping8from dataclasses import replace9import os10import signal11from time import perf_counter1213from .._control_wire import ControlChannel14from .can_block_codec import encode_block, encode_metadata15from ..results.budget import ResourceBudget16from ..sources.text_can_decode import CANTextDecodeRead1718# 条数和估算占用共同决定何时交付；合法记录不能因同批信号较多而被拒绝。19BLOCK_ROWS = 6420WIRE_LIMIT = 52428821BLOCK_RAM = 8 * 104857622RECORD_RAM_LIMIT = BLOCK_RAM // 223BLOCK_OBJECT_TARGET = BLOCK_RAM // 4242526def plain(value):27 """转换有限冻结元数据为可传输结构。参数: JSON式值。返回: 独立容器。异常: 无。28 """29 if isinstance(value, Mapping):30 return {key: plain(item) for key,item in value.items()}31 if isinstance(value,(tuple,list)):32 return tuple(plain(item) for item in value)33 return value343536def facts(reader):37 """摘取同次读取得到的有限事实，不把观测末点当水位。参数: reader。返回: dict。异常: 无。38 """39 return dict(count=reader.count,first_ns=reader.first_ns,last_ns=reader.last_ns,40 decode_calls=reader.decode_calls,first_order_time=reader.first_order_time,41 last_order_time=reader._last_selected_time)424344def read_block(reader):45 """在固定预付空间内读取一块，不为凑满64条继续读取宽记录。4647 参数: reader为逐条解码游标。返回: (records, eof)；只有实际读到EOF才置真。48 异常: 单条记录超过原4MiB估算上限或读取失败；不丢弃、回读或重排记录。49 """50 from .history import _metadata_bytes51 records=[]52 record_cost=053 for _ in range(BLOCK_ROWS):54 try:55 record=next(reader)56 except StopIteration:57 return records,True58 cost=16384+32*_metadata_bytes((record.values,record.frame_values,59 record.frame_layout,str(record.raw.path),getattr(record.raw,'raw_time',None),60 bytes(record.raw.frame.data)))61 if cost>RECORD_RAM_LIMIT:62 raise ValueError('CAN单条记录对象超过已预约空间')63 records.append(record)64 record_cost+=cost65 # 加入前不足2MiB，末条至多4MiB，所以整块不足6MiB；原8MiB预付中66 # 仍有至少2MiB容纳512KiB传输块及编码副本。达到目标就交付，不预读下一条。67 if record_cost>=BLOCK_OBJECT_TARGET:68 break69 return records,False707172def worker_main(control, data):73 """等待文件/读取许可并发送有界结果。参数: 独立控制和数据连接。返回: None。异常: 父断连退出。7475 生命周期由父拥有；正常文件结束关闭解析器，异常先发错误控制消息再关闭。76 任意时刻取消可由父终止工作进程；不等待满数据管道才能报告错误。77 """78 if type(control) is not ControlChannel:79 raise TypeError('CAN工作进程需要固定字节控制端点')80 reader = context = None81 block_number = 082 # 控制台取消由父负责结束完整进程树，不能让子在写块中途单独捕获Ctrl+C。83 signal.signal(signal.SIGINT,signal.SIG_IGN)84 try:85 while True:86 waited=perf_counter()87 command=control.recv()88 wait_seconds=perf_counter()-waited89 if command[0]=='close':90 break91 if command[0]=='file':92 _,spec,signals,options,limit,frame_context=command93 budget=ResourceBudget(limit)94 # 固定块工作空间涵盖构造、序列化及发送副本；父已预约同一额度。95 budget.reserve_ram(BLOCK_RAM)96 if frame_context is not None:97 from .frame_capture import FrameDBCContext98 rules,cycles,integrity=frame_context99 context=FrameDBCContext(spec.dbc_path,rules,budget,cycle_periods=cycles,integrity=integrity)100 reader=CANTextDecodeRead(spec,budget,signals,database_context=context,**options)101 metadata=plain(reader.prepare_metadata())102 packet=encode_metadata(metadata)103 if len(packet)>WIRE_LIMIT:104 raise ValueError('CAN文件元数据超过传输块上限')105 # 先给控制通知，父再接有上限的数据，避免等待相反通道造成死锁。106 control.send(('ready',os.getpid(),len(packet)))107 data.send_bytes(packet)108 block_number=0109 continue110 if command[0]!='next' or reader is None:111 raise ValueError('CAN工作进程收到非法读取许可')112 started=perf_counter()113 records,eof=read_block(reader)114 decoded=perf_counter()-started115 started=perf_counter()116 packet=encode_block(block_number,records,eof,facts(reader))117 serialized=perf_counter()-started118 if len(packet)>WIRE_LIMIT:119 raise ValueError('CAN数据块超过字节上限；不能扩大队列或丢弃记录')120 if eof:121 # 文件结束确认必须晚于解析器和DBC成功关闭，不能先发送成功EOF再报清理错。122 reader.close();reader=None123 if context is not None:124 context.close();context=None125 budget.release_ram(BLOCK_RAM)126 if budget.ram_bytes:127 raise RuntimeError('CAN文件结束后仍持有未释放资源')128 control.send(('block',len(packet),decoded,serialized,wait_seconds))129 started=perf_counter()130 data.send_bytes(packet)131 # 数据管道阻塞与等待下一许可分别计量；父确认发送完成后才复用槽。132 control.send(('sent',perf_counter()-started))133 records=record=packet=None134 block_number+=1135 except (EOFError,BrokenPipeError):136 pass137 except BaseException as error:138 # 仅传定位和原因，不发送完整栈、原始载荷或未完成的累计值。139 try:140 control.send(('error',type(error).__name__,str(error)))141 except (EOFError,BrokenPipeError,OSError):142 pass143 finally:144 if reader is not None:145 reader.close()146 if context is not None:147 context.close()148 control.close();data.close()149