chronaxis.execution.record_sequences
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""收齐同刻的请求、响应和条件后配对，直接交付最终结果。23只保留本次输入组之前的有限条件边界及尚待响应的请求。没有配对请求时不4建立本对象；不会为配对保存整段信号历史或另一时间轴。每个规则的观测终点5由其必要来源决定，响应来源提前结束只使后续响应未知，不提前结束其他来源。6"""78from ..planning.intervals import IntervalChoice9from dataclasses import replace10from ..core.sequence import SequenceRule, SequenceTick11from ..core.sequence_reference import SequenceDetector12from ..core.temporal import ConditionTick13from ..core.types import Truth as T, QualityFlag as Q14from ..results.errors import ResultClosedError, BudgetExceededError15from ..results.spool import _reserve_owned_ram, _release_owned_ram, _cleanup_owner16from .history import _metadata_bytes171819class RecordSequences:20 """拥有同刻合并队列及FIFO候选；计划、状态和结果消费者为借用。"""2122 def __init__(self,plan,state,budget,steps,consumer):23 """登记有限配对配置。参数: 计划、共享状态/预算、周期映射、结果回调。返回: owner。异常: 构造原错附owner。"""24 self._plan,self._state,self._budget,self._steps,self._consumer=plan,state,budget,steps,consumer25 self._held_cost=self._fixed_cost=self._queue_cost=self._pending_cost=self._step_cost=026 self._queue,self._active,self._costs={},{},{}27 self._closed=self._failed=False28 self._rules={r.rule_id:r for r in plan.rules}29 self._needed={name for s in plan.sequences for name in (s.start_rule,s.response_rule)}30 try:31 _reserve_owned_ram(self,65536+32*_metadata_bytes(plan.sequences),'_fixed_cost')32 except BaseException as error:33 _cleanup_owner(self,error,'record_sequences_cleanup_owner','在线配对构造')34 raise3536 def _open(self):37 """拒绝失败实例。参数: 无。返回: None。异常: ResultClosedError。"""38 if self._closed or self._failed:39 raise ResultClosedError('在线配对已关闭或失败')4041 def _entry(self,time_ns):42 """先付费再建立一组。参数: 实际时间。返回: 可变组。异常: 预算原错。"""43 if time_ns not in self._queue:44 _reserve_owned_ram(self,8192,'_queue_cost')45 self._queue[time_ns]=[{},set(),8192]46 return self._queue[time_ns]4748 def bounds(self,cursors,ends):49 """从真实来源范围开始/收窄配对域，在末组交付之前调用。5051 参数: 已前瞻的来源和确认EOF终点。返回: None。异常: 范围/预算错误。52 """53 self._open()54 choices=dict(self._plan.intervals.resolved)55 for spec in self._plan.sequences:56 names=set(self._rules[spec.start_rule].dependencies)|set(self._rules[spec.response_rule].dependencies)57 driver=choices[spec.sequence_id].driver58 if driver:59 names.add(driver)60 from ..planning.transforms import physical_dependencies61 sources={self._state.definitions[name].source for name in physical_dependencies(self._plan,names)}62 if any(not self._rules[name].dependencies and choices.get(name,IntervalChoice('actual')).driver is None63 for name in (spec.start_rule,spec.response_rule)):64 sources.update(source.name for source in self._plan.sources)65 starts=[c.observed_start for c in cursors if c.source_name in sources and c.observed_start is not None]66 if not starts:67 continue68 start=max(min(starts),self._plan.dataset.domain.start_ns)69 end=self._plan.dataset.domain.end_ns70 finished={c.source_name for c in cursors if c._finished}71 final=bool(sources) and sources<=finished72 if final:73 end=min(end,max(ends[source] for source in sources if source in ends))74 if start>end:75 continue76 if spec.sequence_id not in self._active:77 rule=SequenceRule(spec.sequence_id,self._plan.dataset.dataset_id,self._plan.dataset.session_id,78 spec.start_rule,spec.response_rule,spec.min_delay_ns,spec.max_delay_ns,79 self._plan.dataset.entity,record_clock=self._steps[spec.sequence_id] is not None)80 detector=SequenceDetector(rule,start,end,max_pending=spec.max_pending)81 self._active[spec.sequence_id]=[spec,detector,None,False,False]82 self._entry(start)83 detector=self._active[spec.sequence_id][1]84 if not self._plan.dataset.domain_explicit and not detector._ended:85 detector.end_ns=end86 if final and not detector._ended:87 if detector._last_time is not None and end<=detector._last_time:88 raise ValueError('配对末组已提交，不能回退观测范围')89 detector.end_ns=end90 self._entry(end)9192 def observe(self,rule,tick,events):93 """借用规则批时先付费再保留必要点，不保存不参与配对的规则。9495 参数: 规则ID、条件点及事件tuple。返回: None。异常: 预算/重复点错误。96 """97 self._open()98 if rule not in self._needed:99 return100 entry=self._entry(tick.time_ns)101 if rule in entry[0]:102 raise ValueError('配对收到重复规则时间组')103 cost=8192+16*_metadata_bytes(events)104 _reserve_owned_ram(self,cost,'_queue_cost')105 entry[2]+=cost106 entry[0][rule]=(tick,events)107108 def drivers(self,group):109 """仅记录配对实际需要的驱动更新身份。参数: 完整原始组。返回: None。异常: 预算错误。"""110 self._open()111 names={choice.driver for name,choice in self._plan.intervals.resolved112 if name in self._steps and choice.driver is not None}113 selected={sample.name for sample in group.samples}&names114 if selected:115 self._entry(group.time_ns)[1].update(selected)116117 def _consume(self,time_ns,entry,active):118 """一次提交同刻A/B，候选到结果交付完毕才退费。参数: 时间/组/检测状态。返回: None。异常: 原计算错。"""119 spec,detector,last,seen,finished=active120 if finished or not detector.start_ns<=time_ns<=detector.end_ns:121 return122 a=entry[0].get(spec.start_rule)123 b=entry[0].get(spec.response_rule)124 if b is not None:125 condition=b[0]126 elif last is None:127 condition=ConditionTick(time_ns,T.UNKNOWN,Q.MISSING)128 else:129 condition=ConditionTick(time_ns,last.state,last.quality)130 terminal=time_ns==detector.end_ns131 condition=replace(condition,terminal=terminal,132 state=T.UNKNOWN if terminal else condition.state,133 quality=condition.quality|Q.MISSING if terminal else condition.quality)134 starts=() if a is None else a[1]135 responses=() if b is None else b[1]136 _reserve_owned_ram(self,16384+16*_metadata_bytes(responses),'_step_cost')137 for event in starts:138 cost=32768+16*_metadata_bytes((event,detector.rule))139 _reserve_owned_ram(self,cost,'_pending_cost')140 self._costs[(spec.sequence_id,event.event_id)]=cost141 step=self._steps[spec.sequence_id]142 choice=dict(self._plan.intervals.resolved)[spec.sequence_id]143 selected=choice.driver in entry[1]144 advance=step if selected and seen else 0145 active[3]=seen or selected146 try:147 results=detector.push((SequenceTick(condition,starts,responses),),148 advances=None if step is None else (advance,))149 except MemoryError as error:150 raise BudgetExceededError('配对等待候选超过限制：'+spec.sequence_id) from error151 released=0152 for result in results:153 self._consumer(result)154 released+=self._costs.pop((spec.sequence_id,result.request.event_id))155 result=results=None156 _release_owned_ram(self,released,'_pending_cost')157 _release_owned_ram(self,self._step_cost,'_step_cost')158 active[2]=condition159 if terminal:160 detector.finish()161 active[4]=True162163 def flush_before(self,before=None):164 """所有规则已交付边界后，按实际时间处理一次；None仅用于真实全体EOF。165166 参数: 下一个尚未提交的实际组。返回: None。异常: 原错并封闭实例。167 """168 self._open()169 try:170 for time in sorted(self._queue):171 if before is not None and time>=before:172 break173 entry=self._queue[time]174 for active in self._active.values():175 self._consume(time,entry,active)176 cost=entry[2]177 del self._queue[time]178 entry=None179 _release_owned_ram(self,cost,'_queue_cost')180 if before is None and any(not active[4] for active in self._active.values()):181 raise ValueError('配对尚未收到实际终点')182 except BaseException as error:183 self._failed=True184 error.record_sequences_cleanup_owner=self185 raise186187 @property188 def closed(self):189 """返回资源关闭状态。参数: 无。返回: bool。异常: 无。"""190 return self._closed191192 def close(self):193 """放弃未发布候选并退还自有费用。参数: 无。返回: None。异常: 预算释放原错可重试。"""194 if self._closed:195 return196 self._queue.clear();self._active.clear();self._costs.clear()197 self._plan=self._state=self._consumer=self._steps=self._rules=None198 for slot in ('_pending_cost','_step_cost','_queue_cost','_fixed_cost'):199 _release_owned_ram(self,getattr(self,slot),slot)200 self._closed=True201