chronaxis.execution.csv_reader
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""在独立信号CSV读取、引用记录累积和数组分配前申请共享预算。23输入沿用SignalCSVSpec、声明及ReadRequest；输出沿用原SignalBatch和CSVSourceRef。4原csv.reader、解析和文件顺序保持；批次达到样本上限或剩余容量前交付。5物理行readline限制按当时剩余RAM计算，跨物理行的引用记录保守累计，超限失败。6原样本/ref、定宽Unicode提升、owned数组交接与协议pending副本分别计账。7close释放本来源/请求引用和预算，文件保留；不是全进程RSS或外部保留全部输出保证。8默认请求各自保存真实生成器、FD与用途费用槽，关闭取消后保留尚未完成的责任；9未返回构造以csv_source_cleanup_owner、csv_request_cleanup_owner或csv_resources_cleanup_owner恢复。10"""1112# 新来源仍使用既有CSV解释和数值归一化，资源模型不依赖旧CAN链。13from collections.abc import Iterator14from sys import getsizeof, int_info15from ..results.budget import ResourceBudget16from ..results.errors import BudgetExceededError17from ..results.spool import _cleanup_owner, _reserve_owned_ram, _release_owned_ram18from ..sources.signal_csv import SignalCSVSourceAdapter19from ..sources.protocol import SourceClosedError, SourceReadError20from ..sources.models import ReadRequest, SignalBatch21from ..sources.csv_spec import SignalCSVSpec22from ..core.binding import SignalDefinition23from .history import _metadata_bytes242526def _expanded_integer_bytes(text):27 """按十进制文本估算整数展开空间，不重复执行 Decimal 或实际整数转换。2829 参数: text 为待解析单元格。返回: 最多 4096 位十进制整数所需的字节上界。30 异常: 不把坏文本提前当成合法数值；语法和指数边界仍由原解析器拒绝。31 """32 # 普通短整数由记录的固定标量留量覆盖；仅指数或长数值需要额外空间。33 exponent_at = max(text.find('e'), text.find('E'))34 if exponent_at < 0 and len(text) <= 20:35 return 3236 digits = len(text) if exponent_at < 0 else exponent_at37 if exponent_at >= 0:38 exponent = text[exponent_at + 1:].strip()39 if not exponent.startswith('-'):40 exponent = exponent.lstrip('+').lstrip('0')41 if exponent and exponent.isdecimal():42 # 原解析器最多允许 4096 位；不为估算构造任意长度的指数整数。43 digits += int(exponent) if len(exponent) <= 4 else 409644 digits = min(4096, max(1, digits))45 # log2(10) < 3.322；按本 Python 的 PyLong 数字位宽计算，不依赖 float 舍入。46 limbs = (digits * 3322 + int_info.bits_per_digit * 1000 - 1) // (int_info.bits_per_digit * 1000)47 return getsizeof(0) + max(0, limbs - 1) * int_info.sizeof_digit484950class BudgetedSignalCSVSourceAdapter(SignalCSVSourceAdapter):51 """为各请求独立持有检查器，固定声明一次计账，关闭不删除输入。5253 _held_cost只含本来源_input_cost；请求与资源各自拥有自己的费用。54 _closing拒绝继续读取，只有真实子资源及固定槽均已清完才closed。55 未安装请求和检查器保留在明确字段，不借共享余额差额补账。56 """5758 def __init__(self, budget: ResourceBudget, spec: SignalCSVSpec, session_id: str,59 signals: tuple[SignalDefinition, ...], *, backend: str = 'numpy', clock_only=(), retain_original=True, consumer_sample_bytes=0):60 """先计声明，再建立不读取文件的原CSV适配器。6162 参数: budget为共享RAM账本；其余沿用原CSV入口，backend默认numpy。63 clock_only只取时间；retain_original=False省去逐行原文引用，默认保留旧入口行为。64 consumer_sample_bytes为接收方每样本最坏排队费用，只用于容量协调，不重复预约。65 返回: 新预算CSV来源。异常: 类型/配置/预算失败；持续清理失败附csv_source_cleanup_owner。66 """67 if not isinstance(budget, ResourceBudget) or not isinstance(spec, SignalCSVSpec) or type(signals) is not tuple:68 raise TypeError('预算CSV需要ResourceBudget/SignalCSVSpec/信号tuple')69 if type(consumer_sample_bytes) is not int or consumer_sample_bytes < 0:70 raise ValueError('CSV接收方样本费用须为非负整数')71 self._consumer_sample_bytes = consumer_sample_bytes72 # 路径/元数据和choices固定容器在原适配器构造前保守计数。73 # Path/tzinfo不是结果层可序列化模型；路径计原文本，时区只保留原对象引用。74 declaration = (spec.name, spec.timebase,75 spec.time_column, spec.unit, spec.offset_ns, spec.encoding, spec.delimiter,76 spec.missing_values, session_id, signals)77 # 逐路径计数避免在申请前再复制完整路径tuple；每次只持一个路径文本。78 cost = 20480 + _metadata_bytes(declaration) * 16 + sum(512 + _metadata_bytes(str(path)) * 16 for path in spec.paths)79 # 首次预算回调前建立可关闭的真实空壳；父构造尚未返回也不依赖未初始化字段。80 self.budget = self._budget = budget81 self._input_cost = self._held_cost = 082 self.closed = self._closing = False83 self.spec, self.session_id, self.signals, self.backend, self.choices = spec, session_id, signals, backend, {}84 self.active, self.handles = set(), set()85 self._requests, self._resources = set(), set()86 self._uninstalled_request = self._uninstalled_resources = None87 try:88 _reserve_owned_ram(self, cost, '_input_cost')89 super().__init__(spec, session_id, signals, backend=backend,clock_only=clock_only,retain_original=retain_original)90 except BaseException as exc:91 _cleanup_owner(self, exc, 'csv_source_cleanup_owner', '预算CSV来源构造')92 raise9394 def iter_batches(self, request: ReadRequest, *, observe_bounds=False) -> Iterator[SignalBatch]:95 """预留请求/validator固定结构，再建立原参数生成器。9697 参数: request为ReadRequest，投影/会话由原入口验证；observe_bounds仅收计算所需首末时刻。98 返回: 独立预算游标，不生成额外观测报告。99 异常: 关闭/配置/预算/读取错误传播；未返回请求持续清理失败保最高恢复owner，100 不切换后端；batch_rows 是上限，容量不足时提前交付已有样本。101 """102 self._ensure_open()103 # 只限制创建新请求；已有其他活read仍可推进，不能覆盖尚未返回的最高子责任。104 if self._uninstalled_request is not None or self._uninstalled_resources is not None:105 raise SourceClosedError('预算CSV来源尚有未返回请求或资源待清理')106 try:107 iterator = _CSVRequest(self, request, observe_bounds=observe_bounds)108 self._uninstalled_request = iterator109 self._requests.add(iterator)110 self._uninstalled_request = None111 except BaseException as exc:112 iterator = self._uninstalled_request or getattr(exc, 'csv_request_cleanup_owner', None)113 if iterator is not None:114 _cleanup_owner(iterator, exc, 'csv_request_cleanup_owner', '默认CSV请求登记')115 if iterator.closed or iterator in self._requests:116 self._uninstalled_request = None117 else:118 self._uninstalled_request = iterator119 if not iterator.closed:120 exc.csv_source_cleanup_owner = self121 raise122 return iterator123124 def iter_batches_with_observation(self, request: ReadRequest):125 """创建本次独立读取owner，完整耗尽并清理成功后才开放原时间行观测事实。126127 参数: request为原ReadRequest，不改变会话、投影、范围或batch_rows。128 返回: CSVObservationRead；迭代交付原SignalBatch，真正StopIteration后129 observation给出本read独立的首末时间、原行数、请求及读取身份。130 异常: 原配置、预算、读取或取消错误；持续构造清理失败时从原异常的131 csv_observation_read_cleanup_owner恢复close。132 副作用: 增加本read自有固定预算，默认iter_batches不启用此采集。133 """134 # 延迟导入避免预算来源与它的可选资源子类循环；默认入口不导入或构造新owner。135 from .csv_observation import CSVObservationRead136 from ..results.spool import _cleanup_owner137 self._ensure_open()138 # 可选观测同样不能越过默认路径遗留的未返回责任，但不改变其他read的开放检查。139 if self._uninstalled_request is not None or self._uninstalled_resources is not None:140 raise SourceClosedError('预算CSV来源尚有未返回请求或资源待清理')141 read = CSVObservationRead(self, request)142 try:143 # 每个read按对象登记，同一request可独立同时扫描，不用request值作全局key。144 self._requests.add(read)145 except BaseException as exc:146 # 未返回的登记失败也不是可恢复推进状态，最高read只保留关闭责任。147 read._failed = True148 _cleanup_owner(read, exc, 'csv_observation_read_cleanup_owner', 'CSV观测读取登记')149 raise150 return read151152 def _ensure_open(self) -> None:153 """在固定声明已释放后仍抛明确关闭错误，不访问已丢弃spec。154155 参数: 无。返回: None。异常: SourceClosedError为来源已关闭。156 """157 if self.closed or self._closing:158 raise SourceClosedError('预算CSV来源已关闭')159160 def _read_resources(self, request: ReadRequest) -> '_CSVResources':161 """在原迭代器实际打开文件前建立独立资源检查器。162163 参数: request为已校验请求。返回: _CSVResources。164 异常: 固定预算不足或登记失败；持续清理失败保实际检查器及最高来源恢复owner。165 """166 # 此入口只管理未被请求owner接管的原资源，禁止覆盖已有未安装检查器。167 if self._uninstalled_resources is not None:168 raise SourceClosedError('预算CSV来源尚有未返回资源待清理')169 try:170 resources = _CSVResources(self)171 self._uninstalled_resources = resources172 self._resources.add(resources)173 self._uninstalled_resources = None174 except BaseException as exc:175 resources = self._uninstalled_resources or getattr(exc, 'csv_resources_cleanup_owner', None)176 if resources is not None:177 _cleanup_owner(resources, exc, 'csv_resources_cleanup_owner', '默认CSV资源登记')178 self._uninstalled_resources = None if resources.closed or resources in self._resources else resources179 if not resources.closed:180 exc.csv_source_cleanup_owner = self181 raise182 return resources183184 def _close_handle(self, handle, primary):185 """仅在真实FD关闭后移除登记，业务根因不被附加清理错误覆盖。186187 参数: handle实际文件；primary已有原错或None。返回: None。188 异常: 无主错或GeneratorExit时原关闭错误传播；其他主错只增加note。189 """190 try:191 handle.close()192 except BaseException as cleanup:193 if primary is None or isinstance(primary, GeneratorExit):194 raise195 primary.add_note(f'预算CSV文件关闭附加失败: {type(cleanup).__name__}')196 finally:197 if handle.closed:198 self.handles.discard(handle)199200 def close(self) -> None:201 """关闭全部包装请求、原句柄与检查器，最后抵销固定声明预算。202203 参数: 无。返回: None，成功后幂等。异常: 首个关闭错误，仍尝试其他独立请求；204 尚未结束的子责任及费用保留，恢复后可重复close。205 """206 if self.closed:207 return208 self._closing = True209 first_error = None210 # 未登记子对象仍由本来源拥有；已在集合中的同一对象不会再保留第二个未安装槽。211 if self._uninstalled_request is not None:212 try:213 self._uninstalled_request.close()214 self._uninstalled_request = None215 except BaseException as exc:216 first_error = exc217 # 一个生成器关闭失败也不能跳过其他独立请求及原句柄。218 for iterator in tuple(self._requests):219 try:220 iterator.close()221 except BaseException as exc:222 if first_error is None:223 first_error = exc224 else:225 first_error.add_note(f'其他预算CSV请求清理失败: {type(exc).__name__}')226 # 活请求已经负责其生成器/FD/资源；这里只关闭没有更高请求owner的残留资源。227 requests = tuple(self._requests) + (() if self._uninstalled_request is None else (self._uninstalled_request,))228 for iterator in tuple(self.active):229 if any(getattr(request, 'iterator', getattr(request, '_iterator', None)) is iterator for request in requests):230 continue231 try:232 iterator.close()233 self.active.discard(iterator)234 except BaseException as exc:235 if first_error is None:236 first_error = exc237 for handle in tuple(self.handles):238 if any(getattr(request, '_handle', None) is handle for request in requests):239 continue240 try:241 self._close_handle(handle, None)242 except BaseException as exc:243 if first_error is None:244 first_error = exc245 resources_to_close = tuple(self._resources) + (() if self._uninstalled_resources is None else (self._uninstalled_resources,))246 for resources in resources_to_close:247 if any(getattr(request, '_resources', None) is resources for request in requests):248 continue249 try:250 resources.close()251 if resources is self._uninstalled_resources:252 self._uninstalled_resources = None253 except BaseException as exc:254 if first_error is None:255 first_error = exc256 children_closed = (not self._requests and not self._resources and not self.active and not self.handles257 and self._uninstalled_request is None and self._uninstalled_resources is None)258 if children_closed:259 try:260 _release_owned_ram(self, self._input_cost, '_input_cost')261 except BaseException as exc:262 if first_error is None:263 first_error = exc264 finally:265 if self._input_cost == 0:266 self.spec, self.signals, self.choices = None, (), {}267 self.closed = True268 if first_error is not None:269 if not self.closed:270 first_error.csv_source_cleanup_owner = self271 raise first_error272273274class _CSVRequest:275 """拥有默认原生成器、真实FD和资源检查器，半关闭只允许重试close。"""276277 def __init__(self, owner, request, *, observe_bounds=False):278 """先预约原固定请求费用，再接住原生成器，未返回失败保实际恢复owner。279280 参数: owner为预算CSV来源；request沿原结构；observe_bounds按需保存两个时间整数。返回: 独立默认请求。281 异常: 原配置、预算、取消或登记错误；持续清理失败附csv_request_cleanup_owner。282 """283 if type(observe_bounds) is not bool:284 raise TypeError("observe_bounds需要bool")285 self.observe_bounds=observe_bounds286 self.observed_start=self.observed_end=None287 self.owner, self._budget = owner, owner.budget288 self.closed = self._closing = False289 self.current = self.iterator = self._resources = self._handle = None290 self.cost = self._held_cost = 0291 try:292 # 金额与旧请求完全一致；resource_sink只接资源，不采集首末原行事实。293 _reserve_owned_ram(self, 16384 + len(owner.signals) * 4096, 'cost')294 self.iterator = SignalCSVSourceAdapter._request_batches(owner, request, resource_sink=self)295 except BaseException as exc:296 self._adopt_cleanup(exc)297 _cleanup_owner(self, exc, 'csv_request_cleanup_owner', '默认CSV请求构造')298 raise299300 def _adopt_cleanup(self, exc):301 """接住本次尚未返回的检查器，不接他请求或借用来源。302303 参数: exc原异常。返回: None。异常: 无预期异常。304 """305 resources = getattr(exc, 'csv_resources_cleanup_owner', None)306 if resources is not None and self._resources is None:307 self._resources = resources308309 def _created_iterator(self, iterator):310 """在原活动登记前接住同一个实际生成器。311312 参数: iterator原生成器。返回: None。异常: 无预期异常。313 """314 self.iterator = iterator315316 def _read_resources(self):317 """建立本请求唯一检查器，登记前失败仍由本请求保留责任。318319 参数: 无。返回: 真实_CSVResources。异常: 原预算、构造或集合登记错误。320 """321 # 同一次原生成器只建立一个检查器；失败后的旧责任须先由同owner关闭。322 if self._resources is not None:323 raise SourceClosedError('默认CSV请求尚有资源待清理')324 try:325 self._resources = _CSVResources(self.owner)326 self.owner._resources.add(self._resources)327 except BaseException as exc:328 self._adopt_cleanup(exc)329 raise330 return self._resources331332 def _opened_handle(self, handle):333 """在原handles集合登记前接住真实只读FD。334335 参数: handle实际文件。返回: None。异常: 无预期异常。336 """337 self._handle = handle338339 def _close_handle(self, handle, primary):340 """执行原文件关闭责任，失败时保持本请求的实际FD引用。341342 参数: handle实际FD；primary已有业务原错或None。返回: None。343 异常: 无主错或GeneratorExit时传播原关闭错误，其余主错保留并加note。344 """345 try:346 self.owner._close_handle(handle, primary)347 finally:348 if handle.closed:349 self._handle = None350351 def _end_file(self, resources, primary):352 """抵销文件局部费用，已有业务原错或取消不被附加释放错误取代。353354 参数: resources本请求检查器；primary为生成器现场原错。355 返回: None。异常: 正常结束或GeneratorExit时传播原清理错误。356 """357 try:358 resources.end_file()359 except BaseException as cleanup:360 if primary is None or isinstance(primary, GeneratorExit):361 raise362 primary.add_note(f'默认CSV文件费用清理附加失败: {type(cleanup).__name__}')363364 def _close_resources(self, primary):365 """仅在真实FD已关闭后清理本请求检查器，保留原主异常。366367 参数: primary原错或None。返回: None。异常: 正常关闭失败原样传播。368 """369 if self._handle is not None:370 if primary is None:371 raise SourceReadError('默认CSV文件尚未成功关闭')372 return373 try:374 self._resources.close()375 except BaseException as cleanup:376 if primary is None or isinstance(primary, GeneratorExit):377 raise378 primary.add_note(f'默认CSV资源关闭附加失败: {type(cleanup).__name__}')379380 def __iter__(self):381 """返回自有游标。参数: 无。返回: self。异常: 无。"""382 return self383384 def __next__(self) -> SignalBatch:385 """推进原生成器，失败/EOF只清自有资源并保留原异常和最高恢复owner。386387 参数: 无。返回: 原SignalBatch。异常: 原关闭、EOF、预算、读取或取消错误。388 """389 self.owner._ensure_open()390 if self.closed:391 raise StopIteration392 if self._closing:393 raise SourceClosedError('默认CSV请求正在清理')394 # 结束上一批借引用，原生成器及检查器仍持有原费用直到原交接完成。395 self.current = None396 try:397 self.current = next(self.iterator)398 return self.current399 except StopIteration:400 # 真实EOF尚须本请求固定费用关闭成功；此时没有业务主错，不能用StopIteration遮蔽清理取消。401 self.close()402 raise403 except BaseException as exc:404 self._adopt_cleanup(exc)405 _cleanup_owner(self, exc, 'csv_request_cleanup_owner', '默认CSV请求推进')406 raise407408 def close(self) -> None:409 """依次关闭真实生成器/FD/检查器，最后抵销固定费用，成功后才closed。410411 参数: 无。返回: None，幂等。异常: 原关闭或预算错误；尚未清完附最高请求owner。412 """413 if self.closed:414 self.owner._requests.discard(self)415 return416 self._closing = True417 first_error = None418 # 生成器可能已在第一次close中退出finally，FD与检查器字段仍须独立重试。419 if self.iterator is not None:420 try:421 self.iterator.close()422 except BaseException as exc:423 first_error = exc424 if self._handle is not None:425 try:426 self.owner._close_handle(self._handle, None)427 except BaseException as exc:428 if first_error is None:429 first_error = exc430 finally:431 if self._handle.closed:432 self._handle = None433 if self._handle is None and self._resources is not None:434 try:435 self._resources.close()436 except BaseException as exc:437 if first_error is None:438 first_error = exc439 finally:440 if self._resources.closed:441 self._resources = None442 # 实际生成器已关闭且所有子责任结束才释放其固定请求槽；不抵销借用adapter费用。443 iterator_closed = self.iterator is None or self.iterator.gi_frame is None444 if iterator_closed and self._handle is None and self._resources is None:445 self.owner.active.discard(self.iterator)446 self.current = None447 try:448 _release_owned_ram(self, self.cost, 'cost')449 except BaseException as exc:450 if first_error is None:451 first_error = exc452 finally:453 if self.cost == 0:454 self.closed = True455 self.iterator = None456 self.owner._requests.discard(self)457 if first_error is not None:458 if not self.closed:459 first_error.csv_request_cleanup_owner = self460 raise first_error461462463class _CSVResources:464 """每请求按用途保存金额，固定读取缓冲、行/ref、数组和协议状态共同计账。465466 八个实际属性槽是付款依据，charges只提供兼容视图；_held_cost为槽位总和。467 _closing后不再增加费用，所有用途真实抵销完成才closed并解除来源登记。468 """469470 def __init__(self, owner):471 """预留文件解码缓冲及资源容器，不在申请前打开文件。472473 参数: owner为预算CSV来源。返回: 独立检查器。异常: BudgetExceededError。474 """475 self.owner, self.budget, self.closed = owner, owner.budget, False476 self._budget, self._held_cost, self._closing = owner.budget, 0, False477 self._fixed_cost = self._reading_cost = self._header_cost = self._row_cost = 0478 self._pending_cost = self._work_cost = self._array_cost = self._protocol_cost = 0479 self.charges, self.line_source, self.array_cost = {}, None, 0480 self._sample_count, self._column_count = 0, {}481 self._string_layout = {}482 # TextIO解码缓冲和固定容器保守64KiB，不声称整进程RSS。483 try:484 self._add('fixed', 65536)485 except BaseException as exc:486 # 可选观测子类已有同一真实恢复边界，默认构造只处理自己的未返回资源。487 if type(self) is _CSVResources:488 _cleanup_owner(self, exc, 'csv_resources_cleanup_owner', '默认CSV资源构造')489 raise490491 def _sync(self, key):492 """把真实付款槽同步到兼容charges视图，不从视图反推所有权。493494 参数: key为八个固定用途之一。返回: None。异常: 原字典登记错误可重试。495 """496 amount = getattr(self, '_' + key + '_cost')497 if amount:498 self.charges[key] = amount499 else:500 self.charges.pop(key, None)501502 def _add(self, key, amount):503 """事务式登记一次真实预约，未入账超限不增加自有金额。504505 参数: key为用途；amount为bytes。返回: None。异常: 账本超限/金额错误/取消；506 实际已入账而返回取消时仍保留真实付款槽。507 """508 if self.closed or self._closing:509 raise SourceClosedError('默认CSV资源已关闭或正在清理')510 try:511 _reserve_owned_ram(self, amount, '_' + key + '_cost')512 finally:513 self._sync(key)514515 def _release_amount(self, key, amount):516 """只抵销本用途真实费用，释放前后取消均保留正确槽位。517518 参数: key固定用途；amount不超过本槽bytes。返回: None。异常: 原释放或取消。519 """520 try:521 _release_owned_ram(self, amount, '_' + key + '_cost')522 finally:523 self._sync(key)524525 def _move(self, source, target, amount):526 """转移已付款责任槽，总费用不变，不申请或归还共享余额。527528 参数: source/target固定用途，amount为真实已有bytes。返回: None。异常: 金额矛盾ValueError。529 """530 source_slot, target_slot = '_' + source + '_cost', '_' + target + '_cost'531 if amount < 0 or amount > getattr(self, source_slot):532 raise ValueError('默认CSV费用交接超出原用途余额')533 setattr(self, source_slot, getattr(self, source_slot) - amount)534 setattr(self, target_slot, getattr(self, target_slot) + amount)535 self._sync(source)536 self._sync(target)537538 def _release(self, key):539 """抵销某用途全部预留。参数: key用途。返回: None。异常: 账本一致性错误。"""540 self._release_amount(key, getattr(self, '_' + key + '_cost'))541542 def lines(self, handle):543 """把真实TextIO交给限长物理行迭代器，不自己解释CSV。544545 参数: handle为原适配器只读文件。返回: _PhysicalLines。异常: 构造错误传播。546 """547 self.line_source = _PhysicalLines(self, handle)548 return self.line_source549550 def records(self, reader):551 """保持原reader的物理计数和解释，为表头/当前逻辑记录保存累积预算。552553 参数: reader为原csv.reader。返回: _CSVRecords。异常: 构造错误传播。554 """555 return _CSVRecords(self, reader)556557 def _sample_estimate(self, time_text, text, definition):558 """估算新记录独有的容器、数值和文本，不重复计算来源路径或声明。559560 参数: time_text/text 为原单元格；definition 为当前信号声明。561 返回: (记录费用, 字符串原值长度)，长度为零表示没有定宽字符串列。562 异常: 输入已由读取器限定；声明查找错误直接传播。563 """564 # 四项 tuple、list 槽、时间/序号/原值和 CSVSourceRef 的实例/字典留量。565 # 路径、字段、会话和来源名均引用固定声明；不按每条样本重新递归收费。566 cost = 384 + getsizeof(text) + getsizeof(time_text)567 width = 0568 if text not in self.owner.spec.missing_values:569 choices = self.owner.choices[definition.name]570 if text in choices:571 original = choices[text]572 width = len(original) if type(original) is str else 0573 if type(original) is int:574 cost += max(0, getsizeof(original) - 32)575 elif any(type(code) is str for code, _ in definition.choices):576 width = len(text)577 else:578 # 例如 1e4095 只有六个字符，却会在原解析器中展开为 1840B 整数。579 cost += max(0, _expanded_integer_bytes(text) - 32)580 return cost, width581582 def _pack_estimate(self, validator, count, columns, unicode_bytes):583 """分别计算最终列、归一化工作区和跨批同刻检查状态的峰值。584585 参数: validator 为现有协议状态，其余为候选批大小、列数和字节估算。586 返回: (保留列费用, 打包峰值费用)。异常: 无预期异常。587 """588 # 输出五个数组每项至多 27B，附原码列、引用 tuple 和数组对象留量。589 # 输出仍引用的原值/CSVSourceRef/文本已在 pending 付费，交付时转交 array，590 # 不在临时工作区再付第二次；这里只预留新增数组、指针 tuple 和模型对象。591 retained = 4096 + columns * 1024 + count * 64592 # 384B/样本覆盖类型分组 tuple/索引、临时 records、原值指针、数值数组及掩码。593 # Unicode 仅归一化时存在一份原始 UTF32 数组，不是八份长期输出。594 identities = sum(len(state[1]) for state in validator.pending.values())595 state_cost = 4096 + len(self.owner.signals) * 512 + identities * 256 + count * 256596 workspace = retained + count * 384 + unicode_bytes + state_cost597 return retained, workspace598599 def should_flush(self, time_text, text, definition, validator):600 """在再接收一条样本之前判断是否应先交付当前批。601602 参数: 原文本、声明和当前协议状态；不解析值、不修改缓冲或预算。603 返回: bool；已有样本且继续累积会挤占打包空间时为 True。604 异常: 无预期异常；单条无法容纳由实际预约明确失败。605 """606 if not self._sample_count:607 return False608 cost, width = self._sample_estimate(time_text, text, definition)609 columns = len(self._column_count) + (definition.name not in self._column_count)610 old_count, old_width = self._string_layout.get(definition.name, (0, 0))611 unicode_bytes = max((n * w * 4 for name, (n, w) in self._string_layout.items()612 if name != definition.name), default=0)613 if width:614 unicode_bytes = max(unicode_bytes, (old_count + 1) * max(old_width, width) * 4)615 else:616 unicode_bytes = max(unicode_bytes, old_count * old_width * 4)617 if self.owner.backend != 'numpy':618 unicode_bytes = 0619 _, workspace = self._pack_estimate(validator, self._sample_count + 1, columns,620 unicode_bytes)621 # 为下一物理行和当前解析结果交接保留 64KiB；不修改用户的 batch_rows。622 available = self.budget.memory_budget_bytes - self.budget.ram_bytes623 receiver_peak = (self._sample_count + 1) * self.owner._consumer_sample_bytes624 # 有接收队列时，输入打包与队列最多使用当前余量的一半，另一半供正在625 # 计算的积分/统计申请临时数组。此处只决定批大小，不把余量重复登记为费用。626 capacity = available // 2 if self.owner._consumer_sample_bytes else available627 return cost + workspace + receiver_peak + 65536 > capacity628629 def before_sample(self, time_text, text, path, definition):630 """在值解析及原记录 append 前预留记录独有的数据。631632 参数: 原单元格、实际路径及信号声明；路径已在来源构造时计费。633 返回: None。异常: 预算超限或预约取消，仍保留真实已付金额。634 """635 cost, width = self._sample_estimate(time_text, text, definition)636 self._add('pending', cost)637 self._sample_count += 1638 self._column_count[definition.name] = self._column_count.get(definition.name, 0) + 1639 if width:640 count, maximum = self._string_layout.get(definition.name, (0, 0))641 self._string_layout[definition.name] = (count + 1, max(maximum, width))642643 def before_pack(self, pending, validator):644 """在创建数组前预留实际候选批的峰值，不按最大批大小预先拒绝。645646 参数: pending 为有限原缓冲；validator 为跨批协议状态。返回: None。647 异常: 单批真实工作区超限或取消；不丢样本，也不切换计算后端。648 """649 count = sum(len(records) for records in pending.values())650 unicode_bytes = 0651 if self.owner.backend == 'numpy':652 for records in pending.values():653 # 各列依次打包，原始 Unicode 数组在返回数值列前释放，只取最大列峰值。654 widths = [len(item[2]) for item in records if type(item[2]) is str]655 unicode_bytes = max(unicode_bytes, len(widths) * max(widths, default=0) * 4)656 retained, workspace = self._pack_estimate(validator, count, len(pending), unicode_bytes)657 self._add('work', workspace)658 self.array_cost = retained659660 def after_pack(self, validator):661 """校验完成后把工作预留转为当前owned数组和保留协议状态。662663 参数: validator为已accept状态。返回: None。异常: 内部账本一致性错误。664 """665 state_cost = sum(512 + len(state[1]) * 256 for state in validator.pending.values())666 # 旧batch已被原生成器和包装游标交接；新数组预留早已申请，不重复申请。667 self._release('array')668 self._release('protocol')669 # 原对象由缓冲与新列共享，既有付款转交输出责任，不申请第二份费用。670 self._move('pending', 'array', self._pending_cost)671 self._move('work', 'array', self.array_cost)672 self._move('work', 'protocol', state_cost)673 self._release('work')674675 def clear_pending(self):676 """原生成器恢复并清空缓冲后重置计数，不重复抵销已转交的共享费用。677678 参数: 无。返回: None。异常: 账本一致性错误。679 """680 self._release('pending')681 self._sample_count = 0682 self._column_count.clear()683 self._string_layout.clear()684685 def end_file(self):686 """局部行/表头引用丢弃后抵销预算，跨文件样本与协议预算保留。687688 参数: 无。返回: None。异常: 账本一致性错误。689 """690 self.line_source = None691 for key in ('reading', 'header', 'row'):692 self._release(key)693694 def close(self):695 """释放本请求全部金额，不关闭他请求或删除输入文件。696697 参数: 无。返回: None，幂等。异常: 账本一致性错误。698 """699 if self.closed:700 return701 self._closing = True702 self.line_source = None703 for key in ('fixed', 'reading', 'header', 'row', 'pending', 'work', 'array', 'protocol'):704 self._release(key)705 if self._held_cost == 0:706 self.closed = True707 self.owner._resources.discard(self)708709710class _PhysicalLines:711 """每次readline前按剩余RAM申请上限；引用记录跨行继续累积reading金额。"""712713 def __init__(self, resources, handle):714 """保存真实只读句柄。参数: resources检查器/handle文件。返回: 新行游标。异常: 无。"""715 self.resources, self.handle = resources, handle716717 def __iter__(self):718 """返回物理行游标。参数: 无。返回: self。异常: 无。"""719 return self720721 def _retained_line_bytes(self, text, first_line):722 """计算当前逻辑记录仍需保存的解析字符和字段开销。723724 参数: text 为刚读取的物理行；first_line 表示该逻辑记录的第一行。725 返回: 字节估算。异常: 无预期异常。726 """727 if not text:728 return 0729 # 原文本、CSV 的 UTF32 字符缓冲及结果字符串按 16B/字符预留。730 # 每个分隔符另加字段对象/列表槽费用；引号内分隔符保守计入。731 # 固定容器按逻辑记录计一次，不能对引用单元格的每一行重复收 8KiB。732 return (256 if first_line else 0) + len(text) * 16 + text.count(self.resources.owner.spec.delimiter) * 48733734 def __next__(self):735 """有界读完整物理行，超限不把截断内容交给csv.reader。736737 参数: 无。返回: 含原换行的str；EOF停止。异常: BudgetExceededError/原I/O错误。738 """739 budget = self.resources.budget740 # 16B/字符先覆盖文本与字符缓冲；实际分隔符对应的字段费用在交给 CSV 解析器前另加。741 capacity = (budget.memory_budget_bytes - budget.ram_bytes - 8192) // 16 - 1742 if capacity <= 0:743 raise BudgetExceededError('CSV物理行/引用记录累积超出剩余RAM估算')744 amount = 8192 + (capacity + 1) * 16745 first_line = not self.resources._reading_cost746 self.resources._add('reading', amount)747 try:748 # 多读一字符仅判别完整物理行，不能把部分未闭合引号当另一记录。749 text = self.handle.readline(capacity + 1)750 except BaseException as exc:751 try:752 self.resources._release_amount('reading', amount)753 except BaseException as cleanup:754 exc.add_note(f'默认CSV物理行清理附加失败: {type(cleanup).__name__}')755 raise756 retained = self._retained_line_bytes(text, first_line)757 if retained > amount:758 self.resources._add('reading', retained - amount)759 else:760 self.resources._release_amount('reading', amount - retained)761 if len(text) > capacity:762 raise BudgetExceededError('CSV物理行超过剩余RAM允许长度，未解析截断行')763 if not text:764 raise StopIteration765 return text766767768class _CSVRecords:769 """包装原csv.reader，表头固定持有，当前/下一逻辑记录读取时预算重叠。"""770771 def __init__(self, resources, reader):772 """保存原解析器。参数: resources检查器/reader原csv.reader。返回: 新记录游标。异常: 无。"""773 self.resources, self.reader, self.first = resources, reader, True774775 def __iter__(self):776 """返回记录游标。参数: 无。返回: self。异常: 无。"""777 return self778779 def __next__(self):780 """推进原解析器，跨物理行累积到一条完整记录再转移预算。781782 参数: 无。返回: 原单元格list。异常: EOF/原csv错误/预算或I/O错误。783 """784 # 上一cells在调用next期间仍由原生成器保存，先完成新解析再释放旧行预算。785 cells = next(self.reader)786 key = 'header' if self.first else 'row'787 self.resources._release(key)788 self.resources._move('reading', key, self.resources._reading_cost)789 self.first = False790 return cells791