got something working in ch10

Author: cfsamson

ch10/a-coroutines-improved/src/runtime.rs

@@ -1,5 +1,5 @@
 [package]
-name = "a-coroutines-improved"
+name = "a-coroutines-variables"
 version = "0.1.0"
 edition = "2021"
 

ch10/a-coroutines-improved/Cargo.toml renamed to ch10/a-coroutines-variables/Cargo.toml

@@ -0,0 +1,19 @@
+mod future;
+mod http;
+mod runtime;
+use crate::http::Http;
+use future::{Future, PollState};
+use runtime::Waker;
+
+fn main() {
+    let mut executor = runtime::init();
+    executor.block_on(async_main());
+}
+
+coroutine fn async_main() {
+    println!("Program starting");
+    let txt = Http::get("/600/HelloAsyncAwait").wait;
+    println!("{txt}");
+    let txt = Http::get("/400/HelloAsyncAwait").wait;
+    println!("{txt}");
+}

ch10/a-coroutines-variables/main_orig.rs

@@ -1,6 +1,14 @@
+// NEW
+use std::{thread::Thread, sync::Arc};
+
+use crate::runtime::Waker;
+// END NEW
+
+
 pub trait Future {
     type Output;
-    fn poll(&mut self) -> PollState<Self::Output>;
+    ///////////////////////// NEW
+    fn poll(&mut self, waker: &Waker) -> PollState<Self::Output>;
 }
 
 pub enum PollState<T> {
@@ -22,15 +30,15 @@ pub fn join_all<F: Future>(futures: Vec<F>) -> JoinAll<F> {
     }
 
     impl<F: Future> Future for JoinAll<F> {
-        type Output = ();
-
-        fn poll(&mut self) -> PollState<Self::Output> {
+        type Output = String;
+        ////////////////////////// HERE
+        fn poll(&mut self, waker: &Waker) -> PollState<Self::Output> {
             for (finished, fut) in self.futures.iter_mut() {
                 if *finished {
                     continue;
                 }
 
-                match fut.poll() {
+                match fut.poll(waker) {
                     PollState::Ready(_) => {
                         *finished = true;
                         self.finished_count += 1;
@@ -41,7 +49,7 @@ pub fn join_all<F: Future>(futures: Vec<F>) -> JoinAll<F> {
             }
 
             if self.finished_count == self.futures.len() {
-                PollState::Ready(())
+                PollState::Ready(String::new())
             } else {
                 PollState::NotReady
             }

ch10/b-coroutines-problem/src/future.rs renamed to ch10/a-coroutines-variables/src/future.rs

@@ -1,7 +1,13 @@
-use crate::{future::PollState, runtime, Future};
-use mio::{Interest, Token};
 use std::io::{ErrorKind, Read, Write};
 
+use mio::{Interest, Registry, Token};
+
+use crate::{
+    future::PollState,
+    runtime::{self, reactor, Waker},
+    Future,
+};
+
 fn get_req(path: &str) -> String {
     format!(
         "GET {path} HTTP/1.1\r\n\
@@ -18,19 +24,21 @@ impl Http {
         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,
         }
     }
 
@@ -46,30 +54,35 @@ impl HttpGetFuture {
 impl Future for HttpGetFuture {
     type Output = String;
 
-    fn poll(&mut self) -> PollState<Self::Output> {
+    fn poll(&mut self, waker: &Waker) -> PollState<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
         if self.stream.is_none() {
             println!("FIRST POLL - START OPERATION");
             self.write_request();
-
             // CHANGED
-            runtime::registry()
-                .register(self.stream.as_mut().unwrap(), Token(0), Interest::READABLE)
-                .unwrap();
+            let stream = self.stream.as_mut().unwrap();
+            runtime::reactor().register(stream, Interest::READABLE, self.id);
+            runtime::reactor().set_waker(waker, self.id);
             // ============
         }
 
-        let mut buff = vec![0u8; 4096];
+        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);
+                    runtime::reactor().deregister(self.stream.as_mut().unwrap(), self.id);
                     break PollState::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 PollState::NotReady;
                 }
 

ch10/b-coroutines-problem/src/http.rs renamed to ch10/a-coroutines-variables/src/http.rs

@@ -1,33 +1,30 @@
 mod future;
 mod http;
 mod runtime;
-
+use crate::http::Http;
 use future::{Future, PollState};
-use runtime::Runtime;
+use runtime::Waker;
 
 fn main() {
-    let future = async_main();
-    let mut runtime = Runtime::new();
-    runtime.block_on(future);
+    let mut executor = runtime::init();
+    executor.block_on(async_main());
 }
 
-
-
 // =================================
 // We rewrite this:
 // =================================
 
-// coro fn async_main() {
-//     let mut counter = 0; 
+// coroutine fn async_main() {
+//     let mut counter = 0;
 //     println!("Program starting");
 //     let txt = http::Http::get("/600/HelloAsyncAwait").wait;
 //     println!("{txt}");
 //     counter += 1;
 //     let txt = http::Http::get("/400/HelloAsyncAwait").wait;
 //     println!("{txt}");
 //     counter += 1;
-//     
-//     println!("Sent {} GET requests.", counter);
+
+//     println!("Received {} responses.", counter);
 // }
 
 // =================================
@@ -65,37 +62,36 @@ impl Coroutine0 {
 impl Future for Coroutine0 {
     type Output = String;
 
-    fn poll(&mut self) -> PollState<Self::Output> {
+    fn poll(&mut self, waker: &Waker) -> PollState<Self::Output> {
         loop {
         match self.state {
                 State0::Start => {
-                    // restore stack (no stack yet)
-                    
+                    // initialize stack (hoist variables)
+                    self.stack.counter = Some(0);
                     // ---- Code you actually wrote ----
-                    let mut counter = 0;
                     println!("Program starting");
 
                     // ---------------------------------
                     let fut1 = Box::new( http::Http::get("/600/HelloAsyncAwait"));
                     self.state = State0::Wait1(fut1);
-                    
+
                     // save stack
-                    self.stack.counter = Some(counter);
+
                 }
 
                 State0::Wait1(ref mut f1) => {
-                    match f1.poll() {
+                    match f1.poll(waker) {
                         PollState::Ready(txt) => {
                             // Restore stack
                             let mut counter = self.stack.counter.take().unwrap();
-                            
+
                             // ---- Code you actually wrote ----
                             println!("{txt}");
                             counter += 1;
                             // ---------------------------------
                             let fut2 = Box::new( http::Http::get("/400/HelloAsyncAwait"));
                             self.state = State0::Wait2(fut2);
-                            
+
                             // save stack
                             self.stack.counter = Some(counter);
                         }
@@ -104,21 +100,21 @@ impl Future for Coroutine0 {
                 }
 
                 State0::Wait2(ref mut f2) => {
-                    match f2.poll() {
+                    match f2.poll(waker) {
                         PollState::Ready(txt) => {
                             // Restore stack
                             let mut counter = self.stack.counter.take().unwrap();
-                            
+
                             // ---- Code you actually wrote ----
                             println!("{txt}");
                             counter += 1;
-                            
-                            println!("Sent {} GET requests.", counter);
+
+                            println!("Received {} responses.", counter);
                             // ---------------------------------
                             self.state = State0::Resolved;
-                            
+
                             // Save stack (all variables set to None already)
-                            
+
                             break PollState::Ready(String::new());
                         }
                         PollState::NotReady => break PollState::NotReady,

ch10/a-coroutines-improved/src/main.rs renamed to ch10/a-coroutines-variables/src/main.rs

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