Web workers

cli/compiler.rs

@@ -8,6 +8,7 @@ use crate::resources::ResourceId;
 use crate::startup_data;
 use crate::workers;
 use crate::workers::WorkerBehavior;
+use crate::workers::WorkerInit;
 use deno_core::deno_buf;
 use deno_core::deno_mod;
 use deno_core::Behavior;
@@ -70,6 +71,7 @@ impl WorkerBehavior for CompilerBehavior {
       self.state.flags.clone(),
       self.state.argv.clone(),
       Some(worker_channels),
+      None,
     ));
   }
 }
@@ -111,8 +113,9 @@ fn lazy_start(parent_state: Arc<IsolateState>) -> Resource {
         parent_state.flags.clone(),
         parent_state.argv.clone(),
         None,
+        None,
       ))),
-      "compilerMain()".to_string(),
+      WorkerInit::Script("compilerMain()".to_string()),
     );
     resource.rid
   });
@@ -138,10 +141,10 @@ pub fn compile_sync(
 
   let compiler = lazy_start(parent_state);
 
-  let send_future = resources::worker_post_message(compiler.rid, req_msg);
+  let send_future = resources::post_message_to_worker(compiler.rid, req_msg);
   send_future.wait().unwrap();
 
-  let recv_future = resources::worker_recv_message(compiler.rid);
+  let recv_future = resources::get_message_from_worker(compiler.rid);
   let result = recv_future.wait().unwrap();
   assert!(result.is_some());
   let res_msg = result.unwrap();

cli/isolate.rs

@@ -216,7 +216,7 @@ mod tests {
     let argv = vec![String::from("./deno"), filename.clone()];
     let (flags, rest_argv, _) = flags::set_flags(argv).unwrap();
 
-    let state = Arc::new(IsolateState::new(flags, rest_argv, None));
+    let state = Arc::new(IsolateState::new(flags, rest_argv, None, None));
     let state_ = state.clone();
     tokio_util::run(lazy(move || {
       let cli = CliBehavior::new(None, state.clone());
@@ -239,7 +239,7 @@ mod tests {
     let argv = vec![String::from("./deno"), filename.clone()];
     let (flags, rest_argv, _) = flags::set_flags(argv).unwrap();
 
-    let state = Arc::new(IsolateState::new(flags, rest_argv, None));
+    let state = Arc::new(IsolateState::new(flags, rest_argv, None, None));
     let state_ = state.clone();
     tokio_util::run(lazy(move || {
       let cli = CliBehavior::new(None, state.clone());

cli/isolate_state.rs

@@ -43,13 +43,21 @@ pub struct IsolateState {
   pub modules: Mutex<Modules>,
   pub worker_channels: Option<Mutex<WorkerChannels>>,
   pub global_timer: Mutex<GlobalTimer>,
+  pub main_module_specifier: Option<String>,
 }
 
+// TODO(afinch7) abstract out IsolateState to a trait, since
+// some isolates don't require or shouldn't be given some
+// state data
 impl IsolateState {
+  /// main_module should be a fully qualified url
+  /// argv will be used to determine main module
+  /// if not sepcified
   pub fn new(
     flags: flags::DenoFlags,
     argv_rest: Vec<String>,
     worker_channels: Option<WorkerChannels>,
+    main_module_specifier: Option<String>,
   ) -> Self {
     let custom_root = env::var("DENO_DIR").map(|s| s.into()).ok();
 
@@ -63,20 +71,37 @@ impl IsolateState {
       modules: Mutex::new(Modules::new()),
       worker_channels: worker_channels.map(Mutex::new),
       global_timer: Mutex::new(GlobalTimer::new()),
+      main_module_specifier,
     }
   }
 
+  /// Attempt to resolve explicit main module specifier from creation
+  /// if present or extract it from argv_rest if not
   pub fn main_module(&self) -> Option<String> {
-    if self.argv.len() <= 1 {
-      None
-    } else {
-      let specifier = self.argv[1].clone();
-      let referrer = ".";
-      match self.dir.resolve_module_url(&specifier, referrer) {
-        Ok(url) => Some(url.to_string()),
-        Err(e) => {
-          debug!("Potentially swallowed error {}", e);
+    match &self.main_module_specifier {
+      Some(specifier) => {
+        let referrer = ".";
+        match self.dir.resolve_module_url(&specifier, referrer) {
+          Ok(url) => Some(url.to_string()),
+          Err(e) => {
+            debug!("Potentially swallowed error {}", e);
+            None
+          }
+        }
+      }
+      None => {
+        if self.argv.len() <= 1 {
           None
+        } else {
+          let specifier = self.argv[1].clone();
+          let referrer = ".";
+          match self.dir.resolve_module_url(&specifier, referrer) {
+            Ok(url) => Some(url.to_string()),
+            Err(e) => {
+              debug!("Potentially swallowed error {}", e);
+              None
+            }
+          }
         }
       }
     }
@@ -112,7 +137,7 @@ impl IsolateState {
     let argv = vec![String::from("./deno"), String::from("hello.js")];
     // For debugging: argv.push_back(String::from("-D"));
     let (flags, rest_argv, _) = flags::set_flags(argv).unwrap();
-    IsolateState::new(flags, rest_argv, None)
+    IsolateState::new(flags, rest_argv, None, None)
   }
 
   pub fn metrics_op_dispatched(

cli/main.rs

@@ -33,6 +33,7 @@ mod startup_data;
 mod tokio_util;
 mod tokio_write;
 pub mod version;
+pub mod web_worker_behavior;
 pub mod workers;
 
 use crate::cli_behavior::CliBehavior;
@@ -108,7 +109,7 @@ fn main() {
   let should_prefetch = flags.prefetch || flags.info;
   let should_display_info = flags.info;
 
-  let state = Arc::new(IsolateState::new(flags, rest_argv, None));
+  let state = Arc::new(IsolateState::new(flags, rest_argv, None, None));
   let state_ = state.clone();
   let startup_data = startup_data::deno_isolate_init();
   let cli = CliBehavior::new(Some(startup_data), state_);

cli/msg.fbs

@@ -65,6 +65,11 @@ union Any {
   StatRes,
   Symlink,
   Truncate,
+  CreateWorker,
+  CreateWorkerRes,
+  HostGetMessage,
+  HostGetMessageRes,
+  HostPostMessage,
   WorkerGetMessage,
   WorkerGetMessageRes,
   WorkerPostMessage,
@@ -174,6 +179,31 @@ table FormatErrorRes {
   error: string;
 }
 
+// Create worker as host
+table CreateWorker {
+  specifier: string;
+}
+
+table CreateWorkerRes {
+  rid: uint32;
+}
+
+// Get message from guest worker as host
+table HostGetMessage {
+  rid: uint32;
+}
+
+table HostGetMessageRes {
+  data: [ubyte];
+}
+
+// Post message to guest worker as host
+table HostPostMessage {
+  rid: uint32;
+  // data passed thru the zero-copy data parameter.
+}
+
+// Get message from host as guest worker
 table WorkerGetMessage {
   unused: int8;
 }
@@ -182,6 +212,7 @@ table WorkerGetMessageRes {
   data: [ubyte];
 }
 
+// Post message to host as guest worker
 table WorkerPostMessage {
   // data passed thru the zero-copy data parameter.
 }

cli/ops.rs

@@ -18,6 +18,8 @@ use crate::resources::Resource;
 use crate::tokio_util;
 use crate::tokio_write;
 use crate::version;
+use crate::web_worker_behavior;
+use crate::workers;
 use deno_core::deno_buf;
 use deno_core::Buf;
 use deno_core::JSError;
@@ -141,13 +143,24 @@ pub fn dispatch_all(
   (base.sync(), boxed_op)
 }
 
+/// Superset of op_selector_worker for compiler isolates
 pub fn op_selector_compiler(inner_type: msg::Any) -> Option<OpCreator> {
   match inner_type {
     msg::Any::FetchModuleMetaData => Some(op_fetch_module_meta_data),
+    _ => op_selector_worker(inner_type),
+  }
+}
+
+/// Superset of op_selector_std for worker isolates
+pub fn op_selector_worker(inner_type: msg::Any) -> Option<OpCreator> {
+  match inner_type {
+    msg::Any::WorkerGetMessage => Some(op_worker_get_message),
+    msg::Any::WorkerPostMessage => Some(op_worker_post_message),
     _ => op_selector_std(inner_type),
   }
 }
 
+/// Standard ops set for most isolates
 pub fn op_selector_std(inner_type: msg::Any) -> Option<OpCreator> {
   match inner_type {
     msg::Any::Accept => Some(op_accept),
@@ -190,8 +203,9 @@ pub fn op_selector_std(inner_type: msg::Any) -> Option<OpCreator> {
     msg::Any::Stat => Some(op_stat),
     msg::Any::Symlink => Some(op_symlink),
     msg::Any::Truncate => Some(op_truncate),
-    msg::Any::WorkerGetMessage => Some(op_worker_get_message),
-    msg::Any::WorkerPostMessage => Some(op_worker_post_message),
+    msg::Any::CreateWorker => Some(op_create_worker),
+    msg::Any::HostGetMessage => Some(op_host_get_message),
+    msg::Any::HostPostMessage => Some(op_host_post_message),
     msg::Any::Write => Some(op_write),
     msg::Any::WriteFile => Some(op_write_file),
     _ => None,
@@ -1806,6 +1820,7 @@ impl Future for GetMessageFuture {
   }
 }
 
+/// Get message from host as guest worker
 fn op_worker_get_message(
   sc: &IsolateStateContainer,
   base: &msg::Base<'_>,
@@ -1840,6 +1855,7 @@ fn op_worker_get_message(
   Box::new(op)
 }
 
+/// Post message to host as guest worker
 fn op_worker_post_message(
   sc: &IsolateStateContainer,
   base: &msg::Base<'_>,
@@ -1872,3 +1888,103 @@ fn op_worker_post_message(
   });
   Box::new(op)
 }
+
+/// Create worker as the host
+fn op_create_worker(
+  sc: &IsolateStateContainer,
+  base: &msg::Base<'_>,
+  data: deno_buf,
+) -> Box<OpWithError> {
+  assert_eq!(data.len(), 0);
+  let cmd_id = base.cmd_id();
+  let inner = base.inner_as_create_worker().unwrap();
+  let specifier = inner.specifier();
+
+  Box::new(futures::future::result(move || -> OpResult {
+    let parent_state = sc.state().clone();
+    let worker_state = Arc::new(IsolateState::new(
+      parent_state.flags.clone(),
+      parent_state.argv.clone(),
+      None,
+      specifier.map(|v| v.to_string()),
+    ));
+    let behavior = web_worker_behavior::WebWorkerBehavior::new(worker_state);
+    let resource = workers::spawn(behavior, workers::WorkerInit::MainModule());
+    let builder = &mut FlatBufferBuilder::new();
+    let msg_inner = msg::CreateWorkerRes::create(
+      builder,
+      &msg::CreateWorkerResArgs { rid: resource.rid },
+    );
+    Ok(serialize_response(
+      cmd_id,
+      builder,
+      msg::BaseArgs {
+        inner: Some(msg_inner.as_union_value()),
+        inner_type: msg::Any::CreateWorkerRes,
+        ..Default::default()
+      },
+    ))
+  }()))
+}
+
+/// Get message from guest worker as host
+fn op_host_get_message(
+  _sc: &IsolateStateContainer,
+  base: &msg::Base<'_>,
+  data: deno_buf,
+) -> Box<OpWithError> {
+  assert_eq!(data.len(), 0);
+  let cmd_id = base.cmd_id();
+  let inner = base.inner_as_host_get_message().unwrap();
+  let rid = inner.rid();
+
+  let op = resources::get_message_from_worker(rid);
+  let op = op.map_err(move |_| -> DenoError { unimplemented!() });
+  let op = op.and_then(move |maybe_buf| -> DenoResult<Buf> {
+    let builder = &mut FlatBufferBuilder::new();
+
+    let data = maybe_buf.as_ref().map(|buf| builder.create_vector(buf));
+    let msg_inner = msg::HostGetMessageRes::create(
+      builder,
+      &msg::HostGetMessageResArgs { data },
+    );
+    Ok(serialize_response(
+      cmd_id,
+      builder,
+      msg::BaseArgs {
+        inner: Some(msg_inner.as_union_value()),
+        inner_type: msg::Any::HostGetMessageRes,
+        ..Default::default()
+      },
+    ))
+  });
+  Box::new(op)
+}
+
+/// Post message to guest worker as host
+fn op_host_post_message(
+  _sc: &IsolateStateContainer,
+  base: &msg::Base<'_>,
+  data: deno_buf,
+) -> Box<OpWithError> {
+  let cmd_id = base.cmd_id();
+  let inner = base.inner_as_host_post_message().unwrap();
+  let rid = inner.rid();
+
+  let d = Vec::from(data.as_ref()).into_boxed_slice();
+
+  let op = resources::post_message_to_worker(rid, d);
+  let op = op.map_err(|e| errors::new(ErrorKind::Other, e.to_string()));
+  let op = op.and_then(move |_| -> DenoResult<Buf> {
+    let builder = &mut FlatBufferBuilder::new();
+
+    Ok(serialize_response(
+      cmd_id,
+      builder,
+      msg::BaseArgs {
+        ..Default::default()
+      },
+    ))
+  });
+  Box::new(op)
+}

cli/resources.rs

@@ -305,7 +305,8 @@ pub fn add_worker(wc: WorkerChannels) -> Resource {
   Resource { rid }
 }
 
-pub fn worker_post_message(
+/// Post message to worker as a host or privilged overlord
+pub fn post_message_to_worker(
   rid: ResourceId,
   buf: Buf,
 ) -> futures::sink::Send<futures::sync::mpsc::Sender<Buf>> {
@@ -341,7 +342,7 @@ impl Future for WorkerReceiver {
   }
 }
 
-pub fn worker_recv_message(rid: ResourceId) -> WorkerReceiver {
+pub fn get_message_from_worker(rid: ResourceId) -> WorkerReceiver {
   WorkerReceiver { rid }
 }