写一个迷你执行器

主线位置：把定时器主线和跨任务主线自己走一遍。spawn 把 future 放进静态的一格（对应 TaskStorage）→ 任务第一次 poll，Timer 把到期时刻和 waker 登记进定时器表（对应 schedule_wake 和集成定时器队列）→ 执行器无事可做，WFE → SysTick 中断里时钟加一、表头到期就 wake → waker 在运行队列里置一位、SEV（对应 wake_task）→ 执行器醒来，poll 这一位对应的任务。跨任务主线那一段是一个容量为 1 的 channel：满了登记发送方的 waker，接收方取走一个值时唤醒它。

flowchart TB
  sp["spawn"] --> slot["定长任务槽"]
  sp --> mask["u32 位图"]
  isr["SysTick 中断"] --> tick["Clock tick"]
  tick --> head["定时器表表头"]
  head -->|到期| wk["waker wake"]
  ch["Channel 收发"] --> wk
  wk -->|fetch_or| mask
  mask -->|由空变非空| sev["SEV"]
  run["run_until"] --> take["swap 取走位图"]
  take --> poll["按槽号 poll"]
  poll --> wfe["WFE"]
  wfe --> run

mini 是一个 no_std 库 hands-mini-executor，只依赖 critical-section，Cortex-M 上再加 cortex-m（SEV、WFE、SysTick 寄存器）。平台二选一，由 feature 决定：

[features]
std = ["critical-section/std"]
cortex-m = ["dep:cortex-m"]

[dependencies]
critical-section = "1"
cortex-m = { version = "0.7.7", optional = true }

embassy 只出现在另一个 crate hands-qemu 的两个对照程序里。平台 M2 Max、macOS 27、rustc 1.99.0；std 上的测试是 debug 构建，std 示例和 QEMU 程序是 release（opt-level = "s"）。QEMU 是 QEMU 11.1.2 b-l475e-iot01a，用 -icount shift=8,sleep=off 让虚拟时间按指令数走、结果可复现；这几个程序不用 embassy-stm32，所以没有 PWR 那一行补丁，启动后直接用复位时的 4 MHz MSI 时钟开 SysTick。库里没有 unwrap / expect，测试里随便用。

范围

库（不含 demo.rs）474 行代码，加上空行和注释 622 行。执行器、时钟、channel 三个文件和平台无关；平台相关的只有 platform.rs 一个文件，两个平台各一份 Signal，再加 SysTick 的启动和 std 的 tick 线程。

任务槽：一块定长字节放任意 future

static 里不能写 TaskPool<impl Future, N>：稳定版 Rust 没法给 async fn 返回的匿名类型起名字。embassy 的办法是让宏在编译期算出每个任务函数 future 的大小，为它生成一个正好那么大的池（宏、任务存储）。mini 不写宏，换一个更粗的办法：执行器里放 TASKS 个一样大的格子，每格 SLOT 个字节，spawn 时把任意 future 拷进去，只记一个类型擦除后的函数指针。

type PollFn = unsafe fn(*mut u8, &mut Context<'_>) -> Poll<()>;

#[repr(C, align(8))]
struct Bytes<const N: usize>([MaybeUninit<u8>; N]);

struct Slot<const N: usize> {
  header: Header,
  state: AtomicU8,
  poll_fn: UnsafeCell<Option<PollFn>>,
  future: UnsafeCell<Bytes<N>>,
}

/// An executor with `TASKS` slots of `SLOT` bytes each. Meant to live in a `static`.
pub struct Executor<const TASKS: usize, const SLOT: usize> {
  slots: [Slot<SLOT>; TASKS],
  queue: ReadyQueue,
  polling: AtomicBool,
  loops: AtomicU32,
}

poll_fn 是泛型函数 poll_erased::<F> 的一个实例，它知道格子里放的是哪个 F：

unsafe fn poll_erased<F: Future<Output = ()>>(future: *mut u8, cx: &mut Context<'_>) -> Poll<()> {
  let future = future.cast::<F>();
  // SAFETY: the slot sits inside a static executor and is never moved, so the future is pinned.
  let poll = unsafe { Pin::new_unchecked(&mut *future) }.poll(cx);
  if poll.is_ready() {
    // SAFETY: completed futures are never polled again; the slot is marked FREE right after.
    unsafe { ptr::drop_in_place(future) };
  }
  poll
}

这和 embassy 的 TaskStorage::poll 是同一个思路：执行器只认识函数指针和一块内存，Ready 时就地析构 future。格子的状态是 FREE → CLAIMED → SPAWNED → FREE，spawn 用一次 compare_exchange(FREE, CLAIMED) 抢格子，写完 future 和 poll_fn 才 store(SPAWNED, Release)：

  pub fn spawn<F>(&'static self, future: F) -> Result<(), SpawnError>
  where
    F: Future<Output = ()> + Send + 'static,
  {
    const {
      assert!(
        size_of::<F>() <= SLOT && align_of::<F>() <= 8,
        "future does not fit in a task slot"
      )
    };
    for (index, slot) in self.slots.iter().enumerate() {
      if slot
        .state
        .compare_exchange(FREE, CLAIMED, Ordering::Acquire, Ordering::Relaxed)
        .is_err()
      {
        continue;
      }
      // SAFETY: winning the exchange gives this call the slot until SPAWNED is stored;
      // the size and alignment were checked above.
      unsafe {
        slot.future.get().cast::<F>().write(future);
        *slot.poll_fn.get() = Some(poll_erased::<F>);
      }
      let bit = 1 << index;
      // ... header 记下运行队列的地址和自己的位
      slot.state.store(SPAWNED, Ordering::Release);
      self.queue.push(bit);
      return Ok(());
    }
    Err(SpawnError::Full)
  }

三点：

• future 放不下是编译错误。const { assert!(...) } 在 spawn::<F> 被实例化时求值，SLOT 太小就编不过。所有格子都占满了才是运行期的 Err(SpawnError::Full)，对应 embassy 的 SpawnError::Busy。
• 要求 Send。Executor 是 Sync 的，任何线程（或中断）都能 spawn，future 却在执行器那个线程上被 poll。embassy 把这两种情况分成 Spawner 和 SendSpawner（Spawner），mini 只有一种，一律要 Send。
• future 按值传进来。它先在调用者的栈上建好，再拷进格子。embassy 传的是一个返回 future 的闭包，在格子里原地构造（任务存储），省掉这次拷贝和这块栈。

SLOT 要按最大的那个 future 定。QEMU 程序里两个任务的 future 一个 136 字节、一个 144 字节（临时加一行 size_of_val 打印出来的），SLOT 就是 144，生产者那一格空着 8 字节。这 136 字节里，demo::producer 本身只有 56，其余来自外面那层数 poll 次数的 counted（见下文「两个周期任务」）。同样在 thumbv7em 上临时打印几种包法的大小：

producer 56 counted 136 plain-await 120 struct 64

plain-await 是函数体只有 future.await 一行的 async fn plain(future: F)，struct 是手写的 Counted { polls, future } 加 impl Future。把一个 future 当参数传进 async fn，56 字节就变成 120 = 2 × 56 + 8：状态机里参数占一份，被 await 的那个（counted 里是 pin! 之后的局部变量）又占一份。手写的结构体只多 8 字节。所以 SLOT 的大头是测量用的包装，不是任务本身。

把 SLOT 改成 136，cargo build 报（release，thumbv7em）：

error[E0080]: evaluation panicked: future does not fit in a task slot
   --> ~/.rustup/toolchains/stable-aarch64-apple-darwin/lib/rustlib/src/rust/library/core/src/panic.rs:62:9
    |
 62 |         $crate::panicking::panic_fmt($crate::const_format_args!($($t)+));
    |         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ evaluation of `hands_mini_executor::Executor::<2, 136>::spawn::<{async block@src/bin/mini.rs:35:22: 35:27}>::{constant#0}` failed here
...
error: could not compile `hands-qemu` (bin "mini") due to 1 previous error

错误里的类型就是那个装不下的 future：main 里的 async 块（消费者加上结束时打印统计）。

运行队列：一个 u32

embassy 的运行队列是侵入式链表，节点在任务头里，任务多少都行（状态与运行队列）。mini 的任务数在编译期就定了（TASKS ≤ 32，new 里同样用 const 断言），于是运行队列可以就是一个 AtomicU32：第 i 位为 1 表示第 i 格想被 poll。

/// The run queue, shared by the executor and every waker it hands out.
struct ReadyQueue {
  mask: AtomicU32,
  signal: Signal,
}

impl ReadyQueue {
  fn push(&self, bit: u32) {
    // Only the push that makes the queue non-empty has to wake the executor:
    // a non-empty queue means it was already signalled and has not swapped the mask yet.
    if self.mask.fetch_or(bit, Ordering::AcqRel) == 0 {
      self.signal.pend();
    }
  }
}

一条 fetch_or 同时做了 embassy 里两样东西的事：位已经是 1，再唤醒什么也不改，这就是 STATE_RUN_QUEUED 的去重；由 0 变非 0 时才叫醒执行器，这就是 SyncExecutor::enqueue 里「队列原来为空才 pend」。出队也是一条指令，swap(0) 取走整批：

  pub fn poll_ready(&'static self) -> Result<u32, Reentered> {
    if self.polling.swap(true, Ordering::Acquire) {
      return Err(Reentered);
    }
    self.loops.fetch_add(1, Ordering::Relaxed);
    let mut ready = self.queue.mask.swap(0, Ordering::Acquire);
    let taken = ready.count_ones();
    while ready != 0 {
      self.poll_slot(ready.trailing_zeros() as usize);
      ready &= ready - 1;
    }
    self.polling.store(false, Ordering::Release);
    Ok(taken)
  }

• 批内按槽号从小到大。同一批里谁先跑由位的位置决定，和唤醒顺序无关。测试 batches_run_in_slot_order 让 A、B、C、D 四个任务各 yield 三次，顺序是 A0 B0 C0 D0 A1 B1 C1 D1 A2 B2 C2 D2；embassy 的同一个实验是 D0 C0 B0 A0 A1 B1 C1 D1 D2 C2 B2 A2，批内后进先出、每批翻一次（状态与运行队列）。两者都满足「一批跑完才开始下一批」：poll 期间被唤醒的任务进的是已经清零的新位图。
• 重入要拦住。格子用 UnsafeCell 存 future，两个调用同时 poll 同一格就是两个 &mut。embassy 把 raw::Executor::poll 标成 unsafe fn，安全条件写的是「You must NOT call poll reentrantly on the same executor.」，由调用它的平台代码保证；mini 的 poll_ready 是安全函数，用一个 AtomicBool 把第二个调用挡回去，测试 polling_from_inside_a_task_is_rejected 在任务里调 poll_ready，拿到 Err(Reentered)。
• 迟到的唤醒。任务结束后格子回到 FREE，旧 waker 再来置位，poll_slot 看到状态不是 SPAWNED 就跳过；格子被新任务占了，旧 waker 唤醒的就是新任务，多一次空 poll（测试 stale_waker_only_causes_a_spurious_poll）。embassy 用把 poll_fn 换成 poll_exited 处理同一件事。

three_wakes_one_poll 把 waker 存下来连调 wake_by_ref、wake_by_ref、wake，poll_ready 返回 1、任务一共被 poll 2 次，和 embassy 的「唤醒三次只 poll 一次」一样。

waker：指针加一张静态 vtable

waker 的数据指针指向格子的 Header，里面记着运行队列在哪、自己是哪一位：

/// What a waker points at: which queue to push to, and which bit.
struct Header {
  queue: AtomicPtr<ReadyQueue>,
  bit: AtomicU32,
}

static VTABLE: RawWakerVTable = RawWakerVTable::new(clone, wake, wake, drop);

unsafe fn clone(data: *const ()) -> RawWaker {
  RawWaker::new(data, &VTABLE)
}

unsafe fn wake(data: *const ()) {
  // SAFETY: `waker_for` only builds wakers from headers inside a `'static` executor.
  let header = unsafe { &*data.cast::<Header>() };
  let queue = header.queue.load(Ordering::Relaxed);
  // SAFETY: set at spawn to the owning executor's queue, which is `'static` too.
  if let Some(queue) = unsafe { queue.as_ref() } {
    queue.push(header.bit.load(Ordering::Relaxed));
  }
}

unsafe fn drop(_: *const ()) {
  // Slots are never freed, so there is no reference count to release.
}

和 embassy 的 waker.rs 几乎一样（状态与运行队列）：格子是 'static 的、永不释放，clone 只复制指针，drop 什么也不做，wake 和 wake_by_ref 是同一个函数。tokio 的 waker 要维护任务的引用计数（poll 与唤醒），这里不需要。差别在 wake 找队列的方式：embassy 的任务头里存的是执行器指针，mini 存的是运行队列指针加一个位。

睡与醒：WFE / SEV 和条件变量

执行器的主循环和 Cortex-M 线程模式执行器的 Executor::run 一样，poll 一批就睡，不判断有没有活：

  pub fn run_until(&'static self, mut done: impl FnMut() -> bool) -> Result<(), Reentered> {
    loop {
      self.poll_ready()?;
      if done() {
        return Ok(());
      }
      self.queue.signal.sleep();
    }
  }

不丢唤醒靠的是 Signal 自带的记忆。Cortex-M 上 pend 是 SEV，sleep 是 WFE，事件寄存器就是那个标志（为什么 poll 和 WFE 之间的唤醒不会丢，见各平台的执行器）；std 上是一个 Mutex<bool> 加 Condvar，和 embassy 的 platform-std 一样：

  pub(crate) struct Signal {
    flag: Mutex<bool>,
    cond: Condvar,
  }

  impl Signal {
    pub(crate) fn pend(&self) {
      *self.flag.lock().unwrap_or_else(PoisonError::into_inner) = true;
      self.cond.notify_one();
    }

    pub(crate) fn sleep(&self) {
      let mut flag = self.flag.lock().unwrap_or_else(PoisonError::into_inner);
      while !*flag {
        flag = self.cond.wait(flag).unwrap_or_else(PoisonError::into_inner);
      }
      *flag = false;
    }
  }

Signal 放在 ReadyQueue 里，每个执行器一个，所以 std 上可以有多个执行器，测试也能并行跑。embassy 的 std 平台用 Box::leak 一个 Signaler、把地址当 __pender 的 context，是同一件事。

时钟和按到期时刻排序的定时器表

Clock<N> 是一个毫秒计数加一张最多 N 项的表，每项是 (到期时刻, Waker)，前 len 项按到期时刻从小到大排好，相等的按登记先后：

  /// Advances the clock by one millisecond and wakes every timer that is now due.
  pub fn tick(&self) {
    critical_section::with(|cs| {
      let mut table = self.table.borrow_ref_mut(cs);
      table.now += 1;
      while table
        .entries
        .first()
        .is_some_and(|e| e.as_ref().is_some_and(|e| e.at <= table.now))
      {
        if let Some(entry) = table.remove(0) {
          entry.waker.wake();
        }
      }
    });
  }

表是排好序的，每个 tick 只看表头：没有到期的就是一次比较。embassy-time-queue-utils 的集成队列不排序，next_expiration 每次扫整条链表（定时器队列）；代价换到了登记一侧，mini 插入时要找位置、把后面的项挪一格。

Timer 的 poll 把「到没到」和「登记」放进同一个临界区：

  fn poll_at(&self, at: u64, waker: &Waker) -> Poll<()> {
    // Check and register under one critical section: a tick in between would be lost.
    critical_section::with(|cs| {
      let mut table = self.table.borrow_ref_mut(cs);
      if table.now >= at {
        return Poll::Ready(());
      }
      if !table.insert(at, waker) {
        // Table full: degrade to being polled on every executor pass until due.
        waker.wake_by_ref();
      }
      Poll::Pending
    })
  }

分开做的话，检查完「没到」、登记之前来一个 tick 把时刻推过了 at，这个 tick 已经看过表、不会再看，任务就永远睡下去。insert 先找有没有 will_wake 同一个任务的项：有就取较早的到期时刻、只留一项，所以 N 等于任务数时表不会满。满了（比如测试 full_timer_table_degrades_to_busy_polling 里 1 项的表、2 个任务）就退化成每轮都 poll：那个任务 10 个 tick 里被 poll 了 11 次。embassy 的通用队列满了也是提前唤醒，定时器队列一页测出过它的活锁。

挪项最初用的是 rotate_left / rotate_right。QEMU 程序的 llvm-nm --size-sort 里于是多出一个 1410 字节（0x582）的 compiler_builtins::mem::memmove，.text 是 8764 字节；换成逐个 swap 的循环后 memmove 没了，.text 降到 7308，少了 1456 字节。opt-level = "s" 下 rotate 不内联，走通用的 ptr::copy，于是链进来整个 memmove。

mini 的 Timer 和 embassy 的还有一处不同：embassy 的 Timer 带一个 yielded_once，第一次 poll 一定返回 Pending，哪怕时刻已过（Timer、Ticker 与超时）；mini 到了就直接 Ready。

Channel

Channel<T, N> 是 critical_section::Mutex<RefCell<...>> 里的一个环形缓冲，加上发送方、接收方各一个 Option<Waker>，和 embassy-sync 的 ChannelState 每侧一个 WakerRegistration 是同一个结构。登记 waker 的规则也照抄 WakerRegistration::register：

/// Remembers one waker; a different waker replaces it and the old one is woken so it can re-register.
fn register(slot: &mut Option<Waker>, waker: &Waker) {
  if slot.as_ref().is_some_and(|old| old.will_wake(waker)) {
    return;
  }
  if let Some(old) = slot.replace(waker.clone()) {
    old.wake();
  }
}

照抄了规则，也就照抄了它的行为：几个发送方同时等时，它们会互相挤掉对方、互相唤醒（Channel一页在 embassy-sync 上实测过这种忙转，mini 没有单独测）。演示里每侧只有一个任务，碰不到这种情况。发送的 future 把值存在 Option<T> 里，有空位时 take 出来写进缓冲、唤醒接收方；值是被移出来的，不在原地被钉住，所以给它写了 impl Unpin，T 是什么都行。

两个周期任务

demo.rs 里的两个任务：生产者每 100 ms 往容量为 1 的 channel 里送一个数，消费者每 150 ms 取一个。生产者快，channel 很快就满，生产者等在 send 上，之后的节奏被消费者拖住。

pub async fn producer<const T: usize, const C: usize>(clock: &Clock<T>, channel: &Channel<u32, C>, log: Log) {
  for n in 0..ROUNDS {
    clock.after(100).await;
    channel.send(n).await;
    log(format_args!("send {n} at {} ms", clock.now()));
  }
}

pub async fn consumer<const T: usize, const C: usize>(clock: &Clock<T>, channel: &Channel<u32, C>, log: Log) {
  for _ in 0..ROUNDS {
    clock.after(150).await;
    let value = channel.recv().await;
    log(format_args!("recv {value} at {} ms", clock.now()));
  }
}

/// Runs `future`, adding one to `polls` every time it is polled.
pub async fn counted<F: Future>(polls: &AtomicU32, future: F) -> F::Output {
  let mut future = pin!(future);
  poll_fn(|cx| {
    polls.fetch_add(1, Ordering::Relaxed);
    future.as_mut().poll(cx)
  })
  .await
}

counted 不碰执行器，只在任务外面套一层数 poll 次数，同一个包装用在 mini 和 embassy 上，数出来的数可以直接比。QEMU 上的主程序：

static EXECUTOR: Executor<2, 144> = Executor::new();
static CLOCK: Clock<2> = Clock::new();
static CHANNEL: Channel<u32, 1> = Channel::new();

#[exception]
fn SysTick() {
  CLOCK.tick();
}

#[entry]
fn main() -> ! {
  let Some(mut cp) = cortex_m::Peripherals::take() else { panic!("core peripherals taken") };
  start_systick(&mut cp.SYST, 4_000_000);
  let spawned = EXECUTOR
    .spawn(demo::counted(&POLLS[0], demo::producer(&CLOCK, &CHANNEL, log)))
    .and_then(|()| {
      EXECUTOR.spawn(async {
        demo::counted(&POLLS[1], demo::consumer(&CLOCK, &CHANNEL, log)).await;
        // ... 打印 poll 次数、执行器转了几圈，semihosting 退出
      })
    });
  // ... 出错就打印；否则 EXECUTOR.run_until(|| false)
}

QEMU 11.1.2 b-l475e-iot01a，release：

send 0 at 100 ms
recv 0 at 150 ms
send 1 at 200 ms
recv 1 at 300 ms
send 2 at 300 ms
recv 2 at 450 ms
send 3 at 450 ms
recv 3 at 600 ms
send 4 at 600 ms
recv 4 at 750 ms
send 5 at 750 ms
recv 5 at 900 ms
polls: producer 11 consumer 7; executor loops 1801; clock 900 ms

• 300 ms 起，send n 和 recv n-1 总在同一毫秒：生产者 400、550、700 ms 醒来时 channel 是满的，一直等到消费者取走一个值把它唤醒。这是 channel 的背压（backpressure）。
• 消费者 7 次 poll：spawn 后一次，6 次定时器到期各一次，取的时候 channel 里总有值。生产者 11 次：spawn 后一次，6 次定时器到期，加上 4 次被 channel 唤醒（300、450、600、750 ms）。
• 执行器转了 1801 圈，900 ms 里每毫秒两圈：每个 SysTick 中断都让 WFE 返回，哪怕表头没到期；QEMU 处理完异常后事件寄存器是置位的，下一次 WFE 立刻返回，再空转一圈才睡下（一次 Timer::after）。

同一个程序在 std 上（examples/demo.rs，release，连跑 3 次，日志和下面这一行三次都相同，负载 6.8）：

polls: producer 11 consumer 7; executor loops 17; clock 900 ms, wall 900 ms

日志的 12 行和 QEMU 一样。执行器只转了 17 圈：std 的 tick 线程每毫秒调 CLOCK.tick()，表头没到期就什么也不做，不会去碰执行器的条件变量；只有真的 wake 才 pend。Cortex-M 上叫醒 WFE 的不只是 SEV，任何中断都算，用每毫秒一次的 SysTick 当时钟，这笔开销就省不掉；embassy-stm32 的 TIM 驱动用比较通道设闹钟，CPU 只在闹钟和计数器半圈中断时醒（时间驱动）。

tests/mini.rs 的 12 个测试（debug），其中 demo_channel_backpressure 由测试线程自己一毫秒一毫秒地 tick、poll，逐行比对上面的日志，poll 次数断言为 [11, 7]；demo_on_condvar_and_ticker_thread 用真的 tick 线程和条件变量跑一遍。连跑 20 次都通过：

running 12 tests
test three_wakes_one_poll ... ok
test spawn_runs_to_completion_and_reuses_the_slot ... ok
test full_timer_table_degrades_to_busy_polling ... ok
test batches_run_in_slot_order ... ok
test full_pool_is_an_error ... ok
test polling_from_inside_a_task_is_rejected ... ok
test timer_fires_on_the_deadline_tick ... ok
test timers_fire_in_deadline_order ... ok
test stale_waker_only_causes_a_spurious_poll ... ok
test demo_channel_backpressure ... ok
test consumer_in_slot_zero_saves_the_producer_a_poll ... ok
test demo_on_condvar_and_ticker_thread ... ok

test result: ok. 12 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.90s

用真线程的那个测试只比对日志的集合，不比顺序：300 ms 那个 tick 同时唤醒两个任务，tick 线程置第一位和置第二位之间，执行器线程可能已经取走了位图，谁先被 poll 就不一定了。单线程的 Cortex-M 上中断处理函数跑完才回到主循环，没有这个问题。

和 embassy 对照：同样的两个任务

hands-qemu 里另有两个程序，SysTick、semihosting、counted 都一样：

• embassy：embassy-executor 0.10.0（platform-cortex-m、executor-thread）、embassy-time 0.5.1（tick-hz-1_000，时间驱动是模板里的 SysTick 驱动加集成队列）、embassy-sync 0.8.0 的 Channel<CriticalSectionRawMutex, u32, 1>。两个 #[embassy_executor::task]，执行器用 cortex_m_rt::entry 加 StaticCell<Executor> 自己建，不用 #[main]，免得多出一个 main 任务。
• mixed：embassy-executor 跑 mini 的 demo::producer / consumer、Clock、Channel，只有执行器换了。mini 的定时器表存的是 Waker，不需要从 waker 反查任务，所以放在别的执行器上也能用；反过来，embassy-time 的集成队列离开 embassy 执行器会链接失败或 panic（定时器队列）。

三个程序打印的 12 行日志完全一样。poll 次数：

体积（thumbv7em-none-eabihf，release，opt-level = "s"，llvm-size -A，单位字节；三个程序的 .vector_table 都是 1024，.data 都是 0）：

.bss 逐项（llvm-nm -S --size-sort -C，第二列是十六进制字节数）：

== mini
20000038 00000008 b mini::POLLS
200001b8 00000008 b cortex_m_semihosting::export::HSTDOUT
20000040 00000024 b mini::CHANNEL
20000000 00000038 b mini::CLOCK
20000068 00000150 b mini::EXECUTOR
== embassy
20000000 00000008 b embassy::POLLS
20000160 00000008 b cortex_m_semihosting::export::HSTDOUT
20000040 0000000c b embassy::EXECUTOR
20000008 00000010 b embassy::DRIVER
20000018 00000024 b embassy::CHANNEL
20000050 00000088 b embassy::consumer::POOL
200000d8 00000088 b embassy::producer::POOL
== mixed
20000038 00000008 b mixed::POLLS
200001e8 00000008 b cortex_m_semihosting::export::HSTDOUT
20000068 0000000c b mixed::EXECUTOR
20000040 00000024 b mixed::CHANNEL
20000000 00000038 b mixed::CLOCK
20000078 000000b8 b mixed::consumer::POOL
20000130 000000b8 b mixed::producer::POOL

三个程序另有 1 字节的 cortex_m::peripheral::TAKEN（B，上面只列了 b），其余是对齐的空隙。

• 执行器差不多大。mini 和 mixed 的任务、时钟、channel 完全相同，只差执行器：换成 embassy-executor，.text 多 76 字节，.rodata 多 344 字节。.rodata 里多出来的有 embassy 的 panic 信息，例如 SpawnToken instances may not be dropped. You must pass them to Spawner::spawn()、StaticCell 的「already full」和源文件路径。
• 每个任务的额外开销。mini 一格 160 字节 = 16 字节的头（队列指针、位、状态、poll_fn）+ 144 字节的 future 区，两格 320，加上位图、重入标志和圈数计数，EXECUTOR 共 336（0x150）。mixed 里同样的 future 在 embassy 的池里一格 184 = 40 字节任务头 + 144 字节 future（任务函数在 mini 的 future 外面又包了一层 async fn，136 变 144），其中 24 字节是 6 个字的 timer_queue_item。mixed 并不用 embassy-time，可同一个包里的 embassy 程序依赖 embassy-time-queue-utils，它打开的 timer-item-size-6-words 经 Cargo 的 feature 合并作用到包里所有程序（cargo tree -e features -i embassy-executor-timer-queue 可见）。没有这个依赖时 embassy 的任务头是 16 字节（任务存储），和 mini 一样。
• 定时器的内存在哪。embassy 的定时器节点在每个任务头里，驱动只存 tick 计数和链表头（DRIVER 16 字节）；mini 的 CLOCK 56 字节里是计数、长度、借用标志和两项各 16 字节的 (u64, Waker)。
• embassy 比 mixed 小。换上 embassy-time 和 embassy-sync，.text 再少 176 字节，.bss 少 136 字节。.bss 的差在 future：embassy 版两个 future 各 96 字节（池 136 = 40 + 96），mixed 是 144。两边的 Timer 都是 16 字节，发送 future 都是 12 字节、接收 future 都是 4 字节（同样临时打印 size_of），差的是任务本身：mini 的 demo::producer 把时钟、channel、日志函数三个参数存在 future 里，任务又经 counted 套了一层，内层差多少，外层就差两倍。.text 的差在定时器：mini 一侧的 Timer::poll（408 字节，插入内联在里面）、Table::remove（140）、SysTick（136，tick 内联在里面）、Clock::now（46）共 730 字节，embassy-time 一侧的 Timer::poll / Timer::after、_embassy_time_now / _embassy_time_schedule_wake、Queue::schedule_wake / next_expiration 和 SysTick 共 584 字节：排序插入比头插长。这个规模下 mini 没有比 embassy 省出什么：执行器本身几十字节的差别，被 future 的写法、定时器的数据结构和 Cargo feature 盖过去了。

执行器转的圈数：embassy 程序开 trace feature、在 _embassy_trace_poll_start 钩子里计数（钩子会改变体积，所以只在这一次构建里开）：

polls: producer 10 consumer 7; executor loops 1801; clock 900 ms

1801 圈，和 mini 完全相同：两个执行器都是「poll 一批、WFE」，醒几次只由 SysTick 和 QEMU 的事件寄存器决定。

embassy 有而这里没有的

做到这里能看清 embassy 执行器的骨架有多小：一格任务、一个类型擦除的 poll、一个运行队列、一张 waker vtable、一个 poll; wfe 的循环，mini 都写了，体积也相当。多出来的那些——编译期按任务定大小的池、原地构造、不限任务数的侵入式队列、可替换的时间驱动、中断执行器、调度策略——都是为了「任务多少都行、外设和芯片换了也不用改执行器」。和 tenferro、candle 的动手页一样，追不上的是这部分工程。

动脑筋答案

10 次，比 mini 少一次。把 embassy 程序临时改一下，在两个任务的定时器返回后各打印一行当时的时刻和 CHANNEL.len()（QEMU 11.1.2 b-l475e-iot01a，release）：

  producer up at 100 ms, channel len 0
send 0 at 100 ms
  consumer up at 150 ms, channel len 1
recv 0 at 150 ms
  producer up at 200 ms, channel len 0
send 1 at 200 ms
  consumer up at 300 ms, channel len 1
recv 1 at 300 ms
  producer up at 300 ms, channel len 0
send 2 at 300 ms
  producer up at 400 ms, channel len 1
  consumer up at 450 ms, channel len 1
recv 2 at 450 ms
send 3 at 450 ms
...
polls: producer 10 consumer 7; executor loops 0; clock 900 ms

（最后一行的 executor loops 0 是因为这次构建没开 trace。）

• 300 ms 那个 tick，两个任务的定时器同时到期。embassy 先 poll 消费者：它取走 1，channel 空了，生产者随后被 poll 时直接把 2 送进去，不用等。300 ms 这次不是「被 channel 唤醒」，只剩 450、600、750 ms 三次，11 − 1 = 10。
• 顺序由两个后进先出的结构决定：集成定时器队列头插，生产者 200 ms 登记、消费者 150 ms 登记，链表是「生产者 → 消费者」，到期时先唤醒生产者；运行队列又是栈，后入栈的消费者先被 poll。两次反转，结果是先登记的先跑。
• mini 在这一刻先 poll 生产者：两位同时置上，位图按槽号出队，生产者在第 0 格。它看到 channel 里还有 1、等在 send 上，等消费者取走 1 再被唤醒，多了这一次。mixed 也是 11：mini 的定时器表按登记先后唤醒（消费者先），embassy 的运行队列后进先出，又是生产者先跑。
• 测试 consumer_in_slot_zero_saves_the_producer_a_poll 只把 mini 里两个 spawn 的顺序对调，让消费者占第 0 格，日志不变，poll 次数变成 [10, 7]。

谁先跑只改变 poll 次数，不改变日志；但依赖同一 tick 里谁先跑的代码，在 mini 和 embassy 上会得到不同的结果。

小结

• 执行器的骨架：定长任务槽里放类型擦除的 future 和一个 poll_fn，一个运行队列，一张 waker vtable（数据指针指向任务，clone / drop 什么也不做），一个「poll 一批、睡到被叫醒」的循环。mini 库 474 行代码，Cortex-M 和 std 各一个 Signal。
• 任务数固定时，运行队列可以就是一个 AtomicU32：fetch_or 去重并决定要不要 SEV，swap(0) 取走整批；批内按槽号，embassy 是后进先出。
• 定时器表按到期时刻排序，tick 只看表头；「到没到」和「登记」必须在同一个临界区里。rotate 在 opt-level = "s" 下拉进 1410 字节的 memmove，换成 swap 循环 .text 少 1456 字节。
• 同样两个任务（QEMU 11.1.2，release）：mini 7308 / 1192 / 452 字节（.text / .rodata / .bss），embassy 全家 7208 / 1700 / 364；只换执行器，embassy 多 76 字节 .text。两个执行器 900 ms 都转 1801 圈；std 上 mini 只转 17 圈。
• 同一 tick 两个任务同时到期时，mini 先跑槽号小的，embassy 先跑先登记的，生产者的 poll 次数因此是 11 对 10。
• embassy 多出来的是工程：按任务定大小的池、原地构造、不限任务数的队列、可替换的时间驱动和硬件闹钟、InterruptExecutor、调度策略和 trace。