VonBuilderbisRuntime: Eine vollständige Reise der Assemblierung

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.

Von der Queue zur Ausführung: Das Skelett der Worker-Hauptschleife

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