Sync: commaai/msgq:master → sunnypilot/msgq:master

.gitattributes

@@ -0,0 +1 @@
+* text=auto eol=lf

.github/workflows/tests.yml

@@ -6,19 +6,16 @@ on:
       - master
   pull_request:
 
-concurrency:
-  group: ${{ github.workflow }}-${{ github.ref != 'refs/heads/master' && github.ref || github.run_id }}-${{ github.event_name }}
-  cancel-in-progress: true
-
 jobs:
   test:
     name: ./test.sh
     runs-on: ${{ matrix.os }}
     strategy:
       fail-fast: false
       matrix:
-        os: ['macos-latest', 'ubuntu-latest']
+        os: ['macos-latest', 'ubuntu-latest', 'ubuntu-24.04-arm']
     timeout-minutes: 10
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v7
       - run: ./test.sh
+        timeout-minutes: 1

.github/workflows/wheels.yml

@@ -0,0 +1,48 @@
+name: wheels
+
+on:
+  pull_request:
+  release:
+    types: [published]
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  build:
+    runs-on: ${{ matrix.os }}
+    strategy:
+      fail-fast: false
+      matrix:
+        os: [ubuntu-latest, ubuntu-24.04-arm, macos-15-intel, macos-latest]
+    steps:
+      - uses: actions/checkout@v7
+      - uses: pypa/cibuildwheel@e090b81e30c4d855ea63bf4b6e59204c09a101ae # v4.2.1
+      - uses: actions/setup-python@v7
+        with:
+          python-version: '3.11'
+      - name: Build source distribution
+        run: |
+          python -m pip install build
+          python -m build --sdist --outdir wheelhouse
+      - uses: actions/upload-artifact@v7
+        with:
+          name: dist-${{ matrix.os }}
+          path: wheelhouse/*
+          if-no-files-found: error
+
+  publish:
+    if: github.event_name == 'release'
+    needs: build
+    runs-on: ubuntu-latest
+    environment: pypi
+    steps:
+      - uses: actions/download-artifact@v8
+        with:
+          pattern: dist-*
+          merge-multiple: true
+          path: dist
+      - uses: pypa/gh-action-pypi-publish@dc37677b2e1c63e2034f94d8a5b11f265b73ba33 # v1.14.2
+        with:
+          password: ${{ secrets.PYPI_PAT }}

.gitignore

@@ -10,11 +10,16 @@ __pycache__
 *.a
 uv.lock
 catch2/
-msgq.egg-info/
+*.egg-info/
+/build/
+/dist/
 
 test_runner
 
 libmessaging.*
 libmessaging_shared.*
 .sconsign.dblite
-.mypy_cache/
+.mypy_cache/
+
+/msgq/_ipc.pyi
+/msgq/visionipc/_visionipc.pyi

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) comma.ai contributors
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

MANIFEST.in

@@ -0,0 +1,3 @@
+include README.md pyproject.toml setup.py SConstruct SConscript
+recursive-include msgq *.py py.typed *.h *.cc
+global-exclude *.pyi *.so *.o *.os *.a *.pyc

README.md

@@ -1,9 +1,122 @@
-# MSGQ: A lock free single producer multi consumer message queue
+<div align="center" style="text-align: center;">
 
-## What is this library?
-MSGQ is a generic high performance IPC pub sub system with a single publisher and multiple subscribers. It uses a ring buffer in shared memory to efficiently read and write data. Each read requires a copy. Writing can be done without a copy, as long as the size of the data is known in advance. This library also provides a spoofed implementation that can be used for deterministic testing, and visionipc, an IPC system specifically for large contiguous buffers (like images/video).
+<h1>MSGQ</h1>
+<p><b>High-performance <a href="https://en.wikipedia.org/wiki/Publish%E2%80%93subscribe_pattern">pub/sub</a> messaging, made simple.<br>For Python, C, and C++.</b></p>
 
-## Storage
+<h3>
+  <a href="#quickstart">Quickstart</a>
+  <span> · </span>
+  <a href="#cheatsheet">Cheatsheet</a>
+  <span> · </span>
+  <a href="https://github.com/commaai/msgq/tree/master/msgq/examples">Examples</a>
+  <span> · </span>
+  <a href="https://discord.comma.ai">Discord</a>
+</h3>
+
+[![Discord](https://img.shields.io/badge/Discord-Join-5865F2?logo=discord&logoColor=white)](https://discord.comma.ai)
+[![PyPI](https://img.shields.io/pypi/v/msgq-ipc)](https://pypi.org/project/msgq-ipc/)
+[![Ask DeepWiki](https://deepwiki.com/badge.svg)](https://deepwiki.com/commaai/msgq)
+[![Tests](https://github.com/commaai/msgq/actions/workflows/tests.yml/badge.svg?branch=master)](https://github.com/commaai/msgq/actions/workflows/tests.yml)
+[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://github.com/commaai/msgq/blob/master/LICENSE)
+
+</div>
+
+---
+
+MSGQ lets programs on the same machine exchange messages. A publisher sends messages to a named endpoint, and subscribers listen on that same endpoint. Each endpoint supports one publisher and multiple subscribers.
+
+<p align="center">
+  <img src="https://github.com/user-attachments/assets/79bb91cf-c9ad-4fb4-97d9-33359a083f0f" alt="1 KiB cross-process ping-pong benchmark"><br>
+  <sub>1 KiB cross-process ping-pong on x86 Linux. <a href="https://github.com/commaai/msgq/blob/master/msgq/examples/benchmark.py">Benchmark script</a>.</sub>
+</p>
+
+## Quickstart
+
+```sh
+python -m pip install msgq-ipc
+```
+
+Run the included [publisher](https://github.com/commaai/msgq/blob/master/msgq/examples/publisher.py) and [subscriber](https://github.com/commaai/msgq/blob/master/msgq/examples/subscriber.py) examples in separate terminals:
+
+```sh
+python -m msgq.examples.publisher --endpoint demo   # terminal 1
+python -m msgq.examples.subscriber --endpoint demo  # terminal 2
+```
+
+The subscriber prints `Hello from MSGQ!` once per second.
+
+The core API sends and receives bytes:
+
+```python
+import msgq
+
+publisher = msgq.pub_sock("hello")
+subscriber = msgq.sub_sock("hello")
+publisher.send(b"Hello from MSGQ!")
+print(subscriber.receive())  # b'Hello from MSGQ!'
+```
+
+## Cheatsheet
+
+```python
+import msgq
+
+# API reference; calls are not intended to run in sequence.
+
+# === Create sockets ===
+pub = msgq.pub_sock("demo")                       # One publisher per endpoint
+sub = msgq.sub_sock("demo")                       # Receive messages from that endpoint
+sub = msgq.sub_sock("demo", timeout=1000)         # Wait up to 1000 milliseconds per receive
+sub = msgq.sub_sock("demo", conflate=True)        # Receive only the latest available message
+
+
+# === Send & Receive ===
+pub.send(b"hello")                               # Send nonempty bytes; returns None
+sub.receive()                                    # Return bytes; block by default
+sub.receive(non_blocking=True)                   # Return bytes immediately, or None
+sub.setTimeout(1000)                             # Set receive timeout in milliseconds
+sub.setTimeout(-1)                               # Restore indefinite blocking
+msgq.drain_sock_raw(sub)                         # Return a list of all available messages
+msgq.drain_sock_raw(sub, wait_for_one=True)      # Wait for the first message, then drain
+
+# A receive timeout returns None; draining returns [] if no messages arrive.
+# Slow subscribers can miss messages when the ring buffer wraps.
+
+# === Poll multiple subscribers ===
+poller = msgq.Poller()                           # Create a group of subscribers to watch
+sub = msgq.sub_sock("demo", poller=poller)       # Create and register a subscriber
+poller.registerSocket(sub)                       # Alternatively, register an existing socket
+poller.poll(1000)                                # Return readable sockets; timeout in milliseconds
+poller.poll(0)                                   # Check immediately; return [] if none are ready
+poller.poll(-1)                                  # Wait indefinitely for a readable socket
+
+# Register each socket once. Call receive() on the sockets returned by poll().
+
+# === Reader synchronization and errors ===
+pub.all_readers_updated()                        # Check whether tracked readers have caught up
+pub.wait_for_readers(timeout=1.0, interval=0.01) # Wait for that condition; times are in seconds
+msgq.IpcError                                    # Messaging failure exception
+msgq.MultiplePublishersError                     # Publisher conflict; subclass of IpcError
+
+# wait_for_readers() raises TimeoutError if its deadline expires.
+# Synchronization checks queue positions, not application processing. It requires
+# at least one tracked reader and ignores readers invalidated by an overwrite.
+```
+
+## Contributing
+
+Issues and pull requests are welcome on [GitHub](https://github.com/commaai/msgq). Run `./test.sh` to build, lint, and test the package.
+
+## License
+
+MSGQ is available under the [MIT License](https://github.com/commaai/msgq/blob/master/LICENSE).
+
+<details>
+<summary>Under the hood</summary>
+
+The message queue copies data on send and receive. A fake implementation is also available for deterministic testing.
+
+### Storage
 The storage for the queue consists of an area of metadata, and the actual buffer. The metadata contains:
 
 1. A counter to the number of readers that are active
@@ -18,7 +131,7 @@ The counter and the pointer are both 32 bit values, packed into 64 bit so they c
 The data buffer is a ring buffer. All messages are prefixed by an 8 byte size field, followed by the data. A size of -1 indicates a wrap-around, and means the next message is stored at the beginning of the buffer.
 
 
-## Writing
+### Writing
 Writing involves the following steps:
 
 1. Check if the area that is to be written overlaps with any of the read pointers, mark those readers as invalid by clearing the validity flag.
@@ -29,14 +142,14 @@ In case there is not enough space at the end of the buffer, a special empty mess
 
 There always needs to be 8 bytes of empty space at the end of the buffer. By doing this there is always space to write the -1.
 
-## Reset reader
+### Reset reader
 When the reader is lagging too much behind the read pointer becomes invalid and no longer points to the beginning of a valid message. To reset a reader to the current write pointer, the following steps are performed:
 
 1. Set valid flag
 2. Set read cycle counter to that of the writer
 3. Set read pointer to write pointer
 
-## Reading
+### Reading
 Reading involves the following steps:
 
 1. Read the size field at the current read pointer
@@ -52,3 +165,5 @@ If a writer overwrites the data while it's being copied out, the data will be in
 If at steps 2 or 5 the validity flag is not set, the reader is reset. Any data that was already read is discarded. After the reader is reset, the reading starts from the beginning.
 
 If a message with size -1 is encountered, step 3 and 4 are replaced by increasing the cycle counter and setting the read pointer to the beginning of the buffer. After that another read is performed.
+
+</details>

SConscript

@@ -1,18 +1,38 @@
-Import('env', 'envCython', 'common')
+import sys
+import sysconfig
+from pathlib import Path
+
+import nanobind
+
+Import('env', 'arch', 'common')
+
+envNanobind = env.Clone(LIBS=[])
+nb_dir = Path(nanobind.source_dir()).parent
+envNanobind.Append(CPPPATH=[sysconfig.get_paths()['include'], nanobind.include_dir(), str(nb_dir / "ext/robin_map/include")],
+                   CCFLAGS=["-fvisibility=hidden", "-fno-strict-aliasing"])
+if arch == "Darwin":
+  envNanobind.Append(LINKFLAGS=["-bundle", "-undefined", "dynamic_lookup"])
+else:
+  # Python symbols resolve at import time.
+  envNanobind["LINKFLAGS"] = [f for f in env["LINKFLAGS"] if f != "-Wl,--no-undefined"] + ["-pthread", "-shared"]
+
+# Third-party runtime sources do not follow our shadowing warning policy.
+envNanobindRuntime = envNanobind.Clone()
+envNanobindRuntime['CCFLAGS'] = [flag for flag in envNanobindRuntime['CCFLAGS'] if not flag.startswith('-Wshadow')]
+nanobind_lib = envNanobindRuntime.SharedObject('msgq/nb_combined.os', str(nb_dir / "src/nb_combined.cpp"))
 
 
 visionipc_dir = Dir('msgq/visionipc')
+python_suffix = sysconfig.get_config_var('EXT_SUFFIX')
 
 # Build msgq
 msgq_objects = env.SharedObject([
   'msgq/ipc.cc',
   'msgq/event.cc',
-  'msgq/impl_msgq.cc',
-  'msgq/impl_fake.cc',
   'msgq/msgq.cc',
 ])
 msgq = env.Library('msgq', msgq_objects)
-msgq_python = envCython.Program('msgq/ipc_pyx.so', 'msgq/ipc_pyx.pyx', LIBS=envCython["LIBS"]+[msgq]+common)
+msgq_python = envNanobind.Program(f'msgq/_ipc{python_suffix}', ['msgq/ipc_py.cc', nanobind_lib], LIBS=envNanobind["LIBS"]+[msgq]+common)
 
 # Build Vision IPC
 vipc_files = ['visionipc.cc', 'visionipc_server.cc', 'visionipc_client.cc']
@@ -26,10 +46,17 @@ vipc_objects = env.SharedObject(vipc_sources)
 visionipc = env.Library('visionipc', vipc_objects)
 
 
-vipc_libs = envCython["LIBS"] + [visionipc, msgq] + common
-envCython.Program(f'{visionipc_dir.abspath}/visionipc_pyx.so', f'{visionipc_dir.abspath}/visionipc_pyx.pyx',
+vipc_libs = envNanobind["LIBS"] + [visionipc, msgq] + common
+visionipc_python = envNanobind.Program(f'{visionipc_dir.abspath}/_visionipc{python_suffix}', [f'{visionipc_dir.abspath}/visionipc_py.cc', nanobind_lib],
                   LIBS=vipc_libs)
 
+# ASan modules require a preloaded runtime before Python can import them.
+if "-fsanitize=address" not in env["LINKFLAGS"]:
+  root = Dir('.').abspath
+  for module in ('msgq._ipc', 'msgq.visionipc._visionipc'):
+    env.Command(module.replace('.', '/') + '.pyi', [msgq_python, visionipc_python],
+                f'"{sys.executable}" -m nanobind.stubgen -i "{root}" -m {module} -o "$TARGET"')
+
 if GetOption('extras'):
   env.Program('msgq/test_runner', ['msgq/msgq_tests.cc'], LIBS=[msgq]+common)
 

SConstruct

@@ -1,19 +1,18 @@
 import os
 import platform
 import subprocess
-import sysconfig
 
 arch = subprocess.check_output(["uname", "-m"], encoding='utf8').rstrip()
 if platform.system() == "Darwin":
   arch = "Darwin"
 
-common = []
+# shm_open/shm_unlink live in librt on older glibc versions (including manylinux).
+common = [] if arch == "Darwin" else ["rt"]
 
 cpppath = [
   "#/",
   '#msgq/',
   '/usr/lib/include',
-  sysconfig.get_paths()['include'],
 ]
 
 AddOption('--minimal',
@@ -61,21 +60,9 @@ env = Environment(
   CFLAGS="-std=gnu11",
   CXXFLAGS="-std=c++1z",
   CPPPATH=cpppath,
-  CYTHONCFILESUFFIX=".cpp",
-  tools=["default", "cython"]
+  tools=["default"]
 )
 
 Export('env', 'arch', 'common')
 
-envCython = env.Clone(LIBS=[])
-envCython["CCFLAGS"] += ["-Wno-#warnings", "-Wno-cpp", "-Wno-shadow", "-Wno-deprecated-declarations"]
-envCython["CCFLAGS"].remove('-Werror')
-if arch == "Darwin":
-  envCython["LINKFLAGS"] = ["-bundle", "-undefined", "dynamic_lookup"]
-else:
-  envCython["LINKFLAGS"] = ["-pthread", "-shared"]
-
-Export('envCython')
-
-
 SConscript(['SConscript'])