// Copyright 2018-2019 the Deno authors. All rights reserved. MIT license. use core::ops::Deref; use crate::flags::DenoFlags; use crate::isolate_state::*; use crate::js_errors; use crate::msg; use crate::ops; use crate::resources; use crate::resources::ResourceId; use crate::startup_data; use crate::workers; use crate::workers::WorkerBehavior; use crate::workers::WorkerInit; use deno::deno_buf; use deno::Behavior; use deno::Buf; use deno::JSError; use deno::Op; use deno::StartupData; use futures::future::*; use futures::sync::oneshot; use futures::Future; use futures::Stream; use serde_json; use std::collections::HashMap; use std::str; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; use std::sync::Arc; use std::sync::Mutex; use tokio::runtime::Runtime; /// Used for normalization of types on internal future completions type CompilerInnerResult = Result>; type WorkerErrReceiver = oneshot::Receiver; type CmdId = u32; type ResponseSenderTable = HashMap>; /// Shared resources for used to complete compiler operations. /// rid is the resource id for compiler worker resource used for sending it /// compile requests /// worker_err_receiver is a shared future that will compelete when the /// compiler worker future completes, and send back an error if present /// or a None if not #[derive(Clone)] struct CompilerShared { pub rid: ResourceId, pub worker_err_receiver: Shared, } lazy_static! { static ref C_NEXT_CMD_ID: AtomicUsize = AtomicUsize::new(1); // Map of response senders static ref C_RES_SENDER_TABLE: Mutex = Mutex::new(ResponseSenderTable::new()); // Shared worker resources so we can spawn static ref C_SHARED: Mutex> = Mutex::new(None); // tokio runtime specifically for spawning logic that is dependent on // completetion of the compiler worker future static ref C_RUNTIME: Mutex = Mutex::new(Runtime::new().unwrap()); } pub struct CompilerBehavior { pub state: Arc, } impl CompilerBehavior { pub fn new(flags: DenoFlags, argv_rest: Vec) -> Self { Self { state: Arc::new(IsolateState::new(flags, argv_rest, None, true)), } } } impl IsolateStateContainer for CompilerBehavior { fn state(&self) -> Arc { self.state.clone() } } impl IsolateStateContainer for &CompilerBehavior { fn state(&self) -> Arc { self.state.clone() } } impl Behavior for CompilerBehavior { fn startup_data(&mut self) -> Option { Some(startup_data::compiler_isolate_init()) } fn dispatch( &mut self, control: &[u8], zero_copy: deno_buf, ) -> (bool, Box) { ops::dispatch_all(self, control, zero_copy, ops::op_selector_compiler) } } impl WorkerBehavior for CompilerBehavior { fn set_internal_channels(&mut self, worker_channels: WorkerChannels) { self.state = Arc::new(IsolateState::new( self.state.flags.clone(), self.state.argv.clone(), Some(worker_channels), true, )); } } // This corresponds to JS ModuleMetaData. // TODO Rename one or the other so they correspond. #[derive(Debug, Clone)] pub struct ModuleMetaData { pub module_name: String, pub module_redirect_source_name: Option, // source of redirect pub filename: String, pub media_type: msg::MediaType, pub source_code: Vec, pub maybe_output_code_filename: Option, pub maybe_output_code: Option>, pub maybe_source_map_filename: Option, pub maybe_source_map: Option>, } impl ModuleMetaData { pub fn has_output_code_and_source_map(&self) -> bool { self.maybe_output_code.is_some() && self.maybe_source_map.is_some() } pub fn js_source(&self) -> String { if self.media_type == msg::MediaType::Json { return format!( "export default {};", str::from_utf8(&self.source_code).unwrap() ); } match self.maybe_output_code { None => str::from_utf8(&self.source_code).unwrap().to_string(), Some(ref output_code) => str::from_utf8(output_code).unwrap().to_string(), } } } fn new_cmd_id() -> CmdId { let next_rid = C_NEXT_CMD_ID.fetch_add(1, Ordering::SeqCst); next_rid as CmdId } fn parse_cmd_id(res_json: &str) -> CmdId { match serde_json::from_str::(res_json) { Ok(serde_json::Value::Object(map)) => match map["cmdId"].as_u64() { Some(cmd_id) => cmd_id as CmdId, _ => panic!("Error decoding compiler response: expected cmdId"), }, _ => panic!("Error decoding compiler response"), } } fn lazy_start(parent_state: Arc) -> CompilerShared { let mut cell = C_SHARED.lock().unwrap(); cell .get_or_insert_with(|| { let worker_result = workers::spawn( CompilerBehavior::new( parent_state.flags.clone(), parent_state.argv.clone(), ), "TS", WorkerInit::Script("compilerMain()".to_string()), ); match worker_result { Ok(worker) => { let rid = worker.resource.rid; // create oneshot channels and use the sender to pass back // results from worker future let (err_sender, err_receiver) = oneshot::channel::(); let mut runtime = C_RUNTIME.lock().unwrap(); runtime.spawn(lazy(move || { let resource = worker.resource.clone(); worker.then(move |result| -> Result<(), ()> { // Close resource so the future created by // handle_worker_message_stream exits resource.close(); match result { Err(err) => err_sender.send(Err(Some(err))).unwrap(), _ => err_sender.send(Err(None)).unwrap(), }; Ok(()) }) })); runtime.spawn(lazy(move || { debug!("Start worker stream handler!"); let worker_stream = resources::get_message_stream_from_worker(rid); worker_stream .for_each(|msg: Buf| { // All worker responses are handled here first before being sent via // their respective sender. This system can be compared to the // promise system used on the js side. This provides a way to // resolve many futures via the same channel. let res_json = std::str::from_utf8(&msg).unwrap(); debug!("Got message from worker: {}", res_json); // Get the intended receiver's cmd_id from the message. let cmd_id = parse_cmd_id(res_json); let mut table = C_RES_SENDER_TABLE.lock().unwrap(); debug!("Cmd id for get message handler: {}", cmd_id); // Get the corresponding response sender from the table and // send a response. let response_sender = table.remove(&(cmd_id as CmdId)).unwrap(); response_sender.send(msg).unwrap(); Ok(()) }).map_err(|_| ()) })); CompilerShared { rid, worker_err_receiver: err_receiver.shared(), } } Err(err) => { println!("{}", err.to_string()); std::process::exit(1); } } }).clone() } fn req( specifier: &str, referrer: &str, is_worker_main: bool, cmd_id: u32, ) -> Buf { json!({ "specifier": specifier, "referrer": referrer, "isWorker": is_worker_main, "cmdId": cmd_id, }).to_string() .into_boxed_str() .into_boxed_bytes() } pub fn compile_sync( parent_state: Arc, specifier: &str, referrer: &str, module_meta_data: &ModuleMetaData, ) -> Result { debug!( "Running rust part of compile_sync. specifier: {}, referrer: {}", &specifier, &referrer ); let cmd_id = new_cmd_id(); let req_msg = req(specifier, referrer, parent_state.is_worker, cmd_id); let module_meta_data_ = module_meta_data.clone(); let shared = lazy_start(parent_state.clone()); let compiler_rid = shared.rid; let (local_sender, local_receiver) = oneshot::channel::>>(); let (response_sender, response_receiver) = oneshot::channel::(); // Scoping to auto dispose of locks when done using them { let mut table = C_RES_SENDER_TABLE.lock().unwrap(); debug!("Cmd id for response sender insert: {}", cmd_id); // Place our response sender in the table so we can find it later. table.insert(cmd_id, response_sender); let mut runtime = C_RUNTIME.lock().unwrap(); runtime.spawn(lazy(move || { resources::post_message_to_worker(compiler_rid, req_msg) .then(move |_| { debug!("Sent message to worker"); response_receiver.map_err(|_| None) }).and_then(move |res_msg| { debug!("Received message from worker"); let res_json = std::str::from_utf8(res_msg.as_ref()).unwrap(); let res = serde_json::from_str::(res_json) .expect("Error decoding compiler response"); let res_data = res["data"].as_object().expect( "Error decoding compiler response: expected object field 'data'", ); match res["success"].as_bool() { Some(true) => Ok(ModuleMetaData { maybe_output_code: res_data["outputCode"] .as_str() .map(|s| s.as_bytes().to_owned()), maybe_source_map: res_data["sourceMap"] .as_str() .map(|s| s.as_bytes().to_owned()), ..module_meta_data_ }), Some(false) => { let js_error = JSError::from_json_value( serde_json::Value::Object(res_data.clone()), ).expect( "Error decoding compiler response: failed to parse error", ); Err(Some(js_errors::apply_source_map( &js_error, &parent_state.dir, ))) } _ => panic!( "Error decoding compiler response: expected bool field 'success'" ), } }).then(move |result| { local_sender.send(result).expect("Oneshot send() failed"); Ok(()) }) })); } let worker_receiver = shared.worker_err_receiver.clone(); let union = futures::future::select_all(vec![worker_receiver, local_receiver.shared()]); match union.wait() { Ok((result, i, rest)) => { // We got a sucessful finish before any recivers where canceled let mut rest_mut = rest; match ((*result.deref()).clone(), i) { // Either receiver was completed with success. (Ok(v), _) => Ok(v), // Either receiver was completed with a valid error // this should be fatal for now since it is not intended // to be possible to recover from a uncaught error in a isolate (Err(Some(err)), _) => Err(err), // local_receiver finished first with a none error. This is intended // to catch when the local logic can't complete because it is unable // to send and/or receive messages from the compiler worker. // Due to the way that scheduling works it is very likely that the // compiler worker future has already or will in the near future // complete with a valid JSError or a None. (Err(None), 1) => { debug!("Compiler local exited with None error!"); // While technically possible to get stuck here indefinately // in theory it is highly unlikely. debug!( "Waiting on compiler worker result specifier: {} referrer: {}!", specifier, referrer ); let worker_result = (*rest_mut.remove(0).wait().unwrap().deref()).clone(); debug!( "Finished waiting on worker result specifier: {} referrer: {}!", specifier, referrer ); match worker_result { Err(Some(err)) => Err(err), Err(None) => panic!("Compiler exit for an unknown reason!"), Ok(v) => Ok(v), } } // While possible beccause the compiler worker can exit without error // this shouldn't occurr normally and I don't intend to attempt to // handle it right now (_, i) => panic!("Odd compiler result for future {}!", i), } } // This should always a result of a reciver being cancled // in theory but why not give a print out just in case Err((err, i, _)) => panic!("compile_sync {} failed: {}", i, err), } } #[cfg(test)] mod tests { use super::*; #[test] fn test_compile_sync() { let cwd = std::env::current_dir().unwrap(); let cwd_string = cwd.to_str().unwrap().to_owned(); let specifier = "./tests/002_hello.ts"; let referrer = cwd_string + "/"; let mut out = ModuleMetaData { module_name: "xxx".to_owned(), module_redirect_source_name: None, filename: "/tests/002_hello.ts".to_owned(), media_type: msg::MediaType::TypeScript, source_code: include_bytes!("../tests/002_hello.ts").to_vec(), maybe_output_code_filename: None, maybe_output_code: None, maybe_source_map_filename: None, maybe_source_map: None, }; out = compile_sync(Arc::new(IsolateState::mock()), specifier, &referrer, &out) .unwrap(); assert!( out .maybe_output_code .unwrap() .starts_with("console.log(\"Hello World\");".as_bytes()) ); } #[test] fn test_parse_cmd_id() { let cmd_id = new_cmd_id(); let msg = req("Hello", "World", false, cmd_id); let res_json = std::str::from_utf8(&msg).unwrap(); assert_eq!(parse_cmd_id(res_json), cmd_id); } }