diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 000000000..6313b56c5 --- /dev/null +++ b/.gitattributes @@ -0,0 +1 @@ +* text=auto eol=lf diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index cc6d7df65..9e9adccf3 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -6,10 +6,6 @@ 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 @@ -17,8 +13,9 @@ jobs: 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 diff --git a/.github/workflows/wheels.yml b/.github/workflows/wheels.yml new file mode 100644 index 000000000..4cbbeadd8 --- /dev/null +++ b/.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 }} diff --git a/.gitignore b/.gitignore index ff828c476..1963e59d5 100644 --- a/.gitignore +++ b/.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/ \ No newline at end of file +.mypy_cache/ + +/msgq/_ipc.pyi +/msgq/visionipc/_visionipc.pyi diff --git a/LICENSE b/LICENSE new file mode 100644 index 000000000..43af41c78 --- /dev/null +++ b/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. diff --git a/MANIFEST.in b/MANIFEST.in new file mode 100644 index 000000000..5f694e2a0 --- /dev/null +++ b/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 diff --git a/README.md b/README.md index 23d3b9e2e..1609f32a8 100644 --- a/README.md +++ b/README.md @@ -1,9 +1,122 @@ -# MSGQ: A lock free single producer multi consumer message queue +
-## 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). +

MSGQ

+

High-performance pub/sub messaging, made simple.
For Python, C, and C++.

-## Storage +

+ Quickstart + · + Cheatsheet + · + Examples + · + Discord +

+ +[![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) + +
+ +--- + +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. + +

+ 1 KiB cross-process ping-pong benchmark
+ 1 KiB cross-process ping-pong on x86 Linux. Benchmark script. +

+ +## 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). + +
+Under the hood + +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. + +
diff --git a/SConscript b/SConscript index cde36529e..4ecc1e08c 100644 --- a/SConscript +++ b/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) diff --git a/SConstruct b/SConstruct index 0ee7a1dba..e3d22e4bc 100644 --- a/SConstruct +++ b/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']) diff --git a/lefthook.yml b/lefthook.yml index 91b18aa52..43d664a3b 100644 --- a/lefthook.yml +++ b/lefthook.yml @@ -18,12 +18,12 @@ test: ty: run: ty check . codespell: - run: codespell {files} -L ned,stdio,master --builtin clear,rare,informal,usage,code,names,en-GB_to_en-US -S uv.lock,*_pyx.cpp + run: codespell {files} -L ned,stdio,master --builtin clear,rare,informal,usage,code,names,en-GB_to_en-US -S uv.lock files: git ls-tree -r HEAD --name-only cppcheck: run: cppcheck --error-exitcode=1 --inline-suppr --language=c++ --force --quiet -j4 --check-level=exhaustive $(git ls-files '*.cc' | grep -v -E '(msgq_tests|test_runner)\.cc') cpplint: - run: cpplint --exclude=msgq/ipc_pyx.cpp --exclude=msgq/visionipc/visionipc_pyx.cpp --recursive --quiet --counting=detailed --linelength=240 --filter=-build,-legal,-readability,-runtime,-whitespace,+build/include_subdir,+build/forward_decl,+build/include_what_you_use,+build/deprecated,+whitespace/comma,+whitespace/line_length,+whitespace/empty_if_body,+whitespace/empty_loop_body,+whitespace/empty_conditional_body,+whitespace/forcolon,+whitespace/parens,+whitespace/semicolon,+whitespace/tab,+readability/braces msgq/ + run: cpplint --recursive --quiet --counting=detailed --linelength=240 --filter=-build,-legal,-readability,-runtime,-whitespace,+build/include_subdir,+build/forward_decl,+build/include_what_you_use,+build/deprecated,+whitespace/comma,+whitespace/line_length,+whitespace/empty_if_body,+whitespace/empty_loop_body,+whitespace/empty_conditional_body,+whitespace/forcolon,+whitespace/parens,+whitespace/semicolon,+whitespace/tab,+readability/braces msgq/ # *** tests *** test_runner: diff --git a/msgq/.gitignore b/msgq/.gitignore deleted file mode 100644 index 6bd751773..000000000 --- a/msgq/.gitignore +++ /dev/null @@ -1 +0,0 @@ -ipc_pyx.cpp diff --git a/msgq/__init__.py b/msgq/__init__.py index 1f71f3b6d..cb459251a 100644 --- a/msgq/__init__.py +++ b/msgq/__init__.py @@ -1,12 +1,10 @@ -# must be built with scons -from msgq.ipc_pyx import Context, Poller, SubSocket, PubSocket, SocketEventHandle, toggle_fake_events, \ +from msgq._ipc import Poller, SubSocket, PubSocket, SocketEventHandle, toggle_fake_events, \ set_fake_prefix, get_fake_prefix, delete_fake_prefix, wait_for_one_event -from msgq.ipc_pyx import MultiplePublishersError, IpcError +from msgq._ipc import MultiplePublishersError, IpcError -from typing import Optional, List, Union +from typing import Optional, List __all__ = [ - "Context", "Poller", "SubSocket", "PubSocket", @@ -19,7 +17,6 @@ "delete_fake_prefix", "wait_for_one_event", "NO_TRAVERSAL_LIMIT", - "context", "fake_event_handle", "pub_sock", "sub_sock", @@ -28,10 +25,7 @@ NO_TRAVERSAL_LIMIT = 2**64-1 -context = Context() - - -def fake_event_handle(endpoint: str, identifier: Optional[Union[str, bytes]] = None, override: bool = True, enable: bool = False) -> SocketEventHandle: +def fake_event_handle(endpoint: str, identifier: Optional[str] = None, override: bool = True, enable: bool = False) -> SocketEventHandle: ident = identifier if identifier is not None else get_fake_prefix() handle = SocketEventHandle(endpoint, ident, override) if override: @@ -41,14 +35,14 @@ def fake_event_handle(endpoint: str, identifier: Optional[Union[str, bytes]] = N def pub_sock(endpoint: str, segment_size: int = 0) -> PubSocket: sock = PubSocket() - sock.connect(context, endpoint, segment_size) + sock.connect(endpoint, segment_size) return sock -def sub_sock(endpoint: str, poller: Optional[Poller] = None, addr: str = "127.0.0.1", - conflate: bool = False, timeout: Optional[int] = None, segment_size: int = 0) -> SubSocket: +def sub_sock(endpoint: str, poller: Optional[Poller] = None, conflate: bool = False, + timeout: Optional[int] = None, segment_size: int = 0) -> SubSocket: sock = SubSocket() - sock.connect(context, endpoint, addr.encode('utf8'), conflate, segment_size) + sock.connect(endpoint, conflate, segment_size) if timeout is not None: sock.setTimeout(timeout) diff --git a/msgq/examples/__init__.py b/msgq/examples/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/msgq/examples/benchmark.py b/msgq/examples/benchmark.py new file mode 100644 index 000000000..209793525 --- /dev/null +++ b/msgq/examples/benchmark.py @@ -0,0 +1,292 @@ +# /// script +# requires-python = ">=3.11" +# dependencies = ["msgq-ipc>=1.0", "pyzmq", "eclipse-zenoh>=1.10", "lcm", "matplotlib"] +# [tool.uv.sources] +# msgq-ipc = { path = "../.." } +# [tool.ty.rules] +# unresolved-import = "ignore" +# /// + +import os +import lcm +import zmq +import msgq +import time +import zenoh +import random +import argparse +import platform +import tempfile +import matplotlib +import statistics +import multiprocessing + +from pathlib import Path +from threading import Event +from collections import deque # codespell:ignore deque +from functools import partial +from contextlib import ExitStack + +matplotlib.use("Agg") +import matplotlib.pyplot as plt + +BACKENDS = ("msgq", "pyzmq IPC", "Pipe bytes", "zenoh IPC", "LCM UDP") +ROUNDS = 1000 +SECONDS = 2 +REPEATS = 5 +TIMEOUT = 60 + + +def setup_backend(name, role, directory, prefix, barrier, pipe, cleanup, size): + outgoing, incoming = ("request", "reply") if role == "sender" else ("reply", "request") + if name == "msgq": + os.environ["OPENPILOT_PREFIX"] = prefix + publisher = msgq.pub_sock(outgoing) + # Creating a publisher resets its queue, so both must exist before subscribing. + barrier.wait(10) + subscriber = msgq.sub_sock(incoming) + return publisher.send, subscriber.receive + if name == "pyzmq IPC": + context = cleanup.enter_context(zmq.Context()) + publisher = context.socket(zmq.XPUB) + subscriber = context.socket(zmq.SUB) + cleanup.callback(publisher.close, linger=0) + cleanup.callback(subscriber.close, linger=0) + publisher.setsockopt(zmq.RCVTIMEO, 5000) + subscriber.setsockopt(zmq.RCVTIMEO, 5000) + subscriber.setsockopt(zmq.SUBSCRIBE, b"") + publisher.bind(f"ipc://{directory}/{outgoing}") + subscriber.connect(f"ipc://{directory}/{incoming}") + if publisher.recv() != b"\x01": + raise RuntimeError("Unexpected ZeroMQ subscription") + return publisher.send, subscriber.recv + if name == "Pipe bytes": + cleanup.enter_context(pipe) + return pipe.send_bytes, pipe.recv_bytes + if name == "zenoh IPC": + endpoint = f"unixsock-stream/{directory}/zenoh" + config = zenoh.Config() + config.insert_json5("listen/endpoints", repr([endpoint] if role == "sender" else [])) + config.insert_json5("connect/endpoints", repr([] if role == "sender" else [endpoint])) + config.insert_json5("scouting/multicast/enabled", "false") + config.insert_json5("scouting/gossip/enabled", "false") + config.insert_json5("transport/unicast/qos/enabled", "false") + # Low-latency transport cannot fragment a 64 KiB payload. + config.insert_json5("transport/unicast/lowlatency", str(size <= 1024).lower()) + session = cleanup.enter_context(zenoh.open(config)) + zenoh_subscriber = session.declare_subscriber(incoming) + cleanup.callback(zenoh_subscriber.undeclare) + publisher = session.declare_publisher(outgoing) + cleanup.callback(publisher.undeclare) + ready = Event() + listener = publisher.declare_matching_listener(lambda status: ready.set() if status.matching else None) + cleanup.callback(listener.undeclare) + if publisher.matching_status.matching: + ready.set() + if not ready.wait(10): + raise RuntimeError("Zenoh subscriber did not become ready") + return publisher.put, lambda: zenoh_subscriber.recv().payload.to_bytes() + if name == "LCM UDP": + bus = lcm.LCM("udpm://239.255.76.67:7667?ttl=0") + messages = deque() # codespell:ignore deque + token = Path(directory).name + subscriber = bus.subscribe(token + incoming, lambda channel, data: messages.append(data)) + cleanup.callback(bus.unsubscribe, subscriber) + + def receive(): + while not messages: + if bus.handle_timeout(5000) == 0: + raise TimeoutError("LCM receive timed out; check host multicast support") + return messages.popleft() + + return partial(bus.publish, token + outgoing), receive + raise ValueError(name) + + +def payload(sequence, padding): + return sequence.to_bytes(8, "little") + padding + + +def sender(send, receive, size): + padding = b"x" * (size - 8) + sequence = 0 + + def exchange(): + nonlocal sequence + message = payload(sequence, padding) + send(message) + reply = receive() + if reply != message: + raise RuntimeError(f"Invalid reply at sequence {sequence}: {None if reply is None else (len(reply), reply[:8])}") + sequence += 1 + + start = time.perf_counter() + while sequence < 100 or time.perf_counter() - start < 0.1: + exchange() + send(b"BEGIN") + if receive() != b"BEGIN": + raise RuntimeError("Invalid measurement handshake") + warm_up_count = sequence + start = time.perf_counter() + while True: + for _ in range(64): + exchange() + elapsed = time.perf_counter() - start + count = sequence - warm_up_count + if count >= ROUNDS and elapsed >= SECONDS: + break + # Stop timing before the completion handshake. + send(b"END") + if receive() != b"END": + raise RuntimeError("Invalid completion handshake") + return {"round_trips": count, "seconds": elapsed} + + +def echo(send, receive, size): + padding = b"x" * (size - 8) + sequence = 0 + measured_start = None + while True: + message = receive() + if message == b"BEGIN" and measured_start is None: + measured_start = sequence + send(message) + elif message == b"END" and measured_start is not None: + send(message) + return {"round_trips": sequence - measured_start} + else: + if message != payload(sequence, padding): + raise RuntimeError(f"Invalid request at sequence {sequence}: {None if message is None else (len(message), message[:8])}") + sequence += 1 + send(message) + + +def peer(name, role, directory, prefix, barrier, pipe, control, size): + try: + with ExitStack() as cleanup: + send, receive = setup_backend(name, role, directory, prefix, barrier, pipe, cleanup, size) + barrier.wait(10) + result = sender(send, receive, size) if role == "sender" else echo(send, receive, size) + control.send(("ok", result)) + except BaseException as error: + control.send(("error", f"{name} {role}: {type(error).__name__}: {error}")) + finally: + pipe.close() + control.close() + + +def measure(name, size): + context = multiprocessing.get_context("spawn") + shared_root = "/tmp" if platform.system() == "Darwin" else "/dev/shm" + with tempfile.TemporaryDirectory(prefix="mq_", dir="/tmp") as directory, \ + tempfile.TemporaryDirectory(prefix="msgq_", dir=shared_root) as shared: + # An owned prefix isolates and removes MSGQ files without touching existing queues. + prefix = Path(shared).name.removeprefix("msgq_") + barrier = context.Barrier(2) + pipes = context.Pipe() + processes, controls = [], [] + deadline = time.monotonic() + TIMEOUT + try: + for role, pipe in zip(("sender", "echo"), pipes, strict=True): + parent, child = context.Pipe(duplex=False) + controls.append(parent) + process = context.Process(target=peer, args=(name, role, directory, prefix, barrier, pipe, child, size)) + process.start() + processes.append(process) + child.close() + for pipe in pipes: + pipe.close() + results = {} + while len(results) < 2: + if time.monotonic() >= deadline: + raise TimeoutError(f"{name}: trial exceeded {TIMEOUT}s") + for index, control in enumerate(controls): + if index in results: + continue + if control.poll(0.05): + try: + status, result = control.recv() + except EOFError as error: + raise RuntimeError(f"{name}: peer {index} closed its result pipe unexpectedly") from error + if status != "ok": + raise RuntimeError(result) + results[index] = result + elif processes[index].exitcode is not None: + raise RuntimeError(f"{name}: peer exited without results ({processes[index].exitcode})") + for process in processes: + process.join(max(0, deadline - time.monotonic())) + if process.exitcode != 0: + raise RuntimeError(f"{name}: peer failed to exit cleanly") + result = results[0] + if result["round_trips"] != results[1]["round_trips"]: + raise RuntimeError(f"{name}: peers disagree on verified message count") + return result + finally: + for process in processes: + if process.is_alive(): + process.terminate() + for process in processes: + process.join(2) + if process.is_alive(): + process.kill() + process.join() + for connection in (*controls, *pipes): + connection.close() + + +def plot_results(results, path): + from matplotlib.ticker import EngFormatter, MaxNLocator + + labels = {"msgq": "msgq", "pyzmq IPC": "pyzmq", "Pipe bytes": "multiprocessing.Pipe", "zenoh IPC": "zenoh", "LCM UDP": "LCM"} + rows = sorted(((name, rate) for name, size, rate in results if size == 1024), key=lambda row: row[1]) + maximum = max(rate for _, rate in rows) + formatter = EngFormatter(sep="", places=2) + with plt.rc_context({"font.family": "sans-serif", "font.sans-serif": ["Helvetica Neue", "Arial", "DejaVu Sans"]}): + fig, ax = plt.subplots(figsize=(10, max(2.2, len(rows) * 0.48 + 0.7))) + fig.set_facecolor("white") + ax.set(xlim=(0, maximum * 1.2), ylim=(len(rows) - 0.5, -0.5), yticks=[]) + # Keep grid lines out of the values on the right. + ticks = MaxNLocator(nbins=4).tick_values(0, maximum) + ax.set_xticks([tick for tick in ticks if 0 <= tick < maximum]) + ax.xaxis.set_major_formatter(EngFormatter(sep="")) + ax.set_xlabel("messages/sec — higher is better", fontsize=14, labelpad=8, color="#172b3a") + ax.tick_params(length=0, pad=6, labelsize=11, colors="#425b6c") + ax.set_axisbelow(True) + ax.grid(axis="x", color="#b7b7b7", linewidth=1.5) + for side, spine in ax.spines.items(): + spine.set(visible=side == "left", color="#192a32", linewidth=2) + for index, (name, rate) in enumerate(rows): + gray = str(0.88 - 0.28 * index / max(len(rows) - 2, 1)) + ax.barh(index, rate, height=1, color="#31cf43" if name == "msgq" else gray, zorder=2) + ax.text(maximum * 0.029, index, labels[name], va="center", fontsize=16, fontweight="bold", color="#172b3a") + ax.text(maximum * 1.173, index, formatter(rate), ha="right", va="center", fontsize=16, fontweight="bold", color="#172b3a") + fig.subplots_adjust(left=0.055, right=0.98, top=0.98, bottom=0.21) + path.parent.mkdir(parents=True, exist_ok=True) + fig.savefig(path, dpi=180, facecolor=fig.get_facecolor()) + plt.close(fig) + print(f"Chart saved to {path}") + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="Cross-process pub/sub ping-pong benchmark.") + parser.add_argument("--plot", type=Path, help="save a 1 KiB chart") + args = parser.parse_args() + if "CEREAL_FAKE" in os.environ: + parser.error("Unset CEREAL_FAKE to benchmark the real MSGQ backend") + print("Cross-process ping-pong; messages/sec counts requests and replies.", flush=True) + samples = {(name, size): [] for name in BACKENDS for size in (64, 1024, 65536)} + randomizer = random.Random(0) + for _ in range(REPEATS): + jobs = list(samples) + randomizer.shuffle(jobs) + for name, size in jobs: + sample = measure(name, size) + rate = 2 * sample["round_trips"] / sample["seconds"] + samples[name, size].append(rate) + print(f"{name:<12} {size:>6} bytes: {rate:>12,.0f} messages/sec", flush=True) + results = [(name, size, statistics.median(rates)) for (name, size), rates in samples.items()] + print("\nMedians:") + for name, size, rate in results: + print(f"{name:<12} {size:>6} bytes: {rate:>12,.0f} messages/sec") + if args.plot: + plot_results(results, args.plot) diff --git a/msgq/examples/publisher.py b/msgq/examples/publisher.py new file mode 100644 index 000000000..3c9360bb8 --- /dev/null +++ b/msgq/examples/publisher.py @@ -0,0 +1,21 @@ +import argparse +import time + +import msgq + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="Publish a greeting once per second.") + parser.add_argument("--endpoint", default="msgq_example", help="endpoint name (default: %(default)s)") + args = parser.parse_args() + publisher = msgq.pub_sock(args.endpoint) + + print("Ctrl-C to exit") + try: + while True: + message = "Hello from MSGQ!" + publisher.send(message.encode("utf-8")) + print(f"Sent: {message}", flush=True) + time.sleep(1) + except KeyboardInterrupt: + pass diff --git a/msgq/examples/subscriber.py b/msgq/examples/subscriber.py new file mode 100644 index 000000000..9a17e844b --- /dev/null +++ b/msgq/examples/subscriber.py @@ -0,0 +1,19 @@ +import argparse + +import msgq + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="Receive and print greetings from a publisher.") + parser.add_argument("--endpoint", default="msgq_example", help="endpoint name (default: %(default)s)") + args = parser.parse_args() + subscriber = msgq.sub_sock(args.endpoint, timeout=1000) + + print("Ctrl-C to exit") + try: + while True: + message = subscriber.receive() + if message is not None: + print(f"Received: {message.decode('utf-8')}", flush=True) + except KeyboardInterrupt: + pass diff --git a/msgq/impl_fake.cc b/msgq/impl_fake.cc deleted file mode 100644 index 682990109..000000000 --- a/msgq/impl_fake.cc +++ /dev/null @@ -1,10 +0,0 @@ -#include -#include "msgq/impl_fake.h" - -void FakePoller::registerSocket(SubSocket *socket) { - this->sockets.push_back(socket); -} - -std::vector FakePoller::poll(int timeout) { - return this->sockets; -} diff --git a/msgq/impl_fake.h b/msgq/impl_fake.h deleted file mode 100644 index 3884bf6ea..000000000 --- a/msgq/impl_fake.h +++ /dev/null @@ -1,75 +0,0 @@ -#pragma once - -#include -#include -#include -#include -#include - -#include -#include -#include -#include - -#include "msgq/ipc.h" -#include "msgq/event.h" - -template -class FakeSubSocket: public TSubSocket { -private: - EventState *state = nullptr; - int fds[2] = {-1, -1}; - - void ensure_fifos_open() { - for (size_t i = 0; i < 2; i++) { - if (fds[i] < 0 && state->paths[i][0] != '\0') { - fds[i] = open(state->paths[i], O_RDWR | O_NONBLOCK); - } - } - } - -public: - FakeSubSocket(): TSubSocket() {} - ~FakeSubSocket() { - for (int fd : fds) { - if (fd >= 0) close(fd); - } - if (state != nullptr) { - munmap(state, sizeof(EventState)); - } - } - - int connect(Context *context, std::string endpoint, std::string address, bool conflate=false, bool check_endpoint=true, size_t segment_size=0) override { - const char* cereal_prefix = std::getenv("CEREAL_FAKE_PREFIX"); - - char* mem; - std::string identifier = cereal_prefix != nullptr ? std::string(cereal_prefix) : ""; - event_state_shm_mmap(endpoint, identifier, &mem, nullptr); - - this->state = (EventState*)mem; - ensure_fifos_open(); - - return TSubSocket::connect(context, endpoint, address, conflate, check_endpoint, segment_size); - } - - Message *receive(bool non_blocking=false) override { - if (this->state->enabled) { - ensure_fifos_open(); - Event(fds[EventPurpose::RECV_CALLED]).set(); - Event(fds[EventPurpose::RECV_READY]).wait(); - Event(fds[EventPurpose::RECV_READY]).clear(); - } - - return TSubSocket::receive(non_blocking); - } -}; - -class FakePoller: public Poller { -private: - std::vector sockets; - -public: - void registerSocket(SubSocket *socket) override; - std::vector poll(int timeout) override; - ~FakePoller() {} -}; diff --git a/msgq/impl_msgq.cc b/msgq/impl_msgq.cc deleted file mode 100644 index 8243cee2d..000000000 --- a/msgq/impl_msgq.cc +++ /dev/null @@ -1,169 +0,0 @@ -#include -#include -#include -#include -#include -#include -#include - -#include "msgq/impl_msgq.h" - -void Message::init(size_t sz) { - size = sz; - data = new char[size]; -} - -void Message::init(char * d, size_t sz) { - size = sz; - data = new char[size]; - memcpy(data, d, size); -} - -void Message::takeOwnership(char * d, size_t sz) { - size = sz; - data = d; -} - -void Message::close() { - if (size > 0){ - delete[] data; - } - size = 0; -} - -Message::~Message() { - this->close(); -} - -int MSGQSubSocket::connect(Context *context, std::string endpoint, std::string address, bool conflate, bool check_endpoint, size_t segment_size){ - assert(context); - assert(address == "127.0.0.1"); - - q = new msgq_queue_t; - size_t size = segment_size > 0 ? segment_size : DEFAULT_SEGMENT_SIZE; - int r = msgq_new_queue(q, endpoint.c_str(), size); - if (r != 0){ - return r; - } - - msgq_init_subscriber(q); - - if (conflate){ - q->read_conflate = true; - } - - timeout = -1; - - return 0; -} - - -Message * MSGQSubSocket::receive(bool non_blocking){ - msgq_msg_t msg; - - Message *r = NULL; - - int rc = msgq_msg_recv(&msg, q); - - // Hack to implement blocking read with a poller. Don't use this - while (!non_blocking && rc == 0){ - msgq_pollitem_t items[1]; - items[0].q = q; - - int t = (timeout != -1) ? timeout : 100; - - int n = msgq_poll(items, 1, t); - rc = msgq_msg_recv(&msg, q); - - // The poll indicated a message was ready, but the receive failed. Try again - if (n == 1 && rc == 0){ - continue; - } - - if (timeout != -1){ - break; - } - } - - if (rc > 0){ - r = new Message; - r->takeOwnership(msg.data, msg.size); - } - - return r; -} - -void MSGQSubSocket::setTimeout(int t){ - timeout = t; -} - -MSGQSubSocket::~MSGQSubSocket(){ - if (q != NULL){ - msgq_close_queue(q); - delete q; - } -} - -int PubSocket::connect(Context *context, std::string endpoint, bool check_endpoint, size_t segment_size){ - assert(context); - - q = new msgq_queue_t; - size_t size = segment_size > 0 ? segment_size : DEFAULT_SEGMENT_SIZE; - int r = msgq_new_queue(q, endpoint.c_str(), size); - if (r != 0){ - return r; - } - - msgq_init_publisher(q); - - return 0; -} - -int PubSocket::sendMessage(Message *message){ - msgq_msg_t msg; - msg.data = message->getData(); - msg.size = message->getSize(); - - return msgq_msg_send(&msg, q); -} - -int PubSocket::send(char *data, size_t size){ - msgq_msg_t msg; - msg.data = data; - msg.size = size; - - return msgq_msg_send(&msg, q); -} - -bool PubSocket::all_readers_updated() { - return msgq_all_readers_updated(q); -} - -PubSocket::~PubSocket(){ - if (q != NULL){ - msgq_close_queue(q); - delete q; - } -} - - -void MSGQPoller::registerSocket(SubSocket * socket){ - assert(num_polls + 1 < MAX_POLLERS); - polls[num_polls].q = static_cast(socket)->getQueue(); - - sockets.push_back(socket); - num_polls++; -} - -std::vector MSGQPoller::poll(int timeout){ - std::vector r; - - msgq_poll(polls, num_polls, timeout); - for (size_t i = 0; i < num_polls; i++){ - if (polls[i].revents){ - r.push_back(sockets[i]); - } - } - - return r; -} diff --git a/msgq/impl_msgq.h b/msgq/impl_msgq.h deleted file mode 100644 index bd9a9aa03..000000000 --- a/msgq/impl_msgq.h +++ /dev/null @@ -1,33 +0,0 @@ -#pragma once - -#include -#include - -#include "msgq/ipc.h" -#include "msgq/msgq.h" - -#define MAX_POLLERS 128 - -class MSGQSubSocket : public SubSocket { -private: - msgq_queue_t * q = NULL; - int timeout; -public: - int connect(Context *context, std::string endpoint, std::string address, bool conflate=false, bool check_endpoint=true, size_t segment_size=0); - void setTimeout(int timeout); - msgq_queue_t * getQueue() {return q;} - Message *receive(bool non_blocking=false); - ~MSGQSubSocket(); -}; - -class MSGQPoller : public Poller { -private: - std::vector sockets; - msgq_pollitem_t polls[MAX_POLLERS]; - size_t num_polls = 0; - -public: - void registerSocket(SubSocket *socket); - std::vector poll(int timeout); - ~MSGQPoller(){} -}; diff --git a/msgq/ipc.cc b/msgq/ipc.cc index ec4a64e8f..f53f3d28c 100644 --- a/msgq/ipc.cc +++ b/msgq/ipc.cc @@ -1,30 +1,134 @@ #include +#include +#include +#include #include #include +#include + +#include +#include +#include #include "msgq/ipc.h" -#include "msgq/impl_msgq.h" -#include "msgq/impl_fake.h" +#include "msgq/event.h" + +static bool messaging_use_fake(){ + return std::getenv("CEREAL_FAKE") != NULL; +} + +void Message::init(size_t sz) { + size = sz; + data = new char[size]; +} -bool messaging_use_fake(){ - char* fake_enabled = std::getenv("CEREAL_FAKE"); - return fake_enabled != NULL; +void Message::init(char * d, size_t sz) { + size = sz; + data = new char[size]; + memcpy(data, d, size); } -Context * Context::create(){ - return new Context(); +void Message::takeOwnership(char * d, size_t sz) { + size = sz; + data = d; } -SubSocket * SubSocket::create(){ +void Message::close() { + if (size > 0){ + delete[] data; + } + size = 0; +} + +Message::~Message() { + this->close(); +} + +void SubSocket::open_fake_fifos() { + for (size_t i = 0; i < 2; i++) { + if (fake_fds[i] < 0 && fake_state->paths[i][0] != '\0') { + fake_fds[i] = open(fake_state->paths[i], O_RDWR | O_NONBLOCK); + } + } +} + +int SubSocket::connect(std::string endpoint, bool conflate, size_t segment_size){ if (messaging_use_fake()) { - return new FakeSubSocket(); + const char* cereal_prefix = std::getenv("CEREAL_FAKE_PREFIX"); + std::string identifier = cereal_prefix != nullptr ? std::string(cereal_prefix) : ""; + + char* mem; + event_state_shm_mmap(endpoint, identifier, &mem, nullptr); + fake_state = (EventState*)mem; + open_fake_fifos(); + } + + q = new msgq_queue_t; + size_t size = segment_size > 0 ? segment_size : DEFAULT_SEGMENT_SIZE; + int r = msgq_new_queue(q, endpoint.c_str(), size); + if (r != 0){ + return r; } - return new MSGQSubSocket(); + + msgq_init_subscriber(q); + + if (conflate){ + q->read_conflate = true; + } + + return 0; } -SubSocket * SubSocket::create(Context * context, std::string endpoint, std::string address, bool conflate, bool check_endpoint, size_t segment_size){ - SubSocket *s = SubSocket::create(); - int r = s->connect(context, endpoint, address, conflate, check_endpoint, segment_size); + +Message * SubSocket::receive(bool non_blocking){ + if (fake_state != nullptr && fake_state->enabled) { + open_fake_fifos(); + Event(fake_fds[EventPurpose::RECV_CALLED]).set(); + Event(fake_fds[EventPurpose::RECV_READY]).wait(); + Event(fake_fds[EventPurpose::RECV_READY]).clear(); + } + + msgq_msg_t msg; + + Message *r = NULL; + + int rc = msgq_msg_recv(&msg, q); + + // Hack to implement blocking read with a poller. Don't use this + while (!non_blocking && rc == 0){ + msgq_pollitem_t items[1]; + items[0].q = q; + + int t = (timeout != -1) ? timeout : 100; + + int n = msgq_poll(items, 1, t); + rc = msgq_msg_recv(&msg, q); + + // The poll indicated a message was ready, but the receive failed. Try again + if (n == 1 && rc == 0){ + continue; + } + + if (timeout != -1){ + break; + } + } + + if (rc > 0){ + r = new Message; + r->takeOwnership(msg.data, msg.size); + } + + return r; +} + +void SubSocket::setTimeout(int t){ + timeout = t; +} + +SubSocket * SubSocket::create(std::string endpoint, bool conflate, size_t segment_size){ + SubSocket *s = new SubSocket(); + int r = s->connect(endpoint, conflate, segment_size); if (r == 0) { return s; @@ -36,13 +140,56 @@ SubSocket * SubSocket::create(Context * context, std::string endpoint, std::stri } } -PubSocket * PubSocket::create(){ - return new PubSocket(); +SubSocket::~SubSocket(){ + for (int fd : fake_fds) { + if (fd >= 0) ::close(fd); + } + if (fake_state != nullptr) { + munmap(fake_state, sizeof(EventState)); + } + + if (q != NULL){ + msgq_close_queue(q); + delete q; + } +} + +int PubSocket::connect(std::string endpoint, size_t segment_size){ + q = new msgq_queue_t; + size_t size = segment_size > 0 ? segment_size : DEFAULT_SEGMENT_SIZE; + int r = msgq_new_queue(q, endpoint.c_str(), size); + if (r != 0){ + return r; + } + + msgq_init_publisher(q); + + return 0; +} + +int PubSocket::sendMessage(Message *message){ + msgq_msg_t msg; + msg.data = message->getData(); + msg.size = message->getSize(); + + return msgq_msg_send(&msg, q); +} + +int PubSocket::send(char *data, size_t size){ + msgq_msg_t msg; + msg.data = data; + msg.size = size; + + return msgq_msg_send(&msg, q); } -PubSocket * PubSocket::create(Context * context, std::string endpoint, bool check_endpoint, size_t segment_size){ - PubSocket *s = PubSocket::create(); - int r = s->connect(context, endpoint, check_endpoint, segment_size); +bool PubSocket::all_readers_updated() { + return msgq_all_readers_updated(q); +} + +PubSocket * PubSocket::create(std::string endpoint, size_t segment_size){ + PubSocket *s = new PubSocket(); + int r = s->connect(endpoint, segment_size); if (r == 0) { return s; @@ -54,18 +201,40 @@ PubSocket * PubSocket::create(Context * context, std::string endpoint, bool chec } } -Poller * Poller::create(){ - if (messaging_use_fake()) { - return new FakePoller(); +PubSocket::~PubSocket(){ + if (q != NULL){ + msgq_close_queue(q); + delete q; } - return new MSGQPoller(); } -Poller * Poller::create(std::vector sockets){ - Poller * p = Poller::create(); +Poller::Poller() : fake(messaging_use_fake()) {} +Poller::Poller(std::vector sockets) : Poller() { for (auto s : sockets){ - p->registerSocket(s); + registerSocket(s); + } +} + +void Poller::registerSocket(SubSocket * socket){ + sockets.push_back(socket); + polls.push_back({socket->getQueue(), 0}); +} + +std::vector Poller::poll(int timeout){ + // in fake mode, the sockets block in receive() until the test releases them + if (fake) { + return sockets; } - return p; + + std::vector r; + + msgq_poll(polls.data(), polls.size(), timeout); + for (size_t i = 0; i < polls.size(); i++){ + if (polls[i].revents){ + r.push_back(sockets[i]); + } + } + + return r; } diff --git a/msgq/ipc.h b/msgq/ipc.h index 62a13e442..21a0f3226 100644 --- a/msgq/ipc.h +++ b/msgq/ipc.h @@ -1,25 +1,17 @@ #pragma once #include -#include #include #include -#include #include - +#include "msgq/msgq.h" #ifdef __APPLE__ #define CLOCK_BOOTTIME CLOCK_MONOTONIC #endif -#define MSG_MULTIPLE_PUBLISHERS 100 - -class Context { -public: - static Context * create(); - ~Context(){} -}; +struct EventState; class Message { private: @@ -35,35 +27,46 @@ class Message { ~Message(); }; - class SubSocket { +private: + msgq_queue_t * q = nullptr; + int timeout = -1; + + // fake mode (CEREAL_FAKE) + EventState *fake_state = nullptr; + int fake_fds[2] = {-1, -1}; + void open_fake_fifos(); + public: - virtual int connect(Context *context, std::string endpoint, std::string address, bool conflate=false, bool check_endpoint=true, size_t segment_size=0) = 0; - virtual void setTimeout(int timeout) = 0; - virtual Message *receive(bool non_blocking=false) = 0; - static SubSocket * create(); - static SubSocket * create(Context * context, std::string endpoint, std::string address="127.0.0.1", bool conflate=false, bool check_endpoint=true, size_t segment_size=0); - virtual ~SubSocket(){} + int connect(std::string endpoint, bool conflate=false, size_t segment_size=0); + void setTimeout(int timeout); + msgq_queue_t * getQueue() {return q;} + Message *receive(bool non_blocking=false); + static SubSocket * create(std::string endpoint, bool conflate=false, size_t segment_size=0); + ~SubSocket(); }; class PubSocket { private: - struct msgq_queue_t * q = nullptr; + msgq_queue_t * q = nullptr; public: - int connect(Context *context, std::string endpoint, bool check_endpoint=true, size_t segment_size=0); + int connect(std::string endpoint, size_t segment_size=0); int sendMessage(Message *message); int send(char *data, size_t size); bool all_readers_updated(); - static PubSocket * create(); - static PubSocket * create(Context * context, std::string endpoint, bool check_endpoint=true, size_t segment_size=0); + static PubSocket * create(std::string endpoint, size_t segment_size=0); ~PubSocket(); }; class Poller { +private: + bool fake; + std::vector sockets; + std::vector polls; + public: - virtual void registerSocket(SubSocket *socket) = 0; - virtual std::vector poll(int timeout) = 0; - static Poller * create(); - static Poller * create(std::vector sockets); - virtual ~Poller(){} + Poller(); + Poller(std::vector sockets); + void registerSocket(SubSocket *socket); + std::vector poll(int timeout); }; diff --git a/msgq/ipc.pxd b/msgq/ipc.pxd deleted file mode 100644 index 0760561d2..000000000 --- a/msgq/ipc.pxd +++ /dev/null @@ -1,68 +0,0 @@ -# distutils: language = c++ -#cython: language_level=3 - -from libcpp.string cimport string -from libcpp.vector cimport vector -from libcpp cimport bool - - -cdef extern from "msgq/impl_fake.h": - cdef cppclass Event: - @staticmethod - int wait_for_one(vector[Event], int) except + - - Event() - Event(int) - void set() - int clear() - void wait(int) except + - bool peek() - int fd() - - cdef cppclass SocketEventHandle: - @staticmethod - void toggle_fake_events(bool) - @staticmethod - void set_fake_prefix(string) - @staticmethod - string fake_prefix() - - SocketEventHandle(string, string, bool) - bool is_enabled() - void set_enabled(bool) - Event recv_called() - Event recv_ready() - - -cdef extern from "msgq/ipc.h": - cdef cppclass Context: - @staticmethod - Context * create() - - cdef cppclass Message: - void init(size_t) - void init(char *, size_t) - void close() - size_t getSize() - char *getData() - - cdef cppclass SubSocket: - @staticmethod - SubSocket * create() nogil - int connect(Context *, string, string, bool, bool, size_t) nogil - Message * receive(bool) nogil - void setTimeout(int) nogil - - cdef cppclass PubSocket: - @staticmethod - PubSocket * create() - int connect(Context *, string, bool, size_t) - int sendMessage(Message *) - int send(char *, size_t) - bool all_readers_updated() - - cdef cppclass Poller: - @staticmethod - Poller * create() - void registerSocket(SubSocket *) - vector[SubSocket*] poll(int) nogil diff --git a/msgq/ipc_py.cc b/msgq/ipc_py.cc new file mode 100644 index 000000000..e221d3501 --- /dev/null +++ b/msgq/ipc_py.cc @@ -0,0 +1,113 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include + +#include "msgq/event.h" +#include "msgq/ipc.h" + +namespace nb = nanobind; + +namespace { + +struct IpcError : std::runtime_error { + using std::runtime_error::runtime_error; +}; + +struct MultiplePublishersError : IpcError { + using IpcError::IpcError; +}; + +void check_error(bool failed, int error, const std::string &endpoint = "") { + if (!failed) return; + std::string message = "Messaging failure " + (endpoint.empty() ? "" : "with " + endpoint) + ": " + std::strerror(error); + if (error == EADDRINUSE) throw MultiplePublishersError(message); + throw IpcError(message); +} + +} // namespace + +NB_MODULE(_ipc, m) { + // Daemon threads can still hold sockets at exit. + nb::set_leak_warnings(false); + + nb::exception ipc_error(m, "IpcError"); + nb::exception(m, "MultiplePublishersError", ipc_error); + + m.def("toggle_fake_events", &SocketEventHandle::toggle_fake_events, nb::arg("enabled")); + m.def("set_fake_prefix", &SocketEventHandle::set_fake_prefix, nb::arg("prefix")); + m.def("get_fake_prefix", &SocketEventHandle::fake_prefix); + m.def("delete_fake_prefix", [] { SocketEventHandle::set_fake_prefix(""); }); + m.def("wait_for_one_event", &Event::wait_for_one, nb::arg("events"), nb::arg("timeout") = -1, nb::call_guard()); + + nb::class_(m, "Event") + .def(nb::init<>()) + .def("set", &Event::set) + .def("clear", &Event::clear) + .def("wait", &Event::wait, nb::arg("timeout") = -1, nb::call_guard()) + .def("peek", &Event::peek) + .def_prop_ro("fd", &Event::fd) + .def_prop_ro("ptr", [](Event &event) { return reinterpret_cast(&event); }); + + nb::class_(m, "SocketEventHandle") + .def(nb::init(), nb::arg("endpoint"), nb::arg("identifier"), nb::arg("override")) + .def_prop_rw("enabled", &SocketEventHandle::is_enabled, &SocketEventHandle::set_enabled) + .def_prop_ro("recv_called_event", &SocketEventHandle::recv_called, nb::keep_alive<0, 1>()) + .def_prop_ro("recv_ready_event", &SocketEventHandle::recv_ready, nb::keep_alive<0, 1>()); + + nb::class_(m, "SubSocket") + .def(nb::init<>()) + .def("connect", [](SubSocket &socket, const std::string &name, bool conflate, size_t segment_size) { + nb::gil_scoped_release release; + int result = socket.connect(name, conflate, segment_size); + check_error(result != 0, errno, name); + }, nb::arg("endpoint"), nb::arg("conflate") = false, nb::arg("segment_size") = 0) + .def("setTimeout", &SubSocket::setTimeout, nb::arg("timeout"), nb::call_guard()) + .def("receive", [](SubSocket &socket, bool non_blocking) -> std::optional { + std::unique_ptr message; + { + nb::gil_scoped_release release; + message.reset(socket.receive(non_blocking)); + } + if (!message) return std::nullopt; + return nb::bytes(message->getData(), message->getSize()); + }, nb::arg("non_blocking") = false); + + nb::class_(m, "Poller") + .def(nb::init<>()) + .def("registerSocket", &Poller::registerSocket, nb::arg("socket"), nb::keep_alive<1, 2>()) + .def("poll", &Poller::poll, nb::arg("timeout"), nb::rv_policy::reference, nb::call_guard()); + + nb::class_(m, "PubSocket") + .def(nb::init<>()) + .def("connect", [](PubSocket &socket, const std::string &name, size_t segment_size) { + int result = socket.connect(name, segment_size); + check_error(result != 0, errno, name); + }, nb::arg("endpoint"), nb::arg("segment_size") = 0) + .def("send", [](PubSocket &socket, nb::bytes data) { + int result = socket.send(const_cast(data.c_str()), data.size()); + check_error(result != static_cast(data.size()), errno); + }, nb::arg("data")) + .def("all_readers_updated", &PubSocket::all_readers_updated) + .def("wait_for_readers", [](PubSocket &socket, double timeout, double interval) { + // Python's sleep releases the GIL and handles signals while waiting. + nb::object sleep = nb::module_::import_("time").attr("sleep"); + auto deadline = std::chrono::steady_clock::now() + std::chrono::duration(timeout); + while (std::chrono::steady_clock::now() < deadline) { + if (socket.all_readers_updated()) return; + sleep(interval); + } + PyErr_SetString(PyExc_TimeoutError, "subscriber did not connect"); + throw nb::python_error(); + }, nb::arg("timeout") = 1.0, nb::arg("interval") = 0.001); +} diff --git a/msgq/ipc_pyx.pyx b/msgq/ipc_pyx.pyx deleted file mode 100644 index b16e9702e..000000000 --- a/msgq/ipc_pyx.pyx +++ /dev/null @@ -1,256 +0,0 @@ -# distutils: language = c++ -# cython: c_string_encoding=ascii, language_level=3 - -import time -from libcpp.string cimport string -from libcpp.vector cimport vector -from libcpp cimport bool -from libc cimport errno -from libc.string cimport strerror -from cython.operator import dereference - - -from .ipc cimport Context as cppContext -from .ipc cimport SubSocket as cppSubSocket -from .ipc cimport PubSocket as cppPubSocket -from .ipc cimport Poller as cppPoller -from .ipc cimport Message as cppMessage -from .ipc cimport Event as cppEvent, SocketEventHandle as cppSocketEventHandle - - -class IpcError(Exception): - def __init__(self, endpoint=None): - suffix = f"with {endpoint.decode('utf-8')}" if endpoint else "" - message = f"Messaging failure {suffix}: {strerror(errno.errno).decode('utf-8')}" - super().__init__(message) - - -class MultiplePublishersError(IpcError): - pass - - -def toggle_fake_events(bool enabled): - cppSocketEventHandle.toggle_fake_events(enabled) - - -def set_fake_prefix(string prefix): - cppSocketEventHandle.set_fake_prefix(prefix) - - -def get_fake_prefix(): - return cppSocketEventHandle.fake_prefix() - - -def delete_fake_prefix(): - cppSocketEventHandle.set_fake_prefix(b"") - - -def wait_for_one_event(list events, int timeout=-1): - cdef vector[cppEvent] items - for event in events: - items.push_back(dereference(event.ptr)) - return cppEvent.wait_for_one(items, timeout) - - -cdef class Event: - cdef cppEvent event; - - cdef setEvent(self, cppEvent event): - self.event = event - - def set(self): - self.event.set() - - def clear(self): - return self.event.clear() - - def wait(self, int timeout=-1): - self.event.wait(timeout) - - def peek(self): - return self.event.peek() - - @property - def fd(self): - return self.event.fd() - - @property - def ptr(self): - return &self.event - - -cdef class SocketEventHandle: - cdef cppSocketEventHandle * handle; - - def __cinit__(self, string endpoint, string identifier, bool override): - self.handle = new cppSocketEventHandle(endpoint, identifier, override) - - def __dealloc__(self): - del self.handle - - @property - def enabled(self): - return self.handle.is_enabled() - - @enabled.setter - def enabled(self, bool value): - self.handle.set_enabled(value) - - @property - def recv_called_event(self): - e = Event() - e.setEvent(self.handle.recv_called()) - - return e - - @property - def recv_ready_event(self): - e = Event() - e.setEvent(self.handle.recv_ready()) - - return e - - -cdef class Context: - cdef cppContext * context - - def __cinit__(self): - self.context = cppContext.create() - - def term(self): - del self.context - self.context = NULL - - def __dealloc__(self): - pass - # Deleting the context will hang if sockets are still active - # TODO: Figure out a way to make sure the context is closed last - # del self.context - - -cdef class Poller: - cdef cppPoller * poller - cdef list sub_sockets - - def __cinit__(self): - self.sub_sockets = [] - self.poller = cppPoller.create() - - def __dealloc__(self): - del self.poller - - def registerSocket(self, SubSocket socket): - self.sub_sockets.append(socket) - self.poller.registerSocket(socket.socket) - - def poll(self, timeout): - sockets = [] - cdef int t = timeout - - with nogil: - result = self.poller.poll(t) - - for s in result: - socket = SubSocket() - socket.setPtr(s) - sockets.append(socket) - - return sockets - - -cdef class SubSocket: - cdef cppSubSocket * socket - cdef bool is_owner - - def __cinit__(self): - with nogil: - self.socket = cppSubSocket.create() - - self.is_owner = True - if self.socket == NULL: - raise IpcError - - def __dealloc__(self): - if self.is_owner: - with nogil: - del self.socket - - cdef setPtr(self, cppSubSocket * ptr): - if self.is_owner: - with nogil: - del self.socket - - self.is_owner = False - self.socket = ptr - - def connect(self, Context context, string endpoint, string address=b"127.0.0.1", bool conflate=False, size_t segment_size=0): - cdef int r - with nogil: - r = self.socket.connect(context.context, endpoint, address, conflate, True, segment_size) - - if r != 0: - if errno.errno == errno.EADDRINUSE: - raise MultiplePublishersError(endpoint) - else: - raise IpcError(endpoint) - - def setTimeout(self, int timeout): - with nogil: - self.socket.setTimeout(timeout) - - def receive(self, bool non_blocking=False): - cdef cppMessage *msg - with nogil: - msg = self.socket.receive(non_blocking) - - if msg == NULL: - return None - else: - sz = msg.getSize() - m = msg.getData()[:sz] - with nogil: - del msg - - return m - - -cdef class PubSocket: - cdef cppPubSocket * socket - - def __cinit__(self): - self.socket = cppPubSocket.create() - if self.socket == NULL: - raise IpcError - - def __dealloc__(self): - del self.socket - - def connect(self, Context context, string endpoint, size_t segment_size=0): - r = self.socket.connect(context.context, endpoint, True, segment_size) - - if r != 0: - if errno.errno == errno.EADDRINUSE: - raise MultiplePublishersError(endpoint) - else: - raise IpcError(endpoint) - - def send(self, bytes data): - length = len(data) - r = self.socket.send(data, length) - - if r != length: - if errno.errno == errno.EADDRINUSE: - raise MultiplePublishersError - else: - raise IpcError - - def all_readers_updated(self): - return self.socket.all_readers_updated() - - def wait_for_readers(self, double timeout=1.0, double interval=0.001): - deadline = time.monotonic() + timeout - while time.monotonic() < deadline: - if self.all_readers_updated(): - return - time.sleep(interval) - raise TimeoutError("subscriber did not connect") diff --git a/msgq/logger/logger.h b/msgq/logger/logger.h deleted file mode 100644 index 03336f8ff..000000000 --- a/msgq/logger/logger.h +++ /dev/null @@ -1,21 +0,0 @@ -#pragma once - -#ifdef SWAGLOG -// cppcheck-suppress preprocessorErrorDirective -#include SWAGLOG -#else - -#define CLOUDLOG_DEBUG 10 -#define CLOUDLOG_INFO 20 -#define CLOUDLOG_WARNING 30 -#define CLOUDLOG_ERROR 40 -#define CLOUDLOG_CRITICAL 50 - -#define cloudlog(lvl, fmt, ...) printf(fmt "\n", ## __VA_ARGS__) - -#define LOGD(fmt, ...) cloudlog(CLOUDLOG_DEBUG, fmt, ## __VA_ARGS__) -#define LOG(fmt, ...) cloudlog(CLOUDLOG_INFO, fmt, ## __VA_ARGS__) -#define LOGW(fmt, ...) cloudlog(CLOUDLOG_WARNING, fmt, ## __VA_ARGS__) -#define LOGE(fmt, ...) cloudlog(CLOUDLOG_ERROR, fmt, ## __VA_ARGS__) - -#endif diff --git a/msgq/msgq_tests.cc b/msgq/msgq_tests.cc index b8d6824da..7a9fcc1d7 100644 --- a/msgq/msgq_tests.cc +++ b/msgq/msgq_tests.cc @@ -1,3 +1,6 @@ +#include +#include + #include "msgq/test_runner.h" #include "msgq/msgq.h" @@ -424,3 +427,98 @@ TEST_CASE("1 publisher, 2 subscribers") msgq_msg_close(&msg2); } } + +static int64_t millis_since(std::chrono::steady_clock::time_point start) +{ + return std::chrono::duration_cast(std::chrono::steady_clock::now() - start).count(); +} + +static std::thread send_later(msgq_queue_t *q, int delay_ms) +{ + return std::thread([=]() { + std::this_thread::sleep_for(std::chrono::milliseconds(delay_ms)); + uint64_t value = 1; + msgq_msg_t msg; + msgq_msg_init_data(&msg, (char *)&value, sizeof(value)); + msgq_msg_send(&msg, q); + msgq_msg_close(&msg); + }); +} + +TEST_CASE("msgq_poll times out") +{ + cleanup_test_queue(); + msgq_queue_t writer, reader; + msgq_new_queue(&writer, "test_queue", 1024); + msgq_new_queue(&reader, "test_queue", 1024); + msgq_init_publisher(&writer); + msgq_init_subscriber(&reader); + + msgq_pollitem_t items[] = {{&reader, 0}}; + auto start = std::chrono::steady_clock::now(); + REQUIRE(msgq_poll(items, 1, 100) == 0); + REQUIRE(millis_since(start) >= 100); + REQUIRE(msgq_poll(items, 1, 0) == 0); + + send_later(&writer, 0).join(); + REQUIRE(msgq_poll(items, 1, 0) == 1); + + msgq_close_queue(&writer); + msgq_close_queue(&reader); +} + +TEST_CASE("msgq_poll wakes up on send") +{ + cleanup_test_queue(); + msgq_queue_t writer, reader; + msgq_new_queue(&writer, "test_queue", 1024); + msgq_new_queue(&reader, "test_queue", 1024); + msgq_init_publisher(&writer); + msgq_init_subscriber(&reader); + + msgq_pollitem_t items[] = {{&reader, 0}}; + std::thread sender = send_later(&writer, 50); + auto start = std::chrono::steady_clock::now(); + int n = msgq_poll(items, 1, 5000); + int64_t elapsed = millis_since(start); + sender.join(); + + REQUIRE(n == 1); + REQUIRE(elapsed < 1000); + + msgq_close_queue(&writer); + msgq_close_queue(&reader); +} + +TEST_CASE("msgq_poll multiple queues") +{ + cleanup_test_queue(); + remove("/dev/shm/msgq_test_queue2"); + remove("/tmp/msgq_test_queue2"); + msgq_queue_t writer1, writer2, reader1, reader2; + msgq_new_queue(&writer1, "test_queue", 1024); + msgq_new_queue(&writer2, "test_queue2", 1024); + msgq_new_queue(&reader1, "test_queue", 1024); + msgq_new_queue(&reader2, "test_queue2", 1024); + msgq_init_publisher(&writer1); + msgq_init_publisher(&writer2); + msgq_init_subscriber(&reader1); + msgq_init_subscriber(&reader2); + + msgq_pollitem_t items[] = {{&reader1, 0}, {&reader2, 0}}; + std::thread sender = send_later(&writer2, 50); + auto start = std::chrono::steady_clock::now(); + int n = msgq_poll(items, 2, 5000); + int64_t elapsed = millis_since(start); + sender.join(); + + REQUIRE(n == 1); + REQUIRE(elapsed < 1000); + REQUIRE(items[0].revents == 0); + REQUIRE(items[1].revents == 1); + + msgq_close_queue(&writer1); + msgq_close_queue(&writer2); + msgq_close_queue(&reader1); + msgq_close_queue(&reader2); +} diff --git a/msgq/py.typed b/msgq/py.typed new file mode 100644 index 000000000..e69de29bb diff --git a/msgq/tests/test_poller.py b/msgq/tests/test_poller.py index ba1f6b038..9813d46ac 100644 --- a/msgq/tests/test_poller.py +++ b/msgq/tests/test_poller.py @@ -5,12 +5,10 @@ SERVICE_NAME = 'myService' def poller(): - context = msgq.Context() - p = msgq.Poller() sub = msgq.SubSocket() - sub.connect(context, SERVICE_NAME) + sub.connect(SERVICE_NAME) p.registerSocket(sub) socks = p.poll(10000) @@ -21,10 +19,8 @@ def poller(): class TestPoller(unittest.TestCase): def test_poll_once(self): - context = msgq.Context() - pub = msgq.PubSocket() - pub.connect(context, SERVICE_NAME) + pub.connect(SERVICE_NAME) with concurrent.futures.ThreadPoolExecutor() as e: poll = e.submit(poller) @@ -38,23 +34,19 @@ def test_poll_once(self): result = poll.result() del pub - context.term() assert result == [b"a"] def test_poll_and_create_many_subscribers(self): - context = msgq.Context() - pub = msgq.PubSocket() - pub.connect(context, SERVICE_NAME) + pub.connect(SERVICE_NAME) with concurrent.futures.ThreadPoolExecutor() as e: poll = e.submit(poller) pub.wait_for_readers() - c = msgq.Context() for _ in range(10): - msgq.SubSocket().connect(c, SERVICE_NAME) + msgq.SubSocket().connect(SERVICE_NAME) # Send message pub.send(b"a") @@ -63,34 +55,28 @@ def test_poll_and_create_many_subscribers(self): result = poll.result() del pub - context.term() assert result == [b"a"] def test_multiple_publishers_exception(self): - context = msgq.Context() - with self.assertRaises(msgq.MultiplePublishersError): pub1 = msgq.PubSocket() - pub1.connect(context, SERVICE_NAME) + pub1.connect(SERVICE_NAME) pub2 = msgq.PubSocket() - pub2.connect(context, SERVICE_NAME) + pub2.connect(SERVICE_NAME) pub1.send(b"a") del pub1 del pub2 - context.term() def test_multiple_messages(self): - context = msgq.Context() - pub = msgq.PubSocket() - pub.connect(context, SERVICE_NAME) + pub.connect(SERVICE_NAME) sub = msgq.SubSocket() - sub.connect(context, SERVICE_NAME) + sub.connect(SERVICE_NAME) pub.wait_for_readers() @@ -113,16 +99,13 @@ def test_multiple_messages(self): del pub del sub - context.term() def test_conflate(self): - context = msgq.Context() - pub = msgq.PubSocket() - pub.connect(context, SERVICE_NAME) + pub.connect(SERVICE_NAME) sub = msgq.SubSocket() - sub.connect(context, SERVICE_NAME, conflate=True) + sub.connect(SERVICE_NAME, conflate=True) pub.wait_for_readers() pub.send(b'a') @@ -132,4 +115,3 @@ def test_conflate(self): del pub del sub - context.term() diff --git a/msgq/visionipc/.gitignore b/msgq/visionipc/.gitignore deleted file mode 100644 index fa5a2b9f3..000000000 --- a/msgq/visionipc/.gitignore +++ /dev/null @@ -1,2 +0,0 @@ -visionipc_pyx.cpp -*.so diff --git a/msgq/visionipc/__init__.py b/msgq/visionipc/__init__.py index f93bafcc4..2f30a3627 100644 --- a/msgq/visionipc/__init__.py +++ b/msgq/visionipc/__init__.py @@ -1,4 +1,4 @@ -from msgq.visionipc.visionipc_pyx import VisionBuf, VisionIpcClient, VisionIpcServer, get_endpoint_name +from msgq.visionipc._visionipc import VisionBuf, VisionIpcClient, VisionIpcServer, get_endpoint_name __all__ = [ "VisionBuf", diff --git a/msgq/visionipc/tests/test_visionipc.py b/msgq/visionipc/tests/test_visionipc.py index b938df447..1a78b1cf9 100644 --- a/msgq/visionipc/tests/test_visionipc.py +++ b/msgq/visionipc/tests/test_visionipc.py @@ -1,3 +1,4 @@ +import gc import struct import unittest from typing import Optional @@ -16,24 +17,24 @@ def tearDown(self): self.client = None self.server = None - def setup_vipc(self, name, *stream_types, num_buffers=1, width=100, height=100, conflate=False): - self.server = VisionIpcServer(name) - for stream_type in stream_types: - self.server.create_buffers(stream_type, num_buffers, width, height) - self.server.start_listener() - - if len(stream_types): - self.client = VisionIpcClient(name, stream_types[0], conflate) - assert self.client.connect(True) - else: - self.client = None - - return self.server, self.client + def setup_vipc( + self, name: str, stream_type: int, *other_stream_types: int, + num_buffers: int = 1, width: int = 100, height: int = 100, conflate: bool = False, + ) -> tuple[VisionIpcServer, VisionIpcClient, int]: + self.server = server = VisionIpcServer(name) + for stream in (stream_type, *other_stream_types): + server.create_buffers(stream, num_buffers, width, height) + server.start_listener() + + self.client = client = VisionIpcClient(name, stream_type, conflate) + assert client.connect(True) + buffer_len = client.buffer_len + assert buffer_len is not None + return server, client, buffer_len def test_connect(self): - self.setup_vipc("camerad", 0) - assert self.client is not None - assert self.client.is_connected() + _, client, _ = self.setup_vipc("camerad", 0) + assert client.is_connected() def test_available_streams(self): stream_types = (0, 2) @@ -43,61 +44,99 @@ def test_available_streams(self): def test_buffers(self): width, height, num_buffers = 100, 200, 5 - self.setup_vipc("camerad", 0, num_buffers=num_buffers, width=width, height=height) - assert self.client is not None - assert self.client.width == width - assert self.client.height == height - assert self.client.buffer_len is not None and self.client.buffer_len > 0 - assert self.client.num_buffers == num_buffers + _, client, buffer_len = self.setup_vipc("camerad", 0, num_buffers=num_buffers, width=width, height=height) + assert client.width == width + assert client.height == height + assert buffer_len > 0 + assert client.num_buffers == num_buffers def test_send_single_buffer(self): - self.setup_vipc("camerad", 0) - assert self.server is not None - assert self.client is not None - assert self.client.buffer_len is not None - buf = bytearray(self.client.buffer_len) + server, client, buffer_len = self.setup_vipc("camerad", 0) + buf = bytearray(buffer_len) struct.pack_into("registerSocket(sock); } @@ -41,7 +39,7 @@ bool VisionIpcClient::connect(bool blocking) { // Cleanup old buffers on reconnect for (size_t i = 0; i < num_buffers; i++) { if (buffers[i].free() != 0) { - LOGE("Failed to free buffer %zu", i); + std::cerr << "Failed to free buffer " << i << std::endl; } } @@ -118,7 +116,7 @@ VisionBuf * VisionIpcClient::recv(VisionIpcBufExtra * extra, const int timeout_m } if (buf->sync(VISIONBUF_SYNC_TO_DEVICE) != 0) { - LOGE("Failed to sync buffer"); + std::cerr << "Failed to sync buffer" << std::endl; } delete r; @@ -152,11 +150,10 @@ std::set VisionIpcClient::getAvailableStreams(const std::strin VisionIpcClient::~VisionIpcClient() { for (size_t i = 0; i < num_buffers; i++) { if (buffers[i].free() != 0) { - LOGE("Failed to free buffer %zu", i); + std::cerr << "Failed to free buffer " << i << std::endl; } } delete sock; delete poller; - delete msg_ctx; } diff --git a/msgq/visionipc/visionipc_client.h b/msgq/visionipc/visionipc_client.h index f7ecc7450..117351700 100644 --- a/msgq/visionipc/visionipc_client.h +++ b/msgq/visionipc/visionipc_client.h @@ -10,7 +10,6 @@ class VisionIpcClient { private: std::string name; - Context * msg_ctx; SubSocket * sock; Poller * poller; diff --git a/msgq/visionipc/visionipc_py.cc b/msgq/visionipc/visionipc_py.cc new file mode 100644 index 000000000..d82c94052 --- /dev/null +++ b/msgq/visionipc/visionipc_py.cc @@ -0,0 +1,109 @@ +#include +#include +#include +#include + +#include +#include +#include +#include +#include + +#include "msgq/visionipc/visionbuf.h" +#include "msgq/visionipc/visionipc.h" +#include "msgq/visionipc/visionipc_client.h" +#include "msgq/visionipc/visionipc_server.h" + +namespace nb = nanobind; + +namespace { + +struct PythonVisionIpcClient : VisionIpcClient { + using VisionIpcClient::VisionIpcClient; + VisionIpcBufExtra extra = {}; +}; + +using Buffer = nb::ndarray, nb::device::cpu>; +using InputBuffer = nb::ndarray, nb::device::cpu>; + +} // namespace + +NB_MODULE(_visionipc, m) { + // Daemon threads can still hold clients and servers at exit. + nb::set_leak_warnings(false); + + m.def("get_endpoint_name", &get_endpoint_name, nb::arg("name"), nb::arg("stream")); + + nb::class_(m, "VisionBuf") + .def_prop_ro("data", [](VisionBuf &buf) { return Buffer(buf.addr, {buf.len}); }, nb::rv_policy::reference_internal, + nb::sig("def data(self) -> memoryview")) + .def_ro("width", &VisionBuf::width) + .def_ro("height", &VisionBuf::height) + .def_ro("stride", &VisionBuf::stride) + .def_ro("uv_offset", &VisionBuf::uv_offset) + .def_ro("idx", &VisionBuf::idx) + .def_ro("fd", &VisionBuf::fd) + .def_prop_ro("frame_id", &VisionBuf::get_frame_id); + + nb::class_(m, "VisionIpcServer") + .def(nb::init(), nb::arg("name")) + .def("create_buffers", &VisionIpcServer::create_buffers, nb::arg("tp"), nb::arg("num_buffers"), nb::arg("width"), nb::arg("height")) + .def("create_buffers_with_sizes", &VisionIpcServer::create_buffers_with_sizes, + nb::arg("tp"), nb::arg("num_buffers"), nb::arg("width"), nb::arg("height"), nb::arg("size"), nb::arg("stride"), nb::arg("uv_offset")) + .def("send", [](VisionIpcServer &server, VisionStreamType tp, InputBuffer data, + uint32_t frame_id, uint64_t timestamp_sof, uint64_t timestamp_eof) { + VisionBuf *buf = server.get_buffer(tp); + if (buf->len != data.size()) { + PyErr_SetString(PyExc_AssertionError, "buffer size does not match frame size"); + throw nb::python_error(); + } + // The old binding accepted strided byte views as well as contiguous data. + auto *source = data.data(); + if (data.stride(0) == 1) { + std::memcpy(buf->addr, source, data.size()); + } else { + auto *target = static_cast(buf->addr); + for (size_t i = 0; i < data.size(); ++i) target[i] = source[static_cast(i) * data.stride(0)]; + } + buf->set_frame_id(frame_id); + VisionIpcBufExtra extra = {frame_id, timestamp_sof, timestamp_eof, false}; + server.send(buf, &extra, false); + }, nb::arg("tp"), nb::arg("data").noconvert(), nb::arg("frame_id") = 0, nb::arg("timestamp_sof") = 0, nb::arg("timestamp_eof") = 0, + nb::sig("def send(self, tp: int, data: bytes | bytearray | memoryview, frame_id: int = 0, timestamp_sof: int = 0, timestamp_eof: int = 0) -> None")) + .def("start_listener", &VisionIpcServer::start_listener); + + nb::class_(m, "VisionIpcClient") + .def(nb::init(), nb::arg("name"), nb::arg("stream"), nb::arg("conflate")) + .def_prop_ro("width", [](PythonVisionIpcClient &client) -> std::optional { + return client.num_buffers ? std::make_optional(client.buffers[0].width) : std::nullopt; + }) + .def_prop_ro("height", [](PythonVisionIpcClient &client) -> std::optional { + return client.num_buffers ? std::make_optional(client.buffers[0].height) : std::nullopt; + }) + .def_prop_ro("stride", [](PythonVisionIpcClient &client) -> std::optional { + return client.num_buffers ? std::make_optional(client.buffers[0].stride) : std::nullopt; + }) + .def_prop_ro("uv_offset", [](PythonVisionIpcClient &client) -> std::optional { + return client.num_buffers ? std::make_optional(client.buffers[0].uv_offset) : std::nullopt; + }) + .def_prop_ro("buffer_len", [](PythonVisionIpcClient &client) -> std::optional { + return client.num_buffers ? std::make_optional(client.buffers[0].len) : std::nullopt; + }) + .def_prop_ro("num_buffers", [](PythonVisionIpcClient &client) { return client.num_buffers; }) + .def_prop_ro("frame_id", [](PythonVisionIpcClient &client) { return client.extra.frame_id; }) + .def_prop_ro("timestamp_sof", [](PythonVisionIpcClient &client) { return client.extra.timestamp_sof; }) + .def_prop_ro("timestamp_eof", [](PythonVisionIpcClient &client) { return client.extra.timestamp_eof; }) + .def_prop_ro("valid", [](PythonVisionIpcClient &client) { return client.extra.valid; }) + .def("recv", [](PythonVisionIpcClient &client, int timeout_ms) { + nb::gil_scoped_release release; + return client.recv(&client.extra, timeout_ms); + }, nb::arg("timeout_ms") = 100, nb::rv_policy::reference_internal, + nb::sig("def recv(self, timeout_ms: int = 100) -> VisionBuf | None")) + // Holds the GIL: connect remaps the buffers other threads may be reading. + .def("connect", [](PythonVisionIpcClient &client, bool blocking) { + return client.connect(blocking); + }, nb::arg("blocking")) + .def("is_connected", [](PythonVisionIpcClient &client) { return client.is_connected(); }) + .def_static("available_streams", &VisionIpcClient::getAvailableStreams, nb::arg("name"), nb::arg("block"), + nb::call_guard()); +} diff --git a/msgq/visionipc/visionipc_pyx.pxd b/msgq/visionipc/visionipc_pyx.pxd deleted file mode 100644 index 6e2d5ed05..000000000 --- a/msgq/visionipc/visionipc_pyx.pxd +++ /dev/null @@ -1,10 +0,0 @@ -# distutils: language = c++ -#cython: language_level=3 - -from .visionipc cimport VisionBuf as cppVisionBuf - -cdef class VisionBuf: - cdef cppVisionBuf * buf - - @staticmethod - cdef create(cppVisionBuf*) diff --git a/msgq/visionipc/visionipc_pyx.pyx b/msgq/visionipc/visionipc_pyx.pyx deleted file mode 100644 index a274d4f7a..000000000 --- a/msgq/visionipc/visionipc_pyx.pyx +++ /dev/null @@ -1,163 +0,0 @@ -# distutils: language = c++ -# cython: c_string_encoding=ascii, language_level=3 - -from libc.string cimport memcpy -from libc.stdint cimport uint32_t, uint64_t -from libcpp cimport bool -from libcpp.string cimport string - -from .visionipc cimport VisionIpcServer as cppVisionIpcServer -from .visionipc cimport VisionIpcClient as cppVisionIpcClient -from .visionipc cimport VisionBuf as cppVisionBuf -from .visionipc cimport VisionIpcBufExtra, VisionStreamType -from .visionipc cimport get_endpoint_name as cpp_get_endpoint_name - - -def get_endpoint_name(string name, VisionStreamType stream): - return cpp_get_endpoint_name(name, stream).decode('utf-8') - - -cdef class VisionBuf: - @staticmethod - cdef create(cppVisionBuf * cbuf): - buf = VisionBuf() - buf.buf = cbuf - return buf - - @property - def data(self): - cdef unsigned char[:] data = self.buf.addr - return memoryview(data) - - @property - def width(self): - return self.buf.width - - @property - def height(self): - return self.buf.height - - @property - def stride(self): - return self.buf.stride - - @property - def uv_offset(self): - return self.buf.uv_offset - - @property - def idx(self): - return self.buf.idx - - @property - def fd(self): - return self.buf.fd - - @property - def frame_id(self): - return self.buf.get_frame_id() - - -cdef class VisionIpcServer: - cdef cppVisionIpcServer * server - - def __init__(self, string name): - self.server = new cppVisionIpcServer(name) - - def create_buffers(self, VisionStreamType tp, size_t num_buffers, size_t width, size_t height): - self.server.create_buffers(tp, num_buffers, width, height) - - def create_buffers_with_sizes(self, VisionStreamType tp, size_t num_buffers, size_t width, size_t height, size_t size, size_t stride, size_t uv_offset): - self.server.create_buffers_with_sizes(tp, num_buffers, width, height, size, stride, uv_offset) - - def send(self, VisionStreamType tp, const unsigned char[:] data, uint32_t frame_id=0, uint64_t timestamp_sof=0, uint64_t timestamp_eof=0): - cdef cppVisionBuf * buf = self.server.get_buffer(tp) - - # Populate buffer - assert buf.len == len(data) - memcpy(buf.addr, &data[0], len(data)) - buf.set_frame_id(frame_id) - - cdef VisionIpcBufExtra extra - extra.frame_id = frame_id - extra.timestamp_sof = timestamp_sof - extra.timestamp_eof = timestamp_eof - extra.valid = False - - self.server.send(buf, &extra, False) - - def start_listener(self): - self.server.start_listener() - - def __dealloc__(self): - del self.server - - -cdef class VisionIpcClient: - cdef cppVisionIpcClient * client - cdef VisionIpcBufExtra extra - - def __cinit__(self, string name, VisionStreamType stream, bool conflate): - self.client = new cppVisionIpcClient(name, stream, conflate) - - def __dealloc__(self): - del self.client - - @property - def width(self): - return self.client.buffers[0].width if self.client.num_buffers else None - - @property - def height(self): - return self.client.buffers[0].height if self.client.num_buffers else None - - @property - def stride(self): - return self.client.buffers[0].stride if self.client.num_buffers else None - - @property - def uv_offset(self): - return self.client.buffers[0].uv_offset if self.client.num_buffers else None - - @property - def buffer_len(self): - return self.client.buffers[0].len if self.client.num_buffers else None - - @property - def num_buffers(self): - return self.client.num_buffers - - @property - def frame_id(self): - return self.extra.frame_id - - @property - def timestamp_sof(self): - return self.extra.timestamp_sof - - @property - def timestamp_eof(self): - return self.extra.timestamp_eof - - @property - def valid(self): - return self.extra.valid - - def recv(self, int timeout_ms=100): - cdef cppVisionBuf * buf - # release the GIL, this can block for timeout_ms - with nogil: - buf = self.client.recv(&self.extra, timeout_ms) - if not buf: - return None - return VisionBuf.create(buf) - - def connect(self, bool blocking): - return self.client.connect(blocking) - - def is_connected(self): - return self.client.is_connected() - - @staticmethod - def available_streams(string name, bool block): - return cppVisionIpcClient.getAvailableStreams(name, block) diff --git a/msgq/visionipc/visionipc_server.cc b/msgq/visionipc/visionipc_server.cc index a59ea6169..0c7181cbb 100644 --- a/msgq/visionipc/visionipc_server.cc +++ b/msgq/visionipc/visionipc_server.cc @@ -12,7 +12,6 @@ #include "msgq/visionipc/visionipc.h" #include "msgq/visionipc/visionipc_server.h" -#include "msgq/logger/logger.h" std::string get_endpoint_name(std::string name, VisionStreamType type){ return "visionipc_" + name + "_" + std::to_string(type); @@ -27,8 +26,6 @@ std::string get_ipc_path(const std::string& name) { } VisionIpcServer::VisionIpcServer(std::string name_) : name(name_) { - msg_ctx = Context::create(); - std::random_device rd("/dev/urandom"); std::uniform_int_distribution distribution(0, std::numeric_limits::max()); server_id = distribution(rd); @@ -61,7 +58,7 @@ void VisionIpcServer::create_buffers_with_sizes(VisionStreamType type, size_t nu cur_idx[type] = 0; // Create msgq publisher for each of the `name` + type combos - sockets[type] = PubSocket::create(msg_ctx, get_endpoint_name(name, type), false); + sockets[type] = PubSocket::create(get_endpoint_name(name, type)); } @@ -71,8 +68,6 @@ void VisionIpcServer::start_listener(){ void VisionIpcServer::listener(){ - LOGD("Starting listener for: %s", name.c_str()); - const std::string ipc_path = get_ipc_path(name); int sock = ipc_bind(ipc_path.c_str()); assert(sock >= 0); @@ -144,7 +139,6 @@ void VisionIpcServer::listener(){ close(fd); } - LOGD("Stopping listener for: %s", name.c_str()); close(sock); unlink(ipc_path.c_str()); } @@ -154,7 +148,7 @@ void VisionIpcServer::listener(){ VisionBuf * VisionIpcServer::get_buffer(VisionStreamType type, int idx){ // Do we want to keep track if the buffer has been sent out yet and warn user? assert(buffers.count(type)); - auto b = buffers[type]; + const auto &b = buffers[type]; if (idx < 0) { idx = cur_idx[type]++ % b.size(); } else { @@ -167,7 +161,7 @@ VisionBuf * VisionIpcServer::get_buffer(VisionStreamType type, int idx){ void VisionIpcServer::send(VisionBuf * buf, VisionIpcBufExtra * extra, bool sync){ if (sync) { if (buf->sync(VISIONBUF_SYNC_FROM_DEVICE) != 0) { - LOGE("Failed to sync buffer"); + std::cerr << "Failed to sync buffer" << std::endl; } } assert(buffers.count(buf->type)); @@ -196,7 +190,7 @@ VisionIpcServer::~VisionIpcServer(){ for (auto const& [type, buf] : buffers) { for (VisionBuf* b : buf){ if (b->free() != 0) { - LOGE("Failed to free buffer"); + std::cerr << "Failed to free buffer" << std::endl; } delete b; } @@ -206,5 +200,4 @@ VisionIpcServer::~VisionIpcServer(){ for (auto const& [type, sock] : sockets) { delete sock; } - delete msg_ctx; } diff --git a/msgq/visionipc/visionipc_server.h b/msgq/visionipc/visionipc_server.h index ddac145d2..86c8ec70b 100644 --- a/msgq/visionipc/visionipc_server.h +++ b/msgq/visionipc/visionipc_server.h @@ -22,7 +22,6 @@ class VisionIpcServer { std::map > cur_idx; std::map > buffers; - Context * msg_ctx; std::map sockets; void listener(void); diff --git a/pyproject.toml b/pyproject.toml index 8bb946933..3b94dfc4c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,10 +1,14 @@ +[build-system] +requires = ["setuptools>=77", "nanobind>=2.12", "scons"] +build-backend = "setuptools.build_meta" + [project] -name = "msgq" -version = "0.0.1" -description = "Lock-free IPC pub/sub message queue" +name = "msgq-ipc" +version = "1.1" +description = "High-performance pub/sub messaging, made simple. For Python, C, and C++." readme = "README.md" requires-python = ">=3.11" -license = {text = "MIT"} +license = "MIT" authors = [{name = "comma.ai"}] classifiers = [ "Natural Language :: English", @@ -13,11 +17,18 @@ classifiers = [ ] dependencies = [] +[project.urls] +Repository = "https://github.com/commaai/msgq" + [project.optional-dependencies] +# for building from source with SCons +build = [ + "nanobind>=2.12", + "scons", +] dev = [ + "msgq-ipc[build]", "setuptools", # for distutils - "Cython", - "scons", "ruff", "cppcheck", "cpplint", @@ -26,8 +37,23 @@ dev = [ "lefthook", ] +[tool.setuptools] +include-package-data = false + +[tool.setuptools.packages.find] +include = ["msgq", "msgq.*"] +namespaces = false + [tool.setuptools.package-data] -"*" = ["*.so"] +"msgq" = ["*.pyi", "py.typed", "*.h"] +"msgq.visionipc" = ["*.pyi", "*.h"] + +[tool.cibuildwheel] +build = "cp311-* cp312-* cp313-* cp314-*" +test-command = "python -m unittest discover -s msgq" + +[tool.cibuildwheel.macos.environment] +MACOSX_DEPLOYMENT_TARGET = "10.13" # https://beta.ruff.rs/docs/configuration/#using-pyprojecttoml [tool.ruff] @@ -37,10 +63,3 @@ lint.flake8-implicit-str-concat.allow-multiline=false line-length = 160 target-version="py311" - -[tool.ty.src] -exclude = ["site_scons/"] - -[tool.ty.rules] -# Cython modules are compiled at build time, not available for static analysis -unresolved-import = "ignore" diff --git a/setup.py b/setup.py new file mode 100644 index 000000000..ff6cb0b4d --- /dev/null +++ b/setup.py @@ -0,0 +1,30 @@ +import os +from pathlib import Path +import subprocess +import sys +import sysconfig + +from setuptools import Extension, setup +from setuptools.command.build_ext import build_ext + + +class SConsBuildExt(build_ext): + def build_extensions(self): + subprocess.check_call([sys.executable, "-m", "SCons", "--minimal", "-j", str(self.parallel or os.cpu_count() or 1)]) + for ext in self.extensions: + module = Path(*ext.name.split(".")) + source = module.with_suffix(sysconfig.get_config_var("EXT_SUFFIX")) + target = Path(self.get_ext_fullpath(ext.name)) + target.parent.mkdir(parents=True, exist_ok=True) + self.copy_file(str(source), str(target)) + stub = module.with_suffix(".pyi") + self.copy_file(str(stub), str(target.parent / stub.name)) + + +setup( + ext_modules=[ + Extension("msgq._ipc", sources=[]), + Extension("msgq.visionipc._visionipc", sources=[]), + ], + cmdclass={"build_ext": SConsBuildExt}, +) diff --git a/site_scons/site_tools/cython.py b/site_scons/site_tools/cython.py deleted file mode 100644 index c29147553..000000000 --- a/site_scons/site_tools/cython.py +++ /dev/null @@ -1,72 +0,0 @@ -import re -import SCons -from SCons.Action import Action -from SCons.Scanner import Scanner - -pyx_from_import_re = re.compile(r'^from\s+(\S+)\s+cimport', re.M) -pyx_import_re = re.compile(r'^cimport\s+(\S+)', re.M) -cdef_import_re = re.compile(r'^cdef extern from\s+.(\S+).:', re.M) - - -def pyx_scan(node, env, path, arg=None): - contents = node.get_text_contents() - - # from cimport ... - matches = pyx_from_import_re.findall(contents) - # cimport - matches += pyx_import_re.findall(contents) - - # Modules can be either .pxd or .pyx files - files = [m.replace('.', '/') + '.pxd' for m in matches] - files += [m.replace('.', '/') + '.pyx' for m in matches] - - # cdef extern from - files += cdef_import_re.findall(contents) - - # Handle relative imports - cur_dir = str(node.get_dir()) - files = [cur_dir + f if f.startswith('/') else f for f in files] - - # Filter out non-existing files (probably system imports) - files = [f for f in files if env.File(f).exists()] - return env.File(files) - - -pyxscanner = Scanner(function=pyx_scan, skeys=['.pyx', '.pxd'], recursive=True) -cythonAction = Action("$CYTHONCOM") - - -def create_builder(env): - try: - cython = env['BUILDERS']['Cython'] - except KeyError: - cython = SCons.Builder.Builder( - action=cythonAction, - emitter={}, - suffix=cython_suffix_emitter, - single_source=1 - ) - env.Append(SCANNERS=pyxscanner) - env['BUILDERS']['Cython'] = cython - return cython - -def cython_suffix_emitter(env, source): - return "$CYTHONCFILESUFFIX" - -def generate(env): - env["CYTHON"] = "cythonize" - env["CYTHONCOM"] = "$CYTHON $CYTHONFLAGS $SOURCE" - env["CYTHONCFILESUFFIX"] = ".cpp" - - c_file, _ = SCons.Tool.createCFileBuilders(env) - - c_file.suffix['.pyx'] = cython_suffix_emitter - c_file.add_action('.pyx', cythonAction) - - c_file.suffix['.py'] = cython_suffix_emitter - c_file.add_action('.py', cythonAction) - - create_builder(env) - -def exists(env): - return True diff --git a/test.sh b/test.sh index d91afb035..2faa3773f 100755 --- a/test.sh +++ b/test.sh @@ -12,6 +12,16 @@ scons -j8 # *** lint + test *** lefthook run test +# *** test installed package outside the checkout *** +( + TEST_DIR=$(mktemp -d) + trap 'rm -rf "$TEST_DIR"' EXIT + uv venv --python "$DIR/.venv/bin/python" "$TEST_DIR/.venv" + uv pip install --python "$TEST_DIR/.venv/bin/python" "$DIR" + cd "$("$TEST_DIR/.venv/bin/python" -c "import sysconfig; print(sysconfig.get_path('purelib'))")" + "$TEST_DIR/.venv/bin/python" -m unittest discover -t . -s msgq +) + # *** all done *** GREEN='\033[0;32m' NC='\033[0m'