Basic stdin support

Author: shepmaster

compiler/base/orchestrator/src/coordinator.rs

@@ -605,10 +605,12 @@ impl Container {
     ) -> Result<WithOutput<ExecuteResponse>, ExecuteError> {
         let ActiveExecution {
             task,
+            stdin_tx,
             stdout_rx,
             stderr_rx,
         } = self.begin_execute(request).await?;
 
+        drop(stdin_tx);
         WithOutput::try_absorb(task, stdout_rx, stderr_rx).await
     }
 
@@ -636,6 +638,7 @@ impl Container {
 
         let SpawnCargo {
             task,
+            stdin_tx,
             stdout_rx,
             stderr_rx,
         } = self
@@ -660,6 +663,7 @@ impl Container {
 
         Ok(ActiveExecution {
             task,
+            stdin_tx,
             stdout_rx,
             stderr_rx,
         })
@@ -707,13 +711,16 @@ impl Container {
 
         let SpawnCargo {
             task,
+            stdin_tx,
             stdout_rx,
             stderr_rx,
         } = self
             .spawn_cargo_task(execute_cargo)
             .await
             .context(CouldNotStartCargoSnafu)?;
 
+        drop(stdin_tx);
+
         let commander = self.commander.clone();
         let task = async move {
             let ExecuteCommandResponse {
@@ -758,41 +765,53 @@ impl Container {
     ) -> Result<SpawnCargo, SpawnCargoError> {
         use spawn_cargo_error::*;
 
+        let (stdin_tx, mut stdin_rx) = mpsc::channel(8);
         let (stdout_tx, stdout_rx) = mpsc::channel(8);
         let (stderr_tx, stderr_rx) = mpsc::channel(8);
 
-        let mut from_worker_rx = self
+        let (to_worker_tx, mut from_worker_rx) = self
             .commander
             .many(execute_cargo)
             .await
             .context(CouldNotStartCargoSnafu)?;
 
         let task = tokio::spawn({
             async move {
-                while let Some(container_msg) = from_worker_rx.recv().await {
-                    trace!("processing {container_msg:?}");
-
-                    match container_msg {
-                        WorkerMessage::ExecuteCommand(resp) => {
-                            return Ok(resp);
-                        }
-                        WorkerMessage::StdoutPacket(packet) => {
-                            stdout_tx.send(packet).await.ok(/* Receiver gone, that's OK */);
-                        }
-                        WorkerMessage::StderrPacket(packet) => {
-                            stderr_tx.send(packet).await.ok(/* Receiver gone, that's OK */);
-                        }
-                        _ => return UnexpectedMessageSnafu.fail(),
+                loop {
+                    select! {
+                        Some(stdin) = stdin_rx.recv() => {
+                            let msg = CoordinatorMessage::StdinPacket(stdin);
+                            trace!("processing {msg:?}");
+                            to_worker_tx.send(msg).await.context(StdinSnafu)?;
+                        },
+
+                        Some(container_msg) = from_worker_rx.recv() => {
+                            trace!("processing {container_msg:?}");
+
+                            match container_msg {
+                                WorkerMessage::ExecuteCommand(resp) => {
+                                    return Ok(resp);
+                                }
+                                WorkerMessage::StdoutPacket(packet) => {
+                                    stdout_tx.send(packet).await.ok(/* Receiver gone, that's OK */);
+                                }
+                                WorkerMessage::StderrPacket(packet) => {
+                                    stderr_tx.send(packet).await.ok(/* Receiver gone, that's OK */);
+                                }
+                                _ => return UnexpectedMessageSnafu.fail(),
+                            }
+                        },
+
+                        else => return UnexpectedEndOfMessagesSnafu.fail(),
                     }
                 }
-
-                UnexpectedEndOfMessagesSnafu.fail()
             }
             .instrument(trace_span!("cargo task").or_current())
         });
 
         Ok(SpawnCargo {
             task,
+            stdin_tx,
             stdout_rx,
             stderr_rx,
         })
@@ -812,6 +831,7 @@ impl Container {
 
 pub struct ActiveExecution {
     pub task: BoxFuture<'static, Result<ExecuteResponse, ExecuteError>>,
+    pub stdin_tx: mpsc::Sender<String>,
     pub stdout_rx: mpsc::Receiver<String>,
     pub stderr_rx: mpsc::Receiver<String>,
 }
@@ -820,6 +840,7 @@ impl fmt::Debug for ActiveExecution {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("ActiveExecution")
             .field("task", &"<future>")
+            .field("stdin_tx", &self.stdin_tx)
             .field("stdout_rx", &self.stdout_rx)
             .field("stderr_rx", &self.stderr_rx)
             .finish()
@@ -900,6 +921,7 @@ pub enum CompileError {
 
 struct SpawnCargo {
     task: JoinHandle<Result<ExecuteCommandResponse, SpawnCargoError>>,
+    stdin_tx: mpsc::Sender<String>,
     stdout_rx: mpsc::Receiver<String>,
     stderr_rx: mpsc::Receiver<String>,
 }
@@ -915,6 +937,9 @@ pub enum SpawnCargoError {
 
     #[snafu(display("There are no more messages"))]
     UnexpectedEndOfMessages,
+
+    #[snafu(display("Unable to send stdin message"))]
+    Stdin { source: MultiplexedSenderError },
 }
 
 #[derive(Debug, Clone)]
@@ -1173,7 +1198,10 @@ impl Commander {
         }
     }
 
-    async fn many<M>(&self, message: M) -> Result<mpsc::Receiver<WorkerMessage>, CommanderError>
+    async fn many<M>(
+        &self,
+        message: M,
+    ) -> Result<(MultiplexedSender, mpsc::Receiver<WorkerMessage>), CommanderError>
     where
         M: Into<CoordinatorMessage>,
     {
@@ -1190,7 +1218,7 @@ impl Commander {
             .await
             .context(UnableToStartManySnafu)?;
 
-        Ok(from_worker_rx)
+        Ok((to_worker_tx, from_worker_rx))
     }
 }
 
@@ -1509,15 +1537,20 @@ mod tests {
         //Coordinator::new_docker().await
     }
 
+    const ARBITRARY_EXECUTE_REQUEST: ExecuteRequest = ExecuteRequest {
+        channel: Channel::Stable,
+        mode: Mode::Debug,
+        edition: Edition::Rust2021,
+        crate_type: CrateType::Binary,
+        tests: false,
+        backtrace: false,
+        code: String::new(),
+    };
+
     fn new_execute_request() -> ExecuteRequest {
         ExecuteRequest {
-            channel: Channel::Stable,
-            mode: Mode::Debug,
-            edition: Edition::Rust2021,
-            crate_type: CrateType::Binary,
-            tests: false,
-            backtrace: false,
             code: r#"fn main() { println!("Hello, coordinator!"); }"#.into(),
+            ..ARBITRARY_EXECUTE_REQUEST
         }
     }
 
@@ -1724,6 +1757,51 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    #[snafu::report]
+    async fn execute_stdin() -> Result<()> {
+        let coordinator = new_coordinator().await;
+
+        let request = ExecuteRequest {
+            code: r#"
+                fn main() {
+                    let mut input = String::new();
+                    if std::io::stdin().read_line(&mut input).is_ok() {
+                        println!("You entered >>>{input:?}<<<");
+                    }
+                }
+            "#
+            .into(),
+            ..ARBITRARY_EXECUTE_REQUEST
+        };
+
+        let ActiveExecution {
+            task,
+            stdin_tx,
+            stdout_rx,
+            stderr_rx,
+        } = coordinator.begin_execute(request).await.unwrap();
+
+        stdin_tx.send("this is stdin\n".into()).await.unwrap();
+        // Purposefully not dropping stdin_tx early -- a user might forget
+
+        let WithOutput {
+            response,
+            stdout,
+            stderr,
+        } = WithOutput::try_absorb(task, stdout_rx, stderr_rx)
+            .with_timeout()
+            .await
+            .unwrap();
+
+        assert!(response.success, "{stderr}");
+        assert_contains!(stdout, r#">>>"this is stdin\n"<<<"#);
+
+        coordinator.shutdown().await?;
+
+        Ok(())
+    }
+
     const HELLO_WORLD_CODE: &str = r#"fn main() { println!("Hello World!"); }"#;
 
     const ARBITRARY_COMPILE_REQUEST: CompileRequest = CompileRequest {

tests/spec/features/streaming_spec.rb

@@ -35,6 +35,34 @@
     end
   end
 
+  scenario "input can be supplied" do
+    editor.set <<~EOF
+      fn main() {
+          println!("Enter some text!");
+          let mut input = String::new();
+          std::io::stdin().read_line(&mut input).expect("Unable to read input");
+          println!("You entered >>>{input:?}<<<");
+      }
+    EOF
+
+    click_on("Run")
+
+    within(:output, :stdout) do
+      expect(page).to have_content 'Enter some text'
+      expect(page).to_not have_content 'You entered'
+    end
+
+    within(:stdin) do
+      fill_in 'content', with: 'An automated test'
+      click_on 'Send'
+    end
+
+    within(:output, :stdout) do
+      expect(page).to have_content 'Enter some text'
+      expect(page).to have_content 'You entered >>>"An automated test\n"<<<'
+    end
+  end
+
   def editor
     Editor.new(page)
   end

tests/spec/spec_helper.rb

@@ -79,6 +79,10 @@
   end
 end
 
+Capybara.add_selector(:stdin) do
+  css { '[data-test-id = "stdin"]' }
+end
+
 RSpec.configure do |config|
   config.after(:example, :js) do
     page.execute_script <<~JS

ui/frontend/.eslintrc.js

@@ -66,6 +66,7 @@ module.exports = {
         'ButtonSet.tsx',
         'Header.tsx',
         'PopButton.tsx',
+        'Stdin.tsx',
         'compileActions.ts',
         'editor/AceEditor.tsx',
         'editor/SimpleEditor.tsx',

ui/frontend/.prettierignore

@@ -17,6 +17,7 @@ node_modules
 !ButtonSet.tsx
 !Header.tsx
 !PopButton.tsx
+!Stdin.tsx
 !compileActions.ts
 !editor/AceEditor.tsx
 !editor/SimpleEditor.tsx

ui/frontend/Output.module.css

@@ -49,3 +49,7 @@ $background-tab: #fcfcfc; /* desaturate($current-tab, 100%); */
   height: 100%;
   overflow: scroll;
 }
+
+.stdin {
+  margin-top: 0.25em;
+}

ui/frontend/Output.tsx

@@ -13,6 +13,7 @@ import PaneWithMir from './Output/PaneWithMir';
 import * as selectors from './selectors';
 
 import styles from './Output.module.css';
+import Stdin from './Stdin';
 
 const Tab: React.FC<TabProps> = ({ kind, focus, label, onClick, tabProps }) => {
   if (selectors.hasProperties(tabProps)) {
@@ -64,6 +65,8 @@ const Output: React.FC = () => {
   const focusWasm = useCallback(() => dispatch(actions.changeFocus(Focus.Wasm)), [dispatch]);
   const focusGist = useCallback(() => dispatch(actions.changeFocus(Focus.Gist)), [dispatch]);
 
+  const showStdin = useSelector(selectors.showStdinSelector);
+
   if (!somethingToShow) {
     return null;
   }
@@ -74,19 +77,26 @@ const Output: React.FC = () => {
     close = <button className={styles.tabClose} onClick={focusClose}>Close</button>;
 
     body = (
-      <div className={styles.body}>
-        {focus === Focus.Execute && <Execute />}
-        {focus === Focus.Format && <SimplePane {...format} kind="format" />}
-        {focus === Focus.Clippy && <SimplePane {...clippy} kind="clippy" />}
-        {focus === Focus.Miri && <SimplePane {...miri} kind="miri" />}
-        {focus === Focus.MacroExpansion && <SimplePane {...macroExpansion} kind="macro-expansion" />}
-        {focus === Focus.Asm && <PaneWithCode {...assembly} kind="asm" />}
-        {focus === Focus.LlvmIr && <PaneWithCode {...llvmIr} kind="llvm-ir" />}
-        {focus === Focus.Mir && <PaneWithMir {...mir} kind="mir" />}
-        {focus === Focus.Hir && <PaneWithMir {...hir} kind="hir" />}
-        {focus === Focus.Wasm && <PaneWithCode {...wasm} kind="wasm" />}
-        {focus === Focus.Gist && <Gist />}
-      </div>
+      <>
+        <div className={styles.body}>
+          {focus === Focus.Execute && <Execute />}
+          {focus === Focus.Format && <SimplePane {...format} kind="format" />}
+          {focus === Focus.Clippy && <SimplePane {...clippy} kind="clippy" />}
+          {focus === Focus.Miri && <SimplePane {...miri} kind="miri" />}
+          {focus === Focus.MacroExpansion && <SimplePane {...macroExpansion} kind="macro-expansion" />}
+          {focus === Focus.Asm && <PaneWithCode {...assembly} kind="asm" />}
+          {focus === Focus.LlvmIr && <PaneWithCode {...llvmIr} kind="llvm-ir" />}
+          {focus === Focus.Mir && <PaneWithMir {...mir} kind="mir" />}
+          {focus === Focus.Hir && <PaneWithMir {...hir} kind="hir" />}
+          {focus === Focus.Wasm && <PaneWithCode {...wasm} kind="wasm" />}
+          {focus === Focus.Gist && <Gist />}
+        </div>
+        {showStdin && (
+          <div className={styles.stdin}>
+            <Stdin />
+          </div>
+        )}
+      </>
     );
   }
 
@@ -139,7 +149,7 @@ const Output: React.FC = () => {
           tabProps={gist} />
         {close}
       </div>
-      { body}
+      { body }
     </div>
   );
 };

ui/frontend/Output/Execute.tsx

@@ -21,7 +21,6 @@ const Execute: React.FC = () => {
     <SimplePane {...details} kind="execute">
       {isAutoBuild && <Warning addMainFunction={addMainFunction} />}
     </SimplePane>
-
   );
 };