diff options
Diffstat (limited to 'core')
| -rw-r--r-- | core/lib.rs | 7 | ||||
| -rw-r--r-- | core/ops.rs | 119 | ||||
| -rw-r--r-- | core/ops_metrics.rs | 2 | ||||
| -rw-r--r-- | core/runtime.rs | 152 |
4 files changed, 177 insertions, 103 deletions
diff --git a/core/lib.rs b/core/lib.rs index 70dadfc6a..cb16c2654 100644 --- a/core/lib.rs +++ b/core/lib.rs @@ -89,11 +89,8 @@ pub use crate::modules::ModuleType; pub use crate::modules::NoopModuleLoader; pub use crate::modules::ResolutionKind; pub use crate::normalize_path::normalize_path; -pub use crate::ops::Op; -pub use crate::ops::OpAsyncFuture; pub use crate::ops::OpCall; pub use crate::ops::OpError; -pub use crate::ops::OpFn; pub use crate::ops::OpId; pub use crate::ops::OpResult; pub use crate::ops::OpState; @@ -135,6 +132,10 @@ pub mod _ops { pub use super::ops::to_op_result; pub use super::ops::OpCtx; pub use super::ops::OpResult; + pub use super::runtime::map_async_op1; + pub use super::runtime::map_async_op2; + pub use super::runtime::map_async_op3; + pub use super::runtime::map_async_op4; pub use super::runtime::queue_async_op; pub use super::runtime::queue_fast_async_op; pub use super::runtime::V8_WRAPPER_OBJECT_INDEX; diff --git a/core/ops.rs b/core/ops.rs index cceeb5654..b7dcc2663 100644 --- a/core/ops.rs +++ b/core/ops.rs @@ -8,12 +8,10 @@ use crate::runtime::JsRuntimeState; use crate::OpDecl; use crate::OpsTracker; use anyhow::Error; -use futures::future::maybe_done; -use futures::future::FusedFuture; use futures::future::MaybeDone; -use futures::ready; -use futures::task::noop_waker; use futures::Future; +use futures::FutureExt; +use pin_project::pin_project; use serde::Serialize; use std::cell::RefCell; use std::ops::Deref; @@ -22,91 +20,78 @@ use std::pin::Pin; use std::ptr::NonNull; use std::rc::Rc; use std::rc::Weak; -use std::task::Context; -use std::task::Poll; use v8::fast_api::CFunctionInfo; use v8::fast_api::CTypeInfo; -/// Wrapper around a Future, which causes that Future to be polled immediately. -/// -/// Background: ops are stored in a `FuturesUnordered` structure which polls -/// them, but without the `OpCall` wrapper this doesn't happen until the next -/// turn of the event loop, which is too late for certain ops. -pub struct OpCall<T>(MaybeDone<Pin<Box<dyn Future<Output = T>>>>); +pub type RealmIdx = u16; +pub type PromiseId = i32; +pub type OpId = u16; -pub enum EagerPollResult<T> { - Ready(T), - Pending(OpCall<T>), +#[pin_project] +pub struct OpCall { + realm_idx: RealmIdx, + promise_id: PromiseId, + op_id: OpId, + /// Future is not necessarily Unpin, so we need to pin_project. + #[pin] + fut: MaybeDone<Pin<Box<dyn Future<Output = OpResult>>>>, } -impl<T> OpCall<T> { - /// Wraps a future, and polls the inner future immediately. - /// This should be the default choice for ops. - pub fn eager(fut: impl Future<Output = T> + 'static) -> EagerPollResult<T> { - let boxed = Box::pin(fut) as Pin<Box<dyn Future<Output = T>>>; - let mut inner = maybe_done(boxed); - let waker = noop_waker(); - let mut cx = Context::from_waker(&waker); - let mut pinned = Pin::new(&mut inner); - let poll = pinned.as_mut().poll(&mut cx); - match poll { - Poll::Ready(_) => EagerPollResult::Ready(pinned.take_output().unwrap()), - _ => EagerPollResult::Pending(Self(inner)), - } - } - +impl OpCall { /// Wraps a future; the inner future is polled the usual way (lazily). - pub fn lazy(fut: impl Future<Output = T> + 'static) -> Self { - let boxed = Box::pin(fut) as Pin<Box<dyn Future<Output = T>>>; - let inner = maybe_done(boxed); - Self(inner) + pub fn pending( + op_ctx: &OpCtx, + promise_id: PromiseId, + fut: Pin<Box<dyn Future<Output = OpResult> + 'static>>, + ) -> Self { + Self { + realm_idx: op_ctx.realm_idx, + op_id: op_ctx.id, + promise_id, + fut: MaybeDone::Future(fut), + } } /// Create a future by specifying its output. This is basically the same as /// `async { value }` or `futures::future::ready(value)`. - pub fn ready(value: T) -> Self { - Self(MaybeDone::Done(value)) + pub fn ready(op_ctx: &OpCtx, promise_id: PromiseId, value: OpResult) -> Self { + Self { + realm_idx: op_ctx.realm_idx, + op_id: op_ctx.id, + promise_id, + fut: MaybeDone::Done(value), + } } } -impl<T> Future for OpCall<T> { - type Output = T; +impl Future for OpCall { + type Output = (RealmIdx, PromiseId, OpId, OpResult); fn poll( self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>, ) -> std::task::Poll<Self::Output> { - // TODO(piscisaureus): safety comment - #[allow(clippy::undocumented_unsafe_blocks)] - let inner = unsafe { &mut self.get_unchecked_mut().0 }; - let mut pinned = Pin::new(inner); - ready!(pinned.as_mut().poll(cx)); - Poll::Ready(pinned.as_mut().take_output().unwrap()) - } -} - -impl<F> FusedFuture for OpCall<F> -where - F: Future, -{ - fn is_terminated(&self) -> bool { - self.0.is_terminated() + let realm_idx = self.realm_idx; + let promise_id = self.promise_id; + let op_id = self.op_id; + let fut = &mut *self.project().fut; + match fut { + MaybeDone::Done(_) => { + // Let's avoid using take_output as it keeps our Pin::box + let res = std::mem::replace(fut, MaybeDone::Gone); + let MaybeDone::Done(res) = res + else { + unreachable!() + }; + std::task::Poll::Ready(res) + } + MaybeDone::Future(f) => f.poll_unpin(cx), + MaybeDone::Gone => std::task::Poll::Pending, + } + .map(move |res| (realm_idx, promise_id, op_id, res)) } } -pub type RealmIdx = usize; -pub type PromiseId = i32; -pub type OpAsyncFuture = OpCall<(PromiseId, OpId, OpResult)>; -pub type OpFn = - fn(&mut v8::HandleScope, v8::FunctionCallbackArguments, v8::ReturnValue); -pub type OpId = usize; - -pub enum Op { - Sync(OpResult), - Async(OpAsyncFuture), - NotFound, -} - pub enum OpResult { Ok(serde_v8::SerializablePkg), Err(OpError), diff --git a/core/ops_metrics.rs b/core/ops_metrics.rs index c0b8abb51..b25368bd0 100644 --- a/core/ops_metrics.rs +++ b/core/ops_metrics.rs @@ -63,7 +63,7 @@ impl OpsTracker { #[inline] fn metrics_mut(&self, id: OpId) -> RefMut<OpMetrics> { - RefMut::map(self.ops.borrow_mut(), |ops| &mut ops[id]) + RefMut::map(self.ops.borrow_mut(), |ops| &mut ops[id as usize]) } #[inline] diff --git a/core/runtime.rs b/core/runtime.rs index 6820df6bc..3723a917a 100644 --- a/core/runtime.rs +++ b/core/runtime.rs @@ -35,8 +35,10 @@ use futures::channel::oneshot; use futures::future::poll_fn; use futures::future::Future; use futures::future::FutureExt; +use futures::future::MaybeDone; use futures::stream::FuturesUnordered; use futures::stream::StreamExt; +use futures::task::noop_waker; use futures::task::AtomicWaker; use smallvec::SmallVec; use std::any::Any; @@ -45,6 +47,7 @@ use std::collections::HashMap; use std::collections::VecDeque; use std::ffi::c_void; use std::option::Option; +use std::pin::Pin; use std::rc::Rc; use std::sync::Arc; use std::sync::Mutex; @@ -53,8 +56,6 @@ use std::task::Context; use std::task::Poll; use v8::OwnedIsolate; -type PendingOpFuture = OpCall<(RealmIdx, PromiseId, OpId, OpResult)>; - pub enum Snapshot { Static(&'static [u8]), JustCreated(v8::StartupData), @@ -165,7 +166,7 @@ pub struct JsRuntimeState { dyn_module_evaluate_idle_counter: u32, pub(crate) source_map_getter: Option<Rc<Box<dyn SourceMapGetter>>>, pub(crate) source_map_cache: Rc<RefCell<SourceMapCache>>, - pub(crate) pending_ops: FuturesUnordered<PendingOpFuture>, + pub(crate) pending_ops: FuturesUnordered<OpCall>, pub(crate) have_unpolled_ops: bool, pub(crate) op_state: Rc<RefCell<OpState>>, pub(crate) shared_array_buffer_store: Option<SharedArrayBufferStore>, @@ -360,7 +361,7 @@ impl JsRuntime { .into_iter() .enumerate() .map(|(id, decl)| { - OpCtx::new(id, 0, Rc::new(decl), op_state.clone(), weak.clone()) + OpCtx::new(id as u16, 0, Rc::new(decl), op_state.clone(), weak.clone()) }) .collect::<Vec<_>>() .into_boxed_slice(); @@ -610,7 +611,7 @@ impl JsRuntime { /// constructed. pub fn create_realm(&mut self) -> Result<JsRealm, Error> { let realm = { - let realm_idx = self.state.borrow().known_realms.len(); + let realm_idx = self.state.borrow().known_realms.len() as u16; let op_ctxs: Box<[OpCtx]> = self .global_realm() @@ -2231,7 +2232,7 @@ impl JsRuntime { { let (realm_idx, promise_id, op_id, resp) = item; state.op_state.borrow().tracker.track_async_completed(op_id); - responses_per_realm[realm_idx].push((promise_id, resp)); + responses_per_realm[realm_idx as usize].push((promise_id, resp)); } } @@ -2335,7 +2336,7 @@ impl JsRuntime { { let (realm_idx, promise_id, op_id, mut resp) = item; debug_assert_eq!( - state.known_realms[realm_idx], + state.known_realms[realm_idx as usize], state.global_realm.as_ref().unwrap().context() ); realm_state.unrefed_ops.remove(&promise_id); @@ -2382,27 +2383,106 @@ impl JsRuntime { } #[inline] -pub fn queue_fast_async_op( +pub fn queue_fast_async_op<R: serde::Serialize + 'static>( ctx: &OpCtx, - op: impl Future<Output = (RealmIdx, PromiseId, OpId, OpResult)> + 'static, + promise_id: PromiseId, + op: impl Future<Output = Result<R, Error>> + 'static, ) { let runtime_state = match ctx.runtime_state.upgrade() { Some(rc_state) => rc_state, // atleast 1 Rc is held by the JsRuntime. None => unreachable!(), }; - + let get_class = { + let state = RefCell::borrow(&ctx.state); + state.tracker.track_async(ctx.id); + state.get_error_class_fn + }; + let fut = op + .map(|result| crate::_ops::to_op_result(get_class, result)) + .boxed_local(); let mut state = runtime_state.borrow_mut(); - state.pending_ops.push(OpCall::lazy(op)); + state + .pending_ops + .push(OpCall::pending(ctx, promise_id, fut)); state.have_unpolled_ops = true; } #[inline] +pub fn map_async_op1<R: serde::Serialize + 'static>( + ctx: &OpCtx, + op: impl Future<Output = Result<R, Error>> + 'static, +) -> MaybeDone<Pin<Box<dyn Future<Output = OpResult>>>> { + let get_class = { + let state = RefCell::borrow(&ctx.state); + state.tracker.track_async(ctx.id); + state.get_error_class_fn + }; + + let fut = op + .map(|result| crate::_ops::to_op_result(get_class, result)) + .boxed_local(); + MaybeDone::Future(fut) +} + +#[inline] +pub fn map_async_op2<R: serde::Serialize + 'static>( + ctx: &OpCtx, + op: impl Future<Output = R> + 'static, +) -> MaybeDone<Pin<Box<dyn Future<Output = OpResult>>>> { + let state = RefCell::borrow(&ctx.state); + state.tracker.track_async(ctx.id); + + let fut = op.map(|result| OpResult::Ok(result.into())).boxed_local(); + MaybeDone::Future(fut) +} + +#[inline] +pub fn map_async_op3<R: serde::Serialize + 'static>( + ctx: &OpCtx, + op: Result<impl Future<Output = Result<R, Error>> + 'static, Error>, +) -> MaybeDone<Pin<Box<dyn Future<Output = OpResult>>>> { + let get_class = { + let state = RefCell::borrow(&ctx.state); + state.tracker.track_async(ctx.id); + state.get_error_class_fn + }; + + match op { + Err(err) => MaybeDone::Done(OpResult::Err(OpError::new(get_class, err))), + Ok(fut) => MaybeDone::Future( + fut + .map(|result| crate::_ops::to_op_result(get_class, result)) + .boxed_local(), + ), + } +} + +#[inline] +pub fn map_async_op4<R: serde::Serialize + 'static>( + ctx: &OpCtx, + op: Result<impl Future<Output = R> + 'static, Error>, +) -> MaybeDone<Pin<Box<dyn Future<Output = OpResult>>>> { + let get_class = { + let state = RefCell::borrow(&ctx.state); + state.tracker.track_async(ctx.id); + state.get_error_class_fn + }; + + match op { + Err(err) => MaybeDone::Done(OpResult::Err(OpError::new(get_class, err))), + Ok(fut) => MaybeDone::Future( + fut.map(|result| OpResult::Ok(result.into())).boxed_local(), + ), + } +} + pub fn queue_async_op<'s>( ctx: &OpCtx, scope: &'s mut v8::HandleScope, deferred: bool, - op: impl Future<Output = (RealmIdx, PromiseId, OpId, OpResult)> + 'static, + promise_id: PromiseId, + mut op: MaybeDone<Pin<Box<dyn Future<Output = OpResult>>>>, ) -> Option<v8::Local<'s, v8::Value>> { let runtime_state = match ctx.runtime_state.upgrade() { Some(rc_state) => rc_state, @@ -2415,32 +2495,40 @@ pub fn queue_async_op<'s>( // deno_core doesn't currently support such exposure, even though embedders // can cause them, so we panic in debug mode (since the check is expensive). debug_assert_eq!( - runtime_state.borrow().known_realms[ctx.realm_idx].to_local(scope), + runtime_state.borrow().known_realms[ctx.realm_idx as usize].to_local(scope), Some(scope.get_current_context()) ); - match OpCall::eager(op) { - // If the result is ready we'll just return it straight to the caller, so - // we don't have to invoke a JS callback to respond. // This works under the - // assumption that `()` return value is serialized as `null`. - EagerPollResult::Ready((_, _, op_id, mut resp)) if !deferred => { - let resp = resp.to_v8(scope).unwrap(); - ctx.state.borrow_mut().tracker.track_async_completed(op_id); - return Some(resp); - } - EagerPollResult::Ready(op) => { - let ready = OpCall::ready(op); - let mut state = runtime_state.borrow_mut(); - state.pending_ops.push(ready); - state.have_unpolled_ops = true; + // All ops are polled immediately + let waker = noop_waker(); + let mut cx = Context::from_waker(&waker); + + // Note that MaybeDone returns () from the future + let op_call = match op.poll_unpin(&mut cx) { + Poll::Pending => { + let MaybeDone::Future(fut) = op else { + unreachable!() + }; + OpCall::pending(ctx, promise_id, fut) } - EagerPollResult::Pending(op) => { - let mut state = runtime_state.borrow_mut(); - state.pending_ops.push(op); - state.have_unpolled_ops = true; + Poll::Ready(_) => { + let mut op_result = Pin::new(&mut op).take_output().unwrap(); + // If the op is ready and is not marked as deferred we can immediately return + // the result. + if !deferred { + ctx.state.borrow_mut().tracker.track_async_completed(ctx.id); + return Some(op_result.to_v8(scope).unwrap()); + } + + OpCall::ready(ctx, promise_id, op_result) } - } + }; + // Otherwise we will push it to the `pending_ops` and let it be polled again + // or resolved on the next tick of the event loop. + let mut state = runtime_state.borrow_mut(); + state.pending_ops.push(op_call); + state.have_unpolled_ops = true; None } |