Async Nodes

HORUS supports async def node ticks. There is no separate async node class — you write a normal horus.Node and pass a coroutine function as tick. HORUS detects it and schedules the node on its async I/O executor.

import asyncio
import horus

async def poll_sensor(node):
    # Any awaitable works here: HTTP, sockets, database drivers.
    await asyncio.sleep(0.05)
    node.send("reading", {"value": 42})

horus.run(
    horus.Node(name="poller", tick=poll_sensor, rate=10, pubs=["reading"]),
    duration=1.0,
)

horus.asyncio is the standard library's asyncio module, re-exported for convenience. import horus then horus.asyncio.sleep(...) and import asyncio then asyncio.sleep(...) are the same function. Everything you already know about asyncio applies unchanged.

APIs that do not exist

Earlier versions of this page documented an async API that HORUS does not ship. If you have code or notes referring to these names, here is what to use instead:

Topic sends and receives are shared-memory operations measured in microseconds. There is nothing to await, which is why no async topic type exists.

How an async tick runs

When you pass a coroutine function to tick, init, or shutdown, HORUS:

1. Detects it with asyncio.iscoroutinefunction().
2. Creates one event loop for that node and reuses it for the node's whole lifetime.
3. Marks the node for the async_io execution class, so it runs on the async I/O executor instead of the main scheduler thread.
4. Each tick calls loop.run_until_complete(your_coroutine(node)).

Step 4 is the one that shapes everything else, so it is worth stating plainly.

An async tick blocks until it finishes

run_until_complete does not return until your coroutine completes. A tick does not overlap with the next tick of the same node — rate is a ceiling, not a promise.

import asyncio
import horus

async def slow(node):
    await asyncio.sleep(0.05)      # 50 ms of I/O
    node.send("out", {"ok": True})

# rate=100 asks for a 10 ms period, but each tick needs 50 ms.
horus.run(horus.Node(name="slow", tick=slow, rate=100, pubs=["out"]), duration=1.0)

Running this for one second produces about 20 ticks, not 100, and the timing report shows an average tick of roughly 50,000 µs. If you need a node to keep ticking at its configured rate while long I/O is in flight, an async tick is the wrong tool — see Long-lived streams below.

Concurrency inside one tick works normally

You cannot overlap ticks, but you can overlap awaits within a single tick:

import asyncio
import horus

async def fan_out(node):
    # These three run concurrently — total wait is ~50 ms, not 150 ms.
    a, b, c = await asyncio.gather(
        asyncio.sleep(0.05, result="sensor-a"),
        asyncio.sleep(0.05, result="sensor-b"),
        asyncio.sleep(0.05, result="sensor-c"),
    )
    node.send("merged", {"a": a, "b": b, "c": c})

horus.run(horus.Node(name="fan", tick=fan_out, rate=10, pubs=["merged"]), duration=0.5)

The measured tick time is ~50 ms, confirming the three sleeps overlapped.

A slow async node does not stall other nodes

Async nodes run on a separate executor, so blocking inside one does not delay your control loop:

import asyncio
import horus

async def fetch(node):
    await asyncio.sleep(0.05)            # slow network call
    node.send("reading", {"value": 1})

def control(node):
    if node.has_msg("reading"):
        reading = node.recv("reading")
        node.send("cmd", {"speed": reading["value"] * 0.1})

horus.run(
    horus.Node(name="fetch", tick=fetch, rate=100, pubs=["reading"]),
    horus.Node(name="control", tick=control, rate=100, subs=["reading"], pubs=["cmd"]),
    duration=1.0,
)

Over one second the control node gets its full 100 ticks at ~80 µs each, while fetch manages ~20. The sync node consumes whatever the async node last published and never waits for it. This is the pattern to reach for: slow I/O in an async node, timing-critical logic in a sync node, data passed between them over a topic.

Async init and shutdown

init and shutdown accept coroutine functions too. Because the node keeps one event loop for its entire life, objects created in init stay valid in every later tick — connections, sessions, and asyncio primitives all survive.

import asyncio
import horus

class Connection:
    """Stand-in for aiohttp.ClientSession, asyncpg pool, or a websocket."""
    async def open(self):
        await asyncio.sleep(0.01)
        self.seq = 0

    async def read(self):
        await asyncio.sleep(0.01)
        self.seq += 1
        return {"seq": self.seq}

    async def close(self):
        await asyncio.sleep(0.01)

async def connect(node):
    node.conn = Connection()
    await node.conn.open()
    node.log_info("connected")

async def poll(node):
    node.send("telemetry", await node.conn.read())

async def disconnect(node):
    await node.conn.close()
    node.log_info("disconnected")

horus.run(
    horus.Node(
        name="link",
        tick=poll,
        init=connect,
        shutdown=disconnect,
        rate=20,
        pubs=["telemetry"],
    ),
    duration=0.5,
)

Open expensive resources in init, not in tick. Reopening a session every tick is the most common way to make an async node far slower than it needs to be.

Execution classes are mutually exclusive

A node picks exactly one execution class. Combining them raises ValueError at construction:

horus.Node(tick=async_fn, compute=True)     # ValueError
horus.Node(tick=async_fn, on="scan")        # ValueError

Execution class is mutually exclusive: only one of async tick, compute=True, or on='topic' can be set

Long-lived streams: background thread

An async tick opens and closes its work each cycle. That is a poor fit for a connection that should stay open and push messages at its own pace, like a websocket feed. For that, run the event loop in a background thread and hand results to a sync node through a queue:

import asyncio
import queue
import threading
import horus

inbox = queue.Queue(maxsize=256)
stop = threading.Event()

async def stream():
    """One long-lived connection producing messages at its own pace."""
    while not stop.is_set():
        await asyncio.sleep(0.02)        # stands in for `await ws.recv()`
        try:
            inbox.put_nowait({"t": "tick"})
        except queue.Full:
            pass                         # drop when the node falls behind

threading.Thread(target=lambda: asyncio.run(stream()), daemon=True).start()

def drain(node):
    """Sync tick: never blocks, just takes whatever arrived."""
    while True:
        try:
            node.send("stream", inbox.get_nowait())
        except queue.Empty:
            return

horus.run(horus.Node(name="drain", tick=drain, rate=50, pubs=["stream"]), duration=0.5)
stop.set()

The drain node ticks in roughly 0.1 ms because it never waits — it only moves whatever the background loop already produced. Use a bounded queue and drop on overflow so a fast producer cannot grow memory without limit.

Rate: pacing a background loop

horus.Rate is the rate-control primitive for loops you drive yourself, such as the background thread above. It compensates for drift, unlike time.sleep(1 / hz).

import horus

rate = horus.Rate(50)                    # 50 Hz -> 20 ms period
print("period:", rate.period(), "target:", rate.target_hz())

for _ in range(25):
    do_work = sum(range(10_000))         # your loop body
    if rate.is_late():                   # check BEFORE sleeping
        print("overran the 20 ms period")
    rate.sleep()                         # drift-compensated

print("headroom: %.0f Hz" % rate.actual_hz())
rate.reset()

actual_hz() is easy to misread. It reports the frequency the loop could sustain if it never slept, so a healthy 50 Hz loop doing 5 ms of work reports about 190 Hz, not 50. Read it as headroom: when it falls below target_hz(), the loop can no longer keep up. Check is_late() before sleep(), since after sleeping you are on schedule again by definition.

Rate.sleep() blocks the calling thread and does not release the GIL. Use it in dedicated background threads, never inside a node tick — the scheduler already paces ticks via rate=.

See Also

• Python Bindings — core Node, run, and topic API
• Examples — more complete Python programs