merge: load test

app/core/config.py

@@ -18,7 +18,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

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

app/services/ai/track.py

@@ -3,40 +3,79 @@
 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,
     FRAMES_PROCESSED,
     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,34 +84,33 @@ 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",
                 self._session_id,
                 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:

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,
     )
 
 

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()