diff --git a/Cargo.lock b/Cargo.lock index 4bf72cc54eda47..223d7873de631f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1930,9 +1930,9 @@ dependencies = [ [[package]] name = "deno_core" -version = "0.357.0" +version = "0.358.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "aa0cf9203866302d1321e45e3a15f2cd799fdcee469a9e64f314fdeac86ce357" +checksum = "12b2303b5bd690de06b04fcd1b544d06e6a73a3ed30569da2504890223d435d8" dependencies = [ "anyhow", "az", @@ -2648,9 +2648,9 @@ dependencies = [ [[package]] name = "deno_ops" -version = "0.233.0" +version = "0.234.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9c7689c66ee4172042c1da65119905f1d31948e32f75473861a9ff70e254603f" +checksum = "3f0d3a944fa9ee07021e2514aaf2754db6b30c294899730646b9b3bbae4a9583" dependencies = [ "indexmap 2.9.0", "proc-macro-rules", @@ -8252,9 +8252,9 @@ dependencies = [ [[package]] name = "serde_v8" -version = "0.266.0" +version = "0.267.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ca27ce1cc0f7bbe1d0274ce5efbfaf95cb8997503da85daf62076891b6d6e74e" +checksum = "cc490dcfb7c8ded6aa0946af8c2f5833ca8ea22c414b9392366c9a8d5e1283c4" dependencies = [ "deno_error", "num-bigint", diff --git a/Cargo.toml b/Cargo.toml index 925619ce9c45eb..f36bdd47417a56 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -62,7 +62,7 @@ repository = "https://github.com/denoland/deno" [workspace.dependencies] deno_ast = { version = "=0.50.0", features = ["transpiling"] } -deno_core = { version = "0.357.0" } +deno_core = { version = "0.358.0" } deno_cache_dir = "=0.25.0" deno_doc = "=0.183.0" diff --git a/cli/factory.rs b/cli/factory.rs index 052d9e4697dd33..00e0e303123b57 100644 --- a/cli/factory.rs +++ b/cli/factory.rs @@ -99,10 +99,10 @@ use crate::resolver::CliResolver; use crate::resolver::on_resolve_diagnostic; use crate::standalone::binary::DenoCompileBinaryWriter; use crate::sys::CliSys; -use crate::tools::coverage::CoverageCollector; +use crate::tools::coverage::CoverageCollectorState; use crate::tools::installer::BinNameResolver; use crate::tools::lint::LintRuleProvider; -use crate::tools::run::hmr::HmrRunner; +use crate::tools::run::hmr::HmrRunnerState; use crate::tsc::TypeCheckingCjsTracker; use crate::type_checker::TypeChecker; use crate::util::file_watcher::WatcherCommunicator; @@ -1106,8 +1106,8 @@ impl CliFactory { let create_hmr_runner = if cli_options.has_hmr() { let watcher_communicator = self.watcher_communicator.clone().unwrap(); let emitter = self.emitter()?.clone(); - let fn_: crate::worker::CreateHmrRunnerCb = Box::new(move |session| { - HmrRunner::new(emitter.clone(), session, watcher_communicator.clone()) + let fn_: crate::worker::CreateHmrRunnerCb = Box::new(move || { + HmrRunnerState::new(emitter.clone(), watcher_communicator.clone()) }); Some(fn_) } else { @@ -1116,9 +1116,7 @@ impl CliFactory { let create_coverage_collector = if let Some(coverage_dir) = cli_options.coverage_dir() { let fn_: crate::worker::CreateCoverageCollectorCb = - Box::new(move |session| { - CoverageCollector::new(coverage_dir.clone(), session) - }); + Box::new(move || CoverageCollectorState::new(coverage_dir.clone())); Some(fn_) } else { None diff --git a/cli/lib/worker.rs b/cli/lib/worker.rs index 5cef55cea108b8..3cc45a9e5d5e4d 100644 --- a/cli/lib/worker.rs +++ b/cli/lib/worker.rs @@ -816,8 +816,11 @@ impl LibMainWorker { } #[inline] - pub fn create_inspector_session(&mut self) -> LocalInspectorSession { - self.worker.create_inspector_session() + pub fn create_inspector_session( + &mut self, + cb: deno_core::InspectorSessionSend, + ) -> LocalInspectorSession { + self.worker.create_inspector_session(cb) } #[inline] diff --git a/cli/tools/coverage/mod.rs b/cli/tools/coverage/mod.rs index b858cc481e0112..3084a454a07408 100644 --- a/cli/tools/coverage/mod.rs +++ b/cli/tools/coverage/mod.rs @@ -7,6 +7,7 @@ use std::io::Write; use std::path::Path; use std::path::PathBuf; use std::sync::Arc; +use std::sync::atomic::AtomicI32; use deno_ast::MediaType; use deno_ast::ModuleKind; @@ -15,17 +16,15 @@ use deno_config::glob::FileCollector; use deno_config::glob::FilePatterns; use deno_config::glob::PathOrPattern; use deno_config::glob::PathOrPatternSet; -use deno_core::InspectorPostMessageError; -use deno_core::InspectorPostMessageErrorKind; use deno_core::LocalInspectorSession; use deno_core::anyhow::Context; use deno_core::anyhow::anyhow; use deno_core::error::AnyError; use deno_core::error::CoreError; +use deno_core::parking_lot::Mutex; use deno_core::serde_json; use deno_core::sourcemap::SourceMap; use deno_core::url::Url; -use deno_error::JsErrorBox; use deno_resolver::npm::DenoInNpmPackageChecker; use node_resolver::InNpmPackageChecker; use regex::Regex; @@ -55,36 +54,47 @@ pub mod reporter; mod util; use merge::ProcessCoverage; -pub struct CoverageCollector { - pub dir: PathBuf, - session: LocalInspectorSession, +static NEXT_MSG_ID: AtomicI32 = AtomicI32::new(0); + +fn next_msg_id() -> i32 { + NEXT_MSG_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed) } -impl CoverageCollector { - pub fn new(dir: PathBuf, session: LocalInspectorSession) -> Self { - Self { dir, session } - } +#[derive(Debug)] +pub struct CoverageCollectorInner { + dir: PathBuf, + coverage_msg_id: Option, +} - pub async fn start_collecting( - &mut self, - ) -> Result<(), InspectorPostMessageError> { - self.enable_debugger().await?; - self.enable_profiler().await?; - self - .start_precise_coverage(cdp::StartPreciseCoverageArgs { - call_count: true, - detailed: true, - allow_triggered_updates: false, - }) - .await?; +#[derive(Clone, Debug)] +pub struct CoverageCollectorState(Arc>); - Ok(()) +impl CoverageCollectorState { + pub fn new(dir: PathBuf) -> Self { + Self(Arc::new(Mutex::new(CoverageCollectorInner { + dir, + coverage_msg_id: None, + }))) } - pub async fn stop_collecting(&mut self) -> Result<(), CoreError> { - fs::create_dir_all(&self.dir)?; + pub fn callback(&self, msg: deno_core::InspectorMsg) { + let deno_core::InspectorMsgKind::Message(msg_id) = msg.kind else { + return; + }; + let maybe_coverage_msg_id = self.0.lock().coverage_msg_id.as_ref().cloned(); + + if let Some(coverage_msg_id) = maybe_coverage_msg_id + && coverage_msg_id == msg_id + { + let message: serde_json::Value = + serde_json::from_str(&msg.content).unwrap(); + let coverages: cdp::TakePreciseCoverageResponse = + serde_json::from_value(message["result"].clone()).unwrap(); + self.write_coverages(coverages.result); + } + } - let script_coverages = self.take_precise_coverage().await?.result; + fn write_coverages(&self, script_coverages: Vec) { for script_coverage in script_coverages { // Filter out internal and http/https JS files, eval'd scripts, // and scripts with invalid urls from being included in coverage reports @@ -100,92 +110,86 @@ impl CoverageCollector { } let filename = format!("{}.json", Uuid::new_v4()); - let filepath = self.dir.join(filename); - - let mut out = BufWriter::new(File::create(&filepath)?); - let coverage = serde_json::to_string(&script_coverage) - .map_err(JsErrorBox::from_err)?; + let filepath = self.0.lock().dir.join(filename); + + let file = match File::create(&filepath) { + Ok(f) => f, + Err(err) => { + log::error!( + "Failed to create coverage file at {:?}, reason: {:?}", + filepath, + err + ); + continue; + } + }; + let mut out = BufWriter::new(file); + let coverage = serde_json::to_string(&script_coverage).unwrap(); let formatted_coverage = format_json(&filepath, &coverage, &Default::default()) .ok() .flatten() .unwrap_or(coverage); - out.write_all(formatted_coverage.as_bytes())?; - out.flush()?; + if let Err(err) = out.write_all(formatted_coverage.as_bytes()) { + log::error!( + "Failed to write coverage file at {:?}, reason: {:?}", + filepath, + err + ); + continue; + } + if let Err(err) = out.flush() { + log::error!( + "Failed to flush coverage file at {:?}, reason: {:?}", + filepath, + err + ); + continue; + } } - - self.disable_debugger().await?; - self.disable_profiler().await?; - - Ok(()) - } - - async fn enable_debugger(&mut self) -> Result<(), InspectorPostMessageError> { - self - .session - .post_message::<()>("Debugger.enable", None) - .await?; - Ok(()) } +} - async fn enable_profiler(&mut self) -> Result<(), InspectorPostMessageError> { - self - .session - .post_message::<()>("Profiler.enable", None) - .await?; - Ok(()) - } +pub struct CoverageCollector { + pub state: CoverageCollectorState, + session: LocalInspectorSession, +} - async fn disable_debugger( - &mut self, - ) -> Result<(), InspectorPostMessageError> { - self - .session - .post_message::<()>("Debugger.disable", None) - .await?; - Ok(()) +impl CoverageCollector { + pub fn new( + state: CoverageCollectorState, + session: LocalInspectorSession, + ) -> Self { + Self { state, session } } - async fn disable_profiler( - &mut self, - ) -> Result<(), InspectorPostMessageError> { + pub fn start_collecting(&mut self) { self .session - .post_message::<()>("Profiler.disable", None) - .await?; - Ok(()) - } - - async fn start_precise_coverage( - &mut self, - parameters: cdp::StartPreciseCoverageArgs, - ) -> Result { - let return_value = self - .session - .post_message("Profiler.startPreciseCoverage", Some(parameters)) - .await?; - - let return_object = serde_json::from_value(return_value).map_err(|e| { - InspectorPostMessageErrorKind::JsBox(JsErrorBox::from_err(e)).into_box() - })?; - - Ok(return_object) + .post_message::<()>(next_msg_id(), "Profiler.enable", None); + self.session.post_message( + next_msg_id(), + "Profiler.startPreciseCoverage", + Some(cdp::StartPreciseCoverageArgs { + call_count: true, + detailed: true, + allow_triggered_updates: false, + }), + ); } - async fn take_precise_coverage( - &mut self, - ) -> Result { - let return_value = self - .session - .post_message::<()>("Profiler.takePreciseCoverage", None) - .await?; + pub fn stop_collecting(&mut self) -> Result<(), CoreError> { + fs::create_dir_all(&self.state.0.lock().dir)?; + let msg_id = next_msg_id(); + self.state.0.lock().coverage_msg_id.replace(msg_id); - let return_object = serde_json::from_value(return_value).map_err(|e| { - InspectorPostMessageErrorKind::JsBox(JsErrorBox::from_err(e)).into_box() - })?; - - Ok(return_object) + self.session.post_message::<()>( + msg_id, + "Profiler.takePreciseCoverage", + None, + ); + Ok(()) } } diff --git a/cli/tools/repl/session.rs b/cli/tools/repl/session.rs index 14a8c7decdb2d2..d692bbd5c8f15f 100644 --- a/cli/tools/repl/session.rs +++ b/cli/tools/repl/session.rs @@ -1,6 +1,8 @@ // Copyright 2018-2025 the Deno authors. MIT license. +use std::collections::HashMap; use std::sync::Arc; +use std::sync::atomic::AtomicI32; use deno_ast::ImportsNotUsedAsValues; use deno_ast::JsxAutomaticOptions; @@ -27,6 +29,9 @@ use deno_core::error::CoreError; use deno_core::futures::FutureExt; use deno_core::futures::StreamExt; use deno_core::futures::channel::mpsc::UnboundedReceiver; +use deno_core::futures::channel::mpsc::UnboundedSender; +use deno_core::futures::channel::mpsc::unbounded; +use deno_core::parking_lot::Mutex as SyncMutex; use deno_core::serde_json; use deno_core::serde_json::Value; use deno_core::unsync::spawn; @@ -45,6 +50,7 @@ use once_cell::sync::Lazy; use regex::Match; use regex::Regex; use tokio::sync::Mutex; +use tokio::sync::oneshot; use crate::args::CliOptions; use crate::cdp; @@ -172,8 +178,11 @@ pub struct ReplSession { internal_object_id: Option, npm_installer: Option>, resolver: Arc, - pub worker: MainWorker, + // NB: `session` and `state` must come before Worker, so that relevant V8 objects + // are dropped before the isolate is dropped with `worker`. session: LocalInspectorSession, + state: ReplSessionState, + pub worker: MainWorker, pub context_id: u64, pub language_server: ReplLanguageServer, pub notifications: Arc>>, @@ -186,6 +195,101 @@ pub struct ReplSession { decorators: deno_ast::DecoratorsTranspileOption, } +// TODO: duplicated in `cli/tools/run/hmr.rs` +#[derive(Debug)] +enum InspectorMessageState { + Ready(serde_json::Value), + WaitingFor(oneshot::Sender), +} + +#[derive(Debug)] +pub struct ReplSessionInner { + messages: HashMap, + notification_tx: UnboundedSender, +} + +#[derive(Clone, Debug)] +pub struct ReplSessionState(Arc>); + +impl ReplSessionState { + pub fn new(notification_tx: UnboundedSender) -> Self { + Self(Arc::new(SyncMutex::new(ReplSessionInner { + messages: HashMap::new(), + notification_tx, + }))) + } + + fn callback(&self, msg: deno_core::InspectorMsg) { + let deno_core::InspectorMsgKind::Message(msg_id) = msg.kind else { + if let Ok(value) = serde_json::from_str(&msg.content) { + let _ = self.0.lock().notification_tx.unbounded_send(value); + } + return; + }; + + let message: serde_json::Value = match serde_json::from_str(&msg.content) { + Ok(v) => v, + Err(error) => match error.classify() { + serde_json::error::Category::Syntax => serde_json::json!({ + "id": msg_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"), + }, + }; + + let mut state = self.0.lock(); + let Some(message_state) = state.messages.remove(&msg_id) else { + state + .messages + .insert(msg_id, InspectorMessageState::Ready(message)); + return; + }; + let InspectorMessageState::WaitingFor(sender) = message_state else { + return; + }; + let _ = sender.send(message); + } + + async fn wait_for_response( + &self, + msg_id: i32, + ) -> Result { + if let Some(message_state) = self.0.lock().messages.remove(&msg_id) { + let InspectorMessageState::Ready(mut value) = message_state else { + unreachable!(); + }; + return Ok(value["result"].take()); + } + + let (tx, rx) = oneshot::channel(); + self + .0 + .lock() + .messages + .insert(msg_id, InspectorMessageState::WaitingFor(tx)); + let mut value = rx.await.unwrap(); + Ok(value["result"].take()) + } +} + +static NEXT_MSG_ID: AtomicI32 = AtomicI32::new(0); +fn next_msg_id() -> i32 { + NEXT_MSG_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed) +} + impl ReplSession { #[allow(clippy::too_many_arguments)] pub async fn initialize( @@ -198,23 +302,20 @@ impl ReplSession { test_event_receiver: TestEventReceiver, ) -> Result { let language_server = ReplLanguageServer::new_initialized().await?; - let mut session = worker.create_inspector_session(); - worker - .js_runtime - .with_event_loop_future( - session - .post_message::<()>("Runtime.enable", None) - .boxed_local(), - PollEventLoopOptions::default(), - ) - .await?; + let (notification_tx, mut notification_rx) = unbounded(); + let repl_session_state = ReplSessionState::new(notification_tx); + let state = repl_session_state.clone(); + let callback = + Box::new(move |message| repl_session_state.callback(message)); + let mut session = worker.create_inspector_session(callback); + + session.post_message::<()>(next_msg_id(), "Runtime.enable", None); // Enabling the runtime domain will always send trigger one executionContextCreated for each // context the inspector knows about so we grab the execution context from that since // our inspector does not support a default context (0 is an invalid context id). let context_id: u64; - let mut notification_rx = session.take_notification_rx(); loop { let notification = notification_rx.next().await.unwrap(); @@ -260,6 +361,7 @@ impl ReplSession { resolver, worker, session, + state, context_id, language_server, referrer, @@ -316,11 +418,15 @@ impl ReplSession { method: &str, params: Option, ) -> Result { + let msg_id = next_msg_id(); + self.session.post_message(msg_id, method, params); + let fut = self.state.wait_for_response(msg_id).boxed_local(); + self .worker .js_runtime .with_event_loop_future( - self.session.post_message(method, params).boxed_local(), + fut, PollEventLoopOptions { // NOTE(bartlomieju): this is an important bit; we don't want to pump V8 // message loop here, so that GC won't run. Otherwise, the resulting diff --git a/cli/tools/run/hmr.rs b/cli/tools/run/hmr.rs index cc003d3cd88d82..346ae25e304d97 100644 --- a/cli/tools/run/hmr.rs +++ b/cli/tools/run/hmr.rs @@ -3,24 +3,34 @@ use std::collections::HashMap; use std::path::PathBuf; use std::sync::Arc; +use std::sync::atomic::AtomicI32; use deno_core::InspectorPostMessageError; use deno_core::InspectorPostMessageErrorKind; use deno_core::LocalInspectorSession; use deno_core::error::CoreError; -use deno_core::futures::StreamExt; +use deno_core::parking_lot::Mutex; use deno_core::serde_json::json; use deno_core::serde_json::{self}; use deno_core::url::Url; use deno_error::JsErrorBox; use deno_terminal::colors; use tokio::select; +use tokio::sync::mpsc; +use tokio::sync::mpsc::UnboundedReceiver; +use tokio::sync::mpsc::UnboundedSender; +use tokio::sync::oneshot; use crate::cdp; use crate::module_loader::CliEmitter; use crate::util::file_watcher::WatcherCommunicator; use crate::util::file_watcher::WatcherRestartMode; +static NEXT_MSG_ID: AtomicI32 = AtomicI32::new(0); +fn next_id() -> i32 { + NEXT_MSG_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed) +} + fn explain(response: &cdp::SetScriptSourceResponse) -> String { match response.status { cdp::Status::Ok => "OK".to_string(), @@ -62,6 +72,100 @@ fn should_retry(status: &cdp::Status) -> bool { } } +#[derive(Debug)] +enum InspectorMessageState { + Ready(serde_json::Value), + WaitingFor(oneshot::Sender), +} + +#[derive(Debug)] +pub struct HmrRunnerInner { + watcher_communicator: Arc, + script_ids: HashMap, + messages: HashMap, + emitter: Arc, + exception_tx: UnboundedSender, + exception_rx: Option>, +} + +#[derive(Clone, Debug)] +pub struct HmrRunnerState(Arc>); + +impl HmrRunnerState { + pub fn new( + emitter: Arc, + watcher_communicator: Arc, + ) -> Self { + let (exception_tx, exception_rx) = mpsc::unbounded_channel(); + + Self(Arc::new(Mutex::new(HmrRunnerInner { + emitter, + watcher_communicator, + script_ids: HashMap::new(), + messages: HashMap::new(), + exception_tx, + exception_rx: Some(exception_rx), + }))) + } + + pub fn callback(&self, msg: deno_core::InspectorMsg) { + let deno_core::InspectorMsgKind::Message(msg_id) = msg.kind else { + let notification = serde_json::from_str(&msg.content).unwrap(); + self.handle_notification(notification); + return; + }; + + let message: serde_json::Value = + serde_json::from_str(&msg.content).unwrap(); + let mut state = self.0.lock(); + let Some(message_state) = state.messages.remove(&msg_id) else { + state + .messages + .insert(msg_id, InspectorMessageState::Ready(message)); + return; + }; + let InspectorMessageState::WaitingFor(sender) = message_state else { + return; + }; + let _ = sender.send(message); + } + + fn handle_notification(&self, notification: cdp::Notification) { + if notification.method == "Runtime.exceptionThrown" { + let exception_thrown = + serde_json::from_value::(notification.params) + .unwrap(); + // .map_err(JsErrorBox::from_err)?; + let (message, description) = exception_thrown + .exception_details + .get_message_and_description(); + let _ = self + .0 + .lock() + .exception_tx + .send(JsErrorBox::generic(format!("{} {}", message, description))); + } else if notification.method == "Debugger.scriptParsed" { + let params = + serde_json::from_value::(notification.params) + .unwrap(); + // .map_err(JsErrorBox::from_err)?; + if params.url.starts_with("file://") { + let file_url = Url::parse(¶ms.url).unwrap(); + let file_path = file_url.to_file_path().unwrap(); + if let Ok(canonicalized_file_path) = file_path.canonicalize() { + let canonicalized_file_url = + Url::from_file_path(canonicalized_file_path).unwrap(); + self + .0 + .lock() + .script_ids + .insert(canonicalized_file_url.into(), params.script_id); + } + } + } + } +} + /// This structure is responsible for providing Hot Module Replacement /// functionality. /// @@ -79,69 +183,54 @@ fn should_retry(status: &cdp::Status) -> bool { /// force a full restart of a program by notifying the `FileWatcher`. pub struct HmrRunner { session: LocalInspectorSession, - watcher_communicator: Arc, - script_ids: HashMap, - emitter: Arc, + state: HmrRunnerState, } impl HmrRunner { - pub fn new( - emitter: Arc, - session: LocalInspectorSession, - watcher_communicator: Arc, - ) -> Self { - Self { - session, - emitter, - watcher_communicator, - script_ids: HashMap::new(), - } + pub fn new(state: HmrRunnerState, session: LocalInspectorSession) -> Self { + Self { session, state } + } + + pub fn start(&mut self) { + self + .session + .post_message::<()>(next_id(), "Debugger.enable", None); + self + .session + .post_message::<()>(next_id(), "Runtime.enable", None); } - // TODO(bartlomieju): this code is duplicated in `cli/tools/coverage/mod.rs` - pub async fn start(&mut self) -> Result<(), InspectorPostMessageError> { - self.enable_debugger().await + fn watcher(&self) -> Arc { + self.state.0.lock().watcher_communicator.clone() } - // TODO(bartlomieju): this code is duplicated in `cli/tools/coverage/mod.rs` - pub async fn stop(&mut self) -> Result<(), InspectorPostMessageError> { + pub fn stop(&mut self) { self - .watcher_communicator + .watcher() .change_restart_mode(WatcherRestartMode::Automatic); - self.disable_debugger().await } pub async fn run(&mut self) -> Result<(), CoreError> { self - .watcher_communicator + .watcher() .change_restart_mode(WatcherRestartMode::Manual); - let mut session_rx = self.session.take_notification_rx(); + let watcher = self.watcher(); + let mut exception_rx = self.state.0.lock().exception_rx.take().unwrap(); loop { select! { biased; - Some(notification) = session_rx.next() => { - let notification = serde_json::from_value::(notification).map_err(JsErrorBox::from_err)?; - if notification.method == "Runtime.exceptionThrown" { - let exception_thrown = serde_json::from_value::(notification.params).map_err(JsErrorBox::from_err)?; - let (message, description) = exception_thrown.exception_details.get_message_and_description(); - break Err(JsErrorBox::generic(format!("{} {}", message, description)).into()); - } else if notification.method == "Debugger.scriptParsed" { - let params = serde_json::from_value::(notification.params).map_err(JsErrorBox::from_err)?; - if params.url.starts_with("file://") { - let file_url = Url::parse(¶ms.url).unwrap(); - let file_path = file_url.to_file_path().unwrap(); - if let Ok(canonicalized_file_path) = file_path.canonicalize() { - let canonicalized_file_url = Url::from_file_path(canonicalized_file_path).unwrap(); - self.script_ids.insert(canonicalized_file_url.into(), params.script_id); - } - } + + maybe_error = exception_rx.recv() => { + if let Some(err) = maybe_error { + break Err(err.into()); } - } - changed_paths = self.watcher_communicator.watch_for_changed_paths() => { + }, + + changed_paths = watcher.watch_for_changed_paths() => { let changed_paths = changed_paths.map_err(JsErrorBox::from_err)?; let Some(changed_paths) = changed_paths else { - let _ = self.watcher_communicator.force_restart(); + let _ = self.watcher().force_restart(); continue; }; @@ -154,130 +243,111 @@ impl HmrRunner { // we can't HMR or an external file that was passed explicitly to // `--watch-hmr=` path. if filtered_paths.is_empty() { - let _ = self.watcher_communicator.force_restart(); + let _ = self.watcher().force_restart(); continue; } for path in filtered_paths { let Some(path_str) = path.to_str() else { - let _ = self.watcher_communicator.force_restart(); + let _ = self.watcher().force_restart(); continue; }; let Ok(module_url) = Url::from_file_path(path_str) else { - let _ = self.watcher_communicator.force_restart(); + let _ = self.watcher().force_restart(); continue; }; - let Some(id) = self.script_ids.get(module_url.as_str()).cloned() else { - let _ = self.watcher_communicator.force_restart(); + let Some(id) = self.state.0.lock().script_ids.get(module_url.as_str()).cloned() else { + let _ = self.watcher().force_restart(); continue; }; let source_code = tokio::fs::read_to_string(deno_path_util::url_to_file_path(&module_url).unwrap()).await?; - let source_code = self.emitter.emit_for_hmr( + let source_code = self.state.0.lock().emitter.emit_for_hmr( &module_url, source_code, )?; let mut tries = 1; loop { - let result = self.set_script_source(&id, source_code.as_str()).await?; + let msg_id = self.set_script_source(&id, source_code.as_str()); + let value = self.wait_for_response(msg_id).await?; + let result: cdp::SetScriptSourceResponse = serde_json::from_value(value).map_err(|e| { + InspectorPostMessageErrorKind::JsBox(JsErrorBox::from_err(e)).into_box() + })?; + if matches!(result.status, cdp::Status::Ok) { - self.dispatch_hmr_event(module_url.as_str()).await?; - self.watcher_communicator.print(format!("Replaced changed module {}", module_url.as_str())); + self.dispatch_hmr_event(module_url.as_str()); + self.watcher().print(format!("Replaced changed module {}", module_url.as_str())); break; } - self.watcher_communicator.print(format!("Failed to reload module {}: {}.", module_url, colors::gray(&explain(&result)))); + self.watcher().print(format!("Failed to reload module {}: {}.", module_url, colors::gray(&explain(&result)))); if should_retry(&result.status) && tries <= 2 { tries += 1; tokio::time::sleep(std::time::Duration::from_millis(100)).await; continue; } - let _ = self.watcher_communicator.force_restart(); + let _ = self.watcher().force_restart(); break; } } } - _ = self.session.receive_from_v8_session() => {} } } } - // TODO(bartlomieju): this code is duplicated in `cli/tools/coverage/mod.rs` - async fn enable_debugger(&mut self) -> Result<(), InspectorPostMessageError> { - self - .session - .post_message::<()>("Debugger.enable", None) - .await?; - self - .session - .post_message::<()>("Runtime.enable", None) - .await?; - Ok(()) - } + async fn wait_for_response( + &self, + msg_id: i32, + ) -> Result { + if let Some(message_state) = self.state.0.lock().messages.remove(&msg_id) { + let InspectorMessageState::Ready(mut value) = message_state else { + unreachable!(); + }; + return Ok(value["result"].take()); + } - // TODO(bartlomieju): this code is duplicated in `cli/tools/coverage/mod.rs` - async fn disable_debugger( - &mut self, - ) -> Result<(), InspectorPostMessageError> { - self - .session - .post_message::<()>("Debugger.disable", None) - .await?; + let (tx, rx) = oneshot::channel(); self - .session - .post_message::<()>("Runtime.disable", None) - .await?; - Ok(()) + .state + .0 + .lock() + .messages + .insert(msg_id, InspectorMessageState::WaitingFor(tx)); + let mut value = rx.await.unwrap(); + Ok(value["result"].take()) } - async fn set_script_source( - &mut self, - script_id: &str, - source: &str, - ) -> Result { - let result = self - .session - .post_message( - "Debugger.setScriptSource", - Some(json!({ - "scriptId": script_id, - "scriptSource": source, - "allowTopFrameEditing": true, - })), - ) - .await?; - - serde_json::from_value::(result).map_err( - |e| { - InspectorPostMessageErrorKind::JsBox(JsErrorBox::from_err(e)).into_box() - }, - ) + fn set_script_source(&mut self, script_id: &str, source: &str) -> i32 { + let msg_id = next_id(); + self.session.post_message( + msg_id, + "Debugger.setScriptSource", + Some(json!({ + "scriptId": script_id, + "scriptSource": source, + "allowTopFrameEditing": true, + })), + ); + msg_id } - async fn dispatch_hmr_event( - &mut self, - script_id: &str, - ) -> Result<(), InspectorPostMessageError> { + fn dispatch_hmr_event(&mut self, script_id: &str) { let expr = format!( "dispatchEvent(new CustomEvent(\"hmr\", {{ detail: {{ path: \"{}\" }} }}));", script_id ); - let _result = self - .session - .post_message( - "Runtime.evaluate", - Some(json!({ - "expression": expr, - "contextId": Some(1), - })), - ) - .await?; - - Ok(()) + self.session.post_message( + next_id(), + "Runtime.evaluate", + Some(json!({ + "expression": expr, + "contextId": Some(1), + })), + ); } } diff --git a/cli/tools/test/mod.rs b/cli/tools/test/mod.rs index 59cf04e1daa920..bd4e1a99c72d44 100644 --- a/cli/tools/test/mod.rs +++ b/cli/tools/test/mod.rs @@ -36,7 +36,6 @@ use deno_core::error::AnyError; use deno_core::error::CoreError; use deno_core::error::CoreErrorKind; use deno_core::error::JsError; -use deno_core::futures::FutureExt; use deno_core::futures::StreamExt; use deno_core::futures::future; use deno_core::futures::stream; @@ -668,7 +667,7 @@ async fn configure_main_worker( None, ) .await?; - let coverage_collector = worker.maybe_setup_coverage_collector().await?; + let coverage_collector = worker.maybe_setup_coverage_collector(); if options.trace_leaks { worker .execute_script_static( @@ -804,13 +803,7 @@ async fn test_specifier_inner( worker.run_up_to_duration(Duration::from_millis(0)).await?; if let Some(coverage_collector) = &mut coverage_collector { - worker - .js_runtime - .with_event_loop_future( - coverage_collector.stop_collecting().boxed_local(), - PollEventLoopOptions::default(), - ) - .await?; + coverage_collector.stop_collecting()?; } Ok(()) } diff --git a/cli/worker.rs b/cli/worker.rs index e467b4c1aba7ea..52167b01250aef 100644 --- a/cli/worker.rs +++ b/cli/worker.rs @@ -7,7 +7,6 @@ use std::sync::Arc; use deno_ast::ModuleSpecifier; use deno_core::Extension; use deno_core::OpState; -use deno_core::PollEventLoopOptions; use deno_core::error::CoreError; use deno_core::error::JsError; use deno_core::futures::FutureExt; @@ -30,17 +29,17 @@ use crate::npm::CliNpmInstaller; use crate::npm::CliNpmResolver; use crate::sys::CliSys; use crate::tools::coverage::CoverageCollector; +use crate::tools::coverage::CoverageCollectorState; use crate::tools::run::hmr::HmrRunner; +use crate::tools::run::hmr::HmrRunnerState; use crate::util::file_watcher::WatcherCommunicator; use crate::util::file_watcher::WatcherRestartMode; use crate::util::progress_bar::ProgressBar; -pub type CreateHmrRunnerCb = - Box HmrRunner + Send + Sync>; +pub type CreateHmrRunnerCb = Box HmrRunnerState + Send + Sync>; -pub type CreateCoverageCollectorCb = Box< - dyn Fn(deno_core::LocalInspectorSession) -> CoverageCollector + Send + Sync, ->; +pub type CreateCoverageCollectorCb = + Box CoverageCollectorState + Send + Sync>; pub struct CliMainWorkerOptions { pub create_hmr_runner: Option, @@ -73,9 +72,8 @@ impl CliMainWorker { } pub async fn run(&mut self) -> Result { - let mut maybe_coverage_collector = - self.maybe_setup_coverage_collector().await?; - let mut maybe_hmr_runner = self.maybe_setup_hmr_runner().await?; + let mut maybe_coverage_collector = self.maybe_setup_coverage_collector(); + let mut maybe_hmr_runner = self.maybe_setup_hmr_runner(); // WARNING: Remember to update cli/lib/worker.rs to align with // changes made here so that they affect deno_compile as well. @@ -111,6 +109,7 @@ impl CliMainWorker { return Err(e); } } else { + // TODO(bartlomieju): this might not be needed anymore self .worker .run_event_loop(maybe_coverage_collector.is_none()) @@ -130,24 +129,10 @@ impl CliMainWorker { self.worker.dispatch_process_exit_event()?; if let Some(coverage_collector) = maybe_coverage_collector.as_mut() { - self - .worker - .js_runtime() - .with_event_loop_future( - coverage_collector.stop_collecting().boxed_local(), - PollEventLoopOptions::default(), - ) - .await?; + coverage_collector.stop_collecting()?; } if let Some(hmr_runner) = maybe_hmr_runner.as_mut() { - self - .worker - .js_runtime() - .with_event_loop_future( - hmr_runner.stop().boxed_local(), - PollEventLoopOptions::default(), - ) - .await?; + hmr_runner.stop(); } Ok(self.worker.exit_code()) @@ -233,48 +218,36 @@ impl CliMainWorker { self.worker.js_runtime().op_state() } - pub async fn maybe_setup_hmr_runner( - &mut self, - ) -> Result, CoreError> { - let Some(setup_hmr_runner) = self.shared.create_hmr_runner.as_ref() else { - return Ok(None); - }; + pub fn maybe_setup_hmr_runner(&mut self) -> Option { + let setup_hmr_runner = self.shared.create_hmr_runner.as_ref()?; - let session = self.worker.create_inspector_session(); + let hmr_runner_state = setup_hmr_runner(); + let state = hmr_runner_state.clone(); - let mut hmr_runner = setup_hmr_runner(session); + let callback = Box::new(move |message| hmr_runner_state.callback(message)); + let session = self.worker.create_inspector_session(callback); + let mut hmr_runner = HmrRunner::new(state, session); + hmr_runner.start(); - self - .worker - .js_runtime() - .with_event_loop_future( - hmr_runner.start().boxed_local(), - PollEventLoopOptions::default(), - ) - .await?; - Ok(Some(hmr_runner)) + Some(hmr_runner) } - pub async fn maybe_setup_coverage_collector( + pub fn maybe_setup_coverage_collector( &mut self, - ) -> Result, CoreError> { - let Some(create_coverage_collector) = - self.shared.create_coverage_collector.as_ref() - else { - return Ok(None); - }; + ) -> Option { + let create_coverage_collector = + self.shared.create_coverage_collector.as_ref()?; - let session = self.worker.create_inspector_session(); - let mut coverage_collector = create_coverage_collector(session); - self - .worker - .js_runtime() - .with_event_loop_future( - coverage_collector.start_collecting().boxed_local(), - PollEventLoopOptions::default(), - ) - .await?; - Ok(Some(coverage_collector)) + let coverage_collector_state = create_coverage_collector(); + let state = coverage_collector_state.clone(); + + let callback = + Box::new(move |message| coverage_collector_state.callback(message)); + let session = self.worker.create_inspector_session(callback); + let mut coverage_collector = CoverageCollector::new(state, session); + coverage_collector.start_collecting(); + + Some(coverage_collector) } pub fn execute_script_static( diff --git a/ext/node/ops/inspector.rs b/ext/node/ops/inspector.rs index 180ae1e36ca6ac..3937444b1990ca 100644 --- a/ext/node/ops/inspector.rs +++ b/ext/node/ops/inspector.rs @@ -4,11 +4,11 @@ use std::cell::RefCell; use std::rc::Rc; use deno_core::GarbageCollected; +use deno_core::InspectorMsg; use deno_core::InspectorSessionKind; use deno_core::InspectorSessionOptions; use deno_core::JsRuntimeInspector; use deno_core::OpState; -use deno_core::futures::channel::mpsc; use deno_core::op2; use deno_core::v8; use deno_error::JsErrorBox; @@ -81,7 +81,7 @@ pub fn op_inspector_emit_protocol_event( } struct JSInspectorSession { - tx: RefCell>>, + session: RefCell>, } // SAFETY: we're sure this can be GCed @@ -131,39 +131,40 @@ where let context = v8::Global::new(scope, context); let callback = v8::Global::new(scope, callback); - let inspector = state - .borrow::>>() - .borrow_mut(); + let inspector = state.borrow::>>().clone(); + + // The inspector connection does not keep the event loop alive but + // when the inspector sends a message to the frontend, the JS that + // that runs may keep the event loop alive so we have to call back + // synchronously, instead of using the usual LocalInspectorSession + // UnboundedReceiver API. + let callback = Box::new(move |message: InspectorMsg| { + // SAFETY: This function is called directly by the inspector, so + // 1) The isolate is still valid + // 2) We are on the same thread as the Isolate + let scope = unsafe { &mut v8::CallbackScope::new(&mut *isolate) }; + let context = v8::Local::new(scope, context.clone()); + let scope = &mut v8::ContextScope::new(scope, context); + let scope = &mut v8::TryCatch::new(scope); + let recv = v8::undefined(scope); + if let Some(message) = v8::String::new(scope, &message.content) { + let callback = v8::Local::new(scope, callback.clone()); + callback.call(scope, recv.into(), &[message.into()]); + } + }); - let tx = inspector.create_raw_session( + let session = JsRuntimeInspector::create_local_session( + inspector, + callback, InspectorSessionOptions { kind: InspectorSessionKind::NonBlocking { wait_for_disconnect: false, }, }, - // The inspector connection does not keep the event loop alive but - // when the inspector sends a message to the frontend, the JS that - // that runs may keep the event loop alive so we have to call back - // synchronously, instead of using the usual LocalInspectorSession - // UnboundedReceiver API. - Box::new(move |message| { - // SAFETY: This function is called directly by the inspector, so - // 1) The isolate is still valid - // 2) We are on the same thread as the Isolate - let scope = unsafe { &mut v8::CallbackScope::new(&mut *isolate) }; - let context = v8::Local::new(scope, context.clone()); - let scope = &mut v8::ContextScope::new(scope, context); - let scope = &mut v8::TryCatch::new(scope); - let recv = v8::undefined(scope); - if let Some(message) = v8::String::new(scope, &message.content) { - let callback = v8::Local::new(scope, callback.clone()); - callback.call(scope, recv.into(), &[message.into()]); - } - }), ); Ok(JSInspectorSession { - tx: RefCell::new(Some(tx)), + session: RefCell::new(Some(session)), }) } @@ -172,12 +173,12 @@ pub fn op_inspector_dispatch( #[cppgc] session: &JSInspectorSession, #[string] message: String, ) { - if let Some(tx) = &*session.tx.borrow() { - let _ = tx.unbounded_send(message); + if let Some(session) = &mut *session.session.borrow_mut() { + session.dispatch(message); } } #[op2(fast)] pub fn op_inspector_disconnect(#[cppgc] session: &JSInspectorSession) { - drop(session.tx.borrow_mut().take()); + drop(session.session.borrow_mut().take()); } diff --git a/runtime/worker.rs b/runtime/worker.rs index 69734e5ef52859..c7aebef92f2281 100644 --- a/runtime/worker.rs +++ b/runtime/worker.rs @@ -19,6 +19,7 @@ use deno_core::Extension; use deno_core::InspectorSessionKind; use deno_core::InspectorSessionOptions; use deno_core::JsRuntime; +use deno_core::JsRuntimeInspector; use deno_core::LocalInspectorSession; use deno_core::ModuleCodeString; use deno_core::ModuleId; @@ -908,9 +909,16 @@ impl MainWorker { /// Create new inspector session. This function panics if Worker /// was not configured to create inspector. - pub fn create_inspector_session(&mut self) -> LocalInspectorSession { + pub fn create_inspector_session( + &mut self, + cb: deno_core::InspectorSessionSend, + ) -> LocalInspectorSession { self.js_runtime.maybe_init_inspector(); - self.js_runtime.inspector().borrow().create_local_session( + let insp = self.js_runtime.inspector(); + + JsRuntimeInspector::create_local_session( + insp, + cb, InspectorSessionOptions { kind: InspectorSessionKind::Blocking, }, diff --git a/tests/integration/coverage_tests.rs b/tests/integration/coverage_tests.rs index e0eaede7107cf0..c4ae75eb225a44 100644 --- a/tests/integration/coverage_tests.rs +++ b/tests/integration/coverage_tests.rs @@ -173,6 +173,7 @@ fn multifile_coverage() { let tempdir = context.temp_dir(); let tempdir = tempdir.path().join("cov"); + eprintln!("before test"); let output = context .new_command() .args_vec(vec![ @@ -182,8 +183,9 @@ fn multifile_coverage() { format!("coverage/multifile/"), ]) .run(); - + eprintln!("after test"); output.assert_exit_code(0); + eprintln!("output {:#?}", output.print_output()); output.skip_output_check(); let output = context @@ -197,6 +199,7 @@ fn multifile_coverage() { .run(); // Verify there's no "Check" being printed + eprintln!("output2 {:#?}", output.print_output()); assert!(output.stderr().is_empty()); output.assert_stdout_matches_file( diff --git a/tests/integration/inspector_tests.rs b/tests/integration/inspector_tests.rs index 18f5cbd54d0714..0c5c70f7d1855f 100644 --- a/tests/integration/inspector_tests.rs +++ b/tests/integration/inspector_tests.rs @@ -603,6 +603,7 @@ async fn inspector_runtime_evaluate_does_not_crash() { .arg("repl") .arg("--allow-read") .arg(inspect_flag_with_unique_port("--inspect")) + .env("RUST_BACKTRACE", "1") .stdin(std::process::Stdio::piped()) .piped_output() .spawn() @@ -1025,6 +1026,7 @@ async fn inspector_memory() { .arg("run") .arg(inspect_flag_with_unique_port("--inspect-brk")) .arg(script) + .env("RUST_BACKTRACE", "1") .piped_output() .spawn() .unwrap();