From bf3e31bb66eb530a1489793808e2b9d9ea466ae6 Mon Sep 17 00:00:00 2001 From: marshmallow Date: Wed, 25 Mar 2026 01:01:34 +0000 Subject: [PATCH] channel-based status updates (#410) Signed-off-by: marshmallow --- CHANGELOG.md | 4 ++++ index.scip | 0 crates/cli/src/apply.rs | 18 ++++++++++-------- crates/cli/src/tracing_setup.rs | 84 +++++++++++++++++++----------------------------------------------------------------- crates/core/src/lib.rs | 43 ++++++++++++++++++++++++++++++++++--------- crates/core/src/status.rs | 147 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++--------------------------------------------- crates/core/src/hive/executor.rs | 29 ++++++++++++++++++++++------- crates/core/src/commands/pty/mod.rs | 20 ++++++++++---------- 8 file(s) changed, 201 insertion(s)(+), 144 deletion(s)(-) diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ ## [Unreleased] - yyyy-mm-dd +### Fixed + +- Status bar is cleaned every time after execution is completed. + ## [v1.2.0] - 2026-03-18 ### Added diff --git a/index.scip b/index.scip new file mode 100644 --- /dev/null +++ b/index.scip diff --git a/crates/cli/src/apply.rs b/crates/cli/src/apply.rs --- a/crates/cli/src/apply.rs +++ b/crates/cli/src/apply.rs @@ -6,7 +6,7 @@ use miette::{Diagnostic, IntoDiagnostic, Result}; use std::any::Any; use std::collections::HashSet; -use std::io::{Read, stderr}; +use std::io::Read; use std::sync::Arc; use std::sync::atomic::AtomicBool; use thiserror::Error; @@ -15,7 +15,7 @@ use wire_core::hive::node::{Name, Node}; use wire_core::hive::plan::{Goal, plan_for_node}; use wire_core::hive::{Hive, HiveLocation}; -use wire_core::status::STATUS; +use wire_core::status::{UI_SENDER, UiMessage}; use wire_core::{SubCommandModifiers, errors::HiveLibError}; use crate::cli::{ApplyTarget, CommonVerbArgs, Partitions}; @@ -132,9 +132,9 @@ ); } - STATUS - .lock() - .add_many(&partitioned_names.iter().collect::>()); + if let Some(tx) = UI_SENDER.get() { + let _ = tx.send(UiMessage::AddMany(partitioned_names.clone())); + } let mut set = hive .nodes @@ -178,10 +178,12 @@ ); } - if !errors.is_empty() { - // clear the status bar if we are about to print error messages - STATUS.lock().clear(&mut stderr()); + // clear the status bar at the end of execution. + if let Some(tx) = UI_SENDER.get() { + let _ = tx.send(UiMessage::Clear); + } + if !errors.is_empty() { return Err(NodeErrors( errors .into_iter() diff --git a/crates/cli/src/tracing_setup.rs b/crates/cli/src/tracing_setup.rs --- a/crates/cli/src/tracing_setup.rs +++ b/crates/cli/src/tracing_setup.rs @@ -1,14 +1,11 @@ // SPDX-License-Identifier: AGPL-3.0-or-later // Copyright 2024-2025 wire Contributors -use std::{ - collections::VecDeque, - io::{self, Stderr, Write, stderr}, - time::Duration, -}; +use std::io::{self, Write}; use clap_verbosity_flag::{LogLevel, Verbosity}; use owo_colors::{OwoColorize, Stream, Style}; +use tokio::sync::mpsc; use tracing::{Level, Subscriber}; use tracing_log::AsTrace; use tracing_subscriber::{ @@ -22,54 +19,28 @@ registry::LookupSpan, util::SubscriberInitExt, }; -use wire_core::{STDIN_CLOBBER_LOCK, status::STATUS}; +use wire_core::status::{UI_SENDER, UiMessage}; -/// The non-clobbering writer ensures that log lines are held while interactive -/// prompts are shown to the user. If logs where shown, they would "clobber" the -/// sudo / ssh prompt. -/// -/// Additionally, the `STDIN_CLOBBER_LOCK` is used to ensure that no two -/// interactive prompts are shown at the same time. -struct NonClobberingWriter { - queue: VecDeque>, - stderr: Stderr, -} +/// Forwards log lines to the UI worker over `UI_SENDER`. +struct NonClobberingWriter; impl NonClobberingWriter { - fn new() -> Self { - NonClobberingWriter { - queue: VecDeque::with_capacity(100), - stderr: stderr(), - } - } - - /// expects the caller to write the status line - fn dump_previous(&mut self) -> Result<(), io::Error> { - STATUS.lock().clear(&mut self.stderr); - - for buf in self.queue.iter().rev() { - self.stderr.write(buf).map(|_| ())?; - } - - Ok(()) + const fn new() -> Self { + NonClobberingWriter } } impl Write for NonClobberingWriter { - fn write(&mut self, buf: &[u8]) -> std::io::Result { - if let 1.. = STDIN_CLOBBER_LOCK.available_permits() { - self.dump_previous().map(|()| 0)?; - - STATUS.lock().write_above_status(buf, &mut self.stderr) - } else { - self.queue.push_front(buf.to_vec()); - - Ok(buf.len()) + fn write(&mut self, buf: &[u8]) -> io::Result { + if let Some(tx) = UI_SENDER.get() { + let _ = tx.send(UiMessage::LogLine(buf.to_vec())); } + + Ok(buf.len()) } - fn flush(&mut self) -> std::io::Result<()> { - self.stderr.flush() + fn flush(&mut self) -> io::Result<()> { + Ok(()) } } @@ -231,36 +202,19 @@ } } -async fn status_tick_worker() { - let mut interval = tokio::time::interval(Duration::from_secs(1)); - let mut stderr = stderr(); - - loop { - interval.tick().await; - - if STDIN_CLOBBER_LOCK.available_permits() < 1 { - continue; - } - - let mut status = STATUS.lock(); - - status.clear(&mut stderr); - status.write_status(&mut stderr); - } -} - /// Set up logging for the application /// Uses `WireFieldFormat` if -v was never passed pub fn setup_logging(verbosity: &Verbosity, show_progress: bool) { let filter = verbosity.log_level_filter().as_trace(); let registry = tracing_subscriber::registry(); - STATUS.lock().show_progress(show_progress); + let (tx, rx) = mpsc::unbounded_channel(); + UI_SENDER + .set(tx) + .expect("expected setup_logging to the first and only .set() of `UI_SENDER`"); // spawn worker to tick the status bar - if show_progress { - tokio::spawn(status_tick_worker()); - } + tokio::spawn(wire_core::status::status_tick_worker(rx, show_progress)); if verbosity.is_present() { let layer = tracing_subscriber::fmt::layer() diff --git a/crates/core/src/lib.rs b/crates/core/src/lib.rs --- a/crates/core/src/lib.rs +++ b/crates/core/src/lib.rs @@ -6,14 +6,15 @@ #![feature(sync_nonpoison)] #![feature(nonpoison_mutex)] -use std::{ - io::{IsTerminal, stderr}, - sync::LazyLock, +use std::{io::IsTerminal, sync::LazyLock}; + +use tokio::sync::{AcquireError, Semaphore, SemaphorePermit, mpsc::UnboundedSender, oneshot}; + +use crate::{ + errors::HiveLibError, + hive::node::Name, + status::{UI_SENDER, UiMessage}, }; - -use tokio::sync::{AcquireError, Semaphore, SemaphorePermit}; - -use crate::{errors::HiveLibError, hive::node::Name, status::STATUS}; pub mod cache; pub mod commands; @@ -63,9 +64,33 @@ pub static STDIN_CLOBBER_LOCK: LazyLock = LazyLock::new(|| Semaphore::new(1)); -pub async fn acquire_stdin_lock<'a>() -> Result, AcquireError> { +/// `SemaphorePermit` that sends a `UiMessage::Release` on drop +pub struct ClobberGuard<'a>( + #[allow(unused)] SemaphorePermit<'a>, + Option<&'a UnboundedSender>, +); + +impl Drop for ClobberGuard<'_> { + fn drop(&mut self) { + if let Some(tx) = self.1 { + let _ = tx.send(UiMessage::Release); + } + } +} + +pub async fn acquire_stdin_lock<'a>() -> Result, AcquireError> { let result = STDIN_CLOBBER_LOCK.acquire().await?; - STATUS.lock().wipe_out(&mut stderr()); + let (sender, rx) = oneshot::channel(); + let tx = UI_SENDER.get(); + + if let Some(tx) = tx { + let _ = tx.send(UiMessage::Takeover(sender)); + + // wait until takeover is confirmed + let _ = rx.await; + } + + let result = ClobberGuard(result, tx); Ok(result) } diff --git a/crates/core/src/status.rs b/crates/core/src/status.rs --- a/crates/core/src/status.rs +++ b/crates/core/src/status.rs @@ -2,15 +2,21 @@ // Copyright 2024-2025 wire Contributors use owo_colors::OwoColorize; -use std::{fmt::Write, time::Instant}; -use termion::{clear, cursor}; - -use crate::{STDIN_CLOBBER_LOCK, hive::node::Name}; - use std::{ - collections::HashMap, - sync::{LazyLock, nonpoison::Mutex}, + collections::VecDeque, + fmt::Write, + sync::OnceLock, + time::{Duration, Instant}, }; +use termion::{clear, cursor}; +use tokio::sync::{ + mpsc::{self, UnboundedReceiver}, + oneshot, +}; + +use crate::hive::node::Name; + +use std::collections::HashMap; #[derive(Default)] pub enum NodeStatus { @@ -21,14 +27,31 @@ Failed, } +pub enum UiMessage { + /// Initialise the status bar with many nodes at once. + AddMany(Vec), + SetStatus(Name, NodeStatus), + /// Takeover the terminal, blocking new messages from being printed until + /// `Release` is sent. + /// + /// Once the takeover request is completed, the oneshot channel will be + /// consumed. + Takeover(oneshot::Sender<()>), + /// Indicate that the takeover is no longer necessary + Release, + /// Clear the status line, mostly for when the program is about to end + Clear, + /// Writes above the status line + LogLine(Vec), +} + pub struct Status { statuses: HashMap, began: Instant, show_progress: bool, } -/// global status used for the progress bar in the cli crate -pub static STATUS: LazyLock> = LazyLock::new(|| Mutex::new(Status::new())); +pub static UI_SENDER: OnceLock> = OnceLock::new(); impl Status { fn new() -> Self { @@ -41,28 +64,6 @@ pub const fn show_progress(&mut self, show_progress: bool) { self.show_progress = show_progress; - } - - pub fn add_many(&mut self, names: &[&Name]) { - self.statuses.extend( - names - .iter() - .map(|name| (name.0.to_string(), NodeStatus::Pending)), - ); - } - - pub fn set_node_step(&mut self, node: &Name, step: String) { - self.statuses - .insert(node.0.to_string(), NodeStatus::Running(step)); - } - - pub fn mark_node_failed(&mut self, node: &Name) { - self.statuses.insert(node.0.to_string(), NodeStatus::Failed); - } - - pub fn mark_node_succeeded(&mut self, node: &Name) { - self.statuses - .insert(node.0.to_string(), NodeStatus::Succeeded); } #[must_use] @@ -153,21 +154,77 @@ let _ = write!(writer, "{}", self.get_msg()); } } +} - pub fn write_above_status( - &mut self, - buf: &[u8], - writer: &mut T, - ) -> std::io::Result { - if STDIN_CLOBBER_LOCK.available_permits() != 1 { - // skip - return Ok(0); +pub async fn status_tick_worker(mut rx: UnboundedReceiver, show_progress: bool) { + let mut status = Status::new(); + + status.show_progress(show_progress); + + let mut ticker = tokio::time::interval(Duration::from_secs(1)); + let mut stderr = std::io::stderr(); + let mut log_queue: VecDeque> = VecDeque::with_capacity(100); + + // A single boolean represents the "taken over" state, where stdin is being + // accepted from the user. A "depth" is not used as it is expected the + // callers of `Takeover` respect the Semaphore. + // + // If there was ever multiple take overs at once (unlikely), this code would + // need to be updated to track multiple takeovers at once. + let mut taken_over = false; + + loop { + tokio::select! { + Some(msg) = rx.recv() => { + match msg { + UiMessage::AddMany(names) => { + status.statuses.extend( + names + .iter() + .map(|name| (name.0.to_string(), NodeStatus::Pending)), + ); + }, + UiMessage::SetStatus(name, value) => { + status.statuses.insert(name.0.to_string(), value); + }, + UiMessage::Takeover(tx) => { + taken_over = true; + status.wipe_out(&mut stderr); + let _ = tx.send(()); + }, + UiMessage::Release => { + taken_over = false; + for buf in log_queue.drain(..) { + let _ = std::io::Write::write_all(&mut stderr, &buf); + } + status.write_status(&mut stderr); + }, + UiMessage::Clear => { + status.clear(&mut stderr); + }, + UiMessage::LogLine(line) => { + if taken_over { + log_queue.push_back(line); + } else { + status.clear(&mut stderr); + for buf in log_queue.drain(..) { + let _ = std::io::Write::write_all(&mut stderr, &buf); + } + let _ = std::io::Write::write_all(&mut stderr, &line); + status.write_status(&mut stderr); + } + }, + } + } + + _ = ticker.tick() => { + if taken_over { + continue; + } + + status.clear(&mut stderr); + status.write_status(&mut stderr); + } } - - self.clear(writer); - let written = writer.write(buf)?; - self.write_status(writer); - - Ok(written) } } diff --git a/crates/core/src/hive/executor.rs b/crates/core/src/hive/executor.rs --- a/crates/core/src/hive/executor.rs +++ b/crates/core/src/hive/executor.rs @@ -1,4 +1,7 @@ -use crate::hive::node::Step; +use crate::{ + hive::node::Step, + status::{NodeStatus, UI_SENDER, UiMessage}, +}; use std::{assert_matches::debug_assert_matches, sync::Arc}; use tracing::{Instrument, Span, debug, error, event, instrument}; @@ -12,7 +15,6 @@ node::{Context, Derivation, ExecuteStep, Name}, plan::NodePlan, }, - status::STATUS, }; /// returns Err if the application should shut down. @@ -96,9 +98,12 @@ progress = format!("{}/{length}", position + 1) ); - STATUS - .lock() - .set_node_step(&plan.context.name, step.to_string()); + if let Some(tx) = UI_SENDER.get() { + let _ = tx.send(UiMessage::SetStatus( + plan.context.name.clone(), + NodeStatus::Running(step.to_string()), + )); + } if let Err(err) = step.execute(&mut plan.context).await.inspect_err(|_| { error!("Failed to execute `{step}`"); @@ -107,13 +112,23 @@ return Ok(()); } - STATUS.lock().mark_node_failed(&plan.context.name); + if let Some(tx) = UI_SENDER.get() { + let _ = tx.send(UiMessage::SetStatus( + plan.context.name.clone(), + NodeStatus::Failed, + )); + } return Err(err); } } - STATUS.lock().mark_node_succeeded(&plan.context.name); + if let Some(tx) = UI_SENDER.get() { + let _ = tx.send(UiMessage::SetStatus( + plan.context.name.clone(), + NodeStatus::Succeeded, + )); + } Ok(()) } diff --git a/crates/core/src/commands/pty/mod.rs b/crates/core/src/commands/pty/mod.rs --- a/crates/core/src/commands/pty/mod.rs +++ b/crates/core/src/commands/pty/mod.rs @@ -3,7 +3,7 @@ use crate::commands::pty::output::{WatchStdoutArguments, handle_pty_stdout}; use crate::hive::node::SharedTarget; -use crate::status::STATUS; +use crate::status::{UI_SENDER, UiMessage}; use aho_corasick::PatternID; use itertools::Itertools; use nix::sys::termios::{LocalFlags, SetArg, Termios, tcgetattr, tcsetattr}; @@ -12,7 +12,6 @@ use portable_pty::{CommandBuilder, NativePtySystem, PtyPair, PtySize}; use rand::distr::Alphabetic; use std::collections::VecDeque; -use std::io::stderr; use std::sync::{LazyLock, Mutex}; use std::{ io::{Read, Write}, @@ -262,14 +261,15 @@ "localhost (!)".to_string() }; - let _ = STATUS.lock().write_above_status( - &format!( - "{target_display} | Authenticate for \"sudo {}\":\n", - arguments.command_string.as_ref() - ) - .into_bytes(), - &mut stderr(), - ); + if let Some(tx) = UI_SENDER.get() { + let _ = tx.send(UiMessage::LogLine( + format!( + "{target_display} | Authenticate for \"sudo {}\":\n", + arguments.command_string.as_ref() + ) + .into_bytes(), + )); + } Ok(()) } -- tangled.sh