feat: dispatch_json in core

cli/ops/plugins.rs

@@ -8,6 +8,7 @@ use std::collections::HashMap;
 use std::ffi::OsStr;
 use std::path::Path;
 use std::sync::Arc;
+use std::sync::Mutex;
 
 pub fn init(
   i: &mut Isolate,
@@ -38,6 +39,16 @@ impl Resource for PluginResource {}
 
 struct InitContext {
   ops: HashMap<String, Box<OpDispatcher>>,
+  state: PluginState,
+}
+
+impl InitContext {
+  fn new(resource_table: Arc<Mutex<ResourceTable>>) -> Self {
+    Self {
+      ops: HashMap::new(),
+      state: PluginState { resource_table },
+    }
+  }
 }
 
 impl PluginInitContext for InitContext {
@@ -48,6 +59,10 @@ impl PluginInitContext for InitContext {
       format!("Op already registered: {}", name)
     );
   }
+
+  fn state(&self) -> PluginState {
+    self.state.clone()
+  }
 }
 
 #[derive(Deserialize)]
@@ -81,9 +96,7 @@ pub fn op_open_plugin(
       .lib
       .symbol::<PluginInitFn>("deno_plugin_init")
   }?;
-  let mut init_context = InitContext {
-    ops: HashMap::new(),
-  };
+  let mut init_context = InitContext::new(state.resource_table.clone());
   init_fn(&mut init_context);
   for op in init_context.ops {
     // Register each plugin op in the `OpRegistry` with the name

cli/state.rs

@@ -58,7 +58,7 @@ pub struct State {
   pub next_worker_id: AtomicUsize,
   pub start_time: Instant,
   pub seeded_rng: Option<Mutex<StdRng>>,
-  pub resource_table: Mutex<ResourceTable>,
+  pub resource_table: Arc<Mutex<ResourceTable>>,
 }
 
 impl Clone for ThreadSafeState {
@@ -262,8 +262,7 @@ impl ThreadSafeState {
       next_worker_id: AtomicUsize::new(0),
       start_time: Instant::now(),
       seeded_rng,
-
-      resource_table: Mutex::new(ResourceTable::default()),
+      resource_table: Arc::new(Mutex::new(ResourceTable::default())),
     };
 
     Ok(ThreadSafeState(Arc::new(state)))

core/Cargo.toml

@@ -22,6 +22,9 @@ log = "0.4.8"
 serde_json = "1.0.44"
 url = "2.1.0"
 rusty_v8 = "0.2.0"
+# dispatch json stuff
+serde = { version = "1.0.102", features = ["derive"] }
+serde_derive = "1.0.102" 
 
 [[example]]
 name = "deno_core_http_bench"
@@ -31,3 +34,9 @@ path = "examples/http_bench.rs"
 [dev_dependencies]
 tokio = { version = "0.2", features = ["full"] }
 num_cpus = "1.11.1"
+
+[features]
+default = []
+dispatch_json = [ "default" ]
+dispatch_json_ts = [ "dispatch_json" ]
+dispatch_json_js = [ "dispatch_json" ]

core/dispatch_json.js

@@ -0,0 +1,114 @@
+// Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
+
+// These utilities are borrowed from std. We don't really
+// have a better way to include them here yet.
+class AssertionError extends Error {
+  constructor(message) {
+    super(message);
+    this.name = "AssertionError";
+  }
+}
+
+/** Make an assertion, if not `true`, then throw. */
+function assert(expr, msg = "") {
+  if (!expr) {
+    throw new AssertionError(msg);
+  }
+}
+
+/** Creates a Promise with the `reject` and `resolve` functions
+ * placed as methods on the promise object itself. It allows you to do:
+ *
+ *     const p = deferred<number>();
+ *     // ...
+ *     p.resolve(42);
+ */
+function deferred() {
+  let methods;
+  const promise = new Promise((resolve, reject) => {
+    methods = { resolve, reject };
+  });
+  return Object.assign(promise, methods);
+}
+
+function decode(ui8) {
+  const s = new TextDecoder().decode(ui8);
+  return JSON.parse(s);
+}
+
+function encode(args) {
+  const s = JSON.stringify(args);
+  return new TextEncoder().encode(s);
+}
+
+function unwrapResponse(res) {
+  if (res.err != null) {
+    throw new Error(res.err.message);
+  }
+  assert(res.ok != null);
+  return res.ok;
+}
+
+class DispatchJsonOp {
+  constructor(dispatch) {
+    this.dispatch = dispatch;
+    this.promiseTable = new Map();
+    this._nextPromiseId = 1;
+  }
+
+  nextPromiseId() {
+    return this._nextPromiseId++;
+  }
+
+  handleAsync(resUi8) {
+    const res = decode(resUi8);
+    assert(res.promiseId != null);
+    const promise = this.promiseTable.get(res.promiseId);
+    assert(promise != null);
+    this.promiseTable.delete(res.promiseId);
+    promise.resolve(res);
+  }
+
+  dispatchSync(args = {}, zeroCopy) {
+    const argsUi8 = encode(args);
+    const resUi8 = this.dispatch(argsUi8, zeroCopy);
+    assert(resUi8 != null);
+    const res = decode(resUi8);
+    assert(res.promiseId == null);
+    return unwrapResponse(res);
+  }
+
+  async dispatchAsync(args = {}, zeroCopy) {
+    const promiseId = this.nextPromiseId();
+    args = Object.assign(args, { promiseId });
+    const promise = deferred();
+    const argsUi8 = encode(args);
+    const buf = this.dispatch(argsUi8, zeroCopy);
+    if (buf) {
+      // Sync result.
+      const res = decode(buf);
+      promise.resolve(res);
+    } else {
+      // Async result.
+      this.promiseTable.set(promiseId, promise);
+    }
+    const res = await promise;
+    return unwrapResponse(res);
+  }
+}
+
+export class DispatchJsonCoreOp extends DispatchJsonOp {
+  constructor(opId) {
+    super((c, zc) => Deno["core"].dispatch(this.opId, c, zc));
+    this.opId = opId;
+    Deno["core"].setAsyncHandler(this.opId, resUi8 => this.handleAsync(resUi8));
+  }
+}
+
+export class DispatchJsonPluginOp extends DispatchJsonOp {
+  constructor(pluginOp) {
+    super((c, zc) => this.pluginOp.dispatch(c, zc));
+    this.pluginOp = pluginOp;
+    this.pluginOp.setAsyncHandler(resUi8 => this.handleAsync(resUi8));
+  }
+}

core/dispatch_json.rs

@@ -0,0 +1,114 @@
+// Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
+// Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
+use crate::any_error::ErrBox;
+use crate::isolate::ZeroCopyBuf;
+use crate::ops::Buf;
+use crate::ops::CoreOp;
+use futures::future::FutureExt;
+use futures::task::SpawnExt;
+pub use serde_derive::Deserialize;
+use serde_json::json;
+pub use serde_json::Value;
+use std::future::Future;
+use std::pin::Pin;
+
+#[cfg(feature = "dispatch_json_ts")]
+pub static DISPATCH_JSON_TS: &'static str = include_str!("dispatch_json.ts");
+
+#[cfg(feature = "dispatch_json_js")]
+pub static DISPATCH_JSON_JS: &'static str = include_str!("dispatch_json.js");
+
+pub type AsyncJsonOp =
+  Pin<Box<dyn Future<Output = Result<Value, ErrBox>> + Send>>;
+
+pub enum JsonOp {
+  Sync(Value),
+  Async(AsyncJsonOp),
+}
+
+fn json_err(err: ErrBox) -> Value {
+  json!({
+    "message": err.to_string(),
+  })
+}
+
+fn serialize_result(
+  promise_id: Option<u64>,
+  result: Result<Value, ErrBox>,
+) -> Buf {
+  let value = match result {
+    Ok(v) => json!({ "ok": v, "promiseId": promise_id }),
+    Err(err) => json!({ "err": json_err(err), "promiseId": promise_id }),
+  };
+  let mut vec = serde_json::to_vec(&value).unwrap();
+  debug!("JSON response pre-align, len={}", vec.len());
+  // Align to 32bit word, padding with the space character.
+  vec.resize((vec.len() + 3usize) & !3usize, b' ');
+  vec.into_boxed_slice()
+}
+
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct AsyncArgs {
+  promise_id: Option<u64>,
+}
+
+pub fn json_op<D>(d: D) -> impl Fn(&[u8], Option<ZeroCopyBuf>) -> CoreOp
+where
+  D: Fn(Value, Option<ZeroCopyBuf>) -> Result<JsonOp, ErrBox>,
+{
+  move |control: &[u8], zero_copy: Option<ZeroCopyBuf>| {
+    let async_args: AsyncArgs = match serde_json::from_slice(control) {
+      Ok(args) => args,
+      Err(e) => {
+        let buf = serialize_result(None, Err(ErrBox::from(e)));
+        return CoreOp::Sync(buf);
+      }
+    };
+    let promise_id = async_args.promise_id;
+    let is_sync = promise_id.is_none();
+
+    let result = serde_json::from_slice(control)
+      .map_err(ErrBox::from)
+      .and_then(|args| d(args, zero_copy));
+
+    // Convert to CoreOp
+    match result {
+      Ok(JsonOp::Sync(sync_value)) => {
+        assert!(promise_id.is_none());
+        CoreOp::Sync(serialize_result(promise_id, Ok(sync_value)))
+      }
+      Ok(JsonOp::Async(fut)) => {
+        assert!(promise_id.is_some());
+        let fut2 = fut.then(move |result| {
+          futures::future::ok(serialize_result(promise_id, result))
+        });
+        CoreOp::Async(fut2.boxed())
+      }
+      Err(sync_err) => {
+        let buf = serialize_result(promise_id, Err(sync_err));
+        if is_sync {
+          CoreOp::Sync(buf)
+        } else {
+          CoreOp::Async(futures::future::ok(buf).boxed())
+        }
+      }
+    }
+  }
+}
+
+pub fn blocking_json<F>(is_sync: bool, f: F) -> Result<JsonOp, ErrBox>
+where
+  F: 'static + Send + FnOnce() -> Result<Value, ErrBox> + Unpin,
+{
+  if is_sync {
+    Ok(JsonOp::Sync(f()?))
+  } else {
+    //TODO(afinch7) replace this with something more efficent.
+    let pool = futures::executor::ThreadPool::new().unwrap();
+    let handle = pool
+      .spawn_with_handle(futures::future::lazy(move |_cx| f()))
+      .unwrap();
+    Ok(JsonOp::Async(handle.boxed()))
+  }
+}