Asynchronous Input/Output

The previous lesson examined two kinds of task: one that never pauses and one that pauses once by its own decision. In both, the reason for pausing was the task’s own logic — it depended on no outside condition. Most of what a program actually waits on in the real world is not like this: a network connection takes an uncertain amount of time to respond, a read request sent to a file system waits for hardware to become ready. This lesson takes up this kind of wait — but, per the course’s ban on environment-dependent data, without using a real network or file. The wait is built with a model in which a resource that is not ready returns Pending, and wakes itself once it is ready.

The reason for not using a real network socket or file descriptor is not just convenience. The determinism constraint the course has kept since its start requires that a run produce the same number no matter which machine it runs on. How long a real network request takes depends on the network itself; how long a real file read takes depends on the storage device — this would make it impossible for the course to produce an identical output across five separate runs. Instead, the abstract shape of the real-world “resource is not ready yet” situation is captured: a poll call returning Pending and later turning into a Ready. Why the resource is not ready — a network delay, a disk access — falls outside this model; what stays inside the model is only how many times it takes to be asked before it is ready.

Modeling a Resource That Is Not Ready

MockResource is the most reduced form of an input/output resource: it carries how many polls until it becomes ready in a counter. On every poll call, if the counter is nonzero it decreases by one, returns Pending, and wakes itself; once it reaches zero it writes a record to the channel and returns Ready.

// advance.rs — a slice of the core: a future modeling a resource that is not ready.

use std::future::Future;
use std::pin::Pin;
use std::sync::mpsc::Sender;
use std::sync::{Arc, Condvar, Mutex};
use std::task::{Context, Poll, Wake, Waker};

pub struct Notifier {
    pub ready: Mutex<bool>,
    pub condvar: Condvar,
}

impl Wake for Notifier {
    fn wake(self: Arc<Self>) {
        *self.ready.lock().unwrap() = true;
        self.condvar.notify_one();
    }
}

pub fn run<F: Future>(task: F) -> (F::Output, u32) {
    let mut task = Box::pin(task);
    let n = Arc::new(Notifier {
        ready: Mutex::new(false),
        condvar: Condvar::new(),
    });
    let waker: Waker = n.clone().into();
    let mut context = Context::from_waker(&waker);
    let mut poll_count = 0;
    loop {
        poll_count += 1;
        match task.as_mut().poll(&mut context) {
            Poll::Ready(x) => return (x, poll_count),
            Poll::Pending => {
                let mut ready = n.ready.lock().unwrap();
                while !*ready {
                    ready = n.condvar.wait(ready).unwrap();
                }
                *ready = false;
            }
        }
    }
}

// models an input/output resource that is not ready: each poll decreases the remaining
// count by one; once remaining reaches zero it is considered ready and sends the value.
pub struct MockResource {
    pub remaining: u32,
    pub name: &'static str,
    pub channel: Sender<String>,
}

impl Future for MockResource {
    type Output = &'static str;
    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<&'static str> {
        if self.remaining == 0 {
            self.channel.send(self.name.to_string()).unwrap();
            Poll::Ready(self.name)
        } else {
            self.remaining -= 1;
            cx.waker().wake_by_ref();
            Poll::Pending
        }
    }
}

pub async fn read_data(resource: MockResource, channel: Sender<String>) -> &'static str {
    let _ = resource.await;
    channel.send(String::from("read-complete")).unwrap();
    "read"
}

MockResource is the generalized form of the previous lesson’s PausesOnce: PausesOnce paused exactly once, MockResource pauses for however many times the remaining field says. In a real program, this number would stand in for a hardware interrupt, an operating system notification, or a network packet’s arrival; here it is only an integer, and it determines the resource’s behavior — not the real event it stands in for. read_data is an async fn awaiting this resource: it does not advance itself until resource.await completes, exactly like the outer body awaiting an inner future in the previous lesson.

MockResource’s poll method calls cx.waker().wake_by_ref() on every Pending return — just like PausesOnce. This is the most important difference from a real resource: a real network socket does not wake itself; a notification mechanism belonging to the operating system (a subsystem that tells the runtime when a socket is ready) calls the Waker. MockResource imitates this outside notification — having no timer of its own, it chooses to wake itself immediately every time and decrease remaining by one on the next poll call. The result, seen from outside, is the same: the executor sees Pending, gets a chance to poll again after some time, and the resource eventually becomes ready.

How Many Polls, How Much Waiting

The following program runs resources constructed with three different delays: one short (waits one poll), one long (waits four polls), and one awaited from inside an async fn.

// input-output.rs — how many polls it takes a resource that is not ready to complete.
mod advance;
use advance::{run, read_data, MockResource};
use std::sync::mpsc::channel;

fn main() {
    let (sender, receiver) = channel::<String>();

    let (_value_a, poll_count_a) = run(MockResource {
        remaining: 1,
        name: "resource-short-wait",
        channel: sender.clone(),
    });
    let (_value_b, poll_count_b) = run(MockResource {
        remaining: 4,
        name: "resource-long-wait",
        channel: sender.clone(),
    });

    let resource_c = MockResource {
        remaining: 2,
        name: "resource-inner",
        channel: sender.clone(),
    };
    let (_value_c, poll_count_c) = run(read_data(resource_c, sender.clone()));

    drop(sender);
    let mut records: Vec<String> = receiver.try_iter().collect();
    records.sort();

    println!("short wait (remaining=1) -> completed in {} polls", poll_count_a);
    println!("long wait  (remaining=4) -> completed in {} polls", poll_count_b);
    println!("read, waiting on an inner resource -> completed in {} polls", poll_count_c);
    println!();
    println!("records ({}):", records.len());
    for record in &records {
        println!("  {}", record);
    }
}

short wait (remaining=1) -> completed in 2 polls
long wait  (remaining=4) -> completed in 5 polls
read, waiting on an inner resource -> completed in 3 polls

records (4):
  read-complete
  resource-inner
  resource-long-wait
  resource-short-wait

AP5. The poll count is one more than the remaining field’s starting value: remaining: 1 completes in two polls, remaining: 4 in five. The rule belongs to the same family as the $n+1$ rule in the async/await lesson — there it was $n$ consecutive suspension points, here it is $n$ consecutive Pending returns; both count the final call — the one returning Ready — into the total. resource-inner (remaining: 2) is awaited from inside read_data and completes in three polls — two Pending plus one Ready, exactly the count it would have completed in if handed directly to run. This stayed identical across five separate runs.

The Waiting Task’s Class

resource-short-wait and resource-long-wait fall into the when-polled class: both are handed directly to run, both advance according to their own remaining counters. read_data, by contrast, lands in the same position as the previous lesson’s await_inner — its own advancement depends on resource_c completing, and so it falls into the when-something-else-finishes class. resource_c itself is still when-polled — the two classes sit together here too, as in the previous lesson, in different layers of the same task tree.

The Standard Library and Network Applications course measured what a value crossing a network boundary keeps and loses of its type information — in what form the type on one side of the boundary reaches the other side. This lesson measures the same kind of boundary from a different axis: what a task needs in order to advance while it waits on an input/output resource. Type information is never in question here — the value MockResource produces is already &'static str from the start; the boundary separates not a type’s shape, but a task’s progress condition.

Both measurements take up the same boundary — the boundary between a program’s own code and a resource coming from outside — but they ask different questions. There the question was “does the value crossing this boundary keep the same type”; here the question is “what does this side do until the resource on the other side of the boundary is ready.” The difference between resource-short-wait and resource-long-wait is not a type difference — both come from the same MockResource type, both share the same Output; the difference is only in the starting value of the remaining field, that is, in how often the progress condition gets satisfied.

The Executor Still Does Not Matter Here

The Executors lesson showed that two different executor implementations poll the same task the same number of times. The same result holds for a modeled input/output resource too. The following excerpt carries the previous lesson’s run_loop over into this lesson and runs the same MockResource through two different executors.

// advance.rs — the core slice changes: the previous lesson's second executor (run_loop)
// is introduced here, the same MockResource is compared across two executors.

use std::future::Future;
use std::pin::Pin;
use std::sync::mpsc::Sender;
use std::sync::{Arc, Condvar, Mutex};
use std::task::{Context, Poll, Wake, Waker};

pub struct Notifier {
    pub ready: Mutex<bool>,
    pub condvar: Condvar,
}

impl Wake for Notifier {
    fn wake(self: Arc<Self>) {
        *self.ready.lock().unwrap() = true;
        self.condvar.notify_one();
    }
}

pub fn run<F: Future>(task: F) -> (F::Output, u32) {
    let mut task = Box::pin(task);
    let n = Arc::new(Notifier {
        ready: Mutex::new(false),
        condvar: Condvar::new(),
    });
    let waker: Waker = n.clone().into();
    let mut context = Context::from_waker(&waker);
    let mut poll_count = 0;
    loop {
        poll_count += 1;
        match task.as_mut().poll(&mut context) {
            Poll::Ready(x) => return (x, poll_count),
            Poll::Pending => {
                let mut ready = n.ready.lock().unwrap();
                while !*ready {
                    ready = n.condvar.wait(ready).unwrap();
                }
                *ready = false;
            }
        }
    }
}

pub struct EmptyNotifier;

impl Wake for EmptyNotifier {
    fn wake(self: Arc<Self>) {}
}

pub fn run_loop<F: Future>(task: F) -> (F::Output, u32) {
    let mut task = Box::pin(task);
    let n = Arc::new(EmptyNotifier);
    let waker: Waker = n.clone().into();
    let mut context = Context::from_waker(&waker);
    let mut poll_count = 0;
    loop {
        poll_count += 1;
        if let Poll::Ready(x) = task.as_mut().poll(&mut context) {
            return (x, poll_count);
        }
    }
}

pub struct MockResource {
    pub remaining: u32,
    pub name: &'static str,
    pub channel: Sender<String>,
}

impl Future for MockResource {
    type Output = &'static str;
    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<&'static str> {
        if self.remaining == 0 {
            self.channel.send(self.name.to_string()).unwrap();
            Poll::Ready(self.name)
        } else {
            self.remaining -= 1;
            cx.waker().wake_by_ref();
            Poll::Pending
        }
    }
}

// two-executors-io.rs — the same modeled resource, two different executors, the same poll count.
mod advance;
use advance::{run, run_loop, MockResource};
use std::sync::mpsc::channel;

fn main() {
    let (sender, receiver) = channel::<String>();

    let (_value_1, poll_count_1) = run(MockResource {
        remaining: 3,
        name: "resource-run",
        channel: sender.clone(),
    });
    let (_value_2, poll_count_2) = run_loop(MockResource {
        remaining: 3,
        name: "resource-loop",
        channel: sender.clone(),
    });

    drop(sender);
    let mut records: Vec<String> = receiver.try_iter().collect();
    records.sort();

    println!("run={} run_loop={} same: {}", poll_count_1, poll_count_2, poll_count_1 == poll_count_2);
    println!("records ({}):", records.len());
    for record in &records {
        println!("  {}", record);
    }
}

run=4 run_loop=4 same: true
records (2):
  resource-loop
  resource-run

AP6. Two resources constructed with the same remaining: 3 value, run through two utterly different executors, produce an identical poll count (4). MockResource‘s poll method has no idea which executor is calling it — it looks only at its own remaining field. This carries the previous lesson’s finding over to a modeled input/output resource: the executor does not interfere with how long a resource keeps someone waiting, it only interferes with what the calling thread does while it waits — and that second difference lives in an area this course forbids printing.

The Limiting Measurement

The modeled wait does not show a real delay’s duration. The remaining: 4 field requires five polls, but there is no actual waiting at all between these five calls — run’s condition variable wakes up immediately on every wake_by_ref call, because MockResource wakes itself on every Pending return. A real resource (a network socket, a file descriptor) would make this wake call at the moment it actually becomes ready, through an operating system notification; here that notification is imitated by the remaining counter decreasing. What is always measured is the poll count — not the waiting time, because that time depends on the environment, and this course has forbidden printing it from the start.

One consequence of this limit: if the number written into the remaining field is made larger, the poll count grows proportionally, but the run’s actual duration does not change in any measurable way — there is no real wait anywhere, only how many times the loop runs changes. In a real program this relationship reverses: the poll count can stay fixed (an executor can be designed to re-poll a resource only once it is genuinely ready), but the real waiting time changes depending on how fast the resource itself is. Seeing that these two axes — poll count and waiting time — can be independent of each other is necessary for understanding what the model represents and what it does not: this lesson measures only the first of the two, the one number this course has allowed to be printed from the very start.

Summary

• Suspension while waiting on input or output is built with a future modeling a resource that is not ready: if the resource is not ready it returns Pending and wakes itself.
• How many polls a resource takes to complete comes from the resource’s own remaining logic and is deterministic — it stays identical across five separate runs.
• A directly polled resource falls into the when-polled class; an async fn awaiting it falls, for its own completion, into the when-something-else-finishes class.
• The Standard Library and Network Applications course measured a boundary through type information; this lesson measures the same boundary through a task’s progress condition.
• The modeled wait does not show the real delay; what is measured is the poll count, not the time elapsed.
• The same modeled resource run through two different executors also produces the same poll count — the Executors lesson’s finding holds unchanged for modeled input/output too.

Next Step

This topic built, in four steps, what a task needs in order to advance: a task does not advance by being constructed, it advances only when polled; async/await makes visible in the source where these polls pause and from where they resume; an executor decides when these polls happen but does not interfere with the value a task produces; and an input/output resource can extend this polling chain by a number tied to its readiness delay. Across four lessons, the one number measured — how many times poll was called — stayed, every time, derivable in advance from the source.

The remaining question is what happens to a task that is written in the source but never completes, or is left uncompleted. This course’s closing lesson takes up the cancellation established by dropping a task: a dropped task is never polled again, and stays written in the source like this lesson’s MockResource, but shares the same fate as the _never_polled in the Futures lesson — never advancing. A timeout is modeled from the same frame too, not through duration but through poll count.