chronaxis.legacy.process_read
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""在独立进程中逐次读取并解码BLF，确认退出后才交付完整EOF。23输入是冻结的单BLF请求和借用ResourceBudget，输出LegacyDecodedFrame。4构造不启动进程；start加载DBC和公开元数据。每次next只授权一个PULL，5不预取下一帧；子进程通过共用BLFRead读取，并按显式串行政策解码。6子进程在EOF、STOP或失败时关闭自己持有的读取器；父层关闭管道，等待7或终止自己启动的进程，确认退出后才释放对应进程责任。89固定状态、IPC最大包、元数据和当前记录分别计入新层保守RAM费用。10DBC、第三方读取器与解释器的整进程RSS未由本账本限制。11本层不写源文件、不导出BLF、不关闭借用账本或其他read。失败时保留原父12异常；清理仍失败时以can_process_read_cleanup_owner暴露同一真实owner。13跨进程只能保留远端异常事实，不能承诺远端异常对象与父对象相同。14"""1516# 进程和传输对象仅在start中建立，普通导入不加载CAN库或DBC。17from collections.abc import Mapping18import multiprocessing19from ..can_dependencies import runtime_dependency_versions20from sys import getsizeof21from types import MappingProxyType2223# 原样数据模型和有界JSON协议由worker与父进程共用，不接任意pickle对象。24from .records import (MAX_WIRE_BYTES, LegacyCANRequest, decode_packet,25 decoded_from_dict, encode_packet, field_from_dict, field_to_dict)26from .worker import worker_main27from ..results.budget import ResourceBudget28from ..results.errors import AnalysisError, ResultClosedError29from ..results.spool import _cleanup_owner, _release_owned_ram, _reserve_owned_ram30from ..sources.protocol import SourceReadError313233def _wire_metadata_bytes(value):34 """估算已通过有界wire检查的JSON树，复制冻结前先预约保留费用。3536 参数: value为decode_packet得到的dict/list/原生标量。返回: 基础字节估算。37 异常: TypeError为内部未校验类型；不借用只接受冻结tuple的历史元数据估算器。38 """39 # wire的深度与总字节已校验；按原容器计数即可，不先复制成冻结树。40 if type(value) is dict:41 return getsizeof(value) + sum(128 + _wire_metadata_bytes(key)42 + _wire_metadata_bytes(item) for key, item in value.items())43 if type(value) is list:44 return getsizeof(value) + sum(32 + _wire_metadata_bytes(item) for item in value)45 if value is None or type(value) in (str, int, bool, float):46 return getsizeof(value) + 6447 raise TypeError('旧CAN元数据只接受已校验JSON树')484950def _freeze(value):51 """复制一份已经通过JSON边界检查的元数据，拒绝后续原地修改。5253 参数: value为有界协议JSON树。返回: 只读映射、tuple或原标量。54 异常: 原分配错误传播；本函数不读取文件或保留进程对象。55 """56 # 映射和数组逐层冻结，外部消费者不能修改下一次读取的合同。57 if isinstance(value, Mapping):58 return MappingProxyType({key: _freeze(item) for key, item in value.items()})59 if isinstance(value, (list, tuple)):60 return tuple(_freeze(item) for item in value)61 return value626364class RemoteLegacyError(RuntimeError):65 """保存远端错误的有限事实，明确区别于父进程中的原异常对象。"""6667 def __init__(self, facts, phase, request_id):68 """将有界错误包保存为供SourceReadError引用的cause。6970 参数: facts为错误字段映射；phase和request_id为原协议位置。71 返回: 新远端事实异常。异常: 原分配/格式错误，不执行清理。72 """73 # 不动态import远端异常类型，也不根据类型名执行任意代码。74 self.facts = _freeze(facts)75 self.phase, self.request_id = phase, request_id76 module = facts.get('module', 'unknown')77 error_type = facts.get('type', 'unknown')78 message = facts.get('message', '')79 super().__init__(f'旧CAN子进程{phase}失败: {module}.{error_type}: {message}')808182class LegacyCANProcessRead:83 """持有一次读取的真实进程、两端连接及各自费用，不与其他read共享。8485 _waiting表示已有一个发送尝试尚未取得完整回应，禁止叠加许可。86 _process_stopped只在实际确认退出后置真；关闭失败保留真实对象。87 finished仅表示真实协议EOF、正常退出和本owner关闭均成功。88 """8990 def __init__(self, request, budget, *, _worker_target=None):91 """保存请求并预约固定状态，不启动进程或访问DBC。9293 参数: request为LegacyCANRequest；budget为借用ResourceBudget；94 _worker_target仅为隔离协议测试的私有spawn目标，生产使用worker_main。95 返回: 尚未启动的独立read。异常: 配置/预算/分配错误；清理持续失败96 在原异常上提供can_process_read_cleanup_owner，不补共享余额差。97 """98 # 限定初始化包的结构规模，超出当前声明能力时明确拒绝。99 if not isinstance(request, LegacyCANRequest) or not isinstance(budget, ResourceBudget):100 raise TypeError('旧CAN读取需要LegacyCANRequest和ResourceBudget')101 if len(request.fields) > 256:102 raise ValueError('当前单CAN来源最多声明256个字段，且初始化包不能超过64KiB')103 # 第一次预算回调前建立全部真实责任槽，覆盖未返回构造的取消。104 self._budget, self._request = budget, request105 self._worker_target = worker_main if _worker_target is None else _worker_target106 self._connection = self._child_connection = self._process = None107 self._worker_pid = self._exit_code = None108 self._process_stopped = self._waiting = self._started = False109 self._closed = self._closing = self._failed = self._finished = False110 self._eof_received = False111 self._metadata = self._current = self._pending_reply = None112 self._fixed_cost = self._pipe_cost = self._process_cost = 0113 self._wire_cost = self._metadata_cost = self._current_cost = self._held_cost = 0114 self._pulls_issued = self._items_returned = self._items_accepted = 0115 try:116 # 有限初始化字段/路径引用和序列化临时结构均由这一固定预约覆盖。117 _reserve_owned_ram(self, 1048576, '_fixed_cost')118 except BaseException as exc:119 self._failed = True120 _cleanup_owner(self, exc, 'can_process_read_cleanup_owner', '旧CAN进程read构造')121 raise122123 @property124 def closed(self):125 """返回全部真实责任及费用是否已结束。126127 参数: 无。返回: bool。异常: 无，不触碰已关闭Process属性。128 """129 return self._closed130131 @property132 def finished(self):133 """返回本次真实协议EOF是否已完整交付。134135 参数: 无。返回: bool；提前close或失败永远不伪装EOF。异常: 无。136 """137 return self._finished138139 @property140 def worker_pid(self):141 """返回实际启动过的PID，进程对象关闭后仍可用于验证。142143 参数: 无。返回: int或None。异常: 无，不重新启动进程。144 """145 return self._worker_pid146147 @property148 def counters(self):149 """区分许可发送尝试、已接收ITEM和实际交付记录的有限计数。150151 参数: 无。返回: 独立dict；发送中取消时issued是已授予序号的上界，152 不等于已被子进程接收或完成。异常: 原小字典分配错误。153 """154 return {'pulls_issued': self._pulls_issued, 'items_returned': self._items_returned,155 'items_accepted': self._items_accepted}156157 @property158 def metadata(self):159 """读取已完成INIT的同次DBC公开元数据，不隐式启动。160161 参数: 无。返回: 冻结JSON映射，借用至close；长期保留须由消费者计费。162 异常: ResultClosedError为关闭中/已关闭，RuntimeError为尚未READY。163 """164 self._require_open()165 if self._metadata is None:166 raise RuntimeError('start完成前没有旧CAN运行元数据')167 return self._metadata168169 def _require_open(self):170 """阻止失败或半关闭对象再次授权读取。171172 参数: 无。返回: None。异常: ResultClosedError为已关/正在关；173 SourceReadError为失败后继续推进；close仍可重试。174 """175 if self._closed or self._closing:176 raise ResultClosedError('旧CAN进程read已关闭或正在关闭')177 if self._failed:178 raise SourceReadError('失败的旧CAN进程read只允许关闭')179180 def _fail(self, exc, stage):181 """以原父错误为主关闭实际owner，必要时包装来源位置。182183 参数: exc为原错误；stage为有限业务位置。返回: 不返回。184 异常: 父KI/SystemExit、预算及来源错误原对象传播；其他错误作为185 SourceReadError的cause，清理副错只附note/真实恢复owner。186 """187 self._failed = True188 # 先确定对外原错，再把清理证据附在调用者真正会接收到的异常上。189 if isinstance(exc, (KeyboardInterrupt, SystemExit, AnalysisError, SourceReadError)):190 failure = exc191 else:192 failure = SourceReadError(f'旧CAN进程{stage}失败: {exc}')193 failure.__cause__ = exc194 _cleanup_owner(self, failure, 'can_process_read_cleanup_owner', '旧CAN进程read')195 if failure is exc:196 raise failure197 raise failure from exc198199 def start(self):200 """实际启动隔离进程并取得READY，不授权任何消息读取。201202 参数: 无。返回: self，重复成功调用不重启/重载DBC。203 异常: 启动、协议、远端或预算错误；父取消保原对象和关闭责任。204 """205 self._require_open()206 if self._started:207 return self208 try:209 # 两进程的有界wire和解码临时容器预先计费；DBC及整进程RSS另列限制。210 _reserve_owned_ram(self, 128 * MAX_WIRE_BYTES, '_wire_cost')211 _reserve_owned_ram(self, 65536, '_pipe_cost')212 context = multiprocessing.get_context('spawn')213 self._connection, self._child_connection = context.Pipe(duplex=True)214 _reserve_owned_ram(self, 65536, '_process_cost')215 self._process = context.Process(target=self._worker_target,216 args=(self._child_connection,), name='chronaxis-legacy-can')217 try:218 # Process先存真实槽；start成功后取消仍能从同owner回收PID。219 self._process.start()220 finally:221 self._worker_pid = self._process.pid222 self._close_connection('_child_connection')223 packet = {'op': 'init', 'version': 1, 'path': str(self._request.input_path),224 'dbc_path': str(self._request.dbc_path),225 'fields': [field_to_dict(field) for field in self._request.fields]}226 ready = self._exchange(packet)227 self._check_ready(ready)228 # 复制冻结元数据前预约实际保留量，wire预约仍覆盖旧解码包。229 _reserve_owned_ram(self, 4096 + 16 * _wire_metadata_bytes(ready), '_metadata_cost')230 self._metadata = _freeze(ready)231 self._pending_reply = None232 self._started = True233 return self234 except BaseException as exc:235 self._fail(exc, '初始化')236237 def _check_ready(self, packet):238 """核INIT回应来自同一真实PID和同一字段请求，不猜元数据。239240 参数: packet为有界READY dict。返回: None。241 异常: SourceReadError为profile/版本/PID/字段位置不符。242 """243 if (set(packet) != {'op', 'version', 'profile', 'worker_pid', 'fields',244 'dependency_versions'}245 or packet.get('op') != 'ready' or type(packet.get('version')) is not int or packet.get('version') != 1246 or packet.get('profile') != self._request.profile247 or type(packet.get('worker_pid')) is not int248 or packet.get('worker_pid') != self._worker_pid):249 raise SourceReadError('旧CAN READY身份、版本或实际PID不符')250 self._check_ready_metadata(packet)251252 def _check_ready_metadata(self, packet, *, cycle_fields=(), metadata_fields=()):253 """核两种READY共有的字段定位、单位、依赖，不改协议头。254255 参数: packet已由对应读取入口核过完整键集、版本、profile和真实PID；256 cycle_fields仅由确认了周期请求的新版批读取器传入，默认不允许cycle_ns；metadata_fields仅增加有限候选声明。257 返回: None。异常: 原SourceReadError；检查顺序和错误文本沿用逐条入口。258 默认v1和显式按需v2分别检查各自头部，不能伪造v1包绕过策略核对。259 """260 expected_fields = tuple(dict.fromkeys((*self._request.fields,*cycle_fields,*metadata_fields)))261 fields = packet.get('fields')262 if not isinstance(fields, list) or len(fields) != len(expected_fields):263 raise SourceReadError('旧CAN READY字段数量不符')264 for actual, expected in zip(fields, expected_fields):265 if (not isinstance(actual, dict)266 or set(actual) - ({'cycle_ns'} if expected in cycle_fields else set()) - {'dbc_frames'} not in267 ({'locator', 'unit'}, {'locator', 'unit', 'frame_format_compatibility'})268 or actual['unit'] is not None and type(actual['unit']) is not str):269 raise SourceReadError('旧CAN READY没有对应原请求的字段定位')270 if expected in cycle_fields and (type(actual.get('cycle_ns')) is not int271 or not 0 < actual['cycle_ns'] <= (1 << 63)-1):272 raise SourceReadError('DBC周期元数据缺失或不是正int64纳秒')273 # dict相等会把False与0混同；先按原生int/bool合同重建，再比较定位。274 try:275 locator = field_from_dict(actual['locator'])276 except (TypeError, ValueError) as exc:277 raise SourceReadError('旧CAN READY字段定位类型非法') from exc278 if locator != expected:279 raise SourceReadError('旧CAN READY没有对应原请求的字段定位')280 needs_candidates = locator.is_extended is None or locator.is_fd is None281 if ('dbc_frames' in actual) != needs_candidates:282 raise SourceReadError('旧CAN READY候选帧型与省略参数声明不符')283 if 'dbc_frames' in actual:284 pairs = actual['dbc_frames']285 if (type(pairs) is not list or not 1 <= len(pairs) <= 4286 or any(type(pair) is not list or len(pair) != 2287 or any(type(flag) is not bool for flag in pair) for pair in pairs)):288 raise SourceReadError('旧CAN READY候选帧型非法')289 if len({tuple(pair) for pair in pairs}) != len(pairs) or any(290 locator.is_extended is not None and pair[0] != locator.is_extended291 or locator.is_fd is True and pair[1] is not True for pair in pairs):292 raise SourceReadError('旧CAN READY候选帧型与显式参数冲突或重复')293 if 'frame_format_compatibility' in actual:294 # 只承认固定单向提示，不将0/1当bool，也不接受其他兼容或额外字段。295 notice = actual['frame_format_compatibility']296 if (type(notice) is not dict297 or set(notice) != {'code', 'dbc_is_fd', 'requested_is_fd'}298 or type(notice['code']) is not str or notice['code'] != 'dbc_fd_classic_selector'299 or notice['dbc_is_fd'] is not True or notice['requested_is_fd'] is not False300 or locator.is_fd is not False):301 raise SourceReadError('旧CAN READY帧型兼容提示非法或与原定位不同')302 # 安装范围由包元数据约束；父子必须使用同一解释器与实际依赖版本。303 expected_versions = runtime_dependency_versions()304 if packet['dependency_versions'] != expected_versions:305 raise SourceReadError('CAN READY依赖版本与父进程不一致')306307 def _exchange(self, packet):308 """完成一个有界请求/回应；STATUS/ACK也不隐含PULL许可。309310 参数: packet为协议dict。返回: 有界回应dict。311 异常: 原父取消/管道错误、SourceReadError或明确远端cause。312 边界: 最多一项在途，父每50ms回到Python检查取消；本方法发送后立即等待。313 """314 try:315 self._send_request(packet)316 return self._receive_response(packet)317 finally:318 packet = None319320 def _send_request(self, packet):321 """发送唯一在途许可；发送中断时仍保留实际连接和未确认责任。322323 参数: packet为调用方已经为相应工作空间付款的协议dict。324 返回: None。异常: 原编码/发送/取消错误；失败不可按未发送重试。325 边界: 原逐条入口立即接收；批次入口可在消费当前批时保留这一项在途请求。326 """327 payload = None328 try:329 if self._connection is None or self._waiting:330 raise SourceReadError('旧CAN连接不可用或已有一个在途请求')331 payload = encode_packet(packet)332 self._waiting = True333 self._connection.send_bytes(payload)334 finally:335 packet = payload = None336337 def _receive_response(self, packet):338 """接收先前发送的许可，并严格核对远端错误事实与请求身份。339340 参数: packet为同一在途许可，不再次发送。返回: 有界回应dict。341 异常: 原连接/格式/取消错误，或带RemoteLegacyError cause的来源错误。342 边界: 必须已有一个在途请求；没有真实协议EOF的退出仍是错误。343 """344 reply = facts = remote = None345 try:346 if self._connection is None or not self._waiting:347 raise SourceReadError('旧CAN连接没有对应的在途请求')348 # poll不授予读取许可；子进程意外退出也不能当成真实EOF。349 while not self._connection.poll(0.05):350 if self._process is not None and not self._process.is_alive():351 raise SourceReadError(f'旧CAN进程未交付完整协议回应便退出: {self._process.exitcode}')352 reply = self._receive_reply()353 self._waiting = False354 if reply.get('op') == 'error':355 facts = reply.get('error')356 if (set(reply) != {'op', 'request_id', 'phase', 'error'}357 or type(reply.get('phase')) is not str358 or reply.get('request_id') is not None and type(reply.get('request_id')) is not int359 or not isinstance(facts, dict)360 or set(facts) != {'module', 'type', 'message', 'traceback', 'notes'}361 or any(type(facts[key]) is not str for key in ('module', 'type', 'message', 'traceback'))362 or type(facts['notes']) is not list363 or any(type(note) is not str for note in facts['notes'])):364 raise SourceReadError('旧CAN远端错误包缺少明确事实')365 if reply['request_id'] != packet.get('request_id'):366 raise SourceReadError('旧CAN远端错误不属于当前请求')367 remote = RemoteLegacyError(facts, reply.get('phase'), reply.get('request_id'))368 raise SourceReadError(str(remote)) from remote369 return reply370 finally:371 # 原错误可留在调用方；已结束调用的临时引用不继续占有整批对象。372 packet = reply = facts = remote = None373374 def _receive_reply(self):375 """读取本协议的一条严格JSON回应，供批次子类替换物理数据包边界。376377 参数: 无，使用本owner的实际Connection。378 返回: 已校验的回应dict。异常: 原管道、长度、JSON或取消错误。379 边界: 原逐条入口仍只接收64KiB；子类不能绕过共享接收流程的在途及远端错误检查。380 """381 self._pending_reply = self._connection.recv_bytes(MAX_WIRE_BYTES)382 return decode_packet(self._pending_reply)383384 def __iter__(self):385 """返回本次独立read，不因取得迭代器就启动或读取。386387 参数: 无。返回: self。异常: 原错误只在实际next/start时报告。388 """389 return self390391 def __next__(self):392 """只授权并返回一帧，真实完整EOF之后才StopIteration。393394 参数: 无。返回: LegacyDecodedFrame；调用者保留引用须另计费。395 异常: StopIteration为已确认EOF；源/预算/协议错误或父原取消。396 """397 if self._finished:398 raise StopIteration399 self._require_open()400 try:401 self.start()402 # 上一记录停止由本read持有后才归还其费用，失败不会继续发PULL。403 self._current = None404 _release_owned_ram(self, self._current_cost, '_current_cost')405 self._pending_reply = None406 request_id = self._items_accepted407 self._pulls_issued += 1408 reply = self._exchange({'op': 'pull', 'request_id': request_id})409 if type(reply.get('request_id')) is not int or reply.get('request_id') != request_id:410 raise SourceReadError('旧CAN回应不是本次唯一许可序号')411 if reply.get('op') == 'eof':412 if (set(reply) != {'op', 'request_id', 'frame_count'}413 or type(reply.get('frame_count')) is not int414 or reply.get('frame_count') != self._items_accepted):415 raise SourceReadError('旧CAN EOF原帧计数不符')416 self._eof_received = True417 self.close()418 if self._exit_code != 0:419 raise SourceReadError('旧CAN虽然发EOF，但没有正常退出')420 self._finished = True421 elif reply.get('op') == 'item':422 if set(reply) != {'op', 'request_id', 'record'}:423 raise SourceReadError('旧CAN ITEM字段缺失或多余')424 self._items_returned += 1425 raw = reply.get('record')426 _reserve_owned_ram(self, 4096 + 16 * _wire_metadata_bytes(raw), '_current_cost')427 self._current = decoded_from_dict(raw)428 if self._current.sequence != request_id:429 raise SourceReadError('旧CAN原yield序号不连续，禁止重排或补号')430 self._items_accepted += 1431 self._pending_reply = None432 return self._current433 else:434 raise SourceReadError('旧CAN PULL必须得到ITEM或真实EOF')435 except BaseException as exc:436 self._fail(exc, '逐帧读取')437 # 正常StopIteration不进入失败清理，已关闭read也能重复返回EOF。438 raise StopIteration439440 def _close_connection(self, slot):441 """关闭一个真实自有连接，成功后取消也不丢失其实际关闭事实。442443 参数: slot为本owner连接字段。返回: None。444 异常: 原close错误；未关闭对象仍在同槽，费用稍后单独归还。445 """446 connection = getattr(self, slot)447 if connection is None:448 return449 try:450 connection.close()451 finally:452 if connection.closed:453 setattr(self, slot, None)454455 def _close_process(self):456 """确认自己启动的进程退出后关闭其实际句柄，不动其他read。457458 参数: 无。返回: None，实际exitcode保留供EOF判断。459 异常: join/terminate/kill/close错误保原process和停止确认状态以便重试。460 边界: 三次显式join最多约6秒；Windows kill仍是TerminateProcess，461 必须实际取得exitcode，不能把终止请求成功当作进程已经退出。462 """463 process = self._process464 if process is None:465 return466 if not self._process_stopped:467 # start前失败没有PID，仍须关闭Process自身已经建立的描述。468 if process.pid is not None:469 process.join(timeout=0.5)470 if process.exitcode is None:471 process.terminate()472 process.join(timeout=0.5)473 if process.exitcode is None:474 process.kill()475 # Windows终止后的OS清理可能晚于0.5秒；有限等待真实退出信号。476 process.join(timeout=5.0)477 exit_code = process.exitcode478 if exit_code is None:479 failure = SourceReadError('尚未确认旧CAN子进程退出，不能结束关闭责任')480 failure.can_process_exit_state = {'pid': self._worker_pid,481 'exit_code': None, 'join_wait_limits_seconds': (0.5, 0.5, 5.0)}482 raise failure483 self._exit_code = exit_code484 self._process_stopped = True485 # 此确认先于close，覆盖真实close后取消时已关闭Process属性不可再读的情况。486 process.close()487 self._process = None488489 def _release_closed_state(self):490 """实际连接和进程全部关闭后清状态、逐槽退费，供批次子类补充责任。491492 参数: 无；调用方必须已完成本owner实际资源关闭。返回: None。493 异常: 原预算回调错误；费用槽保留实际入账事实，同一close可继续恢复。494 """495 self._metadata = self._current = self._pending_reply = self._request = None496 self._worker_target = None497 self._waiting = False498 for slot in ('_current_cost', '_metadata_cost', '_wire_cost', '_pipe_cost', '_process_cost', '_fixed_cost'):499 _release_owned_ram(self, getattr(self, slot), slot)500501 def close(self):502 """停止授予许可，关闭自有IPC/进程，最后归还各自费用。503504 参数: 无。返回: None，完成后重复调用幂等。505 异常: 实际关闭或预算回调错误；对象仍可同owner重试。506 边界: 关管道令等待命令的child退出；在途旧IO不响应时有限等待后终止，507 不为关闭继续next，也不把强制退出当正常分析完成。508 """509 if self._closed:510 return511 self._closing = True512 try:513 # 任一真实资源未关时即保留父责任，不先释放其内存描述或借用余额。514 self._close_connection('_child_connection')515 self._close_connection('_connection')516 self._close_process()517 # 每笔实际申请有独立槽，释放后取消由既有helper确认，不重复抵销别人的费用。518 self._release_closed_state()519 self._closed = True520 except BaseException as exc:521 # 直接close的调用者也能从同一异常找到真实责任，不制造替代owner。522 exc.can_process_read_cleanup_owner = self523 raise524