Add Isolate::set_wasm_streaming_callback() (#560)

Add the hook necessary to implement `WebAssembly.compileStreaming()`.

Fixes #556

src/binding.cc

@@ -266,6 +266,11 @@ void v8__Isolate__SetAllowAtomicsWait(v8::Isolate* isolate, bool allow) {
   isolate->SetAllowAtomicsWait(allow);
 }
 
+void v8__Isolate__SetWasmStreamingCallback(v8::Isolate* isolate,
+                                           v8::WasmStreamingCallback callback) {
+  isolate->SetWasmStreamingCallback(callback);
+}
+
 void v8__Isolate__CreateParams__CONSTRUCT(
     uninit_t<v8::Isolate::CreateParams>* buf) {
   construct_in_place<v8::Isolate::CreateParams>(buf);
@@ -2029,6 +2034,38 @@ MaybeBool v8__Module__SetSyntheticModuleExport(const v8::Module& self,
       isolate, ptr_to_local(export_name), ptr_to_local(export_value)));
 }
 
+struct WasmStreamingSharedPtr {
+  std::shared_ptr<v8::WasmStreaming> inner;
+};
+
+static_assert(sizeof(WasmStreamingSharedPtr) <= 2 * sizeof(void*),
+              "std::shared_ptr<v8::WasmStreaming> size mismatch");
+
+void v8__WasmStreaming__Unpack(v8::Isolate* isolate, const v8::Value& value,
+                               WasmStreamingSharedPtr* self) {
+  new(self) WasmStreamingSharedPtr();
+  self->inner = v8::WasmStreaming::Unpack(isolate, ptr_to_local(&value));
+}
+
+void v8__WasmStreaming__shared_ptr_DESTRUCT(WasmStreamingSharedPtr* self) {
+  self->~WasmStreamingSharedPtr();
+}
+
+void v8__WasmStreaming__OnBytesReceived(WasmStreamingSharedPtr* self,
+                                        const uint8_t* data,
+                                        size_t len) {
+  self->inner->OnBytesReceived(data, len);
+}
+
+void v8__WasmStreaming__Finish(WasmStreamingSharedPtr* self) {
+  self->inner->Finish();
+}
+
+void v8__WasmStreaming__Abort(WasmStreamingSharedPtr* self,
+                              const v8::Value* exception) {
+  self->inner->Abort(ptr_to_maybe_local(exception));
+}
+
 using HeapSnapshotCallback = bool (*)(void*, const char*, size_t);
 
 void v8__HeapProfiler__TakeHeapSnapshot(v8::Isolate* isolate,

src/function.rs

@@ -166,7 +166,9 @@ pub struct FunctionCallbackArguments<'s> {
 }
 
 impl<'s> FunctionCallbackArguments<'s> {
-  fn from_function_callback_info(info: *const FunctionCallbackInfo) -> Self {
+  pub(crate) fn from_function_callback_info(
+    info: *const FunctionCallbackInfo,
+  ) -> Self {
     Self {
       info,
       phantom: PhantomData,

src/isolate.rs

@@ -1,10 +1,15 @@
 // Copyright 2019-2020 the Deno authors. All rights reserved. MIT license.
+use crate::function::FunctionCallbackInfo;
 use crate::isolate_create_params::raw;
 use crate::isolate_create_params::CreateParams;
 use crate::promise::PromiseRejectMessage;
 use crate::scope::data::ScopeData;
+use crate::scope::HandleScope;
 use crate::support::BuildTypeIdHasher;
 use crate::support::Opaque;
+use crate::support::UnitType;
+use crate::wasm::trampoline;
+use crate::wasm::WasmStreaming;
 use crate::Context;
 use crate::Function;
 use crate::Local;
@@ -191,6 +196,11 @@ extern "C" {
   );
   fn v8__Isolate__SetAllowAtomicsWait(isolate: *mut Isolate, allow: bool);
 
+  fn v8__Isolate__SetWasmStreamingCallback(
+    isolate: *mut Isolate,
+    callback: extern "C" fn(*const FunctionCallbackInfo),
+  );
+
   fn v8__HeapProfiler__TakeHeapSnapshot(
     isolate: *mut Isolate,
     callback: extern "C" fn(*mut c_void, *const u8, usize) -> bool,
@@ -558,6 +568,20 @@ impl Isolate {
     unsafe { v8__Isolate__SetAllowAtomicsWait(self, allow) }
   }
 
+  /// Embedder injection point for `WebAssembly.compileStreaming(source)`.
+  /// The expectation is that the embedder sets it at most once.
+  ///
+  /// The callback receives the source argument (string, Promise, etc.)
+  /// and an instance of [WasmStreaming]. The [WasmStreaming] instance
+  /// can outlive the callback and is used to feed data chunks to V8
+  /// asynchronously.
+  pub fn set_wasm_streaming_callback<F>(&mut self, _: F)
+  where
+    F: UnitType + Fn(&mut HandleScope, Local<Value>, WasmStreaming),
+  {
+    unsafe { v8__Isolate__SetWasmStreamingCallback(self, trampoline::<F>()) }
+  }
+
   /// Disposes the isolate.  The isolate must not be entered by any
   /// thread to be disposable.
   unsafe fn dispose(&mut self) {

src/lib.rs

@@ -70,6 +70,7 @@ mod typed_array;
 mod value;
 mod value_deserializer;
 mod value_serializer;
+mod wasm;
 
 pub mod inspector;
 pub mod json;
@@ -137,6 +138,7 @@ pub use value_deserializer::ValueDeserializerImpl;
 pub use value_serializer::ValueSerializer;
 pub use value_serializer::ValueSerializerHelper;
 pub use value_serializer::ValueSerializerImpl;
+pub use wasm::WasmStreaming;
 
 // TODO(piscisaureus): Ideally this trait would not be exported.
 pub use support::MapFnTo;

src/wasm.rs

@@ -0,0 +1,105 @@
+// Copyright 2019-2020 the Deno authors. All rights reserved. MIT license.
+
+use crate::function::FunctionCallbackArguments;
+use crate::function::FunctionCallbackInfo;
+use crate::scope::CallbackScope;
+use crate::scope::HandleScope;
+use crate::support::UnitType;
+use crate::Isolate;
+use crate::Local;
+use crate::Value;
+use std::ptr::null;
+use std::ptr::null_mut;
+
+// Type-erased std::shared_ptr<v8::WasmStreaming>. Assumes it's safe
+// to move around (no backlinks). Not generally true for shared_ptrs
+// but it is in this case - other shared_ptrs that point to the same
+// v8::WasmStreaming exist but are managed by V8 and don't leak out.
+//
+// We don't use crate::support::SharedPtr because it only allows
+// immutable borrows and derefs to avoid aliasing but that's not
+// a problem here, only a single instance outside V8 exists.
+//
+// Note: uses *mut u8 rather than e.g. usize to enforce !Send and !Sync.
+#[repr(C)]
+struct WasmStreamingSharedPtr([*mut u8; 2]);
+
+/// The V8 interface for WebAssembly streaming compilation.
+/// When streaming compilation is initiated, V8 passes a [Self]
+/// object to the embedder such that the embedder can pass the
+/// input bytes for streaming compilation to V8.
+#[repr(C)]
+pub struct WasmStreaming(WasmStreamingSharedPtr);
+
+impl WasmStreaming {
+  /// Pass a new chunk of bytes to WebAssembly streaming compilation.
+  pub fn on_bytes_received(&mut self, data: &[u8]) {
+    unsafe {
+      v8__WasmStreaming__OnBytesReceived(&mut self.0, data.as_ptr(), data.len())
+    }
+  }
+
+  /// Should be called after all received bytes where passed to
+  /// [`Self::on_bytes_received()`] to tell V8 that there will be no
+  /// more bytes. Does not have to be called after [`Self::abort()`]
+  /// has been called already.
+  pub fn finish(mut self) {
+    unsafe { v8__WasmStreaming__Finish(&mut self.0) }
+  }
+
+  /// Abort streaming compilation. If {exception} has a value, then the promise
+  /// associated with streaming compilation is rejected with that value. If
+  /// {exception} does not have value, the promise does not get rejected.
+  pub fn abort(mut self, exception: Option<Local<Value>>) {
+    let exception = exception.map(|v| &*v as *const Value).unwrap_or(null());
+    unsafe { v8__WasmStreaming__Abort(&mut self.0, exception) }
+  }
+}
+
+impl Drop for WasmStreaming {
+  fn drop(&mut self) {
+    unsafe { v8__WasmStreaming__shared_ptr_DESTRUCT(&mut self.0) }
+  }
+}
+
+pub(crate) fn trampoline<F>() -> extern "C" fn(*const FunctionCallbackInfo)
+where
+  F: UnitType + Fn(&mut HandleScope, Local<Value>, WasmStreaming),
+{
+  extern "C" fn c_fn<F>(info: *const FunctionCallbackInfo)
+  where
+    F: UnitType + Fn(&mut HandleScope, Local<Value>, WasmStreaming),
+  {
+    let scope = &mut unsafe { CallbackScope::new(&*info) };
+    let args = FunctionCallbackArguments::from_function_callback_info(info);
+    let data = args.data().unwrap(); // Always present.
+    let data = &*data as *const Value;
+    let zero = null_mut();
+    let mut that = WasmStreamingSharedPtr([zero, zero]);
+    unsafe {
+      v8__WasmStreaming__Unpack(scope.get_isolate_ptr(), data, &mut that)
+    };
+    let source = args.get(0);
+    (F::get())(scope, source, WasmStreaming(that));
+  }
+  c_fn::<F>
+}
+
+extern "C" {
+  fn v8__WasmStreaming__Unpack(
+    isolate: *mut Isolate,
+    value: *const Value,
+    that: *mut WasmStreamingSharedPtr, // Out parameter.
+  );
+  fn v8__WasmStreaming__shared_ptr_DESTRUCT(this: *mut WasmStreamingSharedPtr);
+  fn v8__WasmStreaming__OnBytesReceived(
+    this: *mut WasmStreamingSharedPtr,
+    data: *const u8,
+    len: usize,
+  );
+  fn v8__WasmStreaming__Finish(this: *mut WasmStreamingSharedPtr);
+  fn v8__WasmStreaming__Abort(
+    this: *mut WasmStreamingSharedPtr,
+    exception: *const Value,
+  );
+}

tests/test_api.rs

@@ -4,6 +4,7 @@
 extern crate lazy_static;
 
 use std::any::type_name;
+use std::cell::RefCell;
 use std::collections::hash_map::DefaultHasher;
 use std::convert::{Into, TryFrom, TryInto};
 use std::ffi::c_void;
@@ -4335,3 +4336,69 @@ fn clear_kept_objects() {
   scope.clear_kept_objects();
   eval(scope, step2).unwrap();
 }
+
+#[test]
+fn wasm_streaming_callback() {
+  thread_local! {
+    static WS: RefCell<Option<v8::WasmStreaming>> = RefCell::new(None);
+  }
+
+  let callback = |scope: &mut v8::HandleScope,
+                  url: v8::Local<v8::Value>,
+                  ws: v8::WasmStreaming| {
+    assert_eq!("https://example.com", url.to_rust_string_lossy(scope));
+    WS.with(|slot| assert!(slot.borrow_mut().replace(ws).is_none()));
+  };
+
+  let _setup_guard = setup();
+
+  let isolate = &mut v8::Isolate::new(v8::CreateParams::default());
+  isolate.set_wasm_streaming_callback(callback);
+
+  let scope = &mut v8::HandleScope::new(isolate);
+  let context = v8::Context::new(scope);
+  let scope = &mut v8::ContextScope::new(scope, context);
+
+  let script = r#"
+    globalThis.result = null;
+    WebAssembly
+      .compileStreaming("https://example.com")
+      .then(result => globalThis.result = result);
+  "#;
+  eval(scope, script).unwrap();
+
+  let global = context.global(scope);
+  let name = v8::String::new(scope, "result").unwrap().into();
+  assert!(global.get(scope, name).unwrap().is_null());
+
+  let mut ws = WS.with(|slot| slot.borrow_mut().take().unwrap());
+  assert!(global.get(scope, name).unwrap().is_null());
+
+  // MVP of WASM modules: contains only the magic marker and the version (1).
+  ws.on_bytes_received(&[0x00, 0x61, 0x73, 0x6d, 0x01, 0x00, 0x00, 0x00]);
+  assert!(global.get(scope, name).unwrap().is_null());
+
+  ws.finish();
+  assert!(global.get(scope, name).unwrap().is_null());
+
+  scope.perform_microtask_checkpoint();
+  assert!(global.get(scope, name).unwrap().is_wasm_module_object());
+
+  let script = r#"
+    globalThis.result = null;
+    WebAssembly
+      .compileStreaming("https://example.com")
+      .catch(result => globalThis.result = result);
+  "#;
+  eval(scope, script).unwrap();
+
+  let ws = WS.with(|slot| slot.borrow_mut().take().unwrap());
+  assert!(global.get(scope, name).unwrap().is_null());
+
+  let exception = v8::Object::new(scope).into(); // Can be anything.
+  ws.abort(Some(exception));
+  assert!(global.get(scope, name).unwrap().is_null());
+
+  scope.perform_microtask_checkpoint();
+  assert!(global.get(scope, name).unwrap().strict_equals(exception));
+}