Rust Build an Executor

Take a task, poll it once, keep it only if it is still Pending. recv() blocks while the queue is empty, so an idle executor uses no CPU.

while let Ok(task) = self.ready_queue.recv() {
    let mut future_slot = task.future.lock().unwrap();
    if let Some(mut future) = future_slot.take() {
        let waker = waker_ref(&task);
        let context = &mut Context::from_waker(&waker);
        if future.as_mut().poll(context).is_pending() {
            *future_slot = Some(future);
        }
    }
}

The future sits in a Mutex<Option<...>> so poll can take it out and (maybe) put it back. The task_sender is the task's own doorbell back into the queue — its waker sends through it.

struct Task {
    future: Mutex<Option<BoxFuture<'static, ()>>>,
    task_sender: SyncSender<Arc<Task>>,
}

This impl is the waker the executor hands to every poll. When the Delay future's timer thread calls wake(), this code runs — and the task lands on the queue again. That is the entire notification system.

impl ArcWake for Task {
    fn wake_by_ref(arc_self: &Arc<Self>) {
        let cloned = arc_self.clone();
        arc_self.task_sender.try_send(cloned).expect("too many tasks queued");
    }
}

Spawning wraps the future in an Arc<Task> and sends it once. That first send is what gets the task its first poll — after that, only wake() can bring it back.

fn spawn(&self, future: impl Future<Output = ()> + Send + 'static) {
    let task = Arc::new(Task {
        future: Mutex::new(Some(future.boxed())),
        task_sender: self.task_sender.clone(),
    });
    self.task_sender.try_send(task).expect("too many tasks queued");
}

use futures::future::BoxFuture;
use futures::task::{waker_ref, ArcWake};
use futures::FutureExt;
use std::future::Future;
use std::pin::Pin;
use std::sync::mpsc::{sync_channel, Receiver, SyncSender};
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll, Waker};
use std::thread;
use std::time::Duration;

// ---------- a small future to run: wait, then finish ----------
struct SharedState {
    completed: bool,
    waker: Option<Waker>,
}

struct Delay {
    state: Arc<Mutex<SharedState>>,
}

impl Delay {
    fn new(wait: Duration) -> Delay {
        let state = Arc::new(Mutex::new(SharedState { completed: false, waker: None }));
        let thread_state = Arc::clone(&state);
        thread::spawn(move || {
            thread::sleep(wait);
            let mut state = thread_state.lock().unwrap();
            state.completed = true;
            if let Some(waker) = state.waker.take() {
                waker.wake();
            }
        });
        Delay { state }
    }
}

impl Future for Delay {
    type Output = ();

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        let mut state = self.state.lock().unwrap();
        if state.completed {
            Poll::Ready(())
        } else {
            state.waker = Some(cx.waker().clone());
            Poll::Pending
        }
    }
}

// ---------- the executor ----------
struct Task {
    future: Mutex<Option<BoxFuture<'static, ()>>>,
    task_sender: SyncSender<Arc<Task>>,
}

impl ArcWake for Task {
    fn wake_by_ref(arc_self: &Arc<Self>) {
        let cloned = arc_self.clone();
        arc_self.task_sender.try_send(cloned).expect("too many tasks queued");
    }
}

struct Executor {
    ready_queue: Receiver<Arc<Task>>,
}

impl Executor {
    fn run(self) {
        while let Ok(task) = self.ready_queue.recv() {
            let mut future_slot = task.future.lock().unwrap();
            if let Some(mut future) = future_slot.take() {
                let waker = waker_ref(&task);
                let context = &mut Context::from_waker(&waker);
                if future.as_mut().poll(context).is_pending() {
                    *future_slot = Some(future);
                }
            }
        }
    }
}

#[derive(Clone)]
struct Spawner {
    task_sender: SyncSender<Arc<Task>>,
}

impl Spawner {
    fn spawn(&self, future: impl Future<Output = ()> + Send + 'static) {
        let task = Arc::new(Task {
            future: Mutex::new(Some(future.boxed())),
            task_sender: self.task_sender.clone(),
        });
        self.task_sender.try_send(task).expect("too many tasks queued");
    }
}

fn new_executor_and_spawner() -> (Executor, Spawner) {
    const MAX_QUEUED_TASKS: usize = 10_000;
    let (task_sender, ready_queue) = sync_channel(MAX_QUEUED_TASKS);
    (Executor { ready_queue }, Spawner { task_sender })
}

// ---------- run two tasks on it ----------
fn main() {
    let (executor, spawner) = new_executor_and_spawner();

    let spawner2 = spawner.clone();
    spawner.spawn(async move {
        println!("task A start");
        Delay::new(Duration::from_millis(60)).await;
        println!("task A done");
    });
    spawner2.spawn(async move {
        println!("task B start");
        Delay::new(Duration::from_millis(20)).await;
        println!("task B done");
    });

    drop(spawner);
    drop(spawner2);
    executor.run();
}