chronaxis.results.spool
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""用有限缓冲暂存当前平坦结果，再按稳定业务键完成有界外部归并。23只为请求的表创建文件，每个缓冲先付编码与排序所需费用，再写JSONL。4活动/片段须已形成最终事实；此模块不读取原日志或调用解码器。5归并每次处理两路且只持当前行，路径按层次和计数生成，不保存全部run列表。6失败关闭句柄并清理自有RAM，保留原错误和可重试的最高所有者。7文件生命周期由结果所有者管理；本层不删除用户文件或仍有消费者的暂存。8"""910# 归并只使用标准库文件和固定模型；文件名根据常量表名/层次/计数构造。11from contextlib import ExitStack12import json13from pathlib import Path14import tempfile15from .budget import ResourceBudget, _reserve_owned_ram, _release_owned_ram, _cleanup_owner16from .codec import decode_row, encode_row, estimate_row_bytes17from .rows import ResultRow, row_table, row_sort_key18from .summary import SUMMARY_MODELS19from .owned_directory import directory_identity20from .errors import AnalysisError, BudgetExceededError, ResultClosedError, ResultFormatError2122TABLES = ('events', 'segments', 'calculations')232425def _buffer_key(item):26 """取已缓存排序键，避免sort期间生成无界的新对象树。2728 参数: item为键/bytes/已付金额的tuple。返回: tuple键。29 异常: 内部结构错误直接传播。30 """31 return item[0]323334class ResultSpool:35 """保存building/analysis_complete/failed/cancelled状态及本次固定启用表计数。3637 序号必须严格递增；每个缓冲排序后写run，seal完成预算内有界归并。38 不接受已有工作目录，不删除任何暂存；close取消未完成构建。39 路径记录只含有限请求表的当前层和run数量，不随结果数保留Python路径列表。40 _fixed_cost/_pending_cost/_buffer_cost/_merge_cost合计为_held_cost，仅含自有RAM。41 _closing拒绝半关闭继续构建；_closed只有全部自有费用实际释放后才成立。42 """4344 def __init__(self, budget: ResourceBudget, work_dir=None, result_buffer_rows: int = 4096, *, enable_sequences: bool = False, requested_tables=None, failure_diagnostics=True, enable_contributions=False, enable_trace=False,enable_selections=False,enable_aggregations=False):45 """创建唯一工作目录并预留固定状态成本。4647 参数: budget为共享账本；work_dir为全新目录或None；result_buffer_rows为行上限；48 enable_sequences为原生bool，按需增加独立配对结果表；requested_tables可明确取其子集，未请求表不创建；failure_diagnostics决定是否另写失败文件。49 enable_contributions/enable_trace/enable_selections按请求增加对应表。50 返回: ResultSpool。异常: ValueError/TypeError为配置；预算/目录/取消原样传播；51 持续清理失败附result_spool_cleanup_owner。已有目录保持原样，不写失败诊断。52 """53 if not isinstance(budget, ResourceBudget) or type(result_buffer_rows) is not int or result_buffer_rows <= 0:54 raise ValueError('暂存需ResourceBudget和正原生缓冲行数')55 if type(enable_sequences) is not bool:56 raise TypeError('enable_sequences需要原生bool')57 if type(failure_diagnostics) is not bool:58 raise TypeError('失败诊断开关需要bool')59 if any(type(flag) is not bool for flag in (enable_contributions,enable_trace,enable_selections,enable_aggregations)):60 raise TypeError('精简贡献/输入追溯开关需要bool')61 available = (TABLES + (('sequences',) if enable_sequences else ())62 + (('contributions',) if enable_contributions else ()) + (('input_trace',) if enable_trace else ())63 + (('selections',) if enable_selections else ()) + (('aggregations',) if enable_aggregations else ()))64 if requested_tables is not None and (type(requested_tables) is not tuple65 or any(type(table) is not str or table not in available for table in requested_tables)66 or len(set(requested_tables)) != len(requested_tables)):67 raise ValueError('请求结果表需已启用白名单中的不重复tuple')68 # 所有可清理字段先于首次预算调用，未返回的对象也能保存真实申请归属。69 self.budget = self._budget = budget70 self.path = None71 self._directory_identity = None72 self._failure_diagnostics = failure_diagnostics73 self._buffer = []74 self._row_limits = None75 self._merge_run_resources = None76 self._fixed_cost = self._pending_cost = self._buffer_cost = self._merge_cost = self._held_cost = 077 self._identity_cost = 078 self._closed = self._closing = False79 self._summary_row_count = 0 # 仅已实际写入的精简行计数，不重扫文件判断格式。80 self._summary_versions = {} # 有限schema版本集合，随成功append登记，不扫描结果。81 self.identity_summary = None # 三个共同身份；出现多个值时该位置为None，不保留所有身份。82 self.state = 'building'83 self.tables = available if requested_tables is None else requested_tables84 try:85 # 固定描述与可选配对表状态先预付后创建。86 _reserve_owned_ram(self, 4096 + (4096 if enable_sequences else 0), '_fixed_cost')87 # 只在成功新建后记录自有目录；已有路径拒绝时不向它写入或删除任何文件。88 if work_dir is None:89 path = Path(tempfile.mkdtemp(prefix='chronaxis-result-'))90 else:91 path = Path(work_dir).absolute()92 path.mkdir(parents=True, exist_ok=False)93 self.path = path94 if requested_tables is not None:95 self._directory_identity = directory_identity(path)96 self.result_buffer_rows = result_buffer_rows97 # 原行上限与两路读取余量保持；这些是需求值，实际已付费用另在槽位中。98 self.max_row_bytes = max(1, budget.memory_budget_bytes // 256)99 self.reader_cost = self.max_row_bytes * 64 + 8192100 self._runs = dict.fromkeys(self.tables, 0)101 self.counts = dict.fromkeys(self.tables, 0)102 # 启用表的固定小字典包含在原固定状态费；只记成功追加的编码行长，103 # 不保存每run/每行索引，也不改变任何表的编码硬上界。104 self._row_limits = dict.fromkeys(self.tables, 0)105 self.files = {}106 self._last_sequence = -1107 self.peak_buffer_rows = 0108 except BaseException as exc:109 _cleanup_owner(self, exc, 'result_spool_cleanup_owner', '结果暂存构造')110 raise111112 def _require_building(self) -> None:113 """拒绝关闭或已封口后的追加，保持最终结果不变。114115 参数: 无。返回: None。异常: ResultClosedError/AnalysisError为状态不符。116 """117 if self._closed or self._closing:118 raise ResultClosedError('结果暂存已关闭或正在清理')119 if self.state != 'building':120 raise AnalysisError(f'不能追加到{self.state}暂存')121122 def row_limit(self, table: str) -> int:123 """提供本表成功追加的实际最大行长，所有表仍受原编码硬限约束。124125 参数: table为本次启用的固定表名。返回: 正bytes上界，空表为1。126 异常: ResultFormatError为未知表或内部行长越界；不读取文件猜大小。127 """128 # 此界仅来自成功encode；损坏文件不能用stat扩容后继续解码。129 if table not in self.tables:130 raise ResultFormatError('未启用的结果表')131 maximum = max(1, self._row_limits[table])132 if type(maximum) is not int or not 1 <= maximum <= self.max_row_bytes:133 raise ResultFormatError('结果表已知最大行长非法')134 return maximum135136 def reader_cost_for(self, table: str) -> int:137 """按已知表上界预付解析及FD空间，原64倍工作费不降低。138139 参数: table为固定表名。返回: 整数bytes。140 异常: 沿row_limit的类型/上界检查；不创建reader或分配数组。141 """142 # 按本表最大已编码行预付解码工作区及一个实际文件缓冲。143 cost = self.row_limit(table) * 64 + 8192144 return max(16384, cost)145146 def append(self, row: ResultRow) -> None:147 """预算检查后编码并暂存一行，必要时先释放已有缓冲。148149 参数: row为最终typed行且sequence严格递增。返回: None。150 异常: AnalysisError/预算/格式错误；失败后不再接受追加。151 """152 try:153 # 公共入口始终使用原模型检查；内部复用不会让调用者提交私有事实。154 self._append_core(row)155 finally:156 row = None157158 def _append_core(self, row) -> None:159 """预算先于编码，成功登记后再按有限行数落盘。160161 参数: row为当前冻结行。返回: None。异常: 原预算/格式/IO/取消错误。162 pending金额在安装时转入buffer槽；失败只清本对象费用和借入引用。163 """164 try:165 self._require_building()166 except BaseException:167 row = None168 raise169 try:170 table = row_table(row)171 if table not in self.tables:172 raise ResultFormatError('本次结果暂存未启用该表')173 if table in self.files:174 raise ResultFormatError('该结果表已完成，不能继续追加')175 if row.sequence <= self._last_sequence:176 raise ResultFormatError('结果sequence必须严格递增且不重复')177 cost = estimate_row_bytes(row)178 if self._buffer and self.budget.ram_bytes + cost > self.budget.memory_budget_bytes:179 self._flush()180 _reserve_owned_ram(self, cost, '_pending_cost')181 payload = encode_row(row, self.max_row_bytes)182 key = (table, row_sort_key(row))183 self._buffer.append((key, payload, cost))184 # 同一owner内转移实际已付费用，不重复申请或提前释放缓冲的归属。185 self._buffer_cost += self._pending_cost186 self._pending_cost = 0187 self._last_sequence = row.sequence188 self.counts[table] += 1189 self._summary_row_count += 1190 self._summary_versions.setdefault(table, set()).add(SUMMARY_MODELS[type(row)][2])191 # 同次已计算行顺带维护固定大小身份摘要，导出无需为省略重复身份再扫描明细。192 identity = (row.dataset_id, row.entity_id, row.session_id)193 if self.identity_summary is None:194 _reserve_owned_ram(self, 256 + 8 * sum(len(value) for value in identity), '_identity_cost')195 self.identity_summary = identity196 else:197 self.identity_summary = tuple(old if old == new else None198 for old, new in zip(self.identity_summary, identity))199 self._row_limits[table] = max(self._row_limits[table], len(payload))200 self.peak_buffer_rows = max(self.peak_buffer_rows, len(self._buffer))201 row = payload = key = None202 if len(self._buffer) >= self.result_buffer_rows:203 self._flush()204 except BaseException as exc:205 row = payload = key = None206 self._fail(exc)207 raise208 finally:209 row = payload = key = None210211 def _run_path(self, table: str, level: int, index: int) -> Path:212 """以受控常量和计数生成run路径，不接受用户行中的路径。213214 参数: table为固定表名，level/index为内部整数。返回: Path。215 异常: 无；只生成路径，不创建文件。216 """217 return self.path / f'{table}-L{level}-{index:012d}.jsonl'218219 def _write_bytes(self, handle, payload: bytes) -> None:220 """配额预留后完整写bytes，短写继续，零写明确失败。221222 参数: handle为有限缓冲或无缓冲二进制输出；payload为有限行。返回: None。223 异常: BudgetExceededError/OSError为配额、磁盘满或短写失败。224 """225 view = None226 try:227 self.budget.reserve_spool(len(payload))228 view = memoryview(payload)229 while view:230 # FileIO可能短写；不能把部分写入当成成功结果。231 written = handle.write(view)232 if not written:233 raise OSError('结果暂存写入未推进')234 view = view[written:]235 finally:236 # 不调用可能再抛错的view.release；只断本层借入引用和真实memoryview。237 view = payload = handle = self = None238239 def _clear_buffer(self) -> None:240 """抵销所有缓冲申请并释放对象，不删除已写文件。241242 参数: 无。返回: None。异常: 原释放/取消传播，真实未抵销费用可由close重试。243 """244 # 先丢本层缓冲引用，再一次抵销实际缓冲总费，避免逐条释放后仍留旧费用tuple。245 self._buffer.clear()246 _release_owned_ram(self, self._buffer_cost, '_buffer_cost')247248 def _flush(self) -> None:249 """将有限缓冲排序并按表依次写run，异常也释放RAM。250251 参数: 无。返回: None。异常: 文件或预算错误原样传播。252 """253 # Timsort键/临时指针已包含每行保守估算；同时只开一个输出句柄。254 failure = None255 try:256 self._buffer.sort(key=_buffer_key)257 with ExitStack() as stack:258 table = None259 handle = None260 for key, payload, _ in self._buffer:261 if key[0] != table:262 if handle is not None:263 handle.close()264 table = key[0]265 run_path = self._run_path(table, 0, self._runs[table])266 handle = stack.enter_context(run_path.open('xb', buffering=0))267 self._runs[table] += 1268 self._write_bytes(handle, payload)269 except BaseException as exc:270 failure = exc271 raise272 finally:273 # 清理取消不能替代写入原错误；未成功清账的buffer费用仍保留在当前owner。274 key = payload = None275 try:276 self._clear_buffer()277 except BaseException as cleanup:278 if failure is None:279 raise280 failure.add_note(f'暂存缓冲清理附加失败: {type(cleanup).__name__}')281282 def _read_entry(self, handle, maximum=None):283 """有界读取一行，按当前格式重建稳定键。284285 参数: handle为实际二进制文件；maximum是已付行上界，省略时用硬限。286 返回: (排序键, 原bytes)，EOF返回None。异常: 超限、缺换行、格式或IO原错。287 """288 maximum = self.max_row_bytes if maximum is None else maximum289 payload = row = None290 try:291 payload = handle.readline(maximum + 1)292 if not payload:293 return None294 if len(payload) > maximum or len(payload) > self.max_row_bytes or not payload.endswith(b'\n'):295 raise ResultFormatError('暂存行超限或缺少完整换行')296 row = decode_row(payload, maximum)297 return row_sort_key(row), payload298 finally:299 # 原TB保留，但释放读槽后不继续借持解码对象或大字节串。300 payload = row = None301302 def _merge(self, left: Path, right: Path | None, destination: Path, *, table=None) -> None:303 """用双路预算，将真实FD关闭与恢复交给共用归并owner。304305 参数: left/right为原有限输入，destination为新run，table为固定表或None（采用全局硬限）。306 返回: None。异常: 原格式、预算、IO或取消；持续关闭失败保实际owner供重试。307 """308 from .merge_runs import merge_runs309 merge_runs(self, (left,) if right is None else (left, right), destination, table=table)310311 def seal(self) -> None:312 """落盘并完成有限外部排序，成功后标analysis_complete。313314 参数: 无；调用侧应已完成结果生成并交付最终行。315 返回: None，files保存本次固定表的最终排序路径。异常: 失败标failed并关闭资源。316 """317 self._require_building()318 try:319 self._flush()320 for table in self.tables:321 self._seal_table(table)322 self.state = 'analysis_complete'323 except BaseException as exc:324 self._fail(exc)325 raise326327 def _seal_table(self,table):328 """完成一个有后续消费者的最终表，其他表仍可追加，不另复制中间阶段。329330 参数: 本次已启用表名。返回: 最终有序路径；重复调用复用同一文件。331 异常: 未启用表/写入/排序失败；调用方按原运行资源树清理。332 """333 self._require_building()334 if table not in self.tables:335 raise ResultFormatError('不能封存未启用结果表')336 if table in self.files:337 return self.files[table]338 self._flush()339 count, level = self._runs[table], 0340 if count == 0:341 # 空表仍有真实文件，发布时可验证三表齐全。342 self._run_path(table, 0, 0).touch(exist_ok=False)343 count = 1344 while count > 1:345 # 只保留整数层次/数量，路径按索引生成，RAM不随run数增长。346 for index in range(0, count, 2):347 destination = self._run_path(table, level + 1, index // 2)348 right = self._run_path(table, level, index + 1) if index + 1 < count else None349 self._merge(self._run_path(table, level, index), right, destination, table=table)350 count = (count + 1) // 2351 level += 1352 self.files[table] = self._run_path(table, level, 0)353 return self.files[table]354355 def _fail(self, exc: BaseException) -> None:356 """保存明确失败状态，诊断写失败不能掩盖原异常。357358 参数: exc为原失败。返回: None。异常: 无正常新增异常。359 """360 self.state = 'cancelled' if isinstance(exc, (KeyboardInterrupt, SystemExit)) else 'failed'361 # 失败状态阻止继续追加，清理错误只作为附注；尚未释放费用由实际owner保留。362 try:363 self._clear_buffer()364 except BaseException as cleanup:365 exc.add_note(f'暂存失败缓冲清理附加失败: {type(cleanup).__name__}')366 if self._merge_run_resources is not None:367 try:368 self._merge_run_resources.close()369 except BaseException as cleanup:370 exc.add_note(f'多路归并真实资源清理附加失败: {type(cleanup).__name__}')371 exc.result_spool_cleanup_owner = self372 self.identity_summary = None373 for slot in ('_pending_cost', '_merge_cost', '_identity_cost'):374 if slot == '_merge_cost' and self._merge_run_resources is not None:375 # helper仍持真实未关FD或未清费用，不能从父层提前抵销同一读槽。376 continue377 try:378 _release_owned_ram(self, getattr(self, slot), slot)379 except BaseException as cleanup:380 exc.add_note(f'暂存失败{slot}清理附加失败: {type(cleanup).__name__}')381 if self._pending_cost or self._buffer_cost or self._merge_cost or self._identity_cost or self._merge_run_resources is not None:382 exc.result_spool_cleanup_owner = self383 if not self._failure_diagnostics:384 return385 try:386 # 小型独立诊断不计作完成清单；磁盘满/配额不足时仅保留对象状态。387 data = json.dumps({'status': self.state, 'error_type': type(exc).__name__,388 'resources': self.budget.snapshot()}, ensure_ascii=False).encode()389 with (self.path / 'failure.json').open('xb', buffering=0) as handle:390 self._write_bytes(handle, data)391 except BaseException as diagnostic:392 # 原诊断失败也不能覆盖业务错误；磁盘/配额原限制保持，其余错误留下附注。393 if not isinstance(diagnostic, (OSError, BudgetExceededError)):394 exc.add_note(f'暂存失败诊断附加失败: {type(diagnostic).__name__}')395396 def close(self) -> None:397 """取消未完成构建，或释放封口对象的固定RAM；重复关闭安全。398399 参数: 无。返回: None。异常: 原释放/取消传播，未释放金额保留，重复关闭可重试。400 """401 if self._closed:402 return403 # 开始清理即拒绝新构建；只释放实际自有槽位，不抵销共享账本中的借用余额。404 self._closing = True405 if self._merge_run_resources is not None:406 try:407 self._merge_run_resources.close()408 except BaseException as exc:409 exc.result_spool_cleanup_owner = self410 raise411 self._clear_buffer()412 if self.state == 'building':413 self.state = 'cancelled'414 self.identity_summary = None415 try:416 for slot in ('_pending_cost', '_merge_cost', '_identity_cost', '_fixed_cost'):417 _release_owned_ram(self, getattr(self, slot), slot)418 finally:419 # 释放后取消同样完成最小关闭；释放前取消仍可重试，不能提前closed。420 if self._held_cost == 0 and self._merge_run_resources is None:421 self._closed = True422