quality for users

README.md

@@ -6,39 +6,41 @@
 <h3>
   <a href="#quickstart">Quickstart</a>
   <span> · </span>
-  <a href="examples/">Examples</a>
+  <a href="#cheatsheet">Cheatsheet</a>
+  <span> · </span>
+  <a href="https://github.com/commaai/msgq/tree/master/msgq/examples">Examples</a>
   <span> · </span>
   <a href="https://discord.comma.ai">Discord</a>
 </h3>
 
 [![Discord](https://img.shields.io/badge/Discord-Join-5865F2?logo=discord&logoColor=white)](https://discord.comma.ai)
+[![PyPI](https://img.shields.io/pypi/v/msgq-ipc)](https://pypi.org/project/msgq-ipc/)
+[![Ask DeepWiki](https://deepwiki.com/badge.svg)](https://deepwiki.com/commaai/msgq)
 [![Tests](https://github.com/commaai/msgq/actions/workflows/tests.yml/badge.svg?branch=master)](https://github.com/commaai/msgq/actions/workflows/tests.yml)
-[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE)
+[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://github.com/commaai/msgq/blob/master/LICENSE)
 
 </div>
 
 ---
 
 MSGQ lets programs on the same machine exchange messages. A publisher sends messages to a named endpoint, and subscribers listen on that same endpoint. Each endpoint supports one publisher and multiple subscribers.
 
-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).
-
 <p align="center">
   <img src="https://github.com/user-attachments/assets/79bb91cf-c9ad-4fb4-97d9-33359a083f0f" alt="1 KiB cross-process ping-pong benchmark"><br>
-  <sub>1 KiB cross-process ping-pong on x86 Linux. <a href="examples/benchmark.py">Benchmark script</a>.</sub>
+  <sub>1 KiB cross-process ping-pong on x86 Linux. <a href="https://github.com/commaai/msgq/blob/master/msgq/examples/benchmark.py">Benchmark script</a>.</sub>
 </p>
 
 ## Quickstart
 
 ```sh
-python -m pip install 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
+<details>
+<summary>Under the hood</summary>
+
+The message queue copies data on send and receive. A fake implementation is also available for deterministic testing.
 
 ### Storage
 The storage for the queue consists of an area of metadata, and the actual buffer. The metadata contains:
@@ -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.
+
+</details>

examples/benchmark.py → 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)

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

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

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"