Valós idejű rendszerek építése RTIC keretrendszerrel Cortex-M-en

Ha eddig cortex-m-rt-vel és nyers interrupt handler-ekkel dolgoztál, valószínűleg találkoztál már azzal a problémával, hogy két megszakítás egyszerre akar hozzáférni ugyanahhoz az állapothoz, és a critical_section::with meg a RefCell<Option<T>> kombók egy idő után kezelhetetlenné válnak. Az RTIC pontosan erre a problémára ad elegáns, fordítási időben ellenőrzött megoldást.

Az RTIC alapfilozófiája: prioritásalapú megszakításkezelés

Az RTIC nem egy hagyományos, preemptív ütemezésű RTOS abban az értelemben, ahogy a FreeRTOS vagy a Zephyr. Nincs benne szálkezelés, nincs stack-enkénti context switch. Ehelyett a Cortex-M NVIC (Nested Vectored Interrupt Controller) hardveres prioritáskezelésére épít, és ezt egészíti ki a Stack Resource Policy (SRP) néven ismert ütemezési algoritmussal.

A lényeg: minden task-hoz (feladathoz) egy prioritási szintet rendelsz, és az RTIC garantálja, hogy egy magasabb prioritású task bármikor megszakíthat egy alacsonyabb prioritásút, de fordítva soha nem történhet meg deadlock vagy priority inversion. Ez determinisztikus viselkedést jelent – pontosan azt, amire egy valós idejű rendszerben szükséged van.

Egy minimális RTIC alkalmazás váza így néz ki:

#![no_main]
#![no_std]

use panic_halt as _;

#[rtic::app(device = stm32f4xx_hal::pac, peripherals = true)]
mod app {
    use stm32f4xx_hal::prelude::*;

    #[shared]
    struct Shared {
        counter: u32,
    }

    #[local]
    struct Local {}

    #[init]
    fn init(cx: init::Context) -> (Shared, Local, init::Monotonics) {
        let _rcc = cx.device.RCC.constrain();

        (Shared { counter: 0 }, Local {}, init::Monotonics())
    }
}

Figyeld meg, hogy nincs main függvény a hagyományos értelemben – az #[init] task veszi át a szerepét, és a makró generálja le a reset handler-t, ami ezt meghívja.

Shared resources és a race condition-mentes hozzáférés

Az RTIC egyik legnagyobb erénye, hogy a megosztott állapotot (#[shared] struct mezőket) statikus elemzéssel kezeli. Amikor egy task hozzá akar férni egy shared resource-hoz, a .lock() metódust kell hívnod, ami a task prioritása alapján ideiglenesen megemeli a hardveres prioritási maszkot, amíg a kritikus szakasz tart.

#[task(binds = TIM2, shared = [counter], priority = 2)]
fn tim2_handler(mut cx: tim2_handler::Context) {
    cx.shared.counter.lock(|counter| {
        *counter += 1;
    });
}

A lényeg, hogy ez a lock nem egy futásidejű mutex — nincs benne semaphore, nincs blokkolás a hagyományos értelemben. A fordító és az RTIC makró már build időben kiszámolja, hogy melyik prioritási szinteket kell maszkolni ahhoz, hogy a hozzáférés biztonságos legyen, és ezt közvetlenül a BASEPRI regiszter (vagy Cortex-M0 esetén a PRIMASK) manipulálásával oldja meg.

Ez azért kritikus, mert Cortex-M-en a klasszikus Mutex<RefCell<T>> minta (amit gyakran látsz nyers cortex-m projektekben) futásidejű borrow-checket igényel, ami paniolhat, ha hibázol. Az RTIC-nél ez a hiba fordítási időben derül ki, mert a borrow checker és a lock rendszer együtt garantálja, hogy nem lehet egyszerre két helyről mutable referenciát szerezni ugyanarra a resource-ra.

Hardware task-ok definiálása megszakítás-vezérlőkhöz

A hardware task-ok közvetlenül egy periféria megszakításához vannak kötve a binds attribútummal. Amikor a hardver (pl. egy timer túlcsordul, vagy egy ADC konverzió befejeződik) kiváltja a megszakítást, az RTIC futásidejű rendszere lefuttatja a hozzá tartozó task-ot.

#[task(binds = EXTI0, local = [button_pressed: bool = false], priority = 3)]
fn button_handler(cx: button_handler::Context) {
    *cx.local.button_pressed = true;
    // itt törölnéd a pending flag-et is a periférián
}

Figyeld meg a local resource-t: ez egy task-ra privát, statikusan allokált állapot, ami nem oszlik meg más task-okkal, ezért nem is kell hozzá lock. Ez egy gyakori minta olyan esetekben, amikor egy interrupt handler-nek meg kell jegyeznie valamit a hívások között (pl. debounce számláló), de más task-nak nincs rá szüksége.

Software task-ok és üzenetküldés köztük

A hardware task-ok mellett az RTIC lehetővé teszi software task-ok definiálását is, amiket nem egy periféria vált ki, hanem explicit spawn() hívással indítasz el más task-okból. Ezek is rendelkeznek prioritással, és paramétereket is át lehet nekik adni – ez gyakorlatilag egy típusbiztos üzenetküldő mechanizmus interruptok között.

#[task(shared = [counter], priority = 1)]
fn process_measurement(mut cx: process_measurement::Context, raw_value: u16) {
    cx.shared.counter.lock(|counter| {
        if raw_value > 2048 {
            *counter += 1;
        }
    });
}

#[task(binds = ADC1, priority = 3)]
fn adc_handler(cx: adc_handler::Context) {
    let value: u16 = 1234; // itt olvasnád ki az ADC regisztert
    process_measurement::spawn(value).ok();
}

A spawn egy előre lefoglalt, lock-mentes queue-ba teszi be az üzenetet, aminek méretét az RTIC build időben számolja ki a lehetséges hívások alapján. Ha a queue megtelik (mert pl. túl gyakran hívod egy alacsonyabb prioritású task-ra), a spawn Err-t ad vissza, amit érdemes explicit kezelni, ne csak .ok()-val eldobni éles kódban.

Egy egyszerű valós idejű mérési projekt bemutatása

Nézzünk egy teljesebb példát: egy hőmérséklet-mérő rendszert, ami periodikusan mintavételez egy ADC-n keresztül, és ha a mért érték túllép egy küszöböt, riasztást ad egy GPIO pin-en keresztül. A monoton időzítéshez az RTIC 1.x-ben a #[monotonic] attribútumot és egy DWT vagy timer-alapú monoton órát használunk (a dwt-systick-monotonic crate-et).

#[rtic::app(device = stm32f4xx_hal::pac, dispatchers = [USART1])]
mod app {
    use dwt_systick_monotonic::{DwtSystick, ExtU32};

    #[monotonic(binds = SysTick, default = true)]
    type MyMono = DwtSystick<48_000_000>;

    #[shared]
    struct Shared {
        last_reading: u16,
    }

    #[local]
    struct Local {
        alarm_pin: bool,
    }

    #[init]
    fn init(mut cx: init::Context) -> (Shared, Local, init::Monotonics) {
        let mono = DwtSystick::new(
            &mut cx.core.DCB,
            cx.core.DWT,
            cx.core.SYST,
            48_000_000,
        );

        periodic_sample::spawn_after(100u32.millis()).ok();

        (
            Shared { last_reading: 0 },
            Local { alarm_pin: false },
            init::Monotonics(mono),
        )
    }

    #[task(shared = [last_reading], local = [alarm_pin])]
    fn periodic_sample(mut cx: periodic_sample::Context) {
        let value: u16 = 1500; // szimulált ADC olvasás

        cx.shared.last_reading.lock(|reading| {
            *reading = value;
        });

        if value > 2000 && !*cx.local.alarm_pin {
            *cx.local.alarm_pin = true;
            // itt húznánk fel a riasztó GPIO-t
        }

        periodic_sample::spawn_after(100u32.millis()).ok();
    }
}

Ebben a példában a dispatchers = [USART1] egy olyan, ténylegesen nem használt megszakítási vonalat jelöl ki, amit az RTIC software task-ok futtatásához használ, ha nincs elég valódi hardware interrupt a rendszerben. A spawn_after az ExtU32 trait segítségével kényelmes, olvasható időegység-jelöléseket tesz lehetővé (100u32.millis()), amit a fugit crate biztosít a háttérben.

Összefoglalás

Az RTIC egy remek példa arra, hogyan lehet a Rust típusrendszerét és fordítási idejű garanciáit kihasználva olyan valós idejű beágyazott rendszereket építeni, amikben a race condition-ök és a priority inversion egyszerűen nem fordulhatnak elő – nem azért, mert körültekintően programoztál, hanem mert a fordító nem is engedi lefordulni a hibás kódot. A shared resource-ok lock rendszere, a hardware és software task-ok közötti explicit üzenetküldés, valamint a build időben generált, statikusan allokált queue-k együtt egy kiszámítható, alacsony overhead-ű keretrendszert adnak, ami ideális választás lehet szenzoradat-gyűjtéstől kezdve motorvezérlésig szinte bármilyen determinisztikus időzítést igénylő Cortex-M projekthez. Ha még nem próbáltad, érdemes egy STM32 vagy nRF52 fejlesztőpanelen elindítani az első #[app] modulodat – a tanulási görbe meredek, de a megtérülés is az.