一次 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 的
tracefeature,定义它要求的七个_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 步能看到。
#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 却又立刻返回了,执行器空跑了第二圈才真正睡下。
已知: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不睡。