diff options
Diffstat (limited to 'core')
| -rw-r--r-- | core/bindings.rs | 125 | ||||
| -rw-r--r-- | core/lib.deno_core.d.ts | 26 | ||||
| -rw-r--r-- | core/runtime.rs | 30 |
3 files changed, 172 insertions, 9 deletions
diff --git a/core/bindings.rs b/core/bindings.rs index bee3ecf6d..143ccda9b 100644 --- a/core/bindings.rs +++ b/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, diff --git a/core/lib.deno_core.d.ts b/core/lib.deno_core.d.ts index 2cec6be76..2c69782a6 100644 --- a/core/lib.deno_core.d.ts +++ b/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; } } diff --git a/core/runtime.rs b/core/runtime.rs index 42651a5ef..f156b60f3 100644 --- a/core/runtime.rs +++ b/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. |