From 321e64e39b572c4111865c66b3782acf49eb1c48 Mon Sep 17 00:00:00 2001 From: Adeeb Shihadeh Date: Sun, 6 Sep 2026 12:49:42 -0700 Subject: [PATCH 1/9] we're on pypi! --- README.md | 3 ++- pyproject.toml | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index b5e0a163c..41e2eeb5c 100644 --- a/README.md +++ b/README.md @@ -12,6 +12,7 @@ [![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/) [![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)](LICENSE) @@ -31,7 +32,7 @@ MSGQ is a generic high performance IPC pub sub system with a single publisher an ## 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: diff --git a/pyproject.toml b/pyproject.toml index 6c2c3aa7d..385b5f267 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta" [project] name = "msgq-ipc" version = "1.0" -description = "Lock-free IPC pub/sub message queue" +description = "High-performance pub/sub messaging, made simple. For Python, C, and C++." readme = "README.md" requires-python = ">=3.11" license = "MIT" From be67128cf88bcf940b767f744000ddc9ed7a1fc4 Mon Sep 17 00:00:00 2001 From: Adeeb Shihadeh Date: Sun, 6 Sep 2026 12:57:19 -0700 Subject: [PATCH 2/9] move examples into the package --- README.md | 15 +++++++++------ examples/publisher.py | 16 ---------------- examples/subscriber.py | 13 ------------- msgq/examples/__init__.py | 0 {examples => msgq/examples}/benchmark.py | 2 +- msgq/examples/publisher.py | 21 +++++++++++++++++++++ msgq/examples/subscriber.py | 18 ++++++++++++++++++ 7 files changed, 49 insertions(+), 36 deletions(-) delete mode 100644 examples/publisher.py delete mode 100644 examples/subscriber.py create mode 100644 msgq/examples/__init__.py rename {examples => msgq/examples}/benchmark.py (99%) create mode 100644 msgq/examples/publisher.py create mode 100644 msgq/examples/subscriber.py diff --git a/README.md b/README.md index 41e2eeb5c..807a5d182 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,7 @@

Quickstart · - Examples + Examples · Discord

@@ -26,7 +26,7 @@ MSGQ is a generic high performance IPC pub sub system with a single publisher an

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

## Quickstart @@ -35,11 +35,11 @@ MSGQ is a generic high performance IPC pub sub system with a single publisher an 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](msgq/examples/publisher.py) and [subscriber](msgq/examples/subscriber.py) in separate terminals: ```sh -python examples/publisher.py # terminal 1 -python examples/subscriber.py # terminal 2 +python -m msgq.examples.publisher # terminal 1 +python -m msgq.examples.subscriber # terminal 2 ``` The subscriber prints `Hello from MSGQ!` once per second. @@ -63,7 +63,8 @@ Issues and pull requests are welcome on [GitHub](https://github.com/commaai/msgq MSGQ is available under the [MIT License](LICENSE). -## Under the hood +
+Under the hood ### Storage The storage for the queue consists of an area of metadata, and the actual buffer. The metadata contains: @@ -114,3 +115,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..d62c963af 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" # /// diff --git a/msgq/examples/publisher.py b/msgq/examples/publisher.py new file mode 100644 index 000000000..18d897872 --- /dev/null +++ b/msgq/examples/publisher.py @@ -0,0 +1,21 @@ +import time + +import msgq + + +def main(): + 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 + + +if __name__ == "__main__": + main() diff --git a/msgq/examples/subscriber.py b/msgq/examples/subscriber.py new file mode 100644 index 000000000..cbf7a81d8 --- /dev/null +++ b/msgq/examples/subscriber.py @@ -0,0 +1,18 @@ +import msgq + + +def main(): + 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 + + +if __name__ == "__main__": + main() From eac284e6a3a09d4baf5b07a3063288bcffd6ec77 Mon Sep 17 00:00:00 2001 From: Adeeb Shihadeh Date: Sun, 6 Sep 2026 12:59:32 -0700 Subject: [PATCH 3/9] abs links for pypi --- README.md | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index 807a5d182..2ea184c78 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,7 @@

Quickstart · - Examples + Examples · Discord

@@ -14,7 +14,7 @@ [![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/) [![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)](LICENSE) +[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://github.com/commaai/msgq/blob/master/LICENSE) @@ -26,7 +26,7 @@ MSGQ is a generic high performance IPC pub sub system with a single publisher an

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

## Quickstart @@ -35,7 +35,7 @@ MSGQ is a generic high performance IPC pub sub system with a single publisher an python -m pip install msgq-ipc ``` -Run the included [publisher](msgq/examples/publisher.py) and [subscriber](msgq/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) in separate terminals: ```sh python -m msgq.examples.publisher # terminal 1 @@ -61,7 +61,7 @@ Issues and pull requests are welcome on [GitHub](https://github.com/commaai/msgq ## 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 From 75076ee7eb3fb23ccaefb410a6e3dbe857669893 Mon Sep 17 00:00:00 2001 From: Adeeb Shihadeh Date: Sun, 6 Sep 2026 13:04:44 -0700 Subject: [PATCH 4/9] lil more --- README.md | 13 +++++++++++-- msgq/examples/publisher.py | 6 +++++- msgq/examples/subscriber.py | 7 ++++++- 3 files changed, 22 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 2ea184c78..012647cc4 100644 --- a/README.md +++ b/README.md @@ -22,7 +22,7 @@ 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). +No locks, no broker, just shared memory.

1 KiB cross-process ping-pong benchmark
@@ -35,7 +35,7 @@ MSGQ is a generic high performance IPC pub sub system with a single publisher an 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) 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 -m msgq.examples.publisher # terminal 1 @@ -44,6 +44,13 @@ python -m msgq.examples.subscriber # terminal 2 The subscriber prints `Hello from MSGQ!` once per second. +To run multiple pairs independently, give each pair a different endpoint name: + +```sh +python -m msgq.examples.publisher --endpoint demo # terminal 1 +python -m msgq.examples.subscriber --endpoint demo # terminal 2 +``` + The core API sends and receives bytes: ```python @@ -66,6 +73,8 @@ MSGQ is available under the [MIT License](https://github.com/commaai/msgq/blob/m

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: diff --git a/msgq/examples/publisher.py b/msgq/examples/publisher.py index 18d897872..153b95e79 100644 --- a/msgq/examples/publisher.py +++ b/msgq/examples/publisher.py @@ -1,10 +1,14 @@ +import argparse import time import msgq def main(): - publisher = msgq.pub_sock("msgq_example") + 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: diff --git a/msgq/examples/subscriber.py b/msgq/examples/subscriber.py index cbf7a81d8..4cb31bd5b 100644 --- a/msgq/examples/subscriber.py +++ b/msgq/examples/subscriber.py @@ -1,8 +1,13 @@ +import argparse + import msgq def main(): - subscriber = msgq.sub_sock("msgq_example", timeout=1000) + 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: From 28d451b9fbd3c58ec3548e074beb30a2edb58750 Mon Sep 17 00:00:00 2001 From: Adeeb Shihadeh Date: Sun, 6 Sep 2026 13:08:22 -0700 Subject: [PATCH 5/9] lil more --- README.md | 1 + msgq/examples/benchmark.py | 48 +++++++++++++++++-------------------- msgq/examples/publisher.py | 6 +---- msgq/examples/subscriber.py | 6 +---- 4 files changed, 25 insertions(+), 36 deletions(-) diff --git a/README.md b/README.md index 012647cc4..299b9f35c 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,7 @@ [![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) diff --git a/msgq/examples/benchmark.py b/msgq/examples/benchmark.py index d62c963af..209793525 100644 --- a/msgq/examples/benchmark.py +++ b/msgq/examples/benchmark.py @@ -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 index 153b95e79..3c9360bb8 100644 --- a/msgq/examples/publisher.py +++ b/msgq/examples/publisher.py @@ -4,7 +4,7 @@ import msgq -def main(): +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() @@ -19,7 +19,3 @@ def main(): time.sleep(1) except KeyboardInterrupt: pass - - -if __name__ == "__main__": - main() diff --git a/msgq/examples/subscriber.py b/msgq/examples/subscriber.py index 4cb31bd5b..9a17e844b 100644 --- a/msgq/examples/subscriber.py +++ b/msgq/examples/subscriber.py @@ -3,7 +3,7 @@ import msgq -def main(): +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() @@ -17,7 +17,3 @@ def main(): print(f"Received: {message.decode('utf-8')}", flush=True) except KeyboardInterrupt: pass - - -if __name__ == "__main__": - main() From addc013bd345c751e984f55c19c088050c3ab561 Mon Sep 17 00:00:00 2001 From: Adeeb Shihadeh Date: Sun, 6 Sep 2026 13:09:39 -0700 Subject: [PATCH 6/9] lil less --- README.md | 13 ++----------- 1 file changed, 2 insertions(+), 11 deletions(-) diff --git a/README.md b/README.md index 299b9f35c..dd0e1eae9 100644 --- a/README.md +++ b/README.md @@ -23,8 +23,6 @@ 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. -No locks, no broker, just shared memory. -

1 KiB cross-process ping-pong benchmark
1 KiB cross-process ping-pong on x86 Linux. Benchmark script. @@ -38,20 +36,13 @@ 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 # terminal 1 -python -m msgq.examples.subscriber # terminal 2 -``` - -The subscriber prints `Hello from MSGQ!` once per second. - -To run multiple pairs independently, give each pair a different endpoint name: - ```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 From 5eae53ffc3d39c955f961dd3fd37738e7adac452 Mon Sep 17 00:00:00 2001 From: Adeeb Shihadeh Date: Sun, 6 Sep 2026 13:12:34 -0700 Subject: [PATCH 7/9] cheatsheet --- README.md | 55 ++++++++++++++++++++++++++++++++++++++++++++++++++ pyproject.toml | 2 +- 2 files changed, 56 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index dd0e1eae9..4b6ebb178 100644 --- a/README.md +++ b/README.md @@ -6,6 +6,8 @@

Quickstart · + API cheatsheet + · Examples · Discord @@ -54,6 +56,59 @@ publisher.send(b"Hello from MSGQ!") print(subscriber.receive()) # b'Hello from MSGQ!' ``` +## API cheatsheet + +Python messaging API (`import msgq`). The calls below are a reference, not a script to run in sequence. + +### Create sockets + +```python +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 +``` + +Both helpers accept `segment_size` in bytes; `0` selects the default 1 MiB ring buffer. Use the same size for all sockets on an endpoint. Messages must fit within roughly one third of the buffer, including metadata. Endpoints are local to the machine; leave `sub_sock`'s `addr` at its default. + +### Send and receive + +```python +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 an empty list if no messages arrive. Slow subscribers can miss messages when the ring buffer wraps. + +### Poll multiple subscribers + +```python +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()` to read their messages. + +### Reader synchronization and errors + +```python +pub.all_readers_updated() # Check whether tracked readers have caught up +pub.wait_for_readers(timeout=1.0, interval=0.001) # Wait for that condition; times are in seconds +msgq.IpcError # Messaging failure +msgq.MultiplePublishersError # Publisher conflict; subclass of IpcError +``` + +`wait_for_readers()` raises `TimeoutError` if its deadline expires. Reader 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. diff --git a/pyproject.toml b/pyproject.toml index 385b5f267..c7e46e37d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "msgq-ipc" -version = "1.0" +version = "1.1" description = "High-performance pub/sub messaging, made simple. For Python, C, and C++." readme = "README.md" requires-python = ">=3.11" From db37a66c6a47d23e9fd8ea53e38939ee011d99be Mon Sep 17 00:00:00 2001 From: Adeeb Shihadeh Date: Sun, 6 Sep 2026 13:13:45 -0700 Subject: [PATCH 8/9] lil simpler --- README.md | 39 ++++++++++++++++++++------------------- 1 file changed, 20 insertions(+), 19 deletions(-) diff --git a/README.md b/README.md index 4b6ebb178..25cb89ce3 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,7 @@

Quickstart · - API cheatsheet + Cheatsheet · Examples · @@ -56,24 +56,26 @@ publisher.send(b"Hello from MSGQ!") print(subscriber.receive()) # b'Hello from MSGQ!' ``` -## API cheatsheet +## Cheatsheet -Python messaging API (`import msgq`). The calls below are a reference, not a script to run in sequence. +```python +import msgq -### Create sockets +# API reference; calls are not intended to run in sequence. + +# Create sockets -```python 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 -``` -Both helpers accept `segment_size` in bytes; `0` selects the default 1 MiB ring buffer. Use the same size for all sockets on an endpoint. Messages must fit within roughly one third of the buffer, including metadata. Endpoints are local to the machine; leave `sub_sock`'s `addr` at its default. +# Both helpers accept segment_size in bytes; 0 selects the default 1 MiB buffer. +# Use the same size for all sockets on an endpoint. Messages, including metadata, +# must fit within roughly one third of the buffer. Endpoints are local; leave addr at its default. -### Send and receive +# Send and receive -```python 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 @@ -81,33 +83,32 @@ sub.setTimeout(1000) # Set receive timeout in millis 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 an empty list if no messages arrive. Slow subscribers can miss messages when the ring buffer wraps. +# A receive timeout returns None; draining returns [] if no messages arrive. +# Slow subscribers can miss messages when the ring buffer wraps. -### Poll multiple subscribers +# Poll multiple subscribers -```python 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()` to read their messages. +# Register each socket once. Call receive() on the sockets returned by poll(). -### Reader synchronization and errors +# Reader synchronization and errors -```python pub.all_readers_updated() # Check whether tracked readers have caught up pub.wait_for_readers(timeout=1.0, interval=0.001) # Wait for that condition; times are in seconds msgq.IpcError # Messaging failure msgq.MultiplePublishersError # Publisher conflict; subclass of IpcError -``` -`wait_for_readers()` raises `TimeoutError` if its deadline expires. Reader synchronization checks queue positions, not application processing; it requires at least one tracked reader and ignores readers invalidated by an overwrite. +# 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 From b2b6742c475f2bb0f580413cf48b04f94701a7aa Mon Sep 17 00:00:00 2001 From: Adeeb Shihadeh Date: Sun, 6 Sep 2026 13:17:12 -0700 Subject: [PATCH 9/9] lil more --- README.md | 29 +++++++++++------------------ 1 file changed, 11 insertions(+), 18 deletions(-) diff --git a/README.md b/README.md index 25cb89ce3..1609f32a8 100644 --- a/README.md +++ b/README.md @@ -63,46 +63,39 @@ import msgq # API reference; calls are not intended to run in sequence. -# Create sockets - +# === 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 -# Both helpers accept segment_size in bytes; 0 selects the default 1 MiB buffer. -# Use the same size for all sockets on an endpoint. Messages, including metadata, -# must fit within roughly one third of the buffer. Endpoints are local; leave addr at its default. - -# Send and receive +# === 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 +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 - +# === 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 +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 - +# === Reader synchronization and errors === pub.all_readers_updated() # Check whether tracked readers have caught up -pub.wait_for_readers(timeout=1.0, interval=0.001) # Wait for that condition; times are in seconds -msgq.IpcError # Messaging failure +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.