wgpu

commit 791d63cd81
Author: royjr <royjr96@gmail.com>
Date:   Sat Jul 25 03:03:28 2026 -0400

    Update host.py

commit f166bfcdc5
Author: royjr <royjr96@gmail.com>
Date:   Sat Jul 25 02:53:13 2026 -0400

    Update process_config.py

commit 418c8154be
Author: royjr <royjr96@gmail.com>
Date:   Sat Jul 25 02:41:08 2026 -0400

    wgpu

openpilot/common/params_keys.h

@@ -134,6 +134,7 @@ inline static std::unordered_map<std::string, ParamKeyAttributes> keys = {
     {"UsbGpuPresent", {CLEAR_ON_MANAGER_START | CLEAR_ON_OFFROAD_TRANSITION, BOOL}},
     {"UsbGpuCompiled", {CLEAR_ON_MANAGER_START | CLEAR_ON_OFFROAD_TRANSITION, BOOL}},
     {"Version", {PERSISTENT, STRING}},
+    {"WgpuEnabled", {CLEAR_ON_MANAGER_START | DEVELOPMENT_ONLY, BOOL}},
 
     // --- sunnypilot params --- //
     {"ApiCache_DriveStats", {PERSISTENT, JSON}},

openpilot/selfdrive/modeld/SConscript

@@ -37,6 +37,9 @@ available = probe_devices()
 if 'CUDA' in available:
   tg_backend = 'CUDA'
   tg_flags = f'DEV={tg_backend}'
+elif 'METAL' in available:
+  tg_backend = 'METAL'
+  tg_flags = f'DEV={tg_backend} FLOAT16=1'
 elif 'QCOM' in available:
   tg_backend = 'QCOM'
   tg_flags = f'DEV={tg_backend} IMAGE=1 FLOAT16=1 NOLOCALS=1 JIT_BATCH_SIZE=0 OPENPILOT_HACKS=1'
@@ -55,6 +58,7 @@ tg_devices = { # which device to put jit inputs to at runtime
 }
 
 USBGPU = usbgpu_present() # or release # TODO always build big model on release
+WGPU = os.getenv('WGPU') == '1'
 if USBGPU:
   usbgpu_tg_flags = f'DEBUG=2 DEV=USB+AMD:LLVM WARP_DEV={tg_backend} FLOAT16=1 JIT_BATCH_SIZE=0 GMMU=0'
   # the USB+AMD GPU takes an exclusive flock; serialize all targets that touch it
@@ -84,13 +88,13 @@ compile_modeld_script = [
 model_w, model_h = MEDMODEL_INPUT_SIZE
 frame_skip = ModelConstants.MODEL_RUN_FREQ // ModelConstants.MODEL_CONTEXT_FREQ
 
-for usbgpu in [False, True] if USBGPU else [False]:
+for usbgpu in [False, True] if USBGPU or WGPU else [False]:
   target_pkl_path = File(modeld_pkl_path(usbgpu)).abspath
   # BIG_INTO_SMALL=1 builds the default target from the big model, e.g. to test it without a USB GPU
-  file_prefix, cmd_flags = ('big_', usbgpu_tg_flags) if usbgpu else ('big_' if os.getenv('BIG_INTO_SMALL') else '', tg_flags)
+  file_prefix, cmd_flags = ('big_', usbgpu_tg_flags if USBGPU else tg_flags) if usbgpu else ('big_' if os.getenv('BIG_INTO_SMALL') else '', tg_flags)
   driving_onnx_deps = get_existing_chunks(File(f"models/{file_prefix}driving_supercombo.onnx").abspath)
   camera_res_args = ' '.join(f'{cw}x{ch}' for cw, ch in CAMERA_CONFIGS)
-  cmd = (f'{cmd_flags} {mac_brew_string} python3 {modeld_dir}/compile_modeld.py '
+  cmd = (f'{cmd_flags} {mac_brew_string} {sys.executable} {modeld_dir}/compile_modeld.py '
         f'--model-size {model_w}x{model_h} '
         f'--camera-resolutions {camera_res_args} '
         f'--onnx {File(f"models/{file_prefix}driving_supercombo.onnx").abspath} '
@@ -104,7 +108,7 @@ for usbgpu in [False, True] if USBGPU else [False]:
     tinygrad_files + compile_modeld_script + driving_onnx_deps + [Value(chunk_targets), chunker_file],
     [cmd, Action(do_chunk, "  [CHUNK] $TARGET")],
   )
-  if usbgpu:
+  if usbgpu and USBGPU:
     lenv.SideEffect(usbgpu_lock, node)
 
 # get model metadata

openpilot/selfdrive/modeld/modeld.py

@@ -26,6 +26,7 @@ from openpilot.common.file_chunker import open_file_chunked, get_manifest_path
 from openpilot.selfdrive.modeld.constants import ModelConstants, Plan
 from openpilot.selfdrive.modeld.helpers import usbgpu_present, modeld_pkl_path, get_tg_input_devices, load_oob
 from openpilot.selfdrive.modeld.usbgpu_link import wait_usbgpu_link
+from openpilot.tools.wgpu.zmq import ZmqPubMaster, ZmqSubMaster
 
 from openpilot.sunnypilot.livedelay.helpers import get_lat_delay
 from openpilot.sunnypilot.modeld_v2.modeld_base import ModelStateBase
@@ -79,12 +80,12 @@ class FrameMeta:
 class ModelState(ModelStateBase):
   prev_desire: np.ndarray  # for tracking the rising edge of the pulse
 
-  def __init__(self, cam_w: int, cam_h: int, usbgpu: bool):
+  def __init__(self, cam_w: int, cam_h: int, usbgpu: bool, big_model: bool = False):
     ModelStateBase.__init__(self)
     self.LAT_SMOOTH_SECONDS = LAT_SMOOTH_SECONDS
     input_devices = get_tg_input_devices(PROCESS_NAME, usbgpu)
     self.WARP_DEV, self.QUEUE_DEV = input_devices['WARP_DEV'], input_devices['QUEUE_DEV']
-    jits = load_oob(open_file_chunked(modeld_pkl_path(usbgpu)))
+    jits = load_oob(open_file_chunked(modeld_pkl_path(usbgpu or big_model)))
     metadata = jits['metadata']
     self.input_shapes = metadata['input_shapes']
     self.vision_input_names = [k for k in self.input_shapes if 'img' in k]
@@ -139,12 +140,14 @@ class ModelState(ModelStateBase):
     return outputs_dict
 
 
-def main(demo=False):
+def main(demo=False, remote_addr: str | None = None, big_model: bool = False):
   cloudlog.warning("modeld init")
 
   _present = usbgpu_present()
   _compiled = os.path.isfile(get_manifest_path(modeld_pkl_path(usbgpu=True)))
   USBGPU = _present and _compiled
+  if big_model and not _compiled:
+    raise FileNotFoundError(f"big model is not compiled: {modeld_pkl_path(usbgpu=True)}")
   params = Params()
   params.put_bool("UsbGpuPresent", _present)
   params.put_bool("UsbGpuCompiled", _compiled)
@@ -178,12 +181,16 @@ def main(demo=False):
     wait_usbgpu_link()
   st = time.monotonic()
   cloudlog.warning("loading model")
-  model = ModelState(vipc_client_main.width, vipc_client_main.height, USBGPU)
+  model = ModelState(vipc_client_main.width, vipc_client_main.height, USBGPU, big_model=big_model)
   cloudlog.warning(f"models loaded in {time.monotonic() - st:.1f}s, modeld starting")
 
   # messaging
-  pm = PubMaster(["modelV2", "drivingModelData", "cameraOdometry", "modelDataV2SP"])
-  sm = SubMaster(["deviceState", "carState", "roadCameraState", "liveCalibration", "driverMonitoringState", "carControl", "liveDelay"])
+  output_services = ["modelV2", "drivingModelData", "cameraOdometry", "modelDataV2SP"]
+  pm = ZmqPubMaster(output_services) if remote_addr is not None else PubMaster(output_services)
+  services = ["deviceState", "carState", "roadCameraState", "liveCalibration", "driverMonitoringState", "carControl", "liveDelay"]
+  if remote_addr is not None:
+    services.append("carParams")
+  sm = ZmqSubMaster(services, remote_addr) if remote_addr is not None else SubMaster(services)
 
   publish_state = PublishState()
   params = Params()
@@ -203,6 +210,11 @@ def main(demo=False):
 
   if demo:
     CP = get_demo_car_params()
+  elif remote_addr is not None:
+    cloudlog.warning("waiting for remote carParams")
+    while not sm.seen["carParams"]:
+      sm.update(100)
+    CP = sm["carParams"]
   else:
     CP = messaging.log_from_bytes(params.get("CarParams", block=True), car.CarParams)
   cloudlog.info("modeld got CarParams: %s", CP.brand)
@@ -332,7 +344,9 @@ if __name__ == "__main__":
     import argparse
     parser = argparse.ArgumentParser()
     parser.add_argument('--demo', action='store_true', help='A boolean for demo mode.')
+    parser.add_argument('--remote', metavar='ADDRESS', help='Run against a remote device over the cereal ZMQ bridge.')
+    parser.add_argument('--big-model', action='store_true', help='Use the locally compiled big driving model.')
     args = parser.parse_args()
-    main(demo=args.demo)
+    main(demo=args.demo, remote_addr=args.remote, big_model=args.big_model)
   except KeyboardInterrupt:
     cloudlog.warning("got SIGINT")

openpilot/system/manager/process_config.py

@@ -95,6 +95,9 @@ def is_stock_model(started, params, CP: car.CarParams) -> bool:
   """Check if the active model runner is stock."""
   return bool(get_active_model_runner(params, not started) == custom.ModelManagerSP.Runner.stock)
 
+def not_wgpu(started: bool, params: Params, CP: car.CarParams) -> bool:
+  return not params.get_bool("WgpuEnabled")
+
 def mapd_ready(started: bool, params: Params, CP: car.CarParams) -> bool:
   return bool(os.path.exists(Paths.mapd_root()))
 
@@ -128,7 +131,7 @@ procs = [
   PythonProcess("micd", "openpilot.system.micd", iscar),
   PythonProcess("timed", "openpilot.system.timed", always_run, enabled=not PC),
 
-  PythonProcess("modeld", "openpilot.selfdrive.modeld.modeld", and_(only_onroad, is_stock_model)),
+  PythonProcess("modeld", "openpilot.selfdrive.modeld.modeld", and_(and_(only_onroad, is_stock_model), not_wgpu)),
   PythonProcess("dmonitoringmodeld", "openpilot.selfdrive.modeld.dmonitoringmodeld", driverview, enabled=(WEBCAM or not PC)),
 
   PythonProcess("sensord", "openpilot.system.sensord.sensord", only_onroad, enabled=not PC),

openpilot/tools/camerastream/compressed_vipc.py

@@ -1,5 +1,4 @@
 #!/usr/bin/env python3
-import os
 import argparse
 import multiprocessing
 import time
@@ -10,6 +9,7 @@ from collections import deque
 import openpilot.cereal.messaging as messaging
 from msgq.visionipc import VisionIpcServer, VisionStreamType
 from openpilot.tools.camerastream.ffmpeg_decoder import Decoder, FFmpegError
+from openpilot.tools.wgpu.zmq import ZmqSubMaster, ZmqSubSocket
 
 V4L2_BUF_FLAG_KEYFRAME = 8
 
@@ -30,10 +30,7 @@ def decoder(addr, vipc_server, vst, W, H, debug=False):
 
   codec = Decoder("hevc")
 
-  os.environ["ZMQ"] = "1"
-  messaging.reset_context()
-  sock = messaging.sub_sock(sock_name, None, addr=addr, conflate=False)
-  cnt = 0
+  sock = ZmqSubSocket(sock_name, addr)
   last_idx = -1
   seen_iframe = False
 
@@ -46,8 +43,9 @@ def decoder(addr, vipc_server, vst, W, H, debug=False):
     time_q.clear()
 
   while 1:
-    msgs = messaging.drain_sock(sock, wait_for_one=True)
-    for evt in msgs:
+    msgs = sock.drain(wait_for_one=True)
+    for raw in msgs:
+      evt = messaging.log_from_bytes(raw)
       evta = getattr(evt, evt.which())
       if last_idx != -1 and evta.idx.encodeId != (last_idx + 1):
         if debug:
@@ -94,8 +92,9 @@ def decoder(addr, vipc_server, vst, W, H, debug=False):
         continue
 
       frame_start_time = time_q.popleft()
-      vipc_server.send(vst, img_yuv.data, cnt, int(frame_start_time*1e9), int(time.monotonic()*1e9))
-      cnt += 1
+      # Preserve the device camera metadata so remote model outputs line up with
+      # the rest of the device's cereal timeline.
+      vipc_server.send(vst, img_yuv.data, evta.idx.frameId, evta.idx.timestampSof, evta.idx.timestampEof)
 
       pc_latency = (time.monotonic()-frame_start_time)*1000
       if debug:
@@ -105,14 +104,10 @@ def decoder(addr, vipc_server, vst, W, H, debug=False):
 
 class CompressedVipc:
   def __init__(self, addr, vision_streams, server_name, debug=False):
-    print("getting frame sizes")
-    os.environ["ZMQ"] = "1"
-    messaging.reset_context()
-    sm = messaging.SubMaster([ENCODE_SOCKETS[s] for s in vision_streams], addr=addr)
+    print("waiting for remote camera stream metadata", flush=True)
+    sm = ZmqSubMaster([ENCODE_SOCKETS[s] for s in vision_streams], addr)
     while min(sm.recv_frame.values()) == 0:
       sm.update(100)
-    os.environ.pop("ZMQ")
-    messaging.reset_context()
 
     self.vipc_server = VisionIpcServer(server_name)
     for vst in vision_streams:
@@ -121,9 +116,10 @@ class CompressedVipc:
     self.vipc_server.start_listener()
 
     self.procs = []
+    process_context = multiprocessing.get_context("fork")
     for vst in vision_streams:
       ed = sm[ENCODE_SOCKETS[vst]]
-      p = multiprocessing.Process(target=decoder, args=(addr, self.vipc_server, vst, ed.width, ed.height, debug))
+      p = process_context.Process(target=decoder, args=(addr, self.vipc_server, vst, ed.width, ed.height, debug))
       p.start()
       self.procs.append(p)
 

openpilot/tools/camerastream/ffmpeg_decoder.py

@@ -46,10 +46,23 @@ def _bind(fn, restype, *argtypes):
   return fn
 
 
+def _library_path(name: str, major: int) -> str:
+  candidates = (
+    f"lib{name}.so.{major}",
+    f"lib{name}.{major}.dylib",
+    f"lib{name}.dylib",
+  )
+  for candidate in candidates:
+    path = os.path.join(ffmpeg.LIB_DIR, candidate)
+    if os.path.isfile(path):
+      return path
+  raise FileNotFoundError(f"FFmpeg library not found in {ffmpeg.LIB_DIR}: {', '.join(candidates)}")
+
+
 def _load_libraries():
-  avutil = ctypes.CDLL(os.path.join(ffmpeg.LIB_DIR, "libavutil.so.59"), mode=ctypes.RTLD_GLOBAL)
-  avcodec = ctypes.CDLL(os.path.join(ffmpeg.LIB_DIR, "libavcodec.so.61"), mode=ctypes.RTLD_GLOBAL)
-  swscale = ctypes.CDLL(os.path.join(ffmpeg.LIB_DIR, "libswscale.so.8"), mode=ctypes.RTLD_GLOBAL)
+  avutil = ctypes.CDLL(_library_path("avutil", 59), mode=ctypes.RTLD_GLOBAL)
+  avcodec = ctypes.CDLL(_library_path("avcodec", 61), mode=ctypes.RTLD_GLOBAL)
+  swscale = ctypes.CDLL(_library_path("swscale", 8), mode=ctypes.RTLD_GLOBAL)
 
   c_int, c_char_p, c_void_p, c_size_t = ctypes.c_int, ctypes.c_char_p, ctypes.c_void_p, ctypes.c_size_t
   c_uint8_p = ctypes.POINTER(ctypes.c_uint8)

openpilot/tools/wgpu/README.md

@@ -0,0 +1,59 @@
+# Wireless modeld proof of concept
+
+This runs driving `modeld` on a laptop and returns its cereal outputs to a comma
+device over the existing Wi-Fi network. It reuses the existing HEVC camera
+stream, VisionIPC decoder, and cereal ZMQ bridge.
+
+This is for offroad/bench testing only. Wi-Fi has no deterministic latency or
+availability guarantee. The device restores local `modeld` when the device-side
+helper exits, but that is not a seamless onroad failover.
+
+## Build
+
+Use the same commit on the laptop and comma device. Build the cereal bridge on
+the device:
+
+```sh
+scons -u openpilot/cereal/messaging/bridge
+```
+
+Build and test the normal model on the laptop first:
+
+```sh
+PATH="$PWD/.venv/bin:$PATH" scons -u
+```
+
+To compile the big external-GPU model for the laptop's local tinygrad backend:
+
+```sh
+PATH="$PWD/.venv/bin:$PATH" WGPU=1 scons -u \
+  openpilot/selfdrive/modeld/models/big_driving_tinygrad.pkl.chunkmanifest
+```
+
+On macOS, the build selects tinygrad's Metal backend when it is available.
+On an 8-GPU-core M5 MacBook Air, the small model's compiled policy pass measured
+about 6–11 ms, while the big model measured about 79–81 ms. The latter already
+misses the 50 ms model cadence before network and codec latency, so start with
+the small model on that class of laptop.
+
+## Run
+
+Find the laptop's LAN IP address that the comma device can reach. While the
+device is offroad, run:
+
+```sh
+cd /data/openpilot
+python3 -m openpilot.tools.wgpu.device LAPTOP_IP
+```
+
+Keep that terminal open. On the laptop, run:
+
+```sh
+cd /path/to/openpilot
+python3 -m openpilot.tools.wgpu.host COMMA_IP
+```
+
+Add `--big-model` after `COMMA_IP` to use the locally compiled big model.
+
+The first remote `carParams` packet can take up to 50 seconds. Stop either side
+with Ctrl+C. Stop the device helper before changing branches or rebooting.

openpilot/tools/wgpu/device.py

@@ -0,0 +1,67 @@
+#!/usr/bin/env python3
+import argparse
+import signal
+import subprocess
+import time
+from pathlib import Path
+
+from openpilot.common.params import Params
+
+
+MODEL_OUTPUTS = "modelV2,drivingModelData,cameraOdometry,modelDataV2SP"
+ROOT = Path(__file__).resolve().parents[3]
+BRIDGE = ROOT / "openpilot/cereal/messaging/bridge"
+
+
+def stop_process(proc: subprocess.Popen) -> None:
+  proc.terminate()
+  try:
+    proc.wait(timeout=5)
+  except subprocess.TimeoutExpired:
+    proc.kill()
+    proc.wait()
+
+
+def handle_sigterm(*_) -> None:
+  raise KeyboardInterrupt
+
+
+def main() -> None:
+  parser = argparse.ArgumentParser(description="Route modeld traffic between this device and a wireless host.")
+  parser.add_argument("host", help="Laptop IP address reachable from this device")
+  args = parser.parse_args()
+
+  if not BRIDGE.is_file():
+    raise FileNotFoundError(f"build the cereal bridge first: {BRIDGE}")
+
+  params = Params()
+  if not params.get_bool("IsOffroad"):
+    raise RuntimeError("start the wgpu bridge while offroad")
+
+  procs: list[subprocess.Popen] = []
+  try:
+    params.put_bool("WgpuEnabled", True, block=True)
+    procs = [
+      subprocess.Popen([str(BRIDGE)]),
+      subprocess.Popen([str(BRIDGE), args.host, MODEL_OUTPUTS]),
+    ]
+    print(f"wgpu enabled; forwarding camera/state to {args.host}")
+    print("keep this process running; Ctrl+C restores local modeld")
+    while all(proc.poll() is None for proc in procs):
+      time.sleep(0.25)
+    failed = next(proc for proc in procs if proc.poll() is not None)
+    raise RuntimeError(f"bridge exited with status {failed.returncode}")
+  finally:
+    params.put_bool("WgpuEnabled", False, block=True)
+    for proc in procs:
+      if proc.poll() is None:
+        stop_process(proc)
+    print("wgpu disabled; local modeld restored")
+
+
+if __name__ == "__main__":
+  signal.signal(signal.SIGTERM, handle_sigterm)
+  try:
+    main()
+  except KeyboardInterrupt:
+    pass

openpilot/tools/wgpu/host.py

@@ -0,0 +1,57 @@
+#!/usr/bin/env python3
+import argparse
+import os
+import signal
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[3]
+CAMERASTREAM = ROOT / "openpilot/tools/camerastream/compressed_vipc.py"
+
+
+def stop_process(proc: subprocess.Popen) -> None:
+  try:
+    os.killpg(proc.pid, signal.SIGTERM)
+  except ProcessLookupError:
+    proc.wait()
+    return
+  try:
+    proc.wait(timeout=5)
+  except subprocess.TimeoutExpired:
+    os.killpg(proc.pid, signal.SIGKILL)
+    proc.wait()
+
+
+def main() -> None:
+  parser = argparse.ArgumentParser(description="Run modeld on this host for a remote comma device.")
+  parser.add_argument("device", help="comma device hostname or IP address")
+  parser.add_argument("--big-model", action="store_true", help="use the locally compiled big driving model")
+  args = parser.parse_args()
+
+  camera = subprocess.Popen([sys.executable, str(CAMERASTREAM), args.device, "--cams", "0,2"], start_new_session=True)
+  model_args = [sys.executable, "-m", "openpilot.selfdrive.modeld.modeld", "--remote", args.device]
+  if args.big_model:
+    model_args.append("--big-model")
+  model = subprocess.Popen(model_args, cwd=ROOT, start_new_session=True)
+
+  procs = [camera, model]
+  try:
+    while all(proc.poll() is None for proc in procs):
+      time.sleep(0.25)
+    failed = next(proc for proc in procs if proc.poll() is not None)
+    raise RuntimeError(f"wgpu host process exited with status {failed.returncode}")
+  finally:
+    for proc in procs:
+      if proc.poll() is None:
+        stop_process(proc)
+    print("wgpu host stopped")
+
+
+if __name__ == "__main__":
+  try:
+    main()
+  except KeyboardInterrupt:
+    pass

openpilot/tools/wgpu/zmq.py

@@ -0,0 +1,90 @@
+import time
+
+import zmq
+
+import openpilot.cereal.messaging as messaging
+
+
+def service_port(endpoint: str) -> int:
+  # Keep this in sync with cereal/messaging/bridge_zmq.cc.
+  value = 0xcbf29ce484222325
+  for char in endpoint.encode():
+    value ^= char
+    value = (value * 0x100000001b3) & 0xffffffffffffffff
+  return 8023 + (value % (65535 - 8023))
+
+
+class ZmqSubSocket:
+  def __init__(self, endpoint: str, address: str, conflate: bool = False):
+    self.context = zmq.Context()
+    self.socket = self.context.socket(zmq.SUB)
+    self.socket.setsockopt(zmq.SUBSCRIBE, b"")
+    self.socket.setsockopt(zmq.RECONNECT_IVL_MAX, 500)
+    if conflate:
+      self.socket.setsockopt(zmq.CONFLATE, 1)
+    self.socket.connect(f"tcp://{address}:{service_port(endpoint)}")
+
+  def receive(self, non_blocking: bool = False) -> bytes | None:
+    try:
+      return self.socket.recv(flags=zmq.NOBLOCK if non_blocking else 0)
+    except zmq.Again:
+      return None
+
+  def drain(self, wait_for_one: bool = False) -> list[bytes]:
+    messages = []
+    if wait_for_one:
+      message = self.receive()
+      if message is not None:
+        messages.append(message)
+    while (message := self.receive(non_blocking=True)) is not None:
+      messages.append(message)
+    return messages
+
+
+class ZmqSubMaster:
+  def __init__(self, services: list[str], address: str):
+    self.services = services
+    self.sockets = {service: ZmqSubSocket(service, address, conflate=True) for service in services}
+    self.poller = zmq.Poller()
+    self.socket_to_service = {}
+    for service, sub in self.sockets.items():
+      self.poller.register(sub.socket, zmq.POLLIN)
+      self.socket_to_service[sub.socket] = service
+
+    self.data = {service: getattr(messaging.new_message(service).as_reader(), service) for service in services}
+    self.seen = dict.fromkeys(services, False)
+    self.updated = dict.fromkeys(services, False)
+    self.recv_frame = dict.fromkeys(services, 0)
+    self.frame = -1
+
+  def __getitem__(self, service: str):
+    return self.data[service]
+
+  def update(self, timeout: int = 100) -> None:
+    self.frame += 1
+    self.updated = dict.fromkeys(self.services, False)
+    for socket, _ in self.poller.poll(timeout):
+      service = self.socket_to_service[socket]
+      raw = self.sockets[service].receive(non_blocking=True)
+      if raw is None:
+        continue
+      event = messaging.log_from_bytes(raw)
+      self.data[service] = getattr(event, service)
+      self.seen[service] = True
+      self.updated[service] = True
+      self.recv_frame[service] = self.frame
+
+
+class ZmqPubMaster:
+  def __init__(self, services: list[str]):
+    context = zmq.Context.instance()
+    self.sockets = {}
+    for service in services:
+      socket = context.socket(zmq.PUB)
+      socket.bind(f"tcp://*:{service_port(service)}")
+      self.sockets[service] = socket
+    # Give already-running bridge subscribers time to finish their handshake.
+    time.sleep(0.1)
+
+  def send(self, service: str, message) -> None:
+    self.sockets[service].send(message.to_bytes(), flags=zmq.NOBLOCK)