chronaxis.fast.session_worker
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""可复用文件worker：流式求贡献，获得父进程许可后直传或直接写暂存。23控制和数据独立，任一时刻只有一个尚未交付块。worker不累计总量、不创建4子进程；EOF包含最终块数与已成功关闭的文件事实。父进程拥有取消和回收责任。5"""6from hashlib import sha2567import os8from pathlib import Path9import signal10from time import perf_counter1112from .._control_wire import ControlChannel13from .blocks import batches, encode, LIMIT14from ..dbc_loading import load_dbc_file151617def worker_main(control, data, cancel):18 """运行父授权的文件任务。参数: 控制连接、单向数据连接、取消事件。返回: None。19 异常: 原错误通过控制通道报告；断管退出，finally释放读取器。20 """21 if type(control) is not ControlChannel:22 raise TypeError('fast工作进程需要固定字节控制端点')23 signal.signal(signal.SIGINT, signal.SIG_IGN)24 reader = None25 database = None26 identity = None27 try:28 while True:29 command = control.recv()30 if command[0] == 'close':31 break32 if command[0] not in ('file', 'program_file'):33 raise ValueError('未知fast文件命令')34 programs_checked = command[0] == 'program_file'35 _, index, spec, directory, expected_digest = command36 from .stream import FileContributions37 import cantools38 # 复用DBC只限同一冻结内容；变化属于任务失败，不能混合解码口径。39 content = Path(spec['dbc']).read_bytes()40 digest = sha256(content).hexdigest()41 if digest != expected_digest:42 raise ValueError('运行期间DBC内容变化')43 if identity != digest:44 database = load_dbc_file(spec['dbc'], cantools)45 if sha256(Path(spec['dbc']).read_bytes()).hexdigest() != digest:46 raise ValueError('加载期间DBC内容变化')47 identity = digest48 reader = FileContributions(spec, database, cancel, _programs_checked=programs_checked)49 count = 050 clocks = dict(compute_seconds=0.0, credit_wait_seconds=0.0, send_seconds=0.0, write_seconds=0.0)51 blocks = iter(batches(reader))52 while True:53 if cancel.is_set():54 raise InterruptedError('fast文件计算取消')55 started = perf_counter()56 try:57 records = next(blocks)58 except StopIteration:59 clocks['compute_seconds'] += perf_counter() - started60 break61 packet = encode(index, count, records)62 records = None63 clocks['compute_seconds'] += perf_counter() - started64 control.send(('offer', index, count, len(packet)))65 started = perf_counter()66 route = control.recv()67 clocks['credit_wait_seconds'] += perf_counter() - started68 if route == ('close',):69 return70 started = perf_counter()71 if route == ('direct',):72 data.send_bytes(packet)73 clocks['send_seconds'] += perf_counter() - started74 control.send(('delivered', index, count, len(packet)))75 elif route == ('spool',):76 path = Path(directory) / f'{index}-{count}.block'77 temporary = path.with_suffix('.part')78 with temporary.open('xb') as output:79 if output.write(packet) != len(packet):80 raise OSError('贡献暂存未完整写入')81 from ..results.platform_files import move_no_replace82 move_no_replace(temporary,path)83 clocks['write_seconds'] += perf_counter() - started84 control.send(('stored', index, count, len(packet)))85 else:86 raise ValueError('未知贡献交付许可')87 packet = None88 count += 189 # 等待父确认，直传缓冲不能随主进程处理速度无界累积。90 if control.recv() != ('next',):91 return92 facts = reader.summary()93 if spec.get('_profile', False):94 clocks.update(reader.timings)95 reader = None96 control.send(('eof', index, count, facts, clocks, os.getpid()))97 except (EOFError, BrokenPipeError):98 pass99 except BaseException as error:100 try:101 control.send(('error', type(error).__name__, str(error)))102 except (EOFError, BrokenPipeError, OSError):103 pass104 finally:105 if reader is not None:106 reader.close()107 control.close()108 data.close()109