Keyboard shortcuts

Press ← or → to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

Real-Time Analysis

This chapter describes static analysis methods and tools for detecting anti-patterns that violate real-time guarantees in Rust embedded applications.

Overview

Real-time systems require deterministic execution times. Common anti-patterns that break real-time guarantees include:

Anti-PatternProblemDetection Method
Unbounded loopsInfinite execution timeClippy + custom lints
RecursionStack overflow, unbounded depthcargo-call-stack
Heap allocationNon-deterministic timing, fragmentationno_std + forbid patterns
Blocking I/OUnbounded wait timesCustom lints
Missing timeoutsOperations can hang foreverCustom lints
Large stack framesStack overflowcargo-call-stack

Built-in Clippy Lints

Loop and Iteration Lints

# Enable all loop-related lints
cargo clippy -- \
    -D clippy::infinite_iter \
    -D clippy::while_immutable_condition \
    -D clippy::never_loop \
    -D clippy::empty_loop
LintDetects
infinite_iterIterator chains guaranteed to be infinite
while_immutable_conditionLoop conditions that can never change
never_loopLoops that exit on first iteration
empty_looploop { } without body (use loop { hint::spin_loop() })

Memory and Performance Lints

cargo clippy -- \
    -W clippy::large_stack_arrays \
    -W clippy::large_types_passed_by_value \
    -W clippy::box_collection \
    -W clippy::rc_buffer

Create clippy.toml in your project root:

# Maximum size for stack-allocated arrays (bytes)
array-size-threshold = 512

# Warn on types larger than this passed by value
trivial-copy-size-limit = 16

# Cognitive complexity threshold
cognitive-complexity-threshold = 15

Running Clippy for Real-Time Code

# Strict mode for real-time code
cargo clippy --all-targets -- \
    -D warnings \
    -D clippy::all \
    -W clippy::pedantic \
    -D clippy::infinite_iter \
    -D clippy::while_immutable_condition \
    -A clippy::module_name_repetitions

Stack Analysis with cargo-call-stack

Installation and Usage

# Install (requires nightly)
cargo +nightly install cargo-call-stack

# Build with stack size info (from your firmware crate's directory)
RUSTFLAGS="-Z emit-stack-sizes" cargo +nightly build --release

# Generate call graph with stack sizes
cargo +nightly call-stack --release > call_graph.dot

# Visualize (requires graphviz)
dot -Tsvg call_graph.dot -o call_graph.svg

Interpreting Results

The output shows:

  • Each function’s stack frame size
  • Call graph relationships
  • Maximum stack depth through any path
  • Cycles (recursion) in the call graph

Example output:

digraph {
    "main" [label="main\n256 bytes"]
    "zenoh_poll" [label="zenoh_poll\n128 bytes"]
    "publisher_task" [label="publisher_task\n512 bytes"]

    "main" -> "zenoh_poll"
    "main" -> "publisher_task"
}

Limitations

  • Requires fat LTO (lto = "fat" in Cargo.toml)
  • Limited support for programs linking std
  • Indirect calls (function pointers, trait objects) may not be analyzed
  • Best for embedded no_std programs

Preventing Heap Allocation

Method 1: no_std Without Allocator

The simplest approach – don’t provide a global allocator:

#![allow(unused)]
#![no_std]
#![no_main]

fn main() {
// No #[global_allocator] defined
// Attempting to use Box, Vec, String will fail to compile
}

Method 2: Compile-Time Enforcement

#![allow(unused)]
#![no_std]
#![forbid(unsafe_code)]  // Also prevents custom allocators

fn main() {
use heapless::{Vec, String};  // Static-sized alternatives

// This will NOT compile:
// let v = alloc::vec::Vec::new();  // Error: no allocator

// This works:
let v: heapless::Vec<u8, 256> = heapless::Vec::new();
}

Method 3: Custom Lint (Dylint)

For projects that need alloc but want to restrict usage in certain modules:

#![allow(unused)]
fn main() {
// In real-time critical code, add:
#![deny(clippy::disallowed_methods)]
}

With clippy.toml:

disallowed-methods = [
    { path = "alloc::vec::Vec::push", reason = "Use heapless::Vec in RT code" },
    { path = "alloc::boxed::Box::new", reason = "No heap in RT code" },
    { path = "alloc::string::String::new", reason = "Use heapless::String" },
]

Detecting Missing Timeouts

Pattern: I/O Operations Without Timeout

Anti-pattern:

#![allow(unused)]
fn main() {
// BAD: No timeout - can block forever
let data = socket.read(&mut buf)?;
}

Correct pattern:

#![allow(unused)]
fn main() {
// GOOD: Explicit timeout
socket.set_read_timeout(Some(Duration::from_millis(100)))?;
let data = socket.read(&mut buf)?;
}

Custom Lint Approach

A Dylint lint can flag an I/O call that has no timeout configured before it in the same scope:

#![allow(unused)]
fn main() {
// Conceptual lint logic (simplified)
fn check_io_without_timeout(expr: &Expr) {
    if is_io_operation(expr) && !has_timeout_in_scope(expr) {
        emit_warning("I/O operation without timeout may block indefinitely");
    }
}
}

Manual Audit Checklist

For code review, check that these operations have timeouts:

  • TcpStream::connect() – use connect_timeout()
  • socket.read() / socket.write() – set socket timeout
  • channel.recv() – use recv_timeout()
  • zenoh operations – configure session timeout
  • Any blocking syscall

Detecting Unbounded Loops

Clippy Detection

cargo clippy -- -D clippy::infinite_iter -D clippy::while_immutable_condition

Manual Patterns to Audit

Potentially unbounded:

#![allow(unused)]
fn main() {
// BAD: No termination guarantee
loop {
    if let Some(msg) = queue.pop() {
        process(msg);
    }
}

// BAD: Condition may never be false
while !flag.load(Ordering::Relaxed) {
    do_work();
}
}

Bounded alternatives:

#![allow(unused)]
fn main() {
// GOOD: Bounded iteration count
for _ in 0..MAX_ITERATIONS {
    if let Some(msg) = queue.pop() {
        process(msg);
    } else {
        break;
    }
}

// GOOD: Timeout-based termination
let deadline = Instant::now() + Duration::from_millis(10);
while Instant::now() < deadline {
    if let Some(msg) = queue.pop() {
        process(msg);
    } else {
        break;
    }
}
}

Custom Lint for Loop Bounds

Clippy cannot prove a loop terminates. A custom Dylint lint can check that:

  1. every loop block has a reachable break or return;
  2. every while condition involves a variable that the loop body modifies;
  3. a loop inside a task function has an explicit iteration limit.

Detecting Recursion

cargo-call-stack Cycle Detection

# Will report cycles in call graph
cargo +nightly call-stack --release 2>&1 | grep -i "cycle"

Clippy Recursion Lint

# Warn on unconditional recursion
cargo clippy -- -D clippy::unconditional_recursion

Manual Pattern Detection

#![allow(unused)]
fn main() {
// BAD: Direct recursion
fn process(node: &Node) {
    process(&node.child);  // May overflow stack
}

// BAD: Mutual recursion
fn a() { b(); }
fn b() { a(); }

// GOOD: Iterative with explicit stack
fn process(root: &Node) {
    let mut stack: heapless::Vec<&Node, 32> = heapless::Vec::new();
    stack.push(root).ok();

    while let Some(node) = stack.pop() {
        // Process node
        if stack.push(&node.child).is_err() {
            // Handle stack full - bounded!
            break;
        }
    }
}
}

Advanced: Custom Dylint Lints

Setting Up Dylint

# Install dylint
cargo install cargo-dylint dylint-link

# Create a new lint library
cargo dylint new realtime_lints
cd realtime_lints

Example: Detect Blocking in Async

A lint is a late lint pass over the compiler’s typed AST. The skeleton below shows the shape; the two checks in check_expr are the part you write:

#![allow(unused)]
fn main() {
// src/lib.rs
use clippy_utils::diagnostics::span_lint;
use rustc_lint::{LateContext, LateLintPass};
use rustc_session::declare_lint_pass;

declare_lint! {
    pub BLOCKING_IN_ASYNC,
    Warn,
    "blocking operations in async context"
}

declare_lint_pass!(BlockingInAsync => [BLOCKING_IN_ASYNC]);

impl<'tcx> LateLintPass<'tcx> for BlockingInAsync {
    fn check_expr(&mut self, cx: &LateContext<'tcx>, expr: &'tcx Expr<'tcx>) {
        // Check if we're in an async function
        // Check if expr is a blocking call (std::thread::sleep, std::fs::*, etc.)
        // Emit lint if both conditions are true
    }
}
}
LintPurpose
blocking_in_asyncDetect std::thread::sleep, blocking I/O in async
unbounded_loop_in_taskFlag loops without bounds in task functions
heap_in_interruptDetect allocation in #[interrupt] handlers
mutex_in_interruptFlag std::sync::Mutex in interrupt context
float_in_isrWarn about FP operations in ISRs (soft-float targets)
missing_timeoutI/O operations without timeout

Running the Checks in CI

The checks above run unattended. A GitHub Actions workflow for your own project, with the firmware crate in firmware/ and the host-testable crates named my_core and my_msgs (substitute your own):

name: Real-Time Static Analysis

on: [push, pull_request]

jobs:
  clippy-realtime:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: dtolnay/rust-toolchain@stable
        with:
          components: clippy
      - name: Clippy with RT lints
        run: |
          cargo clippy --all-targets -- \
            -D warnings \
            -D clippy::infinite_iter \
            -D clippy::while_immutable_condition \
            -D clippy::unconditional_recursion \
            -W clippy::large_stack_arrays

  stack-analysis:
    runs-on: ubuntu-latest
    defaults:
      run:
        working-directory: firmware
    steps:
      - uses: actions/checkout@v4
      - uses: dtolnay/rust-toolchain@nightly
      - name: Install cargo-call-stack
        run: cargo +nightly install cargo-call-stack
      - name: Analyze the firmware
        run: |
          RUSTFLAGS="-Z emit-stack-sizes" cargo +nightly build --release
          cargo +nightly call-stack --release > stack_analysis.txt
          cat stack_analysis.txt
      - name: Check for recursion
        run: |
          if grep -q "cycle" stack_analysis.txt; then
            echo "ERROR: Recursion detected in call graph"
            exit 1
          fi

  miri-safety:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: dtolnay/rust-toolchain@nightly
        with:
          components: miri
      - name: Run Miri
        run: cargo +nightly miri test -p my_core -p my_msgs

Miri runs host tests, so point it at the crates whose logic does not depend on the target: serialization, state machines, protocol code.

Tool Summary

ToolDetectsEffortCI-Ready
ClippyInfinite iters, recursion, large arraysLowYes
cargo-call-stackStack usage, recursion cyclesMediumYes
MiriUndefined behavior, memory errorsLowYes
DylintCustom patterns (blocking, heap, etc.)HighYes
MIRAIAbstract interpretation, all pathsHighExperimental
no_stdPrevents std heap allocationLowAutomatic
heaplessStatic collectionsLowN/A

Quick Reference: Lint Flags

# Copy-paste for strict real-time checking
cargo clippy -- \
    -D warnings \
    -D clippy::all \
    -D clippy::infinite_iter \
    -D clippy::while_immutable_condition \
    -D clippy::never_loop \
    -D clippy::empty_loop \
    -D clippy::unconditional_recursion \
    -W clippy::large_stack_arrays \
    -W clippy::large_types_passed_by_value \
    -W clippy::cognitive_complexity

WCET Benchmark Harness (no measurements yet)

A Worst-Case Execution Time (WCET) benchmark harness for nros core operations, built around DWT cycle counting on Cortex-M3. It has produced no numbers to date — every in-tree run is under QEMU, where DWT reads zero (see the limitation below), so the tables in this section list the benchmark categories, not results. Producing real baselines needs physical hardware (e.g. STM32F4 @ 168 MHz).

Measurement Platform

PropertyValue
TargetARM Cortex-M3 (thumbv7m-none-eabi)
MachineQEMU lm3s6965evb (for infrastructure validation)
ClockDWT CYCCNT (cycle-accurate on real hardware)
Iterations100 per benchmark (10 warmup)
Optimizationrelease profile (opt-level = "s")

DWT limitation on QEMU: QEMU’s Cortex-M3 emulation does not implement the DWT cycle counter – all reads return 0. The benchmark infrastructure is validated by the fact that it compiles, runs, and produces structured output. Actual cycle counts must be collected on real hardware (e.g., STM32F4 at 168 MHz, STM32F7 at 216 MHz).

Benchmark Categories

CDR Serialization:

FunctionNotes
serialize Int32Single i32 field
deserialize Int32Single i32 field
serialize TimeTwo fields (i32 + u32)
roundtrip Int32Serialize + deserialize
serialize w/ headerCDR encapsulation header + Int32

Node API:

FunctionNotes
Node::new()standalone node creation (nros_node::StandaloneNode)
create_publisher()Register Int32 publisher
serialize_message()Node-level serialize to buffer

Safety E2E:

FunctionNotes
crc32 (64B)CRC-32/ISO-HDLC, 64-byte payload
crc32 (256B)CRC-32/ISO-HDLC, 256-byte payload
crc32 (1024B)CRC-32/ISO-HDLC, 1024-byte payload
validate()SafetyValidator sequence check
full pipeline (128B)Extract attachment + CRC + validate

Sanity Bound

The harness asserts a 100,000-cycle ceiling per function for typical payloads — a design assertion baked into the benchmark, not a measured bound (no measured WCET exists yet; see above):

  • CRC-32 is O(n) in payload size with a single table lookup per byte
  • SafetyValidator::validate() is O(1) – a few comparisons and an increment
  • The full pipeline combines CRC computation with attachment parsing and validation

Running the harness in the tree: see Internals → Running the RT tests.

Static WCET Analysis

For certification (ISO 26262 ASIL C/D, DO-178C DAL A/B), dynamic measurement is insufficient – static WCET analysis is required. Candidate tools:

ToolTypeLicenseNotes
PlatinStatic (IPET)Open sourceAnalyzes LLVM IR + machine code
aiTStatic (abstract interpretation)Commercial (AbsInt)Industry standard for DO-178C / ISO 26262
SWEETStatic (flow analysis)AcademicResearch tool; good for Cortex-M
ChronosStatic (IPET)Open sourceAcademic; limited embedded target support

Recommended path: Use Platin for open-source WCET bounding on the LLVM IR produced by rustc. For certification evidence, aiT provides the strongest tool qualification story (pre-qualified per ISO 26262 Part 8).

References