"""Opt-in numeric diagnostics shared by legacy and PR13 comparisons.""" from collections import Counter, OrderedDict from functools import wraps from contextlib import contextmanager import os import json import threading import time class PipelineDiagnostics: INPUT_TIMELINE_LIMIT = 8 INPUT_TIMELINE_BYTES = 8 * 1024 BOUNDS_MS = (1, 2, 5, 10, 20, 50, 75, 100, 150, 200, 300, 500, 1000, 2000, 5000) PAIRS = {'preprocess': 'selected', 'callback': 'preprocess', 'copied': 'callback', 'submitted': 'selected', 'preprocess_start': 'selected', 'infer_output': 'preprocess', 'preprocess_output': 'preprocess_start'} def __init__(self, enabled=False): limit = os.environ.get('CONTEXTRA_INPUT_TIMELINE_LIMIT', '8') if limit not in tuple(str(n) for n in range(1, self.INPUT_TIMELINE_LIMIT + 1)): raise ValueError('Invalid private input timeline limit.') self.input_timeline_limit = int(limit) self.enabled = enabled self.lock = threading.Lock() self.frames = OrderedDict() self.counts = Counter() self.durations = {} self.missed = 0 self.providers = {} self.until = None self.started_ns = time.monotonic_ns() self.last_accepted_ns = None self.queue_levels = {} def collecting(self): return self.enabled and (self.until is None or time.monotonic() < self.until) def configure(self, enabled, seconds=60): if type(enabled) is not bool or type(seconds) is not int or not 1 <= seconds <= 60: raise ValueError('Invalid pipeline diagnostic window.') with self.lock: self.enabled = enabled self.until = time.monotonic()+seconds if enabled else 0 self.started_ns = time.monotonic_ns() self.last_accepted_ns = None self.frames.clear() self.counts.clear() self.durations.clear() self.missed = 0 self.queue_levels.clear() for provider in self.providers.values(): provider.configure(enabled, seconds) def for_provider(self, key): if key not in ('medium', 'pose', 'fire', 'crowd'): raise ValueError('Unknown diagnostic provider.') with self.lock: if key not in self.providers: self.providers[key] = PipelineDiagnostics(self.enabled) self.providers[key].until = self.until return self.providers[key] @contextmanager def phase(self, name): if not self.collecting(): yield return wall, cpu = time.monotonic_ns(), time.thread_time_ns() try: yield finally: elapsed, used = time.monotonic_ns()-wall, time.thread_time_ns()-cpu with self.lock: if self.collecting(): self._duration(name+'_wall', elapsed) self._duration(name+'_cpu', used) @contextmanager def copy_scope(self): previous = getattr(_copy_context, 'value', None) _copy_context.value = self try: yield finally: _copy_context.value = previous def _duration(self, name, ns): if ns < 0: return row = self.durations.setdefault(name, dict(count=0, sum_ns=0, max_ns=0, over_100ms=0, histogram=[0]*(len(self.BOUNDS_MS)+1))) row['count'] += 1 row['sum_ns'] += ns row['max_ns'] = max(row['max_ns'], ns) row['over_100ms'] += int(ns > 100_000_000) index = next((i for i, ms in enumerate(self.BOUNDS_MS) if ns <= ms*1_000_000), len(self.BOUNDS_MS)) row['histogram'][index] += 1 def mark(self, stage, key=None, at=None): if not self.collecting(): return if not self.lock.acquire(blocking=False): self.missed += 1 return try: self.counts[stage] += 1 now = time.monotonic_ns() if at is None else at if stage == 'accepted': self._duration('accepted_interval', now-(self.last_accepted_ns or self.started_ns)) self.last_accepted_ns = now if stage == 'selected' and key is not None: self.frames[key] = {'selected': now} while len(self.frames) > 256: self.frames.popitem(last=False) row = self.frames.get(key) if row is not None: prior = self.PAIRS.get(stage) if prior in row: self._duration(prior + '_to_' + stage, now-row[prior]) extra = {'preprocess': 'preprocess_start', 'callback': 'infer_output'}.get(stage) if extra in row: self._duration(extra + '_to_' + stage, now-row[extra]) if stage == 'preprocess' and 'preprocess_output' in row: self._duration('preprocess_output_to_infer_entry', now-row['preprocess_output']) row[stage] = now finally: self.lock.release() def copy_duration(self, wall, cpu): if not self.collecting(): return with self.lock: self._duration('copy_wall', wall) self._duration('copy_thread_cpu', cpu) def input_timing(self, key, row, ingress_ns): if not self.collecting(): return with self.lock: target = self.frames.get(key) if target is None: return if row is None: self.counts['input_timing_unavailable'] += 1 return target['input'] = dict(row, ingress_ns=ingress_ns, ingress_wall_ns=time.time_ns()) self._duration('depay_to_source_exit', row['source_exit_ns']-row['depay_first_ns']) self._duration('source_exit_to_ingress', ingress_ns-row['source_exit_ns']) for name, start, end in ( ('source_encoded_queue', 'encoded_in_ns', 'encoded_out_ns'), ('decode_between_queues', 'encoded_out_ns', 'raw_in_ns'), ('source_raw_queue', 'raw_in_ns', 'raw_out_ns'), ('raw_queue_to_source_exit', 'raw_out_ns', 'source_exit_ns')): if row.get(start, 0) and row.get(end, 0): self._duration(name, row[end]-row[start]) def queue_sample(self, values): if not self.collecting(): return with self.lock: for key, value in values.items(): row = self.queue_levels.setdefault(key, dict(count=0, total=0, maximum=0, last=0)) row.update(count=row['count']+1, total=row['total']+value, maximum=max(row['maximum'], value), last=value) def snapshot(self): if not self.enabled: return {'enabled': False} with self.lock: # Detailed samples share the child's 64 KiB IPC status envelope. timeline = [dict(frame=k[0], pts=k[1], **v) for k, v in list(self.frames.items())[-self.input_timeline_limit:] if 'input' in v] while timeline and len(json.dumps(timeline, allow_nan=False).encode()) > self.INPUT_TIMELINE_BYTES: timeline.pop(0) return dict(enabled=True, collecting=self.collecting(), counts=dict(self.counts), missed=self.missed, started_ns=self.started_ns, last_accepted_ns=self.last_accepted_ns, acceptance_silence_ns=time.monotonic_ns()-(self.last_accepted_ns or self.started_ns), histogram_bounds_ms=list(self.BOUNDS_MS), retained=len(self.frames), queue_levels={k: dict(v) for k, v in self.queue_levels.items()}, input_timeline=timeline, input_timeline_limit=self.input_timeline_limit, durations={k: dict(v, histogram=list(v['histogram'])) for k, v in self.durations.items()}, providers={k: v.snapshot() for k, v in self.providers.items()}) diagnostics = PipelineDiagnostics(os.environ.get('CONTEXTRA_PIPELINE_DIAGNOSTICS') == 'true') _copy_context = threading.local() def timed_copy(function): @wraps(function) def wrapped(*args, **kwargs): target = getattr(_copy_context, 'value', None) or diagnostics if not target.collecting(): return function(*args, **kwargs) wall, cpu = time.monotonic_ns(), time.thread_time_ns() try: result = function(*args, **kwargs) finally: end = time.monotonic_ns() target.copy_duration(end-wall, time.thread_time_ns()-cpu) target.mark('copied', (result.frame_number, result.pts), end) return result return wrapped