Scheduler

Key Takeaways

After reading this guide, you will understand:

• How the Scheduler orchestrates node execution through init(), tick(), and shutdown() phases
• Scheduler::new() as the single entry point, with builder methods for global settings
• Per-node execution classes: real-time (implied by .rate()), .compute(), .on(topic), .async_io()
• The fluent NodeBuilder API for adding nodes with .add(node).order(0).rate(1000_u64.hz()).done()?
• Per-node configuration: .order(), .rate(), .budget(), .deadline()
• Priority-based execution where lower numbers run first (0 = highest priority)
• Graceful shutdown via Ctrl+C signal handling

The Scheduler is the execution orchestrator in HORUS. It manages the node lifecycle, coordinates priority-based execution, and handles graceful shutdown.

What is the Scheduler?

The Scheduler is responsible for:

Node Registration: Adding nodes with the fluent builder API

Lifecycle Management: Calling init(), tick(), and shutdown() at the right times

Priority-Based Execution: Running nodes in priority order every tick

Signal Handling: Graceful shutdown on Ctrl+C

Performance Monitoring: Tracking execution metrics for all nodes

Fault Tolerance: Circuit breakers and failure policies per node

Creating a Scheduler

Scheduler::new() — Lightweight, No Syscalls

new() detects runtime capabilities (~30-100us) but does not apply any OS-level features. Use builder methods to opt in to features.

use horus::prelude::*;

// Minimal — just capability detection, no syscalls
let mut scheduler = Scheduler::new();
scheduler.add(my_node).order(0).done()?;
scheduler.run()?;

Builder Methods — Global Settings

For production deployments, chain builder methods on Scheduler::new() to configure global scheduler behaviour:

use horus::prelude::*;

let mut scheduler = Scheduler::new()
    .tick_rate(1000_u64.hz())  // 1 kHz control loop
    .blackbox(16)              // Flight recorder
    .watchdog(500_u64.ms())    // Frozen-node watchdog
    .prefer_rt();              // Attempt RT priority + mlockall, degrade if unavailable

scheduler.add(motor_ctrl).order(0).rate(1000_u64.hz()).budget(500_u64.us()).done()?;
scheduler.run()?;

All RT features that cannot be applied at runtime are recorded as degradations, not errors (see Real-Time Features below).

Adding Nodes

Use the fluent builder API to add nodes with configuration:

let mut scheduler = Scheduler::new();

// Basic: just set execution order
scheduler.add(sensor_node).order(0).done()?;

// With per-node tick rate
scheduler.add(fast_sensor).order(0).rate(1000_u64.hz()).done()?;

// Real-time node with an explicit tick budget
scheduler.add(motor_ctrl)
    .order(0)
    .rate(1000_u64.hz())
    .budget(500_u64.us())    // 500us max execution time
    .deadline(1000_u64.us()) // 1ms deadline
    .done()?;

// Chain multiple nodes
scheduler.add(safety_node).order(0).done()?;
scheduler.add(controller).order(10).done()?;
scheduler.add(sensor).order(50).done()?;
scheduler.add(logger).order(200).done()?;

NodeBuilder Methods

Execution classes (mutually exclusive — pick one per node):

Real-time is not declared, it is implied: a node becomes real-time by calling .rate(N.hz()) (or .budget() / .deadline()) without .compute(), .on() or .async_io(). Omit all of these and the node is best-effort, ticking in the main loop.

Per-node configuration:

.done() (like .build()) returns HorusResult<&mut Scheduler>, so write .done()? — a bare .done(); discards the validation result. It rejects a zero or non-finite .rate(), a zero .budget() or .deadline(), an empty .on("") topic, .budget() / .deadline() on a node that .compute() / .on() / .async_io() made non-RT, and .no_alloc() or .subscribe_with_timeout() on an execution class that would never enforce them.

Priority-Based Execution

Priorities are u32 values where lower numbers = higher priority:

order sequences nodes that run on the main tick loop and have no topic relationship between them:

let mut scheduler = Scheduler::new();

// Four BestEffort nodes with no shared topics — order decides the sequence
scheduler.add(safety_monitor).order(0).done()?;    // Runs 1st
scheduler.add(controller).order(10).done()?;        // Runs 2nd
scheduler.add(sensor).order(50).done()?;            // Runs 3rd
scheduler.add(logger).order(200).done()?;           // Runs 4th

TICK Sensor(order 50)  at  1.96ms
TICK Safety(order 0)   at 32.18ms     # subscribes to Sensor's topic

Running the Scheduler

Continuous Mode

Run until Ctrl+C:

scheduler.run()?;

Duration-Limited

Run for a specific duration, then shutdown:

use std::time::Duration;

scheduler.run_for(Duration::from_secs(30))?;

Node-Specific Execution

Execute only specific nodes by name:

// Run only these nodes continuously
scheduler.tick(&["SensorNode", "MotorNode"])?;

// Run specific nodes for a duration
scheduler.tick_for(&["SensorNode"], Duration::from_secs(10))?;

Builder Methods

Tick Rate

// Set global tick rate (default: 100 Hz)
let scheduler = Scheduler::new()
    .tick_rate(1000_u64.hz()); // 1kHz control loop

Real-Time Features

OS-level RT features are requested globally with .prefer_rt() (or .require_rt()), while individual nodes become RT by calling .rate():

let mut scheduler = Scheduler::new()
    .prefer_rt(); // Try RT, degrade gracefully if unavailable
    // .require_rt() instead if you want a panic on a non-RT system

// Per-node: .rate() on a node with no other execution class makes it RT
scheduler.add(motor_ctrl)
    .order(0)
    .rate(1000_u64.hz())  // RT execution class at 1kHz
    .budget(500_u64.us()) // 500us tick budget
    .deadline(1_u64.ms()) // 1ms deadline
    .done()?;

scheduler.add(sensor)
    .order(50)
    .compute()            // CPU-heavy, non-RT
    .done()?;

// Check what degraded (features that couldn't be applied)
for deg in scheduler.degradations() {
    println!("{}: {}", deg.feature, deg.reason);
}

When .prefer_rt() is set, the scheduler attempts, in order:

1. Memory locking (mlockall) — if permitted
2. RT priority (SCHED_FIFO, priority 50) — if available
3. CPU affinity — only if you also called .cores(&[2, 3])

Features that fail are recorded as degradations, not errors. .require_rt() applies the same features but panics when the system supports neither SCHED_FIFO nor mlockall.

BlackBox Flight Recorder

Enable the BlackBox through the builder API:

let mut scheduler = Scheduler::new()
    .blackbox(16);  // 16 MB flight recorder buffer

// Critical events are now captured in the ring buffer

The BlackBox is a crash-forensics ring buffer: it records deadline misses, budget violations, fault-tolerance state changes, and emergency stops — not every node tick. For full tick capture see Recording and Replay.

Safety Monitor

Budget enforcement and deadline monitoring need no flag — they are always active for nodes that have .rate() set. The .watchdog(timeout) builder adds one extra thing on top: detection of frozen nodes that stop ticking altogether.

let mut scheduler = Scheduler::new()
    .watchdog(500_u64.ms());  // Frozen-node detection

// Budget and deadline are derived from .rate() (80% / 95% of the period);
// override them explicitly when you need tighter limits.
scheduler.add(motor_ctrl)
    .order(0)
    .rate(1000_u64.hz())
    .budget(500_u64.us())   // 500us tick budget
    .deadline(900_u64.us()) // 900us deadline
    .done()?;

Other Builder Methods

Per-Node Rate Control

Individual nodes can run at different frequencies:

let mut scheduler = Scheduler::new();
scheduler.add(fast_sensor).order(0).rate(1000_u64.hz()).done()?;  // 1kHz
scheduler.add(slow_logger).order(200).rate(10_u64.hz()).done()?;  // 10Hz

// Or set rates after adding
scheduler.set_node_rate("FastSensor", 100_u64.hz());
scheduler.set_node_rate("SlowLogger", 10_u64.hz());

The scheduler automatically adjusts its internal tick period to be fast enough for the highest-frequency node.

Lifecycle Management

Initialization Phase

When you call run(), the scheduler initializes all nodes by calling init():

• All nodes initialize before the main loop starts
• If init() fails, the node enters Error state and won't tick — see Node Initialization Failure for how that affects the rest of the schedule

Main Execution Loop

1. Initialize all nodes (call init())
2. Hand nodes with an execution class to their executors
     - RT      -> one dedicated thread per node
     - Compute -> parallel thread pool
     - Event   -> per-node watcher thread
     - AsyncIo -> tokio blocking pool
   These run on their own schedules from here on.
3. Build a dependency graph over the remaining (BestEffort) nodes
     - from pub/sub topic metadata where it exists
     - falling back to `order` tiers where it does not
4. Main loop, for the BestEffort nodes only:
   a. Dispatch each node as soon as its predecessors have finished
      (in parallel; strictly sequential only under `.deterministic(true)`)
      - Check if node should tick (rate limiting)
      - Start tick timing
      - Call node.tick()
      - Record metrics, check budget/deadlines
   b. Sleep to maintain target tick rate
5. On shutdown signal:
   a. Set running = false
   b. Call shutdown() on all nodes

Graceful Shutdown

Ctrl+C is automatically caught:

^C
Ctrl+C received! Shutting down HORUS scheduler...
[Nodes shutting down gracefully...]
Scheduler shutdown complete

• Main loop exits
• Each node's shutdown() is called
• Errors during shutdown are logged but don't prevent other nodes from cleaning up
• Shared memory cleaned up

Recording and Replay

Recording Sessions

Enable tick recording with .with_recording(). This is a different feature from the BlackBox: the BlackBox is a crash-forensics ring buffer of critical events, while .with_recording() captures the full tick stream that replay_from() consumes.

let mut scheduler = Scheduler::new()
    .tick_rate(100_u64.hz())
    .with_recording()    // Enable recording (saves to the platform data dir:
                     //   ~/.local/share/horus/recordings/, or $XDG_DATA_HOME/horus/recordings)
    .blackbox(16);       // 16MB flight recorder for crash forensics

scheduler.add(my_node).order(0).done()?;

// Run normally — all node ticks are recorded
scheduler.run()?;

Recording can also be switched on without touching the code: horus run --record <session> sets the HORUS_RECORD_SESSION environment variable, which Scheduler::new() picks up automatically.

Replaying

replay_from() takes the path to a scheduler recording file, not the session directory:

// `~` is not expanded for a PathBuf — pass a real path
let mut replay_scheduler = Scheduler::replay_from(
    "/home/you/.local/share/horus/recordings/my_session/scheduler@abc123.horus".into()
)?;
replay_scheduler.run()?;

Use Scheduler::list_recordings() to enumerate available sessions.

Performance Monitoring

Node Metrics

let metrics = scheduler.metrics();

for m in &metrics {
    println!("Node: {} (order: {})", m.name(), m.order());
    println!("  Ticks: {} total, {} ok, {} failed",
             m.total_ticks(), m.successful_ticks(), m.failed_ticks());
    println!("  Duration: avg={:.2}ms, min={:.2}ms, max={:.2}ms",
             m.avg_tick_duration_ms(), m.min_tick_duration_ms(), m.max_tick_duration_ms());
}

Other Monitoring Methods

Error Handling

Node Initialization Failure

If init() fails, the node enters Error state and won't tick. Other nodes continue — unless the error's Severity is Fatal, in which case the scheduler stops immediately. Panics inside init() are caught and converted to NodeError::InitPanic, which is itself Fatal — so a panicking init() always stops the scheduler.

fn init(&mut self) -> Result<()> {
    Err(Error::node("SensorNode", "Sensor not connected")) // Node won't run, others unaffected
}

Runtime Errors

Handle errors gracefully in tick() — don't panic:

// GOOD: Handle errors
fn tick(&mut self) {
    match self.operation() {
        Ok(_) => {}
        Err(e) => hlog!(error, "Error: {}", e),
    }
}

// BAD: a panic in tick() does not crash the scheduler — it is caught and the
// node keeps being ticked, so this panics again on every tick, forever.
fn tick(&mut self) {
    self.operation().unwrap();
}

A panic inside tick() is caught: it is logged, the node is marked Error, its on_error() runs, and by default the scheduler keeps ticking it — so a node that panics deterministically panics on every tick and floods the log rather than stopping. Set a FailurePolicy if you want it stopped or restarted. A panic in init() is different: that one is fatal (NodeError::InitPanic).

Common Patterns

Layered Architecture

// Layer 1: Safety (Critical)
scheduler.add(collision_detector).order(0).done()?;
scheduler.add(emergency_stop).order(0).done()?;

// Layer 2: Control (High)
scheduler.add(pid_controller).order(10).done()?;
scheduler.add(motor_driver).order(10).done()?;

// Layer 3: Sensing (Normal)
scheduler.add(lidar_node).order(50).done()?;
scheduler.add(camera_node).order(50).done()?;

// Layer 4: Processing (Low)
scheduler.add(path_planner).order(100).done()?;

// Layer 5: Monitoring (Background)
scheduler.add(logger).order(200).done()?;
scheduler.add(diagnostics).order(200).done()?;

Production Deployment

let mut scheduler = Scheduler::new()
    .tick_rate(1000_u64.hz())  // 1 kHz control loop
    .prefer_rt()               // Attempt RT priority + mlockall (degrade if unavailable)
    .cores(&[2, 3])            // Pin to these CPU cores (required for affinity)
    .blackbox(16)              // Flight recorder
    .watchdog(500_u64.ms());   // Frozen-node watchdog

scheduler.add(safety_monitor).order(0).rate(1000_u64.hz()).budget(100_u64.us()).done()?;
scheduler.add(motor_ctrl).order(5).rate(1000_u64.hz()).budget(500_u64.us()).done()?;
scheduler.add(sensor).order(50).compute().rate(1000_u64.hz()).done()?;  // Compute class, ticked at 1 kHz
scheduler.add(logger).order(200).async_io().done()?;

scheduler.run()?;

Best Practices

Initialize heavy resources in init() — not in the constructor:

fn init(&mut self) -> Result<()> {
    self.buffer = vec![0.0; 10000];
    self.connection = connect_to_hardware()?;
    Ok(())
}

Keep tick() fast — each tick should complete within the tick period:

fn tick(&mut self) {
    let data = self.read_sensor();
    self.pub_topic.send(data);  // Fast!
}

Use appropriate priorities — don't make everything order 0:

scheduler.add(emergency_stop).order(0).done()?;   // Critical
scheduler.add(controller).order(10).done()?;       // High
scheduler.add(sensor).order(50).done()?;           // Normal
scheduler.add(logger).order(200).done()?;          // Background

Use builder methods for production — enable RT features, the BlackBox flight recorder, and the frozen-node watchdog:

let mut scheduler = Scheduler::new()
    .prefer_rt()
    .blackbox(16)
    .watchdog(500_u64.ms());

Deadline Miss Policy (Miss)

When a real-time node misses its deadline, the Miss policy determines what happens:

scheduler.add(safety_node)
    .order(0)
    .rate(1000_u64.hz())
    .on_miss(Miss::Stop)
    .done()?;

Miss::SafeMode fires on the transition into safe mode, not on every miss: a node under sustained overload is safed once, not re-safed every cycle. The latch clears as soon as the node meets its deadline again, so a later degradation is caught too — a flapping node produces one entry per episode. The node is not isolated and keeps ticking, so it can hold its safe outputs; the scheduler does not poll is_safe_state().

The default for every node is Miss::Warn, regardless of execution class. Set .on_miss(...) explicitly if you need Skip, SafeMode, or Stop.

Next Steps

• Learn about Message Types for communication
• Explore Examples for complete applications
• Read the API Reference for detailed documentation
• See Scheduler Configuration for advanced options