Add support for stdout output for superconsole emitting.

Summary:
RFC: https://fb.workplace.com/groups/buck2dev/permalink/3976662752621767/

Previously all output of superconsole is stderr by default. This diff add support to emit stdout above the canvas like we emit stderr.

## Why we need output stdout support in superconsole
Superconsole writes to stderr by default. If stdout isn’t managed by it, it would get cleared or users need to stop superconsole. In Buck2 BXL, streaming output is sent to stdout during execution, and users often need to listen on it. At the same time, users still want to see Superconsole’s progress updates.

## This diff
1. adds a `aux_stream` to `NonBlockingSuperConsoleOutput` and `BlockingSuperConsoleOutput`
2. For trait `SuperConsoleOutput` adds `output_to` methods to output to different stream
3. In `SuperConsole`, it adds a new buffer `aux_to_emit` for aux output (stdout) and an api `emit_aux`
4. In render of superconsle, output the stdout when  `aux_to_emit` is not emtpy.

Note: right now when stdout is not compatible with tty, it hasn't handle correctly. Handle this issue at D73564704

Reviewed By: cjhopman

Differential Revision: D73534481

fbshipit-source-id: 38d99da2a8934b0aafb292d7f3697523ba4b12d1

Author: Nero5023

src/builder.rs

@@ -19,7 +19,10 @@ use crate::output::SuperConsoleOutput;
 /// A builder to create SuperConsole, with more options.
 pub struct Builder {
     non_blocking: bool,
+    // The stream that superconsole writes to by default (emit output + canvas). By default is stderr.
     stream: Box<dyn Write + Send + 'static + Sync>,
+    // The stream that superconsole writes to for auxiliary output. By default is stdout.
+    aux_stream: Box<dyn Write + Send + 'static + Sync>,
 }
 
 impl Default for Builder {
@@ -33,6 +36,7 @@ impl Builder {
         Self {
             non_blocking: false,
             stream: Box::new(io::stderr()),
+            aux_stream: Box::new(io::stdout()),
         }
     }
 
@@ -62,14 +66,21 @@ impl Builder {
     }
 
     fn build_inner(self, fallback_size: Option<Dimensions>) -> anyhow::Result<SuperConsole> {
-        Ok(SuperConsole::new_with_output(fallback_size, self.output()?))
+        let output = self.output()?;
+        Ok(SuperConsole::new_with_output(fallback_size, output))
     }
 
     fn output(self) -> anyhow::Result<Box<dyn SuperConsoleOutput>> {
         if self.non_blocking {
-            Ok(Box::new(NonBlockingSuperConsoleOutput::new(self.stream)?))
+            Ok(Box::new(NonBlockingSuperConsoleOutput::new(
+                self.stream,
+                self.aux_stream,
+            )?))
         } else {
-            Ok(Box::new(BlockingSuperConsoleOutput::new(self.stream)))
+            Ok(Box::new(BlockingSuperConsoleOutput::new(
+                self.stream,
+                self.aux_stream,
+            )))
         }
     }
 }

src/content/lines.rs

@@ -285,20 +285,27 @@ impl Lines {
     /// If a limit is specified, no more than that amount will be drained.
     /// The limit is on the number of *lines*, **NOT** the number of *bytes*.
     /// Care should be taken with calling a limit of 0 - this will cause no lines to render and the buffer to never be drained.
+    ///
+    /// Returns the remain limit after rendering.  If the limit is None, means no limit
     pub(crate) fn render_with_limit(
         &mut self,
         writer: &mut Vec<u8>,
         limit: Option<usize>,
-    ) -> anyhow::Result<()> {
-        let limit = limit.unwrap_or(self.len());
-        let amt = cmp::min(limit, self.len());
+    ) -> anyhow::Result<Option<usize>> {
+        let output_limit = limit.unwrap_or(self.len());
+        let amt = cmp::min(output_limit, self.len());
         for line in self.0.drain(..amt) {
             line.render_with_clear_and_nl(writer)?;
         }
-        Ok(())
+        if limit.is_some() {
+            Ok(Some(output_limit - amt))
+        } else {
+            // if the original limit was None, it means no limit, so just return None meaning no limit
+            Ok(None)
+        }
     }
 
-    /// Formats and renders all lines to `stdout`.
+    /// Formats and renders all lines to `buffer`.
     /// Notably, this *queues* the lines for rendering.  You must flush the buffer.
     pub(crate) fn render_from_line(
         &self,

src/output.rs

@@ -20,6 +20,15 @@ use crossbeam_channel::unbounded;
 
 use crate::Dimensions;
 
+/// Represents the target stream for output
+#[derive(Copy, Clone, Debug)]
+pub enum OutputTarget {
+    /// Main output stream (default: stderr)
+    Main,
+    /// Auxiliary output stream (default: stdout)
+    Aux,
+}
+
 pub trait SuperConsoleOutput: Send + Sync + 'static {
     /// Called before rendering will occur. This has a chance to prevent rendering by returning
     /// false.
@@ -29,6 +38,14 @@ pub trait SuperConsoleOutput: Send + Sync + 'static {
     /// clearing. This should flush if possible.
     fn output(&mut self, buffer: Vec<u8>) -> anyhow::Result<()>;
 
+    /// Called to produce to a specific output target.
+    ///This may be called without should_render if we are finalizing or clearing. This should flush if possible.
+    /// Default implementation sends all to main output for backwards compatibility
+    fn output_to(&mut self, buffer: Vec<u8>, target: OutputTarget) -> anyhow::Result<()> {
+        let _ = target;
+        self.output(buffer)
+    }
+
     /// How big is the terminal to write to.
     fn terminal_size(&self) -> anyhow::Result<Dimensions> {
         Ok(crossterm::terminal::size()?.into())
@@ -48,11 +65,16 @@ pub trait SuperConsoleOutput: Send + Sync + 'static {
 pub struct BlockingSuperConsoleOutput {
     /// Stream to write to.
     stream: Box<dyn Write + Send + 'static + Sync>,
+    /// Auxiliary stream to write to.
+    aux_stream: Box<dyn Write + Send + 'static + Sync>,
 }
 
 impl BlockingSuperConsoleOutput {
-    pub fn new(stream: Box<dyn Write + Send + 'static + Sync>) -> Self {
-        Self { stream }
+    pub fn new(
+        stream: Box<dyn Write + Send + 'static + Sync>,
+        aux_stream: Box<dyn Write + Send + 'static + Sync>,
+    ) -> Self {
+        Self { stream, aux_stream }
     }
 }
 
@@ -62,9 +84,20 @@ impl SuperConsoleOutput for BlockingSuperConsoleOutput {
     }
 
     fn output(&mut self, buffer: Vec<u8>) -> anyhow::Result<()> {
-        self.stream.write_all(&buffer)?;
-        self.stream.flush()?;
+        self.output_to(buffer, OutputTarget::Main)
+    }
 
+    fn output_to(&mut self, buffer: Vec<u8>, target: OutputTarget) -> anyhow::Result<()> {
+        match target {
+            OutputTarget::Main => {
+                self.stream.write_all(&buffer)?;
+                self.stream.flush()?;
+            }
+            OutputTarget::Aux => {
+                self.aux_stream.write_all(&buffer)?;
+                self.aux_stream.flush()?;
+            }
+        }
         Ok(())
     }
 
@@ -88,7 +121,7 @@ impl SuperConsoleOutput for BlockingSuperConsoleOutput {
 /// - When an error occurs, the next fallible call will return it.
 pub(crate) struct NonBlockingSuperConsoleOutput {
     /// A channel to send frames for writing.
-    sender: Sender<Vec<u8>>,
+    sender: Sender<(Vec<u8>, OutputTarget)>,
     /// A channel back for errors encountered by the thread doing the writing.
     errors: Receiver<io::Error>,
     /// The thread doing the writing. It owns the other end of the aforementioned channels and will
@@ -97,19 +130,29 @@ pub(crate) struct NonBlockingSuperConsoleOutput {
 }
 
 impl NonBlockingSuperConsoleOutput {
-    pub fn new(stream: Box<dyn Write + Send + 'static + Sync>) -> anyhow::Result<Self> {
-        Self::new_for_writer(stream)
+    pub fn new(
+        stream: Box<dyn Write + Send + 'static + Sync>,
+        aux_stream: Box<dyn Write + Send + 'static + Sync>,
+    ) -> anyhow::Result<Self> {
+        Self::new_for_writer(stream, aux_stream)
     }
 
-    fn new_for_writer(mut stream: Box<dyn Write + Send + 'static + Sync>) -> anyhow::Result<Self> {
-        let (sender, receiver) = bounded::<Vec<u8>>(1);
+    fn new_for_writer(
+        mut stream: Box<dyn Write + Send + 'static + Sync>,
+        mut aux_stream: Box<dyn Write + Send + 'static + Sync>,
+    ) -> anyhow::Result<Self> {
+        let (sender, receiver) = bounded::<(Vec<u8>, OutputTarget)>(1);
         let (error_sender, errors) = unbounded::<io::Error>();
 
         let handle = std::thread::Builder::new()
             .name("superconsole-io".to_owned())
             .spawn(move || {
-                for frame in receiver.into_iter() {
-                    match stream.write_all(&frame).and_then(|()| stream.flush()) {
+                for (data, output_target) in receiver.into_iter() {
+                    let out_stream = match output_target {
+                        OutputTarget::Main => &mut stream,
+                        OutputTarget::Aux => &mut aux_stream,
+                    };
+                    match out_stream.write_all(&data).and_then(|()| stream.flush()) {
                         Ok(()) => {}
                         Err(e) => {
                             // This can only fail if the sender disconnected, in which case they'll
@@ -140,12 +183,16 @@ impl SuperConsoleOutput for NonBlockingSuperConsoleOutput {
     /// Attempt to send out a frame. If we called should_render, this won't block. If we didn't,
     /// then it may block.
     fn output(&mut self, buffer: Vec<u8>) -> anyhow::Result<()> {
+        self.output_to(buffer, OutputTarget::Main)
+    }
+
+    fn output_to(&mut self, buffer: Vec<u8>, target: OutputTarget) -> anyhow::Result<()> {
         if let Ok(err) = self.errors.try_recv() {
             return Err(anyhow::Error::from(err).context("Superconsole I/O thread errored"));
         }
 
         self.sender
-            .send(buffer)
+            .send((buffer, target))
             .context("Superconsole I/O thread has crashed")?;
 
         Ok(())
@@ -221,29 +268,44 @@ mod tests {
 
     #[test]
     fn test_non_blocking_output_errors_on_next_output() -> anyhow::Result<()> {
-        let (writer, drain) = TestWriter::new();
+        fn test_send_target(target0: OutputTarget, target1: OutputTarget) -> anyhow::Result<()> {
+            let (writer, drain) = TestWriter::new();
+            let aux_writer = writer.clone();
+
+            let mut output = NonBlockingSuperConsoleOutput::new_for_writer(
+                Box::new(writer),
+                Box::new(aux_writer),
+            )?;
 
-        let mut output = NonBlockingSuperConsoleOutput::new_for_writer(Box::new(writer))?;
+            // Send a first message, this will go into write()
+            assert!(output.should_render());
+            output.output_to(msg(), target0)?;
 
-        // Send a first message, this will go into write()
-        assert!(output.should_render());
-        output.output(msg())?;
+            // Send a second message, this will stay in the channel.
+            output.output_to(msg(), target1)?;
 
-        // Send a second message, this will stay in the channel.
-        output.output(msg())?;
+            // Now, kill the output
+            assert!(!output.should_render());
+            drop(drain);
 
-        // Now, kill the output
-        assert!(!output.should_render());
-        drop(drain);
+            // We expect that should_render() will eventually return true.
+            while !output.should_render() {
+                std::thread::yield_now();
+            }
 
-        // We expect that should_render() will eventually return true.
-        while !output.should_render() {
-            std::thread::yield_now();
+            // Likewise, we expect that sending output and finalizing wold fail.
+            assert!(output.output(Vec::new()).is_err());
+            assert!(Box::new(output).finalize().is_err());
+
+            Ok(())
         }
 
-        // Likewise, we expect that sending output and finalizing wold fail.
-        assert!(output.output(Vec::new()).is_err());
-        assert!(Box::new(output).finalize().is_err());
+        // Test all combinations of targets
+        for target0 in [OutputTarget::Main, OutputTarget::Aux] {
+            for target1 in [OutputTarget::Main, OutputTarget::Aux] {
+                test_send_target(target0, target1)?
+            }
+        }
 
         Ok(())
     }