"""One channel control owner; independent fenced provider Frame writers.""" from copy import deepcopy from dataclasses import replace from functools import partial import json import multiprocessing import queue import threading import time from app.channel_child import run_channel from app.channel_process import ProcessSlots, ProcessStreamGroup from app.process_metadata import CHANNEL_BYTES, ChildPublisher from app.provider_delivery import provider_spec from app.provider_plan import plan from app.stream_lifecycle import make_stream_failure class ProviderSlots(ProcessSlots): def __init__(self, parent): super().__init__() self.parent = parent def guard(self): # The outer channel already owns the host lock and channel admission slot. self.parent.guard() class ProviderProcessGroup(ProcessStreamGroup): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self._validate_split_storage(self.spec) self._split_plan = None self._split_active = False self._children = {} self._snapshots = {} self._provider_status = {} self._split_publisher = None self._split_stop = threading.Event() self._split_failure = None def check_restart(self, candidate): super().check_restart(candidate) self._validate_split_storage(candidate) @staticmethod def _validate_split_storage(spec): if len(plan(spec).providers) > 1 and (spec.get('store', True) or spec.get('active_learning', False) or spec.get('persist_mode') == 'event_gated'): raise ValueError('Provider processes require store/active_learning Off and no legacy event_gated storage.') def start(self): with self._teardown_lock: if self._split_active: if self.desired_state == 'running' and self._pending_restart is None: return self.run_generation raise RuntimeError('Provider cleanup is pending.') if self._process is not None or not self._cleanup_done.is_set(): return super().start() selected = plan(self.spec) if len(selected.providers) == 1: self._split_plan = None return super().start() self._validate_split_storage(self.spec) if self._cleanup_failure is not None: raise RuntimeError('Provider ownership cleanup is incomplete.') self._slots.acquire(self.stream_id) generation = self.lifecycle.begin() run_id = self.lifecycle.snapshot()['run_id'] self._split_plan = selected self._split_active = True self._split_failure = None self._children, self._snapshots, self._provider_status = {}, {}, {} self._split_stop = threading.Event() self._cleanup_done.clear() self.started_at = time.time() self._monitor_thread = threading.Thread(target=self._run_split, args=(run_id, generation), name='provider-channel-owner', daemon=True) try: self._monitor_thread.start() except Exception: self._slots.release(self.stream_id) self._split_active = False self._cleanup_done.set() self.lifecycle.signal_error(generation, make_stream_failure('PIPELINE_INIT_FAILED', 'provider_owner')) raise return generation def stop(self): with self._teardown_lock: if not self._split_active: return super().stop() self._pending_restart = None self.lifecycle.request_stop() self._split_stop.set() def restart(self, candidate=None): with self._teardown_lock: if not self._split_active: return super().restart(candidate) target = deepcopy(self.spec if candidate is None else candidate) self.check_restart(target) if self._pending_restart is None: self._pending_restart = target self.lifecycle.request_stop() self._split_stop.set() def _fail_split(self, generation, stage): self._split_failure = make_stream_failure('ROLE_RUNTIME_FAILED', stage) self.lifecycle.signal_error(generation, self._split_failure) self._split_stop.set() def _drain(self, run_id, generation): for key, output in self._snapshots.items(): try: payload = output.get_nowait() except queue.Empty: continue snapshot = json.loads(payload) ready_ns = snapshot.pop('_ipc_ready_ns', None) if (str(snapshot.get('cameraId')), snapshot.get('runId'), snapshot.get('runGeneration'), snapshot.get('providerKey')) != (self.spec['source_id'], run_id, generation, key): raise ValueError('Provider snapshot identity mismatch.') if self._split_publisher is not None and not self._split_stop.is_set(): self._split_publisher.channel(self.spec['source_id'], run_id, generation, 'RUNNING') self._split_publisher.submit(snapshot) if ready_ns is not None: self._split_publisher.note_relay(key, time.monotonic_ns()-ready_ns) def _observe(self, run_id, generation): rows = {key: child.status() for key, child in self._children.items()} active, disabled, running = [], [], [] for provider in self._split_plan.providers: row = rows.get(provider.key) if row is None: continue prior = self._provider_status.get(provider.key, {}) if row.get('metadata_path', {}).get('state') == 'NOT_CONNECTED' and prior: row['metadata_path'] = prior.get('metadata_path', {}) failed = bool(row.get('last_error')) or (row['state'] in ('error', 'stopped') and not self._split_stop.is_set()) if failed: disabled.extend(dict(name=name, reason='provider_unavailable') for name in provider.scenarios) elif row['state'] == 'running': running.append(provider.key) active.extend(provider.scenarios) if row.get('cleanup_error') or (row.get('last_cleanup') or {}).get('forced'): self._fail_split(generation, 'provider_cleanup_failed') with self._teardown_lock: self._provider_status = rows self._cached = dict(scenarios=dict(active=active, disabled=disabled), degraded=bool(disabled), required_providers=list(self._split_plan.keys)) if running: self.lifecycle.observe_child(dict(run_id=run_id, run_generation=generation, observed_state='running', roles={})) if self._split_publisher is not None: self._split_publisher.channel(self.spec['source_id'], run_id, generation, 'RUNNING') self._split_publisher.note_provider_sequence(max( (r.get('publisher_metrics', {}).get('lastSeq', -1) for r in rows.values()), default=-1)) if rows and all(r['state'] in ('error', 'stopped') for r in rows.values()): self._fail_split(generation, 'all_providers_unavailable') elif self._split_plan.requested_scenarios and len(disabled) == len(self._split_plan.requested_scenarios): self._fail_split(generation, 'all_scenarios_unavailable') if self.metadata_publisher is not None and self._split_publisher is not None: self.metadata_publisher.update_stats(self.spec['source_id'], self._writer_token, self._publisher_stats()) def _publisher_stats(self): result = self._split_publisher.metrics() if self._split_publisher else {} for row in self._provider_status.values(): for key, value in row.get('publisher_metrics', {}).items(): if key in ('sent', 'sent_frames', 'sent_status', 'replaced', 'expired', 'send_error', 'byte_limit', 'message_limit', 'inactive', 'stale_generation_or_seq', 'shutdown_drop'): result[key] = result.get(key, 0) + value result['workerMode'] = 'provider-direct' return result def _run_split(self, run_id, generation): failed = False try: if self.metadata_publisher is not None: self._writer_token = self.metadata_publisher.claim(self.spec['source_id'], run_id, generation) self._split_publisher = ChildPublisher(self.metadata_publisher.settings, self.spec['source_id'], run_id, generation, self._writer_token, preserve_frames=True) self._split_publisher.start() self._split_publisher.channel(self.spec['source_id'], run_id, generation, 'STARTING') context = multiprocessing.get_context('spawn') limit = min(self.metadata_publisher.settings.max_message_bytes if self.metadata_publisher else 128*1024, CHANNEL_BYTES // (2*len(self._split_plan.providers))) for provider in self._split_plan.providers: if self._split_stop.is_set(): break writer = ((replace(self.metadata_publisher.settings, max_message_bytes=limit), self._writer_token) if self.metadata_publisher is not None else None) child = ProcessStreamGroup(provider_spec(self.spec, provider, self._split_plan.keys), slots=ProviderSlots(self._slots), child_target=(partial(run_channel, provider_writer=writer) if writer else run_channel)) with self._teardown_lock: self._children[provider.key] = child try: child.start(assigned_run_id=run_id, assigned_generation=generation) except Exception: # A reaped local launch failure disables only its dependent scenarios. if not child._cleanup_done.is_set(): raise sampled = 0 while not self._split_stop.is_set(): now = time.monotonic() if now - sampled >= .25: self._observe(run_id, generation) sampled = now self._split_stop.wait(.01) except Exception as error: failed = True print('[PROVIDER-OWNER] ' + type(error).__name__, flush=True) self._fail_split(generation, 'provider_owner') finally: for child in self._children.values(): child.stop() deadline = time.monotonic() + self.CLEANUP_TIMEOUT_S + 10 for child in self._children.values(): child._cleanup_done.wait(max(0, deadline-time.monotonic())) reaped = all(c._cleanup_done.is_set() and c._process is None for c in self._children.values()) failed = failed or self._split_failure is not None cleanup_failure = None if not reaped or any(c._cleanup_failure is not None or (c._last_cleanup or {}).get('forced') for c in self._children.values()): cleanup_failure = make_stream_failure('STOP_CLEANUP_FAILED', 'provider_reap') if self._split_publisher is not None: self._split_publisher.channel(self.spec['source_id'], run_id, generation, 'ERROR' if failed or cleanup_failure else 'STOPPED') self._split_publisher.wait_channel(self.spec['source_id'], run_id, generation, 2) if not self._split_publisher.close(2): cleanup_failure = make_stream_failure('STOP_CLEANUP_FAILED', 'provider_publisher') if reaped and self._writer_token is not None: try: self.metadata_publisher.retire(self.spec['source_id'], self._writer_token, run_id, generation, failed or bool(cleanup_failure)) self._writer_token = None except Exception: cleanup_failure = make_stream_failure('STOP_CLEANUP_FAILED', 'writer_handoff') if reaped: for output in self._snapshots.values(): output.cancel_join_thread() output.close() with self._teardown_lock: self._provider_status = {k: dict(self._provider_status.get(k, {}), child_pid=c.status().get('child_pid'), last_cleanup=deepcopy(c._last_cleanup)) for k, c in self._children.items()} self._cleanup_failure = cleanup_failure self.lifecycle.finish_stop(generation, cleanup_failure) if failed: self.lifecycle.signal_error(generation, self._split_failure or make_stream_failure('ROLE_RUNTIME_FAILED', 'provider_owner')) self._last_cleanup = dict(generation=generation, completed=not failed and not cleanup_failure, child_reaped=reaped, forced=any((c._last_cleanup or {}).get('forced') for c in self._children.values()), writer_released=self._writer_token is None, providers={k: deepcopy(c._last_cleanup) for k,c in self._children.items()}) if reaped: self._slots.release(self.stream_id) self._split_active = False self._cleanup_done.set() self.started_at = None pending, self._pending_restart = self._pending_restart, None if pending is not None and not failed and cleanup_failure is None: self._apply_spec(pending) self.start() def configure_metrics(self, generation, enabled, seconds): if not self._split_active: return super().configure_metrics(generation, enabled, seconds) if generation != self.run_generation: raise RuntimeError('Stale diagnostics generation.') if self._split_publisher is not None: self._split_publisher.configure_metrics(enabled, seconds) return {key: child.configure_metrics(generation, enabled, seconds) for key, child in list(self._children.items()) if child.state == 'running'} def status(self, detail=False): with self._teardown_lock: if self._split_plan is None: return super().status(detail) out = deepcopy(self._cached) out.update(self.lifecycle.snapshot()) out.update(stream_id=self.stream_id, site_id=self.site_id, source_type=self.source_type, cleanup_pending=not self._cleanup_done.is_set(), restart_pending=self._pending_restart is not None, started_at=self.started_at, uptime_s=round(time.time()-self.started_at,1) if self.started_at else None, last_cleanup=deepcopy(self._last_cleanup), execution_mode='provider-process', child_pid=None, provider_processes=deepcopy(self._provider_status), required_providers=list(self._split_plan.keys), publisher_metrics=self._publisher_stats() if self._split_active else {}, metadata_path=dict(state='CONNECTED' if self._split_active else 'NOT_CONNECTED', mode='provider-process', providers={k:v.get('metadata_path',{}) for k,v in self._provider_status.items()})) return out