diff options
Diffstat (limited to 'runtime/inspector/mod.rs')
| -rw-r--r-- | runtime/inspector/mod.rs | 751 |
1 files changed, 751 insertions, 0 deletions
diff --git a/runtime/inspector/mod.rs b/runtime/inspector/mod.rs new file mode 100644 index 000000000..08d388302 --- /dev/null +++ b/runtime/inspector/mod.rs @@ -0,0 +1,751 @@ +// 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<i32>, String)>; +// TODO(bartlomieju): does it even need to send a Result? +// It seems `Vec<u8>` would be enough +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 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<RefCell<v8::UniquePtr<v8::inspector::V8Inspector>>>, + new_session_tx: UnboundedSender<SessionProxy>, + sessions: RefCell<SessionContainer>, + flags: RefCell<InspectorFlags>, + waker: Arc<InspectorWaker>, + deregister_tx: Option<oneshot::Sender<()>>, +} + +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<Self> { + 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::<SessionProxy>(); + + let v8_inspector_client = + v8::inspector::V8InspectorClientBase::new::<Self>(); + + 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<Poll<()>, 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<SessionProxy> { + 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<Self> { + 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<Box<dyn Stream<Item = Box<InspectorSession>> + 'static>>, + handshake: Option<Box<InspectorSession>>, + established: FuturesUnordered<Box<InspectorSession>>, +} + +impl SessionContainer { + fn new( + v8_inspector: Rc<RefCell<v8::UniquePtr<v8::inspector::V8Inspector>>>, + new_session_rx: UnboundedReceiver<SessionProxy>, + ) -> RefCell<Self> { + 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<task::Waker>, + parked_thread: Option<thread::Thread>, + inspector_ptr: Option<NonNull<JsRuntimeInspector>>, + isolate_handle: v8::IsolateHandle, +} + +unsafe impl Send for InspectorWakerInner {} + +struct InspectorWaker(Mutex<InspectorWakerInner>); + +impl InspectorWaker { + fn new(isolate_handle: v8::IsolateHandle) -> Arc<Self> { + 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<F, R>(&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<Self>) { + 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<RefCell<v8::UniqueRef<v8::inspector::V8InspectorSession>>>, + proxy_tx: SessionProxySender, + proxy_rx_handler: Pin<Box<dyn Future<Output = ()> + 'static>>, +} + +impl InspectorSession { + const CONTEXT_GROUP_ID: i32 = 1; + + pub fn new( + v8_inspector_rc: Rc<RefCell<v8::UniquePtr<v8::inspector::V8Inspector>>>, + session_proxy: SessionProxy, + ) -> Box<Self> { + new_box_with(move |self_ptr| { + let v8_channel = v8::inspector::ChannelBase::new::<Self>(); + 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<u8>) { + 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<v8::UniqueRef<v8::inspector::V8InspectorSession>>, + >, + proxy_rx: SessionProxyReceiver, + ) -> Pin<Box<dyn Future<Output = ()> + '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<i32>, + msg: v8::UniquePtr<v8::inspector::StringBuffer>, + ) { + 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<v8::inspector::StringBuffer>, + ) { + self.send_message(Some(call_id), message); + } + + fn send_notification( + &mut self, + message: v8::UniquePtr<v8::inspector::StringBuffer>, + ) { + 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::Output> { + 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<Result<Vec<u8>, AnyError>>, + v8_session_rx: UnboundedReceiver<(Option<i32>, String)>, + response_tx_map: HashMap<i32, oneshot::Sender<serde_json::Value>>, + next_message_id: i32, + notification_queue: Vec<Value>, +} + +impl LocalInspectorSession { + pub fn new( + v8_session_tx: UnboundedSender<Result<Vec<u8>, AnyError>>, + v8_session_rx: UnboundedReceiver<(Option<i32>, 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<Value> { + self.notification_queue.split_off(0) + } + + pub async fn post_message( + &mut self, + method: &str, + params: Option<serde_json::Value>, + ) -> Result<serde_json::Value, AnyError> { + let id = self.next_message_id; + self.next_message_id += 1; + + let (response_tx, mut response_rx) = + oneshot::channel::<serde_json::Value>(); + 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<T>(new_fn: impl FnOnce(*mut T) -> T) -> Box<T> { + let b = Box::new(MaybeUninit::<T>::uninit()); + let p = Box::into_raw(b) as *mut T; + unsafe { ptr::write(p, new_fn(p)) }; + unsafe { Box::from_raw(p) } +} |