Route stdout and stderr through the broker

.github/workflows/ci.yml

@@ -90,6 +90,7 @@ jobs:
       # is the PR this job starts covering it.
       AARCH64_CRATES: >-
         -p litebox
+        -p litebox_broker_userland
         -p litebox_common_linux
         -p litebox_egress_proxy
         -p litebox_syscall_rewriter

dev_bench/unixbench/README.md

@@ -6,11 +6,12 @@ Run [byte-unixbench](https://github.com/kdlucas/byte-unixbench) benchmarks nativ
 
 - The UnixBench source tree at `byte-unixbench-6.0.0/UnixBench/` (extracted from `v6.0.0.zip`).
 - `gcc`, `make`, `ldd`, `tar` on the host.
-- Pre-built LiteBox binaries (`litebox_runner_linux_userland` and `litebox_syscall_rewriter`).
+- Pre-built LiteBox binaries (`litebox-broker-userland`,
+  `litebox_runner_linux_userland`, and `litebox_syscall_rewriter`).
 
 Build LiteBox (from workspace root):
 ```bash
-cargo build --release -p litebox_runner_linux_userland -p litebox_syscall_rewriter
+cargo build --release -p litebox_broker_userland -p litebox_runner_linux_userland -p litebox_syscall_rewriter
 ```
 
 ## Quick Start
@@ -129,8 +130,8 @@ This creates `dev_bench/unixbench/prepared/` containing:
 Build the Windows runner, then run benchmarks using the prepared artifacts:
 
 ```powershell
-# Build the Windows runner
-cargo build -p litebox_runner_linux_on_windows_userland --release
+# Build the Windows runner and broker
+cargo build -p litebox_runner_linux_on_windows_userland -p litebox_broker_userland --release
 
 # Run benchmarks
 python run_unixbench.py --mode litebox --windows --prepared-dir ./prepared --release

dev_bench/unixbench/run_unixbench.py

@@ -39,6 +39,7 @@
 import json
 import os
 import re
+import signal
 import shutil
 import subprocess
 import sys
@@ -350,35 +351,73 @@ def _run_litebox_cmd(
     # Use a shorter timeout for alarm-based benchmarks under LiteBox,
     # since if SIGALRM isn't delivered the process will hang forever.
     timeout = duration * 3 + 30 if bench.uses_alarm else duration * 10 + 60
-    try:
-        result = subprocess.run(
-            cmd, capture_output=True, timeout=timeout,
+    popen_options = {}
+    if sys.platform == "win32":
+        popen_options["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP
+    else:
+        popen_options["start_new_session"] = True
+    with tempfile.TemporaryFile() as stderr_file:
+        process = subprocess.Popen(
+            cmd,
+            stdout=subprocess.DEVNULL,
+            stderr=stderr_file,
+            **popen_options,
         )
-    except subprocess.TimeoutExpired:
-        hint = " (this benchmark uses alarm/SIGALRM)" if bench.uses_alarm else ""
-        print(f"  [TIMEOUT] {bench.name}{hint}")
-        return None
+        try:
+            process.wait(timeout=timeout)
+        except subprocess.TimeoutExpired:
+            _terminate_process_tree(process)
+            hint = " (this benchmark uses alarm/SIGALRM)" if bench.uses_alarm else ""
+            print(f"  [TIMEOUT] {bench.name}{hint}")
+            return None
+        except KeyboardInterrupt:
+            _terminate_process_tree(process)
+            raise
+
+        stderr_file.seek(0)
+        stderr = stderr_file.read()
     elapsed = time.monotonic() - t0
 
-    if result.returncode != 0:
-        stderr = result.stderr.decode("utf-8", errors="replace")
-        print(f"  [FAIL] {bench.name} exited with {result.returncode}")
-        print(f"  stderr (last 300 chars): ...{stderr[-300:]}")
+    if process.returncode != 0:
+        decoded_stderr = stderr.decode("utf-8", errors="replace")
+        print(f"  [FAIL] {bench.name} exited with {process.returncode}")
+        print(f"  stderr (last 300 chars): ...{decoded_stderr[-300:]}")
         return None
 
-    stderr = result.stderr.decode("utf-8", errors="replace")
-    parsed = parse_count_line(stderr)
+    decoded_stderr = stderr.decode("utf-8", errors="replace")
+    parsed = parse_count_line(decoded_stderr)
     if parsed is None:
-        print(f"  [FAIL] {bench.name}: no COUNT line in stderr:\n{stderr[:500]}")
+        print(f"  [FAIL] {bench.name}: no COUNT line in stderr:\n{decoded_stderr[:500]}")
         return None
 
     count, base, unit = parsed
     return BenchmarkResult(
         name=bench.name, count=count, base=base, unit=unit,
-        elapsed=elapsed, raw_stderr=stderr,
+        elapsed=elapsed, raw_stderr=decoded_stderr,
     )
 
 
+def _terminate_process_tree(process: subprocess.Popen) -> None:
+    """Terminate a timed-out broker and the runner process it launched."""
+    if sys.platform == "win32":
+        result = subprocess.run(
+            ["taskkill", "/PID", str(process.pid), "/T", "/F"],
+            capture_output=True,
+        )
+        if process.poll() is None:
+            if result.returncode != 0:
+                error = result.stderr.decode("utf-8", errors="replace").strip()
+                print(f"  Warning: process-tree termination failed: {error}")
+            process.kill()
+    else:
+        try:
+            os.killpg(process.pid, signal.SIGKILL)
+        except ProcessLookupError:
+            pass
+
+    process.wait()
+
+
 def run_litebox(
     pgms_dir: Path,
     bench: BenchmarkDef,
@@ -395,10 +434,11 @@ def run_litebox(
         return None
 
     tar_path, rewritten = prepared
+    broker_path = runner_path.with_name("litebox-broker-userland")
 
     cmd = [
-        str(runner_path),
-        "--unstable",
+        str(broker_path),
+        "--runner", str(runner_path),
         "--env", "LD_LIBRARY_PATH=/lib64:/lib32:/lib",
         "--env", "HOME=/",
     ]
@@ -448,8 +488,10 @@ def run_litebox_windows(
         print(f"  [SKIP] {bench.name}: tar not found at {tar_path}")
         return None
 
+    broker_path = runner_path.with_name("litebox-broker-userland.exe")
     cmd = [
-        str(runner_path),
+        str(broker_path),
+        "--runner", str(runner_path),
         "--env", "LD_LIBRARY_PATH=/lib64:/lib32:/lib",
         "--env", "HOME=/",
     ]
@@ -533,14 +575,15 @@ def build_litebox_binaries(
     workspace_root: Path, release: bool,
 ) -> tuple[Path, Path]:
     """
-    Build litebox_runner_linux_userland and litebox_packager via cargo.
+    Build the LiteBox runner, broker, and packager via cargo.
 
     Returns (runner_path, packager_path).
     """
     build_type = "release" if release else "debug"
     cmd = [
         "cargo", "build",
         "-p", "litebox_runner_linux_userland",
+        "-p", "litebox_broker_userland",
         "-p", "litebox_packager",
     ]
     if release:
@@ -554,8 +597,10 @@ def build_litebox_binaries(
     print("Build complete.")
 
     runner = workspace_root / "target" / build_type / "litebox_runner_linux_userland"
+    broker = workspace_root / "target" / build_type / "litebox-broker-userland"
     packager = workspace_root / "target" / build_type / "litebox_packager"
     assert runner.exists(), f"Runner not found at {runner}"
+    assert broker.exists(), f"Broker not found at {broker}"
     assert packager.exists(), f"Packager not found at {packager}"
     return runner, packager
 
@@ -604,7 +649,8 @@ def main():
     )
     parser.add_argument(
         "--runner-path", type=str, default=None,
-        help="Path to litebox_runner_linux_userland binary (auto-detected if not given)",
+        help="Path to the LiteBox runner binary; litebox-broker-userland must be beside it "
+             "(auto-detected if not given)",
     )
     parser.add_argument(
         "--packager-path", type=str, default=None,
@@ -692,6 +738,7 @@ def main():
                 build_cmd = [
                     "cargo", "build",
                     "-p", "litebox_runner_linux_on_windows_userland",
+                    "-p", "litebox_broker_userland",
                 ]
                 if args.release:
                     build_cmd.append("--release")
@@ -725,6 +772,22 @@ def main():
                 workspace_root, args.release,
             )
 
+        if not runner_path.exists():
+            print(f"Error: LiteBox runner not found at {runner_path}")
+            sys.exit(1)
+        broker_name = (
+            "litebox-broker-userland.exe"
+            if is_windows_mode
+            else "litebox-broker-userland"
+        )
+        broker_path = runner_path.with_name(broker_name)
+        if not broker_path.exists():
+            print(f"Error: LiteBox broker not found at {broker_path}")
+            print("Build it beside the runner:")
+            print("  cargo build -p litebox_broker_userland"
+                  + (" --release" if args.release else ""))
+            sys.exit(1)
+
     # Working directory
     if args.work_dir:
         work_dir = Path(args.work_dir)

litebox/src/broker/mod.rs

@@ -21,6 +21,7 @@ use litebox_broker_protocol::socket::{
     ReceiveFromSocketResponse, ReceiveSocketResponse, SendFlags as BrokerSendFlags, ShutdownMode,
     SocketConnectionStatus, SocketOutcome, SocketStatusResponse, TcpOptionName, TcpOptionValue,
 };
+use litebox_broker_protocol::stdio::{MAX_STDIO_TRANSFER_SIZE, StdioOutputStream};
 use litebox_broker_transport::channel::LocalCallChannel;
 
 use crate::event::{Events, polling::Pollee};
@@ -44,6 +45,12 @@ use shared_buffer::{SlotAllocator, SlotLease};
 pub(crate) trait BrokerControl: Send + Sync {
     fn fill_random(&self, output: &mut [u8]) -> core::result::Result<(), BrokerControlError>;
 
+    fn write_stdio(
+        &self,
+        stream: StdioOutputStream,
+        data: &[u8],
+    ) -> core::result::Result<usize, BrokerControlError>;
+
     fn create_tcp_socket(&self) -> core::result::Result<ObjectHandle, BrokerControlError>;
 
     fn create_udp_socket(&self) -> core::result::Result<ObjectHandle, BrokerControlError>;
@@ -313,6 +320,20 @@ where
         self.request(|local| local.fill_random(lease.descriptor(), output))
     }
 
+    fn write_stdio(
+        &self,
+        stream: StdioOutputStream,
+        data: &[u8],
+    ) -> core::result::Result<usize, BrokerControlError> {
+        if data.len() > MAX_STDIO_TRANSFER_SIZE as usize {
+            return Err(BrokerControlError::Broker(ErrorCode::ResourceExhausted));
+        }
+        let length = u32::try_from(data.len())
+            .expect("validated shared stdio transfer length must fit in u32");
+        let lease = self.acquire_shared_buffer(length)?;
+        self.request(|local| local.write_stdio(stream, lease.descriptor(), data))
+    }
+
     fn create_tcp_socket(&self) -> core::result::Result<ObjectHandle, BrokerControlError> {
         self.request(BrokerLocal::create_tcp_socket)
     }

litebox/src/event/counter.rs

@@ -485,10 +485,7 @@ mod tests {
                 BrokerOperation::CheckReadiness(_) => {
                     BrokerResult::Readiness(ReadinessFlags::WRITE)
                 }
-                request @ (BrokerOperation::Event(_)
-                | BrokerOperation::Pipe(_)
-                | BrokerOperation::Socket(_)
-                | BrokerOperation::FillRandom(_)) => {
+                request => {
                     panic!("unexpected broker request: {request:?}")
                 }
             };

litebox/src/fs/devices.rs

@@ -11,6 +11,7 @@ use alloc::vec::Vec;
 use litebox_broker_protocol::random::MAX_RANDOM_TRANSFER_SIZE;
 
 use crate::LiteBox;
+use crate::stdio::StdioOutputStream;
 use crate::sync::RawSyncPrimitivesProvider;
 
 use super::backend::{
@@ -291,8 +292,8 @@ where
         let h = h.get_typed::<Self>();
         let stream = match h.device {
             Device::Stdin => return Err(WriteError::NotForWriting),
-            Device::Stdout => crate::platform::StdioOutStream::Stdout,
-            Device::Stderr => crate::platform::StdioOutStream::Stderr,
+            Device::Stdout => StdioOutputStream::Stdout,
+            Device::Stderr => StdioOutputStream::Stderr,
             Device::Null | Device::URandom => {
                 // /dev/null discards data: report as if written fully
                 //
@@ -306,12 +307,8 @@ where
             }
         };
         self.litebox
-            .x
-            .platform
-            .write_to(stream, buf)
-            .map_err(|e| match e {
-                crate::platform::StdioWriteError::Closed => WriteError::Io,
-            })
+            .write_stdio(stream, buf)
+            .map_err(|_| WriteError::Io)
     }
 
     fn truncate(&self, _h: &FileHandle, _len: usize) -> Result<(), TruncateError> {