Rust in Everything
基于 embassy-executor 0.10.0

embassy 解决什么问题

主线位置:三条主线的起点。本页回答为什么需要「任务挂起 → 中断唤醒 → 再次 poll」这样一套机制,它和裸机上别的并发写法差在哪;三条主线本身从一次 Timer::after 开始走。

flowchart TB
  q{等待时状态放在哪} --> loop1["主循环 全局变量"]
  q --> rtos["RTOS 每线程一个栈"]
  q --> rtic["RTIC 中断优先级"]
  q --> emb["embassy future 状态机"]
  loop1 --> busy["轮询或 WFI"]
  rtos --> ctx["内核切换上下文"]
  rtic --> nvic["NVIC 抢占"]
  emb --> wake["中断调 waker 再 poll"]

裸机上的四种并发写法

一块单核 MCU 要同时做几件事:每 100 ms 翻转一次 LED,等按键,收串口数据。没有操作系统,没有线程,通常也没有堆。常见的写法有四种,区别在于「一件事等待的时候,它的进度记在哪里,谁来叫醒它」:

  • 轮询主循环(superloop):一个 loop,每圈检查每件事该不该做。进度记在全局变量里(下一次翻转的时刻、串口收到第几个字节)。不想忙等就在圈尾执行 WFI,让任意中断把核心叫醒再转一圈。embassy 书里的 layer_by_layer 一章就是从这种写法出发的,它对 PAC 和 HAL 版的评价是「the application is busy looping and consuming more power than necessary」。
  • RTOS 线程:每件事一个线程,各有一个栈,阻塞调用让内核把它挂起、切到别的线程。进度记在线程栈上,代码可以写成顺序的;代价是每个线程的栈大小要事先估,切换要保存寄存器。
  • RTIC:把每件事写成一个中断处理函数,用中断优先级当任务优先级,靠 NVIC 硬件抢占来调度,共享资源按优先级上限加锁。进度记在静态资源里。
  • embassy:每件事写成一个 async fn,编译器把它变成一个状态机(future),进度就是状态机里跨 .await 存活的那些变量。任务等外设时返回 Pending,外设的中断处理函数调它的 Waker,执行器只 poll 被唤醒的任务;没有任务要跑时 WFE 睡眠。

本卷只对 embassy 做实验。RTOS 和 RTIC 本机没有搭,上面两条只作为对照写出它们各把状态放在哪;和 RTIC 的对比在 08 设计取舍。

embassy 自己怎么说,落在哪段代码上

仓库 README 和书的 runtime.adoc 列了执行器的几条特点。逐条对到 0.10.0 的代码:

原文代码里是什么结论
「No alloc, no heap needed. Tasks are statically allocated.」#[embassy_executor::task] 为每个任务函数生成一个 static POOL: TaskPool<Fut, POOL_SIZE>,TaskPool 就是 [TaskStorage<F>; N];#[main] 生成的 Executor 放在 main 的栈上成立;下面的实验里两个任务池在 .bss 里各 192、72 字节
「runs on a single stack, so no per-task stack size tuning is required」(README)跨 .await 的变量在 future 里,future 在 static 里;poll 时用的是调用它的那个栈成立;中断处理函数也用同一个主栈(Cortex-M 的 MSP)
「No "fixed capacity" data structures, executor works with 1 or 1000 tasks」运行队列是 cordyceps 的侵入式链表,链表节点在每个任务头里(RunQueue);集成定时器队列同样把节点放在任务头里对执行器成立;开 generic-queue-N 时定时器队列就是固定容量的
「No busy-loop polling: CPU sleeps when there's no work to do」Executor::run 是 loop { poll(); wfe }成立,但醒几次取决于时间驱动:本卷的 SysTick 驱动每毫秒一个中断,100 ms 的等待里执行器 poll 了 200 次,只有最后一次有任务可跑,见一次 Timer::after
「Efficient polling: a wake will only poll the woken task, not all of them.」只有 wake_task 会把任务放进运行队列,dequeue_all 只取运行队列里的成立
「Once a task yields, the executor enqueues the task at the end of the run queue」返回 Pending 的任务不会被放回去;而且一批里的任务是倒着取的(TransferStack 是栈,源码注释写着「batches will be iterated in the reverse order as they were enqueued」)书和代码对不上:前半句不成立,「at the end」也不成立

最后一行的倒序能在下面的实验里直接看到。

和 tokio 比

embassy 的任务、Waker、Future 和 tokio 用的是同一套 core::task 接口,差别在接口下面。只列能在代码里核实的:

  • 任务在哪。 tokio 的 spawn 每次在堆上分配一个任务(任务的内存布局);embassy 的任务存储在编译期就定好了,是 .bss 里的 TaskPool。池子满了,调用任务函数拿 SpawnToken 时就返回 SpawnError::Busy,报错文字是「Too many instances of this task are already running. Check the pool_size attribute of the task.」。也没有 JoinHandle:Spawner::spawn 什么都不返回。
  • 谁叫醒任务。 tokio 的 worker 没活干时停在 I/O 驱动里(epoll_wait / kevent,见驱动栈与 park 和 mio 卷的事件循环),事件和定时器都由驱动处理后再唤醒任务。embassy 没有驱动线程,也没有反应器:叫醒任务的是外设自己的中断处理函数,执行器睡在 WFE 上,被唤醒靠的是一条 SEV(__pender)或中断本身。
  • 任务参数必须 'static。 任务的 future 要放进 static,所以宏在展开时就检查参数类型:引用不写生命周期、写了 'static 以外的生命周期,都报「Arguments for tasks must live forever. Try using the 'static lifetime.」;参数里有 impl Trait、函数本身是泛型也不行(check_arg_ty)。所以传给任务的外设是 Peri<'static, _>,见外设单例。
  • 一个执行器就是一个线程。 Cortex-M 的 Executor 带着 PhantomData<*mut ()>,不是 Send;同一个执行器里的任务协作式地轮流跑,一个任务不返回 Pending,别的任务就没有机会跑。要抢占,就再开一个 InterruptExecutor,让它在更高优先级的中断里 poll 自己的任务(各平台的执行器)。tokio 的多线程运行时靠的是多个 worker 线程和任务窃取。

完整的对照在 08 设计取舍。

📝NOTE
没有蠢问题

问: 任务的 future 全在 static 里,那 async fn 里的局部变量放在哪?栈上那么多 let 不也得有地方放吗?

答: 分两种。活过某个 .await 的变量,编译器把它们放进 future 的状态里,也就是 static POOL 里那块 TaskStorage;只在两个 .await 之间用到的,就是普通的栈变量,只在 poll 那一下占用调用者的栈,poll 返回就没了。所以所有任务轮流借用同一个栈,栈只需要够最深的那一次 poll 加上中断嵌套用。代价是 future 的大小在编译期就定了:跨 .await 拿着一个大数组,.bss 就大一块。

实验:两个周期任务有多大

同一件事写两遍,放在 QEMU 11.1.2 b-l475e-iot01a 上跑。两个程序都调 embassy_stm32::init(用的是 QEMU 补丁版的 embassy-stm32:只注释掉 rcc/bd.rs 里等 PWR 的那一行),都开 1 kHz SysTick,都用 semihosting 打印:任务 1 每 100 ms、任务 2 每 150 ms 打一行,各三次,600 ms 时退出。

  • ticker(QEMU 模板原样):两个 #[embassy_executor::task(pool_size = 2)] 的实例加 #[embassy_executor::main],Timer::after_millis 等待,时间驱动是 SysTick 驱动加 embassy-time-queue-utils 的集成队列。
  • poll_loop:cortex_m_rt::entry 加一个主循环,SysTick 中断只把一个 AtomicU32 加一,主循环比较时刻、打印、WFI。不依赖 embassy-executor 和 embassy-time。
loop {
  let now = TICKS.load(Ordering::Relaxed);
  for i in 0..2 {
    if n[i] < 3 && now >= next[i] {
      hprintln!("task {} tick {} at {} ms", i + 1, n[i], now);
      n[i] += 1;
      next[i] += period[i];
    }
  }
  if now >= 600 { /* 打印并退出 */ }
  cortex_m::asm::wfi();
}
❗IMPORTANT
动脑筋:300 ms 那一刻两个任务同时到期,ticker 先打印谁?

两个任务都在 300 ms 到期:任务 1 是在 200 ms 那次登记的,任务 2 是在 150 ms 那次。poll_loop 按下标顺序检查,先打任务 1。ticker 呢?先从定时器队列和运行队列的结构推一推,再看下面的输出。

输出(-icount shift=8,sleep=off):

== ticker
before init
init done, pclk2 4000000 Hz
task 1 tick 0 at 100 ms
task 2 tick 0 at 150 ms
task 1 tick 1 at 200 ms
task 2 tick 1 at 300 ms
task 1 tick 2 at 300 ms
task 2 tick 2 at 450 ms
main done at 600 ms
== poll_loop
before init
init done, pclk2 4000000 Hz
task 1 tick 0 at 100 ms
task 2 tick 0 at 150 ms
task 1 tick 1 at 200 ms
task 1 tick 2 at 300 ms
task 2 tick 1 at 300 ms
task 2 tick 2 at 450 ms
main done at 600 ms

体积(thumbv7em-none-eabihf,release,opt-level = "s",llvm-size -A 里的各段,单位字节):

段tickerpoll_loop差
.text105608180+2380
.rodata36722816+856
.bss384108+276
.data000

两边共有的是 embassy-stm32 的 init(时钟、GPIO、EXTI 初始化)、semihosting 和格式化代码,差出来的是执行器、embassy-time、定时器队列和两个任务的状态机。.bss 的差可以逐个对上(llvm-nm -S --size-sort):

20000000 00000010 b ticker::DRIVER
20000010 00000048 b ticker::__embassy_main::POOL
20000058 000000c0 b ticker::ticker::POOL

ticker::ticker::POOL 是两个 ticker 任务实例,0xc0 = 192 字节,每个 96 字节;main 的任务 72 字节;SysTick 驱动的 tick 计数和队列头 16 字节。192 + 72 + 16 = 280,减去 poll_loop 那个 4 字节的 TICKS,正好是 276。执行器本身不在 .bss 里:它在 main 的栈上。

动脑筋答案

ticker 先打印 task 2 tick 1 at 300 ms,再打印 task 1 tick 2 at 300 ms,和 poll_loop 相反。顺序由两个后进先出的结构决定:

  1. 集成定时器队列把新登记的任务插在链表头(schedule_wake_queue_item 里的 self.head.replace(Some(item_ptr)))。任务 2 在 150 ms 登记,任务 1 在 200 ms 登记,所以链表是「任务 1 → 任务 2」。300 ms 的 SysTick 中断里,next_expiration 从头扫,先唤醒任务 1,再唤醒任务 2。
  2. 唤醒就是压进运行队列。运行队列是一个栈,dequeue_all 一次取走一批,倒着处理:后压进去的任务 2 先被 poll。

两步各反一次,结果和「谁先登记」一致:任务 2 先登记,先打印。每一批里的顺序都是这样定的,不要依赖它;需要先后关系时用 embassy-sync 的原语显式同步。运行队列的细节在状态与运行队列,定时器队列在定时器队列。

小结

  • 裸机并发的差别在「等待时进度记在哪」:主循环记在全局变量里,RTOS 记在每个线程的栈上,RTIC 记在静态资源里,embassy 记在编译器生成的 future 里,future 放在 .bss 的 TaskPool 中。
  • README 和书的说法大多成立:没有堆、一个栈、只 poll 被唤醒的任务、空闲时 WFE;书里「把让出的任务放到运行队列末尾」和代码不符。
  • 和 tokio 能核实的差别:任务存储在编译期定好、池满报 SpawnError::Busy、没有 JoinHandle;没有 I/O 驱动线程,中断处理函数直接调 waker;任务参数必须 'static;一个执行器单线程,要抢占就用 InterruptExecutor。
  • 同样两个周期任务,embassy 版比手写主循环多 2380 字节 .text、856 字节 .rodata、276 字节 .bss(QEMU 11.1.2,PWR 一行补丁,SysTick 驱动,release)。
  • 同时到期的任务按登记顺序被 poll:定时器链表头插、运行队列后进先出,两次反转。

💬 关联讨论

发起讨论
加载中...