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-Pattern | Problem | Detection Method |
|---|---|---|
| Unbounded loops | Infinite execution time | Clippy + custom lints |
| Recursion | Stack overflow, unbounded depth | cargo-call-stack |
| Heap allocation | Non-deterministic timing, fragmentation | no_std + forbid patterns |
| Blocking I/O | Unbounded wait times | Custom lints |
| Missing timeouts | Operations can hang forever | Custom lints |
| Large stack frames | Stack overflow | cargo-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
| Lint | Detects |
|---|---|
infinite_iter | Iterator chains guaranteed to be infinite |
while_immutable_condition | Loop conditions that can never change |
never_loop | Loops that exit on first iteration |
empty_loop | loop { } 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
Recommended Clippy Configuration
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_stdprograms
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()– useconnect_timeout()socket.read()/socket.write()– set socket timeoutchannel.recv()– userecv_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:
- every
loopblock has a reachablebreakorreturn; - every
whilecondition involves a variable that the loop body modifies; - 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
}
}
}
Recommended Custom Lints for Real-Time
| Lint | Purpose |
|---|---|
blocking_in_async | Detect std::thread::sleep, blocking I/O in async |
unbounded_loop_in_task | Flag loops without bounds in task functions |
heap_in_interrupt | Detect allocation in #[interrupt] handlers |
mutex_in_interrupt | Flag std::sync::Mutex in interrupt context |
float_in_isr | Warn about FP operations in ISRs (soft-float targets) |
missing_timeout | I/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
| Tool | Detects | Effort | CI-Ready |
|---|---|---|---|
| Clippy | Infinite iters, recursion, large arrays | Low | Yes |
| cargo-call-stack | Stack usage, recursion cycles | Medium | Yes |
| Miri | Undefined behavior, memory errors | Low | Yes |
| Dylint | Custom patterns (blocking, heap, etc.) | High | Yes |
| MIRAI | Abstract interpretation, all paths | High | Experimental |
| no_std | Prevents std heap allocation | Low | Automatic |
| heapless | Static collections | Low | N/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
| Property | Value |
|---|---|
| Target | ARM Cortex-M3 (thumbv7m-none-eabi) |
| Machine | QEMU lm3s6965evb (for infrastructure validation) |
| Clock | DWT CYCCNT (cycle-accurate on real hardware) |
| Iterations | 100 per benchmark (10 warmup) |
| Optimization | release 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:
| Function | Notes |
|---|---|
serialize Int32 | Single i32 field |
deserialize Int32 | Single i32 field |
serialize Time | Two fields (i32 + u32) |
roundtrip Int32 | Serialize + deserialize |
| serialize w/ header | CDR encapsulation header + Int32 |
Node API:
| Function | Notes |
|---|---|
Node::new() | standalone node creation (nros_node::StandaloneNode) |
create_publisher() | Register Int32 publisher |
serialize_message() | Node-level serialize to buffer |
Safety E2E:
| Function | Notes |
|---|---|
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:
| Tool | Type | License | Notes |
|---|---|---|---|
| Platin | Static (IPET) | Open source | Analyzes LLVM IR + machine code |
| aiT | Static (abstract interpretation) | Commercial (AbsInt) | Industry standard for DO-178C / ISO 26262 |
| SWEET | Static (flow analysis) | Academic | Research tool; good for Cortex-M |
| Chronos | Static (IPET) | Open source | Academic; 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).