refactor(cli/repl): tightly integrate event loop (#7834)

cli/main.rs

@@ -63,7 +63,6 @@ use crate::file_fetcher::SourceFileFetcher;
 use crate::file_fetcher::TextDocument;
 use crate::fs as deno_fs;
 use crate::global_state::GlobalState;
-use crate::inspector::InspectorSession;
 use crate::media_type::MediaType;
 use crate::permissions::Permissions;
 use crate::worker::MainWorker;
@@ -432,26 +431,7 @@ async fn run_repl(flags: Flags) -> Result<(), AnyError> {
   let mut worker = MainWorker::new(&global_state, main_module.clone());
   (&mut *worker).await?;
 
-  let inspector = worker
+  repl::run(&global_state, worker).await
-    .inspector
-    .as_mut()
-    .expect("Inspector is not created.");
-
-  let inspector_session = InspectorSession::new(&mut **inspector);
-  let repl = repl::run(&global_state, inspector_session);
-
-  tokio::pin!(repl);
-
-  loop {
-    tokio::select! {
-      result = &mut repl => {
-          return result;
-      }
-      _ = &mut *worker => {
-          tokio::time::delay_for(tokio::time::Duration::from_millis(10)).await;
-      }
-    }
-  }
 }
 
 async fn run_from_stdin(flags: Flags) -> Result<(), AnyError> {

cli/repl.rs

@@ -2,8 +2,11 @@
 
 use crate::global_state::GlobalState;
 use crate::inspector::InspectorSession;
+use crate::worker::MainWorker;
+use crate::worker::Worker;
 use deno_core::error::AnyError;
 use deno_core::serde_json::json;
+use deno_core::serde_json::Value;
 use rustyline::error::ReadlineError;
 use rustyline::validate::MatchingBracketValidator;
 use rustyline::validate::ValidationContext;
@@ -29,17 +32,76 @@ impl Validator for Helper {
   }
 }
 
+async fn post_message_and_poll(
+  worker: &mut Worker,
+  session: &mut InspectorSession,
+  method: &str,
+  params: Option<Value>,
+) -> Result<Value, AnyError> {
+  let response = session.post_message(method.to_string(), params);
+  tokio::pin!(response);
+
+  loop {
+    tokio::select! {
+      result = &mut response => {
+        return result
+      }
+
+      _ = &mut *worker => {
+        // 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::delay_for(tokio::time::Duration::from_millis(0)).await;
+      }
+    }
+  }
+}
+
+async fn read_line_and_poll(
+  worker: &mut Worker,
+  editor: Arc<Mutex<Editor<Helper>>>,
+) -> Result<String, ReadlineError> {
+  let mut line =
+    tokio::task::spawn_blocking(move || editor.lock().unwrap().readline("> "));
+
+  let mut poll_worker = true;
+  loop {
+    // Because an inspector websocket client may choose to connect at anytime when we have an
+    // inspector server we need to keep polling the worker to pick up new connections.
+    let mut timeout =
+      tokio::time::delay_for(tokio::time::Duration::from_millis(1000));
+
+    tokio::select! {
+      result = &mut line => {
+        return result.unwrap();
+      }
+      _ = &mut *worker, if poll_worker => {
+        poll_worker = false;
+      }
+      _ = &mut timeout => {
+          poll_worker = true
+      }
+    }
+  }
+}
+
 pub async fn run(
   global_state: &GlobalState,
-  mut session: Box<InspectorSession>,
+  mut worker: MainWorker,
 ) -> Result<(), AnyError> {
   // Our inspector is unable to default to the default context id so we have to specify it here.
   let context_id: u32 = 1;
 
+  let inspector = worker
+    .inspector
+    .as_mut()
+    .expect("Inspector is not created.");
+
+  let mut session = InspectorSession::new(&mut **inspector);
+
   let history_file = global_state.dir.root.join("deno_history.txt");
 
-  session
+  post_message_and_poll(&mut *worker, &mut session, "Runtime.enable", None)
-    .post_message("Runtime.enable".to_string(), None)
     .await?;
 
   let helper = Helper {
@@ -90,23 +152,19 @@ pub async fn run(
     });
   "#;
 
-  session
+  post_message_and_poll(
-    .post_message(
+    &mut *worker,
-      "Runtime.evaluate".to_string(),
+    &mut session,
-      Some(json!({
+    "Runtime.evaluate",
-        "expression": prelude,
+    Some(json!({
-        "contextId": context_id,
+      "expression": prelude,
-      })),
+      "contextId": context_id,
-    )
+    })),
-    .await?;
+  )
+  .await?;
 
   loop {
-    let editor2 = editor.clone();
+    let line = read_line_and_poll(&mut *worker, editor.clone()).await;
-    let line = tokio::task::spawn_blocking(move || {
-      editor2.lock().unwrap().readline("> ")
-    })
-    .await?;
-
     match line {
       Ok(line) => {
         // It is a bit unexpected that { "foo": "bar" } is interpreted as a block
@@ -120,16 +178,17 @@ pub async fn run(
           line.clone()
         };
 
-        let evaluate_response = session
+        let evaluate_response = post_message_and_poll(
-          .post_message(
+          &mut *worker,
-            "Runtime.evaluate".to_string(),
+          &mut session,
-            Some(json!({
+          "Runtime.evaluate",
-              "expression": format!("'use strict'; void 0;\n{}", &wrapped_line),
+          Some(json!({
-              "contextId": context_id,
+            "expression": format!("'use strict'; void 0;\n{}", &wrapped_line),
-              "replMode": true,
+            "contextId": context_id,
-            })),
+            "replMode": true,
-          )
+          })),
-          .await?;
+        )
+        .await?;
 
         // If that fails, we retry it without wrapping in parens letting the error bubble up to the
         // user if it is still an error.
@@ -137,35 +196,37 @@ pub async fn run(
           if evaluate_response.get("exceptionDetails").is_some()
             && wrapped_line != line
           {
-            session
+            post_message_and_poll(
-              .post_message(
+              &mut *worker,
-                "Runtime.evaluate".to_string(),
+              &mut session,
-                Some(json!({
+              "Runtime.evaluate",
-                  "expression": format!("'use strict'; void 0;\n{}", &line),
+              Some(json!({
-                  "contextId": context_id,
+                "expression": format!("'use strict'; void 0;\n{}", &line),
-                  "replMode": true,
+                "contextId": context_id,
-                })),
+                "replMode": true,
-              )
+              })),
-              .await?
+            )
+            .await?
           } else {
             evaluate_response
           };
 
-        let is_closing = session
+        let is_closing = post_message_and_poll(
-          .post_message(
+          &mut *worker,
-            "Runtime.evaluate".to_string(),
+          &mut session,
-            Some(json!({
+          "Runtime.evaluate",
-              "expression": "(globalThis.closed)",
+          Some(json!({
-              "contextId": context_id,
+            "expression": "(globalThis.closed)",
-            })),
+            "contextId": context_id,
-          )
+          })),
-          .await?
+        )
-          .get("result")
+        .await?
-          .unwrap()
+        .get("result")
-          .get("value")
+        .unwrap()
-          .unwrap()
+        .get("value")
-          .as_bool()
+        .unwrap()
-          .unwrap();
+        .as_bool()
+        .unwrap();
 
         if is_closing {
           break;
@@ -176,42 +237,49 @@ pub async fn run(
           evaluate_response.get("exceptionDetails");
 
         if evaluate_exception_details.is_some() {
-          session
+          post_message_and_poll(
-            .post_message(
+            &mut *worker,
-            "Runtime.callFunctionOn".to_string(),
+            &mut session,
+            "Runtime.callFunctionOn",
             Some(json!({
               "executionContextId": context_id,
               "functionDeclaration": "function (object) { Deno[Deno.internal].lastThrownError = object; }",
               "arguments": [
                 evaluate_result,
               ],
-            }))).await?;
+            })),
+          ).await?;
         } else {
-          session
+          post_message_and_poll(
-            .post_message(
+            &mut *worker,
-            "Runtime.callFunctionOn".to_string(),
+            &mut session,
+            "Runtime.callFunctionOn",
             Some(json!({
               "executionContextId": context_id,
               "functionDeclaration": "function (object) { Deno[Deno.internal].lastEvalResult = object; }",
               "arguments": [
                 evaluate_result,
               ],
-            }))).await?;
+            })),
+          ).await?;
         }
 
         // TODO(caspervonb) we should investigate using previews here but to keep things
         // consistent with the previous implementation we just get the preview result from
         // Deno.inspectArgs.
-        let inspect_response = session
+        let inspect_response =
-          .post_message(
+          post_message_and_poll(
-            "Runtime.callFunctionOn".to_string(),
+            &mut *worker,
+            &mut session,
+            "Runtime.callFunctionOn",
             Some(json!({
               "executionContextId": context_id,
               "functionDeclaration": "function (object) { return Deno[Deno.internal].inspectArgs(['%o', object], { colors: true}); }",
               "arguments": [
                 evaluate_result,
               ],
-            }))).await?;
+            })),
+          ).await?;
 
         let inspect_result = inspect_response.get("result").unwrap();