chronaxis._control_wire
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""内部有限控制消息；Connection 只收发 bytes，不使用隐含 pickle。23CCP1 的固定头为版本、通道、命令和 uint64 总字节数，主体分片不超过 64KiB。4普通CAN/fast由已有工作预算限定；独立batch沿用无全批RAM硬上限的边界。5任务与固定结果逐片接收，不按声明总长预分配，也不截断合法身份。6"""7import json8from math import isfinite9import struct1011from ._control_models import pack_message, restore_message, native, check_message_fields1213CHUNK = 6553614_HEADER = struct.Struct('!4sBBQ')15_FAMILIES = {'can':1, 'fast':2, 'batch':3}16_COMMANDS = {17 'can': ('file','next','close','ready','block','sent','error'),18 'fast': ('file','program_file','direct','spool','next','close','offer','delivered','stored','eof','error'),19 'batch': ('run','fast','cleanup','close','result','fast_result','cleaned','cleanup_failed'),20}21_PARENT_COMMANDS = {22 'can': {'file','next','close'},23 'fast': {'file','program_file','direct','spool','next','close'},24 'batch': {'run','fast','cleanup','close'},25}26_LARGE = {'file','program_file','run','fast','result','fast_result','eof'}272829class ControlProtocolError(ValueError):30 """内部协议版本、长度、命令或固定字段错误。"""313233def _pairs(items):34 """建立不重复的JSON对象。3536 参数: items为json解析器给出的键值对。37 返回: dict。38 异常: 重复键时ControlProtocolError。39 """40 result = {}41 for key,value in items:42 if key in result:43 raise ControlProtocolError('控制JSON包含重复键')44 result[key] = value45 return result464748def _depth(data):49 """在解析前核JSON容器深度。5051 参数: data为受长度限定的bytes或bytearray。52 返回: None，忽略字符串内的括号。53 异常: 超过320层或结构未闭合时ControlProtocolError。54 """55 depth = 056 quoted = escaped = False57 for char in data:58 if quoted:59 if escaped:60 escaped = False61 elif char == 92:62 escaped = True63 elif char == 34:64 quoted = False65 elif char == 34:66 quoted = True67 elif char in (91,123):68 depth += 169 if depth > 320:70 raise ControlProtocolError('控制JSON嵌套超过320层')71 elif char in (93,125):72 depth -= 173 if depth < 0:74 raise ControlProtocolError('控制JSON容器结构非法')75 if quoted or depth:76 raise ControlProtocolError('控制JSON截断或容器结构非法')777879def encode_json(value, maximum):80 """逐片编码有限JSON结构。8182 参数: value为已检查原生树，maximum为字节上限或父当前计划的None。83 返回: bytearray。84 异常: JSON类型、非有限数或超过额度。85 """86 output = bytearray()87 encoder = json.JSONEncoder(ensure_ascii=True,allow_nan=False,separators=(',',':'))88 for part in encoder.iterencode(value):89 data = part.encode('ascii')90 if maximum is not None and len(output)+len(data) > maximum:91 raise ControlProtocolError('控制消息超过本次接收额度')92 output.extend(data)93 return output949596def decode_json(data):97 """解码有限JSON为原生值。9899 参数: data为已按本次额度接收的字节。100 返回: dict/list/原生标量。101 异常: 重复键、非有限数、语法或深度错误。102 """103 _depth(data)104 def constant(value):105 """拒绝JSON宽松常量。106107 参数: value为NaN/Infinity类原标记。108 返回: 无，始终抛错。109 异常: ControlProtocolError。110 """111 raise ControlProtocolError('控制JSON不允许非有限常量')112 try:113 value = json.loads(data,object_pairs_hook=_pairs,parse_constant=constant)114 # JSON 1e999 不走parse_constant，需要核最终数值是否有限。115 return native(value)116 except (UnicodeError,json.JSONDecodeError,RecursionError) as error:117 raise ControlProtocolError('控制JSON不是完整有效结构') from error118119120def _integer(value, minimum=0):121 """判断原生整数下界。122123 参数: value候选值，minimum默认0。124 返回: bool，不把bool当整数。125 异常: 无。126 """127 return type(value) is int and value >= minimum128129130def _time(value):131 """判断有限非负计时。132133 参数: value为候选秒数。134 返回: bool。135 异常: 无。136 """137 return type(value) in (int,float) and isfinite(value) and value >= 0138139140def _check(family, command, args):141 """核控制命令基础形状。142143 参数: family通道、command命令、args参数列表或元组。144 返回: None。145 异常: 数量或字段基本类型不符时ControlProtocolError。146 """147 if type(args) not in (tuple,list):148 raise ControlProtocolError('控制消息参数非法')149 if command in ('next','close','direct','spool','cleanup'):150 valid = len(args)==0151 elif command == 'error':152 valid = len(args)==2 and all(type(v) is str for v in args)153 elif family == 'can':154 valid = (command=='file' and len(args)==5155 or command=='ready' and len(args)==2 and _integer(args[0],1) and _integer(args[1],1)156 or command=='block' and len(args)==4 and _integer(args[0],1) and all(_time(v) for v in args[1:])157 or command=='sent' and len(args)==1 and _time(args[0]))158 elif family == 'fast':159 valid = (command in ('file','program_file') and len(args)==4 and _integer(args[0])160 and type(args[1]) is dict and type(args[2]) is str and type(args[3]) is str161 or command in ('offer','delivered','stored') and len(args)==3 and all(_integer(v) for v in args)162 or command=='eof' and len(args)==5 and all(_integer(v) for v in args[:2])163 and type(args[2]) is dict and type(args[3]) is dict and _integer(args[4],1))164 else:165 valid = (command in ('run','fast') and len(args)==3 and type(args[0]) is dict166 and type(args[1]) is str and (args[2] is None or type(args[2]) is str)167 or command=='result' and len(args)==2 and type(args[1]) is bool168 or command=='fast_result' and len(args) in (2,3) and type(args[0]) is dict and type(args[1]) is bool169 or command=='cleaned' and len(args)==1 and args[0] is None170 or command=='cleanup_failed' and len(args)==1 and type(args[0]) in (list,tuple)171 and all(type(v) is str for v in args[0]))172 if not valid:173 raise ControlProtocolError('控制命令参数结构不符：'+command)174175176def _short(value, maximum):177 # 只编码有限前缀，不为截短而遍历完整巨型错误文本。178 """按JSON转义后的字节数截短诊断。179180 参数: value文本，maximum为字节上限。181 返回: 短文本，发生截断时明确注明。182 异常: 输入不是文本时类型错误。183 """184 prefix = value[:maximum]185 encoded = json.dumps(prefix,ensure_ascii=True)186 if len(value)<=maximum and len(encoded)<=maximum:187 return value188 while len(encoded)>maximum-100 and prefix:189 prefix = prefix[:max(0,len(prefix)*3//4)]190 encoded = json.dumps(prefix,ensure_ascii=True)191 return prefix+'…[诊断已截断]'192193194def _diagnostics(family, command, args):195 """限制仅供展示的错误信息。196197 参数: family通道、command命令、args已展开参数。198 返回: 参数列表，只修改独立结果字典中的诊断。199 异常: 诊断字段类型非法。200 """201 if command=='error':202 return [_short(args[0],128),_short(args[1],4096)]203 if family=='batch' and command=='cleanup_failed':204 return [[_short(value,1024) for value in args[0][:16]]]205 if family=='batch' and command in ('result','fast_result'):206 args[0] = dict(args[0])207 for name,maximum in (('error_type',128),('error_message',4096)):208 if args[0].get(name) is not None:209 args[0][name] = _short(args[0][name],maximum)210 args[0]['error_chain'] = [_short(value,1024) for value in args[0].get('error_chain',())[:16]]211 return args212213214class ControlChannel:215 """单个管道端的有限控制编码；不拥有额外句柄，close关闭实际Connection。216217 parent表示父端，只允许向worker派发已定义命令；worker只能回报已定义事实。218 max_bytes是CAN/fast已有预算。独立batch任务及固定结果总长没有原有硬上限；219 每片仍至多64KiB，只随实际收到内容增长，不能把分片长度当作总RAM上限。220 """221 def __init__(self, connection, family, *, parent, max_bytes=None):222 """绑定一个已有控制端点。223224 参数: connection真实Connection，family通道，parent方向，max_bytes已有预算或batch的None。225 返回: 新编码端点，不新增句柄。226 异常: 通道、方向或额度错误。227 """228 if family not in _FAMILIES or type(parent) is not bool:229 raise ValueError('控制通道类型非法')230 if max_bytes is not None and not _integer(max_bytes,1):231 raise ValueError('控制接收额度非法')232 self._connection, self.family, self.parent = connection,family,parent233 self.maximum = max_bytes234 self.last_payload_bytes = 0235236 def _limit(self, operation, *, sending):237 """取得当前消息可用额度。238239 参数: operation固定命令，sending区分发送或接收。240 返回: 字节上限。241 异常: batch固定大消息沿用None；CAN/fast未提供预算时ControlProtocolError。242 """243 if operation not in _LARGE:244 return CHUNK245 if self.family != 'batch':246 if self.maximum is None:247 raise ControlProtocolError('CAN/fast控制通道缺少既有预算上限')248 return self.maximum249 # 路径任务的合法身份长度不能从短路径推断；不新增父端配置读取或任意阈值。250 # recv仅按实际片增长，不按对端声明的总长预约或预分配。251 return None252253 def _direction(self, operation, *, sending):254 """核发送者是否有权发此命令。255256 参数: operation命令，sending为本端发送与否。257 返回: None。258 异常: 父子方向不符时ControlProtocolError。259 """260 from_parent = self.parent if sending else not self.parent261 if (operation in _PARENT_COMMANDS[self.family]) != from_parent:262 raise ControlProtocolError('控制命令发送方向非法')263264 def send(self, message):265 """发送一条固定命令。266267 参数: message为现有控制tuple/list。268 返回: None，返回前释放编码副本。269 异常: 字段、额度或实际管道写入错误。270 """271 if type(message) not in (tuple,list) or not message or message[0] not in _COMMANDS[self.family]:272 raise ControlProtocolError('控制命令未知')273 operation = message[0]274 self._direction(operation,sending=True)275 _check(self.family,operation,message[1:])276 arguments = _diagnostics(self.family,operation,pack_message(self.family,message))277 check_message_fields(self.family,operation,arguments)278 payload = encode_json(arguments,self._limit(operation,sending=True))279 self.last_payload_bytes = len(payload)280 try:281 # 小消息只写一次管道，避免每个原控制通知变成两次系统调用。282 # 大消息首片含固定头，其后每片至多64KiB；不会缓存整批任务。283 first_size=min(len(payload),CHUNK-_HEADER.size)284 first=bytearray(_HEADER.pack(b'CCP1',_FAMILIES[self.family],285 _COMMANDS[self.family].index(operation),len(payload)))286 view = memoryview(payload)287 try:288 first.extend(view[:first_size])289 self._connection.send_bytes(first)290 first=None291 for offset in range(first_size,len(payload),CHUNK):292 self._connection.send_bytes(view[offset:offset+CHUNK])293 finally:294 view.release()295 finally:296 payload = arguments = first = None297298 def recv(self):299 """接收一条固定消息。300301 参数: 无。302 返回: 现有消费者使用的tuple。303 异常: 版本、长度、字段、截断或管道错误。304 """305 first = self._connection.recv_bytes(CHUNK)306 if len(first)<_HEADER.size:307 raise ControlProtocolError('控制消息头截断')308 magic,family,code,size = _HEADER.unpack_from(first)309 if magic!=b'CCP1' or family!=_FAMILIES[self.family] or code>=len(_COMMANDS[self.family]):310 raise ControlProtocolError('控制消息版本/通道/命令未知')311 operation = _COMMANDS[self.family][code]312 self._direction(operation,sending=False)313 maximum = self._limit(operation,sending=False)314 if size==0 or maximum is not None and size>maximum:315 raise ControlProtocolError('控制声明长度超过本次接收额度')316 first_size=min(size,CHUNK-_HEADER.size)317 if len(first)!=_HEADER.size+first_size:318 raise ControlProtocolError('控制首片长度不符')319 payload = bytearray(memoryview(first)[_HEADER.size:])320 first=None321 try:322 while len(payload)<size:323 expected = min(CHUNK,size-len(payload))324 packet = self._connection.recv_bytes(expected)325 if len(packet)!=expected:326 raise ControlProtocolError('控制分片截断或长度不符')327 payload.extend(packet)328 arguments = decode_json(payload)329 _check(self.family,operation,arguments)330 check_message_fields(self.family,operation,arguments)331 self.last_payload_bytes = size332 return restore_message(self.family,operation,arguments)333 finally:334 payload = None335336 def __enter__(self):337 """保持原管道上下文用法。338339 参数: 无。340 返回: 当前编码端点。341 异常: 无。342 """343 return self344345 def __exit__(self, *args):346 """按原管道上下文关闭。347348 参数: args为上下文异常三元组。349 返回: 原Connection.__exit__返回值。350 异常: 真实关闭错误。351 """352 return self._connection.__exit__(*args)353354 def poll(self, *args):355 """查询实际管道是否可读。356357 参数: args沿用Connection.poll的等待参数。358 返回: bool。359 异常: 真实管道错误。360 """361 return self._connection.poll(*args)362363 def fileno(self):364 """返回同一实际管道句柄。365366 参数: 无。367 返回: int，供multiprocessing.wait使用。368 异常: 已关闭等底层句柄错误。369 """370 return self._connection.fileno()371372 @property373 def closed(self):374 """报告实际管道关闭状态。375376 参数: 无。377 返回: bool。378 异常: 无。379 """380 return self._connection.closed381382 def close(self):383 """关闭本端实际管道。384385 参数: 无。386 返回: None。387 异常: 真实关闭错误。388 """389 return self._connection.close()390