diff --git a/README.md b/README.md
index b5e0a163c..1609f32a8 100644
--- a/README.md
+++ b/README.md
@@ -6,14 +6,18 @@
[](https://discord.comma.ai)
+[](https://pypi.org/project/msgq-ipc/)
+[](https://deepwiki.com/commaai/msgq)
[](https://github.com/commaai/msgq/actions/workflows/tests.yml)
-[](LICENSE)
+[](https://github.com/commaai/msgq/blob/master/LICENSE)
@@ -21,24 +25,22 @@
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.
-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).
-
- 1 KiB cross-process ping-pong on x86 Linux. Benchmark script.
+ 1 KiB cross-process ping-pong on x86 Linux. Benchmark script.
## Quickstart
```sh
-python -m pip install git+https://github.com/commaai/msgq.git
+python -m pip install msgq-ipc
```
-From a local checkout, run the [publisher](examples/publisher.py) and [subscriber](examples/subscriber.py) in separate terminals:
+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 examples/publisher.py # terminal 1
-python examples/subscriber.py # terminal 2
+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.
@@ -54,15 +56,65 @@ 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](LICENSE).
+MSGQ is available under the [MIT License](https://github.com/commaai/msgq/blob/master/LICENSE).
-## Under the hood
+
+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:
@@ -113,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/examples/publisher.py b/examples/publisher.py
deleted file mode 100644
index ae3bcd01d..000000000
--- a/examples/publisher.py
+++ /dev/null
@@ -1,16 +0,0 @@
-import time
-
-import msgq
-
-
-publisher = msgq.pub_sock("msgq_example")
-
-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/examples/subscriber.py b/examples/subscriber.py
deleted file mode 100644
index 4d48c4822..000000000
--- a/examples/subscriber.py
+++ /dev/null
@@ -1,13 +0,0 @@
-import msgq
-
-
-subscriber = msgq.sub_sock("msgq_example", 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/examples/__init__.py b/msgq/examples/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/examples/benchmark.py b/msgq/examples/benchmark.py
similarity index 99%
rename from examples/benchmark.py
rename to msgq/examples/benchmark.py
index 808fd2450..209793525 100644
--- a/examples/benchmark.py
+++ b/msgq/examples/benchmark.py
@@ -2,7 +2,7 @@
# requires-python = ">=3.11"
# dependencies = ["msgq-ipc>=1.0", "pyzmq", "eclipse-zenoh>=1.10", "lcm", "matplotlib"]
# [tool.uv.sources]
-# msgq-ipc = { path = ".." }
+# msgq-ipc = { path = "../.." }
# [tool.ty.rules]
# unresolved-import = "ignore"
# ///
@@ -234,31 +234,6 @@ def measure(name, size):
connection.close()
-def 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)
-
-
def plot_results(results, path):
from matplotlib.ticker import EngFormatter, MaxNLocator
@@ -293,4 +268,25 @@ def plot_results(results, path):
if __name__ == "__main__":
- 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/pyproject.toml b/pyproject.toml
index 6c2c3aa7d..c7e46e37d 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,8 +4,8 @@ build-backend = "setuptools.build_meta"
[project]
name = "msgq-ipc"
-version = "1.0"
-description = "Lock-free IPC pub/sub message queue"
+version = "1.1"
+description = "High-performance pub/sub messaging, made simple. For Python, C, and C++."
readme = "README.md"
requires-python = ">=3.11"
license = "MIT"