chronaxis.execution.memory_reader
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""在既有内存适配器之外预留输入、批次与协议状态预算。23输入SourceSpec本来已完整存在于调用方内存；封装构造前核算它和信号声明，4不声称控制过去的分配或全进程RSS。输出为既有适配器实际SignalBatch，5归一化次数、backend、分批、原顺序、watermark/EOF和异常原因不改变。6每请求预留最坏两个交接批次及来源pending状态，再创建/推进内部生成器；7NumPy字符串按最大原宽度×N核算Unicode数组/临时副本，不能只计原字符串总长。8超限明确失败，不偷偷减小batch_rows、切换后端或丢样本。所有预算至EOF/close释放，9外部长期保存owned批次不在本封装内部预算保证内；不读取CSV、旧日志或DBC。10"""1112# 原适配器和结构模型负责输入契约，资源账本仅控制新执行层开销。13from collections.abc import Iterator14from ..core.binding import SignalDefinition15from ..core.models import SourceSpec16from ..results.budget import ResourceBudget17from ..results.spool import _cleanup_owner, _reserve_owned_ram, _release_owned_ram18from ..sources.memory import MemorySourceAdapter19from ..sources.models import ReadRequest, SignalBatch, SourceDescription20from ..sources.protocol import SourceClosedError21from .history import _metadata_bytes222324class BudgetedMemorySourceAdapter:25 """共享一个输入预留，为各读取请求保留独立批次/协议最坏预算。2627adapter只在预留成功后建立；info存每别名实际投影总数与最大原字符串字符数。28active登记活动请求，close尝试关闭全部并释放输入引用，原SourceSpec不修改。29_uninstalled_read保构造或登记未返回的真实请求；_closing先拒绝新读，closed仅在30所有子请求、原适配器及本层_input_cost实际释放后成立。31"""3233 def __init__(self, budget: ResourceBudget, source: SourceSpec, session_id: str,34 signals: tuple[SignalDefinition, ...], *, backend: str = 'numpy', clock_only=(), retain_original=True):35 """核算已有完整输入后建立原适配器，不提前生成数组。3637 参数: budget为共享账本；source/session_id/signals/backend与既有内存入口相同。38 clock_only只取时间；retain_original=False不生成原值列，默认仍保留旧行为。39 返回: 新封装。异常: 配置/预算/原适配器初始化原错；持续清理失败附40 memory_source_cleanup_owner，恢复后从该真实owner重试close。41 """42 if not isinstance(budget, ResourceBudget) or not isinstance(source, SourceSpec) or type(signals) is not tuple or any(not isinstance(item, SignalDefinition) for item in signals):43 raise TypeError('预算内存来源需要账本/SourceSpec/typed信号tuple')44 # 所有保留输入和固定描述保守计数，未知对象不参与任意序列化。45 cost = 16384 + _metadata_bytes((source, session_id, signals)) * 1646 self.budget = self._budget = budget47 self.closed = self._closing = False48 self._input_cost = self._held_cost = 049 self._adapter = self._uninstalled_read = None50 self.active: set[_BudgetedBatches] = set()51 self.info = {}52 try:53 # 直接申请返回途中取消也由同一槽登记；不靠子调用后的共享余额补账。54 _reserve_owned_ram(self, cost, '_input_cost')55 self._adapter = MemorySourceAdapter(source, session_id, signals, backend=backend,56 clock_only=clock_only,retain_original=retain_original)57 self.info = {item.name: [0, 0] for item in signals}58 # 只遍历已经提供的memory原行，不重新读取旧输入或进行归一化。59 for row in source.rows if signals else ():60 for item in signals:61 if item.field in row.values:62 self.info[item.name][0] += 163 value = None if item.name in clock_only else row.values[item.field]64 if type(value) is str:65 self.info[item.name][1] = max(self.info[item.name][1], len(value))66 except BaseException as exc:67 _cleanup_owner(self, exc, 'memory_source_cleanup_owner', '预算内存来源构造')68 raise6970 def _ensure_open(self) -> None:71 """拒绝已关闭来源，不以空批掩盖错误。7273 参数: 无。返回: None。异常: SourceClosedError为已关闭。74 """75 if self.closed or self._closing:76 raise SourceClosedError('预算内存来源已关闭或正在清理')7778 def describe(self) -> SourceDescription:79 """直接返回原适配器静态描述，不提升来源能力或观测证据状态。8081 参数: 无。返回: SourceDescription。异常: 关闭/原描述错误传播。82 """83 self._ensure_open()84 return self._adapter.describe()8586 def _request_cost(self, request: ReadRequest) -> int:87 """计算一个读取请求在全部next期间持有的最坏预留。8889 参数: request为当前会话/投影/批次上限。返回: 保守bytes，非RSS。90 异常: ValueError为请求会话/别名错误，其他预算错误由申请点报告。91 """92 self._ensure_open()93 if not isinstance(request, ReadRequest) or request.session_id != self._adapter.session_id or any(name not in self.info for name in request.signal_names):94 raise ValueError('预算内存来源请求会话/投影错误')95 total = sum(self.info[name][0] for name in request.signal_names)96 size = min(total, request.batch_rows)97 max_chars = max((self.info[name][1] for name in request.signal_names), default=0)98 # nbytes至多time/sequence各8、values8、valid1、quality2；4096覆盖复制、装箱/分组。99 per_batch = 65536 + len(request.signal_names) * 8192 + size * 4096100 if self._adapter.backend == 'numpy':101 # 原str可能形成(N,) UTF32定宽列，八倍覆盖原列/转换及模型验证临时副本。102 per_batch += size * max_chars * 4 * 8103 # yield时原生成器仍保留当前批，next构造下一批存在交接重叠，不只申请一套数组。104 # 协议pending最坏保留全部原投影身份并复制set；memory总样本数是已知有限事实。105 return per_batch * 2 + total * 512 + len(request.signal_names) * 2048106107 def iter_batches(self, request: ReadRequest) -> Iterator[SignalBatch]:108 """预留整次请求的最坏缓冲/状态，再创建实际批次生成器。109110 参数: request为既有读取请求，不改变batch_rows。返回: 独立预算游标。111 异常: 请求/预算/原来源错误传播；超限在数组归一化之前失败，持续清理112 失败附memory_read_cleanup_owner，来源也保留未返回read槽。113 """114 cost = self._request_cost(request)115 if self._uninstalled_read is not None:116 raise SourceClosedError('预算内存来源尚有未返回请求待清理')117 try:118 iterator = _BudgetedBatches(self, request, cost)119 except BaseException as exc:120 # 构造未返回的真实read先归来源，不能只保费用而遗失原生成器。121 self._uninstalled_read = getattr(exc, 'memory_read_cleanup_owner', None)122 raise123 self._uninstalled_read = iterator124 try:125 self.active.add(iterator)126 except BaseException as exc:127 # add可能在登记前或后抛错；只关当前read，原注册错误不被close取消替代。128 _cleanup_owner(iterator, exc, 'memory_read_cleanup_owner', '预算内存请求登记')129 raise130 self._uninstalled_read = None131 return iterator132133 def close(self) -> None:134 """关闭所有请求及原适配器，释放本封装预留并保留调用方原输入。135136 参数: 无。返回: None，幂等。异常: 首个原关闭错误附memory_source_cleanup_owner；137 其他独立read也尝试关闭，子责任未完成时保留来源合同与全部未释放槽。138 """139 if self.closed:140 return141 self._closing = True142 try:143 first_error = None144 # 未登记read仍属于本来源；显式槽与活动set可指向同对象，close幂等。145 if self._uninstalled_read is not None:146 try:147 self._uninstalled_read.close()148 except BaseException as exc:149 first_error = exc150 for iterator in tuple(self.active):151 try:152 iterator.close()153 except BaseException as exc:154 if first_error is None:155 first_error = exc156 else:157 first_error.add_note(f'其他内存请求关闭失败: {type(exc).__name__}')158 if first_error is not None:159 raise first_error160 if self._adapter is not None:161 self._adapter.close()162 self._adapter = None163 # 子生成器与原来源全关后才丢输入引用；固定费逐笔确认释放前后取消。164 self.info.clear()165 _release_owned_ram(self, self._input_cost, '_input_cost')166 if self._held_cost == 0:167 self.closed = True168 except BaseException as exc:169 exc.memory_source_cleanup_owner = self170 raise171172173class _BudgetedBatches:174 """持有一请求的预留/原生成器/当前批，EOF或close时抵销一次预留。"""175176 def __init__(self, owner: BudgetedMemorySourceAdapter, request: ReadRequest, cost: int):177 """先申请再创建原请求，构造失败不保留请求状态。178179 参数: owner为预算来源；request为原请求；cost为最坏预留。返回: 独立游标。180 异常: 预算或原请求初始化原错，持续清理失败附memory_read_cleanup_owner；181 未实际入账不抵销，返回途中取消仍登记原cost槽。182 """183 self.owner, self.closed, self.current = owner, False, None184 self._budget, self._held_cost, self.cost = owner.budget, 0, 0185 self.iterator = None186 self._closing = False187 try:188 # 先装恢复槽再申请；旧最坏整次请求费用公式保持原数值。189 _reserve_owned_ram(self, cost, 'cost')190 self.iterator = owner._adapter.iter_batches(request)191 except BaseException as exc:192 _cleanup_owner(self, exc, 'memory_read_cleanup_owner', '预算内存请求构造')193 raise194195 def __iter__(self) -> '_BudgetedBatches':196 """返回本独立游标。197198 参数: 无。返回: self。异常: 无。199 """200 return self201202 def __next__(self) -> SignalBatch:203 """在整次预留覆盖下推进原适配器，不重复归一化或改进度。204205 参数: 无。返回: 原适配器实际批次。异常: EOF/关闭/原来源错误，故障清理保留根因。206 """207 self.owner._ensure_open()208 if self.closed:209 raise StopIteration210 if self._closing:211 raise SourceClosedError('预算内存请求正在清理')212 try:213 self.current = next(self.iterator)214 return self.current215 except StopIteration:216 # 真实耗尽后关闭失败必须传播该失败，不能用StopIteration伪报正常完成。217 self.close()218 raise219 except BaseException as exc:220 _cleanup_owner(self, exc, 'memory_read_cleanup_owner', '预算内存请求推进')221 raise222223 def close(self) -> None:224 """关闭本请求原生成器，释放当前批引用和最坏预留，其他请求保持独立。225226 参数: 无。返回: None，幂等。异常: 原生成器或释放错误附memory_read_cleanup_owner；227 原生成器未成功关闭时保留原对象和费用，其他请求保持独立。228 """229 if self.closed:230 return231 self._closing = True232 try:233 if self.iterator is not None:234 self.iterator.close()235 # 未开始生成器close不执行finally，只解除本read的原登记，不动平行read。236 if self.owner._adapter is not None:237 self.owner._adapter.active.discard(self.iterator)238 self.current = self.iterator = None239 _release_owned_ram(self, self.cost, 'cost')240 # 费用真正抵销后才从父解除，释放后取消可重试而不会重复扣账。241 self.owner.active.discard(self)242 if self.owner._uninstalled_read is self:243 self.owner._uninstalled_read = None244 self.closed = True245 except BaseException as exc:246 exc.memory_read_cleanup_owner = self247 raise248