chronaxis._process_start
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""Linux工作进程在业务入口前登记进程组；此模块只依赖标准库。23独立限长bytes管道只用于启动握手，GO后立即关闭；业务通信仍使用原协议。4"""5import os6import secrets7import signal8import sys9from multiprocessing.connection import wait1011def registered_entry(connection, token, target, args):12 """建立独立进程组，在父进程确认身份后调用原业务入口。1314 参数: connection为启动管道；token为本次随机标记；target及args为原入口与参数。15 返回: 原业务入口的返回值。16 异常: 握手、管道或原入口错误；未收到GO时不执行业务。17 """18 primary = None19 try:20 os.setsid()21 pid = os.getpid()22 connection.send_bytes(b'READY' + token + pid.to_bytes(8, 'big') + os.getpgrp().to_bytes(8, 'big'))23 if connection.recv_bytes(64) != b'PROBE' + token:24 raise ValueError('进程组登记PROBE不匹配')25 connection.send_bytes(b'ACK' + token)26 if connection.recv_bytes(64) != b'GO' + token:27 raise ValueError('进程组登记GO不匹配')28 except BaseException as error:29 primary = error30 raise31 finally:32 # GO之后立即关闭启动通道；业务通信继续使用独立控制端点。33 try:34 connection.close()35 except BaseException as cleanup:36 if primary is None:37 raise38 primary.add_note('启动管道关闭失败: ' + type(cleanup).__name__)39 return target(*args)4041class StartupGate:42 """仅拥有启动管道；父工作槽在分配完成后立即登记本对象。"""4344 def __init__(self, context):45 """创建并拥有本次启动管道。4647 参数: context为multiprocessing的spawn上下文。48 返回: 新的启动资源对象。49 异常: 创建错误原样传播；未关闭端点通过process_start_cleanup_owner保留。50 """51 self.parent = self.child = None52 self.token = secrets.token_bytes(16)53 try:54 self.parent, self.child = context.Pipe()55 except BaseException as error:56 try:57 self.close()58 except BaseException as cleanup:59 error.add_note('启动管道清理失败: ' + type(cleanup).__name__)60 error.process_start_cleanup_owner = self61 raise6263 @property64 def closed(self):65 """查询启动管道是否全部关闭。6667 参数: 无。68 返回: bool；True只表示本对象不再持有端点，不表示业务进程退出。69 异常: 无。70 """71 return self.parent is None and self.child is None7273 def receive(self, process, timeout=5.0):74 """限时等待本次进程的一条有限启动消息。7576 参数: process为本次Process；timeout为等待秒数。77 返回: 最多64字节的bytes。78 异常: 超时、提前退出、消息过长或断管时失败。79 """80 ready = wait((self.parent, process.sentinel), timeout=timeout)81 if self.parent in ready:82 return self.parent.recv_bytes(64)83 if process.sentinel in ready:84 raise ChildProcessError('工作进程在启动登记期间退出')85 raise TimeoutError('等待工作进程登记超时')8687 def close_child(self):88 """关闭父进程持有的启动子端副本。8990 参数: 无。91 返回: None。92 异常: 真实关闭错误；尚未关闭的端点仍由本对象持有。93 """94 if self.child is not None:95 try:96 self.child.close()97 finally:98 if self.child.closed:99 self.child = None100101 def close(self):102 """关闭本对象持有的两个端点。103104 参数: 无。105 返回: None；重复关闭安全。106 异常: 报告首个关闭错误，其余写入notes；未关闭端点保留供恢复。107 """108 errors = []109 for name in ('child', 'parent'):110 connection = getattr(self, name)111 if connection is None:112 continue113 try:114 connection.close()115 except BaseException as error:116 errors.append(error)117 finally:118 if connection.closed:119 setattr(self, name, None)120 if errors:121 for error in errors[1:]:122 errors[0].add_note('另一启动端点关闭失败: ' + type(error).__name__)123 raise errors[0]124125def spawn_entry(context, target, args):126 """为当前平台准备原业务入口，不启动进程。127128 参数: context为spawn上下文；target及args为原入口与参数。129 返回: (入口函数, 参数元组, 启动owner)；Windows原样返回入口，owner为None。130 异常: 平台或pidfd接口不支持时明确拒绝；创建启动管道失败时传播原错。131 """132 if os.name == 'nt':133 return (target, args, None)134 if not sys.platform.startswith('linux') or not hasattr(os, 'pidfd_open') or (not hasattr(signal, 'pidfd_send_signal')):135 raise NotImplementedError('进程树管理需要Windows或支持pidfd组信号的Linux 6.9及以上')136 gate = StartupGate(context)137 return (registered_entry, (gate.child, gate.token, target, args), gate)138