Rust in Everything
基于 embassy-executor 0.10.0

一次 Timer::after

主线位置:定时器主线全程。从 Timer::after(100ms) 被创建,到任务在 .await 之后继续执行;每一步只讲到看清全貌的深度,细节链到 02 执行器和 03 时间。

sequenceDiagram
  participant T as 任务
  participant TM as Timer
  participant D as 时间驱动
  participant Q as 定时器队列
  participant E as 执行器
  participant I as SysTick 中断
  T->>TM: Timer after 记到期时刻
  T->>TM: 第一次 poll
  TM->>D: schedule_wake at waker
  D->>Q: 任务头里的节点挂上链表
  TM-->>T: Pending
  E->>E: 运行队列空 执行 WFE
  I->>Q: next_expiration 到期
  Q->>E: waker wake 入运行队列 SEV
  E->>T: poll
  T->>TM: poll
  TM-->>T: Ready

程序

实验在 QEMU 11.1.2 b-l475e-iot01a 上跑。和全卷的 QEMU 实验一样,embassy-stm32 用的是只注释掉 rcc/bd.rs 里等 PWR 那一行的补丁版,时间驱动是 QEMU 模板里手写的 1 kHz SysTick 驱动(QEMU 没有 TIM)。驱动就这么几行:

struct SysTickDriver {
  ticks: Mutex<Cell<u64>>,
  queue: Mutex<RefCell<Queue>>, // embassy_time_queue_utils::Queue
}

impl embassy_time_driver::Driver for SysTickDriver {
  fn now(&self) -> u64 {
    critical_section::with(|cs| self.ticks.borrow(cs).get())
  }
  fn schedule_wake(&self, at: u64, waker: &Waker) {
    critical_section::with(|cs| {
      self.queue.borrow(cs).borrow_mut().schedule_wake(at, waker);
    });
  }
}

embassy_time_driver::time_driver_impl!(static DRIVER: SysTickDriver = SysTickDriver { /* ... */ });

#[exception]
fn SysTick() {
  critical_section::with(|cs| {
    let t = DRIVER.ticks.borrow(cs).get() + 1;
    DRIVER.ticks.borrow(cs).set(t);
    DRIVER.queue.borrow(cs).borrow_mut().next_expiration(t);
  });
}

被测的是 main 任务里的一句 Timer::after_millis(100).await。打点的办法(intro-qemu 的 timer,release):

  • 在任务里 .await 前后、驱动的 schedule_wake 入口、SysTick 中断入口各记一次时刻;
  • 打开 embassy-executor 的 trace feature,定义它要求的七个 _embassy_trace_* 钩子,在 poll_start、task_exec_begin、task_exec_end、task_ready_begin(任务进运行队列)、executor_idle(执行器要去 WFE 了)里也记一次;
  • 时刻 = tick 数 × 1 ms + SysTick 当前计数值换算出的纳秒(4 MHz,一个计数 250 ns),这是 QEMU 的虚拟时间;
  • QEMU 加 -icount shift=8,sleep=off:每条指令固定占 2⁸ = 256 ns 虚拟时间,CPU 空闲时虚拟时钟直接跳到下一个定时器事件。所以两个打点之间只要没经过睡眠,纳秒数除以 256 就是这段执行的指令条数。这是 QEMU 数的指令,不是真芯片上的周期。同一个程序跑两次,输出逐字节相同。

完整输出:

     2101.000 us  +      0 ns (~     0 insn)  stamp A
     2120.250 us  +  19250 ns (~    75 insn)  stamp B
---
     3153.750 us  +      0 ns (~     0 insn)  task: before Timer::after(100ms)
     3190.750 us  +  37000 ns (~   144 insn)  driver schedule_wake
                                            schedule_wake at = 103
                                            queue changed = 1
     3247.250 us  +  56500 ns (~   220 insn)  exec task_exec_end
     3269.250 us  +  22000 ns (~    85 insn)  exec executor_idle (-> wfe)
    53005.500 us  +49736250 ns (~194282 insn)  idle tick: SysTick ISR entry
    53049.000 us  +  43500 ns (~   169 insn)  exec poll_start
    53072.000 us  +  23000 ns (~    89 insn)  exec executor_idle (-> wfe)
    53102.500 us  +  30500 ns (~   119 insn)  exec poll_start
    53125.500 us  +  23000 ns (~    89 insn)  exec executor_idle (-> wfe)
   103005.500 us  +49880000 ns (~194843 insn)  (last SysTick ISR entry)
   103044.750 us  +  39250 ns (~   153 insn)  wake -> task_ready_begin
   103081.750 us  +  37000 ns (~   144 insn)  exec poll_start
   103105.500 us  +  23750 ns (~    92 insn)  exec task_exec_begin
   103142.250 us  +  36750 ns (~   143 insn)  task: after .await
SysTick interrupts during the wait: 100
executor polls (poll_start) during the wait: 200
executor idle -> wfe while waiting: 200

开头两行是连着打的两个点:一次打点本身约 75 条指令,下面每一行的增量都含一次打点。idle tick 那五行是等待期间随便挑的一个 tick(第 50 个),完整记下来看执行器空闲时在做什么。跨过睡眠的两行(+49736250 ns 那种)是虚拟时间,不是指令数。

逐步走

1. 任务第一次 poll:Timer 把时刻交给驱动

#[embassy_executor::main] 把 async fn main 变成第一个任务,Executor::run 先 poll 它一次(怎么展开见宏)。任务跑到 Timer::after_millis(100):Timer::after 只是 Instant::now() + duration,Instant::now 调驱动的 now(),此刻是 tick 3。然后 .await 第一次 poll 这个 Timer:

fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
    if self.yielded_once && self.expires_at <= Instant::now() {
        Poll::Ready(())
    } else {
        embassy_time_driver::schedule_wake(self.expires_at.as_ticks(), cx.waker());
        self.yielded_once = true;
        Poll::Pending
    }
}

(impl Future for Timer。)第一次 poll 一定走 else:哪怕到期时刻已经过了,也要先让出一次(yielded_once)。schedule_wake 调的是外部符号 _embassy_time_schedule_wake,由驱动里的 time_driver_impl! 定义,链接器把两边接上(源码地图)。

实测:从任务里打点到驱动的 schedule_wake 入口 144 条指令,去掉一次打点约 70 条。这 70 条里是 Instant::now()(一次临界区)、构造 Timer、进 poll、调驱动。到期时刻 at = 103:tick 3 加 100。

Instant::now() 只有 tick 精度,此刻真实时间是 3.154 ms,到期 tick 103 对应 103.0 ms,所以这次等待实际只有 99.85 ms 就会被叫醒。Duration::from_millis 换算 tick 时向上取整,但 now() 是向下取整的,等待最多会比要求短不到一个 tick。tick 率怎么选、怎么换算,见 Instant 与 Duration。

2. 进定时器队列:节点就在任务头里

驱动把 (at, waker) 交给 embassy-time-queue-utils 的 Queue::schedule_wake。没开 generic-queue-*,用的是集成队列:先用 TimerQueueItem::from_embassy_waker(waker) 拿到这个任务的任务头里那 6 个字的节点(executor 定义的 __embassy_time_queue_item_from_waker 检查 waker 的 vtable 是不是 embassy 的,再从数据指针还原任务),然后插到链表头:

None => {
    // If not in the queue, add it and update.
    let mut item_ptr = NonNull::from(item);
    let prev = self.head.replace(Some(item_ptr));

    let item = unsafe { item_ptr.as_mut() };

    item.expires_at = at;
    item.waker = Some(waker.clone());
    item.next.set(prev);
    // ...
    true
}

(schedule_wake_queue_item。)没有分配,也没有容量上限:一个任务同一时刻最多在队列里出现一次,它自己带着节点。返回 true 表示「最早到期时刻可能变了,驱动该重设闹钟」,输出里的 queue changed = 1 就是它。同一个任务又登记一个更早的时刻怎么办、更晚的又怎么办,见定时器队列。

3. 设硬件闹钟:SysTick 驱动没有这一步

真正的驱动在 schedule_wake 返回 true 时重设闹钟。embassy-stm32 的 TIM 驱动(schedule_wake):

fn schedule_wake(&self, at: u64, waker: &core::task::Waker) {
    critical_section::with(|cs| {
        let mut queue = self.queue.borrow(cs).borrow_mut();

        if queue.schedule_wake(at, waker) {
            let mut next = queue.next_expiration(self.now());
            while !self.set_alarm(cs, next) {
                next = queue.next_expiration(self.now());
            }
        }
    })
}

set_alarm 把到期时刻的低 16 位写进通道 2 的比较寄存器(ccr(1);通道 1 的比较值固定是 0x8000,用来产生半溢出中断),离得够近就打开这个比较中断;set_alarm 返回 false 说明时刻已经过了,就再处理一次队列。这样两次闹钟之间 CPU 不用醒(计数器溢出和半溢出各有一次中断,用来把 16 位计数拼成 64 位,见时间驱动)。

QEMU 模板的 SysTick 驱动忽略了 schedule_wake 的返回值,因为它根本没有闹钟:SysTick 每 1 ms 中断一次,每次都调 next_expiration 扫一遍队列。这是为了在没有 TIM 的 QEMU 上跑通而做的最简实现,代价在第 5 步能看到。

📝NOTE
冷知识:定时器队列原来在执行器里

#3593「Rework time-driver contract.」2024-12-16 合并,首次发布于 embassy-executor 0.7.0、embassy-time 0.4.0、embassy-time-driver 0.2.0。在这之前,执行器有一个 integrated-timers feature,定时器队列由执行器维护,时间驱动只提供底层闹钟:0.1.0 的 Driver trait 是 now、allocate_alarm、set_alarm_callback、set_alarm 四个方法,HAL 按闹钟个数分配,执行器把最早到期的时刻设进去。#3593 里的提交「Merge time-driver and time-queue-driver traits, make HALs own and handle the queue.」把这套换成了今天的 now 加 schedule_wake:队列归驱动所有,驱动自己决定怎么设闹钟(甚至像 SysTick 驱动这样不设)。embassy-time-driver 0.2.0 的 CHANGELOG 只有两行:「The allocate_alarm, set_alarm_callback, set_alarm functions have been removed.」「schedule_wake has been added to the Driver trait.」执行器那边只留下任务头里的一块内存,后来又在 0.9.0 搬进了 embassy-executor-timer-queue(源码地图)。

4. 让出:任务返回 Pending,执行器 WFE

Timer::poll 返回 Pending,任务的 async fn 状态机跟着返回 Pending,TaskStorage::poll 什么都不做就回来了:任务不会被放回运行队列。运行队列空了,Executor::run 的循环走到 wfe:

pub fn run(&'static mut self, init: impl FnOnce(Spawner)) -> ! {
    init(self.inner.spawner());

    loop {
        unsafe {
            self.inner.poll();
            asm!("wfe");
        };
    }
}

实测:驱动返回后到 task_exec_end 220 条指令(其中有两次往日志里记 at 和返回值),再到 executor_idle 85 条。从任务调 Timer::after 到执行器准备睡下,一共约 450 条指令,含三次打点。

5. 等待期间:SysTick 每次都把执行器叫醒

这 100 ms 里有 100 次 SysTick 中断。第 50 个 tick 的完整记录:

    53005.500 us  +49736250 ns (~194282 insn)  idle tick: SysTick ISR entry
    53049.000 us  +  43500 ns (~   169 insn)  exec poll_start
    53072.000 us  +  23000 ns (~    89 insn)  exec executor_idle (-> wfe)
    53102.500 us  +  30500 ns (~   119 insn)  exec poll_start
    53125.500 us  +  23000 ns (~    89 insn)  exec executor_idle (-> wfe)

中断让 wfe 返回,循环回到 poll():运行队列是空的,poll 什么都没做,又去 wfe。奇怪的是第二个 poll_start:中间没有新的中断,wfe 却又立刻返回了,执行器空跑了第二圈才真正睡下。

❗IMPORTANT
动脑筋:100 ms 的等待里,执行器一共 poll 了几次?

已知:SysTick 1 kHz,从任务让出到被唤醒一共 100 次 SysTick 中断(最后一次就是叫醒任务的那次);上面那个空闲 tick 里 poll_start 出现了两次。把被唤醒后的那次 poll 也算上,执行器一共 poll 了几次?先算出你的答案,再想一想第二次是谁让 wfe 返回的。

为了看清 wfe 什么时候返回,单独写了一个不带 embassy 执行器的小程序(wfe_probe,同样的 QEMU、同样的 -icount),量四种情况下一条 wfe 停了多少个 SysTick 计数(4000 个计数是 1 ms):

plain wfe:              4008 counts
wfe after sev:          6 counts
wfe after PendSV ran:   6 counts
next wfe:               3949 counts

先执行过 sev 的 wfe 立刻返回:SEV 把事件寄存器置位,WFE 看到置位就清掉并继续。刚处理完一个异常(这里是手动挂起的 PendSV,处理函数是空的)之后的 wfe 也立刻返回:在 QEMU 11.1.2 上,处理过一个异常之后事件寄存器也是置位的(是进入、返回还是挂起时置的位,这个实验分不出来)。之后的 wfe 才一直睡到下一个 SysTick。执行器的空跑就是这么来的:睡着时 SysTick 来了,wfe 因中断返回;中断处理完事件寄存器又是置位的,于是下一次 wfe 立刻返回;第三次 wfe 才真正睡下。真芯片上是否也这样,本卷没有板子验证。

所以 WFE 期间 CPU 做的事是:停在 wfe 这条指令上,直到有事件或中断;中断处理完,执行器空 poll 两次,再睡。打点版里一个空闲 tick 从中断入口到第二次 executor_idle 是 469 条指令,其中 4 次钩子打点约占 300 条。用 SysTick 驱动,这笔开销每毫秒都要付;用 TIM 驱动,CPU 只在闹钟和计数器溢出时醒。

6. 到期:中断里唤醒任务

第 103 个 tick 的 SysTick 中断里,next_expiration(103) 扫到这个任务的节点已经到期,把它摘下链表,调存在节点里的 waker(next_expiration)。embassy 的 waker 的 wake 就是 wake_task:

pub fn wake_task(task: TaskRef) {
    let header = task.header();
    header.state.run_enqueue(|l| {
        // We have just marked the task as scheduled, so enqueue it.
        unsafe {
            let executor = header.executor.load(Ordering::Relaxed).as_ref().unwrap_unchecked();
            executor.enqueue(task, l);
        }
    });
}

run_enqueue 在 Cortex-M 上是一次原子 swap:任务已经在运行队列里就什么都不做,同一个任务被唤醒两次只入队一次。SyncExecutor::enqueue 把任务压进运行队列,压之前队列是空的才调 pender;线程模式执行器的 __pender 就是一条 sev。这些都发生在中断处理函数里。状态位、运行队列和 waker 的细节见状态与运行队列。

实测:SysTick 中断入口到 task_ready_begin(enqueue 的开头)153 条指令,含中断里的 tick 计数、扫队列、一次打点。

7. 再次 poll:Ready

中断返回,wfe 返回,poll() 取走运行队列(只有这一个任务),调它的 poll 函数。Timer::poll 这次 yielded_once 是 true,expires_at(103)<= Instant::now()(103),返回 Ready,任务继续执行 .await 后面的代码。

实测:task_ready_begin 到 poll_start 144 条,到 task_exec_begin 92 条,到任务里 .await 之后的打点 143 条。

每一步多少条指令

把钩子里的打点全部去掉再跑一次(trace 仍开着,七个钩子是空函数;只留任务里两个点、驱动入口一个点,以及 SysTick 中断入口记下的时刻):

     3114.500 us  +      0 ns (~     0 insn)  task: before Timer::after(100ms)
     3150.000 us  +  35500 ns (~   138 insn)  driver schedule_wake
                                            schedule_wake at = 103
                                            queue changed = 1
   103067.000 us  +99917000 ns (~390300 insn)  task: after .await
                                          last SysTick ISR entry (ns) = 103005250

汇总(QEMU 11.1.2 b-l475e-iot01a,PWR 一行补丁,SysTick 驱动,release opt-level = "s",-icount shift=8,sleep=off;一次打点约 70 条,已扣除):

区间指令数
任务调 Timer::after → 驱动 schedule_wake 入口约 70
SysTick 中断入口 → 任务在 .await 之后继续(103005.25 → 103067.0 µs)约 170(含中断里记时刻的开销)
一个空闲 tick:中断入口 → 两次空 poll → 再次睡下469(打点版,含 4 次打点约 300 条)

第二行就是「一次定时器唤醒」在执行器和队列上的全部开销:中断里的 tick 计数和扫队列、wake_task、异常返回、wfe 返回、poll、Timer::poll 返回 Ready。tokio 的同一段路要经过时间轮、驱动锁和 worker 的 unpark(Sleep 的一生),这里没有锁,只有关中断的临界区和一次原子交换。

动脑筋答案

200 次。输出最后两行:

executor polls (poll_start) during the wait: 200
executor idle -> wfe while waiting: 200

但不是简单的「100 次中断 × 2」。再跑一次,把执行器第一次睡下之后的四个事件也记下来(probe2,其余相同):

     3266.750 us  +  21250 ns (~    83 insn)  exec executor_idle (-> wfe)
     3298.250 us  +  31500 ns (~   123 insn)  exec poll_start
     3323.500 us  +  25250 ns (~    98 insn)  exec executor_idle (-> wfe)
     4044.000 us  + 720500 ns (~  2814 insn)  exec poll_start
     4069.000 us  +  25000 ns (~    97 insn)  exec executor_idle (-> wfe)

任务让出后,执行器第一次 wfe 就没睡:main 任务这一轮本身是被 tick 3 的 SysTick 中断叫醒的(热身用的 Timer::after_millis(2)),那次中断里的 sev(以及处理中断本身)留下的事件还在,wfe 直接返回,执行器空 poll 一次(3298 µs,还在 tick 3 里),第二次 wfe 才睡到 tick 4。于是:

  • poll 次数 200 = 让出后那 1 次空跑 + tick 4 到 102 这 99 个空闲 tick × 2 + tick 103 被唤醒后真正 poll 任务的 1 次;
  • executor_idle 次数 200 = 让出后的 2 次 + 99 × 2。

每个空闲 tick 的两次都来自同一个机制:中断让 wfe 返回,而处理完中断事件寄存器又是置位的,下一条 wfe 不睡。wfe_probe 的第三行(wfe after PendSV ran: 6 counts)单独证实了后一点。200 次里只有最后一次有任务可跑。如果驱动不是每毫秒都中断,这个数会完全不同,这也是 SysTick 驱动不适合低功耗的原因。

小结

  • Timer::after 只算到期时刻;第一次 poll 一定 Pending,经链接期符号把 (at, waker) 交给时间驱动。
  • 集成定时器队列的节点在任务头里,从 waker 还原出来,插到链表头;没有分配,没有容量上限。
  • stm32 的 TIM 驱动在队列变化时设比较寄存器,两次闹钟之间 CPU 不醒;本卷的 SysTick 驱动每毫秒扫一次队列,没有闹钟。
  • 任务返回 Pending 后不在任何队列里,直到 waker 被调用;执行器 poll 完就 wfe。
  • 到期时中断里的 wake_task 做一次原子交换、压进运行队列、sev;中断返回后执行器只 poll 这一个任务。
  • QEMU 上实测:Timer::after 到驱动约 70 条指令,中断入口到任务继续约 170 条;100 ms 等待里执行器 poll 了 200 次,只有最后一次有任务可跑,因为处理完一次中断后事件寄存器是置位的,每个空闲 tick 的第二条 wfe 不睡。

💬 关联讨论

发起讨论
加载中...