Real-Time Interrupt-driven Concurrency

Когда ресурсы (статические переменные) разделяются между двумя или более задачами, которые выполняются с разными приоритетами, некая форма запрета изменений необходима, чтобы изменять память без гонки данных. В RTIC мы используем основанные на приоритетах критические секции, чтобы гарантировать запрет изменений (см. Протокол немедленного максимального приоритета).

Критическия секция состоит во временном увеличении динамического приоритета задачи. Пока задача находится в критической секции, все другие задачи, которые могут послать запрос переменной не могут запуститься.

Насколько большим должен быть динамический приориткт, чтобы гарантировать запрет изменений определенного ресурса? Анализ приоритетов отвечает на этот вопрос и будет обсужден в следующем разделе. В этом разделе мы сфокусируемся на реализации критической секции.

Прокси-ресурсы

Для упрощения, давайте взглянем на ресурс, разделяемый двумя задачами, запускаемыми с разными приоритетами. Очевидно, что одна задача может вытеснить другую; чтобы предотвратить гонку данных задача с низким приоритетом должна использовать критическую секцию, когда необходимо изменять разделяемую память. С другой стороны, высокоприоритетная задача может напрямую изменять разделяемую память, поскольку не может быть вытеснена низкоприоритетной задачей. Чтобы заставить использовать критическую секцию на задаче с низким приоритетом, мы предоставляем прокси-ресурсы, в которых мы отдаем уникальную ссылку (&mut-) высокоприоритетной задаче.

Пример ниже показывает разные типы, передаваемые каждой задаче:

#![allow(unused)]

fn main() {

#[rtic::app(device = ..)]

mut app {

struct Resources {

#[init(0)]

x: u64,

}

#[interrupt(binds = UART0, priority = 1, resources = [x])]

fn foo(c: foo::Context) {

// прокси-ресурс

let mut x: resources::x = c.resources.x;

x.lock(|x: &mut u64| {

// критическая секция

*x += 1

});

}

#[interrupt(binds = UART1, priority = 2, resources = [x])]

fn bar(c: bar::Context) {

let mut x: &mut u64 = c.resources.x;

*x += 1;

}

// ..

}

}

Теперь давайте посмотрим. как эти типы создаются фреймворком.

#![allow(unused)]

fn main() {

fn foo(c: foo::Context) {

// .. пользовательский код ..

}

fn bar(c: bar::Context) {

// .. пользовательский код ..

}

pub mod resources {

pub struct x {

// ..

}

}

pub mod foo {

pub struct Resources {

pub x: resources::x,

}

pub struct Context {

pub resources: Resources,

// ..

}

}

pub mod bar {

pub struct Resources<'a> {

pub x: &'a mut u64,

}

pub struct Context {

pub resources: Resources,

// ..

}

}

mod app {

static mut x: u64 = 0;

impl rtic::Mutex for resources::x {

type T = u64;

fn lock(&mut self, f: impl FnOnce(&mut u64) -> R) -> R {

// мы рассмотрим это детально позднее

}

}

#[no_mangle]

unsafe fn UART0() {

foo(foo::Context {

resources: foo::Resources {

x: resources::x::new(/* .. */),

},

// ..

})

}

#[no_mangle]

unsafe fn UART1() {

bar(bar::Context {

resources: bar::Resources {

x: &mut x,

},

// ..

})

}

}

}

lock

Теперь давайте рассмотрим непосредственно критическую секцию. В этом примере мы должны увеличить динамический приоритет минимум до 2, чтобы избежать гонки данных. В архитектуре Cortex-M динамический приоритет можно изменить записью в регистр BASEPRI.

Семантика регистра BASEPRI такова:

• Запись 0 в BASEPRI отключает его функциональность.

• Запись ненулевого значения в BASEPRI изменяет уровень приоритета, требуемого для вытеснения прерывания. Однако, это имеет эффект, только когда записываемое значение меньше, чем уровень приоритета текущего контекста выполнения, но обращаем внимание, что более низкий уровень аппаратного приоритета означает более высокий логический приоритет

Таким образом, динамический приоритет в любой момент времени может быть рассчитан как

#![allow(unused)]

fn main() {

dynamic_priority = max(hw2logical(BASEPRI), hw2logical(static_priority))

}

Где static_priority - приоритет, запрограммированный в NVIC для текущего прерывания, или логический 0, когда текущий контекств - это idle.

В этом конкретном примере мы можем реализовать критическую секцию так:

#![allow(unused)]

fn main() {

impl rtic::Mutex for resources::x {

type T = u64;

fn lock(&mut self, f: F) -> R

where

F: FnOnce(&mut u64) -> R,

{

unsafe {

// начать критическую секцию: увеличить динамический приоритет до `2`

asm!("msr BASEPRI, 192" : : : "memory" : "volatile");

// запустить пользовательский код в критической секции

let r = f(&mut x);

// окончить критическую секцию: восстановить динамический приоритет до статического значения (`1`)

asm!("msr BASEPRI, 0" : : : "memory" : "volatile");

r

}

}

}

}

В данном случае важно указать "memory" в блоке asm!. Это не даст компилятору менять местами операции вокруг него. Это важно, поскольку доступ к переменной x вне критической секции привело бы к гонке данных.

Важно отметить, что сигнатура метода lock препятствет его вложенным вызовам. Это необходимо для безопасности памяти, так как вложенные вызовы привели бы к созданию множественных уникальных ссылок (&mut-) на x, ломая правила заимствования Rust. Смотреть ниже:

#![allow(unused)]

fn main() {

#[interrupt(binds = UART0, priority = 1, resources = [x])]

fn foo(c: foo::Context) {

// resource proxy

let mut res: resources::x = c.resources.x;

res.lock(|x: &mut u64| {

res.lock(|alias: &mut u64| {

//~^ ошибка: `res` уже был заимствован уникально (`&mut-`)

// ..

});

});

}

}

Вложенность

Вложенные вызовы lock на том же ресурсе должны отклоняться компилятором для безопасности памяти, однако вложенные вызовы lock на разных ресурсах - нормальная операция. В этом случае мы хотим убедиться, что вложенные критические секции никогда не приведут к понижению динамического приоритета, так как это плохо, и мы хотим оптимизировать несколько записей в регистр BASEPRI и compiler fences. Чтобы справиться с этим, мы проследим динамический приоритет задачи, с помощью стековой переменной и используем ее, чтобы решить, записывать BASEPRI или нет. На практике, стековая переменная будет соптимизирована компилятором, но все еще будет предоставлять информацию компилятору.

Рассмотрим такую программу:

#![allow(unused)]

fn main() {

#[rtic::app(device = ..)]

mod app {

struct Resources {

#[init(0)]

x: u64,

#[init(0)]

y: u64,

}

#[init]

fn init() {

rtic::pend(Interrupt::UART0);

}

#[interrupt(binds = UART0, priority = 1, resources = [x, y])]

fn foo(c: foo::Context) {

let mut x = c.resources.x;

let mut y = c.resources.y;

y.lock(|y| {

*y += 1;

*x.lock(|x| {

x += 1;

});

*y += 1;

});

// середина

x.lock(|x| {

*x += 1;

y.lock(|y| {

*y += 1;

});

*x += 1;

})

}

#[interrupt(binds = UART1, priority = 2, resources = [x])]

fn bar(c: foo::Context) {

// ..

}

#[interrupt(binds = UART2, priority = 3, resources = [y])]

fn baz(c: foo::Context) {

// ..

}

// ..

}

}

Код, сгенерированный фреймворком, выглядит так:

#![allow(unused)]

fn main() {

// опущено: пользовательский код

pub mod resources {

pub struct x<'a> {

priority: &'a Cell,

}

impl<'a> x<'a> {

pub unsafe fn new(priority: &'a Cell) -> Self {

x { priority }

}

pub unsafe fn priority(&self) -> &Cell {

self.priority

}

}

// repeat for `y`

}

pub mod foo {

pub struct Context {

pub resources: Resources,

// ..

}

pub struct Resources<'a> {

pub x: resources::x<'a>,

pub y: resources::y<'a>,

}

}

mod app {

use cortex_m::register::basepri;

#[no_mangle]

unsafe fn UART1() {

// статический приоритет прерывания (определено пользователем)

const PRIORITY: u8 = 2;

// сделать снимок BASEPRI

let initial = basepri::read();

let priority = Cell::new(PRIORITY);

bar(bar::Context {

resources: bar::Resources::new(&priority),

// ..

});

// вернуть BASEPRI значение из снимка, сделанного ранее

basepri::write(initial); // то же, что и `asm!` блок, виденный ранее

}

// так же для `UART0` / `foo` и `UART2` / `baz`

impl<'a> rtic::Mutex for resources::x<'a> {

type T = u64;

fn lock(&mut self, f: impl FnOnce(&mut u64) -> R) -> R {

unsafe {

// определение максимального приоритет ресурса

const CEILING: u8 = 2;

let current = self.priority().get();

if current < CEILING {

// увеличить динамический приоритет

self.priority().set(CEILING);

basepri::write(logical2hw(CEILING));

let r = f(&mut y);

// восстановить динамический приоритет

basepri::write(logical2hw(current));

self.priority().set(current);

r

} else {

// динамический приоритет достаточно высок

f(&mut y)

}

}

}

}

// повторить для ресурса `y`

}

}

Наконец, компилятор оптимизирует функцию foo во что-то наподобие такого:

#![allow(unused)]

fn main() {

fn foo(c: foo::Context) {

// ПРИМЕЧАНИЕ: BASEPRI содержит значение `0` (значение сброса) в этот момент

// увеличить динамический приоритет до `3`

unsafe { basepri::write(160) }

// две операции над `y` объединены в одну

y += 2;

// BASEPRI не изменяется для доступа к `x`, потому что динамический приоритет достаточно высок

x += 1;

// уменьшить (восстановить) динамический приоритет до `1`

unsafe { basepri::write(224) }

// средина

// увеличить динамический приоритет до `2`

unsafe { basepri::write(192) }

x += 1;

// увеличить динамический приоритет до `3`

unsafe { basepri::write(160) }

y += 1;

// уменьшить (восстановить) динамический приоритет до `2`

unsafe { basepri::write(192) }

// ПРИМЕЧАНИЕ: было вы правильно объединить эту операцию над  `x` с предыдущей, но

// compiler fences грубые и предотвращают оптимизацию

x += 1;

// уменьшить (восстановить) динамический приоритет до `1`

unsafe { basepri::write(224) }

// ПРИМЕЧАНИЕ: BASEPRI содержит значение `224` в этот момент

// обработчик UART0 восстановит значение `0` перед завершением

}

}

Инвариант BASEPRI

Инвариант, который фреймворк RTIC должен сохранять в том, что значение BASEPRI в начале обработчика прерывания должно быть таким же, как и при выходе из него. BASEPRI может изменяться в процессе выполнения обработчика прерывания, но но выполнения обработчика прерывания в начале и конце не должно вызвать наблюдаемого изменения BASEPRI.

Этот инвариант нужен, чтобы избежать уеличения динамического приоритета до значений, при которых обработчик не сможет быть вытеснен. Лучше всего это видно на следующем примере:

#![allow(unused)]

fn main() {

#[rtic::app(device = ..)]

mod app {

struct Resources {

#[init(0)]

x: u64,

}

#[init]

fn init() {

// `foo` запустится сразу после завершения `init`

rtic::pend(Interrupt::UART0);

}

#[task(binds = UART0, priority = 1)]

fn foo() {

// BASEPRI равен `0` в этот момент; динамический приоритет равен `1`

// `bar` вытеснит `foo` в этот момент

rtic::pend(Interrupt::UART1);

// BASEPRI равен `192` в этот момент (из-за бага); динамический приоритет равен `2`

// эта функция возвращается в `idle`

}

#[task(binds = UART1, priority = 2, resources = [x])]

fn bar() {

// BASEPRI равен `0` (динамический приоритет = 2)

x.lock(|x| {

// BASEPRI увеличен до `160` (динамический приоритет = 3)

// ..

});

// BASEPRI восстановлен до `192` (динамический приоритет = 2)

}

#[idle]

fn idle() -> ! {

// BASEPRI равен `192` (из-за бага); динамический приоритет = 2

// это не оказывает эффекта, из-за значени BASEPRI

// задача `foo` не будет выполнена снова никогда

rtic::pend(Interrupt::UART0);

loop {

// ..

}

}

#[task(binds = UART2, priority = 3, resources = [x])]

fn baz() {

// ..

}

}

}

ВАЖНО: давайте например мы забудем восстановить BASEPRI в UART1 -- из-за какого нибудь бага в генераторе кода RTIC.

#![allow(unused)]

fn main() {

// код, сгенерированный RTIC

mod app {

// ..

#[no_mangle]

unsafe fn UART1() {

// статический приоритет этого прерывания (определен пользователем)

const PRIORITY: u8 = 2;

// сделать снимок BASEPRI

let initial = basepri::read();

let priority = Cell::new(PRIORITY);

bar(bar::Context {

resources: bar::Resources::new(&priority),

// ..

});

// БАГ: ЗАБЫЛИ восстановить BASEPRI на значение из снимка

basepri::write(initial);

}

}

}

В результате, idle запустится на динамическом приоритете 2 и на самом деле система больше никогда не перейдет на динамический приоритет ниже 2. Это не компромис для безопасности памяти программы, а влияет на диспетчеризацию задач: в этом конкретном случае задачи с приоритетом 1 никогда не получат шанс на запуск.