finished ch 11

Author: cfsamson

ch11/a-async-await/Cargo.toml renamed to ch11/a-rust-futures/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "a-async-await"
+name = "a-rust-futures"
 version = "0.1.0"
 edition = "2021"
 

ch11/a-rust-futures/src/http.rs

@@ -0,0 +1,95 @@
+use crate::runtime::{self, reactor};
+use mio::Interest;
+use std::{
+    future::Future,
+    io::{ErrorKind, Read, Write},
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+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>, cx: &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(cx, 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(cx, self.id);
+                    break Poll::Pending;
+                }
+
+                Err(e) => panic!("{e:?}"),
+            }
+        }
+    }
+}

ch11/a-rust-futures/src/main.rs

@@ -0,0 +1,45 @@
+mod http;
+mod runtime;
+use crate::http::Http;
+
+fn main() {
+    let mut executor = runtime::init();
+    executor.block_on(async_main());
+}
+
+async fn async_main() {
+    println!("Program starting");
+    let txt = Http::get("/600/HelloAsyncAwait").await;
+    println!("{txt}");
+    let txt = Http::get("/400/HelloAsyncAwait").await;
+    println!("{txt}");
+}
+
+// use isahc::prelude::*;
+
+// async fn async_main2() {
+//     let mut buffer = String::from("\nBUFFER:\n----\n");
+//     let writer = &mut buffer;
+//     println!("Program starting");
+//     let mut res = isahc::get_async("http://127.0.0.1:8080/600/HelloAsyncAwait").await.unwrap();
+//     let txt = res.text().await.unwrap();
+//     writeln!(writer, "{txt}").unwrap();
+//     let mut res = isahc::get_async("http://127.0.0.1:8080/400/HelloAsyncAwait").await.unwrap();
+//     let txt = res.text().await.unwrap();
+//     writeln!(writer, "{txt}").unwrap();
+
+//     println!("{}", buffer);
+// }
+
+// async fn spawn_many() {
+//     for i in 0..100 {
+//         let delay = i * 10;
+//         let req = format!("http://127.0.0.1:8080/{delay}/HelloAsyncAwait{i}");
+
+//         runtime::spawn(async move {
+//             let mut res = isahc::get_async(&req).await.unwrap();
+//             let txt = res.text().await.unwrap();
+//             println!("{txt}");
+//         });
+//     }
+// }

ch11/a-async-await/src/runtime.rs renamed to ch11/a-rust-futures/src/runtime.rs

@@ -0,0 +1,128 @@
+use std::{
+    cell::{Cell, RefCell},
+    collections::HashMap,
+    future::Future,
+    pin::Pin,
+    sync::{Arc, Mutex},
+    task::{Context, Poll, Wake, Waker},
+    thread::{self, Thread},
+};
+
+type Task = Pin<Box<dyn Future<Output = ()>>>;
+
+thread_local! {
+    static CURRENT_EXEC: ExecutorCore = ExecutorCore::default();
+}
+
+#[derive(Default)]
+struct ExecutorCore {
+    tasks: RefCell<HashMap<usize, Task>>,
+    ready_queue: Arc<Mutex<Vec<usize>>>,
+    next_id: Cell<usize>,
+}
+
+pub fn spawn<F>(future: F)
+where
+    F: Future<Output = ()> + 'static,
+{
+    CURRENT_EXEC.with(|e| {
+        let id = e.next_id.get();
+        e.tasks.borrow_mut().insert(id, Box::pin(future));
+        e.ready_queue.lock().map(|mut q| q.push(id)).unwrap();
+        e.next_id.set(id + 1);
+    });
+}
+
+pub struct Executor;
+
+impl Executor {
+    pub fn new() -> Self {
+        Self {}
+    }
+
+    fn pop_ready(&self) -> Option<usize> {
+        CURRENT_EXEC.with(|q| q.ready_queue.lock().map(|mut q| q.pop()).unwrap())
+    }
+
+    fn get_future(&self, id: usize) -> Option<Task> {
+        CURRENT_EXEC.with(|q| q.tasks.borrow_mut().remove(&id))
+    }
+
+    fn get_waker(&self, id: usize) -> Arc<MyWaker> {
+        Arc::new(MyWaker {
+            id,
+            thread: thread::current(),
+            ready_queue: CURRENT_EXEC.with(|q| q.ready_queue.clone()),
+        })
+    }
+
+    fn insert_task(&self, id: usize, task: Task) {
+        CURRENT_EXEC.with(|q| q.tasks.borrow_mut().insert(id, task));
+    }
+
+    fn task_count(&self) -> usize {
+        CURRENT_EXEC.with(|q| q.tasks.borrow().len())
+    }
+
+    pub fn block_on<F>(&mut self, future: F)
+    where
+        F: Future<Output = ()> + 'static,
+    {
+        // ===== OPTIMIZATION, ASSUME READY
+        // let waker = self.get_waker(usize::MAX);
+        // let mut future = future;
+        // match future.poll(&waker) {
+        //     PollState::Pending => (),
+        //     PollState::Ready(_) => return,
+        // }
+        // ===== END
+
+        spawn(future);
+
+        loop {
+            while let Some(id) = self.pop_ready() {
+                let mut future = match self.get_future(id) {
+                    Some(f) => f,
+                    // guard against false wakeups
+                    None => continue,
+                };
+
+                let waker: Waker = self.get_waker(id).into();
+                let mut cx = Context::from_waker(&waker);
+
+                match future.as_mut().poll(&mut cx) {
+                    Poll::Pending => self.insert_task(id, future),
+                    Poll::Ready(_) => continue,
+                }
+            }
+
+            let task_count = self.task_count();
+            let name = thread::current().name().unwrap_or_default().to_string();
+
+            if task_count > 0 {
+                println!("{name}: {task_count} pending tasks. Sleep until notified.");
+                thread::park();
+            } else {
+                println!("{name}: All tasks are finished");
+                break;
+            }
+        }
+    }
+}
+
+#[derive(Clone)]
+pub struct MyWaker {
+    thread: Thread,
+    id: usize,
+    ready_queue: Arc<Mutex<Vec<usize>>>,
+}
+
+impl Wake for MyWaker {
+    fn wake(self: Arc<Self>) {
+        self.ready_queue
+            .lock()
+            .map(|mut q| q.push(self.id))
+            .unwrap();
+        self.thread.unpark();
+    }
+}

ch11/a-rust-futures/src/runtime/executor.rs

@@ -0,0 +1,82 @@
+use mio::{net::TcpStream, Events, Interest, Poll, Registry, Token};
+use std::{
+    collections::HashMap,
+    sync::{
+        atomic::{AtomicUsize, Ordering},
+        Arc, Mutex, OnceLock,
+    },
+    thread, task::{Context, Waker},
+};
+
+
+type Wakers = Arc<Mutex<HashMap<usize, Waker>>>;
+
+static REACTOR: OnceLock<Reactor> = OnceLock::new();
+
+pub fn reactor() -> &'static Reactor {
+    REACTOR.get().expect("Called outside an runtime context")
+}
+
+pub struct Reactor {
+    wakers: Wakers,
+    registry: Registry,
+    next_id: AtomicUsize,
+}
+
+impl Reactor {
+    pub fn register(&self, stream: &mut TcpStream, interest: Interest, id: usize) {
+        self.registry.register(stream, Token(id), interest).unwrap();
+    }
+
+    pub fn set_waker(&self, cx: &Context, id: usize) {
+        let _ = self
+            .wakers
+            .lock()
+            .map(|mut w| w.insert(id, cx.waker().clone()).is_none())
+            .unwrap();
+    }
+
+    pub fn deregister(&self, stream: &mut TcpStream, id: usize) {
+        self.wakers.lock().map(|mut w| w.remove(&id)).unwrap();
+        self.registry.deregister(stream).unwrap();
+    }
+
+    pub fn next_id(&self) -> usize {
+        self.next_id.fetch_add(1, Ordering::Relaxed)
+    }
+}
+
+fn event_loop(mut poll: Poll, wakers: Wakers) {
+    let mut events = Events::with_capacity(100);
+    loop {
+        poll.poll(&mut events, None).unwrap();
+        for e in events.iter() {
+            // Optimization for Windows since we get unneeded wakeups
+            // if !e.is_readable() && e.is_read_closed() {
+            //     continue;
+            // }
+            let Token(id) = e.token();
+            let wakers = wakers.lock().unwrap();
+
+            if let Some(waker) = wakers.get(&id) {
+                waker.wake_by_ref();
+            }
+        }
+    }
+}
+
+pub fn start() {
+    use thread::spawn;
+    let wakers = Arc::new(Mutex::new(HashMap::new()));
+    let poll = Poll::new().unwrap();
+    let registry = poll.registry().try_clone().unwrap();
+    let next_id = AtomicUsize::new(1);
+    let reactor = Reactor {
+        wakers: wakers.clone(),
+        registry,
+        next_id,
+    };
+
+    REACTOR.set(reactor).ok().expect("Reactor already running");
+    spawn(move || event_loop(poll, wakers));
+}

ch11/a-rust-futures/src/runtime/reactor.rs

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