still half work on switching to async

Author: adwinwhite

ui/Cargo.lock

@@ -1,12 +1,13 @@
 use std::collections::HashMap;
+use std::convert::TryFrom;
 use std::io::{Read, Write};
-use std::os::unix::net::UnixStream;
 
 use bincode::{deserialize_from, serialize_into};
 use log::info;
 
+use crate::{Error, parse_channel, parse_mode, parse_edition, parse_crate_type, parse_target};
 use crate::sandbox::{
-    Channel, CompileRequest, CompileTarget, CrateType, Edition, ExecuteRequest, Mode,
+    Channel, CompileRequest, CompileTarget, CrateType, Edition, ExecuteRequest, Mode, self,
 };
 
 type JobBatch = Vec<worker_message::Request>;
@@ -138,42 +139,135 @@ fn execute_request_to_batch(req: ExecuteRequest) -> JobBatch {
     batch
 }
 
-fn process_batch(stream: &mut UnixStream, batch: JobBatch) -> Option<worker_message::Response> {
-    let mut response = None;
-    for job in batch {
-        let res = work(stream, job);
-        dbg!(&res);
-        if !res.is_ok() {
-            return Some(res);
-        } else {
-            response = Some(res);
-        }
+// fn process_batch(stream: &mut UnixStream, batch: JobBatch) -> Option<worker_message::Response> {
+    // let mut response = None;
+    // for job in batch {
+        // let res = work(stream, job);
+        // dbg!(&res);
+        // if !res.is_ok() {
+            // return Some(res);
+        // } else {
+            // response = Some(res);
+        // }
+    // }
+    // response
+// }
+
+// fn work(stream: &mut UnixStream, req: worker_message::Request) -> worker_message::Response {
+    // serialize_into(&mut *stream, &req).unwrap();
+    // deserialize_from(&mut *stream).unwrap()
+// }
+//
+#[derive(serde::Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct WSExecuteRequest {
+    channel: String,
+    mode: String,
+    edition: String,
+    crate_type: String,
+    tests: bool,
+    code: String,
+    backtrace: bool,
+    extra: serde_json::Value,
+}
+
+impl TryFrom<WSExecuteRequest> for (sandbox::ExecuteRequest, serde_json::Value) {
+    type Error = Error;
+
+    fn try_from(value: WSExecuteRequest) -> Result<Self, Self::Error> {
+        let WSExecuteRequest {
+            channel,
+            mode,
+            edition,
+            crate_type,
+            tests,
+            code,
+            backtrace,
+            extra,
+        } = value;
+
+        let req = sandbox::ExecuteRequest {
+            channel: parse_channel(&channel)?,
+            mode: parse_mode(&mode)?,
+            edition: parse_edition(&edition)?,
+            crate_type: parse_crate_type(&crate_type)?,
+            tests,
+            backtrace,
+            code,
+        };
+
+        Ok((req, extra))
     }
-    response
 }
 
-fn work(stream: &mut UnixStream, req: worker_message::Request) -> worker_message::Response {
-    serialize_into(&mut *stream, &req).unwrap();
-    deserialize_from(&mut *stream).unwrap()
+#[derive(serde::Serialize, serde::Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct WSCompileRequest {
+    target: String,
+    channel: String,
+    mode: String,
+    edition: String,
+    crate_type: String,
+    tests: bool,
+    code: String,
+    backtrace: bool,
+    extra: serde_json::Value,
+}
+
+impl TryFrom<WSCompileRequest> for (sandbox::CompileRequest, serde_json::Value) {
+    type Error = Error;
+
+    fn try_from(value: WSCompileRequest) -> Result<Self, Self::Error> {
+        let WSCompileRequest {
+            target,
+            channel,
+            mode,
+            edition,
+            crate_type,
+            tests,
+            code,
+            backtrace,
+            extra,
+        } = value;
+
+        let req = sandbox::CompileRequest {
+            target: parse_target(&target)?,
+            channel: parse_channel(&channel)?,
+            mode: parse_mode(&mode)?,
+            edition: parse_edition(&edition)?,
+            crate_type: parse_crate_type(&crate_type)?,
+            tests,
+            backtrace,
+            code,
+        };
+
+        Ok((req, extra))
+    }
 }
 
 #[cfg(test)]
 mod tests {
     use std::io::{Read, Write};
-    use std::path::Path;
+    use std::process::Stdio;
     use std::str::from_utf8;
     use std::time::SystemTime;
-    use std::{fs, os::unix::net::UnixStream, thread, time::Duration};
+    use std::{fs, thread, time::Duration};
 
+    const WORKER_FILEPATH: &str = "../worker_message/target/debug/worker";
     use crate::{
         coordinator::{compile_request_to_batch, execute_request_to_batch},
         sandbox::{
             Channel, CompileRequest, CompileTarget, CrateType, Edition, ExecuteRequest, Mode,
         },
     };
-    use worker_message::worker;
+    use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
+    use tokio::process::{ChildStdin, ChildStdout, Command};
+    use tokio::sync::mpsc;
+    use worker_message::{worker, CoordinatorMessage, WorkerMessage};
+
+    use super::WSCompileRequest;
 
-    use super::process_batch;
+    // use super::process_batch;
     fn new_execute_request() -> ExecuteRequest {
         ExecuteRequest {
             channel: Channel::Stable,
@@ -186,85 +280,155 @@ mod tests {
         }
     }
 
-    fn new_compile_request() -> CompileRequest {
-        CompileRequest {
-            target: CompileTarget::Mir,
-            channel: Channel::Stable,
-            crate_type: CrateType::Binary,
-            mode: Mode::Release,
-            edition: Some(Edition::Rust2021),
+    fn new_compile_request() -> WSCompileRequest {
+        WSCompileRequest {
+            target: "mir".to_owned(),
+            channel: "stable".to_owned(),
+            crate_type: "binary".to_owned(),
+            mode: "release".to_owned(),
+            edition: "2021".to_owned(),
             tests: false,
             backtrace: true,
             code: r#"fn main() { println!("Hello World!"); }"#.to_owned(),
+            extra: serde_json::json!({})
         }
     }
 
-    fn generate_temporary_socket() -> String {
-        let timestamp = SystemTime::now()
-            .duration_since(SystemTime::UNIX_EPOCH)
-            .unwrap()
-            .as_nanos()
-            .to_string();
-        format!("/tmp/playground_test_socket_{timestamp}")
+    async fn run_worker_in_background() -> (ChildStdin, ChildStdout) {
+        let mut child = Command::new(WORKER_FILEPATH)
+            .stdin(Stdio::piped())
+            .stdout(Stdio::piped())
+            .spawn()
+            .unwrap();
+        (child.stdin.take().unwrap(), child.stdout.take().unwrap())
     }
 
-    fn run_worker_in_background() -> (UnixStream, UnixStream) {
-        let test_request_socket = generate_temporary_socket();
-        let test_stdio_socket = generate_temporary_socket();
-        if Path::new(&test_request_socket).exists() {
-            fs::remove_file(&test_request_socket).unwrap();
-        }
-        if Path::new(&test_stdio_socket).exists() {
-            fs::remove_file(&test_stdio_socket).unwrap();
-        }
-        {
-            let test_request_socket = test_request_socket.clone();
-            let test_stdio_socket = test_stdio_socket.clone();
-            thread::spawn(move || {
-                worker::listen(&test_request_socket, &test_stdio_socket);
-            });
-        }
-        // Let listener get ready.
-        thread::sleep(Duration::from_secs(1));
-        let request_socket = UnixStream::connect(test_request_socket).unwrap();
-        let stdio_socket = UnixStream::connect(test_stdio_socket).unwrap();
-        (request_socket, stdio_socket)
+    async fn spawn_io_queue(
+        mut sender: ChildStdin,
+        mut receiver: ChildStdout,
+    ) -> (
+        mpsc::Sender<CoordinatorMessage>,
+        mpsc::Receiver<WorkerMessage>,
+    ) {
+        let (tx, worker_rx) = mpsc::channel(32);
+        tokio::spawn(async move {
+            let mut buffer = Vec::new();
+            loop {
+                let payload_len = receiver.read_u64().await.unwrap();
+                buffer.resize(payload_len as usize, 0);
+                receiver.read_exact(&mut buffer).await.unwrap();
+                let worker_msg: WorkerMessage = bincode::deserialize(&buffer).unwrap();
+                tx.send(worker_msg).await.unwrap();
+            }
+        });
+        let (worker_tx, mut rx) = mpsc::channel(32);
+        tokio::spawn(async move {
+            loop {
+                let worker_msg = rx.recv().await.unwrap();
+                let encoded = bincode::serialize(&worker_msg).unwrap();
+                sender.write_u64(encoded.len() as u64).await.unwrap();
+                sender.write_all(&encoded).await.unwrap();
+            }
+        });
+        (worker_tx, worker_rx)
     }
 
-    #[test]
-    fn compile_request() {
-        let (mut request_socket, stdio_socket) = run_worker_in_background();
-
-        let request = new_compile_request();
-        let batch = compile_request_to_batch(request);
-        let result = process_batch(&mut request_socket, batch).unwrap();
-        println!("{result}");
-        assert!(result.is_ok());
+    async fn make_worker_channel() -> (
+        mpsc::Sender<CoordinatorMessage>,
+        mpsc::Receiver<WorkerMessage>,
+    ) {
+        let (stdin, stdout) = run_worker_in_background().await;
+        spawn_io_queue(stdin, stdout).await
     }
 
-    #[test]
-    fn execute_request() {
-        let (mut request_socket, mut stdio_socket) = run_worker_in_background();
+    async fn handle_websocket_message(
+        ws_msg: String,
+        worker_sender: &mpsc::Sender<CoordinatorMessage>,
+    ) {
+    }
+    async fn handle_worker_message(msg: WorkerMessage, websocket_sender: &mpsc::Sender<String>) {}
 
-        let request = new_execute_request();
-        let batch = execute_request_to_batch(request);
-        let result = process_batch(&mut request_socket, batch).unwrap();
-        let handle = thread::spawn(move || {
-            let mut buffer = vec![0; 1024];
-            loop {
-                let n = stdio_socket.read(&mut buffer).unwrap();
-                if n != 0 {
-                    println!("From stdio {} - {}", n, from_utf8(&buffer[..n]).unwrap());
+    async fn pair_websocket_worker(ws_sender: mpsc::Sender<String>, mut ws_receiver: mpsc::Receiver<String>, worker_sender: mpsc::Sender<CoordinatorMessage>, mut worker_receiver: mpsc::Receiver<WorkerMessage>) {
+        loop {
+            tokio::select! {
+                ws_request = ws_receiver.recv() => {
+                    match ws_request {
+                        None => {
+                            break;
+                        }
+                        Some(txt) => {
+                            handle_websocket_message(txt, &worker_sender).await;
+                        }
+                    }
+                },
+                worker_msg = worker_receiver.recv() => {
+                    match worker_msg {
+                        None => {
+                            break;
+                        }
+                        Some(msg) => {
+                            handle_worker_message(msg, &ws_sender).await;
+                        }
+                    }
                 }
             }
-        });
-        println!("{result}");
-        assert!(result.is_ok());
-        // handle.join().unwrap();
-        thread::sleep(Duration::from_secs(1));
+        }
     }
 
-    #[test]
-    fn consecutive_compile_execute() {
+    #[tokio::test]
+    async fn compile_request() {
+        let (worker_sender, worker_receiver) = make_worker_channel().await;
+        let (tx, ws_receiver) = mpsc::channel(32);
+        let (ws_sender, mut rx) = mpsc::channel(32);
+        tokio::spawn(async move {
+            pair_websocket_worker(ws_sender, ws_receiver, worker_sender, worker_receiver).await;
+        });
+
+        // Simulate user input.
+        tx.send(serde_json::to_string(&new_compile_request()).unwrap()).await.unwrap();
+        loop {
+            let server_msg = rx.recv().await;
+            match server_msg {
+                None => {
+                    panic!("server disconnected");
+                }
+                Some(msg) => {
+                }
+            }
+        }
+
+
+
+        // let request = new_compile_request();
+        // let batch = compile_request_to_batch(request);
+        // let result = process_batch(&mut request_socket, batch).unwrap();
+
+        // println!("{result}");
+        // assert!(result.is_ok());
     }
+
+    // #[test]
+    // fn execute_request() {
+    // let (mut request_socket, mut stdio_socket) = run_worker_in_background();
+
+    // let request = new_execute_request();
+    // let batch = execute_request_to_batch(request);
+    // let result = process_batch(&mut request_socket, batch).unwrap();
+    // let handle = thread::spawn(move || {
+    // let mut buffer = vec![0; 1024];
+    // loop {
+    // let n = stdio_socket.read(&mut buffer).unwrap();
+    // if n != 0 {
+    // println!("From stdio {} - {}", n, from_utf8(&buffer[..n]).unwrap());
+    // }
+    // }
+    // });
+    // println!("{result}");
+    // assert!(result.is_ok());
+    // // handle.join().unwrap();
+    // thread::sleep(Duration::from_secs(1));
+    // }
+
+    // #[test]
+    // fn consecutive_compile_execute() {}
 }