IPC and Processes

Sources:

• fes_framework/orchestrator.py
• fes_framework/ipc/buffer.py
• fes_framework/ipc/queue_handler.py

The two-process model#

┌────────────────── Main process ───────────────────────┐
│  Hosts the optional PyQt5 GUI (or just waits on Ctrl-C │
│  for headless apps).                                   │
│                                                        │
│  - PyQt5 widgets                                       │
│  - reads SharedRingBuffer for sensor data (~30 Hz)     │
│  - writes control_queue / utility_queue                │
└─────────────┬──────────────────────────────────────────┘
              │  queues + shared memory
              │
┌─────────────▼─────── Backend subprocess ──────────────┐
│  Hosts ClosedLoopEngine.run() — the full closed loop. │
│                                                        │
│  - SuitHandler / DataStreamer                          │
│  - ControlStrategy.process()                           │
│  - Stimulator / LSLStreamer                            │
│  - QueueHandler (drains queues each cycle)             │
│  - SharedRingBuffer writer                             │
└────────────────────────────────────────────────────────┘

The split is non-optional. Even in headless mode, the engine runs in a backend subprocess; the main process just calls Process.join() and waits for Ctrl-C.

Why two processes#

Python's GIL prevents true parallel execution within a single process. PyQt5's event loop blocks regularly, especially during plot redraws. If the engine ran in the same process as the GUI, every dropped frame on the GUI side would delay a stimulation pulse. A subprocess is the only reliable isolation.

The OS also gives us guaranteed cleanup: if the GUI crashes, the backend continues. If the backend crashes, the GUI sees its SharedRingBuffer go stale and can show "backend disconnected" rather than crashing too. Each process can be killed independently; the SDK's hardware connections are released by _cleanup() either way.

See Design principle 5.

Two communication channels#

The two processes exchange information through two complementary channels — different rates, different patterns, different uses:

Queues — for low-frequency events#

The two multiprocessing.Queue instances are created in the launcher and passed to both processes. The backend's QueueHandler drains them at the start of every cycle:

# Backend (engine.py):
def _poll_messages(self) -> None:
    # Drain utility queue to latest message
    while True:
        msg = self.queue_handler.get_utility_message()
        if msg is None: break
        if msg == STOP_SENTINEL: self.stop(); return
        self.utility_message = msg
        self.on_utility_message(msg)
    # Same for control queue
    ...

"Drain to latest" matters: if the GUI emits 50 control messages per second (slider drag) and the backend sees 100 cycles per second, the backend needs to skip the stale ones and act only on the most recent. The drain pattern guarantees that.

SharedRingBuffer — for high-frequency data#

For sensor data flowing the other way (backend → GUI), a queue would either drop frames or back-pressure the producer. The framework uses a fixed-size ring buffer in shared memory: the backend writes one frame per cycle, the GUI reads the most recent N frames. There's no locking on the read side — the GUI accepts that the very newest frame might be torn (rare; bounded by one cycle).

┌──────────────── shared memory ────────────────┐
│ frame[0] frame[1] frame[2] ... frame[N-1]    │
│   ▲                                            │
│   └── write_index (atomic increment)          │
└────────────────────────────────────────────────┘
       ▲                            ▲
   backend writes              GUI reads last N
   (~100 Hz)                  (~30 Hz)

Frame layout is application-defined as a numpy structured dtype. The backend (your ClosedLoopEngine subclass, in on_cycle_complete()) writes the frame; the GUI specifies the same dtype when attaching to the buffer.

A typical layout:

FRAME_DTYPE = np.dtype([
    ("timestamp",      "f8"),
    ("knee_R",         "f8"),
    ("hip_R",          "f8"),
    ("contact_L",      "u1"),
    ("contact_R",      "u1"),
    ("quad_R_amp",     "f4"),
    # ...
])

See examples/walking_fes/ for a production buffer layout.

Bringing it all together — the orchestrator#

fes_framework.orchestrator.launch(StrategyClass) is the standard way to start an application. It:

1. Creates control_queue and utility_queue.
2. Forks a backend subprocess that runs ClosedLoopEngine against StrategyClass.
3. (Optional) starts a GUI process with the same queues plus a SharedRingBuffer reader.
4. Waits for Ctrl-C; then sends STOP_SENTINEL over the utility queue to ask the backend to exit cleanly.

For custom orchestration (e.g. an Electron front-end instead of PyQt5), call the engine directly:

from fes_framework.engine import ClosedLoopEngine

engine = ClosedLoopEngine(
    control_strategy=MyStrategy(),
    control_queue=control_queue,
    utility_queue=utility_queue,
    lsl_enabled=True,
)
engine.run()

Things to know when crossing the process boundary#

• Pickling: anything sent over a queue must be picklable. Dataclasses are; lambdas and database connections aren't.
• Module imports happen twice: once in main, once in the subprocess. Heavy imports done at module scope cost twice; do them inside the subprocess entry function if they're large.
• Logging: subprocess logs go to a different stream by default. Use logging.basicConfig(...) inside the subprocess entry, or a queue-based handler if you need consolidated logs.
• if __name__ == "__main__": — every multiprocessing script on Windows needs this guard. The subprocess re-imports the script at startup; without the guard you fork-bomb yourself.
• SharedRingBuffer lifecycle: the backend creates the buffer in on_start(). The GUI attaches with create=False after the backend has created it — so attach late, or wait for an "engine ready" signal. The orchestrator handles this with a startup delay.

Headless mode#

For headless applications (no GUI), only the backend subprocess runs. The main process calls process.join() and exits when the backend does.

You can use orchestrator.launch(StrategyClass) without passing a GUI; the launcher detects no GUI process is needed and just waits.

Or you can skip the orchestrator entirely:

from fes_framework.engine import ClosedLoopEngine

if __name__ == "__main__":
    engine = ClosedLoopEngine(control_strategy=MyStrategy())
    engine.run()   # blocks until Ctrl-C

This collapses the model to a single process. For development and quick experiments it's often fine. Keep in mind:

• No GUI is possible without orchestrator (the engine's blocking loop won't let a Qt event loop spin).
• Ctrl-C works; the engine's _cleanup() runs on KeyboardInterrupt.

See also#

• Architecture — full multi-process design diagram
• Messaging — what flows over the queues
• Implementation Guide → Step 8 — adding a PyQt5 GUI
• Examples → Generic GUI — reference template
• Source: fes_framework/orchestrator.py, fes_framework/ipc/buffer.py, fes_framework/ipc/queue_handler.py