chronaxis.execution.state
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""把已经收齐的同刻更新原子提交为各别名的最新规范观测。23输入为当前SignalDefinition/session及DecodedTimeGroup，输出选定更新和冲突数量。4先检查整个组的绑定、dtype/单位和递增时间，再应用既有同刻error/last_by_sequence政策。5组内任一失败不会更新任何别名；无效更新同样覆盖旧值，不重做归一化或回退有效观测。6共享ResourceBudget覆盖声明、最新记录和提交临时容器，不保存全历史。7这是两后端可共享的控制协议；TTL/来源范围/表达式查询和独立对齐内核留后续。8无文件操作或旧链调用，close释放自有引用；调用方保存返回对象的空间另负责。9"""1011# 控制协议只处理已经归一化的标量事实，不调用参考或NumPy数值内核。12from dataclasses import dataclass, replace13from math import isnan14from types import MappingProxyType15from ..core.binding import SignalDefinition16from ..core.numeric import ScalarValue17from ..core.scalar import _scalar_from_validated_fields18from ..core.types import SignalDType19from ..core.validation import require_identifier20from ..results.budget import ResourceBudget, _reserve_owned_ram21from ..results.errors import ResultClosedError22from .groups import DecodedTimeGroup23from .history import _metadata_bytes24from .history_records import HistorySample, estimate_record_bytes252627@dataclass(frozen=True)28class CommittedUpdates:29 """保存一次原子提交结果，别名唯一，原seq/ref/质量及显式冲突数量保留。"""3031 time_ns: int32 updates: tuple[HistorySample, ...]33 conflict_count: int3435 def __post_init__(self):36 """复用完整组检查时间/身份，再限制每别名仅有一个选定观测。3738 参数: 无，读取字段。返回: None。异常: TypeError/ValueError为非法结果。39 """40 DecodedTimeGroup(self.time_ns, self.updates)41 if len({item.name for item in self.updates}) != len(self.updates) or type(self.conflict_count) is not int or self.conflict_count < 0:42 raise ValueError('提交结果需唯一别名与非负原生冲突数')434445def _committed_from_validated_fields(time_ns, updates, conflict_count):46 """包装本次已完成绑定/冲突选择的结果，不再遍历一次时间组。4748 参数: 原完整组时刻、每别名唯一的选定tuple、非负冲突数。49 返回: CommittedUpdates。异常: 分配错误传播，不安装部分状态。50 只由LatestSignalState.apply在全部检查后使用；公开构造仍完整验证。51 """52 result = object.__new__(CommittedUpdates)53 object.__setattr__(result, 'time_ns', time_ns)54 object.__setattr__(result, 'updates', updates)55 object.__setattr__(result, 'conflict_count', conflict_count)56 return result575859def _same_fact(left: HistorySample, right: HistorySample) -> bool:60 """按现有核心比较逻辑值/类型/valid/quality，NaN诊断可以同因一致。6162 参数: left/right已检查的同别名同刻样本。返回: bool事实一致。63 异常: 内部已校验类型错误传播；不把不同CSV位置本身算数值冲突。64 """65 a, b = left.value, right.value66 # dtype/单位另在绑定检查，原位置不同不会改变数值冲突政策。67 return (a.valid == b.valid and a.quality == b.quality and a.code == b.code and type(a.value) is type(b.value)68 and (a.value == b.value or type(a.value) is float and isnan(a.value) and isnan(b.value)))697071@dataclass(frozen=True)72class _StateSnapshot:73 """持有一次完整提交的只读映射、时间、冲突计数与预算量；安装后字段不可改。"""7475 latest: MappingProxyType76 last_time_ns: int | None77 conflict_count: int78 retained_cost: int79 revision: int # 成功完整组提交次数，仅用于内部两段查询证明提交顺序。80 costs: MappingProxyType # 每别名最新记录的已核费用；与latest在同一快照内替换。81 prior: dict | None = None # 仅v4最近一组被替换的(时间,规范标量)，无原记录定位。828384class LatestSignalState:85 """持有固定声明与每别名一个最新记录；成功完整组严格递增，不按批次重置。"""8687 def __init__(self, budget: ResourceBudget, definitions: tuple[SignalDefinition, ...], session_id: str,88 *, declaration_owner=None):89 """先申请声明/容器预算，再检查唯一别名并创建空状态。9091 参数: budget共享账本；definitions当前冻结声明，常量计划可为空tuple；session_id当前会话。92 declaration_owner仅供运行树传入仍持有同一声明的RunSession；独立调用省略。93 返回: 新状态。异常: 配置/预算/取消；保留原异常并清理成功申请，包括申请返回取消。94 持续清理失败时，原异常的state_cleanup_owner提供本次状态的恢复close入口。95 """96 # 没有信号仍需固定付费的时间游标；真实原行/范围由上层证明，不伪造占位信号。97 if not isinstance(budget, ResourceBudget) or type(definitions) is not tuple or any(not isinstance(item, SignalDefinition) for item in definitions):98 raise TypeError('状态需要账本与SignalDefinition tuple')99 # 周期须在来源初始化后核实；最新值状态不接受尚待DBC的声明。100 if any(item.valid_for_ns is None for item in definitions):101 raise ValueError('状态需要已确定的有效期；请先读取DBC周期或提供period/valid_for')102 require_identifier(session_id, 'state.session_id')103 # 新增费用映射只借用声明中的名称，自己拥有字典槽、只读代理和整数费用。104 # 固定申请覆盖一份持有映射；apply另外申请新副本，不能把新旧同时存在算成一份。105 cache_cost = 512 + len(definitions) * 128106 from .static_metadata import static_metadata_bytes107 borrowed = ()108 if declaration_owner is not None:109 from .run_session import RunSession110 if (not isinstance(declaration_owner, RunSession) or declaration_owner.closed or declaration_owner._closing111 or declaration_owner._budget is not budget):112 raise ValueError('共享声明必须由同一账本中仍开放的运行会话持有')113 borrowed = declaration_owner._plan.bindings.signals114 # 运行树保证先关子state再释放父声明；不接受数字/布尔标志作为共享证明。115 # 只有与父会话完全相同的声明对象才借用；DBC补全或类型变换产生的新对象照付。116 # 独立使用时没有父owner，完整声明在本状态中计一次，驱动只借此状态。117 cost = (32768 + len(definitions) * 8192 + cache_cost118 + static_metadata_bytes((definitions, session_id), borrowed=borrowed)119 + sum(512 + 64 * len(item.choices) for item in definitions))120 # 自有状态只经私有指针更新，公开属性不能绕过绑定检查或改写预算所有权。121 self._budget, self._closed, self._session_id = budget, False, session_id122 self._fixed_cost, self._definitions, self._labels, self._snapshot = 0, None, None, None123 # _held_cost记录本对象实际已付总额；与当前快照费用的差额是未提交事务。124 # 这项差额从当前快照推导，成功时只需安装一个快照指针，不再同步第二个标志。125 self._held_cost = 0126 try:127 # 申请也在保护区；单线程账本可辨别操作成功但返回被取消的交接。128 _reserve_owned_ram(self, cost, '_fixed_cost')129 # 同一别名不能隐式选首声明；模型本身已校验别名/策略/单位。130 if len({item.name for item in definitions}) != len(definitions):131 raise ValueError('状态定义别名重复')132 self._definitions = MappingProxyType({item.name: item for item in definitions})133 # ENUM物理下标按声明标签的首次出现顺序，连无效观测也必须携带同一域。134 self._labels = MappingProxyType({item.name: tuple(dict.fromkeys(label for _, label in item.choices))135 for item in definitions if item.dtype is SignalDType.ENUM})136 self._snapshot = _StateSnapshot(MappingProxyType({item.name: None for item in definitions}), None, 0, 0, 0,137 MappingProxyType({item.name: 0 for item in definitions}))138 except BaseException as exc:139 # 构造异常不能被释放取消替换；只重试一次，未抵销费用仍归本状态owner。140 try:141 self.close()142 except BaseException as cleanup:143 exc.add_note(f'状态初始化清理附加失败: {type(cleanup).__name__}')144 if not self._closed:145 try:146 self.close()147 except BaseException as retry:148 exc.add_note(f'状态初始化清理重试失败: {type(retry).__name__}')149 if not self._closed:150 exc.state_cleanup_owner = self151 exc.add_note('释放恢复后调用state_cleanup_owner.close()')152 raise153154 @property155 def budget(self) -> ResourceBudget:156 """返回共享账本引用。参数: 无。返回: ResourceBudget。异常: 无；不转移所有权。"""157 return self._budget158159 @property160 def session_id(self) -> str:161 """返回固定会话标识。参数: 无。返回: str。异常: 无。"""162 return self._session_id163164 @property165 def closed(self) -> bool:166 """返回关闭状态。参数: 无。返回: bool。异常: 无。"""167 return self._closed168169 @property170 def definitions(self) -> MappingProxyType:171 """返回固定声明只读视图。参数: 无。返回: 映射。异常: ResultClosedError为关闭。"""172 self._ensure_open()173 return self._definitions174175 @property176 def latest(self) -> MappingProxyType:177 """返回当前快照只读视图。参数: 无。返回: 映射。异常: ResultClosedError为关闭。178179 调用方跨提交保存旧视图会延长旧观测寿命，该外部保留量不由状态预算负责。180 """181 self._ensure_open()182 return self._snapshot.latest183184 @property185 def fixed_cost(self) -> int:186 """返回声明预算量。参数: 无。返回: int bytes，关闭后0。异常: 无。"""187 return self._fixed_cost188189 @property190 def retained_cost(self) -> int:191 """返回最新记录预算量。参数: 无。返回: int bytes，关闭后0。异常: 无。"""192 return 0 if self._snapshot is None else self._snapshot.retained_cost193194 @property195 def _transaction_delta(self) -> int:196 """取得实际已付量与当前快照的差额。参数: 无。返回: 带符号bytes。异常: 无。"""197 return self._held_cost - self._fixed_cost - self.retained_cost198199 @property200 def last_time_ns(self) -> int | None:201 """返回最后完整组时间。参数: 无。返回: int ns或None。异常: 无。"""202 return None if self._snapshot is None else self._snapshot.last_time_ns203204 @property205 def conflict_count(self) -> int:206 """返回累计显式冲突数。参数: 无。返回: int，关闭后0。异常: 无。"""207 return 0 if self._snapshot is None else self._snapshot.conflict_count208209 @property210 def revision(self) -> int:211 """读取成功完整组的提交版本，失败提交不推进该版本。212213 参数: 无。返回: 原生非负int，初始0，每成功提交增加1。214 异常: ResultClosedError为状态关闭；不能通过此属性修改快照。215 """216 # 版本与latest/time在同一冻结快照中安装，不建立独立可变计数器。217 self._ensure_open()218 return self._snapshot.revision219220 def _ensure_open(self) -> None:221 """拒绝关闭后访问。参数: 无。返回: None。异常: ResultClosedError。"""222 if self.closed:223 raise ResultClosedError('最新信号状态已关闭')224225 def get(self, name: str) -> HistorySample | None:226 """返回当前原样规范观测；无更新是None，无效更新不会回退旧值。227228 参数: name当前简单别名。返回: HistorySample或None。229 异常: KeyError为未知别名；ResultClosedError为关闭。不检查TTL或观测范围。230 """231 self._ensure_open()232 return self.latest[name]233234 def _before_group(self, name):235 """查组前标量。参数: v4别名。返回: (时间,标量)或None。异常: 未声明别名。"""236 snapshot = self._snapshot237 if snapshot.prior is not None and name in snapshot.prior:238 return snapshot.prior[name]239 sample = snapshot.latest[name]240 return None if sample is None else (sample.time_ns, sample.value)241242 def apply(self, group: DecodedTimeGroup) -> CommittedUpdates:243 """整个组检查/选择成功后一次替换最新状态，任何失败保持旧状态。244245 参数: group已收齐的同刻更新，time_ns严格晚于上次提交。246 返回: CommittedUpdates，选定更新按别名排序、每别名一次，冲突数遵循核心政策。247 异常: 类型/时间/绑定/同刻冲突错误或预算超限；不部分更新或重做转换。248 """249 self._ensure_open()250 if self._transaction_delta:251 raise ResultClosedError('最新信号状态尚有未恢复的事务费用，请先关闭state_cleanup_owner')252 if not isinstance(group, DecodedTimeGroup):253 raise TypeError('状态只能提交完整DecodedTimeGroup')254 if self.last_time_ns is not None and group.time_ns <= self.last_time_ns:255 raise ValueError('状态完整组必须严格递增，不能重复提交或回退')256 # 先付分组引用、费用索引及固定别名副本，再建立临时容器。257 # 额外每条256覆盖本组id→费用索引；不持有跨组/跨运行的记录缓存。258 work = (4096 + len(self.definitions) * 512 + 512 + len(self.definitions) * 128259 + 768 * len(group.samples))260 # 旧历史只有原始整数时，当前组要额外构造并保留一个定宽整数 ScalarValue。261 # 每字段包装及安全倍数与 estimate_record_bytes 一致，uint64 最大值覆盖两种原码。262 enrich = sum(1 for item in group.samples if item.name in self.definitions and263 self.definitions[item.name].capture_code and item.value.code is None)264 costs = snapshot = record_costs = None265 try:266 self._change_budget(work)267 # 当前输入对象不可变，临时费用与最终保留费用可共用本次估算。268 record_costs = {id(item): estimate_record_bytes(item) for item in group.samples}269 payload_cost = sum(record_costs.values())270 if enrich:271 from ..results.codec import _estimate272 payload_cost += enrich * 16 * (256 + _estimate(ScalarValue((1 << 64) - 1, SignalDType.UINT64)))273 self._change_budget(payload_cost)274 work += payload_cost275 candidates, identities = {}, set()276 for sample in group.samples:277 definition = self.definitions.get(sample.name)278 if definition is None:279 raise ValueError('完整组包含未声明别名')280 # 完整来源/会话/字段保留，CSV额外定位不能用来冒充其他信号。281 ref = sample.source_ref282 if (ref.session_id != self.session_id or ref.source_name != definition.source or ref.field != definition.field283 or sample.value.dtype is not definition.dtype or sample.unit != definition.unit):284 raise ValueError('完整组当前绑定/dtype/单位不匹配')285 if definition.dtype is SignalDType.ENUM and sample.value.labels != self._labels[sample.name]:286 raise ValueError('完整组ENUM标签域与当前声明不匹配')287 if definition.capture_code and sample.value.code is None:288 from ..core.enum_codes import history_code289 sample = replace(sample, value=replace(sample.value, code=history_code(definition, sample)))290 # 一个别名绑定单一来源；不同CSV定位不能为重复原seq提供合法平手。291 identity = (sample.name, sample.sequence)292 if identity in identities:293 raise ValueError('完整组同别名原sequence重复')294 identities.add(identity)295 candidates.setdefault(sample.name, []).append(sample)296 selected, conflicts = [], 0297 for name in sorted(candidates):298 records = candidates[name]299 same = all(_same_fact(records[0], item) for item in records)300 if not same and self.definitions[name].same_time_policy == 'error':301 raise ValueError(f'同信号同时间冲突：{name}@{group.time_ns}ns')302 # 按原来源sequence选择，与组内遍历顺序无关；相同事实不记冲突。303 selected.append(max(records, key=_sequence))304 conflicts += 0 if same else len(records) - 1305 result = _committed_from_validated_fields(group.time_ns, tuple(selected), conflicts)306 latest = dict(self.latest)307 costs = dict(self._snapshot.costs)308 retained = self.retained_cost309 # v4 where 查询可以在任意次状态提交后才开始，不能靠上次查询缓存区分310 # “条件刷新”和“切换”。每别名最多一项及已计费的规范标签域，包含在311 # 按声明数量/宽度预付的固定状态内；不保留原始定位及无效输入大原文。312 prior = None313 for item in selected:314 if prior is None:315 prior = {}316 old = latest[item.name]317 prior[item.name] = None318 if old is not None:319 value = old.value320 prior[item.name] = (old.time_ns, _scalar_from_validated_fields(value.value if value.valid else None,321 value.dtype, value.valid, value.quality, self._labels.get(item.name, ()), value.code,322 partial=value.partial))323 # 只估算本组真正选中的替换记录；未更新别名沿用不可变快照中的已核费用。324 # 原码补齐发生在选择前，因此这里收费的是最终保留对象，不是原始输入副本。325 cost = record_costs.get(id(item))326 # 补齐原码会产生新对象；只有该情况重新核算最终保留内容。327 if cost is None:328 cost = estimate_record_bytes(item)329 retained += cost - costs[item.name]330 costs[item.name] = cost331 latest[item.name] = item332 # 工作申请包含所有新增记录；原未变记录由原持有申请覆盖。333 change = retained - self.retained_cost334 if change > work:335 raise ValueError('状态内部预估不足，拒绝提交')336 # 构造映射、累计整数和整帧均可能分配；必须在安装前完成才能保持失败原子性。337 snapshot = _StateSnapshot(MappingProxyType(latest), group.time_ns, self.conflict_count + conflicts,338 retained, self.revision + 1, MappingProxyType(costs), prior)339 release_amount = work - change340 record_costs = None341 # 账本转账也先完成；之后仅赋值一个已有指针，没有计数算术或容器构造。342 self._change_budget(-release_amount)343 self._snapshot = snapshot344 return result345 except BaseException as error:346 # 新缓存没有交付时不由异常栈延长其寿命；外部输入和旧只读快照仍归原所有者。347 costs = snapshot = record_costs = None348 self._restore_transaction(error)349 raise350351 def _change_budget(self, delta):352 """为一次直接预算操作登记实际增减，操作完成后取消也不会遗失归属。353354 参数: delta为带符号费用，正数申请、负数退款。返回: None。355 异常: 原预算或取消传播；只观察本次直接调用，不按共享余额计算其他费用。356 """357 if delta == 0:358 return359 before = self._budget.ram_bytes360 try:361 if delta > 0:362 self._budget.reserve_ram(delta)363 else:364 self._budget.release_ram(-delta)365 finally:366 if self._budget.ram_bytes == before + delta:367 self._held_cost += delta368369 def _restore_transaction(self, primary):370 """恢复未提交事务的实际费用差额，清理失败不替换最先发生的错误。371372 参数: primary为提交中的原异常。返回: None。异常: 不主动替换主异常；373 两次恢复仍失败时附state_cleanup_owner，禁止再次apply，由同一对象close处理。374 """375 for attempt in range(2):376 delta = self._transaction_delta377 if delta == 0:378 return379 try:380 # 正差额退回未交付工作区；负差额则补回刚退掉的旧记录费用。381 # 后者不增加实际数据，所需余量正是本事务先前释放的量。382 self._change_budget(-delta)383 except BaseException as cleanup:384 primary.add_note(f'状态事务恢复第{attempt + 1}次附加失败: {type(cleanup).__name__}')385 if self._transaction_delta:386 primary.state_cleanup_owner = self387 primary.add_note('费用恢复后调用state_cleanup_owner.close()')388389 def close(self) -> None:390 """丢弃自有声明与最新记录，抵销账本，不删除历史或修改返回对象。391392 参数: 无。返回: None，幂等。异常: 预算/取消传播；抵销前可重试，已抵销仍最小关闭。393 """394 if self.closed:395 return396 # 外部只读旧视图不改；以账本实际变化区分操作前后取消，避免二次扣账。397 # 失败事务可能尚有正/负差额；只退本对象真实已付金额，不拿旧快照的逻辑费用398 # 代替账本，也不从共享总余额反推本对象需要退多少。399 cost, before, released = self._held_cost, self.budget.ram_bytes, False400 try:401 self.budget.release_ram(cost)402 released = True403 finally:404 if released or self.budget.ram_bytes == before - cost:405 self._closed = True406 self._snapshot = self._definitions = self._labels = None407 self._fixed_cost = self._held_cost = 0408409410def _sequence(sample: HistorySample) -> int:411 """选定稳定原序号。参数: sample规范观测。返回: int sequence。异常: 无。"""412 return sample.sequence413