brokerd: move the container's pipe handling into pipes.rs

crates/brokerd/src/container.rs

@@ -3,16 +3,15 @@
 //! said goes only to `brokerd`'s log. M3b spec, section 6.
 
 use std::ffi::OsString;
-use std::io::{Read, Write};
 use std::os::unix::fs::{DirBuilderExt, PermissionsExt};
 use std::path::{Path, PathBuf};
 use std::process::{Child, Command, ExitStatus, Stdio};
 use std::sync::Arc;
 use std::sync::atomic::{AtomicU64, Ordering};
-use std::thread::JoinHandle;
 use std::time::{Duration, Instant};
 
 use crate::config::Runner;
+use crate::pipes::Io;
 use crate::podman;
 use crate::runner::{RunError, RunOutput, RunSpec, Runtime};
 
@@ -349,84 +348,3 @@ impl<'a> EgressGuard<'a> {
         }
     }
 }
-
-/// Step 4: the three threads that feed and read one container, so no pipe can block another.
-struct Io {
-    writer: Option<JoinHandle<()>>,
-    stdout: Option<JoinHandle<(Vec<u8>, bool)>>,
-    stderr: Option<JoinHandle<(Vec<u8>, bool)>>,
-}
-
-impl Io {
-    /// Take `child`'s three pipes (`child.stdin.take()` and so on) and start one thread for each:
-    /// write `input` to standard input and then drop it; `read_capped` standard output with `cap`;
-    /// `read_capped` standard error with `STDERR_KEPT`. A pipe that is `None` gets no thread.
-    fn start(child: &mut Child, input: String, cap: usize) -> Io {
-        let writer = child.stdin.take().map(|mut stdin| {
-            std::thread::spawn(move || {
-                let _ = stdin.write_all(input.as_bytes());
-            })
-        });
-        let stdout = child
-            .stdout
-            .take()
-            .map(|stdout| std::thread::spawn(move || read_capped(stdout, cap)));
-        let stderr = child
-            .stderr
-            .take()
-            .map(|stderr| std::thread::spawn(move || read_capped(stderr, STDERR_KEPT)));
-        Io {
-            writer,
-            stdout,
-            stderr,
-        }
-    }
-
-    /// Step 6: join the three threads. A thread that is missing or panicked counts as empty
-    /// (`join().ok()`, `unwrap_or_default()`). Returns the kept standard output, whether there was
-    /// more, and the kept standard error decoded with `from_utf8_lossy`.
-    fn finish(self) -> (Vec<u8>, bool, String) {
-        if let Some(handle) = self.writer {
-            let _ = handle.join();
-        }
-        let (out, truncated) = match self.stdout {
-            Some(handle) => handle.join().ok().unwrap_or_default(),
-            None => (Vec::new(), false),
-        };
-        let err = match self.stderr {
-            Some(handle) => {
-                let (bytes, _) = handle.join().ok().unwrap_or_default();
-                String::from_utf8_lossy(&bytes).into_owned()
-            }
-            None => String::new(),
-        };
-        (out, truncated, err)
-    }
-}
-
-/// Everything `from` gives, keeping the first `cap` bytes; `true` if there was more. Reads on past
-/// the cap, so the writer is never blocked or broken by a closed pipe. No indexing: `buf.get(..n)`.
-fn read_capped(from: impl Read, cap: usize) -> (Vec<u8>, bool) {
-    let mut from = from;
-    let mut kept: Vec<u8> = Vec::new();
-    let mut truncated = false;
-    let mut chunk = [0u8; 8192];
-    loop {
-        let n = match from.read(&mut chunk) {
-            Ok(0) => break,
-            Ok(n) => n,
-            Err(_) => break,
-        };
-        let remaining = cap - kept.len();
-        if remaining == 0 {
-            truncated = true;
-        } else {
-            let take = remaining.min(n);
-            kept.extend_from_slice(&chunk[..take]);
-            if n > take {
-                truncated = true;
-            }
-        }
-    }
-    (kept, truncated)
-}

crates/brokerd/src/lib.rs

@@ -9,6 +9,7 @@ pub mod config;
 pub mod container;
 pub mod grants;
 pub mod ledger;
+pub mod pipes;
 pub mod podman;
 pub mod policy;
 pub mod runner;

crates/brokerd/src/pipes.rs

@@ -0,0 +1,86 @@
+//! The three pipes of one container: its arguments go in on standard input, its output and errors
+//! come back, each on its own thread so no pipe can block another.
+
+use std::io::{Read, Write};
+use std::process::Child;
+use std::thread::JoinHandle;
+
+/// Step 4: the three threads that feed and read one container, so no pipe can block another.
+pub(crate) struct Io {
+    writer: Option<JoinHandle<()>>,
+    stdout: Option<JoinHandle<(Vec<u8>, bool)>>,
+    stderr: Option<JoinHandle<(Vec<u8>, bool)>>,
+}
+
+impl Io {
+    /// Take `child`'s three pipes (`child.stdin.take()` and so on) and start one thread for each:
+    /// write `input` to standard input and then drop it; `read_capped` standard output with `cap`;
+    /// `read_capped` standard error with `STDERR_KEPT`. A pipe that is `None` gets no thread.
+    pub(crate) fn start(child: &mut Child, input: String, cap: usize) -> Io {
+        let writer = child.stdin.take().map(|mut stdin| {
+            std::thread::spawn(move || {
+                let _ = stdin.write_all(input.as_bytes());
+            })
+        });
+        let stdout = child
+            .stdout
+            .take()
+            .map(|stdout| std::thread::spawn(move || read_capped(stdout, cap)));
+        let stderr = child.stderr.take().map(|stderr| {
+            std::thread::spawn(move || read_capped(stderr, crate::container::STDERR_KEPT))
+        });
+        Io {
+            writer,
+            stdout,
+            stderr,
+        }
+    }
+
+    /// Step 6: join the three threads. A thread that is missing or panicked counts as empty
+    /// (`join().ok()`, `unwrap_or_default()`). Returns the kept standard output, whether there was
+    /// more, and the kept standard error decoded with `from_utf8_lossy`.
+    pub(crate) fn finish(self) -> (Vec<u8>, bool, String) {
+        if let Some(handle) = self.writer {
+            let _ = handle.join();
+        }
+        let (out, truncated) = match self.stdout {
+            Some(handle) => handle.join().ok().unwrap_or_default(),
+            None => (Vec::new(), false),
+        };
+        let err = match self.stderr {
+            Some(handle) => {
+                let (bytes, _) = handle.join().ok().unwrap_or_default();
+                String::from_utf8_lossy(&bytes).into_owned()
+            }
+            None => String::new(),
+        };
+        (out, truncated, err)
+    }
+}
+
+/// Everything `from` gives, keeping the first `cap` bytes; `true` if there was more. Reads on past
+/// the cap, so the writer is never blocked or broken by a closed pipe. No indexing: `buf.get(..n)`.
+pub(crate) fn read_capped(from: impl Read, cap: usize) -> (Vec<u8>, bool) {
+    let mut from = from;
+    let mut kept: Vec<u8> = Vec::new();
+    let mut truncated = false;
+    let mut chunk = [0u8; 8192];
+    loop {
+        let n = match from.read(&mut chunk) {
+            Ok(0) => break,
+            Ok(n) => n,
+            Err(_) => break,
+        };
+        let remaining = cap - kept.len();
+        if remaining == 0 {
+            truncated = true;
+        } else {
+            let take = remaining.min(n);
+            kept.extend_from_slice(&chunk[..take]);
+            if n > take {
+                truncated = true;
+            }
+        }
+    }
+    (kept, truncated)
+}