foster/denoland-deno
1
0
Fork 0
mirror of https://github.com/denoland/deno.git synced 2024-11-21 15:04:11 -05:00
Code Issues Packages 1 Wiki Activity

refactor: move JsRuntimeInspector to deno_core (#10763)

Browse source 
This commit moves implementation of "JsRuntimeInspector" to "deno_core" crate.

To achieve that following changes were made:

* "Worker" and "WebWorker" no longer own instance of "JsRuntimeInspector",
instead it is now owned by "deno_core::JsRuntime".

* Consequently polling of inspector is no longer done in "Worker"/"WebWorker",
instead it's done in "deno_core::JsRuntime::poll_event_loop".

* "deno_core::JsRuntime::poll_event_loop" and "deno_core::JsRuntime::run_event_loop",
now accept "wait_for_inspector" boolean that tells if event loop should still be 
"pending" if there are active inspector sessions - this change fixes the problem 
that inspector disconnects from the frontend and process exits once the code has
stopped executing.
This commit is contained in:
Bartek Iwańczuk 2021-05-26 21:07:12 +02:00 committed by GitHub
parent e9edd7e14d
commit e5beb800c9
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
18 changed files with 171 additions and 145 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
bench_util/src/js_runtime.rs
View file

@ -72,5 +72,5 @@ pub fn bench_js_async(
async fn inner_async(src: &str, runtime: &mut JsRuntime) {
runtime.execute("inner_loop", src).unwrap();
runtime.run_event_loop().await.unwrap();
runtime.run_event_loop(false).await.unwrap();
}

12
cli/main.rs
View file

@ -534,7 +534,7 @@ async fn eval_command(
debug!("main_module {}", &main_module);
worker.execute_module(&main_module).await?;
worker.execute("window.dispatchEvent(new Event('load'))")?;
worker.run_event_loop().await?;
worker.run_event_loop(false).await?;
worker.execute("window.dispatchEvent(new Event('unload'))")?;
Ok(())
}
@ -737,7 +737,7 @@ async fn run_repl(flags: Flags) -> Result<(), AnyError> {
let program_state = ProgramState::build(flags).await?;
let mut worker =
create_main_worker(&program_state, main_module.clone(), permissions, false);
worker.run_event_loop().await?;
worker.run_event_loop(false).await?;
tools::repl::run(&program_state, worker).await
}
@ -770,7 +770,7 @@ async fn run_from_stdin(flags: Flags) -> Result<(), AnyError> {
debug!("main_module {}", main_module);
worker.execute_module(&main_module).await?;
worker.execute("window.dispatchEvent(new Event('load'))")?;
worker.run_event_loop().await?;
worker.run_event_loop(false).await?;
worker.execute("window.dispatchEvent(new Event('unload'))")?;
Ok(())
}
@ -839,7 +839,7 @@ async fn run_with_watch(flags: Flags, script: String) -> Result<(), AnyError> {
debug!("main_module {}", main_module);
worker.execute_module(&main_module).await?;
worker.execute("window.dispatchEvent(new Event('load'))")?;
worker.run_event_loop().await?;
worker.run_event_loop(false).await?;
worker.execute("window.dispatchEvent(new Event('unload'))")?;
Ok(())
}
@ -881,7 +881,9 @@ async fn run_command(flags: Flags, script: String) -> Result<(), AnyError> {
debug!("main_module {}", main_module);
worker.execute_module(&main_module).await?;
worker.execute("window.dispatchEvent(new Event('load'))")?;
worker.run_event_loop().await?;
worker
.run_event_loop(maybe_coverage_collector.is_none())
.await?;
worker.execute("window.dispatchEvent(new Event('unload'))")?;
if let Some(coverage_collector) = maybe_coverage_collector.as_mut() {

2
cli/program_state.rs
View file

@ -17,7 +17,7 @@ use crate::specifier_handler::FetchHandler;
use crate::version;
use deno_runtime::deno_broadcast_channel::InMemoryBroadcastChannel;
use deno_runtime::deno_file::BlobUrlStore;
use deno_runtime::inspector::InspectorServer;
use deno_runtime::inspector_server::InspectorServer;
use deno_runtime::permissions::Permissions;
use deno_core::error::anyhow;

2
cli/standalone.rs
View file

@ -202,7 +202,7 @@ pub async fn run(
worker.bootstrap(&options);
worker.execute_module(&main_module).await?;
worker.execute("window.dispatchEvent(new Event('load'))")?;
worker.run_event_loop().await?;
worker.run_event_loop(true).await?;
worker.execute("window.dispatchEvent(new Event('unload'))")?;
std::process::exit(0);
}

2
cli/tools/coverage.rs
View file

@ -13,7 +13,7 @@ use deno_core::error::AnyError;
use deno_core::serde_json;
use deno_core::serde_json::json;
use deno_core::url::Url;
use deno_runtime::inspector::LocalInspectorSession;
use deno_core::LocalInspectorSession;
use deno_runtime::permissions::Permissions;
use regex::Regex;
use serde::Deserialize;

4
cli/tools/repl.rs
View file

@ -9,7 +9,7 @@ use deno_core::error::AnyError;
use deno_core::futures::FutureExt;
use deno_core::serde_json::json;
use deno_core::serde_json::Value;
use deno_runtime::inspector::LocalInspectorSession;
use deno_core::LocalInspectorSession;
use deno_runtime::worker::MainWorker;
use rustyline::completion::Completer;
use rustyline::error::ReadlineError;
@ -287,7 +287,7 @@ async fn read_line_and_poll(
result = &mut line => {
return result.unwrap();
}
_ = worker.run_event_loop(), if poll_worker => {
_ = worker.run_event_loop(false), if poll_worker => {
poll_worker = false;
}
_ = timeout => {

4
cli/tools/test_runner.rs
View file

@ -307,7 +307,9 @@ pub async fn run_test_file(
let execute_result = worker.execute_module(&test_module).await;
execute_result?;
worker.run_event_loop().await?;
worker
.run_event_loop(maybe_coverage_collector.is_none())
.await?;
worker.execute("window.dispatchEvent(new Event('unload'))")?;
if let Some(coverage_collector) = maybe_coverage_collector.as_mut() {

2
core/examples/http_bench_json_ops.rs
View file

@ -223,7 +223,7 @@ fn main() {
include_str!("http_bench_json_ops.js"),
)
.unwrap();
js_runtime.run_event_loop().await
js_runtime.run_event_loop(false).await
};
runtime.block_on(future).unwrap();
}

72
runtime/inspector/mod.rs → core/inspector.rs
View file

@ -4,25 +4,25 @@
//! https://chromedevtools.github.io/devtools-protocol/
//! https://hyperandroid.com/2020/02/12/v8-inspector-from-an-embedder-standpoint/
use deno_core::error::generic_error;
use deno_core::error::AnyError;
use deno_core::futures::channel::mpsc;
use deno_core::futures::channel::mpsc::UnboundedReceiver;
use deno_core::futures::channel::mpsc::UnboundedSender;
use deno_core::futures::channel::oneshot;
use deno_core::futures::future::select;
use deno_core::futures::future::Either;
use deno_core::futures::future::Future;
use deno_core::futures::prelude::*;
use deno_core::futures::stream::FuturesUnordered;
use deno_core::futures::stream::StreamExt;
use deno_core::futures::task;
use deno_core::futures::task::Context;
use deno_core::futures::task::Poll;
use deno_core::serde_json;
use deno_core::serde_json::json;
use deno_core::serde_json::Value;
use deno_core::v8;
use crate::error::generic_error;
use crate::error::AnyError;
use crate::futures::channel::mpsc;
use crate::futures::channel::mpsc::UnboundedReceiver;
use crate::futures::channel::mpsc::UnboundedSender;
use crate::futures::channel::oneshot;
use crate::futures::future::select;
use crate::futures::future::Either;
use crate::futures::future::Future;
use crate::futures::prelude::*;
use crate::futures::stream::FuturesUnordered;
use crate::futures::stream::StreamExt;
use crate::futures::task;
use crate::futures::task::Context;
use crate::futures::task::Poll;
use crate::serde_json;
use crate::serde_json::json;
use crate::serde_json::Value;
use crate::v8;
use std::cell::BorrowMutError;
use std::cell::RefCell;
use std::collections::HashMap;
@ -38,10 +38,6 @@ use std::sync::Arc;
use std::sync::Mutex;
use std::thread;
mod server;
pub use server::InspectorServer;
/// If first argument is `None` then it's a notification, otherwise
/// it's a message.
pub type SessionProxySender = UnboundedSender<(Option<i32>, String)>;
@ -51,12 +47,12 @@ pub type SessionProxyReceiver = UnboundedReceiver<Result<Vec<u8>, AnyError>>;
/// Encapsulates an UnboundedSender/UnboundedReceiver pair that together form
/// a duplex channel for sending/receiving messages in V8 session.
pub struct SessionProxy {
pub struct InspectorSessionProxy {
pub tx: SessionProxySender,
pub rx: SessionProxyReceiver,
}
impl SessionProxy {
impl InspectorSessionProxy {
pub fn split(self) -> (SessionProxySender, SessionProxyReceiver) {
(self.tx, self.rx)
}
@ -84,7 +80,7 @@ enum PollState {
pub struct JsRuntimeInspector {
v8_inspector_client: v8::inspector::V8InspectorClientBase,
v8_inspector: Rc<RefCell<v8::UniquePtr<v8::inspector::V8Inspector>>>,
new_session_tx: UnboundedSender<SessionProxy>,
new_session_tx: UnboundedSender<InspectorSessionProxy>,
sessions: RefCell<SessionContainer>,
flags: RefCell<InspectorFlags>,
waker: Arc<InspectorWaker>,
@ -153,11 +149,14 @@ impl JsRuntimeInspector {
/// and thus it's id is provided as an associated contant.
const CONTEXT_GROUP_ID: i32 = 1;
pub fn new(js_runtime: &mut deno_core::JsRuntime) -> Box<Self> {
let context = js_runtime.global_context();
let scope = &mut v8::HandleScope::new(js_runtime.v8_isolate());
pub fn new(
isolate: &mut v8::OwnedIsolate,
context: v8::Global<v8::Context>,
) -> Box<Self> {
let scope = &mut v8::HandleScope::new(isolate);
let (new_session_tx, new_session_rx) = mpsc::unbounded::<SessionProxy>();
let (new_session_tx, new_session_rx) =
mpsc::unbounded::<InspectorSessionProxy>();
let v8_inspector_client =
v8::inspector::V8InspectorClientBase::new::<Self>();
@ -198,6 +197,11 @@ impl JsRuntimeInspector {
self_
}
pub fn has_active_sessions(&self) -> bool {
let sessions = self.sessions.borrow();
!sessions.established.is_empty() || sessions.handshake.is_some()
}
fn poll_sessions(
&self,
mut invoker_cx: Option<&mut Context>,
@ -323,7 +327,7 @@ impl JsRuntimeInspector {
/// After a proxy is sent inspector will wait for a "handshake".
/// Frontend must send "Runtime.runIfWaitingForDebugger" message to
/// complete the handshake.
pub fn get_session_sender(&self) -> UnboundedSender<SessionProxy> {
pub fn get_session_sender(&self) -> UnboundedSender<InspectorSessionProxy> {
self.new_session_tx.clone()
}
@ -349,7 +353,7 @@ impl JsRuntimeInspector {
// The 'inbound' channel carries messages received from the session.
let (inbound_tx, inbound_rx) = mpsc::unbounded();
let proxy = SessionProxy {
let proxy = InspectorSessionProxy {
tx: outbound_tx,
rx: inbound_rx,
};
@ -395,7 +399,7 @@ struct SessionContainer {
impl SessionContainer {
fn new(
v8_inspector: Rc<RefCell<v8::UniquePtr<v8::inspector::V8Inspector>>>,
new_session_rx: UnboundedReceiver<SessionProxy>,
new_session_rx: UnboundedReceiver<InspectorSessionProxy>,
) -> RefCell<Self> {
let new_incoming = new_session_rx
.map(move |session_proxy| {
@ -506,7 +510,7 @@ impl InspectorSession {
pub fn new(
v8_inspector_rc: Rc<RefCell<v8::UniquePtr<v8::inspector::V8Inspector>>>,
session_proxy: SessionProxy,
session_proxy: InspectorSessionProxy,
) -> Box<Self> {
new_box_with(move |self_ptr| {
let v8_channel = v8::inspector::ChannelBase::new::<Self>();

4
core/lib.rs
View file

@ -6,6 +6,7 @@ pub mod error;
mod extensions;
mod flags;
mod gotham_state;
mod inspector;
mod module_specifier;
mod modules;
mod normalize_path;
@ -37,6 +38,9 @@ pub use crate::async_cell::AsyncRefFuture;
pub use crate::async_cell::RcLike;
pub use crate::async_cell::RcRef;
pub use crate::flags::v8_set_flags;
pub use crate::inspector::InspectorSessionProxy;
pub use crate::inspector::JsRuntimeInspector;
pub use crate::inspector::LocalInspectorSession;
pub use crate::module_specifier::resolve_import;
pub use crate::module_specifier::resolve_path;
pub use crate::module_specifier::resolve_url;

26
core/modules.rs
View file

@ -902,7 +902,7 @@ mod tests {
let a_id = futures::executor::block_on(a_id_fut).expect("Failed to load");
runtime.mod_evaluate(a_id);
futures::executor::block_on(runtime.run_event_loop()).unwrap();
futures::executor::block_on(runtime.run_event_loop(false)).unwrap();
let l = loads.lock().unwrap();
assert_eq!(
l.to_vec(),
@ -1130,7 +1130,7 @@ mod tests {
assert_eq!(count.load(Ordering::Relaxed), 0);
// We should get an error here.
let result = runtime.poll_event_loop(cx);
let result = runtime.poll_event_loop(cx, false);
if let Poll::Ready(Ok(_)) = result {
unreachable!();
}
@ -1223,14 +1223,20 @@ mod tests {
.unwrap();
// First poll runs `prepare_load` hook.
assert!(matches!(runtime.poll_event_loop(cx), Poll::Pending));
assert!(matches!(runtime.poll_event_loop(cx, false), Poll::Pending));
assert_eq!(prepare_load_count.load(Ordering::Relaxed), 1);
// Second poll actually loads modules into the isolate.
assert!(matches!(runtime.poll_event_loop(cx), Poll::Ready(Ok(_))));
assert!(matches!(
runtime.poll_event_loop(cx, false),
Poll::Ready(Ok(_))
));
assert_eq!(resolve_count.load(Ordering::Relaxed), 4);
assert_eq!(load_count.load(Ordering::Relaxed), 2);
assert!(matches!(runtime.poll_event_loop(cx), Poll::Ready(Ok(_))));
assert!(matches!(
runtime.poll_event_loop(cx, false),
Poll::Ready(Ok(_))
));
assert_eq!(resolve_count.load(Ordering::Relaxed), 4);
assert_eq!(load_count.load(Ordering::Relaxed), 2);
})
@ -1261,10 +1267,10 @@ mod tests {
)
.unwrap();
// First poll runs `prepare_load` hook.
let _ = runtime.poll_event_loop(cx);
let _ = runtime.poll_event_loop(cx, false);
assert_eq!(prepare_load_count.load(Ordering::Relaxed), 1);
// Second poll triggers error
let _ = runtime.poll_event_loop(cx);
let _ = runtime.poll_event_loop(cx, false);
})
}
@ -1283,7 +1289,7 @@ mod tests {
assert!(result.is_ok());
let circular1_id = result.unwrap();
runtime.mod_evaluate(circular1_id);
runtime.run_event_loop().await.unwrap();
runtime.run_event_loop(false).await.unwrap();
let l = loads.lock().unwrap();
assert_eq!(
@ -1356,7 +1362,7 @@ mod tests {
assert!(result.is_ok());
let redirect1_id = result.unwrap();
runtime.mod_evaluate(redirect1_id);
runtime.run_event_loop().await.unwrap();
runtime.run_event_loop(false).await.unwrap();
let l = loads.lock().unwrap();
assert_eq!(
l.to_vec(),
@ -1505,7 +1511,7 @@ mod tests {
futures::executor::block_on(main_id_fut).expect("Failed to load");
runtime.mod_evaluate(main_id);
futures::executor::block_on(runtime.run_event_loop()).unwrap();
futures::executor::block_on(runtime.run_event_loop(false)).unwrap();
let l = loads.lock().unwrap();
assert_eq!(

2
core/ops_json.rs
View file

@ -134,7 +134,7 @@ mod tests {
"#,
)
.unwrap();
let e = runtime.run_event_loop().await.unwrap_err().to_string();
let e = runtime.run_event_loop(false).await.unwrap_err().to_string();
println!("{}", e);
assert!(e.contains("Error: foo"));
assert!(e.contains("at async f1 (<init>:"));

59
core/runtime.rs
View file

@ -8,6 +8,7 @@ use crate::error::generic_error;
use crate::error::AnyError;
use crate::error::ErrWithV8Handle;
use crate::error::JsError;
use crate::inspector::JsRuntimeInspector;
use crate::module_specifier::ModuleSpecifier;
use crate::modules::ModuleId;
use crate::modules::ModuleLoadId;
@ -23,6 +24,7 @@ use crate::OpState;
use crate::PromiseId;
use futures::channel::mpsc;
use futures::future::poll_fn;
use futures::future::FutureExt;
use futures::stream::FuturesUnordered;
use futures::stream::StreamExt;
use futures::task::AtomicWaker;
@ -75,6 +77,7 @@ pub struct JsRuntime {
// This is an Option<OwnedIsolate> instead of just OwnedIsolate to workaround
// an safety issue with SnapshotCreator. See JsRuntime::drop.
v8_isolate: Option<v8::OwnedIsolate>,
inspector: Option<Box<JsRuntimeInspector>>,
snapshot_creator: Option<v8::SnapshotCreator>,
has_snapshotted: bool,
allocations: IsolateAllocations,
@ -113,6 +116,10 @@ pub(crate) struct JsRuntimeState {
impl Drop for JsRuntime {
fn drop(&mut self) {
// The Isolate object must outlive the Inspector object, but this is
// currently not enforced by the type system.
self.inspector.take();
if let Some(creator) = self.snapshot_creator.take() {
// TODO(ry): in rusty_v8, `SnapShotCreator::get_owned_isolate()` returns
// a `struct OwnedIsolate` which is not actually owned, hence the need
@ -198,6 +205,9 @@ pub struct RuntimeOptions {
/// V8 platform instance to use. Used when Deno initializes V8
/// (which it only does once), otherwise it's silenty dropped.
pub v8_platform: Option<v8::UniquePtr<v8::Platform>>,
/// Create a V8 inspector and attach to the runtime.
pub attach_inspector: bool,
}
impl JsRuntime {
@ -258,6 +268,14 @@ impl JsRuntime {
(isolate, None)
};
let maybe_inspector = if options.attach_inspector {
let inspector =
JsRuntimeInspector::new(&mut isolate, global_context.clone());
Some(inspector)
} else {
None
};
let loader = options
.module_loader
.unwrap_or_else(|| Rc::new(NoopModuleLoader));
@ -298,6 +316,7 @@ impl JsRuntime {
let mut js_runtime = Self {
v8_isolate: Some(isolate),
inspector: maybe_inspector,
snapshot_creator: maybe_snapshot_creator,
has_snapshotted: false,
allocations: IsolateAllocations::default(),
@ -328,6 +347,10 @@ impl JsRuntime {
self.v8_isolate.as_mut().unwrap()
}
pub fn inspector(&mut self) -> Option<&mut Box<JsRuntimeInspector>> {
self.inspector.as_mut()
}
pub fn handle_scope(&mut self) -> v8::HandleScope {
let context = self.global_context();
v8::HandleScope::with_context(self.v8_isolate(), context)
@ -568,15 +591,26 @@ impl JsRuntime {
/// This future resolves when:
/// - there are no more pending dynamic imports
/// - there are no more pending ops
pub async fn run_event_loop(&mut self) -> Result<(), AnyError> {
poll_fn(|cx| self.poll_event_loop(cx)).await
/// - there are no more active inspector sessions (only if `wait_for_inspector` is set to true)
pub async fn run_event_loop(
&mut self,
wait_for_inspector: bool,
) -> Result<(), AnyError> {
poll_fn(|cx| self.poll_event_loop(cx, wait_for_inspector)).await
}
/// Runs a single tick of event loop
///
/// If `wait_for_inspector` is set to true event loop
/// will return `Poll::Pending` if there are active inspector sessions.
pub fn poll_event_loop(
&mut self,
cx: &mut Context,
wait_for_inspector: bool,
) -> Poll<Result<(), AnyError>> {
// We always poll the inspector if it exists.
let _ = self.inspector().map(|i| i.poll_unpin(cx));
let state_rc = Self::state(self.v8_isolate());
let module_map_rc = Self::module_map(self.v8_isolate());
{
@ -617,12 +651,21 @@ 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 inspector_has_active_sessions = self
.inspector
.as_ref()
.map(|i| i.has_active_sessions())
.unwrap_or(false);
if !has_pending_ops
&& !has_pending_dyn_imports
&& !has_pending_dyn_module_evaluation
&& !has_pending_module_evaluation
{
if wait_for_inspector && inspector_has_active_sessions {
return Poll::Pending;
}
return Poll::Ready(Ok(()));
}
@ -1562,7 +1605,7 @@ pub mod tests {
"#,
)
.unwrap();
if let Poll::Ready(Err(_)) = runtime.poll_event_loop(&mut cx) {
if let Poll::Ready(Err(_)) = runtime.poll_event_loop(&mut cx, false) {
unreachable!();
}
});
@ -1588,7 +1631,7 @@ pub mod tests {
include_str!("encode_decode_test.js"),
)
.unwrap();
if let Poll::Ready(Err(_)) = runtime.poll_event_loop(&mut cx) {
if let Poll::Ready(Err(_)) = runtime.poll_event_loop(&mut cx, false) {
unreachable!();
}
});
@ -1604,7 +1647,7 @@ pub mod tests {
include_str!("serialize_deserialize_test.js"),
)
.unwrap();
if let Poll::Ready(Err(_)) = runtime.poll_event_loop(&mut cx) {
if let Poll::Ready(Err(_)) = runtime.poll_event_loop(&mut cx, false) {
unreachable!();
}
});
@ -1637,7 +1680,7 @@ pub mod tests {
include_str!("error_builder_test.js"),
)
.unwrap();
if let Poll::Ready(Err(_)) = runtime.poll_event_loop(&mut cx) {
if let Poll::Ready(Err(_)) = runtime.poll_event_loop(&mut cx, false) {
unreachable!();
}
});
@ -1817,7 +1860,7 @@ pub mod tests {
.unwrap();
runtime.mod_evaluate(module_id);
futures::executor::block_on(runtime.run_event_loop()).unwrap();
futures::executor::block_on(runtime.run_event_loop(false)).unwrap();
let _snapshot = runtime.snapshot();
}
@ -1896,7 +1939,7 @@ main();
at async error_async_stack.js:4:5
at async error_async_stack.js:10:5"#;
match runtime.poll_event_loop(cx) {
match runtime.poll_event_loop(cx, false) {
Poll::Ready(Err(e)) => {
assert_eq!(e.to_string(), expected_error);
}

2
runtime/examples/hello_runtime.rs
View file

@ -55,6 +55,6 @@ async fn main() -> Result<(), AnyError> {
MainWorker::from_options(main_module.clone(), permissions, &options);
worker.bootstrap(&options);
worker.execute_module(&main_module).await?;
worker.run_event_loop().await?;
worker.run_event_loop(false).await?;
Ok(())
}

13
runtime/inspector/server.rs → runtime/inspector_server.rs
View file

@ -16,6 +16,7 @@ use deno_core::futures::task::Poll;
use deno_core::serde_json;
use deno_core::serde_json::json;
use deno_core::serde_json::Value;
use deno_core::InspectorSessionProxy;
use deno_websocket::tokio_tungstenite::tungstenite;
use std::cell::RefCell;
use std::collections::HashMap;
@ -26,8 +27,6 @@ use std::rc::Rc;
use std::thread;
use uuid::Uuid;
use super::SessionProxy;
/// Websocket server that is used to proxy connections from
/// devtools to the inspector.
pub struct InspectorServer {
@ -63,7 +62,7 @@ impl InspectorServer {
pub fn register_inspector(
&self,
session_sender: UnboundedSender<SessionProxy>,
session_sender: UnboundedSender<InspectorSessionProxy>,
deregister_rx: oneshot::Receiver<()>,
) {
let info = InspectorInfo::new(self.host, session_sender, deregister_rx);
@ -288,14 +287,14 @@ fn create_websocket_proxy(
websocket: deno_websocket::tokio_tungstenite::WebSocketStream<
hyper::upgrade::Upgraded,
>,
) -> (SessionProxy, impl Future<Output = ()> + Send) {
) -> (InspectorSessionProxy, impl Future<Output = ()> + Send) {
// The 'outbound' channel carries messages sent to the websocket.
let (outbound_tx, outbound_rx) = mpsc::unbounded();
// The 'inbound' channel carries messages received from the websocket.
let (inbound_tx, inbound_rx) = mpsc::unbounded();
let proxy = SessionProxy {
let proxy = InspectorSessionProxy {
tx: outbound_tx,
rx: inbound_rx,
};
@ -332,14 +331,14 @@ pub struct InspectorInfo {
pub host: SocketAddr,
pub uuid: Uuid,
pub thread_name: Option<String>,
pub new_session_tx: UnboundedSender<SessionProxy>,
pub new_session_tx: UnboundedSender<InspectorSessionProxy>,
pub deregister_rx: oneshot::Receiver<()>,
}
impl InspectorInfo {
pub fn new(
host: SocketAddr,
new_session_tx: mpsc::UnboundedSender<SessionProxy>,
new_session_tx: mpsc::UnboundedSender<InspectorSessionProxy>,
deregister_rx: oneshot::Receiver<()>,
) -> Self {
Self {

2
runtime/lib.rs
View file

@ -16,7 +16,7 @@ pub use deno_webstorage;
pub mod colors;
pub mod errors;
pub mod fs_util;
pub mod inspector;
pub mod inspector_server;
pub mod js;
pub mod metrics;
pub mod ops;

44
runtime/web_worker.rs
View file

@ -1,7 +1,6 @@
// Copyright 2018-2021 the Deno authors. All rights reserved. MIT license.
use crate::colors;
use crate::inspector::InspectorServer;
use crate::inspector::JsRuntimeInspector;
use crate::inspector_server::InspectorServer;
use crate::js;
use crate::metrics;
use crate::ops;
@ -199,7 +198,6 @@ fn create_handles(
/// `WebWorker`.
pub struct WebWorker {
id: WorkerId,
inspector: Option<Box<JsRuntimeInspector>>,
pub js_runtime: JsRuntime,
pub name: String,
internal_handle: WebWorkerInternalHandle,
@ -320,23 +318,18 @@ impl WebWorker {
startup_snapshot: Some(js::deno_isolate_init()),
js_error_create_fn: options.js_error_create_fn.clone(),
get_error_class_fn: options.get_error_class_fn,
attach_inspector: options.attach_inspector,
extensions,
..Default::default()
});
let inspector = if options.attach_inspector {
let mut inspector = JsRuntimeInspector::new(&mut js_runtime);
if let Some(inspector) = js_runtime.inspector() {
if let Some(server) = options.maybe_inspector_server.clone() {
let session_sender = inspector.get_session_sender();
let deregister_rx = inspector.add_deregister_handler();
server.register_inspector(session_sender, deregister_rx);
}
Some(inspector)
} else {
None
};
}
let (internal_handle, external_handle) = {
let handle = js_runtime.v8_isolate().thread_safe_handle();
@ -349,7 +342,6 @@ impl WebWorker {
Self {
id: worker_id,
inspector,
js_runtime,
name,
internal_handle,
@ -424,7 +416,7 @@ impl WebWorker {
return result;
}
event_loop_result = self.run_event_loop() => {
event_loop_result = self.run_event_loop(false) => {
if self.internal_handle.is_terminated() {
return Ok(());
}
@ -444,15 +436,14 @@ impl WebWorker {
pub fn poll_event_loop(
&mut self,
cx: &mut Context,
wait_for_inspector: bool,
) -> Poll<Result<(), AnyError>> {
// If awakened because we are terminating, just return Ok
if self.internal_handle.is_terminated() {
return Poll::Ready(Ok(()));
}
// We always poll the inspector if it exists.
let _ = self.inspector.as_mut().map(|i| i.poll_unpin(cx));
match self.js_runtime.poll_event_loop(cx) {
match self.js_runtime.poll_event_loop(cx, wait_for_inspector) {
Poll::Ready(r) => {
// If js ended because we are terminating, just return Ok
if self.internal_handle.is_terminated() {
@ -478,16 +469,11 @@ impl WebWorker {
}
}
pub async fn run_event_loop(&mut self) -> Result<(), AnyError> {
poll_fn(|cx| self.poll_event_loop(cx)).await
}
}
impl Drop for WebWorker {
fn drop(&mut self) {
// The Isolate object must outlive the Inspector object, but this is
// currently not enforced by the type system.
self.inspector.take();
pub async fn run_event_loop(
&mut self,
wait_for_inspector: bool,
) -> Result<(), AnyError> {
poll_fn(|cx| self.poll_event_loop(cx, wait_for_inspector)).await
}
}
@ -543,7 +529,7 @@ pub fn run_web_worker(
return Ok(());
}
let result = rt.block_on(worker.run_event_loop());
let result = rt.block_on(worker.run_event_loop(true));
debug!("Worker thread shuts down {}", &name);
result
}
@ -613,7 +599,7 @@ mod tests {
worker.execute(source).unwrap();
let handle = worker.thread_safe_handle();
handle_sender.send(handle).unwrap();
let r = tokio_util::run_basic(worker.run_event_loop());
let r = tokio_util::run_basic(worker.run_event_loop(false));
assert!(r.is_ok())
});
@ -660,7 +646,7 @@ mod tests {
worker.execute("onmessage = () => { close(); }").unwrap();
let handle = worker.thread_safe_handle();
handle_sender.send(handle).unwrap();
let r = tokio_util::run_basic(worker.run_event_loop());
let r = tokio_util::run_basic(worker.run_event_loop(false));
assert!(r.is_ok())
});

62
runtime/worker.rs
View file

@ -1,8 +1,6 @@
// Copyright 2018-2021 the Deno authors. All rights reserved. MIT license.
use crate::inspector::InspectorServer;
use crate::inspector::JsRuntimeInspector;
use crate::inspector::LocalInspectorSession;
use crate::inspector_server::InspectorServer;
use crate::js;
use crate::metrics;
use crate::ops;
@ -11,7 +9,6 @@ use deno_broadcast_channel::InMemoryBroadcastChannel;
use deno_core::error::AnyError;
use deno_core::error::Context as ErrorContext;
use deno_core::futures::future::poll_fn;
use deno_core::futures::future::FutureExt;
use deno_core::futures::stream::StreamExt;
use deno_core::futures::Future;
use deno_core::serde_json;
@ -21,6 +18,7 @@ use deno_core::Extension;
use deno_core::GetErrorClassFn;
use deno_core::JsErrorCreateFn;
use deno_core::JsRuntime;
use deno_core::LocalInspectorSession;
use deno_core::ModuleId;
use deno_core::ModuleLoader;
use deno_core::ModuleSpecifier;
@ -42,7 +40,6 @@ use std::task::Poll;
/// All `WebWorker`s created during program execution
/// are descendants of this worker.
pub struct MainWorker {
inspector: Option<Box<JsRuntimeInspector>>,
pub js_runtime: JsRuntime,
should_break_on_first_statement: bool,
}
@ -145,28 +142,22 @@ impl MainWorker {
js_error_create_fn: options.js_error_create_fn.clone(),
get_error_class_fn: options.get_error_class_fn,
extensions,
attach_inspector: options.attach_inspector,
..Default::default()
});
let inspector = if options.attach_inspector {
let mut inspector = JsRuntimeInspector::new(&mut js_runtime);
let mut should_break_on_first_statement = false;
if let Some(inspector) = js_runtime.inspector() {
if let Some(server) = options.maybe_inspector_server.clone() {
let session_sender = inspector.get_session_sender();
let deregister_rx = inspector.add_deregister_handler();
server.register_inspector(session_sender, deregister_rx);
}
Some(inspector)
} else {
None
};
let should_break_on_first_statement =
inspector.is_some() && options.should_break_on_first_statement;
should_break_on_first_statement = options.should_break_on_first_statement;
}
Self {
inspector,
js_runtime,
should_break_on_first_statement,
}
@ -229,7 +220,7 @@ impl MainWorker {
return result;
}
event_loop_result = self.run_event_loop() => {
event_loop_result = self.run_event_loop(false) => {
event_loop_result?;
let maybe_result = receiver.next().await;
let result = maybe_result.expect("Module evaluation result not provided.");
@ -241,8 +232,8 @@ impl MainWorker {
fn wait_for_inspector_session(&mut self) {
if self.should_break_on_first_statement {
self
.inspector
.as_mut()
.js_runtime
.inspector()
.unwrap()
.wait_for_session_and_break_on_next_statement()
}
@ -251,21 +242,23 @@ impl MainWorker {
/// Create new inspector session. This function panics if Worker
/// was not configured to create inspector.
pub async fn create_inspector_session(&mut self) -> LocalInspectorSession {
let inspector = self.inspector.as_ref().unwrap();
let inspector = self.js_runtime.inspector().unwrap();
inspector.create_local_session()
}
pub fn poll_event_loop(
&mut self,
cx: &mut Context,
wait_for_inspector: bool,
) -> Poll<Result<(), AnyError>> {
// We always poll the inspector if it exists.
let _ = self.inspector.as_mut().map(|i| i.poll_unpin(cx));
self.js_runtime.poll_event_loop(cx)
self.js_runtime.poll_event_loop(cx, wait_for_inspector)
}
pub async fn run_event_loop(&mut self) -> Result<(), AnyError> {
poll_fn(|cx| self.poll_event_loop(cx)).await
pub async fn run_event_loop(
&mut self,
wait_for_inspector: bool,
) -> Result<(), AnyError> {
poll_fn(|cx| self.poll_event_loop(cx, wait_for_inspector)).await
}
/// A utility function that runs provided future concurrently with the event loop.
@ -280,25 +273,12 @@ impl MainWorker {
result = &mut fut => {
return result;
}
_ = self.run_event_loop() => {
// A zero delay is long enough to yield the thread in order to prevent the loop from
// running hot for messages that are taking longer to resolve like for example an
// evaluation of top level await.
tokio::time::sleep(tokio::time::Duration::from_millis(0)).await;
}
_ = self.run_event_loop(false) => {}
};
}
}
}
impl Drop for MainWorker {
fn drop(&mut self) {
// The Isolate object must outlive the Inspector object, but this is
// currently not enforced by the type system.
self.inspector.take();
}
}
#[cfg(test)]
mod tests {
use super::*;
@ -347,7 +327,7 @@ mod tests {
if let Err(err) = result {
eprintln!("execute_mod err {:?}", err);
}
if let Err(e) = worker.run_event_loop().await {
if let Err(e) = worker.run_event_loop(false).await {
panic!("Future got unexpected error: {:?}", e);
}
}
@ -364,7 +344,7 @@ mod tests {
if let Err(err) = result {
eprintln!("execute_mod err {:?}", err);
}
if let Err(e) = worker.run_event_loop().await {
if let Err(e) = worker.run_event_loop(false).await {
panic!("Future got unexpected error: {:?}", e);
}
}