final example finished, just had to double check that the changes were as easy as I hoped

Author: cfsamson

ch11/a-async-await/Cargo.toml

@@ -0,0 +1,9 @@
+[package]
+name = "a-async-await"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+mio = { version = "0.8", features = ["net", "os-poll"] }

ch11/a-async-await/src/future.rs

@@ -0,0 +1,46 @@
+// NEW
+use std::{future::Future, pin::Pin, task::{Poll, Context}};
+
+
+pub fn join_all<F: Future>(futures: Vec<F>) -> JoinAll<F> {
+    let futures = futures.into_iter().map(|f| (false, Box::pin(f))).collect();
+    JoinAll {
+        futures,
+        finished_count: 0,
+    }
+}
+
+pub struct JoinAll<F: Future> {
+    futures: Vec<(bool, Pin<Box<F>>)>,
+    finished_count: usize,
+}
+
+impl<F: Future> Future for JoinAll<F> {
+    type Output = String;
+    fn poll(mut self: Pin<&mut Self>, waker: &mut Context) -> Poll<Self::Output> {
+        let Self {
+            futures,
+            finished_count,
+        } = &mut *self;
+        for (finished, fut) in futures.iter_mut() {
+            if *finished {
+                continue;
+            }
+
+            match fut.as_mut().poll(waker) {
+                Poll::Ready(_) => {
+                    *finished = true;
+                    *finished_count += 1;
+                }
+
+                Poll::Pending => continue,
+            }
+        }
+
+        if self.finished_count == self.futures.len() {
+            Poll::Ready(String::new())
+        } else {
+            Poll::Pending
+        }
+    }
+}

ch11/a-async-await/src/http.rs

@@ -0,0 +1,99 @@
+use std::{
+    io::{ErrorKind, Read, Write},
+    pin::Pin,
+    task::{Poll, Context},
+};
+
+use mio::Interest;
+
+use crate::{
+    runtime::{self, reactor},
+    Future,
+};
+
+fn get_req(path: &str) -> String {
+    format!(
+        "GET {path} HTTP/1.1\r\n\
+             Host: localhost\r\n\
+             Connection: close\r\n\
+             \r\n"
+    )
+}
+
+pub struct Http;
+
+impl Http {
+    pub fn get(path: &str) -> impl Future<Output = String> {
+        HttpGetFuture::new(path.to_string())
+    }
+}
+struct HttpGetFuture {
+    stream: Option<mio::net::TcpStream>,
+    buffer: Vec<u8>,
+    path: String,
+    id: usize,
+}
+
+impl HttpGetFuture {
+    fn new(path: String) -> Self {
+        let id = reactor().next_id();
+        Self {
+            stream: None,
+            buffer: vec![],
+            path,
+            id,
+        }
+    }
+
+    fn write_request(&mut self) {
+        let stream = std::net::TcpStream::connect("127.0.0.1:8080").unwrap();
+        stream.set_nonblocking(true).unwrap();
+        let mut stream = mio::net::TcpStream::from_std(stream);
+        stream.write_all(get_req(&self.path).as_bytes()).unwrap();
+        self.stream = Some(stream);
+    }
+}
+
+impl Future for HttpGetFuture {
+    type Output = String;
+
+    fn poll(mut self: Pin<&mut Self>, waker: &mut Context) -> Poll<Self::Output> {
+        // If this is first time polled, start the operation
+        // see: https://users.rust-lang.org/t/is-it-bad-behaviour-for-a-future-or-stream-to-do-something-before-being-polled/61353
+        // Avoid dns lookup this time
+        //let this = self.get_mut();
+
+        let id = self.id;
+        if self.stream.is_none() {
+            println!("FIRST POLL - START OPERATION");
+            self.write_request();
+            // CHANGED
+            let stream = (&mut self).stream.as_mut().unwrap();
+            runtime::reactor().register(stream, Interest::READABLE, id);
+            runtime::reactor().set_waker(waker, self.id);
+            // ============
+        }
+
+        let mut buff = vec![0u8; 147];
+        loop {
+            match self.stream.as_mut().unwrap().read(&mut buff) {
+                Ok(0) => {
+                    let s = String::from_utf8_lossy(&self.buffer).to_string();
+                    runtime::reactor().deregister(self.stream.as_mut().unwrap(), id);
+                    break Poll::Ready(s.to_string());
+                }
+                Ok(n) => {
+                    self.buffer.extend(&buff[0..n]);
+                    continue;
+                }
+                Err(e) if e.kind() == ErrorKind::WouldBlock => {
+                    // always store the last given Waker
+                    runtime::reactor().set_waker(waker, self.id);
+                    break Poll::Pending;
+                }
+
+                Err(e) => panic!("{e:?}"),
+            }
+        }
+    }
+}

ch11/a-async-await/src/main.rs

@@ -0,0 +1,134 @@
+mod future;
+mod http;
+mod runtime;
+use crate::http::Http;
+
+
+fn main() {
+    let mut executor = runtime::init();
+    executor.block_on(async_main());
+}
+
+use std::{fmt::Write, future::Future, marker::PhantomPinned, mem, pin::Pin, task::{Poll, Context}};
+// =================================
+// We rewrite this:
+// =================================
+
+// coro fn async_main() {
+//     let mut buffer = String::from("\nBUFFER:\n----\n");
+//     let writer = &mut buffer;
+//     println!("Program starting");
+//     let txt = http::Http::get("/600/HelloAsyncAwait").wait;
+//     writeln!(writer, "{txt}").unwrap();
+//     let txt = http::Http::get("/400/HelloAsyncAwait").wait;
+//     writeln!(writer, "{txt}").unwrap();
+//
+//     println!("{}", buffer);
+// }
+
+// =================================
+// Into this:
+// =================================
+
+fn async_main() -> impl Future<Output = String> {
+    Coroutine0::new()
+}
+
+enum State0 {
+    Start,
+    Wait1(Pin<Box<dyn Future<Output = String>>>),
+    Wait2(Pin<Box<dyn Future<Output = String>>>),
+    Resolved,
+}
+
+#[derive(Default)]
+struct Stack0 {
+    buffer: Option<String>,
+    writer: Option<*mut String>,
+}
+
+struct Coroutine0 {
+    stack: Stack0,
+    state: State0,
+    _pin: PhantomPinned,
+}
+
+impl Coroutine0 {
+    fn new() -> Self {
+        Self {
+            state: State0::Start,
+            stack: Stack0::default(),
+            _pin: PhantomPinned,
+        }
+    }
+}
+
+impl Future for Coroutine0 {
+    type Output = String;
+
+    fn poll(self: Pin<&mut Self>, waker: &mut Context) -> Poll<Self::Output> {
+        let this = unsafe { self.get_unchecked_mut() };
+        loop {
+            match this.state {
+                State0::Start => {
+                    // initialize stack (hoist declarations - no stack yet)
+
+                    this.stack.buffer = Some(String::from("\nBUFFER:\n----\n"));
+                    this.stack.writer = Some(this.stack.buffer.as_mut().unwrap());
+                    // ---- Code you actually wrote ----
+                    println!("Program starting");
+
+                    // ---------------------------------
+                    let fut1 = Box::pin(http::Http::get("/600/HelloAsyncAwait"));
+                    this.state = State0::Wait1(fut1);
+
+                    // save stack
+                    // nothing to save
+                }
+
+                State0::Wait1(ref mut f1) => {
+                    match f1.as_mut().poll(waker) {
+                        Poll::Ready(txt) => {
+                            // Restore stack
+                            let writer = unsafe { &mut *this.stack.writer.unwrap() };
+
+                            // ---- Code you actually wrote ----
+                            writeln!(writer, "{txt}").unwrap();
+                            // ---------------------------------
+                            let fut2 = Box::pin(http::Http::get("/400/HelloAsyncAwait"));
+                            this.state = State0::Wait2(fut2);
+
+                            // save stack
+                            this.stack.writer = Some(writer);
+                        }
+                        Poll::Pending => break Poll::Pending,
+                    }
+                }
+
+                State0::Wait2(ref mut f2) => {
+                    match f2.as_mut().poll(waker) {
+                        Poll::Ready(txt) => {
+                            // Restore stack
+                            let buffer = this.stack.buffer.as_ref().take().unwrap();
+                            let writer = unsafe { &mut *this.stack.writer.unwrap() };
+
+                            // ---- Code you actually wrote ----
+                            writeln!(writer, "{txt}").unwrap();
+
+                            println!("{}", buffer);
+                            // ---------------------------------
+                            this.state = State0::Resolved;
+
+                            // Save stack (all variables set to None already)
+
+                            break Poll::Ready(String::new());
+                        }
+                        Poll::Pending => break Poll::Pending,
+                    }
+                }
+
+                State0::Resolved => panic!("Polled a resolved future"),
+            }
+        }
+    }
+}

ch11/a-async-await/src/runtime.rs

@@ -0,0 +1,10 @@
+pub use executor::{spawn, Executor};
+pub use reactor::reactor;
+
+mod executor;
+mod reactor;
+
+pub fn init() -> Executor {
+    reactor::start();
+    Executor::new()
+}