Im vorherigen Kapitel haben wir die Verantwortungsgrenzen von Future, Waker und Executor geklärt. Aber eine real nutzbare Laufzeit ist weit mehr als „ein Executor" – sie benötigt auch eine I/O-Ereignisschleife, Timer, einen Blocking-Thread-Pool, und diese Komponenten müssen dieselbe Menge von Handles und denselben Lebenszyklus teilen. Dieses Kapitel verfolgtBuilder::builddie vollständige Assemblierungskette und beantwortet eine Kernfrage:Welche Komponenten befinden sich tatsächlich im Inneren einesRuntime, wie werden sie zusammengesetzt und teilen sich Handles。

Tokios Assemblierungseinstiegspunkt istBuilder. Es selbst ist ein reiner Konfigurationscontainer, alle Felder sind „Absichtserklärungen" und halten keine Laufzeitressourcen. Die tatsächliche Ressourcenerstellung erfolgt beim Aufruf vonbuild().

Intuitives Modell: Builder ist der „Renovierungsplan", Runtime ist das „Haus nach der Übergabe"

Builderist wie ein Renovierungsplan: Man markiert darauf „wie viele Zimmer (worker_threads)", „ob Wasseranschluss (enable_io)", „ob Stromanschluss (enable_time)", „Obergrenze für ausgelagerte Hilfskräfte (max_blocking_threads)". Der Plan selbst erzeugt keine physischen Entitäten. Erst wennbuild()aufgerufen wird, baut das Bautrupp nach Plan und errichtet die „Zimmer" wie Scheduler, Treiber, Thread-Pool tatsächlich und übergibt eineRuntimeInstanz.

Ohne die Ebene vonBuildermüsste der Benutzer jede Komponente manuell new-en, manuell verdrahten, manuell Fehler-Rollback handhaben – jede Reihenfolgefehler würde zu hängenden Handles oder Ressourcenlecks führen.BuilderDer Wert von liegt darin:„Konfiguration" und „Konstruktion" vollständig zu trennen, sodass der Konstruktionsprozess zentral Validierung, Fehlerbereinigung und Handle-Sharing durchführen kann。

Speicherlayout:Builderdie Feldpartitionierung von

BuilderDie Felder von können nach Verantwortlichkeit in vier Gruppen unterteilt werden. Die erste Gruppe istForm und Schalter：kindbestimmt die Scheduler-Form,enable_io / enable_timebestimmt, ob der entsprechende Treiber erstellt wird.

📎 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,
    // ...
}

Die zweite Gruppe istThread-Pool-Parameter：worker_threadsistOption<usize>，Nonebedeutet „beim Build auf Basis der CPU-Kernzahl automatisch erkennen";max_blocking_threadsStandard ist 512.

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

worker_threads: Option<usize>,
max_blocking_threads: usize,

Die dritte Gruppe istCallback-Hooks, alle sindOption<Arc<dyn Fn ...>>. Beachten Sie, dass sieArcstattBoxverwenden, weil diese Callbacks in denConfig。

📎 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>,

KopierenDie vierte Gruppe ist：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,

KopierenKindHier gibt es ein bemerkenswertes Design:Copyist ein

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

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

MultiThreadKopierenrt-multi-threadDie Variante wird durch dasrtFeature gesteuert. Das bedeutet, in einem Build, bei dem nur dasKindFeature aktiviert ist, hatbuild()nur eine Variante, und dasmatchvonwird vom Compiler zu einem einzigen Zweig optimiert –。

Verwendung des Typsystems statt Laufzeitprüfung, um die Codegröße des Multithread-Schedulers zu eliminieren

Builder::newDie Philosophie der Standardwerte: Warum I/O und time standardmäßig deaktiviert sindenable_ioist der gemeinsame Einstiegspunkt für alle Konstruktionen. Es setztenable_timeundfalse。

📎 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()Das

📎 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
}

aufruft. Die Implementierung vonenable_io()offenbart, wie Feature-Gating die Semantik von „alles an" beeinflusst.net、processKopierensignalBeachten Sie, dasstime feature，enable_all()nur aufgerufen wird, wenn

oder dasbuild()Feature aktiviert ist. Wenn der Benutzer nur

build()aktiviert hat, wirdkindden I/O-Treiber nicht öffnen – weil im Kompilat überhaupt kein I/O-Treiber-Code vorhanden ist.

📎 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(),
    }
}

die Verzweigung von

ist der Startpunkt der Assemblierung, es verzweigt nach

build_current_thread_runtimein zwei völlig unterschiedliche Pfade.build_current_thread_runtime_componentsKopierenRuntime。

📎 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,
    ))
}

Pfad eins: Assemblierung von current_threadbuild_current_thread_runtime_componentsselbst ist sehr dünn, es delegiert an

📎 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();

KopierendriverDie eigentliche Assemblierungslogik befindet sich in(driver, driver_handle). Ihre Ausführungsreihenfolge ist entscheidend:?KopierenbuildDer erste Schritt erstelltErrund gibt ein Paar

zurück. Beachten Sie, dass hierspawnerFehler direkt nach oben propagiert – wenn die I/O-Treiber-Initialisierung fehlschlägt (z. B. epoll-Erstellung fehlschlägt), gibt das gesamtespawner

zurück, zu diesem Zeitpunkt wurde der Blocking-Pool noch nicht erstellt, keine Bereinigung erforderlich.

📎 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();

zufällige Verzweigungsreihenfolge);Configwird anCurrentThread::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(),
);

gewährleistet wird.enable_eager_driver_handoffDer vierte Schritt ist der Kern: driver, driver_handle, blocking_spawner, Seeds undfalse。

📎 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,

Schließlich wirdCurrentThread::newdas vonhandlezurückgegebenescheduler::Handle::CurrentThreadinHandle。

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

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

Ok((scheduler, handle, blocking_pool))

Kopie

build_threaded_runtimePfad zwei: Die Assemblierung von multi_thread

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

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

Noneähnelt dem von current_thread, weist jedoch drei wesentliche Unterschiede auf. Der erste Unterschied ist die Bestimmung der Worker-Thread-Anzahl:num_cpus()KopieBuilder::newwird hier zu

aufgelöst. Dies ist der Ort, an dem die „verzögerte automatische Erkennung" greift – die Erkennung erfolgt zur build-Zeit und nicht zur

📎 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();

Der zweite Unterschied liegt in der Kapazitätsberechnung des blocking pool:max_blocking_threads + worker_threadsKopieself.max_blocking_threadsBeachten Sie0。

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

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

) ist höchstwahrscheinlich ein Hinweis auf die „Anzahl reservierter Threads" oder „Anzahl initialer Threads". Dieses Design sorgt dafür, dass die Semantik vonMultiThread::newin beiden Formen konsistent bleibt: Sie beschreibt, „wie viele zusätzliche Blocking-Threads über die Kern-Worker hinaus geöffnet werden können".

📎 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(),
);

ein Tripel statt eines Tupels zurückgibt:launchKopieMultiThread::newDas zusätzlicheist ein „Start-Handle".ist nur für die Konstruktion der Scheduler-Struktur verantwortlich,

📎 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(). Der eigentliche Start erfolgt später:launch.launch()Kopie

bereits bereit

Der Guard stellt sicher, dass sich Worker-Threads im Moment des Starts bereits im korrekten Runtime-Kontext befinden.driver::Driver::newAssemblierungs-FlussdiagrammErrDie folgende Abbildung zeigt die Assemblierungsreihenfolge, die wichtigsten Verzweigungen und die Fehlerpfade beider Pfade zusammen. Beachten Sie, dass bei einem Fehlschlag von

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)"]

zurückgegeben wird, wobei der blocking pool zu diesem Zeitpunkt noch nicht erstellt wurde.HandleKopie

Handle-Sharing:RuntimeWiescheduler、handle、blocking_poolzum „Passierschein" über Komponentengrenzen hinweg wirdhandleNach Abschluss der Assemblierung hält

📎 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,
}

. Davon istArcder gemeinsame Kern. Sein Inneres ist eine Enum:HandleKopieHandleBeachten Sie, dass beide Variantenmatchumschließen. Das bedeutet, dass das Klonen vondriver()：

📎 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()bietet eine einheitliche Zugriffsschnittstelle, die die Formunterschiede im Inneren vonmatch_flavor!kapselt. Zum Beispiel

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

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

verwendet dasdriver()-Makro zur Eliminierung von Wiederholungen:matchKopiematch_flavor!Dieses Makro expandiert zu einemmatchwie dem obigen

. Sein Wert liegt darin: Wenn ein neuer Accessor hinzugefügt wird, der nach Form verteilen muss, genügt eine ZeileHandle, anstatt zweimal diescheduler::Handle-Verzweigung von Hand zu schreiben.

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

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

ist ein dünner Wrapper um das interneHandle:spawnKopieblock_on。spawnDas vom Benutzer erhalteneAutoBoxkann threadübergreifend geklont werden, kann

📎 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_BOXDie Implementierung vonsize_of::<F>()zeigt die Compile-Zeit-Verzweigung von

📎 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;
}

verwendet wird: Bei einer Laufzeitprüfung würde

zweimal monomorphisiert (einmal für, einmal fürdriver -> blocking_pool -> scheduler), was dazu führt, dass für jede gespawnte Future zwei Task-Harnesses generiert werden und sich die Codegröße verdoppelt. Mit einer konstanten Verzweigung behält der Monomorphisierungs-Sammler nur den tatsächlich durchlaufenen Zweig.

Design-Überlegungen: Assemblierungsreihenfolge, Fehlerwiederherstellung und Produktions-Fallstrickelocal_tidReihenfolge ist Vertrag。build_local. Die Assemblierungsreihenfolgebuild_current_thread_local_runtimeist nicht willkürlich. Der Treiber wird zuerst erstellt, da er der einzige Schritt ist, der aufgrund unzureichender OS-Ressourcen fehlschlagen kann und bei einem Fehlschlag keine Bereinigung anderer Komponenten erfordert. blocking_pool kommt nach dem Treiber und vor dem Scheduler, da der Scheduler blocking_spawner benötigt. Wenn die Erstellung von blocking_pool fehlschlägt (was praktisch kaum vorkommt), wird der Treiber durch Drop automatisch bereinigt.

📎 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,
    ))
}

-Zweig von current_threadtidverwendetHandle, wobei die aktuelle Thread-ID übergeben wird:can_spawn_local_on_local_runtimeKopie

📎 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)Die Future von。worker_threadsdarf nur auf ihrem Owner-Thread gepollt werden, und

📎 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
}

Diese Assertion schlägt bereits in der Konfigurationsphase fehl, nicht erst beim Build. Der Vorteil ist eine frühere Fehlerlokalisierung, der Nachteil ist, dass der Benutzer, wenn die Thread-Anzahl aus einem dynamischen Wert der Konfigurationsdatei stammt, sie vor dem Aufruf selbst validieren muss.

Produktions-Fallstrick Zwei:max_blocking_threadsZu klein eingestellt führt zu Hängen. Die Dokumentation warnt ausdrücklich:

📎 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`].

Produktions-Fallstrick Drei:UnhandledPanic::ShutdownRuntimeNur current_thread wird unterstützt。

📎 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
}

Zusammenfassung dieses Kapitels

Dieses Kapitel hatBuilder::builddie vollständige Assemblierungskette nachverfolgt. Kernschlussfolgerungen:

1. Builderist ein reiner Konfigurationscontainer,build()erst erstellt Ressourcen. Die Assemblierungsreihenfolgedriver -> blocking_pool -> schedulerwird durch die Fehlerwiederherstellungsanforderung bestimmt.

2. Der Unterschied zwischen current_thread und multi_thread beschränkt sich nicht auf die Thread-Anzahl: Die Kapazitätsberechnung des blocking pool ist unterschiedlich (max_blocking_threads vs max_blocking_threads + worker_threads), multi_thread hat zusätzlich einenlaunchzweiphasigen Start,enable_eager_driver_handoffwird unter current_thread zwangsweise deaktiviert.

3. Handleist der Kern, der komponentenübergreifend geteilt wird, intern wirdArcverwendet, um form-spezifische Handles zu umschließen, und übermatchodermatch_flavor!Makros einheitlich zugegriffen.

4. AutoBoxverwendet assoziierte Konstanten, um zur Kompilierzeit zu entscheiden, ob das Future geboxt wird, wodurch eine Verdopplung der Codegröße vermieden wird.

5. local_tidistLocalRuntimeder Laufzeitprüfpunkt für Sicherheit.

Im nächsten Kapitel betreten wir den Lebenszyklus von Aufgaben:spawnwie ein Future in eine schedulierbare Entität umgewandelt wird,JoinHandlewie mit der Aufgaben-Zustandsmaschine interagiert wird, und die Zustandsübergänge von Aufgaben zwischenPENDING / RUNNING / COMPLETE.

Gedanken und Selbsttest dieses Kapitels

Im vorherigen Kapitel haben wir die Montage der Runtime abgeschlossen: I/O driver, time driver, blocking pool und Scheduler werden in dieselbeRuntimeInstanz injiziert,Handleund werden zu einem gemeinsamen Handle für den threadübergreifenden Zugriff auf diese Komponenten. Doch die montierte Runtime ist zu diesem Zeitpunkt noch eine leere Hülle – sie besitzt die Engine zum Antreiben von Aufgaben, aber es gibt keine Aufgaben, die angetrieben werden könnten. Die Frage, die dieses Kapitel beantworten will, ist genau: Wenn dutokio::spawn(async { ... })eingibst, was durchläuft dann dieserasyncBlock, um von einem gewöhnlichen Rust-Code-Stück zu einer Entität zu werden, die „vom Scheduler übernommen, geweckt und gejoint werden kann"? Dies ist die erste Halbzeit von „Das Leben einer Aufgabe", wir konzentrieren uns auf die Geburt: Ausgehend vonHandle::spawndurchlaufen wirnew_taskdie Referenzzählungs-Allokation und landen beiCell<T, S>dem Speicherlayout, um schließlich klar zu sehen, wie die Aufgabe in die lokale Warteschlange eines Workers oder in die globale Injektions-Warteschlange eingereiht wird. Die zweite Halbzeit (Kapitel 4) wird dann in die Scheduling-Schleife und den poll/wake-Kreislauf eintreten.

3.1 Future ist keine Aufgabe: Was genau erzeugt ein spawn

Intuitives Modell

Stelle dirFutureals ein „Rezept" vor und eine Aufgabe als „ein Gericht, das gerade in der Küche gekocht wird". Das Rezept selbst ist statisch, kopierbar und hat keinerlei Ausführungszustand; erst wenn die Küche (der Scheduler) entscheidet „jetzt dieses Gericht zubereiten", ihm einen Herd (Worker), eine Bestellnummer (TaskId) und eine Ausgabestation (JoinHandle) zuweist, wird es zu einem „Gericht in Zubereitung". Ohne diese Verpackungsschicht kann der Scheduler nicht wissen, „bis zu welchem Schritt dieses Gericht zubereitet ist", „wer darauf wartet", „wen es nach Fertigstellung benachrichtigen soll" – er sieht nur ein Rezept und kann nichts verwalten.

Datenstruktur und Speicherlayout

Tokio verwendetTask<S>um „eine von der Runtime besessene Aufgabenreferenz" darzustellen, es ist ein transparenter Wrapper umRawTask

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

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

#[repr(transparent)]bedeutet, dassTask<S>undRawTaskim Speicher vollständig identisch sind, ohne zusätzlichen Overhead.PhantomData<S>ist nur eine Typmarkierung zur Kompilierungszeit, die markiert, zu welchem Scheduler-Typ diese Aufgabe gehörtS。

Was tatsächlich den gesamten Zustand der Aufgabe trägt, istCell<T, S>dessen Layout der Grundstein des gesamten Aufgabenmoduls ist:

#[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

Die drei Felder sind nach „heiß-warm-kalt" angeordnet.Headersind heiße Daten (bei jedem Scheduling, bei jedem Zustandsübergang zugegriffen),Coresind warme Daten (beim Poll zugegriffen),Trailersind kalte Daten (nur bei Erstellung und Zerstörung zugegriffen). Der Kommentar schreibt ausdrücklich:Headermuss das erste Feld sein, weil die Aufgabenstruktur gleichzeitig von*mut Cellund*mut Headerreferenziert wird📎 tokio/src/runtime/task/core.rs:37-43。

Noch entscheidender ist die Cache-Line-Ausrichtung.Cellträgt eine lange Reihe von#[cfg_attr(..., repr(align(...)))]die je nach Zielarchitektur die Anzahl der Ausrichtungsbytes wählt: x86_64/aarch64/powerpc64 verwenden 128 Bytes, arm/mips/sparc/hexagon verwenden 32 Bytes, m68k verwendet 16 Bytes, s390x verwendet 256 Bytes, der Rest standardmäßig 64 Bytes📎 tokio/src/runtime/task/core.rs:64-125Der Kommentar erklärt, warum x86_64 128 statt 64 verwenden muss: Seit Intel Sandy Bridge lädt der Spatial Prefetcherpaarweise64-Byte-Cache-Lines, daher muss auf 128 Bytes ausgerichtet werden, um False Sharing zu vermeiden📎 tokio/src/runtime/task/core.rs:45-53。

Headerselbst ist auf 8 Zeigergrößen beschränkt:

#[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

Dieser Test stellt sicher, dassHeader64 Bytes (8 × 8) nicht überschreitet, sodass es auf Architekturen mit 64-Byte-Cache-Lines vollständig in eine Zeile passt.HeaderDie Felder von umfassen:state: State(atomare Statusbits),queue_next: UnsafeCell<Option<NonNull<Header>>>(Verkettungszeiger der Injektions-Warteschlange),vtable: &'static Vtable(Funktionszeigertabelle),owner_id: UnsafeCell<Option<NonZeroU64>>(ID der zugehörigenOwnedTasksListe),scheduled_at: UnsafeCell<ScheduleLatencyInstant>(Scheduling-Latenzmessung)📎 tokio/src/runtime/task/core.rs:169-198。

Core<T, S>hält das Scheduler-Handlescheduler: Sdie Aufgaben-IDtask_id: Idsowie das Kernstückstage: CoreStage<T> 📎 tokio/src/runtime/task/core.rs:148-165。Stageist eine dreizuständige Enum:

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

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

Genau das ist der Schlüssel dafür, dass „Future und Output denselben Speicher wiederverwenden": Während die Aufgabe läuft, hältStage::Runningden Future, nach Abschluss wird er an Ort und Stelle durchStage::Finished(output)ersetzt, nach Entnahme durchJoinHandlewird er zuStage::Consumed。#[repr(C)]Der Kommentar verweist auf ein Miri-Issue, das zeigt, dass dieses Layout harte Anforderungen an die Korrektheit von unsafe-Code stellt📎 tokio/src/runtime/task/core.rs:225-229。

Trailerspeichert kalte Daten:owned: linked_list::Pointers<Header>（OwnedTasksVerkettungszeiger),waker: UnsafeCell<Option<Waker>>(Consumer-Waker, der auf die Fertigstellung der Aufgabe wartet),hooks: TaskHarnessScheduleHooks 📎 tokio/src/runtime/task/core.rs:205-213。

Step-by-Step: Von spawn bis zur Einreihung

Wir versetzen uns in ein konkretes Szenario: In einer multi_thread-Runtime führt Worker-Thread Atokio::spawn(async { 42 })。

aus new_taskErster Schritt: Die drei Teile der Aufgabe konstruieren.

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

KopierenRawTask::new::<T, S>Es ruftCellauf, umrawzu allokieren, und leitet dann aus demselbenTaskZeiger drei Referenzen ab:OwnedTasks）、Notified(Owned-Referenz, wird normalerweise sofort inJoinHandle(Benachrichtigungsreferenz, wird dem Scheduler übergeben),📎 tokio/src/runtime/task/mod.rs:347-363. Beachten Sie, dass alle drei dasselberawteilen, wobei jeder eine Referenzzählung hält.

Zweiter Schritt: Allokieren vonCellund Schreiben des Anfangszustands. Cell::newAllokieren der gesamten Struktur auf dem Heap:

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

vtablewird vonraw::vtable::<T, S>()generiert und ist eine auf konkreteTundSmonomorphisierte Funktionstabelle📎 tokio/src/runtime/task/core.rs:260. Das Future wird direkt inStage::Runningverschoben, ohne zusätzliches Boxing.

Dritter Schritt: Debug-Assertion zur Layout-Verifikation.Unterdebug_assertions,Cell::newwird diecheck-Funktion aufgerufen, die mitHeader::get_trailer、Header::get_scheduler、Header::get_id_ptrund anderen auf vtable-Offsets basierenden Zeigerarithmetiken einzeln assertiert, dass „die über den Header zurückermittelte Feldadresse" mit „der tatsächlichen Feldadresse" übereinstimmt📎 tokio/src/runtime/task/core.rs:280-321. Dies ist eine Laufzeit-Selbstprüfung der Korrektheit der vtable-Offsets.

Vierter Schritt: Zustellung an den Scheduler.Der Scheduler erhältNotified<S>und ruftSchedule::schedule 📎 tokio/src/runtime/task/mod.rs:315auf. Unter multi_thread geht dies überpush_back_or_overflow, wobei die Aufgabe in die lokale Warteschlange des aktuellen Workers eingereiht wird; bei voller Warteschlange wird in die Injection-Queue überlaufen.

Die folgende Abbildung skizziert den Kontrollfluss und die Verzweigungen vonnew_taskbis zur Einreihung:

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

Diese Abbildung offenbart einige Schlüsselverzweigungen: Debug-Assertions wirken nur in Debug-Builds; bei voller lokaler Warteschlange wird nicht direkt überlaufen, sondern zuerst geprüft, ob es nebenläufige Stealer gibt (steal != real), und falls ja, wird nur die aktuelle Aufgabe in die Injection-Queue eingereiht, da der von Stealern freigegebene Platz bald verfügbar ist.

Designüberlegung: Warum drei Referenzen statt einer

new_taskgibt drei Referenzen zurück, nicht eine. Dies ist der Kern des Referenzzählungsdesigns:Taskrepräsentiert „die Laufzeit besitzt diese Aufgabe",Notifiedrepräsentiert „diese Aufgabe wurde benachrichtigt, wartet auf Scheduling",JoinHandlerepräsentiert „jemand interessiert sich für ihr Ergebnis". Die drei haben unabhängige Lebensdauern——JoinHandlekann gedroppt werden (Aufgabe läuft weiter, Ergebnis wird verworfen),Notifiedverschwindet nach poll,Taskwird freigegeben, nachdem die Aufgabe abgeschlossen und ausOwnedTasksentfernt wurde. Mit nur einer Referenz ließe sich der Zustand „Aufgabe läuft noch, aber niemand joint" nicht ausdrücken.

UnownedTaskist eine weitere wichtige Verzweigung: Sie hältzweiReferenzzählungen, für Blocking-Aufgaben (nicht inOwnedTasks）📎 tokio/src/runtime/task/mod.rs:286-295。unownedgespeichert). Diemem::forget(task)-Funktion führt übermem::forget(notified)undUnownedTask 📎 tokio/src/runtime/task/mod.rs:388-397die beiden Referenzen inOwnedTaskszusammen. Die Designmotivation für „zwei Referenzen" ist: Blocking-Aufgaben haben keine

3.2 Statusbits: Wie ein usize den gesamten Lebenszyklus einer Aufgabe kodiert

Intuitives Modell

Stellen Sie sich den Aufgabenstatus als einen „Gesundheitsbericht" mit mehreren unabhängigen Kontrollkästchen vor: Wird gerade gepollt, ist abgeschlossen, wurde benachrichtigt, wurde abgebrochen, hat jemand gejoint. Tokio verwendet nicht mehrere boolesche Felder, sondern presst diese Markierungsbits ineinAtomicUsize. So benötigt jeder Statusübergang nur ein CAS statt mehrerer Lockings. Ohne dieses Design würden Aufgabenstatusübergänge zu verschachtelten mehreren Locks werden, was Deadlock-Risiko und Overhead stark erhöhen würde.

Bitfeld-Layout

StateDas Bitfeld von📎 tokio/src/runtime/task/mod.rs:32-53：

• RUNNINGist in der Moduldokumentation vollständig definiert: ob die Aufgabe gerade gepollt oder abgebrochen wird. 📎 tokio/src/runtime/task/mod.rs:37-38。
• COMPLETEDieses Bit dient gleichzeitig als Lock der AufgabeRUNNING: Das Future ist vollständig abgeschlossen und gedroppt. Einmal gesetzt, wird es nie gelöscht und nie gleichzeitig mit📎 tokio/src/runtime/task/mod.rs:40-41。
• NOTIFIEDgesetztNotified: ob derzeit ein📎 tokio/src/runtime/task/mod.rs:43。
• CANCELLED-Objekt existiert📎 tokio/src/runtime/task/mod.rs:45-46。
• JOIN_INTEREST: Die Aufgabe sollte so schnell wie möglich abgebrochen werdenJoinHandle 📎 tokio/src/runtime/task/mod.rs:48。
• JOIN_WAKER: existiert📎 tokio/src/runtime/task/mod.rs:50-51。

: als Zugriffskontrollbit für den join-handle-waker📎 tokio/src/runtime/task/mod.rs:53。

RUNNINGDie restlichen Bits dienen der ReferenzzählungRUNNINGDass das📎 tokio/src/runtime/task/mod.rs:130-133-Bit als Lock dient, verdient Ausführung. Der Safety-Abschnitt der Moduldokumentation stellt fest: Jeder mutierende Zugriff auf das Future muss nach Erlangen des Locks durch Modifikation desRUNNING-Bits erfolgen, um exklusiven Zugriff zu gewährleisten

. Das bedeutet, beim Pollen einer Aufgabe setzt der Thread zuerst per CAS

JOIN_WAKER, und bei Erfolg hat er exklusiven Zugriff auf das Future; bei Fehlschlag pollt ein anderer Thread, und dieser Poll kehrt direkt zurück. Dies vereint „Poll-Mutex" und „Statusübergang" in einer einzigen atomaren Operation und vermeidet ein separates Mutex.wakerZugriffskontrollprotokoll für JOIN_WAKERTrailerDas-Bit ist der raffinierteste Teil der gesamten Zustandsmaschine. Es löst das Problem:DasJoinHandle-Feld (in) wird von zwei Threads nebenläufig zugegriffen——die Laufzeitliest📎 tokio/src/runtime/task/mod.rs:75-120：

1. JOIN_WAKERes beim Abschluss der Aufgabe, um den Joiner zu wecken,

schreibtJoinHandlees beim Pollen, um den Waker zu registrieren. Die Moduldokumentation gibt 7 Regeln an

ist initial 0.JoinHandle2. Wenn 0,

hat exklusiven (mutierenden) Zugriff auf das waker-Feld.COMPLETE3. Wenn 1,

5. JoinHandlehat nur gemeinsamen (nur lesenden) Zugriff.JOIN_WAKER4. Wenn 1 undJOIN_WAKER1 ist, hat die Laufzeit gemeinsamen (nur lesenden) Zugriff auf das waker-Feld.

6. JoinHandleUm waker zu schreiben, muss man: (i) erfolgreichCOMPLETEauf 0 setzen, um exklusiven Zugriff zu erlangen, (ii) waker schreiben, (iii) erfolgreichJOIN_WAKERauf 1 setzen.COMPLETEdarf

nur ändern, wennJOIN_INTEREST0 ist; die Laufzeit darf nur ändern, wennCOMPLETE1 ist.

7. WennCOMPLETE0 ist und📎 tokio/src/runtime/task/mod.rs:110-1201 ist, hat die Laufzeit exklusiven Zugriff auf das waker-Feld (zum Droppen des wakers).

Regel 6 impliziert eine Race-Condition: Schritt (i) oder (iii) kann fehlschlagen. Wenn (i) fehlschlägt, wird das Schreiben des wakers aufgegeben; wenn (iii) fehlschlägt (ein anderer Thread hat in der Zwischenzeit

Taskgesetzt), wird das waker-Feld geleertUnownedTaskden Drop zweimal dekrementiert:

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_decgibt zurücktruezeigt an, dass dies die letzte Referenz ist, und erst dann wird tatsächlichCellSpeicher freigegeben.ref_dec_twiceistUnownedTaskder direkte Ausdruck dafür, dass zwei Zähler gehalten werden.

Designüberlegung: Warum Statusbits und Referenzzähler ein gemeinsames Atom verwenden

3.3 JoinHandle: Wie Ergebnisse über Task-Grenzen hinweg zurückgegeben werden

Intuitives Modell

JoinHandleist wie der „Abholschein“, den Ihnen ein Restaurant gibt. Wenn der Task (die Küche) fertig ist, wird das Gericht (output) an die Ausgabe (output) gestelltStage::Finishedund dann Ihr Abholsignal (waker) ausgelöst. Sie kommen mit dem Schein zum Abholen; der Schein selbst enthält nicht das Gericht, sondern ist nur ein Zeiger auf die Ausgabe. Wenn Sie den Schein verlieren (dropJoinHandle), wird das Gericht direkt weggeworfen (output wird gedroppt), aber die Küche stellt deshalb nicht die Arbeit ein.

Datenstruktur

JoinHandle<T>ist ebenfalls eine transparente Verpackung vonRawTask:

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

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

PhantomData<T>markiert den Ausgabetyp.JoinHandle<T>ist erst beiT: SendSend/Sync 📎 tokio/src/runtime/task/join.rs:169-170, was garantiert, dass Nicht-Send-Ausgaben nicht über Threads hinweg verschoben werden.

Schritt für Schritt: await auf einem JoinHandle

JoinHandleimplementiertFuture, dessenpollder Kern der Ergebnisrückgabe ist:

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

Beachten Sie einige Details:trace_leafdient der Tracing-Instrumentierung;coop::poll_proceedverbraucht das Kooperationsbudget (ausführlich in Kapitel 12);try_read_outputlöscht Generics über die vtable, legt den Rückgabewert auf den Stack und übergibt ihn mit*mut ()an📎 tokio/src/runtime/task/join.rs:327-354. Diese Technik, den Rückgabewert auf den Stack zu legen, ist nötig, weil vtable-Funktionen den Rückgabetyp nicht generisch machen könnenTund nur über einen Rohzeiger zurückschreiben können.

Eigentumsübertragung des Ergebnisses

Der Abschnitt „Non-Send output“ der Moduldokumentation beschreibt präzise die Eigentumsregeln des Ergebnisses📎 tokio/src/runtime/task/mod.rs:151-170：

• Wenn der Task abgeschlossen ist, wird output inStagegelegt, dann wird die Umwandlung „COMPLETE setzen“ ausgeführt und der aktuelleJOIN_INTEREST-Wert gelesen.
• WennJOIN_INTEREST0 ist (keinJoinHandle), wird output sofort gedroppt📎 tokio/src/runtime/task/mod.rs:157-158。
• WennJOIN_INTEREST1 ist,JoinHandlefür die Bereinigung von output verantwortlich📎 tokio/src/runtime/task/mod.rs:160-161。

Für Nicht-Send-output gibt die Dokumentation eine dreistufige Argumentation: output wird auf dem Thread erzeugt, der das Future pollt;JoinHandle<Output>ist ebenfalls Nicht-Send, wenn Output Nicht-Send ist, also liegt es auch auf dem Spawn-Thread; daher wirdJoinHandlebeim Entnehmen oder Droppen von output nicht über Threads hinweg verschoben📎 tokio/src/runtime/task/mod.rs:164-170。

Drop von JoinHandle: zwei Pfade, schnell und langsam

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_fastversucht, mit einem CAS „JOIN_INTEREST-Bit löschen + Referenzzähler dekrementieren“ abzuschließen. Wenn das fehlschlägt (z. B. der Task schließt gerade ab und das Statusbit ist belegt), wird der langsame Pfad vondrop_join_handle_slowgenommen. Dies ist das typische Muster „optimistischer schneller Pfad + pessimistischer langsamer Pfad“.

Designüberlegung: Warum JoinHandle output nicht direkt hält

3.4 Lokale Warteschlange: Produzenten-Konsumenten-Struktur für Work-Stealing

Intuitives Modell

Jeder Worker hat eine „private To-do-Liste“ (lokale Warteschlange) mit Kapazität 256. Der Worker selbst entnimmt Aufgaben vomKopf(LIFO, nutzt Cache-Lokalität), andere Worker stehlen Aufgaben vomEnde(FIFO, nehmen die ältesten und wahrscheinlich bereits abgeschlossenen Aufgaben). Ohne lokale Warteschlange würden alle Aufgaben in der globalen Warteschlange liegen, jede Aufgabenentnahme müsste um das globale Lock konkurrieren, und die Mehrkern-Skalierbarkeit würde zusammenbrechen.

Speicherlayout: Trennung von head und 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

headistAtomicUnsignedLong(64 Bit, falls die Plattform u64 unterstützt),tailistAtomicUnsignedShort(32 Bit). Der Kommentar erklärt, warum die Indizes breiter als eigentlich nötig sind: zur ABA-Milderung und zur Unterscheidung zwischen „voll“ und „leer“ des Puffers📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:37-49。

headpackt internzwei UnsignedShort：Das niedrige Bit ist der „echte Kopf“ (real head), das hohe Bit ist die „erste Position, die der Dieb gerade verarbeitet“ (steal head). Wenn beide gleich sind, gibt es keinen aktiven Dieb📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:39-49. Diese Doppelwert-Packung ist der Kerntrick der work-stealing Queue: Der Dieb aktualisiert zuerst per CAS den steal-Wert, um eine Charge von Aufgaben zu „beanspruchen“, und nach Abschluss zieht er den steal-Wert auf den real-Wert nach, was das Ende des Stealens anzeigt.

LOCAL_QUEUE_CAPACITYist unter Nicht-loom 256, unter loom auf 4 reduziert, um mehr Randfälle zu testen📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:62-69。MASK = LOCAL_QUEUE_CAPACITY - 1, verwendet für den Ringpuffer-Index📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:71。

Schritt für Schritt: Die vollständigen Verzweigungen von push_back_or_overflow

Dies ist die komplexeste Funktion der lokalen Queue; wir analysieren sie Zweig für Zweig:

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

Drei Verzweigungen:

1. Kapazität vorhanden（tail - steal < CAPACITY）：break tail, nach Verlassen der Schleife wirdpush_back_finishin den Puffer geschrieben.

2. Keine Kapazität, aber gleichzeitige Diebe（steal != real): Der Dieb schafft Platz, also wird nur die aktuelle Aufgabe in die Injektions-Queue geschoben und sofort zurückgekehrt📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:204-208。

3. Keine Kapazität und keine Diebe: Aufruf vonpush_overflow, um die hintere Hälfte der Aufgaben in die Injektions-Queue überlaufen zu lassen📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:209-219. Wenn CAS fehlschlägt (gegen einen gleichzeitigen Dieb verliert),push_overflowgibtErr(task)zurück, Schleife wiederholt.

push_back_finishschreibt die Aufgabe und aktualisiert 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

ReleaseDie Reihenfolge garantiert, dass die geschriebene Aufgabe für Diebe sichtbar ist.

push_overflow: Warum die hintere Hälfte überlaufen lassen

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

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

Beim Überlauf werden 128 Aufgaben entnommen. Der Kommentar erklärt ausführlich, warumdie hintere Hälfteund nicht die vordere Hälfte entnommen wird📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:295-306: Beim Entnehmen von Aufgaben aus der Injektions-Queue werden sie immer im vorderen Teil platziert. Wenn eine Aufgabe also im hinteren Teil liegt, kann man sicher sein, dass sie nicht gerade aus der Injektions-Queue entnommen wurde. Dies garantiert, dass „eine aus der Injektions-Queue entnommene Aufgabe nicht sofort wieder in die Injektions-Queue zurückgelegt wird“ (zumindest bevor sie einmal gepollt wurde).

CAS beansprucht die hintere Hälfte:

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

Aktualisiertheadvon(head, head)auf(tail, tail), d. h. steal und real werden gleichzeitig auf tail vorgerückt und alle Aufgaben beansprucht. Nach Erfolg wird tail auftail + NUM_TASKS_TAKENzurückgesetzt, was anzeigt, dass die vordere Hälfte in der lokalen Queue verbleibt📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:314-316。

pop und steal_into: Die zwei Pfade zum Abholen von Aufgaben

popist, wenn der Worker selbst eine Aufgabe abholt (vom Kopf, 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

Kritische Verzweigung: Wennsteal == real(kein Dieb), werden beide gleichzeitig vorgerückt; andernfalls wird nur real vorgerückt und steal unverändert gelassen📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:377-384。assert_ne!(steal, next_real)stellt sicher, dass real nicht auf die Position von steal vorgerückt wird, da sonst der Beanspruchungszustand des Diebes zerstört würde.

steal_intoist der Steal-Pfad; zuerst wird geprüft, ob die Ziel-Queue genügend Platz hat:

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

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

Wenn die Ziel-Queue mehr als halb voll ist, wird nicht gestohlen, um zu vermeiden, dass nach dem Stehlen sofort wieder überlaufen wird.

steal_into2ist der Kern des Stealens und berechnet die Anzahl der zu stehlenden Aufgaben:

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

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

Die Hälfte stehlen (aufgerundet). Dann per CAS den steal-Wert von head aktualisieren, um zu beanspruchen:

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

Beachten Sie, dass hier nur der real-Wert aktualisiert wird (pack(src_head_steal, steal_to)steal bleibt unverändert), wodurch real aufsteal_tovorgerückt wird. Dies zeigt an: „Diese Aufgaben wurden beansprucht, andere Diebe dürfen sie nicht mehr anfassen.“ Nach Abschluss des Stealens wird steal auf real nachgezogen:

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

Das folgende Sequenzdiagramm stellt die gleichzeitige Interaktion der drei Parteien „Produzent pusht, Konsument poppt, Dieb stiehlt“ dar:

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"

Designüberlegung: Warum die lokale Queue LIFO ist und das Stehlen FIFO

Damit hat die Aufgabe ihre Verwandlung von Future zu einer planbaren Entität abgeschlossen: Sie hat einen Referenzzähler erhalten, wurde in dasCellSpeicherlayout eingeordnet und erfolgreich an die lokale Queue des Workers oder die globale Injektions-Queue zugestellt. Aber die Aufgabe in die Queue zu legen ist nur der Anfang; was sie wirklich zum Laufen bringt, ist die Scheduling-Schleife des Worker-Threads. Im nächsten Kapitel betreten wir die zweite Hälfte von „Das Leben einer Aufgabe“ und verfolgen, wie der Worker Aufgaben aus der Queue holt,Future::pollaufruft und bei Rückgabe vonPendingdurchWakereine Weckung registriert, was schließlichscheduleerneut in die Queue einreiht – der vollständige Aufrufpfad des geschlossenen Kreises „Wecken → Einreihen → erneut pollen“ sowie die work-stealing-Strategie und die LIFO-Slot-Optimierung werden dort enthüllt.

Im vorherigen Kapitel haben wir die Aufgabe in dieLocalQueue oder die globale Injektions-Queue geschickt. Aber die Queue ist nur eine „To-do-Liste“; was die Aufgabe wirklich zum Laufen bringt, ist die niemals endende Schleife im Worker-Thread. In diesem Kapitel verfolgen wirContext::run– sie ist das Herz des gesamten Multithread-Schedulers.

Zuerst eine Intuition: Der Worker-Thread ist wie ein Koch, vor sich einen Stapel eigener Bestellungen (run_queue), daneben ein öffentliches Bestellregal (inject). Der Koch schaut zuerst auf die ihm am nächsten liegende Bestellung (lifo_slot), wenn nicht, nimmt er von seinem eigenen Stapel, wenn auch dort nichts ist, greift er sich eine Handvoll vom öffentlichen Regal, und wenn das auch nicht funktioniert, stiehlt er ein paar Blätter vom Stapel eines anderen Kochs. Erst wenn alles leer ist, geht er sich ausruhen, aber beim Ausruhen bleiben seine Ohren gespitzt – sobald eine Bestellung eingeht, wacht er sofort auf.

Ohne diese Schleife würde eine Aufgabe nach dem Einreihen in die Warteschlange für immer dort liegen,Future::pollniemals aufgerufen werden, und die gesamte Laufzeit wäre nur ein Haufen toter Daten.

Speicherlayout und Zustandsfelder des Core

Der gesamte veränderliche Zustand des Workers ist inCoreuntergebracht, der vonBoxauf dem Heap allokiert wird und überAtomicCell<Core>zwischenWorkerund thread-lokalemContextübergeben wird.

CoreDie Schlüsselfelder von📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:113-167：

• tick: u32sind folgende: Bei jeder Schleifeniteration inkrementiert, um periodisch Wartung (maintenance) und globale Warteschlangenprüfung auszulösen.
• lifo_slot: Option<Notified>：LIFO-Slot, dies ist das raffinierteste Design dieses Kapitels. Wenn ein Worker selbst eine Aufgabe einplant, geht sie nicht inrun_queue, sondern wird in diesen Slot gelegt, und beim nächsten Abrufen einer Aufgabewirdbevorzugt von hier genommen.
• lifo_enabled: bool: Der Schalter für den LIFO-Slot, um Hunger in Ping-Pong-Szenarien zu verhindern.
• run_queue: queue::Local<Arc<Handle>>: Lokale Warteschlange, die im vorherigen Kapitel analysierteLocal-Struktur.
• is_searching: bool: Ob der Worker gerade nach stehlbaren Aufgaben sucht.
• is_shutdown: bool / is_traced: bool: Shutdown- und Tracing-Flags.
• park: Option<Parker>: Parker, mitOptionumhüllt, um ihn unter dem Borrow-Checker bequem herausnehmen/zurücklegen zu können.
• global_queue_interval: u32: Wie oft die globale Warteschlange überprüft wird.
• rand: FastRand: Schneller Zufallszahlengenerator, um den Startpunkt für das Stehlen zufällig zu wählen.

prüft, mit dem Effekt „die zuletzt eingeplante Aufgabe läuft als nächste" (LIFO). Dies dient der Verbesserung der Lokalität, ist besonders effektiv für Message-Passing-Muster und kann die Latenz senken.

Warum kann LIFO die Latenz senken? Betrachten Sie ein typisches Message-Passing-Szenario: Aufgabe A weckt nach der Verarbeitung einer Nachricht Aufgabe B, B weckt nach der Verarbeitung wieder A. Wenn B sofort läuft, nachdem A es geweckt hat, befinden sich die von B benötigten Daten wahrscheinlich noch im CPU-Cache (da A sie gerade berührt hat). Wenn B ans Ende der Warteschlange gestellt wird und darauf wartet, dass Dutzende Aufgaben davor abgearbeitet werden, ist der Cache längst überschrieben.MAX_LIFO_POLLS_PER_TICK = 3Aber LIFO birgt ein Hunger-Risiko. Der Quellcode verwendet📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:263-263, um

zu begrenzen: Pro Tick wird der LIFO-Slot höchstens 3 Mal bevorzugt, danach wird er deaktiviert, damit andere Aufgaben eine Chance zur Ausführung bekommen.

Walkthrough der Hauptschleife: Ein vollständiger Scheduling-ZyklusparkWir versetzen uns in ein konkretes Szenario: Worker 0 ist gerade ausrun_queueaufgewacht,lifo_slotenthält 5 Aufgaben,

enthält 1 Aufgabe, die globale Warteschlange enthält 3 Aufgaben.Context::run 📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:570-642Der Einstieg in die Hauptschleife istlifo_enabled. Zuerst wirdblock_in_placezurückgesetzt (da der Core möglicherweise von📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:571-573gestohlen wurde und der Zustand zurückgesetzt werden muss),while !core.is_shutdowndann wird in die

-Schleife eingetreten.

Jede Schleifeniteration erledigt vier Dinge: core.tick()Erster Schritt: Tick und Wartung.📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:587inkrementiert den Zählerself.maintenance(core). Dann prüfttick % event_interval == 0park_yield, und falls ja, wird📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:809-826。

mit 0-Timeout aufgerufen, um I/O und Timer anzutreiben. core.next_task(&self.worker)Zweiter Schritt: Aufgabe abrufen.📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1090-1156ist die zentrale Aufgabenabruflogik

• . Sie hat zwei Pfade:tick % global_queue_interval == 0Wenn,wird📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1091-1098bevorzugt aus der globalen Warteschlange abgerufen, und wenn das fehlschlägt, wird die lokale
• abgerufen. Dies dient dazu, zu verhindern, dass Aufgaben in der globalen Warteschlange verhungern.Andernfallswird📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1090-1156。

bevorzugt die lokale Aufgabe abgerufennext_local_task. Das lokale Abrufen von Aufgaben wird von📎 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())
}

Kopieren

Zuerst wird der LIFO-Slot abgerufen, dann der Kopf der Warteschlange (LIFO-Pop). Das ist das im vorherigen Kapitel erwähnte „lokale LIFO".Wenn lokal leer ist, aber die globale Warteschlange nicht leer ist, wird der Workerbatchweise📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1110-1154Aufgaben aus der globalen Warteschlange ziehenn. Die Berechnung der Batch-Größemin(inject.len() / remotes.len() + 1, cap)ist wohlüberlegt:cap, wobeimin(remaining_slots, max_capacity / 2)wiederum📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1120-1131annimmt. Die Quellcode-Kommentare erklären, warum auf die Hälfte der Warteschlangenkapazität begrenzt wird: Um sicherzustellen, dass die gezogenen Aufgaben in dervorderen Hälfte

der lokalen Warteschlange landen, sodass diese Aufgaben selbst bei späterem Überlauf nicht zurück in die globale Warteschlange geschoben werden (Überlauf betrifft nur die hintere Hälfte).Dritter Schritt: Aufgabe ausführen.run_task 📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:647-796Nachdem

eine Aufgabe erhalten hat, wirdaufgerufen. Dies ist die komplexeste Funktion dieses Kapitels, die wir im nächsten Abschnitt gesondert behandeln.next_taskVierter Schritt: Stehlen oder Parken.NoneWennsteal_work 📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1167-1195parkzurückgibt, bedeutet dies, dass weder lokal noch global Arbeit vorhanden ist, undpark_yield 📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:613-621。

wird aufgerufen. Bei fehlgeschlagenem Stehlen wird

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

betreten. Der gesamte Kontrollfluss ist wie folgt:

run_taskKopierenpollrun_task: Der geschlossene Kreislauf von poll und LIFO-Slot

ist der Ort, an dem die Aufgabe tatsächlichassert_owner 📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:648wird, und auch der Schließpunkt des Kreislaufs „Aufwecken → Einreihen → erneut pollen".NotifiedDas Erste, was nach dem Betreten der Funktion geschieht, istTask, wobei

intransition_from_searching 📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:652umgewandelt wird, während gleichzeitig per Debug-Assertion sichergestellt wird, dass der aktuelle Thread tatsächlich der Owner dieser Aufgabe ist.

Danach📎 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();
    }
})

Dann folgt die entscheidende Budget-Umhüllungtask.run()KopierenFuture::pollDieser Codeabschnitt offenbart den vollständigen geschlossenen Kreislauf des LIFO-Slots:schedule_localführtlifo_slot 📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1396-1408aus, und wenn die Aufgabe während des Polls sich selbst oder eine andere Aufgabe aufweckt,lifo_slotlegtdie neue Aufgabe in. Nach der Rückkehr des Polls prüft die Schleife sofort

, und falls eine Aufgabe vorhanden ist, wird weiter ausgeführt –lifo_slotohne zur Hauptschleife zurückzukehren

, wird direkt innerhalb desselben Budgets kontinuierlich gepollt.self.core.borrow_mut().take()Dies ist die Verkörperung von „Aufwecken → Einreihen → erneut pollen" auf dem LIFO-Pfad: Beim Aufwecken wird die Aufgabe inNonegelegt, und nach der Rückkehr des Polls wird sie sofort entnommen und erneut gepollt, wodurch ein enger geschlossener Kreislauf entsteht.📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:716-724Beachten Sie denblock_in_place-Zweig vonControlFlow::Break(()): Wenn der Core gestohlen wurde (z. B. wenn in der AufgabeContext::runaufgerufen wurde), muss der Workerblock_in_placeInteraktionspunkte mit der Scheduler-Schleife.

Aufwachpfad: Wie der Waker das erneute Einreihen auslöst

WennFuture::pollzurückgibtPending, muss die Task einenWakerregistrieren, um beim Bereitwerden des Events aufgeweckt zu werden. TokiosWaker-Implementierung ist extrem schlank – sie ist lediglich ein Rohzeiger auf die TaskHeaderplus eine vtable.

waker_refKonstruiertWakerRef 📎 tokio/src/runtime/task/waker.rs:11-34, umhüllt mitManuallyDropdieWaker, um beim Drop das Dekrementieren des Referenzzählers zu vermeiden. Die vtable ist statisch📎 tokio/src/runtime/task/waker.rs:119-119：

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

Alle vier Funktionen stellen lediglich den Rohzeiger wieder alsHeaderher und rufen dann die entsprechende Methode vonRawTaskauf📎 tokio/src/runtime/task/waker.rs:70-116. Zum Beispiel ruftwake_by_refletztendlichraw.wake_by_ref() 📎 tokio/src/runtime/task/waker.rs:106-116。

wake_by_refauf. Die Semantik ist: Den Task-Status vonPENDINGnachSCHEDULEDüberführen, und wenn die Überführung erfolgreich ist (d. h. zuvor tatsächlich PENDING war),Schedule::scheduleaufrufen, um die Task erneut einzureihen.

Für den Multithread-Scheduler,scheduledie Implementierung inHandle::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();
    });
}

Die Logik teilt sich in zwei Zweige:

• Wenn der aktuelle Thread der Worker dieses Schedulers ist und core hält, gehe überschedule_local 📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1385-1417– ablegen im LIFO-Slot oder in der lokalen Queue.
• Andernfalls (Aufwecken von einem externen Thread oder core wurde gestohlen), gehe überpush_remote_taskEinfügen in die globale Injektions-Queue undnotify_parked_remoteeinen geparkten Worker aufwecken📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1379-1383。

schedule_localteilt sich intern wiederum in zwei Zweige📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1385-1417: Wenn esyieldist oder LIFO deaktiviert wurde, einfügen amrun_queueEnde; andernfalls ablegen inlifo_slot, und die Task aus dem ursprünglichen Slot an das Ende der Queue verdrängen.

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 und unpark: Zustandsmaschine und Atomarität des Aufweckens

Wenn der Worker nichts zu tun hat, muss er parken, aber park/unpark ist die Stelle, die am anfälligsten für Race Conditions ist. Tokio löst dies mit einerAtomicUsize-Zustandsmaschine plusCondvarals Fallback.

InnerDie Felder von📎 tokio/src/runtime/scheduler/multi_thread/park.rs:31-43：state: AtomicUsize、mutex: Mutex<()>、condvar: Condvar、shared: Arc<Shared>. Es gibt vier Zustandskonstanten📎 tokio/src/runtime/scheduler/multi_thread/park.rs:36-45：

• EMPTY = 0: nicht geparkt.
• PARKED_CONDVAR = 1: geparkt auf der condvar.
• PARKED_DRIVER = 2: geparkt auf dem I/O driver.
• NOTIFIED = 3: wurde bereits aufgeweckt.

Dies ist eine explizite Zustandsmaschine; wir verwenden sie, um das Zustandsdiagramm zu zeichnen (dies ist die einzige Stelle in diesem Kapitel, die diestateDiagram-v2-Zulassungsbedingung erfüllt – im Quellcode existieren tatsächlich diese vier Zustandskonstanten):

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)"

unparkDie Implementierung von📎 tokio/src/runtime/scheduler/multi_thread/park.rs:277-290verwendetswapstatt CAS; der Quellcode-Kommentar erklärt den Grund📎 tokio/src/runtime/scheduler/multi_thread/park.rs:277-290: Es muss eine Release-Operation ausgeführt werden, damit der parkende Thread die Schreibvorgänge vor dem unpark beobachten kann, daher muss selbst wenn state bereitsNOTIFIEDist, einmal geschrieben werden.

parkVersucht zunächst, eine vorhandene Benachrichtigung zu konsumieren📎 tokio/src/runtime/scheduler/multi_thread/park.rs:132-149: Wenn CASNOTIFIED -> EMPTYerfolgreich ist, bedeutet dies, dass zuvor bereits aufgeweckt wurde, und es wird direkt zurückgekehrt ohne zu blockieren. Andernfalls wird versucht, das driver-Lock zu erlangen; wenn erhalten, wird auf dem driver geparkt, andernfalls wird die condvar als Fallback verwendet📎 tokio/src/runtime/scheduler/multi_thread/park.rs:143-148。

park_condvarEs gibt eine klassische doppelte Prüfung📎 tokio/src/runtime/scheduler/multi_thread/park.rs:162-180: Zuerst CASEMPTY -> PARKED_CONDVAR, wenn dies fehlschlägt und esNOTIFIEDist, bedeutet dies, dass vor dem Setzen des Zustands bereits aufgeweckt wurde; in diesem Fall mussswap(EMPTY)ausgeführt werden, um die Schreibvorgänge des unpark zu synchronisieren📎 tokio/src/runtime/scheduler/multi_thread/park.rs:167-177. Der Kommentar betont besonders: Selbst wenn bekannt ist, dass esNOTIFIEDist, muss einmal gelesen werden, da unpark möglicherweise nach unserem Lesen vonNOTIFIEDerneut aufgerufen wurde.

unpark_condvarDer Kommentar von📎 tokio/src/runtime/scheduler/multi_thread/park.rs:292-307weist auf die klassische Falle der condvar hin: Zwischen dem Setzen desPARKED-Zustands durch den parkenden Thread und dem tatsächlichenwaitgibt es ein Zeitfenster; wenn in diesem Zeitraum notify aufgerufen wird, wird es ignoriert. Die Lösung ist, dass der parkende Thread zu diesem Zeitpunktmutexhält, und der unparkende Thread zuerstdrop(self.mutex.lock())das Lock erwirbt (und somit darauf wartet, dass der parkende Thread es freigibt), dannnotify_one。

Designüberlegung: Warum der LIFO-Slot ein einzelner Slot und keine Queue ist

MAX_LIFO_POLLS_PER_TICK = 3Diese magische Zahl📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:263-263ist ebenfalls ein Erfahrungswert. Der Quellcode-Kommentar besagt: „Ein paar Durchläufe durch den LIFO-Slot scheinen auszureichen, um von der Lokalität zu profitieren; mehr als 3 könnten übergewichten." Dies verhindert, dass ein Ping-Pong-Szenario, bei dem A B aufweckt und B A aufweckt, andere Tasks aushungert.

Ein weiteres bemerkenswertes Design iststeal_workdie „Halbierungs-Such"-Strategie📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1158-1160: Nur wenn weniger als die Hälfte der Worker suchen, versucht ein neuer Worker tatsächlich zu stehlen. Dies vermeidet CAS-Konkurrenz, die entsteht, wenn alle Worker gleichzeitig wild stehlen.transition_to_searchingKoordiniert durchidle.transition_worker_to_searching()wird📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1197-1203。

Das Stehlen beginnt an einem zufälligen Startpunkt📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1172-1174, durchläuft alle remote, überspringt sich selbst📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1179-1182, ruftsteal_intoauf, um einen Diebstahl zu versuchen. Nachdem alles fehlgeschlagen ist, wird auf die globale Queue zurückgegriffen📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1197-1203。

Zusammenfassung dieses Kapitels

Die Worker-HauptschleifeContext::runist das Herz des Schedulers: Nach jedem Tick wird zuerst eine Task geholt (LIFO-Slot → lokale Queue → globale Queue); wenn eine gefunden wird, wirdrun_taskausgeführt, um poll aufzurufen; wenn keine gefunden wird, wird gestohlen; wenn der Diebstahl fehlschlägt, wird geparkt.run_taskDie interne LIFO-Schleife komprimiert „Aufwecken → Einreihen → erneut poll" in dasselbe Budget und bildet so einen geschlossenen Regelkreis mit niedriger Latenz.WakerIst ein Rohzeiger plus statische vtable,wake_by_refLöst durch Zustandsübergangscheduleaus, und je nachdem, ob der aktuelle Thread derselbe Worker ist, wird entschieden, ob die lokale Queue oder die globale Queue verwendet wird.park/unparkVerwendet eine Vier-Zustands-Atommaschine plus condvar als Fallback und löst die klassische Race Condition des verlorenen Aufweckens.

Im nächsten Kapitel verlassen wir den Scheduler und betreten die I/O-Welt: Wie der Reactor epoll-Events inWakerAufwecken übersetzt, sodassAsyncFddiePendingzuReady。

Denkfragen und Selbsttests dieses Kapitels

Im vorherigen Kapitel haben wir die Hauptschleife des Worker-Threads verfolgt: Eine Aufgabe wird gepollt, bei Rückgabe von Pending wird der Waker irgendwo gespeichert, nach Eintritt der Bereitschaft wird der Waker ausgelöst und die Aufgabe erneut in die Warteschlange eingereiht. Aber wo genau ist dieses „irgendwo“? Wie wird der Waker gefunden, wenn ein epoll-Ereignis eintritt? Genau das ist die Frage, die der Reactor beantworten soll. Zuerst ein intuitives Modell: Stellen Sie sich den gesamten I/O-Bereitschaftsbenachrichtigungsmechanismus wie ein Nummernaufrufsystem in einem Restaurant vor – der Gast (die Aufgabe) stellt sich nach der Bestellung nicht ans Fenster und wartet, sondern nimmt einen Summer (Waker) mit zurück an den Platz; wenn die Küche (der Kernel-epoll) das Essen zubereitet hat, findet die Rezeption (der Reactor) anhand der Bestellnummer (Token) den entsprechenden Summer und drückt den Knopf. Ohne dieses System könnte jede Aufgabe nur den Socket pollen, und die CPU würde verbrannt; oder man würde blockierende Threads zum Warten verwenden, ein Thread pro Verbindung, was nicht skalierbar ist. Der Reactor von Tokio besteht aus drei Dateien mit einer dreischichtigen Struktur und strikter Trennung der Zuständigkeiten: driver.rs ist der Ereignisschleifenkern, hält mio::Poll, ist für den Aufruf von poll() zum blockierenden Warten auf Kernel-Ereignisse verantwortlich und übersetzt Ereignisse in Lese-/Schreibvorgänge auf ScheduledIo; registration.rs ist das benutzerseitige Registrierungshandle, das TcpStream intern hält, und bietet APIs wie poll_read_ready / poll_write_ready; scheduled_io.rs ist der Status-Slot jedes fd, speichert Lese-/Schreibbereitschaftsbits und Waker-Listen und ist die Brücke zwischen Ereignissen und Aufgaben. Die Modulzusammensetzung ist in tokio/src/runtime/io/mod.rs:5-16 zu finden: driver exportiert Driver, Handle, ReadyEvent, registration exportiert Registration, scheduled_io exportiert ScheduledIo. Die folgende Abbildung verankert den vollständigen Datenfluss, der in diesem Kapitel verfolgt werden soll: TcpStream → Registration → ScheduledIo → Handle/Driver → Kernel → zurück zu ScheduledIo → Waker. Als Nächstes zerlegen wir dies Schicht für Schicht.

Treiberschicht:DriverundHandleAufgabenteilung

Intuitives Modell

Driveristdie einzige Entität, diemio::Pollbesitzt, und kann nur in einem einzelnen Thread&mutzugegriffen werden – dies ist die Exklusivitätsanforderung der Ereignisschleife. Hingegen istHandleeinklonbarer, threadübergreifend gemeinsam nutzbarer Registrierungseinstiegspunkt, jeder Thread, der einen neuen fd registrieren möchte, tut dies darüber. Ohne diese Aufteilung müsste man entwedermio::Pollsperren (bei jeder Registrierung Konkurrenz), oder alle Registrierungen zurück zum Driver-Thread leiten (Einführung einer Cross-Thread-Message-Queue). Tokio entscheidet sich,Handledirekt einen Klon vonmio::Registryhalten zu lassen, Registrierungsoperationen können parallel ablaufen, nur das tatsächliche Warten auf Events benötigt Exklusivzugriff.

Speicherlayout und Felder

Zuerst schauen wir unsDriverdie Felder von📎 tokio/src/runtime/io/driver.rs:25-38：

• signal_ready: boolan: Ob ein Unix-Signal-Event angekommen ist, wird für signal-getrieben verwendet.
• events: mio::Events: Haupt-Event-Puffer, wird überturnAufrufe hinweg wiederverwendet, um Allokation bei jedem Aufruf zu vermeiden.
• events_busy: Option<mio::Events>：Dedizierter Puffer für nicht-blockierendes poll, existiert nur, wennmax_io_events_per_busy_tickgesetzt ist.
• poll: mio::Poll: Kapselung der Kernel-Event-Queue.

Schauen wir uns nunHandle 📎 tokio/src/runtime/io/driver.rs:41-75：

• registry: mio::Registry：mio::Poll::registry()den Klon vonregister/deregister。
• registrations: RegistrationSetan, verwendet fürToken: Menge aller aktiven Registrierungen, verantwortlich für die Zuweisung vonScheduledIo。
• synced: Mutex<registration_set::Synced>undRegistrationSet: Schützt den Synchronisationszustand von
• waker: mio::Waker: Wird verwendet, um aus beliebigen Threads den Driver aufzuwecken, der inturnblockiert.
• metrics: IoDriverMetrics: Zählt die Anzahl der fds und der bereiten Events.

Hier gibt es ein entscheidendes Design:events_busyDie Existenz von📎 tokio/src/runtime/io/driver.rs:25-38dient dazu,das Problem zu lösen, dass nicht-blockierendes poll Events verschluckt. Der Kommentar📎 tokio/src/runtime/io/driver.rs:189-190sagt es ganz klar: Wenn Events, die durch nicht-blockierendes poll entnommen wurden, im Hauptpuffer verbleiben, sind sie beim nächsten poll nicht mehr sichtbar; mit einem separaten Puffer bleiben unbehandelte Events in der Kernel-Queue und werden beim nächsten poll erneut zurückgegeben.

Step-by-Step: Eine Ausführung vonturn

turnist die Kernfunktion des Drivers📎 tokio/src/runtime/io/driver.rs:184-261. Angenommen, ein Worker-Thread stellt fest, dass keine Aufgaben auszuführen sind, und ruftpark → turn(handle, None)auf, um blockierend zu warten:

Erster Schritt: Assertion, dass nicht heruntergefahren📎 tokio/src/runtime/io/driver.rs:185, und Freigabe der zu bereinigenden Registrierungen📎 tokio/src/runtime/io/driver.rs:187。release_pending_registrationsPrüftneeds_release(), falls vorhanden, wirdregistrations.release() 📎 tokio/src/runtime/io/driver.rs:336-340。

aufgerufen. Zweiter Schritt: Auswahl des Event-Puffers📎 tokio/src/runtime/io/driver.rs:191-194. Fallsmax_waitnull ist undevents_busyexistiert, wird der busy-Puffer verwendet; andernfalls der Hauptpuffer.

Dritter Schritt: Aufruf vonself.poll.poll(events, max_wait) 📎 tokio/src/runtime/io/driver.rs:198. Hier wird tatsächlich in epoll_wait blockiert. Die Fehlerbehandlung ist sehr zurückhaltend:Interruptedwird direkt ignoriert (Signalunterbrechung ist normal)📎 tokio/src/runtime/io/driver.rs:200, unter WASI wirdInvalidInputebenfalls ignoriert📎 tokio/src/runtime/io/driver.rs:201-205, andere Fehler führen direkt zu panic📎 tokio/src/runtime/io/driver.rs:206。

Vierter Schritt: Iteration über die Events📎 tokio/src/runtime/io/driver.rs:211-233. Für jedesevent：

• fallstoken == TOKEN_WAKEUP(Wert 0)📎 tokio/src/runtime/io/driver.rs:214, wird nichts getan – dies wird vonunparkverwendet, um die Blockierung zu unterbrechen.
• Fallstoken == TOKEN_SIGNAL(Wert 1)📎 tokio/src/runtime/io/driver.rs:216, wirdsignal_ready = true。
• gesetzt. Andernfalls handelt es sich um ein normales I/O-Event📎 tokio/src/runtime/io/driver.rs:218-231:mio::Readywird in TokiosReadyumgewandelt, mitEXPOSE_IO.from_exposed_addr(token.0)wird das Token zurück in einen*const ScheduledIo-Zeiger umgewandelt, dannset_readiness(Tick::Set, |curr| curr | ready)werden die Bereitschaftsbits akkumuliert, anschließendio.wake(ready)wird die entsprechende Richtung vonWaker。

ausgelöst. Hier istEXPOSE_IOeinPtrExposeDomain<ScheduledIo> 📎 tokio/src/runtime/io/mod.rs:21-22, das den Zeiger alsusize„exponiert“ alsmio::Token. Der Sicherheitskommentar📎 tokio/src/runtime/io/driver.rs:222-225erklärt, warum diese unsafe-Konvertierung sicher ist: Der Zeiger wird nicht freigegeben, bevor er bei mio abgemeldetundder Driver nicht mehr parallel pollt, und der Driver besitzt das Eigentum anArc<ScheduledIo>.

Fünfter Schritt: Verarbeitung der io_uring Completion-Queue (nur Linux + tokio_unstable)📎 tokio/src/runtime/io/driver.rs:235-258, einschließlich der Flush-Schleife bei CQ-Überlauf.

Sechster Schritt: Akkumulation der Metriken📎 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)"]

Designüberlegung: WarumHandlehalten mussmio::Waker

unpark 📎 tokio/src/runtime/io/driver.rs:280-283aufrufenself.waker.wake(). Diesesmio::Wakerwird beiDriver::newmitTOKEN_WAKEUPregistriert📎 tokio/src/runtime/io/driver.rs:124. Wenn der Driver inpoll.poll()blockiert, wird durch einen anderen Thread, derunparkaufruft, einTOKEN_WAKEUP-Event in epoll eingefügt,pollkehrt sofort zurück, bei der Iteration wird dieses Token direkt übersprungen📎 tokio/src/runtime/io/driver.rs:214-215。

Ein weiteres Detail:deregister_sourceruft zuerstself.registry.deregister(source) 📎 tokio/src/runtime/io/driver.rs:322auf, dann wirdregistrations 📎 tokio/src/runtime/io/driver.rs:315-334bereinigt. Der Kommentar📎 tokio/src/runtime/io/driver.rs:320-321sagt „Cleanup ALWAYS happens“ – selbst wenn die Deregistrierung auf OS-Ebene fehlschlägt, muss der interne Zustand bereinigt werden, erst danach wird der OS-Fehler zurückgegeben📎 tokio/src/runtime/io/driver.rs:336-340. Dies ist ein typischesMuster, bei dem Ressourcenbereinigung Vorrang vor Fehlerpropagierung hat.

Registrierungsschicht:RegistrationWieWakerinScheduledIo

gespeichert wird. Intuitives Modell

Registrationistein Vertrag zwischen Task und fd. Es hält zwei Dinge: einscheduler::Handle(verwendet, um bei Bedarf auf die Runtime zuzugreifen), einArc<ScheduledIo>(der Zustandsslot des fd). Wenn eine Taskpoll_read_readyaufruft,RegistrationübergibtWakeranScheduledIozur Verwahrung; wenn der Driver ein Event empfängt, wirdScheduledIoausWakerentnommen und aufgeweckt.

Speicherlayout und Felder

Registrationhat nur zwei Felder📎 tokio/src/runtime/io/registration.rs:46-54：

• handle: scheduler::Handle: Runtime-Handle, Kommentar📎 tokio/src/runtime/io/registration.rs:46-54sagt „TODO: this can probably be moved into ScheduledIo“, was zeigt, dass der Autor der Meinung ist, dass die Position dieses Feldes optimiert werden kann.
• shared: Arc<ScheduledIo>: Gemeinsamer Zustand,Arcstellt sicher, dass sowohl Driver als auch Task darauf zugreifen können.

Step-by-Step：poll_read_readyDie Aufrufkette von

Angenommen, eine Task ist inTcpStream::poll_readstellt fest, dass der Socket keine Daten hat, und muss ein Lese-Interesse registrieren. Die Aufrufkette istTcpStream::poll_read_priv → PollEvented::poll_read → Registration::poll_read_io → poll_io → poll_ready。

poll_readyist der Kern📎 tokio/src/runtime/io/registration.rs:155-171：

Erster Schritt：trace_leaf() 📎 tokio/src/runtime/io/registration.rs:160, verwendet für Tracing-Instrumentierung.

Zweiter Schritt：coop::poll_proceed(cx) 📎 tokio/src/runtime/io/registration.rs:155-171. Dies ist der kooperative Budget-Mechanismus, der in Kapitel 12 behandelt wird. Wenn das Budget erschöpft ist, wird zurückgegebenPendingund ein speziellesWakerregistriert, damit die Aufgabe in der nächsten Runde neu geplant wird.

Dritter Schritt：self.shared.poll_readiness(cx, direction) 📎 tokio/src/runtime/io/registration.rs:155-171. Dies ist der Ort, an dem tatsächlich mitScheduledIointeragiert wird: Der aktuelle Bereitschaftsstatus wird geprüft; wenn bereits bereit, wird sofort zurückgegebenReady; andernfalls wirdcx.waker()in den entsprechenden Richtungs-Slot vonScheduledIogespeichert und zurückgegebenPending。

Vierter Schritt: Prüfen vonev.is_shutdown 📎 tokio/src/runtime/io/registration.rs:155-171. Wenn die Runtime gerade heruntergefahren wird, wird zurückgegebenRUNTIME_SHUTTING_DOWN_ERROR。

Fünfter Schritt：coop.made_progress() 📎 tokio/src/runtime/io/registration.rs:169, Budgetverbrauch markieren und das Bereitschaftsereignis zurückgeben.

poll_iofügt überpoll_readyeine Retry-Schleife hinzu📎 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)),
    }
}

Hier zeigt sichReadiness ist ein Hinweis, keine GarantieDie Kernidee:poll_readymeldet lesbar, aber beim tatsächlichenread()kannWouldBlockzurückgegeben werden (z. B. wenn ein anderer Thread die Daten zuerst weggelesen hat). In diesem Fall mussclear_readiness(ev) 📎 tokio/src/runtime/io/registration.rs:187das Bereitschaftsbit gelöscht und die Schleife erneut auf Warten gesetzt werden. Wenn nicht gelöscht wird, gerät die Aufgabe in eine Busy-Schleife von „glaubt lesbar → read schlägt fehl → glaubt wieder lesbar“.

Designüberlegung:try_ioundasync_ioArbeitsteilung

try_io 📎 tokio/src/runtime/io/registration.rs:194-213ist die synchrone Version: Zuerstready_event(interest)das Bereitschaftsbit prüfen; wenn leer, direkt zurückgebenWouldBlock 📎 tokio/src/runtime/io/registration.rs:194-213; andernfallsf()ausführen; wennf()zurückgibtWouldBlock, dann das Bereitschaftsbit löschen📎 tokio/src/runtime/io/registration.rs:207-210. Esregistriert keinen Wakerund eignet sich fürtry_readSzenarien wie „einmal versuchen und gehen“.

async_io 📎 tokio/src/runtime/io/registration.rs:225-245ist die asynchrone Version:readiness(interest).awaitregistriert einen Waker und wartet, dann wird bei der Ausführung vonf()，WouldBlockdas Bereitschaftsbit gelöscht und die Schleife fortgesetzt. Beachten Sie, dass in der Schleife auchcoop::poll_proceed 📎 tokio/src/runtime/io/registration.rs:233aufgerufen wird, um zu verhindern, dass bei vielenWouldBlockRetries das Budget erschöpft wird.

Produktions-Fallstricke:DropWaker-Bereinigung in

Registration::drop 📎 tokio/src/runtime/io/registration.rs:253-262ruftself.shared.clear_wakers()auf. Der Kommentar📎 tokio/src/runtime/io/registration.rs:253-262erklärt den Grund:ScheduledIoDas inWakergespeicherteArc<driver::Inner>kanndriver::Innerhalten, undScheduledIohält wiederumRegistration, was einen Zirkelbezug bildet. Die Waker-Bereinigung ist ein Mittel, um den Zyklus zu durchbrechen. Der Kommentar gibt jedoch zu, dass dies eine „imperfect solution“ ist – wennWakerselbst in

beachtet werden.TcpStream::readVonWakerbis

vollständige Kette des Aufweckens

Intuitives ModellTcpStreamNun werden die drei Ebenen verbunden. Der Benutzer ruft auf.read().awaitaufAsyncRead::poll_read → PollEvented::poll_read → Registration::poll_read_io, tatsächlich ausgeführt wirdWaker. Wenn keine Daten ankommen,ScheduledIowird inScheduledIogespeichert; wenn epoll Lesbarkeit meldet, entnimmt der Driver ausWakerdenpoll_readinessund weckt auf, die Aufgabe wird neu geplant, und beim erneuten Poll entdecktReady，read(), dass das Bereitschaftsbit gesetzt ist, und gibt direkt

erfolgreich zurück.

Step-by-Step: Ein vollständiges Lese-Warten。TcpStream::new 📎 tokio/src/net/tcp/stream.rs:166-169Phase eins: Interesse registrierenPollEvented::new(connected)ruftRegistration::new_with_interest_and_handle 📎 tokio/src/runtime/io/registration.rs:73-81auf, letzteres ruft internhandle.driver().io().add_source(io, interest) 📎 tokio/src/runtime/io/registration.rs:73-81。

add_source 📎 tokio/src/runtime/io/driver.rs:288-312auf, und

1. registrations.allocate(&mut synced.lock())erledigt drei Dinge:ScheduledIoweist einentoken 📎 tokio/src/runtime/io/driver.rs:293-294。

2. self.registry.register(source, token, interest.to_mio())zu, holt📎 tokio/src/runtime/io/driver.rs:298registriertbeim Kernel. Bei FehlschlagmussScheduledIoder gerade zugewiesene📎 tokio/src/runtime/io/driver.rs:300-303aus der Menge entfernt werden

3. metrics.incr_fd_count(), sonst Leck.📎 tokio/src/runtime/io/driver.rs:309。

zähltPhase zwei: Auf Bereitschaft wartenTcpStream::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. Die Aufgabe polltWaker. Wenn zu diesem Zeitpunkt nicht bereit,ScheduledIowird in den Lese-Slot vonPending。

gespeichert und zurückgegebenPhase drei: Ereignis trifft einturn. Derpoll.poll()des Drivers holt das Ereignis📎 tokio/src/runtime/io/driver.rs:198ausio.set_readiness(Tick::Set, |curr| curr | ready), und beim Durchlaufen wird für jedes fd-Ereignisio.wake(ready) 📎 tokio/src/runtime/io/driver.rs:228-229。wakeausgeführt undWakerintern derwake()。

der entsprechenden Richtung entnommen und。Waker::wake()aufgerufenpoll_readinessPhase vier: Aufgabe neu planenReady，read()Die Aufgabe wird erneut in die lokale Queue des Workers eingereiht (im vorherigen Kapitel behandelt). Der Worker pollt die Aufgabe erneut,

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() 成功返回数据"

erfolgreich zurück.assume_readyKopieren

TcpStream::new_accepted 📎 tokio/src/net/tcp/stream.rs:174-181Wichtiger Zweig:acceptOptimierungnew_acceptedist eine bemerkenswerte Optimierung.assume_ready(Ready::READABLE | Ready::WRITABLE) 📎 tokio/src/net/tcp/stream.rs:174-181。

assume_readyDer von📎 tokio/src/runtime/io/registration.rs:103-105zurückgegebene Socket ist natürlich beschreibbar und hält normalerweise bereits die ersten Bytes der Gegenseite. Wenn man auf das erste Ereignis des Drivers wartet, könnte dieses Ereignis unter hoher Last hinter allen Ereignissen bereits aufgebauter Verbindungen stehen und Verzögerungen verursachen. Daher ruftWouldBlockdirektWouldBlock，poll_ioauf. Der Kommentarsagt: „A wrong guess costs one, which clears the readiness again.“ – Der Preis für eine falsche Vermutung ist nur eine

Schleife, die das Bereitschaftsbit löscht und erneut wartet. Dies ist ein Design von

1. wird an einer speziellen Stelle der Runtime platziert (normalerweise：HandleThread oder dedizierter I/O-Thread), während Worker-Threads nurmio::Registryhalten. Diese Entkopplung bringt mehrere Vorteile:

2. Lock-freie Registrierunghält einenepoll_wait-Klon, jeder Worker kann parallel neue fds registrieren, ohne zum Driver-Thread zurückzukehren.

3. Zentralisiertes Ereignis-Warten: Nur ein Thread blockiert aufScheduledIo, wodurch das Thundering-Herd-Problem vermieden wird, bei dem mehrere Threads gleichzeitig denselben epoll-fd pollen.Waker::wake()，wake()Kurzer Aufweckpfad

: Nach Erhalt eines Ereignisses operiert der Driver direkt aufScheduledIound ruftset_readinessauf; intern wird die Aufgabe in die Worker-Queue geschoben, ohne threadübergreifende Nachrichtenübermittlung.poll_readinessDer Preis ist, dass

konkurrierende Zugriffe behandeln muss (is_shutdownundRUNTIME_SHUTTING_DOWN_ERROR

poll_readykönnen gleichzeitig auftreten), was durch atomare Operationen und interne Locks gelöst wird.ev.is_shutdown 📎 tokio/src/runtime/io/registration.rs:155-171Produktions-Fallstricke:gone() 📎 tokio/src/runtime/io/registration.rs:265-267undRUNTIME_SHUTTING_DOWN_ERROR。

Eine weitere Falle istderegister_sourcevonunpark 📎 tokio/src/runtime/io/driver.rs:328. Wenn der Driver blockiert inpollwartet und zu diesem Zeitpunkt der letzteRegistrationgedroppt wird,unparkweckt den Driver auf. Aber wenn der Driver nicht blockiert ist (z. B. gerade andere Events verarbeitet),unparklässt nur den nächstenturnsofort📎 tokio/src/runtime/io/driver.rs:280-283zurückgeben. Diese Semantik ist in den Dokumentationskommentaren vonHandle::unparkbeschrieben.

Designüberlegung: Die drei entscheidenden Kompromisse des Reactors

Kompromiss eins:Tokenverwendet Zeiger statt Indizes。EXPOSE_IO.from_exposed_addr(token.0) 📎 tokio/src/runtime/io/driver.rs:220behandeltmio::Tokendirekt als*const ScheduledIoAdresse. Dies vermeidet die Pflege einerToken → ScheduledIoMapping-Tabelle, die Suche ist O(1) und lockfrei. Der Preis ist, dass die Sicherheit von striktem Lebenszeitmanagement abhängt: Der Zeiger darf erst freigegeben werden, nachdem die Registrierung aufgehoben wurde und der Driver nicht mehr pollt📎 tokio/src/runtime/io/driver.rs:222-225。

Kompromiss zwei: Zwei getrennte Waker-Slots für Lesen und Schreiben。RegistrationDokumentation📎 tokio/src/runtime/io/registration.rs:24-26sagt „A registration instance represents two separate readiness streams“ – Lesen und Schreiben haben jeweils einen unabhängigenWakerSlot. Dies erlaubt, dass Lese- und Schreib-Tasks desselben Sockets sich separat registrieren, ohne sich gegenseitig zu stören. Aberpoll_read_readyKommentar📎 tokio/src/net/tcp/stream.rs:549-552erinnert daran: Mehrfache Aufrufe vonpoll_read_ready/poll_read/poll_peekbehalten nur den letztenWaker– die Lese-Richtung hat nur einen Slot.

Kompromiss drei:events_busyunabhängiger Puffer. Test📎 tokio/src/runtime/io/driver.rs:364-386verifiziert dieses Verhalten:Driver::new(16, Some(2))Erstellt einen Driver mit busy-Kapazität 2, registriert 5 lesbare Sources, nicht-blockierendesturnnimmt nur 2 Events📎 tokio/src/runtime/io/driver.rs:375-376, die restlichen 3 bleiben in der Kernel-Queue, beim nächsten blockierendenturnwerden📎 tokio/src/runtime/io/driver.rs:379-380geholt. Dies verhindert, dass ein nicht-blockierender Poll alle Events auf einmal verschlingt und nachfolgende Polls aushungert.

Zusammenfassung dieses Kapitels

Dieses Kapitel verfolgte die vollständige Reactor-Kette hinterTcpStream::read:

• Treiber-Schicht：Driverexklusivmio::Poll，turnblockierendes Warten auf Events, mitEXPOSE_IOwirdTokenzurück inScheduledIoZeiger umgewandelt, Aufruf vonset_readiness + wakelöstWaker。Handleaus, bietet einen thread-übergreifenden Registrierungseingang,unparkdient zum Unterbrechen der Blockierung.
• Registrierungs-Schicht：RegistrationhältArc<ScheduledIo>，poll_readyprüft Ready-Bits oder speichert inWaker，poll_iomitWouldBlockRetry-Schleife behandelt False Positives,try_io/async_iobedient jeweils synchrone und asynchrone Szenarien.
• Zustands-Schicht：ScheduledIoist der Zustands-Slot des fd, speichert Lese-/Schreib-Ready-Bits und zweiWakerSlots, ist die einzige Brücke zwischen Events und Tasks.

Kapitel-Überlegungen und Selbsttest

Zugehöriges Projekt: tokio-rs/tokio

Fortschritt des Buches: Kapitel 7 / 14

Verifikationsstatus: FACT-Zeilennummern echt verankert

std::sync::MutexDas vorherige Kapitel hat gezeigt, wie Zeit als eine Art I/O-Ereignis abstrahiert wird, sodass Timer und fd-Bereitschaft denselben park/unpark-Warteeingang teilen. Wenn jedoch mehrere Aufgaben um dieselbe Sperre konkurrieren oder Nachrichten über Kanäle austauschen, ist das Objekt des Wartens nicht mehr ein fd oder eine Uhr, sondern die Zustandsänderung einer anderen Aufgabe. Dieses Kapitel betritt die tokio::sync-Familie und untersucht, wo ein lock().await oder recv().await beim Blockieren tatsächlich den Waker speichert und wie er beim Aufwecken neu eingeplant wird.lock()Warum der asynchrone Mutex nicht die std-Implementierung wiederverwenden kannIntuitives Modell: Von „den Platz besetzen“ zu „den Sitzplatz freigeben“Dasvonblockiert den aktuellen Thread, wenn die Sperre belegt istPending– der Thread wird vom Betriebssystem angehalten, bis die Sperre freigegeben wird. In einer asynchronen Laufzeit ist das katastrophal: Ein Worker-Thread kann gleichzeitig Hunderte oder Tausende von Aufgaben antreiben; wenn er wegen des Wartens auf eine Sperre blockiert, stehen alle anderen von ihm getragenen Aufgaben still. Das Kernanliegen des asynchronen Mutex ist: beim Warten auf die Sperre

den Thread freigebenMutex, die Tatsache „ich warte auf diese Sperre“ in eine Warteschlange eintragen und dannVollständig auf Semaphoren aufgebaut。

Datenstruktur und Speicherlayout

Mutex<T>Die Felder von sind minimalistisch:

📎 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>,
}

Die drei Felder erfüllen jeweils ihren Zweck:sist einSemaphore mit einer Genehmigungsanzahl von 1，cistUnsafeCell<T>die geschützten Daten, die umschlossen werden. Beachten Sie, dass hiersemaphoreein Alias fürbatch_semaphoreist📎 tokio/src/sync/mutex.rs:3-3, also die zugrunde liegende Implementierung, nicht die öffentliche Kapselungsync::Semaphore.

MutexGuard<'a, T>hält lediglich eine Referenz aufMutex:

📎 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>,
}

Hier gibt es ein entscheidendes Design:MutexGuard hält kein Semaphore-Genehmigungsobjekt, sondern nur&Mutex. Die Freigabe der Sperre erfolgt inDrop, indem direktself.lock.s.release(1) 📎 tokio/src/sync/mutex.rs:959-961aufgerufen wird. Dies unterscheidet sich vonSemaphorePermit, daspermits: usizeZähler hält und diese bei Drop zurückgibt – die Genehmigungsanzahl von Mutex ist konstant 1, es ist kein Zähler nötig.

Send/SyncDie Grenzen von sind es wert, separat betrachtet zu werden:

📎 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 {}

Syncerfordert nurT: Sendund nichtT: Sync– das ist sinnvoll, da gegenseitiger Ausschluss garantiert, dass nur ein Thread gleichzeitig aufTzugreifen kann. Die Übertragung der Eigentümerschaft vonTüber Threads hinweg (Send) reicht aus, es ist nicht nötig, dassTselbst geteilt werden kann (Sync). Genau das ist der Grund, warumMutex<T>ein nicht-Sync-Tin einSyncverwandeln kann.

Schritt für Schritt: Die vollständige Reise eineslock().await

Szenario: Aufgabe A ruftmutex.lock().awaitauf, die Sperre ist zu diesem Zeitpunkt frei.

Erster Schritt:lock()konstruiert einen async-Block, der intern zuerstself.acquire().awaitausführt und nach ErfolgMutexGuard 📎 tokio/src/sync/mutex.rs:434-443。

konstruiert. Zweiter Schritt:acquire()delegiert direkt an die Semaphore:

📎 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!())Diese Zeile Kommentar offenbart die Design-Einschränkung: Mutex schließt die Semaphore niemals explizit und hält sie exklusiv, daher wirdacquireniemalsErrzurückgeben. Dies schließt den Fehlerpfad „Semaphore geschlossen" auf Typebene aus.

Dritter Schritt: Wenn die Sperre belegt ist,s.acquire(1)gibtPendingzurück, der Waker der aktuellen Aufgabe wird in die Warteschlange der Semaphore eingetragen.Wo wird der Waker gespeichert?Die Antwort liegt in der Warteschlange vonbatch_semaphore(das Quellmaterial dieses Kapitels führt diese Datei nicht weiter aus, aber ihre Rolle ist: Jeder Wartende hält einen Waker und wird in FIFO-Reihenfolge eingereiht).

Vierter Schritt: Wenn Aufgabe B, die die Sperre hält, diese freigibt,MutexGuard::droprufts.release(1) 📎 tokio/src/sync/mutex.rs:965-975auf, die Semaphore übergibt die Genehmigung an den ersten Wartenden in der Schlange und weckt dessen Waker, Aufgabe A wird neu geplant,acquiregibtOkzurück und konstruiertMutexGuard。

Der gesamte Ablauf lässt sich mit dem folgenden Sequenzdiagramm darstellen:

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

Designüberlegungen: FIFO-Fairness und Abbruchsicherheit

Die Dokumentation erklärt ausdrücklich, dass Tokios Mutex FIFO📎 tokio/src/sync/mutex.rs:20-22garantiert. Diese Fairness stammt aus der Warteschlangensemantik der zugrunde liegenden Semaphore. Der Preis der Fairness ist: Wenn einlockabgebrochen wird (z. B. bei einer Niederlage inselect!), verlieren SieIhren Platz in der Warteschlange 📎 tokio/src/sync/mutex.rs:415-419. Das ist kein Bug, sondern eine Notwendigkeit der FIFO-Warteschlange – ein Abbruch bedeutet Entfernung aus der Warteschlange, ein erneuteslockerfordert ein erneutes Einreihen.

Ein weiteres kontraintuitives Design istkeine Vergiftung（no poisoning）。std::sync::Mutexwird bei einem Panic des sperrenden Threads als poisoned markiert, nachfolgendelockgebenErrzurück. Tokios Mutex macht das nicht: Bei einem Panic des Sperrenden wird die Sperre normal freigegeben📎 tokio/src/sync/mutex.rs:122-125. Die Dokumentation warnt, dass die geschützten Daten in einem inkonsistenten Zustand sein können, wenn der Panic abgefangen wird. Dies ist ein pragmatischer Kompromiss im asynchronen Kontext – ein Panic in einer asynchronen Aufgabe bedeutet normalerweise das Ende der Aufgabe, und der Vergiftungsmechanismus würde nur Komplexität hinzufügen.

MutexGuard::mapDieMutexGuard<T>-Methodenfamilie ist erwähnenswert. Sie ermöglicht es, ein ganzesMappedMutexGuard<U>in eindataherabzustufen, das nur ein bestimmtes Unterfeld schützt. In der Implementierung wird zuerst per Closure der Zeiger auf das Unterfeldskip_dropberechnet, dann überMutexGuardInnerder ursprüngliche Guard in ein📎 tokio/src/sync/mutex.rs:869-883。skip_dropzerlegt, das kein Drop auslöst, und schließlich ein neuer Guard konstruiertManuallyDrop + ptr::read. MitDropwird die Feld-Eigentümerschaft übertragen, um zu vermeiden, dass📎 tokio/src/sync/mutex.rs:827-836zweimal aufgerufen wird. Dies ist eine klassische Technik in Rust, um „Eigentümerschaft zu übertragen, ohne den Destruktor auszulösen".

Semaphore: Wie Genehmigungszähler und Warteschlange Backpressure implementieren

Intuitives Modell: Parkplätze auf einem Parkplatz

Eine Semaphore ist wie ein Parkplatz:acquireist die Einfahrt, bei freiem Platz wird eingefahren, sonst wird am Eingang gewartet;releaseist die Ausfahrt, ein frei werdender Platz benachrichtigt das erste wartende Auto einzufahren. Die Genehmigungsanzahl ist die Gesamtzahl der Parkplätze,acquire_many(n)ist ein großes Auto, das n Parkplätze belegt.

Datenstruktur und Speicherlayout

Die öffentlicheSemaphoreist nur eine dünne Kapselung der zugrunde liegendenbatch_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>hält eine Semaphore-Referenz und einen Genehmigungszähler:

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

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

permitsDas Feld ist der Schlüssel zum Verständnis vonforget/merge/split.forgetsetztpermitsauf null📎 tokio/src/sync/semaphore.rs:1193-1195, sodass bei Drop 0 Genehmigungen zurückgegeben werden – äquivalent zu „permanentem Verbrauch" dieser Genehmigungen.splitschneidet n Genehmigungen aus den aktuellen Genehmigungen für das neue Permit ab📎 tokio/src/sync/semaphore.rs:1260-1271。mergeführt den Zähler eines anderen Permits zusammen und stellt sicher, dass beide von derselben Semaphore stammen📎 tokio/src/sync/semaphore.rs:1230-1240。

Schritt für Schritt: Genehmigungsfluss von acquire und release

Szenario: Semaphore startet mit 2 Genehmigungen, Aufgabe Aacquire(), Aufgabe Bacquire_many(2)。

acquire()delegiert anll_sem.acquire(1), nach Erfolg wirdSemaphorePermit { permits: 1 } 📎 tokio/src/sync/semaphore.rs:614-631。acquire_many(2)konstruiert. Ähnlich, aber mit Übergabe von 2📎 tokio/src/sync/semaphore.rs:661-679。

Wenn Genehmigungen nicht ausreichen,ll_sem.acquire(n)gibtPendingzurück, der Waker wird eingereiht. Hier gibt es ein Fairness-Detail: Die Dokumentation weist darauf hin, dass, wenn der erste in der Schlange einacquire_many(5)ist und aktuell nur noch 3 Genehmigungen übrig sind, selbst wenn dahinter einacquire(1)sofort erfüllt werden könnte, dieser warten muss – weil das große Auto an der Spitze die Schlange blockiert📎 tokio/src/sync/semaphore.rs:19-24. Dies ist der Preis strikter FIFO, der Hunger vermeidet.

Der Freigabepfad liegt in Drop:

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

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

add_permitsdelegiert anll_sem.release(n) 📎 tokio/src/sync/semaphore.rs:568-570, die zugrunde liegende Implementierung gibt die Genehmigung an die Warteschlange zurück und weckt Wartende, die genügend Genehmigungen ansammeln können.

Bezüglich der Speicherordnung gibt die Dokumentation eine starke Garantie: acquire, release und close sind alleAcqRel-Operationen, total geordnet zueinander, äquivalent zu denen auf einer einzelnen atomaren VariableAcqRel 📎 tokio/src/sync/semaphore.rs:35-42. Das bedeutet, dass Schreibvorgänge, die „zuerst Daten schreiben und dann die Berechtigung freigeben“, für Aufgaben sichtbar sind, die „danach die Berechtigung erwerben“ – Semaphore können sicher Daten zwischen Aufgaben übertragen.

Designüberlegungen: close und Backpressure

close()Lässt alle Wartenden empfangenAcquireError, und nachfolgendetry_acquiregeben zurückClosed 📎 tokio/src/sync/semaphore.rs:1161-1163. Dies ist die Grundlage für elegantes Herunterfahren: Wenn die Empfängerseite keine Daten mehr benötigt, kann das close-Semaphor alle blockierten Sender sofort mit einem Fehler zurückkehren lassen, anstatt ewig zu warten.

Das Wesen der Backpressure zeigt sich am deutlichsten in mpsc. Im nächsten Abschnitt wird deutlich, dass die Kapazitätssteuerung von mpsc durch ein Semaphor implementiert wird, dessen Anzahl an Berechtigungen der Puffergröße entspricht.

Die Kanal-Familie: Unterschiedliche Kompromisse bei Warteschlangen und Waker-Aufweckung

Intuitives Modell: Vier Kanäle, vier Wartestrategien

oneshotist ein „Einmal-Umschlag“ – es kann nur ein einziges Mal gesendet werden, der Sender wartet nicht (sendist synchron), der Empfängerawaitwartet auf die Nachricht.mpscist ein „begrenztes Förderband“ – der Sender wartet, wenn das Förderband voll ist, der Empfänger wartet, wenn es leer ist, die Kapazität wird durch ein Semaphor gesteuert.broadcastundwatchsind „Rundruf-Lautsprecher“ – ein Sender, mehrere Empfänger, aber beide gehen mit „Zurückbleiben“ völlig unterschiedlich um.

Der Quellcode dieses Abschnitts konzentriert sich aufoneshotundmpsc::bounded, wir zerlegen sie einzeln.

oneshot: Minimaler Handshake mit Zustandsbits kodiert

oneshotDieInner-Struktur ist der Kern zum Verständnis des Designs:

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

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

stateist einAtomicUsize, das den gesamten Zustand des Kanals mit Bitflags kodiert. Die vier Flags sind am Ende der Datei definiert:

📎 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;

valueistUnsafeCell<Option<T>>，tx_taskundrx_tasksindTask-Typen, internUnsafeCell<MaybeUninit<Waker>> 📎 tokio/src/sync/oneshot.rs:411-411. Beachten SieMaybeUninit– der Waker ist möglicherweise nicht initialisiert, ob er gültig ist, wird durch dasstate-Bit inRX_TASK_SET/TX_TASK_SETentschieden📎 tokio/src/sync/oneshot.rs:396-399。

Die Essenz dieses Designs：VALUE_SENTDas Bit zeigt nicht nur „Wert wurde gesendet“ an, sondern bestimmt auch, wem der Zugriff aufUnsafeCellgehört. Der Kommentar ist sehr klar📎 tokio/src/sync/oneshot.rs:1491-1496: WennVALUE_SENTgesetzt ist,UnsafeCellkann nur vom Empfänger zugegriffen werden; wenn nicht gesetzt, nur vom Sender. So wird mit einem einzigen atomaren Bit eine lockere Eigentumsübertragung erreicht, ohne zusätzliche Sperren.

sendDer Ablauf von

📎 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(())
}

Zuerst wird der Wert inUnsafeCellgeschrieben (zu diesem Zeitpunkt istVALUE_SENTnicht gesetzt, der Empfänger greift nicht zu), dann wirdcomplete()aufgerufen, um zu versuchen,VALUE_SENT。complete()zu setzen. Es ist eine CAS-Schleife:

📎 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)
}

Warum CAS statt eines einfachenfetch_or? Der Kommentar erklärt es klar📎 tokio/src/sync/oneshot.rs:1517-1529: Wenn der Kanal bereitsCLOSED, danndarfnichtVALUE_SENTerneut gesetzt werden. Denn sobald es gesetzt ist, geht der Empfänger davon aus, dass aufUnsafeCellzugegriffen werden kann, während der Sender gerade dabei ist, den Wert zurückzunehmen (consume_value), und gleichzeitiger Zugriff von beiden Seiten würde zu einem Datenrennen führen. Daher bricht die CAS-Schleife vorzeitig ab, wennCLOSEDentdeckt wird, und setzt nicht.

complete()Nach der Rückkehr vonRX_TASK_SET, wenn erfolgreich gesetzt und

📎 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
}

Kopierenpoll_recvDer

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

des Empfängers ist der Kern der Zustandsmaschine:is_complete()Er lädt zuerst den Zustand, wennconsume_valuedann direktis_closed()zurückgeben; wennErrgibtis_rx_task_set()zurück; andernfalls wird der Zweig „Waker registrieren“ betreten. Bei der Registrierung wird zuerstwill_wakegeprüft, wenn bereits gesetzt undis_complete()feststellt, dass es derselbe Waker ist, wird nicht erneut gesetzt; wenn unterschiedlich, wird zuerst unset und dann set. Hier gibt es eine subtile Race-Behandlung: Nach dem unset, wenn festgestellt wird, dasswahr geworden ist, muss das Flag 📎 tokio/src/sync/oneshot.rs:1342-1344wieder gesetzt werden

, sonst leckt der Waker beim Drop (da Drop vom Flag abhängt, um zu entscheiden, ob der Waker gedroppt werden soll).poll_closedDieses Muster „nach unset erneut set“ erscheint auch in📎 tokio/src/sync/oneshot.rs:839-848, es ist die Standardmethode von oneshot zur Behandlung gleichzeitiger Aufweckungen.

mpsc::bounded: Semaphor-gesteuerte Backpressure

Die Kapazitätssteuerung von mpsc wird vollständig dem Semaphor überlassen.channelDie Funktion erstellt ein Semaphor, dessen Anzahl an Berechtigungen der Puffergröße entspricht:

📎 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)
}

Semaphoreist eine interne Verpackung von mpsc, die gleichzeitig das zugrunde liegende Semaphor undbound(maximale Kapazität) hält📎 tokio/src/sync/mpsc/bounded.rs:176-179。boundwird fürmax_capacity-Abfragen verwendet, währendavailable_permitsdie aktuelle Kapazität angibt📎 tokio/src/sync/mpsc/bounded.rs:591-593。

Der Sendepfadsendzuerstreservedannsend：

📎 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)),
    }
}

reserveruft internreserve_inner(1)auf, letzteres prüft zuerstn > max_capacityund gibt direkt einen Fehler zurück, dannacquire(n) 📎 tokio/src/sync/mpsc/bounded.rs:1272-1311. Hier gibt es einen raffiniertenWakeReceiverOnDrop-Guard:

📎 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();
        }
    }
}

Der Kommentar erklärt die Motivation📎 tokio/src/sync/mpsc/bounded.rs:1279-1285: Wennreservenach Erhalt eines Teils der Berechtigungen abgebrochen wird (z. B.select!verliert), gibt das zugrunde liegendeAcquirediese Berechtigungen beim Drop zurück, aberbenachrichtigtnichtPermitwiemem::forget(guard)den Empfänger. Wenn der Kanal zu diesem Zeitpunkt geschlossen und leer ist, könnte der Empfänger nie die Benachrichtigung „Kanal geschlossen“ erhalten. Dieser Guard holt diese Aufweckung beim Drop nach. Bei Erfolg wird der Guard mit📎 tokio/src/sync/mpsc/bounded.rs:1306-1306abgebrochenPermit, da der Erfolgspfad von

Permitdie Benachrichtigungsverantwortung übernimmt.

📎 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::sendmacht dasselbe:mem::forgetKopieren📎 tokio/src/sync/mpsc/bounded.rs:1721-1728。

überspringt Drop mitrecv, um die Rückgabe der Berechtigungen zu vermeidenpoll_fnDer Empfangspfadchan.recv(cx) 📎 tokio/src/sync/mpsc/bounded.rs:243-246。poll_recvverwendet📎 tokio/src/sync/mpsc/bounded.rs:650-652, umchanzu verpackenchan::Rxund direkt ansendzu delegieren. Die eigentliche Warteschlangenlogik befindet sich im

try_send-Modul (in diesem Kapitel nicht behandelt), aber man kann ableiten: Der Empfänger-Waker wird in

📎 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_acquireaufruft, wird aufgeweckt.Closedzeigt den nicht-blockierenden Pfad:FullKopieren

Die beiden Fehler von

werden präzise auf📎 tokio/src/sync/mpsc/bounded.rs:776-784：sendundselect!abgebildet, wodurch „Kanal geschlossen“ und „Puffer voll“ als zwei Fehlerarten unterschieden werden.Designüberlegungen: Abbruchsicherheit und NachrichtenverlustDie mpsc-Dokumentation betont wiederholt AbbruchsicherheitreserveWennPermitinsendverliert,Permitwird die Nachricht verworfensend. Um Verlust zu vermeiden, muss man

recvverwenden, um📎 tokio/src/sync/mpsc/bounded.rs:199-204zu erhalten, dannrecv– daselect!bereits Kapazität reserviert hat,recvist synchron und wird nicht unterbrochen.poll_recvist abbruchsicherReady，Pending: Wenn

oneshotinReceiververliert, wird garantiert keine Nachricht konsumiert. Dies liegt daran, dass📎 tokio/src/sync/oneshot.rs:246-251vononeshotnur dannsendzurückgibt, wenn tatsächlich eine Nachricht abgerufen wurde, und beiErrdie Warteschlange nicht verändert.

Das

Falle eins: Asynchroner Mutex zum Schutz reiner Daten.Die Dokumentation empfiehlt ausdrücklich📎 tokio/src/sync/mutex.rs:26-36: Wenn es sich um reine Daten handelt (ohne.awaitAnforderung), iststd::sync::Mutexoderparking_lotschneller. Der Overhead des asynchronen Mutex liegt in den atomaren Operationen des Semaphors und der möglichen Task-Scheduling. Nur wenn während des Haltens der Sperre.awaiterforderlich ist (z. B. Zugriff auf eine Datenbankverbindung unter Sperre), sollte ein asynchroner Mutex verwendet werden.

Falle zwei: Sperre über.awaithinweg halten führt zu Deadlock.Dies ist die gefährlichste Falle des asynchronen Mutex. Wenn Task A nach dem Erwerb der Sperre auf ein Ereignis wartet,.awaitdas von Task B abgeschlossen werden muss, und Task B auf dieselbe Sperre wartet, entsteht ein Deadlock.std::sync::MutexDer Guard vonSendist nicht.await(in beweglichen Tasks), der Compiler verhindert das Halten überSend 📎 tokio/src/sync/mutex.rs:314-314hinweg; aber der Guard des asynchronen Mutex ist

, der Compiler hindert Sie nicht, Sie müssen selbst sicherstellen, dass keine zirkuläre Wartesituation entsteht.reserveFalle drei:send。 PermitNach📎 tokio/src/sync/mpsc/bounded.rs:1732-1745vergessenes

Der Drop gibt die Erlaubnis zurückoneshot, daher wird keine Kapazität verloren gehen. Aber wenn der Kanal geschlossen und leer ist, weckt Drop den Empfänger – dieses Wecken ist notwendig, sonst könnte der Empfänger möglicherweise nie die Schließbenachrichtigung erhalten.pollFalle vier:Pending。Das📎 tokio/src/sync/oneshot.rs:236-242vonpollkann fälschlicherweisePendingDie Dokumentation erklärt

: Selbst wenn die Nachricht gesendet wurde,forget_permitskann forget_permits(n)zurückgeben📎 tokio/src/sync/semaphore.rs:576-578. Dies ist kein Bug, sondern ein normales Phänomen unter Konkurrenzbedingungen – der Aufrufer wird geweckt und versucht es erneut, die Nachricht geht nicht verloren, nur verzögert.

Falle fünf:

Die Semantik vontokio::sync.Versucht, n Erlaubnisse zu reduzieren, gibt die tatsächlich reduzierte Anzahl zurück。

• Mutex. Es blockiert nicht und weckt keine Wartenden – es „schluckt" einfach die Erlaubnisse. Wird verwendet, um die Semaphor-Kapazität dynamisch zu verkleinern.MutexGuardZusammenfassung dieses Kapitelsrelease(1)Dieses Kapitel enthüllt
• Semaphoredas Kernmuster vonSemaphorePermit:permitsAlle asynchronen Warteprimitive basieren auf „Warteschlange + Waker-Aufwecken", und die konkrete Implementierung der Warteschlange variiert je nach Szenarioforget/merge/split，MAX_PERMITSWiederverwendung eines Semaphors mit Erlaubniszahl 1,
• oneshothält nur Referenzen, bei DropAtomicUsize, FIFO-fair aber nicht vergiftend.VALUE_SENTist Erlaubniszähler + Warteschlange,UnsafeCellverwendetCLOSEDZählung zur Unterstützung von
• mpsc::boundedRechtsverschiebung um 3 Bits, um Platz für Statusflags zu lassen.WakeReceiverOnDropVerwendet ein einzelnes

Bit-Flag zur Kodierung des Status,

auf. Wenn man es ändern würde, um ein Erlaubnisobjekt zu halten, würden mehrere Probleme entstehen. Erstens,tokio::syncdiespawn_blocking-Methodenfamilie muss den Guard inblock_onzerlegen, um nur das Unterfeld

Die Ablageposition des Wakers variiert je nach Primitiv: Bei Mutex/Semaphore liegt er in der Warteschlange des zugrunde liegenden Semaphors, bei oneshot in den Feldern tx_task/rx_task von Inner, bei mpsc in den Sende- und Empfangswarteschlangen des chan-Moduls. Der Weckmechanismus ist jedoch einheitlich: Bei einer Zustandsänderung wird der Waker entnommen und wake_by_ref aufgerufen, woraufhin der Executor die Aufgabe neu einplant. Damit sind Warten und Aufwecken innerhalb asynchroner Primitive klar erkennbar. Doch nicht jeder Code lässt sich asynchronisieren – das nächste Kapitel untersucht, wie blockierende Operationen mit spawn_blocking überbrückt werden und wie block_on Futures in einem nicht-asynchronen Kontext antreibt.

Im vorherigen Kapitel haben wir gesehen, dass asynchrone Mutexe und Kanäle beim Warten keinen Thread belegen, weil sie den Waker in eine Warteschlange legen und die Aufgabe erst nach Erfüllung der Bedingung vom Wecker neu eingeplant wird. All dies setzt jedoch voraus, dass eine Aufgabe bei Pending den Thread aktiv freigibt. Sobald Code std::fs::read, libsqlite3 oder eine reine CPU-Kompressionsschleife aufruft, blockiert er den Worker-Thread bis zur Rückkehr, und alle anderen Aufgaben auf diesem Thread verhungern. Tokios Lösung besteht darin, solche Arbeiten an einen separaten blockierenden Threadpool auszulagern und mit block_on Futures in einem nicht-asynchronen Kontext anzutreiben. Dieses Kapitel zerlegt diese beiden Grenzen.

8.1 Speicherlayout des blockierenden Threadpools: Inner und die Queue mit zwei Implementierungen

Intuitives Modell：spawn_blockingDer Threadpool gleicht einem „ausgelagerten Hilfskräfte-Pool“ eines Restaurants. Die Kellner (Worker-Threads) nehmen nur Bestellungen auf und servieren; bei Gerichten, die lange schmoren müssen, schreiben sie einen Arbeitsauftrag und werfen ihn in das Durchreichefenster (Queue) der Küche, wo die Hilfskräfte (blockierende Threads) den Auftrag entgegennehmen. Ohne diesen Pool müsste der Kellner selbst kochen, und das ganze Restaurant stünde still.

Kernstruktur. Der gesamte Pool wird vonBlockingPoolgehalten, das nur zwei Dinge speichert: ein klonbaresSpawner(Einlieferungseingang) und einshutdown_rx(Empfangsende des Shutdown-Signals)📎 tokio/src/runtime/blocking/pool.rs:20-23。Spawnerist internArc<Inner>, alle Einlieferer teilen denselben Zustand📎 tokio/src/runtime/blocking/pool.rs:26-28。

Innerist der gesamte Zustand des Pools, die Felder verdienen eine Einzelbetrachtung📎 tokio/src/runtime/blocking/pool.rs:77-104：

• inner_impl: InnerImpl: Implementierung von Queue + Benachrichtigung + Lock-Topologie, ein Enum mit den beiden VariantenLockedundSharded. Dies ist die entscheidendste Abstraktion dieses Kapitels – sie vereint die beiden Topologien „Single-Lock-Queue“ und „Sharded Queue“ unter einer Schnittstelle.📎 tokio/src/runtime/blocking/pool.rs:107-110: Obergrenze der Threadanzahl, also
• thread_cap: usize: Anzahl der Scheduler-Worker-Threads, wird verwendet, um sie in den Metriken abzuziehen, sodassmax_blocking_threads。
• scheduler_threads: usizenur blockierende Threads zähltnum_blocking_threads: Überlebensdauer eines Leerlauf-Threads, Standard📎 tokio/src/runtime/blocking/pool.rs:455-460。
• keep_alive: Duration: drei atomare Zähler –KEEP_ALIVE = 10s 📎 tokio/src/runtime/blocking/pool.rs:231。
• metrics: SpawnerMetrics〔Design-Inferenz und Architekturabwägung〕num_threads、num_idle_threads、queue_depth 📎 tokio/src/runtime/blocking/pool.rs:31-35。

wird separat herausgezogen, damit beide Queue-Implementierungen ihn wiederverwenden können。ThreadManagementState: Shutdown-Flag.📎 tokio/src/runtime/blocking/pool.rs:135-150：

• shutdown: bool: Jeder Worker-Thread hält einen Klon; nach dem Drop aller
• shutdown_tx: Option<shutdown::Sender>wird eine Benachrichtigung empfangen.shutdown_rx: Handle des zuletzt durch Timeout beendeten Threads.
• last_exiting_thread: Option<JoinHandle<()>>: Handles aller lebenden Worker.
• worker_threads: HashMap<usize, JoinHandle<()>>: Monoton steigender Thread-ID-Zuteiler.
• worker_thread_index: usizeDie Design-Motivation ist im Kommentar klar beschrieben: Ein durch Timeout beendeter Thread joint den zuletzt durch Timeout beendeten Thread, um Valgrind-Fehlerkennungen zu vermeiden

last_exiting_threadGenau das ist die Implementierung dieses verketteten Joins – er entfernt sein eigenes Handle und tauscht das alte📎 tokio/src/runtime/blocking/pool.rs:135-150。worker_timed_outaus und gibt es an den Aufrufer zum Joinen zurücklast_exiting_threadAufgabenkapselung📎 tokio/src/runtime/blocking/pool.rs:172-178。

. In der Queue liegt, das einTaskund einUnownedTask<BlockingSchedule>-Flag umschließtMandatoryentscheidet, ob diese Aufgabe beim Shutdown verworfen oder zwangsweise ausgeführt wird:📎 tokio/src/runtime/blocking/pool.rs:187-191。Mandatoryruft beishutdown_or_run_if_mandatoryauf, beiNonMandatoryruftshutdown()auf. Das ist der Unterschied zwischenMandatory(nicht erzwungen) undrun() 📎 tokio/src/runtime/blocking/pool.rs:223-228(erzwungen, für fs verwendet)spawn_blockingSpeicherlayout der Single-Lock-Implementierungspawn_mandatory_blockingist die ursprünglichste Topologie: ein📎 tokio/src/runtime/blocking/pool.rs:233-265。

plus ein。LockedImplenthältMutex<LockedInner>undCondvar 📎 tokio/src/runtime/blocking/pool.rs:113-116。LockedInner. Beachte, dassVecDeque<Task>、num_notify: u32undthread_mgmt_state 📎 tokio/src/runtime/blocking/pool.rs:118-124unter demselben Lock liegen, währendnum_notifyeine atomare Größe außerhalb des Locks ist – dieses gemischte Layout „teils Zustand im Lock, teils außerhalb“ ist die Wurzel aller späteren Nebenläufigkeitsfeinheiten.thread_mgmt_state8.2 Einlieferungspfad: von spawn_blocking bis zum Thread-Weckennum_idle_threadsSzenario

: Ein asynchroner Task ruft

auf – was geschieht in diesem Moment?Erster Schritt: Boxing-Entscheidung und Aufgabenerstellungtokio::task::spawn_blocking(move || heavy_compute(data))misst zunächst die Closure-Größe

und entscheidet anhand von。Spawner::spawn_blocking, ob die Closurefn_sizewirdAutoBox::<F>::SHOULD_BOX. Dies ist Tokios allgemeine Strategie „automatisches Boxing großer Futures“: Bei zu großen Closures wird geboxt, um eine Aufblähung der Aufgabenstruktur zu vermeiden.BoxBeim Eintritt in📎 tokio/src/runtime/blocking/pool.rs:359-389wird zuerst eine Aufgaben-ID vergeben, dann die Closure mit

zu einem Future verpackt und schließlich mitspawn_blocking_innereinblocking_taskundtask::unownedkonstruiert. Beachte, dass hier einUnownedTask-Tupel zurückgegeben wird – Handle und Einlieferungsergebnis werden getrennt zurückgegeben.JoinHandle 📎 tokio/src/runtime/blocking/pool.rs:440-449Zweiter Schritt: Drei Behandlungen des Einlieferungsergebnisses(JoinHandle<R>, Result<(), SpawnError>). Zurück zu

, es wird ein Match aufdurchgeführtspawn_blocking: Normal, Handle zurückgeben.spawn_result 做匹配 📎 tokio/src/runtime/blocking/pool.rs:381-388：

• Ok(())：正常，返回句柄。
• Err(ShuttingDown)：Kein Panic, gibt trotzdem ein Handle zurück. Der Kommentar erklärt, dass dies aus Kompatibilitätsgründen geschieht – das Handle wird niemals aufgelöst, aber der Aufrufer stürzt nicht ab, weil die Laufzeit gerade heruntergefahren wird.
• Err(NoThreads(e)): Das OS kann keinen Thread erstellen und niemand im Pool übernimmt, direktes Panic.

Dritter Schritt: Einreihung und Aufweckentscheidung。spawn_taskÜbergibt dason_no_idleClosure anInnerImpl::spawn_task, wobei die konkrete Implementierung entscheidet, wann es aufgerufen wird📎 tokio/src/runtime/blocking/pool.rs:462-506. Betrachtet manLockedImpl::spawn_taskden kritischen Abschnitt von📎 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();
}

Hier gibt es zwei entscheidende Punkte. Erstens: Die Shutdown-Prüfung erfolgt vor der Einreihung, und selbst wenn die AufgabeMandatoryist, wird sie direktshutdown()– der Kommentar erklärt: Sie wurde erst nach Beginn des Shutdowns eingeplant, daher ist das Verwerfen legitim📎 tokio/src/runtime/blocking/pool.rs:614-620. Zweitens: Die Aufweckentscheidung hängt vomnum_idle_threadsaußerhalb des Locks ab: Ist es 0, wirdon_no_idleaufgerufen, um einen neuen Thread zu starten; andernfalls wird der Idle-Zähler dekrementiert undnum_notify、notify_one。

num_notifyinkrementiert. Warum mussexistieren?CondvarWeilnotify_onespurious wakeups (falsche Aufwachvorgänge) erzeugen kann. Wenn man nurnum_notifyohne Zählung verwendet, könnte ein fälschlich aufgeweckter Thread irrtümlich glauben, es gäbe Aufgaben zu holen, stellt dann fest, dass die Warteschlange leer ist, und schläft wieder ein – während der tatsächlich aufgeweckte Thread möglicherweise nie benachrichtigt wird.+1Verwandelt „legitime Aufwachvorgänge" in zählbare Tokens: Die einreichende Seitenum_notify != 0, die aufgeweckte Seite betrachtet den Aufwachvorgang erst bei-1 📎 tokio/src/runtime/blocking/pool.rs:674-684。

als legitim und。on_no_idleVierter Schritt: Neuen Thread starten📎 tokio/src/runtime/blocking/pool.rs:462-506Das Closure wird unter Halten des Queue-Locks ausgeführtnum_threads == thread_cap. Es prüft zuerstOk(()), und bei Erreichen des Limits wird direkt zurückgekehrtshutdown_tx– die Aufgabe bleibt in der Warteschlange und wartet auf die Bearbeitung durch vorhandene Threads; das ist Backpressure. Andernfalls wirdspawn_threadgeklont,num_threadsaufgerufen, um einen Thread zu erstellen, und bei Erfolgworker_thread_indexinkrementiert,worker_threads。

spawn_threadinkrementiert und das Handle inthread::Buildereingefügtrt.enter()Mitinner.run(id)werden Thread-Name und Stack-Größe festgelegt, dann wird ein Closure gespawnt: Eintritt in den Laufzeitkontextshutdown_tx 📎 tokio/src/runtime/blocking/pool.rs:508-528。

, Aufruf von。spawn_thread, schließlich Drop📎 tokio/src/runtime/blocking/pool.rs:488-500Fehlertoleranz bei OS-Thread-ErstellungsfehlernWouldBlockkann fehlschlagen. Der Code klassifiziert den Fehleris_temporary_os_thread_error: Wenn es sich um📎 tokio/src/runtime/blocking/pool.rs:750-752handelt (temporärer Fehler, bestimmt durch) und bereits blockierende Threads im Pool vorhanden sind, dannstillschweigend ignorierenSpawnError::NoThreads– die Aufgabe wird schließlich von einem aktuell beschäftigten Thread übernommen. Andernfalls wird

zurückgegeben, was letztendlich zu einem Panic führt.

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"]

Kopie

8.3 Worker-Hauptschleife: BUSY/IDLE-Zustandsmaschine und Timeout-RückgewinnungIntuitives Modellkeep_alive: Jeder blockierende Thread ist ein „Bereitschaftshelfer". Bei Aufträgen wird kontinuierlich gearbeitet (BUSY), ohne Aufträge wird geschlafen (IDLE), und nach Überschreiten von

wird Feierabend gemacht (Timeout-Beendigung). Ohne Timeout-Rückgewinnung würde der Pool dauerhaft alle bei Spitzenlast erstellten Threads behalten, was Speicher und Kernel-Scheduling-Overhead verschwendet.。LockedImpl::run_workerStruktur der Hauptschleife'mainist eine📎 tokio/src/runtime/blocking/pool.rs:642-735Schleife, die intern abwechselnd die beiden Phasen BUSY und IDLE durchläuft. Hinweis: BUSY/IDLE sind hierPhasen

innerhalb der Schleife, keine expliziten Enum-Zustände, daher wird im Folgenden ein Flussdiagramm statt eines Zustandsdiagramms verwendet.BUSY-Phasewhile let Some(task) = locked.queue.pop_front(): Die innere📎 tokio/src/runtime/blocking/pool.rs:655-661holt kontinuierlich Aufgabenqueue_depth，. Nach dem Abholen wirddekrementiert,task.run()das Lock gedroppt

, die Ausführungdurchgeführt und dann das Lock erneut erworben. Der Schritt des Lock-Drops ist entscheidend – blockierende Aufgaben können sehr lange laufen und dürfen niemals unter Lock-Haltung ausgeführt werden.num_idle_threadsIDLE-Phaseis_counted_idle = true: Die Warteschlange ist leer,📎 tokio/src/runtime/blocking/pool.rs:663-696wird inkrementiert,condvar.wait_timeout(locked, keep_alive)gesetzt, dann wird die Warteschleife

1. num_notify != 0betreten. Kern istnum_notify; nach der Rückkehr werden drei Dinge geprüft:is_counted_idle = false: Legitimer Aufwachvorgang. Dekrementiertnum_idle_threads, setzt📎 tokio/src/runtime/blocking/pool.rs:674-684。

(da die einreichende Seite bereitsworker_timed_outdekrementiert hat), break zurück zu BUSYbreak 'main2. Nicht heruntergefahren und Timeout: Aufruf von📎 tokio/src/runtime/blocking/pool.rs:689-693。

, um das Handle des zuletzt beendeten Threads zu erhalten,

die Schleife verlassen3. Andernfalls handelt es sich um einen spurious wakeup, weiter warten.thread_mgmt_state.shutdownLeeren der Warteschlange beim Shutdown📎 tokio/src/runtime/blocking/pool.rs:698-710. Wenntask.shutdown_or_run_if_mandatory()wahr ist, wird die Leerungslogik betreten

: Aufgaben werden einzeln herausgepoppt, das Lock gedroppt,aufgerufen – nicht-erzwungene Aufgaben werden verworfen, erzwungene Aufgaben wie gewohnt ausgeführt. Dann break, um die Hauptschleife zu verlassen.num_threads 📎 tokio/src/runtime/blocking/pool.rs:714Bereinigung beim Beendenis_counted_idle. Vor dem Thread-Ende wirdnum_idle_threadsdekrementiert. Wennassert_ne!(prev_idle, 0)wahr ist, muss außerdem📎 tokio/src/runtime/blocking/pool.rs:716-726dekrementiert werden, und mitnum_idle_threadswird per Assertion geprüft, dass kein Unterlauf stattfindet

. Diese Assertion ist ein Schutzgeländer in der Debug-Phase: Sobald die Buchführung vonnum_threads == 0fehlerhaft ist, wird hier sofort ein Panic ausgelöst, anstatt den Fehler stillschweigend zu propagieren.notify_oneSchließlich, wenn gerade heruntergefahren wird und📎 tokio/src/runtime/blocking/pool.rs:728-730(der letzte Thread),join_on_threadwird der möglicherweise wartende Shutdown-InitiatorInner::runaufgeweckt. Gibt📎 tokio/src/runtime/blocking/pool.rs:755-771。

zurück, und。BlockingPool::shutdownjoint vor dem Beendenbegin_shutdownShutdown-Handshake📎 tokio/src/runtime/blocking/pool.rs:310-312。LockedImpl::begin_shutdownruft zuerstshutdown_tx、notify_allauf, um alle Worker-Handles zu erhalten📎 tokio/src/runtime/blocking/pool.rs:740-745setzt das Shutdown-Flag, dropptshutdown_rx.wait(timeout)weckt alle wartenden Threads auf📎 tokio/src/runtime/blocking/pool.rs:324。

shutdown::Receiver::wait. Dann blockiert📎 tokio/src/runtime/blocking/shutdown.rs:37-70und wartet auftimeout == 0Die Implementierung vontry_enter_blocking_region()ist sehr durchdacht📎 tokio/src/runtime/blocking/shutdown.rs:44-57: Zuerst wird der schnelle Pfad vonblock_on_timeoutbehandelt und direkt false zurückgegeben; dann wirdblock_onaufgerufen, um den Blockierbereich zu betreten; bei Fehler und aktuellem Panic wird false zurückgegeben, andernfalls Panic mit dem Hinweis „Runtime kann nicht in einem asynchronen Kontext gedroppt werden"

shutdown_tx. Schließlich wird je nach TimeoutArc<oneshot::Sender<()>>oder📎 tokio/src/runtime/blocking/shutdown.rs:12-14aufgerufen, um das Oneshot anzutreiben.ArcDer Mechanismus vononeshot::Senderist: Jeder Worker-Thread hält einen Klon vonReceiver. Nachdem alle Threads beendet sind, werden alle Klone gedroppt,

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 句柄

wird gedroppt,

erhält die Benachrichtigung. Das ist das klassische Muster „Receiver wird aufgeweckt, nachdem alle Sender gedroppt wurden".：block_onKopiemain8.4 block_on: Future in einem nicht-asynchronen Kontext antreiben

Intuitives Modell。Runtime::block_onist das „Haupttor" der Laufzeit. Es verwandelt den aktuellen Thread in einen temporären Executor und pollt das übergebene Future wiederholt, bis es abgeschlossen ist. Ohne es könnte dieSHOULD_BOXFunktion keinen asynchronen Code starten.Box::pinEinstieg und Boxingblock_on_inner 📎 tokio/src/runtime/runtime.rs:343-350。block_on_innerZuerst wird ebenfalls die Größe gemessen und je nachself.enter()entschieden, ob📎 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),
}

. Darin befinden sich zwei bedingt kompilierte Trace-Wrapper (taskdump und tracing), dannblock_onDie Semantik ist unterschiedlich, die Dokumentation sagt es ganz klar📎 tokio/src/runtime/runtime.rs:302-320：

• Multithread-Scheduler: Futures laufen im Kontext des I/O-Treibers und Timers,block_onnach der Rückkehr laufen bereits gespawnte Tasks weiter.
• Current-Thread-Scheduler：block_onkann von mehreren Threads gleichzeitig aufgerufen werden, der erste Aufrufer erlangt das Eigentum am I/O- und Timer-Treiber, andere Threads „hängen sich ein".block_onNach Abschluss des ersten können andere Threads den Treiber „stehlen".block_onNach der Rückkehr werden bereits gespawnte Tasks angehalten, ein erneuter Aufruf vonblock_onsetzt sie fort.

Kritische Einschränkung: Darf nicht in einem asynchronen Kontext aufgerufen werden. Die Dokumentation stellt klarblock_on, dass ein Aufruf in einem asynchronen Ausführungskontext panicked📎 tokio/src/runtime/runtime.rs:321-324. Der Grund ist unmittelbar:block_onblockiert den aktuellen Thread, bis das Future abgeschlossen ist. Wenn der aktuelle Thread selbst ein Worker-Thread ist, wird der gesamte Executor blockiert – genau das ist das Problem, dasspawn_blockinglösen soll, daher schließen sich beide gegenseitig aus.

Herunterfahrpfad。Runtime::dropwird je nach Scheduler-Typ verteilt📎 tokio/src/runtime/runtime.rs:506-521: Der Current-Thread-Scheduler muss zuersttry_set_currentin den Kontext eintreten und dann herunterfahren (um sicherzustellen, dass Tasks im Laufzeitkontext gedroppt werden); der Multithread-Scheduler fährt direkt herunter (Worker-Threads befinden sich bereits im Kontext).shutdown_timeoutZuerst den Scheduler herunterfahren, dann den Blocking-Pool📎 tokio/src/runtime/runtime.rs:457-461，shutdown_backgroundist äquivalent zushutdown_timeout(Duration::from_nanos(0)) 📎 tokio/src/runtime/runtime.rs:494-496。

Designüberlegungen, Fehlerbehandlung und Produktions-Fallstricke

Warumspawn_blocking'sShuttingDownnicht panicked? 📎 tokio/src/runtime/blocking/pool.rs:383-384Der Kommentar nennt Kompatibilitätsgründe.spawn_blockinggibtJoinHandlestattResultzurück. Ein Panic beim Herunterfahren würde einen vorhersehbaren Zustand – „die Laufzeit fährt gerade herunter" – in einen Absturz verwandeln. Die Rückgabe eines Handles, das nie aufgelöst wird, führt dazu, dass der Aufrufer beiawaitdauerhaft hängt – aber zu diesem Zeitpunkt ist die Laufzeit bereits heruntergefahren, und das gesamteblock_onwird ebenfalls beendet, sodass es praktisch nicht zu einem dauerhaften Leck kommt.

max_blocking_threadsDie Backpressure-Semantik von. Der Standardwert ist sehr groß (512), weilspawn_blockinghäufig für Datei-I/O verwendet wird. Die Dokumentation warnt jedoch: Bei CPU-intensiven Tasks muss die Parallelität mit einem Semaphore begrenzt werden, sonst werden massenhaft Threads erstellt📎 tokio/src/task/blocking.rs:94-100. Nach Erreichen des Limits werden Tasks in der Warteschlange eingereiht, was Backpressure erzeugt – aber beachten Sie, dass diese Backpressure nur auf den Blocking-Pool wirkt und nicht auf den asynchronen Scheduler zurückdrückt.

spawn_blockingist nicht abbrechbar. Die Dokumentation stellt klar:aborthat keine Wirkung auf bereits laufende Blocking-Tasks, die Tasks laufen weiter bis zum Ende📎 tokio/src/task/blocking.rs:106-120. Nur noch nicht gestartete Tasks können durch abort verhindert werden. Beim Herunterfahren wartet die Laufzeit auf alle bereits gestarteten Blocking-Tasks,shutdown_timeoutnach dem Timeout werden diese Threads geleakt.

num_idle_threadsDie Buchhaltungsfalle von。is_counted_idleDie Existenz des Flags zeigt, dass dieser Zähler leicht fehleranfällig ist. Die einreichende Seite dekrementiert beim Aufweckennum_idle_threads, die aufgeweckte Seite siehtnum_notify != 0und setzt dannis_counted_idle = false, um ein doppeltes Dekrementieren zu vermeiden📎 tokio/src/runtime/blocking/pool.rs:679-682. Wenn dieser Pfad einen Bug hat,assert_ne!(prev_idle, 0)panicked📎 tokio/src/runtime/blocking/pool.rs:722-725beim Beenden. Wenn in der Produktion „num_idle_threadsunderflowed on thread exit" auftritt, ist die Buchhaltungslogik des Pools beschädigt.

InnerImplDie Bedeutung der Enum-Abstraktion. Der Kommentar erklärt, dassLocked-Variante sich exakt wie vor dem Refactoring verhält, während dieSharded-Variante symmetrische Slots für zukünftige nebenläufige Warteschlangen reserviert📎 tokio/src/runtime/blocking/pool.rs:537-539。spawn_task、run_worker、begin_shutdownAlle drei Methoden werden über Enum verteilt📎 tokio/src/runtime/blocking/pool.rs:548-582. Dieses Design aus „Enum-Verteilung + jeder Variante mit eigener kritischer Sektion" ermöglicht es, neue Warteschlangen-Topologien hinzuzufügen, ohne die Aufrufer ändern zu müssen.

Kapitelzusammenfassung

Dieses Kapitel hat die beiden Grenzen von Tokio für die Aufnahme von synchronem Code analysiert.spawn_blockingClosures werden an einen separaten Blocking-Thread-Pool übergeben:Innerhält Warteschlange, Thread-Limit, Lebensdauer und atomare Metriken;LockedImplverwendet Single-Lock +Condvarzur Implementierung der Warteschlange,num_notifyder Zähler kompensiert falsche Aufweckungen; Worker zirkulieren zwischen BUSY/IDLE, nach Idle-Timeout beenden sie sich durch verkettetes join;max_blocking_threadsnach Erreichen des Limits werden Tasks eingereiht und bilden Backpressure.block_ontreibt Futures in einem nicht-asynchronen Kontext an, Multithread- und Current-Thread-Scheduler haben unterschiedliche Semantik und dürfen keinesfalls in einem asynchronen Kontext aufgerufen werden. Der Herunterfahrpfad wird durchshutdown_tx'sArc-Zählerstand auf Null ausgelöst, wasoneshottriggert und einen Handshake implementiert: „Nachdem alle Worker beendet sind, wird der Initiator des Herunterfahrens aufgeweckt".

Kapitel-Überlegungen und Selbsttest

Der blockierende Thread-Pool und block_on markieren die Fähigkeitsgrenzen der asynchronen Runtime: Ersterer isoliert Arbeit, die den Thread nicht abgeben kann, in dedizierte Threads, Letzterer ermöglicht es auch nicht-asynchronen Einstiegspunkten, Futures anzutreiben. Aber diese beiden Grenzen werden im Code oft nicht von Hand geschrieben – im nächsten Kapitel betreten wir die Welt der Makros und schauen, wie #[tokio::main], select! und join! diesen Runtime-Code zur Kompilierzeit generieren.

Im vorherigen Kapitel haben wir gesehen,block_onund wie der blockierende Thread-Pool die Fähigkeitsgrenzen der asynchronen Runtime absteckt, während Nutzer diese Grenzen fast nie von Hand schreiben – sie schreiben#[tokio::main]、select!、join!, damit das Makro diesen Boilerplate-Code zur Kompilierzeit ausbreitet. Makros sind die erste Zuckerhülle, die Tokio dem Nutzer bietet, und auch der Ort, an dem zur Kompilierzeit tatsächlich Runtime-Code generiert wird. Dieses Kapitel konzentriert sich auftokio-macrosCrate undtokio/src/macros/select.rs, zerlegt die drei am häufigsten verwendeten Makro-Expansionspfade und beantwortet vor allem eine Frage: Wie sieht die reale Aufrufkette nach der Makro-Expansion aus, und warumselect!muss die Cancel-Safety-Semantik gesondert beachtet werden.

9.1 #[tokio::main]: async fn in Runtime::block_on umschreiben

Intuitives Modell：#[tokio::main]Es ist wie eine „Renovierungsvollmacht“. Du übergibst einen Rohbau (async fn main), es verlegt für dich Wasser und Strom (baut die Runtime), montiert Türen und Fenster (enable_all) und stellt schließlich deine ursprünglichen Möbel (den Funktionskörper) hinein. Ohne es müsste jedermainvon HandBuilder::new_multi_thread().enable_all().build().unwrap().block_on(...)schreiben, und Boilerplate-Code würde die Geschäftslogik überschwemmen.

Datenstrukturen und Speicherlayout

Das Makro selbst erzeugt keine Runtime-Datenstrukturen, aber die von ihm geparste Konfiguration wird in zwei Strukturen untergebracht.Configurationist ein „veränderlicher Akkumulator zur Parse-Zeit“, dessen Felder alleOptionsind, weil Attributparameter fehlen, doppelt vorkommen oder ungültig sein können📎 tokio-macros/src/entry.rs:74-84. Beachte, dassworker_threads、start_paused、unhandled_panicalleSpantragen – dies dient dazu, Fehler bei der Fehlermeldung auf die vom Nutzer geschriebene Zeile zu lokalisieren und nicht auf das Makro-Innere📎 tokio-macros/src/entry.rs:74-84。FinalConfigist dagegen das „validierte unveränderliche Ergebnis“,flavorist nicht mehrOption, weilbuild()bereits mitdefault_flavorabgesichert wurde📎 tokio-macros/src/entry.rs:55-62。

RuntimeFlavorhat nur drei Varianten:CurrentThread、Threaded、Local 📎 tokio-macros/src/entry.rs:10-14。from_strenthält absichtlich freundliche Fehlermeldungen für historische Namen:single_threadweist darauf hin, dass escurrent_thread，basic_schedulerheißen sollte, weist darauf hin, dass es umbenannt wurde,threaded_schedulerweist darauf hin, dass es umbenannt wurde📎 tokio-macros/src/entry.rs:17-27. Dies ist ein typisches Design von Makros als „erste Kontaktfläche für den Nutzer“: Fehlermeldungen sind Dokumentation.

Schritt-für-Schritt-Expansionsablauf

Szenario einsetzen: Der Nutzer schreibt#[tokio::main(flavor = "multi_thread", worker_threads = 4)] async fn main() { ... }。

Erster Schritt,mainDer Einstieg parst zuerst das Item als benutzerdefiniertesItemFn 📎 tokio-macros/src/entry.rs:577-580. DiesesItemFnist nichtsyn::ItemFn, sondern ein von Tokio selbst implementierter Parser; der Grund steht im Kommentar: Er will nicht die gesamte Anweisung rekursiv parsen, sondern nur ein leichtgewichtiges Parsing „nach Token-Tree puffern, bei Semikolon trennen“ durchführen📎 tokio-macros/src/entry.rs:720-764. Dies vermeidet den Aufwand, im Makro einen vollständigen AST für den Funktionskörper aufzubauen.

Zweiter Schritt,build_configvalidiert, ob dasasyncSchlüsselwort existiert; fehlt es, wird „theasync keyword is missing" 📎 tokio-macros/src/entry.rs:346-349gemeldet. Danach werden die Attributparameter durchlaufen undworker_threads、flavor、start_paused、crate、unhandled_panic、namean den entsprechenden Setter verteilt📎 tokio-macros/src/entry.rs:369-399. Beachte, dasscore_threadsexplizit abgelehnt und auf die Umbenennung hingewiesen wird📎 tokio-macros/src/entry.rs:379-382。

Dritter Schritt,Configuration::buildführt eine feldübergreifende Konsistenzprüfung durch. Hier gibt es drei entscheidende Einschränkungen:worker_threadserlaubt nurmulti_thread 📎 tokio-macros/src/entry.rs:197-217；start_pausederlaubt nurcurrent_thread/local 📎 tokio-macros/src/entry.rs:219-229；unhandled_panicerlaubt ebenfalls nurcurrent_thread/local 📎 tokio-macros/src/entry.rs:231-241. Wenn der Nutzermulti_threadwählt, aber dasrt-multi-threadFeature nicht aktiviert ist, unterscheidet sich die Fehlermeldung danach, ob flavor explizit angegeben wurde📎 tokio-macros/src/entry.rs:209-216。

Vierter Schritt,parse_knobsgeneriert Code. Es entfernt zuerstasyncness 📎 tokio-macros/src/entry.rs:441und wählt dann je nach flavor den Builder-Ausgangspunkt:CurrentThread/LocalverwendetBuilder::new_current_thread()，ThreadedverwendetBuilder::new_multi_thread() 📎 tokio-macros/src/entry.rs:468-477。LocalDie Besonderheit liegt darin, dass der build-Aufrufbuild_local(Default::default())stattbuild() 📎 tokio-macros/src/entry.rs:479-483ist. Danach wird bei Bedarf verkettet.worker_threads(#v)、.start_paused(#v)、.unhandled_panic(...)、.name(#v) 📎 tokio-macros/src/entry.rs:485-497。

Fünfter Schritt, der endgültige Funktionskörper wird generiert. Der Kern istlast_block：return #rt.enable_all().#build.expect("Failed building the Runtime").block_on(body) 📎 tokio-macros/src/entry.rs:509-522. Beachte das explizitereturn; der Kommentar verweist auf tokio-rs/tokio#4636 und dient der Behebung eines Typinferenzproblems📎 tokio-macros/src/entry.rs:508。

Sechster Schritt, der Funktionskörper wird inasync #bodyverpackt und einer Typprüfung unterzogen. Im Nicht-Test-Pfad wird, wenn der Rückgabetyp nicht!ist undimpl Traitnicht enthält,if false { let _: &dyn Future<Output = #output_type> = &body; }eingefügt, um eine Compile-Zeit-Assertion durchzuführen📎 tokio-macros/src/entry.rs:551-571. Der Test-Pfad verwendetpin!Den body auf den Stack pinnen und inPin<&mut dyn Future>umwandeln,block_onDer Kommentar erklärt, dass dies dazu dient, den Kompilierungsaufwand für📎 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"]

Designüberlegungen und Fallstricke im Produktivbetrieb

mainundtestteilenparse_knobs, aber der Standard-Flavor ist unterschiedlich:testStandardCurrentThread，mainStandardThreaded 📎 tokio-macros/src/entry.rs:91-94. Das erklärt, warum#[tokio::test]standardmäßig single-threaded ist – Tests benötigen normalerweise keine Mehrkernunterstützung, und Single-Threaded ist leichter reproduzierbar.

Eine leicht zu übersehende Falle: Nach der Makro-Expansion wird bei jedem Funktionsaufruf eine neue Runtime erstellt. Die Dokumentation warnt ausdrücklich davor, dass bei häufig aufgerufenen Funktionen stattdessen ein Builder zur Wiederverwendung der Runtime verwendet werden sollte📎 tokio-macros/src/lib.rs:31-35. Die Verwendung von#[tokio::main]in einer normalen Funktion ist legal, aber jeder Aufruf kostet einmal die Runtime-Erstellung.

Eine weitere Falle ist die Umbenennung voncrate. Wenn der Benutzeruse tokio as tokio1, kann das standardmäßig vom Makro intern generiertetokio::runtime::Builderden Pfad nicht finden und muss explizitcrate = "tokio1" 📎 tokio-macros/src/lib.rs:239-264。parse_knobsincrate_pathDer Standardwert von istIdent::new("tokio", ...) 📎 tokio-macros/src/entry.rs:456-462, was genau die Ursache für Fehler im Umbenennungsszenario ist.

9.2 select!: Mehrzweig-Polling, Bitmaske und zufällige Fairness

Intuitives Modell：select!ist wie ein „Kellner, der gleichzeitig mehrere Ausgabefenster im Blick behält". Welches Fenster zuerst Essen ausgibt, von dort nimmt er es mit, und die Warteschlangen der anderen Fenster verfallen. Ohne es müsste der Benutzer manuellpoll_fnmehrere Futures in ein Tupel packen und einzeln pollen, und auch selbst die Logik handhaben, dass nach Bereitschaft eines Zweigs die übrigen Zweige verworfen werden müssen.

Datenstruktur und Speicherlayout

select!erzeugt nach der Expansion ein lokales Modul__tokio_select_util, das ein EnumOutund einen TypaliasMask 📎 tokio/src/macros/select.rs:615-619。Outenthält. Die Variantennamen sind_0、_1… einer pro Zweig, plus einDisabled, das anzeigt, dass alle Zweige ungültig sind📎 tokio-macros/src/select.rs:33-39。Mask. Der zugrunde liegende Typ wird dynamisch nach Zweiganzahl gewählt: ≤8 verwendetu8, ≤16 verwendetu16, ≤32 verwendetu32, ≤64 verwendetu64, über 64 führt direkt zu einem Panic📎 tokio-macros/src/select.rs:17-31. Diese Bitmaske ist der zentrale Zustand vonselect!: Bit i auf 1 bedeutet, dass der i-te Zweig deaktiviert wurde.

Alle Futures werden in einem Tupel gespeichertfutures, wobei jedes Element zuerst durchIntoFuture::into_futurekonvertiert wird📎 tokio/src/macros/select.rs:654-656. Beachten Sie, dass hier zuerstfutures_initkonstruiert und dann einzelninto_futurewird. Der Kommentar erklärt, dass dies die Verlängerung temporärer Lebensdauern ausnutzt📎 tokio/src/macros/select.rs:641-646. Anschließendlet mut futures = &mut futures;wird das Tupel auf veränderliche Referenzen herabgestuft, um zu vermeiden, dasspoll_fnClosures das Eigentum übernehmen📎 tokio/src/macros/select.rs:658-662。

Schritt-für-Schritt-Polling-Ablauf

Szenario einsetzen:select! { v = stream1.next() => ..., v = stream2.next() => ..., else => break }。

Erster Schritt: Makro-Eingangsregelabgleich. Fallsbiased;Präfix vorhanden,start=0 📎 tokio/src/macros/select.rs:801-803; andernfallsstartist ein zufälliger Ausdruckthread_rng_n(BRANCHES) 📎 tokio/src/macros/select.rs:805-809. Dies ist die Fairness-Quelle, die die Dokumentation als „standardmäßig zufällige Auswahl eines Zweigs zur Überprüfung" bezeichnet📎 tokio/src/macros/select.rs:61-65。

Zweiter Schritt: Normalisierung. Der tt-muncher normalisiert jeden Zweig in die Form(skip) pat = fut, if cond => handler,,skipist eine Folge von_, deren Länge der Anzahl der Branches vor diesem Zweig entspricht📎 tokio/src/macros/select.rs:770-793。skip. Sie wird sowohl zur Generierung des Tupelfeldzugriffsfutures_init.$($skip)*als auch zur Berechnung des Zweigindex durchcount!verwendet.

Dritter Schritt: Auswertung der Vorbedingungen. Für jedeif $cjedes Zweigs gilt: Falls false, danndisabled |= 1 << index 📎 tokio/src/macros/select.rs:631-636. Beachten Sie: Selbst wenn ein Zweig deaktiviert ist, wird sein$fut-Ausdruck weiterhin ausgewertet, nur nicht gepollt📎 tokio/src/macros/select.rs:39-41。

Vierter Schritt: Eintritt in diepoll_fn-Closure. Zuerst wird das Kooperationsbudget geprüft:ready!(poll_budget_available(cx)). Bei erschöpftem Budget wird direktPending 📎 tokio/src/macros/select.rs:664-667zurückgegeben. Dies stellt sicher, dassselect!den Worker nicht monopolisiert.

Fünfter Schritt: Schleifefor i in 0..BRANCHES，branch = (start + i) % BRANCHES 📎 tokio/src/macros/select.rs:680-685. Für jeden Branch: Zuerstdisabled & mask == maskprüfen; falls bereits deaktiviert, danncontinue 📎 tokio/src/macros/select.rs:694-699; andernfalls das Future aus dem Tupel entnehmen und mitPin::new_uncheckedumhüllen (die Sicherheit hängt davon ab, dass das Future auf dem Stack liegt und nicht verschoben wird)📎 tokio/src/macros/select.rs:701-707; es pollen,Ready(out)dann zuerstdisabled |= maskund dann das Muster abgleichen📎 tokio/src/macros/select.rs:710-730。

Sechster Schritt: Musterabgleich. Fallsoutauf$bindpasst, wirdPoll::Ready(Out::_i(out)) 📎 tokio/src/macros/select.rs:727-733zurückgegeben; falls nicht,continueweiter andere Zweige pollen – genau das, was die Dokumentation in Schritt 5 als „bei Nichtübereinstimmung des Musters den aktuellen Zweig deaktivieren" beschreibt📎 tokio/src/macros/select.rs:44-47。

Siebter Schritt: Schleifenende. Fallsis_pendingwahr ist, wirdPendingzurückgegeben; andernfalls sind alle Zweige ungültig und es wirdOut::Disabled 📎 tokio/src/macros/select.rs:740-745zurückgegeben. Das äußerematch outputmapptOut::_iauf den entsprechenden Handler,Disabledmappt auf denelse-Ausdruck📎 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

Designüberlegungen und Fallstricke im Produktivbetrieb

Warum Bitmaske stattVec<bool>？? Die Bitmaske ist eine einzelne Ganzzahl auf dem Stack, ohne Heap-Allokation, unddisabled |= maskist eine einzelne Instruktion. Fürselect!auf dem Hot Path vermeidet dies Heap-Zugriffe bei jeder Iteration.

Warum bei Nichtübereinstimmung des Musters den Zweig deaktivieren?Dies ist der entscheidende Unterschied zwischenselect!und einem „einfachen race". Betrachten SieSome(v) = stream.next() => ...: Fallsstream.next()(Ende des Streams) zurückgibt, passt das Muster nicht, der Zweig wird dauerhaft deaktiviert, wodurch ein endloses Polling eines bereits beendeten Streams vermieden wird. Das Dokumentationsbeispiel nutzt genau diese Semantik, um zwei Streams zu sammeln, bis beide beendet sindNoneDie wahre Bedeutung von Cancel-Safety📎 tokio/src/macros/select.rs:198-223。

Sobald ein Zweig bereit ist, werden die Futures der übrigen Zweige gedroppt. Falls ein gedropptes Future bereits Daten konsumiert, aber noch nicht zurückgegeben hat, gehen die Daten verloren. Die Dokumentation listet ausdrücklich auf, dass：select!nicht cancel-safe istread_exact、read_to_end、write_all, während📎 tokio/src/macros/select.rs:119-124aufgrund der Warteschlangen-Fairness bei Abbruch die Warteschlangenposition verliertMutex::lock、Semaphore::acquire. Bestimmungsmethode: Suchen Sie den📎 tokio/src/macros/select.rs:126-133-Punkt; falls ein Neustart der Funktion an.awaitweiterhin korrekt ist, ist sie cancel-safe.awaitRace-Falle bei Vorbedingungen📎 tokio/src/macros/select.rs:135-139。

if: Die Dokumentation gibt ein klassisches Fehlerbeispiel – mitwird derif !sleep.is_elapsed()-Zweig abgesichert, abersleepkönnte zwischen deris_elapsed()-Prüfung undwhiletrue werden, wodurch das Timeout übersehen wirdselect!. Die korrekte Schreibweise besteht darin,📎 tokio/src/macros/select.rs:336-376zu entfernen und denif-Zweig stets am Polling teilnehmen zu lassen, sodass nach dem Timeoutsleepdie Kosten vonbreak 📎 tokio/src/macros/select.rs:378-405。

biased;: Zufälliger RNG hat CPU-Kosten, und in manchen Szenarien ist eine deterministische Polling-Reihenfolge erforderlich. Aber📎 tokio/src/macros/select.rs:67-74legt die Fairness-Verantwortung in die Hände des Benutzers: Falls ein Zweig immer bereit ist, verhungern die nachfolgenden Zweigebiased;9.3 join! und die technischen Einschränkungen der Makro-Expansion📎 tokio/src/macros/select.rs:75-81。

Intuitives Modell

ist wie „gleichzeitig warten, bis alle Pakete angekommen sind". Anders als：join!, das bei früherer Ankunft die übrigen abbricht, aggregiert es dieselect!-Werte aller Futures zu einem Tupel. Ohne es müsste der Benutzer manuellReadyden Abschlussstatus jedes Futures verwalten.poll_fnDatenstruktur und Speicherlayout

数据结构与内存布局

join!Die Expansion von basiert ebenfalls auf dem Speichern von Futures in einem Tupel, aber der Zustand ist keine Bitmaske, sondern ein Tupel von „abgeschlossenen Werten". Nachdem jedes Future abgeschlossen ist, wird sein Wert entnommen und im Ergebnis-Tupel gespeichert, wobei der entsprechende Slot als abgeschlossen markiert wird. Im Gegensatz zuselect!anders,join!werden unvollständige Futures nicht gedroppt – es muss warten, bis alle Futures abgeschlossen sind, bevor es zurückkehrt.

Schritt-für-Schritt-Ablauf

join!Die Poll-Logik von teilt das Gerüst „Tupel speichert Futures +select!-getrieben" mit , aber die Semantik ist entgegengesetzt:poll_fnist „gib zurück, sobald irgendeines bereit ist",select!ist „gib erst zurück, wenn alle bereit sind". Jede Poll-Runde durchläuft alle unvollständigen Futures; wenn irgendeinesjoin!zurückgibt, wird das GesamtergebnisPending, wenn allePending, wird aggregiert zurückgegeben.ReadyKopieren

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"]

Die Cancel-Safety-Semantik von unterscheidet sich von

join!:select!Wenn gedroppt wird, werden alle unvollständigen Futures gedroppt, was ebenfalls Datenverlust verursachen kann. Dajoin!jedoch keinen Zweig aktiv abbricht, wird es nicht wiejoin!„einen Zweig abbrechen, weil ein anderer Zweig bereit ist". Das eigentliche Risiko besteht darin, dassselect!als Ganzes durch äußeresjoin!oder Timeout abgebrochen wird.select!Der Unterschied zwischen und

join!ist beachtenswert:try_join!gibt sofort zurück, wenn irgendein Futuretry_join!zurückgibt, und bricht die übrigen Futures ab, wodurch es das Cancel-Safety-Risiko vonErrerbt.select!Designüberlegungen

Die Grenzen von Makros als Compile-Zeit-Codegeneratoren

Die Konfigurationsvalidierung wird in die Compile-Zeit verlagert; illegale Kombinationen (wie。#[tokio::main]) führen direkt zu einem Compile-Fehler statt zu einer Laufzeit-Panic. Das ist der Kernvorteil von Makros gegenüber Buildern: Fehler frühzeitig.multi_thread + start_pausedHybride Architektur aus deklarativen Makros + prozeduralen Makros

Der Hauptteil von ist。select!, aber zwei Schlüssellogiken werden an prozedurale Makros delegiert:macro_rules!generiertselect_priv_declare_output_enumdas Enum undOutden TypMaskim Clear-Modus📎 tokio-macros/src/lib.rs:658-660，select_priv_clean_pattern. Warum? Die Kommentare erklären: Deklarative Makros können nur schwer Code generieren, der „je nach Zweiganzahl dynamisch den Integer-Typ auswählt", und auch nur schwer Token-Level-Bereinigung an Musterpositionen durchführenref/mut 📎 tokio-macros/src/lib.rs:666-668Notwendigkeit von📎 tokio/src/macros/select.rs:577-579。

clean_patternmatcht。select!in Form vonoutgegen das Muster&out; wenn der Benutzer📎 tokio/src/macros/select.rs:727schreibt, wird darausref v, was zu einem Typfehler führt.&ref vRekursives Löschen vonclean_patternsowieby_ref、mutabilitydesReference-Mustersmutability 📎 tokio-macros/src/select.rs:68-73📎 tokio-macros/src/select.rs:100-103. Dies ist der Kompromiss, den das Makro zwischen „Benutzerintuition" und „Borrow-Checker" eingeht.

Die technische Realität des 64-Zweig-Limits。count!、count_field!、select_variant!Die drei Makros haben jeweils manuell Match-Regeln von 0 bis 64 geschrieben📎 tokio/src/macros/select.rs:821-1017📎 tokio/src/macros/select.rs:1021-1217📎 tokio/src/macros/select.rs:1221-1414. Die Kommentare sagen offen „I'm not happy about it either"📎 tokio/src/macros/select.rs:816-817. Das ist der Preis dafür, dass deklarative Makros keine Arithmetik beherrschen: Man kann nur die Token-Anzahl hart auf Integer abbilden.

Zusammenfassung dieses Kapitels

Überlegungen und Selbsttests zu diesem Kapitel

Das Makro nimmt dem Benutzer den Boilerplate-Code und die Compile-Time-Validierung ab, aber was es erzeugt, sind weiterhin gewöhnliche Futures undpoll-Aufrufe. Im nächsten Kapitel verlassen wir die Compile-Time-Welt der Makros und betreten die Laufzeit-I/O-Abstraktionsschicht, um zu sehen, wieAsyncRead/AsyncWriteden Byte-Stream in Frames zerlegt und wie dasFramed-Codec-Framework unter den Cancel-Safety-Einschränkungen vonselect!korrekt funktioniert.

#[tokio::main]Im Wesentlichen ist es „Konfigurationsparsing + Builder-Kettengenerierung +block_on-Umhüllung". Die Konfigurationsvalidierung erfolgt zur Compile-Zeit, und der Flavor bestimmt den Builder-Startpunkt und die build-Methode.select!Der Kern ist „Tupel speichert Futures + Bitmaske markiert Deaktivierungen + zufälliger Startpunkt gewährleistet Fairness". Bei Nichtübereinstimmung des Musters wird der Zweig deaktiviert, und die Cancel-Safety hängt davon ab, ob das gedroppte Future bei.awaitneu gestartet werden kann.join!undselect!teilen dasselbe Skelett, haben aber entgegengesetzte Semantik: Ersteres wartet auf die Fertigstellung aller, Letzteres kehrt zurück, sobald eines bereit ist. Zusammen zeigen die drei die zentrale Abwägung im Design der Tokio-Makros: Boilerplate-Code und Compile-Time-Validierung dem Makro überlassen, die Komplexität der Laufzeitsemantik (insbesondere Cancel-Safety) dem Benutzer zur expliziten Verständnis überlassen. Nachdem man verstanden hat, wie Makros Laufzeitcode generieren, stellt sich die nächste natürliche Frage: Welche Abstraktionen bietet Tokio, wenn dieser Code tatsächlich beginnt, Byte-Streams zu lesen und zu schreiben? Kapitel 10 wirdAsyncRead/AsyncWriteund das Codec-Framework analysieren und untersuchen, wieBufReader/BufWriterSystemaufrufe reduziert,copy_bidirectionaldie bidirektionale Weiterleitung antreibt undFramedden Byte-Stream in Frames aufteilt, um die Frage zu beantworten: „Wo liegt die Abstraktionsgrenze für asynchrones I/O?"

Das vorherige Kapitel hat den Expansionsprozess von tokio-macros zerlegt, und wir haben gesehen, wie #[tokio::main], select! und join! dem Benutzer Boilerplate-Code und Compile-Time-Validierung abnehmen. Aber was die Makros erzeugen, sind weiterhin gewöhnliche Futures und poll-Aufrufe – wenn diese Futures tatsächlich beginnen, Bytes zu lesen und zu schreiben, bietet Tokio als zugrunde liegende Abstraktion nur zwei Traits: AsyncRead und AsyncWrite. Ihr Problem ist, dass sie „zu niedrig" sind: Ein poll_read garantiert nur „einige Bytes gelesen", nicht „eine vollständige Nachricht gelesen". Und die allermeisten Protokolle (HTTP, Redis, gRPC, benutzerdefiniertes RPC) sind auf „Frames" statt auf „Byte-Streams" ausgerichtet. Die zentrale Frage, die dieses Kapitel beantworten will, ist: Wo sollte die Abstraktionsgrenze für asynchrones I/O gezogen werden? Tokios Antwort ist zweischichtig: tokio::io bietet Traits und Werkzeuge auf Byte-Stream-Ebene (BufReader/BufWriter/copy_bidirectional), und das Codec-Framework von tokio-util bietet darauf aufbauend Frame-Level-Stream/Sink-Adapter (Framed/LengthDelimitedCodec). Wenn man die Arbeitsteilung dieser beiden Schichten versteht, versteht man, „warum Protokollimplementierungen fast alle mit Framed beginnen".

I. AsyncRead/AsyncWrite: Warum std::io::Read nicht direkt wiederverwendet werden kann

Intuitives Modell

std::io::Read::readist „blockierende Abholung": Man steht vor dem Fenster und wartet, bis die Ware ankommt; der Thread wird angehalten.AsyncRead::poll_readist „Abholung mit Essensmarke": Man fragt einmal „Ist es fertig?", und wenn nicht (Poll::Pending), erledigt man zuerst etwas anderes und hinterlässt gleichzeitig einen Waker, damit das System einen benachrichtigt, wenn die Ware ankommt. Ohne diesen Trait müsste man bei jedem asynchronen I/O manuellepoll-Registrierung und Waker-Mapping schreiben – genau das, was der Reactor in Kapitel 5 tut, undAsyncReadist die einheitliche Fassade, die er nach oben hin exponiert.

Datenstruktur und Speicherlayout

AsyncReadDie Definition von ist extrem knapp, mit nur einer Methode:

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

Die drei Parameter haben jeweils ihre Besonderheiten.self: Pin<&mut Self>statt&mut self: WeilAsyncReadoft von dem durchasync fngenerierten Future gehalten wird, und ein Future, sobald es gepollt wird, nicht mehr verschoben werden kann (self-referential), istPinein vom Compiler erzwungener Vertrag.cx: &mut Context<'_>trägt den Waker und ist der Übertragungskanal des „Abholgeräts".buf: &mut ReadBuf<'_>ist Tokios Kapselung von&mut [u8]– es zeichnet gleichzeitig „bereits gefüllte Länge" und „nicht initialisierte Kapazität" auf, um zu vermeiden, dassstd::io::ReadDiese Mehrdeutigkeit von „gibt die Anzahl gelesener Bytes zurück, aber der Puffer ist möglicherweise nicht initialisiert“.

Die Dokumentation listet explizit drei Rückgabesemantiken auf:📎 tokio/src/io/async_read.rs:15-32：Ready(Ok(()))bedeutet, dass Daten geschrieben wurden,bufdie Lesemenge wird durch das Längeninkrement vonReadBuf::filledbestimmt; wenn das Inkrement 0 ist, handelt es sich entweder um EOF oder umbuf.remaining() == 0(Puffer mit Nullkapazität);Pendingbedeutet, dass derzeit nicht lesbar, aber eine Weckbenachrichtigung registriert wurde;Ready(Err(e))ist ein zugrundeliegender I/O-Fehler. Hier gibt es eine leicht zu übersehende Falle:„Lesemenge 0“ ist nicht gleichbedeutend mit EOF– wenn der Aufrufer einen Puffer mit Nullkapazität übergibt,poll_readwird sofortReady(Ok(()))zurückgegeben, aber nichts gelesen. Wenn die obere Ebene „0 Bytes“ als EOF behandelt, wird fälschlicherweise eine Verbindungsschließung angenommen.

Szenario-getriebener Walkthrough: Lesen eines Byte-Abschnitts aus&[u8]Betrachten wir die einfachste Implementierung – eine Kopie von

für&[u8]Schrittweise Analyse: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

ist die verbleibende Kapazität des Zielpuffers; der kleinere Wertself.len()wird genommen. Der Slice wird aufgeteilt in „die diesmal zu kopierendenbuf.remaining()“ und „die verbleibenden zu lesendenamt。split_at(amt)“. Dann wirdainb」。buf.put_slice(a)kopiert und dessen filled-Zeiger vorgerückt.aDer Slice selbst wird auf den verbleibenden Teil vorgerückt – das ist der Schlüssel zuReadBufals „Cursor“: Nach jedem poll zeigt*self = bauf den ungelesenen Teil. Schließlich wird&[u8]zurückgegeben, weil ein Speicher-Slice immer „bereit“ ist und niemalsself. Beachten Sie, dassReady(Ok(()))ignoriert wird: Eine Speicherdatenquelle benötigt keinen Waker. Dies steht im Gegensatz zu einem Netzwerk-Socket – letzterer gibt bei fehlenden DatenPending。

zurück und registriert Leseinteresse._cxDie Implementierung vonPendinghat eine zusätzliche Grenzprüfung

io::Cursor<T>: Zuerst wird📎 tokio/src/io/async_read.rs:113-134genommen; wennposition()(Position außerhalb des gültigen Bereichs), wird direktpos > slice.len()zurückgegeben, ohne zu panicenReady(Ok(())). Dies ist defensives Design:📎 tokio/src/io/async_read.rs:113-134Die Position vonCursorkann extern durchset_positionauf einen beliebigen Wert gesetzt werden; bei Überschreitung des Bereichs ist die Behandlung als „bereits vollständig gelesen“ konformer mit der I/O-Semantik als ein panic.

Designüberlegungen: deref-Makro und die Propagation von Pin

AsyncReadbietet Weiterleitungsimplementierungen fürBox<T>、&mut T、Pin<P>. Die ersten beiden erzeugenderef_async_read!über das📎 tokio/src/io/async_read.rs:64-70-Makro; der Kern istPin::new(&mut **self).poll_read(cx, buf)– Dereferenzierung vonPin<&mut Box<T>>zuPin<&mut T>und dann Weiterleitung.Pin<P>Die Implementierung von📎 tokio/src/io/async_read.rs:87-93ist subtilercrate::util::pin_as_deref_mut(self): Sie ruftPin<&mut Pin<P>>auf und projiziertPin<&mut P::Target>zuPin. Diese Projektionsebene ist notwendig, da sonst verschachtelte

---

-Semantik korrekt bleibt. Der Preis ist, dass jede Weiterleitungsebene einen indirekten Aufruf einführt, den der Compiler normalerweise durch Inlining eliminieren kann.

II. copy_bidirectional: Die Zustandsmaschine der bidirektionalen Weiterleitung

copy_bidirectionalIntuitives Modellcopyist ein „bidirektionaler Kellner“: Er beobachtet gleichzeitig die Richtungen A→B und B→A; sobald auf einer Seite Daten gelesen werden, werden sie auf die Gegenseite geschrieben. Ohne ihn müsste man für einen TCP-Proxy zweiselect!-Futures manuell schreiben und mitselect!kombinieren – doch die Cancel-Safety-Einschränkung voncopy_bidirectional(Kapitel 9) würde dazu führen, dass „mitten im Lesen abgebrochene“ Daten verloren gehen.

verwendet eine explizite Zustandsmaschine, um die Zwischenzustände von „Lesen-Schreiben-Schließen“ zu speichern und so Cancel-Safety zu erreichen.

Datenstruktur und Speicherlayout

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

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

RunningKopierenCopyBufferhältShuttingDown(u64)(enthält einen 8KB-Puffer sowie Lese-/Schreibzähler) und repräsentiert „Daten werden gerade übertragen“.Done(u64)trägt die Anzahl der bereits kopierten Bytes und repräsentiert „Leseseite hat EOF erreicht, Schreibseite wird geschlossen“.repräsentiert „Schließen abgeschlossen, endgültige Byteanzahl wird aufgezeichnet“. Dieses Enum ist der Schlüssel zur Cancel-Safety:。

CopyBufferBei einem Drop zu jedem Zeitpunkt bleibt der Zustand im Enum erhalten, und der nächste poll kann vom Unterbrechungspunkt fortfahren.copy.rsstammt ausDEFAULT_BUF_SIZE, die Standardgröße wird durch📎 tokio/src/io/util/copy_bidirectional.rs:76-88bestimmt (8KB)CopyBuffer. Jede Richtung hält einen unabhängigen

, daher beträgt der Speicheraufwand 16KB.

copy_bidirectional_implSzenario-getriebener Walkthrough: Der vollständige Lebenszyklus einer bidirektionalen Weiterleitungpoll_fnkombiniert die Zustandsmaschinen beider Richtungen mit

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

Kopierentransfer_one_directionBeachten Sie die Aufrufreihenfolge vonPoll。ready!: Zuerst wird a→b vorangetrieben, dann b→a; beide gebenPendingzurück. Das Makro kehrt sofort zurück, wenn eine Richtung nicht abgeschlossen ist– aberder Zustand der anderen Richtung wurde bereits vorangetrieben📎 tokio/src/io/util/copy_bidirectional.rs:143-144. Genau das betont der Kommentarready!: Selbst wennDone(count)vorzeitig zurückkehrt, wird die andere Richtung beim nächsten poll immer noch

transfer_one_directionzurückgeben und keinen Fortschritt verlieren.loopIntern ist

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, die zustandsabhängig voranschreitet:poll_copyKopierenShuttingDown。ShuttingDownIm Zustandpoll_shutdownwirdDone。Doneaufgerufen; intern wird in einer Schleife „ein Block gelesen, ein Block geschrieben“, bis die Leseseite EOF erreicht oder die Schreibseite blockiert. Bei EOF wird die Gesamtzahl der kopierten Bytes zurückgegeben und der Zustand wechselt zu

. Dann wird

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))"]

und gibt direkt den Zähler zurück.

verwenden, würde bei Abschluss einer Richtung die andere gedroppt, ihr interner Puffer und Zähler gingen verloren, was Cancel-Safety verletzt. Die explizitepoll_copylegt den Zustand auf dem Stack offen;Errbei jedem erneuten Eintritt ist der Zustand noch vorhanden, wodurch „Wiederaufnahme vom Unterbrechungspunkt nach Abbruch“ gewährleistet wird.?Bei der Fehlerbehandlung wird das von📎 tokio/src/io/util/copy_bidirectional.rs:32zurückgegebene📎 tokio/src/io/util/copy_bidirectional.rs:67-70sofort übernach oben propagiert. Die Dokumentation stellt klarcopy_bidirectional: Unterbrochene Lese-/Schreibvorgänge werden erneut versucht, andere Fehler werden sofort zurückgegeben, und

copy_bidirectional_with_sizesteilweise gelesene Daten können verloren gehen📎 tokio/src/io/util/copy_bidirectional.rs:99-125(nicht auf die Gegenseite geschrieben). Dies ist ein Punkt, der in Produktionsumgebungen beachtet werden muss:poll_copyImmer zurückgebenReady(Ok(0))fälschlicherweise als EOF interpretiert, was zu einer Busy-Loop führt.

---

Drei, Framed: Den Byte-Stream in Frames aufteilen

Intuitives Modell

Framedist eine „Wurstmaschine": Upstream ist ein kontinuierlicher Wasserfluss (AsyncRead/AsyncWrite), Downstream sind geschnittene Wurststücke (Stream<Item = Frame> / Sink<Frame>）。Decoderist für „ein Stück aus dem Wasserfluss herausschneiden" zuständig,Encoderist für „ein Stück in einen Wasserfluss verpacken" zuständig. OhneFramedmüsste jede Protokollimplementierung manuell „Pufferverwaltung + Halbpaket-Verarbeitung + Klebepaket-Aufteilung" schreiben – genau die repetitive Arbeit, die das Codec-Framework beseitigen soll.

Datenstruktur und Speicherlayout

Framedselbst ist nur ein dünner Wrapper:

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

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

Der eigentliche Zustand befindet sich inFramedImplvonstate: RWFramesund enthältread: ReadFrameundwrite: WriteFramezwei Teile.ReadFrameDie Felder vonwith_capacitysind in📎 tokio-util/src/codec/framed.rs:107-126：eof: boolsichtbaris_readable: bool(ob das Leseende EOF ist),buffer: BytesMut(ob lesbares Interesse registriert ist),has_errored: bool(Lesepuffer),WriteFrame(ob bereits ein Fehler aufgetreten ist, um wiederholtes Lesen zu verhindern).📎 tokio-util/src/codec/framed.rs:119-122：buffer: BytesMutFelderbackpressure_boundary: usize(Schreibpuffer),

backpressure_boundary(Backpressure-Schwellenwert).poll_readyist der Schlüssel zum Backpressure-Mechanismus: Wenn der Schreibpuffer diesen Schwellenwert überschreitet,PendingwirdSinkzurückgegeben, bis die Daten herausgeschrieben sind, wodurch Backpressure auf den Upstreamcapacity 📎 tokio-util/src/codec/framed.rs:121ausgeübt wird. Standardmäßig gleichset_backpressure_boundary, anpassbar über📎 tokio-util/src/codec/framed.rs:271-273。

Szenario-getriebener Walkthrough: Einen Frame vom Socket lesen

FramedDieStreamImplementierung leitet lediglich weiter anFramedImpl::poll_next 📎 tokio-util/src/codec/framed.rs:309-311. Die eigentliche Logik befindet sich inFramedImpl(diese Datei wird in diesem Kapitel nicht bereitgestellt, aber die Aufrufkette lässt sich aus der Schnittstelle vonFramedableiten):

1. poll_nextprüft zuerst, obread.bufferbereits einen vollständigen Frame enthält (Aufruf voncodec.decode）；

2. WenndecodezurückgibtSome(frame), direkt ausgeben, ohne die zugrunde liegende I/O zu berühren;

3. WennNonezurückgegeben wird (Halbpaket), prüfenread.eof: Wenn bereits EOF und der Puffer nicht leer ist, bedeutet dies, dass Restdaten nicht dekodiert werden können; Fehler zurückgeben oderNone；

4. Andernfalls das zugrunde liegendeAsyncRead::poll_readaufrufen, um weitere Bytes inread.buffer；

5. Die gelesenen Bytes erneut mitdecodeversuchen, in einer Schleife, bis ein Frame produziert wird oderPending。

Diese Reihenfolge „zuerst decode, dann read" ist wichtig: Sie stellt sicher, dassein read mehrere Frames produzieren kann(Klebepaket), und dassein Frame sich über mehrere reads erstrecken kann(Halbpaket).is_readableDas Flag verhindert doppelte Registrierung von lesbarem Interesse – wenn beim letzten poll bereits registriert und nicht bereit, wird diesmal direktPendingzurückgegeben, ohne das zugrunde liegende erneut aufzurufen.

SinkImplementierte Aufrufkette📎 tokio-util/src/codec/framed.rs:315-338：start_sendruftcodec.encode(item, &mut write.buffer)auf, um den Frame in den Schreibpuffer zu kodieren;poll_flushschreibtwrite.bufferin das zugrunde liegendeAsyncWrite；poll_readyprüftwrite.buffer.len() >= backpressure_boundary, bei Überschreitung des Schwellenwerts zuerst flush und dann bereit zurückgeben.

Das folgende Sequenzdiagramm zeigtFrameddie komponentenübergreifende Zusammenarbeit während eines „Frame lesen – Frame schreiben"-Roundtrips:

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(()))

Cancel-Safety: Die Dokumentationswarnung von Framed

FramedDie Dokumentation listet speziell die Cancel-Safety-Semantik auf📎 tokio-util/src/codec/framed.rs:23-30：SinkExt::sendWenn inselect!von einem anderen Branch zuerst abgeschlossen,ist die Nachricht garantiert nicht gesendet, aber die Nachricht selbst geht verloren– weilsendintern zuerstpoll_readydannstart_send, wenn in derpoll_ready-Phase gedroppt,itembereits konsumiert aber nicht kodiert. WohingegenStreamExt::nextcancel-sicher ist: Es hält nur eine Referenz auf den zugrunde liegenden Stream; ein Drop verliert keine bereits dekodierten Frames.

Design-Überlegung:into_partsundmap_codec

Framedbieteninto_parts/from_partsfür „Codec wechseln aber Puffer behalten"📎 tokio-util/src/codec/framed.rs:290-298 📎 tokio-util/src/codec/framed.rs:155-166。map_codecist basierend auf diesem Methodenpaar implementiert📎 tokio-util/src/codec/framed.rs:221-234: Zuerstinto_partsherauslösenio/codec/read_buf/write_buf, dann mit dermap-Funktion den Codec konvertieren, schließlichfrom_partswieder zusammensetzen. Dieses Design erlaubt es, bei einem Protokoll-Upgrade (z. B. Wechsel von Klartext zu TLS) bereits gepufferte Daten zu behalten und ein erneutes Lesen zu vermeiden.

FramedPartsDas_priv: ()Feld📎 tokio-util/src/codec/framed.rs:373-375ist die „nicht-exhaustive Struct"-Technik: Private Felder verhindern direkte externe Konstruktion und erzwingen den Weg übernew/from_parts, wodurch in Zukunft Felder hinzugefügt werden können, ohne die Kompatibilität zu brechen.

---

Vier, LengthDelimitedCodec: Die Zustandsmaschine des längenpräfixierten Codecs

Intuitives Modell

LengthDelimitedCodecist ein spezielles Messer zum „Schneiden der Wurst nach Länge": Es nimmt an, dass jedem Frame ein Längenfeld mit fester Byte-Anzahl vorausgeht; zuerst wird die Länge gelesen, dann der Payload. Ohne es müsste man für ein längenpräfixiertes Protokoll manuell eine Zustandsmaschine „4 Bytes lesen → Länge parsen → N Bytes lesen → Schleife" schreiben – genau das, was internDecodeStatetut.

Datenstruktur und Speicherlayout

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

enum DecodeState {
    Head,
    Data(usize),
}

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

DecodeStateist eine explizite Zustandsmaschine:Headbedeutet „Längenfeld wird gerade gelesen",Data(n)bedeutet „Länge n wurde geparst, Payload wird gerade gelesen". Dieser Zustand bleibt überdecodeAufrufe hinweg erhalten, dahergeht im Halbpaket-Szenario kein Fortschritt verloren。

Builderhält die gesamte Konfiguration📎 tokio-util/src/codec/length_delimited.rs:416-435：max_frame_len(Standard 8MB),length_field_len(Standard 4 Bytes),length_field_offset(Standard 0),length_adjustment(Standard 0),num_skip(StandardNone, d. h.offset + len）、length_field_is_big_endian(Standard true).

Szenario-getriebener Walkthrough: Einen längenpräfixierten Frame dekodieren

decodeist der Einstiegspunkt der Zustandsmaschine:

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

HeadIm Zustanddecode_headaufrufen. WennNonezurückgegeben wird (unzureichende Daten), direktOk(None)zurückgeben und auf mehr Daten warten; wennSome(n)zurückgegeben wird, wechselt der Zustand zuData(n)。DataIm Zustanddecode_data(n, src)direkt n nehmen. Dannsplit_to(n)aufrufen: Wenn der Puffer bereits n Bytes enthält,Headden Frame herausschneiden, Zustand zurück zuNone, und Platz für den nächsten Frame-Header reservieren; andernfalls

decode_headzurückgeben und warten.

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

Kopierensrc.len() >= head_lenSchrittweises Parsen: ZuerstNone 📎 tokio-util/src/codec/length_delimited.rs:499-502prüfen, bei UnzulänglichkeitCursorzurückgeben. Mitsrcumadvance/get_uintwickeln, umadvance(length_field_offset)Operationen durchzuführen, ohne den ursprünglichen Puffer zu konsumieren.📎 tokio-util/src/codec/length_delimited.rs:517Den Header-Präfix überspringenfield_len. Entsprechend der Endianness die📎 tokio-util/src/codec/length_delimited.rs:520-524。

-Byte-LängenwertSchlüsselverteidigungn > max_frame_len: WennInvalidData, sofort📎 tokio-util/src/codec/length_delimited.rs:526-531Fehler zurückgeben

. Dies verhindert, dass ein bösartiger Peer einen Frame mit „Längenfeld = 4GB" sendet und damit Speichererschöpfung verursacht – dies ist die klassischste DoS-Angriffsfläche bei längenpräfixierten Protokollen.checked_sub/checked_addLängenanpassung mit📎 tokio-util/src/codec/length_delimited.rs:537-541statt nackter OperationInvalidInputFehler statt Panic.get_num_skip()Rückgabe vonnum_skipoder dem Standard-offset + len 📎 tokio-util/src/codec/length_delimited.rs:1070-1073, überspringt den Rest des Headers. Schließlichreserve(n.saturating_sub(src.len()))reserviert Payload-Speicher📎 tokio-util/src/codec/length_delimited.rs:559— verwendetsaturating_sub, weilsrcmöglicherweise bereits einen Teil der Payload enthält.

Das folgende Flussdiagramm zeigtdecodeden vollständigen Entscheidungspfad von:

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))"]

Designüberlegung: max_frame_len-Kürzung und Überlaufschutz

Builder::adjust_max_frame_lenBeim Erstellen des Codecs wirdmax_frame_lenauf den maximalen Wert gekürzt, den das Längenfeld darstellen kann📎 tokio-util/src/codec/length_delimited.rs:1075-1081。max_allowed_frame_lenBerechnung vonmax_length_field_value + length_adjustment 📎 tokio-util/src/codec/length_delimited.rs:1083-1089, wobeimax_length_field_valuemitchecked_shlbehandelt wirdlength_field_len == 8den Shift-Überlauf bei📎 tokio-util/src/codec/length_delimited.rs:1091-1096. Diese Kürzung verhindert widersprüchliche Konfigurationen wie „Längenfeld 2 Bytes, aber max_frame_len auf 1MB gesetzt" — 2 Bytes können maximal 65535 darstellen, nach der Kürzung wird max_frame_len zu 65535.

Symmetrischer Schutz im Kodierungspfad:encodePrüfung vonn > max_frame_lenRückgabe vonInvalidInput 📎 tokio-util/src/codec/length_delimited.rs:607-607, Längenanpassung ebenfalls mitchecked_add/checked_sub 📎 tokio-util/src/codec/length_delimited.rs:620-631. Beachten Sie: Die Anpassungsrichtung bei der Kodierung ist umgekehrt zur Dekodierung: Dekodierung ist „gelesene Länge ± adjustment = Payload-Länge", Kodierung ist „Payload-Länge ∓ adjustment = geschriebenes Längenfeld"📎 tokio-util/src/codec/length_delimited.rs:620-624。

---

Designüberlegung: Drei Ebenen der Abstraktionsgrenze

Rückblickend auf dieses Kapitel zeigt die I/O-Abstraktion von Tokio eine klare dreischichtige Struktur:

Erste Ebene: Byte-Stream-Trait (AsyncRead/AsyncWrite）. Verspricht nur „einige Bytes lesen/schreiben", keine Frame-Grenzen. Dies ist die minimale Schnittstelle, die jede I/O-Quelle (Socket, Datei, Speicher-Slice) implementieren kann. Der Preis ist, dass die obere Ebene Halb-Pakete/Klebe-Pakete selbst behandeln muss.

Zweite Ebene: Byte-Stream-Werkzeuge (BufReader/BufWriter/copy_bidirectional）. Bieten auf dem Trait allgemeine Fähigkeiten wie „Systemaufrufe reduzieren" und „bidirektionale Weiterleitung".copy_bidirectionalDie explizite Zustandsmaschine von zeigt, wie „Cancel-Safety" auf der Werkzeug-Ebene implementiert wird — der Zustand wird auf dem Stack statt im Future gespeichert.

Dritte Ebene: Frame-Adaption (Framed/Decoder/Encoder）. Hebt den Byte-Stream aufStream<Frame>/Sink<Frame>an, sodass die Protokollimplementierung sich nur um „Frame-Kodierung/-Dekodierung" kümmern muss statt um „Pufferverwaltung".LengthDelimitedCodecist das Standardbeispiel dieser Ebene, dessenDecodeStateZustandsmaschine undmax_frame_lenSchutz Muster sind, die alle längenpräfixierten Protokolle wiederverwenden sollten.

---

in ihren jeweiligen Crates.

• AsyncRead::poll_readZusammenfassung dieses KapitelsPin<&mut Self> + Context + ReadBufverwendetstd::io::Read::readdrei Parameter stattReady(Ok(())), um „blockierendes Warten" in „Waker registrieren + Pending zurückgeben" zu verwandeln.
• copy_bidirectionalund bei Lesemenge 0 muss zwischen EOF und Null-Kapazitäts-Puffer unterschieden werden.TransferStateverwendetRunning/ShuttingDown/Donedie dreizuständige Enum (select!), um Zwischenzustände zu speichern, sodass bidirektionale Weiterleitung auch bei
• FramedAbbruch wiederhergestellt werden kann. Bei Fehlern können teilweise Daten verloren gehen.AsyncRead/AsyncWriteadaptiertStream/Sink，ReadFrame/WriteFramezuSinkExt::sendund verwaltet Lese-/Schreibpuffer und Backpressure getrennt.StreamExt::nextist nicht cancel-safe (Nachrichtenverlust),
• LengthDelimitedCodecist cancel-safe.DecodeState（Head/Data(n)verwendetmax_frame_len) Zustandsmaschine zur Behandlung von Halb-Paketen,checked_add/checked_subschützt vor Längenfeld-DoS,

schützt vor Anpassungsüberlauf.

Im vorherigen Kapitel haben wir die Byte-Ebene von Framed zerlegt: Decoder schneidet BytesMut in Frames, Sink schreibt Frames zurück, und damit wird die Abstraktionsgrenze der asynchronen I/O klar. Doch Frames sind nur Container für Daten. Eine echte Protokollimplementierung stößt unmittelbar danach auf drei Probleme, die weder tokio::io noch Framed lösen: asynchrone Iteration – Framed implementiert Stream, aber Stream hat nur poll_next, kein next().await, filter, take, merge; handgeschriebenes poll_fn ist nicht nur umständlich, sondern führt auch leicht zu Fehlern bei der Cancel-Sicherheit; dynamische Aufgabenmengen – ein Chat-Dienst muss gleichzeitig N Kanäle abonnieren, Kanäle treten jederzeit bei oder aus, während die Anzahl der Zweige von select! zur Compile-Zeit festgelegt ist und daher keine zur Laufzeit wachsende oder schrumpfende Stream-Menge ausdrücken kann; strukturierter Abbruch – select! kann einen einzelnen Zweig abbrechen, aber es kann nicht die Stilllegung eines gesamten Aufgabenbaums weiterreichen und auch nicht darauf warten, dass alle Aufgaben tatsächlich beendet sind. tokio-stream und tokio-util wurden genau für diese drei Dinge geschaffen. Ihr zentrales Designprinzip ist, nichts von Grund auf neu zu bauen: Jeder Kombinator von StreamExt ist nur eine Hülle um poll_next, StreamMap verwendet die Registrierungssemantik des Waker wieder, CancellationToken baut direkt auf tokio::sync::Notify auf, und TaskTracker kodiert den gesamten Zustand in einem AtomicUsize. Sie zu verstehen bedeutet im Wesentlichen zu verstehen, wie man auf den bestehenden Waker- und Scheduling-Mechanismen Zero-Cost-Abstraktionen baut. Dieses Kapitel schreitet in drei Ebenen voran: Iteration, Mengen, Abbruch. Zuerst wird gezeigt, wie StreamExt poll_next in einen komponierbaren Iterator verwandelt, dann, wie StreamMap und TaskTracker dynamische Mengen verwalten, und schließlich, wie CancellationToken mit einem Baum ein Abbruchsignal im gesamten Aufgabenbaum verbreitet.

StreamExt: poll_next in einen komponierbaren Iterator verwandeln

Intuitives Modell

Streamverhält sich zuFuture, wieIteratorsich zu Werten verhält:Futureerzeugt „einen Wert“,Streamerzeugt „eine Folge von Werten“. AberStreamdefiniert nurpoll_nextals einziges Primitiv, so wieIteratornurnextdefiniert. OhneStreamExtmüsste man für jedes Filtern, Mappen und Abschneiden manuellpoll_fn-Closures schreiben undPinmanuell verwalten – genau das war für frühe Nutzer desfutures-Crates am schmerzhaftesten.StreamExtDie Rolle vonStreambesteht darin,Iteratorein Kombinator-Ökosystem wie das von

zu geben.Ohne es steht das System nicht vor fehlender Funktionalität, sondern voreinem systematischen Zusammenbruch der Cancel-Sicherheitpoll_fn: Jedes handgeschriebeneselect!kann bei einem Abbruch durchpollein Element verlieren, das bereits

wurde.

StreamExtDatenstruktur und Speicherlayoutist einErweiterungs-Trait

📎 tokio-stream/src/stream_ext.rs:106-106

pub trait StreamExt: Stream {

KopierenAlle seine Methoden geben einkonkretes Kombinator-StructBox<dyn Stream>zurück, nichtmap. Das ist das entscheidende Design:Map<Self, F>，filtergibtFilter<Self, F>，takezurück,Take<Self>gibtpoll_nextzurück,

gibt

📎 tokio-stream/src/stream_ext.rs:1213-1213

impl<St: ?Sized> StreamExt for St where St: Stream {}

-Aufrufen inline expandieren.StreamBeachten Sie die Blanket-Impl des Traits:?SizedKopierendyn StreamJedes

erhält automatisch alle Kombinatoren, ohne manuelle Implementierung.

📎 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;

die Erweiterungsmethoden nutzen kann.next、try_next、all、any、fold、collectDie Moduldeklaration der Kombinatoren offenbart die vollständige Fähigkeitsoberfläche dieses Traits:Future（Next、TryNext、AllFutureKopierenmap、filter、takeHier gibt es eine bemerkenswerte Unterscheidung:Streamgibtnextzurück …), weil sie den gesamten Stream zu einem Wert konsumieren; währendNext<'_, Self>usw.

📎 tokio-stream/src/stream_ext.rs:144-149

fn next(&mut self) -> Next<'_, Self>
where
    Self: Unpin,
{
    Next::new(self)
}

Self: UnpinDer Rückgabetyp vonnextistPin, mit einem Lifetime-Parameter, weil es den Stream nur ausleiht:!UnpinKopierenBox::pinDiepin_mut!-Beschränkung ist absichtlich:

📎 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.

werden. Wenn der StreammergePolling von

mergeist das beste Beispiel dafür, wie ein Kombinator Waker wiederverwendet. Er verschränkt die Ausgabe zweier Streams undgarantiert Fairness– wenn beide Streams gleichzeitig bereit sind, wird abwechselnd ausgegeben. Die Dokumentation warnt ausdrücklich davor, verkettete Aufrufe zu verwendenmerge：

📎 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.

mergeDie Signatur von erfordert, dass beide Streams denselbenItemTyp haben:

📎 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)
}

Wenn der Aufrufer.next().awaitaufruft, läuft die Ausführung wie folgt ab:

1. Next::pollruft aufMerge::poll_next。

2. MergeIntern wird ein boolesches Flag „wer zuletzt an der Reihe war" verwaltet. Zuerst wirdpollder Stream, der zuletzt nichts ausgegeben hat; wennPending, dannpollder andere.

3. Wenn beidePending，MergezurückgebenPending, aberdie jeweiligen Waker beider Streams bereits registriert sind– jede Bereitschaft weckt die aktuelle Aufgabe auf.

4. Wenn ein StreamReady(None)zurückgibt (Ende),Mergewird vermerkt, dass dieser Stream beendet ist, und danach nur nochpollder andere Stream, bis auch dieser endet.

Der Schlüssel hierbei ist:Mergehat keine eigene Waker-Verwaltungslogik; es gibtcxunverändert an die internen beiden Streams weiterpoll_next。Die Waker-Registrierung liegt vollständig in der Verantwortung der zugrunde liegenden Streams，Mergeentscheidet nur, „wen diesmal zuerst gefragt wird". Genau das bedeutet „Wiederverwendung des zugrunde liegenden Waker-Mechanismus" wörtlich.

merge_size_hintsDie Hilfsfunktion zeigt, wie Kombinatoren Kapazitätshinweise zusammenführen:

📎 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)
}

Beachten Sie die Wahl vonsaturating_addundchecked_add: Für die untere Schranke wird saturierende Addition verwendet (lieber unterschätzen als bei Überlauf panicen), für die obere Schranke geprüfte Addition (wenn irgendeine unbekannt ist, ist das Ganze unbekannt). Dies ist die typische Behandlung dessize_hint-Vertrags.

Designüberlegung: Abbruchsicherheit undchunks_timeoutPanic-Schutz

StreamExtDie Dokumentation von annotiert jede Methode mitCancel safety. Nehmen wirnextals Beispiel:

📎 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.

nextist abbruchsicher, weil es den Stream nur ausleiht und keine Elemente konsumiert –Nextwenn das Future gedroppt wird, bleibt der Zustand des Streams unverändert, beim nächstennextwird erneutpoll。

Aber nicht alle Kombinatoren sind abbruchsicher.chunks_timeoutführt bereits bei der Konstruktion eine Parameterprüfung durch:

📎 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)
}

timeoutDer Unterschied zwischentimeout_repeatingundtimeoutist ebenfalls beachtenswert:gibt nach einem Timeout einen Fehler zurück, aber；timeout_repeatingpollt den inneren Stream weiterIntervalproduziert gemäß

📎 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.

---

Kopie

StreamMap: Dynamische Stream-Mengen und faires Polling

select!Intuitives ModellStreamMapDie Anzahl der Zweige von ist zur Kompilierzeit festgelegt. Aber die Anzahl der Kanäle, die ein Chat-Dienst abonnieren muss, oder die Anzahl der Verbindungen, die ein Crawler verfolgen muss, sind zur Laufzeit bekannt.select!ist ein „zur Laufzeit erweiterbares und reduzierbaresnext": Es legt beliebig viele Streams in eine Menge, und jedes(key, value)gibtmpsczurück und teilt mit, von welchem Stream der Wert stammt. Ohne es müsste man alle Streams in einen

-Kanal stopfen, mit zusätzlichem Weiterleitungsaufwand.

StreamMapDatenstruktur und SpeicherlayoutVec：

📎 tokio-stream/src/stream_map.rs:204-208

#[derive(Debug)]
pub struct StreamMap<K, V> {
    /// Streams stored in the map
    entries: Vec<(K, V)>,
}

Kopie

📎 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.

insertund

📎 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
}

removeDie Implementierung von spiegelt die Semantik „erst löschen, dann einfügen" wider:swap_removeKopie

📎 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
}

, um das gelöschte Element mit dem letzten Element zu tauschen und dann zu entfernen, wodurch O(n)-Verschiebungen vermieden werden:

StreamMapKopiepoll_next_entrySzenario-getriebener Walkthrough: Zufälliger Startpunkt und Cursor-Korrektur von poll_next_entryDer Kern von ist. Es beginnt mit

📎 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
    }
}

, um Fairness zu gewährleisten – wenn immer bei Index 0 begonnen würde, würde der erste Stream die nachfolgenden Streams aushungern:

Kopie thread_rng_nDieser Code hat drei Feinheiten, die wir einzeln aufschlüsseln:FastRandErstens, der zufällige Startpunkt.xorshift64+verwendet thread-lokales

📎 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-Algorithmus:% 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
}

verwendet Lemires multiplikative Modulo-Operation anstelle vonswap_removeKopieZweitens, die Cursor-Korrektur nachidx. Wenn der Stream bei IndexNonezurückgibtswap_removeund entfernt wird,idxverschiebt das letzte Element nach. Dieses verschobene Element könntebereits gepollt worden seinstart(wenn sein ursprünglicher Index voridx < start && start <= self.entries.len()lag). Der Code erkennt dies mitidx = idx.wrapping_add(1) % lenund überspringt es gegebenenfalls (idx == len). Wenn das entfernte Element das letzte war (

), wird der Cursor auf 0 zurückgesetzt.Poll::PendingDrittens, die Semantik von. Wenn eine vollständige Runde keinen bereiten Stream findet und die Menge nicht leer ist, wirdPendingzurückgegeben. Zu diesem Zeitpunkt sind die Waker aller Streams registriert, und jede Bereitschaft weckt auf.

poll_nextergänzt den Key aufpoll_next_entry:

📎 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)
    }
}

Beachten Sie dasready!-Makro: Wennpoll_next_entryzurückgibtPending, gibt das gesamtepoll_nextsofortPending。K: Clonezurück. Die Einschränkung stammt aus dem hier verwendetenkey.clone()。

Designüberlegung: Batch-Semantik und Abbruchsicherheit von next_many

next_manyist die Batch-Version vonStreamMapund sammelt in einem Durchgang so viele bereite Elemente wie möglich:

📎 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
}

Seine Abbruchsicherheitsgarantie ist entscheidend:

📎 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.

Warum istnext_manyabbruchsicher? Weil es Elementesofort in den vom Aufrufer bereitgestelltenbufferpusht, statt sie intern zwischenzuspeichern. Wenn das Future gedroppt wird, bleiben die bereits gepushten Elemente imbuffererhalten und gehen nicht verloren. Das bedeutet aber auch: Beim Drop kannbufferbereits teilweise Elemente enthalten – der Aufrufer muss das wissen.

poll_next_manyDie Schleifenstruktur von ist komplexer als die vonpoll_next_entry, weil es in einer Runde so viele Elemente wie möglich sammeln muss:

📎 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
    }
}

Das äußerewhile added < limitbildet zusammen mit dem innerenforeinen „mehrrundigen Scan": Solange in der vorherigen Runde ein Stream einen Wert produziert hat (should_loop = true), wird eine weitere Runde gescannt, bis genügendlimitangesammelt sind oder eine Runde keine Ausgabe liefert. Die drei Fälle des Rückgabewerts entsprechen präzise der Dokumentation:

📎 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_hintDie Implementierung zeigt, wie man Kapazitätshinweise mehrerer Streams aggregiert:

📎 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
}

Dasselbe Muster wie beimerge_size_hints: Untere Schranke sättigend addieren, obere Schranke prüfend addieren, bei Unbekanntem bleibt das Ganze unbekannt.

Im Folgenden wird der Entscheidungspfad vonpoll_next_entryanhand eines Flussdiagramms dargestellt:

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: Alle Zustände in einem einzigen AtomicUsize kodieren

Intuitives Modell

Graceful Shutdown erfordert zwei Dinge:Aufgaben zum Stoppen benachrichtigen（CancellationTokenist dafür zuständig), sowiewarten, bis Aufgaben tatsächlich beendet sind（TaskTrackerist dafür zuständig).TaskTrackerist wie eine Kombination aus „Aufgabenzähler + Shutdown-Schalter": Solange noch Aufgaben laufen oderclose，wait()nicht aufgerufen wurde, kehrt es nicht zurück. Ohne es könnte man nurJoinSetverwenden, aberJoinSetakkumuliert die Rückgabewerte jeder Aufgabe, und lang laufende Dienste würden OOM bekommen.

Datenstruktur und Speicherlayout

TaskTrackerist einArc-Wrapper:

📎 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,
}

Dies ist das raffinierteste Speicherlayout dieses Kapitels:EinAtomicUsizekodiert gleichzeitig „ob geschlossen" und „Aufgabenzähler". Das niedrigste Bit ist das Shutdown-Flag, die übrigen Bits sind die Aufgabenanzahl (da der Aufgabenzähler bei jedem+2, ist das niedrigste Bit immer 0). So benötigtis_closed_and_emptynur einen einzigen atomaren Ladevorgang:

📎 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
}

Szenario-getriebener Walkthrough: Race zwischen close und drop_task

Betrachten wir ein typisches Szenario: Der Hauptthread rufttracker.close()auf, während gleichzeitig die letzte Aufgabe beendet wird (TaskTrackerToken::dropruftdrop_taskauf). Beide können nebenläufig sein, und es muss garantiert werden, dass unabhängig davon, wer zuerst kommt,wait()aufgeweckt werden kann.

Zuerstset_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)setzt atomar das Shutdown-Bit und gibt den alten Wert zurück. Wenn der alte Wert 0 ist (vorher nicht geschlossen und keine Aufgaben), bedeutet dies „nach dem Schließen sofort leer + geschlossen erfüllt", undnotify_nowwird aufgerufen. Der Rückgabewert(state & 1) == 0bedeutet „dieser Aufruf hat den Zustand tatsächlich verändert".

Nun zudrop_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)dekrementiert den Zähler. Wenn der alte Wert 3 ist (binär11: Shutdown-Bit 1 + Zähler 1), bedeutet dies „dies ist die letzte Aufgabe und bereits geschlossen", undnotify_now。

wird aufgerufen. Race-Analyse der beiden Pfade:

• close wird zuerst ausgeführt：set_closedsieht den alten Wert2(Zähler 1, nicht geschlossen), benachrichtigt nicht. Anschließend siehtdrop_taskden alten Wert3, benachrichtigt. ✓
• drop_task wird zuerst ausgeführt：drop_tasksieht den alten Wert2(Zähler 1, nicht geschlossen), benachrichtigt nicht. Anschließend siehtset_closedden alten Wert0(Zähler 0, nicht geschlossen), benachrichtigt. ✓
• Nebenläufig：fetch_orundfetch_subsind atomar; unabhängig von der Verschachtelungsreihenfolge wird immer einer die Kombination „geschlossen + leer" sehen und benachrichtigen. ✓

notify_nowenthält einen leicht zu übersehendenAcquire-Ladevorgang:

📎 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();
}

Warum verwendetdrop_taskReleasestattAcqRel? Weildrop_tasksfetch_subnur „vorherige Schreibvorgänge für nachfolgende Leser sichtbar machen" muss (Release-Semantik) und nicht „Schreibvorgänge anderer Threads von zuvor sehen" muss (Acquire-Semantik). Abernotify_nowbenötigt Acquire, um happens-before herzustellen: Es stellt sicher, dass alle Aufräumarbeiten, die vor dem Beenden der Aufgabe durchgeführt wurden, für den Code nach der Rückkehr vonwait()sichtbar sind. Das Ergebnis diesesloadwird verworfen, einzig wegen seines Speicherordnungs-Nebeneffekts – dies ist eine typische Verwendung eines „fence-artigen Ladevorgangs" in Rust-Atomoperationen.

Design-Überlegungen: ABA-Widerstand von wait und drop-Semantik von TrackedFuture

waitgibt einTaskTrackerWaitFuturezurück, das internNotified：

📎 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)
        },
    }
}

KopiereninnerBeachten Sie das FeldNone，poll: Wenn es beim Erstellen bereits „geschlossen und leer" ist, wird es direkt aufReadygesetzt und kehrt sofort zurück

. Dies ist der schnelle Pfad.

📎 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.

KopierenNotify::notified()Diese Garantie stammt aus der Semantik vonNotified: Der Future registriert sich bei der Erstellung als „Wartender", und selbst wennnotify_waitersaufgerufen wird, bevor erpollwird, wird er beim erstenpolldie Benachrichtigung sehen.TaskTrackerWaitFuture::pollDie Implementierung von

📎 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
    }
}

KopierenpollJedes Malis_closed_and_empty()wird zuerstpoll Notifiedgeprüft, dannNotified. Diese Reihenfolge garantiert: Selbst wenn

TrackedFutureaus irgendeinem Grund nicht aufgeweckt wird, fängt die Zustandsprüfung es auf.TaskTrackerDie drop-Semantik vonJoinSetist der zentrale Unterschied zwischen

📎 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`].

:ReadyKopierenTrackedFutureDies bedeutet: Selbst wenn der Future bereitsTaskTrackerzurückgegeben hat, solange

📎 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.

TaskTrackerTokendie Aufgabe als noch laufend. Die Dokumentation erklärt, warum dieses Design wichtig ist:DropKopieren

📎 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();
    }
}

TrackedFuturevonpin_project!ist der Auslösepunkt für die Zählerdekrementierung:tokenKopierenfutureDurchtokenwerdenspawn_blockingund

📎 tokio-util/src/task/task_tracker.rs:452-464

zusammengepackt, und der drop von

Im vorherigen Kapitel haben wir gesehen, wie tokio-stream und tokio-util die zugrunde liegenden Waker- und Scheduling-Mechanismen wiederverwenden, um Kernfähigkeiten zu erweitern. Doch egal wie viele Kombinatoren erweitert werden, der Kernwiderspruch der asynchronen Laufzeit bleibt bestehen: Der Scheduler muss CPU-Zeit fair zwischen mehreren Tasks aufteilen, während die Tasks selbst nicht präemptiv sind – sobald die poll-Ausführung eines Future beginnt, kann der Scheduler es nicht von außen unterbrechen. Wenn ein Task in einem einzigen poll hunderttausend Nachrichten in einer Schleife verarbeitet oder in einer loop wiederholt ein Future awaitet, das immer bereit ist, monopolisiert es den Worker-Thread und lässt andere Tasks auf demselben Thread nie zum Zuge kommen. Das ist das klassische Problem „Task hungert Scheduler aus". Tokios Lösung ist nicht Präemption, sondern Kooperation: Jedem Task wird innerhalb eines Planungszyklus ein begrenztes Budget zugewiesen, Ressourcenoperationen verbrauchen Budget, und wenn das Budget erschöpft ist, muss der Task aktiv abgeben. Dieses Kapitel taucht tief in die Implementierung dieses coop-Mechanismus ein.

12.1 Der Träger des Budgets: Thread-lokaler Speicher und die Budget-Struktur

Das Budget muss zwei Einschränkungen erfüllen: Erstens muss es von beliebig tiefenpollAufrufstapeln aus zugänglich sein, ohne Parameter durch alle Ebenen zu reichen; zweitens muss es unterscheiden können, ob man sich „gerade innerhalb der Tokio-Laufzeit befindet" – außerhalb der Laufzeit aufgerufenblock_onsollte es nicht durch das Budget eingeschränkt werden. Tokio wähltThread-lokalen Speicher (TLS)als Träger des Budgets und verwaltet es einheitlich über dascontextModul.

Der Kerntyp des Budgets istcoop::Budget. Obwohl der Quellcode-Ausschnitt dieses Kapitels nicht direkt die vollständige Definition voncoop.rsliefert, lässt sich aus den Verwendungsstellen vonworker.rssein Schnittstellenvertrag ableiten:

📎 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);
        }
        // ...
    }
})

Hier erscheinen drei zentrale APIs:coop::budget(closure)errichtet einen Budget-Gültigkeitsbereich,coop::has_budget_remaining()fragt das verbleibende Budget ab, und später werden wircoop::stop()undcoop::set()。budgetsehen. Die Semantik ist: Beim Eintritt in die Closure wird das Budget des aktuellen Threads auf einen vollen Wert zurückgesetzt (Standard 128), während der Ausführung der Closure teilen sich alle Ressourcenoperationen dieses Kontingent, und beim Verlassen der Closure wird das äußere Budget wiederhergestellt.

Budgetexistiert im TLS üblicherweise in Form vonCell<Option<Budget>>.OptionDie äußere Semantik vonNoneist „ob sich der aktuelle Thread im Tokio-Laufzeitkontext befindet":block_onbedeutet, nicht innerhalb der Laufzeit zu sein (z. B. außerhalb der Laufzeit

), wobei alle Budget-Prüfungen direkt durchgelassen werden.

12.2 Die Verbrauchspunkte des Budgets: Wie Ressourcenoperationen es verringernDas Budget wird nicht aus dem Nichts verbraucht, nurRessourcenoperationensend/recvverringern es. Unter Ressourcenoperationen versteht man APIs, die potenziell in Endlosschleifen aufgerufen werden und mit der Außenwelt interagieren – channelyield_now, I/O-Lese- und Schreibvorgänge,mpsc::Sender::reserveusw. Nehmen wir

📎 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_innerKopierencrate::trace::async_trace_leaf()Bevorasync_trace_leaftatsächlich die Semaphore-Berechtigung erwirbt, durchläuft escoop::poll_proceed. Dieser scheinbar nur Tracing-Aufruf ist tatsächlich einer der Anknüpfungspunkte für die Budget-Verringerung.Proceedruft intern eine Funktion wiePendingauf: Wenn das Budget ausreicht, wird 1 abgezogen und

zurückgegeben; wenn das Budget erschöpft ist, wird eine „Abgabe"-Aktion registriert – der Waker des aktuellen Tasks wird an den Scheduler übergeben, undzurückgegeben, wodurch der Task in diesem poll vorzeitig beendet wird.PendingDas ist die Raffinesse von coop:PendingBudget-Erschöpfung wirft keinen Fehler, sondern tarnt die „Abgabe" als gewöhnliches

yield_now. Das übergeordnete Future siehtund kehrt natürlich zurück, der Scheduler reiht den Task wieder ein, und beim nächsten Scheduling ist das Budget bereits zurückgesetzt, sodass der Task ab der letzten Unterbrechungsstelle fortfährt. Der gesamte Prozess ist für den Geschäftscode völlig transparent.：

📎 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
}

löst aktiv eine Abgabe auscontext::defer(cx.waker())KopierenwakeBeachten Sie die Zeile. Sie ruft nicht direktauf, sondern übergibt den Waker an die

defer-WarteschlangeContextdes Schedulers. Warum? Der Quellcode-Kommentar sagt es klar: Bei sofortigem Aufwecken würde der Task sofort wieder in die Ausführungswarteschlange gestellt und könnte erneut gepollt werden, bevor der I/O-/Timer-Treiber läuft, wodurch die Abgabe ihren Sinn verliert. Die Semantik der defer-Warteschlange ist: „Diese Tasks erst aufwecken, nachdem der aktuelle Worker die bereiten Tasks abgearbeitet und die Treiber gepollt hat."

📎 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,
}

deferdes Workers definiert:

📎 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();
}

Wenn die defer-Warteschlange nicht leer ist, ruft der Workerpark_yieldauf – parkt mit 0 Timeout, was I/O und Timer antreibt und dann die Aufgaben im defer aufweckt. Dies garantiert, dass eine „abgegebene" Aufgabe erst nach einem Durchlauf des Treibers neu geplant wird.

12.3 Aufbau und Wiederherstellung des Budget-Geltungsbereichs: run_task und block_in_place

Der Budget-Geltungsbereich wird inrun_taskeingerichtet. Wenn jede Aufgabe gepollt wird,coop::budgetumschließt den gesamten Poll-Vorgang:

📎 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::budgetBeim Eintritt wird das Budget im TLS auf den vollen Betrag gesetzt, beim Austritt wiederhergestellt. Das bedeutet,jede Aufgabe erhält bei jedem Poll einen völlig neuen Budgetbetrag. Unabhängig davon, wie oft innerhalb einer AufgabeawaitRessourcenoperationen durchgeführt werden, wird die Aufgabe zwangsweise abgegeben, sobald innerhalb eines einzelnenpollmehr als 128 verbraucht werden.

Doch hier gibt es ein subtiles Problem: Aufgaben im LIFO-Slot werden innerhalbderselbenbudgetClosuregepollt. Betrachten wir die Schleife vonrun_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);
    }
    // ...
}

Der entscheidende Punkt: Aufgaben im LIFO-Slotteilen sich das Budget der äußeren Aufgabe. Der Kommentar am Anfang vonrun_taskbesagt: „Tasks from the LIFO slot inherit the 'parent's limits". Dies ist beabsichtigtes Design – wenn jede LIFO-Aufgabe das Budget zurücksetzen würde, würden im Ping-Pong-Szenario (Aufgabe A weckt B, B weckt wieder A) beide Aufgaben sich gegenseitig unendlich oft planen, das Budget würde stets zurückgesetzt, und das Verhungerungsproblem bliebe bestehen. Geteiltes Budget bedeutet, dass A und B zusammen höchstens 128 Ressourcenoperationen verbrauchen können, danach muss abgegeben werden.

Der LIFO-Slot selbst hat noch einen unabhängigen RatenbegrenzerMAX_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_TICKDer Wert von ist 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;

Dies istdie zweite Verteidigungslinie: Selbst wenn das Budget noch nicht erschöpft ist, wird der LIFO-Slot nach dreimaliger aufeinanderfolgender Bevorzugung deaktiviert, und nachfolgende Aufgaben laufen über die normale Warteschlange. Das Budget regelt die „Gesamtmenge der Ressourcenoperationen", die LIFO-Ratenbegrenzung regelt die „Anzahl der gegenseitigen Aufweckungen desselben Aufgabenpaares" – beide ergänzen sich.

Der Budget-Geltungsbereich hat inblock_in_placeeine wichtige Ausnahme.block_in_placeübergibt den Worker-Core an einen anderen Thread, und der aktuelle Thread geht in den blockierenden Zustand über. Blockierender Code unterliegt nicht der Budgetbeschränkung, daher muss das Budgetpausiertwerden:

📎 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()gibt das aktuelle Budget zurück und setzt es aufNone(d. h. „nicht innerhalb der Laufzeit"),ResetdieDropvon stellt es nach Ende der Blockierung wieder her:

📎 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)stellt das zuvorstop()gespeicherte Budget wieder her. Auf diese Weise verbraucht synchroner blockierender Code innerhalb vonblock_in_placekein Budget und löst auch nicht fälschlicherweise eine Abgabe wegen erschöpften Budgets aus; nach Ende der Blockierung setzt die Aufgabe ihre Ausführung mit dem ursprünglichen Restbudget fort.

Die folgende Abbildung zeigt den vollständigen Kontrollfluss von der Planung einer Aufgabe bis zur Abgabe wegen erschöpften Budgets:

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

In der Abbildung sind zwei Abgabepfade zu erkennen: Bei erschöpftem Budget wird die LIFO-Aufgabe zurück in die Warteschlange geschoben (push_back_or_overflow), und bei Überschreitung der aufeinanderfolgenden LIFO-Bevorzugung wird der LIFO-Slot deaktiviert. Beide führen zurück zur Hauptschleife, sodass der Worker Gelegenheit hat, andere Aufgaben zu bearbeiten oder den Treiber anzutreiben.

12.4 Designüberlegungen, Fehlerbehandlung und Produktions-Fallstricke

Warum TLS statt expliziter Parameterübergabe?Die Budget-Prüfpunkte sind tief in verschiedenen Modulen wie channel, I/O, time verstreut. Bei expliziter Parameterübergabe müsste jede API einen zusätzlichenBudgetParameter erhalten, was die gesamte öffentliche Schnittstelle verschmutzt. TLS macht das Budget für den Geschäftscode völlig transparent, zum Preis eines TLS-Zugriffsaufwands pro Prüfung. Tokio verwendet#[thread_local]oder plattformspezifisches schnelles TLS, um diesen Aufwand zu minimieren.

Zusammenspiel von Budget-Erschöpfung und Abbruchsicherheit.Wenn die Budget-Erschöpfung dazu führt, dassreserve_innerzurückgibtPending, könnte sich die Aufgabe gerade in einem bestimmten Zweig vonselect!befinden. Wenn zu diesem Zeitpunkt ein anderer Zweig bereit ist,select!wird der aktuelle Zweig abgebrochen –reserve_innerderWakeReceiverOnDrop-Guard prüft beim Drop, ob „das Semaphore geschlossen und leer ist", und weckt die Empfängerseite auf:

📎 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();
        }
    }
}

Die Existenz dieses Guards zeigt: Das durch das Budget ausgelöstePendingund das echte „keine Berechtigung"Pendingmüssen sich auf dem Abbruchpfad identisch verhalten, sonst könnte die Empfängerseite nie die Benachrichtigung „channel wurde geschlossen" erhalten.

Produktions-Fallstrick: Versteckte Latenz durch Budget-Erschöpfung.Ein häufiges Phänomen ist: Die Geschwindigkeit, mit der eine Aufgabe Nachrichten verarbeitet, wird plötzlich langsamer, aber die CPU-Auslastung ist nicht hoch. Bei der Fehlersuche verdächtigt man leicht Lock-Konkurrenz oder I/O, tatsächlich aber hat die Aufgabe möglicherweise innerhalb eines einzelnen Poll mehr als 128 Nachrichten verarbeitet, was eine Budget-Abgabe ausgelöst hat, und jede Abgabe durchläuft einen vollständigen Zyklus von „Zurückschieben in die Warteschlange → Neuplanung → Treiber-Poll". Wenn die Nachrichtenverarbeitung selbst schnell ist, kann dieser Planungsaufwand einen hohen Anteil ausmachen. Die Lösung besteht darin, die Massenverarbeitung in mehrerespawnAufgaben aufzuteilen oder explizit im Loopyield_now。

einzufügen. Die Grenze zwischen Budget undblock_in_place.Wie zuvor gesehen,block_in_placewirdcoop::stop()das Budget pausiert. Aber Vorsicht:coop::stop()wird nur aufgerufen, wennhad_enteredwahr ist, d. h. nur pausiert, wenn man sich tatsächlich auf einem Laufzeit-Worker-Thread befindet. Wennblock_in_placeaußerhalb der Laufzeit aufgerufen wird,f()direkt ausgeführt, und der Budget-Zustand bleibt unverändert. Diese Verzweigungsprüfung erfolgt inmaybe_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(());
        }
    }
    // ...
})

Die vier Kombinationen entsprechen: innerhalb eines Worker-Threads,block_ondem Thread-Pool-Eingang,block_in_placeverschachteltem

Zusammenfassung dieses Kapitels

Der coop-Mechanismus löst das Fairness-Problem nicht-präemptiver Scheduler mit einem dreischichtigen Design:

1. Budget-Träger：coop::BudgetIn TLS gespeichert,OptionDie äußere Schicht unterscheidet innerhalb/außerhalb der Laufzeit,coop::budgetErstellt einen Ganzbudget-Gültigkeitsbereich,coop::stop/coop::setUnterstützt Pausieren und Fortsetzen (block_in_placeSzenario).

2. Verbrauchspunkte: Ressourcenoperationen (Channel-Senden/-Empfangen, I/O,yield_now) übercoop::poll_proceedreduzieren das Budget; bei Erschöpfung wird das „Abgeben" alsPendinggetarnt, transparent für die Geschäftslogik.

3. Abgabepfad：yield_nowÜbergibt den Waker übercontext::deferan die defer-Warteschlange, um sicherzustellen, dass erst nach dem Treiber-Polling neu geplant wird; LIFO-Slot-Aufgaben teilen das Budget der Elternaufgabe und haben eineMAX_LIFO_POLLS_PER_TICK = 3unabhängige Ratenbegrenzung.

Die zentrale Erkenntnis dieses Mechanismus ist:Fairness erfordert keine Präemption, sondern nur, dass eine „Endlosschleife" nach einer endlichen Anzahl von Schritten natürlich unterbrochen wird. Das Budget ist das Maß für diese „endliche Anzahl von Schritten".

Denkanstöße und Selbsttests zu diesem Kapitel

ausführen. Wenn die innere Schicht erneut

Im vorherigen Kapitel haben wir das coop-Kooperationsbudget aufgeschlüsselt: Jede Aufgabe hat innerhalb eines Scheduling-Zyklus nur ein begrenztes Budget und muss nach dessen Erschöpfung abgeben, wodurch verhindert wird, dass eine einzelne Aufgabe andere aushungert. Doch der Budgetmechanismus löst nur das Problem der „fairen Planung". In realen Produktionsumgebungen gibt es eine weitere, verstecktere Fallgrube – Abbruchsicherheit, panic-Ausbreitung und Shutdown-Reihenfolge. Wenn select! ein Future abbricht, wenn ein Aufgaben-panic abgefangen wird, wenn die Runtime heruntergefahren wird – das Grenzverhalten des Codes widerspricht oft der Intuition. Dieses Kapitel beginnt mit der Abbruchsicherheit und schaut zuerst, was ein gedropptes Future tatsächlich verliert.

13.2 panic-Ausbreitung: Wie JoinError einen Absturz abfängt

Intuitives Modell

Ein Tokio-Aufgaben-panic lässt nicht den gesamten Prozess abstürzen (außer bei panic=abort), sondern wird abgefangen, verpackt inJoinError, überJoinHandle::awaitzurückgegeben. Das ist wie ein Unfall an einer Station am Fließband: Das Sicherheitsnetz fängt den Arbeiter auf, aber das Produkt ist Ausschuss – du bekommst einen „Unfallbericht" statt des Produkts.

Datenstruktur und Zustand

JoinHandle<T>DasFuture::Outputvonsuper::Result<T>istResult<T, JoinError> 📎 tokio/src/runtime/task/join.rs:325。JoinError, d.h.

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

KopierenRawTaskDer Mechanismus, mit dem panic abgefangen wird, liegt im poll-Pfad voncatch_unwind: Beim poll der Aufgabe wird sie mitJoinHandle::pollumschlossen; nach einem panic wird die Payload in den Ausgabe-Slot der Aufgabe gespeichert, der Zustand als complete markiert und dann der join-Waker geweckt.try_read_outputWas überErr(JoinError::panic(payload))。

gelesen wird, ist

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)

KopierenJoinErrorKernpunkt: Die panic-Payload bleibt vollständig erhalten,std::error::Errorimplementiertinto_panic(), kann überBox<dyn Any + Send>dasdowncast_ref::<&str>()zurückholen und dann mit

die panic-Nachricht extrahieren.

Designüberlegungen und StolperfallenJoinHandleFalle 1:UnwindSafeDas

impl<T> UnwindSafe for JoinHandle<T> {}
impl<T> RefUnwindSafe for JoinHandle<T> {}

📎 tokio/src/runtime/task/join.rs:176-181

ist manuell implementiert.T: UnwindSafeKopierenJoinHandleDies ist eine bedingungslose Implementierung, die keinT，Terfordert. Grund:catch_unwindselbst hält keinTIn der Heap-Zuweisung der Aufgabe wurde panic bereits durchUnwindSafe，JoinHandleisoliert. Also ist es auch dann sicher, wenn

nichtist.JoinHandleFalle 2: panic breitet sich nicht automatisch auf die Elternaufgabe aus.

Wenn Aufgabe A Aufgabe B spawnt und B panicked, wird A nicht automatisch benachrichtigt, es sei denn A awaited dasspawn_blockingvon B. Wenn A nicht awaited, wird Bs panic stillschweigend verschluckt. Dies ist eine der verstecktesten Bug-Quellen in Produktionsumgebungen.Falle 3:catch_unwindDer panic vonMutexwird ebenfalls abgefangen.std::sync::MutexDie Worker des Blocking-Threadpools umschließen Aufgaben ebenfalls mit

; nach einem panic stirbt der Thread nicht, sondern kehrt in den Pool zurück und nimmt weiter Arbeit an. Aber wenn du in einer Blocking-Aufgabe einhältst und es bei panic nicht freigibst, führt das zu Lock-Vergiftung – das ist das inhärente Verhalten voncatch_unwind, Tokio greift nicht ein.

Falle 4: panic beim Runtime-Drop.

Wenn eine Aufgabe während des Runtime-Drops panicked, bleibt

wirksam, aber der join-Waker ist möglicherweise bereits ungültig, und die panic-Payload wird verworfen. Dies ist eine Teilmenge des Shutdown-Reihenfolge-Problems, das im nächsten Abschnitt entfaltet wird.

13.3 Shutdown-Reihenfolge: Bereinigung von Blocking-Threads und I/O-Ressourcen

RuntimeIntuitives Modell

pub struct Runtime {
    scheduler: Scheduler,
    handle: Handle,
    blocking_pool: BlockingPool,
}

📎 tokio/src/runtime/runtime.rs:97-106

DropDatenstruktur und Shutdown-Pfad

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

bestimmen die Shutdown-Reihenfolge:DropKopierenscheduler，Die Implementierung vonblocking_pool。blocking_pool:DropKopierenRuntime::dropBeachte:scheduler → handle → blocking_poolbehandelt nur

, es wird nicht explizitshutdown_timeoutbehandelt. Das Schließen von

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

statt und wird nach der Rückkehr vonhandle.inner.shutdown()durch die Feld-Drop-Reihenfolge ausgelöst. Die Feld-Drop-Reihenfolge ist die Deklarationsreihenfolge:blocking_pool.shutdown(Some(duration)). Daher wird der Blocking-Pool zuletzt geschlossen.duration。

Aber

blocking/shutdown.rssteuert die Reihenfolge explizit:

pub(super) struct Sender {
    _tx: Arc<oneshot::Sender<()>>,
}

pub(super) struct Receiver {
    rx: oneshot::Receiver<()>,
}

📎 tokio/src/runtime/blocking/shutdown.rs:13-19

ZuerstSender, um Scheduler und I/O-Treiber zum Stoppen zu benachrichtigen, dannArc<oneshot::Sender>, um auf Blocking-Aufgaben zu warten, maximalSenderDer zugrundeliegende Mechanismus des Schließens des Blocking-PoolsReceiververwendet einen raffinierten oneshot-Channel:waitKopieren

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

-Klon (intern ein

1. timeout == Some(0)). Wenn alle Worker beendet sind und alleshutdown_backgroundgedroppt wurden, empfängt

2. try_enter_blocking_region()die Benachrichtigung.None。

Die

-Methode:block_on_timeoutKopieren

Schrittweise Analyse:

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"]

, es wird nicht gewartet.

Versucht, in den Blocking-Bereich einzutreten. Wenn man sich gerade im asynchronen Kontext befindet (z.B. Runtime in einer async-Aufgabe droppen), wirdDie Fehlermeldung ist eindeutig: „Cannot drop a runtime in a context where blocking is not allowed"📎 tokio/src/runtime/blocking/shutdown.rs:51-54. Die Lösung ist die Verwendung vonshutdown_background(), was äquivalent zushutdown_timeout(Duration::from_nanos(0)) 📎 tokio/src/runtime/runtime.rs:494-496ist und nicht auf blockierende Tasks wartet.

Falle 2:shutdown_backgroundlässt blockierende Tasks auslaufen.Die Dokumentation warnt ausdrücklich: „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. Blockierende Tasks laufen weiter, bis sie natürlich zurückkehren, aber die Runtime wurde bereits gedroppt, und die von ihnen gehaltenen Ressourcen sind möglicherweise ungültig geworden.

Falle 3: I/O-Ressourcen werden nach dem Drop der Runtime ungültig.Die Dokumentation erklärt: „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_errDie Funktion dient zum Erkennen solcher Fehler.📎 tokio/src/runtime/runtime.rs:585-593。

Falle 4:Dropwartet standardmäßig unbegrenzt.Die Dokumentation weist darauf hin: „TheDrop implementation waits forever for this」📎 tokio/src/runtime/runtime.rs:43-44. Wenn ein blockierender Task hängt (z. B. in einer Endlosschleife), wird das Droppen der Runtime dauerhaft blockiert. In der Produktion sollteshutdown_timeoutmit einer Obergrenze verwendet werden.

13.4 Signalverarbeitung und Konflikte mit mehreren Runtimes

Intuitives Modell

Unix-Signale sind prozessweit, aber TokiosSignalist an die Runtime gebunden. Das ist wie ein gemeinsamer Feueralarm im gesamten Gebäude, aber jeder Raum hat einen eigenen Empfänger – die erste Person, die einen Empfänger installiert, ändert die Verkabelung des Alarms, und alle nachfolgenden können diese Änderung nur noch teilen.

Datenstrukturen und globaler Zustand

signal_enableist der Einstiegspunkt für die Registrierung von Signal-Handlern:

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

Kernpunkte:

1. signal <= 0 || FORBIDDEN.contains(&signal)lehnt ungültige Signale ab.

2. handle.check_inner()prüft, ob der Signal-Treiber läuft – wenn die Runtime bereits heruntergefahren ist, schlägt dies hier fehl.

3. siginfo.init.get_or_init(...)verwendetOnceLock, um sicherzustellen, dass jeder Signal-OS-Handler nur einmal registriert wird.get_or_initDie Closure von ruftsignal_hook_registry::registerauf, was eine globale, prozessweite Registrierung ist.

4. Der registrierte Handler istaction(globals, signal), er erledigt zwei Dinge:globals.record_event(signal)Ereignis aufzeichnen und dann ein Byte in die Pipe schreiben, um den Treiber aufzuwecken📎 tokio/src/signal/unix.rs:252-259。

Ursache von Konflikten mit mehreren Runtimes

globals()gibt einen prozessweiten globalenGlobals，OsExtraDatazurück. DieUnixStreamin

pub(crate) struct OsExtraData {
    sender: UnixStream,
    pub(crate) receiver: UnixStream,
}

📎 tokio/src/signal/unix.rs:61-64

DefaultKopierenUnixStream 📎 tokio/src/signal/unix.rs:61-64Die Implementierung erstellt ein Paar

. Diese Pipe ist global eindeutig, alle Signal-Treiber aller Runtimes teilen sie.signal_enableDas Problem:handle.check_inner()Inprüftden Signal-Treiber der aktuellen Runtime. Aber der vonsignal_hook_registry::registerregistrierte Handler ist prozessweitund schreibt in die globalePipe. Wenn Runtime A zuerst SIGINT registriert und dann Runtime B ebenfalls SIGINT registriert,gibt direkt die vorhandenezurück und registriert nicht erneut. Aber der Signal-Treiber von Runtime B liest Daten aus der globalen Pipe – beide Runtimes konkurrieren um die Bytes derselben Pipe.get_or_initSzenariobasierter Walkthrough: Signalwettlauf bei mehreren RuntimesOk(())Kopieren

Designüberlegungen und Fallstricke

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 竞争读取，只有一个能读到字节

Die Dokumentation warnt ausdrücklich: „Once a signal handler is registered with the process the underlying libc signal handler is never unregistered"

. Selbst wenn die-Instanz gedroppt wird, werden nachfolgende Signale weiterhin von Tokio abgefangen, das Standardverhalten wird nicht wiederhergestellt.📎 tokio/src/signal/unix.rs:379-380Falle 2: Signale werden zusammengefasst.SignalDie Dokumentation erklärt: „before📎 tokio/src/signal/unix.rs:338-340。

. Wenn Sie 10 SIGINT empfangen, aber nur einmal pollen, sehen Sie nur ein Ereignis. Das ist eine Eigenschaft von Unix-Signalen selbst (Standard-Signale werden nicht in eine Warteschlange gestellt), Tokio führt keine zusätzliche Zusammenfassung durch.Falle 3: Signale können bei mehreren Runtimes verloren gehen.poll is called, all signal notifications are coalesced into one item returned from poll」📎 tokio/src/signal/unix.rs:312-315Da die globale Pipe von mehreren Runtimes konkurrierend gelesen wird, kann eine Runtime die Bytes weglaufen, während eine andere nie darauf wartet. In der Produktion sollte man Signale nur in einer Runtime verarbeiten oder

selbst verwalten.Falle 4:signal_hookPanic-Bedingungen der Funktion.

Die Dokumentation erklärt: „This function panics if there is no current reactor set, or if thesignal. Ein Aufruf vonaußerhalb einer Runtime führt zu einem Panic.rt feature flag is not enabled」📎 tokio/src/signal/unix.rs:398-405Falle 5:signal()Cancel-Sicherheit von

. Die Dokumentation garantiert: „This method is cancel safe. If you use it as a branch inrecv(). Der Grund ist, dass Signalereignisse im globalengespeichert werden,tokio::select! and another branch completes first, then it is guaranteed that no signal is lost」📎 tokio/src/signal/unix.rs:423-427nur liest und den zugrunde liegenden Zustand nicht konsumiert.EventInfoDesignüberlegungenrecv()Die drei Themen dieses Kapitels teilen ein zugrunde liegendes Muster:

Die Eigentümerschaft des Zustands bestimmt die Sicherheit von Abbruch/Herunterfahren/Signalen

ist cancel-sicher, weil die Ausgabe auf dem Heap liegt und das Handle nur eine Referenz ist.Die Reihenfolge des Herunterfahrens der Runtime ist kritisch, weil der Blocking-Pool und der Scheduler gemeinsam genutzt werden。

• JoinHandle 取消安全，因为输出在堆上，handle 只是引用。
• Runtime 关闭顺序敏感，因为阻塞池和调度器共享 Handle, eine falsche Reihenfolge führt zu Deadlock oder Panic.
• Signale kollidieren mit mehreren Runtimes, weil Handler und Pipe prozessweiter globaler Zustand sind, währendSignaleine Runtime-weite Sicht ist.

Nachdem man dieses Muster verstanden hat, lassen sich die Fallstricke in drei Prinzipien zusammenfassen:

1. Cancel-Safety = Zustand außerhalb des Future.Wenn das Future intern einen Puffer hat, geht beim Drop Daten verloren.JoinHandle、Signal::recv、tokio::sync::mpsc::Receiver::recvAlle erfüllen diese Bedingung.

2. Shutdown-Reihenfolge = Umgekehrte Abhängigkeitsrichtung.Wer von wem abhängt, zuerst wird der Abhängige geschlossen. Der Scheduler hängt vom I/O-Treiber ab, also wird zuerst der Scheduler geschlossen; der Blocking-Pool ist unabhängig und wird zuletzt geschlossen.

3. Globaler Zustand = Konflikte bei mehreren Instanzen.Jede prozessweite Ressource (Signal-Handler, Pipe, Dateideskriptor-Tabelle) kollidiert bei mehreren Runtimes; entweder auf eine einzelne Runtime beschränken oder externe Synchronisation verwenden.

Zusammenfassung dieses Kapitels

Fragen und Selbsttests zu diesem Kapitel

in Runtime B möglicherweise

noch Signale empfangen, aber es hängt davon ab, ob Runtime B einen laufenden Signal-Treiber hat. Wenn Runtime B keinen Signal-Treiber hat (z. B. das Signal-Feature nicht aktiviert ist oder der Treiber geschlossen wurde), liest niemand die Pipe-Daten, undcatch_unwindwartet ewig auf ein Wecken. Dies ist die Fragilität der Signalverarbeitung mit mehreren Runtimes.GlobalsKapitelübergang

Cancel-Safety, Panic-Propagierung, Shutdown-Reihenfolge, Signalkonflikte – die gemeinsame Wurzel dieser vier Probleme ist die Unschärfe der „Zustands-Eigentümerschaft" an asynchronen Grenzen. Tokio gibt eine praktisch nutzbare Antwort, indem es Zustand auf dem Heap ablegt, Lebenszyklen über Referenzzählung verwaltet,

Damit haben wir die tückischsten Randbereiche der Tokio-Produktionsumgebung durchlaufen: Cancel-Safety hängt davon ab, dass Ausgaben auf dem Heap gespeichert werden, die Atomarität von try_read_output; JoinHandle::drop bricht die Aufgabe nicht ab, erst abort bricht sie wirklich ab, aber nicht bei spawn_blocking; ein Panic wird von catch_unwind abgefangen und in einen JoinError verpackt, und ohne await geht er still verloren; das Herunterfahren der Runtime hat eine strikte Reihenfolge, und ein Drop im async-Kontext führt zu einem Panic; Signal-Handler sind prozessweiter globaler Zustand und werden nach der Registrierung nie wieder entfernt. Hinter diesen Regeln steht Tokios wiederholtes Abwägen zwischen Korrektheit und Leistung. Im nächsten Kapitel verlassen wir die konkreten Mechanismen, blicken aus der Architekturperspektive auf die Herkunft dieser Abwägungen zurück und werfen einen Blick darauf, wohin io_uring, Treiber-Refactoring und die Schnittstelle für benutzerdefinierte Executoren Tokio führen werden.

Im vorherigen Kapitel haben wir die vier Arten von Produktionsfallen sortiert: Cancel-Safety, Panic-Weiterleitung, Shutdown-Reihenfolge und Signal-Konflikte. Sie wirken verstreut, weisen aber alle auf dasselbe Architekturproblem hin: Wie wird die Zustands-Eigentümerschaft an asynchronen Grenzen klar aufgeteilt? Und die Art der Aufteilung der Eigentümerschaft wird genau durch die drei grundlegendsten Architekturentscheidungen der Runtime bestimmt – wie Aufgaben geplant werden, wie I/O-Ereignisse verteilt werden und wie Nebenläufigkeits-Korrektheit verifiziert wird. Dieses Kapitel taucht nicht mehr in die Implementierungsdetails einer konkreten Funktion ein, sondern stellt sich auf die Architekturebene, blickt auf Tokios Abwägungen bei diesen Entscheidungen zurück und folgt den in offizieller Dokumentation und Quellcode bereits angelegten Entwicklungslinien, um zu sehen, wohin io_uring, Treiber-Refactoring und die Schnittstelle für benutzerdefinierte Executoren Tokio führen werden. Nach diesem Kapitel sollten Sie eine praktische Frage beantworten können: Wann sollte man Tokio erweitern, und wann sollte man es umgehen.

I. Drei historische Abwägungen: Warum es so ist, wie es ist

Intuitives Modell

Stellen Sie sich Tokio als ein Restaurant vor, das seit zehn Jahren geöffnet ist. Die Schichtplanung der Küche (work-stealing), die eigenständige Aufstellung der Kellner (Trennung von I/O-Treiber und Scheduler) und das Hygienekontrollsystem der Küche (loom-Nebenläufigkeitsverifikation) wurden nicht am ersten Öffnungstag entworfen, sondern haben sich schrittweise im Prozess „mehr Gäste, komplexere Gerichte“ entwickelt. Nur wenn man diese Evolution versteht, kann man beurteilen, welche Designs vorausschauende Planung und welche historischer Ballast sind.

Abwägung eins: work-stealing statt globaler Warteschlange

Die Abwägung bei work-stealing ist: Jeder Worker hält eine lokale Warteschlange,spawnbevorzugt in die lokale Warteschlange eingefügt wird (lock-free, cache-freundlich), und erst wenn die lokale Warteschlange leer ist, wird vom Ende der Warteschlange eines anderen Workers gestohlen. Der Preis ist, dass die Lastverteilung Verzögerungen hat und das Stehlen selbst atomare Operationen und Speicherbarrieren erfordert. Tokio wählt Letzteres, weil moderne Server leicht mehrere Dutzend Kerne haben und die Kosten der Sperrkonkurrenz weit über den gelegentlichen Stehlkosten liegen.

Abwägung zwei: I/O-Treiber unabhängig vom Scheduler

Dies ist die interessanteste Stelle im Quellcode-Material dieses Kapitels. Betrachten Sie die Modulstruktur vontokio/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();

Beachten Sie, dassdriver、registration、scheduled_iodrei unabhängige Module sind und nach außen nur die TypenDriver、Handle、ReadyEvent、Registrationexponiert werden.ScheduledIoistpub(crate)– es wird vonPtrExposeDomainumschlossen, um unter loom-Tests rohe Zeiger der Nebenläufigkeitsprüfung auszusetzen.

Abwägung drei: loom zur Prüfung des Nebenläufigkeitsmodells

tokio/src/loom/mod.rshat nur 14 Zeilen, offenbart aber Tokios Strategie zur Verifikation der Nebenläufigkeits-Korrektheit:

📎 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::*;

Entscheidend ist die Bedingung#[cfg(all(test, loom))]: Nur wenn gleichzeitigtestundloomzwei cfgs aktiviert sind, wird das Modulmockeddurchstdersetzt. Das bedeutet, dass im Produktions-Build überhaupt kein loom-Code enthalten ist, also null Laufzeit-Overhead.

Designüberlegungen

Diese drei Abwägungen haben ein gemeinsames Merkmal:Sie alle wählen den „komplexeren, aber skalierbareren" Ansatz und beschränken die Komplexität auf das Innere. Die Komplexität von work-stealing ist im Scheduler verborgen, die Komplexität der I/O-Treiber ist inScheduledIoverborgen, und die Komplexität von loom ist in cfg-Bedingungen verborgen. Die nach außen exponierte API ist stetsspawn、TcpStream::readdiese einfachen Schnittstellen.

---

Zwei. Treiber-Refactoring: Von „ein Waker, eine Richtung" zu „beliebige Interessenmengen"

Intuitives Modell

Die frühen Tokio-I/O-Typen hatten eine harte Einschränkung:async fn read(&mut self)benötigt&mut self. Das ist wie ein Restaurant mit nur einem Ausgabefenster, an dem sich jeweils nur eine Person anstellen kann – weil der Waker im Inneren der I/O-Ressource gespeichert wurde und nicht im Future, das der Operation entspricht.tokio/docs/reactor-refactor.mddokumentiert vollständig die Ursache dieser Einschränkung und den Refactoring-Plan.

Die Schmerzpunkte der alten Architektur

Das Dokument benennt das Problem gleich zu Beginn:

📎 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).

Neue Architektur: Den Waker in das Future verschieben

Der Kernansatz des Refactorings ist „den Waker vom Ressourcenzustand in das Operations-Future zu verschieben", um so die Registrierung mehrerer Waker pro Operation zu ermöglichen:

📎 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.

Die neueScheduledIo-Struktur sieht wie folgt aus:

📎 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,
}

Hier gibt es einige raffinierte Designpunkte, die es wert sind, näher erläutert zu werden:

Erstens,readinessistAtomicUsize，waitersistMutex<Waiters>。Warum nicht eine einzige Sperre zum Schutz beider verwenden? Weil Leseoperationen aufreadinessextrem häufig sind (bei jedemreadiness()-Aufruf muss geprüft werden), während Schreiboperationen nur beim Empfang von mio-Ereignissen auftreten. Atomare Variablen für einen lock-freien Lesepfad zu verwenden, ist eine typische Optimierung der Trennung von Lesen und Schreiben.

Zweitens,Waiterist ein intrusiver Listenknoten. pointers: linked_list::Pointers<Waiter>lässtWaiterselbst Teil der Liste werden, ohne dass ein zusätzlicher Knoten allokiert werden muss._p: PhantomPinnedmarkiert explizit, dass es nichtUnpinwerden kann – denn sobald sich die Adresse eines Knotens in einer intrusiven Liste verschiebt, bricht die Liste.

Drittens,readerundwriterzweiOption<Waker>sind fürAsyncRead/AsyncWritegedacht.Das Dokument erklärt den Grund:

📎 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.

-Pfad verwendet feste Slots (unterstützt kein Abbrechen, ist aber trait-kompatibel). Diese „Koexistenz zweier Mechanismen" ist der typische Preis eines schrittweisen Refactorings.

Race Conditions und der tick-Mechanismus

📎 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.

KopierenreadinessDie Lösung ist die Einführung eines tick-Mechanismus, derAtomicUsizedieses

📎 tokio/docs/reactor-refactor.md:199-199

| shutdown | generation |  driver tick | readiness |
|----------+------------+--------------+-----------|
|   1 bit  |   7 bits   +    8 bits    +  16 bits  |

löscht den Bereitschaftszustand nur, wenn der tick übereinstimmt – wenn der tick nicht übereinstimmt, bedeutet das, dass in der Zwischenzeit ein neues Ereignis eingetroffen ist, und es darf nicht gelöscht werden. Auf diese Weise wird die Race Condition zwischen „Löschen" und „Eintreffen eines neuen Ereignisses" in einem einzigen atomaren Lese-Ändere-Schreibe-Vorgang aufgelöst.readiness()Das folgende Flussdiagramm veranschaulicht den Entscheidungspfad zwischenclear_readiness()und

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

Kopierentick_matchDer entscheidende Zweig in diesem Diagramm liegt beiclear_readiness: Wenn der tick nicht übereinstimmt, mussreadiness()das Löschen aufgeben, sonst geht das gerade eingetroffene Ereignis verloren, was dazu führt, dass die nächste Runde

dauerhaft blockiert.

Interessen-Abmeldung und Speicherlecksreadiness()Die intrusive Liste bringt ein neues Problem mit sich: Wenn das von

📎 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.

-Implementierung selbst aus der Liste entfernen, sonst bleibt der Knoten dauerhaft in

, was sowohl Speicher leckt als auch beim nächsten Eintreffen eines Ereignisses fälschlicherweise aufgeweckt wird.Vec<Waker>Designüberlegungen und Stolperfallen in der ProduktionWarum nicht&Resource, sondern eine intrusive Liste verwenden?

📎 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.

〔Designschlussfolgerungen und Architekturabwägungen〕：TcpStream::by_ref()Das Problem beiTcpStreamRefist: Nachdem das Future gedroppt wurde, bleibt der entsprechende Waker im Vec und kann nicht lokalisiert und entfernt werden; man kann nur beim nächsten Eintreffen eines Ereignisses feststellen, dass „dieser Waker bereits ungültig ist". Die intrusive Liste macht die Knotenadresse zur Adresse des Feldes im Inneren des Futures, sodass beim Drop präzise entfernt werden kann.read_waiterStolperfallen in der Produktionwrite_waiterDas von

📎 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,
}

---

über Branches hinweg teilst, sei vorsichtig mit den Lebensdauern –

Intuitionsmodell

Manchmal möchtest du nicht den Scheduler von Tokio verwenden, sondern nur dessen I/O und Timer nutzen. Das ist, als ob du nicht im Restaurant essen möchtest, sondern nur den Lieferfenster-Service nutzen willst.examples/custom-executor.rszeigt dieses „Hybridmodell": Verwendefutures::executor::ThreadPoolfür die Planung, Tokio für I/O.

Kernmechanismus: TokioContext

Der Schlüssel des gesamten Beispiels liegt inTokioContextdiesem Wrapper-Typ:

📎 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));
    }
}

Betrachte die Struktur des gesamten Beispiels:

📎 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 }
});

Datenfluss: Die executorübergreifende Reise eines 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 重新调度该任务

Der Schlüssel dieses Sequenzdiagramms liegt darin:Das Polling der Aufgabe findet im futures-Thread-Pool statt, aber das Warten auf I/O-Ereignisse findet in einem Tokio-Hintergrundthread statt. Beide werden durchHandleund den Waker verbunden.

Designüberlegung: Wann sollte man Tokio umgehen

1. Wenn du dich in ein bestehendes Executor-Ökosystem integrieren musst(zum Beispiel verlangen manche Frameworks zwingendfutures::executor), istTokioContextdie am wenigsten invasive Lösung.

2. Wenn du die Planungsstrategie vollständig kontrollieren musst(zum Beispiel erfordern Echtzeitsysteme deterministische Planung), erfüllt Tokios work-stealing die Anforderungen nicht, aber sein I/O-Treiber ist weiterhin nutzbar.

3. Wenn du nur die Komplexität von Tokios API ablehnst, dann solltest du es nicht umgehen——TokioContextdie eingeführte executorübergreifende Grenze bringt neue Debugging-Schwierigkeiten mit sich, was den Aufwand nicht lohnt.

Fallstricke im Produktivbetrieb：TokioContextImblock_on-Modus wird Tokios Laufzeit-Runtime::shutdownnie aufgerufen, was bedeutet, dass die Bereinigungslogik vonRuntimenicht automatisch ausgelöst wird. Du musst

vor dem Programmende explizit droppen, sonst wird der Hintergrundthread des I/O-Treibers möglicherweise nicht ordnungsgemäß beendet.

📎 tokio/src/runtime/io/mod.rs:1-4

#![cfg_attr(
    not(all(feature = "rt", feature = "net", feature = "io-uring", tokio_unstable)),
    allow(dead_code)
)]

verrät, wie io_uring eingebunden wird:feature = "io-uring"Kopierentokio_unstableBeachte, dassundgleichzeitig auftreten. Das bedeutet, dass die io_uring-Unterstützung derzeitallow(dead_code)experimentellallowist und nur kompiliert werden kann, wenn gleichzeitig das unstable-Feature aktiviert wird.

---

-Modell——

die Semantik von

ist unter io_uring nicht mehr anwendbar, und es braucht eine völlig neue „Submit-Complete"-Abstraktion. Deshalb bleibt die io_uring-Unterstützung so lange unstable: Es ist nicht einfach, ein Backend hinzuzufügen, sondern die gesamte Abstraktionsschicht des I/O-Treibers umzustrukturieren.：

• Zusammenfassung dieses Kapitels
• Dieses Kapitel hat aus architektonischer Sicht die drei Kernabwägungen von Tokio rekapituliert und drei Entwicklungspfade aufgezeigt:block_onHistorische Abwägungen
• work-stealing tauscht Planungskomplexität gegen Mehrkern-Skalierbarkeit ein, mit der Grenze, dass die Aufgabengranularität nicht zu fein sein darf;

Der I/O-Treiber ist unabhängig vom Scheduler, sodass（reactor-refactor.md）：

• und die Multithread-Laufzeit dieselbe I/O-Implementierung wiederverwenden können;ScheduledIoloom verschwindet durch cfg-Bedingungen vollständig aus Produktions-Builds und erschöpft Thread-Interleavings nur während Tests.
• Treiber-RefactoringAtomicUsizeDer Waker wird aus dem Inneren vonclear_readinessin das Operation-Future verschoben, um mit einer intrusiven verketteten Liste mehrere Wartende zu unterstützen;
• AsyncRead/AsyncWriteDas Bitfeld-Layout vonreader/writer(shutdown/generation/tick/readiness) löst die Race-Condition von

auf;：

• Da die Poll-Semantik keine intrusive verkettete Liste verwenden kann, bleibttokio_unstablemit festen Slots als Kompromiss erhalten.
• TokioContextZukünftige Entwicklung
• io_uring benötigt eine neue „Submit-Complete"-Abstraktion und ist derzeit durch

geschützt;

Damit haben wir den Aufstieg von konkreten Fallstricken zu Architekturabwägungen abgeschlossen. Wenn wir auf das gesamte Buch zurückblicken, von der lazy Evaluation von Futures über die Fairness des Schedulers, von Cancel-Safety über die Shutdown-Reihenfolge bis hin zu io_uring und austauschbaren Treibern in diesem Kapitel – alle Diskussionen drehen sich um einen Kern: die klare Aufteilung der Zustandsbesitzverhältnisse an asynchronen Grenzen. Die Architektur von Tokio ist nicht unveränderlich; das Zero-Copy-I/O von io_uring, die Entkopplung der Treiberschicht und die Öffnung der Schnittstelle für benutzerdefinierte Executoren treiben sie in eine flexiblere und effizientere Richtung voran. Wenn du dieses Buch zuklappst, hoffe ich, dass nicht eine Sammlung von API-Nutzungen zurückbleibt, sondern ein Urteilsvermögen: zu wissen, wann man der Laufzeitumgebung vertrauen sollte, wann man in die unterste Ebene eingreifen muss und wie man in der Produktionsumgebung die Kombinationen vermeidet, die zubeißen. Das Ökosystem von asynchronem Rust wächst weiterhin schnell – den Quellcode und die offiziellen Dokumentationen im Auge zu behalten ist wichtiger, als sich an irgendeine Schlussfolgerung zu erinnern.