diff --git a/app/core/config.py b/app/core/config.py index 35b71f6..e11f80d 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -22,7 +22,7 @@ class Settings(BaseSettings): webrtc_udp_port_max: int | None = 60000 webrtc_disconnected_grace_seconds: float = 10.0 ai_privacy_enabled: bool = True - ai_privacy_mode: Literal["real", "bypass", "fixed_delay"] = "real" + ai_privacy_mode: Literal["real", "bypass", "fixed_delay", "echo"] = "real" ai_privacy_fixed_delay_ms: float = 20.0 ai_privacy_device: str | None = None ai_privacy_require_gpu: bool = False diff --git a/app/services/ai/filter.py b/app/services/ai/filter.py index 5bf2da2..42f83dd 100644 --- a/app/services/ai/filter.py +++ b/app/services/ai/filter.py @@ -9,7 +9,6 @@ from app.core.config import Settings, get_settings from app.services.metrics import AI_STAGE_DURATION from privacy_blur import ( - FacePrivacyFilter, PrivacyBlurConfig, PrivacyBlurNotReadyError, initialize_runtime, @@ -18,6 +17,7 @@ ) if TYPE_CHECKING: + from privacy_blur import FacePrivacyFilter as FacePrivacyFilterType from privacy_blur.runtime import SharedGpuRuntime logger = logging.getLogger(__name__) @@ -29,7 +29,10 @@ _config_initialized = False _runtime: SharedGpuRuntime | None = None _state_lock = threading.RLock() -_filters: list[FacePrivacyFilter] = [] +# Importing FacePrivacyFilter resolves privacy_blur.core, which imports Torch. +# Keep bypass and fixed-delay server startup independent of that runtime. +FacePrivacyFilter: Any | None = None +_filters: list[FacePrivacyFilterType] = [] _filters_lock = threading.Lock() _identity_features_by_client: dict[str, Any] = {} @@ -152,7 +155,18 @@ def initialize_privacy_runtime() -> PrivacyBlurConfig | None: return config -def get_privacy_filter(client_id: str | None = None) -> FacePrivacyFilter | None: +def get_privacy_filter(client_id: str | None = None) -> Any | None: + global FacePrivacyFilter + + settings = get_settings() + if settings.ai_privacy_enabled and settings.ai_privacy_mode == "echo": + from app.services.ai.track import EchoPrivacyFilter + + privacy_filter = EchoPrivacyFilter() + with _filters_lock: + _filters.append(privacy_filter) + return privacy_filter + from app.services.ai.identity import ( _reference_face_lock, normalize_reference_client_id, @@ -186,7 +200,13 @@ def get_privacy_filter(client_id: str | None = None) -> FacePrivacyFilter | None } if reference_features is not None: kwargs["reference_features"] = reference_features - privacy_filter = FacePrivacyFilter(**kwargs) + filter_type = FacePrivacyFilter + if filter_type is None: + from privacy_blur import FacePrivacyFilter as imported_filter_type + + filter_type = imported_filter_type + FacePrivacyFilter = filter_type + privacy_filter = filter_type(**kwargs) with _filters_lock: _filters.append(privacy_filter) return privacy_filter @@ -408,12 +428,17 @@ def close_privacy_filter() -> None: ): return _filters.clear() + runtime_started = _runtime is not None for privacy_filter in snapshot: try: privacy_filter.close() except Exception: logger.exception("Failed to close privacy filter instance") - shutdown_runtimes() + # bypass/fixed_delay intentionally never load the model runtime. Calling + # shutdown_runtimes() in that state imports the model stack only to + # discover that there is nothing to close. + if runtime_started: + shutdown_runtimes() _runtime = None _config = None _config_initialized = False diff --git a/app/services/ai/track.py b/app/services/ai/track.py index e6a6c07..ae0d626 100644 --- a/app/services/ai/track.py +++ b/app/services/ai/track.py @@ -3,13 +3,14 @@ import asyncio import logging from time import perf_counter -from typing import TYPE_CHECKING +from typing import Protocol import numpy as np from aiortc.mediastreams import MediaStreamTrack, VideoStreamTrack from av import VideoFrame from app.services.metrics import ( + AI_STAGE_DURATION, FRAME_FAILURES, FRAME_PROCESSING_DURATION, FRAMES_DROPPED, @@ -17,26 +18,64 @@ FRAMES_RECEIVED, ) -if TYPE_CHECKING: - from privacy_blur import FacePrivacyFilter - logger = logging.getLogger(__name__) +class PrivacyFilter(Protocol): + """Minimal per-track filter contract used by real and benchmark echo modes.""" + + async def apply_async(self, frame: VideoFrame) -> VideoFrame: ... + + def close(self) -> None: ... + + +class EchoPrivacyFilter: + """No-op filter that retains the real track's BGR frame conversions. + + It deliberately creates a new ``VideoFrame`` from the BGR ndarray. This + preserves the Python in-process conversion cost measured by ``real`` while + omitting model initialization, detection, identity matching, and blur. + """ + + def __init__(self) -> None: + self._closed = False + + async def apply_async(self, frame: VideoFrame) -> VideoFrame: + if self._closed: + raise RuntimeError("The echo privacy filter is closed") + + decode_started = perf_counter() + image = frame.to_ndarray(format="bgr24") + AI_STAGE_DURATION.labels("decode").observe(perf_counter() - decode_started) + + encode_started = perf_counter() + echoed = VideoFrame.from_ndarray(image, format="bgr24") + AI_STAGE_DURATION.labels("encode").observe(perf_counter() - encode_started) + echoed.pts = frame.pts + if frame.time_base is not None: + echoed.time_base = frame.time_base + return echoed + + def close(self) -> None: + self._closed = True + + class ProtectedVideoTrack(VideoStreamTrack): kind = "video" def __init__( self, source: MediaStreamTrack, - privacy_filter: FacePrivacyFilter | None, + privacy_filter: PrivacyFilter | None, *, session_id: str, + mode: str = "real", ) -> None: super().__init__() self._source = source self._filter = privacy_filter self._session_id = session_id + self._mode = mode self._fallback = privacy_filter is None @property @@ -45,25 +84,24 @@ def is_fallback_active(self) -> bool: async def recv(self): frame, dropped = await _recv_latest_frame_with_drop_count(self._source) - mode = "real" - FRAMES_RECEIVED.labels(mode).inc() + FRAMES_RECEIVED.labels(self._mode).inc() if dropped: - FRAMES_DROPPED.labels(mode).inc(dropped) + FRAMES_DROPPED.labels(self._mode).inc(dropped) started = perf_counter() if self._fallback: result = _blackout_frame(frame) - FRAMES_PROCESSED.labels(mode).inc() - FRAME_PROCESSING_DURATION.labels(mode).observe(perf_counter() - started) + FRAMES_PROCESSED.labels(self._mode).inc() + FRAME_PROCESSING_DURATION.labels(self._mode).observe(perf_counter() - started) return result try: assert self._filter is not None result = await self._filter.apply_async(frame) - FRAMES_PROCESSED.labels(mode).inc() - FRAME_PROCESSING_DURATION.labels(mode).observe(perf_counter() - started) + FRAMES_PROCESSED.labels(self._mode).inc() + FRAME_PROCESSING_DURATION.labels(self._mode).observe(perf_counter() - started) return result except Exception: self._fallback = True - FRAME_FAILURES.labels(mode).inc() + FRAME_FAILURES.labels(self._mode).inc() logger.warning( "AI filter failed, blocking video instead of exposing raw frames: " "session_id=%s", @@ -71,8 +109,8 @@ async def recv(self): exc_info=True, ) result = _blackout_frame(frame) - FRAMES_PROCESSED.labels(mode).inc() - FRAME_PROCESSING_DURATION.labels(mode).observe(perf_counter() - started) + FRAMES_PROCESSED.labels(self._mode).inc() + FRAME_PROCESSING_DURATION.labels(self._mode).observe(perf_counter() - started) return result def stop(self) -> None: diff --git a/app/services/sessions/manager.py b/app/services/sessions/manager.py index bcbf349..59566bf 100644 --- a/app/services/sessions/manager.py +++ b/app/services/sessions/manager.py @@ -920,6 +920,7 @@ def _wrap_with_ai( source=track, privacy_filter=privacy_filter, session_id=session_id, + mode=settings.ai_privacy_mode, ) diff --git a/loadtest/pion-client/main.go b/loadtest/pion-client/main.go index c477c96..2efd01f 100644 --- a/loadtest/pion-client/main.go +++ b/loadtest/pion-client/main.go @@ -118,31 +118,41 @@ func (c *controller) scaleTo(target int, ramp time.Duration) error { if ramp > 0 { delay = ramp / time.Duration(target-current) } + ordinals := make([]int, target-current) + for attempt := range ordinals { + c.mu.Lock() + c.nextOrdinal++ + ordinals[attempt] = c.nextOrdinal + c.mu.Unlock() + } + + if delay == 0 { + // A spike is intended to create all PeerConnections concurrently. Starting + // them serially multiplies the per-client connect timeout and makes high + // session counts a load-generator setup benchmark rather than a server test. + results := make(chan error, len(ordinals)) + for _, ordinal := range ordinals { + go func(ordinal int) { results <- c.startClient(ordinal) }(ordinal) + } + failures := 0 + for range ordinals { + if err := <-results; err != nil { + failures++ + } + } + if failures > 0 { + return fmt.Errorf("%d of %d new clients failed", failures, len(ordinals)) + } + return nil + } + var failures int - for attempt := 0; attempt < target-current; attempt++ { + for attempt, ordinal := range ordinals { if c.ctx.Err() != nil { return c.ctx.Err() } - c.mu.Lock() - c.nextOrdinal++ - ordinal := c.nextOrdinal - c.mu.Unlock() - client, err := startLoadClient(c.ctx, c.cfg, ordinal, c.iceServers, c.http) - if err != nil { + if err := c.startClient(ordinal); err != nil { failures++ - log.Printf("client=%d start failed: %v", ordinal, err) - _, cancel := context.WithCancel(c.ctx) - failed := &loadClient{ordinal: ordinal, startedAt: time.Now(), pcState: "failed", iceState: "failed", cancel: cancel, http: c.http, baseURL: c.cfg.BaseURL} - failed.setError(err) - c.mu.Lock() - c.attempts = append(c.attempts, failed) - c.mu.Unlock() - } else { - c.mu.Lock() - c.clients = append(c.clients, client) - c.attempts = append(c.attempts, client) - c.mu.Unlock() - log.Printf("client=%d session=%s connected", ordinal, client.sessionID) } if delay > 0 && attempt+1 < target-current { if err := waitContext(c.ctx, delay); err != nil { @@ -156,6 +166,26 @@ func (c *controller) scaleTo(target int, ramp time.Duration) error { return nil } +func (c *controller) startClient(ordinal int) error { + client, err := startLoadClient(c.ctx, c.cfg, ordinal, c.iceServers, c.http) + if err != nil { + log.Printf("client=%d start failed: %v", ordinal, err) + _, cancel := context.WithCancel(c.ctx) + failed := &loadClient{ordinal: ordinal, startedAt: time.Now(), pcState: "failed", iceState: "failed", cancel: cancel, http: c.http, baseURL: c.cfg.BaseURL} + failed.setError(err) + c.mu.Lock() + c.attempts = append(c.attempts, failed) + c.mu.Unlock() + return err + } + c.mu.Lock() + c.clients = append(c.clients, client) + c.attempts = append(c.attempts, client) + c.mu.Unlock() + log.Printf("client=%d session=%s connected", ordinal, client.sessionID) + return nil +} + func (c *controller) collectStats() { ticker := time.NewTicker(c.cfg.StatsInterval) defer ticker.Stop() diff --git a/loadtest/scripts/analyze_bypass_runs.py b/loadtest/scripts/analyze_bypass_runs.py index 839859c..69586eb 100644 --- a/loadtest/scripts/analyze_bypass_runs.py +++ b/loadtest/scripts/analyze_bypass_runs.py @@ -17,8 +17,30 @@ RUN_RE = re.compile( - r"^bypass-(?P1|2|4|8)-clients-5m-cycle(?P[123])\.json$" + r"^(?:(?:go|python)-)?(?Pbypass|fixed_delay|real|echo)" + r"(?:-[a-z0-9][a-z0-9_-]*)?-" + r"(?P\d+)-clients-(?P\d+[ms])-cycle(?P\d+)\.json$" ) +STABLE_WINDOW_MIN_COVERAGE = 0.9 + + +def covers_requested_window( + before: datetime | None, + after: datetime | None, + requested_start: datetime, + requested_end: datetime, +) -> bool: + """Return whether target-session samples cover the requested stable window.""" + if before is None or after is None or after <= before: + return False + requested_duration = (requested_end - requested_start).total_seconds() + covered_duration = (after - before).total_seconds() + return ( + requested_duration > 0 + and covered_duration >= requested_duration * STABLE_WINDOW_MIN_COVERAGE + ) + + CONNECTED_RE = re.compile( r"^(?P