Basic WebWorker implementation.

Only accessible from Rust currently.

BUILD.gn

@@ -102,6 +102,7 @@ ts_sources = [
   "js/types.ts",
   "js/url_search_params.ts",
   "js/util.ts",
+  "js/workers.ts",
   "js/write_file.ts",
   "tsconfig.json",
 

js/globals.ts

@@ -17,6 +17,7 @@ import * as headers from "./headers";
 import * as textEncoding from "./text_encoding";
 import * as timers from "./timers";
 import * as urlSearchParams from "./url_search_params";
+import * as workers from "./workers";
 
 // These imports are not exposed and therefore are fine to just import the
 // symbols required.
@@ -71,3 +72,8 @@ window.TextEncoder = textEncoding.TextEncoder;
 export type TextEncoder = textEncoding.TextEncoder;
 window.TextDecoder = textEncoding.TextDecoder;
 export type TextDecoder = textEncoding.TextDecoder;
+
+window.workerMain = workers.workerMain;
+// TODO These shouldn't be available in main isolate.
+window.postMessage = workers.postMessage;
+window.close = workers.workerClose;

js/workers.ts

@@ -0,0 +1,72 @@
+import * as dispatch from "./dispatch";
+import * as msg from "gen/msg_generated";
+import * as flatbuffers from "./flatbuffers";
+import { window } from "./globals";
+import { libdeno } from "./libdeno";
+import { assert, log, setLogDebug } from "./util";
+
+export async function postMessage(data: Uint8Array): Promise<void> {
+  const builder = flatbuffers.createBuilder();
+  msg.WorkerPostMessage.startWorkerPostMessage(builder);
+  const inner = msg.WorkerPostMessage.endWorkerPostMessage(builder);
+  const baseRes = await dispatch.sendAsync(
+    builder,
+    msg.Any.WorkerPostMessage,
+    inner,
+    data
+  );
+  assert(baseRes != null);
+}
+
+export async function getMessage(): Promise<null | Uint8Array> {
+  log("getMessage");
+  // Send CodeFetch message
+  const builder = flatbuffers.createBuilder();
+  msg.WorkerGetMessage.startWorkerGetMessage(builder);
+  const inner = msg.WorkerGetMessage.endWorkerGetMessage(builder);
+  const baseRes = await dispatch.sendAsync(
+    builder,
+    msg.Any.WorkerGetMessage,
+    inner
+  );
+  assert(baseRes != null);
+  assert(
+    msg.Any.WorkerGetMessageRes === baseRes!.innerType(),
+    `base.innerType() unexpectedly is ${baseRes!.innerType()}`
+  );
+  const res = new msg.WorkerGetMessageRes();
+  assert(baseRes!.inner(res) != null);
+
+  const dataArray = res.dataArray();
+  if (dataArray == null) {
+    return null;
+  } else {
+    return new Uint8Array(dataArray!);
+  }
+}
+
+let isClosing = false;
+
+export function workerClose(): void {
+  isClosing = true;
+}
+
+export async function workerMain() {
+  libdeno.recv(dispatch.handleAsyncMsgFromRust);
+  setLogDebug(true);
+  log("workerMain");
+
+  while (!isClosing) {
+    let data = await getMessage();
+    if (data == null) {
+      log("workerMain got null message. quitting.");
+      break;
+    }
+    if (window["onmessage"]) {
+      let event = { data };
+      window.onmessage(event);
+    } else {
+      break;
+    }
+  }
+}

src/isolate.rs

@@ -12,6 +12,7 @@ use js_errors::JSError;
 use libdeno;
 use permissions::DenoPermissions;
 
+use futures::sync::mpsc as async_mpsc;
 use futures::Future;
 use libc::c_void;
 use std;
@@ -22,6 +23,7 @@ use std::ffi::CString;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::mpsc;
 use std::sync::Arc;
+use std::sync::Mutex;
 use std::time::Duration;
 use std::time::Instant;
 use tokio;
@@ -52,6 +54,10 @@ pub struct Isolate {
   pub state: Arc<IsolateState>,
 }
 
+pub type WorkerSender = async_mpsc::Sender<Buf>;
+pub type WorkerReceiver = async_mpsc::Receiver<Buf>;
+pub type WorkerChannels = (WorkerSender, WorkerReceiver);
+
 // Isolate cannot be passed between threads but IsolateState can.
 // IsolateState satisfies Send and Sync.
 // So any state that needs to be accessed outside the main V8 thread should be
@@ -63,17 +69,24 @@ pub struct IsolateState {
   pub permissions: DenoPermissions,
   pub flags: flags::DenoFlags,
   pub metrics: Metrics,
+  pub worker_channels: Option<Mutex<WorkerChannels>>,
 }
 
 impl IsolateState {
-  pub fn new(flags: flags::DenoFlags, argv_rest: Vec<String>) -> Self {
+  pub fn new(
+    flags: flags::DenoFlags,
+    argv_rest: Vec<String>,
+    worker_channels: Option<WorkerChannels>,
+  ) -> Self {
     let custom_root = env::var("DENO_DIR").map(|s| s.into()).ok();
+
     Self {
       dir: deno_dir::DenoDir::new(flags.reload, custom_root).unwrap(),
       argv: argv_rest,
       permissions: DenoPermissions::new(&flags),
       flags,
       metrics: Metrics::default(),
+      worker_channels: worker_channels.map(|wc| Mutex::new(wc)),
     }
   }
 
@@ -397,7 +410,7 @@ mod tests {
     let argv = vec![String::from("./deno"), String::from("hello.js")];
     let (flags, rest_argv, _) = flags::set_flags(argv).unwrap();
 
-    let state = Arc::new(IsolateState::new(flags, rest_argv));
+    let state = Arc::new(IsolateState::new(flags, rest_argv, None));
     let snapshot = libdeno::deno_buf::empty();
     let isolate = Isolate::new(snapshot, state, dispatch_sync);
     tokio_util::init(|| {
@@ -438,7 +451,7 @@ mod tests {
   fn test_metrics_sync() {
     let argv = vec![String::from("./deno"), String::from("hello.js")];
     let (flags, rest_argv, _) = flags::set_flags(argv).unwrap();
-    let state = Arc::new(IsolateState::new(flags, rest_argv));
+    let state = Arc::new(IsolateState::new(flags, rest_argv, None));
     let snapshot = libdeno::deno_buf::empty();
     let isolate = Isolate::new(snapshot, state, metrics_dispatch_sync);
     tokio_util::init(|| {
@@ -474,7 +487,7 @@ mod tests {
   fn test_metrics_async() {
     let argv = vec![String::from("./deno"), String::from("hello.js")];
     let (flags, rest_argv, _) = flags::set_flags(argv).unwrap();
-    let state = Arc::new(IsolateState::new(flags, rest_argv));
+    let state = Arc::new(IsolateState::new(flags, rest_argv, None));
     let snapshot = libdeno::deno_buf::empty();
     let isolate = Isolate::new(snapshot, state, metrics_dispatch_async);
     tokio_util::init(|| {

src/main.rs

@@ -47,6 +47,7 @@ pub mod snapshot;
 mod tokio_util;
 mod tokio_write;
 pub mod version;
+pub mod workers;
 
 #[cfg(unix)]
 mod eager_unix;
@@ -96,7 +97,7 @@ fn main() {
     log::LevelFilter::Info
   });
 
-  let state = Arc::new(isolate::IsolateState::new(flags, rest_argv));
+  let state = Arc::new(isolate::IsolateState::new(flags, rest_argv, None));
   let snapshot = snapshot::deno_snapshot();
   let isolate = isolate::Isolate::new(snapshot, state, ops::dispatch);
   tokio_util::init(|| {

src/msg.fbs

@@ -1,6 +1,9 @@
 union Any {
   Start,
   StartRes,
+  WorkerGetMessage,
+  WorkerGetMessageRes,
+  WorkerPostMessage,
   CodeFetch,
   CodeFetchRes,
   CodeCache,
@@ -145,6 +148,18 @@ table StartRes {
   v8_version: string;
 }
 
+table WorkerGetMessage {
+  unused: int8;
+}
+
+table WorkerGetMessageRes {
+  data: [ubyte];
+}
+
+table WorkerPostMessage {
+  // data passed thru the zero-copy data parameter.
+}
+
 table CodeFetch {
   specifier: string;
   referrer: string;