DesdeBuilderhastaRuntime: un viaje completo de ensamblaje

En el capítulo anterior dejamos claros los límites de responsabilidad entre Future, Waker y Executor. Pero un runtime realmente utilizable es mucho más que «un Executor»: también necesita un bucle de eventos de I/O, temporizadores, un pool de hilos bloqueantes, y todos estos componentes deben compartir el mismo conjunto de handles y el mismo ciclo de vida. Este capítulo rastrea la cadena completa de ensamblaje deBuilder::buildy responde a una pregunta central:¿Qué componentes hay dentro de unRuntimey cómo se ensamblan y comparten handles?。

El punto de entrada del ensamblaje de Tokio esBuilder. En sí mismo es un contenedor de configuración puro; todos sus campos son «declaraciones de intención» y no poseen ningún recurso de runtime. La creación real de recursos ocurre cuando se llama abuild().

Modelo intuitivo: Builder es el «plano de reforma», Runtime es «la casa tras la entrega»

Builderes como un plano de reforma: en él anotas «cuántas habitaciones quiero (worker_threads)», «si quiero agua corriente (enable_io)», «si quiero electricidad (enable_time)», «límite de ayudantes subcontratados (max_blocking_threads)». El plano en sí no produce ninguna entidad. Hasta que se llama abuild(), el equipo de construcción no construye según el plano, levantando realmente las «habitaciones» —scheduler, driver, pool de hilos— y entregando una instancia deRuntime.

Si no existiera la capa deBuilder, el usuario tendría que hacer new manualmente de cada componente, cablear manualmente, gestionar manualmente el rollback ante fallos; cualquier error de orden provocaría handles colgantes o fugas de recursos.BuilderEl valor de radica en:Separar por completo «configuración» y «construcción», permitiendo que el proceso de construcción realice de forma centralizada validación, limpieza ante fallos y compartición de handles.。

Diseño de memoria:Builderpartición de campos de

BuilderLos campos de pueden dividirse por responsabilidad en cuatro grupos. El primer grupo esforma y conmutadores：kinddetermina la forma del scheduler,enable_io / enable_timedetermina si se crea el driver correspondiente.

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

El segundo grupo esparámetros del pool de hilos：worker_threadsesOption<usize>，Nonelo que significa «diferir hasta build para detectar automáticamente según el número de núcleos de CPU»;max_blocking_threadspor defecto 512.

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

worker_threads: Option<usize>,
max_blocking_threads: usize,

El tercer grupo eshooks de callback, todos sonOption<Arc<dyn Fn ...>>. Nótese que usanArcen lugar deBox, porque estos callbacks deben clonarse en elConfig。

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

CopiarEl cuarto grupo es：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,

CopiarKindAquí hay un diseño digno de mención:Copyes un pequeño enum de

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

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

MultiThreadCopiarrt-multi-threadLa variante está condicionada por el featurert. Esto significa que en una compilación donde solo se habilita el featureKind,build()tiene una sola variante, y elmatchdeserá optimizado por el compilador a una sola rama:。

usar el sistema de tipos en lugar de juicios en runtime para eliminar el tamaño de código del scheduler multihilo.

Builder::newLa filosofía de los valores por defecto: por qué I/O y time están desactivados por defectoenable_ioes el punto de entrada común para todas las construcciones. Estableceenable_timeyfalse。

📎 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()La razón por la que la macro es «plug-and-play» es porque internamente llama a

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

Copiarenable_io()Nótese quenet、processsolo se llama cuando se ha habilitado el featuresignalotime feature，enable_all(). Si el usuario solo habilita

no abrirá el driver de I/O, porque simplemente no hay código del driver de I/O en el artefacto compilado.build()Ruta principal de ensamblaje:

build()la bifurcación dekindes el punto de partida del ensamblaje; bifurca según

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

Copiar

La diferencia entre estas dos rutas va mucho más allá de «un hilo vs múltiples hilos». A continuación se detallan por separado.

build_current_thread_runtimeRuta uno: ensamblaje de current_threadbuild_current_thread_runtime_componentsen sí mismo es muy delgado; delega enRuntime。

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

Copiarbuild_current_thread_runtime_componentsLa lógica real de ensamblaje está en

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

CopiardriverEl primer paso crea(driver, driver_handle), devolviendo un par?. Nótese que aquíbuildpropaga el error directamente hacia arriba: si la inicialización del driver de I/O falla (por ejemplo, falla la creación de epoll), todoErrdevuelve

, y en ese momento el blocking pool aún no se ha creado, por lo que no hay nada que limpiar.spawnerEl segundo paso crea el blocking pool y extrae inmediatamente suspawnerclonado. Este

se inyectará en el scheduler, dándole la capacidad de enviar tareas bloqueantes al pool de hilos.

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

.ConfigEl cuarto paso es el núcleo: entregar juntos driver, driver_handle, blocking_spawner, las semillas yCurrentThread::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(),
);

Copiarenable_eager_driver_handoffAquí hay un detalle clave:false。

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

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

Finalmente,CurrentThread::newelhandledevuelto se envuelve enscheduler::Handle::CurrentThread, y luego se envuelve en elHandle。

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

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

Ok((scheduler, handle, blocking_pool))

Ruta dos: el ensamblaje de multi_thread

build_threaded_runtimeEl esqueleto de es similar al de current_thread, pero hay tres diferencias esenciales. La primera es la determinación del número de hilos worker:

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

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

NoneAquí se resuelve comonum_cpus(). Este es el punto de aterrizaje de la «detección automática diferida»: la detección ocurre en el momento de build y no en el momento deBuilder::new, porque la afinidad de CPU puede cambiar entre ambos momentos.

La segunda diferencia está en el cálculo de la capacidad del blocking pool:

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

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

Nótese quemax_blocking_threads + worker_threads. En contraste, la ruta de current_thread pasaself.max_blocking_threadsy0。

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

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

La tercera diferencia es queMultiThread::newdevuelve una tupla de tres elementos en lugar de dos:

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

Ellaunchadicional es un «handle de arranque».MultiThread::newSolo se encarga de construir la estructura del planificador,y no inicia inmediatamente los hilos worker. El arranque real ocurre después:

📎 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()entra en el contexto de runtime, y entonceslaunch.launch()realmente hace spawn de todos los hilos worker. Este diseño de dos fases de «construir primero, arrancar después» es muy crítico.

Diagrama de flujo del ensamblaje

La siguiente figura reúne el orden de ensamblaje, las ramas clave y las rutas de error de ambas rutas. Nótese que cuandodriver::Driver::newfalla, se devuelve directamenteErr, y en ese momento el blocking pool aún no se ha creado.

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

Compartición de handles:Handlecómo se convierte en un «pase» entre componentes

Una vez completado el ensamblaje,Runtimeposee el conjunto de tres piezasscheduler、handle、blocking_pool. Entre ellas,handlees el núcleo compartido. En su interior hay una enumeración:

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

Nótese que ambas variantes envuelvenArc. Esto significa que el clonado deHandlees un incremento barato de conteo de referencias, y puede distribuirse libremente a cualquier hilo.Handleproporciona una interfaz de acceso unificada, encapsulando las diferencias de forma dentro dematch. Por ejemplodriver()：

📎 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()usa la macromatch_flavor!para eliminar la repetición:

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

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

Esta macro, al expandirse, es exactamente eldriver()como el de arribamatch. Su valor radica en que: al añadir un nuevo accesor que necesite despacharse según la forma, basta con una línea dematch_flavor!, sin tener que escribir a mano dos veces la ramamatch.

ElHandlepúblico es un envoltorio delgado delscheduler::Handleinterno:

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

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

ElHandleque obtiene el usuario puede clonarse entre hilos, puedespawn, puedeblock_on。spawn. La implementación deAutoBoxmuestra la rama en tiempo de compilación de

📎 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_BOXCopiarsize_of::<F>()es una constante asociada, obtenida al comparar

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

), lo que provocaría que cada future de spawn generara dos copias del task harness, duplicando el tamaño del código. Con la rama constante, el recolector de monomorfización solo conserva la rama que realmente se recorre.

Reflexiones de diseño: orden de ensamblaje, recuperación de errores y trampas en producciónEl orden es el contratodriver -> blocking_pool -> scheduler. El orden de ensamblaje

no es arbitrario. El driver se crea primero, porque es el único paso que puede fallar por recursos insuficientes del SO y que, tras fallar, no requiere limpiar otros componentes. blocking_pool va después del driver y antes del scheduler, porque el scheduler necesita blocking_spawner. Si la creación de blocking_pool falla (en realidad no suele fallar), el driver se limpiará automáticamente al hacer drop.local_tidLa rama。build_localde current_threadbuild_current_thread_local_runtimetoma la ruta de

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

CopiartidEsteHandlese almacena encan_spawn_local_on_local_runtime, y posteriormente

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

fueran accedidos concurrentemente, causando UB.worker_threads(0)Trampa en producción uno:。worker_threadshará panic

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

Esta aserción falla en la fase de configuración, en lugar de esperar hasta el build. La ventaja es que el error se localiza antes; la desventaja es que si el número de hilos proviene de un valor dinámico del archivo de configuración, el usuario debe validarlo por su cuenta antes de la llamada.

Problema en producción n.º 2:max_blocking_threadsSi se configura demasiado pequeño, se cuelga. La documentación advierte explícitamente:

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

Problema en producción n.º 3:UnhandledPanic::ShutdownRuntimeSolo soporta current_thread。

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

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

    self.unhandled_panic = behavior;
    self
}

Resumen del capítulo

Este capítulo rastreóBuilder::buildla cadena completa de ensamblaje de . Conclusiones clave:

1. Builderes un contenedor de configuración puro,build()es quien crea los recursos. El orden de ensamblajedriver -> blocking_pool -> schedulerestá determinado por los requisitos de recuperación de errores.

2. La diferencia entre current_thread y multi_thread no es solo el número de hilos: el cálculo de la capacidad del blocking pool es diferente (max_blocking_threads vs max_blocking_threads + worker_threads), multi_thread tiene unlauncharranque en dos fases adicional,enable_eager_driver_handoffse fuerza su cierre bajo current_thread.

3. Handlees el núcleo compartido entre componentes, internamente usaArcpara envolver handles específicos de la forma, y se accede de manera unificada mediantematchomatch_flavor!macros.

4. AutoBoxusa constantes asociadas para decidir en tiempo de compilación si se boxea el future, evitando duplicar el tamaño del código.

5. local_tidesLocalRuntimeel punto de verificación en tiempo de ejecución de la seguridad.

En el próximo capítulo, entraremos en el ciclo de vida de las tareas:spawncómo convertir un Future en una entidad programable,JoinHandlecómo interactuar con la máquina de estados de la tarea, y las transiciones de estado de la tarea entrePENDING / RUNNING / COMPLETE.

Reflexión y autoevaluación de este capítulo

En el capítulo anterior completamos el ensamblaje del Runtime: el driver de I/O, el driver de tiempo, el blocking pool y el planificador se inyectan en la misma instancia deRuntime,Handleconvirtiéndose en el handle compartido para acceder a estos componentes entre hilos. Pero el runtime ya ensamblado en este punto sigue siendo solo un cascarón vacío: posee el motor que impulsa las tareas, pero no tiene ninguna tarea que impulsar. La pregunta que este capítulo responde es precisamente: cuando escribestokio::spawn(async { ... }), ese bloqueasync¿qué experimenta exactamente para pasar de ser código Rust ordinario a una entidad «que puede ser tomada por el planificador, despertada y unida con join»? Esta es la primera mitad de «la vida de una tarea»; nos centramos en el nacimiento: partiendo deHandle::spawn, atravesando la asignación por conteo de referencias denew_task, hasta aterrizar en el diseño de memoria deCell<T, S>, y finalmente ver cómo la tarea se entrega a la cola local de algún worker o a la cola de inyección global. La segunda mitad (Capítulo 4) entrará en el bucle de planificación y el ciclo cerrado poll/wake.

3.1 Future no es una tarea: qué crea exactamente un spawn

Modelo intuitivo

ImaginaFuturecomo una «receta de cocina», y una tarea como «un plato que se está cocinando en la cocina». La receta en sí es estática, copiable y sin ningún estado de ejecución; solo cuando la cocina (el planificador) decide «hacer este plato ahora», le asigna un fogón (worker), un número de pedido (TaskId) y una ventanilla de despacho (JoinHandle), y entonces se convierte en un «plato en preparación». Sin esta capa de envoltura, el planificador no tendría forma de saber «en qué paso va este plato», «quién lo espera», «a quién notificar cuando esté listo»; solo vería una receta y no podría gestionarla.

Estructuras de datos y diseño de memoria

Tokio usaTask<S>para representar «una referencia a una tarea poseída por el runtime», que es una envoltura transparente sobreRawTask:

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

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

#[repr(transparent)]significa queTask<S>yRawTaskson completamente idénticos en memoria, sin sobrecarga adicional.PhantomData<S>es solo una marca de tipo en tiempo de compilación que indica a qué tipo de planificadorS。

pertenece esta tarea. Lo que realmente soporta todo el estado de la tarea esCell<T, S>, cuyo diseño es la piedra angular de todo el módulo de tareas:

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

Los tres campos están ordenados como «caliente-tibio-frío».Headerson datos calientes (se accede a ellos en cada planificación y en cada transición de estado),Coreson datos tibios (se accede a ellos durante poll),Trailerson datos fríos (solo se accede a ellos al crear y destruir). El comentario dice explícitamente:Headerdebe ser el primer campo, porque la estructura de la tarea será referenciada simultáneamente por*mut Celly*mut HeaderAún más crítico es la alineación a la línea de caché.📎 tokio/src/runtime/task/core.rs:37-43。

tiene colgada una larga lista deCell, que selecciona el número de bytes de alineación según la arquitectura objetivo: x86_64/aarch64/powerpc64 usan 128 bytes, arm/mips/sparc/hexagon usan 32 bytes, m68k usa 16 bytes, s390x usa 256 bytes, y el resto usa 64 bytes por defecto#[cfg_attr(..., repr(align(...)))]. El comentario explica por qué x86_64 debe usar 128 en lugar de 64: desde Intel Sandy Bridge, el prefetcher espacial trae de una vez📎 tokio/src/runtime/task/core.rs:64-125paresde líneas de caché de 64 bytes, por lo que es necesario alinear a 128 bytes para evitar el falso compartido〔Inferencia de diseño y compensaciones arquitectónicas〕📎 tokio/src/runtime/task/core.rs:45-53。

HeaderCopiar

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

no supere los 64 bytes (8 × 8), de modo que en arquitecturas con líneas de caché de 64 bytes pueda caber completamente en una línea.Headerincluye los campos:Header(bits de estado atómicos),state: State(puntero de lista enlazada de la cola de inyección),queue_next: UnsafeCell<Option<NonNull<Header>>>(tabla de punteros a funciones),vtable: &'static Vtable(ID de la lista deowner_id: UnsafeCell<Option<NonZeroU64>>a la que pertenece),OwnedTasks(medición de latencia de planificación)scheduled_at: UnsafeCell<ScheduleLatencyInstant>contiene el handle del planificador📎 tokio/src/runtime/task/core.rs:169-198。

Core<T, S>, el ID de tareascheduler: S, y lo más central,task_id: Ides una enumeración de tres estados:stage: CoreStage<T> 📎 tokio/src/runtime/task/core.rs:148-165。StageCopiar

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

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

contiene el future; al completarse, se reemplaza in situ porStage::Running, y tras ser tomado porStage::Finished(output)pasa a serJoinHandleEl comentario apunta a un issue de Miri, indicando que este diseño impone requisitos estrictos de corrección al código unsafeStage::Consumed。#[repr(C)]almacena datos fríos:📎 tokio/src/runtime/task/core.rs:225-229。

Trailerpuntero de lista enlazada),owned: linked_list::Pointers<Header>（OwnedTasks(waker del consumidor que espera a que la tarea se complete),waker: UnsafeCell<Option<Waker>>Paso a paso: de spawn a encoladohooks: TaskHarnessScheduleHooks 📎 tokio/src/runtime/task/core.rs:205-213。

Nos ponemos en un escenario concreto: en un runtime multi_thread, el hilo worker A ejecuta

Primer paso: construir el trío de la tarea.tokio::spawn(async { 42 })。

es la única entrada para el nacimiento de una tarea: new_taskCopiar

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

para asignarRawTask::new::<T, S>, y luego deriva tres referencias desde el mismo punteroCell:raw(referencia owned, normalmente se coloca inmediatamente enTask(referencia de notificación, entregada al planificador),OwnedTasks）、Notified(handle de lectura de resultados)JoinHandle（结果读取句柄）📎 tokio/src/runtime/task/mod.rs:347-363. Ten en cuenta que los tres comparten el mismoraw, cada uno mantiene un conteo de referencias.

Segundo paso: asignarCelly escribir el estado inicial. Cell::newAsignar toda la estructura en el 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

vtablegenerado porraw::vtable::<T, S>(), es una tabla de punteros a funciones monomorfizada paraTySespecíficos📎 tokio/src/runtime/task/core.rs:260. El future se mueve directamente aStage::Running, sin boxing adicional.

Tercer paso: aserciones de debug para verificar el layout.bajodebug_assertions,Cell::newllamará a la funcióncheck, usandoHeader::get_trailer、Header::get_scheduler、Header::get_id_ptry otras operaciones de punteros basadas en desplazamientos de vtable, para aseverar una por una que «la dirección del campo obtenida a través del header» coincide con «la dirección real del campo»📎 tokio/src/runtime/task/core.rs:280-321. Esta es una autoverificación en tiempo de ejecución de la correctitud de los desplazamientos de la vtable.

Cuarto paso: enviar al planificador.Una vez que el planificador obtieneNotified<S>, llama aSchedule::schedule 📎 tokio/src/runtime/task/mod.rs:315. Bajo multi_thread, esto pasará porpush_back_or_overflow, empujando la tarea a la cola local del worker actual, y cuando la cola está llena se desborda a la cola de inyección.

La siguiente figura describe el flujo de control y las ramificaciones desdenew_taskhasta el encolamiento:

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

Esta figura revela varias ramas clave: las aserciones de debug solo tienen efecto en compilaciones de depuración; cuando la cola local está llena no se desborda directamente, sino que primero se determina si hay un ladrón concurrente (steal != real), y si lo hay, solo se empuja la tarea actual a la cola de inyección, porque el espacio liberado por el ladrón estará disponible pronto.

Reflexión de diseño: por qué tres referencias en lugar de una

new_taskdevuelve tres referencias, no una. Este es el núcleo del diseño del conteo de referencias:Taskrepresenta «el runtime posee esta tarea»,Notifiedrepresenta «esta tarea ha sido notificada, pendiente de planificación»,JoinHandlerepresenta «alguien está interesado en su resultado». Los tres tienen ciclos de vida independientes——JoinHandlepuede ser dropeado (la tarea continúa ejecutándose, el resultado se descarta),Notifieddesaparece tras el poll,Taskse libera cuando la tarea se completa y se elimina deOwnedTasks. Si solo hubiera una referencia, no se podría expresar el estado «la tarea sigue corriendo pero nadie hace join».

UnownedTaskes otra rama importante: mantienedosconteos de referencias, usados para tareas blocking (no se almacenan enOwnedTasks）📎 tokio/src/runtime/task/mod.rs:286-295。unownedla función combina ambas referencias enmem::forget(task)mediantemem::forget(notified)yUnownedTask 📎 tokio/src/runtime/task/mod.rs:388-397. La motivación de diseño de estas «dos referencias» es: las tareas blocking no tienen una listaOwnedTasksque mantenga la referencia owned, por lo que se necesita un conteo de referencias adicional para garantizar que la tarea no sea liberada durante su ejecución.

3.2 Bits de estado: cómo un usize codifica todo el ciclo de vida de una tarea

Modelo intuitivo

Imagina el estado de la tarea como un «informe de chequeo médico» con varias casillas independientes: si está siendo poll, si se completó, si fue notificada, si fue cancelada, si alguien hace join. Tokio no usa múltiples campos booleanos, sino que comprime estos bits enunAtomicUsize. Así cada transición de estado requiere solo un CAS, en lugar de múltiples bloqueos. Sin este diseño, las transiciones de estado de la tarea se convertirían en un anidamiento de múltiples locks, disparando el riesgo de deadlock y la sobrecarga.

Layout de bits

StateLos bits de📎 tokio/src/runtime/task/mod.rs:32-53：

• RUNNINGestán completamente definidos en la documentación del módulo: si la tarea está siendo poll o cancelada. 📎 tokio/src/runtime/task/mod.rs:37-38。
• COMPLETEEste bit actúa simultáneamente como el lock de la tareaRUNNING: el future se ha completado por completo y ha sido dropeado. Una vez establecido nunca se limpia, y nunca se establece junto con📎 tokio/src/runtime/task/mod.rs:40-41。
• NOTIFIED: si actualmente existe un objetoNotified📎 tokio/src/runtime/task/mod.rs:43。
• CANCELLED: la tarea debe cancelarse lo antes posible📎 tokio/src/runtime/task/mod.rs:45-46。
• JOIN_INTEREST: existeJoinHandle 📎 tokio/src/runtime/task/mod.rs:48。
• JOIN_WAKER: bit de control de acceso como join handle waker📎 tokio/src/runtime/task/mod.rs:50-51。

Los bits restantes se usan para el conteo de referencias📎 tokio/src/runtime/task/mod.rs:53。

RUNNINGEl hecho de que el bit actúe como lock merece desarrollarse. La sección Safety de la documentación del módulo señala: cualquier acceso mutable al future debe realizarse después de adquirir el lock modificando el bitRUNNING, garantizando así acceso exclusivo📎 tokio/src/runtime/task/mod.rs:130-133. Esto significa que al hacer poll de una tarea, el hilo primero hace CAS para establecerRUNNING, y tras el éxito obtiene acceso exclusivo al future; si falla, significa que otro hilo está haciendo poll, y este poll retorna directamente. Esto fusiona «la exclusión mutua del poll» y «la transición de estado» en una sola operación atómica, evitando un mutex separado.

Protocolo de control de acceso de JOIN_WAKER

JOIN_WAKEREl bitwakeres la parte más ingeniosa de toda la máquina de estados. El problema que resuelve es:Trailerel campo(en) será accedido concurrentemente por dos hilos——el runtime al completar la tareaJoinHandleloleepara despertar al joiner,📎 tokio/src/runtime/task/mod.rs:75-120：

1. JOIN_WAKERal hacer poll

loJoinHandleescribe

para registrar el waker. La documentación del módulo proporciona 7 reglasJoinHandleinicialmente es 0.

2. Cuando es 0,COMPLETEtiene acceso exclusivo (mutable) al campo waker.

5. JoinHandle3. Cuando es 1,JOIN_WAKERsolo tiene acceso compartido (solo lectura).JOIN_WAKER4. Cuando es 1 y

6. JoinHandlees 1, el runtime tiene acceso compartido (solo lectura) al campo waker.COMPLETEPara escribir el waker, se debe: (i) establecer exitosamenteJOIN_WAKERa 0 para obtener acceso exclusivo, (ii) escribir el waker, (iii) establecer exitosamenteCOMPLETEa 1.

solo puede modificarJOIN_INTERESTcuandoCOMPLETEes 0; el runtime solo puede modificar cuando

es 1.COMPLETE7. Si📎 tokio/src/runtime/task/mod.rs:110-120es 0 y

es 1, el runtime tiene acceso exclusivo al campo waker (para dropear el waker).

TaskLa regla 6 implica una condición de carrera: los pasos (i) o (iii) pueden fallar. Si (i) falla, se abandona la escritura del waker; si (iii) falla (otro hilo establecióUnownedTaskel drop decrementa dos veces:

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_decdevuelvetrueindica que esta es la última referencia, y solo entonces se libera realmenteCellla memoria.ref_dec_twiceesUnownedTaskla manifestación directa de mantener dos conteos.

Reflexión de diseño: por qué el bit de estado y el conteo de referencias comparten un mismo atómico

3.3 JoinHandle: cómo el resultado cruza la frontera de la tarea para ser devuelto

Modelo intuitivo

JoinHandlees como el «comprobante de recogida» que te da el restaurante. Cuando la tarea (la cocina) termina, coloca el plato (output) en la ventanilla de salida (Stage::Finished), y luego activa tu localizador de recogida (waker). Tú vienes a recogerlo con el comprobante; el comprobante en sí no contiene el plato, solo es un puntero a la ventanilla de salida. Si pierdes el comprobante (dropJoinHandle), el plato se desechará directamente (output se dropea), pero la cocina no se detendrá por ello.

Estructura de datos

JoinHandle<T>es igualmente un envoltorio transparente sobreRawTask:

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

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

PhantomData<T>marca el tipo de salida.JoinHandle<T>solo esT: SendenSend/Sync 📎 tokio/src/runtime/task/join.rs:169-170, lo que garantiza que una salida no Send no se mueva entre hilos.

Paso a paso: await sobre un JoinHandle

JoinHandleimplementaFuture, cuyopolles el núcleo de la devolución del resultado:

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

Nótese varios detalles:trace_leafse usa para instrumentación de tracing;coop::poll_proceedconsume el presupuesto de cooperación (detallado en el capítulo 12);try_read_outputborra los genéricos mediante vtable, coloca el valor de retorno en la pila y lo pasa con*mut ()a📎 tokio/src/runtime/task/join.rs:327-354. Esta técnica de «colocar el valor de retorno en la pila» se debe a que la función de vtable no puede genericizar el tipo de retornoT, y solo puede escribir de vuelta mediante un puntero crudo.

Transferencia de propiedad del resultado

La sección «Non-Send output» de la documentación del módulo describe con precisión las reglas de propiedad del resultado📎 tokio/src/runtime/task/mod.rs:151-170：

• Cuando la tarea se completa, output se coloca enStage, luego se ejecuta la transición de «establecer COMPLETE» y se lee el valor deJOIN_INTERESTen ese momento.
• SiJOIN_INTERESTes 0 (sinJoinHandle), output se dropea inmediatamente📎 tokio/src/runtime/task/mod.rs:157-158。
• SiJOIN_INTERESTes 1,JoinHandlese encarga de limpiar output📎 tokio/src/runtime/task/mod.rs:160-161。

Para output no Send, la documentación ofrece una argumentación en tres pasos: output se crea en el hilo que hace poll del future;JoinHandle<Output>tampoco es Send cuando Output no es Send, por lo que también está en el hilo de spawn; por lo tantoJoinHandleno mueve output entre hilos al tomarlo o dropearlo📎 tokio/src/runtime/task/mod.rs:164-170。

Drop de JoinHandle: dos rutas, rápida y lenta

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_fastintenta completar en un solo CAS «limpiar el bitJOIN_INTEREST+ decrementar el conteo de referencias». Si falla (por ejemplo, la tarea se está completando y el bit de estado está ocupado), entonces toma la ruta lenta dedrop_join_handle_slow. Este es el patrón típico de «ruta rápida optimista + ruta lenta pesimista».

Reflexión de diseño: por qué JoinHandle no contiene directamente output

3.4 Cola local: la estructura productor-consumidor del work-stealing

Modelo intuitivo

Cada worker tiene una «lista privada de tareas pendientes» (cola local), con capacidad 256. El propio worker toma tareas desdela cabeza(LIFO, aprovechando la localidad de caché), mientras que otros workersrobantareas desde la cola (FIFO, tomando las más antiguas, las más probablemente ya completadas). Sin la cola local, todas las tareas se agolparían en la cola global, y cada toma de tarea competiría por el lock global, lo que haría colapsar la escalabilidad multinúcleo.

Diseño de memoria: separación de head y 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

headesAtomicUnsignedLong(64 bits, si la plataforma soporta u64),tailesAtomicUnsignedShort(32 bits). El comentario explica por qué los índices son más anchos de lo estrictamente necesario: para mitigar ABA y distinguir entre búfer «lleno» y «vacío»📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:37-49。

headempaqueta internamentedos UnsignedShort：la posición baja es la «cabeza real» (real head), la posición alta es «la primera posición que el ladrón está procesando» (steal head). Cuando ambas son iguales, no hay ladrones activos📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:39-49. Este empaquetado de dos valores es la técnica central de la cola work-stealing: el ladrón primero actualiza el valor steal mediante CAS para «reclamar» un lote de tareas, y al terminar hace que el valor steal alcance el valor real, indicando el fin del robo.

LOCAL_QUEUE_CAPACITYEn modo no-loom es 256, en loom se reduce a 4 para probar más casos límite📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:62-69。MASK = LOCAL_QUEUE_CAPACITY - 1, usado para el índice del búfer circular📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:71。

Paso a paso: las ramas completas de push_back_or_overflow

Esta es la función más compleja de la cola local, la analizamos rama por rama:

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

Tres ramas:

1. Hay capacidad（tail - steal < CAPACITY）：break tail, tras salir del bucle se llama apush_back_finishpara escribir en el búfer.

2. Sin capacidad pero con ladrones concurrentes（steal != real): el ladrón liberará espacio, así que solo se empuja la tarea actual a la cola de inyección y se retorna inmediatamente📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:204-208。

3. Sin capacidad y sin ladrones: se llama apush_overflowpara desbordar la segunda mitad de tareas a la cola de inyección📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:209-219. Si el CAS falla (pierde ante un ladrón concurrente),push_overflowretornaErr(task), y el bucle reintenta.

push_back_finishescribe la tarea y actualiza 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

ReleaseEl orden garantiza que la tarea escrita sea visible para los ladrones.

push_overflow: por qué desbordar la segunda mitad

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

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

Al desbordar se toman 128 tareas. El comentario explica en detalle por qué se tomala segunda mitaden lugar de la primera mitad📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:295-306: al tomar tareas de la cola de inyección, siempre se colocan en la primera mitad. Así que si una tarea está en la segunda mitad, se puede determinar que no acaba de ser tomada de la cola de inyección. Esto garantiza que «una tarea tomada de la cola de inyección no sea devuelta inmediatamente a la cola de inyección» (al menos antes de ser poll una vez).

CAS para reclamar la segunda mitad:

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

Se actualizaheaddesde(head, head)hasta(tail, tail), es decir, se avanzan simultáneamente steal y real hasta tail, reclamando todas las tareas. Tras el éxito se retrocede tail hastatail + NUM_TASKS_TAKEN, indicando que la primera mitad permanece en la cola local📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:314-316。

pop y steal_into: las dos rutas para tomar tareas

popes cuando el worker toma tareas por sí mismo (desde la cabeza, 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

Rama clave: sisteal == real(sin ladrones), se avanzan ambos; de lo contrario solo se avanza real, dejando steal intacto📎 tokio/src/runtime/scheduler/multi_thread/queue.rs:377-384。assert_ne!(steal, next_real)Garantiza que no se avance real hasta la posición de steal, pues se destruiría el estado de reclamación del ladrón.

steal_intoes la ruta de robo, primero verifica si la cola objetivo tiene suficiente espacio:

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

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

Si la cola objetivo está más de medio llena no se roba, evitando que tras robar se desborde inmediatamente.

steal_into2es el núcleo del robo, calcula la cantidad a robar:

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

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

Se roba la mitad (redondeando hacia arriba). Luego se actualiza el valor steal de head mediante CAS para reclamar:

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

Nótese que aquí solo se actualiza el valor real (pack(src_head_steal, steal_to)en steal permanece sin cambios), avanzando real hastasteal_to. Esto indica que «estas tareas ya han sido reclamadas, otros ladrones no pueden tocarlas». Tras completar el robo, se hace que steal alcance real:

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

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

El siguiente diagrama de secuencia describe la interacción concurrente de tres partes: «productor push, consumidor pop, ladrón steal»:

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

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

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

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

Reflexión de diseño: por qué la cola local es LIFO y el robo es FIFO

Hasta aquí, la tarea ha completado su transformación de Future a entidad planificable: se le asignó un conteo de referencias, se colocó enCellel diseño de memoria, y se entregó con éxito a la cola local del worker o a la cola de inyección global. Pero poner la tarea en la cola es solo el comienzo, lo que realmente la hace funcionar es el bucle de planificación del hilo worker. En el próximo capítulo entraremos en la segunda mitad de «la vida de una tarea», rastreando cómo el worker saca tareas de la cola, llama aFuture::poll, y al retornarPendingregistra el despertar medianteWaker, finalmente disparandoscheduleel reencolado — la ruta de llamada completa del ciclo cerrado «despertar → encolar → re-poll», así como la estrategia work-stealing y la optimización de ranuras LIFO, se revelarán allí.

De la cola a la ejecución: el esqueleto del bucle principal del worker

En el capítulo anterior enviamos la tarea aLocalla cola o a la cola de inyección global. Pero la cola es solo una «lista de pendientes», lo que realmente hace correr la tarea es ese bucle interminable en el hilo worker. En este capítulo rastreamosContext::run—es el corazón de todo el planificador multihilo.

Primero establezcamos la intuición: el hilo worker es como un chef, frente a él tiene una pila de sus propios pedidos (run_queue), y al lado hay un estante de pedidos público (inject). El chef primero mira la hoja más cercana a su mano (lifo_slot), si no hay, toma de su propia pila; si tampoco hay, agarra un puñado del estante público; si aún así no funciona, roba algunas hojas de la pila de otro cocinero. Solo cuando todo está vacío va a descansar, pero mientras descansa mantiene las orejas alerta — en cuanto entra un pedido, se despierta de inmediato.

Sin este bucle, la tarea quedaría en la cola para siempre tras ser encolada,Future::pollnunca sería invocada, y todo el runtime sería un montón de datos muertos.

Diseño de memoria y campos de estado de Core

Todo el estado mutable del worker está contenido enCore, que es asignadoBoxen el heap, y se pasa a través deAtomicCell<Core>entreWorkery el almacenamiento local del hiloContext.

CoreLos campos clave de📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:113-167：

• tick: u32son los siguientes: se incrementa en cada iteración del bucle, usado para disparar periódicamente el mantenimiento (maintenance) y la verificación de la cola global.
• lifo_slot: Option<Notified>：Ranura LIFO, este es el diseño más ingenioso de este capítulo. Cuando el worker programa una tarea por sí mismo, no la pone enrun_queue, sino en esta ranura, y la próxima vez que toma una tareapriorizatomarla de aquí.
• lifo_enabled: bool: interruptor de la ranura LIFO, usado para prevenir inanición en escenarios de ping-pong.
• run_queue: queue::Local<Arc<Handle>>: cola local, la estructuraLocalanalizada en el capítulo anterior.
• is_searching: bool: indica si el worker está buscando tareas que pueda robar.
• is_shutdown: bool / is_traced: bool: indicadores de cierre y seguimiento.
• park: Option<Parker>: parker, envuelto conOptionpara facilitar su extracción/reinserción bajo el borrow checker.
• global_queue_interval: u32: cada cuánto tiempo verificar la cola global.
• rand: FastRand: generador rápido de números aleatorios, usado para elegir aleatoriamente el punto de inicio del robo.

¿Por qué LIFO reduce la latencia? Considera un escenario típico de paso de mensajes: la tarea A termina de procesar un mensaje y despierta a la tarea B, B termina de procesar y despierta a A. Si después de que A despierta a B, B se ejecuta inmediatamente, los datos que B necesita probablemente aún estén en la caché de CPU (porque A acaba de tocarlos). Si B se coloca al final de la cola, tras ejecutarse decenas de tareas anteriores, la caché ya habrá sido desplazada.

Pero LIFO tiene riesgo de inanición. El código fuente usaMAX_LIFO_POLLS_PER_TICK = 3para limitar📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:263-263: en cada tick se prioriza la ranura LIFO como máximo 3 veces; superado eso se deshabilita, dando oportunidad a otras tareas de ejecutarse.

Recorrido del bucle principal: un ciclo completo de planificación

Nos situamos en un escenario concreto: el worker 0 acaba de despertar depark,run_queuetiene 5 tareas,lifo_slottiene 1 tarea, la cola global tiene 3 tareas.

La entrada del bucle principal esContext::run 📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:570-642. Primero reinicialifo_enabled(porque el core puede haber sido robado porblock_in_place, el estado necesita restablecerse)📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:571-573, luego entra en el buclewhile !core.is_shutdown.

Cada iteración del bucle hace cuatro cosas:

Primer paso: tick y mantenimiento. core.tick()incrementa el contador📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:587. Luegoself.maintenance(core)verificatick % event_interval == 0, y si es así llama apark_yieldpara impulsar I/O y temporizadores con timeout 0📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:809-826。

Segundo paso: tomar tarea. core.next_task(&self.worker)es la lógica central de toma de tareas📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1090-1156. Se divide en dos rutas:

• Cuandotick % global_queue_interval == 0,priorizatomar de la cola global, y si no hay, toma de la local📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1091-1098. Esto es para evitar que las tareas de la cola global mueran de inanición.
• De lo contrariopriorizatomar tareas locales📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1090-1156。

La toma local la realizanext_local_task📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1158-1160：

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

primero toma la ranura LIFO, luego la cabeza de la cola (extracción LIFO). Esto es lo que se mencionó en el capítulo anterior como «LIFO local».

Si la local está vacía pero la cola global no, el workeren loteextrae tareas de la cola global📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1110-1154. El cálculo del tamaño del lotenes muy cuidadoso:min(inject.len() / remotes.len() + 1, cap), dondecapa su vez tomamin(remaining_slots, max_capacity / 2). Los comentarios del código fuente explican por qué se limita a la mitad de la capacidad de la cola📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1120-1131: asegurar que las tareas extraídas caigan en laprimera mitadde la cola local, de modo que incluso si ocurre un desbordamiento posterior, estas tareas no sean devueltas a la cola global (el desbordamiento solo afecta a la segunda mitad).

Tercer paso: ejecutar tarea.Tras obtener la tarea, llama arun_task 📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:647-796. Esta es la función más compleja del capítulo, que desarrollaremos en la siguiente sección.

Cuarto paso: robar o park.Sinext_taskdevuelveNone, indica que no hay trabajo ni local ni global, llama asteal_work 📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1167-1195. Si el robo falla, entra enparkopark_yield 📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:613-621。

Todo el flujo de control es el siguiente:

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

run_task: el ciclo cerrado entre poll y la ranura LIFO

run_taskes donde la tarea realmente espoll, y también el punto de cierre del ciclo «despertar → encolar → re-poll».

Lo primero que hace al entrar en la función esassert_owner 📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:648, convierteNotifiedenTask, y al mismo tiempo afirma que el hilo actual es efectivamente el owner de esta tarea (aserción de debug).

Luegotransition_from_searching 📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:652— si el worker estaba antes en estado de búsqueda, ahora que encontró la tarea, debe salir del estado de búsqueda y posiblemente despertar a otros workers en park.

Después viene el envoltorio clave de budget📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:695-795：

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

Este fragmento de código revela el ciclo cerrado completo de la ranura LIFO:task.run()ejecutaFuture::poll, si durante el poll la tarea se despierta a sí misma o a otra tarea,schedule_localpondrá la nueva tarea enlifo_slot 📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1396-1408. Tras retornar el poll, el bucle verifica inmediatamentelifo_slot, y si hay tarea continúa ejecutando —sin volver al bucle principal, haciendo poll continuo dentro del mismo budget.

Esto es la manifestación de «despertar → encolar → re-poll» en la ruta LIFO: al despertar, la tarea se coloca enlifo_slot, y tras retornar el poll se extrae inmediatamente para re-poll, formando un ciclo cerrado estrecho.

Nota la ramaself.core.borrow_mut().take()deNone📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:716-724: si el core fue robado (por ejemplo, si dentro de la tarea se llamó ablock_in_place), el worker debe devolverControlFlow::Break(()), haciendo queContext::runsalga. Esto esblock_in_placePunto de interacción con el bucle de planificación.

Ruta de activación: cómo Waker desencadena la reincorporación a la cola

CuandoFuture::polldevuelvePendingla tarea necesita registrar unWakery ser activada cuando el evento esté listo. La implementación deWakeren Tokio es extremadamente compacta: es simplemente un puntero crudo a la tareaHeadermás una vtable.

waker_refConstruyeWakerRef 📎 tokio/src/runtime/task/waker.rs:11-34usandoManuallyDroppara envolverWakery evitar decrementar el contador de referencias al hacer drop. La vtable es estática📎 tokio/src/runtime/task/waker.rs:119-119：

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

Las cuatro funciones simplemente convierten el puntero crudo de vuelta aHeadery luego llaman al método correspondiente deRawTask📎 tokio/src/runtime/task/waker.rs:70-116Por ejemplo,wake_by_reffinalmente llama araw.wake_by_ref() 📎 tokio/src/runtime/task/waker.rs:106-116。

wake_by_refLa semántica es: cambiar el estado de la tarea dePENDINGaSCHEDULEDy, si la conversión tiene éxito (es decir, si efectivamente estaba en PENDING), llamar aSchedule::schedulepara reincorporar la tarea a la cola.

Para el planificador multihilo,schedulela implementación deHandle::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();
    });
}

La lógica se divide en dos ramas:

• Si el hilo actual es un worker de este planificador y posee el core, se usaschedule_local 📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1385-1417— se coloca en la ranura LIFO o en la cola local.
• En caso contrario (activación desde un hilo externo, o el core fue robado), se usapush_remote_taskpara insertar en la cola de inyección global ynotify_parked_remoteactivar un worker en park📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1379-1383。

schedule_localInternamente se divide en dos ramas📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1385-1417: si esyieldo LIFO está deshabilitado, se inserta al final derun_queue; de lo contrario, se coloca enlifo_sloty la tarea que estaba en la ranura se empuja al final de la cola.

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 y unpark: atomicidad de la máquina de estados y la activación

El worker debe hacer park cuando no tiene trabajo, pero park/unpark es donde más fácilmente ocurren condiciones de carrera. Tokio usa una máquina de estadosAtomicUsizemásCondvarcomo respaldo para resolverlo.

InnerLos campos de📎 tokio/src/runtime/scheduler/multi_thread/park.rs:31-43：state: AtomicUsize、mutex: Mutex<()>、condvar: Condvar、shared: Arc<Shared>. Hay cuatro constantes de estado📎 tokio/src/runtime/scheduler/multi_thread/park.rs:36-45：

• EMPTY = 0: no está en park.
• PARKED_CONDVAR = 1: en park sobre el condvar.
• PARKED_DRIVER = 2: en park sobre el I/O driver.
• NOTIFIED = 3: ya fue activado.

Esta es una máquina de estados explícita; la usamos para dibujar el diagrama de estados (este es el único lugar del capítulo que cumple con los criterios de admisión destateDiagram-v2— en el código fuente efectivamente existen estas cuatro constantes de estado):

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

unparkLa implementación de📎 tokio/src/runtime/scheduler/multi_thread/park.rs:277-290usaswapen lugar de CAS; los comentarios del código fuente explican la razón📎 tokio/src/runtime/scheduler/multi_thread/park.rs:277-290: se debe ejecutar una operación release para que el hilo en park observe las escrituras anteriores al unpark, así que incluso si state ya esNOTIFIEDhay que escribir una vez.

parkPrimero intenta consumir una notificación existente📎 tokio/src/runtime/scheduler/multi_thread/park.rs:132-149: si el CASNOTIFIED -> EMPTYtiene éxito, significa que ya fue activado antes, y retorna directamente sin bloquear. De lo contrario, intenta adquirir el lock del driver; si lo obtiene, hace park sobre el driver; si no, usa el condvar como respaldo📎 tokio/src/runtime/scheduler/multi_thread/park.rs:143-148。

park_condvarHay una doble verificación clásica en📎 tokio/src/runtime/scheduler/multi_thread/park.rs:162-180: primero CASEMPTY -> PARKED_CONDVAR; si falla y esNOTIFIED, significa que fue activado antes de establecer el estado, y en ese momento se debeswap(EMPTY)para sincronizar la escritura del unpark📎 tokio/src/runtime/scheduler/multi_thread/park.rs:167-177. El comentario enfatiza especialmente: incluso sabiendo que esNOTIFIEDtambién hay que leer una vez, porque unpark puede haber sido llamado otra vez después de nuestra lectura deNOTIFIED.

unpark_condvarEl comentario de📎 tokio/src/runtime/scheduler/multi_thread/park.rs:292-307señala la trampa clásica del condvar: entre que el hilo en park establece el estadoPARKEDy realmentewaithay una ventana, y si se hace notify durante ese período se ignora. La solución es que el hilo que hace park poseemutexen ese momento, y el hilo que hace unpark primerodrop(self.mutex.lock())adquiere el lock (esperando así a que el hilo en park lo libere), y luegonotify_one。

Reflexión de diseño: por qué la ranura LIFO es una ranura única y no una cola

MAX_LIFO_POLLS_PER_TICK = 3Este número mágico📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:263-263también es un valor empírico. El comentario del código fuente dice que "ejecutar unas pocas veces la ranura LIFO parece suficiente para beneficiarse de la localidad; más de 3 veces puede sobreponderar". Esto evita que el escenario ping-pong en que A activa a B y B activa a A mate de hambre a otras tareas.

Otro diseño digno de mención es la estrategia de "búsqueda por mitad" desteal_work📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1158-1160: solo cuando menos de la mitad de los workers están buscando, un nuevo worker realmente intenta robar. Esto evita la contención de CAS causada por todos los workers robando frenéticamente al mismo tiempo.transition_to_searchingCoordina medianteidle.transition_worker_to_searching()📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1197-1203。

El robo comienza desde un punto de partida aleatorio📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1172-1174, recorre todos los remote, se salta a sí mismo📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1179-1182y llama asteal_intopara intentar robar. Tras fallar todo, recurre a la cola global📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:1197-1203。

Resumen del capítulo

El bucle principal del workerContext::runes el corazón del planificador: tras cada tick primero toma una tarea (ranura LIFO → cola local → cola global); si la obtiene,run_taskejecuta poll; si no, roba; si el robo falla, hace park.run_taskEl bucle LIFO interno comprime "activar → encolar → volver a hacer poll" dentro del mismo budget, formando un ciclo cerrado de baja latencia.Wakeres un puntero crudo más una vtable estática,wake_by_refdesencadenaschedulemediante transiciones de estado, y según si el hilo actual es el mismo worker decide ir a la cola local o a la global.park/unparkusa una máquina atómica de cuatro estados más un condvar de respaldo, resolviendo la clásica condición de carrera de pérdida de activación.

En el próximo capítulo dejaremos el planificador y entraremos al mundo de I/O: cómo el Reactor traduce eventos de epoll aWakeractivaciones, haciendo queAsyncFddePendingse convierta enReady。

Reflexión y autoevaluación del capítulo

Proyecto: tokio-rs/tokio

Progreso del libro: Capítulo 5 / 14DriverEstado de verificación: líneas FACT con anclaje realHandleEn el capítulo anterior rastreamos el bucle principal del hilo worker: la tarea es poll, al retornar Pending el Waker se guarda en algún lugar, y cuando el evento está listo el Waker se dispara y la tarea se reencola. Pero ¿dónde está exactamente ese «algún lugar»? ¿Cómo se recupera el Waker cuando llega un evento epoll? Esta es precisamente la pregunta que Reactor debe responder. Primero construyamos un modelo intuitivo: imaginemos todo el mecanismo de notificación de E/S lista como el sistema de llamada por número de un restaurante — el cliente (tarea) tras pedir no se queda esperando de pie en la ventanilla, sino que toma un vibrador (Waker) y vuelve a su asiento; cuando la cocina (epoll del kernel) termina el pedido, la recepción (Reactor) busca el vibrador correspondiente según el número de pedido (Token) y pulsa el botón. Sin este sistema, cada tarea solo podría hacer polling del socket, quemando la CPU; o usar hilos bloqueantes en espera, un hilo por conexión, sin poder escalar. El Reactor de Tokio se compone de tres archivos en una estructura de tres capas con responsabilidades estrictamente separadas: driver.rs es el cuerpo del bucle de eventos, posee mio::Poll, se encarga de llamar a poll() bloqueándose a la espera de eventos del kernel y traduce los eventos en lecturas/escrituras sobre ScheduledIo; registration.rs es el handle de registro orientado al usuario, que es lo que TcpStream mantiene internamente, ofreciendo APIs como poll_read_ready / poll_write_ready; scheduled_io.rs es el slot de estado de cada fd, almacena los bits de readiness de lectura/escritura y la lista de Wakers, y es el puente entre eventos y tareas. La relación de ensamblaje de módulos puede consultarse en tokio/src/runtime/io/mod.rs:5-16: driver exporta Driver, Handle, ReadyEvent; registration exporta Registration; scheduled_io exporta ScheduledIo. El siguiente diagrama ancla el flujo de datos completo que este capítulo va a rastrear: TcpStream → Registration → ScheduledIo → Handle/Driver → kernel → de vuelta a ScheduledIo → Waker. A continuación lo desglosamos capa por capa.

Capa de driver:

Driverydivisión de responsabilidadesmio::PollModelo intuitivoes&mutla única entidad que poseeHandle, solo puede ser accedidadesde un único hilo — este es el requisito de exclusividad del bucle de eventos. Mientras que, cualquier hilo que quiera registrar un nuevo fd lo hace a través de él. Sin esta división, o bien se tendría quemio::Pollbloquear con lock (compitiendo en cada registro), o bien hacer que todos los registros vuelvan al hilo driver (introduciendo una cola de mensajes entre hilos). Tokio elige queHandlemantenga directamente un clon demio::Registry, las operaciones de registro pueden realizarse de forma concurrente, y solo la espera real de eventos requiere exclusividad.

Diseño de memoria y campos

Primero veamosDriverlos campos de📎 tokio/src/runtime/io/driver.rs:25-38：

• signal_ready: bool: si ha llegado un evento de señal Unix, usado para el driver de señales.
• events: mio::Events: búfer principal de eventos, reutilizado entre llamadas aturn, evitando asignaciones en cada ocasión.
• events_busy: Option<mio::Events>：Búfer dedicado para poll no bloqueante, existe solo cuandomax_io_events_per_busy_tickestá configurado.
• poll: mio::Poll: envoltorio de la cola de eventos del kernel.

Ahora veamosHandle 📎 tokio/src/runtime/io/driver.rs:41-75：

• registry: mio::Registry：mio::Poll::registry()el clon deregister/deregister。
• registrations: RegistrationSet, usado paraToken: conjunto de todos los registros activos, responsable de asignarScheduledIo。
• synced: Mutex<registration_set::Synced>yRegistrationSet: protege el estado de sincronización de
• waker: mio::Waker: se usa para despertar desde cualquier hilo al driver que está bloqueado enturn.
• metrics: IoDriverMetrics: cuenta el número de fds y de eventos listos.

Aquí hay un diseño clave:events_busyla existencia de📎 tokio/src/runtime/io/driver.rs:25-38es para resolverel problema de que el poll no bloqueante se traga eventos. El comentario📎 tokio/src/runtime/io/driver.rs:189-190lo dice claramente: si los eventos tomados por el poll no bloqueante se quedaran en el búfer principal, el siguiente poll no los vería; usando un búfer independiente, los eventos no procesados permanecen en la cola del kernel y el siguiente poll los devolverá de nuevo.

Paso a paso: una ejecución deturn

turnes la función central del driver📎 tokio/src/runtime/io/driver.rs:184-261. Supongamos que un hilo worker descubre que no hay tareas que ejecutar y llama apark → turn(handle, None)para bloquearse en espera:

Primer paso: afirmar que no se ha hecho shutdown📎 tokio/src/runtime/io/driver.rs:185, y liberar los registros pendientes de limpieza📎 tokio/src/runtime/io/driver.rs:187。release_pending_registrationscomprobarneeds_release(), y si los hay, llamar aregistrations.release() 📎 tokio/src/runtime/io/driver.rs:336-340。

Segundo paso: elegir el búfer de eventos📎 tokio/src/runtime/io/driver.rs:191-194. Simax_waites cero yevents_busyexiste, usar el búfer busy; en caso contrario, usar el búfer principal.

Tercer paso: llamar aself.poll.poll(events, max_wait) 📎 tokio/src/runtime/io/driver.rs:198. Este es el lugar donde realmente se bloquea en epoll_wait. El manejo de errores es muy contenido:Interruptedse ignora directamente (una interrupción por señal es normal)📎 tokio/src/runtime/io/driver.rs:200, bajo WASIInvalidInputtambién se ignora📎 tokio/src/runtime/io/driver.rs:201-205, otros errores provocan panic directamente📎 tokio/src/runtime/io/driver.rs:206。

Cuarto paso: recorrer los eventos📎 tokio/src/runtime/io/driver.rs:211-233. Para cadaevent：

• sitoken == TOKEN_WAKEUP(valor 0)📎 tokio/src/runtime/io/driver.rs:214, no hacer nada: esto es lo que usaunparkpara interrumpir el bloqueo.
• Sitoken == TOKEN_SIGNAL(valor 1)📎 tokio/src/runtime/io/driver.rs:216, establecersignal_ready = true。
• ; en caso contrario, es un evento de E/S normal📎 tokio/src/runtime/io/driver.rs:218-231: convertirmio::Readyen elReadyde Tokio, usarEXPOSE_IO.from_exposed_addr(token.0)para restaurar el token a un puntero*const ScheduledIo, luegoset_readiness(Tick::Set, |curr| curr | ready)acumular los bits de listo, y despuésio.wake(ready)disparar elWaker。

en la dirección correspondienteEXPOSE_IOAquíPtrExposeDomain<ScheduledIo> 📎 tokio/src/runtime/io/mod.rs:21-22es unusize, que «expone» el puntero como unmio::Tokencomo📎 tokio/src/runtime/io/driver.rs:222-225. El comentario de seguridadexplica por qué esta conversión unsafe es segura: el puntero no se libera antes de darse de baja de mioyArc<ScheduledIo>que el driver deje de hacer poll de forma concurrente, y el driver posee la propiedad de

Quinto paso: procesar la cola de completación de io_uring (solo Linux + tokio_unstable)📎 tokio/src/runtime/io/driver.rs:235-258, incluido el bucle de flush cuando hay desbordamiento de CQ.

Sexto paso: acumular métricas📎 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)"]

Reflexión de diseño: por quéHandledebe poseermio::Waker

unpark 📎 tokio/src/runtime/io/driver.rs:280-283llamar aself.waker.wake(). Estemio::WakerenDriver::newse registra conTOKEN_WAKEUPusando📎 tokio/src/runtime/io/driver.rs:124. Cuando el driver está bloqueado enpoll.poll(), otro hilo que llame aunparkinsertará un eventoTOKEN_WAKEUPen epoll,pollregresará inmediatamente, y al recorrer verá este token y lo saltará directamente📎 tokio/src/runtime/io/driver.rs:214-215。

Otro detalle:deregister_sourceprimero llama aself.registry.deregister(source) 📎 tokio/src/runtime/io/driver.rs:322, y luego limpiaregistrations 📎 tokio/src/runtime/io/driver.rs:315-334. El comentario📎 tokio/src/runtime/io/driver.rs:320-321dice «Cleanup ALWAYS happens»: incluso si el deregister a nivel de SO falla, también hay que limpiar el estado interno, y solo al final devolver el error del SO📎 tokio/src/runtime/io/driver.rs:336-340. Este es el típico patrón delimpieza de recursos antes que propagación de errores.

Capa de registro:Registrationcómo almacenarWakerenScheduledIo

Modelo intuitivo

Registrationesel contrato entre la tarea y el fd. Mantiene dos cosas: unscheduler::Handle(usado para acceder al runtime cuando sea necesario), y unArc<ScheduledIo>(la ranura de estado del fd). Cuando una tarea llama apoll_read_ready,RegistrationentregaWakeraScheduledIopara que lo custodie; cuando el driver recibe un evento, sacaScheduledIodeWakerpara despertar.

Diseño de memoria y campos

Registrationsolo tiene dos campos📎 tokio/src/runtime/io/registration.rs:46-54：

• handle: scheduler::Handle: el handle del runtime, el comentario📎 tokio/src/runtime/io/registration.rs:46-54dice «TODO: this can probably be moved into ScheduledIo», lo que indica que el autor cree que la posición de este campo puede optimizarse.
• shared: Arc<ScheduledIo>: estado compartido,Arcgarantiza que tanto el driver como la tarea puedan acceder.

Step-by-Step：poll_read_readyla cadena de llamadas de

Supongamos que la tarea enTcpStream::poll_readse descubre que el socket no tiene datos, es necesario registrar el interés de lectura. La cadena de llamadas esTcpStream::poll_read_priv → PollEvented::poll_read → Registration::poll_read_io → poll_io → poll_ready。

poll_readyes el núcleo📎 tokio/src/runtime/io/registration.rs:155-171：

Primer paso：trace_leaf() 📎 tokio/src/runtime/io/registration.rs:160, utilizado para el trazado de instrumentación.

Segundo paso：coop::poll_proceed(cx) 📎 tokio/src/runtime/io/registration.rs:155-171. Este es el mecanismo de presupuesto cooperativo que se explicará en el capítulo 12. Si el presupuesto se agota, devuelvePendingy registra unWakerespecial, para que la tarea sea reprogramada en la siguiente ronda.

Tercer paso：self.shared.poll_readiness(cx, direction) 📎 tokio/src/runtime/io/registration.rs:155-171. Este es el lugar donde realmente se interactúa conScheduledIo: verifica el bit de listo actual, si ya está listo devuelve inmediatamenteReady; de lo contrario almacenacx.waker()enScheduledIola ranura de dirección correspondiente dePending。

, devuelveCuarto pasoev.is_shutdown 📎 tokio/src/runtime/io/registration.rs:155-171: verificaRUNTIME_SHUTTING_DOWN_ERROR。

. Si el runtime se está cerrando, devuelve：coop.made_progress() 📎 tokio/src/runtime/io/registration.rs:169Quinto paso

poll_io, marca el consumo de presupuesto, devuelve el evento de listo.poll_readyañade una capa de bucle de reintento sobre📎 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)),
    }
}

Aquí se reflejareadiness es una sugerencia, no una garantíala idea central depoll_ready: dice que es legible, pero al realmenteread()puede devolverWouldBlock(por ejemplo, otro hilo se adelantó y leyó los datos). En este caso se debeclear_readiness(ev) 📎 tokio/src/runtime/io/registration.rs:187limpiar el bit de listo, y luego repetir el bucle de espera. Si no se limpia, la tarea caerá en un bucle ocupado de «cree que es legible → read falla → vuelve a creer que es legible».

Reflexión de diseño:try_ioyasync_iola división de trabajo

try_io 📎 tokio/src/runtime/io/registration.rs:194-213es la versión síncrona: primeroready_event(interest)verifica el bit de listo, si está vacío devuelve directamenteWouldBlock 📎 tokio/src/runtime/io/registration.rs:194-213; de lo contrario ejecutaf(), sif()devuelveWouldBlockentonces limpia el bit de listo📎 tokio/src/runtime/io/registration.rs:207-210. Esteno registra Waker, es adecuado paratry_readescenarios de «intentar y salir» como

async_io 📎 tokio/src/runtime/io/registration.rs:225-245es la versión asíncrona:readiness(interest).awaitregistra el Waker y espera, luego al ejecutarf()，WouldBlocklimpia el bit de listo y repite el bucle. Nótese que dentro del bucle también llama acoop::poll_proceed 📎 tokio/src/runtime/io/registration.rs:233, para evitar agotar el presupuesto en una gran cantidad deWouldBlockreintentos.

Problemas en producción:Dropla limpieza del Waker en

Registration::drop 📎 tokio/src/runtime/io/registration.rs:253-262llama aself.shared.clear_wakers(). El comentario📎 tokio/src/runtime/io/registration.rs:253-262explica la razón:ScheduledIoelWakeralmacenado enArc<driver::Inner>puede contenerdriver::Inner, yScheduledIoa su vez contieneRegistration, formando una referencia circular. Limpiar el Waker es un medio para romper el ciclo. Pero el comentario también admite que es una «imperfect solution» — siWakermismo se almacena en

DesdeTcpStream::readhastaWakerla cadena completa de activación

Modelo intuitivo

Ahora conectemos las tres capas. El usuario llama aTcpStreamsobre.read().await, lo que realmente se ejecuta esAsyncRead::poll_read → PollEvented::poll_read → Registration::poll_read_io. Cuando los datos no han llegado,Wakerse almacena enScheduledIo; cuando epoll reporta legible, el driver sacaScheduledIodeWakery activa, la tarea es reprogramada, y al hacer poll de nuevopoll_readinessdescubre que el bit de listo ya está puesto, devuelve directamenteReady，read()con éxito.

Paso a paso: una espera de lectura completa

Fase uno: registrar interés。TcpStream::new 📎 tokio/src/net/tcp/stream.rs:166-169llama aPollEvented::new(connected), este internamente llama aRegistration::new_with_interest_and_handle 📎 tokio/src/runtime/io/registration.rs:73-81, y luegohandle.driver().io().add_source(io, interest) 📎 tokio/src/runtime/io/registration.rs:73-81。

add_source 📎 tokio/src/runtime/io/driver.rs:288-312hace tres cosas:

1. registrations.allocate(&mut synced.lock())asigna unScheduledIo, obtienetoken 📎 tokio/src/runtime/io/driver.rs:293-294。

2. self.registry.register(source, token, interest.to_mio())registra📎 tokio/src/runtime/io/driver.rs:298ante el kernel. Si falla,debeeliminar elScheduledIorecién asignado del conjunto📎 tokio/src/runtime/io/driver.rs:300-303, de lo contrario hay fuga.

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

Fase dos: esperar a que esté listo. La tarea hace pollTcpStream::poll_read 📎 tokio/src/net/tcp/stream.rs:1492-1498 → poll_read_priv 📎 tokio/src/net/tcp/stream.rs:1451-1458 → PollEvented::poll_read → Registration::poll_read_io 📎 tokio/src/runtime/io/registration.rs:133-139 → poll_io → poll_ready → ScheduledIo::poll_readiness. En este momento si no está listo,Wakerse almacena enScheduledIola ranura de lectura dePending。

, devuelveFase tres: llega el eventoturn. Elpoll.poll()del driver obtiene el evento📎 tokio/src/runtime/io/driver.rs:198deio.set_readiness(Tick::Set, |curr| curr | ready), al recorrer ejecuta para cada evento de fdio.wake(ready) 📎 tokio/src/runtime/io/driver.rs:228-229。wakeyWakerinternamente saca elwake()。

de la dirección correspondiente y llama a。Waker::wake()Fase cuatro: reprogramación de la tareapoll_readinessvuelve a encolar la tarea en la cola local del worker (explicado en el capítulo anterior). El worker hace poll de nuevo a esa tarea,Ready，read()descubre que el bit de listo ya está puesto, devuelve

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

copiarassume_readyRama importante:

TcpStream::new_accepted 📎 tokio/src/net/tcp/stream.rs:174-181optimizaciónacceptes una optimización que vale la pena notar.new_acceptedel socket devuelto es naturalmente escribible, y normalmente ya tiene el primer lote de bytes del par. Si se espera al primer evento del driver, bajo alta carga este evento puede quedar detrás de todos los eventos de conexiones ya establecidas, causando latencia. Por esoassume_ready(Ready::READABLE | Ready::WRITABLE) 📎 tokio/src/net/tcp/stream.rs:174-181。

assume_readyllama directamente a📎 tokio/src/runtime/io/registration.rs:103-105el comentarioWouldBlockdice: «A wrong guess costs oneWouldBlock，poll_io, which clears the readiness again.» — el costo de adivinar mal es solo unel bucle delimpiará el bit de listo y volverá a esperar. Este es un diseño de

conjetura optimista + corrección rápida de errores

1. . Este desacoplamiento trae varias ventajas:：HandleRegistro sin bloqueosmio::Registrytiene

2. clonado, cualquier worker puede registrar nuevos fd concurrentemente, sin necesidad de volver al hilo del driver.Centralización de la espera de eventosepoll_wait: solo un hilo se bloquea en

3. , evitando el problema de thundering herd al hacer poll del mismo epoll fd desde múltiples hilos.Ruta de activación cortaScheduledIo: el driver tras recibir el evento opera directamente sobreWaker::wake()，wake()y llama a

internamente empuja la tarea a la cola del worker, sin necesidad de paso de mensajes entre hilos.ScheduledIoEl costo es queset_readinessnecesita manejar acceso concurrente (poll_readinessy

pueden ocurrir simultáneamente), esto se resuelve mediante operaciones atómicas y bloqueos internos.is_shutdownProblemas en producción:RUNTIME_SHUTTING_DOWN_ERROR

poll_readyyev.is_shutdown 📎 tokio/src/runtime/io/registration.rs:155-171verificagone() 📎 tokio/src/runtime/io/registration.rs:265-267, si es verdadero devuelveRUNTIME_SHUTTING_DOWN_ERROR。

Otra trampa esderegister_sourcedeunpark 📎 tokio/src/runtime/io/driver.rs:328. Si el driver está bloqueado enpolly en ese momento se destruye el últimoRegistration,unparkdespertará al driver. Pero si el driver no está bloqueado (por ejemplo, está procesando otros eventos),unparksolo hace que la siguienteturndevuelva inmediatamente📎 tokio/src/runtime/io/driver.rs:280-283. Esta semántica está documentada en los comentarios deHandle::unpark.

Reflexión de diseño: las tres compensaciones clave del Reactor

Compensación uno:Tokenusar punteros en lugar de índices。EXPOSE_IO.from_exposed_addr(token.0) 📎 tokio/src/runtime/io/driver.rs:220tratarmio::Tokendirectamente como la dirección de*const ScheduledIo. Esto evita mantener una tabla de asignación deToken → ScheduledIo, y la búsqueda es O(1) y sin bloqueos. El coste es que la seguridad depende de una gestión estricta del ciclo de vida: el puntero solo puede liberarse después de darse de baja y de que el driver deje de hacer poll📎 tokio/src/runtime/io/driver.rs:222-225。

Compensación dos: dos ranuras de Waker para lectura y escritura。RegistrationLa documentación de📎 tokio/src/runtime/io/registration.rs:24-26dice «A registration instance represents two separate readiness streams»: lectura y escritura tienen cada una una ranura independiente deWaker. Esto permite que las tareas de lectura y escritura del mismo socket se registren por separado sin interferirse. Pero el comentario depoll_read_ready📎 tokio/src/net/tcp/stream.rs:549-552advierte: llamar varias veces apoll_read_ready/poll_read/poll_peeksolo conserva elWakerde la última llamada; la dirección de lectura solo tiene una ranura.

Compensación tres:events_busybúfer independiente de. El test📎 tokio/src/runtime/io/driver.rs:364-386verifica este comportamiento:Driver::new(16, Some(2))crea un driver con capacidad busy de 2, registra 5 fuentes legibles y, en unturnno bloqueante, solo toma 2 eventos📎 tokio/src/runtime/io/driver.rs:375-376, dejando los 3 restantes en la cola del kernel; en el siguienteturnbloqueante se obtienen📎 tokio/src/runtime/io/driver.rs:379-380. Esto evita que un poll no bloqueante consuma todos los eventos de una vez y provoque inanición en polls posteriores.

Resumen del capítulo

Este capítulo ha seguido la cadena completa del Reactor detrás deTcpStream::read:

• Capa de driver：Drivermonopolizamio::Poll，turny espera eventos de forma bloqueante, usaEXPOSE_IOpara convertirTokende nuevo en el punteroScheduledIo, llama aset_readiness + wakepara activarWaker。Handleproporciona un punto de entrada de registro que puede cruzar hilos,unparkse usa para interrumpir el bloqueo.
• Capa de registro：RegistrationmantieneArc<ScheduledIo>，poll_readycomprueba los bits de readiness o los almacena enWaker，poll_iousaWouldBlockun bucle de reintento para manejar falsos positivos,try_io/async_iosirve por separado a escenarios síncronos y asíncronos.
• Capa de estado：ScheduledIoes la ranura de estado del fd, almacena los bits de readiness de lectura/escritura y las dos ranuras deWaker, y es el único puente entre eventos y tareas.

Reflexión y autoevaluación de este capítulo

En el capítulo anterior rastreamos la cadena completa de TcpStream::read y vimos cómo ScheduledIo traduce los eventos de disponibilidad del fd de epoll en despertares de Waker. Pero el runtime asíncrono aún necesita manejar otro tipo de «disponibilidad»: un Future de sleep(100ms) debe ser despertado después de 100ms. Este tipo de eventos no proviene de un fd del kernel, sino del «tiempo mismo». La decisión de diseño de Tokio es tratar el tiempo también como un evento de E/S: en la estructura Driver solo hay un campo park: IoStack, que reutiliza el mecanismo park/unpark del driver de E/S. Cuando la rueda de tiempo calcula el «próximo instante de vencimiento», el driver llama a park_timeout para que el hilo duerma hasta ese instante; una vez despertado, extrae de la rueda de tiempo las entradas vencidas y dispara sus Waker. Así, el planificador solo necesita un punto de entrada unificado de park para esperar simultáneamente ambos tipos de eventos: «fd listo» y «temporizador vencido». Este capítulo responde a tres preguntas: ¿cómo se insertan los temporizadores en la rueda de tiempo? ¿Cómo se clasifica la rueda de tiempo por tiempo de vencimiento? ¿Cómo calcula el driver el timeout del próximo park y dispara las tareas vencidas?

I. Rueda de tiempo: estructura jerárquica hash de seis niveles y 64 ranuras

Modelo intuitivo

Imagina un reloj mecánico: el segundero da una vuelta y arrastra al minutero, el minutero da una vuelta y arrastra a la manecilla de las horas. Si solo hubiera un segundero, para representar «dentro de 12 días» habría que contar 1 millón de marcas; pero tras la estratificación, el segundero solo se encarga de la precisión dentro de 64 segundos, el minutero de 64 minutos, la manecilla de las horas de 64 horas; cada nivel solo necesita 64 ranuras para cubrir hasta 2 años en el futuro.

Sin estratificación, insertar un temporizador lejano requeriría un recorrido O(N) o un array enorme. La rueda de tiempo usa la «clasificación por tiempo de vencimiento» para reducir tanto la inserción como el disparo a un orden cercano a O(1).

Diseño de memoria y campos

WheelLos campos centrales de solo son tres📎 tokio/src/runtime/time/wheel/mod.rs:22-40：

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

NUM_LEVELS = 6，BITS_PER_LEVEL = 6(es decir, 64 ranuras por nivel)📎 tokio/src/runtime/time/wheel/mod.rs:45-47。MAX_DURATION = 1 << (6 * 6) = 1 << 36milisegundos, aproximadamente 2 años📎 tokio/src/runtime/time/wheel/mod.rs:50。

La granularidad de los seis niveles, según los comentarios de la documentación, es📎 tokio/src/runtime/time/wheel/mod.rs:22-40：

pendinges una lista enlazada intrusiva (LinkedList<TimerShared>), que almacena las entradas que ya han sido extraídas de la rueda y esperan disparar el Waker. Ten en cuenta que esLinkedListy noVec: la entrada en sí está incrustada enTimerShared, por lo que insertar/eliminar no requiere asignación.

Guiado por escenario: insertar un sleep de 100ms

Cuandosleep(100ms)se sondea por primera vez,Sleep::poll_elapsedconstruyeTimer::newy llama ainit 📎 tokio/src/time/sleep.rs:436-440。initfinalmente llama aHandle::reregister, y a su vez llama aWheel::insert。

insertEl primer paso es comprobar si ya ha vencido📎 tokio/src/runtime/time/wheel/mod.rs:90-98：

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

Siwhenya ha caído antes deelapsed(por ejemplo, el deadline ya pasó), devuelve directamenteElapsed, y el llamador disparará inmediatamente ese temporizador.

En caso contrario, calcula en qué nivel debe colocarse la entrada📎 tokio/src/runtime/time/wheel/mod.rs:90-114：

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

level_fores el núcleo del algoritmo de clasificación📎 tokio/src/runtime/time/wheel/mod.rs:276-289：

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

Aquí se usaelapsed ^ whenen lugar dewhen - elapsed, lo cual es un truco ingenioso: el bit más significativo del XOR refleja «a partir de qué bit difieren las dos marcas de tiempo», es decir, «qué granularidad se necesita para distinguirlas».| SLOT_MASKfuerza a 1 los 6 bits bajos, evitando queilog2calcule un nivel demasiado pequeño cuando cae en la misma ranura.ilog2() / 6asigna el ancho de bits al número de nivel. Si el resultado del XOR superaMAX_DURATION(es decir, supera los 2 años), se fuerza a meterlo en el nivel más alto; esto es «fudge the timer into the top level».

Para un sleep de 100ms, suponiendo queelapsedestá cerca de 0,when ≈ 100，elapsed ^ when ≈ 100，ilog2(100) = 6，6 / 6 = 1, por lo que cae en el nivel 1 (granularidad de 64 ms). Esto significa que esperará en una ranura del nivel 1 hasta que el tiempo avance hasta el límite de esa ranura, momento en el cual será descendido al nivel 0.

Descenso por niveles: process_expiration

Cuandopoll(now)avanza el tiempo,Wheel::pollllama cíclicamente anext_expirationyprocess_expiration 📎 tokio/src/runtime/time/wheel/mod.rs:142-166：

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

process_expirationse encarga de "descender" las entradas expiradas de un nivel al siguiente, o (en el nivel 0) marcarlas como pending📎 tokio/src/runtime/time/wheel/mod.rs:218-251：

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

mark_pendinges clave: verifica si el deadline real de la entrada ya ha llegado. Si llegó, devuelveOk(()), la entrada entra en lapendinglista enlazada; si aún no llegó (solo se alcanzó el límite de la ranura donde se encuentra), devuelveErr(expiration_tick), y la entrada se reinserta en un nivel más fino.

Nótese un punto enfatizado en los comentarios📎 tokio/src/runtime/time/wheel/mod.rs:219-228: se deben extraer todas las entradas de la ranura completa antes de procesarlas, porque algunas entradas pueden ser reinsertadas en la misma ranura (esto ocurre cuando el tiempo de inserción excedeMAX_DURATION, produciendo un wraparound). Si se extrae e inserta a la vez, se puede caer en un bucle infinito.

Cálculo del próximo instante de expiración

next_expirationEscanea desde el nivel más bajo al más alto, devolviendo el primer punto de expiración no vacío📎 tokio/src/runtime/time/wheel/mod.rs:169-191：

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

Sipendingno está vacío, significa que hay entradas ya expiradas pendientes de disparo, y devuelve inmediatamente elelapsedactual como deadline (así el driver hará park con timeout 0 y volverá de inmediato a procesarlas). En caso contrario, escanea nivel por nivel y devuelve el deadline de la primera ranura con contenido.debug_assertSe verifica un invariante: ningún nivel superior puede tener un punto de expiración más temprano que el nivel actual.

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

---

II. El bucle park del Driver: conectando el temporizador jerárquico a la pila de I/O

Modelo intuitivo

El temporizador jerárquico por sí solo no "avanza solo". Necesita un bucle externo que le pregunte repetidamente: "¿Cuándo es la próxima expiración?" y luego duerma hasta ese momento, y al despertar avance el tiempo. Este bucle esDriver::park_internal. Traduce "la próxima expiración del temporizador jerárquico" en una duración depark_timeout, que se entrega a la pila de I/O subyacente para dormir.

Sin este bucle, los temporizadores nunca se dispararían—el temporizador jerárquico es solo una estructura de datos estática que necesita que alguien lo "accione".

Estructuras de datos: Driver e InnerState

Driversolo tiene un campopark: IoStack 📎 tokio/src/runtime/time/mod.rs:90-93. El estado real está enHandle, distinguiendo entre la implementación tradicional y la experimental mediante la enumeraciónInner. La implementación tradicional de📎 tokio/src/runtime/time/mod.rs:95-127contiene dos camposInnerStateCopia📎 tokio/src/runtime/time/mod.rs:130-136：

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

next_wakeen lugar deNonZeroU64anidadosOption<u64>para aprovechar la optimización de niche—Option<NonZeroU64>yu64tienen el mismo tamaño. Registra "antes de qué tick el driver promete despertar", usado parareregisteral determinar si se necesitaunpark。

is_shutdownes unAtomicBoolindependiente, y los comentarios explican por qué se separó del Mutex📎 tokio/src/runtime/time/mod.rs:90-93：HandleSe necesita verificaris_shutdownsin bloquear el mutex. Esta es una optimización típica de "muchas lecturas, pocas escrituras"—shutdown solo ocurre una vez, pero la verificación puede ser frecuente.

Guiado por escenarios: el flujo completo de un park

park_internales el núcleo📎 tokio/src/runtime/time/mod.rs:213-256：

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

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

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

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

Análisis paso a paso:

1. Tomar el lock, leer la próxima expiración：lock.wheel.next_expiration_time()devuelveOption<u64>, es decir, el próximo tick de expiración. A la vez lo escribe enlock.next_wake, para quereregisterdetermine si se necesita unpark.

2. Liberar el lock：drop(lock)debe hacerse antes del park, de lo contrario otros hilos no podrían insertar temporizadores durante el park.

3. Calcular la duración del park：when.saturating_sub(now)obtiene el número de ticks restantes,tick_to_durationse convierte aDuration. Los comentarios señalan que aquí en realidad se redondea hacia arriba a 1 ms📎 tokio/src/runtime/time/mod.rs:228-230, para evitar que un sleep de nivel microsegundo sea tratado por el SO como de longitud cero.

4. Manejar el limit: si el llamador pasólimit(por ejemplo, el timeout explícito depark_timeout), se tomamin(limit, duration), garantizando no dormir de más.

5. Caso especial: siduration == 0(ya expirado), se usapark_timeout(0)para retornar inmediatamente, sin dormir realmente.

6. Sin temporizadores: sinext_wakeesNone, conlimitse hacepark_thread_timeout(limit), de lo contrariopark。

7. infinito：handle.process(clock)Procesar tras despertar

avanza el temporizador jerárquico y dispara las entradas expiradas.

processprocess_at_time: disparar entradas expiradasprocess_at_time 📎 tokio/src/runtime/time/mod.rs:296-337：

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

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

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

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

Copia

• Puntos clave: 📎 tokio/src/runtime/time/mod.rs:301-309Protección contra retroceso del tiemponow < wheel.elapsed(): siInstant, significa que el reloj retrocedió. Los comentarios señalan que esto normalmente no debería ocurrir (Rust garantiza quenowes monótono), pero sí ocurre en VMs Linux sobre hosts Windows, porque std confía erróneamente en que el reloj de hardware es monótono. La forma de protegerse es fijarelapsed。
• a：WakeListDespertar por lotes!can_push()se recolectan Wakers, y cuando se llena (📎 tokio/src/runtime/time/mod.rs:319) se libera temporalmente el lock, se despierta un lote, y se vuelve a tomar el lock. Los comentarios enfatizan que esto es para evitar deadlocks
• . Si se llama a un Waker mientras se tiene el lock, y el Waker intenta operar sobre el temporizador jerárquico (por ejemplo, reregistrar un temporizador), se produciría un deadlock.Actualizar next_wakepoll_at(): tras procesar, recalcularnext_wake。

, actualizar

reregister: reregistro y unparkSleep::resetCuando se llama areregister, el temporizador necesita reregistrarse.📎 tokio/src/runtime/time/mod.rs:398-450：

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

Copianext_wakeLógica clave: tras una inserción exitosa, si el nuevo instante de expiración es más temprano queunpark.unpark(), se llama a

para despertar al driver. Esto se debe a que el driver podría estar durmiendo hasta un momento más tardío, y necesita ser despertado antes para recalcular la duración del park.unparkNótese quese llamacon el lock tomadowaker.wake(), mientras quese llamatras liberar el lock📎 tokio/src/runtime/time/mod.rs:441. Los comentarios explicanunpark: se debe liberar el lock antes de llamar al Waker para evitar deadlocks. Pero

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

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

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

---

Copia

III. Sleep y Timeout: la capa de API visible al usuario

SleepModelo intuitivo.awaites el Future que el usuarioTimeoutes un adaptador que envuelve otro Future. Por sí mismos no gestionan la rueda de tiempo, solo traducen el «deadline» a tick y lo delegan aTimeryHandle。

Diseño de memoria de Sleep

Sleepusapin_project!macro define📎 tokio/src/time/sleep.rs:221-227：

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

timeresOption<Timer>y con#[pin]: antes del primer poll esNone, solo en el primer poll se creaTimery se registra. Esta «inicialización perezosa» evita acceder al runtime en el momento de la llamada asleep()—sleep()puede llamarse fuera del runtime, siempre que el registro real ocurra en.await.

PinnedDropimplementación garantiza cancelar el temporizador al hacer drop📎 tokio/src/time/sleep.rs:230-235：

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

Flujo completo de poll_elapsed

poll_elapsedesSleepnúcleo de📎 tokio/src/time/sleep.rs:396-454：

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

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

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

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

Paso a paso:

1. Verificación del presupuesto coop：poll_proceed(cx)consume un presupuesto de cooperación. Si el presupuesto se agota, devuelvePendingy cede la ejecución. Este es el mecanismo de Tokio para evitar que una sola tarea mate de hambre a las demás.

2. Creación perezosa del Timer: sitimeresNone, conviertedeadlinea tick, creaTimery llama ainitpara registrarlo en la rueda de tiempo.

3. Delegación a Timer::poll_elapsed: la verificación real de expiración la realizaTimer.

4. Marcar progreso tras el éxito：coop.made_progress()indica que este poll tuvo progreso real.

Poll de Timeout: primero poll del valor, luego poll del delay

Timeoutel orden de poll es crítico📎 tokio/src/time/timeout.rs:210-224：

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

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

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

El comentario indica explícitamente📎 tokio/src/time/timeout.rs:24-26: el future se poll primero, y solo después se verifica el timeout. Así que si el future se completa sin ceder, puede devolverOkincluso después de superar el timeout. Esto es una decisión de diseño, no un bug.

poll_delaymaneja un escenario sutil📎 tokio/src/time/timeout.rs:229-251：

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

    let has_budget_now = coop::has_budget_remaining();

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

Lógica: si al entrar enpollaún queda presupuesto, pero tras hacer poll del value el presupuesto se agota, significa que fue el value quien consumió el presupuesto. En ese caso, si se hace poll del delay con presupuesto restringido, el delay podría devolverPendinginmediatamente, haciendo imposible determinar si se alcanzó el timeout. Por eso se usawith_unconstrainedpara levantar temporalmente la restricción de presupuesto. El comentario lo llama «pathological cases»📎 tokio/src/time/timeout.rs:243-246。

Manejo de desbordamiento del deadline en timeout

timeoutla función usachecked_addpara manejar el desbordamiento📎 tokio/src/time/timeout.rs:86-99：

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

SiInstant::now() + durationse desborda (duration extremadamente grande),delayesNone, y en el poll devuelve directamentePoll::Pending 📎 tokio/src/time/timeout.rs:222. Esto equivale a «nunca expira», un comportamiento de degradación razonable.

---

Reflexiones de diseño y trampas en producción

¿Por qué usar XOR en lugar de resta para calcular el nivel? elapsed ^ whenel bit más significativo refleja directamente «desde qué bit difieren dos timestamps», que es precisamente la medida de «qué granularidad se necesita». La restawhen - elapsedcuandoelapsedestá cerca dewhendeja los bits altos en 0,ilog2calcularía un nivel demasiado pequeño. XOR maneja naturalmente el escenario de wraparound.

Necesidad de la protección contra retroceso del tiempo 📎 tokio/src/runtime/time/mod.rs:301-309: Rust garantiza queInstantes monótono, pero el SO subyacente puede no garantizarlo. En una VM Linux sobre host Windows, std confía en el reloj de hardware provocando queInstantretroceda. Tokio usanow = lock.wheel.elapsed()para acotar, evitando que falle el assert deset_elapsed.

Despertar en lote y deadlock 📎 tokio/src/runtime/time/mod.rs:319: llamar a Waker mientras se tiene el lock de la rueda de tiempo es peligroso — el Waker puede disparar un re-poll de la tarea, que a su vez llama aSleep::reset, intentando adquirir de nuevo el lock de la rueda de tiempo, causando un deadlock.WakeListel mecanismo de lote libera temporalmente el lock cuando está lleno, es el patrón estándar de «callback fuera del lock».

next_wakeoptimización niche de 📎 tokio/src/runtime/time/mod.rs:130-136：Option<NonZeroU64>yu64tienen el mismo tamaño, porque 0 se usa como niche deNone. Pero tick 0 es un valor válido, así que el código usaNonZeroU64::new(t).unwrap_or_else(|| NonZeroU64::new(1).unwrap())para mapear 0 a 1📎 tokio/src/runtime/time/mod.rs:221. Este es un manejo de borde sutil: tick 0 se trata como tick 1, causando como máximo un despertar extra de 1ms.

process_expirationel «extraer primero, procesar después» de 📎 tokio/src/runtime/time/wheel/mod.rs:219-228: hay que extraer todas las entradas del slot antes de procesarlas, porque las entradas que superanMAX_DURATIONdan la vuelta y se reinsertan en el mismo slot. Si se extrae e inserta a la vez, se produce un bucle infinito.

Timeoutla trampa del orden de poll en 📎 tokio/src/time/timeout.rs:24-26: el future se poll primero, el timeout se verifica después. Si el future es intensivo en CPU y no cede, puede devolverOkincluso tras superar el timeout. En producción no dependas detimeoutpara forzar la interrupción de un future que no coopera.

---

Resumen del capítulo

Este capítulo desglosó la estructura de tres capas del driver de tiempo de Tokio:

1. Rueda de tiempo（Wheel): estructura hash jerárquica de seis niveles con 64 slots, usando el ancho de bits deelapsed ^ whenpara determinar el nivel de la entrada, con inserción y disparo aproximadamente O(1).pendingla lista enlazada almacena las entradas ya expiradas,process_expirationse encarga de descender nivel por nivel.

2. Driver（Driver::park_internal): traduce elnext_expiration_timede la rueda de tiempo a una duración depark_timeout, reutilizando el park/unpark de la pila de I/O.process_at_timetras el despertar avanza la rueda de tiempo, dispara Wakers en lote, y maneja el retroceso del tiempo y la protección contra deadlock.

3. API de usuario（Sleep / Timeout）：Sleepcreación perezosa deTimery registro,Timeoutprimero poll del value y luego poll del delay, usandowith_unconstrainedpara manejar el escenario de presupuesto agotado.

El diseño central es que «el tiempo también es un evento de I/O»: el driver tiene una sola entrada de park, que espera simultáneamente a que un fd esté listo y a que expire un temporizador.next_wakeregistra el instante de despertar prometido,reregisteral insertar un temporizador más tempranounparkdespierta al driver para recalcular.

En el próximo capítulo entraremos en las primitivas de sincronización:Mutex、Semaphorey cómo los canales implementan la espera asíncrona. Verás cómo reutilizan el mecanismo Waker de este capítulo, y cómo colaboran el «conteo de permisos» y la «cola de espera».

Reflexión y autoevaluación de este capítulo

es el caso normal, con presupuesto suficiente, se hace poll del delay directamente.

El capítulo anterior reveló cómo el tiempo se abstrae como un evento de I/O, haciendo que los temporizadores y la disponibilidad de fd compartan el mismo punto de entrada de espera park/unpark. Sin embargo, cuando múltiples tareas compiten por el mismo lock o se pasan mensajes a través de canales, el objeto de espera ya no es un fd o un reloj, sino el cambio de estado de otra tarea. Este capítulo entra en la familia tokio::sync, para averiguar dónde se almacena exactamente el Waker cuando un lock().await o recv().await se bloquea, y cómo se reprograma al ser despertado.

Por qué el Mutex asíncrono no puede reutilizar la implementación de std

Modelo intuitivo: de «ocupar el puesto» a «ceder el asiento»

std::sync::MutexEllock()debloquea el hilo actualcuando el lock está ocupado — el hilo es suspendido por el sistema operativo hasta que el lock se libera. Esto es catastrófico en un runtime asíncrono: un hilo worker puede estar impulsando cientos o miles de tareas simultáneamente; si se bloquea esperando un lock, todas las demás tareas que soporta se detienen. La exigencia central del Mutex asíncrono es: al esperar el lock,ceder el hilo, registrar el hecho de «estoy esperando este lock» en una cola, y luego devolverPending, dejando que el ejecutor ejecute otras tareas.

ElMutexde Tokio no implementa su propia cola de espera, sino queConstruido completamente sobre semáforos。

Estructura de datos y diseño de memoria

Mutex<T>Los campos de son minimalistas:

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

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

Los tres campos cumplen cada uno su función:ses unsemáforo con un contador de permisos de 1，cesUnsafeCell<T>el dato protegido envuelto por . Nótese que aquísemaphoreesbatch_semaphoreun alias de📎 tokio/src/sync/mutex.rs:3-3, es decir, la implementación subyacente, no lasync::Semaphorecapa de encapsulación pública.

MutexGuard<'a, T>solo contiene una referencia aMutex:

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

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

Aquí hay un diseño clave:MutexGuard no posee el objeto de permiso del semáforo, solo posee&Mutex. La acción de liberar el lock ocurre enDrop, llamando directamente aself.lock.s.release(1) 📎 tokio/src/sync/mutex.rs:959-961. Esto difiere deSemaphorePermitque mantiene unpermits: usizecontador y lo devuelve en Drop — el contador de permisos de Mutex es siempre 1, no necesita conteo.

Send/SyncLos límites de merecen un análisis aparte:

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

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

Syncsolo requiereT: Senden lugar deT: Sync— esto es razonable, porque el acceso mutuo garantiza que solo un hilo puede tocarTa la vez, transferir la propiedad deTentre hilos (Send) es suficiente, no se necesita queTen sí mismo sea compartible (Sync). Esta es precisamente la razón por la queMutex<T>puede convertir unSyncque no esTenSync.

Paso a paso: el viaje completo de unlock().await

Escenario: la tarea A llama amutex.lock().await, en este momento el lock está libre.

Primer paso,lock()construye un bloque async, internamente primeroself.acquire().await, tras el éxito construyeMutexGuard 📎 tokio/src/sync/mutex.rs:434-443。

Segundo paso,acquire()delega directamente al semáforo:

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

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

unwrap_or_else(|_| unreachable!())Esta línea de comentario revela la restricción de diseño: Mutex nunca cierra explícitamente el semáforo, y lo posee en exclusiva, por lo queacquirenunca devolveráErr. Esto excluye a nivel de tipos la ruta de error de «cierre del semáforo».

Tercer paso, si el lock está ocupado,s.acquire(1)devuelvePending, el Waker de la tarea actual se registra en la cola de espera del semáforo.¿Dónde se almacena el Waker?La respuesta está enbatch_semaphorela cola de espera de (el material fuente de este capítulo no expande ese archivo, pero su rol es: cada esperador mantiene un Waker, en cola FIFO).

Cuarto paso, cuando la tarea B que posee el lock lo libera,MutexGuard::dropllama as.release(1) 📎 tokio/src/sync/mutex.rs:965-975, el semáforo entrega el permiso al primero de la cola y despierta su Waker, la tarea A es re-programada,acquiredevuelveOk, construyendoMutexGuard。

Todo el flujo puede describirse con el siguiente diagrama de secuencia:

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

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

Reflexión de diseño: equidad FIFO y seguridad ante cancelación

La documentación declara explícitamente que el Mutex de Tokio garantiza FIFO📎 tokio/src/sync/mutex.rs:20-22. Esta equidad proviene de la semántica de cola del semáforo subyacente. El costo de la equidad es: unalockcancelada (por ejemplo, al perder enselect!) te haráperder tu posición en la cola 📎 tokio/src/sync/mutex.rs:415-419. Esto no es un bug, sino una consecuencia inevitable de la cola FIFO — cancelar significa salir de la cola, y para volver alockhay que reencolarse.

Otro diseño contraintuitivo esno envenenar（no poisoning）。std::sync::Mutexse marca como poisoned cuando el hilo que posee el lock entra en panic, y las siguienteslockdevuelvenErr. El Mutex de Tokio no hace esto: cuando el poseedor del lock entra en panic, el lock se libera normalmente📎 tokio/src/sync/mutex.rs:122-125. La documentación advierte que si el panic es capturado, los datos protegidos pueden quedar en un estado inconsistente. Esta es una concesión pragmática en escenarios asíncronos — un panic en una tarea asíncrona normalmente significa la terminación de la tarea, y el mecanismo de envenenamiento solo añadiría complejidad.

MutexGuard::mapLa serie de métodos merece mención. Permite degradar todo elMutexGuard<T>a unMappedMutexGuard<U>que solo protege un subcampo. En implementación, primero usa un closure para calcular el puntero al subcampodata, luego medianteskip_dropdescompone el guard original en unMutexGuardInnerque no dispara Drop, y finalmente construye un nuevo guard📎 tokio/src/sync/mutex.rs:869-883。skip_dropusandoManuallyDrop + ptr::readpara transferir la propiedad del campo, evitando queDropsea llamado dos veces📎 tokio/src/sync/mutex.rs:827-836. Esta es la técnica clásica en Rust de «transferir propiedad sin disparar el destructor».

Semaphore: cómo el conteo de permisos y la cola de espera implementan backpressure

Modelo intuitivo: las plazas de un estacionamiento

El semáforo es como un estacionamiento:acquirees entrar conduciendo, si hay plaza libre entras, si no haces cola en la entrada;releasees salir conduciendo, al liberar una plaza se notifica al primero de la cola para que entre. El número de permisos es el total de plazas,acquire_many(n)es un vehículo grande que ocupa n plazas.

Estructura de datos y diseño de memoria

El públicoSemaphorees solo una fina envoltura sobre elbatch_semaphore::Semaphoresubyacente:

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

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

SemaphorePermit<'a>mantiene la referencia al semáforo y el conteo de permisos:

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

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

permitsEl campo es la clave para entenderforget/merge/split.forgetponepermitsa cero📎 tokio/src/sync/semaphore.rs:1193-1195, así al hacer Drop devuelve 0 permisos — equivalente a «consumir permanentemente» esos permisos.splitCorta n permisos del conteo actual para el nuevo permit📎 tokio/src/sync/semaphore.rs:1260-1271。mergefusiona el conteo de otro permit, y afirma que ambos provienen del mismo semáforo📎 tokio/src/sync/semaphore.rs:1230-1240。

Paso a paso: el flujo de permisos en acquire y release

Escenario: el semáforo inicia con 2 permisos, la tarea Aacquire(), la tarea Bacquire_many(2)。

acquire()delega all_sem.acquire(1), tras el éxito construyeSemaphorePermit { permits: 1 } 📎 tokio/src/sync/semaphore.rs:614-631。acquire_many(2)similar, pero pasa 2📎 tokio/src/sync/semaphore.rs:661-679。

Si los permisos son insuficientes,ll_sem.acquire(n)devuelvePending, el Waker se encola. Aquí hay un detalle de equidad: la documentación señala que si el primero de la cola es unacquire_many(5)y solo quedan 3 permisos, aunque detrás haya unacquire(1)que podría satisfacerse de inmediato, también debe esperar — porque el vehículo grande al frente ocupa la cola📎 tokio/src/sync/semaphore.rs:19-24. Este es el costo del FIFO estricto, que evita la inanición.

La ruta de liberación está en Drop:

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

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

add_permitsdelega all_sem.release(n) 📎 tokio/src/sync/semaphore.rs:568-570, la capa subyacente devuelve los permisos a la cola de espera, despertando a los esperadores que pueden completar los permisos necesarios.

En cuanto al orden de memoria, la documentación da garantías fuertes: acquire, release, close son todas operacionesAcqRel, totalmente ordenadas entre sí, equivalentes a las de una única variable atómicaAcqRel 📎 tokio/src/sync/semaphore.rs:35-42. Esto significa que la escritura de «escribir datos primero y luego release del permiso» es visible para la tarea que «adquiere el permiso después»: el semáforo puede transferir datos de forma segura entre tareas.

Reflexión de diseño: close y contrapresión

close()Hace que todos los que esperan recibanAcquireError, y posteriormentetry_acquiredevuelveClosed 📎 tokio/src/sync/semaphore.rs:1161-1163. Esta es la base del cierre elegante: cuando el receptor ya no necesita datos, el close del semáforo permite que todos los emisores bloqueados fallen y retornen de inmediato, en lugar de esperar para siempre.

La esencia de la contrapresión se manifiesta con mayor claridad en mpsc. En la siguiente sección se verá que el control de capacidad de mpsc se implementa con un semáforo cuyo número de permisos es igual al tamaño del buffer.

La familia de canales: diferentes compensaciones entre cola de espera y activación por Waker

Modelo intuitivo: cuatro tipos de canales, cuatro estrategias de espera

oneshotes un «sobre de un solo uso»: solo puede enviar una carta, el emisor no espera (sendes síncrono), el receptorawaitespera la carta.mpsces una «cinta transportadora acotada»: el emisor espera cuando la cinta está llena, el receptor espera cuando está vacía, y la capacidad está controlada por un semáforo.broadcastywatchson un «altavoz de difusión»: un emisor, múltiples receptores, pero ambos manejan el «atraso» de forma completamente distinta.

El material fuente de esta sección se centra enoneshotympsc::bounded, los desglosamos uno por uno.

oneshot: un handshake minimalista codificado con bits de estado

oneshotLa estructuraInnerde es el núcleo para entender su diseño:

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

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

statees unAtomicUsize, que codifica todo el estado del canal con bits de bandera. Las cuatro banderas se definen al final del archivo:

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

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

valueesUnsafeCell<Option<T>>，tx_taskyrx_taskson de tipoTask, internamente esUnsafeCell<MaybeUninit<Waker>> 📎 tokio/src/sync/oneshot.rs:411-411. NóteseMaybeUninit—el Waker puede no estar inicializado, y si es válido lo determina el bitstatedentro deRX_TASK_SET/TX_TASK_SET📎 tokio/src/sync/oneshot.rs:396-399。

La esencia de este diseño：VALUE_SENTEl bit no solo indica «el valor ya fue enviado», sino que también determina a quién pertenece el acceso aUnsafeCell. El comentario lo deja muy claro📎 tokio/src/sync/oneshot.rs:1491-1496: siVALUE_SENTestá activado,UnsafeCellsolo puede ser accedido por el receptor; si no está activado, solo puede ser accedido por el emisor. Así se logra una transferencia de propiedad sin bloqueos usando un solo bit atómico, evitando bloqueos adicionales.

sendEl flujo de

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

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

Primero se escribe el valor enUnsafeCell(en este momentoVALUE_SENTno está activado, el receptor no accederá), luego se llama acomplete()para intentar activarVALUE_SENT。complete()es un bucle CAS:

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

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

¿Por qué usar CAS en lugar de un simplefetch_or? El comentario lo explica con claridad📎 tokio/src/sync/oneshot.rs:1517-1529: si el canal ya estáCLOSED, entoncesno se puedevolver a activarVALUE_SENT. Porque una vez activado, el receptor creerá que puede acceder aUnsafeCell, y en ese momento el emisor está preparándose para recuperar el valor (consume_value), y el acceso simultáneo de ambos lados provocaría una condición de carrera de datos. Por eso el bucle CAS hace break anticipado al detectarCLOSED, sin activar.

complete()Después de que retorna, si se activó con éxito yRX_TASK_SETya estaba activado, se despierta al receptor:

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

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

Elpoll_recvdel receptor es el núcleo de la máquina de estados:

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

Primero carga el estado; siis_complete()entonces directamenteconsume_valueretorna; siis_closed()devuelveErr; de lo contrario entra en la rama de «registrar Waker». Al registrar, primero verificais_rx_task_set(); si ya está configurado ywill_wakedetermina que es el mismo Waker, no lo configura de nuevo; si es diferente, primero hace unset y luego set. Aquí hay un manejo sutil de condición de carrera: después del unset, si se descubre queis_complete()se volvió verdadero, hay quevolver a setear la bandera 📎 tokio/src/sync/oneshot.rs:1342-1344, de lo contrario el Waker se filtrará en el Drop (porque el Drop depende de la bandera para decidir si debe hacer drop del Waker).

Este patrón de «unset y luego set de nuevo» también aparece enpoll_closed📎 tokio/src/sync/oneshot.rs:839-848, y es la técnica estándar de oneshot para manejar despertares concurrentes.

mpsc::bounded: contrapresión impulsada por semáforo

El control de capacidad de mpsc se delega por completo al semáforo.channelLa función crea un semáforo cuyo número de permisos es igual al buffer:

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

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

Semaphorees un envoltorio interno de mpsc, que mantiene simultáneamente el semáforo subyacente ybound(capacidad máxima)📎 tokio/src/sync/mpsc/bounded.rs:176-179。boundse usa paramax_capacityconsultas, mientras queavailable_permitsda la capacidad actual📎 tokio/src/sync/mpsc/bounded.rs:591-593。

La ruta de envíosendprimeroreservey luegosend：

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

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

reserveInternamente llama areserve_inner(1), que primero verifican > max_capacityy retorna error directamente, luegoacquire(n) 📎 tokio/src/sync/mpsc/bounded.rs:1272-1311. Aquí hay un sutilWakeReceiverOnDropguardián:

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

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

El comentario explica la motivación📎 tokio/src/sync/mpsc/bounded.rs:1279-1285: sireservese cancela después de obtener permisos parciales (por ejemplo,select!pierde), elAcquiresubyacente devolverá esos permisos en el Drop, perononotificará al receptor como lo hacePermit. Si en ese momento el canal ya está cerrado y ocioso, el receptor podría nunca recibir la notificación de «canal cerrado». Este guardián añade ese despertar en el Drop. En caso de éxito se usamem::forget(guard)para cancelar el guardián📎 tokio/src/sync/mpsc/bounded.rs:1306-1306, porque la ruta de éxito pasa aPermitpara asumir la responsabilidad de notificar.

PermitEl Drop de también hace lo mismo:

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

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

Permit::sendEn cambio, usamem::forgetpara omitir el Drop, evitando devolver permisos📎 tokio/src/sync/mpsc/bounded.rs:1721-1728。

La ruta de recepciónrecvusapoll_fnpara envolverchan.recv(cx) 📎 tokio/src/sync/mpsc/bounded.rs:243-246。poll_recvy delega directamente en📎 tokio/src/sync/mpsc/bounded.rs:650-652. La lógica real de la cola de espera está en el módulochan(no desarrollado en este capítulo), pero se puede inferir: el Waker del receptor se almacena enchan::Rx, y se despierta cuando el emisor hacesend.

try_sendmuestra la ruta no bloqueante:

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

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

try_acquireLos dos tipos de error de se mapean con precisión aClosedyFull, distinguiendo los dos tipos de fallo: «canal cerrado» y «búfer lleno».

Reflexión de diseño: seguridad ante cancelación y pérdida de mensajes

La documentación de mpsc enfatiza repetidamente la seguridad ante cancelación📎 tokio/src/sync/mpsc/bounded.rs:776-784：sendCuandoselect!pierde en,el mensaje se descartareserve. Para evitar la pérdida, hay que usarPermitpara obtenersendy luegoPermit—porquesendya reservó la capacidad,

recves síncrono y no puede ser interrumpido.📎 tokio/src/sync/mpsc/bounded.rs:199-204En cambio, es seguro ante cancelaciónrecv: siselect!pierde enrecv, se garantiza que ningún mensaje fue consumido. Esto se debe a que elpoll_recvdeReady，Pendingsolo retorna cuando realmente obtiene el mensaje

oneshoty no toca la cola.ReceiverEl📎 tokio/src/sync/oneshot.rs:246-251deoneshotcomo Future también es seguro ante cancelaciónsend. Pero hay que tener en cuenta:Errel

de

Trampa uno: usar un Mutex asíncrono para proteger datos puros.La documentación recomienda explícitamente📎 tokio/src/sync/mutex.rs:26-36: si lo protegido son datos puros (sin.awaitrequisitos), usarstd::sync::Mutexoparking_lotes más rápido. El costo de un Mutex asíncrono radica en las operaciones atómicas del semáforo y la posible programación de tareas. Solo cuando se necesita mantener el lock durante.await(por ejemplo, mantener el lock para acceder a una conexión de base de datos), se usa un Mutex asíncrono.

Trampa dos: mantener el lock a través de.awaitprovoca un deadlock.Esta es la trampa más peligrosa de un Mutex asíncrono. Si la tarea A, tras adquirir el lock,.awaitespera un evento que requiere que la tarea B se complete, y la tarea B a su vez espera ese mismo lock, se produce un deadlock.std::sync::MutexEl guard deSendno es.await(en tareas movibles), el compilador impide mantenerlo a través deSend 📎 tokio/src/sync/mutex.rs:314-314; pero el guard de un Mutex asíncrono es

, el compilador no te detiene, debes garantizar tú mismo que no se forme una espera circular.reserveTrampa tres:send。 Permitolvidar📎 tokio/src/sync/mpsc/bounded.rs:1732-1745El Drop de

devuelve el permisooneshot, por lo que no se filtra capacidad. Pero si el canal ya está cerrado y ocioso, el Drop despertará al receptor; este despertar es necesario, de lo contrario el receptor podría nunca recibir la notificación de cierre.pollTrampa cuatro:Pending。el📎 tokio/src/sync/oneshot.rs:236-242depollpuede ser falsoPendingLa documentación indica

: incluso si el mensaje ya fue enviado,forget_permitspuede devolver forget_permits(n). Esto no es un bug, sino un fenómeno normal bajo una condición de carrera concurrente; el llamador será despertado para reintentar, el mensaje no se pierde, solo se retrasa.📎 tokio/src/sync/semaphore.rs:576-578Trampa cinco:

la semántica de

.tokio::syncIntenta reducir n permisos y devuelve la cantidad realmente reducida. No bloquea ni despierta a los esperadores; simplemente "se traga" los permisos. Se usa para reducir dinámicamente la capacidad del semáforo.。

• MutexResumen del capítuloMutexGuardEste capítulo revelarelease(1)los patrones centrales de
• Semaphore: todas las primitivas de espera asíncrona se construyen sobre "cola de esperadores + despertar mediante Waker", y la implementación concreta de la cola varía según el escenarioSemaphorePermitreutiliza un semáforo con un número de permisos igual a 1,permitssolo mantiene una referencia, al hacer Dropforget/merge/split，MAX_PERMITS, FIFO justo pero sin envenenamiento.
• oneshotes un contador de permisos + cola de espera,AtomicUsizeusaVALUE_SENTel contador para soportarUnsafeCelldesplazamiento a la derecha de 3 bits para dejar espacio a los flags de estado.CLOSEDusa un único
• mpsc::boundedde flags de bits para codificar el estado,WakeReceiverOnDroplos bits determinan simultáneamente

a quién pertenece el derecho de acceso, el bucle CAS evita que se establezca después de

sería más complejo. Segundo, el objeto de permiso normalmente lleva un contadortokio::sync, para un Mutex este contador es siempre 1, lo cual es redundante. Tercero, el límitespawn_blockingdeblock_onya está controlado con precisión mediante

La ubicación del Waker varía según la primitiva: Mutex/Semaphore lo almacenan en la cola de espera del semáforo subyacente, oneshot en los campos tx_task/rx_task de Inner, mpsc en las colas de envío/recepción del módulo chan. Pero el mecanismo de activación es uniforme: al cambiar el estado se extrae el Waker y se llama a wake_by_ref, y el ejecutor reencola la tarea. Hasta aquí, la espera y la activación dentro de las primitivas asíncronas quedan claramente visibles. Sin embargo, no todo el código puede volverse asíncrono: el siguiente capítulo explorará cómo usar spawn_blocking para puentear operaciones bloqueantes, y cómo block_on impulsa un Future en un contexto no asíncrono.

En el capítulo anterior vimos que la clave por la que el Mutex asíncrono y los canales pueden esperar sin ocupar un hilo es almacenar el Waker en la cola de espera y, una vez satisfecha la condición, dejar que el activador reencole la tarea. Pero todo esto presupone que la tarea puede ceder el hilo voluntariamente cuando está en Pending. En cuanto el código llama a std::fs::read, a libsqlite3 o a un bucle de compresión puramente de CPU, acapara el hilo worker hasta retornar, y durante ese tiempo todas las demás tareas de ese hilo mueren de inanición. La solución de Tokio es externalizar ese tipo de trabajo a un grupo de hilos bloqueantes independiente y usar block_on para impulsar Futures en contextos no asíncronos. Este capítulo desglosa ambas fronteras.

8.1 Diseño de memoria del grupo de hilos bloqueantes: Inner y la cola de doble implementación

Modelo intuitivo：spawn_blockingEl grupo de hilos es como el «pool de ayudantes subcontratados» de un restaurante. Los camareros de sala (hilos worker) solo se encargan de tomar pedidos y servir platos; cuando aparece un plato que requiere cocción lenta, escriben una orden de trabajo y la arrojan a la ventanilla de paso a cocina (cola), y los ayudantes (hilos bloqueantes) toman la orden desde la ventanilla. Sin este pool, el camarero tendría que cocinar él mismo y todo el restaurante se detendría.

Estructura central. Todo el pool es sostenido porBlockingPool, que solo almacena dos cosas: unSpawnerclonable (punto de entrada de envío) y unshutdown_rx(extremo receptor de la señal de cierre)📎 tokio/src/runtime/blocking/pool.rs:20-23。Spawnerinternamente esArc<Inner>, todos los emisores comparten el mismo estado📎 tokio/src/runtime/blocking/pool.rs:26-28。

Inneres todo el estado del pool, y sus campos merecen revisarse uno por uno📎 tokio/src/runtime/blocking/pool.rs:77-104：

• inner_impl: InnerImpl: implementación de cola + notificación + topología de bloqueo, es un enum conLockedyShardeddos variantes📎 tokio/src/runtime/blocking/pool.rs:107-110. Esta es la abstracción más crítica del capítulo: unifica bajo una sola interfaz las dos topologías, «cola de un solo lock» y «cola fragmentada».
• thread_cap: usize: límite superior de hilos, es decirmax_blocking_threads。
• scheduler_threads: usize: número de hilos worker del planificador, usado para descontarlos en las métricas, de modo quenum_blocking_threadssolo cuente hilos bloqueantes📎 tokio/src/runtime/blocking/pool.rs:455-460。
• keep_alive: Duration: tiempo de vida de los hilos inactivos, por defectoKEEP_ALIVE = 10s 📎 tokio/src/runtime/blocking/pool.rs:231。
• metrics: SpawnerMetrics: tres contadores atómicos—num_threads、num_idle_threads、queue_depth 📎 tokio/src/runtime/blocking/pool.rs:31-35。

Estado de gestión de hilos。ThreadManagementStatese extrae por separado para que lo reutilicen las dos implementaciones de cola📎 tokio/src/runtime/blocking/pool.rs:135-150：

• shutdown: bool: bandera de cierre.
• shutdown_tx: Option<shutdown::Sender>: cada hilo worker mantiene una copia clonada; cuando todas se dropean,shutdown_rxrecibe la notificación.
• last_exiting_thread: Option<JoinHandle<()>>: handle del último hilo que salió por timeout.
• worker_threads: HashMap<usize, JoinHandle<()>>: handles de todos los workers vivos.
• worker_thread_index: usize: asignador de IDs de hilo monótonamente creciente.

last_exiting_threadLa motivación de diseño de📎 tokio/src/runtime/blocking/pool.rs:135-150。worker_timed_outestá claramente escrita en los comentarios: un hilo que sale por timeout hará join sobre el hilo que salió por timeout anteriormente, evitando falsos positivos de Valgrindlast_exiting_threades precisamente la implementación de ese join encadenado—elimina su propio handle y saca el antiguo📎 tokio/src/runtime/blocking/pool.rs:172-178。

para devolvérselo al llamador y que haga joinEncapsulación de tareasTask. En la cola se almacenaUnownedTask<BlockingSchedule>, que envuelve unMandatoryy una📎 tokio/src/runtime/blocking/pool.rs:187-191。Mandatorybanderashutdown_or_run_if_mandatorydecide si al cerrar esta tarea se descarta o se fuerza su ejecución:NonMandatoryenshutdown()se llama aMandatory, enrun() 📎 tokio/src/runtime/blocking/pool.rs:223-228se llama aspawn_blocking. Esta es la diferencia entrespawn_mandatory_blocking(no forzado) y📎 tokio/src/runtime/blocking/pool.rs:233-265。

(forzado, usado por fs)。LockedImplDiseño de memoria de la implementación de un solo lockMutex<LockedInner>es la topología más primitiva: unCondvar 📎 tokio/src/runtime/blocking/pool.rs:113-116。LockedInnermás unVecDeque<Task>、num_notify: u32dentro haythread_mgmt_state 📎 tokio/src/runtime/blocking/pool.rs:118-124ynum_notify. Nótese quethread_mgmt_stateynum_idle_threadsestán bajo el mismo lock, mientras que

es una cantidad atómica fuera del lock—este diseño híbrido con «parte del estado dentro del lock y parte fuera» es precisamente la raíz de todas las sutilezas de concurrencia posteriores.

8.2 Ruta de envío: de spawn_blocking al despertar del hiloEscenariotokio::task::spawn_blocking(move || heavy_compute(data)): se llama a

dentro de una tarea asíncrona; ¿qué ocurre en ese momento?。Spawner::spawn_blockingPrimer paso: decisión de boxeo y construcción de la tareafn_sizeprimero mide el tamaño del closureAutoBox::<F>::SHOULD_BOX, y luego, segúnBox, decide si boxear el closure📎 tokio/src/runtime/blocking/pool.rs:359-389. Esta es la estrategia genérica de Tokio de «boxeo automático de Futures grandes»: cuando el closure es demasiado grande se boxea, evitando que la estructura de la tarea se infle.

Se entra enspawn_blocking_inner, primero se asigna un ID de tarea, luego se usablocking_taskpara envolver el closure en un Future, y finalmente se usatask::unownedpara construirUnownedTaskyJoinHandle 📎 tokio/src/runtime/blocking/pool.rs:440-449. Nótese que aquí se devuelve la tupla(JoinHandle<R>, Result<(), SpawnError>)—el handle y el resultado del envío se devuelven por separado.

Segundo paso: los tres tratamientos del resultado del envío. De vuelta enspawn_blocking, se hace match sobrespawn_result:📎 tokio/src/runtime/blocking/pool.rs:381-388：

• Ok(()): normal, se devuelve el handle.
• Err(ShuttingDown)：No entra en pánico, y aún así devuelve el handle. El comentario indica que esto es por consideración de compatibilidad: el handle nunca se resolverá, pero quien lo llama no fallará porque el runtime se esté cerrando.
• Err(NoThreads(e)): el SO no puede crear el hilo y nadie en el pool lo asume, así que entra directamente en pánico.

Tercer paso: decisión de encolado y despertar。spawn_taskPasaon_no_idleel closure aInnerImpl::spawn_task, y la implementación concreta decide cuándo invocarlo📎 tokio/src/runtime/blocking/pool.rs:462-506. MiraLockedImpl::spawn_taskla sección crítica de📎 tokio/src/runtime/blocking/pool.rs:603-639：

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

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

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

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

Aquí hay dos puntos clave. Primero, la comprobación de cierre ocurre antes del encolado, e incluso si la tarea esMandatorytambién directamenteshutdown()—el comentario explica: se programó después de que comenzara el cierre, así que descartarla es legítimo📎 tokio/src/runtime/blocking/pool.rs:614-620. Segundo, la decisión de despertar depende denum_idle_threadsfuera del lock: si es 0, llama aon_no_idlepara intentar iniciar un nuevo hilo; de lo contrario, decrementa el contador de inactivos, incrementanum_notify、notify_one。

num_notify¿Por qué debe existir?PorqueCondvarpuede producir despertares espurios (spurious wakeup). Si solo se usaranotify_onesin contar, un hilo despertado espuriamente creería erróneamente que hay una tarea disponible, descubriría que la cola está vacía y volvería a dormirse, mientras que el hilo realmente despertado podría no recibir nunca la notificación.num_notifyConvierte el «despertar legítimo» en un token contable: el emisor+1, y el despertado solo ennum_notify != 0considera el despertar legítimo y-1 📎 tokio/src/runtime/blocking/pool.rs:674-684。

Cuarto paso: iniciar un nuevo hilo。on_no_idleEl closure se ejecuta mientras se mantiene el lock de la cola📎 tokio/src/runtime/blocking/pool.rs:462-506. Primero compruebanum_threads == thread_cap, y si alcanza el límite superior simplemente devuelveOk(())—la tarea permanece en la cola esperando que la procesen los hilos existentes; esto es contrapresión. De lo contrario, clonashutdown_tx, llama aspawn_threadpara crear el hilo y, tras el éxito, incrementanum_threads, incrementaworker_thread_index, inserta el handle enworker_threads。

spawn_threadUsathread::Builderpara establecer el nombre del hilo y el tamaño de pila, y luego hace spawn de un closure: entra en el contexto del runtimert.enter(), llama ainner.run(id), y finalmente dropshutdown_tx 📎 tokio/src/runtime/blocking/pool.rs:508-528。

Tolerancia a fallos al crear hilos del SO。spawn_threadpuede fallar. El código clasifica el error📎 tokio/src/runtime/blocking/pool.rs:488-500: si esWouldBlock(error temporal, determinado poris_temporary_os_thread_error) y ya hay hilos bloqueados en el pool, entonces📎 tokio/src/runtime/blocking/pool.rs:750-752se ignora silenciosamente—la tarea será tomada finalmente por algún hilo actualmente ocupado. De lo contrario, devuelve, lo que finalmente provoca un pánico.SpawnError::NoThreadsResume con un diagrama de flujo de control las ramas de decisión de la ruta de envío:

Copiar

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

Modelo intuitivo

: cada hilo bloqueado es un «ayudante en espera». Cuando hay pedidos trabaja continuamente (BUSY), cuando no hay pedidos dormita (IDLE), y si dormita más desale del trabajo (salida por timeout). Sin recuperación por timeout, el pool conservaría permanentemente todos los hilos creados en el pico, desperdiciando memoria y sobrecarga de planificación del kernel.keep_aliveEstructura del bucle principal

es un bucle。LockedImpl::run_worker, internamente alterna entre las dos fases BUSY e IDLE'main. Nota: aquí BUSY/IDLE son📎 tokio/src/runtime/blocking/pool.rs:642-735fasesdentro del bucle, no estados de un enum explícito, así que a continuación se describe con un diagrama de flujo en lugar de un diagrama de estados.Fase BUSY

: el bucle internotoma tareas continuamentewhile let Some(task) = locked.queue.pop_front(). Tras tomarla, decrementa📎 tokio/src/runtime/blocking/pool.rs:655-661drop del lockqueue_depth，, ejecuta, y vuelve a adquirir el lock. El paso de soltar el lock es crucial—una tarea bloqueante puede tardar mucho, y nunca se debe ejecutar manteniendo el lock.task.run()Fase IDLE

: la cola está vacía, incrementa, establecenum_idle_threads, y luego entra en el bucle de esperais_counted_idle = true. El núcleo es📎 tokio/src/runtime/blocking/pool.rs:663-696, y tras retornar comprueba tres cosas:condvar.wait_timeout(locked, keep_alive): despertar legítimo. Decrementa

1. num_notify != 0, establecenum_notify(porque el emisor ya decrementóis_counted_idle = false), break de vuelta a BUSYnum_idle_threads2. No cerrado y timeout: llama a📎 tokio/src/runtime/blocking/pool.rs:674-684。

para obtener el handle del último hilo que salió,worker_timed_outsale del buclebreak 'main3. De lo contrario es un despertar espurio, sigue esperando.📎 tokio/src/runtime/blocking/pool.rs:689-693。

Vaciado de la cola al cerrar

. Sies verdadero, entra en la lógica de vaciadothread_mgmt_state.shutdown: extrae tareas una a una, drop del lock, llama a📎 tokio/src/runtime/blocking/pool.rs:698-710—las tareas no forzadas se descartan, las forzadas se ejecutan como de costumbre. Luego break para salir del bucle principal.task.shutdown_or_run_if_mandatory()Limpieza al salir

. Antes de que el hilo salga, decrementa. Sinum_threads 📎 tokio/src/runtime/blocking/pool.rs:714es verdadero, también decrementais_counted_idle, y usanum_idle_threadspara afirmar que no hay underflowassert_ne!(prev_idle, 0). Esta aserción es una barandilla en depuración: una vez que📎 tokio/src/runtime/blocking/pool.rs:716-726la contabilidad falla, aquí entrará inmediatamente en pánico en lugar de dejar que el error se propague silenciosamente.num_idle_threadsFinalmente, si se está cerrando y

(el último hilo),num_threads == 0despierta al iniciador del cierre que podría estar esperandonotify_one. Devuelve📎 tokio/src/runtime/blocking/pool.rs:728-730, yjoin_on_threadhace join antes de salirInner::runHandshake de cierre📎 tokio/src/runtime/blocking/pool.rs:755-771。

primero llama a。BlockingPool::shutdownpara obtener todos los handles de workerbegin_shutdown, establece la bandera de cierre, drop📎 tokio/src/runtime/blocking/pool.rs:310-312。LockedImpl::begin_shutdown, despierta a todos los hilos en esperashutdown_tx、notify_all. Luego📎 tokio/src/runtime/blocking/pool.rs:740-745bloquea esperandoshutdown_rx.wait(timeout). La implementación de📎 tokio/src/runtime/blocking/pool.rs:324。

shutdown::Receiver::waites muy cuidadosa📎 tokio/src/runtime/blocking/shutdown.rs:37-70: primero manejatimeout == 0la ruta rápida devuelve directamente false; luego llama atry_enter_blocking_region()para entrar en la región de bloqueo, y si falla y actualmente se está en pánico devuelve false, de lo contrario entra en pánico con el mensaje «no se puede hacer drop del runtime en un contexto asíncrono»📎 tokio/src/runtime/blocking/shutdown.rs:44-57. Finalmente, según el timeout, llama ablock_on_timeoutoblock_onpara impulsar ese oneshot.

shutdown_txEl mecanismo deArc<oneshot::Sender<()>>es: cada hilo worker mantiene un clon de📎 tokio/src/runtime/blocking/shutdown.rs:12-14. Después de que todos los hilos salen, todos los clones se dropean,Arcel contador llega a cero,oneshot::Senderse dropea,Receiverrecibe la notificación. Este es el patrón clásico de «el Receiver se despierta después de que todos los Sender hacen drop».

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

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

8.4 block_on: impulsar un Future en un contexto no asíncrono

Modelo intuitivo：block_ones la «puerta principal» del runtime. Convierte el hilo actual en un ejecutor temporal, haciendo poll repetidamente sobre el Future pasado hasta que se completa. Sin él,mainla función no podría iniciar ningún código asíncrono.

Entrada y boxing。Runtime::block_onigualmente primero mide el tamaño, segúnSHOULD_BOXdecide siBox::pin, y luego entra enblock_on_inner 📎 tokio/src/runtime/runtime.rs:343-350。block_on_inner. Dentro hay dos envoltorios de trace con compilación condicional (taskdump y tracing), luegoself.enter()entra en el contexto del runtime, y finalmente despacha según el tipo de scheduler📎 tokio/src/runtime/runtime.rs:353-383：

let _enter = self.enter();

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

Los dos schedulersblock_onLa semántica es diferente, la documentación lo dice claramente📎 tokio/src/runtime/runtime.rs:302-320：

• Programador de múltiples hilos: Future se ejecuta en el contexto del controlador de E/S y del temporizador,block_onlas tareas ya lanzadas con spawn continúan ejecutándose tras el retorno.
• Programador de hilo actual：block_onpuede ser invocado concurrentemente por múltiples hilos, el primer invocador obtiene la propiedad del controlador de E/S y del temporizador, los demás hilos se "enganchan" a él. El primerblock_ontras completarse, los demás hilos pueden "robar" el controlador.block_onlas tareas ya lanzadas con spawn quedan suspendidas tras el retorno, una nueva invocación deblock_onlas reanudará.

Restricción clave: no se puede invocar en un contexto asíncrono. La documentación especifica claramente queblock_oninvocarlo en un contexto de ejecución asíncrono provocará un panic📎 tokio/src/runtime/runtime.rs:321-324. La razón es directa:block_onbloquea el hilo actual hasta que el Future se complete; si el hilo actual es en sí mismo un hilo worker, bloqueará todo el ejecutor — esto es precisamentespawn_blockinglo que se pretende resolver, por lo que ambos son mutuamente excluyentes.

Ruta de cierre。Runtime::dropse despacha según el tipo de programador📎 tokio/src/runtime/runtime.rs:506-521: el programador de hilo actual necesita primerotry_set_currententrar en el contexto y luego shutdown (garantiza que las tareas se destruyan dentro del contexto de ejecución); el programador de múltiples hilos hace shutdown directamente (los hilos worker ya están en el contexto).shutdown_timeoutPrimero cerrar el programador y luego cerrar el pool de bloqueo📎 tokio/src/runtime/runtime.rs:457-461，shutdown_backgroundes equivalente ashutdown_timeout(Duration::from_nanos(0)) 📎 tokio/src/runtime/runtime.rs:494-496。

Reflexiones de diseño, recuperación de errores y trampas en producción

¿Por quéspawn_blockingdeShuttingDownno entra en panic? 📎 tokio/src/runtime/blocking/pool.rs:383-384El comentario indica que es por consideraciones de compatibilidad.spawn_blockingdevuelveJoinHandleen lugar deResult, si entrara en panic al cerrarse, convertiría el estado predecible de "el runtime se está cerrando" en un crash. Devolver un handle que nunca se resuelve hace que el invocadorawaitquede suspendido indefinidamente — pero en ese momento el runtime ya está cerrado, todo elblock_ontambién saldrá, por lo que en la práctica no habrá fuga permanente.

max_blocking_threadsLa semántica de contrapresión de. El valor por defecto es muy grande (512), porquespawn_blockingse usa frecuentemente para E/S de archivos. Pero la documentación advierte: al ejecutar tareas intensivas en CPU hay que usar un semáforo para limitar la concurrencia, de lo contrario se crearán una gran cantidad de hilos📎 tokio/src/task/blocking.rs:94-100. Al alcanzar el límite, las tareas se encolan, formando contrapresión — pero hay que notar que esta contrapresión solo afecta al pool de bloqueo, no se propaga al programador asíncrono.

spawn_blockingno es cancelable. La documentación especifica claramente:abortno tiene efecto sobre tareas de bloqueo que ya han comenzado a ejecutarse, la tarea continuará hasta completarse📎 tokio/src/task/blocking.rs:106-120. Solo las tareas que aún no han comenzado pueden ser detenidas por abort. Al cerrarse, el runtime esperará a todas las tareas de bloqueo ya iniciadas,shutdown_timeouttras el timeout se filtrarán estos hilos.

num_idle_threadsLa trampa de contabilidad de。is_counted_idleLa existencia del flag indica que este conteo es propenso a errores. El emisor decrementa al despertarnum_idle_threads, el receptor al vernum_notify != 0estableceis_counted_idle = false, evitando el decremento duplicado📎 tokio/src/runtime/blocking/pool.rs:679-682. Si esta ruta tiene un bug,assert_ne!(prev_idle, 0)entrará en panic al salir📎 tokio/src/runtime/blocking/pool.rs:722-725. Si en producción se observa "num_idle_threadsunderflowed on thread exit", significa que la lógica de contabilidad del pool está corrompida.

InnerImplEl significado de la abstracción por enumeración. El comentario indica queLockedla variante se comporta exactamente igual que antes de la refactorización, mientras queShardedla variante reserva un slot simétrico para futuras colas concurrentes📎 tokio/src/runtime/blocking/pool.rs:537-539。spawn_task、run_worker、begin_shutdownlos tres métodos se despachan mediante la enumeración📎 tokio/src/runtime/blocking/pool.rs:548-582. Este diseño de "despacho por enumeración + sección crítica propia por variante" permite añadir nuevas topologías de cola sin modificar el invocador.

Resumen del capítulo

Este capítulo desglosa las dos fronteras de Tokio para acomodar código síncrono.spawn_blockingentrega closures a un pool de hilos de bloqueo independiente:Innermantiene la cola, el límite de hilos, el tiempo de vida y las métricas atómicas;LockedImplimplementa la cola con un solo lock +Condvar,num_notifyel contador compensa los despertares espurios; el worker cicla entre BUSY/IDLE, tras el timeout de inactividad sale mediante join encadenado;max_blocking_threadsal alcanzar el límite las tareas se encolan formando contrapresión.block_onen cambio impulsa Future en contextos no asíncronos, la semántica del programador de múltiples hilos y del de hilo actual es diferente, y está terminantemente prohibido invocarlo en contextos asíncronos. La ruta de cierre medianteshutdown_txdeArcel conteo llegando a cero disparaoneshot, implementando el handshake de "despertar al iniciador del cierre tras la salida de todos los workers".

Reflexiones y autoevaluación del capítulo

El pool de hilos bloqueantes y block_on delimitan las fronteras de capacidad del runtime asíncrono: el primero aísla en hilos dedicados el trabajo que no puede ceder el hilo, y el segundo permite que puntos de entrada no asíncronos también impulsen Futures. Pero estas dos fronteras a menudo no están escritas a mano en el código; en el próximo capítulo entraremos en el mundo de las macros para ver cómo #[tokio::main], select! y join! generan este código de runtime en tiempo de compilación.

En el capítulo anterior vimosblock_ony cómo el pool de hilos bloqueantes delimita las fronteras de capacidad del runtime asíncrono, mientras que los usuarios casi nunca escriben a mano estas fronteras: escriben#[tokio::main]、select!、join!, dejando que la macro despliegue este código repetitivo en tiempo de compilación. Las macros son la primera capa de azúcar que Tokio ofrece al usuario y también el lugar donde realmente se genera el código de runtime en tiempo de compilación. Este capítulo se centra entokio-macroscrate ytokio/src/macros/select.rs, desglosando las tres rutas de expansión de macros más utilizadas, y responde con especial atención a una pregunta: tras la expansión de la macro, ¿cómo es realmente la cadena de llamadas, y por qué la semántica de cancelación segura deselect!debe vigilarse por separado?

9.1 #[tokio::main]: reescribir async fn como Runtime::block_on

Modelo intuitivo：#[tokio::main]es como una «carta de encargo de reformas». Entregas una habitación en bruto (async fn main), y ella te instala la fontanería y la electricidad (construye el Runtime), coloca puertas y ventanas (enable_all), y finalmente mete dentro tus muebles originales (el cuerpo de la función). Sin ella, cadamaintendría que escribir a manoBuilder::new_multi_thread().enable_all().build().unwrap().block_on(...), y el código repetitivo ahogaría la lógica de negocio.

Estructuras de datos y diseño de memoria

La macro en sí no produce estructuras de datos en tiempo de ejecución, pero la configuración que analiza se coloca en dos structs.Configurationes un «acumulador mutable en fase de análisis», todos sus campos sonOption, porque los parámetros del atributo pueden faltar, repetirse o ser ilegales📎 tokio-macros/src/entry.rs:74-84. Nótese queworker_threads、start_paused、unhandled_panicllevanSpan—esto es para localizar el error en la línea que escribió el usuario al reportar un error, y no dentro de la macro📎 tokio-macros/src/entry.rs:74-84。FinalConfigen cambio es el «resultado inmutable tras la validación»,flavorya no esOption, porquebuild()ya ha usadodefault_flavorcomo respaldo📎 tokio-macros/src/entry.rs:55-62。

RuntimeFlavorsolo tiene tres variantes:CurrentThread、Threaded、Local 📎 tokio-macros/src/entry.rs:10-14。from_strincluye deliberadamente mensajes amigables para nombres heredados:single_threadindica que debería llamarsecurrent_thread，basic_schedulerindica que ha cambiado de nombre,threaded_schedulerindica que ha cambiado de nombre📎 tokio-macros/src/entry.rs:17-27. Este es un diseño típico de la macro como «primer punto de contacto del usuario»: el mensaje de error es documentación.

Proceso de expansión paso a paso

Escenario: el usuario escribe#[tokio::main(flavor = "multi_thread", worker_threads = 4)] async fn main() { ... }。

Primer paso,mainla entrada primero analiza el item como unItemFn 📎 tokio-macros/src/entry.rs:577-580personalizado. EsteItemFnno essyn::ItemFn, sino un analizador implementado por el propio Tokio, cuya razón está escrita en los comentarios: no quiere analizar recursivamente toda la sentencia, solo hace un análisis ligero de «almacenar en búfer por token tree y dividir al encontrar punto y coma»📎 tokio-macros/src/entry.rs:720-764. Esto evita la sobrecarga de construir un AST completo del cuerpo de la función dentro de la macro.

Segundo paso,build_configvalidaasyncsi la palabra clave existe; si falta, reporta "theasync keyword is missing" 📎 tokio-macros/src/entry.rs:346-349. Luego recorre los parámetros del atributo, enviandoworker_threads、flavor、start_paused、crate、unhandled_panic、nameal setter correspondiente📎 tokio-macros/src/entry.rs:369-399. Nótese quecore_threadsse rechaza explícitamente y se indica que ha cambiado de nombre📎 tokio-macros/src/entry.rs:379-382。

Tercer paso,Configuration::buildrealiza la validación de consistencia entre campos. Aquí hay tres restricciones clave:worker_threadssolo permitemulti_thread 📎 tokio-macros/src/entry.rs:197-217；start_pausedsolo permitecurrent_thread/local 📎 tokio-macros/src/entry.rs:219-229；unhandled_panicigualmente solo permitecurrent_thread/local 📎 tokio-macros/src/entry.rs:231-241. Si el usuario eligiómulti_threadperort-multi-threadla feature no está activada, el mensaje de error será diferente según si se especificó explícitamente el flavor📎 tokio-macros/src/entry.rs:209-216。

Cuarto paso,parse_knobsgenera el código. Primero eliminaasyncness 📎 tokio-macros/src/entry.rs:441, y luego elige el punto de partida del builder según el flavor:CurrentThread/LocalusaBuilder::new_current_thread()，ThreadedusaBuilder::new_multi_thread() 📎 tokio-macros/src/entry.rs:468-477。Locallo especial es que la llamada a build esbuild_local(Default::default())en lugar debuild() 📎 tokio-macros/src/entry.rs:479-483. Después añade encadenadamente según sea necesario.worker_threads(#v)、.start_paused(#v)、.unhandled_panic(...)、.name(#v) 📎 tokio-macros/src/entry.rs:485-497。

Quinto paso, genera el cuerpo final de la función. El núcleo eslast_block：return #rt.enable_all().#build.expect("Failed building the Runtime").block_on(body) 📎 tokio-macros/src/entry.rs:509-522. Nótese esereturnexplícito, cuyo comentario apunta a tokio-rs/tokio#4636, para corregir un problema de inferencia de tipos📎 tokio-macros/src/entry.rs:508。

Sexto paso, el cuerpo de la función se envuelve enasync #bodyy se somete a comprobación de tipos. En la ruta que no es test, si el tipo de retorno no es!y no contieneimpl Trait, se insertaif false { let _: &dyn Future<Output = #output_type> = &body; }para hacer una aserción en tiempo de compilación📎 tokio-macros/src/entry.rs:551-571. La ruta test en cambio usapin!fijar body en la pila y convertirlo aPin<&mut dyn Future>, el comentario explica que esto es para reducirblock_onla sobrecarga de compilación de la instanciación genérica📎 tokio-macros/src/entry.rs:526-548。

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

Reflexiones de diseño y trampas en producción

mainytestcompartenparse_knobs, pero el flavor predeterminado es diferente:testpor defectoCurrentThread，mainpor defectoThreaded 📎 tokio-macros/src/entry.rs:91-94. Esto explica por qué#[tokio::test]es de un solo hilo por defecto — las pruebas normalmente no necesitan múltiples núcleos, y el modo de un solo hilo es más fácil de reproducir.

Una trampa fácil de pasar por alto: después de la expansión del macro, cada llamada a la función crea un nuevo Runtime. La documentación advierte explícitamente que si la función se llama con frecuencia, se debería usar Builder para reutilizar el Runtime📎 tokio-macros/src/lib.rs:31-35. Usar#[tokio::main]en una función normal es legal, pero cada llamada paga el costo de construir un Runtime.

Otra trampa escrateel renombrado. Cuando el usuariouse tokio as tokio1, eltokio::runtime::Buildergenerado por defecto dentro del macro no encontrará la ruta, y se debe especificar explícitamentecrate = "tokio1" 📎 tokio-macros/src/lib.rs:239-264。parse_knobsencrate_pathel valor predeterminado deIdent::new("tokio", ...) 📎 tokio-macros/src/entry.rs:456-462es

, que es precisamente la raíz del error en escenarios de renombrado.

9.2 select!: sondeo multi-rama, máscara de bits y equidad aleatoria：select!Modelo intuitivopoll_fnes como «un mesero que vigila múltiples ventanillas de recogida al mismo tiempo». La ventanilla que sirva primero, de esa se lleva la comida, y la cola de las demás ventanillas se descarta. Sin él, el usuario tendría que escribir manualmente

para meter múltiples Future en una tupla y hacer poll uno por uno, además de manejar por su cuenta la lógica de «una vez que una rama está lista, las demás ramas deben descartarse».

select!Estructura de datos y diseño de memoria__tokio_select_utiltras expandirse genera un módulo localOut, dentro del cual hay un enumMask 📎 tokio/src/macros/select.rs:615-619。Outy un alias de tipo_0、_1el nombre de la variante esDisabled…… una por cada rama, más un📎 tokio-macros/src/select.rs:33-39。Maskque representa que todas las ramas han falladou8el tipo subyacente se elige dinámicamente según el número de ramas: ≤8 usau16, ≤16 usau32, ≤32 usau64, ≤64 usa📎 tokio-macros/src/select.rs:17-31, y si supera 64 hace panic directamenteselect!. Esta máscara de bits es

el estado central: el bit i en 1 indica que la i-ésima rama ha sido deshabilitada.futuresTodos los Future se almacenan en una tuplaIntoFuture::into_future, cada elemento primero pasa por📎 tokio/src/macros/select.rs:654-656la conversiónfutures_init. Nótese que aquí primero se construyeinto_futurey luego se aplica uno por uno📎 tokio/src/macros/select.rs:641-646, el comentario explica que esto es para aprovechar la extensión del tiempo de vida temporallet mut futures = &mut futures;. Posteriormentepoll_fndegrada la tupla a una referencia mutable, evitando que📎 tokio/src/macros/select.rs:658-662。

el closure se apropie de la propiedad

Flujo de sondeo paso a pasoselect! { v = stream1.next() => ..., v = stream2.next() => ..., else => break }。

Contextualizando el escenario:biased;Primer paso, coincidencia de reglas de entrada del macro. Si haystart=0 📎 tokio/src/macros/select.rs:801-803prefijo,start; de lo contrariothread_rng_n(BRANCHES) 📎 tokio/src/macros/select.rs:805-809es una expresión aleatoria📎 tokio/src/macros/select.rs:61-65。

. Esta es la fuente de equidad que la documentación describe como «seleccionar aleatoriamente una rama para verificar primero»(skip) pat = fut, if cond => handler,Segundo paso, normalización. tt-muncher normaliza cada rama a la formaskip,_es una secuencia de📎 tokio/src/macros/select.rs:770-793。skip, cuya longitud es igual al número de branches anteriores a esa ramafutures_init.$($skip)*se usa tanto para generar el acceso a los campos de la tuplacount!, como para

calcular el índice de la rama.if $cTercer paso, evaluación de precondiciones. Para cada ramadisabled |= 1 << index 📎 tokio/src/macros/select.rs:631-636, si es false, entonces$fut. Nota: incluso si la rama está deshabilitada, su📎 tokio/src/macros/select.rs:39-41。

expresión aún se evalúa, solo que no se hace pollpoll_fnCuarto paso, entrar enready!(poll_budget_available(cx))el closure. Primero se verifica el presupuesto de cooperación:Pending 📎 tokio/src/macros/select.rs:664-667, si el presupuesto se agota, se retorna directamenteselect!. Esto garantiza que

no acapare el worker.for i in 0..BRANCHES，branch = (start + i) % BRANCHES 📎 tokio/src/macros/select.rs:680-685Quinto paso, bucledisabled & mask == mask. Para cada branch: primero se consultacontinue 📎 tokio/src/macros/select.rs:694-699, si ya está deshabilitada entoncesPin::new_unchecked; de lo contrario se extrae ese Future de la tupla, se envuelve con📎 tokio/src/macros/select.rs:701-707(la seguridad depende de que el Future esté en la pila y no se mueva)Ready(out); se hace poll,disabled |= maskentonces primero📎 tokio/src/macros/select.rs:710-730。

y luego se hace coincidir el patrónoutSexto paso, coincidencia de patrones. Si$bindcoincide conPoll::Ready(Out::_i(out)) 📎 tokio/src/macros/select.rs:727-733, retornacontinue; si no coincide,📎 tokio/src/macros/select.rs:44-47。

continúa sondeando otras ramas — esto es precisamente lo que el paso 5 de la documentación describe como «si el patrón no coincide, deshabilitar la rama actual»is_pendingSéptimo paso, fin del bucle. SiPendinges true, retornaOut::Disabled 📎 tokio/src/macros/select.rs:740-745, de lo contrario todas las ramas han fallado, retornamatch output. ElOut::_iexterno mapeaDisabledal handler correspondiente,elsese mapea a📎 tokio/src/macros/select.rs:749-755。

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

Copiar

Reflexiones de diseño y trampas en producciónVec<bool>？¿Por qué usar una máscara de bits en lugar dedisabled |= mask? La máscara de bits es un único entero en la pila, sin asignación en el heap, yselect!es una sola instrucción. Para

en la ruta caliente, esto evita el acceso al heap en cada iteración.¿Por qué deshabilitar la rama cuando el patrón no coincide?select!Esta esSome(v) = stream.next() => ...la diferencia clave con una «race simple». Considéresestream.next(), siNoneretorna📎 tokio/src/macros/select.rs:198-223。

(fin del stream), el patrón no coincide, esa rama se deshabilita permanentemente, evitando sondear infinitamente un stream que ya terminó. El ejemplo de la documentación se basa precisamente en esta semántica para recolectar dos streams hasta que ambos terminen：select!El verdadero significado de cancel safetyread_exact、read_to_end、write_allUna vez que una rama está lista, los Future de las demás ramas se dropean. Si el Future dropeado ya consumió datos pero aún no ha retornado, los datos se pierden. La documentación enumera explícitamente📎 tokio/src/macros/select.rs:119-124no es cancel safeMutex::lock、Semaphore::acquire, mientras que📎 tokio/src/macros/select.rs:126-133debido a la equidad de la cola, la cancelación perderá la posición en la cola.await. Método de determinación: buscar.awaitel punto, si reiniciar la función en📎 tokio/src/macros/select.rs:135-139。

ifsigue siendo correcto, entonces es cancel safeLa trampa de carrera en las precondicionesif !sleep.is_elapsed(): la documentación da un ejemplo clásico de error — usarsleeppara protegeris_elapsed()la rama, perowhilepuede volverse true entre la verificación deselect!y📎 tokio/src/macros/select.rs:336-376, provocando que el timeout se pierdaif. La forma correcta es eliminarsleep, dejar quebreak 📎 tokio/src/macros/select.rs:378-405。

biased;la rama siempre participe en el sondeo, y después del timeoutel costo de📎 tokio/src/macros/select.rs:67-74: el RNG aleatorio tiene costo de CPU, y algunos escenarios requieren un orden de sondeo deterministabiased;. Pero📎 tokio/src/macros/select.rs:75-81。

deja la responsabilidad de la equidad al usuario: si una rama siempre está lista, las ramas posteriores se morirán de inanición

9.3 join! y las restricciones de ingeniería de la expansión de macros：join!Modelo intuitivoselect!es como «esperar al mismo tiempo a que lleguen todos los paquetes». No comoReadydonde quien llega primero cancela a los demás, sino que agrega los valorespoll_fnde todos los Future en una tupla. Sin él, el usuario tendría que escribir manualmente

para mantener el estado de finalización de cada Future.

join!La expansión de también se basa en almacenar Future en tuplas, pero el estado no es una máscara de bits, sino una tupla de «valores completados». Cuando cada Future se completa, su valor se extrae y se almacena en la tupla de resultados, y la ranura correspondiente se marca como completada. A diferencia deselect!,join!no descarta los Future no completados — debe esperar a que todos los Future se completen antes de retornar.

Flujo paso a paso

join!La lógica de sondeo de comparte el esqueleto de «tupla almacena Future +select!impulsado» conpoll_fn, pero con semántica opuesta:select!es «retorna cuando cualquiera esté listo»,join!es «retorna solo cuando todos estén listos». Cada ronda de poll recorre todos los Future no completados, si cualquiera retornaPendingentonces el conjuntoPending, si todosReadyentonces agrega y retorna.

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

Reflexiones de diseño y trampas en producción

join!La semántica de seguridad ante cancelación de es diferente deselect!:join!cuando se descarta, todos los Future no completados también se descartan, igualmente puede perderse datos. Pero comojoin!no cancela activamente ninguna rama, no hace comoselect!que «cancela esta rama porque otra rama está lista». El riesgo real está en quejoin!en su conjunto sea cancelado por unselect!externo o por timeout.

join!La diferencia entre ytry_join!merece atención:try_join!retorna inmediatamente cuando cualquier Future retornaErr, cancelando los demás Future, por lo que hereda el riesgo de seguridad ante cancelación deselect!.

Reflexiones de diseño

El macro como frontera de un generador de código en tiempo de compilación。#[tokio::main]coloca la validación de configuración en tiempo de compilación, combinaciones ilegales (comomulti_thread + start_paused) fallan directamente en compilación, en lugar de panic en tiempo de ejecución. Esta es la ventaja central del macro frente al Builder: errores anticipados.

Arquitectura híbrida de macro declarativo + macro procedural。select!El cuerpo principal de esmacro_rules!, pero dos lógicas clave se delegan a macros procedurales:select_priv_declare_output_enumgenera el enumOuty el tipoMasklimpia el📎 tokio-macros/src/lib.rs:658-660，select_priv_clean_patternen el patrón. ¿Por qué? Los comentarios explican: los macros declarativos difícilmente pueden generar código que «seleccione dinámicamente el tipo entero según el número de ramas», ni pueden hacer limpieza a nivel de token en posiciones de patrónref/mut 📎 tokio-macros/src/lib.rs:666-668La necesidad de📎 tokio/src/macros/select.rs:577-579。

clean_patternhace coincidir。select!conouten forma de&outcon el patrón📎 tokio/src/macros/select.rs:727, si el usuario escriberef v, se convierte en&ref vcausando error de tipo.clean_patternelimina recursivamenteby_ref、mutability, así como elReferencedel patrónmutability 📎 tokio-macros/src/select.rs:68-73📎 tokio-macros/src/select.rs:100-103. Este es el compromiso que hace el macro entre la «intuición del usuario» y el «borrow checker».

La realidad ingenieril del límite de 64 ramas。count!、count_field!、select_variant!Los tres macros escriben a mano cada uno las reglas de coincidencia de 0 a 64📎 tokio/src/macros/select.rs:821-1017📎 tokio/src/macros/select.rs:1021-1217📎 tokio/src/macros/select.rs:1221-1414. El comentario dice directamente «I'm not happy about it either»📎 tokio/src/macros/select.rs:816-817. Este es el precio de que los macros declarativos no puedan hacer aritmética: solo se puede mapear a enteros codificando la cantidad de tokens.

Resumen del capítulo

Reflexiones y autoevaluación del capítulo

La macro se encarga del código repetitivo y la validación en tiempo de compilación por parte del usuario, pero lo que genera sigue siendo un Future normal y una llamada apoll. En el próximo capítulo abandonaremos el mundo de compilación de las macros para entrar en la capa de abstracción de E/S en tiempo de ejecución, y veremos cómoAsyncRead/AsyncWritedivide el flujo de bytes en frames, y cómo el marco de códec deFramedfunciona correctamente bajo las restricciones de cancelación segura deselect!.

#[tokio::main]La esencia deblock_ones «análisis de configuración + generación de cadena Builder +select!envoltura», la validación de configuración se completa en tiempo de compilación, y el flavor determina el punto de inicio del builder y el método build..awaitEl núcleo dejoin!es «almacenar Future en tuplas + registrar deshabilitaciones con máscara de bits + punto de inicio aleatorio para garantizar la equidad», si el patrón no coincide se deshabilita la rama, y la seguridad ante cancelación depende de si el Future descartado puede reiniciarse enselect!.AsyncRead/AsyncWriteyBufReader/BufWritercomparten el esqueleto pero tienen semántica opuesta: el primero espera a que todo se complete, el segundo devuelve en cuanto cualquiera esté listo. Los tres juntos muestran la compensación central del diseño de macros de Tokio: delegar el código repetitivo y la validación en tiempo de compilación a las macros, y dejar la complejidad de la semántica en tiempo de ejecución (especialmente la seguridad ante cancelación) para que el usuario la entienda explícitamente. Después de entender cómo las macros generan código en tiempo de ejecución, la siguiente pregunta natural es: cuando este código realmente empieza a leer y escribir flujos de bytes, ¿qué abstracciones proporciona Tokio? El capítulo 10 analizarácopy_bidirectionaly el marco de códec, para ver cómoFramedreduce las llamadas al sistema, cómo

Progreso del libro: Capítulo 10 / 14

Estado de verificación: líneas FACT con anclaje real

El capítulo anterior desglosó el proceso de expansión de tokio-macros, y vimos cómo #[tokio::main], select! y join! se encargan del código repetitivo y la validación en tiempo de compilación por parte del usuario. Pero lo que generan las macros sigue siendo un Future normal y una llamada a poll: cuando estos Future realmente empiezan a leer y escribir bytes, las únicas abstracciones de bajo nivel que proporciona Tokio son dos traits: AsyncRead y AsyncWrite. Su problema es que son «demasiado de bajo nivel»: un poll_read solo garantiza «se han leído algunos bytes», no garantiza «se ha leído un mensaje completo». Y la gran mayoría de protocolos (HTTP, Redis, gRPC, RPC personalizado) están orientados a «frames» en lugar de a «flujos de bytes». La pregunta central que este capítulo debe responder es: ¿dónde debería trazarse el límite de abstracción de la E/S asíncrona? La respuesta de Tokio tiene dos capas: tokio::io proporciona traits y utilidades a nivel de flujo de bytes (BufReader/BufWriter/copy_bidirectional), y el marco codec de tokio-util proporciona sobre ello adaptadores Stream/Sink a nivel de frame (Framed/LengthDelimitedCodec). Entender la división de trabajo entre estas dos capas es entender «por qué casi todas las implementaciones de protocolos empiezan por Framed».

std::io::Read::read1. AsyncRead/AsyncWrite: por qué no se puede reutilizar directamente std::io::ReadAsyncRead::poll_readModelo intuitivoPoll::Pendinges «recogida bloqueante»: te quedas frente a la ventana y, si la mercancía no llega, esperas indefinidamente, y el hilo queda suspendido.epolles «recogida con comprobante de comida»: preguntas una vez «¿ya está?», si no está (AsyncRead) te vas a hacer otras cosas, y al mismo tiempo dejas un Waker para que el sistema te avise cuando llegue la mercancía. Sin este trait, toda la E/S asíncrona tendría que implementar manualmente el registro de

y el mapeo de Waker; esto es exactamente lo que hace el Reactor del capítulo 5, y

AsyncReades la fachada unificada que expone hacia las capas superiores.

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

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

La definición deself: Pin<&mut Self>es extremadamente concisa, con un solo método:&mut selfCopiarAsyncReadLos tres parámetros tienen su razón de ser.async fnen lugar dePin: porquecx: &mut Context<'_>a menudo es retenido por el Future generado porbuf: &mut ReadBuf<'_>, y un Future, una vez que es poll, no puede moverse (auto-referencia),&mut [u8]es un contrato impuesto por el compilador.std::io::ReadEsa ambigüedad de «devuelve el número de bytes leídos pero el búfer puede no estar inicializado».

La documentación enumera explícitamente tres semánticas de retorno📎 tokio/src/io/async_read.rs:15-32：Ready(Ok(()))indica que los datos se han escritobuf, la cantidad leída viene determinada por el incremento de longitud deReadBuf::filled; si el incremento es 0, o bien es EOF, o bien esbuf.remaining() == 0(búfer de capacidad cero);Pendingindica que actualmente no es legible pero ya se ha registrado un despertar;Ready(Err(e))es un error de E/S subyacente. Aquí hay una trampa fácil de pasar por alto:«cantidad leída igual a 0» no equivale a EOF—si el llamador pasa un búfer de capacidad cero,poll_readdevolverá inmediatamenteReady(Ok(()))pero no habrá leído nada. Si la capa superior trata «0 bytes» como EOF, juzgará erróneamente que la conexión se ha cerrado.

Walkthrough guiado por escenarios: leer un fragmento de bytes desde&[u8]Consideremos la implementación más simple: la copia de

hacia&[u8]deAsyncRead：

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

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

es la longitud del segmento restante sin leer,self.len()es la capacidad restante del búfer destino, se toma el menor de ambosbuf.remaining()se divide el segmento en «elamt。split_at(amt)que se copiará esta vez» y «elarestante por leer»b」。buf.put_slice(a)se copiaaenReadBufy se avanza su puntero filled.*self = bse avanza el propio segmento a la parte restante—esta es la clave de&[u8]como «cursor»: tras cada poll,selfapunta a la parte no leída. Finalmente se devuelveReady(Ok(())), porque un segmento en memoria siempre está «listo», nuncaPending。

Obsérvese que_cxse ignora: una fuente de datos en memoria no necesita Waker. Esto contrasta con un socket de red—este último, cuando no hay datos, devuelvePendingy registra interés de legibilidad.

io::Cursor<T>La implementación de📎 tokio/src/io/async_read.rs:113-134añade una capa de verificación de límitesposition(): primero se tomapos > slice.len(), siReady(Ok(()))(posición fuera de límites) se devuelve directamente📎 tokio/src/io/async_read.rs:113-134sin panicCursor. Este es un diseño defensivo:set_positionla position de

puede ser establecida a cualquier valor por un

AsyncReadexterno; cuando está fuera de límites, tratarlo como «ya leído por completo» se ajusta más a la semántica de E/S que un panic.Box<T>、&mut T、Pin<P>Reflexión de diseño: el macro deref y la propagación de Pinderef_async_read!proporciona a📎 tokio/src/io/async_read.rs:64-70una implementación de reenvío. Los dos primeros generanPin::new(&mut **self).poll_read(cx, buf)mediante el macroPin<&mut Box<T>>, cuya esencia esPin<&mut T>—desreferenciarPin<P>a📎 tokio/src/io/async_read.rs:87-93y luego reenviar.crate::util::pin_as_deref_mut(self)La implementación dePin<&mut Pin<P>>es más sutilPin<&mut P::Target>: llama aPin, proyectando

---

La motivación de diseño aquí es la «abstracción de coste cero»: las implementaciones de reenvío permiten que tipos envoltorio como

no necesiten escribir a mano

copy_bidirectional, manteniendo al mismo tiempo la semántica correcta decopy. El coste es que cada capa de reenvío introduce una llamada indirecta, que el compilador normalmente puede eliminar mediante inline.select!II. copy_bidirectional: la máquina de estados del reenvío bidireccionalselect!Modelo intuitivocopy_bidirectionales un «camarero bidireccional»: vigila simultáneamente las dos direcciones A→B y B→A, y cualquier dato leído en un lado se escribe en el opuesto. Sin él, implementar un proxy TCP requeriría escribir a mano dos

Future y combinarlos con

—y la restricción de seguridad ante cancelación de

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

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

Runningutiliza una máquina de estados explícita para conservar los estados intermedios de «leer-escribir-cerrar», logrando así seguridad ante cancelación.CopyBufferEstructura de datos y diseño de memoriaShuttingDown(u64)El núcleo es una enumeración de tres estados:Done(u64)Copiarposee。

CopyBuffer(que contiene un búfer de 8KB y contadores de lectura/escritura), indicando «se están transfiriendo datos».copy.rslleva el número de bytes ya copiados, indicando «el lado de lectura ha llegado a EOF, se está cerrando el lado de escritura».DEFAULT_BUF_SIZEindica «cierre completado, se registra el número final de bytes». Esta enumeración es la clave de la seguridad ante cancelación:📎 tokio/src/io/util/copy_bidirectional.rs:76-88en cualquier momento en que se haga drop, el estado se conserva en la enumeración, y el siguiente poll puede continuar desde el punto de interrupciónCopyBufferproviene de

, el tamaño predeterminado lo determina

copy_bidirectional_impl(8KB)poll_fn. Cada dirección posee un

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

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

Walkthrough guiado por escenarios: el ciclo de vida completo de un reenvío bidireccionaltransfer_one_directionutilizaPoll。ready!para combinar las máquinas de estados de ambas direcciones:PendingCopiarObsérvese el orden de llamada de: primero se avanza a→b, luego b→a, ambos devuelven📎 tokio/src/io/util/copy_bidirectional.rs:143-144El macro devuelve inmediatamente cuando cualquiera de las direcciones no ha terminadoready!—peroDone(count)el estado de la otra dirección ya ha sido avanzado

transfer_one_direction. Esto es precisamente lo que el comentario enfatiza comoloop: aunque

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

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

Runningen el siguiente poll, sin perder progreso.poll_copyinternamente es unShuttingDown。ShuttingDown, que avanza según el estado:poll_shutdownCopiarDone。DoneEn el estado

se llama a

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

se llama a

avanzar ambas direcciones simultáneamente—si se usaran dospoll_copymásErr, al completarse una dirección la otra se haría drop, perdiéndose su búfer interno y su contador, lo que violaría la seguridad ante cancelación. Un?explícito📎 tokio/src/io/util/copy_bidirectional.rs:32expone el estado en la pila,📎 tokio/src/io/util/copy_bidirectional.rs:67-70y al reentrar cada vez el estado sigue ahí, garantizando así que «tras ser cancelado se puede reanudar desde el punto de interrupción».En cuanto al manejo de errores,elcopy_bidirectionaldevuelto por

copy_bidirectional_with_sizesse propaga inmediatamente hacia arriba mediante📎 tokio/src/io/util/copy_bidirectional.rs:99-125. La documentación especifica claramentepoll_copySiempre devuelveReady(Ok(0))se malinterpreta como EOF, formando un bucle ocupado.

---

Tres, Framed: dividir el flujo de bytes en frames

Modelo intuitivo

Framedes la «máquina de salchichas»: aguas arriba es un flujo continuo de agua (AsyncRead/AsyncWrite), aguas abajo son los segmentos de salchicha ya cortados (Stream<Item = Frame> / Sink<Frame>）。Decoderse encarga de «cortar un segmento del flujo de agua»,Encoderse encarga de «empaquetar un segmento en un flujo de agua». SinFramed, cada implementación de protocolo tendría que escribir a mano «gestión de búfer + manejo de medio paquete + división de paquetes pegados» — precisamente el trabajo repetitivo que el framework codec busca eliminar.

Estructura de datos y diseño de memoria

Frameden sí mismo es solo un envoltorio delgado:

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

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

El estado real está enFramedImpldestate: RWFrames, que contieneread: ReadFrameywrite: WriteFramedos partes.ReadFrameLos campos dewith_capacityson visibles en📎 tokio-util/src/codec/framed.rs:107-126：eof: bool(si el lado de lectura está en EOF),is_readable: bool(si ya se registró interés de lectura),buffer: BytesMut(búfer de lectura),has_errored: bool(si ya ocurrió un error, para prevenir lecturas repetidas).WriteFrameLos campos📎 tokio-util/src/codec/framed.rs:119-122：buffer: BytesMut(búfer de escritura),backpressure_boundary: usize(umbral de contrapresión).

backpressure_boundaryes la clave del mecanismo de contrapresión: cuando el búfer de escritura supera ese umbral,poll_readydevolveráPendinghasta que los datos se vacíen, aplicando así contrapresión alSinkaguas arriba. Por defecto es igual acapacity 📎 tokio-util/src/codec/framed.rs:121, y se puede ajustar medianteset_backpressure_boundaryajustar📎 tokio-util/src/codec/framed.rs:271-273。

Recorrido guiado por escenarios: leer un frame desde el socket

FrameddeStreamla implementación solo reenvía aFramedImpl::poll_next 📎 tokio-util/src/codec/framed.rs:309-311. La lógica real está enFramedImpl(este capítulo no proporciona ese archivo, pero la cadena de llamadas se puede inferir de la interfaz deFramed):

1. poll_nextprimero verificaread.buffersi ya hay un frame completo en (llamando acodec.decode）；

2. SidecodedevuelveSome(frame), se produce directamente, sin tocar la E/S subyacente;

3. Si devuelveNone(medio paquete), verificaread.eof: si ya está en EOF y el búfer no está vacío, significa que hay datos residuales que no se pueden decodificar, devuelve error oNone；

4. De lo contrario, llama alAsyncRead::poll_readsubyacente para leer más bytes enread.buffer；

5. Los bytes leídos intentan de nuevodecode, en bucle hasta producir un frame oPending。

Este orden de «primero decode, luego read» es importante: garantiza queuna sola read puede producir múltiples frames(paquetes pegados), y queun frame puede abarcar múltiples reads(medio paquete).is_readableEl flag evita registrar repetidamente el interés de lectura — si el poll anterior ya lo registró y no está listo, esta vez devuelve directamentePendingsin volver a llamar a la capa subyacente.

SinkLa cadena de llamadas de la implementación📎 tokio-util/src/codec/framed.rs:315-338：start_sendllama acodec.encode(item, &mut write.buffer)codifica el frame en el búfer de escritura;poll_flushvuelcawrite.buffera la capa subyacenteAsyncWrite；poll_readyverificawrite.buffer.len() >= backpressure_boundary, si supera el umbral primero hace flush y luego devuelve listo.

El siguiente diagrama de secuencia muestraFramedla colaboración entre componentes en un ciclo de ida y vuelta de «leer frame - escribir frame»:

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

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

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

Seguridad ante cancelación: advertencia de la documentación de Framed

FramedLa documentación de enumera específicamente la semántica de seguridad ante cancelación📎 tokio-util/src/codec/framed.rs:23-30：SinkExt::sendSi enselect!es completado primero por otra rama,el mensaje garantiza no haber sido enviado, pero el mensaje en sí se pierde— porquesendinternamente primeropoll_readyluegostart_send, si en la etapapoll_readyes drop,itemya fue consumido pero no codificado. Mientras queStreamExt::nextes seguro ante cancelación: solo mantiene una referencia al stream subyacente, el drop no pierde los frames ya decodificados.

Reflexión de diseño:into_partsymap_codec

Framedproporcionaninto_parts/from_partspara «cambiar codec pero conservar el búfer»📎 tokio-util/src/codec/framed.rs:290-298 📎 tokio-util/src/codec/framed.rs:155-166。map_codecestá implementado sobre este par de métodos📎 tokio-util/src/codec/framed.rs:221-234: primerointo_partsseparaio/codec/read_buf/write_buf, luego usa la funciónmappara convertir el codec, finalmentefrom_partsrecompone. Este diseño permite conservar los datos ya almacenados en búfer durante una actualización de protocolo (como cambiar de texto plano a TLS), evitando releer.

FramedPartsde_priv: ()el campo📎 tokio-util/src/codec/framed.rs:373-375es la técnica de «struct no exhaustivo»: los campos privados impiden la construcción directa desde fuera, forzando a pasar pornew/from_parts, permitiendo así añadir campos en el futuro sin romper la compatibilidad.

---

Cuatro, LengthDelimitedCodec: la máquina de estados de la codificación/decodificación con prefijo de longitud

Modelo intuitivo

LengthDelimitedCodeces la herramienta especializada para «cortar salchichas por longitud»: asume que cada frame tiene delante un campo de longitud de bytes fijos, primero lee la longitud y luego el payload. Sin él, implementar un protocolo con prefijo de longitud requeriría escribir a mano la máquina de estados «leer 4 bytes → parsear longitud → leer N bytes → repetir» — precisamente lo que hace internamenteDecodeState.

Estructura de datos y diseño de memoria

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

enum DecodeState {
    Head,
    Data(usize),
}

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

DecodeStatees una máquina de estados explícita:Headindica «se está leyendo el campo de longitud»,Data(n)indica «ya se parseó la longitud n, se está leyendo el payload». Este estado se mantiene a través de las llamadas adecode, por lo tantoen escenarios de medio paquete no se pierde el progreso。

Buildercontiene toda la configuración📎 tokio-util/src/codec/length_delimited.rs:416-435：max_frame_len(por defecto 8MB),length_field_len(por defecto 4 bytes),length_field_offset(por defecto 0),length_adjustment(por defecto 0),num_skip(por defectoNone, es deciroffset + len）、length_field_is_big_endian(por defecto true).

Recorrido guiado por escenarios: decodificar un frame con prefijo de longitud

decodees la entrada de la máquina de estados:

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

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

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

Headen el estado se llama adecode_head. Si devuelveNone(datos insuficientes), devuelve directamenteOk(None)esperando más datos; si devuelveSome(n), el estado cambia aData(n)。Dataen el estado se toma directamente n. Luego se llama adecode_data(n, src): si el búfer ya tiene n bytes,split_to(n)corta el frame, el estado vuelve aHead, y reserva espacio para la cabecera del siguiente frame; de lo contrario devuelveNoneesperando.

decode_heades la lógica central de parseo:

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

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

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

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

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

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

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

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

Parseo paso a paso: primero verificasrc.len() >= head_len, si es insuficiente devuelveNone 📎 tokio-util/src/codec/length_delimited.rs:499-502. UsaCursorpara envolversrca fin deadvance/get_uintoperar sin consumir el búfer original.advance(length_field_offset)salta el prefijo de cabecera📎 tokio-util/src/codec/length_delimited.rs:517. Lee según el endiannessfield_lenel valor de longitud de bytes📎 tokio-util/src/codec/length_delimited.rs:520-524。

Defensa clave: sin > max_frame_len, devuelve inmediatamenteInvalidDatael error📎 tokio-util/src/codec/length_delimited.rs:526-531. Esto evita que un par malicioso envíe un frame con «campo de longitud de 4GB» causando agotamiento de memoria — esta es la superficie de ataque DoS más clásica de los protocolos con prefijo de longitud.

El ajuste de longitud usachecked_sub/checked_adden lugar de una operación cruda📎 tokio-util/src/codec/length_delimited.rs:537-541, en caso de desbordamiento devuelveInvalidInputerror en lugar de panic.get_num_skip()devuelvenum_skipo el valor predeterminadooffset + len 📎 tokio-util/src/codec/length_delimited.rs:1070-1073, omitiendo el resto de la cabecera. Finalmentereserve(n.saturating_sub(src.len()))reserva espacio para el payload📎 tokio-util/src/codec/length_delimited.rs:559——se usasaturating_subporquesrcpuede que ya contenga parte del payload.

El siguiente diagrama de flujo muestradecodela ruta de decisión completa de :

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

Reflexión de diseño: recorte de max_frame_len y protección contra desbordamiento

Builder::adjust_max_frame_lenal construir el codec se recortamax_frame_lenal valor máximo que puede representar el campo de longitud📎 tokio-util/src/codec/length_delimited.rs:1075-1081。max_allowed_frame_lense calculamax_length_field_value + length_adjustment 📎 tokio-util/src/codec/length_delimited.rs:1083-1089, dondemax_length_field_valuese usachecked_shlpara manejarlength_field_len == 8el desbordamiento de desplazamiento en . Este recorte evita que el usuario configure una combinación contradictoria como «campo de longitud de 2 bytes pero max_frame_len establecido en 1MB»——2 bytes como máximo representan 65535, y tras el recorte max_frame_len pasa a ser 65535.📎 tokio-util/src/codec/length_delimited.rs:1091-1096Protección simétrica en la ruta de codificación:

se compruebaencodedevuelven > max_frame_len, el ajuste de longitud también usaInvalidInput 📎 tokio-util/src/codec/length_delimited.rs:607-607. Nótese que la dirección del ajuste al codificar es opuesta a la de decodificar: al decodificar es «longitud leída ± adjustment = longitud del payload», al codificar es «longitud del payload ∓ adjustment = campo de longitud escrito»checked_add/checked_sub 📎 tokio-util/src/codec/length_delimited.rs:620-631〔Inferencia de diseño y compensaciones arquitectónicas〕📎 tokio-util/src/codec/length_delimited.rs:620-624。

---

Repasando este capítulo, la abstracción de E/S de Tokio presenta una estructura clara de tres capas:

Primera capa: traits de flujo de bytes (

. Solo promete «leer/escribir algunos bytes», no promete fronteras de trama. Esta es la interfaz mínima, cualquier fuente de E/S (socket, archivo, slice en memoria) puede implementarla. El coste es que la capa superior debe gestionar por sí misma los paquetes parciales/pegados.AsyncRead/AsyncWrite）Segunda capa: utilidades de flujo de bytes (

. Sobre el trait proporcionan capacidades genéricas como «reducir llamadas al sistema» y «reenvío bidireccional».BufReader/BufWriter/copy_bidirectional）La máquina de estados explícita de muestra cómo se implementa la «seguridad ante cancelación» en la capa de utilidades——el estado se guarda en la pila y no dentro del Future.copy_bidirectionalTercera capa: adaptación de tramas (

. Eleva el flujo de bytes aFramed/Decoder/Encoder）, de modo que la implementación del protocolo solo necesita preocuparse por «la codificación/decodificación de la trama» y no por «la gestión de búferes».Stream<Frame>/Sink<Frame>es el ejemplo estándar de esta capa, suLengthDelimitedCodecmáquina de estados yDecodeStatela protección son patrones que todo protocolo con prefijo de longitud debería reutilizar.max_frame_len〔Inferencia de diseño y compensaciones arquitectónicas〕

---

Resumen del capítulo

• AsyncRead::poll_readusaPin<&mut Self> + Context + ReadBuftres parámetros en lugar destd::io::Read::read, convirtiendo «espera bloqueante» en «registrar Waker + devolver Pending».Ready(Ok(()))y cuando la cantidad leída es 0 hay que distinguir entre EOF y búfer de capacidad cero.
• copy_bidirectionalusaTransferStateel enum de tres estados (Running/ShuttingDown/Done) para guardar el estado intermedio, de modo que el reenvío bidireccional pueda recuperarse incluso bajoselect!cancelación. Cuando ocurre un error, parte de los datos puede perderse.
• FramedadaptaAsyncRead/AsyncWriteaStream/Sink，ReadFrame/WriteFramegestionando por separado los búferes de lectura/escritura y la contrapresión.SinkExt::sendno es seguro ante cancelación (pérdida de mensajes),StreamExt::nextes seguro ante cancelación.
• LengthDelimitedCodecusaDecodeState（Head/Data(n)) la máquina de estados para manejar paquetes parciales,max_frame_lenprotege el campo de longitud contra DoS,checked_add/checked_subprotege contra desbordamiento en los ajustes.

Reflexiones y autoevaluación del capítulo

En el capítulo anterior desglosamos el mecanismo a nivel de bytes de Framed: Decoder corta BytesMut en tramas, Sink escribe las tramas de vuelta, y así el límite de abstracción de la E/S asíncrona queda claro. Pero una trama es solo un contenedor de datos; una implementación real de protocolo se encontrará inmediatamente con tres problemas que ni tokio::io ni Framed resuelven: iteración asíncrona — Framed implementa Stream, pero Stream solo tiene poll_next, no tiene next().await, filter, take, merge; escribir poll_fn a mano es verboso y propenso a errores de seguridad ante cancelación; conjuntos dinámicos de tareas — un servicio de chat necesita suscribirse simultáneamente a N canales, los canales se unen y salen en cualquier momento, y el número de ramas de select! es fijo en tiempo de compilación, incapaz de expresar conjuntos de flujos que aumentan o disminuyen en tiempo de ejecución; cancelación estructurada — select! puede cancelar una sola rama, pero no puede propagar la detención de todo el árbol de tareas, ni puede esperar a que todas las tareas realmente terminen. tokio-stream y tokio-util nacen precisamente para estas tres cosas, y su principio de diseño clave es no empezar desde cero: cada combinador de StreamExt es solo un envoltorio sobre poll_next, StreamMap reutiliza la semántica de registro de Waker, CancellationToken se construye directamente sobre tokio::sync::Notify, TaskTracker codifica todo el estado con un AtomicUsize. Entenderlos es, en esencia, entender cómo hacer abstracciones de coste cero sobre los mecanismos existentes de Waker y scheduling. Este capítulo avanza progresivamente en tres capas: iteración, colecciones y cancelación: primero veremos cómo StreamExt convierte poll_next en un iterador componible, luego cómo StreamMap y TaskTracker gestionan colecciones dinámicas, y finalmente cómo CancellationToken propaga la señal de cancelación a todo el árbol de tareas mediante un árbol.

StreamExt: convertir poll_next en un iterador componible

Modelo intuitivo

Streames aFuture, comoIteratores a un valor:Futureproduce «un valor»,Streamproduce «una secuencia de valores». PeroStreamsolo definepoll_nextesta única primitiva, igual queIteratorsolo definenext. Si no existieraStreamExt, cada filtrado, mapeo o truncamiento requeriría escribir a mano un cierrepoll_fny gestionar manualmentePin—esto es precisamente lo más doloroso para los primeros usuarios del cratefutures.StreamExtEl papel deStreames dotar aIteratorde un ecosistema de combinadores como el de

. Sin él, el desastre al que se enfrenta el sistema no es la falta de funcionalidad, sinoel colapso sistémico de la seguridad ante cancelación: cadapoll_fnescrito a mano puede, al ser cancelado porselect!, perder un elemento que ya había sidopoll.

Estructura de datos y diseño de memoria

StreamExtes untrait de extensión, que en sí mismo no contiene datos:

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

pub trait StreamExt: Stream {

Todos sus métodos devuelven unaestructura concreta de combinador, en lugar deBox<dyn Stream>. Esta es la decisión de diseño clave:mapdevuelveMap<Self, F>，filterdevuelveFilter<Self, F>，takedevuelveTake<Self>. Estas estructuras son envoltorios genéricos sin asignación en heap, y el compilador puede inline toda la cadena en capas de llamadas apoll_next.

Nótese el blanket impl del trait:

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

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

CualquierStreamobtiene automáticamente todos los combinadores, sin necesidad de implementación manual.?Sizedpermite quedyn Streamtambién disfrute de los métodos de extensión.

La declaración de módulos de los combinadores revela la superficie completa de capacidades de este trait:

📎 tokio-stream/src/stream_ext.rs:4-59

mod all; use all::AllFuture;
mod any; use any::AnyFuture;
mod chain; pub use chain::Chain;
pub(crate) mod collect; use collect::{Collect, FromStream};
mod filter; pub use filter::Filter;
mod filter_map; pub use filter_map::FilterMap;
mod fold; use fold::FoldFuture;
mod fuse; pub use fuse::Fuse;
mod map; pub use map::Map;
mod map_while; pub use map_while::MapWhile;
mod merge; pub use merge::Merge;
mod next; use next::Next;
mod skip; pub use skip::Skip;
mod skip_while; pub use skip_while::SkipWhile;
mod take; pub use take::Take;
mod take_while; pub use take_while::TakeWhile;
mod then; pub use then::Then;
mod try_next; use try_next::TryNext;
mod peekable; pub use peekable::Peekable;

Aquí hay una distinción que merece atención:next、try_next、all、any、fold、collectdevuelveFuture（Next、TryNext、AllFuture…), porque consumen todo el flujo en un solo valor; mientras quemap、filter、takeetc. devuelvenStream, porque mantienen la forma del flujo.nextEl tipo de retorno deNext<'_, Self>es

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

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

Self: UnpinCopiarnextLa restricción dePines deliberada:!Unpinno toma posesión del flujo, solo lo toma prestado, por lo que no puedeBox::pinel flujo. Si el flujo espin_mut!, el usuario debe primero

📎 tokio-stream/src/stream_ext.rs:116-121

/// Note that because `next` doesn't take ownership over the stream,
/// the [`Stream`] type must be [`Unpin`]. If you want to use `next` with
/// a [`!Unpin`](Unpin) stream, you'll first have to pin the stream. This can
/// be done by boxing the stream using [`Box::pin`] or
/// pinning it to the stack using the `pin_mut!` macro from the `pin_utils`
/// crate.

. La documentación señala explícitamente este compromiso:mergedel polling

mergees el mejor ejemplo para entender cómo los combinadores reutilizan el Waker. Intercala la producción de dos flujos, ygarantiza la equidad——si ambos flujos están listos a la vez, produce alternadamente. La documentación advierte específicamente contra el encadenamiento de llamadasmerge：

📎 tokio-stream/src/stream_ext.rs:319-321

/// simultaneously, the merge stream alternates between them. This provides
/// some level of fairness. You should not chain calls to `merge`, as this
/// will break the fairness of the merging.

mergede la firma requiere que ambos flujos tengan el mismo tipoItem:

📎 tokio-stream/src/stream_ext.rs:398-404

fn merge<U>(self, other: U) -> Merge<Self, U>
where
    U: Stream<Item = Self::Item>,
    Self: Sized,
{
    Merge::new(self, other)
}

Cuando el llamador.next().await, el flujo de ejecución es el siguiente:

1. Next::pollllama aMerge::poll_next。

2. Mergemantiene internamente una bandera booleana de «a quién le tocó la última vez». Primeropollel flujo que no produjo la última vez; siPending, entoncespollel otro.

3. Si ambosPending，MergedevuelvenPending, perolos Wakers de ambos flujos ya están registrados——cualquiera que esté listo despertará la tarea actual.

4. Si un flujo devuelveReady(None)(fin),Mergeregistra que ese flujo terminó, y a partir de entonces solopollel otro flujo, hasta que también termine.

La clave aquí es:Mergeno tiene su propia lógica de gestión de Waker; pasacxtal cual a los dos flujos internospoll_next。El registro del Waker corre completamente a cargo de los flujos subyacentes，Mergesolo decide «a quién preguntar primero esta vez». Este es el significado literal de «reutilizar el mecanismo de Waker subyacente».

merge_size_hintsLa función auxiliar muestra cómo los combinadores fusionan las pistas de capacidad:

📎 tokio-stream/src/stream_ext.rs:1216-1226

fn merge_size_hints(
    (left_low, left_high): (usize, Option<usize>),
    (right_low, right_high): (usize, Option<usize>),
) -> (usize, Option<usize>) {
    let low = left_low.saturating_add(right_low);
    let high = match (left_high, right_high) {
        (Some(h1), Some(h2)) => h1.checked_add(h2),
        _ => None,
    };
    (low, high)
}

Nótese la elección entresaturating_addychecked_add: para el límite inferior se usa suma saturante (mejor subestimar que desbordar con panic), para el límite superior se usa suma verificada (si alguno es desconocido, el total es desconocido). Esta es la forma típica de manejar el contrato desize_hint.

Reflexión de diseño: seguridad ante cancelación y protección contra panic dechunks_timeout

StreamExtLa documentación deCancel safetyanotanexten cada método. Tomemos

📎 tokio-stream/src/stream_ext.rs:123-127

/// # Cancel safety
///
/// This method is cancel safe. The returned future only
/// holds onto a reference to the underlying stream,
/// so dropping it will never lose a value.

nextCopiaNextes seguro ante cancelación porque solo toma prestado el flujo, no consume elementos——nextcuando el future se descarta, el estado del propio flujo no cambia, y la próxima vezpoll。

volverá achunks_timeoutPero no todos los combinadores son seguros ante cancelación.

📎 tokio-stream/src/stream_ext.rs:1178-1185

#[track_caller]
fn chunks_timeout(self, max_size: usize, duration: Duration) -> ChunksTimeout<Self>
where
    Self: Sized,
{
    assert!(max_size > 0, "`max_size` must be non-zero.");
    ChunksTimeout::new(self, max_size, duration)
}

timeout, la lógica de procesamiento por lotes caería en un bucle infinito de «nunca acumular un lote completo» o produciría lotes vacíos, y este tipo de bug es extremadamente difícil de localizar en tiempo de ejecución. El panic en la fase de construcción adelanta el error al punto observable más temprano.timeout_repeatingLa diferencia entretimeoutytambién merece atención:；timeout_repeatingdevuelve un error tras el timeout, peroIntervalcontinúa haciendo polling del flujo interno

📎 tokio-stream/src/stream_ext.rs:985-1001

/// Once a timeout error is received, no further events will be received
/// unless the wrapped stream yields a value (timeouts do not repeat).

📎 tokio-stream/src/stream_ext.rs:1071-1072

/// Timeout errors will be continuously produced at the specified interval
/// until the wrapped stream yields a value.

---

Copia

Copia

select!StreamMap: colección dinámica de flujos y polling equitativoStreamMapModelo intuitivoselect!tiene un número de ramas fijo en tiempo de compilación. Pero el número de canales a los que se debe suscribir un servicio de chat, o el número de conexiones que debe rastrear un crawler, solo se conocen en tiempo de ejecución.nextes precisamente «(key, value)que se puede añadir o eliminar en tiempo de ejecución»: coloca cualquier cantidad de flujos en una colección, y cada vezmpscdevuelve

, indicándote de qué flujo proviene el valor. Sin él, solo podrías meter todos los flujos en un

StreamMapcanal, con una capa adicional de sobrecarga de reenvío.Vec：

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

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

El almacenamiento de

📎 tokio-stream/src/stream_map.rs:38-44

/// `StreamMap` is backed by a `Vec<(K, V)>`. There is no guarantee that this
/// internal implementation detail will persist in future versions, but it is
/// important to know the runtime implications. In general, `StreamMap` works
/// best with a "smallish" number of streams as all entries are scanned on
/// insert, remove, and polling. In cases where a large number of streams need
/// to be merged, it may be advisable to use tasks sending values on a shared
/// [`mpsc`] channel.

insert, cada

📎 tokio-stream/src/stream_map.rs:446-454

pub fn insert(&mut self, k: K, stream: V) -> Option<V>
where
    K: Hash + Eq,
{
    let ret = self.remove(&k);
    self.entries.push((k, stream));

    ret
}

removeEl escaneo O(n) deswap_removey

📎 tokio-stream/src/stream_map.rs:471-483

pub fn remove<Q>(&mut self, k: &Q) -> Option<V>
where
    K: Borrow<Q>,
    Q: Hash + Eq + ?Sized,
{
    for i in 0..self.entries.len() {
        if self.entries[i].0.borrow() == k {
            return Some(self.entries.swap_remove(i).1);
        }
    }

    None
}

La implementación de

StreamMaprefleja la semántica de «eliminar primero, insertar después»:poll_next_entryCopiausapara intercambiar el elemento eliminado con el último elemento y luego hacer pop, evitando el desplazamiento O(n):

📎 tokio-stream/src/stream_map.rs:515-550

fn poll_next_entry(&mut self, cx: &mut Context<'_>) -> Poll<Option<(usize, V::Item)>> {
    let start = self::rand::thread_rng_n(self.entries.len() as u32) as usize;
    let mut idx = start;

    for _ in 0..self.entries.len() {
        let (_, stream) = &mut self.entries[idx];

        match Pin::new(stream).poll_next(cx) {
            Poll::Ready(Some(val)) => return Poll::Ready(Some((idx, val))),
            Poll::Ready(None) => {
                // Remove the entry
                self.entries.swap_remove(idx);

                // Check if this was the last entry, if so the cursor needs
                // to wrap
                if idx == self.entries.len() {
                    idx = 0;
                } else if idx < start && start <= self.entries.len() {
                    // The stream being swapped into the current index has
                    // already been polled, so skip it.
                    idx = idx.wrapping_add(1) % self.entries.len();
                }
            }
            Poll::Pending => {
                idx = idx.wrapping_add(1) % self.entries.len();
            }
        }
    }

    // If the map is empty, then the stream is complete.
    if self.entries.is_empty() {
        Poll::Ready(None)
    } else {
        Poll::Pending
    }
}

Walkthrough guiado por escenarios: punto de inicio aleatorio y corrección del cursor en poll_next_entry

El núcleo de thread_rng_nesFastRand. Comienza el polling desdexorshift64+un punto de inicio aleatorio

📎 tokio-stream/src/stream_map.rs:765-768

/// Implement `xorshift64+`: 2 32-bit `xorshift` sequences added together.
/// Shift triplet `[17,7,16]` was calculated as indicated in Marsaglia's
/// `Xorshift` paper

fastrand_nCopia% n：

📎 tokio-stream/src/stream_map.rs:787-792

pub(crate) fn fastrand_n(&self, n: u32) -> u32 {
    // This is similar to fastrand() % n, but faster.
    // See https://lemire.me/blog/2016/06/27/a-fast-alternative-to-the-modulo-reduction/
    let mul = (self.fastrand() as u64).wrapping_mul(n as u64);
    (mul >> 32) as u32
}

Primero, el punto de inicio aleatorio.swap_removeusathread-localidx, basado en el algoritmoNone:swap_removeCopiaidxusa la multiplicación y módulo de Lemire en lugar deCopiaSegundo,startla corrección del cursor trasidx < start && start <= self.entries.len(). Cuando el flujo en el índiceidx = idx.wrapping_add(1) % lendevuelveidx == leny es eliminado,

mueve el último elemento aPoll::Pending. Este elemento movido puedeya haber sido sometido a pollingPending(si su índice original estaba antes de

poll_next). El código usapoll_next_entrypara detectar esta situación y, si es así, lo salta (

📎 tokio-stream/src/stream_map.rs:676-683

fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
    if let Some((idx, val)) = ready!(self.poll_next_entry(cx)) {
        let key = self.entries[idx].0.clone();
        Poll::Ready(Some((key, val)))
    } else {
        Poll::Ready(None)
    }
}

), el cursor vuelve a 0.ready!Tercero,poll_next_entryla semántica dePending. Si tras recorrer una vuelta ningún flujo está listo y la colección no está vacía, devuelvepoll_next. En ese momento los Wakers de todos los flujos ya están registrados, y cualquiera que esté listo despertará.Pending。K: Cloneañade la clave sobrekey.clone()。

:

next_manyCopiaStreamMapNótese la macro

📎 tokio-stream/src/stream_map.rs:581-583

pub async fn next_many(&mut self, buffer: &mut Vec<(K, V::Item)>, limit: usize) -> usize {
    poll_fn(|cx| self.poll_next_many(cx, buffer, limit)).await
}

devuelve

📎 tokio-stream/src/stream_map.rs:573-578

/// # Cancel safety
///
/// This method is cancel safe. If `next_many` is used as the event in a
/// [`tokio::select!`] statement and some other branch completes first,
/// it is guaranteed that no items were received on any of the underlying
/// streams.

devuelve inmediatamentenext_manyLa restricción proviene de aquíReflexión de diseño: semántica por lotes y seguridad ante cancelación de next_manybufferes la versión por lotes debuffer, que recoge tantos elementos listos como sea posible de una vez:bufferCopia

poll_next_manySu garantía de seguridad ante cancelación es crucial:poll_next_entryCopia

📎 tokio-stream/src/stream_map.rs:597-666

pub fn poll_next_many(
    &mut self,
    cx: &mut Context<'_>,
    buffer: &mut Vec<(K, V::Item)>,
    limit: usize,
) -> Poll<usize> {
    if limit == 0 || self.entries.is_empty() {
        return Poll::Ready(0);
    }

    let mut added = 0;

    let start = self::rand::thread_rng_n(self.entries.len() as u32) as usize;
    let mut idx = start;

    while added < limit {
        // Indicates whether at least one stream returned a value when polled or not
        let mut should_loop = false;

        for _ in 0..self.entries.len() {
            let (_, stream) = &mut self.entries[idx];

            match Pin::new(stream).poll_next(cx) {
                Poll::Ready(Some(val)) => {
                    added += 1;

                    let key = self.entries[idx].0.clone();
                    buffer.push((key, val));

                    should_loop = true;

                    idx = idx.wrapping_add(1) % self.entries.len();

                    if added == limit {
                        break;
                    }
                }
                Poll::Ready(None) => {
                    // Remove the entry
                    self.entries.swap_remove(idx);

                    // Check if this was the last entry, if so the cursor needs
                    // to wrap
                    if idx == self.entries.len() {
                        idx = 0;
                    } else if idx < start && start <= self.entries.len() {
                        // The stream being swapped into the current index has
                        // already been polled, so skip it.
                        idx = idx.wrapping_add(1) % self.entries.len();
                    }
                }
                Poll::Pending => {
                    idx = idx.wrapping_add(1) % self.entries.len();
                }
            }
        }

        if !should_loop {
            break;
        }
    }

    if added > 0 {
        Poll::Ready(added)
    } else if self.entries.is_empty() {
        Poll::Ready(0)
    } else {
        Poll::Pending
    }
}

es seguro ante cancelación? Porque hacewhile added < limitpush de los elementos inmediatamente en elforproporcionado por el llamador, en lugar de almacenarlos temporalmente en el interior. Si el future se descarta, los elementos ya insertados siguen enshould_loop = true, sin perderse. Pero esto también implica que: al descartarse,limitpuede que ya tenga algunos elementos——el llamador necesita saberlo.

📎 tokio-stream/src/stream_map.rs:588-591

/// * `Poll::Pending` if no items are available but the `StreamMap` is not empty.
/// * `Poll::Ready(count)` where `count` is the number of items successfully received and
///   stored in `buffer`. This can be less than, or equal to, `limit`.
/// * `Poll::Ready(0)` if `limit` is set to zero or when the `StreamMap` is empty.

size_hintLa implementación de muestra cómo agregar las pistas de capacidad de múltiples flujos:

📎 tokio-stream/src/stream_map.rs:685-701

fn size_hint(&self) -> (usize, Option<usize>) {
    let mut ret: (usize, Option<usize>) = (0, Some(0));

    for (_, stream) in &self.entries {
        let hint = stream.size_hint();

        ret.0 = ret.0.saturating_add(hint.0);

        match (ret.1, hint.1) {
            (Some(a), Some(b)) => ret.1 = a.checked_add(b),
            (Some(_), None) => ret.1 = None,
            _ => {}
        }
    }

    ret
}

Igual quemerge_size_hintsel mismo patrón: saturación del límite inferior con suma, verificación del límite superior con suma, y si alguno es desconocido, el conjunto es desconocido.

A continuación, se describe con un diagrama de flujopoll_next_entryla ruta de decisión de:

flowchart TD
    start["poll_next_entry(cx)"] --> rand["start = thread_rng_n(len)"]
    rand --> loop{"遍历 len 次?"}
    loop -->|"未完成"| poll["Pin::new(stream).poll_next(cx)"]
    poll -->|"Ready(Some(val))"| ret_val["返回 Ready(Some((idx, val)))"]
    poll -->|"Ready(None)"| remove["entries.swap_remove(idx)"]
    remove --> wrap{"idx == entries.len()?"}
    wrap -->|"是"| set_zero["idx = 0"]
    wrap -->|"否"| check_swap{"idx < start && start <= len?"}
    check_swap -->|"是"| skip["idx = idx.wrapping_add(1) % len"]
    check_swap -->|"否"| loop
    set_zero --> loop
    skip --> loop
    poll -->|"Pending"| advance["idx = idx.wrapping_add(1) % len"]
    advance --> loop
    loop -->|"遍历完成"| empty{"entries.is_empty()?"}
    empty -->|"是"| ret_none["返回 Ready(None)"]
    empty -->|"否"| ret_pending["返回 Pending"]

---

TaskTracker: codificar todo el estado con un único AtomicUsize

Modelo intuitivo

El cierre elegante requiere dos cosas:notificar a las tareas que se detengan（CancellationTokense encarga de ello), yesperar a que las tareas realmente salgan（TaskTrackerse encarga de ello).TaskTrackerEs como una combinación de «contador de tareas + interruptor de cierre»: mientras haya tareas en ejecución, o no se haya llamado aclose，wait()no retornará. Sin él, solo podrías usarJoinSetperoJoinSetacumularía el valor de retorno de cada tarea, y un servicio de larga duración sufriría OOM.

Estructura de datos y diseño de memoria

TaskTrackeres un envoltorio deArc:

📎 tokio-util/src/task/task_tracker.rs:158-178

pub struct TaskTracker {
    inner: Arc<TaskTrackerInner>,
}

/// Represents a task tracked by a [`TaskTracker`].
#[must_use]
#[derive(Debug)]
pub struct TaskTrackerToken {
    task_tracker: TaskTracker,
}

struct TaskTrackerInner {
    /// Keeps track of the state.
    ///
    /// The lowest bit is whether the task tracker is closed.
    ///
    /// The rest of the bits count the number of tracked tasks.
    state: AtomicUsize,
    /// Used to notify when the last task exits.
    on_last_exit: Notify,
}

Este es el diseño de memoria más ingenioso de este capítulo:unAtomicUsizecodifica simultáneamente «si está cerrado» y «el conteo de tareas». El bit menos significativo es la bandera de cierre, y los bits restantes son el número de tareas (porque el conteo de tareas se incrementa cada vez+2, el bit menos significativo siempre es 0). Así,is_closed_and_emptysolo necesita una carga atómica:

📎 tokio-util/src/task/task_tracker.rs:216-222

fn is_closed_and_empty(&self) -> bool {
    // If empty and closed bit set, then we are done.
    //
    // The acquire load will synchronize with the release store of any previous call to
    // `set_closed` and `drop_task`.
    self.state.load(Ordering::Acquire) == 1
}

Walkthrough guiado por escenarios: la carrera entre close y drop_task

Considera un escenario típico: el hilo principal llama atracker.close(), mientras la última tarea está saliendo (TaskTrackerToken::dropllama adrop_task). Ambos pueden ser concurrentes, y debe garantizarse que, sin importar quién vaya primero,wait()pueda ser despertado.

Primero veamosset_closed：

📎 tokio-util/src/task/task_tracker.rs:225-249

fn set_closed(&self) -> bool {
    // The AcqRel ordering makes the closed bit behave like a `Mutex<bool>` for synchronization
    // purposes. ...
    let state = self.state.fetch_or(1, Ordering::AcqRel);

    // If there are no tasks, and if it was not already closed:
    if state == 0 {
        self.notify_now();
    }

    (state & 1) == 0
}

fetch_or(1, AcqRel)establece atómicamente el bit de cierre y devuelve el valor antiguo. Si el valor antiguo es 0 (no estaba cerrado y no había tareas), significa que «tras el cierre se cumple inmediatamente vacío + cerrado», y se llama anotify_now. El valor de retorno(state & 1) == 0indica que «esta llamada realmente cambió el estado».

Ahora veamosdrop_task：

📎 tokio-util/src/task/task_tracker.rs:264-271

fn drop_task(&self) {
    let state = self.state.fetch_sub(2, Ordering::Release);

    // If this was the last task and we are closed:
    if state == 3 {
        self.notify_now();
    }
}

fetch_sub(2, Release)decrementa el conteo. Si el valor antiguo es 3 (binario11: bit de cierre 1 + conteo 1), significa que «esta es la última tarea y ya está cerrado», y se llama anotify_now。

Análisis de la carrera entre las dos rutas:

• close se ejecuta primero：set_closedve el valor antiguo2(conteo 1, no cerrado), no notifica. Luegodrop_taskve el valor antiguo3, notifica. ✓
• drop_task se ejecuta primero：drop_taskve el valor antiguo2(conteo 1, no cerrado), no notifica. Luegoset_closedve el valor antiguo0(conteo 0, no cerrado), notifica. ✓
• Concurrencia：fetch_oryfetch_subson atómicas; sin importar el orden de intercalado, siempre habrá una que vea la combinación «cerrado + vacío» y notifique. ✓

notify_nowHay una cargaAcquirefácil de pasar por alto en

📎 tokio-util/src/task/task_tracker.rs:274-285

#[cold]
fn notify_now(&self) {
    // Insert an acquire fence. This matters for `drop_task` but doesn't matter for
    // `set_closed` since it already uses AcqRel.
    //
    // This synchronizes with the release store of any other call to `drop_task`, and with the
    // release store in the call to `set_closed`. That ensures that everything that happened
    // before those other calls to `drop_task` or `set_closed` will be visible after this load,
    // and those things will also be visible to anything woken by the call to `notify_waiters`.
    self.state.load(Ordering::Acquire);

    self.on_last_exit.notify_waiters();
}

Copiardrop_task¿Por quéReleaseusaAcqRelen lugar dedrop_task? Porque elfetch_subdenotify_nowsolo necesita «hacer visibles las escrituras previas para lectores posteriores» (semántica Release), no necesita «ver las escrituras previas de otros hilos» (semántica Acquire). Perowait()necesita Acquire para establecer happens-before: garantizar que todo el trabajo de limpieza realizado antes de que la tarea salga sea visible para el código posterior al retorno deload. El resultado de este

se descarta, puramente por su efecto secundario de ordenamiento de memoria; este es el uso típico de «carga tipo fence» en las operaciones atómicas de Rust.

waitReflexión de diseño: la resistencia a ABA de wait y la semántica de drop de TrackedFutureTaskTrackerWaitFuturedevuelve unNotified：

📎 tokio-util/src/task/task_tracker.rs:318-327

pub fn wait(&self) -> TaskTrackerWaitFuture<'_> {
    TaskTrackerWaitFuture {
        future: self.inner.on_last_exit.notified(),
        inner: if self.inner.is_closed_and_empty() {
            None
        } else {
            Some(&self.inner)
        },
    }
}

CopiarinnerNota el campoNone，poll: si al crearse ya está «cerrado y vacío», se establece directamente comoReadyy retorna inmediatamente

. Esta es la ruta rápida.

📎 tokio-util/src/task/task_tracker.rs:304-307

/// The `wait` future is resistant against [ABA problems][aba]. That is, if the `TaskTracker`
/// becomes both closed and empty for a short amount of time, then it is guarantee that all
/// `wait` futures that were created before the short time interval will trigger, even if they
/// are not polled during that short time interval.

CopiarNotify::notified()Esta garantía proviene de la semántica deNotified:notify_waitersel future registra su identidad de «esperador» en el momento de su creación; incluso sipollse llama antes de que seapoll, verá la notificación en su primerTaskTrackerWaitFuture::poll.

📎 tokio-util/src/task/task_tracker.rs:697-712

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

    let inner = match me.inner.as_ref() {
        None => return Poll::Ready(()),
        Some(inner) => inner,
    };

    let ready = inner.is_closed_and_empty() || me.future.poll(cx).is_ready();
    if ready {
        *me.inner = None;
        Poll::Ready(())
    } else {
        Poll::Pending
    }
}

CopiarpollCada vez queis_closed_and_empty()primero verificapoll Notified, y luegoNotified. Este orden garantiza que: incluso si

TrackedFutureno es despertado por alguna razón, la verificación de estado también sirve como respaldo.TaskTrackerLa semántica de drop deJoinSetes la diferencia central entre

📎 tokio-util/src/task/task_tracker.rs:488-494

/// The task is removed from the collection when it is dropped, not when [`poll`] returns
/// [`Poll::Ready`].

:ReadyCopiarTrackedFutureEsto significa que: incluso si el future ya ha retornadoTaskTracker, mientras

📎 tokio-util/src/task/task_tracker.rs:33-35

/// When a call to [`wait`] returns, it is guaranteed that all tracked tasks have exited and that
/// the destructor of the future has finished running. However, there might be a short amount of
/// time where [`JoinHandle::is_finished`] returns false.

TaskTrackerTokenconsidera que la tarea sigue existiendo. La documentación explica por qué este diseño es importante:DropCopiar

📎 tokio-util/src/task/task_tracker.rs:670-672

impl Drop for TaskTrackerToken {
    /// Dropping the token indicates to the [`TaskTracker`] that the task has exited.
    #[inline]
    fn drop(&mut self) {
        self.task_tracker.inner.drop_task();
    }
}

TrackedFuturedepin_project!es el punto de activación del decremento del conteo:tokenCopiarfutureempaquetatokenyspawn_blockingmediante

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

, y el drop de

En el capítulo anterior vimos cómo tokio-stream y tokio-util reutilizan el Waker y el mecanismo de planificación subyacentes para extender las capacidades del núcleo. Pero sin importar cuántos combinadores se extiendan, la contradicción central del runtime asíncrono siempre existe: el planificador debe distribuir el tiempo de CPU de forma justa entre múltiples tareas, y las tareas en sí no son apropiativas — una vez que el poll de un Future comienza a ejecutarse, el planificador no puede interrumpirlo desde fuera. Si una tarea procesa cien mil mensajes en bucle dentro de un solo poll, o hace await repetidamente sobre un Future siempre listo dentro de un loop, acaparará el hilo worker y hará que las demás tareas del mismo hilo nunca obtengan oportunidad de ser sondeadas. Este es el clásico problema de «la tarea mata de inanición al planificador». La solución de Tokio no es la apropiatividad, sino la cooperación: asignar a cada tarea un presupuesto limitado dentro de un ciclo de planificación; las operaciones de recursos consumen presupuesto, y cuando este se agota, la tarea debe ceder voluntariamente. Este capítulo profundiza en la implementación de este mecanismo coop.

12.1 El portador del presupuesto: almacenamiento local de hilo y la estructura Budget

El presupuesto debe cumplir dos restricciones: primera, debe poder ser accedido desde una pila de llamadas de cualquier profundidad, sin necesidad de pasar parámetros capa por capa; segunda, debe poder distinguir «si actualmente se está dentro del runtime de Tokio» — fuera del runtime, al llamar apollno debería estar sujeto a restricciones de presupuesto. Tokio elige usarblock_onalmacenamiento local de hilo (TLS)para portar el presupuesto, y lo gestiona de forma unificada a través del módulo.contextEl tipo central del presupuesto es

. Aunque el fragmento de código fuente de este capítulo no proporciona directamente la definición completa decoop::Budget, a partir de los puntos de uso decoop.rsse puede inferir su contrato de interfaz:worker.rsCopiar

📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:695-795

coop::budget(|| {
    // ... 轮询任务 ...
    task.run();
    // ...
    loop {
        // ...
        if !coop::has_budget_remaining() {
            // 预算耗尽，把 LIFO 任务推回队列
            core.run_queue.push_back_or_overflow(task, ...);
            return ControlFlow::Continue(core);
        }
        // ...
    }
})

establece un ámbito de presupuesto,coop::budget(closure)consulta el presupuesto restante, y más adelante se veráncoop::has_budget_remaining()ycoop::stop()La semántica es: al entrar en el closure, se restablece el presupuesto del hilo actual a un valor completo (por defecto 128); durante la ejecución del closure, todas las operaciones de recursos comparten esta cuota; al salir del closure, se restaura el presupuesto exterior.coop::set()。budget〔Inferencia de diseño y compensaciones arquitectónicas〕

Budget.Cell<Option<Budget>>La semántica externa deOptiones «si el hilo actual está en el contexto del runtime de Tokio»:Noneindica que no está dentro del runtime (por ejemplo,block_onfuera del runtime), en cuyo caso todas las comprobaciones de presupuesto se dejan pasar directamente.

12.2 Los puntos de consumo del presupuesto: cómo las operaciones de recursos lo descuentan

El presupuesto no se consume de la nada; solo lasoperaciones de recursoslo descuentan. Las llamadas operaciones de recursos son aquellas API que interactúan con el mundo exterior y que pueden ser invocadas en bucles infinitos — elsend/recvde channel, la lectura/escritura de I/O,yield_now, etc. Tomemos como ejemplompsc::Sender::reserve, que es el punto de entrada común de todas las rutas de envío:

📎 tokio/src/sync/mpsc/bounded.rs:1272-1311

async fn reserve_inner(&self, n: usize) -> Result<(), SendError<()>> {
    crate::trace::async_trace_leaf().await;

    if n > self.max_capacity() {
        return Err(SendError(()));
    }
    // ... WakeReceiverOnDrop guard ...
    let guard = WakeReceiverOnDrop { chan: &self.chan };
    let result = self.chan.semaphore().semaphore.acquire(n).await;
    // ...
}

reserve_innerAntes de obtener realmente el permiso del semáforo, pasa porcrate::trace::async_trace_leaf(). Esta llamada, que aparentemente solo sirve para tracing, es en realidad uno de los puntos de enganche para el descuento de presupuesto.async_trace_leafInternamente llama a una función del tipocoop::poll_proceed: si el presupuesto es suficiente, descuenta 1 y devuelveProceed; si el presupuesto se agota, registra una acción de «cesión» — entrega el Waker de la tarea actual al planificador, devuelvePending, y hace que la tarea termine anticipadamente en este poll.

Aquí está la sutileza de coop:que el presupuesto se agote no lanza un error, sino que disfraza la «cesión» como unPendingordinario. El Future superior, al verPending, retornará naturalmente; el planificador reencola la tarea, y cuando sea planificada de nuevo el presupuesto ya se habrá restablecido, y la tarea continuará desde donde se interrumpió. Todo el proceso es completamente transparente para el código de negocio.

yield_nowes la manifestación más directa del mecanismo de presupuesto; no consume presupuesto, sino queactiva proactivamente la cesión：

📎 tokio/src/task/yield_now.rs:38-60

pub async fn yield_now() {
    let mut yielded = false;
    poll_fn(|cx| {
        ready!(crate::trace::trace_leaf());

        if yielded {
            return Poll::Ready(());
        }

        yielded = true;

        // Don't wake the task immediately, as that would push it right back
        // onto the run queue and it could be polled again before other tasks
        // or the IO/timer driver get a chance to run. Instead, hand the waker
        // to the scheduler, which wakes deferred tasks only after it has run
        // out of ready tasks and polled the driver. When polled from outside
        // a Tokio runtime, the waker is woken immediately.
        context::defer(cx.waker());

        Poll::Pending
    })
    .await
}

Nótese la líneacontext::defer(cx.waker()). No hacewakedirectamente, sino que entrega el Waker a lacola deferdel planificador. ¿Por qué? Los comentarios del código fuente lo dicen claramente: si se despertara inmediatamente, la tarea sería empujada de vuelta a la cola de ejecución de inmediato, y podría ser sondeada otra vez antes de que el driver de I/O/timer se ejecute, haciendo que la cesión pierda sentido. La semántica de la cola defer es «despertar estas tareas después de que el worker actual termine de ejecutar las tareas listas y haya sondeado los drivers».

La cola defer está definida en elContextdel worker:

📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:247-257

pub(crate) struct Context {
    worker: Arc<Worker>,
    core: RefCell<Option<Box<Core>>>,
    /// Tasks to wake after resource drivers are polled. This is mostly to
    /// handle yielded tasks.
    pub(crate) defer: Defer,
}

deferEl comentario del campo señala directamente su propósito: «mostly to handle yielded tasks». En el bucle principal del worker, cuando ni la cola local ni el robo tienen trabajo que hacer, se comprueba la cola defer:

📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:613-621

} else {
    // Wait for work
    core = if !self.defer.is_empty() {
        self.park_yield(core)
    } else {
        self.park(core)
    };
    core.stats.start_processing_scheduled_tasks();
}

Si la cola de defer no está vacía, el worker llama apark_yield——con un timeout de 0, lo que impulsa la E/S y el timer, y luego despierta las tareas en defer. Esto garantiza que la tarea que "cede" sea reprogramada solo después de que el driver haya corrido.

12.3 Establecimiento y restauración del ámbito de presupuesto: run_task y block_in_place

El ámbito de presupuesto se establece enrun_task. Cuando cada tarea es sondeada,coop::budgetenvuelve todo el proceso de sondeo:

📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:691-704

// Make the core available to the runtime context
*self.core.borrow_mut() = Some(core);

// Run the task
coop::budget(|| {
    // ...
    task.run();
    // ...
})

coop::budgetAl entrar, establece el presupuesto en TLS al máximo, y al salir lo restaura. Esto significa quecada tarea obtiene un presupuesto completamente nuevo cada vez que es sondeada. Dentro de la tarea, sin importarawaitcuántas operaciones de recursos se hayan realizado, siempre que dentro de un solopollel consumo supere 128, se verá forzada a ceder.

Pero aquí hay un problema sutil: las tareas en el LIFO slot son sondeadas dentro deel mismobudgetcierre. Observarun_taskel bucle de:

📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:709-750

let mut lifo_polls = 0;

// As long as there is budget remaining and a task exists in the
// `lifo_slot`, then keep running.
loop {
    let mut core = match self.core.borrow_mut().take() {
        Some(core) => core,
        None => {
            return ControlFlow::Break(());
        }
    };

    let task = match core.lifo_slot.take() {
        Some(task) => task,
        None => {
            self.reset_lifo_enabled(&mut core);
            core.stats.end_poll();
            return ControlFlow::Continue(core);
        }
    };

    if !coop::has_budget_remaining() {
        core.stats.end_poll();
        // Not enough budget left to run the LIFO task, push it to
        // the back of the queue and return.
        core.run_queue.push_back_or_overflow(task, ...);
        debug_assert!(core.lifo_enabled);
        return ControlFlow::Continue(core);
    }
    // ...
}

Punto clave: las tareas en el LIFO slotcomparten el presupuesto de la tarea externa. El comentario al inicio derun_taskdice: "Tasks from the LIFO slot inherit the 'parent''s limits". Este es un diseño intencional——si cada tarea LIFO reiniciara el presupuesto, entonces en escenarios ping-pong (la tarea A despierta a B, B despierta a A), ambas tareas se programarían mutuamente de forma infinita, el presupuesto nunca se reiniciaría y el problema de inanición persistiría. Compartir el presupuesto significa que A y B juntas consumen como máximo 128 operaciones de recursos, tras lo cual deben ceder.

El LIFO slot en sí también tiene un limitador de tasa independienteMAX_LIFO_POLLS_PER_TICK：

📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:756-766

// Disable the LIFO slot if we reach our limit
//
// In ping-ping style workloads where task A notifies task B,
// which notifies task A again, continuously prioritizing the
// LIFO slot can cause starvation as these two tasks will
// repeatedly schedule the other. To mitigate this, we limit the
// number of times the LIFO slot is prioritized.
if lifo_polls >= MAX_LIFO_POLLS_PER_TICK {
    core.lifo_enabled = false;
    super::counters::inc_lifo_capped();
}

MAX_LIFO_POLLS_PER_TICKEl valor de es 3:

📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:263-263

/// Value picked out of thin-air. Running the LIFO slot a handful of times
/// seems sufficient to benefit from locality. More than 3 times probably is
/// over-weighting. The value can be tuned in the future with data that shows
/// improvements.
const MAX_LIFO_POLLS_PER_TICK: usize = 3;

Esta esla segunda línea de defensa: incluso si el presupuesto no se ha agotado, el LIFO slot se deshabilita tras ser priorizado 3 veces consecutivas, y las tareas posteriores van a la cola normal. El presupuesto gestiona el "total de operaciones de recursos", el limitador LIFO gestiona el "número de veces que el mismo par de tareas se despierta mutuamente", ambos son complementarios.

El ámbito de presupuesto tiene una excepción importante enblock_in_place.block_in_placetransfiere el worker core a otro hilo, y el hilo actual entra en estado de bloqueo. El código bloqueante no está sujeto al presupuesto, por lo que debepausarel presupuesto:

📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:406-417

if had_entered {
    // Unset the current task's budget. Blocking sections are not
    // constrained by task budgets.
    let _reset = Reset {
        take_core,
        budget: coop::stop(),
    };

    crate::runtime::context::exit_runtime(f)
} else {
    f()
}

coop::stop()devuelve el presupuesto actual y lo establece enNone(es decir, "fuera del runtime"),ResetelDropde se restaura después de que termina el bloqueo:

📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:374-397

impl Drop for Reset {
    fn drop(&mut self) {
        with_current(|maybe_cx| {
            if let Some(cx) = maybe_cx {
                if self.take_core {
                    let core = cx.worker.core.take();
                    // ...
                    *cx_core = core;
                }

                // Reset the task budget as we are re-entering the
                // runtime.
                coop::set(self.budget);
            }
        });
    }
}

coop::set(self.budget)restaura el presupuesto previamentestop()guardado. De esta forma,block_in_placeel código de bloqueo síncrono dentro de no consume presupuesto, ni dispara falsamente una cesión por agotamiento del presupuesto; después de que termina el bloqueo, la tarea continúa ejecutándose con su presupuesto restante original.

La siguiente figura muestra el flujo de control completo desde que una tarea es programada hasta que cede por agotamiento del presupuesto:

flowchart TD
    start["Context::run 主循环"] --> next["core.next_task()"]
    next --> has_task{"有本地任务?"}
    has_task -->|是| run_task["run_task(task, core)"]
    has_task -->|否| steal["core.steal_work()"]
    steal --> stolen{"窃取到任务?"}
    stolen -->|是| run_task
    stolen -->|否| defer_check{"defer 队列非空?"}
    defer_check -->|是| park_yield["park_yield: 驱动 IO/timer 后唤醒"]
    defer_check -->|否| park["park: 阻塞等待"]
    park_yield --> start
    park --> start

    run_task --> budget["coop::budget 建立满额预算"]
    budget --> poll["task.run() 轮询"]
    poll --> lifo_check{"lifo_slot 有任务?"}
    lifo_check -->|否| done["返回 ControlFlow::Continue"]
    lifo_check -->|是| budget_rem{"coop::has_budget_remaining()?"}
    budget_rem -->|否| push_back["push_back_or_overflow 推回队列"]
    push_back --> done
    budget_rem -->|是| lifo_limit{"lifo_polls >= 3?"}
    lifo_limit -->|是| disable["core.lifo_enabled = false"]
    lifo_limit -->|否| poll_lifo["task.run() 轮询 LIFO 任务"]
    disable --> poll_lifo
    poll_lifo --> lifo_check
    done --> start

En la figura se pueden ver dos rutas de cesión: cuando el presupuesto se agota, se empuja la tarea LIFO de vuelta a la cola (push_back_or_overflow), y cuando se excede el límite de priorización consecutiva del LIFO, se deshabilita el LIFO slot. Ambas vuelven al bucle principal, dando al worker la oportunidad de procesar otras tareas o el driver.

12.4 Reflexiones de diseño, recuperación de errores y trampas en producción

¿Por qué usar TLS en lugar de paso explícito de parámetros?Los puntos de verificación de presupuesto están dispersos en lo profundo de módulos como channel, I/O, time, etc. Si se pasaran explícitamente como parámetros, cada API necesitaría unBudgetparámetro adicional, contaminando toda la interfaz pública. TLS hace que el presupuesto sea completamente transparente para el código de negocio, a costa de un acceso TLS por cada verificación. Tokio usa#[thread_local]o TLS rápido específico de plataforma para reducir esta sobrecarga.

Interacción entre agotamiento de presupuesto y seguridad ante cancelación.Cuando el agotamiento del presupuesto hace quereserve_innerdevuelvaPending, la tarea puede estar en alguna rama deselect!. Si en ese momento otra rama está lista,select!cancela la rama actual——reserve_innerelWakeReceiverOnDropguard de verifica en el drop si "el semáforo está cerrado y ocioso" y despierta al receptor:

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

struct WakeReceiverOnDrop<'a, T> {
    chan: &'a chan::Tx<T, Semaphore>,
}

impl<T> Drop for WakeReceiverOnDrop<'_, T> {
    fn drop(&mut self) {
        use chan::Semaphore;

        let semaphore = self.chan.semaphore();
        if semaphore.is_closed() && semaphore.is_idle() {
            self.chan.wake_rx();
        }
    }
}

La existencia de este guard indica que: elPendingdisparado por el presupuesto y elPendingreal de "sin permiso"

deben comportarse de manera consistente en la ruta de cancelación, de lo contrario el receptor podría nunca recibir la notificación de "channel cerrado".Trampa en producción: latencia oculta causada por agotamiento del presupuesto.spawnUn fenómeno común es: cierta tarea de repente se vuelve más lenta procesando mensajes, pero el uso de CPU no es alto. Al investigar, es fácil sospechar de contención de locks o I/O, pero en realidad puede ser que la tarea procesó más de 128 mensajes en un solo poll, disparando la cesión por presupuesto, y cada cesión requiere un ciclo completo de "empujar de vuelta a la cola → reprogramar → sondeo del driver". Si el procesamiento de mensajes en sí es rápido, esta sobrecarga de programación puede representar una proporción alta. La solución es dividir el procesamiento por lotes grandes en múltiplesyield_now。

tareas, o insertar explícitamenteblock_in_placeen el bucle. Frontera entre presupuesto y. Anteriormente vimos queblock_in_placehacecoop::stop()pausar el presupuesto. Pero hay que tener en cuenta:coop::stop()solo se llama cuandohad_enteredes verdadero, es decir, solo pausa cuando realmente está en un hilo worker del runtime. Siblock_in_placese llama fuera del runtime,f()se ejecuta directamente y el estado del presupuesto no cambia. Esta decisión de rama se realiza enmaybe_move_runtime:

📎 tokio/src/runtime/scheduler/multi_thread/worker.rs:424-464

with_current(|maybe_cx| {
    match (
        crate::runtime::context::current_enter_context(),
        maybe_cx.is_some(),
    ) {
        (context::EnterRuntime::Entered { .. }, true) => {
            had_entered = true;
        }
        (
            context::EnterRuntime::Entered {
                allow_block_in_place,
            },
            false,
        ) => {
            if allow_block_in_place {
                had_entered = true;
                return Ok(());
            } else {
                return Err(
                    "can call blocking only when running on the multi-threaded runtime",
                );
            }
        }
        (context::EnterRuntime::NotEntered, true) => {
            return Ok(());
        }
        (context::EnterRuntime::NotEntered, false) => {
            return Ok(());
        }
    }
    // ...
})

Las cuatro combinaciones corresponden a: dentro del hilo worker,block_onentrada del thread pool deblock_in_place, anidado

Resumen del capítulo

El mecanismo coop resuelve el problema de equidad del planificador no apropiativo con un diseño de tres capas:

1. Portador del presupuesto：coop::Budgetexiste en el TLS,Optionla capa externa distingue dentro y fuera del runtime,coop::budgetestablece un ámbito de presupuesto completo,coop::stop/coop::setsoporta pausa y reanudación (block_in_placeescenario).

2. Puntos de consumo: operaciones de recursos (envío/recepción por channel, I/O,yield_now) mediantecoop::poll_proceeddecrementan el presupuesto; al agotarse, disfrazan el «ceder el turno» comoPending, de forma transparente para el negocio.

3. Ruta de cesión：yield_nowmediantecontext::deferentrega el Waker a la cola defer, asegurando que la reprogramación ocurra solo después de que el driver haya hecho polling; las tareas del LIFO slot comparten el presupuesto de la tarea padre y tienenMAX_LIFO_POLLS_PER_TICK = 3de limitación independiente.

La idea clave de este mecanismo es:la equidad no requiere apropiación; solo necesita que el «bucle infinito» se interrumpa naturalmente tras un número finito de pasos. El presupuesto es la medida de ese «número finito de pasos».

Reflexiones y autoevaluación de este capítulo

en lugar del presupuesto original de la capa externa), provocando que el presupuesto se pierda permanentemente y que todas las operaciones de recursos posteriores de la tarea queden sin restricción.

En el capítulo anterior desglosamos el presupuesto de cooperación coop: cada tarea solo tiene un presupuesto limitado dentro de un ciclo de planificación y, una vez agotado, debe ceder, evitando así que una sola tarea mate de hambre a las demás. Pero el mecanismo de presupuesto solo resuelve el problema de la "planificación justa"; en entornos de producción reales hay otra clase de trampas más sutiles: seguridad ante cancelación, propagación de panic y orden de cierre. Cuando select! cancela un Future, cuando un panic de una tarea es capturado, cuando el Runtime comienza a cerrarse, el comportamiento límite del código suele ser contrario a la intuición. Este capítulo comienza con la seguridad ante cancelación y primero examina qué se pierde realmente cuando un Future es drop.

13.2 Propagación de panic: cómo JoinError captura un colapso

Modelo intuitivo

El panic de una tarea de Tokio no hace que todo el proceso colapse (salvo con panic=abort), sino que es capturado, empaquetado comoJoinError, y devuelto medianteJoinHandle::await. Es como cuando ocurre un accidente en una estación de una línea de montaje de una fábrica: la red de seguridad atrapa al trabajador, pero el producto se descarta; lo que obtienes es un "informe de accidente" en lugar del producto.

Estructura de datos y estados

JoinHandle<T>ElFuture::Outputdesuper::Result<T>esResult<T, JoinError> 📎 tokio/src/runtime/task/join.rs:325。JoinError, es decir,

let join_handle = tokio::spawn(async { panic!("boom"); });
let err = join_handle.await.unwrap_err();
assert!(err.is_panic());

📎 tokio/src/runtime/task/join.rs:121-127

CopiarRawTaskEl mecanismo por el cual se captura el panic está en la ruta de poll decatch_unwind: al hacer poll de la tarea se envuelve conJoinHandle::poll, y tras ocurrir el panic se guarda el payload en la ranura de salida de la tarea, se marca el estado como complete y luego se despierta al join waker.try_read_outputLo que se lee medianteErr(JoinError::panic(payload))。

es

sequenceDiagram
    participant App as 应用任务
    participant Worker as Worker 线程
    participant Raw as RawTask
    participant JH as JoinHandle

    App->>Worker: spawn(async { panic!("boom") })
    Worker->>Raw: poll 任务 Future
    Raw->>Raw: catch_unwind 捕获 panic
    Raw->>Raw: 存储 panic payload 到输出槽
    Raw->>Raw: state 标记 complete
    Raw->>JH: 唤醒 join waker
    JH->>App: await 返回 Err(JoinError::panic)

CopiarJoinErrorPunto clave: el payload del panic se conserva por completo,std::error::Errorimplementainto_panic(), se puede recuperarBox<dyn Any + Send>mediantedowncast_ref::<&str>(), y luego extraer el mensaje del panic con

Reflexiones de diseño y trampas

Trampa 1:JoinHandleElUnwindSafede

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

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

CopiarT: UnwindSafeEsta es una implementación incondicional, no requiereJoinHandle. Razón:T，Ten sí no contienecatch_unwindEn la asignación de tareas en el heap, durante el panic ya ha sido aislado porT. Por lo tanto, incluso siUnwindSafe，JoinHandleno es

también es seguro.Trampa 2: el panic no se propaga automáticamente a la tarea padre.JoinHandleSi la tarea A hace spawn de la tarea B y B entra en panic, A no recibe notificación automáticamente, a menos que A haya hecho await del

de B. Si A no hizo await, el panic de B se traga silenciosamente. Esta es una de las fuentes de bugs más sutiles en entornos de producción.spawn_blockingTrampa 3:El panic decatch_unwindtambién se captura.MutexLos workers del pool de hilos bloqueantes también envuelven las tareas constd::sync::Mutex, y tras el panic el hilo no muere, sino que vuelve al pool para seguir aceptando trabajo. Pero si dentro de una tarea bloqueante mantienes

y no lo liberas durante el panic, se produce envenenamiento de lock; este es el comportamiento inherente de, Tokio no interviene.catch_unwindTrampa 4: panic durante el drop del Runtime.

Si una tarea entra en panic durante el proceso de drop del Runtime,

sigue teniendo efecto, pero en ese momento el join waker puede haber dejado de ser válido y el payload del panic se descarta. Este es un subconjunto del problema del orden de cierre, que se desarrolla en la siguiente sección.

13.3 Orden de cierre: limpieza de hilos bloqueantes y recursos de I/O

Modelo intuitivo

RuntimeEl cierre del Runtime es como el cierre de un restaurante: primero se hace que la recepción deje de aceptar clientes (dejar de aceptar nuevas tareas), luego se espera a que la cocina termine los platos en curso (las tareas asíncronas llegan al siguiente punto de yield), y finalmente se espera a que los ayudantes subcontratados terminen su turno (los hilos bloqueantes retornan). Si el orden es incorrecto surgen problemas; por ejemplo, si primero se despide a los ayudantes, los platos de la cocina nunca terminarán de prepararse.

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

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

DropLos tres campos de

impl Drop for Runtime {
    fn drop(&mut self) {
        match &mut self.scheduler {
            Scheduler::CurrentThread(current_thread) => {
                let _guard = context::try_set_current(&self.handle.inner);
                current_thread.shutdown(&self.handle.inner);
            }
            Scheduler::MultiThread(multi_thread) => {
                multi_thread.shutdown(&self.handle.inner);
            }
        }
    }
}

📎 tokio/src/runtime/runtime.rs:506-521

CopiarDropImplementación descheduler，:blocking_pool。blocking_poolCopiarDropNota:Runtime::dropsolo manejascheduler → handle → blocking_poolno maneja explícitamente

El cierre deshutdown_timeoutocurre en su propio

pub fn shutdown_timeout(mut self, duration: Duration) {
    self.handle.inner.shutdown();
    self.blocking_pool.shutdown(Some(duration));
}

📎 tokio/src/runtime/runtime.rs:457-461

se dispara por el orden de drop de los campos. El orden de drop de los campos es el orden de declaración:handle.inner.shutdown(). Por lo tanto, el pool bloqueante es el último en cerrarse.blocking_pool.shutdown(Some(duration))Peroduration。

controla explícitamente el orden:

blocking/shutdown.rsCopiar

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

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

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

notifica al planificador y al driver de I/O que se detengan, luegoSenderespera las tareas bloqueantes, como máximoArc<oneshot::Sender>Mecanismo subyacente del cierre del pool bloqueanteSenderutiliza un ingenioso oneshot channel:ReceiverCopiarwaitCada worker bloqueante mantiene un clon de

pub(crate) fn wait(&mut self, timeout: Option<Duration>) -> bool {
    use crate::runtime::context::try_enter_blocking_region;

    if timeout == Some(Duration::from_nanos(0)) {
        return false;
    }

    let mut e = match try_enter_blocking_region() {
        Some(enter) => enter,
        _ => {
            if std::thread::panicking() {
                return false;
            } else {
                panic!(
                    "Cannot drop a runtime in a context where blocking is not allowed. \
                    This happens when a runtime is dropped from within an asynchronous context."
                );
            }
        }
    };

    if let Some(timeout) = timeout {
        e.block_on_timeout(&mut self.rx, timeout).is_ok()
    } else {
        let _ = e.block_on(&mut self.rx);
        true
    }
}

📎 tokio/src/runtime/blocking/shutdown.rs:37-70

). Cuando todos los workers salen y todos los

1. timeout == Some(0)son drop,shutdown_backgroundrecibe la notificación.

2. try_enter_blocking_region()MétodoNone。

:

Copiarblock_on_timeoutAnálisis paso a paso:

devuelve directamente false; esta es la ruta de

flowchart TD
    start["Runtime::drop 或 shutdown_timeout"] --> sched{"scheduler 类型?"}
    sched -->|CurrentThread| ct["try_set_current + current_thread.shutdown"]
    sched -->|MultiThread| mt["multi_thread.shutdown"]
    ct --> handle_drop["handle 字段 drop"]
    mt --> handle_drop
    handle_drop --> bp_drop["blocking_pool 字段 drop"]
    bp_drop --> bp_wait{"shutdown_timeout 已调用?"}
    bp_wait -->|是| explicit["blocking_pool.shutdown(Some(duration))"]
    bp_wait -->|否| implicit["BlockingPool::drop 默认等待"]
    explicit --> wait_check{"try_enter_blocking_region 成功?"}
    implicit --> wait_check
    wait_check -->|否且在 panic| skip["返回 false 不等待"]
    wait_check -->|否且不在 panic| panic_err["panic: Cannot drop a runtime in async context"]
    wait_check -->|是| block_on["block_on 等待所有 Sender drop"]

intenta entrar en la región bloqueante. Si actualmente está en un contexto asíncrono (por ejemplo, hacer drop del Runtime dentro de una tarea async), devuelve

3. Cuando falla la entrada, si se está en panic, devuelve false (no entrar en panic de nuevo durante un panic); de lo contrario, entra en panic y da un mensaje de error claro.El mensaje de error es muy claro: «Cannot drop a runtime in a context where blocking is not allowed»📎 tokio/src/runtime/blocking/shutdown.rs:51-54. La solución es usarshutdown_background(), que es equivalente ashutdown_timeout(Duration::from_nanos(0)) 📎 tokio/src/runtime/runtime.rs:494-496, sin esperar a las tareas bloqueantes.

Trampa 2:shutdown_backgroundfiltrará tareas bloqueantes.La documentación advierte explícitamente «this may result in a resource leak (in that any blocking tasks are still running until they return)»📎 tokio/src/runtime/runtime.rs:470-472. Las tareas bloqueantes seguirán ejecutándose hasta que retornen de forma natural, pero el Runtime ya se ha destruido, y los recursos que poseen pueden haber quedado invalidados.

Trampa 3: Los recursos de E/S quedan invalidados tras destruir el Runtime.La documentación indica «Once the runtime has been dropped, any outstanding I/O resources bound to it will no longer function»📎 tokio/src/runtime/runtime.rs:52-54。is_rt_shutdown_errLa función sirve para detectar este tipo de errores📎 tokio/src/runtime/runtime.rs:585-593。

Trampa 4:Dropespera indefinidamente por defecto.La documentación señala «TheDrop implementation waits forever for this」📎 tokio/src/runtime/runtime.rs:43-44. Si una tarea bloqueante se queda atascada (por ejemplo, en un bucle infinito), destruir el Runtime se colgará permanentemente. En producción se debería usarshutdown_timeoutpara establecer un límite.

13.4 Manejo de señales y conflictos entre múltiples Runtimes

Modelo intuitivo

Las señales Unix son a nivel de proceso, pero elSignalde Tokio está vinculado al Runtime. Es como si todo el edificio compartiera una alarma de incendios, pero cada habitación tuviera su propio receptor: la primera persona que instala un receptor cambia el cableado de la alarma, y los demás solo pueden compartir ese cambio.

Estructuras de datos y estado global

signal_enablees el punto de entrada para registrar manejadores de señales:

fn signal_enable(signal: SignalKind, handle: &Handle) -> io::Result<()> {
    let signal = signal.0;
    if signal <= 0 || signal_hook_registry::FORBIDDEN.contains(&signal) {
        return Err(Error::other(format!(
            "Refusing to register signal {signal}"
        )));
    }

    handle.check_inner()?;

    let globals = globals();
    let siginfo = match globals.storage().get(signal as EventId) {
        Some(slot) => slot,
        None => return Err(io::Error::other("signal too large")),
    };

    siginfo
        .init
        .get_or_init(|| {
            unsafe { signal_hook_registry::register(signal, move || action(globals, signal)) }
                .map(|_| ())
                .map_err(|e| e.raw_os_error())
        })
        .map_err(|e| {
            e.map_or_else(
                || Error::other("registering signal handler failed"),
                || Error::from_raw_os_error,
            )
        })
}

📎 tokio/src/signal/unix.rs:266-296

Puntos clave:

1. signal <= 0 || FORBIDDEN.contains(&signal)rechaza señales inválidas.

2. handle.check_inner()verifica si el controlador de señales está en ejecución; si el Runtime ya se ha cerrado, aquí fallará.

3. siginfo.init.get_or_init(...)usaOnceLockpara garantizar que cada señal registre un único manejador del SO.get_or_initEl cierre designal_hook_registry::registerinvoca

, que es un registro global a nivel de proceso.action(globals, signal)4. El manejador registrado esglobals.record_event(signal), que hace dos cosas:📎 tokio/src/signal/unix.rs:252-259。

registra el evento y luego escribe un byte en el pipe para despertar al controlador

globals()La raíz del conflicto entre múltiples RuntimesGlobals，OsExtraDatalo que devuelve es unUnixStreamglobal a nivel de proceso

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

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

DefaultEl par también es global:UnixStream 📎 tokio/src/signal/unix.rs:61-64Copiar

La implementación crea un par designal_enable. Este pipe es único a nivel global, y todos los controladores de señales de los Runtimes lo comparten.handle.check_inner()El problema surge:dentro delo que verifica es elsignal_hook_registry::registercontrolador de señales delRuntime actual. Pero el manejador que registraesa nivel de procesoget_or_init, y escribe en elOk(())pipe

global. Si el Runtime A registra primero SIGINT, y luego el Runtime B también registra SIGINT,

sequenceDiagram
    participant OS as 操作系统
    participant Handler as 全局 signal handler
    participant Pipe as 全局 UnixStream pipe
    participant RtA as Runtime A 信号驱动
    participant RtB as Runtime B 信号驱动

    Note over RtA: signal(SIGINT) 注册
    RtA->>Handler: signal_hook_registry::register(SIGINT, action)
    Note over RtB: signal(SIGINT) 注册
    RtB->>Handler: get_or_init 返回已有 Ok，不重复注册
    OS->>Handler: 投递 SIGINT
    Handler->>Pipe: write(&[1])
    Pipe->>RtA: 可读事件
    Pipe->>RtB: 可读事件
    Note over RtA,RtB: 两个 Runtime 竞争读取，只有一个能读到字节

existente, sin volver a registrarlo. Pero el controlador de señales del Runtime B leerá datos del pipe global: ambos Runtimes competirán por los bytes del mismo pipe.

Walkthrough guiado por escenarios: competencia de señales entre múltiples RuntimesCopiar📎 tokio/src/signal/unix.rs:379-380Reflexiones de diseño y trampasSignalTrampa 1: El manejador de señales nunca se desinstala.📎 tokio/src/signal/unix.rs:338-340。

La documentación advierte explícitamente «Once a signal handler is registered with the process the underlying libc signal handler is never unregistered». Incluso si la instancia depoll is called, all signal notifications are coalesced into one item returned from poll」📎 tokio/src/signal/unix.rs:312-315se destruye, las señales posteriores seguirán siendo capturadas por Tokio, y el comportamiento por defecto no se restaurará

Trampa 2: Las señales se fusionan.La documentación indica «beforesignal_hook. Si recibes 10 SIGINT pero solo haces poll una vez, solo verás un evento. Esta es una característica de las propias señales Unix (las señales estándar no se encolan), Tokio no fusiona adicionalmente.

Trampa 3: Las señales pueden perderse con múltiples Runtimes.signalDado que el pipe global es leído de forma competitiva por múltiples Runtimes, un Runtime puede consumir el byte mientras otro nunca lo recibe. En producción se debería manejar las señales en un solo Runtime, o usarpara gestionarlas por cuenta propia.rt feature flag is not enabled」📎 tokio/src/signal/unix.rs:398-405Trampa 4:signal()Condiciones de panic de la función.

La documentación indica «This function panics if there is no current reactor set, or if therecv(). Llamar afuera del Runtime provocará un panic.tokio::select! and another branch completes first, then it is guaranteed that no signal is lost」📎 tokio/src/signal/unix.rs:423-427Trampa 5:EventInfoLa seguridad de cancelación derecv()La documentación garantiza «This method is cancel safe. If you use it as a branch in

. Esto se debe a que los eventos de señal se almacenan en el

global,solo lee, no consume el estado subyacente.。

• JoinHandleReflexiones de diseño
• Los tres temas de este capítulo comparten un patrón subyacente:Handle, el orden incorrecto provocará interbloqueo o panic.
• Señales con múltiples Runtime en conflicto, porque el handler y el pipe son estado global a nivel de proceso, mientras queSignales una vista a nivel de Runtime.

Después de entender este patrón, la lista de trampas a evitar se puede resumir en tres principios:

1. Cancelación segura = el estado está fuera del Future.Si el Future tiene un búfer interno, el drop perderá datos.JoinHandle、Signal::recv、tokio::sync::mpsc::Receiver::recvTodos cumplen esta condición.

2. Orden de cierre = orden inverso a la dirección de dependencia.Quien depende de quién, primero se cierra el dependido. El planificador depende del driver de I/O, así que primero se cierra el planificador; el pool de bloqueo es independiente, se cierra al final.

3. Estado global = conflicto entre múltiples instancias.Cualquier recurso a nivel de proceso (signal handler, pipe, tabla de descriptores de archivo) entrará en conflicto con múltiples Runtime; o se limita a un solo Runtime, o se usa sincronización externa.

Resumen del capítulo

Reflexión y autoevaluación del capítulo

todavía pueda recibir la señal, pero depende de si Runtime B tiene un driver de señales en ejecución. Si Runtime B no tiene driver de señales (por ejemplo, no habilitó la feature signal o el driver ya está cerrado), nadie leerá los datos del pipe,

y nunca llegará el despertar. Esta es la fragilidad del manejo de señales con múltiples Runtime.catch_unwindTransición al final del capítuloGlobalsCancelación segura, propagación de panic, orden de cierre, conflicto de señales: la raíz común de estos cuatro problemas es la ambigüedad de la «propiedad del estado» en los límites asíncronos. Tokio, al poner el estado en el heap, gestionar el ciclo de vida con conteo de referencias, aislar el panic con

, y compartir el estado de señales con

Con esto, hemos recorrido los terrenos limítrofes más propensos a errores en entornos de producción de Tokio: la atomicidad de try_read_output y el almacenamiento en el heap de las salidas de dependencias seguras ante cancelación; JoinHandle::drop no cancela la tarea, solo abort la cancela realmente, pero no tiene efecto sobre spawn_blocking; un panic capturado por catch_unwind se empaqueta como JoinError y se pierde silenciosamente si no se hace await; el cierre del Runtime tiene un orden estricto, y hacer drop en un contexto async provoca panic; los manejadores de señales son estado global a nivel de proceso y nunca se desregistran una vez registrados. Detrás de estas reglas está la reiterada ponderación de Tokio entre corrección y rendimiento. En el próximo capítulo dejaremos atrás los mecanismos concretos para revisar, desde una perspectiva arquitectónica, el origen de estas ponderaciones y vislumbrar hacia dónde llevarán a Tokio io_uring, la refactorización de drivers y la interfaz de ejecutores personalizados.

En el capítulo anterior revisamos cuatro tipos de trampas de producción: seguridad ante cancelación, propagación de panic, orden de cierre y conflictos de señales. Aunque parecen dispersas, todas apuntan al mismo problema arquitectónico: cómo se divide claramente la propiedad del estado en los límites asíncronos. Y la forma de dividir la propiedad está determinada precisamente por las tres decisiones arquitectónicas más fundamentales del runtime: cómo se planifican las tareas, cómo se distribuyen los eventos de E/S y cómo se verifica la corrección de la concurrencia. Este capítulo ya no se sumerge en los detalles de implementación de una función concreta, sino que se sitúa en una perspectiva arquitectónica para revisar las concesiones que Tokio hizo en estas decisiones y, siguiendo las pistas de evolución ya sembradas en la documentación oficial y el código fuente, ver hacia dónde llevarán a Tokio io_uring, la refactorización de drivers y la interfaz de ejecutores personalizados. Al terminar este capítulo, deberías poder responder una pregunta práctica: cuándo conviene extender Tokio y cuándo conviene evitarlo.

I. Tres ponderaciones históricas: por qué es como es ahora

Modelo intuitivo

Imagina Tokio como un restaurante que lleva diez años abierto. La forma de organizar los turnos en la cocina (work-stealing), la plantilla independiente de los camareros (separación entre el driver de E/S y el planificador) y el sistema de inspección higiénica de la cocina (verificación de concurrencia con loom) no se diseñaron el primer día, sino que evolucionaron gradualmente a medida que «llegaban más clientes y los platos se volvían más complejos». Entender estas evoluciones permite juzgar qué diseños son apuestas visionarias y qué diseños son lastre histórico.

Ponderación uno: work-stealing en lugar de cola global

La concesión de work-stealing es: cada worker mantiene una cola local,spawnal hacer push prioriza la cola local (sin locks, amigable con la caché), y solo cuando la local está vacía roba desde la cola de otro worker. El costo es que el balanceo de carga tiene latencia y que el robo en sí requiere operaciones atómicas y barreras de memoria. Tokio eligió lo segundo porque los servidores modernos tienen decenas de núcleos y el costo de la contención de locks es mucho mayor que el gasto ocasional de robo.

Ponderación dos: el driver de E/S es independiente del planificador

Este es el punto más interesante del material fuente de este capítulo. Observa la estructura de módulos detokio/src/runtime/io/mod.rs:

📎 tokio/src/runtime/io/mod.rs:5-22

mod driver;
use driver::{Direction, Tick};
pub(crate) use driver::{Driver, Handle, ReadyEvent};

mod registration;
pub(crate) use registration::Registration;

mod registration_set;
use registration_set::RegistrationSet;

mod scheduled_io;
use scheduled_io::ScheduledIo;

mod metrics;
use metrics::IoDriverMetrics;

use crate::util::ptr_expose::PtrExposeDomain;
static EXPOSE_IO: PtrExposeDomain<ScheduledIo> = PtrExposeDomain::new();

Fíjate en quedriver、registration、scheduled_ioson tres módulos independientes y que hacia fuera solo exponen los tiposDriver、Handle、ReadyEvent、Registration.ScheduledIoespub(crate)dePtrExposeDomain: está envuelto por

el runtime de un solo hilo también necesita un driver de E/S, pero no necesita un planificador work-stealing: la separación permite que ambos runtimes reutilicen la misma implementación de E/S.

tokio/src/loom/mod.rsPonderación tres: loom para verificar el modelo de concurrencia

📎 tokio/src/loom/mod.rs:1-14

//! This module abstracts over `loom` and `std::sync` depending on whether we
//! are running tests or not.

#![allow(unused)]

#[cfg(not(all(test, loom)))]
mod std;
#[cfg(not(all(test, loom)))]
pub(crate) use self::std::*;

#[cfg(all(test, loom))]
mod mocked;
#[cfg(all(test, loom))]
pub(crate) use self::mocked::*;

Copiar#[cfg(all(test, loom))]La clave está en la condicióntest: solo cuando se activan simultáneamente los dos cfgloomymockedse reemplaza el módulostdpor

Reflexiones de diseño

Estos tres trade-offs comparten una característica común:Todos eligieron la solución «más compleja pero más escalable», y limitaron la complejidad al interior. La complejidad de work-stealing está oculta en el planificador, la complejidad del I/O dirigido por eventos está oculta enScheduledIo, y la complejidad de loom está oculta en las condiciones cfg. La API expuesta al exterior siempre esspawn、TcpStream::readestas interfaces simples.

---

, significa que te has atado a la implementación interna de Tokio, y pagarás un precio al actualizar.

II. Refactorización del driver: de «un waker, una dirección» a «conjunto de intereses arbitrario»

Modelo intuitivoasync fn read(&mut self)Los primeros tipos de I/O de Tokio tenían una restricción rígida:&mut selfrequeríatokio/docs/reactor-refactor.md. Esto es como un restaurante con una sola ventanilla de recogida, donde solo una persona puede hacer fila a la vez — porque el waker se almacenaba dentro del recurso de I/O, no en el Future correspondiente a la operación.

documenta completamente la causa de esta restricción y el plan de refactorización.

Los puntos débiles de la arquitectura antigua

📎 tokio/docs/reactor-refactor.md:16-20

Currently, I/O types require `&mut self` for `async` functions. The reason for
this is the task's waker is stored in the I/O resource's internal state
(`ScheduledIo`) instead of in the future returned by the `async` function.
Because of this limitation, I/O types limit the number of wakers to one per
direction (a direction is either read-related events or write-related events).

— no es una preferencia de diseño de API, sino una restricción directa de la estructura de datos interna.

Nueva arquitectura: mover el waker al Future

📎 tokio/docs/reactor-refactor.md:22-25

Moving the waker from the internal I/O resource's state to the operation's
future enables multiple wakers to be registered per operation. The "intrusive
wake list" strategy used by `Notify` applies to this case, though there are some
concerns unique to the I/O driver.

CopiarScheduledIoLa nueva estructura

📎 tokio/docs/reactor-refactor.md:97-134

#[derive(Debug)]
pub(crate) struct ScheduledIo {
    /// Resource's known state packed with other state that must be
    /// atomically updated.
    readiness: AtomicUsize,

    /// Tracks tasks waiting on the resource
    waiters: Mutex<Waiters>,
}

#[derive(Debug)]
struct Waiters {
    // List of intrusive waiters.
    list: LinkedList<Waiter>,

    /// Waiter used by `AsyncRead` implementations.
    reader: Option<Waker>,

    /// Waiter used by `AsyncWrite` implementations.
    writer: Option<Waker>,
}

// This struct is contained by the **future** returned by `readiness()`.
#[derive(Debug)]
struct Waiter {
    /// Intrusive linked-list pointers
    pointers: linked_list::Pointers<Waiter>,

    /// Waker for task waiting on I/O resource
    waiter: Option<Waker>,

    /// Readiness events being waited on. This is
    /// the value passed to `readiness()`
    interest: mio::Ready,

    /// Should not be `Unpin`.
    _p: PhantomPinned,
}

Copiar

Aquí hay varios puntos de diseño ingeniosos que vale la pena desarrollar:readinessPrimero,AtomicUsize，waitersesMutex<Waiters>。esreadiness¿Por qué no usar un solo lock para proteger ambos? Porque las operaciones de lectura dereadiness()son extremadamente frecuentes (cada llamada a

debe verificarlo), mientras que las operaciones de escritura solo ocurren al recibir eventos de mio. Usar variables atómicas para hacer que la ruta de lectura sea lock-free es una optimización típica de separación lectura/escritura.WaiterSegundo, pointers: linked_list::Pointers<Waiter>es un nodo de lista enlazada intrusiva.Waiterhace que_p: PhantomPinnedsea parte de la lista enlazada, sin necesidad de asignar nodos adicionales.Unpinlo marca explícitamente como no

— porque una vez que la dirección de un nodo de una lista enlazada intrusiva se mueve, la lista se rompe.readerTercero,writeryOption<Waker>dosAsyncRead/AsyncWriteson para. El documento explica la razón:

📎 tokio/docs/reactor-refactor.md:210-213

The `AsyncRead` and `AsyncWrite` traits use a "poll" based API. This means that
it is not possible to use an intrusive linked list to track the waker.
Additionally, there is no future associated with the operation which means it is
not possible to cancel interest in the readiness events.

la ruta

usa ranuras fijas (no soporta cancelación, pero es compatible con el trait). Esta «coexistencia de dos mecanismos» es el costo típico de una refactorización incremental.

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

If care is not taken, if between `mio_socket.read(buf)` returning and
`clear_readiness(event)` is called, a readiness event arrives, the `read()`
function could deadlock. This happens because the readiness event is received,
`clear_readiness()` unsets the readiness event, and on the next iteration,
`readiness().await` will block forever as a new readiness event is not received.

El problema más espinoso de la refactorización son las condiciones de carrera. El documento da un escenario concreto de deadlock:readinessCopiarAtomicUsizeLa solución es introducir el mecanismo de tick, dividiendo

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

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

lleva el tick del momento de la lectura.readiness()solo limpia el estado de readiness cuando el tick coincide — si el tick no coincide, significa que llegaron nuevos eventos durante ese tiempo y no se puede limpiar. Así, la condición de carrera entre «limpiar» y «llegada de nuevos eventos» se resuelve dentro de una única lectura-modificación-escritura atómica.clear_readiness()El siguiente diagrama de flujo describe la ruta de decisión entre

flowchart TD
    start["readiness(interest).await"] --> check_ready{"已知 readiness<br/>与 interest 有交集?"}
    check_ready -->|是| ret_event["返回 ReadyEvent<br/>携带当前 tick"]
    check_ready -->|否| wait["注册 Waiter 到<br/>ScheduledIo.waiters"]
    wait --> mio_poll["mio.poll() 收到事件<br/>tick 递增"]
    mio_poll --> notify["遍历 waiters<br/>interest 匹配者唤醒"]
    notify --> ret_event
    ret_event --> do_read["mio_socket.read(buf)"]
    do_read --> read_ok{"read 结果?"}
    read_ok -->|Ok| done["返回 Ok(v)"]
    read_ok -->|WouldBlock| clear["clear_readiness(event)"]
    read_ok -->|其他 Err| err["返回 Err(e)"]
    clear --> tick_match{"event.tick ==<br/>当前 readiness.tick?"}
    tick_match -->|是| clear_ok["清除 readiness 位"]
    tick_match -->|否| skip["跳过清除<br/>保留新事件"]
    clear_ok --> start
    skip --> start

:tick_matchCopiarclear_readinessLa rama clave de este diagrama está enreadiness(): si el tick no coincide,

debe abandonar la limpieza, de lo contrario perdería el evento recién llegado, provocando que la siguiente ronda de

se bloquee permanentemente.readiness()Cancelación de interés y fuga de memoria

📎 tokio/docs/reactor-refactor.md:144-148

The future returned by `readiness()` uses an intrusive linked list to store the
waker with `ScheduledIo`. Because `readiness()` can be called concurrently, many
wakers may be stored simultaneously in the list. If the `readiness()` future is
dropped early, it is essential that the waker is removed from the list. This
prevents leaking memory.

debe removerse a sí mismo de la lista en la implementación de

, de lo contrario el nodo permanecerá para siempre enVec<Waker>, fugando memoria y además siendo despertado erróneamente la próxima vez que llegue un evento.Reflexiones de diseño y trampas en producción&Resource¿Por qué no usar

📎 tokio/docs/reactor-refactor.md:228-233

It is only possible to implement `AsyncRead` and `AsyncWrite` for resource types
themselves and not for `&Resource`. Implementing the traits for `&Resource`
would permit concurrent operations to the resource. Because only a single waker
is stored per direction, any concurrent usage would result in deadlocks. An
alternate implementation would call for a `Vec<Waker>` but this would result in
memory leaks.

Copiar：TcpStream::by_ref()〔Inferencias de diseño y trade-offs arquitectónicos〕TcpStreamRefEl problema deread_waiteres que, tras descartar el Future, el waker correspondiente queda en el Vec sin poder localizarse para eliminarlo, y solo se descubre «este waker ya no es válido» cuando llega el siguiente evento. La lista enlazada intrusiva hace que la dirección del nodo sea la dirección del campo interno del Future, permitiendo una remoción precisa al hacer drop.write_waiterPuntos problemáticos en producción

📎 tokio/docs/reactor-refactor.md:238-244

struct TcpStreamRef<'a> {
    stream: &'a TcpStream,

    // `Waiter` is the node in the intrusive waiter linked-list
    read_waiter: Waiter,
    write_waiter: Waiter,
}

---

usas una referencia a

Modelo intuitivo

A veces no quieres usar el planificador de Tokio, solo quieres aprovechar su I/O y sus temporizadores. Es como cuando no quieres comer en el restaurante, solo quieres usar su ventanilla de comida para llevar.examples/custom-executor.rsMuestra este «modo híbrido»: usarfutures::executor::ThreadPoolpara la planificación y Tokio para la I/O.

Mecanismo central: TokioContext

La clave de todo el ejemplo está enTokioContexteste tipo envoltorio:

📎 examples/custom-executor.rs:51-54

impl ThreadPool {
    fn spawn(&self, f: impl Future<Output = ()> + Send + 'static) {
        let handle = self.rt.handle().clone();
        self.inner.spawn_ok(TokioContext::new(f, handle));
    }
}

, puede encontrar el driver de I/O de Tokio.

📎 examples/custom-executor.rs:38-48

static EXECUTOR: Lazy<ThreadPool> = Lazy::new(|| {
    // Spawn tokio runtime on a single background thread
    // enabling IO and timers.
    let rt = tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .unwrap();
    let inner = futures::executor::ThreadPool::builder().create().unwrap();

    ThreadPool { inner, rt }
});

Flujo de datos: el viaje de un TcpListener::bind a través de ejecutores

sequenceDiagram
    participant App as "应用 (main)"
    participant FE as "futures::ThreadPool"
    participant TC as "TokioContext"
    participant TR as "tokio::Runtime (后台线程)"
    participant IO as "I/O 驱动 (mio)"

    App->>FE: spawn_ok(TokioContext::new(f, handle))
    FE->>TC: poll(cx)
    TC->>TC: enter(handle) 设置线程局部上下文
    TC->>TC: f.poll(cx) 执行 TcpListener::bind
    TC->>TR: 通过 Handle 访问 I/O 驱动
    TR->>IO: Registration::new 注册 fd
    IO-->>TR: 注册完成
    TR-->>TC: 返回 Pending 或 Ready
    TC-->>FE: 返回 poll 结果
    Note over FE,TR: I/O 就绪时，Tokio 驱动唤醒 waker<br/>FE 重新调度该任务

La clave de este diagrama de secuencia es:el poll de la tarea ocurre en el pool de hilos de futures, pero la espera de eventos de I/O ocurre en el hilo de fondo de Tokio. Ambos se conectan medianteHandley el waker.

Reflexión de diseño: cuándo evitar Tokio

1. Si necesitas integrarte con un ecosistema de ejecutores existente(por ejemplo, algunos frameworks exigenfutures::executor), usarTokioContextes la solución de mínima intrusión.

2. Si necesitas control total sobre la estrategia de planificación(por ejemplo, sistemas en tiempo real que requieren planificación determinista), el work-stealing de Tokio no satisface la necesidad, pero su driver de I/O sigue siendo utilizable.

3. Si solo te parece que la API de Tokio es complicada, entonces no deberías evitarlo—TokioContextla frontera entre ejecutores que introduce

Puntos problemáticos en producción：TokioContextEn el modoblock_on, elRuntime::shutdowndel runtime de Tokio nunca se llama, lo que significa que la lógica de limpieza deRuntimeno se activará automáticamente. Debes hacer drop explícito de

antes de que el programa termine, de lo contrario los hilos de fondo del driver de I/O podrían no cerrarse de forma ordenada.

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

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

Copiarfeature = "io-uring"Observa quetokio_unstableyaparecen simultáneamente. Esto significa que el soporte de io_uring actualmente esexperimentalallow(dead_code), y se debe habilitar también la característica unstable para compilar.allowPor otro lado,

---

la semántica de

ya no aplica bajo io_uring, y se necesita un conjunto completamente nuevo de abstracción «submit-complete». Esta es también la razón por la que el soporte de io_uring permanece en unstable: no se trata simplemente de añadir un backend, sino de reestructurar toda la capa de abstracción del driver de I/O.

Resumen del capítulo：

• Este capítulo revisa desde una perspectiva arquitectónica las tres compensaciones centrales de Tokio, y vislumbra tres rutas de evolución:
• Compensaciones históricasblock_onwork-stealing intercambia complejidad de planificación por escalabilidad multi-núcleo, con el límite de que la granularidad de las tareas no puede ser demasiado fina;
• el driver de I/O es independiente del planificador, permitiendo que

y el runtime multi-hilo reutilicen la misma implementación de I/O;（reactor-refactor.md）：

• loom desaparece por completo en las compilaciones de producción mediante condiciones cfg, y solo enumera exhaustivamente los entrelazados de hilos durante las pruebas.ScheduledIoReestructuración del driver
• Mover el waker desde el interior deAtomicUsizehacia el Future de operación, usando listas enlazadas intrusivas para soportar múltiples esperadores;clear_readinessusar el diseño de campos de bits de
• AsyncRead/AsyncWrite(shutdown/generation/tick/readiness) para resolver la condición de carrera dereader/writer;

como la semántica de poll no permite usar listas enlazadas intrusivas, se conserva：

• con ranuras fijas como compromiso.tokio_unstableEvolución futura
• TokioContextio_uring necesita una nueva abstracción de «submit-complete», actualmente protegida por
• ;

permite usar solo el driver de I/O sin el planificador, pero requiere gestionar manualmente el ciclo de vida del Runtime;

: no reemplaza el backend, sino que cambia el contrato de abstracción del driver de I/O.