webrtc: speed ups (#38163)

- patch `aioice` to only get host address of active network interface (wlan0 when on wifi, ppp0 when not)
- use common webrtc session between `startStream` and `addIceCandidates`
- first time connection speed ups
  - move imports to top of file to move delay to webrtcd server start up not first connection
  - warm up webrtc stack by creating a webrtc connection object to import all submodule classes

system/athena/athenad.py

@@ -40,6 +40,7 @@ from openpilot.system.loggerd.xattr_cache import getxattr, setxattr
 from openpilot.common.swaglog import cloudlog
 from openpilot.system.version import get_build_metadata
 from openpilot.system.hardware.hw import Paths
+from openpilot.system.webrtc.webrtcd import StreamRequestBody
 
 
 ATHENA_HOST = os.getenv('ATHENA_HOST', 'wss://athena.comma.ai')
@@ -84,6 +85,9 @@ UPLOAD_SESS = requests.Session()
 UPLOAD_SESS.mount("http://", UploadTOSAdapter())
 UPLOAD_SESS.mount("https://", UploadTOSAdapter())
 
+WEBRTCD_SESS = requests.Session()
+WEBRTCD_SESS.mount("http://", HTTPAdapter(max_retries=0))
+
 
 @dataclass
 class UploadFile:
@@ -578,7 +582,6 @@ def getNetworks():
 
 @dispatcher.add_method
 def startStream(sdp: str) -> dict:
-  from openpilot.system.webrtc.webrtcd import StreamRequestBody
   bridge_services_in = []
 
   # get live car params to avoid stale notCar edge case
@@ -590,7 +593,7 @@ def startStream(sdp: str) -> dict:
 
   body = StreamRequestBody(sdp, "wideRoad", bridge_services_in, ["carState"])
   try:
-    resp = requests.post(f"http://localhost:{WEBRTCD_PORT}/stream",
+    resp = WEBRTCD_SESS.post(f"http://localhost:{WEBRTCD_PORT}/stream",
                        json=asdict(body), timeout=10)
     if not resp.ok:
       try:
@@ -600,7 +603,7 @@ def startStream(sdp: str) -> dict:
         resp.raise_for_status()
     return resp.json()
   except requests.ConnectTimeout as e:
-    raise Exception("webrtc took too long to respond. is it on?") from e
+    raise Exception("webrtc took too long to respond. is the comma body on?") from e
   except requests.ConnectionError as e:
     raise Exception("webrtc is not running. turn on comma body ignition.") from e
 
@@ -608,8 +611,8 @@ def startStream(sdp: str) -> dict:
 def addIceCandidate(session_id: str, candidate: dict | None) -> dict:
   if session_id is None:
     return Exception("cannot add ice candidate without session_id")
-  resp = requests.post(f"http://localhost:{WEBRTCD_PORT}/candidate",
-                       json={"session_id": session_id, "candidate": candidate}, timeout=10)
+  resp = WEBRTCD_SESS.post(f"http://localhost:{WEBRTCD_PORT}/candidate",
+                            json={"session_id": session_id, "candidate": candidate}, timeout=10)
   return resp.json()
 
 @dispatcher.add_method

system/webrtc/device/video.py

@@ -4,6 +4,7 @@ import time
 
 import av
 from teleoprtc.tracks import TiciVideoStreamTrack
+from aiortc import MediaStreamError
 
 from cereal import messaging
 from openpilot.common.realtime import DT_MDL, DT_DMON
@@ -32,6 +33,10 @@ class LiveStreamVideoStreamTrack(TiciVideoStreamTrack):
     self._t0_ns = time.monotonic_ns()
     self.timing_sei_enabled = False
 
+  def stop(self) -> None:
+    super().stop()
+    self._sock = None
+
   def _make_sock(self, camera_type: str) -> messaging.SubSocket:
     return messaging.sub_sock(self.camera_to_sock_mapping[camera_type], conflate=True)
 
@@ -54,6 +59,8 @@ class LiveStreamVideoStreamTrack(TiciVideoStreamTrack):
 
   async def recv(self):
     while True:
+      if self.readyState != "live":
+        raise MediaStreamError
       msg = messaging.recv_one_or_none(self._sock)
       if msg is not None:
         break

system/webrtc/webrtcd.py

@@ -2,6 +2,7 @@
 
 from abc import abstractmethod
 import os
+import socket
 import time
 import argparse
 import asyncio
@@ -26,6 +27,36 @@ from openpilot.system.webrtc.schema import generate_field
 from openpilot.common.params import Params
 from cereal import messaging, log
 
+from aiortc import RTCBundlePolicy, RTCConfiguration, RTCPeerConnection
+from aiortc.mediastreams import VideoStreamTrack
+from openpilot.system.webrtc.device.video import LiveStreamVideoStreamTrack
+from teleoprtc import WebRTCAnswerBuilder
+
+import aioice.ice
+
+
+# socket trick: route lookup for 8.8.8.8 (nothing is sent or actually connected to)
+# return the source interfaces IP which is the default interface of the device
+def _default_route_ip() -> str | None:
+  s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+  try:
+    s.connect(("8.8.8.8", 53))  # selects a route, sends nothing
+    return s.getsockname()[0]
+  except OSError:
+    return None
+  finally:
+    s.close()
+
+# aioice patch: gather ICE candidates only on the default-route interface
+_get_host_addresses = aioice.ice.get_host_addresses
+def _primary_host_addresses(use_ipv4: bool, use_ipv6: bool) -> list[str]:
+  addresses = _get_host_addresses(use_ipv4, use_ipv6)
+  primary = _default_route_ip()
+  if primary not in addresses:
+    return addresses
+  return [a for a in addresses if a == primary]
+aioice.ice.get_host_addresses = _primary_host_addresses
+
 
 class AsyncTaskRunner:
   def __init__(self):
@@ -206,10 +237,6 @@ class StreamSession:
   shared_pub_master = DynamicPubMaster([])
 
   def __init__(self, sdp: str, init_camera: str, incoming_services: list[str], outgoing_services: list[str], debug_mode: bool = False):
-    from aiortc.mediastreams import VideoStreamTrack
-    from openpilot.system.webrtc.device.video import LiveStreamVideoStreamTrack
-    from teleoprtc import WebRTCAnswerBuilder
-
     builder = WebRTCAnswerBuilder(sdp)
 
     self.video_track = LiveStreamVideoStreamTrack(init_camera) if not debug_mode else VideoStreamTrack()
@@ -327,6 +354,8 @@ class StreamSession:
         await self.bitrate_controller.stop()
       if self.outgoing_bridge is not None:
         await self.outgoing_bridge.stop()
+      if self.video_track is not None:
+        self.video_track.stop()
       await self.stream.stop()
 
 
@@ -353,7 +382,7 @@ async def get_stream(request: 'web.Request'):
         except Exception:
           pass
       await s.stop()
-      del stream_dict[sid]
+      stream_dict.pop(sid, None)
 
     session = StreamSession(body.sdp, body.initCamera, body.bridge_services_in, body.bridge_services_out, debug_mode)
     try:
@@ -367,9 +396,14 @@ async def get_stream(request: 'web.Request'):
     except Exception:
       await session.stop()
       raise
+    stream_dict[session.identifier] = session
     session.start()
 
-    stream_dict[session.identifier] = session
+    session_id = session.identifier
+
+    def remove_finished_session(_: asyncio.Task) -> None:
+      stream_dict.pop(session_id, None)
+    session.run_task.add_done_callback(remove_finished_session)
 
   return web.json_response({"sdp": answer.sdp, "type": answer.type, "session_id": session.identifier})
 
@@ -409,6 +443,26 @@ async def post_notify(request: 'web.Request'):
   return web.Response(status=200, text="OK")
 
 
+async def on_startup(app: 'web.Application'):
+  logger = logging.getLogger("webrtcd")
+  start_time = time.monotonic()
+  pc = None
+
+  # warmup imports for webrtc stack
+  try:
+    pc = RTCPeerConnection(RTCConfiguration(bundlePolicy=RTCBundlePolicy.MAX_BUNDLE))
+    pc.addTransceiver("video", direction="recvonly")
+    pc.createDataChannel("data", ordered=True)
+    offer = await pc.createOffer()
+    await asyncio.wait_for(pc.setLocalDescription(offer), timeout=1.5)
+    logger.info("Warmed WebRTC stack in %.1f ms", (time.monotonic() - start_time) * 1000)
+  except Exception:
+    logger.exception("WebRTC stack warmup failed")
+  finally:
+    if pc is not None:
+      await pc.close()
+
+
 async def on_shutdown(app: 'web.Application'):
   for session in app['streams'].values():
     await session.stop()
@@ -426,6 +480,7 @@ def webrtcd_thread(host: str, port: int, debug: bool):
   app['streams'] = dict()
   app['stream_lock'] = asyncio.Lock()
   app['debug'] = debug
+  app.on_startup.append(on_startup)
   app.on_shutdown.append(on_shutdown)
   app.router.add_post("/stream", get_stream)
   app.router.add_post("/candidate", post_candidate)