Dust Riven

Dust Riven is a small Rust extension for Python, built with PyO3. It provides a Signal class, a simple event system similar to observer/pub-sub patterns. You create a Signal, connect callbacks to it, and emit it with arguments to call all connected callbacks.

What it does

Signal is a thread-safe event emitter. It lets you register functions that run when the signal is emitted, and it supports:

• normal connections
• one-time connections
• connections that expire after a fixed number of calls
• weak references, so a callback can be garbage collected without needing to manually disconnect it
• priority ordering, so higher-priority callbacks run before lower-priority ones
• scoped connections via a context manager, so a callback is automatically disconnected when the with block exits

Basic usage

import dust_riven

signal = dust_riven.Signal("on_update")

name is optional and defaults to None - dust_riven.Signal() works just as well as dust_riven.Signal("on_update"). Naming a signal is mainly useful for repr(signal), which is handy in logs and debuggers when a program has many signals.

• Signal(name=None) - creates a new signal, optionally with a name
• connect(callback, weak=False, priority=0) - registers a callback, returns an id used to disconnect it later. Higher priority values run first; callbacks with equal priority run in the order they were connected.
• connect_once(callback, weak=False, priority=0) - callback runs on the first emit only
• connect_finite(callback, times, weak=False, priority=0) - callback runs for a fixed number of emits, then is removed
• disconnect(id) - removes a specific callback
• connected(callback, weak=False, priority=0) - like connect, but returns a context manager instead of an id. The callback is automatically disconnected when the with block exits, even if an exception is raised inside it.
• emit(*args, on_error="fast_fail", **kwargs) - calls all connected callbacks with the given arguments. With on_error="fast_fail" (default), the first exception raised stops execution and propagates immediately, and remaining callbacks are not called. With on_error="collect", every callback runs regardless of exceptions; the returned list contains each callback's return value, or the exception instance itself in place of a return value for any callback that raised.
• emit_async(*args, **kwargs) - like emit, but if a callback returns an awaitable (e.g. it's an async def), those are collected and run concurrently with asyncio.gather; must be awaited
• on_error(handler) - registers handler(exception, callback), called every time any connected callback raises during emit() or emit_async(), regardless of on_error= mode. Independent of emit(on_error=...): hooks always observe the error, on_error= only controls whether emit()/emit_async() also re-raises (fast_fail) or continues and collects it (collect). Meant for centralized logging/metrics without touching every call site. If a hook itself raises, that error is printed (not propagated) and the remaining hooks still run.
• clear_error_hooks() - removes all hooks registered via on_error
• len(signal) - number of callbacks currently connected

Pass weak=True on any connect method to hold a weak reference instead of a strong one, so the callback can be garbage collected normally if nothing else references it. If the callback has no other strong reference at the time you connect it (so it would be garbage collected immediately), the connect call raises TypeError instead of silently connecting a listener that can never fire.

Examples

Connect and emit

import dust_riven

signal = dust_riven.Signal("on_update")

def handler(message):
    print("got:", message)

signal.connect(handler)
signal.emit("hello world")

Connect once

def setup_handler():
    print("setup ran")

signal.connect_once(setup_handler)
signal.emit()  # prints "setup ran"
signal.emit()  # does nothing, handler already removed

Connect finite

def limited_handler():
    print("called")

signal.connect_finite(limited_handler, 3)
signal.emit()  # called
signal.emit()  # called
signal.emit()  # called
signal.emit()  # nothing happens, removed after 3 calls

Collecting errors instead of failing fast

def ok():
    return "fine"

def boom():
    raise ValueError("bad listener")

signal.connect(ok)
signal.connect(boom)

results = signal.emit(on_error="collect")
# results == ["fine", ValueError("bad listener")]

Centralized error logging with on_error

def log_listener_error(exc, callback):
    print(f"listener {callback!r} failed: {exc!r}")

signal.on_error(log_listener_error)

def boom():
    raise ValueError("bad listener")

signal.connect(boom)

# Hook fires no matter which on_error mode you pick:
signal.emit(on_error="collect")     # hook fires, emit() also returns [ValueError(...)]
signal.emit(on_error="fast_fail")   # hook fires, then emit() re-raises ValueError

Async emit

import asyncio
import dust_riven

signal = dust_riven.Signal("on_update")

def sync_handler(value):
    print("sync:", value)

async def async_handler(value):
    await asyncio.sleep(0.1)
    print("async:", value)

signal.connect(sync_handler)
signal.connect(async_handler)

async def main():
    # sync_handler runs immediately; async_handler is awaited
    # concurrently alongside any other async listeners
    await signal.emit_async(42)

asyncio.run(main())

Weak connection

class Listener:
    def handle(self, value):
        print("received", value)

listener = Listener()
signal.connect(listener.handle, weak=True)

del listener
signal.emit(42)  # dead reference is dropped silently, no crash

Disconnect

cb_id = signal.connect(handler)
signal.disconnect(cb_id)

Priority ordering

def logs_first(message):
    print("log:", message)

def runs_after(message):
    print("handle:", message)

signal.connect(runs_after, priority=0)
signal.connect(logs_first, priority=10)

signal.emit("hello")
# prints "log: hello" then "handle: hello"

Scoped connection

def handler(message):
    print("got:", message)

with signal.connected(handler):
    signal.emit("hello")  # handler runs
# handler is automatically disconnected here, even if emit raised

signal.emit("hello")  # handler no longer runs

Listener count

print(len(signal))

Behavior notes

• Callbacks are collected into a snapshot before being called, so connecting or disconnecting callbacks from inside another callback during emit is safe.
• Callbacks run in priority order (highest first); callbacks connected with equal priority (the default is 0 for all) run in the order they were connected, so existing code that doesn't pass priority keeps its original FIFO ordering.
• connected() returns a context manager rather than an id. It's equivalent to calling connect() and then disconnect() yourself in a finally block - useful for listeners that should only be active for the duration of a single operation or test.
• If a weakly referenced callback has already been garbage collected, it is silently removed the next time emit runs.
• Callbacks connected with connect_once or connect_finite are automatically removed once they've been called the requested number of times.
• By default, emit uses on_error="fast_fail": the first exception raised by a callback propagates immediately and any callbacks after it are skipped. With on_error="collect", all callbacks run no matter what, and exceptions are returned in the result list rather than raised.
• emit_async calls every callback the same way emit does; any callback that returns an awaitable (e.g. an async def) has that awaitable scheduled via asyncio.gather and run concurrently once you await the result. Purely synchronous callbacks run immediately, before the returned value is awaited.
• emit_async must be called from a thread with an already-running asyncio event loop (i.e. await signal.emit_async() from inside a coroutine) - this applies even if every connected callback is synchronous. Calling it with no loop running (e.g. asyncio.run(signal.emit_async()), where the call happens before run starts its loop) raises RuntimeError.
• With emit_async, an exception from a synchronous callback is raised as soon as it's called (before you even reach the await), while an exception from an async callback surfaces when the gathered result is awaited.
• If a callback raises during emit_async, remaining callbacks are not called, and any async callback's coroutine already created before the error is closed rather than left dangling. The same cleanup also happens if emit_async is called with no running event loop: any coroutines already created from earlier async callbacks are closed before the RuntimeError propagates, instead of being left as unawaited, leaked coroutines.
• on_error hooks run for every callback failure in both emit and emit_async, independently of on_error="fast_fail"/"collect". They're observers, not error handlers: they can't suppress or transform the error, and if a hook itself raises, that's printed rather than propagated so one broken hook can't block emit or other hooks.
• Signal participates in Python's cyclic garbage collector, so a callback that holds a reference back to its own Signal (e.g. a closure or bound method capturing the signal it's connected to) doesn't leak - gc.collect() can still find and break that cycle even though nothing was ever explicitly disconnected. This also applies to hooks registered via on_error.

Building

This project uses PyO3 and is built as a native Python extension, typically with maturin. Once built, it exposes a Python module named dust_riven containing the Signal class.

Requirements

• Rust toolchain
• Python with a PyO3-compatible version
• parking_lot, once_cell, and smallvec crates

Running the benchmark against blinker

The repository includes bench.py, a script that compares dust_riven against the blinker library for connect and emit performance, across different listener counts, emit counts, and strong vs weak callback variants.

1. Create and activate a virtual environment

python -m venv .venv
source .venv/bin/activate

On Windows, activate with .venv\Scripts\activate instead.

2. Install the needed Python packages

pip install blinker tqdm maturin

3. Build and install dust_riven into the same venv

maturin develop --release

Run this from the project root (the folder containing Cargo.toml and pyproject.toml). It compiles the Rust code and installs the dust_riven module directly into the active venv.

4. Run the benchmark

python bench.py

This runs the full matrix of listener counts and emit counts for both strong and weak variants, and prints comparison tables with minimum/median timings plus a speedup ratio.

For a fast smoke test instead of the full matrix:

python bench.py --quick

You can also narrow the run with options, for example:

python bench.py --listeners 1,100,1000 --emits 100,1000 --repeats 5 --variants strong --measure emit

Available options: --listeners, --emits, --repeats, --variants, --libraries, --measure. Run python bench.py --help for the full list with descriptions.

License

MIT - see LICENSE for details.