feat: enable WebAssembly.instantiateStreaming and wasm async compilation (#11200)

The WebAssembly streaming APIs used to be enabled, but used to take
buffer sources as their first argument (see #6154 and #7259). This
change re-enables them, requiring a Promise<Response> instead, as well as
enabling asynchronous compilation of WebAssembly modules.

cli/dts/lib.deno.shared_globals.d.ts

@@ -224,6 +224,18 @@ declare namespace WebAssembly {
    */
   export function compile(bytes: BufferSource): Promise<Module>;
 
+  /**
+   * The `WebAssembly.compileStreaming()` function compiles a `WebAssembly.Module`
+   * directly from a streamed underlying source. This function is useful if it is
+   * necessary to a compile a module before it can be instantiated (otherwise, the
+   * `WebAssembly.instantiateStreaming()` function should be used).
+   *
+   * [MDN](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/WebAssembly/compileStreaming)
+   */
+  export function compileStreaming(
+    source: Response | Promise<Response>,
+  ): Promise<Module>;
+
   /**
    * The WebAssembly.instantiate() function allows you to compile and instantiate
    * WebAssembly code.
@@ -255,6 +267,18 @@ declare namespace WebAssembly {
     importObject?: Imports,
   ): Promise<Instance>;
 
+  /**
+   * The `WebAssembly.instantiateStreaming()` function compiles and instantiates a
+   * WebAssembly module directly from a streamed underlying source. This is the most
+   * efficient, optimized way to load wasm code.
+   *
+   * [MDN](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/WebAssembly/instantiateStreaming)
+   */
+  export function instantiateStreaming(
+    response: Response | PromiseLike<Response>,
+    importObject?: Imports,
+  ): Promise<WebAssemblyInstantiatedSource>;
+
   /**
    * The `WebAssembly.validate()` function validates a given typed array of
    * WebAssembly binary code, returning whether the bytes form a valid wasm

cli/tests/unit/wasm_test.ts

@@ -0,0 +1,80 @@
+import {
+  assert,
+  assertEquals,
+  assertThrowsAsync,
+  unitTest,
+} from "./test_util.ts";
+
+// The following blob can be created by taking the following s-expr and pass
+// it through wat2wasm.
+//    (module
+//      (func $add (param $a i32) (param $b i32) (result i32)
+//        local.get $a
+//        local.get $b
+//        i32.add)
+//      (export "add" (func $add))
+//    )
+// deno-fmt-ignore
+const simpleWasm = new Uint8Array([
+  0x00, 0x61, 0x73, 0x6d, 0x01, 0x00, 0x00, 0x00, 0x01, 0x07, 0x01, 0x60,
+  0x02, 0x7f, 0x7f, 0x01, 0x7f, 0x03, 0x02, 0x01, 0x00, 0x07, 0x07, 0x01,
+  0x03, 0x61, 0x64, 0x64, 0x00, 0x00, 0x0a, 0x09, 0x01, 0x07, 0x00, 0x20,
+  0x00, 0x20, 0x01, 0x6a, 0x0b
+]);
+
+unitTest(async function wasmInstantiateWorksWithBuffer(): Promise<void> {
+  const { module, instance } = await WebAssembly.instantiate(simpleWasm);
+  assertEquals(WebAssembly.Module.exports(module), [{
+    name: "add",
+    kind: "function",
+  }]);
+  assertEquals(WebAssembly.Module.imports(module), []);
+  assert(typeof instance.exports.add === "function");
+  const add = instance.exports.add as (a: number, b: number) => number;
+  assertEquals(add(1, 3), 4);
+});
+
+// V8's default implementation of `WebAssembly.instantiateStreaming()` if you
+// don't set the WASM streaming callback, is to take a byte source. Here we
+// check that our implementation of the callback disallows it.
+unitTest(
+  async function wasmInstantiateStreamingFailsWithBuffer(): Promise<void> {
+    await assertThrowsAsync(async () => {
+      await WebAssembly.instantiateStreaming(
+        // Bypassing the type system
+        simpleWasm as unknown as Promise<Response>,
+      );
+    }, TypeError);
+  },
+);
+
+unitTest(async function wasmInstantiateStreaming(): Promise<void> {
+  let isomorphic = "";
+  for (const byte of simpleWasm) {
+    isomorphic += String.fromCharCode(byte);
+  }
+  const base64Url = "data:application/wasm;base64," + btoa(isomorphic);
+
+  const { module, instance } = await WebAssembly.instantiateStreaming(
+    fetch(base64Url),
+  );
+  assertEquals(WebAssembly.Module.exports(module), [{
+    name: "add",
+    kind: "function",
+  }]);
+  assertEquals(WebAssembly.Module.imports(module), []);
+  assert(typeof instance.exports.add === "function");
+  const add = instance.exports.add as (a: number, b: number) => number;
+  assertEquals(add(1, 3), 4);
+});
+
+unitTest(
+  { perms: { net: true } },
+  async function wasmStreamingNonTrivial(): Promise<void> {
+    // deno-dom's WASM file is a real-world non-trivial case that gave us
+    // trouble when implementing this.
+    await WebAssembly.instantiateStreaming(fetch(
+      "http://localhost:4545/cli/tests/deno_dom_0.1.3-alpha2.wasm",
+    ));
+  },
+);

core/bindings.rs

@@ -9,15 +9,18 @@ use crate::OpId;
 use crate::OpPayload;
 use crate::OpTable;
 use crate::PromiseId;
+use crate::ResourceId;
 use crate::ZeroCopyBuf;
 use log::debug;
 use rusty_v8 as v8;
 use serde::Deserialize;
 use serde::Serialize;
 use serde_v8::to_v8;
+use std::cell::RefCell;
 use std::convert::TryFrom;
 use std::convert::TryInto;
 use std::option::Option;
+use std::rc::Rc;
 use url::Url;
 use v8::MapFnTo;
 
@@ -63,6 +66,12 @@ lazy_static::lazy_static! {
       v8::ExternalReference {
         function: call_console.map_fn_to(),
       },
+      v8::ExternalReference {
+        function: set_wasm_streaming_callback.map_fn_to()
+      },
+      v8::ExternalReference {
+        function: wasm_streaming_feed.map_fn_to()
+      }
     ]);
 }
 
@@ -140,6 +149,13 @@ pub fn initialize_context<'s>(
   set_func(scope, core_val, "memoryUsage", memory_usage);
   set_func(scope, core_val, "callConsole", call_console);
   set_func(scope, core_val, "createHostObject", create_host_object);
+  set_func(
+    scope,
+    core_val,
+    "setWasmStreamingCallback",
+    set_wasm_streaming_callback,
+  );
+  set_func(scope, core_val, "wasmStreamingFeed", wasm_streaming_feed);
 
   // Direct bindings on `window`.
   set_func(scope, global, "queueMicrotask", queue_microtask);
@@ -514,6 +530,115 @@ fn call_console(
   deno_console_method.call(scope, receiver.into(), &call_args);
 }
 
+struct WasmStreamingResource(RefCell<v8::WasmStreaming>);
+impl crate::Resource for WasmStreamingResource {}
+
+fn set_wasm_streaming_callback(
+  scope: &mut v8::HandleScope,
+  args: v8::FunctionCallbackArguments,
+  _rv: v8::ReturnValue,
+) {
+  let state_rc = JsRuntime::state(scope);
+  let mut state = state_rc.borrow_mut();
+
+  let cb = match v8::Local::<v8::Function>::try_from(args.get(0)) {
+    Ok(cb) => cb,
+    Err(err) => return throw_type_error(scope, err.to_string()),
+  };
+
+  // The callback to pass to the v8 API has to be a unit type, so it can't
+  // borrow or move any local variables. Therefore, we're storing the JS
+  // callback in a JsRuntimeState slot.
+  if let slot @ None = &mut state.js_wasm_streaming_cb {
+    slot.replace(v8::Global::new(scope, cb));
+  } else {
+    return throw_type_error(
+      scope,
+      "Deno.core.setWasmStreamingCallback() already called",
+    );
+  }
+
+  scope.set_wasm_streaming_callback(|scope, arg, wasm_streaming| {
+    let (cb_handle, streaming_rid) = {
+      let state_rc = JsRuntime::state(scope);
+      let state = state_rc.borrow();
+      let cb_handle = state.js_wasm_streaming_cb.as_ref().unwrap().clone();
+      let streaming_rid = state
+        .op_state
+        .borrow_mut()
+        .resource_table
+        .add(WasmStreamingResource(RefCell::new(wasm_streaming)));
+      (cb_handle, streaming_rid)
+    };
+
+    let undefined = v8::undefined(scope);
+    let rid = serde_v8::to_v8(scope, streaming_rid).unwrap();
+    cb_handle
+      .get(scope)
+      .call(scope, undefined.into(), &[arg, rid]);
+  });
+}
+
+fn wasm_streaming_feed(
+  scope: &mut v8::HandleScope,
+  args: v8::FunctionCallbackArguments,
+  _rv: v8::ReturnValue,
+) {
+  #[derive(Deserialize)]
+  #[serde(rename_all = "snake_case")]
+  enum MessageType {
+    Bytes,
+    Abort,
+    Finish,
+  }
+
+  let rid: ResourceId = match serde_v8::from_v8(scope, args.get(0)) {
+    Ok(rid) => rid,
+    Err(_) => return throw_type_error(scope, "Invalid argument"),
+  };
+  let message_type = match serde_v8::from_v8(scope, args.get(1)) {
+    Ok(message_type) => message_type,
+    Err(_) => return throw_type_error(scope, "Invalid argument"),
+  };
+
+  let wasm_streaming = {
+    let state_rc = JsRuntime::state(scope);
+    let state = state_rc.borrow();
+    // If message_type is not Bytes, we'll be consuming the WasmStreaming
+    // instance, so let's also remove it from the resource table.
+    let wasm_streaming: Option<Rc<WasmStreamingResource>> = match message_type {
+      MessageType::Bytes => state.op_state.borrow().resource_table.get(rid),
+      _ => state.op_state.borrow_mut().resource_table.take(rid),
+    };
+    match wasm_streaming {
+      Some(wasm_streaming) => wasm_streaming,
+      None => return throw_type_error(scope, "Invalid resource ID."),
+    }
+  };
+
+  match message_type {
+    MessageType::Bytes => {
+      let bytes: ZeroCopyBuf = match serde_v8::from_v8(scope, args.get(2)) {
+        Ok(bytes) => bytes,
+        Err(_) => return throw_type_error(scope, "Invalid resource ID."),
+      };
+      wasm_streaming.0.borrow_mut().on_bytes_received(&bytes);
+    }
+    _ => {
+      // These types need to consume the WasmStreaming instance.
+      let wasm_streaming = match Rc::try_unwrap(wasm_streaming) {
+        Ok(streaming) => streaming.0.into_inner(),
+        Err(_) => panic!("Couldn't consume WasmStreamingResource."),
+      };
+      match message_type {
+        MessageType::Bytes => unreachable!(),
+        MessageType::Finish => wasm_streaming.finish(),
+        MessageType::Abort => wasm_streaming.abort(Some(args.get(2))),
+      }
+    }
+  }
+}
+
 fn encode(
   scope: &mut v8::HandleScope,
   args: v8::FunctionCallbackArguments,

core/lib.deno_core.d.ts

@@ -41,5 +41,31 @@ declare namespace Deno {
 
     /** Encode a string to its Uint8Array representation. */
     function encode(input: string): Uint8Array;
+
+    /**
+     * Set a callback that will be called when the WebAssembly streaming APIs
+     * (`WebAssembly.compileStreaming` and `WebAssembly.instantiateStreaming`)
+     * are called in order to feed the source's bytes to the wasm compiler.
+     * The callback is called with the source argument passed to the streaming
+     * APIs and an rid to use with `Deno.core.wasmStreamingFeed`.
+     */
+    function setWasmStreamingCallback(
+      cb: (source: any, rid: number) => void,
+    ): void;
+
+    /**
+     * Affect the state of the WebAssembly streaming compiler, by either passing
+     * it bytes, aborting it, or indicating that all bytes were received.
+     * `rid` must be a resource ID that was passed to the callback set with
+     * `Deno.core.setWasmStreamingCallback`. Calling this function with `type`
+     * set to either "abort" or "finish" invalidates the rid.
+     */
+    function wasmStreamingFeed(
+      rid: number,
+      type: "bytes",
+      bytes: Uint8Array,
+    ): void;
+    function wasmStreamingFeed(rid: number, type: "abort", error: any): void;
+    function wasmStreamingFeed(rid: number, type: "finish"): void;
   }
 }

core/runtime.rs

@@ -104,6 +104,7 @@ pub(crate) struct JsRuntimeState {
   pub global_context: Option<v8::Global<v8::Context>>,
   pub(crate) js_recv_cb: Option<v8::Global<v8::Function>>,
   pub(crate) js_macrotask_cb: Option<v8::Global<v8::Function>>,
+  pub(crate) js_wasm_streaming_cb: Option<v8::Global<v8::Function>>,
   pub(crate) pending_promise_exceptions:
     HashMap<v8::Global<v8::Promise>, v8::Global<v8::Value>>,
   pending_dyn_mod_evaluate: VecDeque<DynImportModEvaluate>,
@@ -155,12 +156,8 @@ fn v8_init(v8_platform: Option<v8::SharedRef<v8::Platform>>) {
   v8::V8::initialize();
 
   let flags = concat!(
-    // TODO(ry) This makes WASM compile synchronously. Eventually we should
-    // remove this to make it work asynchronously too. But that requires getting
-    // PumpMessageLoop and RunMicrotasks setup correctly.
-    // See https://github.com/denoland/deno/issues/2544
     " --experimental-wasm-threads",
-    " --no-wasm-async-compilation",
+    " --wasm-test-streaming",
     " --harmony-import-assertions",
     " --no-validate-asm",
   );
@@ -290,6 +287,7 @@ impl JsRuntime {
       pending_mod_evaluate: None,
       js_recv_cb: None,
       js_macrotask_cb: None,
+      js_wasm_streaming_cb: None,
       js_error_create_fn,
       pending_ops: FuturesUnordered::new(),
       pending_unref_ops: FuturesUnordered::new(),
@@ -600,6 +598,8 @@ impl JsRuntime {
     ) {
       // do nothing
     }
+
+    scope.perform_microtask_checkpoint();
   }
 
   /// Runs event loop to completion
@@ -634,6 +634,8 @@ impl JsRuntime {
       state.waker.register(cx.waker());
     }
 
+    self.pump_v8_message_loop();
+
     // Ops
     {
       let async_responses = self.poll_pending_ops(cx);
@@ -658,8 +660,6 @@ impl JsRuntime {
     // Top level module
     self.evaluate_pending_module();
 
-    self.pump_v8_message_loop();
-
     let state = state_rc.borrow();
     let module_map = module_map_rc.borrow();
 
@@ -669,6 +669,8 @@ impl JsRuntime {
     let has_pending_dyn_module_evaluation =
       !state.pending_dyn_mod_evaluate.is_empty();
     let has_pending_module_evaluation = state.pending_mod_evaluate.is_some();
+    let has_pending_background_tasks =
+      self.v8_isolate().has_pending_background_tasks();
     let inspector_has_active_sessions = self
       .inspector
       .as_ref()
@@ -679,6 +681,7 @@ impl JsRuntime {
       && !has_pending_dyn_imports
       && !has_pending_dyn_module_evaluation
       && !has_pending_module_evaluation
+      && !has_pending_background_tasks
     {
       if wait_for_inspector && inspector_has_active_sessions {
         return Poll::Pending;
@@ -689,7 +692,12 @@ impl JsRuntime {
 
     // Check if more async ops have been dispatched
     // during this turn of event loop.
-    if state.have_unpolled_ops {
+    // If there are any pending background tasks, we also wake the runtime to
+    // make sure we don't miss them.
+    // TODO(andreubotella) The event loop will spin as long as there are pending
+    // background tasks. We should look into having V8 notify us when a
+    // background task is done.
+    if state.have_unpolled_ops || has_pending_background_tasks {
       state.waker.wake();
     }
 
@@ -697,6 +705,7 @@ impl JsRuntime {
       if has_pending_ops
         || has_pending_dyn_imports
         || has_pending_dyn_module_evaluation
+        || has_pending_background_tasks
       {
         // pass, will be polled again
       } else {
@@ -706,7 +715,10 @@ impl JsRuntime {
     }
 
     if has_pending_dyn_module_evaluation {
-      if has_pending_ops || has_pending_dyn_imports {
+      if has_pending_ops
+        || has_pending_dyn_imports
+        || has_pending_background_tasks
+      {
         // pass, will be polled again
       } else {
         let mut msg = "Dynamically imported module evaluation is still pending but there are no pending ops. This situation is often caused by unresolved promise.

extensions/fetch/26_fetch.js

@@ -432,6 +432,63 @@
     return error;
   }
 
+  /**
+   * Handle the Promise<Response> argument to the WebAssembly streaming
+   * APIs. This function should be registered through
+   * `Deno.core.setWasmStreamingCallback`.
+   *
+   * @param {any} source The source parameter that the WebAssembly
+   * streaming API was called with.
+   * @param {number} rid An rid that can be used with
+   * `Deno.core.wasmStreamingFeed`.
+   */
+  function handleWasmStreaming(source, rid) {
+    // This implements part of
+    // https://webassembly.github.io/spec/web-api/#compile-a-potential-webassembly-response
+    (async () => {
+      try {
+        const res = webidl.converters["Response"](await source, {
+          prefix: "Failed to call 'WebAssembly.compileStreaming'",
+          context: "Argument 1",
+        });
+
+        // 2.3.
+        // The spec is ambiguous here, see
+        // https://github.com/WebAssembly/spec/issues/1138. The WPT tests
+        // expect the raw value of the Content-Type attribute lowercased.
+        if (
+          res.headers.get("Content-Type")?.toLowerCase() !== "application/wasm"
+        ) {
+          throw new TypeError("Invalid WebAssembly content type.");
+        }
+
+        // 2.5.
+        if (!res.ok) {
+          throw new TypeError(`HTTP status code ${res.status}`);
+        }
+
+        // 2.6.
+        // Rather than consuming the body as an ArrayBuffer, this passes each
+        // chunk to the feed as soon as it's available.
+        if (res.body !== null) {
+          const reader = res.body.getReader();
+          while (true) {
+            const { value: chunk, done } = await reader.read();
+            if (done) break;
+            Deno.core.wasmStreamingFeed(rid, "bytes", chunk);
+          }
+        }
+
+        // 2.7.
+        Deno.core.wasmStreamingFeed(rid, "finish");
+      } catch (err) {
+        // 2.8 and 3
+        Deno.core.wasmStreamingFeed(rid, "abort", err);
+      }
+    })();
+  }
+
   window.__bootstrap.fetch ??= {};
   window.__bootstrap.fetch.fetch = fetch;
+  window.__bootstrap.fetch.handleWasmStreaming = handleWasmStreaming;
 })(this);

runtime/js/99_main.js

@@ -176,6 +176,7 @@ delete Object.prototype.__proto__;
 
   function runtimeStart(runtimeOptions, source) {
     core.setMacrotaskCallback(timers.handleTimerMacrotask);
+    core.setWasmStreamingCallback(fetch.handleWasmStreaming);
     version.setVersions(
       runtimeOptions.denoVersion,
       runtimeOptions.v8Version,

test_util/src/lib.rs

@@ -948,6 +948,8 @@ fn custom_headers(p: &str, body: Vec<u8>) -> Response<Body> {
     Some("application/javascript")
   } else if p.ends_with(".json") {
     Some("application/json")
+  } else if p.ends_with(".wasm") {
+    Some("application/wasm")
   } else {
     None
   };

tools/wpt/expectation.json

@@ -603,23 +603,23 @@
       }
     },
     "webapi": {
-      "abort.any.html": false,
-      "body.any.html": false,
-      "contenttype.any.html": false,
-      "empty-body.any.html": false,
+      "abort.any.html": true,
+      "body.any.html": true,
+      "contenttype.any.html": true,
+      "empty-body.any.html": true,
       "historical.any.html": false,
-      "idlharness.any.html": [
-        "WebAssembly namespace: operation compileStreaming(Promise<Response>)",
-        "WebAssembly namespace: operation instantiateStreaming(Promise<Response>, optional object)"
+      "idlharness.any.html": true,
+      "instantiateStreaming-bad-imports.any.html": true,
+      "instantiateStreaming.any.html": true,
+      "invalid-args.any.html": true,
+      "invalid-code.any.html": true,
+      "modified-contenttype.any.html": true,
+      "origin.sub.any.html": [
+        "Opaque response: compileStreaming",
+        "Opaque response: instantiateStreaming"
       ],
-      "instantiateStreaming-bad-imports.any.html": false,
-      "instantiateStreaming.any.html": false,
-      "invalid-args.any.html": false,
-      "invalid-code.any.html": false,
-      "modified-contenttype.any.html": false,
-      "origin.sub.any.html": false,
-      "rejected-arg.any.html": false,
-      "status.any.html": false
+      "rejected-arg.any.html": true,
+      "status.any.html": true
     }
   },
   "WebIDL": {