How Nallely's Thread-Based Device Architecture Achieves Asynchrony Without Asyncio

Nallely implements a thread-per-device model where each virtual device runs in its own OS thread, using cooperative generator suspension and thread-safe queues to achieve deterministic, low-latency asynchronous behavior without relying on Python's asyncio library.

Nallely-MIDI is a Python framework for building real-time MIDI processing systems. According to the source code in dr-schlange/nallely-midi, the project deliberately avoids async/await patterns in favor of classic threading combined with cooperative multitasking, ensuring predictable timing critical for live music applications.

Thread-Per-Device Foundation

At the core of Nallely's architecture is the VirtualDevice class defined in nallely/core/virtual_device.py. Every virtual device subclasses VirtualDevice, which itself inherits from threading.Thread (line 70). When start() is called, a dedicated OS thread begins executing the device's run() method, creating true parallelism between devices.

This design means multiple devices run simultaneously without interfering with each other's execution cycles. While Python's GIL prevents true CPU parallelism for pure Python code, the I/O-bound nature of MIDI processing and the framework's use of efficient waiting mechanisms ensure responsive performance across dozens of modules.

Cooperative Suspension and Generator-Based Scheduling

The run() loop implements a novel cooperative multitasking system by treating the device's main() generator like a coroutine. When main() or an OnChange handler yields the special token "__suspend__" (line 57), the scheduler stores the generator in self.suspended_tasks and returns control to the main loop.

On the next cycle, resume_suspended_tasks() (line 112) calls next() on the saved generator, effectively resuming the paused task exactly where it left off. This mechanism allows long-running or timing-sensitive code to be non-blocking while remaining single-threaded within the device context.

Thread-Safe Communication Mechanisms

Nallely ensures safe inter-thread communication through several synchronized primitives:

• Bounded Input Queues: Each parameter has a dedicated queue.Queue with maxsize=2000 (line 18). External threads push parameter changes into these queues via set_parameter(), while the device's run() loop consumes them.
• Batch Processing: To minimize latency while preventing backlog overflow, the loop reads a maximum of 10 items per cycle (max_batch_size = 10), processing inputs efficiently without blocking indefinitely.
• Reentrant Locking: Mutable device state is protected by self._lock = threading.RLock() (line 102), used by OnChange wrappers and internal helpers to prevent race conditions during concurrent access.

Pause, Resume, and Adaptive Timing

Devices support graceful pausing through threading.Event objects. Calling pause() clears self.pause_event, causing the main loop to block on self.pause_event.wait() and consume zero CPU while idle. Resuming wakes the thread immediately without losing state.

The architecture maintains deterministic timing through adaptive sleeping. Each cycle targets a specific duration (target_cycle_time, default 2ms). After processing inputs and outputs, the loop calculates elapsed time and sleeps only the remaining duration (line 108), ensuring stable real-time cadence for MIDI synchronization.

Event-Driven Programming with OnChange

The framework provides event-driven reactivity through the @on decorator. When registered parameters change, the system automatically invokes wrapped handlers:

from nallely.core.virtual_device import VirtualDevice, on

class Counter(VirtualDevice):
    count_cv = VirtualParameter(name="count", range=(0, 127))

    @on(count_cv, edge="rising")
    def on_increase(self, value, ctx):
        print(f"Count increased to {value}")

The OnChange implementation (line 86) handles edge detection and condition checking synchronously within the device's thread cycle, triggering callbacks only when specified conditions (like rising edges) are met.

Generator Suspension in Practice

The following example demonstrates cooperative suspension using self.sleep():

class Blinker(VirtualDevice):
    blink_cv = VirtualParameter(name="blink", range=(0, 1))

    def main(self, ctx):
        while True:
            self.blink = 1 - self.blink
            yield from self.sleep(500, consider_target_time=True)

Here, self.sleep() yields "__suspend__", causing the framework to store the generator and resume it after 500ms have elapsed, all without blocking other devices or the host application.

External Async I/O Integration

For communication with external systems, Nallely utilizes separate background threads that interact safely with device threads. The WebSocket connector in nallely/distributed/remote_ws_connector.py (line 70) runs its own threading.Thread, pushing received data into device input queues via set_parameter():

from nallely.distributed import NallelyWebsocketBus

bus = NallelyWebsocketBus(address="192.168.1.100:6789")
service = bus.register(
    kind="external",
    name="my_neuron",
    parameters={"note": {"min": 0, "max": 127}},
)

class RemoteNoteReceiver(VirtualDevice):
    note_cv = VirtualParameter(name="note", range=(0, 127))

    def __init__(self, **kw):
        super().__init__(**kw)
        service.onmessage = lambda msg: self.set_parameter("note", msg["value"])

    def main(self, ctx):
        while True:
            yield self.note

This thread-safe integration allows external async I/O to coexist with the deterministic device thread model without introducing callback complexity.

Summary

• Thread-per-device architecture: Each VirtualDevice inherits from threading.Thread, creating isolated execution contexts in nallely/core/virtual_device.py.
• Cooperative multitasking: Generators yielding "__suspend__" are stored in suspended_tasks and resumed by resume_suspended_tasks(), enabling non-blocking delays without asyncio.
• Thread-safe queues: Input changes flow through bounded queue.Queue objects with batch processing (max_batch_size=10) to balance throughput and latency.
• Deterministic timing: target_cycle_time (default 2ms) with adaptive sleeping ensures stable real-time performance.
• Event-driven callbacks: The @on decorator registers edge-sensitive handlers that execute synchronously within the device thread.
• External integration: Background threads like the WebSocket connector push data safely into devices using thread-safe parameter setting.

Frequently Asked Questions

How does Nallely avoid asyncio while still being asynchronous?

Nallely uses OS threads (threading.Thread) combined with generator-based cooperative multitasking. Each device runs in its own thread, and within that thread, the main() generator can yield "__suspend__" to pause execution temporarily. The scheduler stores these suspended generators and resumes them on subsequent cycles, achieving asynchrony through classic threading and generator state machines rather than async/await syntax.

What happens when a device is paused?

When pause() is called, the device clears its internal threading.Event (pause_event). Inside the run() loop, the thread blocks on self.pause_event.wait(), entering an efficient wait state that consumes no CPU cycles. Calling resume() sets the event, immediately waking the thread to continue processing from its previous state without losing any context or queued messages.

How does Nallely handle concurrent access to device parameters?

All parameter modifications flow through thread-safe queue.Queue objects with a maximum size of 2000 items. External threads call set_parameter(), which enqueues the change, while the device's internal thread dequeues and processes up to 10 items per cycle (max_batch_size). For direct attribute access, VirtualDevice provides a reentrant lock (RLock) that OnChange wrappers and internal methods use to protect mutable state from race conditions.

Can external I/O operations block the device thread?

No. External I/O operations, such as WebSocket communication, run in separate background threads defined in files like nallely/distributed/remote_ws_connector.py. These threads push data into the device's input queue via thread-safe set_parameter() calls. The device thread never blocks waiting for network operations; it only processes already-queued data during its regular cycle, maintaining deterministic timing for MIDI processing.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}