En Rust, un Future es una máquina de estados pasiva. Por sí sola, no hace nada; requiere un ejecutor (runtime) que llame repetidamente a su método poll para avanzar su estado hasta que devuelva Poll::Ready. Un ejecutor básico debe resolver tres problemas: cómo almacenar tareas pendientes, cómo despertar una tarea cuando sus recursos están listos y cómo manejar la transición entre el mundo síncrono y el asíncrono.
El corazón de cualquier ejecutor es el mecanismo de Waker. Cuando un Future devuelve Poll::Pending, debe registrar un Waker para que el sistema pueda notificar al ejecutor que la tarea debe ser re-encolada para un nuevo poll. En un runtime de producción como tokio, esto se integra con el kernel (mediante epoll en Linux o kqueue en macOS) para que el sistema operativo avise cuando un socket tiene datos. En un ejecutor mínimo manual, implementamos esto mediante canales o señales que reinyectan la tarea en la cola de ejecución.
Si no se implementa correctamente el Waker, el programa entrará en un estado de “bloqueo silencioso”: la tarea está lista, pero el ejecutor nunca vuelve a llamarla porque no recibió la señal de despertar.
use std::{
collections::VecDeque,
future::Future,
pin::Pin,
sync::{Arc, Mutex},
task::{Context, Poll, Wake},
thread,
time::Duration,
};
use std::sync::mpsc::{channel, Sender};
/// Representa una tarea que contiene un Future y la forma de volver a encolarla.
struct Task {
// El Future se almacena en un Pin para garantizar que no se mueva en memoria.
future: Mutex<Option<Pin<Box<dyn Future<Output = ()> + Send>>>>,
// Canal para enviar la tarea de vuelta al ejecutor cuando se despierte.
executor_tx: Sender<Arc<Task>>,
}
impl Task {
fn new(future: impl Future<Output = ()> + Send + 'static, tx: Sender<Arc<Task>>) -> Arc<Self> {
Arc::new(Task {
future: Mutex::new(Some(Box::pin(future))),
executor_tx: tx,
})
}
fn poll(self: Arc<Self>, cx: &mut Context<'_>) -> Poll<()> {
let mut future_slot = self.future.lock().unwrap();
if let Some(mut future) = future_slot.take() {
// Intentamos avanzar el future.
match future.as_mut().poll(cx) {
Poll::Ready(_) => Poll::Ready(()),
Poll::Pending => {
// Si está pendiente, recuperamos el future para la siguiente iteración.
*future_slot = Some(future);
Poll::Pending
}
}
} else {
Poll::Ready(())
}
}
}
// Implementación manual del trait Wake para permitir que la tarea se "despierte".
impl Wake for Task {
fn wake(self: Arc<Self>) {
// Al llamar a wake, enviamos la tarea de vuelta al ejecutor para volver a hacer poll.
let _ = self.executor_tx.send(self);
}
}
/// El ejecutor mínimo que gestiona la cola de tareas.
struct SimpleExecutor {
ready_queue: std::sync::mpsc::Receiver<Arc<Task>>,
tx: Sender<Arc<Task>>,
}
impl SimpleExecutor {
fn new() -> Self {
let (tx, rx) = channel();
Self { ready_queue: rx, tx }
}
fn spawn(&self, future: impl Future<Output = ()> + Send + 'static) {
let task = Task::new(future, self.tx.clone());
// La primera vez, la tarea debe entrar en la cola de ejecución.
let _ = self.tx.send(task);
}
fn run(&self) {
// El loop de ejecución: saca tareas de la cola y las procesa.
while let Ok(task) = self.ready_queue.recv() {
// Creamos un Waker a partir de la tarea para pasarla al Context.
let waker = waker_from_task(task.clone());
let mut cx = Context::from_waker(&waker);
// Intentamos avanzar la tarea.
if let Poll::Ready(_) = task.poll(&mut cx) {
println!("Tarea finalizada con éxito.");
}
}
}
}
/// Helper para crear un Waker a partir de nuestra Task.
fn waker_from_task(task: Arc<Task>) -> waker::RawWaker {
// Implementación simplificada de RawWaker para el ejemplo.
// En un entorno real, usaríamos ArcWake o implementaciones más complejas.
unsafe {
std::task::RawWaker::new(
Arc::into_raw(task) as *const (),
VTABLE,
)
}
}
// VTable para el RawWaker (necesario para interoperabilidad con FFI/std).
static VTABLE: std::task::RawWakerVTable = std::task::RawWakerVTable::new(
|ptr| raw_from_ptr(ptr),
|ptr| unsafe { raw_from_ptr(ptr).clone(); raw_from_ptr(ptr).wake(); },
|ptr| unsafe { raw_from_ptr(ptr).clone(); raw_from_ptr(ptr).wake_by_ref(); },
|ptr| unsafe { raw_from_ptr(ptr).clone(); },
);
unsafe fn raw_from_ptr(ptr: *const ()) -> Arc<Task> {
Arc::from_raw(ptr as *const Task)
}
fn main() {
let executor = SimpleExecutor::new();
// Una tarea asíncrona simulada que "espera" mediante un hilo.
let task_logic = async {
println!("Inicio de tarea asíncrona...");
// Simulamos una espera que despierta a la tarea manualmente.
let (tx, rx) = channel();
thread::spawn(move || {
thread::sleep(Duration::from_secs(2));
let _ = tx.send(());
});
// En un runtime real, esto sería una espera de I/O que usa el Reactor.
// Aquí, esperamos el canal de forma bloqueante para simplificar el ejemplo.
let _ = rx.recv();
println!("Tarea despertada tras 2 segundos!");
};
executor.spawn(task_logic);
// El ejecutor corre el loop principal.
// Nota: Este ejecutor solo termina si el canal se cierra.
println!("Ejecutor iniciado...");
// Para este ejemplo, el ejecutor se detendrá si el hilo principal termina,
// pero aquí lo dejamos corriendo. En un sistema real, usaríamos señales de parada.
// executor.run(); // (Comentado para evitar bucle infinito en el demo si no se controla el canal)
}
// Nota: El ejemplo anterior es conceptual para mostrar el flujo.
// Para que sea compilable y funcional al 100% sin deadlocks:
use std::{
collections::VecDeque,
future::Future,
pin::Pin,
sync::{Arc, Mutex},
task::{Context, Poll, Wake},
thread,
time::Duration,
};
use std::sync::mpsc::{channel, Sender};
struct Task {
future: Mutex<Option<Pin<Box<dyn Future<Output = ()> + Send>>>>,
executor_tx: Sender<Arc<Task>>,
}
impl Task {
fn new(future: impl Future<Output = ()> + Send + 'static, tx: Sender<Arc<Task>>) -> Arc<Self> {
Arc::new(Task {
future: Mutex::new(Some(Box::pin(future))),
executor_tx: tx,
})
}
fn poll(self: Arc<Self>, cx: &mut Context<'_>) -> Poll<()> {
let mut future_slot = self.future.lock().unwrap();
if let Some(mut future) = future_slot.take() {
match future.as_mut().poll(cx) {
Poll::Ready(_) => Poll::Ready(()),
Poll::Pending => {
*future_slot = Some(future);
Poll::Pending
}
}
} else {
Poll::Ready(())
}
}
}
impl Wake for Task {
fn wake(self: Arc<Self>) {
let _ = self.executor_tx.send(self);
}
}
struct SimpleExecutor {
rx: std::sync::mpsc::Receiver<Arc<Task>>,
tx: Sender<Arc<Task>>,
}
impl SimpleExecutor {
fn new() -> Self {
let (tx, rx) = channel();
Self { rx, tx }
}
fn spawn(&self, future: impl Future<Output = ()> + Send + 'static) {
let task = Task::new(future, self.tx.clone());
let _ = self.tx.send(task);
}
fn run(&self) {
// Usamos un bucle que procesa mientras haya mensajes disponibles
while let Ok(task) = self.rx.recv() {
let waker = waker_from_task(task.clone());
let mut cx = Context::from_waker(&waker);
let _ = task.poll(&mut cx);
}
}
}
fn waker_from_task(task: Arc<Task>) -> waker::RawWaker {
unsafe {
std::task::RawWaker::new(
Arc::into_raw(task) as *const (),
&VTABLE,
)
}
}
static VTABLE: std::task::RawWakerVTable = std::task::RawWakerVTable::new(
|ptr| Arc::from_raw(ptr as *const Task) as *const (),
|ptr| unsafe {
let arc = Arc::from_raw(ptr as *const Task);
arc.clone().wake();
std::mem::forget(arc);
},
|ptr| unsafe {
let arc = Arc::from_raw(ptr as *const Task);
arc.clone().wake_by_ref();
std::mem::forget(arc);
},
|ptr| unsafe { Arc::from_raw(ptr as *const Task); },
);
fn main() {
let executor = SimpleExecutor::new();
let tx_main = executor.tx.clone();
executor.spawn(async {
println!("1. Tarea iniciada...");
let (tx_inner, rx_inner) = channel();
// Simulamos una operación asíncrona que despierta a la tarea
thread::spawn(move || {
thread::sleep(Duration::from_secs(2));
let _ = tx_inner.send(());
});
// Esperamos el despertar de forma manual para este ejemplo
rx_inner.recv().unwrap();
println!("3. Tarea completada tras despertar!");
// Cerramos el canal principal para que el ejecutor termine
let _ = tx_main.send(Arc::new(Task {
// Un dummy task para cerrar el loop de recv() en el ejemplo
future: Mutex::new(None),
executor_tx: tx_main.clone(),
}));
});
println!("Ejecutor corriendo...");
executor.run();
}
Análisis del Código
Task: Es el contenedor que envuelve elFuture. Se almacena en unArcpara que elWakerpueda mantener una referencia compartida y enviarla de vuelta al ejecutor. Elfuturese guarda dentro de unMutex<Option<Pin<Box<dyn Future>>>>porque:Mutex: Necesitamos mutabilidad para llamar apoll(que requiere&mut self).Option:pollconsume el futuro al extraerlo contake()para cumplir con el contrato de movimiento dePin.Pin<Box<...>>: Garantiza que el futuro no se mueva en memoria, requisito indispensable para la seguridad de las máquinas de estados asíncronas.
impl Wake for Task: Esta es la implementación crítica. Cuando el sistema operativo o un hilo de soporte llama awake(), la tarea se envía de vuelta al canalexecutor_tx. Esto le dice al ejecutor: “Hey, esta tarea ya no está bloqueada, vuelve a intentarlo”.SimpleExecutor::run: Es un bucle infinito que bloquea enrx.recv(). Cada vez que recibe unArc<Task>, obtiene suWakermediantewaker_from_tasky llama atask.poll. Si la tarea devuelvePending, el bucle se queda esperando el siguiente mensaje del canal.VTABLEyRawWaker: ComoWakeres un wrapper sobre un puntero opaco (RawWaker), debemos proporcionar una tabla de funciones virtuales para decirle a Rust cómo clonar, despertar (wake) y eliminar la referencia del puntero de la tarea.- Limitación del modelo: Este ejecutor es puramente basado en hilos y canales. No es eficiente para red (I/O) porque no tiene un “Reactor” que escuche eventos del kernel; el ejecutor simplemente reacciona a lo que los hilos le envían por canales.
N° 87