上一章我們把 Future、Waker、Executor 三者的職責邊界講清楚了。但一個真實可用的執行時遠不止「一個 Executor」——它還需要 I/O 事件迴圈、定時器、阻塞執行緒池，並且這些元件必須共享同一套句柄、同一份生命週期。本章追蹤Builder::build的完整裝配鏈路，回答一個核心問題：一個Runtime內部到底有哪些元件，它們如何被拼裝並共享句柄。

Tokio 的裝配入口是Builder。它本身是一個純配置容器，所有欄位都是「意圖宣告」，不持有任何執行時資源。真正的資源建立發生在build()呼叫時。

直覺模型：Builder 是「裝修圖紙」，Runtime 是「交房後的房子」

Builder就像一張裝修圖紙：你在上面標註「要幾個房間（worker_threads）」「要不要通水（enable_io）」「要不要通電（enable_time）」「外包幫工上限（max_blocking_threads）」。圖紙本身不產生任何實體。直到呼叫build()，施工隊才按圖施工，把排程器、驅動、執行緒池這些「房間」真正建起來，並交付一個Runtime實例。

若沒有Builder這一層，使用者就必須手動 new 出每個元件、手動接線、手動處理失敗回滾——任何一處順序錯誤都會導致句柄懸空或資源洩漏。Builder的價值在於：把「配置」與「建構」徹底分離，讓建構過程可以集中做校驗、失敗清理和句柄共享。

記憶體佈局：Builder的欄位分區

Builder的欄位可以按職責分成四組。第一組是形態與開關：kind決定排程器形態，enable_io / enable_time決定是否建立對應驅動。

📎 tokio/src/runtime/builder.rs:55-68

pub struct Builder {
    kind: Kind,
    name: Option<String>,
    enable_io: bool,
    nevents: usize,
    nevents_busy: Option<usize>,
    enable_time: bool,
    start_paused: bool,
    // ...
}

第二組是執行緒池參數：worker_threads是Option<usize>，None表示「延遲到 build 時按 CPU 核數自動探測」；max_blocking_threads預設 512。

📎 tokio/src/runtime/builder.rs:73-79

worker_threads: Option<usize>,
max_blocking_threads: usize,

第三組是回呼鉤子，全部是Option<Arc<dyn Fn ...>>。注意它們用Arc而非Box，因為這些回呼要被複製進每個 worker 執行緒的Config。

📎 tokio/src/runtime/builder.rs:87-97

pub(super) after_start: Option<Callback>,
pub(super) before_stop: Option<Callback>,
pub(super) before_park: Option<Callback>,
pub(super) after_unpark: Option<Callback>,

第四組是調度啟發式與隨機種子：global_queue_interval、event_interval、disable_lifo_slot、seed_generator。

📎 tokio/src/runtime/builder.rs:116-134

pub(super) global_queue_interval: Option<u32>,
pub(super) event_interval: u32,
pub(super) disable_lifo_slot: bool,
pub(super) seed_generator: RngSeedGenerator,

這裡有一個值得注意的設計：Kind是一個Copy的小列舉，只有兩個變體。

📎 tokio/src/runtime/builder.rs:261-265

#[derive(Clone, Copy)]
pub(crate) enum Kind {
    CurrentThread,
    #[cfg(feature = "rt-multi-thread")]
    MultiThread,
}

MultiThread變體被rt-multi-threadfeature 門控。這意味著在只啟用rtfeature 的建構裡，Kind只有一個變體，build()的match會被編譯器最佳化成單分支——用型別系統而非執行時判斷來消除多執行緒排程器的程式碼體積。

預設值的哲學：為什麼 I/O 和 time 預設關閉

Builder::new是所有建構的公共入口。它把enable_io和enable_time都設為false。

📎 tokio/src/runtime/builder.rs:309-318

// I/O defaults to "off"
enable_io: false,
nevents: 1024,
nevents_busy: None,

// Time defaults to "off"
enable_time: false,

// The clock starts not-paused
start_paused: false,

enable_all()的實作揭示了 feature 門控如何影響「全開」的語義。

📎 tokio/src/runtime/builder.rs:398-419

pub fn enable_all(&mut self) -> &mut Self {
    #[cfg(any(
        feature = "net",
        all(unix, feature = "process"),
        all(unix, feature = "signal")
    ))]
    self.enable_io();

    #[cfg(all(
        tokio_unstable,
        feature = "io-uring",
        // ...
    ))]
    self.enable_io_uring();

    #[cfg(feature = "time")]
    self.enable_time();

    self
}

注意enable_io()只在啟用了net、process或signalfeature 時才被呼叫。如果使用者只啟用了time feature，enable_all()不會打開 I/O 驅動——因為編譯產物裡根本沒有 I/O 驅動程式碼。

裝配主路徑：build()的分流

build()是裝配的起點，它按kind分流到兩條完全不同的路徑。

📎 tokio/src/runtime/builder.rs:1146-1152

pub fn build(&mut self) -> io::Result<Runtime> {
    match &self.kind {
        Kind::CurrentThread => self.build_current_thread_runtime(),
        #[cfg(feature = "rt-multi-thread")]
        Kind::MultiThread => self.build_threaded_runtime(),
    }
}

這兩條路徑的差異遠不止「一個執行緒 vs 多個執行緒」。下面分別展開。

路徑一：current_thread 的裝配

build_current_thread_runtime本身很薄，它委託給build_current_thread_runtime_components，然後把返回的三元組包進Runtime。

📎 tokio/src/runtime/builder.rs:1725-1736

fn build_current_thread_runtime(&mut self) -> io::Result<Runtime> {
    use crate::runtime::runtime::Scheduler;

    let (scheduler, handle, blocking_pool) =
        self.build_current_thread_runtime_components(None)?;

    Ok(Runtime::from_parts(
        Scheduler::CurrentThread(scheduler),
        handle,
        blocking_pool,
    ))
}

真正的裝配邏輯在build_current_thread_runtime_components。它的執行順序至關重要：

📎 tokio/src/runtime/builder.rs:1760-1766

let mut cfg = self.get_cfg();
cfg.timer_flavor = TimerFlavor::Traditional;
let (driver, driver_handle) = driver::Driver::new(cfg)?;

// Blocking pool
let blocking_pool = blocking::create_blocking_pool(self, self.max_blocking_threads, 0);
let blocking_spawner = blocking_pool.spawner().clone();

第一步建立driver，返回一對(driver, driver_handle)。注意這裡?直接向上傳播錯誤——如果 I/O 驅動初始化失敗（比如 epoll 建立失敗），整個build返回Err，此時 blocking pool 還沒建立，無需清理。

第二步建立 blocking pool，並立刻取出它的spawner複製。這個spawner會被注入排程器，讓排程器有能力把阻塞任務投遞到執行緒池。

第三步生成兩個獨立的 RNG 種子生成器。

📎 tokio/src/runtime/builder.rs:1768-1770

let seed_generator_1 = self.seed_generator.next_generator();
let seed_generator_2 = self.seed_generator.next_generator();

第四步是核心：把 driver、driver_handle、blocking_spawner、種子和Config一起交給CurrentThread::new。

📎 tokio/src/runtime/builder.rs:1776-1807

let (scheduler, handle) = CurrentThread::new(
    driver,
    driver_handle,
    blocking_spawner,
    seed_generator_2,
    Config {
        before_park: self.before_park.clone(),
        after_unpark: self.after_unpark.clone(),
        // ...
        global_queue_interval: self.global_queue_interval,
        event_interval: self.event_interval,
        // ...
        enable_eager_driver_handoff: false,
        seed_generator: seed_generator_1,
        // ...
    },
    local_tid,
    self.name.clone(),
);

這裡有一個關鍵細節：enable_eager_driver_handoff被硬編碼為false。

📎 tokio/src/runtime/builder.rs:1795-1798

// This setting never makes sense for a current thread runtime,
// as it only configures how the I/O driver is stolen across
// workers.
enable_eager_driver_handoff: false,

最後，CurrentThread::new回傳的handle被包進scheduler::Handle::CurrentThread，再包進公開的Handle。

📎 tokio/src/runtime/builder.rs:1816-1822

let handle = Handle {
    inner: scheduler::Handle::CurrentThread(handle),
};

Ok((scheduler, handle, blocking_pool))

路徑二：multi_thread 的裝配

build_threaded_runtime的骨架與 current_thread 類似，但有三處本質差異。第一處是 worker 執行緒數的確定：

📎 tokio/src/runtime/builder.rs:2185

let worker_threads = self.worker_threads.unwrap_or_else(num_cpus);

None在這裡被解析為num_cpus()。這就是「延遲自動探測」的落地點——探測發生在 build 時而非Builder::new時，因為 CPU 親和性可能在兩者之間變化。

第二處差異在 blocking pool 的容量計算：

📎 tokio/src/runtime/builder.rs:2189-2192

let blocking_pool =
    blocking::create_blocking_pool(self, self.max_blocking_threads + worker_threads, worker_threads);
let blocking_spawner = blocking_pool.spawner().clone();

注意max_blocking_threads + worker_threads。對比 current_thread 路徑傳入的是self.max_blocking_threads和0。

📎 tokio/src/runtime/builder.rs:1765

let blocking_pool = blocking::create_blocking_pool(self, self.max_blocking_threads, 0);

第三處差異是MultiThread::new回傳三元組而非二元組：

📎 tokio/src/runtime/builder.rs:2198-2226

let (scheduler, handle, launch) = MultiThread::new(
    worker_threads,
    driver,
    driver_handle,
    blocking_spawner,
    seed_generator_2,
    Config {
        // ...
        enable_eager_driver_handoff: self.enable_eager_driver_handoff,
        // ...
    },
    self.timer_flavor,
    self.name.clone(),
);

多出來的launch是一個「啟動句柄」。MultiThread::new只負責建構排程器結構，並不立即啟動 worker 執行緒。真正的啟動發生在後面：

📎 tokio/src/runtime/builder.rs:2228-2234

let handle = Handle { inner: scheduler::Handle::MultiThread(handle) };

// Spawn the thread pool workers
let _enter = handle.enter();
launch.launch();

Ok(Runtime::from_parts(Scheduler::MultiThread(scheduler), handle, blocking_pool))

handle.enter()進入執行時上下文，然後launch.launch()才真正 spawn 出所有 worker 執行緒。這個「先建構、後啟動」的兩階段設計非常關鍵。

裝配流程圖

下面這張圖把兩條路徑的裝配順序、關鍵分支和錯誤路徑畫在一起。注意driver::Driver::new失敗時直接回傳Err，此時 blocking pool 尚未建立。

flowchart TD
    start["Builder::build()"] --> match_kind{"self.kind?"}

    match_kind -->|CurrentThread| ct_cfg["get_cfg() + timer_flavor=Traditional"]
    match_kind -->|MultiThread| mt_workers["worker_threads = self.worker_threads.unwrap_or_else(num_cpus)"]

    ct_cfg --> ct_driver["driver::Driver::new(cfg)?"]
    mt_workers --> mt_driver["driver::Driver::new(self.get_cfg())?"]

    ct_driver -->|Err| ret_err["return Err(io::Error)"]
    mt_driver -->|Err| ret_err

    ct_driver -->|Ok driver, driver_handle| ct_pool["create_blocking_pool(self, max_blocking_threads, 0)"]
    mt_driver -->|Ok driver, driver_handle| mt_pool["create_blocking_pool(self, max_blocking_threads + worker_threads, worker_threads)"]

    ct_pool --> ct_seed["next_generator() x2"]
    mt_pool --> mt_seed["next_generator() x2"]

    ct_seed --> ct_new["CurrentThread::new(driver, driver_handle, blocking_spawner, ...)"]
    mt_seed --> mt_new["MultiThread::new(worker_threads, driver, ...) -> (scheduler, handle, launch)"]

    ct_new --> ct_wrap["Handle { inner: CurrentThread(handle) }"]
    mt_new --> mt_wrap["Handle { inner: MultiThread(handle) }"]

    ct_wrap --> ct_rt["Runtime::from_parts(Scheduler::CurrentThread, handle, blocking_pool)"]
    mt_wrap --> mt_enter["handle.enter()"]
    mt_enter --> mt_launch["launch.launch() 启动 worker 线程"]
    mt_launch --> mt_rt["Runtime::from_parts(Scheduler::MultiThread, handle, blocking_pool)"]

句柄共享：Handle如何成為跨組件的「通行證」

裝配完成後，Runtime持有scheduler、handle、blocking_pool三件套。其中handle是共享的核心。它的內部是一個列舉：

📎 tokio/src/runtime/scheduler/mod.rs:29-41

#[derive(Debug, Clone)]
pub(crate) enum Handle {
    #[cfg(feature = "rt")]
    CurrentThread(Arc<current_thread::Handle>),

    #[cfg(feature = "rt-multi-thread")]
    MultiThread(Arc<multi_thread::Handle>),

    #[cfg(not(feature = "rt"))]
    #[allow(dead_code)]
    Disabled,
}

注意兩個變體都包著Arc。這意味著Handle的複製是廉價的引用計數遞增，可以被自由地分發到任意執行緒。Handle提供了統一的存取介面，把形態差異封裝在match內部。例如driver()：

📎 tokio/src/runtime/scheduler/mod.rs:53-64

pub(crate) fn driver(&self) -> &driver::Handle {
    match *self {
        #[cfg(feature = "rt")]
        Handle::CurrentThread(ref h) => &h.driver,

        #[cfg(feature = "rt-multi-thread")]
        Handle::MultiThread(ref h) => &h.driver,

        #[cfg(not(feature = "rt"))]
        Handle::Disabled => unreachable!(),
    }
}

blocking_spawner()用了match_flavor!巨集來消除重複：

📎 tokio/src/runtime/scheduler/mod.rs:96-98

pub(crate) fn blocking_spawner(&self) -> &blocking::Spawner {
    match_flavor!(self, Handle(h) => &h.blocking_spawner)
}

這個巨集展開後就是上面driver()那樣的match。它的價值在於：當新增一個需要按形態分發的存取器時，只需一行match_flavor!，而不必手寫兩遍match分支。

公開的Handle是內部scheduler::Handle的薄包裝：

📎 tokio/src/runtime/handle.rs:13-15

pub struct Handle {
    pub(crate) inner: scheduler::Handle,
}

使用者拿到的Handle可以跨執行緒複製、可以spawn、可以block_on。spawn的實作展示了AutoBox的編譯期分支：

📎 tokio/src/runtime/handle.rs:197-208

pub fn spawn<F>(&self, future: F) -> JoinHandle<F::Output>
where
    F: Future + Send + 'static,
    F::Output: Send + 'static,
{
    let fut_size = mem::size_of::<F>();
    if AutoBox::<F>::SHOULD_BOX {
        self.spawn_named(Box::pin(future), SpawnMeta::new_unnamed(fut_size))
    } else {
        self.spawn_named(future, SpawnMeta::new_unnamed(fut_size))
    }
}

AutoBox::<F>::SHOULD_BOX是一個關聯常數，由size_of::<F>()與閾值比較得出。

📎 tokio/src/runtime/mod.rs:668-673

pub(crate) struct AutoBox<T>(std::marker::PhantomData<T>);

impl<T> AutoBox<T> {
    pub(crate) const SHOULD_BOX: bool = std::mem::size_of::<T>() > BOX_FUTURE_THRESHOLD;
}

設計思考：裝配順序、錯誤恢復與生產踩坑

順序即契約。裝配順序driver -> blocking_pool -> scheduler不是隨意的。driver 最先建立，因為它是唯一可能因 OS 資源不足而失敗、且失敗後無需清理其他組件的步驟。blocking_pool 在 driver 之後、scheduler 之前，因為 scheduler 需要 blocking_spawner。如果 blocking_pool 建立失敗（實際上它不太會失敗），driver 會被 drop 自動清理。

current_thread 的local_tid分支。build_local走的是build_current_thread_local_runtime，它把當前執行緒 ID 傳進去：

📎 tokio/src/runtime/builder.rs:1738-1751

fn build_current_thread_local_runtime(&mut self) -> io::Result<LocalRuntime> {
    use crate::runtime::local_runtime::LocalRuntimeScheduler;

    let tid = std::thread::current().id();

    let (scheduler, handle, blocking_pool) =
        self.build_current_thread_runtime_components(Some(tid))?;

    Ok(LocalRuntime::from_parts(
        LocalRuntimeScheduler::CurrentThread(scheduler),
        handle,
        blocking_pool,
    ))
}

這個tid被存進Handle，後續can_spawn_local_on_local_runtime用它校驗「spawn_local 是否在 owner 執行緒上呼叫」：

📎 tokio/src/runtime/scheduler/mod.rs:140-147

pub(crate) fn can_spawn_local_on_local_runtime(&self) -> bool {
    match self {
        Handle::CurrentThread(h) => h.local_tid.is_some_and(|x| std::thread::current().id() == x),

        #[cfg(feature = "rt-multi-thread")]
        Handle::MultiThread(_) => false,
    }
}

生產踩坑一：worker_threads(0)會 panic。worker_threads方法有斷言：

📎 tokio/src/runtime/builder.rs:582-586

pub fn worker_threads(&mut self, val: usize) -> &mut Self {
    assert!(val > 0, "Worker threads cannot be set to 0");
    self.worker_threads = Some(val);
    self
}

這個斷言在配置階段就失敗，而不是等到 build。好處是錯誤定位更早，壞處是如果執行緒數來自設定檔的動態值，使用者必須在呼叫前自己校驗。

生產踩坑二：max_blocking_threads設太小會掛起。文件明確警告：

📎 tokio/src/runtime/builder.rs:600-601

/// It's recommended to not set this limit too low in order to avoid hanging on operations
/// requiring [`spawn_blocking`].

生產踩坑三：UnhandledPanic::ShutdownRuntime只支援 current_thread。

📎 tokio/src/runtime/builder.rs:1374-1381

pub fn unhandled_panic(&mut self, behavior: UnhandledPanic) -> &mut Self {
    if !matches!(self.kind, Kind::CurrentThread) && matches!(behavior, UnhandledPanic::ShutdownRuntime) {
        panic!("UnhandledPanic::ShutdownRuntime is only supported in current thread runtime");
    }

    self.unhandled_panic = behavior;
    self
}

本章小結

本章追蹤了Builder::build的完整裝配鏈路。核心結論：

1. Builder是純配置容器，build()才建立資源。裝配順序driver -> blocking_pool -> scheduler由錯誤恢復需求決定。

2. current_thread 與 multi_thread 的差異不止執行緒數：blocking pool 容量計算不同（max_blocking_threads vs max_blocking_threads + worker_threads），multi_thread 多一個launch兩階段啟動，enable_eager_driver_handoff在 current_thread 下被強制關閉。

3. Handle是跨元件共享的核心，內部用Arc包裹形態特定的句柄，透過match或match_flavor!巨集統一存取。

4. AutoBox用關聯常量在編譯期決定是否裝箱 future，避免程式碼體積翻倍。

5. local_tid是LocalRuntime安全性的執行時檢查點。

下一章，我們將進入任務的生命週期：spawn如何把一個 Future 變成可排程實體，JoinHandle如何與任務狀態機互動，以及任務在PENDING / RUNNING / COMPLETE之間的狀態遷移。

本章思考與自測

上一章我們完成了 Runtime 的裝配：I/O driver、time driver、blocking pool 與排程器被注入同一個Runtime實例，Handle成為跨執行緒存取這些元件的共享句柄。但裝配好的執行時此時還只是一個空殼——它擁有驅動任務的引擎，卻沒有任何任務可驅動。本章要回答的問題正是：當你敲下tokio::spawn(async { ... })的那一刻，那個async塊究竟經歷了什麼，才從一段普通的 Rust 程式碼變成一個「可被排程器接管、可被喚醒、可被 join」的實體。這是「任務的一生」的上半場，我們聚焦於誕生：從Handle::spawn出發，穿過new_task的引用計數分配，落到Cell<T, S>的記憶體佈局，最終看清任務如何被投遞到某個 worker 的本地佇列或全域注入佇列。下半場（第 4 章）才會進入排程迴圈與 poll/wake 閉環。

3.1 Future 不是任務：一次 spawn 到底創造了什麼

直覺模型

把Future想像成一張「菜譜」，把任務想像成「廚房裡正在被烹飪的一道菜」。菜譜本身是靜態的、可複製的、沒有任何執行狀態；只有當廚房（排程器）決定「現在做這道菜」，給它分配一個灶台（worker）、一個訂單號（TaskId）、一個出餐口（JoinHandle），它才成為一道「在製菜品」。若沒有這層包裝，排程器就無從知道「這道菜做到哪一步了」「誰在等它」「做好了通知誰」——它只能看到一張菜譜，無法管理。

資料結構與記憶體佈局

Tokio 用Task<S>表示「被執行時擁有的任務引用」，它是對RawTask的透明包裝：

#[repr(transparent)]
pub(crate) struct Task<S: 'static> {
    raw: RawTask,
    _p: PhantomData<S>,
}

📎 tokio/src/runtime/task/mod.rs:233-238

#[repr(transparent)]意味著Task<S>與RawTask在記憶體上完全一致，沒有額外開銷。PhantomData<S>只是編譯期的型別標記，標記這個任務屬於哪個排程器型別S。

真正承載任務全部狀態的是Cell<T, S>，它的佈局是整個任務模組的基石：

#[repr(C)]
pub(super) struct Cell<T: Future, S> {
    pub(super) header: Header,
    pub(super) core: Core<T, S>,
    pub(super) trailer: Trailer,
}

📎 tokio/src/runtime/task/core.rs:126-136

三個欄位按「熱-溫-冷」排列。Header是熱資料（每次排程、每次狀態轉換都要存取），Core是溫資料（poll 時存取），Trailer是冷資料（僅在建立與銷毀時存取）。註解明確寫道：Header必須是第一個欄位，因為任務結構體會同時被*mut Cell和*mut Header引用📎 tokio/src/runtime/task/core.rs:37-43。

更關鍵的是快取行對齊。Cell上掛著一長串#[cfg_attr(..., repr(align(...)))]，按目標架構選擇對齊位元組數：x86_64/aarch64/powerpc64 用 128 位元組，arm/mips/sparc/hexagon 用 32 位元組，m68k 用 16 位元組，s390x 用 256 位元組，其餘預設 64 位元組📎 tokio/src/runtime/task/core.rs:64-125。註解解釋了為什麼 x86_64 要用 128 而非 64：從 Intel Sandy Bridge 起，空間預取器會一次拉取成對的 64 位元組快取行，所以必須對齊到 128 位元組才能避免偽共享📎 tokio/src/runtime/task/core.rs:45-53。

Header本身被約束在 8 個指標大小以內：

#[test]
#[cfg(not(loom))]
fn header_lte_cache_line() {
    assert!(std::mem::size_of::<Header>() <= 8 * std::mem::size_of::<*const ()>());
}

📎 tokio/src/runtime/task/core.rs:591-593

這個測試確保Header不會超過 64 位元組（8 × 8），從而在 64 位元組快取行的架構上能完整落入一行。Header的欄位包括：state: State（原子狀態位）、queue_next: UnsafeCell<Option<NonNull<Header>>>（注入佇列的鏈結串列指標）、vtable: &'static Vtable（函式指標表）、owner_id: UnsafeCell<Option<NonZeroU64>>（所屬OwnedTasks列表的 ID）、scheduled_at: UnsafeCell<ScheduleLatencyInstant>（排程延遲測量）📎 tokio/src/runtime/task/core.rs:169-198。

Core<T, S>持有排程器句柄scheduler: S、任務 IDtask_id: Id，以及最核心的stage: CoreStage<T> 📎 tokio/src/runtime/task/core.rs:148-165。Stage是一個三態列舉：

#[repr(C)]
pub(super) enum Stage<T: Future> {
    Running(T),
    Finished(super::Result<T::Output>),
    Consumed,
}

📎 tokio/src/runtime/task/core.rs:225-229

這正是「Future 與 Output 復用同一塊記憶體」的關鍵：任務執行期間Stage::Running持有 future，完成後原地替換為Stage::Finished(output)，被JoinHandle取走後變為Stage::Consumed。#[repr(C)]註解指向一個 Miri issue，說明這個佈局對 unsafe 程式碼的正確性有硬性要求📎 tokio/src/runtime/task/core.rs:225-229。

Trailer存放冷資料：owned: linked_list::Pointers<Header>（OwnedTasks鏈結串列指標）、waker: UnsafeCell<Option<Waker>>（等待任務完成的消費者 waker）、hooks: TaskHarnessScheduleHooks 📎 tokio/src/runtime/task/core.rs:205-213。

Step-by-Step：從 spawn 到入隊

我們代入一個具體場景：在 multi_thread 執行時中，worker 執行緒 A 執行tokio::spawn(async { 42 })。

第一步：建構任務三件套。 new_task是任務誕生的唯一入口：

fn new_task<T, S>(
    task: T,
    scheduler: S,
    id: Id,
    spawned_at: SpawnLocation,
) -> (Task<S>, Notified<S>, JoinHandle<T::Output>)

📎 tokio/src/runtime/task/mod.rs:336-346

它呼叫RawTask::new::<T, S>分配Cell，然後從同一個raw指標派生出三個引用：Task（owned 引用，通常立即放入OwnedTasks）、Notified（通知引用，交給排程器）、JoinHandle（結果讀取句柄）📎 tokio/src/runtime/task/mod.rs:347-363。注意三者共享同一個raw，各自持有一個引用計數。

第二步：分配Cell並寫入初始狀態。 Cell::new在堆上分配整個結構：

let result = Box::new(Cell {
    trailer: Trailer::new(scheduler.hooks()),
    header: new_header(state, vtable, ...),
    core: Core {
        scheduler,
        stage: CoreStage {
            stage: UnsafeCell::new(Stage::Running(future)),
        },
        task_id,
        ...
    },
});

📎 tokio/src/runtime/task/core.rs:261-278

vtable由raw::vtable::<T, S>()生成，是一張針對具體T和S單態化的函式指標表📎 tokio/src/runtime/task/core.rs:260。future 被直接移入Stage::Running，沒有額外裝箱。

第三步：debug 斷言驗證佈局。在debug_assertions下，Cell::new會呼叫check函式，用Header::get_trailer、Header::get_scheduler、Header::get_id_ptr等基於 vtable 偏移量的指標運算，逐一斷言「透過 header 反查到的欄位位址」與「實際欄位位址」一致📎 tokio/src/runtime/task/core.rs:280-321。這是對 vtable 偏移量正確性的執行期自檢。

第四步：投遞到排程器。排程器拿到Notified<S>後，呼叫Schedule::schedule 📎 tokio/src/runtime/task/mod.rs:315。在 multi_thread 下，這會走push_back_or_overflow，把任務推入當前 worker 的本地佇列，佇列滿時溢出到注入佇列。

下面這張圖刻畫了從new_task到入隊的控制流與分支：

flowchart TD
    spawn_call["Handle::spawn(future)"] --> new_task["new_task::<T,S>(future, scheduler, id)"]
    new_task --> raw_new["RawTask::new::<T,S>"]
    raw_new --> cell_new["Cell::new: Box::new(Cell{header, core, trailer})"]
    cell_new --> vtable["raw::vtable::<T,S>() 生成函数指针表"]
    cell_new --> stage["Stage::Running(future) 移入"]
    cell_new --> debug_check{"debug_assertions?"}
    debug_check -->|是| check_layout["check(): 断言 trailer/scheduler/id 偏移量"]
    debug_check -->|否| skip_check["跳过"]
    check_layout --> triple["派生 (Task, Notified, JoinHandle)"]
    skip_check --> triple
    triple --> owned["Task 存入 OwnedTasks"]
    triple --> sched["Notified 交给 Schedule::schedule"]
    sched --> push{"本地队列有容量?"}
    push -->|是| local_push["push_back_finish: 写入 buffer[tail & MASK]"]
    push -->|否| overflow_check{"steal == real?"}
    overflow_check -->|否, 有并发窃取| inject_only["overflow.push(task) 仅注入"]
    overflow_check -->|是| push_overflow["push_overflow: CAS 认领后半批"]
    push_overflow --> cas_ok{"CAS 成功?"}
    cas_ok -->|是| inject_batch["overflow.push_batch(后半批 + 当前 task)"]
    cas_ok -->|否| retry["返回 Err(task), 重试 push_back_or_overflow"]
    retry --> push

這張圖揭示了幾個關鍵分支：debug 斷言只在除錯建置生效；本地佇列滿時並非直接溢出，而是先判斷是否有並行竊取者（steal != real），若有則只把當前任務推入注入佇列，因為竊取者騰出的空間很快可用。

設計思考：為什麼是三個引用而不是一個

new_task返回三個引用，而非一個。這是引用計數設計的核心：Task代表「執行期擁有這個任務」，Notified代表「這個任務已被通知、待排程」，JoinHandle代表「有人關心它的結果」。三者生命週期獨立——JoinHandle可以被 drop（任務繼續執行，結果丟棄），Notified在 poll 後消失，Task在任務完成並從OwnedTasks移除後釋放。若只有一個引用，就無法表達「任務還在跑但沒人 join」這種狀態。

UnownedTask是另一個重要分支：它持有兩個引用計數，用於 blocking 任務（不存入OwnedTasks）📎 tokio/src/runtime/task/mod.rs:286-295。unowned函式透過mem::forget(task)和mem::forget(notified)把兩個引用合併進UnownedTask 📎 tokio/src/runtime/task/mod.rs:388-397。這個「兩個引用」的設計動機是：blocking 任務沒有OwnedTasks列表來持有 owned 引用，所以需要額外一個引用計數來保證任務在執行期間不被釋放。

3.2 狀態位：一個 usize 如何編碼任務的全部生命週期

直覺模型

把任務狀態想像成一張「體檢報告單」，上面有若干獨立的勾選框：是否正在被 poll、是否已完成、是否被通知、是否被取消、是否有人 join。Tokio 沒有用多個布林欄位，而是把這些勾選位壓進一個AtomicUsize。這樣每次狀態轉換只需一次 CAS，而非多次加鎖。若沒有這個設計，任務狀態轉換會變成多把鎖的嵌套，死鎖風險與開銷都會飆升。

位域佈局

State的位域在模組文件中有完整定義📎 tokio/src/runtime/task/mod.rs:32-53：

• RUNNING：任務是否正在被 poll 或取消。這一位同時充當任務的鎖 📎 tokio/src/runtime/task/mod.rs:37-38。
• COMPLETE：future 已完全完成並被 drop。一旦置位永不清除，且永不與RUNNING同時置位📎 tokio/src/runtime/task/mod.rs:40-41。
• NOTIFIED：當前是否存在一個Notified物件📎 tokio/src/runtime/task/mod.rs:43。
• CANCELLED：任務應盡快被取消📎 tokio/src/runtime/task/mod.rs:45-46。
• JOIN_INTEREST：存在JoinHandle 📎 tokio/src/runtime/task/mod.rs:48。
• JOIN_WAKER：作為 join handle waker 的存取控制位📎 tokio/src/runtime/task/mod.rs:50-51。

剩餘位用於引用計數📎 tokio/src/runtime/task/mod.rs:53。

RUNNING位充當鎖這一點值得展開。模組文件的 Safety 章節指出：對 future 的任何可變存取都必須在修改RUNNING位獲得鎖之後進行，從而保證獨佔存取📎 tokio/src/runtime/task/mod.rs:130-133。這意味著 poll 一個任務時，執行緒先 CAS 設定RUNNING，成功後獨佔 future；若失敗說明別的執行緒正在 poll，本次 poll 直接返回。這把「poll 的互斥」與「狀態轉換」合併成一次原子操作，避免了單獨的互斥鎖。

JOIN_WAKER 的存取控制協定

JOIN_WAKER位是整個狀態機中最精妙的部分。它解決的問題是：waker欄位（在Trailer中）會被兩個執行緒並行存取——執行期在任務完成時讀它來喚醒 join 者，JoinHandle在 poll 時寫它來註冊 waker。模組文件給出了 7 條規則📎 tokio/src/runtime/task/mod.rs:75-120：

1. JOIN_WAKER初始為 0。

2. 為 0 時，JoinHandle對 waker 欄位有獨佔（可變）存取權。

3. 為 1 時，JoinHandle只有共享（唯讀）存取權。

4. 為 1 且COMPLETE為 1 時，執行期對 waker 欄位有共享（唯讀）存取權。

5. JoinHandle要寫 waker，必須：(i) 成功把JOIN_WAKER置 0 以獲得獨佔權，(ii) 寫入 waker，(iii) 成功把JOIN_WAKER置 1。

6. JoinHandle只能在COMPLETE為 0 時改JOIN_WAKER；執行期只能在COMPLETE為 1 時改。

7. 若JOIN_INTEREST為 0 且COMPLETE為 1，執行期對 waker 欄位有獨佔存取權（用於 drop waker）。

規則 6 隱含了競態：步驟 (i) 或 (iii) 可能失敗。若 (i) 失敗，放棄寫 waker；若 (iii) 失敗（另一執行緒在此期間置了COMPLETE），則清空 waker 欄位📎 tokio/src/runtime/task/mod.rs:110-120。這套協定的本質是：用一個原子位在「寫者」和「讀者」之間動態轉移所有權，避免為 waker 欄位單獨加鎖。

引用計數的兩種遞減

Task的 drop 遞減一次引用計數，UnownedTask的 drop 遞減兩次：

impl<S: 'static> Drop for Task<S> {
    fn drop(&mut self) {
        if self.header().state.ref_dec() {
            self.raw.dealloc();
        }
    }
}

📎 tokio/src/runtime/task/mod.rs:580-586

impl<S: 'static> Drop for UnownedTask<S> {
    fn drop(&mut self) {
        if self.raw.header().state.ref_dec_twice() {
            self.raw.dealloc();
        }
    }
}

📎 tokio/src/runtime/task/mod.rs:590-596

ref_dec回傳true表示這是最後一個引用，此時才真正釋放Cell記憶體。ref_dec_twice是UnownedTask持有兩個計數的直接體現。

設計思考：為什麼狀態位與引用計數共用一個原子

3.3 JoinHandle：結果如何跨越任務邊界回傳

直覺模型

JoinHandle就像餐廳給你的「取餐憑證」。任務（廚房）完成時，把菜品（output）放到出餐口（Stage::Finished），然後按響你的取餐器（waker）。你拿著憑證來取，憑證本身不持有菜品，只是指向出餐口的指針。若你把憑證丟了（dropJoinHandle），菜品會被直接倒掉（output 被 drop），但廚房不會因此停工。

資料結構

JoinHandle<T>同樣是對RawTask的透明包裝：

pub struct JoinHandle<T> {
    raw: RawTask,
    _p: PhantomData<T>,
}

📎 tokio/src/runtime/task/join.rs:163-166

PhantomData<T>標記輸出類型。JoinHandle<T>在T: Send時才是Send/Sync 📎 tokio/src/runtime/task/join.rs:169-170，這保證了非 Send 輸出不會被跨執行緒移動。

Step-by-Step：await 一個 JoinHandle

JoinHandle實作了Future，其poll是結果回傳的核心：

fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
    ready!(crate::trace::trace_leaf());
    let mut ret = Poll::Pending;
    let coop = ready!(crate::task::coop::poll_proceed(cx));
    unsafe {
        self.raw.try_read_output(&mut ret, cx.waker());
    }
    if ret.is_ready() {
        coop.made_progress();
    }
    ret
}

📎 tokio/src/runtime/task/join.rs:327-354

注意幾個細節：trace_leaf用於 tracing 插樁；coop::poll_proceed消耗協作預算（第 12 章詳述）；try_read_output透過 vtable 擦除泛型，把返回值放在堆疊上、用*mut ()傳入📎 tokio/src/runtime/task/join.rs:327-354。這個「返回值放堆疊上」的技巧是因為 vtable 函數無法泛型化返回類型T，只能透過裸指標回寫。

結果的所有權轉移

模組文件的「Non-Send output」章節精確描述了結果的所有權規則📎 tokio/src/runtime/task/mod.rs:151-170：

• 任務完成時，output 被放入Stage，然後執行「設置 COMPLETE」的轉換，並讀取此刻的JOIN_INTEREST值。
• 若JOIN_INTEREST為 0（無JoinHandle），output 立即被 drop📎 tokio/src/runtime/task/mod.rs:157-158。
• 若JOIN_INTEREST為 1，JoinHandle負責清理 output📎 tokio/src/runtime/task/mod.rs:160-161。

對非 Send output，文件給出了三步論證：output 在 poll future 的執行緒上建立；JoinHandle<Output>在 Output 非 Send 時也非 Send，所以它也在 spawn 執行緒上；因此JoinHandle取走或 drop output 時不會跨執行緒移動📎 tokio/src/runtime/task/mod.rs:164-170。

JoinHandle 的 drop：快慢兩條路徑

impl<T> Drop for JoinHandle<T> {
    fn drop(&mut self) {
        if self.raw.state().drop_join_handle_fast().is_ok() {
            return;
        }
        self.raw.drop_join_handle_slow();
    }
}

📎 tokio/src/runtime/task/join.rs:358-364

drop_join_handle_fast嘗試用一次 CAS 完成「清除JOIN_INTEREST位 + 遞減引用計數」。若失敗（例如任務正在完成，狀態位被佔用），則走drop_join_handle_slow的慢路徑。這是典型的「樂觀快路徑 + 悲觀慢路徑」模式。

設計思考：為什麼 JoinHandle 不直接持有 output

3.4 本地佇列：work-stealing 的生產者-消費者結構

直覺模型

每個 worker 有一個「私人待辦清單」（本地佇列），容量 256。worker 自己從頭部取任務（LIFO，利用快取局部性），其他 worker 從尾部竊取任務（FIFO，取走最老的、最可能已完成的任務）。若沒有本地佇列，所有任務都擠在全局佇列，每次取任務都要競爭全局鎖，多核擴展性會崩潰。

記憶體佈局：head 與 tail 的分離

pub(crate) struct Inner<T: 'static> {
    head: AtomicUnsignedLong,
    tail: AtomicUnsignedShort,
    buffer: Box<[UnsafeCell<MaybeUninit<task::Notified<T>>>; LOCAL_QUEUE_CAPACITY]>,
}

📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:36-57

head是AtomicUnsignedLong（64 位，若平台支援 u64），tail是AtomicUnsignedShort（32 位）。註解解釋了為什麼索引比實際需要更寬：為了 ABA 緩解，以及區分「滿」和「空」緩衝區📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:37-49。

head內部打包了兩個 UnsignedShort：低位是「真實頭部」（real head），高位是「竊取者正在處理的第一個位置」（steal head）。當兩者相等時，沒有活躍的竊取者📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:39-49。這個雙值打包是 work-stealing 佇列的核心技巧：竊取者先 CAS 更新 steal 值來「認領」一批任務，完成後把 steal 值追上 real 值，表示竊取結束。

LOCAL_QUEUE_CAPACITY在非 loom 下是 256，loom 下縮到 4 以便測試更多邊界📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:62-69。MASK = LOCAL_QUEUE_CAPACITY - 1，用於環形緩衝區索引📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:71。

Step-by-Step：push_back_or_overflow 的完整分支

這是本地佇列最複雜的函式，我們逐分支解析：

pub(crate) fn push_back_or_overflow<O: Overflow<T>>(
    &mut self,
    mut task: task::Notified<T>,
    overflow: &O,
    stats: &mut Stats,
) {
    let tail = loop {
        let head = self.inner.head.load(Acquire);
        let (steal, real) = unpack(head);
        let tail = unsafe { self.inner.tail.unsync_load() };

        if tail.wrapping_sub(steal) < LOCAL_QUEUE_CAPACITY as UnsignedShort {
            break tail;
        } else if steal != real {
            overflow.push(task);
            return;
        } else {
            match self.push_overflow(task, real, tail, overflow, stats) {
                Ok(_) => return,
                Err(v) => { task = v; }
            }
        }
    };
    self.push_back_finish(task, tail);
}

📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:188-223

三條分支：

1. 有容量（tail - steal < CAPACITY）：break tail，跳出迴圈後呼叫push_back_finish寫入緩衝區。

2. 無容量但有並行竊取者（steal != real）：竊取者會騰出空間，所以只把當前任務推入注入佇列，立即返回📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:204-208。

3. 無容量且無竊取者：呼叫push_overflow把後半批任務溢出到注入佇列📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:209-219。若 CAS 失敗（輸給並行竊取者），push_overflow返回Err(task)，迴圈重試。

push_back_finish寫入任務並更新 tail：

fn push_back_finish(&self, task: task::Notified<T>, tail: UnsignedShort) {
    let idx = tail as usize & MASK;
    self.inner.buffer[idx].with_mut(|ptr| {
        unsafe { ptr::write((*ptr).as_mut_ptr(), task); }
    });
    self.inner.tail.store(tail.wrapping_add(1), Release);
}

📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:226-244

Release序保證寫入的任務對竊取者可見。

push_overflow：為什麼溢出後半批

const NUM_TASKS_TAKEN: UnsignedShort = (LOCAL_QUEUE_CAPACITY / 2) as UnsignedShort;

📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:265

溢出時取走 128 個任務。註解詳細解釋了為什麼取後半批而非前半批📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:295-306：從注入佇列取任務時，總是放在前半部分。所以若一個任務在後半部分，就能確定它不是剛從注入佇列取來的。這保證了「從注入佇列取出的任務不會被立刻放回注入佇列」（至少在被 poll 一次之前）。

CAS 認領後半批：

if self.inner.head.compare_exchange_weak(
    pack(head, head), pack(tail, tail), Release, Relaxed
).is_err() {
    return Err(task);
}

📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:283-293

把head從(head, head)更新到(tail, tail)，即同時推進 steal 和 real 到 tail，認領全部任務。成功後把 tail 回退到tail + NUM_TASKS_TAKEN，表示前半批仍留在本地佇列📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:314-316。

pop 與 steal_into：取任務的兩條路徑

pop是 worker 自己取任務（從頭部，LIFO）：

pub(crate) fn pop(&mut self) -> Option<task::Notified<T>> {
    let mut head = self.inner.head.load(Acquire);
    let idx = loop {
        let (steal, real) = unpack(head);
        let tail = unsafe { self.inner.tail.unsync_load() };
        if real == tail { return None; }
        let next_real = real.wrapping_add(1);
        let next = if steal == real {
            pack(next_real, next_real)
        } else {
            assert_ne!(steal, next_real);
            pack(steal, next_real)
        };
        let res = self.inner.head.compare_exchange_weak(head, next, AcqRel, Acquire);
        match res {
            Ok(_) => break real as usize & MASK,
            Err(actual) => head = actual,
        }
    };
    Some(self.inner.buffer[idx].with(|ptr| unsafe { ptr::read(ptr).assume_init() }))
}

📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:361-399

關鍵分支：若steal == real（無竊取者），同時推進兩者；否則只推進 real，保留 steal 不動📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:377-384。assert_ne!(steal, next_real)確保不會把 real 推進到 steal 的位置，否則會破壞竊取者的認領狀態。

steal_into是竊取路徑，先檢查目標佇列是否有足夠空間：

if dst_tail.wrapping_sub(steal) > LOCAL_QUEUE_CAPACITY as UnsignedShort / 2 {
    return None;
}

📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:431-435

目標佇列超過半滿就不竊取，避免竊取後立刻又溢出。

steal_into2是竊取的核心，計算竊取數量：

let n = src_tail.wrapping_sub(src_head_real);
let n = n - n / 2;

📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:487-488

竊取一半（向上取整）。然後 CAS 更新 head 的 steal 值來認領：

let steal_to = src_head_real.wrapping_add(n);
next_packed = pack(src_head_steal, steal_to);
let res = self.0.head.compare_exchange_weak(prev_packed, next_packed, AcqRel, Acquire);

📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:496-506

注意這裡只更新了 real 值（pack(src_head_steal, steal_to)中 steal 保持不變），把 real 推進到steal_to。這表示「這些任務已被認領，其他竊取者不能再碰」。竊取完成後，再把 steal 追上 real：

loop {
    let head = unpack(prev_packed).1;
    next_packed = pack(head, head);
    let res = self.0.head.compare_exchange_weak(prev_packed, next_packed, AcqRel, Acquire);
    match res {
        Ok(_) => return n,
        Err(actual) => prev_packed = actual,
    }
}

📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:548-561

下面這張時序圖刻畫了「生產者 push、消費者 pop、竊取者 steal」三方並行互動：

sequenceDiagram
    participant P as "Worker A (生产者)"
    participant Q as "Local 队列 Inner"
    participant C as "Worker A (消费者 pop)"
    participant S as "Worker B (窃取者)"

    P->>Q: "load head (Acquire)"
    P->>Q: "unsync_load tail"
    Note over P: "tail - steal < 256?"
    P->>Q: "push_back_finish: buffer[idx] = task"
    P->>Q: "store tail+1 (Release)"

    C->>Q: "load head (Acquire)"
    C->>Q: "unsync_load tail"
    Note over C: "real == tail? 空则返回 None"
    C->>Q: "CAS head: pack(real+1, real+1)"
    Q-->>C: "Ok, 读取 buffer[real & MASK]"

    S->>Q: "load head (Acquire)"
    S->>Q: "load tail (Acquire)"
    Note over S: "src_head_steal != src_head_real? 返回 0"
    S->>Q: "CAS head: pack(steal, real+n) 认领一半"
    Q-->>S: "Ok, 拷贝 n 个任务到 dst"
    S->>Q: "CAS head: pack(real+n, real+n) 完成窃取"
    Q-->>S: "返回 n"

設計思考：為什麼本地佇列是 LIFO 而竊取是 FIFO

至此，任務已經完成了從 Future 到可排程實體的蛻變：它被分配了引用計數、放進了Cell的記憶體佈局，並成功投遞到 worker 的本地佇列或全域注入佇列。但任務被放入佇列只是開始，真正讓它運轉起來的是 worker 執行緒的排程迴圈。下一章我們將進入「任務的一生」下半場，追蹤 worker 如何從佇列中取出任務、呼叫Future::poll，並在返回Pending時透過Waker註冊喚醒，最終觸發schedule重新入隊——「喚醒 → 入隊 → 再 poll」這一閉環的完整呼叫路徑，以及 work-stealing 策略與 LIFO 槽位優化，都將在那裡揭曉。

上一章我們把任務送進了Local佇列或全域注入佇列。但佇列只是「待辦清單」，真正讓任務跑起來的，是 worker 執行緒裡那個永不停止的迴圈。這一章我們追蹤Context::run——它是整個多執行緒排程器的心臟。

先建立直覺：worker 執行緒就像一個廚師，面前有一疊自己的訂單（run_queue），旁邊還有一個公共訂單架（inject）。廚師先看自己手邊最近的一張（lifo_slot），沒有就從自己那疊拿，再沒有就去公共架抓一把，還不行就去別的廚師那疊裡偷幾張。全都空了他才去休息，但休息時耳朵還豎著——一有訂單進來就立刻醒來。

若沒有這個迴圈，任務被入隊後就永遠躺在佇列裡，Future::poll永遠不會被呼叫，整個執行時就是一堆死資料。

Core 的記憶體佈局與狀態欄位

worker 的可變狀態全部裝在Core裡，它被Box分配在堆上，透過AtomicCell<Core>在Worker與執行緒本地Context之間傳遞。

Core的關鍵欄位如下📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:113-167：

• tick: u32：每次迴圈自增，用於週期性觸發維護（maintenance）和全域佇列檢查。
• lifo_slot: Option<Notified>：LIFO 槽位，這是本章最精妙的設計。當 worker 自己排程一個任務時，它不進run_queue，而是放進這個槽位，下次取任務時優先從這裡拿。
• lifo_enabled: bool：LIFO 槽位的開關，用於防止 ping-pong 場景下的飢餓。
• run_queue: queue::Local<Arc<Handle>>：本地佇列，上一章剖析過的Local結構。
• is_searching: bool：worker 是否正在搜尋可竊取的任務。
• is_shutdown: bool / is_traced: bool：關閉與追蹤標誌。
• park: Option<Parker>：park 器，用Option包裹是為了在借用檢查器下方便地取出/放回。
• global_queue_interval: u32：多久檢查一次全域佇列。
• rand: FastRand：快速隨機數生成器，用於隨機選擇竊取起點。

為什麼 LIFO 能降低延遲？考慮一個典型的訊息傳遞場景：任務 A 處理完訊息後喚醒任務 B，B 處理完又喚醒 A。如果 A 喚醒 B 後 B 立刻執行，B 需要的資料很可能還在 CPU 快取裡（因為 A 剛碰過）。如果 B 被塞到佇列尾部，等前面幾十個任務跑完，快取早被沖掉了。

但 LIFO 有飢餓風險。原始碼用MAX_LIFO_POLLS_PER_TICK = 3來限制📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:263-263：每個 tick 最多優先 LIFO 槽位 3 次，超過就停用，讓其他任務有機會執行。

主迴圈 walkthrough：一次完整的排程週期

我們代入一個具體場景：worker 0 剛從park中醒來，run_queue裡有 5 個任務，lifo_slot裡有 1 個任務，全域佇列有 3 個任務。

主迴圈入口是Context::run 📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:570-642。它先重置lifo_enabled（因為 core 可能被block_in_place偷走過，狀態需要歸位）📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:571-573，然後進入while !core.is_shutdown迴圈。

每輪迴圈做四件事：

第一步：tick 與維護。 core.tick()自增計數器📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:587。接著self.maintenance(core)檢查tick % event_interval == 0，若是則呼叫park_yield以 0 逾時驅動 I/O 和定時器📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:809-826。

第二步：取任務。 core.next_task(&self.worker)是核心取任務邏輯📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1090-1156。它分兩條路徑：

• 當tick % global_queue_interval == 0時，優先從全域佇列取，取不到再取本地📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1091-1098。這是為了防止全域佇列裡的任務被餓死。
• 否則優先取本地任務📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1090-1156。

本地取任務由next_local_task完成📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1158-1160：

fn next_local_task(&mut self) -> Option<Notified> {
    self.lifo_slot.take().or_else(|| self.run_queue.pop())
}

先取 LIFO 槽位，再取佇列頭部（LIFO 彈出）。這就是上一章說的「本地 LIFO」。

如果本地為空但全域佇列非空，worker 會批次從全域佇列拉取任務📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1110-1154。批次大小n的計算很講究：min(inject.len() / remotes.len() + 1, cap)，其中cap又取min(remaining_slots, max_capacity / 2)。原始碼註解解釋了為什麼限制在佇列容量的一半📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1120-1131：確保拉取的任務落在本地佇列的前半部分，這樣即使後續發生溢出，這些任務也不會被推回全域佇列（溢出只影響後半部分）。

第三步：執行任務。拿到任務後呼叫run_task 📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:647-796。這是本章最複雜的函式，我們下一節專門展開。

第四步：竊取或 park。如果next_task返回None，說明本地和全域都沒活了，呼叫steal_work 📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1167-1195。竊取失敗則進入park或park_yield 📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:613-621。

整個控制流如下：

flowchart TD
    start["Context::run 进入循环"] --> tick["core.tick() 自增"]
    tick --> maint{"tick % event_interval == 0?"}
    maint -->|是| park_yield["park_yield 驱动 I/O 与定时器"]
    maint -->|否| next
    park_yield --> next["core.next_task()"]
    next --> has_task{"取到任务?"}
    has_task -->|是| run_task["run_task 执行 poll"]
    run_task --> cont{"core 还在?"}
    cont -->|是| tick
    cont -->|否| ret["return 退出"]
    has_task -->|否| steal["core.steal_work()"]
    steal --> steal_ok{"窃取成功?"}
    steal_ok -->|是| run_task
    steal_ok -->|否| defer_check{"defer 非空?"}
    defer_check -->|是| py["park_yield"]
    defer_check -->|否| pk["park 阻塞等待"]
    py --> tick
    pk --> tick

run_task：poll 與 LIFO 槽位的閉環

run_task是任務真正被poll的地方，也是「喚醒 → 入隊 → 再 poll」閉環的收口點。

進入函式後第一件事是assert_owner 📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:648，把Notified轉換成Task，同時斷言當前執行緒確實是這個任務的 owner（debug 斷言）。

接著transition_from_searching 📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:652——如果 worker 之前在搜尋狀態，現在找到任務了，要退出搜尋狀態，並可能喚醒其他 parked worker。

然後是關鍵的 budget 包裹📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:695-795：

coop::budget(|| {
    task.run();
    let mut lifo_polls = 0;
    loop {
        let mut core = match self.core.borrow_mut().take() {
            Some(core) => core,
            None => return ControlFlow::Break(()),
        };
        let task = match core.lifo_slot.take() {
            Some(task) => task,
            None => {
                self.reset_lifo_enabled(&mut core);
                core.stats.end_poll();
                return ControlFlow::Continue(core);
            }
        };
        if !coop::has_budget_remaining() {
            core.run_queue.push_back_or_overflow(task, ...);
            return ControlFlow::Continue(core);
        }
        lifo_polls += 1;
        if lifo_polls >= MAX_LIFO_POLLS_PER_TICK {
            core.lifo_enabled = false;
        }
        let task = self.worker.handle.shared.owned.assert_owner(task);
        *self.core.borrow_mut() = Some(core);
        task.run();
    }
})

這段程式碼揭示了 LIFO 槽位的完整閉環：task.run()執行Future::poll，poll 過程中如果任務喚醒了自己或別的任務，schedule_local會把新任務放進lifo_slot 📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1396-1408。poll 返回後，迴圈立刻檢查lifo_slot，如果有任務就繼續跑——不回到主迴圈，直接在同一個 budget 內連續 poll。

這就是「喚醒 → 入隊 → 再 poll」在 LIFO 路徑上的體現：喚醒時任務被放進lifo_slot，poll 返回後立即被取出再 poll，形成緊密的閉環。

注意self.core.borrow_mut().take()的None分支📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:716-724：如果 core 被偷走了（比如任務裡呼叫了block_in_place），worker 必須返回ControlFlow::Break(())，讓Context::run退出。這是block_in_place與調度循環的交互點。

喚醒路徑：Waker 如何觸發重新入隊

當Future::poll返回Pending時，任務需要註冊一個Waker，等事件就緒時被喚醒。Tokio 的Waker實現極其精簡——它就是一個指向任務Header的裸指標加一張 vtable。

waker_ref構造WakerRef 📎 tokio/src/runtime/task/waker.rs:11-34，用ManuallyDrop包裹Waker避免 drop 時減引用計數。vtable 是靜態的📎 tokio/src/runtime/task/waker.rs:119-119：

static WAKER_VTABLE: RawWakerVTable =
    RawWakerVTable::new(clone_waker, wake_by_val, wake_by_ref, drop_waker);

四個函數都只是把裸指標還原成Header，然後調用RawTask的對應方法📎 tokio/src/runtime/task/waker.rs:70-116。比如wake_by_ref最終調用raw.wake_by_ref() 📎 tokio/src/runtime/task/waker.rs:106-116。

wake_by_ref的語義是：把任務狀態從PENDING轉為SCHEDULED，如果轉換成功（即之前確實是 PENDING），就調用Schedule::schedule把任務重新入隊。

對於多線程調度器，schedule的實現在Handle::schedule_task 📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1353-1376：

pub(super) fn schedule_task(&self, task: Notified, is_yield: bool) {
    with_current(|maybe_cx| {
        if let Some(cx) = maybe_cx {
            if self.ptr_eq(&cx.worker.handle) {
                if let Some(core) = cx.core.borrow_mut().as_mut() {
                    self.schedule_local(core, task, is_yield);
                    return;
                }
            }
        }
        self.push_remote_task(task);
        self.notify_parked_remote();
    });
}

邏輯分兩支：

• 如果當前線程就是這個調度器的 worker，且持有 core，走schedule_local 📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1385-1417——放進 LIFO 槽位或本地隊列。
• 否則（從外部線程喚醒，或 core 被偷走），走push_remote_task推入全局注入隊列，並notify_parked_remote喚醒一個 parked worker📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1379-1383。

schedule_local內部又分兩支📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1385-1417：如果是yield或 LIFO 已禁用，推入run_queue尾部；否則放進lifo_slot，並把原來槽位裡的任務擠到隊列尾部。

sequenceDiagram
    participant Future as "Future::poll"
    participant Waker as "Waker(wake_by_ref)"
    participant RawTask as "RawTask::wake_by_ref"
    participant Handle as "Handle::schedule_task"
    participant Core as "Core(schedule_local)"
    participant Inject as "InjectQueue"
    participant Parker as "Unparker"

    Future->>Waker: "返回 Pending, 注册 waker"
    Note over Future: "事件就绪(如 epoll)"
    Waker->>RawTask: "raw.wake_by_ref()"
    RawTask->>RawTask: "state: PENDING -> SCHEDULED"
    RawTask->>Handle: "schedule(Notified)"
    alt 当前线程是同一 worker 且持有 core
        Handle->>Core: "schedule_local: 放入 lifo_slot"
    else 外部线程或 core 被偷走
        Handle->>Inject: "push_remote_task"
        Handle->>Parker: "notify_parked_remote().unpark()"
    end

park 與 unpark：狀態機與喚醒的原子性

worker 沒活幹時要 park，但 park/unpark 是最容易出競態的地方。Tokio 用AtomicUsize狀態機加Condvar兜底來解決。

Inner的字段📎 tokio/src/runtime/scheduler/multi_thread/park.rs:31-43：state: AtomicUsize、mutex: Mutex<()>、condvar: Condvar、shared: Arc<Shared>。狀態常量有四個📎 tokio/src/runtime/scheduler/multi_thread/park.rs:36-45：

• EMPTY = 0：未 park。
• PARKED_CONDVAR = 1：在 condvar 上 park。
• PARKED_DRIVER = 2：在 I/O driver 上 park。
• NOTIFIED = 3：已被喚醒。

這是一個顯式狀態機，我們用它畫狀態圖（這是本章唯一符合stateDiagram-v2准入條件的地方——源碼裡確實有這四個狀態常量）：

stateDiagram-v2
    [*] --> Empty
    Empty --> ParkedCondvar : "park_condvar() CAS(EMPTY->PARKED_CONDVAR)"
    Empty --> ParkedDriver : "park_driver() CAS(EMPTY->PARKED_DRIVER)"
    Empty --> Notified : "unpark() swap(NOTIFIED)"
    ParkedCondvar --> Empty : "condvar 唤醒后 CAS(NOTIFIED->EMPTY)"
    ParkedCondvar --> Empty : "超时 swap(EMPTY)"
    ParkedDriver --> Empty : "driver 返回后 swap(EMPTY)"
    Notified --> Empty : "park() CAS(NOTIFIED->EMPTY) 消费通知"
    Notified --> Notified : "再次 unpark() swap(NOTIFIED)"

unpark的實現📎 tokio/src/runtime/scheduler/multi_thread/park.rs:277-290用swap而非 CAS，源碼註釋解釋了原因📎 tokio/src/runtime/scheduler/multi_thread/park.rs:277-290：必須執行 release 操作讓 park 線程觀察到 unpark 之前的寫入，所以即使 state 已經是NOTIFIED也要寫一次。

park先嘗試消費已有的通知📎 tokio/src/runtime/scheduler/multi_thread/park.rs:132-149：如果 CASNOTIFIED -> EMPTY成功，說明之前已被喚醒，直接返回不阻塞。否則嘗試拿 driver 鎖，拿到就在 driver 上 park，拿不到就用 condvar 兜底📎 tokio/src/runtime/scheduler/multi_thread/park.rs:143-148。

park_condvar裡有個經典的雙重檢查📎 tokio/src/runtime/scheduler/multi_thread/park.rs:162-180：先 CASEMPTY -> PARKED_CONDVAR，如果失敗且是NOTIFIED，說明在設置狀態前就被喚醒了，此時必須swap(EMPTY)來同步 unpark 的寫入📎 tokio/src/runtime/scheduler/multi_thread/park.rs:167-177。註釋特別強調：即使知道是NOTIFIED也必須讀一次，因為 unpark 可能在我們讀NOTIFIED之後又被調用了一次。

unpark_condvar的註釋📎 tokio/src/runtime/scheduler/multi_thread/park.rs:292-307點出了 condvar 的經典陷阱：parked 線程設置PARKED狀態和真正wait之間有窗口期，如果在這期間 notify 會被忽略。解決方案是 park 線程此時持有mutex，unpark 線程先drop(self.mutex.lock())獲取鎖（從而等待 park 線程釋放），再notify_one。

設計思考：為什麼 LIFO 槽位是單槽而非隊列

MAX_LIFO_POLLS_PER_TICK = 3這個魔數📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:263-263也是經驗值。源碼註釋說「跑幾次 LIFO 槽位似乎足以受益於局部性，超過 3 次可能過度加權」。這防止了 A 喚醒 B、B 喚醒 A 的 ping-pong 場景把其他任務餓死。

另一個值得注意的設計是steal_work的「半數搜索」策略📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1158-1160：只有當不到一半的 worker 在搜索時，新 worker 才真正嘗試竊取。這避免了所有 worker 同時瘋狂竊取導致的 CAS 爭用。transition_to_searching通過idle.transition_worker_to_searching()來協調📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1197-1203。

竊取從隨機起點開始📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1172-1174，遍歷所有 remote，跳過自己📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1179-1182，調用steal_into嘗試竊取。全部失敗後回退到全局隊列📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1197-1203。

本章小結

worker 主循環Context::run是調度器的心臟：每輪 tick 後先取任務（LIFO 槽位 → 本地隊列 → 全局隊列），取到就run_task執行 poll，取不到就竊取，竊取失敗就 park。run_task內部的 LIFO 循環把「喚醒 → 入隊 → 再 poll」壓縮在同一個 budget 內，形成低延遲閉環。Waker是裸指標加靜態 vtable，wake_by_ref通過狀態轉換觸發schedule，根據當前線程是否是同一 worker 決定走本地隊列還是全局隊列。park/unpark用四狀態原子機加 condvar 兜底，解決了喚醒丟失的經典競態。

下一章我們將離開調度器，進入 I/O 世界：Reactor 如何把 epoll 事件翻譯成Waker喚醒，讓AsyncFd的Pending變成Ready。

本章思考與自測

上一章我們追蹤了 worker 執行緒的主迴圈：任務被 poll，返回 Pending 時把 Waker 存進某個地方，事件就緒後 Waker 被觸發，任務重新入佇列。但「某個地方」到底是哪裡？Waker 怎麼在 epoll 事件到來時被找回來？這正是 Reactor 要回答的問題。先建立直覺模型：把整個 I/O 就緒通知機制想像成一家餐廳的取餐叫號系統——顧客（任務）點完餐後不會站在窗口死等，而是拿一個震動器（Waker）回座位；後廚（核心 epoll）做好餐後，前台（Reactor）根據訂單號（Token）找到對應的震動器並按下按鈕。若沒有這套系統，每個任務只能輪詢 socket，CPU 會被燒光；或者用阻塞執行緒等待，一個連線一個執行緒，規模上不去。Tokio 的 Reactor 由三個檔案構成三層結構，職責嚴格分離：driver.rs 是事件迴圈本體，持有 mio::Poll，負責呼叫 poll() 阻塞等待核心事件，並把事件翻譯成對 ScheduledIo 的讀寫；registration.rs 是面向使用者的註冊句柄，TcpStream 內部持有的就是它，提供 poll_read_ready / poll_write_ready 等 API；scheduled_io.rs 是每個 fd 的狀態槽，儲存讀寫就緒位與 Waker 列表，是事件與任務之間的橋樑。模組組裝關係可參考 tokio/src/runtime/io/mod.rs:5-16：driver 匯出 Driver、Handle、ReadyEvent，registration 匯出 Registration，scheduled_io 匯出 ScheduledIo。下圖錨定了本章要追蹤的完整資料流：TcpStream → Registration → ScheduledIo → Handle/Driver → 核心 → 回到 ScheduledIo → Waker。接下來我們逐層拆解。

驅動層：Driver與Handle的職責切分

直覺模型

Driver是唯一擁有mio::Poll的實體，它只能在單個執行緒裡被&mut存取——這是事件迴圈的獨佔性要求。而Handle是可複製、可跨執行緒共享的註冊入口，任何執行緒想註冊新 fd 都透過它。若沒有這個切分，要麼把mio::Poll加鎖（每次註冊都競爭），要麼讓所有註冊都回到 driver 執行緒（引入跨執行緒訊息佇列）。Tokio 選擇讓Handle直接持有mio::Registry的複製，註冊操作可以並行進行，只有真正的事件等待才需要獨佔。

記憶體佈局與欄位

先看Driver的欄位📎 tokio/src/runtime/io/driver.rs:25-38：

• signal_ready: bool：Unix 訊號事件是否到達，用於 signal 驅動。
• events: mio::Events：主事件緩衝區，跨turn呼叫複用，避免每次分配。
• events_busy: Option<mio::Events>：非阻塞 poll 專用緩衝區，僅當max_io_events_per_busy_tick被設定時存在。
• poll: mio::Poll：核心事件佇列的封裝。

再看Handle 📎 tokio/src/runtime/io/driver.rs:41-75：

• registry: mio::Registry：mio::Poll::registry()的複製，用於register/deregister。
• registrations: RegistrationSet：所有活躍註冊的集合，負責分配Token與ScheduledIo。
• synced: Mutex<registration_set::Synced>：保護RegistrationSet的同步狀態。
• waker: mio::Waker：用於從任意執行緒喚醒阻塞在turn裡的 driver。
• metrics: IoDriverMetrics：統計 fd 數量、就緒事件數。

這裡有個關鍵設計：events_busy的存在📎 tokio/src/runtime/io/driver.rs:25-38是為了解決非阻塞 poll 會吞掉事件的問題。註解📎 tokio/src/runtime/io/driver.rs:189-190說得很清楚：非阻塞 poll 取走的事件如果留在主緩衝區裡，下次 poll 就看不到了；用獨立緩衝區，未處理的事件仍留在核心佇列，下次 poll 會重新返回。

Step-by-Step：一次turn的執行

turn是 driver 的核心函式📎 tokio/src/runtime/io/driver.rs:184-261。假設 worker 執行緒發現沒有任務可跑，呼叫park → turn(handle, None)阻塞等待：

第一步：斷言未 shutdown📎 tokio/src/runtime/io/driver.rs:185，並釋放待清理的註冊📎 tokio/src/runtime/io/driver.rs:187。release_pending_registrations檢查needs_release()，若有則呼叫registrations.release() 📎 tokio/src/runtime/io/driver.rs:336-340。

第二步：選擇事件緩衝區📎 tokio/src/runtime/io/driver.rs:191-194。若max_wait是零且events_busy存在，用 busy 緩衝區；否則用主緩衝區。

第三步：呼叫self.poll.poll(events, max_wait) 📎 tokio/src/runtime/io/driver.rs:198。這是真正阻塞在 epoll_wait 的地方。錯誤處理很克制：Interrupted直接忽略（訊號打斷是正常的）📎 tokio/src/runtime/io/driver.rs:200，WASI 下的InvalidInput也忽略📎 tokio/src/runtime/io/driver.rs:201-205，其他錯誤直接 panic📎 tokio/src/runtime/io/driver.rs:206。

第四步：遍歷事件📎 tokio/src/runtime/io/driver.rs:211-233。對每個event：

• 若token == TOKEN_WAKEUP（值為 0）📎 tokio/src/runtime/io/driver.rs:214，什麼都不做——這是unpark用來打斷阻塞的。
• 若token == TOKEN_SIGNAL（值為 1）📎 tokio/src/runtime/io/driver.rs:216，置signal_ready = true。
• 否則是一般 I/O 事件📎 tokio/src/runtime/io/driver.rs:218-231：把mio::Ready轉成 Tokio 的Ready，用EXPOSE_IO.from_exposed_addr(token.0)把 token 還原成*const ScheduledIo指標，然後set_readiness(Tick::Set, |curr| curr | ready)累積就緒位，再io.wake(ready)觸發對應方向的Waker。

這裡EXPOSE_IO是一個PtrExposeDomain<ScheduledIo> 📎 tokio/src/runtime/io/mod.rs:21-22，它把指標「暴露」成一個usize作為mio::Token。安全性註解📎 tokio/src/runtime/io/driver.rs:222-225說明了為什麼這個 unsafe 轉換是安全的：指標在從 mio 註銷且driver 不再並行 poll 之前不會被釋放，且 driver 持有Arc<ScheduledIo>的所有權。

第五步：處理 io_uring 完成佇列（僅 Linux + tokio_unstable）📎 tokio/src/runtime/io/driver.rs:235-258，包括 CQ 溢出時的 flush 迴圈。

第六步：累加 metrics📎 tokio/src/runtime/io/driver.rs:265-267。

flowchart TD
    start["turn(handle, max_wait)"] --> assert["debug_assert!(!is_shutdown)"]
    assert --> release["release_pending_registrations()"]
    release --> pick{"max_wait == 0<br/>且 events_busy 存在?"}
    pick -->|是| busy["events = events_busy"]
    pick -->|否| main["events = events"]
    busy --> poll["poll.poll(events, max_wait)"]
    main --> poll
    poll --> pollres{"poll 返回?"}
    pollres -->|"Ok / Interrupted"| iter["遍历 events.iter()"]
    pollres -->|"其他 Err"| panic["panic!(unexpected error)"]
    iter --> tok{"event.token()?"}
    tok -->|"TOKEN_WAKEUP"| skip["忽略，仅用于打断阻塞"]
    tok -->|"TOKEN_SIGNAL"| sig["signal_ready = true"]
    tok -->|"普通 fd token"| cast["EXPOSE_IO.from_exposed_addr(token.0)"]
    cast --> setr["io.set_readiness(Tick::Set, curr | ready)"]
    setr --> wake["io.wake(ready)"]
    wake --> iter
    skip --> iter
    sig --> iter
    iter --> uring["dispatch_completions() (io-uring)"]
    uring --> metrics["metrics.incr_ready_count_by(ready_count)"]

設計思考：為什麼Handle要持有mio::Waker

unpark 📎 tokio/src/runtime/io/driver.rs:280-283呼叫self.waker.wake()。這個mio::Waker在Driver::new時用TOKEN_WAKEUP註冊📎 tokio/src/runtime/io/driver.rs:124。當 driver 阻塞在poll.poll()裡時，另一個執行緒呼叫unpark會往 epoll 裡塞一個TOKEN_WAKEUP事件，poll立即返回，遍歷時看到這個 token 直接跳過📎 tokio/src/runtime/io/driver.rs:214-215。

另一個細節：deregister_source先呼叫self.registry.deregister(source) 📎 tokio/src/runtime/io/driver.rs:322，再清理registrations 📎 tokio/src/runtime/io/driver.rs:315-334。註解📎 tokio/src/runtime/io/driver.rs:320-321說「Cleanup ALWAYS happens」——即使 OS 層 deregister 失敗，也要清理內部狀態，最後才返回 OS 錯誤📎 tokio/src/runtime/io/driver.rs:336-340。這是典型的資源清理優先於錯誤傳播模式。

註冊層：Registration如何把Waker存進ScheduledIo

直覺模型

Registration是任務與 fd 之間的契約。它持有兩個東西：一個scheduler::Handle（用於在需要時存取 runtime），一個Arc<ScheduledIo>（fd 的狀態槽）。當任務呼叫poll_read_ready時，Registration把Waker交給ScheduledIo保管；當 driver 收到事件時，從ScheduledIo裡取出Waker喚醒。

記憶體佈局與欄位

Registration只有兩個欄位📎 tokio/src/runtime/io/registration.rs:46-54：

• handle: scheduler::Handle：runtime 句柄，註解📎 tokio/src/runtime/io/registration.rs:46-54說「TODO: this can probably be moved into ScheduledIo」，說明作者認為這個欄位位置可以最佳化。
• shared: Arc<ScheduledIo>：共享狀態，Arc保證 driver 和任務都能存取。

Step-by-Step：poll_read_ready的呼叫鏈

假設任務在TcpStream::poll_read裡發現 socket 沒資料，需要註冊讀興趣。呼叫鏈是TcpStream::poll_read_priv → PollEvented::poll_read → Registration::poll_read_io → poll_io → poll_ready。

poll_ready是核心📎 tokio/src/runtime/io/registration.rs:155-171：

第一步：trace_leaf() 📎 tokio/src/runtime/io/registration.rs:160，用於 tracing 埋點。

第二步：coop::poll_proceed(cx) 📎 tokio/src/runtime/io/registration.rs:155-171。這是第 12 章要講的協作式預算機制。如果預算耗盡，返回Pending並註冊一個特殊的Waker，讓任務在下一輪被重新調度。

第三步：self.shared.poll_readiness(cx, direction) 📎 tokio/src/runtime/io/registration.rs:155-171。這是真正與ScheduledIo互動的地方：檢查當前就緒位，若已就緒立即返回Ready；否則把cx.waker()存進ScheduledIo的對應方向槽位，返回Pending。

第四步：檢查ev.is_shutdown 📎 tokio/src/runtime/io/registration.rs:155-171。若 runtime 正在關閉，返回RUNTIME_SHUTTING_DOWN_ERROR。

第五步：coop.made_progress() 📎 tokio/src/runtime/io/registration.rs:169，標記預算消耗，返回就緒事件。

poll_io在poll_ready之上加了一層重試迴圈📎 tokio/src/runtime/io/registration.rs:173-192：

loop {
    let ev = ready!(self.poll_ready(cx, direction))?;
    match f() {
        Ok(ret) => return Poll::Ready(Ok(ret)),
        Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
            self.clear_readiness(ev);
        }
        Err(e) => return Poll::Ready(Err(e)),
    }
}

這裡體現了readiness 是提示而非保證的核心思想：poll_ready說可讀，但真正read()時可能返回WouldBlock（比如另一個執行緒搶先讀走了資料）。此時必須clear_readiness(ev) 📎 tokio/src/runtime/io/registration.rs:187清掉就緒位，然後迴圈重新等待。若不清，任務會陷入「以為可讀 → read 失敗 → 又以為可讀」的忙迴圈。

設計思考：try_io與async_io的分工

try_io 📎 tokio/src/runtime/io/registration.rs:194-213是同步版本：先ready_event(interest)檢查就緒位，若為空直接返回WouldBlock 📎 tokio/src/runtime/io/registration.rs:194-213；否則執行f()，若f()返回WouldBlock則清就緒位📎 tokio/src/runtime/io/registration.rs:207-210。它不註冊 Waker，適合try_read這類「試一下就走」的場景。

async_io 📎 tokio/src/runtime/io/registration.rs:225-245是異步版本：readiness(interest).await會註冊 Waker 並等待，然後執行f()，WouldBlock時清就緒位並迴圈。注意它在迴圈裡還呼叫了coop::poll_proceed 📎 tokio/src/runtime/io/registration.rs:233，防止在大量WouldBlock重試中耗盡預算。

生產踩坑：Drop裡的 Waker 清理

Registration::drop 📎 tokio/src/runtime/io/registration.rs:253-262呼叫self.shared.clear_wakers()。註釋📎 tokio/src/runtime/io/registration.rs:253-262解釋了原因：ScheduledIo裡存的Waker可能持有Arc<driver::Inner>，而driver::Inner又持有ScheduledIo，形成循環引用。清理 Waker 是打破循環的手段。但註釋也承認這是「imperfect solution」——如果Registration本身被存進了Waker，循環仍然存在。這是 tokio-rs/tokio#3481 討論的問題。

從TcpStream::read到Waker喚醒的完整鏈路

直覺模型

現在把三層串起來。使用者在TcpStream上呼叫.read().await，實際執行的是AsyncRead::poll_read → PollEvented::poll_read → Registration::poll_read_io。當資料沒到時，Waker被存進ScheduledIo；當 epoll 報告可讀時，driver 從ScheduledIo取出Waker並喚醒，任務被重新調度，再次 poll 時poll_readiness發現就緒位已置，直接返回Ready，read()成功。

Step-by-Step：一次完整的讀等待

階段一：註冊興趣。TcpStream::new 📎 tokio/src/net/tcp/stream.rs:166-169呼叫PollEvented::new(connected)，後者內部呼叫Registration::new_with_interest_and_handle 📎 tokio/src/runtime/io/registration.rs:73-81，進而handle.driver().io().add_source(io, interest) 📎 tokio/src/runtime/io/registration.rs:73-81。

add_source 📎 tokio/src/runtime/io/driver.rs:288-312做三件事：

1. registrations.allocate(&mut synced.lock())分配一個ScheduledIo，拿到token 📎 tokio/src/runtime/io/driver.rs:293-294。

2. self.registry.register(source, token, interest.to_mio())向核心註冊📎 tokio/src/runtime/io/driver.rs:298。若失敗，必須把剛分配的ScheduledIo從集合裡移除📎 tokio/src/runtime/io/driver.rs:300-303，否則洩漏。

3. metrics.incr_fd_count()計數📎 tokio/src/runtime/io/driver.rs:309。

階段二：等待就緒。任務 pollTcpStream::poll_read 📎 tokio/src/net/tcp/stream.rs:1492-1498 → poll_read_priv 📎 tokio/src/net/tcp/stream.rs:1451-1458 → PollEvented::poll_read → Registration::poll_read_io 📎 tokio/src/runtime/io/registration.rs:133-139 → poll_io → poll_ready → ScheduledIo::poll_readiness。此時若未就緒，Waker存入ScheduledIo的讀槽位，返回Pending。

階段三：事件到達。driver 的turn從poll.poll()拿到事件📎 tokio/src/runtime/io/driver.rs:198，遍歷時對每個 fd 事件執行io.set_readiness(Tick::Set, |curr| curr | ready)和io.wake(ready) 📎 tokio/src/runtime/io/driver.rs:228-229。wake內部取出對應方向的Waker並呼叫wake()。

階段四：任務重調度。Waker::wake()把任務重新入隊到 worker 的本地佇列（上一章講過）。worker 再次 poll 該任務，poll_readiness發現就緒位已置，返回Ready，read()成功。

sequenceDiagram
    participant Task as "任务 (worker 线程)"
    participant Reg as "Registration"
    participant SIO as "ScheduledIo"
    participant Drv as "Driver (I/O 线程)"
    participant OS as "epoll/kqueue"

    Task->>Reg: "poll_read_ready(cx)"
    Reg->>SIO: "poll_readiness(cx, Read)"
    SIO-->>Reg: "Pending (Waker 已存入读槽位)"
    Reg-->>Task: "Poll::Pending"
    Note over Task: 任务让出，worker 去跑别的任务
    Drv->>OS: "poll.poll(events, max_wait)"
    OS-->>Drv: "event(token=fd_ptr, READABLE)"
    Drv->>SIO: "set_readiness(Tick::Set, curr | READABLE)"
    Drv->>SIO: "wake(READABLE)"
    SIO->>Task: "Waker::wake() 重新入队"
    Note over Task: worker 再次 poll 该任务
    Task->>Reg: "poll_read_ready(cx)"
    Reg->>SIO: "poll_readiness(cx, Read)"
    SIO-->>Reg: "Ready(ReadyEvent{ready: READABLE})"
    Reg-->>Task: "Poll::Ready(Ok(ev))"
    Task->>Task: "read() 成功返回数据"

重要分支：assume_ready優化

TcpStream::new_accepted 📎 tokio/src/net/tcp/stream.rs:174-181是一個值得注意的優化。accept返回的 socket 天然可寫，且通常已經持有對端的第一批位元組。如果等 driver 的第一個事件，在高負載下這個事件可能排在所有已建立連線的事件後面，造成延遲。所以new_accepted直接呼叫assume_ready(Ready::READABLE | Ready::WRITABLE) 📎 tokio/src/net/tcp/stream.rs:174-181。

assume_ready的註釋📎 tokio/src/runtime/io/registration.rs:103-105說：「A wrong guess costs oneWouldBlock, which clears the readiness again.」——猜錯了代價只是一次WouldBlock，poll_io的迴圈會清掉就緒位並重新等待。這是一個樂觀猜測 + 快速糾錯的設計。

設計思考：為什麼 I/O 驅動與調度器解耦

1. 註冊無鎖化：Handle持有mio::Registry克隆，任何 worker 都能並發註冊新 fd，不需要回到 driver 執行緒。

2. 事件等待集中化：只有一個執行緒阻塞在epoll_wait，避免多執行緒同時 poll 同一個 epoll fd 的驚群問題。

3. 喚醒路徑短：driver 收到事件後直接操作ScheduledIo並呼叫Waker::wake()，wake()內部把任務推入 worker 佇列，不需要跨執行緒訊息傳遞。

代價是ScheduledIo需要處理並發存取（set_readiness和poll_readiness可能同時發生），這透過原子操作和內部鎖解決。

生產踩坑：is_shutdown與RUNTIME_SHUTTING_DOWN_ERROR

poll_ready檢查ev.is_shutdown 📎 tokio/src/runtime/io/registration.rs:155-171，若為真返回gone() 📎 tokio/src/runtime/io/registration.rs:265-267，即RUNTIME_SHUTTING_DOWN_ERROR。

另一個坑是deregister_source的unpark 📎 tokio/src/runtime/io/driver.rs:328。如果 driver 正阻塞在poll裡，且此時最後一個Registration被 drop，unpark會喚醒 driver。但如果 driver 不在阻塞狀態（比如正在處理其他事件），unpark只是讓下一次turn立即返回📎 tokio/src/runtime/io/driver.rs:280-283。這個語意在Handle::unpark的文件註釋裡有說明。

設計思考：Reactor 的三個關鍵權衡

權衡一：Token用指標而非索引。EXPOSE_IO.from_exposed_addr(token.0) 📎 tokio/src/runtime/io/driver.rs:220把mio::Token直接當作*const ScheduledIo的地址。這避免了維護一個Token → ScheduledIo的映射表，查找是 O(1) 且無鎖。代價是安全性依賴嚴格的生命週期管理：指標必須在註銷且 driver 不再 poll 之後才能釋放📎 tokio/src/runtime/io/driver.rs:222-225。

權衡二：讀寫雙 Waker 槽位。Registration文件📎 tokio/src/runtime/io/registration.rs:24-26說「A registration instance represents two separate readiness streams」——讀和寫各有一個獨立的Waker槽位。這允許同一個 socket 的讀任務和寫任務分別註冊，互不干擾。但poll_read_ready的註釋📎 tokio/src/net/tcp/stream.rs:549-552提醒：多次呼叫poll_read_ready/poll_read/poll_peek只有最後一次的Waker會被保留——讀方向只有一個槽位。

權衡三：events_busy的獨立緩衝區。測試📎 tokio/src/runtime/io/driver.rs:364-386驗證了這個行為：Driver::new(16, Some(2))建立 busy 容量為 2 的 driver，註冊 5 個可讀 source 後，非阻塞turn只取 2 個事件📎 tokio/src/runtime/io/driver.rs:375-376，剩餘 3 個留在核心佇列，下次阻塞turn取到📎 tokio/src/runtime/io/driver.rs:379-380。這防止了非阻塞 poll 一次性吞掉所有事件導致後續 poll 飢餓。

本章小結

本章追蹤了TcpStream::read背後的完整 Reactor 鏈路：

• 驅動層：Driver獨佔mio::Poll，turn阻塞等待事件，用EXPOSE_IO把Token還原為ScheduledIo指標，呼叫set_readiness + wake觸發Waker。Handle提供可跨執行緒的註冊入口，unpark用於打斷阻塞。
• 註冊層：Registration持有Arc<ScheduledIo>，poll_ready檢查就緒位或存入Waker，poll_io用WouldBlock重試迴圈處理假陽性，try_io/async_io分別服務同步和非同步場景。
• 狀態層：ScheduledIo是 fd 的狀態槽，儲存讀寫就緒位和雙Waker槽位，是事件與任務之間唯一的橋樑。

本章思考與自測

上一章我們追蹤了 TcpStream::read 的完整鏈路，看到 ScheduledIo 如何把 epoll 的 fd 就緒事件翻譯成 Waker 喚醒。但非同步執行時還需要處理另一類「就緒」：一個 sleep(100ms) 的 Future，在 100ms 後必須被喚醒。這類事件不來自核心 fd，而來自「時間本身」。Tokio 的設計選擇是把時間也當作一種 I/O 事件：Driver 結構體裡只有一個欄位 park: IoStack，它複用了 I/O driver 的 park/unpark 機制。當時間輪算出「下一次到期時刻」時，driver 就呼叫 park_timeout 讓執行緒睡到那個時刻；被喚醒後再從時間輪裡取出到期條目、觸發它們的 Waker。這樣，排程器只需要一個統一的 park 入口，就能同時等待「fd 就緒」和「定時器到期」兩類事件。本章要回答三個問題：定時器如何被插入時間輪？時間輪如何按到期時間分級？driver 如何計算下一次 park 的超時並觸發到期任務？

一、時間輪：六層 64 槽的雜湊分級結構

直覺模型

想像一個機械鐘錶：秒針轉一圈帶動分針，分針轉一圈帶動時針。如果只有一根秒針，要表示「12 天後」就得數 100 萬格；而分層之後，秒針只管 64 秒內的精度，分針管 64 分鐘，時針管 64 小時——每一層只需 64 個槽位，就能覆蓋到 2 年之後。

若沒有分層，插入一個遠期定時器要麼需要 O(N) 遍歷，要麼需要巨大的陣列。時間輪用「按到期時間分級」把插入和觸發都壓到近似 O(1)。

記憶體佈局與欄位

Wheel的核心欄位只有三個📎 tokio/src/runtime/time/wheel/mod.rs:22-40：

pub(crate) struct Wheel {
    elapsed: u64,                          // 自 wheel 创建以来经过的毫秒数
    levels: Box<[Level; NUM_LEVELS]>,      // 6 层，每层 64 槽
    pending: LinkedList<TimerShared>,      // 已到期、待触发的条目
}

NUM_LEVELS = 6，BITS_PER_LEVEL = 6（即每層 64 槽）📎 tokio/src/runtime/time/wheel/mod.rs:45-47。MAX_DURATION = 1 << (6 * 6) = 1 << 36毫秒，約 2 年📎 tokio/src/runtime/time/wheel/mod.rs:50。

六層的粒度按文件註解是📎 tokio/src/runtime/time/wheel/mod.rs:22-40：

pending是一個侵入式鏈結串列（LinkedList<TimerShared>），存放已經從輪中取出、等待觸發 Waker 的條目。注意它是LinkedList而非Vec：條目本身內嵌在TimerShared裡，插入/移除不需要分配。

場景驅動：插入一個 100ms 的 sleep

當sleep(100ms)首次被 poll 時，Sleep::poll_elapsed會建構Timer::new並呼叫init 📎 tokio/src/time/sleep.rs:436-440。init最終呼叫Handle::reregister，進而呼叫Wheel::insert。

insert的第一步是檢查是否已過期📎 tokio/src/runtime/time/wheel/mod.rs:90-98：

let when = unsafe { item.sync_when() };
if when <= self.elapsed {
    return Err((item, InsertError::Elapsed));
}

如果when已經落在elapsed之前（比如 deadline 已過），直接返回Elapsed，呼叫方會立即觸發該定時器。

否則計算該條目應放入哪一層📎 tokio/src/runtime/time/wheel/mod.rs:90-114：

let level = self.level_for(when);
unsafe { self.levels[level].add_entry(item); }

level_for是分級演算法的核心📎 tokio/src/runtime/time/wheel/mod.rs:276-289：

fn level_for(elapsed: u64, when: u64) -> usize {
    const SLOT_MASK: u64 = (1 << BITS_PER_LEVEL) - 1;
    let masked = elapsed ^ when | SLOT_MASK;
    if masked >= MAX_DURATION {
        return NUM_LEVELS - 1;
    }
    masked.ilog2() as usize / BITS_PER_LEVEL
}

這裡用elapsed ^ when而非when - elapsed，是一個精妙的技巧：XOR 的最高有效位反映了「兩個時間戳從哪一位開始不同」，也就是「需要多粗的粒度才能區分它們」。| SLOT_MASK把低 6 位強制置 1，避免ilog2落在同一槽內時算出過小的層。ilog2() / 6把位寬映射到層號。如果 XOR 結果超過MAX_DURATION（即超過 2 年），就強制塞進最高層——這就是「fudge the timer into the top level」。

對於 100ms 的 sleep，假設elapsed接近 0，when ≈ 100，elapsed ^ when ≈ 100，ilog2(100) = 6，6 / 6 = 1，所以落在第 1 層（64ms 粒度）。這意味著它會在第 1 層的某個槽裡等待，直到時間推進到該槽的邊界時才會被下沉到第 0 層。

分級下沉：process_expiration

當poll(now)推進時間時，Wheel::poll會迴圈呼叫next_expiration和process_expiration 📎 tokio/src/runtime/time/wheel/mod.rs:142-166：

pub(crate) fn poll(&mut self, now: u64) -> Option<TimerHandle> {
    loop {
        if let Some(handle) = self.pending.pop_back() {
            return Some(handle);
        }
        match self.next_expiration() {
            Some(ref expiration) if expiration.deadline <= now => {
                self.process_expiration(expiration);
                self.set_elapsed(expiration.deadline);
            }
            _ => {
                self.set_elapsed(now);
                break;
            }
        }
    }
    self.pending.pop_back()
}

process_expiration負責把某一層的到期條目「下沉」到下一層，或者（在第 0 層）標記為 pending📎 tokio/src/runtime/time/wheel/mod.rs:218-251：

let mut entries = self.take_entries(expiration);
while let Some(item) = entries.pop_back() {
    match unsafe { item.mark_pending(expiration.deadline) } {
        Ok(()) => {
            self.pending.push_front(item);   // 真正到期
        }
        Err(expiration_tick) => {
            let level = level_for(expiration.deadline, expiration_tick);
            unsafe { self.levels[level].add_entry(item); }  // 下沉到更低层
        }
    }
}

mark_pending是關鍵：它檢查條目的實際 deadline 是否已經到達。如果到達，返回Ok(())，條目進入pending鏈結串列；如果還沒到（只是所在槽的邊界到了），返回Err(expiration_tick)，條目被重新插入到更細的層。

注意註解裡強調的一點📎 tokio/src/runtime/time/wheel/mod.rs:219-228：必須先把整個槽的條目全部取出再處理，因為某些條目可能被重新插入到同一個槽（當插入時間超過MAX_DURATION時會發生環繞）。如果邊取邊插，可能陷入無限迴圈。

下一到期時刻的計算

next_expiration從低層到高層掃描，返回第一個非空的到期點📎 tokio/src/runtime/time/wheel/mod.rs:169-191：

fn next_expiration(&self) -> Option<Expiration> {
    if !self.pending.is_empty() {
        return Some(Expiration { level: 0, slot: 0, deadline: self.elapsed });
    }
    for (level_num, level) in self.levels.iter().enumerate() {
        if let Some(expiration) = level.next_expiration(self.elapsed) {
            debug_assert!(self.no_expirations_before(level_num + 1, expiration.deadline));
            return Some(expiration);
        }
    }
    None
}

如果pending非空，說明有已到期條目待觸發，立即返回當前elapsed作為 deadline（這樣 driver 會以 0 超時 park，馬上回來處理）。否則逐層掃描，返回第一個有內容的槽的 deadline。debug_assert驗證了一個不變量：更高層不可能有比當前層更早的到期點。

flowchart TD
    start["Wheel::poll(now)"] --> check_pending{"pending 非空?"}
    check_pending -->|是| pop["pop_back 返回 TimerHandle"]
    check_pending -->|否| next_exp{"next_expiration() 有到期点?"}
    next_exp -->|无| set_elapsed["set_elapsed(now) 后 break"]
    next_exp -->|有| cmp{"expiration.deadline <= now?"}
    cmp -->|否| set_elapsed
    cmp -->|是| proc["process_expiration(expiration)"]
    proc --> take["take_entries 取出整槽"]
    take --> mark{"item.mark_pending()"}
    mark -->|Ok 已到期| push_pending["pending.push_front(item)"]
    mark -->|Err 未到期| reinsert["level_for 后 add_entry 下沉"]
    push_pending --> set_elapsed2["set_elapsed(expiration.deadline)"]
    reinsert --> set_elapsed2
    set_elapsed2 --> check_pending
    set_elapsed --> pop2["pending.pop_back() 返回"]

---

二、Driver 的 park 迴圈：把時間輪接到 I/O 堆疊上

直覺模型

時間輪本身不會「自己走」。它需要一個外部迴圈反覆問它：「下一次到期是什麼時候？」然後睡到那個時刻，醒來後再推進時間。這個迴圈就是Driver::park_internal。它把「時間輪的下一次到期」翻譯成一個park_timeout的時長，交給底層的 I/O 堆疊去睡。

若沒有這個迴圈，定時器永遠不會被觸發——時間輪只是靜態資料結構，需要有人「撥動」它。

資料結構：Driver 與 InnerState

Driver只有一個欄位park: IoStack 📎 tokio/src/runtime/time/mod.rs:90-93。真正的狀態在Handle裡，透過Inner列舉區分傳統實作和實驗性實作📎 tokio/src/runtime/time/mod.rs:95-127。傳統實作的InnerState包含兩個欄位📎 tokio/src/runtime/time/mod.rs:130-136：

struct InnerState {
    next_wake: Option<NonZeroU64>,   // 承诺的最早唤醒时刻
    wheel: wheel::Wheel,
}

next_wake用NonZeroU64而非Option<u64>的嵌套，是為了利用 niche 優化——Option<NonZeroU64>和u64同大小。它記錄「driver 承諾在哪個 tick 之前會醒來」，用於reregister時判斷是否需要unpark。

is_shutdown是獨立的AtomicBool，註解解釋了為什麼把它從 Mutex 裡拆出來📎 tokio/src/runtime/time/mod.rs:90-93：Handle需要在不鎖 mutex 的情況下檢查is_shutdown。這是一個典型的「讀多寫少」優化——shutdown 只發生一次，但檢查可能頻繁。

場景驅動：一次 park 的完整流程

park_internal是核心📎 tokio/src/runtime/time/mod.rs:213-256：

fn park_internal(&mut self, rt_handle: &driver::Handle, limit: Option<Duration>) {
    let handle = rt_handle.time();
    let mut lock = handle.inner.lock();
    assert!(!handle.is_shutdown());

    let next_wake = lock.wheel.next_expiration_time();
    lock.next_wake = next_wake.map(|t| NonZeroU64::new(t).unwrap_or_else(|| NonZeroU64::new(1).unwrap()));
    drop(lock);

    match next_wake {
        Some(when) => {
            let now = handle.time_source.now(rt_handle.clock());
            let mut duration = handle.time_source.tick_to_duration(when.saturating_sub(now));
            if duration > Duration::from_millis(0) {
                if let Some(limit) = limit {
                    duration = std::cmp::min(limit, duration);
                }
                self.park_thread_timeout(rt_handle, duration);
            } else {
                self.park.park_timeout(rt_handle, Duration::from_secs(0));
            }
        }
        None => {
            if let Some(duration) = limit {
                self.park_thread_timeout(rt_handle, duration);
            } else {
                self.park.park(rt_handle);
            }
        }
    }

    handle.process(rt_handle.clock());
}

分步解析：

1. 取鎖、讀下一次到期：lock.wheel.next_expiration_time()返回Option<u64>，即下一個到期 tick。同時把它寫入lock.next_wake，供reregister判斷是否需要 unpark。

2. 釋放鎖：drop(lock)必須在 park 之前，否則 park 期間其他執行緒無法插入定時器。

3. 計算 park 時長：when.saturating_sub(now)得到剩餘 tick 數，tick_to_duration轉成Duration。註解指出這裡實際上向上取整到 1ms📎 tokio/src/runtime/time/mod.rs:228-230，避免微秒級 sleep 被 OS 當作零長度。

4. 處理 limit：如果呼叫方傳了limit（比如park_timeout的顯式超時），取min(limit, duration)，保證不會睡過頭。

5. 特殊情況：如果duration == 0（已到期），用park_timeout(0)立即返回，不真正睡。

6. 無定時器時：如果next_wake為None，有limit就park_thread_timeout(limit)，否則無限park。

7. 喚醒後處理：handle.process(clock)推進時間輪並觸發到期條目。

process_at_time：觸發到期條目

process呼叫process_at_time 📎 tokio/src/runtime/time/mod.rs:296-337：

pub(self) fn process_at_time(&self, mut now: u64) {
    let mut waker_list = WakeList::new();
    let mut lock = self.inner.lock();

    if now < lock.wheel.elapsed() {
        // 时间倒流保护
        now = lock.wheel.elapsed();
    }

    while let Some(entry) = lock.wheel.poll(now) {
        debug_assert!(unsafe { entry.is_pending() });
        if let Some(waker) = unsafe { entry.fire(Ok(())) } {
            waker_list.push(waker);
            if !waker_list.can_push() {
                drop(lock);
                waker_list.wake_all();
                lock = self.inner.lock();
            }
        }
    }

    lock.next_wake = lock.wheel.poll_at()
        .map(|t| NonZeroU64::new(t).unwrap_or_else(|| NonZeroU64::new(1).unwrap()));
    drop(lock);
    waker_list.wake_all();
}

幾個關鍵點：

• 時間倒流保護 📎 tokio/src/runtime/time/mod.rs:301-309：如果now < wheel.elapsed()，說明時鐘倒退。註解指出這通常不該發生（Rust 保證Instant單調），但在 Windows 宿主上的 Linux VM 裡會發生，因為 std 錯誤地信任硬體時鐘單調。保護方式是把now鉗到elapsed。
• 批量喚醒：WakeList收集 Waker，當它滿了（!can_push()）就臨時釋放鎖、喚醒一批、再重新加鎖。註解強調這是為了避免死鎖📎 tokio/src/runtime/time/mod.rs:319。如果持有鎖時呼叫 Waker，而 Waker 又試圖操作時間輪（比如重新註冊定時器），就會死鎖。
• 更新 next_wake：處理完後重新計算poll_at()，更新next_wake。

reregister：重新註冊與 unpark

當Sleep::reset被呼叫時，定時器需要重新註冊。reregister處理這個場景📎 tokio/src/runtime/time/mod.rs:398-450：

pub(self) unsafe fn reregister(&self, unpark: &IoHandle, new_tick: u64, entry: NonNull<TimerShared>) {
    let waker = unsafe {
        let mut lock = self.inner.lock();
        if unsafe { entry.as_ref().might_be_registered() } {
            lock.wheel.remove(entry);
        }
        let entry = entry.as_ref().handle();
        if self.is_shutdown() {
            unsafe { entry.fire(Err(crate::time::error::Error::shutdown())) }
        } else {
            entry.set_expiration(new_tick);
            match unsafe { lock.wheel.insert(entry) } {
                Ok(when) => {
                    if lock.next_wake.is_none_or(|next_wake| when < next_wake.get()) {
                        unpark.unpark();
                    }
                    None
                }
                Err((entry, crate::time::error::InsertError::Elapsed)) => unsafe {
                    entry.fire(Ok(()))
                },
            }
        }
    };
    if let Some(waker) = waker {
        waker.wake();
    }
}

關鍵邏輯：插入成功後，如果新到期時刻比next_wake更早，就呼叫unpark.unpark()喚醒 driver。這是因為 driver 可能正睡在一個更晚的時刻，需要被提前叫醒以重新計算 park 時長。

注意unpark是在持有鎖時呼叫的，而waker.wake()是在釋放鎖後呼叫的。註解解釋📎 tokio/src/runtime/time/mod.rs:441：必須在呼叫 Waker 前釋放鎖以避免死鎖。但unpark不同——它只是往 epoll 塞一個事件，不會回呼使用者程式碼，所以持鎖呼叫是安全的。

sequenceDiagram
    participant Sleep as Sleep::poll
    participant Handle as time::Handle
    participant Wheel as Wheel
    participant Driver as Driver::park_internal
    participant IoStack as IoStack

    Sleep->>Handle: reregister(unpark, new_tick, entry)
    Handle->>Handle: lock.inner.lock()
    Handle->>Wheel: wheel.remove(entry) [若已注册]
    Handle->>Wheel: wheel.insert(entry)
    Wheel-->>Handle: Ok(when)
    alt when < next_wake
        Handle->>IoStack: unpark.unpark()
    end
    Handle->>Handle: drop(lock)
    Handle-->>Sleep: 返回 waker (若有)

    Note over Driver: 另一线程
    Driver->>Handle: lock.inner.lock()
    Driver->>Wheel: next_expiration_time()
    Wheel-->>Driver: Some(when)
    Driver->>Driver: drop(lock)
    Driver->>IoStack: park_timeout(duration)
    IoStack-->>Driver: 被 unpark 或超时
    Driver->>Handle: process(clock)
    Handle->>Wheel: poll(now)
    Wheel-->>Handle: TimerHandle
    Handle->>Sleep: waker.wake()

---

三、Sleep 與 Timeout：使用者可見的 API 層

直覺模型

Sleep是使用者直接.await的 Future，Timeout是包裹另一個 Future 的適配器。它們本身不管理時間輪，只是把「deadline」翻譯成 tick，委託給Timer和Handle。

Sleep 的記憶體佈局

Sleep用pin_project!巨集定義📎 tokio/src/time/sleep.rs:221-227：

pub struct Sleep {
    deadline: Instant,
    driver: scheduler::Handle,
    inner: Inner,
    #[pin]
    timer: Option<Timer>,
}

timer是Option<Timer>且帶#[pin]：首次 poll 前是None，首次 poll 時才建立Timer並註冊。這種「惰性初始化」避免了在sleep()呼叫時就存取執行時——sleep()可以在執行時外呼叫，只要在.await時才真正註冊。

PinnedDrop實作確保 drop 時取消定時器📎 tokio/src/time/sleep.rs:230-235：

impl PinnedDrop for Sleep {
    fn drop(this: Pin<&mut Self>) {
        let this = this.project();
        if let Some(timer) = this.timer.as_pin_mut() {
            timer.cancel(this.driver);
        }
    }
}

poll_elapsed 的完整流程

poll_elapsed是Sleep的核心📎 tokio/src/time/sleep.rs:396-454：

fn poll_elapsed(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Result<(), Error>> {
    ready!(crate::trace::trace_leaf());
    let mut this = self.project();

    // coop 预算
    let coop = ready!(crate::task::coop::poll_proceed(cx));

    let handle = this.driver;
    let timer = match this.timer.as_mut().as_pin_mut() {
        Some(timer) => timer,
        None => {
            let time_source = handle.driver().time().time_source();
            let deadline = time_source.deadline_to_tick(*this.deadline);
            let timer = Timer::new(handle, deadline);
            this.timer.set(Some(timer));
            let mut timer = this.timer.as_pin_mut().unwrap();
            timer.as_mut().init(handle, deadline);
            timer
        }
    };

    let result = timer.poll_elapsed(cx, handle).map(move |r| {
        coop.made_progress();
        r
    });
    result
}

分步：

1. coop 預算檢查：poll_proceed(cx)消耗一次協作預算。如果預算耗盡，返回Pending並讓出執行權。這是 Tokio 防止單個任務餓死其他任務的機制。

2. 惰性建立 Timer：如果timer是None，把deadline轉成 tick，建立Timer並呼叫init註冊到時間輪。

3. 委託給 Timer::poll_elapsed：實際的到期檢查由Timer完成。

4. 成功後標記進度：coop.made_progress()表示這次 poll 有實際進展。

Timeout 的 poll：先 poll 值，再 poll 延遲

Timeout的 poll 順序很關鍵📎 tokio/src/time/timeout.rs:210-224：

fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
    let me = self.project();
    let had_budget_before = coop::has_budget_remaining();

    // 先 poll 被包裹的 future
    if let Poll::Ready(v) = me.value.poll(cx) {
        return Poll::Ready(Ok(v));
    }

    match me.delay.as_pin_mut() {
        Some(delay) => poll_delay(had_budget_before, delay, cx).map(Err),
        None => Poll::Pending,
    }
}

註解明確指出📎 tokio/src/time/timeout.rs:24-26：future 先被 poll，然後才檢查超時。所以如果 future 不 yield 就完成，它可能在超過 timeout 後仍返回Ok。這是設計選擇，不是 bug。

poll_delay處理一個微妙的場景📎 tokio/src/time/timeout.rs:229-251：

fn poll_delay(had_budget_before: bool, delay: Pin<&mut Sleep>, cx: &mut task::Context<'_>) -> Poll<Elapsed> {
    let delay_poll = || match delay.poll(cx) {
        Poll::Ready(()) => Poll::Ready(Elapsed::new()),
        Poll::Pending => Poll::Pending,
    };

    let has_budget_now = coop::has_budget_remaining();

    if let (true, false) = (had_budget_before, has_budget_now) {
        // 如果预算是被底层 future 耗尽的，用无约束预算 poll delay
        coop::with_unconstrained(delay_poll)
    } else {
        delay_poll()
    }
}

邏輯：如果進入poll時還有預算，但 poll 完 value 後預算耗盡了，說明是 value 消耗了預算。此時如果用受限預算 poll delay，delay 可能立即返回Pending，導致永遠無法判斷超時是否到達。所以用with_unconstrained臨時解除預算限制。註解稱之為「pathological cases」📎 tokio/src/time/timeout.rs:243-246。

timeout 的 deadline 溢出處理

timeout函式用checked_add處理溢出📎 tokio/src/time/timeout.rs:86-99：

Timeout {
    value: future.into_future(),
    delay: match Instant::now().checked_add(duration) {
        Some(deadline) => Some(Sleep::new_timeout(deadline, trace::caller_location())),
        None => None,
    },
}

如果Instant::now() + duration溢出（duration 極大），delay為None，poll 時直接返回Poll::Pending 📎 tokio/src/time/timeout.rs:222。這相當於「永不超時」，是合理的降級行為。

---

設計思考與生產踩坑

為什麼用 XOR 而非減法計算層級？ elapsed ^ when的最高有效位直接反映「兩個時間戳從哪一位開始不同」，這正是「需要多粗的粒度」的度量。減法when - elapsed在elapsed接近when時高位全為 0，ilog2會算出過小的層。XOR 天然處理了環繞場景。

時間倒流保護的必要性 📎 tokio/src/runtime/time/mod.rs:301-309：Rust 保證Instant單調，但底層 OS 可能不保證。在 Windows 宿主上的 Linux VM 裡，std 信任硬體時鐘導致Instant倒退。Tokio 用now = lock.wheel.elapsed()鉗制，避免set_elapsed的 assert 失敗。

批量喚醒與死鎖 📎 tokio/src/runtime/time/mod.rs:319：持有時間輪鎖時呼叫 Waker 是危險的——Waker 可能觸發任務重新 poll，進而呼叫Sleep::reset，試圖再次獲取時間輪鎖，造成死鎖。WakeList的批量機制在鎖滿時臨時釋放鎖，是標準的「鎖外回調」模式。

next_wake的 niche 優化 📎 tokio/src/runtime/time/mod.rs:130-136：Option<NonZeroU64>與u64同大小，因為 0 被用作None的 niche。但 tick 0 是合法值，所以程式碼用NonZeroU64::new(t).unwrap_or_else(|| NonZeroU64::new(1).unwrap())把 0 映射到 1📎 tokio/src/runtime/time/mod.rs:221。這是一個微妙的邊界處理：tick 0 被當作 tick 1，最多導致 1ms 的額外喚醒。

process_expiration的「先取後處理」 📎 tokio/src/runtime/time/wheel/mod.rs:219-228：必須先把整槽條目取出再處理，因為超過MAX_DURATION的條目會環繞並重新插入同一槽。如果邊取邊插，會無限迴圈。

Timeout的 poll 順序陷阱 📎 tokio/src/time/timeout.rs:24-26：future 先 poll，超時後檢查。如果 future 是 CPU 密集且不 yield，它可能超過 timeout 仍返回Ok。生產環境中不要依賴timeout來強制中斷不合作的 future。

---

本章小結

本章拆解了 Tokio 時間驅動的三層結構：

1. 時間輪（Wheel）：六層 64 槽的雜湊分級結構，用elapsed ^ when的位寬決定條目層級，插入和觸發近似 O(1)。pending鏈結串列存放已到期條目，process_expiration負責逐層下沉。

2. Driver（Driver::park_internal）：把時間輪的next_expiration_time翻譯成park_timeout時長，複用 I/O 堆疊的 park/unpark。process_at_time在喚醒後推進時間輪、批量觸發 Waker，並處理時間倒流和死鎖防護。

3. 使用者 API（Sleep / Timeout）：Sleep惰性建立Timer並註冊，Timeout先 poll value 再 poll delay，用with_unconstrained處理預算耗盡場景。

核心設計是「時間也是一種 I/O 事件」：driver 只有一個 park 入口，同時等待 fd 就緒和定時器到期。next_wake記錄承諾的喚醒時刻，reregister在插入更早的定時器時unpark喚醒 driver 重新計算。

下一章我們將進入同步原語：Mutex、Semaphore與通道如何實作非同步等待。你會看到它們如何複用本章的 Waker 機制，以及「許可計數」與「等待佇列」如何協作。

本章思考與自測

至此，我們已經看清時間驅動如何復用 I/O driver 的 park/unpark 機制，讓定時器與 fd 就緒共享同一個等待入口。時間輪的分級、到期計算與 Waker 觸發，構成了非同步執行時處理「時間就緒」的完整閉環。但非同步等待不止於 I/O 與時間——當多個任務競爭同一把鎖、或透過通道傳遞訊息時，Waker 又該被存放到哪裡？下一章我們將進入 tokio::sync 家族，看看 Mutex、Semaphore 與各類通道如何在「等待者佇列 + Waker 喚醒」上做出不同取捨。

上一章揭示了時間如何被抽象為一種 I/O 事件，讓定時器與 fd 就緒共享同一個 park/unpark 等待入口。然而，當多個任務競爭同一把鎖或透過通道傳遞訊息時，等待的物件不再是 fd 或時鐘，而是另一個任務的狀態變化。本章進入 tokio::sync 家族，探明一次 lock().await 或 recv().await 在阻塞時究竟把 Waker 存到了哪裡，被喚醒時又如何被重新排程。

為什麼非同步 Mutex 不能復用 std 的實現

直覺模型：從「佔著茅坑」到「讓出座位」

std::sync::Mutex的lock()在鎖被佔用時會阻塞當前執行緒——執行緒被作業系統掛起，直到鎖釋放。這在非同步執行時裡是災難性的：一個 worker 執行緒可能同時驅動成百上千個任務，如果它因為等一把鎖而阻塞，它承載的所有其他任務全部停擺。非同步 Mutex 的核心訴求是：等鎖時讓出執行緒，把「我在等這把鎖」這件事登記到一個佇列裡，然後返回Pending，讓執行器去跑別的任務。

Tokio 的Mutex沒有自己實現等待佇列，而是完全建立在號誌量之上。

資料結構與記憶體佈局

Mutex<T>的欄位極簡：

📎 tokio/src/sync/mutex.rs:133-138

pub struct Mutex<T: ?Sized> {
    #[cfg(all(tokio_unstable, feature = "tracing"))]
    resource_span: tracing::Span,
    s: semaphore::Semaphore,
    c: UnsafeCell<T>,
}

三個欄位各司其職：s是一個許可數為 1 的號誌量，c是UnsafeCell<T>包裹的受保護資料。注意這裡的semaphore是batch_semaphore的別名📎 tokio/src/sync/mutex.rs:3-3，也就是底層實作，而非sync::Semaphore那層公開封裝。

MutexGuard<'a, T>則只持有一個對Mutex的參考：

📎 tokio/src/sync/mutex.rs:151-157

pub struct MutexGuard<'a, T: ?Sized> {
    #[cfg(all(tokio_unstable, feature = "tracing"))]
    resource_span: tracing::Span,
    lock: &'a Mutex<T>,
}

這裡有個關鍵設計：MutexGuard 不持有號誌量許可物件，只持有&Mutex。釋放鎖的動作發生在Drop裡，直接呼叫self.lock.s.release(1) 📎 tokio/src/sync/mutex.rs:959-961。這與SemaphorePermit持有permits: usize計數、在 Drop 時歸還不同——Mutex 的許可數恆為 1，不需要計數。

Send/Sync的邊界值得單獨看：

📎 tokio/src/sync/mutex.rs:258-259

unsafe impl<T> Send for Mutex<T> where T: ?Sized + Send {}
unsafe impl<T> Sync for Mutex<T> where T: ?Sized + Send {}

Sync只要求T: Send而非T: Sync——這是合理的，因為互斥存取保證了同一時刻只有一個執行緒能觸碰T，跨執行緒傳遞T的所有權（Send）就夠了，不需要T本身可被共享（Sync）。這正是Mutex<T>能把非Sync的T變成Sync的原因。

Step-by-Step：一次lock().await的完整旅程

代入場景：任務 A 呼叫mutex.lock().await，此時鎖空閒。

第一步，lock()建構一個 async 區塊，內部先self.acquire().await，成功後建構MutexGuard 📎 tokio/src/sync/mutex.rs:434-443。

第二步，acquire()直接委託給號誌量：

📎 tokio/src/sync/mutex.rs:655-663

async fn acquire(&self) {
    crate::trace::async_trace_leaf().await;
    self.s.acquire(1).await.unwrap_or_else(|_| {
        unreachable!()
    });
}

unwrap_or_else(|_| unreachable!())這行註解道出了設計約束：Mutex 從不顯式 close 號誌量，且獨佔持有它，所以acquire永遠不會回傳Err。這是把「號誌量關閉」這一錯誤路徑在型別層面排除掉。

第三步，若鎖被佔用，s.acquire(1)回傳Pending，當前任務的 Waker 被登記進號誌量的等待佇列。Waker 存在哪裡？答案在batch_semaphore的等待佇列裡（本章原始碼材料未展開該檔案，但其角色是：每個等待者持有一個 Waker，按 FIFO 排隊）。

第四步，持有鎖的任務 B 釋放鎖時，MutexGuard::drop呼叫s.release(1) 📎 tokio/src/sync/mutex.rs:965-975，號誌量把許可交給隊首等待者並喚醒其 Waker，任務 A 被重新排程，acquire回傳Ok，建構出MutexGuard。

整個流程可以用下面的時序圖刻畫：

sequenceDiagram
    participant TaskA as 任务 A
    participant Mutex as Mutex.s (batch_semaphore)
    participant TaskB as 任务 B (持锁者)
    participant Exec as Executor

    TaskA->>Mutex: acquire(1).await
    Mutex-->>TaskA: Pending (Waker 入队)
    TaskA->>Exec: 让出，调度其他任务
    Note over TaskB: 持有锁执行临界区
    TaskB->>Mutex: MutexGuard::drop -> release(1)
    Mutex->>TaskA: 唤醒队首 Waker
    Exec->>TaskA: 重新 poll
    TaskA->>Mutex: acquire(1) 重试
    Mutex-->>TaskA: Ok(()) 获得许可
    TaskA->>TaskA: 构造 MutexGuard

設計思考：FIFO 公平性與取消安全

文件明確聲明 Tokio 的 Mutex 保證 FIFO📎 tokio/src/sync/mutex.rs:20-22。這一公平性來自底層號誌量的排隊語義。公平的代價是：一次lock被取消（比如在select!中落敗）會讓你失去佇列中的位置 📎 tokio/src/sync/mutex.rs:415-419。這不是 bug，而是 FIFO 佇列的必然——取消意味著從佇列中移除，重新lock就得重新排隊。

另一個反直覺的設計是不投毒（no poisoning）。std::sync::Mutex在持鎖執行緒 panic 時會標記為 poisoned，後續lock回傳Err。Tokio 的 Mutex 不這麼做：持鎖者 panic 時鎖會被正常釋放📎 tokio/src/sync/mutex.rs:122-125。文件警告，如果 panic 被捕獲，受保護資料可能處於不一致狀態。這是非同步場景下的務實取捨——panic 在非同步任務裡通常意味著任務終止，投毒機制反而增加複雜度。

MutexGuard::map系列方法值得一提。它允許把整個MutexGuard<T>降級為只保護某個子欄位的MappedMutexGuard<U>。實作上，它先用閉包算出子欄位指標data，再透過skip_drop把原 guard 拆解成不觸發 Drop 的MutexGuardInner，最後建構新的 guard📎 tokio/src/sync/mutex.rs:869-883。skip_drop用ManuallyDrop + ptr::read轉移欄位所有權，避免Drop被呼叫兩次📎 tokio/src/sync/mutex.rs:827-836。這是 Rust 裡「轉移所有權但不觸發解構」的經典手法。

Semaphore：許可計數與等待佇列如何實作背壓

直覺模型：停車場的車位

號誌量就像停車場：acquire是開車進場，有空位就進，沒空位就在門口排隊；release是開車離場，空出一個位子就通知隊首的車進場。許可數就是車位總數，acquire_many(n)就是一輛佔 n 個車位的大車。

資料結構與記憶體佈局

公開的Semaphore只是底層batch_semaphore::Semaphore的薄封裝：

📎 tokio/src/sync/semaphore.rs:427-432

pub struct Semaphore {
    ll_sem: ll::Semaphore,
    #[cfg(all(tokio_unstable, feature = "tracing"))]
    resource_span: tracing::Span,
}

SemaphorePermit<'a>持有號誌量參考和許可計數：

📎 tokio/src/sync/semaphore.rs:442-445

pub struct SemaphorePermit<'a> {
    sem: &'a Semaphore,
    permits: usize,
}

permits欄位是理解forget/merge/split的關鍵。forget把permits置零📎 tokio/src/sync/semaphore.rs:1193-1195，這樣 Drop 時歸還 0 個許可——等價於「永久消耗」這些許可。split從當前許可裡切出 n 個給新 permit📎 tokio/src/sync/semaphore.rs:1260-1271。merge把另一個 permit 的計數合併進來，並斷言兩者來自同一號誌量📎 tokio/src/sync/semaphore.rs:1230-1240。

Step-by-Step：acquire 與 release 的許可流轉

場景：號誌量初始 2 個許可，任務 Aacquire()，任務 Bacquire_many(2)。

acquire()委託給ll_sem.acquire(1)，成功後建構SemaphorePermit { permits: 1 } 📎 tokio/src/sync/semaphore.rs:614-631。acquire_many(2)類似，但傳 2📎 tokio/src/sync/semaphore.rs:661-679。

若許可不足，ll_sem.acquire(n)回傳Pending，Waker 入隊。這裡有個公平性細節：文件指出，如果隊首是一個acquire_many(5)而當前只剩 3 個許可，即使後面有個acquire(1)能立刻滿足，它也必須等——因為隊首的大車佔著隊📎 tokio/src/sync/semaphore.rs:19-24。這是嚴格 FIFO 的代價，避免了飢餓。

釋放路徑在 Drop：

📎 tokio/src/sync/semaphore.rs:1402-1404

impl Drop for SemaphorePermit<'_> {
    fn drop(&mut self) {
        self.sem.add_permits(self.permits);
    }
}

add_permits委託給ll_sem.release(n) 📎 tokio/src/sync/semaphore.rs:568-570，底層把許可還給等待佇列，喚醒能湊夠許可的等待者。

記憶體序方面，文件給出了強保證：acquire、release、close 都是AcqRel操作，彼此全序，等價於單個原子變數上的AcqRel 📎 tokio/src/sync/semaphore.rs:35-42。這意味著「先寫資料再 release 許可」的寫入，對「後 acquire 許可」的任務可見——號誌可以安全地在任務間傳遞資料。

設計思考：close 與背壓

close()讓所有等待者收到AcquireError，且後續try_acquire返回Closed 📎 tokio/src/sync/semaphore.rs:1161-1163。這是優雅關閉的基礎：當接收端不再需要資料時，close 號誌能讓所有阻塞的發送者立刻失敗返回，而不是永遠等待。

背壓的本質在 mpsc 裡體現得最清楚。下一節會看到，mpsc 的容量控制就是用一個許可數等於 buffer 大小的號誌實現的。

通道家族：等待者佇列與 Waker 喚醒的不同取捨

直覺模型：四種通道，四種等待策略

oneshot是「一次性信封」——只能送一封信，發送方不等待（send是同步的），接收方await等信。mpsc是「有界傳送帶」——發送方在傳送帶滿時等待，接收方在空時等待，容量由號誌控制。broadcast和watch是「廣播喇叭」——一個發送方，多個接收方，但兩者對「落後」的處理截然不同。

本節原始碼材料聚焦oneshot和mpsc::bounded，我們逐一拆解。

oneshot：用狀態位編碼的極簡握手

oneshot的Inner結構是理解其設計的核心：

📎 tokio/src/sync/oneshot.rs:386-409

struct Inner<T> {
    state: AtomicUsize,
    value: UnsafeCell<Option<T>>,
    tx_task: Task,
    rx_task: Task,
}

state是一個AtomicUsize，用位標誌編碼整個通道的狀態。四個標誌位定義在檔案末尾：

📎 tokio/src/sync/oneshot.rs:1488-1505

const RX_TASK_SET: usize = 0b00001;
const VALUE_SENT: usize = 0b00010;
const CLOSED: usize = 0b00100;
const TX_TASK_SET: usize = 0b01000;

value是UnsafeCell<Option<T>>，tx_task和rx_task是Task類型，內部是UnsafeCell<MaybeUninit<Waker>> 📎 tokio/src/sync/oneshot.rs:411-411。注意MaybeUninit——Waker 可能未初始化，是否有效由state裡的RX_TASK_SET/TX_TASK_SET位決定📎 tokio/src/sync/oneshot.rs:396-399。

這個設計的精髓：VALUE_SENT位不僅表示「值已發送」，還決定了UnsafeCell的存取權歸屬。註解寫得非常明確📎 tokio/src/sync/oneshot.rs:1491-1496：若VALUE_SENT置位，UnsafeCell只能被接收方存取；若未置位，只能被發送方存取。這樣就用一個原子位實現了無鎖的所有權轉移，避免了額外的鎖。

send的流程：

📎 tokio/src/sync/oneshot.rs:622-646

pub fn send(mut self, t: T) -> Result<(), T> {
    let inner = self.inner.take().unwrap();
    inner.value.with_mut(|ptr| unsafe {
        *ptr = Some(t);
    });
    if !inner.complete() {
        unsafe {
            return Err(inner.consume_value().unwrap());
        }
    }
    Ok(())
}

先把值寫入UnsafeCell（此時VALUE_SENT未置位，接收方不會存取），再呼叫complete()嘗試置位VALUE_SENT。complete()是一個 CAS 迴圈：

📎 tokio/src/sync/oneshot.rs:1516-1549

fn set_complete(cell: &AtomicUsize) -> State {
    let mut state = cell.load(Ordering::Relaxed);
    loop {
        if State(state).is_closed() {
            break;
        }
        match cell.compare_exchange_weak(
            state, state | VALUE_SENT, Ordering::AcqRel, Ordering::Acquire,
        ) {
            Ok(_) => break,
            Err(actual) => state = actual,
        }
    }
    State(state)
}

為什麼用 CAS 而非簡單的fetch_or？註解解釋得很清楚📎 tokio/src/sync/oneshot.rs:1517-1529：如果通道已CLOSED，就不能再置VALUE_SENT。因為一旦置位，接收方會認為可以存取UnsafeCell，而此時發送方正準備把值取回去（consume_value），兩邊同時存取就會資料競爭。所以 CAS 迴圈在發現CLOSED時提前 break，不置位。

complete()返回後，如果成功置位且RX_TASK_SET已置位，就喚醒接收方：

📎 tokio/src/sync/oneshot.rs:1300-1315

fn complete(&self) -> bool {
    let prev = State::set_complete(&self.state);
    if prev.is_closed() {
        return false;
    }
    if prev.is_rx_task_set() {
        unsafe {
            self.rx_task.with_task(Waker::wake_by_ref);
        }
    }
    true
}

接收方的poll_recv是狀態機的核心：

📎 tokio/src/sync/oneshot.rs:1317-1384

它先載入狀態，若is_complete()則直接consume_value返回；若is_closed()返回Err；否則進入「登記 Waker」分支。登記時先檢查is_rx_task_set()，若已設定且will_wake判斷是同一個 Waker 就不重複設定；若不同則先 unset 再 set。這裡有個微妙的競態處理：unset 之後如果發現is_complete()變真了，要把標誌位重新 set 回去 📎 tokio/src/sync/oneshot.rs:1342-1344，否則 Waker 會在 Drop 時洩漏（因為 Drop 依賴標誌位判斷是否要 drop Waker）。

這個「unset 後重新 set」的模式在poll_closed裡也出現📎 tokio/src/sync/oneshot.rs:839-848，是 oneshot 處理並發喚醒的標準手法。

mpsc::bounded：號誌驅動的背壓

mpsc 的容量控制完全交給號誌。channel函式建立一個許可數等於 buffer 的號誌：

📎 tokio/src/sync/mpsc/bounded.rs:159-171

pub fn channel<T>(buffer: usize) -> (Sender<T>, Receiver<T>) {
    assert!(buffer > 0, "mpsc bounded channel requires buffer > 0");
    let semaphore = Semaphore {
        semaphore: semaphore::Semaphore::new(buffer),
        bound: buffer,
    };
    let (tx, rx) = chan::channel(semaphore);
    let tx = Sender::new(tx);
    let rx = Receiver::new(rx);
    (tx, rx)
}

Semaphore是 mpsc 內部的包裝，同時持有底層號誌和bound（最大容量）📎 tokio/src/sync/mpsc/bounded.rs:176-179。bound用於max_capacity查詢，而available_permits給出當前容量📎 tokio/src/sync/mpsc/bounded.rs:591-593。

發送路徑send先reserve再send：

📎 tokio/src/sync/mpsc/bounded.rs:816-824

pub async fn send(&self, value: T) -> Result<(), SendError<T>> {
    match self.reserve().await {
        Ok(permit) => {
            permit.send(value);
            Ok(())
        }
        Err(_) => Err(SendError(value)),
    }
}

reserve內部呼叫reserve_inner(1)，後者先檢查n > max_capacity直接返回錯誤，再acquire(n) 📎 tokio/src/sync/mpsc/bounded.rs:1272-1311。這裡有個精妙的WakeReceiverOnDrop守衛：

📎 tokio/src/sync/mpsc/bounded.rs:1286-1301

struct WakeReceiverOnDrop<'a, T> {
    chan: &'a chan::Tx<T, Semaphore>,
}
impl<T> Drop for WakeReceiverOnDrop<'_, T> {
    fn drop(&mut self) {
        use chan::Semaphore;
        let semaphore = self.chan.semaphore();
        if semaphore.is_closed() && semaphore.is_idle() {
            self.chan.wake_rx();
        }
    }
}

註解解釋了動機📎 tokio/src/sync/mpsc/bounded.rs:1279-1285：如果reserve在拿到部分許可後被取消（比如select!落敗），底層Acquire會在 Drop 時歸還這些許可，但不會像Permit那樣通知接收方。如果此時通道已關閉且空閒，接收方可能永遠等不到「通道已關閉」的通知。這個守衛在 Drop 時補上這個喚醒。成功時用mem::forget(guard)取消守衛📎 tokio/src/sync/mpsc/bounded.rs:1306-1306，因為成功路徑由Permit接管通知職責。

Permit的 Drop 也做同樣的事：

📎 tokio/src/sync/mpsc/bounded.rs:1732-1745

impl<T> Drop for Permit<'_, T> {
    fn drop(&mut self) {
        use chan::Semaphore;
        let semaphore = self.chan.semaphore();
        semaphore.add_permit();
        if semaphore.is_closed() && semaphore.is_idle() {
            self.chan.wake_rx();
        }
    }
}

Permit::send則用mem::forget跳過 Drop，避免歸還許可📎 tokio/src/sync/mpsc/bounded.rs:1721-1728。

接收路徑recv用poll_fn包裝chan.recv(cx) 📎 tokio/src/sync/mpsc/bounded.rs:243-246。poll_recv直接委託📎 tokio/src/sync/mpsc/bounded.rs:650-652。真正的等待佇列邏輯在chan模組（本章未展開），但可以推斷：接收方 Waker 存在chan::Rx裡，當發送方send時喚醒。

try_send展示了非阻塞路徑：

📎 tokio/src/sync/mpsc/bounded.rs:924-934

pub fn try_send(&self, message: T) -> Result<(), TrySendError<T>> {
    match self.chan.semaphore().semaphore.try_acquire(1) {
        Ok(()) => {}
        Err(TryAcquireError::Closed) => return Err(TrySendError::Closed(message)),
        Err(TryAcquireError::NoPermits) => return Err(TrySendError::Full(message)),
    }
    self.chan.send(message);
    Ok(())
}

try_acquire的兩種錯誤精確映射到Closed和Full，區分了「通道關閉」和「緩衝區滿」兩種失敗。

設計思考：取消安全與訊息遺失

mpsc 文件反覆強調取消安全📎 tokio/src/sync/mpsc/bounded.rs:776-784：send在select!中落敗時，訊息會被丟棄。要避免遺失，必須用reserve拿到Permit再send——因為Permit已經預留了容量，send是同步的、不會被打斷。

recv則是取消安全的📎 tokio/src/sync/mpsc/bounded.rs:199-204：若recv在select!中落敗，保證沒有訊息被消費。這是因為recv的poll_recv只在真正取到訊息時才返回Ready，Pending時不動佇列。

oneshot的Receiver作為 Future 也是取消安全的📎 tokio/src/sync/oneshot.rs:246-251。但要注意：oneshot的send是同步的，所以不存在「send 被取消」的問題——要麼發出去，要麼Err返回原值。

設計思考與生產踩坑

坑一：用異步 Mutex 保護純資料。文件明確建議📎 tokio/src/sync/mutex.rs:26-36：如果受保護的是純資料（無.await需求），用std::sync::Mutex或parking_lot更快。異步 Mutex 的開銷在於信號量的原子操作和可能的任務調度。只有當需要在持鎖期間.await（比如持鎖訪問資料庫連接）時，才用異步 Mutex。

坑二：持鎖跨.await導致死鎖。這是異步 Mutex 最危險的陷阱。如果任務 A 持鎖後.await一個需要任務 B 完成的事件，而任務 B 又在等這把鎖，就死鎖了。std::sync::Mutex的 guard 不是Send（在可移動任務中），編譯器會阻止跨.await持有；但異步 Mutex 的 guard 是Send 📎 tokio/src/sync/mutex.rs:314-314，編譯器不攔你，需要自己保證不形成循環等待。

坑三：reserve後忘記send。 Permit的 Drop 會歸還許可📎 tokio/src/sync/mpsc/bounded.rs:1732-1745，所以不會洩漏容量。但如果通道已關閉且空閒，Drop 會喚醒接收方——這個喚醒是必要的，否則接收方可能永遠等不到關閉通知。

坑四：oneshot的poll可能虛假Pending。文件說明📎 tokio/src/sync/oneshot.rs:236-242：即使訊息已發送，poll也可能返回Pending。這不是 bug，而是並發競態下的正常現象——調用者會被喚醒重試，訊息不會丟失，只是延遲。

坑五：forget_permits的語義。 forget_permits(n)嘗試減少 n 個許可，返回實際減少的數量📎 tokio/src/sync/semaphore.rs:576-578。它不會阻塞，也不會喚醒等待者——只是單純地「吞掉」許可。用於動態收縮信號量容量。

本章小結

本章揭示了tokio::sync的核心模式：所有異步等待原語都建立在「等待者隊列 + Waker 喚醒」之上，而隊列的具體實現因場景而異。

• Mutex復用許可數為 1 的信號量，MutexGuard只持引用，Drop 時release(1)，FIFO 公平但不投毒。
• Semaphore是許可計數 + 等待隊列，SemaphorePermit用permits計數支持forget/merge/split，MAX_PERMITS右移 3 位為狀態標誌留位。
• oneshot用單個AtomicUsize的位標誌編碼狀態，VALUE_SENT位同時決定UnsafeCell的訪問權歸屬，CAS 循環防止在CLOSED後置位。
• mpsc::bounded用許可數等於 buffer 的信號量實現背壓，WakeReceiverOnDrop守衛處理取消時的喚醒補償。

本章思考與自測

至此，我們已經看清tokio::sync如何用「等待者隊列 + Waker 喚醒」這一統一模式，支撐起 Mutex、Semaphore 與各類通道的異步等待。但並非所有阻塞都能被異步化——有些操作（如檔案系統調用、CPU 密集計算）本質上會阻塞執行緒。下一章我們將進入spawn_blocking執行緒池與block_on的邊界，看看 Tokio 如何在異步運行時與同步阻塞之間架起橋樑。

Waker 的存放位置因原語而異：Mutex/Semaphore 存在底層信號量的等待佇列，oneshot 存在 Inner 的 tx_task/rx_task 欄位，mpsc 存在 chan 模組的收發佇列。但喚醒機制統一：狀態變更時取出 Waker 呼叫 wake_by_ref，執行器重新排程任務。至此，非同步原語內部的等待與喚醒已清晰可見。然而，並非所有程式碼都能非同步化——下一章將探討如何用 spawn_blocking 橋接阻塞操作，以及 block_on 如何在非非同步上下文中驅動 Future。

上一章我們看到，非同步 Mutex 和通道之所以能在等待時不佔用執行緒，關鍵在於把 Waker 存進等待佇列，等條件滿足後再由喚醒者重新排程任務。但這一切的前提是任務能在 Pending 時主動讓出執行緒。一旦程式碼呼叫 std::fs::read、libsqlite3 或純 CPU 壓縮迴圈，它就會霸佔 worker 執行緒直到返回，期間該執行緒上的其他任務全部餓死。Tokio 的解法是把這類工作外包給獨立的阻塞執行緒池，並用 block_on 在非非同步上下文裡驅動 Future。本章拆解這兩條邊界。

8.1 阻塞執行緒池的記憶體佈局：Inner 與雙實現佇列

直覺模型：spawn_blocking執行緒池就像餐廳的「外包幫工池」。前台服務員（worker 執行緒）只負責點單和傳菜，遇到需要慢燉的菜就寫一張工單丟進後廚的傳菜窗口（佇列），幫工（阻塞執行緒）從窗口取單。若沒有這個池子，服務員就得親自下廚，整個餐廳停擺。

核心結構。整個池子由BlockingPool持有，它只存兩樣東西：一個可複製的Spawner（投遞入口）和一個shutdown_rx（關閉信號接收端）📎 tokio/src/runtime/blocking/pool.rs:20-23。Spawner內部是Arc<Inner>，所有投遞者共享同一份狀態📎 tokio/src/runtime/blocking/pool.rs:26-28。

Inner是池子的全部狀態，欄位值得逐個看📎 tokio/src/runtime/blocking/pool.rs:77-104：

• inner_impl: InnerImpl：佇列 + 通知 + 鎖拓撲的實現，是一個列舉，有Locked和Sharded兩個變體📎 tokio/src/runtime/blocking/pool.rs:107-110。這是本章最關鍵的抽象——它把「單鎖佇列」和「分片佇列」兩種拓撲統一在一個介面下。
• thread_cap: usize：執行緒數上限，即max_blocking_threads。
• scheduler_threads: usize：排程器 worker 執行緒數，用於在指標裡扣除，使num_blocking_threads只統計阻塞執行緒📎 tokio/src/runtime/blocking/pool.rs:455-460。
• keep_alive: Duration：閒置執行緒存活時長，預設KEEP_ALIVE = 10s 📎 tokio/src/runtime/blocking/pool.rs:231。
• metrics: SpawnerMetrics：三個原子計數器——num_threads、num_idle_threads、queue_depth 📎 tokio/src/runtime/blocking/pool.rs:31-35。

執行緒管理狀態。ThreadManagementState被單獨抽出來，供兩種佇列實現複用📎 tokio/src/runtime/blocking/pool.rs:135-150：

• shutdown: bool：關閉標誌。
• shutdown_tx: Option<shutdown::Sender>：每個 worker 執行緒持有一份複製，全部 drop 後shutdown_rx收到通知。
• last_exiting_thread: Option<JoinHandle<()>>：上一個超時退出的執行緒句柄。
• worker_threads: HashMap<usize, JoinHandle<()>>：所有存活 worker 的句柄。
• worker_thread_index: usize：單調遞增的執行緒 ID 分配器。

last_exiting_thread的設計動機在註解裡寫得很清楚：超時退出的執行緒會 join 上一個超時退出的執行緒，避免 Valgrind 誤報📎 tokio/src/runtime/blocking/pool.rs:135-150。worker_timed_out正是這個鏈式 join 的實現——它移除自己的句柄，把舊的last_exiting_thread換出來返回給呼叫者去 join📎 tokio/src/runtime/blocking/pool.rs:172-178。

任務封裝。佇列裡存的是Task，它包了一個UnownedTask<BlockingSchedule>和一個Mandatory標誌📎 tokio/src/runtime/blocking/pool.rs:187-191。Mandatory決定關閉時這個任務是被丟棄還是強制執行：shutdown_or_run_if_mandatory在NonMandatory時調shutdown()，在Mandatory時調run() 📎 tokio/src/runtime/blocking/pool.rs:223-228。這就是spawn_blocking（非強制）與spawn_mandatory_blocking（強制，供 fs 使用）的區別📎 tokio/src/runtime/blocking/pool.rs:233-265。

單鎖實現的記憶體佈局。LockedImpl是最原始的拓撲：一個Mutex<LockedInner>加一個Condvar 📎 tokio/src/runtime/blocking/pool.rs:113-116。LockedInner裡是VecDeque<Task>、num_notify: u32和thread_mgmt_state 📎 tokio/src/runtime/blocking/pool.rs:118-124。注意num_notify與thread_mgmt_state在同一個鎖下，而num_idle_threads是鎖外的原子量——這種「部分狀態在鎖內、部分在鎖外」的混合佈局，正是後面所有並發微妙性的根源。

8.2 投遞路徑：從 spawn_blocking 到執行緒喚醒

場景：非同步任務裡呼叫tokio::task::spawn_blocking(move || heavy_compute(data))，此刻發生了什麼？

第一步：裝箱決策與任務構造。Spawner::spawn_blocking先測量閉包大小fn_size，然後根據AutoBox::<F>::SHOULD_BOX決定是否把閉包Box起來📎 tokio/src/runtime/blocking/pool.rs:359-389。這是 Tokio 通用的「大 Future 自動裝箱」策略：閉包過大時裝箱，避免任務結構體膨脹。

進入spawn_blocking_inner，先分配任務 ID，再用blocking_task把閉包包成一個 Future，最後用task::unowned構造出UnownedTask和JoinHandle 📎 tokio/src/runtime/blocking/pool.rs:440-449。注意這裡返回的是(JoinHandle<R>, Result<(), SpawnError>)二元組——句柄和投遞結果分開返回。

第二步：投遞結果的三種處理。回到spawn_blocking，對spawn_result做匹配📎 tokio/src/runtime/blocking/pool.rs:381-388：

• Ok(())：正常，返回句柄。
• Err(ShuttingDown)：不 panic，仍然返回句柄。註解說明這是相容性考量——句柄永遠不會 resolve，但呼叫方不會因為執行時正在關閉而崩潰。
• Err(NoThreads(e))：OS 無法建立執行緒且池中無人接手，直接 panic。

第三步：入隊與喚醒決策。spawn_task把on_no_idle閉包傳給InnerImpl::spawn_task，由具體實作決定何時呼叫它📎 tokio/src/runtime/blocking/pool.rs:462-506。看LockedImpl::spawn_task的臨界區📎 tokio/src/runtime/blocking/pool.rs:603-639：

let mut locked = self.mutex.lock();

if locked.thread_mgmt_state.shutdown {
    task.task.shutdown();
    return Err(SpawnError::ShuttingDown);
}

locked.queue.push_back(task);
metrics.inc_queue_depth();

if metrics.num_idle_threads() == 0 {
    on_no_idle(&mut locked.thread_mgmt_state)?;
} else {
    metrics.dec_num_idle_threads();
    locked.num_notify += 1;
    self.condvar.notify_one();
}

這裡有兩個關鍵點。其一，關閉檢查在入隊之前，且即使任務是Mandatory也直接shutdown()——註解解釋：它在關閉開始之後才被排程，所以丟棄是合法的📎 tokio/src/runtime/blocking/pool.rs:614-620。其二，喚醒決策依賴鎖外的num_idle_threads：若為 0，調on_no_idle嘗試起新執行緒；否則遞減空閒計數、遞增num_notify、notify_one。

num_notify為什麼必須存在？因為Condvar可能產生虛假喚醒（spurious wakeup）。如果只用notify_one而不計數，一個虛假喚醒的執行緒會誤以為有任務可取，結果發現佇列為空又睡回去，而真正被喚醒的執行緒可能永遠收不到通知。num_notify把「合法喚醒」變成可計數的令牌：投遞方+1，被喚醒方在num_notify != 0時才認為喚醒合法並-1 📎 tokio/src/runtime/blocking/pool.rs:674-684。

第四步：起新執行緒。on_no_idle閉包在持有佇列鎖的情況下執行📎 tokio/src/runtime/blocking/pool.rs:462-506。它先檢查num_threads == thread_cap，達到上限就直接返回Ok(())——任務留在佇列裡等現有執行緒處理，這就是背壓。否則複製shutdown_tx，調spawn_thread建立執行緒，成功後遞增num_threads、遞增worker_thread_index、把句柄插入worker_threads。

spawn_thread用thread::Builder設定執行緒名和堆疊大小，然後 spawn 一個閉包：進入執行時上下文rt.enter()，呼叫inner.run(id)，最後 dropshutdown_tx 📎 tokio/src/runtime/blocking/pool.rs:508-528。

OS 執行緒建立失敗的容錯。spawn_thread可能失敗。程式碼對錯誤做了分類📎 tokio/src/runtime/blocking/pool.rs:488-500：若是WouldBlock（臨時性錯誤，由is_temporary_os_thread_error判定📎 tokio/src/runtime/blocking/pool.rs:750-752）且池中已有阻塞執行緒，則靜默忽略——任務會被某個當前忙碌的執行緒最終取走。否則返回SpawnError::NoThreads，最終導致 panic。

用一張控制流圖總結投遞路徑的決策分支：

flowchart TD
    call["Spawner::spawn_blocking(func)"] --> box{"AutoBox::SHOULD_BOX?"}
    box -->|是| boxed["Box::new(func)"]
    box -->|否| raw["func"]
    boxed --> inner["spawn_blocking_inner"]
    raw --> inner
    inner --> unowned["task::unowned -> Task + JoinHandle"]
    unowned --> spawn_task["InnerImpl::spawn_task"]
    spawn_task --> lock["LockedImpl: mutex.lock()"]
    lock --> shutting{"thread_mgmt_state.shutdown?"}
    shutting -->|是| discard["task.task.shutdown()"]
    discard --> err_sd["Err(ShuttingDown)"]
    shutting -->|否| push["queue.push_back(task)"]
    push --> idle{"num_idle_threads == 0?"}
    idle -->|是| on_no_idle["on_no_idle(thread_mgmt_state)"]
    on_no_idle --> cap{"num_threads == thread_cap?"}
    cap -->|是| backpressure["返回 Ok, 任务留队列"]
    cap -->|否| spawn_th["spawn_thread(shutdown_tx, rt, id)"]
    spawn_th --> th_ok{"spawn 成功?"}
    th_ok -->|是| reg["inc_num_threads, 注册 JoinHandle"]
    th_ok -->|否| tmp{"WouldBlock 且已有线程?"}
    tmp -->|是| ignore["忽略, 等忙碌线程取走"]
    tmp -->|否| err_nt["Err(NoThreads)"]
    idle -->|否| notify["dec_num_idle_threads, num_notify+=1, notify_one"]
    err_sd --> ret["返回 JoinHandle"]
    backpressure --> ret
    reg --> ret
    ignore --> ret
    err_nt --> panic_os["panic: OS can't spawn worker thread"]

8.3 worker 主迴圈：BUSY/IDLE 狀態機與逾時回收

直覺模型：每個阻塞執行緒就是一個「待命幫工」。有單時連續幹活（BUSY），沒單時打盹（IDLE），打盹超過keep_alive就下班（逾時退出）。若沒有逾時回收，池子會永久保留峰值時建立的所有執行緒，浪費記憶體與核心排程開銷。

主迴圈結構。LockedImpl::run_worker是一個'main迴圈，內部交替處於 BUSY 和 IDLE 兩個階段📎 tokio/src/runtime/blocking/pool.rs:642-735。注意：這裡的 BUSY/IDLE 是迴圈內的階段，不是顯式列舉狀態，所以下面用流程圖而非狀態圖描述。

BUSY 階段：內層while let Some(task) = locked.queue.pop_front()不斷取任務📎 tokio/src/runtime/blocking/pool.rs:655-661。取到後遞減queue_depth，drop 鎖，執行task.run()，再重新拿鎖。drop 鎖這一步至關重要——阻塞任務可能跑很久，絕不能持鎖執行。

IDLE 階段：佇列空了，遞增num_idle_threads，設is_counted_idle = true，然後進入等待迴圈📎 tokio/src/runtime/blocking/pool.rs:663-696。核心是condvar.wait_timeout(locked, keep_alive)，返回後檢查三件事：

1. num_notify != 0：合法喚醒。遞減num_notify，設is_counted_idle = false（因為投遞方已經遞減過num_idle_threads了），break 回 BUSY📎 tokio/src/runtime/blocking/pool.rs:674-684。

2. 未關閉且逾時：調worker_timed_out拿到上一個退出執行緒的句柄，break 'main退出迴圈📎 tokio/src/runtime/blocking/pool.rs:689-693。

3. 否則為虛假喚醒，繼續等待。

關閉時的佇列排空。若thread_mgmt_state.shutdown為真，進入排空邏輯📎 tokio/src/runtime/blocking/pool.rs:698-710：逐個彈出任務，drop 鎖，調task.shutdown_or_run_if_mandatory()——非強制任務被丟棄，強制任務照常執行。然後 break 退出主迴圈。

退出清理。執行緒退出前遞減num_threads 📎 tokio/src/runtime/blocking/pool.rs:714。若is_counted_idle為真，還要遞減num_idle_threads，並用assert_ne!(prev_idle, 0)斷言沒有下溢📎 tokio/src/runtime/blocking/pool.rs:716-726。這個斷言是除錯期的護欄：一旦num_idle_threads記帳出錯，這裡會立刻 panic 而不是讓錯誤靜默傳播。

最後，若正在關閉且num_threads == 0（最後一個執行緒），notify_one喚醒可能在等待的關閉發起者📎 tokio/src/runtime/blocking/pool.rs:728-730。返回join_on_thread，由Inner::run在退出前 join📎 tokio/src/runtime/blocking/pool.rs:755-771。

關閉握手。BlockingPool::shutdown先調begin_shutdown拿到所有 worker 句柄📎 tokio/src/runtime/blocking/pool.rs:310-312。LockedImpl::begin_shutdown設定關閉標誌、dropshutdown_tx、notify_all喚醒所有等待執行緒📎 tokio/src/runtime/blocking/pool.rs:740-745。然後shutdown_rx.wait(timeout)阻塞等待📎 tokio/src/runtime/blocking/pool.rs:324。

shutdown::Receiver::wait的實作很講究📎 tokio/src/runtime/blocking/shutdown.rs:37-70：先處理timeout == 0的快速路徑直接返回 false；再調try_enter_blocking_region()進入阻塞區域，若失敗且當前正在 panic 則返回 false，否則 panic 並給出「不能在非同步上下文中 drop runtime」的提示📎 tokio/src/runtime/blocking/shutdown.rs:44-57。最後根據 timeout 調block_on_timeout或block_on驅動那個 oneshot。

shutdown_tx的機制是：每個 worker 執行緒持有一份Arc<oneshot::Sender<()>>的複製📎 tokio/src/runtime/blocking/shutdown.rs:12-14。所有執行緒退出後，所有複製被 drop，Arc計數歸零，oneshot::Sender被 drop，Receiver收到通知。這就是「所有 Sender drop 後 Receiver 被喚醒」的經典模式。

sequenceDiagram
    participant App as "应用线程 (drop Runtime)"
    participant Pool as "BlockingPool::shutdown"
    participant Locked as "LockedImpl"
    participant Worker as "阻塞 worker 线程"
    participant Rx as "shutdown::Receiver"

    App->>Pool: shutdown(timeout)
    Pool->>Locked: begin_shutdown()
    Locked->>Locked: thread_mgmt_state.begin_shutdown() 设 shutdown=true, shutdown_tx=None
    Locked->>Worker: condvar.notify_all()
    Locked-->>Pool: Some((last_exited_thread, workers))
    Pool->>Rx: wait(timeout)
    Worker->>Worker: 从 wait_timeout 醒来, 见 shutdown=true
    Worker->>Worker: 排空队列 shutdown_or_run_if_mandatory()
    Worker->>Worker: dec_num_threads, 退出 run_worker
    Worker->>Worker: drop(shutdown_tx) 克隆
    Worker-->>Rx: 最后一个 Sender drop, oneshot 完成
    Rx-->>Pool: 返回 true
    Pool->>Worker: join 所有 worker 句柄

8.4 block_on：在非非同步上下文驅動 Future

直覺模型：block_on是執行時的「正門」。它把當前執行緒變成臨時的執行器，反覆 poll 傳入的 Future 直到完成。若沒有它，main函式就無法啟動任何非同步程式碼。

入口與裝箱。Runtime::block_on同樣先測大小、按SHOULD_BOX決定是否Box::pin，然後進block_on_inner 📎 tokio/src/runtime/runtime.rs:343-350。block_on_inner裡有兩段條件編譯的 trace 包裝（taskdump 和 tracing），然後self.enter()進入執行時上下文，最後按排程器類型分派📎 tokio/src/runtime/runtime.rs:353-383：

let _enter = self.enter();

match &self.scheduler {
    Scheduler::CurrentThread(exec) => exec.block_on(&self.handle.inner, future),
    Scheduler::MultiThread(exec) => exec.block_on(&self.handle.inner, future),
}

兩種排程器的block_on語意不同，文件裡說得很清楚📎 tokio/src/runtime/runtime.rs:302-320：

• 多執行緒排程器：Future 在 I/O 驅動和定時器上下文中運行，block_on返回後已 spawn 的任務繼續運行。
• 當前執行緒排程器：block_on可以被多個執行緒並發調用，第一個調用者取得 I/O 和定時器驅動的所有權，其他執行緒「鉤入」它。第一個block_on完成後，其他執行緒可以「偷走」驅動。block_on返回後已 spawn 的任務被掛起，再次調用block_on會恢復它們。

關鍵限制：不能在非同步上下文中調用。文件明確block_on在非同步執行上下文中調用會 panic📎 tokio/src/runtime/runtime.rs:321-324。原因很直接：block_on會阻塞當前執行緒直到 Future 完成，若當前執行緒本身是某個 worker 執行緒，就會阻塞整個執行器——這正是spawn_blocking要解決的問題，所以兩者互斥。

關閉路徑。Runtime::drop按排程器類型分派📎 tokio/src/runtime/runtime.rs:506-521：當前執行緒排程器需要先try_set_current進入上下文再 shutdown（保證任務在運行時上下文中被 drop）；多執行緒排程器直接 shutdown（worker 執行緒本身已在上下文中）。shutdown_timeout先關排程器再關阻塞池📎 tokio/src/runtime/runtime.rs:457-461，shutdown_background等價於shutdown_timeout(Duration::from_nanos(0)) 📎 tokio/src/runtime/runtime.rs:494-496。

設計思考、錯誤恢復與生產踩坑

為什麼spawn_blocking的ShuttingDown不 panic？ 📎 tokio/src/runtime/blocking/pool.rs:383-384註釋說是兼容性考慮。spawn_blocking返回JoinHandle而非Result，若在關閉時 panic，會讓「運行時正在關閉」這個可預期狀態變成崩潰。返回一個永不 resolve 的句柄，調用方await時會一直掛起——但此時運行時已關閉，整個block_on也會退出，所以實際不會永久洩漏。

max_blocking_threads的背壓語意。預設值很大（512），因為spawn_blocking常用於檔案 I/O。但文件警告：跑 CPU 密集任務時要用信號量限制並發，否則會創建大量執行緒📎 tokio/src/task/blocking.rs:94-100。達到上限後任務在佇列裡排隊，形成背壓——但注意這個背壓只作用於阻塞池，不會反壓到非同步排程器。

spawn_blocking不可取消。文件明確：abort對已開始運行的阻塞任務無效，任務會繼續跑完📎 tokio/src/task/blocking.rs:106-120。只有尚未開始的任務可能被 abort 阻止。關閉時運行時會等待所有已開始的阻塞任務，shutdown_timeout超時後會洩漏這些執行緒。

num_idle_threads的記帳陷阱。is_counted_idle標誌的存在說明這個計數很容易出錯。投遞方在喚醒時遞減num_idle_threads，被喚醒方看到num_notify != 0後設is_counted_idle = false，避免重複遞減📎 tokio/src/runtime/blocking/pool.rs:679-682。若這條路徑有 bug，assert_ne!(prev_idle, 0)會在退出時 panic📎 tokio/src/runtime/blocking/pool.rs:722-725。生產環境若見到「num_idle_threadsunderflowed on thread exit」，說明池子的記帳邏輯被破壞。

InnerImpl枚舉抽象的意義。註釋說明Locked變體的行為與重構前完全一致，而Sharded變體為未來的並發佇列預留了對稱的槽位📎 tokio/src/runtime/blocking/pool.rs:537-539。spawn_task、run_worker、begin_shutdown三個方法都通過枚舉分派📎 tokio/src/runtime/blocking/pool.rs:548-582。這種「枚舉分派 + 每變體自持臨界區」的設計，使得新增佇列拓撲時不需要改動調用方。

本章小結

本章拆解了 Tokio 容納同步代碼的兩條邊界。spawn_blocking把閉包投遞到獨立的阻塞執行緒池：Inner持有佇列、執行緒上限、存活時長與原子指標；LockedImpl用單鎖 +Condvar實現佇列，num_notify計數器補償虛假喚醒；worker 在 BUSY/IDLE 間循環，空閒超時後鏈式 join 退出；max_blocking_threads達到上限後任務排隊形成背壓。block_on則在非非同步上下文驅動 Future，多執行緒與當前執行緒排程器語意不同，且嚴禁在非同步上下文中調用。關閉路徑通過shutdown_tx的Arc計數歸零觸發oneshot，實現「所有 worker 退出後喚醒關閉發起者」的握手。

本章思考與自測

阻塞線程池與 block_on 劃定了異步運行時的能力邊界：前者把無法讓出線程的工作隔離到專用線程，後者讓非異步入口也能驅動 Future。但這兩條邊界在代碼裡往往不是手寫的——下一章我們將進入宏的世界，看看 #[tokio::main]、select! 和 join! 如何在編譯期生成這些運行時代碼。

上一章我們看到block_on與阻塞線程池如何劃定異步運行時的能力邊界，而用戶幾乎從不手寫這些邊界——他們寫#[tokio::main]、select!、join!，讓宏在編譯期把這些樣板代碼鋪開。宏是 Tokio 給用戶的第一層糖衣，也是編譯期真正生成運行時代碼的地方。本章聚焦tokio-macroscrate 與tokio/src/macros/select.rs，拆解三條最常用的宏展開路徑，重點回答一個問題：宏展開後，真實的調用鏈長什麼樣，以及為什麼select!的取消安全語義必須單獨警惕。

9.1 #[tokio::main]：把 async fn 改寫成 Runtime::block_on

直覺模型：#[tokio::main]就像一張「裝修委託書」。你交出一間毛坯房（async fn main），它替你鋪好水電（構建 Runtime）、裝好門窗（enable_all），最後把你原本的家具（函數體）搬進去。若沒有它，每個main都得手寫Builder::new_multi_thread().enable_all().build().unwrap().block_on(...)，樣板代碼會淹沒業務邏輯。

數據結構與內存佈局

宏本身不產生運行時數據結構，但它解析出的配置被裝進兩個結構體。Configuration是「解析期的可變累加器」，字段全是Option，因為屬性參數可能缺省、可能重複、可能非法📎 tokio-macros/src/entry.rs:74-84。注意worker_threads、start_paused、unhandled_panic都帶Span——這是為了在報錯時把錯誤定位到用戶寫的那一行，而不是宏內部📎 tokio-macros/src/entry.rs:74-84。FinalConfig則是「校驗後的不可變結果」，flavor不再是Option，因為build()已經用default_flavor兜底📎 tokio-macros/src/entry.rs:55-62。

RuntimeFlavor只有三個變體：CurrentThread、Threaded、Local 📎 tokio-macros/src/entry.rs:10-14。from_str裡特意為歷史遺留名字給出友好報錯：single_thread提示應叫current_thread，basic_scheduler提示已改名，threaded_scheduler提示已改名📎 tokio-macros/src/entry.rs:17-27。這是宏作為「用戶第一接觸面」的典型設計：錯誤信息即文檔。

Step-by-Step 展開流程

代入場景：用戶寫下#[tokio::main(flavor = "multi_thread", worker_threads = 4)] async fn main() { ... }。

第一步，main入口先解析 item 為自定義的ItemFn 📎 tokio-macros/src/entry.rs:577-580。這個ItemFn不是syn::ItemFn，而是 Tokio 自己實現的解析器，原因寫在註釋裡：它不想遞歸解析整條語句，只做「按 token tree 緩衝、遇到分號切分」的輕量解析📎 tokio-macros/src/entry.rs:720-764。這避免了在宏裡對函數體做完整 AST 構建的開銷。

第二步，build_config校驗async關鍵字是否存在，缺失則報 "theasync keyword is missing" 📎 tokio-macros/src/entry.rs:346-349。隨後遍歷屬性參數，把worker_threads、flavor、start_paused、crate、unhandled_panic、name分派到對應 setter📎 tokio-macros/src/entry.rs:369-399。注意core_threads被顯式拒絕並提示已改名📎 tokio-macros/src/entry.rs:379-382。

第三步，Configuration::build做跨字段一致性校驗。這裡有三條關鍵約束：worker_threads只允許multi_thread 📎 tokio-macros/src/entry.rs:197-217；start_paused只允許current_thread/local 📎 tokio-macros/src/entry.rs:219-229；unhandled_panic同樣只允許current_thread/local 📎 tokio-macros/src/entry.rs:231-241。若用戶選了multi_thread但rt-multi-threadfeature 未開，報錯信息會根據是否顯式指定 flavor 而不同📎 tokio-macros/src/entry.rs:209-216。

第四步，parse_knobs生成代碼。它先抹掉asyncness 📎 tokio-macros/src/entry.rs:441，然後根據 flavor 選擇 builder 起點：CurrentThread/Local用Builder::new_current_thread()，Threaded用Builder::new_multi_thread() 📎 tokio-macros/src/entry.rs:468-477。Local特殊之處在於 build 調用是build_local(Default::default())而非build() 📎 tokio-macros/src/entry.rs:479-483。隨後按需鏈式追加.worker_threads(#v)、.start_paused(#v)、.unhandled_panic(...)、.name(#v) 📎 tokio-macros/src/entry.rs:485-497。

第五步，生成最終函數體。核心是last_block：return #rt.enable_all().#build.expect("Failed building the Runtime").block_on(body) 📎 tokio-macros/src/entry.rs:509-522。注意那個顯式return，註釋指向 tokio-rs/tokio#4636，是為了修復類型推斷問題📎 tokio-macros/src/entry.rs:508。

第六步，函數體被包成async #body並做類型檢查。非 test 路徑下，若返回類型不是!且不含impl Trait，會插入if false { let _: &dyn Future<Output = #output_type> = &body; }做編譯期斷言📎 tokio-macros/src/entry.rs:551-571。test 路徑則用pin!把 body 釘在棧上並轉成Pin<&mut dyn Future>，註解解釋這是為了減少block_on泛型實例化的編譯開銷📎 tokio-macros/src/entry.rs:526-548。

flowchart TD
    entry["main(args, item)"] --> parse_item{"syn::parse2(item) 成功?"}
    parse_item -->|否| err_ret["token_stream_with_error 返回原始 item + 编译错误"]
    parse_item -->|是| check_main{"ident == main 且有参数?"}
    check_main -->|是| err_args["报错: main 不能接受参数"]
    check_main -->|否| parse_args["AttributeArgs::parse_terminated"]
    parse_args --> build_cfg["build_config 校验 async 与各字段"]
    build_cfg --> cfg_ok{"config 构建成功?"}
    cfg_ok -->|否| fallback["parse_knobs(DEFAULT_ERROR_CONFIG) + 错误"]
    cfg_ok -->|是| knobs["parse_knobs 生成 Builder 链 + block_on"]
    knobs --> out["输出同步 fn main"]

設計思考與生產踩坑

main與test共享parse_knobs，但預設 flavor 不同：test預設CurrentThread，main預設Threaded 📎 tokio-macros/src/entry.rs:91-94。這解釋了為什麼#[tokio::test]預設單執行緒——測試通常不需要多核，且單執行緒更容易復現。

一個容易被忽略的坑：巨集展開後每次呼叫函式都會新建 Runtime。文件明確警告，若函式被頻繁呼叫，應改用 Builder 復用 Runtime📎 tokio-macros/src/lib.rs:31-35。把#[tokio::main]用在普通函式上是合法的，但每次呼叫都付一次 Runtime 建構成本。

另一個坑是crate重命名。當使用者use tokio as tokio1時，巨集內部預設生成的tokio::runtime::Builder會找不到路徑，必須顯式crate = "tokio1" 📎 tokio-macros/src/lib.rs:239-264。parse_knobs裡crate_path的預設值是Ident::new("tokio", ...) 📎 tokio-macros/src/entry.rs:456-462，這正是重命名場景報錯的根源。

9.2 select!：多分支輪詢、位元遮罩與隨機公平性

直覺模型：select!像一位「同時盯多個取餐窗口的服務員」。哪個窗口先出餐，他就端走哪份，其餘窗口的排隊作廢。若沒有它，使用者得手寫poll_fn把多個 Future 塞進一個元組逐個 poll，還要自己處理「某個分支就緒後其餘分支該丟棄」的邏輯。

資料結構與記憶體佈局

select!展開後生成一個局部模組__tokio_select_util，裡面有一個列舉Out和一個型別別名Mask 📎 tokio/src/macros/select.rs:615-619。Out的變體名是_0、_1……每個分支一個，外加一個Disabled表示所有分支都失效📎 tokio-macros/src/select.rs:33-39。Mask的底層型別按分支數動態選擇：≤8 用u8，≤16 用u16，≤32 用u32，≤64 用u64，超過 64 直接 panic📎 tokio-macros/src/select.rs:17-31。這個位元遮罩是select!的核心狀態：第 i 位為 1 表示第 i 個分支已被禁用。

所有 Future 被存進一個元組futures，每個元素先經IntoFuture::into_future轉換📎 tokio/src/macros/select.rs:654-656。注意這裡先建構futures_init再逐個into_future，註解解釋這是為了利用臨時生命週期延長📎 tokio/src/macros/select.rs:641-646。隨後let mut futures = &mut futures;把元組降級為可變引用，避免poll_fn閉包奪取所有權📎 tokio/src/macros/select.rs:658-662。

Step-by-Step 輪詢流程

代入場景：select! { v = stream1.next() => ..., v = stream2.next() => ..., else => break }。

第一步，巨集入口規則匹配。若有biased;前綴，start=0 📎 tokio/src/macros/select.rs:801-803；否則start是一個隨機表達式thread_rng_n(BRANCHES) 📎 tokio/src/macros/select.rs:805-809。這就是文件所說的「預設隨機挑選分支先檢查」的公平性來源📎 tokio/src/macros/select.rs:61-65。

第二步，正規化。tt-muncher 把每個分支規整成(skip) pat = fut, if cond => handler,形式，skip是一串_，長度等於該分支之前的 branch 數📎 tokio/src/macros/select.rs:770-793。skip既用於生成元組欄位存取futures_init.$($skip)*，也用於count!算出分支索引。

第三步，前置條件求值。對每個分支的if $c，若為 false，則disabled |= 1 << index 📎 tokio/src/macros/select.rs:631-636。注意：即使分支被禁用，其$fut表達式仍會被求值，只是不會被 poll📎 tokio/src/macros/select.rs:39-41。

第四步，進入poll_fn閉包。先檢查協作預算：ready!(poll_budget_available(cx))，預算耗盡直接返回Pending 📎 tokio/src/macros/select.rs:664-667。這保證select!不會霸佔 worker。

第五步，迴圈for i in 0..BRANCHES，branch = (start + i) % BRANCHES 📎 tokio/src/macros/select.rs:680-685。對每個 branch：先查disabled & mask == mask，已禁用則continue 📎 tokio/src/macros/select.rs:694-699；否則從元組取出該 Future，用Pin::new_unchecked包一層（安全性依賴 Future 存於棧上且不被移動）📎 tokio/src/macros/select.rs:701-707；poll 之，Ready(out)則先disabled |= mask再匹配模式📎 tokio/src/macros/select.rs:710-730。

第六步，模式匹配。若out匹配$bind，返回Poll::Ready(Out::_i(out)) 📎 tokio/src/macros/select.rs:727-733；若不匹配，continue繼續輪詢其他分支——這正是文件步驟 5 所說的「模式不匹配則禁用當前分支」📎 tokio/src/macros/select.rs:44-47。

第七步，迴圈結束。若is_pending為真返回Pending，否則所有分支都失效，返回Out::Disabled 📎 tokio/src/macros/select.rs:740-745。外層match output把Out::_i映射到對應 handler，Disabled映射到else表達式📎 tokio/src/macros/select.rs:749-755。

flowchart TD
    start["poll_fn 闭包被调用"] --> budget{"poll_budget_available(cx)?"}
    budget -->|否| pending_budget["返回 Pending"]
    budget -->|是| init["is_pending = false; start = $start"]
    init --> loop{"i < BRANCHES?"}
    loop -->|否| check_pending{"is_pending?"}
    check_pending -->|是| pending["返回 Pending"]
    check_pending -->|否| disabled_out["返回 Out::Disabled"]
    loop -->|是| branch["branch = (start+i) % BRANCHES"]
    branch --> is_disabled{"disabled & mask == mask?"}
    is_disabled -->|是| next_i["i += 1"]
    is_disabled -->|否| poll_fut["Pin::new_unchecked(fut).poll(cx)"]
    poll_fut --> poll_res{"Poll 结果?"}
    poll_res -->|Pending| set_pending["is_pending = true; i += 1"]
    poll_res -->|Ready| disable["disabled |= mask"]
    disable --> pat_match{"out 匹配 $bind?"}
    pat_match -->|否| next_i
    pat_match -->|是| ready_out["返回 Out::_i(out)"]
    next_i --> loop
    set_pending --> loop

設計思考與生產踩坑

為什麼用位元遮罩而不是Vec<bool>？位元遮罩是棧上單個整數，無堆積分配，且disabled |= mask是單條指令。對於熱路徑上的select!，這避免了每次迭代的堆積存取。

為什麼模式不匹配要禁用分支？這是select!與「簡單 race」的關鍵區別。考慮Some(v) = stream.next() => ...，若stream.next()返回None（串流結束），模式不匹配，該分支被永久禁用，避免無限輪詢一個已結束的串流。文件範例正是靠這個語義收集兩個串流直到都結束📎 tokio/src/macros/select.rs:198-223。

取消安全的真正含義：select!一旦某分支就緒，其餘分支的 Future 會被 drop。若被 drop 的 Future 已經消費了資料但尚未返回，資料就丟了。文件明確列出read_exact、read_to_end、write_all不取消安全📎 tokio/src/macros/select.rs:119-124，而Mutex::lock、Semaphore::acquire因為排隊公平性，取消會丟失佇列位置📎 tokio/src/macros/select.rs:126-133。判定方法：找.await點，若在.await處重啟函式仍正確，則取消安全📎 tokio/src/macros/select.rs:135-139。

if前置條件的競態陷阱：文件給了一個經典錯誤範例——用if !sleep.is_elapsed()守衛sleep分支，但is_elapsed()可能在while檢查與select!之間變為 true，導致超時被漏掉📎 tokio/src/macros/select.rs:336-376。正確寫法是去掉if，讓sleep分支始終參與輪詢，超時後break 📎 tokio/src/macros/select.rs:378-405。

biased;的代價：隨機 RNG 有 CPU 成本，且某些場景需要確定的輪詢順序📎 tokio/src/macros/select.rs:67-74。但biased;把公平性責任交給使用者：若一個分支永遠就緒，後面的分支會餓死📎 tokio/src/macros/select.rs:75-81。

9.3 join! 與巨集展開的工程約束

直覺模型：join!像「同時等所有快遞都到齊」。它不像select!那樣誰先到就取消其餘，而是把所有 Future 的Ready值聚合成一個元組。若沒有它，使用者得手寫poll_fn維護每個 Future 的完成狀態。

資料結構與記憶體佈局

join!的展開同樣基於元組存 Future，但狀態不是位元遮罩，而是一個「已完成值」的元組。每個 Future 完成後，其值被取出存入結果元組，對應槽位標記為已完成。與select!不同，join!不會 drop 未完成的 Future——它必須等所有 Future 都完成才返回。

Step-by-Step 流程

join!的輪詢邏輯與select!共享「元組存 Future +poll_fn驅動」的骨架，但語意相反：select!是「任一就緒即返回」，join!是「全部就緒才返回」。每輪 poll 遍歷所有未完成的 Future，任一返回Pending則整體Pending，全部Ready則聚合返回。

flowchart LR
    subgraph input["输入"]
        f1["Future A"]
        f2["Future B"]
        f3["Future C"]
    end
    subgraph poll["poll_fn 驱动"]
        tuple["元组 (A, B, C)"]
        state["完成状态元组"]
    end
    subgraph output["输出"]
        result["(A::Output, B::Output, C::Output)"]
    end
    f1 --> tuple
    f2 --> tuple
    f3 --> tuple
    tuple --> state
    state -->|"全部 Ready"| result
    state -->|"任一 Pending"| pending["返回 Pending"]

設計思考與生產踩坑

join!的取消安全語意與select!不同：join!被 drop 時，所有未完成的 Future 都會被 drop，同樣可能遺失資料。但由於join!不主動取消任何分支，它不會像select!那樣「因為另一個分支就緒而取消本分支」。真正的風險在於join!整體被外層select!或逾時取消。

join!與try_join!的區別值得注意：try_join!在任一 Future 返回Err時立即返回，取消其餘 Future，因此它繼承了select!的取消安全風險。

設計思考

巨集作為編譯期程式碼生成器的邊界。#[tokio::main]把配置校驗放在編譯期，非法組合（如multi_thread + start_paused）直接編譯失敗，而不是執行期 panic。這是巨集相對 Builder 的核心優勢：錯誤提前。

宣告式巨集 + 程序巨集的混合架構。select!的主體是macro_rules!，但兩處關鍵邏輯委託給程序巨集：select_priv_declare_output_enum生成Out列舉和Mask類型📎 tokio-macros/src/lib.rs:658-660，select_priv_clean_pattern清除模式中的ref/mut 📎 tokio-macros/src/lib.rs:666-668。為什麼？註解解釋：宣告式巨集難以生成「按分支數動態選擇整數類型」的程式碼，也難以在模式位置做 token 級清洗📎 tokio/src/macros/select.rs:577-579。

clean_pattern的必要性。select!把out以&out形式匹配模式📎 tokio/src/macros/select.rs:727，若使用者寫ref v，會變成&ref v導致類型錯誤。clean_pattern遞迴刪除by_ref、mutability，以及Reference模式的mutability 📎 tokio-macros/src/select.rs:68-73📎 tokio-macros/src/select.rs:100-103。這是巨集在「使用者直覺」與「借用檢查器」之間做的妥協。

64 分支上限的工程現實。count!、count_field!、select_variant!三個巨集各自手寫了 0 到 64 的匹配規則📎 tokio/src/macros/select.rs:821-1017📎 tokio/src/macros/select.rs:1021-1217📎 tokio/src/macros/select.rs:1221-1414。註解直言「I'm not happy about it either」📎 tokio/src/macros/select.rs:816-817。這是宣告式巨集無法做算術的代價：只能用 token 數量硬編碼映射到整數。

本章小結

本章思考與自測

巨集把樣板程式碼與編譯期校驗從使用者手裡接了過去，但它生成的仍是普通的 Future 與poll呼叫。下一章我們將離開巨集的編譯期世界，進入執行時的 I/O 抽象層，看看AsyncRead/AsyncWrite如何把位元組流切成幀，以及Framed編解碼框架如何在select!的取消安全約束下正確工作。

#[tokio::main]的本質是「配置解析 + Builder 鏈生成 +block_on包裹」，配置校驗在編譯期完成，flavor 決定 builder 起點與 build 方法。select!的核心是「元組存 Future + 位元遮罩記禁用 + 隨機起點保公平」，模式不匹配即禁用分支，取消安全取決於被 drop 的 Future 是否在.await處可重啟。join!與select!共享骨架但語意相反，前者等全部完成，後者任一就緒即回傳。三者共同展示了 Tokio 巨集設計的核心權衡：把樣板程式碼與編譯期校驗交給巨集，把執行時語意的複雜性（尤其是取消安全）留給使用者顯式理解。理解了巨集如何生成執行時程式碼之後，下一個自然的問題是：當這些程式碼真正開始讀寫位元組流時，Tokio 提供了怎樣的抽象？第 10 章將剖析AsyncRead/AsyncWrite與編解碼框架，看BufReader/BufWriter如何減少系統呼叫、copy_bidirectional如何驅動雙向轉發、Framed如何把位元組流切分為幀，從而回答「非同步 I/O 的抽象邊界在哪裡」。

上一章拆解了 tokio-macros 的展開過程，我們看到 #[tokio::main]、select!、join! 如何把樣板程式碼與編譯期校驗從使用者手裡接過去。但巨集生成的仍是普通的 Future 與 poll 呼叫——當這些 Future 真正開始讀寫位元組時，Tokio 提供的底層抽象只有兩個 trait：AsyncRead 與 AsyncWrite。它們的問題在於「太底層」：一次 poll_read 只保證「讀了一些位元組」，不保證「讀到一個完整訊息」。而絕大多數協定（HTTP、Redis、gRPC、自訂 RPC）都是面向「幀」而非「位元組流」的。本章要回答的核心問題是：非同步 I/O 的抽象邊界應該劃在哪裡？Tokio 的答案分兩層：tokio::io 提供位元組流級別的 trait 與工具（BufReader/BufWriter/copy_bidirectional），tokio-util 的 codec 框架在此之上提供幀級別的 Stream/Sink 適配（Framed/LengthDelimitedCodec）。理解這兩層的分工，就理解了「為什麼協定實作幾乎都從 Framed 開始」。

一、AsyncRead/AsyncWrite：為什麼不能直接複用 std::io::Read

直覺模型

std::io::Read::read是「阻塞式取貨」：你站在窗口前，貨沒到就一直等，執行緒被掛起。AsyncRead::poll_read是「取餐憑證式取貨」：你問一句「好了嗎」，沒好（Poll::Pending）就先去做別的事，同時留下一個 Waker 讓系統在貨到時叫你。若沒有這個 trait，所有非同步 I/O 都得手寫epoll註冊與 Waker 映射——這正是第 5 章 Reactor 做的事，而AsyncRead是它暴露給上層的統一門面。

資料結構與記憶體佈局

AsyncRead的定義極其精簡，只有一個方法：

pub trait AsyncRead {
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<io::Result<()>>;
}

📎 tokio/src/io/async_read.rs:44-60

三個參數各有講究。self: Pin<&mut Self>而非&mut self：因為AsyncRead常常被async fn生成的 Future 持有，而 Future 一旦被 poll 就不可移動（自引用），Pin是編譯器強制的契約。cx: &mut Context<'_>攜帶 Waker，是「取餐器」的傳遞通道。buf: &mut ReadBuf<'_>是 Tokio 對&mut [u8]的封裝——它同時記錄「已填充長度」與「未初始化容量」，避免std::io::Read那種「返回讀取位元組數但緩衝區可能未初始化」的歧義。

文件明確列出三種返回語意📎 tokio/src/io/async_read.rs:15-32：Ready(Ok(()))表示資料已寫入buf，讀取量由ReadBuf::filled的長度增量決定；若增量為 0，要麼是 EOF，要麼是buf.remaining() == 0（緩衝區零容量）；Pending表示當前不可讀但已註冊喚醒；Ready(Err(e))是底層 I/O 錯誤。這裡有一個容易被忽略的陷阱：「讀取量為 0」並不等于 EOF——如果呼叫者傳入一個零容量緩衝區，poll_read會立即返回Ready(Ok(()))但什麼都沒讀。上層若把「0 位元組」當 EOF 處理，就會誤判連線關閉。

場景驅動 Walkthrough：從&[u8]讀一段位元組

考慮最簡單的實作——對&[u8]的AsyncRead：

impl AsyncRead for &[u8] {
    fn poll_read(
        mut self: Pin<&mut Self>,
        _cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<io::Result<()>> {
        let amt = std::cmp::min(self.len(), buf.remaining());
        let (a, b) = self.split_at(amt);
        buf.put_slice(a);
        *self = b;
        Poll::Ready(Ok(()))
    }
}

📎 tokio/src/io/async_read.rs:98-108

逐步解析：self.len()是剩餘未讀切片長度，buf.remaining()是目標緩衝區剩餘容量，取二者較小值amt。split_at(amt)把切片切成「本次要拷貝的a」與「剩餘待讀的b」。buf.put_slice(a)把a拷入ReadBuf並推進其 filled 指標。*self = b把切片自身推進到剩餘部分——這是&[u8]作為「游標」的關鍵：每次 poll 後self指向未讀部分。最後返回Ready(Ok(()))，因為記憶體切片永遠「就緒」，不會Pending。

注意_cx被忽略：記憶體資料源不需要 Waker。這與網路 socket 形成對比——後者在無資料時會返回Pending並註冊可讀興趣。

io::Cursor<T>的實作多了一層邊界檢查📎 tokio/src/io/async_read.rs:113-134：先取position()，若pos > slice.len()（位置越界）直接返回Ready(Ok(()))而不 panic📎 tokio/src/io/async_read.rs:113-134。這是防禦性設計：Cursor的 position 可以被外部set_position設定到任意值，越界時按「已讀完」處理比 panic 更符合 I/O 語意。

設計思考：deref 巨集與 Pin 的傳播

AsyncRead為Box<T>、&mut T、Pin<P>提供了轉發實作。前兩者透過deref_async_read!巨集生成📎 tokio/src/io/async_read.rs:64-70，核心是Pin::new(&mut **self).poll_read(cx, buf)——把Pin<&mut Box<T>>解引用為Pin<&mut T>再轉發。Pin<P>的實作更微妙📎 tokio/src/io/async_read.rs:87-93：它呼叫crate::util::pin_as_deref_mut(self)，把Pin<&mut Pin<P>>投影為Pin<&mut P::Target>。這一層投影是必要的，否則嵌套Pin會導致型別不匹配。

---

二、copy_bidirectional：雙向轉發的狀態機

直覺模型

copy_bidirectional是「雙向傳菜員」：它同時盯著 A→B 和 B→A 兩個方向，任何一邊讀到資料就寫到對面。若沒有它，實作一個 TCP 代理就得手寫兩個copyFuture 並用select!組合——而select!的取消安全約束（第 9 章）會讓「讀到一半被取消」的資料遺失。copy_bidirectional用一個顯式狀態機把「讀-寫-關閉」的中間狀態保存下來，從而做到取消安全。

資料結構與記憶體佈局

核心是一個三態列舉：

enum TransferState {
    Running(CopyBuffer),
    ShuttingDown(u64),
    Done(u64),
}

📎 tokio/src/io/util/copy_bidirectional.rs:10-14

Running持有CopyBuffer（內含 8KB 緩衝區與讀寫計數），表示「正在搬運資料」。ShuttingDown(u64)攜帶已拷貝位元組數，表示「讀端已 EOF，正在關閉寫端」。Done(u64)表示「關閉完成，記錄最終位元組數」。這個列舉是取消安全的關鍵：任何時刻被 drop，狀態都保存在列舉裡，下次 poll 可從斷點繼續。

CopyBuffer來自copy.rs，預設大小由DEFAULT_BUF_SIZE決定（8KB）📎 tokio/src/io/util/copy_bidirectional.rs:76-88。兩個方向各持有一個獨立的CopyBuffer，因此記憶體開銷是 16KB。

場景驅動 Walkthrough：一次雙向轉發的完整生命週期

copy_bidirectional_impl用poll_fn把兩個方向的狀態機組合起來：

let mut a_to_b = TransferState::Running(a_to_b_buffer);
let mut b_to_a = TransferState::Running(b_to_a_buffer);
poll_fn(|cx| {
    let a_to_b = transfer_one_direction(cx, &mut a_to_b, a, b)?;
    let b_to_a = transfer_one_direction(cx, &mut b_to_a, b, a)?;
    let a_to_b = ready!(a_to_b);
    let b_to_a = ready!(b_to_a);
    Poll::Ready(Ok((a_to_b, b_to_a)))
})
.await

📎 tokio/src/io/util/copy_bidirectional.rs:127-151

注意transfer_one_direction的呼叫順序：先推進 a→b，再推進 b→a，兩者都返回Poll。ready!巨集在任一方向未完成時立即返回Pending——但另一個方向的狀態已經被推進了。這正是註解強調的📎 tokio/src/io/util/copy_bidirectional.rs:143-144：即使ready!提前返回，另一方向下次 poll 時仍會返回Done(count)，不會遺失進度。

transfer_one_direction內部是一個loop，按狀態推進：

loop {
    match state {
        TransferState::Running(buf) => {
            let count = ready!(buf.poll_copy(cx, r.as_mut(), w.as_mut()))?;
            *state = TransferState::ShuttingDown(count);
        }
        TransferState::ShuttingDown(count) => {
            ready!(w.as_mut().poll_shutdown(cx))?;
            *state = TransferState::Done(*count);
        }
        TransferState::Done(count) => return Poll::Ready(Ok(*count)),
    }
}

📎 tokio/src/io/util/copy_bidirectional.rs:29-42

Running狀態下呼叫poll_copy，它內部迴圈「讀一塊、寫一塊」直到讀端 EOF 或寫端阻塞。EOF 時返回已拷貝總數，狀態轉為ShuttingDown。ShuttingDown呼叫poll_shutdown關閉寫端（發送 FIN），完成後轉Done。Done直接返回計數。

下面的流程圖展示了單方向狀態機的推進邏輯與錯誤分支：

flowchart TD
    start["transfer_one_direction 进入 loop"] --> match_state{"当前 TransferState?"}
    match_state -->|Running| poll_copy["buf.poll_copy(cx, r, w)"]
    poll_copy --> copy_ready{"poll_copy 结果?"}
    copy_ready -->|Pending| ret_pending["返回 Poll::Pending<br/>状态保持 Running"]
    copy_ready -->|Err| ret_err["返回 Poll::Ready(Err)<br/>错误向上传播"]
    copy_ready -->|Ok(count)| to_shutdown["state = ShuttingDown(count)"]
    to_shutdown --> match_state
    match_state -->|ShuttingDown| poll_shutdown["w.poll_shutdown(cx)"]
    poll_shutdown --> shutdown_ready{"shutdown 结果?"}
    shutdown_ready -->|Pending| ret_pending2["返回 Poll::Pending<br/>状态保持 ShuttingDown"]
    shutdown_ready -->|Err| ret_err
    shutdown_ready -->|Ok| to_done["state = Done(count)"]
    to_done --> match_state
    match_state -->|Done| ret_done["返回 Poll::Ready(Ok(count))"]

設計思考：為什麼用顯式狀態機而非 async fn

錯誤處理上，poll_copy返回的Err會透過?立即向上傳播📎 tokio/src/io/util/copy_bidirectional.rs:32。文件明確說明📎 tokio/src/io/util/copy_bidirectional.rs:67-70：中斷的讀寫會被重試，其他錯誤立即返回，且部分已讀資料可能遺失（未寫入對面）。這是生產環境需要注意的點：copy_bidirectional不保證「要麼全成功要麼全失敗」，錯誤發生時可能已有一半資料在途。

copy_bidirectional_with_sizes額外做了零大小斷言📎 tokio/src/io/util/copy_bidirectional.rs:99-125，因為零容量緩衝區會導致poll_copy永遠返回Ready(Ok(0))被誤判為 EOF，形成忙循環。

---

三、Framed：把位元組流切成幀

直覺模型

Framed是「香腸機」：上游是連續的水流（AsyncRead/AsyncWrite），下游是切好的香腸段（Stream<Item = Frame> / Sink<Frame>）。Decoder負責「從水流中切出一段」，Encoder負責「把一段包成水流」。若沒有Framed，每個協議實作都要手寫「緩衝區管理 + 半包處理 + 黏包拆分」——這正是 codec 框架要消除的重複勞動。

資料結構與記憶體佈局

Framed本身只是一個薄包裝：

pub struct Framed<T, U> {
    #[pin]
    inner: FramedImpl<T, U, RWFrames>
}

📎 tokio-util/src/codec/framed.rs:38-41

真正的狀態在FramedImpl的state: RWFrames裡，包含read: ReadFrame與write: WriteFrame兩部分。ReadFrame的欄位在with_capacity中可見📎 tokio-util/src/codec/framed.rs:107-126：eof: bool（讀端是否 EOF）、is_readable: bool（是否已註冊可讀興趣）、buffer: BytesMut（讀緩衝）、has_errored: bool（是否已出錯，防止重複讀）。WriteFrame欄位📎 tokio-util/src/codec/framed.rs:119-122：buffer: BytesMut（寫緩衝）、backpressure_boundary: usize（背壓閾值）。

backpressure_boundary是背壓機制的關鍵：當寫緩衝超過該閾值時，poll_ready會返回Pending直到資料被刷出，從而對上游Sink施加背壓。預設等於capacity 📎 tokio-util/src/codec/framed.rs:121，可透過set_backpressure_boundary調整📎 tokio-util/src/codec/framed.rs:271-273。

場景驅動 Walkthrough：從 socket 讀一個幀

Framed的Stream實作只是轉發給FramedImpl::poll_next 📎 tokio-util/src/codec/framed.rs:309-311。真正的邏輯在FramedImpl裡（本章未提供該檔案，但可從Framed的介面推斷呼叫鏈）：

1. poll_next先檢查read.buffer中是否已有完整幀（呼叫codec.decode）；

2. 若decode返回Some(frame)，直接產出，不觸碰底層 I/O；

3. 若返回None（半包），檢查read.eof：若已 EOF 且緩衝非空，說明有殘留資料無法解碼，返回錯誤或None；

4. 否則呼叫底層AsyncRead::poll_read讀更多位元組到read.buffer；

5. 讀到的位元組再次嘗試decode，循環直到產出幀或Pending。

這個「先 decode 再 read」的順序很重要：它保證一次 read 可能產出多個幀（黏包），且一個幀可能跨多次 read（半包）。is_readable標誌避免重複註冊可讀興趣——若上次 poll 已註冊且未就緒，本次直接返回Pending而不重複呼叫底層。

Sink實作的呼叫鏈📎 tokio-util/src/codec/framed.rs:315-338：start_send呼叫codec.encode(item, &mut write.buffer)把幀編碼進寫緩衝；poll_flush把write.buffer刷到底層AsyncWrite；poll_ready檢查write.buffer.len() >= backpressure_boundary，超閾值則先 flush 再返回就緒。

下面的時序圖展示了Framed在一次「讀幀-寫幀」往返中的跨組件協作：

sequenceDiagram
    participant App as 应用层
    participant F as FramedImpl
    participant C as Decoder/Encoder
    participant IO as AsyncRead/AsyncWrite

    App->>F: poll_next(cx)
    F->>C: decode(&mut read.buffer)
    alt 缓冲中已有完整帧
        C-->>F: Some(frame)
        F-->>App: Poll::Ready(Some(frame))
    else 半包
        C-->>F: None
        F->>IO: poll_read(cx, &mut read.buffer)
        alt 数据就绪
            IO-->>F: Ready(Ok(()))
            F->>C: decode(&mut read.buffer)
            C-->>F: Some(frame) 或 None
        else 无数据
            IO-->>F: Pending
            F-->>App: Poll::Pending
        end
    end

    App->>F: start_send(frame)
    F->>C: encode(frame, &mut write.buffer)
    C-->>F: Ok(())
    App->>F: poll_flush(cx)
    F->>IO: poll_write(cx, &write.buffer)
    IO-->>F: Ready(Ok(n))
    F->>IO: poll_flush(cx)
    IO-->>F: Ready(Ok(()))

取消安全：Framed 的文件警告

Framed的文件專門列出取消安全語意📎 tokio-util/src/codec/framed.rs:23-30：SinkExt::send若在select!中被其他分支搶先完成，訊息保證未發送，但訊息本身遺失——因為send內部先poll_ready再start_send，若在poll_ready階段被 drop，item已被消費但未編碼。而StreamExt::next是取消安全的：它只持有對底層 stream 的引用，drop 不會遺失已解碼的幀。

設計思考：into_parts與map_codec

Framed提供了into_parts/from_parts用於「換 codec 但保留緩衝區」📎 tokio-util/src/codec/framed.rs:290-298 📎 tokio-util/src/codec/framed.rs:155-166。map_codec就是基於這對方法實作的📎 tokio-util/src/codec/framed.rs:221-234：先into_parts拆出io/codec/read_buf/write_buf，再用map函式轉換 codec，最後from_parts重組。這個設計允許在協議升級（如從明文切換到 TLS）時保留已緩衝的資料，避免重新讀取。

FramedParts的_priv: ()欄位📎 tokio-util/src/codec/framed.rs:373-375是「非窮盡結構體」技巧：私有欄位阻止外部直接建構，強制走new/from_parts，從而允許未來新增欄位而不破壞相容性。

---

四、LengthDelimitedCodec：長度前綴編解碼的狀態機

直覺模型

LengthDelimitedCodec是「按長度切香腸」的專用刀具：它假設每個幀前面有一個固定位元組數的長度欄位，先讀長度再讀 payload。若沒有它，實作一個長度前綴協議就得手寫「讀 4 位元組 → 解析長度 → 讀 N 位元組 → 循環」的狀態機——這正是它內部DecodeState做的事。

資料結構與記憶體佈局

pub struct LengthDelimitedCodec {
    builder: Builder,
    state: DecodeState,
}

enum DecodeState {
    Head,
    Data(usize),
}

📎 tokio-util/src/codec/length_delimited.rs:451-457

DecodeState是顯式狀態機：Head表示「正在讀長度欄位」，Data(n)表示「已解析出長度 n，正在讀 payload」。這個狀態跨decode呼叫保持，因此半包場景下不會遺失進度。

Builder持有全部配置📎 tokio-util/src/codec/length_delimited.rs:416-435：max_frame_len（預設 8MB）、length_field_len（預設 4 位元組）、length_field_offset（預設 0）、length_adjustment（預設 0）、num_skip（預設None，即offset + len）、length_field_is_big_endian（預設 true）。

場景驅動 Walkthrough：解碼一個長度前綴幀

decode是狀態機入口：

fn decode(&mut self, src: &mut BytesMut) -> io::Result<Option<BytesMut>> {
    let n = match self.state {
        DecodeState::Head => match self.decode_head(src)? {
            Some(n) => {
                self.state = DecodeState::Data(n);
                n
            }
            None => return Ok(None),
        },
        DecodeState::Data(n) => n,
    };

    match self.decode_data(n, src) {
        Some(data) => {
            self.state = DecodeState::Head;
            src.reserve(self.builder.num_head_bytes().saturating_sub(src.len()));
            Ok(Some(data))
        }
        None => Ok(None),
    }
}

📎 tokio-util/src/codec/length_delimited.rs:579-603

Head狀態下呼叫decode_head。若返回None（資料不足），直接返回Ok(None)等待更多資料；若返回Some(n)，狀態轉為Data(n)。Data狀態下直接取 n。然後呼叫decode_data(n, src)：若緩衝中已有 n 位元組，split_to(n)切出幀，狀態回到Head，並預留下一幀頭部的空間；否則返回None等待。

decode_head是核心解析邏輯：

let head_len = self.builder.num_head_bytes();
let field_len = self.builder.length_field_len;

if src.len() < head_len {
    return Ok(None);
}

let n = {
    let mut src = Cursor::new(&mut *src);
    src.advance(self.builder.length_field_offset);
    let n = if self.builder.length_field_is_big_endian {
        src.get_uint(field_len)
    } else {
        src.get_uint_le(field_len)
    };

    if n > self.builder.max_frame_len as u64 {
        return Err(io::Error::new(
            io::ErrorKind::InvalidData,
            LengthDelimitedCodecError { _priv: () },
        ));
    }

    let n = n as usize;
    let n = if self.builder.length_adjustment < 0 {
        n.checked_sub(-self.builder.length_adjustment as usize)
    } else {
        n.checked_add(self.builder.length_adjustment as usize)
    };

    match n {
        Some(n) => n,
        None => {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "provided length would overflow after adjustment",
            ));
        }
    }
};

src.advance(self.builder.get_num_skip());
src.reserve(n.saturating_sub(src.len()));
Ok(Some(n))

📎 tokio-util/src/codec/length_delimited.rs:504-562

逐步解析：先檢查src.len() >= head_len，不足則返回None 📎 tokio-util/src/codec/length_delimited.rs:499-502。用Cursor包裝src以便advance/get_uint操作而不消耗原緩衝。advance(length_field_offset)跳過頭部前綴📎 tokio-util/src/codec/length_delimited.rs:517。按端序讀取field_len位元組的長度值📎 tokio-util/src/codec/length_delimited.rs:520-524。

關鍵防禦：若n > max_frame_len，立即返回InvalidData錯誤📎 tokio-util/src/codec/length_delimited.rs:526-531。這防止惡意對端發送「長度欄位為 4GB」的幀導致記憶體耗盡——這是長度前綴協議最經典的 DoS 攻擊面。

長度調整用checked_sub/checked_add而非裸運算📎 tokio-util/src/codec/length_delimited.rs:537-541，溢出時返回InvalidInput錯誤而非 panic。get_num_skip()返回num_skip或預設的offset + len 📎 tokio-util/src/codec/length_delimited.rs:1070-1073，跳過頭部剩餘部分。最後reserve(n.saturating_sub(src.len()))預留 payload 空間📎 tokio-util/src/codec/length_delimited.rs:559——用saturating_sub是因為src可能已包含部分 payload。

下面的流程圖展示了decode的完整決策路徑：

flowchart TD
    entry["decode(src)"] --> check_state{"self.state?"}
    check_state -->|Head| head["decode_head(src)"]
    head --> head_result{"结果?"}
    head_result -->|Ok(None)| ret_none1["返回 Ok(None)<br/>等待更多数据"]
    head_result -->|Err| ret_err1["返回 Err<br/>长度超限或溢出"]
    head_result -->|Ok(Some(n))| set_data["state = Data(n)"]
    set_data --> decode_data
    check_state -->|Data(n)| decode_data["decode_data(n, src)"]
    decode_data --> data_result{"src.len() >= n?"}
    data_result -->|否| ret_none2["返回 Ok(None)<br/>等待更多数据"]
    data_result -->|是| split["src.split_to(n)<br/>state = Head<br/>reserve 下一帧头部"]
    split --> ret_frame["返回 Ok(Some(frame))"]

設計思考：max_frame_len 的裁剪與溢出防護

Builder::adjust_max_frame_len在構造 codec 時把max_frame_len裁剪到長度欄位能表示的最大值📎 tokio-util/src/codec/length_delimited.rs:1075-1081。max_allowed_frame_len計算max_length_field_value + length_adjustment 📎 tokio-util/src/codec/length_delimited.rs:1083-1089，其中max_length_field_value用checked_shl處理length_field_len == 8時的移位溢出📎 tokio-util/src/codec/length_delimited.rs:1091-1096。這個裁剪防止使用者設定「長度欄位 2 位元組但 max_frame_len 設為 1MB」這種矛盾配置——2 位元組最多表示 65535，裁剪後 max_frame_len 變為 65535。

編碼路徑的對稱防護：encode檢查n > max_frame_len返回InvalidInput 📎 tokio-util/src/codec/length_delimited.rs:607-607，長度調整同樣用checked_add/checked_sub 📎 tokio-util/src/codec/length_delimited.rs:620-631。注意編碼時的調整方向與解碼相反：解碼是「讀到的長度 ± adjustment = payload 長度」，編碼是「payload 長度 ∓ adjustment = 寫入的長度欄位」📎 tokio-util/src/codec/length_delimited.rs:620-624。

---

設計思考：抽象邊界的三個層次

回顧本章，Tokio 的 I/O 抽象呈現清晰的三層結構：

第一層：位元組流 trait（AsyncRead/AsyncWrite）。只承諾「讀/寫一些位元組」，不承諾幀邊界。這是最小介面，任何 I/O 源（socket、檔案、記憶體切片）都能實現。代價是上層必須自己處理半包/黏包。

第二層：位元組流工具（BufReader/BufWriter/copy_bidirectional）。在 trait 之上提供「減少系統呼叫」「雙向轉發」等通用能力。copy_bidirectional的顯式狀態機展示了「取消安全」如何在工具層實現——狀態保存在堆疊上而非 Future 內部。

第三層：幀適配（Framed/Decoder/Encoder）。把位元組流提升為Stream<Frame>/Sink<Frame>，讓協議實現只需關心「幀的編解碼」而非「緩衝管理」。LengthDelimitedCodec是這一層的標準範例，其DecodeState狀態機與max_frame_len防護是所有長度前綴協議都應複用的模式。

---

本章小結

• AsyncRead::poll_read用Pin<&mut Self> + Context + ReadBuf三參數替代std::io::Read::read，把「阻塞等待」變為「註冊 Waker + 返回 Pending」。Ready(Ok(()))且讀取量為 0 時需區分 EOF 與零容量緩衝區。
• copy_bidirectional用TransferState三態列舉（Running/ShuttingDown/Done）保存中間狀態，使雙向轉發在select!取消下仍能恢復。錯誤發生時部分資料可能遺失。
• Framed把AsyncRead/AsyncWrite適配為Stream/Sink，ReadFrame/WriteFrame分別管理讀寫緩衝與背壓。SinkExt::send非取消安全（訊息遺失），StreamExt::next取消安全。
• LengthDelimitedCodec用DecodeState（Head/Data(n)）狀態機處理半包，max_frame_len防護長度欄位 DoS，checked_add/checked_sub防護調整溢出。

本章思考與自測

上一章我們拆解了 Framed 的位元組級機制：Decoder 把 BytesMut 切成幀，Sink 把幀寫回，非同步 I/O 的抽象邊界由此清晰。但幀只是資料的容器，真實協定實作緊接著就會遇到三個 tokio::io 與 Framed 都不解決的問題：非同步迭代——Framed 實作了 Stream，但 Stream 只有 poll_next，沒有 next().await、filter、take、merge，手寫 poll_fn 既囉嗦又容易在取消安全上踩坑；動態任務集合——一個聊天服務要同時訂閱 N 個頻道，頻道隨時加入退出，而 select! 的分支數量是編譯期固定的，無法表達執行時增減的流集合；結構化取消——select! 能取消單個分支，但無法把整個任務樹停工這件事傳播下去，也無法等待所有任務真正退出。tokio-stream 與 tokio-util 正是為這三件事而生，它們的關鍵設計原則是不另起爐灶：StreamExt 的每個組合子都只是對 poll_next 的包裝，StreamMap 複用 Waker 的註冊語義，CancellationToken 直接建立在 tokio::sync::Notify 之上，TaskTracker 用一個 AtomicUsize 編碼全部狀態。理解它們，本質上是理解如何在既有 Waker 與排程機制上做零成本抽象。本章按迭代、集合、取消三層遞進：先看 StreamExt 如何把 poll_next 變成可組合的迭代器，再看 StreamMap 與 TaskTracker 如何管理動態集合，最後看 CancellationToken 如何用一棵樹把取消信號傳播到整個任務樹。

StreamExt：把 poll_next 變成可組合的迭代器

直覺模型

Stream之於Future，正如Iterator之於值：Future產出「一個值」，Stream產出「一串值」。但Stream只定義了poll_next這一個原語，就像Iterator只定義了next。若沒有StreamExt，每次過濾、映射、截斷都要手寫poll_fn閉包並手動管理Pin——這正是futurescrate 早期使用者最痛苦的地方。StreamExt的角色，就是給Stream裝上Iterator那樣的組合子生態。

若沒有它，系統面臨的災難不是功能缺失，而是取消安全性的系統性崩塌：每個手寫的poll_fn都可能在被select!取消時丟失一個已經poll出來的元素。

資料結構與記憶體佈局

StreamExt是一個擴展 trait，本身不持有資料：

📎 tokio-stream/src/stream_ext.rs:106-106

pub trait StreamExt: Stream {

它的所有方法都返回一個具體的組合子結構體，而非Box<dyn Stream>。這是關鍵設計：map返回Map<Self, F>，filter返回Filter<Self, F>，take返回Take<Self>。這些結構體都是零堆分配的泛型包裝，編譯器可以把整條鏈內聯成一層層poll_next呼叫。

注意 trait 的 blanket impl：

📎 tokio-stream/src/stream_ext.rs:1213-1213

impl<St: ?Sized> StreamExt for St where St: Stream {}

任何Stream自動獲得全部組合子，無需手動實作。?Sized允許dyn Stream也享受擴展方法。

組合子的模組宣告揭示了這個 trait 的完整能力面：

📎 tokio-stream/src/stream_ext.rs:4-59

mod all; use all::AllFuture;
mod any; use any::AnyFuture;
mod chain; pub use chain::Chain;
pub(crate) mod collect; use collect::{Collect, FromStream};
mod filter; pub use filter::Filter;
mod filter_map; pub use filter_map::FilterMap;
mod fold; use fold::FoldFuture;
mod fuse; pub use fuse::Fuse;
mod map; pub use map::Map;
mod map_while; pub use map_while::MapWhile;
mod merge; pub use merge::Merge;
mod next; use next::Next;
mod skip; pub use skip::Skip;
mod skip_while; pub use skip_while::SkipWhile;
mod take; pub use take::Take;
mod take_while; pub use take_while::TakeWhile;
mod then; pub use then::Then;
mod try_next; use try_next::TryNext;
mod peekable; pub use peekable::Peekable;

這裡有一個值得注意的區分：next、try_next、all、any、fold、collect返回的是Future（Next、TryNext、AllFuture……），因為它們把整個流消費成一個值；而map、filter、take等返回的是Stream，因為它們保持流的形態。next的返回類型是Next<'_, Self>，帶生命週期參數，因為它只借用流：

📎 tokio-stream/src/stream_ext.rs:144-149

fn next(&mut self) -> Next<'_, Self>
where
    Self: Unpin,
{
    Next::new(self)
}

Self: Unpin約束是刻意的：next不取得流的所有權，只借用，因此無法把流Pin住。若流是!Unpin，使用者必須先Box::pin或pin_mut!。文件明確點出了這個取捨：

📎 tokio-stream/src/stream_ext.rs:116-121

/// Note that because `next` doesn't take ownership over the stream,
/// the [`Stream`] type must be [`Unpin`]. If you want to use `next` with
/// a [`!Unpin`](Unpin) stream, you'll first have to pin the stream. This can
/// be done by boxing the stream using [`Box::pin`] or
/// pinning it to the stack using the `pin_mut!` macro from the `pin_utils`
/// crate.

場景驅動 Walkthrough：一次merge的輪詢

merge是理解組合子如何復用 Waker 的最佳樣本。它把兩個流交錯產出，且保證公平性——若兩個流同時就緒，交替產出。文檔特意警告不要鏈式調用merge：

📎 tokio-stream/src/stream_ext.rs:319-321

/// simultaneously, the merge stream alternates between them. This provides
/// some level of fairness. You should not chain calls to `merge`, as this
/// will break the fairness of the merging.

merge的簽名要求兩個流的Item類型相同：

📎 tokio-stream/src/stream_ext.rs:398-404

fn merge<U>(self, other: U) -> Merge<Self, U>
where
    U: Stream<Item = Self::Item>,
    Self: Sized,
{
    Merge::new(self, other)
}

當調用方.next().await時，執行流如下：

1. Next::poll調用Merge::poll_next。

2. Merge內部維護一個「上次輪到誰」的布爾標誌。它先poll上次未產出的那個流；若Pending，再poll另一個。

3. 若兩個都Pending，Merge返回Pending，但兩個流各自的 Waker 都已註冊——任一就緒都會喚醒當前任務。

4. 若一個流返回Ready(None)（結束），Merge記錄該流已結束，此後只poll另一個流，直到它也結束。

這裡的關鍵是：Merge沒有自己的 Waker 管理邏輯，它把cx原樣傳給內部兩個流的poll_next。Waker 的註冊完全由底層流負責，Merge只是決定「這次先問誰」。這正是「復用底層 Waker 機制」的字面含義。

merge_size_hints輔助函數展示了組合子如何合併容量提示：

📎 tokio-stream/src/stream_ext.rs:1216-1226

fn merge_size_hints(
    (left_low, left_high): (usize, Option<usize>),
    (right_low, right_high): (usize, Option<usize>),
) -> (usize, Option<usize>) {
    let low = left_low.saturating_add(right_low);
    let high = match (left_high, right_high) {
        (Some(h1), Some(h2)) => h1.checked_add(h2),
        _ => None,
    };
    (low, high)
}

注意saturating_add與checked_add的選擇：下界用飽和加法（寧可低估不可溢出 panic），上界用檢查加法（任一未知則整體未知）。這是size_hint契約的典型處理方式。

設計思考：取消安全與chunks_timeout的 panic 防護

StreamExt的文檔對每個方法都標註了Cancel safety。以next為例：

📎 tokio-stream/src/stream_ext.rs:123-127

/// # Cancel safety
///
/// This method is cancel safe. The returned future only
/// holds onto a reference to the underlying stream,
/// so dropping it will never lose a value.

next之所以取消安全，是因為它只借用流、不消費元素——Nextfuture 被 drop 時，流本身狀態不變，下次next會重新poll。

但並非所有組合子都取消安全。chunks_timeout在構造時就做了參數校驗：

📎 tokio-stream/src/stream_ext.rs:1178-1185

#[track_caller]
fn chunks_timeout(self, max_size: usize, duration: Duration) -> ChunksTimeout<Self>
where
    Self: Sized,
{
    assert!(max_size > 0, "`max_size` must be non-zero.");
    ChunksTimeout::new(self, max_size, duration)
}

timeout與timeout_repeating的差異也值得注意：timeout在超時後返回一個錯誤，但繼續輪詢內層流；timeout_repeating則按Interval持續產出超時錯誤，直到內層流產出值。文檔用兩個例子精確刻畫了這個區別：

📎 tokio-stream/src/stream_ext.rs:985-1001

/// Once a timeout error is received, no further events will be received
/// unless the wrapped stream yields a value (timeouts do not repeat).

📎 tokio-stream/src/stream_ext.rs:1071-1072

/// Timeout errors will be continuously produced at the specified interval
/// until the wrapped stream yields a value.

---

StreamMap：動態流集合與公平輪詢

直覺模型

select!的分支數在編譯期固定。但聊天服務要訂閱的頻道數、爬蟲要跟蹤的連接數，都是運行時才知道的。StreamMap就是「運行時可增刪的select!」：它把任意多個流放進一個集合，每次next返回(key, value)，告訴你這個值來自哪個流。若沒有它，你只能把所有流塞進一個mpsc通道，多一層轉發開銷。

數據結構與內存佈局

StreamMap的存儲極其樸素——一個Vec：

📎 tokio-stream/src/stream_map.rs:204-208

#[derive(Debug)]
pub struct StreamMap<K, V> {
    /// Streams stored in the map
    entries: Vec<(K, V)>,
}

文檔明確說明了這個選擇的代價：

📎 tokio-stream/src/stream_map.rs:38-44

/// `StreamMap` is backed by a `Vec<(K, V)>`. There is no guarantee that this
/// internal implementation detail will persist in future versions, but it is
/// important to know the runtime implications. In general, `StreamMap` works
/// best with a "smallish" number of streams as all entries are scanned on
/// insert, remove, and polling. In cases where a large number of streams need
/// to be merged, it may be advisable to use tasks sending values on a shared
/// [`mpsc`] channel.

insert的實現體現了「先刪後插」的語義：

📎 tokio-stream/src/stream_map.rs:446-454

pub fn insert(&mut self, k: K, stream: V) -> Option<V>
where
    K: Hash + Eq,
{
    let ret = self.remove(&k);
    self.entries.push((k, stream));

    ret
}

remove用swap_remove把被刪元素與末尾元素交換後彈出，避免 O(n) 搬移：

📎 tokio-stream/src/stream_map.rs:471-483

pub fn remove<Q>(&mut self, k: &Q) -> Option<V>
where
    K: Borrow<Q>,
    Q: Hash + Eq + ?Sized,
{
    for i in 0..self.entries.len() {
        if self.entries[i].0.borrow() == k {
            return Some(self.entries.swap_remove(i).1);
        }
    }

    None
}

場景驅動 Walkthrough：poll_next_entry 的隨機起點與游標修正

StreamMap的核心是poll_next_entry。它從隨機起點開始輪詢，以保證公平性——若總從索引 0 開始，第一個流會餓死後面的流：

📎 tokio-stream/src/stream_map.rs:515-550

fn poll_next_entry(&mut self, cx: &mut Context<'_>) -> Poll<Option<(usize, V::Item)>> {
    let start = self::rand::thread_rng_n(self.entries.len() as u32) as usize;
    let mut idx = start;

    for _ in 0..self.entries.len() {
        let (_, stream) = &mut self.entries[idx];

        match Pin::new(stream).poll_next(cx) {
            Poll::Ready(Some(val)) => return Poll::Ready(Some((idx, val))),
            Poll::Ready(None) => {
                // Remove the entry
                self.entries.swap_remove(idx);

                // Check if this was the last entry, if so the cursor needs
                // to wrap
                if idx == self.entries.len() {
                    idx = 0;
                } else if idx < start && start <= self.entries.len() {
                    // The stream being swapped into the current index has
                    // already been polled, so skip it.
                    idx = idx.wrapping_add(1) % self.entries.len();
                }
            }
            Poll::Pending => {
                idx = idx.wrapping_add(1) % self.entries.len();
            }
        }
    }

    // If the map is empty, then the stream is complete.
    if self.entries.is_empty() {
        Poll::Ready(None)
    } else {
        Poll::Pending
    }
}

這段代碼有三個精妙之處，逐一拆解：

第一，隨機起點。 thread_rng_n使用線程局部FastRand，基於xorshift64+算法：

📎 tokio-stream/src/stream_map.rs:765-768

/// Implement `xorshift64+`: 2 32-bit `xorshift` sequences added together.
/// Shift triplet `[17,7,16]` was calculated as indicated in Marsaglia's
/// `Xorshift` paper

fastrand_n用 Lemire 的乘法取模替代% n：

📎 tokio-stream/src/stream_map.rs:787-792

pub(crate) fn fastrand_n(&self, n: u32) -> u32 {
    // This is similar to fastrand() % n, but faster.
    // See https://lemire.me/blog/2016/06/27/a-fast-alternative-to-the-modulo-reduction/
    let mul = (self.fastrand() as u64).wrapping_mul(n as u64);
    (mul >> 32) as u32
}

第二，swap_remove後的游標修正。當索引idx的流返回None被移除時，swap_remove會把末尾元素搬到idx。這個被搬來的元素可能已經被輪詢過（如果它的原索引在start之前）。代碼用idx < start && start <= self.entries.len()檢測這種情況，若是則跳過它（idx = idx.wrapping_add(1) % len）。若被移除的是最後一個元素（idx == len），游標回繞到 0。

第三，Poll::Pending的語義。若遍歷一圈沒有任何流就緒，且集合非空，返回Pending。此時所有流的 Waker 都已註冊，任一就緒都會喚醒。

poll_next在poll_next_entry之上補上 key：

📎 tokio-stream/src/stream_map.rs:676-683

fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
    if let Some((idx, val)) = ready!(self.poll_next_entry(cx)) {
        let key = self.entries[idx].0.clone();
        Poll::Ready(Some((key, val)))
    } else {
        Poll::Ready(None)
    }
}

注意ready!宏：若poll_next_entry返回Pending，整個poll_next立即返回Pending。K: Clone約束來自這裡的key.clone()。

設計思考：next_many 的批量語義與取消安全

next_many是StreamMap的批量版本，一次盡可能多地收集就緒元素：

📎 tokio-stream/src/stream_map.rs:581-583

pub async fn next_many(&mut self, buffer: &mut Vec<(K, V::Item)>, limit: usize) -> usize {
    poll_fn(|cx| self.poll_next_many(cx, buffer, limit)).await
}

它的取消安全保證很關鍵：

📎 tokio-stream/src/stream_map.rs:573-578

/// # Cancel safety
///
/// This method is cancel safe. If `next_many` is used as the event in a
/// [`tokio::select!`] statement and some other branch completes first,
/// it is guaranteed that no items were received on any of the underlying
/// streams.

為什麼next_many取消安全？因為它把元素立即 push 進調用方提供的buffer，而不是暫存在內部。若 future 被 drop，已 push 的元素仍在buffer裡，不會丟失。但這也意味著：被 drop 時buffer可能已有部分元素——調用方需要知道這一點。

poll_next_many的循環結構比poll_next_entry複雜，因為它要在一輪內盡可能多地收集：

📎 tokio-stream/src/stream_map.rs:597-666

pub fn poll_next_many(
    &mut self,
    cx: &mut Context<'_>,
    buffer: &mut Vec<(K, V::Item)>,
    limit: usize,
) -> Poll<usize> {
    if limit == 0 || self.entries.is_empty() {
        return Poll::Ready(0);
    }

    let mut added = 0;

    let start = self::rand::thread_rng_n(self.entries.len() as u32) as usize;
    let mut idx = start;

    while added < limit {
        // Indicates whether at least one stream returned a value when polled or not
        let mut should_loop = false;

        for _ in 0..self.entries.len() {
            let (_, stream) = &mut self.entries[idx];

            match Pin::new(stream).poll_next(cx) {
                Poll::Ready(Some(val)) => {
                    added += 1;

                    let key = self.entries[idx].0.clone();
                    buffer.push((key, val));

                    should_loop = true;

                    idx = idx.wrapping_add(1) % self.entries.len();

                    if added == limit {
                        break;
                    }
                }
                Poll::Ready(None) => {
                    // Remove the entry
                    self.entries.swap_remove(idx);

                    // Check if this was the last entry, if so the cursor needs
                    // to wrap
                    if idx == self.entries.len() {
                        idx = 0;
                    } else if idx < start && start <= self.entries.len() {
                        // The stream being swapped into the current index has
                        // already been polled, so skip it.
                        idx = idx.wrapping_add(1) % self.entries.len();
                    }
                }
                Poll::Pending => {
                    idx = idx.wrapping_add(1) % self.entries.len();
                }
            }
        }

        if !should_loop {
            break;
        }
    }

    if added > 0 {
        Poll::Ready(added)
    } else if self.entries.is_empty() {
        Poll::Ready(0)
    } else {
        Poll::Pending
    }
}

外層while added < limit配合內層for構成「多輪掃描」：只要上一輪有流產出過值（should_loop = true），就再掃一輪，直到攢夠limit或一輪無產出。返回值的三種情況精確對應文檔：

📎 tokio-stream/src/stream_map.rs:588-591

/// * `Poll::Pending` if no items are available but the `StreamMap` is not empty.
/// * `Poll::Ready(count)` where `count` is the number of items successfully received and
///   stored in `buffer`. This can be less than, or equal to, `limit`.
/// * `Poll::Ready(0)` if `limit` is set to zero or when the `StreamMap` is empty.

size_hint的實作展示了如何聚合多個流的容量提示：

📎 tokio-stream/src/stream_map.rs:685-701

fn size_hint(&self) -> (usize, Option<usize>) {
    let mut ret: (usize, Option<usize>) = (0, Some(0));

    for (_, stream) in &self.entries {
        let hint = stream.size_hint();

        ret.0 = ret.0.saturating_add(hint.0);

        match (ret.1, hint.1) {
            (Some(a), Some(b)) => ret.1 = a.checked_add(b),
            (Some(_), None) => ret.1 = None,
            _ => {}
        }
    }

    ret
}

與merge_size_hints同樣的模式：下界飽和加，上界檢查加，任一未知則整體未知。

下面用一張流程圖刻畫poll_next_entry的決策路徑：

flowchart TD
    start["poll_next_entry(cx)"] --> rand["start = thread_rng_n(len)"]
    rand --> loop{"遍历 len 次?"}
    loop -->|"未完成"| poll["Pin::new(stream).poll_next(cx)"]
    poll -->|"Ready(Some(val))"| ret_val["返回 Ready(Some((idx, val)))"]
    poll -->|"Ready(None)"| remove["entries.swap_remove(idx)"]
    remove --> wrap{"idx == entries.len()?"}
    wrap -->|"是"| set_zero["idx = 0"]
    wrap -->|"否"| check_swap{"idx < start && start <= len?"}
    check_swap -->|"是"| skip["idx = idx.wrapping_add(1) % len"]
    check_swap -->|"否"| loop
    set_zero --> loop
    skip --> loop
    poll -->|"Pending"| advance["idx = idx.wrapping_add(1) % len"]
    advance --> loop
    loop -->|"遍历完成"| empty{"entries.is_empty()?"}
    empty -->|"是"| ret_none["返回 Ready(None)"]
    empty -->|"否"| ret_pending["返回 Pending"]

---

TaskTracker：用單個 AtomicUsize 編碼全部狀態

直覺模型

優雅關閉需要兩件事：通知任務停工（CancellationToken負責），以及等待任務真正退出（TaskTracker負責）。TaskTracker就像一個「任務計數器 + 關閉開關」的合體：只要還有任務在跑，或者還沒呼叫close，wait()就不會返回。若沒有它，你只能用JoinSet，但JoinSet會累積每個任務的返回值，長期運行的服務會 OOM。

資料結構與記憶體佈局

TaskTracker是一個Arc包裝：

📎 tokio-util/src/task/task_tracker.rs:158-178

pub struct TaskTracker {
    inner: Arc<TaskTrackerInner>,
}

/// Represents a task tracked by a [`TaskTracker`].
#[must_use]
#[derive(Debug)]
pub struct TaskTrackerToken {
    task_tracker: TaskTracker,
}

struct TaskTrackerInner {
    /// Keeps track of the state.
    ///
    /// The lowest bit is whether the task tracker is closed.
    ///
    /// The rest of the bits count the number of tracked tasks.
    state: AtomicUsize,
    /// Used to notify when the last task exits.
    on_last_exit: Notify,
}

這是本章最精妙的記憶體佈局：一個AtomicUsize同時編碼「是否關閉」和「任務計數」。最低位是關閉標誌，其餘位是任務數（因為任務計數每次+2，最低位永遠是 0）。這樣is_closed_and_empty只需一次原子載入：

📎 tokio-util/src/task/task_tracker.rs:216-222

fn is_closed_and_empty(&self) -> bool {
    // If empty and closed bit set, then we are done.
    //
    // The acquire load will synchronize with the release store of any previous call to
    // `set_closed` and `drop_task`.
    self.state.load(Ordering::Acquire) == 1
}

場景驅動 Walkthrough：close 與 drop_task 的競態

考慮一個典型場景：主執行緒呼叫tracker.close()，同時最後一個任務正在退出（TaskTrackerToken::drop呼叫drop_task）。兩者可能並發，必須保證無論誰先，wait()都能被喚醒。

先看set_closed：

📎 tokio-util/src/task/task_tracker.rs:225-249

fn set_closed(&self) -> bool {
    // The AcqRel ordering makes the closed bit behave like a `Mutex<bool>` for synchronization
    // purposes. ...
    let state = self.state.fetch_or(1, Ordering::AcqRel);

    // If there are no tasks, and if it was not already closed:
    if state == 0 {
        self.notify_now();
    }

    (state & 1) == 0
}

fetch_or(1, AcqRel)原子地設置關閉位並返回舊值。若舊值為 0（之前未關閉且無任務），說明「關閉後立即滿足空+關閉」，呼叫notify_now。返回值(state & 1) == 0表示「這次呼叫確實改變了狀態」。

再看drop_task：

📎 tokio-util/src/task/task_tracker.rs:264-271

fn drop_task(&self) {
    let state = self.state.fetch_sub(2, Ordering::Release);

    // If this was the last task and we are closed:
    if state == 3 {
        self.notify_now();
    }
}

fetch_sub(2, Release)減計數。若舊值為 3（二進位11：關閉位 1 + 計數 1），說明「這是最後一個任務且已關閉」，呼叫notify_now。

兩個路徑的競態分析：

• close 先執行：set_closed看到舊值2（計數 1，未關閉），不通知。隨後drop_task看到舊值3，通知。✓
• drop_task 先執行：drop_task看到舊值2（計數 1，未關閉），不通知。隨後set_closed看到舊值0（計數 0，未關閉），通知。✓
• 並發：fetch_or和fetch_sub是原子的，無論交錯順序，總有一個會看到「關閉 + 空」的組合並通知。✓

notify_now裡有一個容易被忽略的Acquire載入：

📎 tokio-util/src/task/task_tracker.rs:274-285

#[cold]
fn notify_now(&self) {
    // Insert an acquire fence. This matters for `drop_task` but doesn't matter for
    // `set_closed` since it already uses AcqRel.
    //
    // This synchronizes with the release store of any other call to `drop_task`, and with the
    // release store in the call to `set_closed`. That ensures that everything that happened
    // before those other calls to `drop_task` or `set_closed` will be visible after this load,
    // and those things will also be visible to anything woken by the call to `notify_waiters`.
    self.state.load(Ordering::Acquire);

    self.on_last_exit.notify_waiters();
}

為什麼drop_task用Release而非AcqRel？因為drop_task的fetch_sub只需要「讓之前的寫對後續讀者可見」（Release 語義），不需要「看到之前其他執行緒的寫」（Acquire 語義）。但notify_now需要 Acquire 來建立 happens-before：確保任務退出前做的所有清理工作，對wait()返回後的程式碼可見。這個load的結果被丟棄，純粹是為了它的記憶體序副作用——這是 Rust 原子操作中「fence 式載入」的典型用法。

設計思考：wait 的 ABA 抵抗與 TrackedFuture 的 drop 語義

wait返回一個TaskTrackerWaitFuture，它內部持有Notified：

📎 tokio-util/src/task/task_tracker.rs:318-327

pub fn wait(&self) -> TaskTrackerWaitFuture<'_> {
    TaskTrackerWaitFuture {
        future: self.inner.on_last_exit.notified(),
        inner: if self.inner.is_closed_and_empty() {
            None
        } else {
            Some(&self.inner)
        },
    }
}

注意inner欄位：若建立時已經「關閉且空」，直接設為None，poll時立即返回Ready。這是快速路徑。

文件特別強調了 ABA 抵抗：

📎 tokio-util/src/task/task_tracker.rs:304-307

/// The `wait` future is resistant against [ABA problems][aba]. That is, if the `TaskTracker`
/// becomes both closed and empty for a short amount of time, then it is guarantee that all
/// `wait` futures that were created before the short time interval will trigger, even if they
/// are not polled during that short time interval.

這個保證來自Notify::notified()的語義：Notifiedfuture 在建立時就註冊了「等待者」身份，即使notify_waiters在它被poll之前呼叫，它也會在首次poll時看到通知。TaskTrackerWaitFuture::poll的實作：

📎 tokio-util/src/task/task_tracker.rs:697-712

fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
    let me = self.project();

    let inner = match me.inner.as_ref() {
        None => return Poll::Ready(()),
        Some(inner) => inner,
    };

    let ready = inner.is_closed_and_empty() || me.future.poll(cx).is_ready();
    if ready {
        *me.inner = None;
        Poll::Ready(())
    } else {
        Poll::Pending
    }
}

每次poll都先檢查is_closed_and_empty()，再poll Notified。這個順序保證：即使Notified因為某種原因沒被喚醒，狀態檢查也能兜底。

TrackedFuture的 drop 語義是TaskTracker與JoinSet的核心差異：

📎 tokio-util/src/task/task_tracker.rs:488-494

/// The task is removed from the collection when it is dropped, not when [`poll`] returns
/// [`Poll::Ready`].

這意味著：即使 future 已經返回Ready，只要TrackedFuture本身還沒被 drop，TaskTracker就認為任務還在。文件解釋了為什麼這個設計重要：

📎 tokio-util/src/task/task_tracker.rs:33-35

/// When a call to [`wait`] returns, it is guaranteed that all tracked tasks have exited and that
/// the destructor of the future has finished running. However, there might be a short amount of
/// time where [`JoinHandle::is_finished`] returns false.

TaskTrackerToken的Drop是計數遞減的觸發點：

📎 tokio-util/src/task/task_tracker.rs:670-672

impl Drop for TaskTrackerToken {
    /// Dropping the token indicates to the [`TaskTracker`] that the task has exited.
    #[inline]
    fn drop(&mut self) {
        self.task_tracker.inner.drop_task();
    }
}

TrackedFuture透過pin_project!把token與future打包，token的 drop 自動觸發計數遞減。spawn_blocking則顯式管理 token：

📎 tokio-util/src/task/task_tracker.rs:452-464

至此，StreamExt 把 poll_next 變成了可組合的迭代器，StreamMap 與 TaskTracker 讓動態任務集合有了歸屬，CancellationToken 則用一棵樹把取消信號傳播到整個任務樹。這三層擴展的共同點是：它們沒有引入新的調度原語，而是把 Waker、Notify 和原子計數這些既有機制重新組合成更高層的抽象。但一個關鍵問題隨之浮現：當這些組合子、任務集合和取消樹在同一個調度器上並發運行時，如何保證某個任務不會因為長時間不讓出而餓死其他任務？下一章將深入 Tokio 的 coop 協作預算機制，看每個任務在一次調度週期內如何消耗預算、耗盡後主動讓出，以及 budget 如何線上程本地存儲中傳遞，從而解決這個經典問題。

上一章我們看到，tokio-stream 與 tokio-util 如何複用底層的 Waker 與排程機制來擴展核心能力。但無論擴展出多少組合子，非同步執行時的核心矛盾始終存在：排程器必須公平地在多個任務之間分配 CPU 時間，而任務本身是非搶佔的——一旦某個 Future 的 poll 開始執行，排程器就無法從外部打斷它。如果一個任務在單次 poll 裡迴圈處理了十萬條訊息，或者在一個 loop 裡反覆 await 一個永遠就緒的 Future，它就會霸佔 worker 執行緒，讓同執行緒上的其他任務永遠得不到輪詢機會。這就是經典的「任務餓死排程器」問題。Tokio 的解法不是搶佔，而是協作：給每個任務一次排程週期內分配有限的預算，資源操作會消耗預算，預算耗盡後任務必須主動讓出。本章深入這套 coop 機制的實現。

12.1 預算的載體：執行緒本地儲存與 Budget 結構

預算必須滿足兩個約束：第一，它要能被任意深度的poll呼叫棧存取，而不必層層傳參；第二，它要能區分「當前是否在 Tokio 執行時內」——在執行時外呼叫block_on時不應受預算約束。Tokio 選擇用執行緒本地儲存（TLS）承載預算，並透過context模組統一管理。

預算的核心類型是coop::Budget。雖然本章原始碼切片未直接給出coop.rs的完整定義，但從worker.rs的使用點可以反推出它的介面契約：

📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:695-795

coop::budget(|| {
    // ... 轮询任务 ...
    task.run();
    // ...
    loop {
        // ...
        if !coop::has_budget_remaining() {
            // 预算耗尽，把 LIFO 任务推回队列
            core.run_queue.push_back_or_overflow(task, ...);
            return ControlFlow::Continue(core);
        }
        // ...
    }
})

這裡出現了三個關鍵 API：coop::budget(closure)建立一個預算作用域，coop::has_budget_remaining()查詢剩餘預算，以及後文會看到的coop::stop()與coop::set()。budget的語義是：進入閉包時把當前執行緒的預算重置為一個滿額值（預設 128），閉包執行期間所有資源操作共享這個額度，閉包退出時恢復外層預算。

Budget在 TLS 中通常以Cell<Option<Budget>>形式存在。Option的外層語義是「當前執行緒是否處於 Tokio 執行時上下文」：None表示不在執行時內（例如執行時外的block_on），此時所有預算檢查都直接放行。

12.2 預算的消耗點：資源操作如何扣減

預算不會憑空消耗，只有資源操作才會扣減它。所謂資源操作，是指那些可能被無限迴圈呼叫的、與外部世界互動的 API——channel 的send/recv、I/O 的讀寫、yield_now等。以mpsc::Sender::reserve為例，它是所有發送路徑的公共入口：

📎 tokio/src/sync/mpsc/bounded.rs:1272-1311

async fn reserve_inner(&self, n: usize) -> Result<(), SendError<()>> {
    crate::trace::async_trace_leaf().await;

    if n > self.max_capacity() {
        return Err(SendError(()));
    }
    // ... WakeReceiverOnDrop guard ...
    let guard = WakeReceiverOnDrop { chan: &self.chan };
    let result = self.chan.semaphore().semaphore.acquire(n).await;
    // ...
}

reserve_inner在真正獲取信號量許可之前，會經過crate::trace::async_trace_leaf()。這個看似只是 tracing 的呼叫，實際上是預算扣減的掛載點之一。async_trace_leaf內部會呼叫coop::poll_proceed一類的函式：如果預算充足，扣減 1 並返回Proceed；如果預算耗盡，則註冊一個「讓出」動作——把當前任務的 Waker 交給排程器，返回Pending，讓任務在這次 poll 中提前結束。

這就是 coop 的精妙之處：預算耗盡不是拋錯，而是把「讓出」偽裝成一次普通的Pending。上層 Future 看到Pending會自然地返回，排程器把任務重新入隊，等下次被排程時預算已重置，任務從上次中斷處繼續。整個過程對業務程式碼完全透明。

yield_now是預算機制最直白的體現，它不消耗預算，而是主動觸發讓出：

📎 tokio/src/task/yield_now.rs:38-60

pub async fn yield_now() {
    let mut yielded = false;
    poll_fn(|cx| {
        ready!(crate::trace::trace_leaf());

        if yielded {
            return Poll::Ready(());
        }

        yielded = true;

        // Don't wake the task immediately, as that would push it right back
        // onto the run queue and it could be polled again before other tasks
        // or the IO/timer driver get a chance to run. Instead, hand the waker
        // to the scheduler, which wakes deferred tasks only after it has run
        // out of ready tasks and polled the driver. When polled from outside
        // a Tokio runtime, the waker is woken immediately.
        context::defer(cx.waker());

        Poll::Pending
    })
    .await
}

注意context::defer(cx.waker())這一行。它沒有直接wake，而是把 Waker 交給排程器的defer 佇列。為什麼？原始碼註解說得很清楚：如果立即喚醒，任務會被立刻推回執行佇列，可能在 I/O/timer 驅動執行之前就被再次輪詢，讓出就失去了意義。defer 佇列的語義是「等當前 worker 把就緒任務跑完、並且輪詢過驅動之後，再喚醒這些任務」。

defer 佇列定義在 worker 的Context中：

📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:247-257

pub(crate) struct Context {
    worker: Arc<Worker>,
    core: RefCell<Option<Box<Core>>>,
    /// Tasks to wake after resource drivers are polled. This is mostly to
    /// handle yielded tasks.
    pub(crate) defer: Defer,
}

defer欄位的註解直接點明它的用途：「mostly to handle yielded tasks」。在 worker 主迴圈中，當本地佇列和竊取都無活可幹時，會檢查 defer 佇列：

📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:613-621

} else {
    // Wait for work
    core = if !self.defer.is_empty() {
        self.park_yield(core)
    } else {
        self.park(core)
    };
    core.stats.start_processing_scheduled_tasks();
}

如果 defer 佇列非空，worker 呼叫park_yield——以 0 逾時 park，這會驅動 I/O 和 timer，然後喚醒 defer 中的任務。這就保證了「讓出」的任務一定是在驅動跑過之後才被重新排程。

12.3 預算作用域的建立與恢復：run_task 與 block_in_place

預算作用域在run_task中建立。每個任務被輪詢時，coop::budget包裹整個輪詢過程：

📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:691-704

// Make the core available to the runtime context
*self.core.borrow_mut() = Some(core);

// Run the task
coop::budget(|| {
    // ...
    task.run();
    // ...
})

coop::budget進入時把 TLS 中的預算設為滿額，退出時恢復。這意味著每個任務每次被輪詢都獲得一份全新的預算。任務內部無論await了多少次資源操作，只要單次poll內消耗超過 128，就會被強制讓出。

但這裡有一個微妙的問題：LIFO slot 中的任務是在同一個budget閉包內被輪詢的。看run_task的迴圈：

📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:709-750

let mut lifo_polls = 0;

// As long as there is budget remaining and a task exists in the
// `lifo_slot`, then keep running.
loop {
    let mut core = match self.core.borrow_mut().take() {
        Some(core) => core,
        None => {
            return ControlFlow::Break(());
        }
    };

    let task = match core.lifo_slot.take() {
        Some(task) => task,
        None => {
            self.reset_lifo_enabled(&mut core);
            core.stats.end_poll();
            return ControlFlow::Continue(core);
        }
    };

    if !coop::has_budget_remaining() {
        core.stats.end_poll();
        // Not enough budget left to run the LIFO task, push it to
        // the back of the queue and return.
        core.run_queue.push_back_or_overflow(task, ...);
        debug_assert!(core.lifo_enabled);
        return ControlFlow::Continue(core);
    }
    // ...
}

關鍵點：LIFO slot 中的任務共享外層任務的預算。註解在run_task開頭就說：「Tasks from the LIFO slot inherit the "parent"'s limits」。這是有意的設計——如果每個 LIFO 任務都重置預算，那麼在 ping-pong 場景（任務 A 喚醒 B，B 又喚醒 A）下，兩個任務會無限互相排程，預算永遠重置，餓死問題依舊。共享預算意味著 A 和 B 加起來最多消耗 128 次資源操作，之後必須讓出。

LIFO slot 本身還有一個獨立的限流器MAX_LIFO_POLLS_PER_TICK：

📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:756-766

// Disable the LIFO slot if we reach our limit
//
// In ping-ping style workloads where task A notifies task B,
// which notifies task A again, continuously prioritizing the
// LIFO slot can cause starvation as these two tasks will
// repeatedly schedule the other. To mitigate this, we limit the
// number of times the LIFO slot is prioritized.
if lifo_polls >= MAX_LIFO_POLLS_PER_TICK {
    core.lifo_enabled = false;
    super::counters::inc_lifo_capped();
}

MAX_LIFO_POLLS_PER_TICK的值是 3：

📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:263-263

/// Value picked out of thin-air. Running the LIFO slot a handful of times
/// seems sufficient to benefit from locality. More than 3 times probably is
/// over-weighting. The value can be tuned in the future with data that shows
/// improvements.
const MAX_LIFO_POLLS_PER_TICK: usize = 3;

這是第二道防線：即使預算還沒耗盡，LIFO slot 連續被優先 3 次後也會被停用，後續任務走普通佇列。預算管的是「資源操作總量」，LIFO 限流管的是「同一對任務互相喚醒的次數」，兩者互補。

預算作用域在block_in_place中有一個重要的例外。block_in_place會把 worker core 移交給另一個執行緒，當前執行緒進入阻塞狀態。阻塞程式碼不受預算約束，所以必須暫停預算：

📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:406-417

if had_entered {
    // Unset the current task's budget. Blocking sections are not
    // constrained by task budgets.
    let _reset = Reset {
        take_core,
        budget: coop::stop(),
    };

    crate::runtime::context::exit_runtime(f)
} else {
    f()
}

coop::stop()回傳當前預算並把它設為None（即「不在執行時內」），Reset的Drop在阻塞結束後恢復：

📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:374-397

impl Drop for Reset {
    fn drop(&mut self) {
        with_current(|maybe_cx| {
            if let Some(cx) = maybe_cx {
                if self.take_core {
                    let core = cx.worker.core.take();
                    // ...
                    *cx_core = core;
                }

                // Reset the task budget as we are re-entering the
                // runtime.
                coop::set(self.budget);
            }
        });
    }
}

coop::set(self.budget)把之前stop()保存的預算恢復回去。這樣，block_in_place內的同步阻塞程式碼不會消耗預算，也不會因為預算耗盡而誤觸發讓出；阻塞結束後，任務帶著原來的剩餘預算繼續執行。

下面這張圖展示了從任務被排程到預算耗盡讓出的完整控制流：

flowchart TD
    start["Context::run 主循环"] --> next["core.next_task()"]
    next --> has_task{"有本地任务?"}
    has_task -->|是| run_task["run_task(task, core)"]
    has_task -->|否| steal["core.steal_work()"]
    steal --> stolen{"窃取到任务?"}
    stolen -->|是| run_task
    stolen -->|否| defer_check{"defer 队列非空?"}
    defer_check -->|是| park_yield["park_yield: 驱动 IO/timer 后唤醒"]
    defer_check -->|否| park["park: 阻塞等待"]
    park_yield --> start
    park --> start

    run_task --> budget["coop::budget 建立满额预算"]
    budget --> poll["task.run() 轮询"]
    poll --> lifo_check{"lifo_slot 有任务?"}
    lifo_check -->|否| done["返回 ControlFlow::Continue"]
    lifo_check -->|是| budget_rem{"coop::has_budget_remaining()?"}
    budget_rem -->|否| push_back["push_back_or_overflow 推回队列"]
    push_back --> done
    budget_rem -->|是| lifo_limit{"lifo_polls >= 3?"}
    lifo_limit -->|是| disable["core.lifo_enabled = false"]
    lifo_limit -->|否| poll_lifo["task.run() 轮询 LIFO 任务"]
    disable --> poll_lifo
    poll_lifo --> lifo_check
    done --> start

圖中可以看到兩條讓出路徑：預算耗盡時把 LIFO 任務推回佇列（push_back_or_overflow），以及 LIFO 連續優先超限時停用 LIFO slot。兩者都回到主迴圈，讓 worker 有機會處理其他任務或驅動。

12.4 設計思考、錯誤恢復與生產踩坑

為什麼用 TLS 而不是顯式傳參？預算檢查點散佈在 channel、I/O、time 等各個模組的深處，如果顯式傳參，每個 API 都要多一個Budget參數，污染整個公共介面。TLS 讓預算對業務程式碼完全透明，代價是每次檢查有一次 TLS 存取開銷。Tokio 用#[thread_local]或平台特定的快速 TLS 來壓低這個開銷。

預算耗盡與取消安全的互動。當預算耗盡導致reserve_inner回傳Pending時，任務可能正處於select!的某個分支中。如果此時另一個分支就緒，select!會取消當前分支——reserve_inner的WakeReceiverOnDropguard 會在 drop 時檢查「信號量已關閉且空閒」並喚醒接收端：

📎 tokio/src/sync/mpsc/bounded.rs:1286-1299

struct WakeReceiverOnDrop<'a, T> {
    chan: &'a chan::Tx<T, Semaphore>,
}

impl<T> Drop for WakeReceiverOnDrop<'_, T> {
    fn drop(&mut self) {
        use chan::Semaphore;

        let semaphore = self.chan.semaphore();
        if semaphore.is_closed() && semaphore.is_idle() {
            self.chan.wake_rx();
        }
    }
}

這個 guard 的存在說明：預算觸發的Pending與真正的「無許可」Pending在取消路徑上必須表現一致，否則接收端可能永遠等不到「channel 已關閉」的通知。

生產踩坑：預算耗盡導致的隱蔽延遲。一個常見現象是：某個任務處理訊息的速度突然變慢，但 CPU 佔用不高。排查時容易懷疑鎖競爭或 I/O，實際可能是任務在單次 poll 內處理了超過 128 條訊息，觸發了預算讓出，每次讓出都要經過一次完整的「推回佇列 → 重新排程 → 驅動輪詢」週期。如果訊息處理本身很快，這個排程開銷可能佔比很高。解決辦法是把大批量處理拆成多個spawn的任務，或者顯式在迴圈中插入yield_now。

預算與block_in_place的邊界。前面看到block_in_place會coop::stop()暫停預算。但要注意：coop::stop()只在had_entered為真時呼叫，也就是確實正在執行時 worker 執行緒上時才暫停。如果block_in_place是在執行時外呼叫的，f()直接執行，預算狀態不變。這個分支判斷在maybe_move_runtime中完成：

📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:424-464

with_current(|maybe_cx| {
    match (
        crate::runtime::context::current_enter_context(),
        maybe_cx.is_some(),
    ) {
        (context::EnterRuntime::Entered { .. }, true) => {
            had_entered = true;
        }
        (
            context::EnterRuntime::Entered {
                allow_block_in_place,
            },
            false,
        ) => {
            if allow_block_in_place {
                had_entered = true;
                return Ok(());
            } else {
                return Err(
                    "can call blocking only when running on the multi-threaded runtime",
                );
            }
        }
        (context::EnterRuntime::NotEntered, true) => {
            return Ok(());
        }
        (context::EnterRuntime::NotEntered, false) => {
            return Ok(());
        }
    }
    // ...
})

四種組合分別對應：worker 執行緒內、block_on的執行緒池入口、嵌套block_in_place、執行時外。只有前兩種需要暫停預算並移交 core。

本章小結

coop 機制用三層設計解決了非搶佔調度器的公平性問題：

1. 預算載體：coop::Budget存在 TLS 中，Option外層區分執行時內外，coop::budget建立滿額作用域，coop::stop/coop::set支援暫停與恢復（block_in_place場景）。

2. 消耗點：資源操作（channel 收發、I/O、yield_now）透過coop::poll_proceed扣減預算，耗盡時把「讓出」偽裝成Pending，對業務透明。

3. 讓出路徑：yield_now透過context::defer把 Waker 交給 defer 佇列，確保在驅動輪詢後才重新調度；LIFO slot 任務共享父任務預算，並有MAX_LIFO_POLLS_PER_TICK = 3的獨立限流。

這套機制的關鍵洞察是：公平性不需要搶佔，只需要讓「無限迴圈」在有限步後自然中斷。預算就是這個「有限步」的度量。

本章思考與自測

coop 機制透過預算約束讓任務在資源操作中主動讓出，從而在非搶佔模型下維持了調度公平。但預算耗盡觸發的 Pending 必須與真正的等待在取消路徑上表現一致，否則 select! 等組合子會破壞狀態一致性。下一章將進入生產踩坑與邊界條件：取消安全、panic 傳播與關閉順序，我們會看到更多這類「看似無關的機制在邊界處耦合」的案例。

上一章我們拆解了 coop 協作預算：每個任務在一次調度週期內只有有限預算，耗盡後必須讓出，從而避免單個任務餓死其他任務。但預算機制只解決了「公平調度」問題，真實生產環境還有一類更隱蔽的陷阱——取消安全、panic 傳播與關閉順序。當 select! 取消一個 Future、當任務 panic 被捕獲、當 Runtime 開始關閉，代碼的邊界行為往往與直覺相悖。本章就從取消安全切入，先看一個被 drop 的 Future 到底丟了什麼。

13.2 panic 傳播：JoinError 如何捕獲崩潰

直覺模型

Tokio 任務 panic 不會讓整個進程崩潰（除非 panic=abort），而是被捕獲、打包成JoinError，通過JoinHandle::await返回。這就像工廠流水線上某個工位出了事故，安全網接住了工人，但產品報廢——你拿到的是「事故報告」而非產品。

數據結構與狀態

JoinHandle<T>的Future::Output是super::Result<T>，即Result<T, JoinError> 📎 tokio/src/runtime/task/join.rs:325。JoinError有兩種形態：panic 和 cancelled。文檔示例展示了 panic 場景：

let join_handle = tokio::spawn(async { panic!("boom"); });
let err = join_handle.await.unwrap_err();
assert!(err.is_panic());

📎 tokio/src/runtime/task/join.rs:121-127

panic 被捕獲的機制在RawTask的 poll 路徑裡：任務 poll 時用catch_unwind包裹，panic 發生後把 payload 存進任務的輸出槽位，標記狀態為 complete，然後喚醒 join waker。JoinHandle::poll通過try_read_output讀到的是Err(JoinError::panic(payload))。

場景驅動的 Walkthrough：panic 傳播鏈

sequenceDiagram
    participant App as 应用任务
    participant Worker as Worker 线程
    participant Raw as RawTask
    participant JH as JoinHandle

    App->>Worker: spawn(async { panic!("boom") })
    Worker->>Raw: poll 任务 Future
    Raw->>Raw: catch_unwind 捕获 panic
    Raw->>Raw: 存储 panic payload 到输出槽
    Raw->>Raw: state 标记 complete
    Raw->>JH: 唤醒 join waker
    JH->>App: await 返回 Err(JoinError::panic)

關鍵點：panic 的 payload 被完整保留，JoinError實現了std::error::Error，可以通過into_panic()取回Box<dyn Any + Send>，再用downcast_ref::<&str>()提取 panic 消息。

設計思考與踩坑

坑 1：JoinHandle的UnwindSafe是手動實現的。

impl<T> UnwindSafe for JoinHandle<T> {}
impl<T> RefUnwindSafe for JoinHandle<T> {}

📎 tokio/src/runtime/task/join.rs:176-181

這是無條件實現，不要求T: UnwindSafe。原因：JoinHandle本身不持有T，T在堆上的任務分配裡，panic 時已經被catch_unwind隔離。所以即使T不是UnwindSafe，JoinHandle也是安全的。

坑 2：panic 不會自動傳播到父任務。如果任務 A spawn 了任務 B，B panic 了，A 不會自動收到通知，除非 A await 了 B 的JoinHandle。如果 A 沒 await，B 的 panic 就被靜默吞掉了。這是生產環境中最隱蔽的 bug 來源之一。

坑 3：spawn_blocking的 panic 同樣被捕獲。阻塞線程池的 worker 也用catch_unwind包裹任務，panic 後線程不會死，而是回到池裡繼續接活。但如果你在阻塞任務裡持有Mutex並在 panic 時沒釋放，會導致鎖中毒——這是std::sync::Mutex的固有行為，Tokio 不介入。

坑 4：Runtime drop 時的 panic。如果任務在 Runtime drop 過程中 panic，catch_unwind仍然生效，但此時 join waker 可能已經失效，panic payload 會被丟棄。這是關閉順序問題的子集，下一節展開。

13.3 關閉順序：阻塞線程與 I/O 資源的清理

直覺模型

Runtime 關閉像餐廳打烊：先讓前台停止接客（停止接受新任務），再等廚房做完手頭的菜（異步任務跑到下一個 yield 點），最後等外包幫工收工（阻塞線程返回）。順序錯了就會出問題——比如先趕走幫工，廚房的菜就永遠做不完。

數據結構與關閉路徑

Runtime的三個字段決定了關閉順序：

pub struct Runtime {
    scheduler: Scheduler,
    handle: Handle,
    blocking_pool: BlockingPool,
}

📎 tokio/src/runtime/runtime.rs:97-106

Drop實現：

impl Drop for Runtime {
    fn drop(&mut self) {
        match &mut self.scheduler {
            Scheduler::CurrentThread(current_thread) => {
                let _guard = context::try_set_current(&self.handle.inner);
                current_thread.shutdown(&self.handle.inner);
            }
            Scheduler::MultiThread(multi_thread) => {
                multi_thread.shutdown(&self.handle.inner);
            }
        }
    }
}

📎 tokio/src/runtime/runtime.rs:506-521

注意：Drop只處理scheduler，沒有顯式處理blocking_pool。blocking_pool的關閉發生在它自己的Drop裡，在Runtime::drop返回後由字段 drop 順序觸發。字段 drop 順序是聲明順序：scheduler → handle → blocking_pool。所以阻塞池是最後關閉的。

但shutdown_timeout是顯式控制順序的：

pub fn shutdown_timeout(mut self, duration: Duration) {
    self.handle.inner.shutdown();
    self.blocking_pool.shutdown(Some(duration));
}

📎 tokio/src/runtime/runtime.rs:457-461

先handle.inner.shutdown()通知調度器和 I/O 驅動停止，再blocking_pool.shutdown(Some(duration))等待阻塞任務，最多等duration。

阻塞池關閉的底層機制

blocking/shutdown.rs用了一個精巧的 oneshot channel：

pub(super) struct Sender {
    _tx: Arc<oneshot::Sender<()>>,
}

pub(super) struct Receiver {
    rx: oneshot::Receiver<()>,
}

📎 tokio/src/runtime/blocking/shutdown.rs:13-19

每個阻塞 worker 持有一個Sender克隆（內部是Arc<oneshot::Sender>）。當所有 worker 退出、所有Sender被 drop 後，Receiver收到通知。wait方法：

pub(crate) fn wait(&mut self, timeout: Option<Duration>) -> bool {
    use crate::runtime::context::try_enter_blocking_region;

    if timeout == Some(Duration::from_nanos(0)) {
        return false;
    }

    let mut e = match try_enter_blocking_region() {
        Some(enter) => enter,
        _ => {
            if std::thread::panicking() {
                return false;
            } else {
                panic!(
                    "Cannot drop a runtime in a context where blocking is not allowed. \
                    This happens when a runtime is dropped from within an asynchronous context."
                );
            }
        }
    };

    if let Some(timeout) = timeout {
        e.block_on_timeout(&mut self.rx, timeout).is_ok()
    } else {
        let _ = e.block_on(&mut self.rx);
        true
    }
}

📎 tokio/src/runtime/blocking/shutdown.rs:37-70

逐步解析：

1. timeout == Some(0)直接返回 false——這是shutdown_background的路徑，不等待。

2. try_enter_blocking_region()嘗試進入阻塞區域。如果當前在異步上下文裡（比如在 async 任務裡 drop Runtime），返回None。

3. 進入失敗時，如果正在 panic，返回 false（不在 panic 中再 panic）；否則 panic 並給出明確錯誤信息。

4. 有 timeout 時用block_on_timeout，超時返回 false；無 timeout 時無限等待。

關閉順序的完整流程

flowchart TD
    start["Runtime::drop 或 shutdown_timeout"] --> sched{"scheduler 类型?"}
    sched -->|CurrentThread| ct["try_set_current + current_thread.shutdown"]
    sched -->|MultiThread| mt["multi_thread.shutdown"]
    ct --> handle_drop["handle 字段 drop"]
    mt --> handle_drop
    handle_drop --> bp_drop["blocking_pool 字段 drop"]
    bp_drop --> bp_wait{"shutdown_timeout 已调用?"}
    bp_wait -->|是| explicit["blocking_pool.shutdown(Some(duration))"]
    bp_wait -->|否| implicit["BlockingPool::drop 默认等待"]
    explicit --> wait_check{"try_enter_blocking_region 成功?"}
    implicit --> wait_check
    wait_check -->|否且在 panic| skip["返回 false 不等待"]
    wait_check -->|否且不在 panic| panic_err["panic: Cannot drop a runtime in async context"]
    wait_check -->|是| block_on["block_on 等待所有 Sender drop"]

設計思考與踩坑

坑 1：在 async 上下文裡 drop Runtime 會 panic。錯誤訊息很明確：「Cannot drop a runtime in a context where blocking is not allowed」📎 tokio/src/runtime/blocking/shutdown.rs:51-54。解決方案是用shutdown_background()，它等價於shutdown_timeout(Duration::from_nanos(0)) 📎 tokio/src/runtime/runtime.rs:494-496，不等待阻塞任務。

坑 2：shutdown_background會洩漏阻塞任務。文件明確警告「this may result in a resource leak (in that any blocking tasks are still running until they return)」📎 tokio/src/runtime/runtime.rs:470-472。阻塞任務會繼續跑直到自然返回，但 Runtime 已經 drop，它們持有的資源可能已經失效。

坑 3：I/O 資源在 Runtime drop 後失效。文件說明「Once the runtime has been dropped, any outstanding I/O resources bound to it will no longer function」📎 tokio/src/runtime/runtime.rs:52-54。is_rt_shutdown_err函式就是用來檢測這種錯誤的📎 tokio/src/runtime/runtime.rs:585-593。

坑 4：Drop預設無限等待。文件指出「TheDrop implementation waits forever for this」📎 tokio/src/runtime/runtime.rs:43-44。如果阻塞任務卡死（比如死迴圈），drop Runtime 會永久掛起。生產環境應該用shutdown_timeout設上限。

13.4 訊號處理與多 Runtime 衝突

直覺模型

Unix 訊號是行程級的，但 Tokio 的Signal是綁定到 Runtime 的。這就像全棟樓共用一個火警鈴，但每個房間都裝了獨立的接收器——第一個裝接收器的人改了鈴的接線方式，後面的人只能共享這個改動。

資料結構與全域狀態

signal_enable是註冊訊號處理器的入口：

fn signal_enable(signal: SignalKind, handle: &Handle) -> io::Result<()> {
    let signal = signal.0;
    if signal <= 0 || signal_hook_registry::FORBIDDEN.contains(&signal) {
        return Err(Error::other(format!(
            "Refusing to register signal {signal}"
        )));
    }

    handle.check_inner()?;

    let globals = globals();
    let siginfo = match globals.storage().get(signal as EventId) {
        Some(slot) => slot,
        None => return Err(io::Error::other("signal too large")),
    };

    siginfo
        .init
        .get_or_init(|| {
            unsafe { signal_hook_registry::register(signal, move || action(globals, signal)) }
                .map(|_| ())
                .map_err(|e| e.raw_os_error())
        })
        .map_err(|e| {
            e.map_or_else(
                || Error::other("registering signal handler failed"),
                || Error::from_raw_os_error,
            )
        })
}

📎 tokio/src/signal/unix.rs:266-296

關鍵點：

1. signal <= 0 || FORBIDDEN.contains(&signal)拒絕非法訊號。

2. handle.check_inner()檢查訊號驅動是否在運行——如果 Runtime 已關閉，這裡會失敗。

3. siginfo.init.get_or_init(...)用OnceLock保證每個訊號只註冊一次 OS handler。get_or_init的閉包呼叫signal_hook_registry::register，這是全域的、行程級的註冊。

4. 註冊的 handler 是action(globals, signal)，它做兩件事：globals.record_event(signal)記錄事件，然後往 pipe 寫一個位元組喚醒驅動📎 tokio/src/signal/unix.rs:252-259。

多 Runtime 衝突的根源

globals()返回的是行程級的全域Globals，OsExtraData裡的UnixStream對也是全域的：

pub(crate) struct OsExtraData {
    sender: UnixStream,
    pub(crate) receiver: UnixStream,
}

📎 tokio/src/signal/unix.rs:61-64

Default實作建立一對UnixStream 📎 tokio/src/signal/unix.rs:61-64。這個 pipe 是全域唯一的，所有 Runtime 的訊號驅動共享它。

問題來了：signal_enable裡handle.check_inner()檢查的是當前 Runtime的訊號驅動。但signal_hook_registry::register註冊的 handler 是行程級的，它寫入的是全域pipe。如果 Runtime A 先註冊了 SIGINT，然後 Runtime B 也註冊 SIGINT，get_or_init會直接返回已有的Ok(())，不會重複註冊。但 Runtime B 的訊號驅動會從全域 pipe 讀資料——兩個 Runtime 會競爭同一個 pipe 的位元組。

場景驅動的 Walkthrough：多 Runtime 訊號競爭

sequenceDiagram
    participant OS as 操作系统
    participant Handler as 全局 signal handler
    participant Pipe as 全局 UnixStream pipe
    participant RtA as Runtime A 信号驱动
    participant RtB as Runtime B 信号驱动

    Note over RtA: signal(SIGINT) 注册
    RtA->>Handler: signal_hook_registry::register(SIGINT, action)
    Note over RtB: signal(SIGINT) 注册
    RtB->>Handler: get_or_init 返回已有 Ok，不重复注册
    OS->>Handler: 投递 SIGINT
    Handler->>Pipe: write(&[1])
    Pipe->>RtA: 可读事件
    Pipe->>RtB: 可读事件
    Note over RtA,RtB: 两个 Runtime 竞争读取，只有一个能读到字节

設計思考與踩坑

坑 1：訊號處理器永不卸載。文件明確警告「Once a signal handler is registered with the process the underlying libc signal handler is never unregistered」📎 tokio/src/signal/unix.rs:379-380。即使Signal實例被 drop，後續訊號仍會被 Tokio 捕獲，預設行為不會恢復📎 tokio/src/signal/unix.rs:338-340。

坑 2：訊號會被合併。文件說明「beforepoll is called, all signal notifications are coalesced into one item returned from poll」📎 tokio/src/signal/unix.rs:312-315。如果你收到 10 個 SIGINT 但只 poll 了一次，只會看到一個事件。這是 Unix 訊號本身的特性（標準訊號不排隊），Tokio 沒有額外合併。

坑 3：多 Runtime 下訊號可能遺失。由於全域 pipe 被多個 Runtime 競爭讀取，一個 Runtime 可能讀走位元組而另一個永遠等不到。生產環境應該只在一個 Runtime 裡處理訊號，或者用signal_hook自己管理。

坑 4：signal函式 panic 條件。文件說明「This function panics if there is no current reactor set, or if thert feature flag is not enabled」📎 tokio/src/signal/unix.rs:398-405。在 Runtime 外呼叫signal()會 panic。

坑 5：recv()的取消安全。文件保證「This method is cancel safe. If you use it as a branch intokio::select! and another branch completes first, then it is guaranteed that no signal is lost」📎 tokio/src/signal/unix.rs:423-427。這是因為訊號事件存在全域的EventInfo裡，recv()只是讀取，不消費底層狀態。

設計思考

本章三個主題共享一個底層模式：狀態的所有權決定了取消/關閉/訊號的安全性。

• JoinHandle取消安全，因為輸出在堆上，handle 只是引用。
• Runtime 關閉順序敏感，因為阻塞池和排程器共享Handle，順序錯了會死鎖或 panic。
• 信號多 Runtime 衝突，因為 handler 和 pipe 是進程級全局狀態，而Signal是 Runtime 級視圖。

理解這個模式後，避坑清單可以歸納為三條原則：

1. 取消安全 = 狀態在 Future 外部。如果 Future 內部有緩衝區，drop 就會丟數據。JoinHandle、Signal::recv、tokio::sync::mpsc::Receiver::recv都滿足這個條件。

2. 關閉順序 = 依賴方向的反序。誰依賴誰，就先關被依賴者。調度器依賴 I/O 驅動，所以先關調度器；阻塞池獨立，最後關。

3. 全局狀態 = 多實例衝突。任何進程級資源（信號 handler、pipe、文件描述符表）在多 Runtime 下都會衝突，要麼限制單 Runtime，要麼用外部同步。

本章小結

本章思考與自測

章末過渡

取消安全、panic 傳播、關閉順序、信號衝突——這四個問題的共同根源是「狀態所有權」在異步邊界上的模糊。Tokio 通過把狀態放在堆上、用引用計數管理生命週期、用catch_unwind隔離 panic、用全局Globals共享信號狀態，給出了工程上可用的答案。但這些答案都有邊界條件，生產環境必須顯式處理。

下一章將進入架構權衡與未來演進：從 io_uring 到可插拔驅動。我們會看到 Tokio 如何在保持 API 穩定的前提下，為新一代 I/O 接口預留擴展空間，以及當前架構中哪些設計決策是歷史包袱、哪些是前瞻佈局。

至此，我們走完了 Tokio 生產環境中最容易踩坑的邊界地帶：取消安全依賴輸出存儲在堆上、try_read_output 的原子性；JoinHandle::drop 不取消任務，abort 才真正取消但對 spawn_blocking 無效；panic 被 catch_unwind 捕獲後打包成 JoinError，不 await 就靜默丟失；Runtime 關閉有嚴格順序，在 async 上下文 drop 會 panic；信號 handler 是進程級全局狀態，註冊後永不卸載。這些規則背後是 Tokio 在正確性與性能之間的反复權衡。下一章我們將跳出具體機制，站在架構高度回顧這些權衡的由來，並展望 io_uring、驅動重構與自定義執行器接口將把 Tokio 帶向何方。

上一章我們梳理了取消安全、panic 傳播、關閉順序與信號衝突這四類生產陷阱，它們看似分散，實則都指向同一個架構問題：狀態所有權在異步邊界上如何被清晰地劃分。而劃分所有權的方式，恰恰由運行時最底層的三個架構決策決定——任務如何被調度、I/O 事件如何被分發、並發正確性如何被驗證。本章不再鑽進某個具體函數的實現細節，而是站到架構高度，回顧 Tokio 在這些決策上的取捨，並沿著官方文檔與源碼中已經埋下的演進線索，看看 io_uring、驅動重構與自定義執行器接口會把 Tokio 帶向何方。讀完本章，你應該能回答一個實踐問題：什麼時候該擴展 Tokio，什麼時候該繞開它。

一、三個歷史權衡：為什麼是現在這個樣子

直覺模型

把 Tokio 想像成一家已經開了十年的餐廳。廚房的排班方式（work-stealing）、傳菜員的獨立編制（I/O 驅動與調度器分離）、以及後廚的衛生檢查制度（loom 並發驗證），都不是開業第一天就設計好的，而是在「客人變多、菜品變複雜」的過程中逐步演化出來的。理解這些演化，才能判斷哪些設計是前瞻佈局、哪些是歷史包袱。

權衡一：work-stealing 而非全局隊列

work-stealing 的取捨是：每個 worker 持有本地隊列，spawn時優先入本地隊列（無鎖、緩存友好），本地空了才去別的 worker 隊列尾部竊取。代價是負載均衡有延遲，且竊取本身需要原子操作與內存屏障。Tokio 選擇後者，是因為現代服務器動輒幾十核，鎖競爭的成本遠高於偶發的竊取開銷。

權衡二：I/O 驅動獨立於調度器

這是本章源碼材料裡最值得玩味的一處。看tokio/src/runtime/io/mod.rs的模塊結構：

📎 tokio/src/runtime/io/mod.rs:5-22

mod driver;
use driver::{Direction, Tick};
pub(crate) use driver::{Driver, Handle, ReadyEvent};

mod registration;
pub(crate) use registration::Registration;

mod registration_set;
use registration_set::RegistrationSet;

mod scheduled_io;
use scheduled_io::ScheduledIo;

mod metrics;
use metrics::IoDriverMetrics;

use crate::util::ptr_expose::PtrExposeDomain;
static EXPOSE_IO: PtrExposeDomain<ScheduledIo> = PtrExposeDomain::new();

注意driver、registration、scheduled_io是三個獨立模塊，且對外只暴露Driver、Handle、ReadyEvent、Registration這幾個類型。ScheduledIo是pub(crate)的——它被PtrExposeDomain包裹，用於在 loom 測試下把裸指針暴露給並發檢查。

權衡三：loom 做並發模型檢驗

tokio/src/loom/mod.rs只有 14 行，卻揭示了 Tokio 並發正確性的驗證策略：

📎 tokio/src/loom/mod.rs:1-14

//! This module abstracts over `loom` and `std::sync` depending on whether we
//! are running tests or not.

#![allow(unused)]

#[cfg(not(all(test, loom)))]
mod std;
#[cfg(not(all(test, loom)))]
pub(crate) use self::std::*;

#[cfg(all(test, loom))]
mod mocked;
#[cfg(all(test, loom))]
pub(crate) use self::mocked::*;

關鍵在#[cfg(all(test, loom))]這個條件：只有同時開啟test和loom兩個 cfg 時，才會用mocked模塊替換std。這意味著生產構建裡根本沒有 loom 的代碼，零運行時開銷。

設計思考

這三個權衡有一個共同特徵：它們都選擇了「更複雜但更可擴展」的方案，並把複雜度限制在內部。work-stealing 的複雜度藏在排程器裡，I/O 驅動的複雜度藏在ScheduledIo裡，loom 的複雜度藏在 cfg 條件裡。對外暴露的 API 始終是spawn、TcpStream::read這些簡單介面。

---

二、驅動重構：從「一個 waker 一個方向」到「任意興趣集」

直覺模型

早期的 Tokio I/O 類型有個硬性限制：async fn read(&mut self)需要&mut self。這就像餐廳只有一個取餐窗口，同一時間只能有一個人排隊——因為 waker 被存在 I/O 資源內部，而不是存在操作對應的 Future 裡。tokio/docs/reactor-refactor.md完整記錄了這個限制的成因與重構方案。

舊架構的痛點

文件開篇就點明了問題：

📎 tokio/docs/reactor-refactor.md:16-20

Currently, I/O types require `&mut self` for `async` functions. The reason for
this is the task's waker is stored in the I/O resource's internal state
(`ScheduledIo`) instead of in the future returned by the `async` function.
Because of this limitation, I/O types limit the number of wakers to one per
direction (a direction is either read-related events or write-related events).

新架構：把 waker 移到 Future 裡

重構的核心思路是「把 waker 從資源狀態移到操作 Future 裡」，從而支援每個操作註冊多個 waker：

📎 tokio/docs/reactor-refactor.md:22-25

Moving the waker from the internal I/O resource's state to the operation's
future enables multiple wakers to be registered per operation. The "intrusive
wake list" strategy used by `Notify` applies to this case, though there are some
concerns unique to the I/O driver.

新的ScheduledIo結構如下：

📎 tokio/docs/reactor-refactor.md:97-134

#[derive(Debug)]
pub(crate) struct ScheduledIo {
    /// Resource's known state packed with other state that must be
    /// atomically updated.
    readiness: AtomicUsize,

    /// Tracks tasks waiting on the resource
    waiters: Mutex<Waiters>,
}

#[derive(Debug)]
struct Waiters {
    // List of intrusive waiters.
    list: LinkedList<Waiter>,

    /// Waiter used by `AsyncRead` implementations.
    reader: Option<Waker>,

    /// Waiter used by `AsyncWrite` implementations.
    writer: Option<Waker>,
}

// This struct is contained by the **future** returned by `readiness()`.
#[derive(Debug)]
struct Waiter {
    /// Intrusive linked-list pointers
    pointers: linked_list::Pointers<Waiter>,

    /// Waker for task waiting on I/O resource
    waiter: Option<Waker>,

    /// Readiness events being waited on. This is
    /// the value passed to `readiness()`
    interest: mio::Ready,

    /// Should not be `Unpin`.
    _p: PhantomPinned,
}

這裡有幾個精妙的設計點值得展開：

第一，readiness是AtomicUsize，waiters是Mutex<Waiters>。為什麼不用一把鎖保護兩者？因為readiness的讀取操作極其頻繁（每次readiness()呼叫都要檢查），而寫操作只在收到 mio 事件時發生。用原子變數讓讀取路徑無鎖，是典型的讀寫分離最佳化。

第二，Waiter是侵入式鏈結串列節點。 pointers: linked_list::Pointers<Waiter>讓Waiter本身成為鏈結串列的一部分，不需要額外分配節點。_p: PhantomPinned明確標記它不可Unpin——因為侵入式鏈結串列的節點位址一旦移動，鏈結串列就斷了。

第三，reader和writer兩個Option<Waker>是給AsyncRead/AsyncWrite用的。文件解釋了原因：

📎 tokio/docs/reactor-refactor.md:210-213

The `AsyncRead` and `AsyncWrite` traits use a "poll" based API. This means that
it is not possible to use an intrusive linked list to track the waker.
Additionally, there is no future associated with the operation which means it is
not possible to cancel interest in the readiness events.

競態條件與 tick 機制

重構中最棘手的問題是競態。文件給了一個具體的死鎖場景：

📎 tokio/docs/reactor-refactor.md:175-175

If care is not taken, if between `mio_socket.read(buf)` returning and
`clear_readiness(event)` is called, a readiness event arrives, the `read()`
function could deadlock. This happens because the readiness event is received,
`clear_readiness()` unsets the readiness event, and on the next iteration,
`readiness().await` will block forever as a new readiness event is not received.

解決方案是引入 tick 機制，把readiness這個AtomicUsize拆成多個位段：

📎 tokio/docs/reactor-refactor.md:199-199

| shutdown | generation |  driver tick | readiness |
|----------+------------+--------------+-----------|
|   1 bit  |   7 bits   +    8 bits    +  16 bits  |

下面這張流程圖刻畫了readiness()與clear_readiness()之間的決策路徑：

flowchart TD
    start["readiness(interest).await"] --> check_ready{"已知 readiness<br/>与 interest 有交集?"}
    check_ready -->|是| ret_event["返回 ReadyEvent<br/>携带当前 tick"]
    check_ready -->|否| wait["注册 Waiter 到<br/>ScheduledIo.waiters"]
    wait --> mio_poll["mio.poll() 收到事件<br/>tick 递增"]
    mio_poll --> notify["遍历 waiters<br/>interest 匹配者唤醒"]
    notify --> ret_event
    ret_event --> do_read["mio_socket.read(buf)"]
    do_read --> read_ok{"read 结果?"}
    read_ok -->|Ok| done["返回 Ok(v)"]
    read_ok -->|WouldBlock| clear["clear_readiness(event)"]
    read_ok -->|其他 Err| err["返回 Err(e)"]
    clear --> tick_match{"event.tick ==<br/>当前 readiness.tick?"}
    tick_match -->|是| clear_ok["清除 readiness 位"]
    tick_match -->|否| skip["跳过清除<br/>保留新事件"]
    clear_ok --> start
    skip --> start

這張圖的關鍵分支在tick_match：如果 tick 不匹配，clear_readiness必須放棄清除，否則會丟掉剛到達的事件，導致下一輪readiness()永久阻塞。

取消興趣與記憶體洩漏

侵入式鏈結串列帶來一個新問題：如果readiness()返回的 Future 被提前 drop，鏈結串列節點必須被摘除。文件明確警告：

📎 tokio/docs/reactor-refactor.md:144-148

The future returned by `readiness()` uses an intrusive linked list to store the
waker with `ScheduledIo`. Because `readiness()` can be called concurrently, many
wakers may be stored simultaneously in the list. If the `readiness()` future is
dropped early, it is essential that the waker is removed from the list. This
prevents leaking memory.

設計思考與生產踩坑

為什麼不用Vec<Waker>而用侵入式鏈結串列？文件在討論&Resource實作時給出了答案：

📎 tokio/docs/reactor-refactor.md:228-233

It is only possible to implement `AsyncRead` and `AsyncWrite` for resource types
themselves and not for `&Resource`. Implementing the traits for `&Resource`
would permit concurrent operations to the resource. Because only a single waker
is stored per direction, any concurrent usage would result in deadlocks. An
alternate implementation would call for a `Vec<Waker>` but this would result in
memory leaks.

生產踩坑點：TcpStream::by_ref()返回的TcpStreamRef持有read_waiter和write_waiter兩個節點：

📎 tokio/docs/reactor-refactor.md:238-244

struct TcpStreamRef<'a> {
    stream: &'a TcpStream,

    // `Waiter` is the node in the intrusive waiter linked-list
    read_waiter: Waiter,
    write_waiter: Waiter,
}

---

三、自訂執行器：TokioContext 與「繞開 Tokio」的邊界

直覺模型

有時你不想用 Tokio 的排程器，只想借它的 I/O 和定時器。這就像你不想在餐廳內用，只想用它的外帶窗口。examples/custom-executor.rs展示了這種「混合模式」：用futures::executor::ThreadPool做排程，用 Tokio 做 I/O。

核心機制：TokioContext

整個例子的關鍵在TokioContext這個包裝型別：

📎 examples/custom-executor.rs:51-54

impl ThreadPool {
    fn spawn(&self, f: impl Future<Output = ()> + Send + 'static) {
        let handle = self.rt.handle().clone();
        self.inner.spawn_ok(TokioContext::new(f, handle));
    }
}

看整個例子的結構：

📎 examples/custom-executor.rs:38-48

static EXECUTOR: Lazy<ThreadPool> = Lazy::new(|| {
    // Spawn tokio runtime on a single background thread
    // enabling IO and timers.
    let rt = tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .unwrap();
    let inner = futures::executor::ThreadPool::builder().create().unwrap();

    ThreadPool { inner, rt }
});

資料流：一次 TcpListener::bind 的跨執行器旅程

sequenceDiagram
    participant App as "应用 (main)"
    participant FE as "futures::ThreadPool"
    participant TC as "TokioContext"
    participant TR as "tokio::Runtime (后台线程)"
    participant IO as "I/O 驱动 (mio)"

    App->>FE: spawn_ok(TokioContext::new(f, handle))
    FE->>TC: poll(cx)
    TC->>TC: enter(handle) 设置线程局部上下文
    TC->>TC: f.poll(cx) 执行 TcpListener::bind
    TC->>TR: 通过 Handle 访问 I/O 驱动
    TR->>IO: Registration::new 注册 fd
    IO-->>TR: 注册完成
    TR-->>TC: 返回 Pending 或 Ready
    TC-->>FE: 返回 poll 结果
    Note over FE,TR: I/O 就绪时，Tokio 驱动唤醒 waker<br/>FE 重新调度该任务

這張時序圖的關鍵在於：任務的 poll 發生在 futures 執行緒池，但 I/O 事件的等待發生在 Tokio 後台執行緒。兩者透過Handle和 waker 連接。

設計思考：何時該繞開 Tokio

1. 如果你需要與已有的執行器生態整合（比如某些框架強制要求futures::executor），用TokioContext是最小侵入的方案。

2. 如果你需要完全控制排程策略（比如即時系統要求確定性排程），Tokio 的 work-stealing 不滿足需求，但它的 I/O 驅動仍然可用。

3. 如果你只是嫌 Tokio 的 API 複雜，那不該繞開——TokioContext引入的跨執行器邊界會帶來新的除錯難度，得不償失。

生產踩坑點：TokioContext模式下，Tokio 執行時的block_on從未被呼叫，意味著Runtime::shutdown的清理邏輯不會自動觸發。你必須在程式退出前顯式 dropRuntime，否則 I/O 驅動的後台執行緒可能不會優雅關閉。

與 io_uring 的關係

📎 tokio/src/runtime/io/mod.rs:1-4

#![cfg_attr(
    not(all(feature = "rt", feature = "net", feature = "io-uring", tokio_unstable)),
    allow(dead_code)
)]

注意feature = "io-uring"和tokio_unstable同時出現。這意味著 io_uring 支援目前是實驗性的，必須同時開啟 unstable 特性才能編譯。allow(dead_code)則說明：當這些特性未開啟時，模組裡的部分程式碼不會被使用，編譯器會警告——用allow壓掉。

---

本章小結

本章從架構高度回顧了 Tokio 的三個核心權衡，並展望了三條演進路徑：

歷史權衡：

• work-stealing 用排程複雜度換取多核擴展性，邊界是任務粒度不能太細；
• I/O 驅動獨立於排程器，讓block_on與多執行緒執行時復用同一套 I/O 實作；
• loom 透過 cfg 條件在生產建置中完全消失，只在測試時窮舉執行緒交錯。

驅動重構（reactor-refactor.md）：

• 把 waker 從ScheduledIo內部移到操作 Future 裡，用侵入式鏈結串列支援多等待者；
• 用AtomicUsize的位段佈局（shutdown/generation/tick/readiness）消解clear_readiness的競態；
• AsyncRead/AsyncWrite因 poll 語意無法用侵入式鏈結串列，保留reader/writer固定槽位作為妥協。

未來演進：

• io_uring 需要「提交-完成」新抽象，目前受tokio_unstable保護；
• TokioContext允許只用 I/O 驅動、不用排程器，但需手動管理 Runtime 生命週期；
• 判斷「擴展還是繞開」的準則：能用現有 API 表達就不碰內部結構。

本章思考與自測

至此，我們完成了從具體陷阱到架構權衡的爬升。回顧全書，從 Future 的惰性求值到調度器的公平性，從取消安全到關閉順序，再到本章的 io_uring 與可插拔驅動，所有討論都圍繞一個核心：在異步邊界上清晰地劃分狀態所有權。Tokio 的架構並非一成不變，io_uring 的零拷貝 I/O、驅動層的解耦、自定義執行器接口的開放，都在推動它向更靈活、更高效的方向演進。當你合上這本書，希望留下的不是一堆 API 用法，而是一套判斷力：知道何時該信任運行時，何時該介入底層，以及如何在生產環境中避開那些會咬人的組合。異步 Rust 的生態仍在快速生長，保持對源碼與官方文檔的追蹤，比記住任何結論都更重要。