Periodic Framework — Rust API

Declare a period; the framework owns the timing loop, topic admission, freshness and the publish. A node is a function with an attribute — there is no main to write.

A periodic node is what you would otherwise create by hand as a thread running a timer loop — spawn a task, read the clock, do the work, sleep for the remainder, repeat. The framework replaces that loop, so you never write the thread, the sleep, or the drift correction the hand-written version needs.

For why the framework works this way, see the design document. This page is the API.

Quick start

Two programs: one that produces a reading on a topic, one that consumes it. Neither writes a main, a thread, or a sleep.

Add the crate as a direct dependency. It is standalone — not re-exported through tail_core — so adding it forces no toolchain or std rebuild.

[dependencies]
periodic = { path = "<relative path to>/framework/periodic/lib" }

A publisher

Reads a sensor every 20 ms and publishes the reading. This is the whole program:

#![feature(tail_core)]

use periodic::periodic;

#[derive(Clone, Copy)]
#[repr(C)]
struct Range { stamp: u64, range_mm: u32, quality: u32 }

// SAFETY: `repr(C)`, all-integer, every bit pattern valid.
unsafe impl std::tail_core::sync::shared_mutex::ShmSafe for Range {}

#[periodic(every = "20ms", publish = "sensor/range", lane = safe, on_overrun = Skip)]
fn lidar() -> Range {
    Range { stamp: 0x1234_5678, range_mm: 4200, quality: 97 }
}

A subscriber

Reads that topic every 20 ms and brakes on it. stale_after says how old a reading may be before it counts as absent, so the fail-safe branch is the one that runs when the sensor goes quiet:

#![feature(tail_core)]

use periodic::{periodic, Cycle};

// Must match the publisher's payload exactly.
#[derive(Clone, Copy)]
#[repr(C)]
struct Range { stamp: u64, range_mm: u32, quality: u32 }

// SAFETY: `repr(C)`, all-integer, every bit pattern valid.
unsafe impl std::tail_core::sync::shared_mutex::ShmSafe for Range {}

#[periodic(every = "20ms", subscribe = "sensor/range", lane = safe,
           stale_after = "60ms", on_overrun = Skip)]
fn brake(lidar: Cycle<Range>) {
    match lidar.get() {
        Some(reading) => brake_for(reading.range_mm),
        None          => brake_for(0),   // absent or stale
    }
}

Neither program names the other. The framework opens each topic and retries, so the two can start in any order.

#[periodic] keys

every — required

The release period. Releases fall on start + n * period, never drifting.

#[periodic(every = "20ms", ...)]

lane — required

Which transport carries the topic. All three move the same payload and converge on one API, so changing lane is a change to this line and nothing else.

Choose the one with the least machinery that meets the requirement:

more than one node publishes the topic?            ->  kernel   (the only lane that allows it)
payload under ~2 KB                                ->  kernel
large payload, a reader has a safety requirement   ->  safe
large payload, nothing downstream is safety-rated  ->  fast

Zero-copy is not "the fast one, always". Below that crossover safe is slower than the copy — two kernel calls and a mapping teardown against one call and a small memcpy — and it buys a safety argument that did not need to be made.

#[periodic(lane = safe, ...)]

Full detail under Lanes.

on_overrun — required

What to do when a cycle runs past its next release: Skip, CatchUp, or Fault. No default — omitting it does not compile.

#[periodic(on_overrun = Skip, ...)]

publish

Topic to publish the body's return value on.

#[periodic(every = "20ms", publish = "sensor/range", lane = safe, on_overrun = Skip)]
fn lidar() -> Range { read_sensor() }

subscribe

Topic to read from. The body receives a Cycle<T>.

#[periodic(every = "20ms", subscribe = "sensor/range", lane = safe,
           stale_after = "60ms", on_overrun = Skip)]
fn brake(lidar: Cycle<Range>) { ... }

stale_after — required with subscribe

How old a sample may be before it reads as absent.

#[periodic(subscribe = "sensor/range", stale_after = "60ms", ...)]

init

Builds the node's state once, before the first release. The body takes it &mut, first.

#[periodic(every = "20ms", publish = "diag/count", lane = safe, on_overrun = Skip,
           init = Cycles::default())]
fn counter(cycles: &mut Cycles) -> Count {
    cycles.0 += 1;
    Count { cycles: cycles.0 }
}

safety

Bare flag: this topic carries safety-relevant data. Binding it to the QM-only fast lane fails the build.

#[periodic(every = "10ms", publish = "brake/command", lane = safe,
           on_overrun = Fault, safety)]
fn brake_command() -> Command { ... }

Keys as constants

String-valued keys are expressions, so a test can assert on the same constant.

const PERIOD: &str = "20ms";
const TOPIC: &str = "sensor/range";

#[periodic(every = PERIOD, publish = TOPIC, lane = safe, on_overrun = Skip)]
fn lidar() -> Range { read_sensor() }

Node shapes

Publisher

#[periodic(every = "20ms", publish = "sensor/range", lane = safe, on_overrun = Skip)]
fn lidar() -> Range { read_sensor() }

Publisher that may skip a cycle

Return Option<T>. None publishes nothing and is not an overrun.

#[periodic(every = "20ms", publish = "sensor/range", lane = safe, on_overrun = Skip)]
fn lidar() -> Option<Range> { read_sensor_if_ready() }

Prefer T when the node always produces a sample: Option<T> costs a copy that T does not.

Subscriber

#[periodic(every = "20ms", subscribe = "sensor/range", lane = safe,
           stale_after = "60ms", on_overrun = Skip)]
fn brake(lidar: Cycle<Range>) {
    brake_for(lidar.get().map_or(0, |r| r.range_mm));
}

Pipe — reads and publishes

#[periodic(every = "20ms", subscribe = "sensor/range", publish = "estimate/distance",
           lane = safe, stale_after = "60ms", on_overrun = Skip)]
fn filter(raw: Cycle<Range>) -> Distance {
    Distance { mm: raw.get().map_or(0, |r| r.range_mm) }
}

With state

State comes first, then the Cycle if the node subscribes.

#[periodic(every = "5ms", subscribe = "input/pad", lane = kernel, stale_after = "300ms",
           on_overrun = Skip, init = Motor::attach())]
fn drive(motor: &mut Motor, pad: Cycle<Command>) {
    motor.set(pad.get().map_or(Command::Stop, |c| *c));
}

Cycle<T>

Everything about this activation, on one value. Passed to any node that subscribes.

get() -> Option<&T>

The sample, if one arrived and is still fresher than stale_after. Stale reads as absent, so the fail-safe branch is the default branch.

match lidar.get() {
    Some(reading) => brake_for(reading.range_mm),
    None          => brake_for(0),   // absent or stale
}

copied() -> Option<T>

The same sample, copied out, for a node that would rather own it.

let reading = lidar.copied().unwrap_or_default();

since_received() -> Option<Duration>

How long since a sample last arrived at this node. Not the sample's publication age — no transport timestamps a publication.

if let Some(gap) = lidar.since_received() {
    if gap > Duration::from_millis(100) { raise_sensor_warning(); }
}

was_stale() -> bool

Whether data was held but had aged past stale_after. Only for a node that needs the distinction get() collapses.

if lidar.was_stale() {
    log::warn!("using fail-safe: last reading too old");
}

missed() -> u32

Publications lost since the previous cycle. Not a running total.

if cycle.missed() > 0 {
    count_dropped(cycle.missed());
}

rejected() -> u32

Publications the transport refused since the previous cycle. Not a running total.

if cycle.rejected() > 0 {
    count_rejected(cycle.rejected());
}

overruns() -> u32

Releases missed since the previous cycle ran. Delivered once, then cleared.

if cycle.overruns() > 0 {
    log::warn!("missed {} release(s)", cycle.overruns());
}

is_healthy() -> bool

No loss, no rejection, no overrun. The one-call form of the three above.

if !cycle.is_healthy() {
    raise_degraded();
}

Cycle borrows its sample and is deliberately not Send: a topic must be read on the thread that subscribed, so moving it across threads is a compile error.

OverrunPolicy

Skip

Abandon the missed releases, resume on the grid. Preserves phase. The default choice for a control loop, where a late command computed from stale inputs is worse than none.

#[periodic(on_overrun = Skip, ...)]

CatchUp

Run the missed releases back to back until caught up. Preserves count. For an integrator or accumulator that must see every interval exactly once.

#[periodic(on_overrun = CatchUp, ...)]

Fault

Stop the node and report. For a node whose deadline is the safety requirement. Exits with NodeError::DeadlineMissed rather than success, so a supervisor can tell a missed deadline from a clean shutdown.

#[periodic(on_overrun = Fault, ...)]

Lanes

Cost and writer count are in lane. What decides a lane beyond those is the safety classification, which the compiler enforces.

fast is Quality Managed by its own design: its writer holds a standing read-write mapping of every slot for the region's lifetime, so it can alter a frame a reader is parsing. That is detection, not prevention.

Declaring safety on a fast topic therefore fails the build rather than a review:

#[periodic(every = "10ms", publish = "brake/command", lane = fast,
           on_overrun = Fault, safety)]
fn brake_command() -> Command { ... }
//  error: a safety-classified topic cannot be carried on a QM-only lane

Known gap. The lane is not compared between publisher and subscriber. Two processes that disagree fail as a subscriber that silently never receives anything.

Durations

An integer and a unit. No fractional form — write "1500us", not "1.5ms".

Payloads

A payload crosses shared memory: repr(C), Copy, and every bit pattern valid. The publisher's and subscriber's types must match exactly.

#[derive(Clone, Copy)]
#[repr(C)]
struct Range { stamp: u64, range_mm: u32, quality: u32 }

// SAFETY: `repr(C)`, all-integer, every bit pattern valid.
unsafe impl std::tail_core::sync::shared_mutex::ShmSafe for Range {}

NodeError

A node that cannot start says which topic and why, then exits non-zero. It does not run forever reading None.

Admission retries because the kernel refuses a name nobody publishes yet — so a misspelled topic fails loudly at startup rather than becoming a topic that never delivers.

Executor

Most nodes never touch this; the attribute drives it. It is public because the cadence is testable without a topic.

Executor::new(clock, period, policy)

Start releasing every period, from now. Fails with StartError::UnusablePeriod if the period is zero or rounds to zero ticks.

let mut executor = Executor::new(clock, Duration::from_millis(20), OverrunPolicy::Skip)?;

run(body)

Run until the policy stops the node. The closure receives this cycle's overrun count.

executor.run(|overruns| {
    if overruns > 0 { log::warn!("missed {overruns} release(s)"); }
    do_work();
});

step(body) -> bool

Run one cycle. Returns false once the policy has stopped the node — only Fault does.

while executor.step(|_overruns| do_work()) {}

release() / period()

The next release, and the declared period, in clock ticks.

let next = executor.release();

Clock

What the executor sleeps against: now(), ticks_per_second(), sleep_until(deadline). Releases are slept to, never "now + period" slept for — a relative sleep adds each cycle's scheduling delay to the period and never recovers it.

next_release(policy, release, period, now)

Pure function over ticks: how many releases were missed, and where the next one falls.

let (missed, next) = next_release(OverrunPolicy::Skip, release, period, now);

period_ticks(period, ticks_per_second)

A declared period in ticks, or None if it is zero or unrepresentable.

let ticks = period_ticks(Duration::from_millis(20), clock.ticks_per_second())?;