From e5beb800c94099852964d482a32a13f5c29ec147 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bartek=20Iwa=C5=84czuk?= Date: Wed, 26 May 2021 21:07:12 +0200 Subject: refactor: move JsRuntimeInspector to deno_core (#10763) 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. --- bench_util/src/js_runtime.rs | 2 +- cli/main.rs | 12 +- cli/program_state.rs | 2 +- cli/standalone.rs | 2 +- cli/tools/coverage.rs | 2 +- cli/tools/repl.rs | 4 +- cli/tools/test_runner.rs | 4 +- core/examples/http_bench_json_ops.rs | 2 +- core/inspector.rs | 755 +++++++++++++++++++++++++++++++++++ core/lib.rs | 4 + core/modules.rs | 26 +- core/ops_json.rs | 2 +- core/runtime.rs | 59 ++- runtime/examples/hello_runtime.rs | 2 +- runtime/inspector/mod.rs | 751 ---------------------------------- runtime/inspector/server.rs | 389 ------------------ runtime/inspector_server.rs | 388 ++++++++++++++++++ runtime/lib.rs | 2 +- runtime/web_worker.rs | 44 +- runtime/worker.rs | 62 +-- 20 files changed, 1270 insertions(+), 1244 deletions(-) create mode 100644 core/inspector.rs delete mode 100644 runtime/inspector/mod.rs delete mode 100644 runtime/inspector/server.rs create mode 100644 runtime/inspector_server.rs diff --git a/bench_util/src/js_runtime.rs b/bench_util/src/js_runtime.rs index f0280b072..a801137fa 100644 --- a/bench_util/src/js_runtime.rs +++ b/bench_util/src/js_runtime.rs @@ -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(); } diff --git a/cli/main.rs b/cli/main.rs index 185de5bfe..c44c002b2 100644 --- a/cli/main.rs +++ b/cli/main.rs @@ -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() { diff --git a/cli/program_state.rs b/cli/program_state.rs index 9f7ddc749..c3c399b69 100644 --- a/cli/program_state.rs +++ b/cli/program_state.rs @@ -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; diff --git a/cli/standalone.rs b/cli/standalone.rs index f281c5336..f79adb510 100644 --- a/cli/standalone.rs +++ b/cli/standalone.rs @@ -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); } diff --git a/cli/tools/coverage.rs b/cli/tools/coverage.rs index 204141b25..9a64cd5f3 100644 --- a/cli/tools/coverage.rs +++ b/cli/tools/coverage.rs @@ -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; diff --git a/cli/tools/repl.rs b/cli/tools/repl.rs index ba69c9234..47f46c361 100644 --- a/cli/tools/repl.rs +++ b/cli/tools/repl.rs @@ -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 => { diff --git a/cli/tools/test_runner.rs b/cli/tools/test_runner.rs index 7d1eae9ee..24bea3ff8 100644 --- a/cli/tools/test_runner.rs +++ b/cli/tools/test_runner.rs @@ -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() { diff --git a/core/examples/http_bench_json_ops.rs b/core/examples/http_bench_json_ops.rs index 88556cea7..a29ebfcf2 100644 --- a/core/examples/http_bench_json_ops.rs +++ b/core/examples/http_bench_json_ops.rs @@ -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(); } diff --git a/core/inspector.rs b/core/inspector.rs new file mode 100644 index 000000000..f6e348d88 --- /dev/null +++ b/core/inspector.rs @@ -0,0 +1,755 @@ +// Copyright 2018-2021 the Deno authors. All rights reserved. MIT license. + +//! The documentation for the inspector API is sparse, but these are helpful: +//! https://chromedevtools.github.io/devtools-protocol/ +//! https://hyperandroid.com/2020/02/12/v8-inspector-from-an-embedder-standpoint/ + +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; +use std::ffi::c_void; +use std::mem::replace; +use std::mem::take; +use std::mem::MaybeUninit; +use std::pin::Pin; +use std::ptr; +use std::ptr::NonNull; +use std::rc::Rc; +use std::sync::Arc; +use std::sync::Mutex; +use std::thread; + +/// If first argument is `None` then it's a notification, otherwise +/// it's a message. +pub type SessionProxySender = UnboundedSender<(Option, String)>; +// TODO(bartlomieju): does it even need to send a Result? +// It seems `Vec` would be enough +pub type SessionProxyReceiver = UnboundedReceiver, AnyError>>; + +/// Encapsulates an UnboundedSender/UnboundedReceiver pair that together form +/// a duplex channel for sending/receiving messages in V8 session. +pub struct InspectorSessionProxy { + pub tx: SessionProxySender, + pub rx: SessionProxyReceiver, +} + +impl InspectorSessionProxy { + pub fn split(self) -> (SessionProxySender, SessionProxyReceiver) { + (self.tx, self.rx) + } +} + +#[derive(Clone, Copy)] +enum PollState { + Idle, + Woken, + Polling, + Parked, + Dropped, +} + +/// This structure is used responsible for providing inspector interface +/// to the `JsRuntime`. +/// +/// It stores an instance of `v8::inspector::V8Inspector` and additionally +/// implements `v8::inspector::V8InspectorClientImpl`. +/// +/// After creating this structure it's possible to connect multiple sessions +/// to the inspector, in case of Deno it's either: a "websocket session" that +/// provides integration with Chrome Devtools, or an "in-memory session" that +/// is used for REPL or converage collection. +pub struct JsRuntimeInspector { + v8_inspector_client: v8::inspector::V8InspectorClientBase, + v8_inspector: Rc>>, + new_session_tx: UnboundedSender, + sessions: RefCell, + flags: RefCell, + waker: Arc, + deregister_tx: Option>, +} + +impl Drop for JsRuntimeInspector { + fn drop(&mut self) { + // Since the waker is cloneable, it might outlive the inspector itself. + // Set the poll state to 'dropped' so it doesn't attempt to request an + // interrupt from the isolate. + self.waker.update(|w| w.poll_state = PollState::Dropped); + // TODO(bartlomieju): this comment is out of date + // V8 automatically deletes all sessions when an `V8Inspector` instance is + // deleted, however InspectorSession also has a drop handler that cleans + // up after itself. To avoid a double free, make sure the inspector is + // dropped last. + take(&mut *self.sessions.borrow_mut()); + + // Notify counterparty that this instance is being destroyed. Ignoring + // result because counterparty waiting for the signal might have already + // dropped the other end of channel. + if let Some(deregister_tx) = self.deregister_tx.take() { + let _ = deregister_tx.send(()); + } + } +} + +impl v8::inspector::V8InspectorClientImpl for JsRuntimeInspector { + fn base(&self) -> &v8::inspector::V8InspectorClientBase { + &self.v8_inspector_client + } + + fn base_mut(&mut self) -> &mut v8::inspector::V8InspectorClientBase { + &mut self.v8_inspector_client + } + + fn run_message_loop_on_pause(&mut self, context_group_id: i32) { + assert_eq!(context_group_id, JsRuntimeInspector::CONTEXT_GROUP_ID); + self.flags.borrow_mut().on_pause = true; + let _ = self.poll_sessions(None); + } + + fn quit_message_loop_on_pause(&mut self) { + self.flags.borrow_mut().on_pause = false; + } + + fn run_if_waiting_for_debugger(&mut self, context_group_id: i32) { + assert_eq!(context_group_id, JsRuntimeInspector::CONTEXT_GROUP_ID); + self.flags.borrow_mut().session_handshake_done = true; + } +} + +/// `JsRuntimeInspector` implements a Future so that it can poll for new incoming +/// connections and messages from the WebSocket server. The Worker that owns +/// this `JsRuntimeInspector` will call this function from `Worker::poll()`. +impl Future for JsRuntimeInspector { + type Output = (); + fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<()> { + self.poll_sessions(Some(cx)).unwrap() + } +} + +impl JsRuntimeInspector { + /// Currently Deno supports only a single context in `JsRuntime` + /// and thus it's id is provided as an associated contant. + const CONTEXT_GROUP_ID: i32 = 1; + + pub fn new( + isolate: &mut v8::OwnedIsolate, + context: v8::Global, + ) -> Box { + let scope = &mut v8::HandleScope::new(isolate); + + let (new_session_tx, new_session_rx) = + mpsc::unbounded::(); + + let v8_inspector_client = + v8::inspector::V8InspectorClientBase::new::(); + + let flags = InspectorFlags::new(); + let waker = InspectorWaker::new(scope.thread_safe_handle()); + + // Create JsRuntimeInspector instance. + let mut self_ = Box::new(Self { + v8_inspector_client, + v8_inspector: Default::default(), + sessions: Default::default(), + new_session_tx, + flags, + waker, + deregister_tx: None, + }); + self_.v8_inspector = Rc::new(RefCell::new( + v8::inspector::V8Inspector::create(scope, &mut *self_).into(), + )); + self_.sessions = + SessionContainer::new(self_.v8_inspector.clone(), new_session_rx); + + // Tell the inspector about the global context. + let context = v8::Local::new(scope, context); + let context_name = v8::inspector::StringView::from(&b"global context"[..]); + self_ + .v8_inspector + .borrow_mut() + .as_mut() + .unwrap() + .context_created(context, Self::CONTEXT_GROUP_ID, context_name); + + // Poll the session handler so we will get notified whenever there is + // new_incoming debugger activity. + let _ = self_.poll_sessions(None).unwrap(); + + 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>, + ) -> Result, BorrowMutError> { + // The futures this function uses do not have re-entrant poll() functions. + // However it is can happpen that poll_sessions() gets re-entered, e.g. + // when an interrupt request is honored while the inspector future is polled + // by the task executor. We let the caller know by returning some error. + let mut sessions = self.sessions.try_borrow_mut()?; + + self.waker.update(|w| { + match w.poll_state { + PollState::Idle | PollState::Woken => w.poll_state = PollState::Polling, + _ => unreachable!(), + }; + }); + + // Create a new Context object that will make downstream futures + // use the InspectorWaker when they are ready to be polled again. + let waker_ref = task::waker_ref(&self.waker); + let cx = &mut Context::from_waker(&waker_ref); + + loop { + loop { + // Do one "handshake" with a newly connected session at a time. + if let Some(session) = &mut sessions.handshake { + let poll_result = session.poll_unpin(cx); + let handshake_done = + replace(&mut self.flags.borrow_mut().session_handshake_done, false); + match poll_result { + Poll::Pending if handshake_done => { + let session = sessions.handshake.take().unwrap(); + sessions.established.push(session); + take(&mut self.flags.borrow_mut().waiting_for_session); + } + Poll::Ready(_) => sessions.handshake = None, + Poll::Pending => break, + }; + } + + // Accept new connections. + match sessions.new_incoming.poll_next_unpin(cx) { + Poll::Ready(Some(session)) => { + let prev = sessions.handshake.replace(session); + assert!(prev.is_none()); + continue; + } + Poll::Ready(None) => {} + Poll::Pending => {} + } + + // Poll established sessions. + match sessions.established.poll_next_unpin(cx) { + Poll::Ready(Some(_)) => continue, + Poll::Ready(None) => break, + Poll::Pending => break, + }; + } + + let should_block = sessions.handshake.is_some() + || self.flags.borrow().on_pause + || self.flags.borrow().waiting_for_session; + + let new_state = self.waker.update(|w| { + match w.poll_state { + PollState::Woken => { + // The inspector was woken while the session handler was being + // polled, so we poll it another time. + w.poll_state = PollState::Polling; + } + PollState::Polling if !should_block => { + // The session handler doesn't need to be polled any longer, and + // there's no reason to block (execution is not paused), so this + // function is about to return. + w.poll_state = PollState::Idle; + // Register the task waker that can be used to wake the parent + // task that will poll the inspector future. + if let Some(cx) = invoker_cx.take() { + w.task_waker.replace(cx.waker().clone()); + } + // Register the address of the inspector, which allows the waker + // to request an interrupt from the isolate. + w.inspector_ptr = NonNull::new(self as *const _ as *mut Self); + } + PollState::Polling if should_block => { + // Isolate execution has been paused but there are no more + // events to process, so this thread will be parked. Therefore, + // store the current thread handle in the waker so it knows + // which thread to unpark when new events arrive. + w.poll_state = PollState::Parked; + w.parked_thread.replace(thread::current()); + } + _ => unreachable!(), + }; + w.poll_state + }); + match new_state { + PollState::Idle => break Ok(Poll::Pending), // Yield to task. + PollState::Polling => {} // Poll the session handler again. + PollState::Parked => thread::park(), // Park the thread. + _ => unreachable!(), + }; + } + } + + /// This function blocks the thread until at least one inspector client has + /// established a websocket connection and successfully completed the + /// handshake. After that, it instructs V8 to pause at the next statement. + pub fn wait_for_session_and_break_on_next_statement(&mut self) { + loop { + match self.sessions.get_mut().established.iter_mut().next() { + Some(session) => break session.break_on_next_statement(), + None => { + self.flags.get_mut().waiting_for_session = true; + let _ = self.poll_sessions(None).unwrap(); + } + }; + } + } + + /// Obtain a sender for proxy channels. + /// + /// 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 { + self.new_session_tx.clone() + } + + /// Create a channel that notifies the frontend when inspector is dropped. + /// + /// NOTE: Only a single handler is currently available. + pub fn add_deregister_handler(&mut self) -> oneshot::Receiver<()> { + let (tx, rx) = oneshot::channel::<()>(); + let prev = self.deregister_tx.replace(tx); + assert!( + prev.is_none(), + "Only a single deregister handler is allowed" + ); + rx + } + + /// Create a local inspector session that can be used on + /// the same thread as the isolate. + pub fn create_local_session(&self) -> LocalInspectorSession { + // The 'outbound' channel carries messages sent to the session. + let (outbound_tx, outbound_rx) = mpsc::unbounded(); + + // The 'inbound' channel carries messages received from the session. + let (inbound_tx, inbound_rx) = mpsc::unbounded(); + + let proxy = InspectorSessionProxy { + tx: outbound_tx, + rx: inbound_rx, + }; + + // InspectorSessions for a local session is added directly to the "established" + // sessions, so it doesn't need to go through the session sender and handshake + // phase. + let inspector_session = + InspectorSession::new(self.v8_inspector.clone(), proxy); + self + .sessions + .borrow_mut() + .established + .push(inspector_session); + take(&mut self.flags.borrow_mut().waiting_for_session); + + LocalInspectorSession::new(inbound_tx, outbound_rx) + } +} + +#[derive(Default)] +struct InspectorFlags { + waiting_for_session: bool, + session_handshake_done: bool, + on_pause: bool, +} + +impl InspectorFlags { + fn new() -> RefCell { + let self_ = Self::default(); + RefCell::new(self_) + } +} + +/// A helper structure that helps coordinate sessions during different +/// parts of their lifecycle. +struct SessionContainer { + new_incoming: Pin> + 'static>>, + handshake: Option>, + established: FuturesUnordered>, +} + +impl SessionContainer { + fn new( + v8_inspector: Rc>>, + new_session_rx: UnboundedReceiver, + ) -> RefCell { + let new_incoming = new_session_rx + .map(move |session_proxy| { + InspectorSession::new(v8_inspector.clone(), session_proxy) + }) + .boxed_local(); + let self_ = Self { + new_incoming, + ..Default::default() + }; + RefCell::new(self_) + } +} + +impl Default for SessionContainer { + fn default() -> Self { + Self { + new_incoming: stream::empty().boxed_local(), + handshake: None, + established: FuturesUnordered::new(), + } + } +} + +struct InspectorWakerInner { + poll_state: PollState, + task_waker: Option, + parked_thread: Option, + inspector_ptr: Option>, + isolate_handle: v8::IsolateHandle, +} + +unsafe impl Send for InspectorWakerInner {} + +struct InspectorWaker(Mutex); + +impl InspectorWaker { + fn new(isolate_handle: v8::IsolateHandle) -> Arc { + let inner = InspectorWakerInner { + poll_state: PollState::Idle, + task_waker: None, + parked_thread: None, + inspector_ptr: None, + isolate_handle, + }; + Arc::new(Self(Mutex::new(inner))) + } + + fn update(&self, update_fn: F) -> R + where + F: FnOnce(&mut InspectorWakerInner) -> R, + { + let mut g = self.0.lock().unwrap(); + update_fn(&mut g) + } +} + +impl task::ArcWake for InspectorWaker { + fn wake_by_ref(arc_self: &Arc) { + arc_self.update(|w| { + match w.poll_state { + PollState::Idle => { + // Wake the task, if any, that has polled the Inspector future last. + if let Some(waker) = w.task_waker.take() { + waker.wake() + } + // Request an interrupt from the isolate if it's running and there's + // not unhandled interrupt request in flight. + if let Some(arg) = w + .inspector_ptr + .take() + .map(|ptr| ptr.as_ptr() as *mut c_void) + { + w.isolate_handle.request_interrupt(handle_interrupt, arg); + } + extern "C" fn handle_interrupt( + _isolate: &mut v8::Isolate, + arg: *mut c_void, + ) { + let inspector = unsafe { &*(arg as *mut JsRuntimeInspector) }; + let _ = inspector.poll_sessions(None); + } + } + PollState::Parked => { + // Unpark the isolate thread. + let parked_thread = w.parked_thread.take().unwrap(); + assert_ne!(parked_thread.id(), thread::current().id()); + parked_thread.unpark(); + } + _ => {} + }; + w.poll_state = PollState::Woken; + }); + } +} + +/// An inspector session that proxies messages to concrete "transport layer", +/// eg. Websocket or another set of channels. +struct InspectorSession { + v8_channel: v8::inspector::ChannelBase, + v8_session: Rc>>, + proxy_tx: SessionProxySender, + proxy_rx_handler: Pin + 'static>>, +} + +impl InspectorSession { + const CONTEXT_GROUP_ID: i32 = 1; + + pub fn new( + v8_inspector_rc: Rc>>, + session_proxy: InspectorSessionProxy, + ) -> Box { + new_box_with(move |self_ptr| { + let v8_channel = v8::inspector::ChannelBase::new::(); + let mut v8_inspector = v8_inspector_rc.borrow_mut(); + let v8_inspector_ptr = v8_inspector.as_mut().unwrap(); + let v8_session = Rc::new(RefCell::new(v8_inspector_ptr.connect( + Self::CONTEXT_GROUP_ID, + // Todo(piscisaureus): V8Inspector::connect() should require that + // the 'v8_channel' argument cannot move. + unsafe { &mut *self_ptr }, + v8::inspector::StringView::empty(), + ))); + + let (proxy_tx, proxy_rx) = session_proxy.split(); + let proxy_rx_handler = + Self::receive_from_proxy(v8_session.clone(), proxy_rx); + + Self { + v8_channel, + v8_session, + proxy_tx, + proxy_rx_handler, + } + }) + } + + // Dispatch message to V8 session + #[allow(unused)] + fn dispatch_message(&mut self, msg: Vec) { + let msg = v8::inspector::StringView::from(msg.as_slice()); + let mut v8_session = self.v8_session.borrow_mut(); + let v8_session_ptr = v8_session.as_mut(); + v8_session_ptr.dispatch_protocol_message(msg); + } + + // TODO(bartlomieju): this function should be reworked into `impl Future` + // or `impl Stream` + /// Returns a future that receives messages from the proxy and dispatches + /// them to the V8 session. + fn receive_from_proxy( + v8_session_rc: Rc< + RefCell>, + >, + proxy_rx: SessionProxyReceiver, + ) -> Pin + 'static>> { + async move { + let result = proxy_rx + .map_ok(move |msg| { + let msg = v8::inspector::StringView::from(msg.as_slice()); + let mut v8_session = v8_session_rc.borrow_mut(); + let v8_session_ptr = v8_session.as_mut(); + v8_session_ptr.dispatch_protocol_message(msg); + }) + .try_collect::<()>() + .await; + + // TODO(bartlomieju): ideally these prints should be moved + // to `server.rs` as they are unwanted in context of REPL/coverage collection + // but right now they do not pose a huge problem. Investigate how to + // move them to `server.rs`. + match result { + Ok(_) => eprintln!("Debugger session ended."), + Err(err) => eprintln!("Debugger session ended: {}.", err), + }; + } + .boxed_local() + } + + fn send_message( + &self, + maybe_call_id: Option, + msg: v8::UniquePtr, + ) { + let msg = msg.unwrap().string().to_string(); + let _ = self.proxy_tx.unbounded_send((maybe_call_id, msg)); + } + + pub fn break_on_next_statement(&mut self) { + let reason = v8::inspector::StringView::from(&b"debugCommand"[..]); + let detail = v8::inspector::StringView::empty(); + self + .v8_session + .borrow_mut() + .as_mut() + .schedule_pause_on_next_statement(reason, detail); + } +} + +impl v8::inspector::ChannelImpl for InspectorSession { + fn base(&self) -> &v8::inspector::ChannelBase { + &self.v8_channel + } + + fn base_mut(&mut self) -> &mut v8::inspector::ChannelBase { + &mut self.v8_channel + } + + fn send_response( + &mut self, + call_id: i32, + message: v8::UniquePtr, + ) { + self.send_message(Some(call_id), message); + } + + fn send_notification( + &mut self, + message: v8::UniquePtr, + ) { + self.send_message(None, message); + } + + fn flush_protocol_notifications(&mut self) {} +} + +impl Future for InspectorSession { + type Output = (); + fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll { + self.proxy_rx_handler.poll_unpin(cx) + } +} + +/// A local inspector session that can be used to send and receive protocol messages directly on +/// the same thread as an isolate. +pub struct LocalInspectorSession { + v8_session_tx: UnboundedSender, AnyError>>, + v8_session_rx: UnboundedReceiver<(Option, String)>, + response_tx_map: HashMap>, + next_message_id: i32, + notification_queue: Vec, +} + +impl LocalInspectorSession { + pub fn new( + v8_session_tx: UnboundedSender, AnyError>>, + v8_session_rx: UnboundedReceiver<(Option, String)>, + ) -> Self { + let response_tx_map = HashMap::new(); + let next_message_id = 0; + + let notification_queue = Vec::new(); + + Self { + v8_session_tx, + v8_session_rx, + response_tx_map, + next_message_id, + notification_queue, + } + } + + pub fn notifications(&mut self) -> Vec { + self.notification_queue.split_off(0) + } + + pub async fn post_message( + &mut self, + method: &str, + params: Option, + ) -> Result { + let id = self.next_message_id; + self.next_message_id += 1; + + let (response_tx, mut response_rx) = + oneshot::channel::(); + self.response_tx_map.insert(id, response_tx); + + let message = json!({ + "id": id, + "method": method, + "params": params, + }); + + let raw_message = serde_json::to_string(&message).unwrap(); + self + .v8_session_tx + .unbounded_send(Ok(raw_message.as_bytes().to_vec())) + .unwrap(); + + loop { + let receive_fut = self.receive_from_v8_session().boxed_local(); + match select(receive_fut, &mut response_rx).await { + Either::Left(_) => continue, + Either::Right((result, _)) => { + let response = result?; + if let Some(error) = response.get("error") { + return Err(generic_error(error.to_string())); + } + + let result = response.get("result").unwrap().clone(); + return Ok(result); + } + } + } + } + + async fn receive_from_v8_session(&mut self) { + let (maybe_call_id, message) = self.v8_session_rx.next().await.unwrap(); + // If there's no call_id then it's a notification + if let Some(call_id) = maybe_call_id { + let message: serde_json::Value = match serde_json::from_str(&message) { + Ok(v) => v, + Err(error) => match error.classify() { + serde_json::error::Category::Syntax => json!({ + "id": call_id, + "result": { + "result": { + "type": "error", + "description": "Unterminated string literal", + "value": "Unterminated string literal", + }, + "exceptionDetails": { + "exceptionId": 0, + "text": "Unterminated string literal", + "lineNumber": 0, + "columnNumber": 0 + }, + }, + }), + _ => panic!("Could not parse inspector message"), + }, + }; + + self + .response_tx_map + .remove(&call_id) + .unwrap() + .send(message) + .unwrap(); + } else { + let message = serde_json::from_str(&message).unwrap(); + self.notification_queue.push(message); + } + } +} + +fn new_box_with(new_fn: impl FnOnce(*mut T) -> T) -> Box { + let b = Box::new(MaybeUninit::::uninit()); + let p = Box::into_raw(b) as *mut T; + unsafe { ptr::write(p, new_fn(p)) }; + unsafe { Box::from_raw(p) } +} diff --git a/core/lib.rs b/core/lib.rs index 90b2cdfc4..7d49c1420 100644 --- a/core/lib.rs +++ b/core/lib.rs @@ -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; diff --git a/core/modules.rs b/core/modules.rs index a21286941..642dcc26a 100644 --- a/core/modules.rs +++ b/core/modules.rs @@ -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!( diff --git a/core/ops_json.rs b/core/ops_json.rs index bc04d9dd4..12fd9fdbc 100644 --- a/core/ops_json.rs +++ b/core/ops_json.rs @@ -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 (:")); diff --git a/core/runtime.rs b/core/runtime.rs index 0dbca0ae3..428add8e3 100644 --- a/core/runtime.rs +++ b/core/runtime.rs @@ -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 instead of just OwnedIsolate to workaround // an safety issue with SnapshotCreator. See JsRuntime::drop. v8_isolate: Option, + inspector: Option>, snapshot_creator: Option, 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>, + + /// 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> { + 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> { + // 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); } diff --git a/runtime/examples/hello_runtime.rs b/runtime/examples/hello_runtime.rs index e8abaffb8..34b249e0b 100644 --- a/runtime/examples/hello_runtime.rs +++ b/runtime/examples/hello_runtime.rs @@ -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(()) } diff --git a/runtime/inspector/mod.rs b/runtime/inspector/mod.rs deleted file mode 100644 index 08d388302..000000000 --- a/runtime/inspector/mod.rs +++ /dev/null @@ -1,751 +0,0 @@ -// Copyright 2018-2021 the Deno authors. All rights reserved. MIT license. - -//! The documentation for the inspector API is sparse, but these are helpful: -//! 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 std::cell::BorrowMutError; -use std::cell::RefCell; -use std::collections::HashMap; -use std::ffi::c_void; -use std::mem::replace; -use std::mem::take; -use std::mem::MaybeUninit; -use std::pin::Pin; -use std::ptr; -use std::ptr::NonNull; -use std::rc::Rc; -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, String)>; -// TODO(bartlomieju): does it even need to send a Result? -// It seems `Vec` would be enough -pub type SessionProxyReceiver = UnboundedReceiver, AnyError>>; - -/// Encapsulates an UnboundedSender/UnboundedReceiver pair that together form -/// a duplex channel for sending/receiving messages in V8 session. -pub struct SessionProxy { - pub tx: SessionProxySender, - pub rx: SessionProxyReceiver, -} - -impl SessionProxy { - pub fn split(self) -> (SessionProxySender, SessionProxyReceiver) { - (self.tx, self.rx) - } -} - -#[derive(Clone, Copy)] -enum PollState { - Idle, - Woken, - Polling, - Parked, - Dropped, -} - -/// This structure is used responsible for providing inspector interface -/// to the `JsRuntime`. -/// -/// It stores an instance of `v8::inspector::V8Inspector` and additionally -/// implements `v8::inspector::V8InspectorClientImpl`. -/// -/// After creating this structure it's possible to connect multiple sessions -/// to the inspector, in case of Deno it's either: a "websocket session" that -/// provides integration with Chrome Devtools, or an "in-memory session" that -/// is used for REPL or converage collection. -pub struct JsRuntimeInspector { - v8_inspector_client: v8::inspector::V8InspectorClientBase, - v8_inspector: Rc>>, - new_session_tx: UnboundedSender, - sessions: RefCell, - flags: RefCell, - waker: Arc, - deregister_tx: Option>, -} - -impl Drop for JsRuntimeInspector { - fn drop(&mut self) { - // Since the waker is cloneable, it might outlive the inspector itself. - // Set the poll state to 'dropped' so it doesn't attempt to request an - // interrupt from the isolate. - self.waker.update(|w| w.poll_state = PollState::Dropped); - // TODO(bartlomieju): this comment is out of date - // V8 automatically deletes all sessions when an `V8Inspector` instance is - // deleted, however InspectorSession also has a drop handler that cleans - // up after itself. To avoid a double free, make sure the inspector is - // dropped last. - take(&mut *self.sessions.borrow_mut()); - - // Notify counterparty that this instance is being destroyed. Ignoring - // result because counterparty waiting for the signal might have already - // dropped the other end of channel. - if let Some(deregister_tx) = self.deregister_tx.take() { - let _ = deregister_tx.send(()); - } - } -} - -impl v8::inspector::V8InspectorClientImpl for JsRuntimeInspector { - fn base(&self) -> &v8::inspector::V8InspectorClientBase { - &self.v8_inspector_client - } - - fn base_mut(&mut self) -> &mut v8::inspector::V8InspectorClientBase { - &mut self.v8_inspector_client - } - - fn run_message_loop_on_pause(&mut self, context_group_id: i32) { - assert_eq!(context_group_id, JsRuntimeInspector::CONTEXT_GROUP_ID); - self.flags.borrow_mut().on_pause = true; - let _ = self.poll_sessions(None); - } - - fn quit_message_loop_on_pause(&mut self) { - self.flags.borrow_mut().on_pause = false; - } - - fn run_if_waiting_for_debugger(&mut self, context_group_id: i32) { - assert_eq!(context_group_id, JsRuntimeInspector::CONTEXT_GROUP_ID); - self.flags.borrow_mut().session_handshake_done = true; - } -} - -/// `JsRuntimeInspector` implements a Future so that it can poll for new incoming -/// connections and messages from the WebSocket server. The Worker that owns -/// this `JsRuntimeInspector` will call this function from `Worker::poll()`. -impl Future for JsRuntimeInspector { - type Output = (); - fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<()> { - self.poll_sessions(Some(cx)).unwrap() - } -} - -impl JsRuntimeInspector { - /// Currently Deno supports only a single context in `JsRuntime` - /// 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 { - let context = js_runtime.global_context(); - let scope = &mut v8::HandleScope::new(js_runtime.v8_isolate()); - - let (new_session_tx, new_session_rx) = mpsc::unbounded::(); - - let v8_inspector_client = - v8::inspector::V8InspectorClientBase::new::(); - - let flags = InspectorFlags::new(); - let waker = InspectorWaker::new(scope.thread_safe_handle()); - - // Create JsRuntimeInspector instance. - let mut self_ = Box::new(Self { - v8_inspector_client, - v8_inspector: Default::default(), - sessions: Default::default(), - new_session_tx, - flags, - waker, - deregister_tx: None, - }); - self_.v8_inspector = Rc::new(RefCell::new( - v8::inspector::V8Inspector::create(scope, &mut *self_).into(), - )); - self_.sessions = - SessionContainer::new(self_.v8_inspector.clone(), new_session_rx); - - // Tell the inspector about the global context. - let context = v8::Local::new(scope, context); - let context_name = v8::inspector::StringView::from(&b"global context"[..]); - self_ - .v8_inspector - .borrow_mut() - .as_mut() - .unwrap() - .context_created(context, Self::CONTEXT_GROUP_ID, context_name); - - // Poll the session handler so we will get notified whenever there is - // new_incoming debugger activity. - let _ = self_.poll_sessions(None).unwrap(); - - self_ - } - - fn poll_sessions( - &self, - mut invoker_cx: Option<&mut Context>, - ) -> Result, BorrowMutError> { - // The futures this function uses do not have re-entrant poll() functions. - // However it is can happpen that poll_sessions() gets re-entered, e.g. - // when an interrupt request is honored while the inspector future is polled - // by the task executor. We let the caller know by returning some error. - let mut sessions = self.sessions.try_borrow_mut()?; - - self.waker.update(|w| { - match w.poll_state { - PollState::Idle | PollState::Woken => w.poll_state = PollState::Polling, - _ => unreachable!(), - }; - }); - - // Create a new Context object that will make downstream futures - // use the InspectorWaker when they are ready to be polled again. - let waker_ref = task::waker_ref(&self.waker); - let cx = &mut Context::from_waker(&waker_ref); - - loop { - loop { - // Do one "handshake" with a newly connected session at a time. - if let Some(session) = &mut sessions.handshake { - let poll_result = session.poll_unpin(cx); - let handshake_done = - replace(&mut self.flags.borrow_mut().session_handshake_done, false); - match poll_result { - Poll::Pending if handshake_done => { - let session = sessions.handshake.take().unwrap(); - sessions.established.push(session); - take(&mut self.flags.borrow_mut().waiting_for_session); - } - Poll::Ready(_) => sessions.handshake = None, - Poll::Pending => break, - }; - } - - // Accept new connections. - match sessions.new_incoming.poll_next_unpin(cx) { - Poll::Ready(Some(session)) => { - let prev = sessions.handshake.replace(session); - assert!(prev.is_none()); - continue; - } - Poll::Ready(None) => {} - Poll::Pending => {} - } - - // Poll established sessions. - match sessions.established.poll_next_unpin(cx) { - Poll::Ready(Some(_)) => continue, - Poll::Ready(None) => break, - Poll::Pending => break, - }; - } - - let should_block = sessions.handshake.is_some() - || self.flags.borrow().on_pause - || self.flags.borrow().waiting_for_session; - - let new_state = self.waker.update(|w| { - match w.poll_state { - PollState::Woken => { - // The inspector was woken while the session handler was being - // polled, so we poll it another time. - w.poll_state = PollState::Polling; - } - PollState::Polling if !should_block => { - // The session handler doesn't need to be polled any longer, and - // there's no reason to block (execution is not paused), so this - // function is about to return. - w.poll_state = PollState::Idle; - // Register the task waker that can be used to wake the parent - // task that will poll the inspector future. - if let Some(cx) = invoker_cx.take() { - w.task_waker.replace(cx.waker().clone()); - } - // Register the address of the inspector, which allows the waker - // to request an interrupt from the isolate. - w.inspector_ptr = NonNull::new(self as *const _ as *mut Self); - } - PollState::Polling if should_block => { - // Isolate execution has been paused but there are no more - // events to process, so this thread will be parked. Therefore, - // store the current thread handle in the waker so it knows - // which thread to unpark when new events arrive. - w.poll_state = PollState::Parked; - w.parked_thread.replace(thread::current()); - } - _ => unreachable!(), - }; - w.poll_state - }); - match new_state { - PollState::Idle => break Ok(Poll::Pending), // Yield to task. - PollState::Polling => {} // Poll the session handler again. - PollState::Parked => thread::park(), // Park the thread. - _ => unreachable!(), - }; - } - } - - /// This function blocks the thread until at least one inspector client has - /// established a websocket connection and successfully completed the - /// handshake. After that, it instructs V8 to pause at the next statement. - pub fn wait_for_session_and_break_on_next_statement(&mut self) { - loop { - match self.sessions.get_mut().established.iter_mut().next() { - Some(session) => break session.break_on_next_statement(), - None => { - self.flags.get_mut().waiting_for_session = true; - let _ = self.poll_sessions(None).unwrap(); - } - }; - } - } - - /// Obtain a sender for proxy channels. - /// - /// 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 { - self.new_session_tx.clone() - } - - /// Create a channel that notifies the frontend when inspector is dropped. - /// - /// NOTE: Only a single handler is currently available. - pub fn add_deregister_handler(&mut self) -> oneshot::Receiver<()> { - let (tx, rx) = oneshot::channel::<()>(); - let prev = self.deregister_tx.replace(tx); - assert!( - prev.is_none(), - "Only a single deregister handler is allowed" - ); - rx - } - - /// Create a local inspector session that can be used on - /// the same thread as the isolate. - pub fn create_local_session(&self) -> LocalInspectorSession { - // The 'outbound' channel carries messages sent to the session. - let (outbound_tx, outbound_rx) = mpsc::unbounded(); - - // The 'inbound' channel carries messages received from the session. - let (inbound_tx, inbound_rx) = mpsc::unbounded(); - - let proxy = SessionProxy { - tx: outbound_tx, - rx: inbound_rx, - }; - - // InspectorSessions for a local session is added directly to the "established" - // sessions, so it doesn't need to go through the session sender and handshake - // phase. - let inspector_session = - InspectorSession::new(self.v8_inspector.clone(), proxy); - self - .sessions - .borrow_mut() - .established - .push(inspector_session); - take(&mut self.flags.borrow_mut().waiting_for_session); - - LocalInspectorSession::new(inbound_tx, outbound_rx) - } -} - -#[derive(Default)] -struct InspectorFlags { - waiting_for_session: bool, - session_handshake_done: bool, - on_pause: bool, -} - -impl InspectorFlags { - fn new() -> RefCell { - let self_ = Self::default(); - RefCell::new(self_) - } -} - -/// A helper structure that helps coordinate sessions during different -/// parts of their lifecycle. -struct SessionContainer { - new_incoming: Pin> + 'static>>, - handshake: Option>, - established: FuturesUnordered>, -} - -impl SessionContainer { - fn new( - v8_inspector: Rc>>, - new_session_rx: UnboundedReceiver, - ) -> RefCell { - let new_incoming = new_session_rx - .map(move |session_proxy| { - InspectorSession::new(v8_inspector.clone(), session_proxy) - }) - .boxed_local(); - let self_ = Self { - new_incoming, - ..Default::default() - }; - RefCell::new(self_) - } -} - -impl Default for SessionContainer { - fn default() -> Self { - Self { - new_incoming: stream::empty().boxed_local(), - handshake: None, - established: FuturesUnordered::new(), - } - } -} - -struct InspectorWakerInner { - poll_state: PollState, - task_waker: Option, - parked_thread: Option, - inspector_ptr: Option>, - isolate_handle: v8::IsolateHandle, -} - -unsafe impl Send for InspectorWakerInner {} - -struct InspectorWaker(Mutex); - -impl InspectorWaker { - fn new(isolate_handle: v8::IsolateHandle) -> Arc { - let inner = InspectorWakerInner { - poll_state: PollState::Idle, - task_waker: None, - parked_thread: None, - inspector_ptr: None, - isolate_handle, - }; - Arc::new(Self(Mutex::new(inner))) - } - - fn update(&self, update_fn: F) -> R - where - F: FnOnce(&mut InspectorWakerInner) -> R, - { - let mut g = self.0.lock().unwrap(); - update_fn(&mut g) - } -} - -impl task::ArcWake for InspectorWaker { - fn wake_by_ref(arc_self: &Arc) { - arc_self.update(|w| { - match w.poll_state { - PollState::Idle => { - // Wake the task, if any, that has polled the Inspector future last. - if let Some(waker) = w.task_waker.take() { - waker.wake() - } - // Request an interrupt from the isolate if it's running and there's - // not unhandled interrupt request in flight. - if let Some(arg) = w - .inspector_ptr - .take() - .map(|ptr| ptr.as_ptr() as *mut c_void) - { - w.isolate_handle.request_interrupt(handle_interrupt, arg); - } - extern "C" fn handle_interrupt( - _isolate: &mut v8::Isolate, - arg: *mut c_void, - ) { - let inspector = unsafe { &*(arg as *mut JsRuntimeInspector) }; - let _ = inspector.poll_sessions(None); - } - } - PollState::Parked => { - // Unpark the isolate thread. - let parked_thread = w.parked_thread.take().unwrap(); - assert_ne!(parked_thread.id(), thread::current().id()); - parked_thread.unpark(); - } - _ => {} - }; - w.poll_state = PollState::Woken; - }); - } -} - -/// An inspector session that proxies messages to concrete "transport layer", -/// eg. Websocket or another set of channels. -struct InspectorSession { - v8_channel: v8::inspector::ChannelBase, - v8_session: Rc>>, - proxy_tx: SessionProxySender, - proxy_rx_handler: Pin + 'static>>, -} - -impl InspectorSession { - const CONTEXT_GROUP_ID: i32 = 1; - - pub fn new( - v8_inspector_rc: Rc>>, - session_proxy: SessionProxy, - ) -> Box { - new_box_with(move |self_ptr| { - let v8_channel = v8::inspector::ChannelBase::new::(); - let mut v8_inspector = v8_inspector_rc.borrow_mut(); - let v8_inspector_ptr = v8_inspector.as_mut().unwrap(); - let v8_session = Rc::new(RefCell::new(v8_inspector_ptr.connect( - Self::CONTEXT_GROUP_ID, - // Todo(piscisaureus): V8Inspector::connect() should require that - // the 'v8_channel' argument cannot move. - unsafe { &mut *self_ptr }, - v8::inspector::StringView::empty(), - ))); - - let (proxy_tx, proxy_rx) = session_proxy.split(); - let proxy_rx_handler = - Self::receive_from_proxy(v8_session.clone(), proxy_rx); - - Self { - v8_channel, - v8_session, - proxy_tx, - proxy_rx_handler, - } - }) - } - - // Dispatch message to V8 session - #[allow(unused)] - fn dispatch_message(&mut self, msg: Vec) { - let msg = v8::inspector::StringView::from(msg.as_slice()); - let mut v8_session = self.v8_session.borrow_mut(); - let v8_session_ptr = v8_session.as_mut(); - v8_session_ptr.dispatch_protocol_message(msg); - } - - // TODO(bartlomieju): this function should be reworked into `impl Future` - // or `impl Stream` - /// Returns a future that receives messages from the proxy and dispatches - /// them to the V8 session. - fn receive_from_proxy( - v8_session_rc: Rc< - RefCell>, - >, - proxy_rx: SessionProxyReceiver, - ) -> Pin + 'static>> { - async move { - let result = proxy_rx - .map_ok(move |msg| { - let msg = v8::inspector::StringView::from(msg.as_slice()); - let mut v8_session = v8_session_rc.borrow_mut(); - let v8_session_ptr = v8_session.as_mut(); - v8_session_ptr.dispatch_protocol_message(msg); - }) - .try_collect::<()>() - .await; - - // TODO(bartlomieju): ideally these prints should be moved - // to `server.rs` as they are unwanted in context of REPL/coverage collection - // but right now they do not pose a huge problem. Investigate how to - // move them to `server.rs`. - match result { - Ok(_) => eprintln!("Debugger session ended."), - Err(err) => eprintln!("Debugger session ended: {}.", err), - }; - } - .boxed_local() - } - - fn send_message( - &self, - maybe_call_id: Option, - msg: v8::UniquePtr, - ) { - let msg = msg.unwrap().string().to_string(); - let _ = self.proxy_tx.unbounded_send((maybe_call_id, msg)); - } - - pub fn break_on_next_statement(&mut self) { - let reason = v8::inspector::StringView::from(&b"debugCommand"[..]); - let detail = v8::inspector::StringView::empty(); - self - .v8_session - .borrow_mut() - .as_mut() - .schedule_pause_on_next_statement(reason, detail); - } -} - -impl v8::inspector::ChannelImpl for InspectorSession { - fn base(&self) -> &v8::inspector::ChannelBase { - &self.v8_channel - } - - fn base_mut(&mut self) -> &mut v8::inspector::ChannelBase { - &mut self.v8_channel - } - - fn send_response( - &mut self, - call_id: i32, - message: v8::UniquePtr, - ) { - self.send_message(Some(call_id), message); - } - - fn send_notification( - &mut self, - message: v8::UniquePtr, - ) { - self.send_message(None, message); - } - - fn flush_protocol_notifications(&mut self) {} -} - -impl Future for InspectorSession { - type Output = (); - fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll { - self.proxy_rx_handler.poll_unpin(cx) - } -} - -/// A local inspector session that can be used to send and receive protocol messages directly on -/// the same thread as an isolate. -pub struct LocalInspectorSession { - v8_session_tx: UnboundedSender, AnyError>>, - v8_session_rx: UnboundedReceiver<(Option, String)>, - response_tx_map: HashMap>, - next_message_id: i32, - notification_queue: Vec, -} - -impl LocalInspectorSession { - pub fn new( - v8_session_tx: UnboundedSender, AnyError>>, - v8_session_rx: UnboundedReceiver<(Option, String)>, - ) -> Self { - let response_tx_map = HashMap::new(); - let next_message_id = 0; - - let notification_queue = Vec::new(); - - Self { - v8_session_tx, - v8_session_rx, - response_tx_map, - next_message_id, - notification_queue, - } - } - - pub fn notifications(&mut self) -> Vec { - self.notification_queue.split_off(0) - } - - pub async fn post_message( - &mut self, - method: &str, - params: Option, - ) -> Result { - let id = self.next_message_id; - self.next_message_id += 1; - - let (response_tx, mut response_rx) = - oneshot::channel::(); - self.response_tx_map.insert(id, response_tx); - - let message = json!({ - "id": id, - "method": method, - "params": params, - }); - - let raw_message = serde_json::to_string(&message).unwrap(); - self - .v8_session_tx - .unbounded_send(Ok(raw_message.as_bytes().to_vec())) - .unwrap(); - - loop { - let receive_fut = self.receive_from_v8_session().boxed_local(); - match select(receive_fut, &mut response_rx).await { - Either::Left(_) => continue, - Either::Right((result, _)) => { - let response = result?; - if let Some(error) = response.get("error") { - return Err(generic_error(error.to_string())); - } - - let result = response.get("result").unwrap().clone(); - return Ok(result); - } - } - } - } - - async fn receive_from_v8_session(&mut self) { - let (maybe_call_id, message) = self.v8_session_rx.next().await.unwrap(); - // If there's no call_id then it's a notification - if let Some(call_id) = maybe_call_id { - let message: serde_json::Value = match serde_json::from_str(&message) { - Ok(v) => v, - Err(error) => match error.classify() { - serde_json::error::Category::Syntax => json!({ - "id": call_id, - "result": { - "result": { - "type": "error", - "description": "Unterminated string literal", - "value": "Unterminated string literal", - }, - "exceptionDetails": { - "exceptionId": 0, - "text": "Unterminated string literal", - "lineNumber": 0, - "columnNumber": 0 - }, - }, - }), - _ => panic!("Could not parse inspector message"), - }, - }; - - self - .response_tx_map - .remove(&call_id) - .unwrap() - .send(message) - .unwrap(); - } else { - let message = serde_json::from_str(&message).unwrap(); - self.notification_queue.push(message); - } - } -} - -fn new_box_with(new_fn: impl FnOnce(*mut T) -> T) -> Box { - let b = Box::new(MaybeUninit::::uninit()); - let p = Box::into_raw(b) as *mut T; - unsafe { ptr::write(p, new_fn(p)) }; - unsafe { Box::from_raw(p) } -} diff --git a/runtime/inspector/server.rs b/runtime/inspector/server.rs deleted file mode 100644 index 8cbf11d57..000000000 --- a/runtime/inspector/server.rs +++ /dev/null @@ -1,389 +0,0 @@ -// Copyright 2018-2021 the Deno authors. All rights reserved. MIT license. - -use core::convert::Infallible as Never; // Alias for the future `!` type. -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; -use deno_core::futures::future::Future; -use deno_core::futures::pin_mut; -use deno_core::futures::prelude::*; -use deno_core::futures::select; -use deno_core::futures::stream::StreamExt; -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_websocket::tokio_tungstenite::tungstenite; -use std::cell::RefCell; -use std::collections::HashMap; -use std::convert::Infallible; -use std::net::SocketAddr; -use std::process; -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 { - pub host: SocketAddr, - register_inspector_tx: UnboundedSender, - shutdown_server_tx: Option>, - thread_handle: Option>, -} - -impl InspectorServer { - pub fn new(host: SocketAddr, name: String) -> Self { - let (register_inspector_tx, register_inspector_rx) = - mpsc::unbounded::(); - - let (shutdown_server_tx, shutdown_server_rx) = oneshot::channel(); - - let thread_handle = thread::spawn(move || { - let rt = crate::tokio_util::create_basic_runtime(); - let local = tokio::task::LocalSet::new(); - local.block_on( - &rt, - server(host, register_inspector_rx, shutdown_server_rx, name), - ) - }); - - Self { - host, - register_inspector_tx, - shutdown_server_tx: Some(shutdown_server_tx), - thread_handle: Some(thread_handle), - } - } - - pub fn register_inspector( - &self, - session_sender: UnboundedSender, - deregister_rx: oneshot::Receiver<()>, - ) { - let info = InspectorInfo::new(self.host, session_sender, deregister_rx); - self.register_inspector_tx.unbounded_send(info).unwrap(); - } -} - -impl Drop for InspectorServer { - fn drop(&mut self) { - if let Some(shutdown_server_tx) = self.shutdown_server_tx.take() { - shutdown_server_tx - .send(()) - .expect("unable to send shutdown signal"); - } - - if let Some(thread_handle) = self.thread_handle.take() { - thread_handle.join().expect("unable to join thread"); - } - } -} - -// Needed so hyper can use non Send futures -#[derive(Clone)] -struct LocalExecutor; - -impl hyper::rt::Executor for LocalExecutor -where - Fut: Future + 'static, - Fut::Output: 'static, -{ - fn execute(&self, fut: Fut) { - tokio::task::spawn_local(fut); - } -} - -fn handle_ws_request( - req: http::Request, - inspector_map: Rc>>, -) -> http::Result> { - let (parts, body) = req.into_parts(); - let req = http::Request::from_parts(parts, ()); - - if let Some(new_session_tx) = req - .uri() - .path() - .strip_prefix("/ws/") - .and_then(|s| Uuid::parse_str(s).ok()) - .and_then(|uuid| { - inspector_map - .borrow() - .get(&uuid) - .map(|info| info.new_session_tx.clone()) - }) - { - let resp = tungstenite::handshake::server::create_response(&req) - .map(|resp| resp.map(|_| hyper::Body::empty())) - .or_else(|e| match e { - tungstenite::error::Error::HttpFormat(http_error) => Err(http_error), - _ => http::Response::builder() - .status(http::StatusCode::BAD_REQUEST) - .body("Not a valid Websocket Request".into()), - }); - - let (parts, _) = req.into_parts(); - let req = http::Request::from_parts(parts, body); - - if resp.is_ok() { - tokio::task::spawn_local(async move { - let upgraded = hyper::upgrade::on(req).await.unwrap(); - let websocket = - deno_websocket::tokio_tungstenite::WebSocketStream::from_raw_socket( - upgraded, - tungstenite::protocol::Role::Server, - None, - ) - .await; - let (proxy, pump) = create_websocket_proxy(websocket); - eprintln!("Debugger session started."); - let _ = new_session_tx.unbounded_send(proxy); - pump.await; - }); - } - resp - } else { - http::Response::builder() - .status(http::StatusCode::NOT_FOUND) - .body("No Valid inspector".into()) - } -} - -fn handle_json_request( - inspector_map: Rc>>, -) -> http::Result> { - let data = inspector_map - .borrow() - .values() - .map(|info| info.get_json_metadata()) - .collect::>(); - http::Response::builder() - .status(http::StatusCode::OK) - .header(http::header::CONTENT_TYPE, "application/json") - .body(serde_json::to_string(&data).unwrap().into()) -} - -fn handle_json_version_request( - version_response: Value, -) -> http::Result> { - http::Response::builder() - .status(http::StatusCode::OK) - .header(http::header::CONTENT_TYPE, "application/json") - .body(serde_json::to_string(&version_response).unwrap().into()) -} - -async fn server( - host: SocketAddr, - register_inspector_rx: UnboundedReceiver, - shutdown_server_rx: oneshot::Receiver<()>, - name: String, -) { - let inspector_map_ = - Rc::new(RefCell::new(HashMap::::new())); - - let inspector_map = Rc::clone(&inspector_map_); - let register_inspector_handler = register_inspector_rx - .map(|info| { - eprintln!( - "Debugger listening on {}", - info.get_websocket_debugger_url() - ); - if inspector_map.borrow_mut().insert(info.uuid, info).is_some() { - panic!("Inspector UUID already in map"); - } - }) - .collect::<()>(); - - let inspector_map = Rc::clone(&inspector_map_); - let deregister_inspector_handler = future::poll_fn(|cx| { - inspector_map - .borrow_mut() - .retain(|_, info| info.deregister_rx.poll_unpin(cx) == Poll::Pending); - Poll::::Pending - }) - .fuse(); - - let json_version_response = json!({ - "Browser": name, - "Protocol-Version": "1.3", - "V8-Version": deno_core::v8_version(), - }); - - let make_svc = hyper::service::make_service_fn(|_| { - let inspector_map = Rc::clone(&inspector_map_); - let json_version_response = json_version_response.clone(); - - future::ok::<_, Infallible>(hyper::service::service_fn( - move |req: http::Request| { - future::ready({ - match (req.method(), req.uri().path()) { - (&http::Method::GET, path) if path.starts_with("/ws/") => { - handle_ws_request(req, inspector_map.clone()) - } - (&http::Method::GET, "/json/version") => { - handle_json_version_request(json_version_response.clone()) - } - (&http::Method::GET, "/json") => { - handle_json_request(inspector_map.clone()) - } - (&http::Method::GET, "/json/list") => { - handle_json_request(inspector_map.clone()) - } - _ => http::Response::builder() - .status(http::StatusCode::NOT_FOUND) - .body("Not Found".into()), - } - }) - }, - )) - }); - - // Create the server manually so it can use the Local Executor - let server_handler = hyper::server::Builder::new( - hyper::server::conn::AddrIncoming::bind(&host).unwrap_or_else(|e| { - eprintln!("Cannot start inspector server: {}.", e); - process::exit(1); - }), - hyper::server::conn::Http::new().with_executor(LocalExecutor), - ) - .serve(make_svc) - .with_graceful_shutdown(async { - shutdown_server_rx.await.ok(); - }) - .unwrap_or_else(|err| { - eprintln!("Cannot start inspector server: {}.", err); - process::exit(1); - }) - .fuse(); - - pin_mut!(register_inspector_handler); - pin_mut!(deregister_inspector_handler); - pin_mut!(server_handler); - - select! { - _ = register_inspector_handler => {}, - _ = deregister_inspector_handler => unreachable!(), - _ = server_handler => {}, - } -} - -/// Creates a future that proxies messages sent and received on a warp WebSocket -/// to a UnboundedSender/UnboundedReceiver pair. We need this to sidestep -/// Tokio's task budget, which causes issues when JsRuntimeInspector::poll_sessions() -/// needs to block the thread because JavaScript execution is paused. -/// -/// This works because UnboundedSender/UnboundedReceiver are implemented in the -/// 'futures' crate, therefore they can't participate in Tokio's cooperative -/// task yielding. -/// -/// A tuple is returned, where the first element is a duplex channel that can -/// be used to send/receive messages on the websocket, and the second element -/// is a future that does the forwarding. -fn create_websocket_proxy( - websocket: deno_websocket::tokio_tungstenite::WebSocketStream< - hyper::upgrade::Upgraded, - >, -) -> (SessionProxy, impl Future + 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 { - tx: outbound_tx, - rx: inbound_rx, - }; - - // The pump future takes care of forwarding messages between the websocket - // and channels. It resolves to () when either side disconnects, ignoring any - // errors. - let pump = async move { - let (websocket_tx, websocket_rx) = websocket.split(); - - let outbound_pump = outbound_rx - .map(|(_maybe_call_id, msg)| tungstenite::Message::text(msg)) - .map(Ok) - .forward(websocket_tx) - .map_err(|_| ()); - - let inbound_pump = websocket_rx - .map(|result| { - let result = result.map(|msg| msg.into_data()).map_err(AnyError::from); - inbound_tx.unbounded_send(result) - }) - .map_err(|_| ()) - .try_collect::<()>(); - - let _ = future::try_join(outbound_pump, inbound_pump).await; - }; - - (proxy, pump) -} - -/// Inspector information that is sent from the isolate thread to the server -/// thread when a new inspector is created. -pub struct InspectorInfo { - pub host: SocketAddr, - pub uuid: Uuid, - pub thread_name: Option, - pub new_session_tx: UnboundedSender, - pub deregister_rx: oneshot::Receiver<()>, -} - -impl InspectorInfo { - pub fn new( - host: SocketAddr, - new_session_tx: mpsc::UnboundedSender, - deregister_rx: oneshot::Receiver<()>, - ) -> Self { - Self { - host, - uuid: Uuid::new_v4(), - thread_name: thread::current().name().map(|n| n.to_owned()), - new_session_tx, - deregister_rx, - } - } - - fn get_json_metadata(&self) -> Value { - json!({ - "description": "deno", - "devtoolsFrontendUrl": self.get_frontend_url(), - "faviconUrl": "https://deno.land/favicon.ico", - "id": self.uuid.to_string(), - "title": self.get_title(), - "type": "node", - // TODO(ry): "url": "file://", - "webSocketDebuggerUrl": self.get_websocket_debugger_url(), - }) - } - - pub fn get_websocket_debugger_url(&self) -> String { - format!("ws://{}/ws/{}", &self.host, &self.uuid) - } - - fn get_frontend_url(&self) -> String { - format!( - "devtools://devtools/bundled/js_app.html?ws={}/ws/{}&experiments=true&v8only=true", - &self.host, &self.uuid - ) - } - - fn get_title(&self) -> String { - format!( - "[{}] deno{}", - process::id(), - self - .thread_name - .as_ref() - .map(|n| format!(" - {}", n)) - .unwrap_or_default() - ) - } -} diff --git a/runtime/inspector_server.rs b/runtime/inspector_server.rs new file mode 100644 index 000000000..a205455fa --- /dev/null +++ b/runtime/inspector_server.rs @@ -0,0 +1,388 @@ +// Copyright 2018-2021 the Deno authors. All rights reserved. MIT license. + +use core::convert::Infallible as Never; // Alias for the future `!` type. +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; +use deno_core::futures::future::Future; +use deno_core::futures::pin_mut; +use deno_core::futures::prelude::*; +use deno_core::futures::select; +use deno_core::futures::stream::StreamExt; +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; +use std::convert::Infallible; +use std::net::SocketAddr; +use std::process; +use std::rc::Rc; +use std::thread; +use uuid::Uuid; + +/// Websocket server that is used to proxy connections from +/// devtools to the inspector. +pub struct InspectorServer { + pub host: SocketAddr, + register_inspector_tx: UnboundedSender, + shutdown_server_tx: Option>, + thread_handle: Option>, +} + +impl InspectorServer { + pub fn new(host: SocketAddr, name: String) -> Self { + let (register_inspector_tx, register_inspector_rx) = + mpsc::unbounded::(); + + let (shutdown_server_tx, shutdown_server_rx) = oneshot::channel(); + + let thread_handle = thread::spawn(move || { + let rt = crate::tokio_util::create_basic_runtime(); + let local = tokio::task::LocalSet::new(); + local.block_on( + &rt, + server(host, register_inspector_rx, shutdown_server_rx, name), + ) + }); + + Self { + host, + register_inspector_tx, + shutdown_server_tx: Some(shutdown_server_tx), + thread_handle: Some(thread_handle), + } + } + + pub fn register_inspector( + &self, + session_sender: UnboundedSender, + deregister_rx: oneshot::Receiver<()>, + ) { + let info = InspectorInfo::new(self.host, session_sender, deregister_rx); + self.register_inspector_tx.unbounded_send(info).unwrap(); + } +} + +impl Drop for InspectorServer { + fn drop(&mut self) { + if let Some(shutdown_server_tx) = self.shutdown_server_tx.take() { + shutdown_server_tx + .send(()) + .expect("unable to send shutdown signal"); + } + + if let Some(thread_handle) = self.thread_handle.take() { + thread_handle.join().expect("unable to join thread"); + } + } +} + +// Needed so hyper can use non Send futures +#[derive(Clone)] +struct LocalExecutor; + +impl hyper::rt::Executor for LocalExecutor +where + Fut: Future + 'static, + Fut::Output: 'static, +{ + fn execute(&self, fut: Fut) { + tokio::task::spawn_local(fut); + } +} + +fn handle_ws_request( + req: http::Request, + inspector_map: Rc>>, +) -> http::Result> { + let (parts, body) = req.into_parts(); + let req = http::Request::from_parts(parts, ()); + + if let Some(new_session_tx) = req + .uri() + .path() + .strip_prefix("/ws/") + .and_then(|s| Uuid::parse_str(s).ok()) + .and_then(|uuid| { + inspector_map + .borrow() + .get(&uuid) + .map(|info| info.new_session_tx.clone()) + }) + { + let resp = tungstenite::handshake::server::create_response(&req) + .map(|resp| resp.map(|_| hyper::Body::empty())) + .or_else(|e| match e { + tungstenite::error::Error::HttpFormat(http_error) => Err(http_error), + _ => http::Response::builder() + .status(http::StatusCode::BAD_REQUEST) + .body("Not a valid Websocket Request".into()), + }); + + let (parts, _) = req.into_parts(); + let req = http::Request::from_parts(parts, body); + + if resp.is_ok() { + tokio::task::spawn_local(async move { + let upgraded = hyper::upgrade::on(req).await.unwrap(); + let websocket = + deno_websocket::tokio_tungstenite::WebSocketStream::from_raw_socket( + upgraded, + tungstenite::protocol::Role::Server, + None, + ) + .await; + let (proxy, pump) = create_websocket_proxy(websocket); + eprintln!("Debugger session started."); + let _ = new_session_tx.unbounded_send(proxy); + pump.await; + }); + } + resp + } else { + http::Response::builder() + .status(http::StatusCode::NOT_FOUND) + .body("No Valid inspector".into()) + } +} + +fn handle_json_request( + inspector_map: Rc>>, +) -> http::Result> { + let data = inspector_map + .borrow() + .values() + .map(|info| info.get_json_metadata()) + .collect::>(); + http::Response::builder() + .status(http::StatusCode::OK) + .header(http::header::CONTENT_TYPE, "application/json") + .body(serde_json::to_string(&data).unwrap().into()) +} + +fn handle_json_version_request( + version_response: Value, +) -> http::Result> { + http::Response::builder() + .status(http::StatusCode::OK) + .header(http::header::CONTENT_TYPE, "application/json") + .body(serde_json::to_string(&version_response).unwrap().into()) +} + +async fn server( + host: SocketAddr, + register_inspector_rx: UnboundedReceiver, + shutdown_server_rx: oneshot::Receiver<()>, + name: String, +) { + let inspector_map_ = + Rc::new(RefCell::new(HashMap::::new())); + + let inspector_map = Rc::clone(&inspector_map_); + let register_inspector_handler = register_inspector_rx + .map(|info| { + eprintln!( + "Debugger listening on {}", + info.get_websocket_debugger_url() + ); + if inspector_map.borrow_mut().insert(info.uuid, info).is_some() { + panic!("Inspector UUID already in map"); + } + }) + .collect::<()>(); + + let inspector_map = Rc::clone(&inspector_map_); + let deregister_inspector_handler = future::poll_fn(|cx| { + inspector_map + .borrow_mut() + .retain(|_, info| info.deregister_rx.poll_unpin(cx) == Poll::Pending); + Poll::::Pending + }) + .fuse(); + + let json_version_response = json!({ + "Browser": name, + "Protocol-Version": "1.3", + "V8-Version": deno_core::v8_version(), + }); + + let make_svc = hyper::service::make_service_fn(|_| { + let inspector_map = Rc::clone(&inspector_map_); + let json_version_response = json_version_response.clone(); + + future::ok::<_, Infallible>(hyper::service::service_fn( + move |req: http::Request| { + future::ready({ + match (req.method(), req.uri().path()) { + (&http::Method::GET, path) if path.starts_with("/ws/") => { + handle_ws_request(req, inspector_map.clone()) + } + (&http::Method::GET, "/json/version") => { + handle_json_version_request(json_version_response.clone()) + } + (&http::Method::GET, "/json") => { + handle_json_request(inspector_map.clone()) + } + (&http::Method::GET, "/json/list") => { + handle_json_request(inspector_map.clone()) + } + _ => http::Response::builder() + .status(http::StatusCode::NOT_FOUND) + .body("Not Found".into()), + } + }) + }, + )) + }); + + // Create the server manually so it can use the Local Executor + let server_handler = hyper::server::Builder::new( + hyper::server::conn::AddrIncoming::bind(&host).unwrap_or_else(|e| { + eprintln!("Cannot start inspector server: {}.", e); + process::exit(1); + }), + hyper::server::conn::Http::new().with_executor(LocalExecutor), + ) + .serve(make_svc) + .with_graceful_shutdown(async { + shutdown_server_rx.await.ok(); + }) + .unwrap_or_else(|err| { + eprintln!("Cannot start inspector server: {}.", err); + process::exit(1); + }) + .fuse(); + + pin_mut!(register_inspector_handler); + pin_mut!(deregister_inspector_handler); + pin_mut!(server_handler); + + select! { + _ = register_inspector_handler => {}, + _ = deregister_inspector_handler => unreachable!(), + _ = server_handler => {}, + } +} + +/// Creates a future that proxies messages sent and received on a warp WebSocket +/// to a UnboundedSender/UnboundedReceiver pair. We need this to sidestep +/// Tokio's task budget, which causes issues when JsRuntimeInspector::poll_sessions() +/// needs to block the thread because JavaScript execution is paused. +/// +/// This works because UnboundedSender/UnboundedReceiver are implemented in the +/// 'futures' crate, therefore they can't participate in Tokio's cooperative +/// task yielding. +/// +/// A tuple is returned, where the first element is a duplex channel that can +/// be used to send/receive messages on the websocket, and the second element +/// is a future that does the forwarding. +fn create_websocket_proxy( + websocket: deno_websocket::tokio_tungstenite::WebSocketStream< + hyper::upgrade::Upgraded, + >, +) -> (InspectorSessionProxy, impl Future + 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 = InspectorSessionProxy { + tx: outbound_tx, + rx: inbound_rx, + }; + + // The pump future takes care of forwarding messages between the websocket + // and channels. It resolves to () when either side disconnects, ignoring any + // errors. + let pump = async move { + let (websocket_tx, websocket_rx) = websocket.split(); + + let outbound_pump = outbound_rx + .map(|(_maybe_call_id, msg)| tungstenite::Message::text(msg)) + .map(Ok) + .forward(websocket_tx) + .map_err(|_| ()); + + let inbound_pump = websocket_rx + .map(|result| { + let result = result.map(|msg| msg.into_data()).map_err(AnyError::from); + inbound_tx.unbounded_send(result) + }) + .map_err(|_| ()) + .try_collect::<()>(); + + let _ = future::try_join(outbound_pump, inbound_pump).await; + }; + + (proxy, pump) +} + +/// Inspector information that is sent from the isolate thread to the server +/// thread when a new inspector is created. +pub struct InspectorInfo { + pub host: SocketAddr, + pub uuid: Uuid, + pub thread_name: Option, + pub new_session_tx: UnboundedSender, + pub deregister_rx: oneshot::Receiver<()>, +} + +impl InspectorInfo { + pub fn new( + host: SocketAddr, + new_session_tx: mpsc::UnboundedSender, + deregister_rx: oneshot::Receiver<()>, + ) -> Self { + Self { + host, + uuid: Uuid::new_v4(), + thread_name: thread::current().name().map(|n| n.to_owned()), + new_session_tx, + deregister_rx, + } + } + + fn get_json_metadata(&self) -> Value { + json!({ + "description": "deno", + "devtoolsFrontendUrl": self.get_frontend_url(), + "faviconUrl": "https://deno.land/favicon.ico", + "id": self.uuid.to_string(), + "title": self.get_title(), + "type": "node", + // TODO(ry): "url": "file://", + "webSocketDebuggerUrl": self.get_websocket_debugger_url(), + }) + } + + pub fn get_websocket_debugger_url(&self) -> String { + format!("ws://{}/ws/{}", &self.host, &self.uuid) + } + + fn get_frontend_url(&self) -> String { + format!( + "devtools://devtools/bundled/js_app.html?ws={}/ws/{}&experiments=true&v8only=true", + &self.host, &self.uuid + ) + } + + fn get_title(&self) -> String { + format!( + "[{}] deno{}", + process::id(), + self + .thread_name + .as_ref() + .map(|n| format!(" - {}", n)) + .unwrap_or_default() + ) + } +} diff --git a/runtime/lib.rs b/runtime/lib.rs index 3cc73bfff..91bea8303 100644 --- a/runtime/lib.rs +++ b/runtime/lib.rs @@ -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; diff --git a/runtime/web_worker.rs b/runtime/web_worker.rs index 6bb26e555..d246843b4 100644 --- a/runtime/web_worker.rs +++ b/runtime/web_worker.rs @@ -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>, 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> { // 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()) }); diff --git a/runtime/worker.rs b/runtime/worker.rs index 1ffd32b48..57f4b532f 100644 --- a/runtime/worker.rs +++ b/runtime/worker.rs @@ -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>, 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> { - // 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); } } -- cgit v1.2.3