use crate::child_class::{CHILD_WAIT_VARIABLE, ChildClass};
use crate::wait_ceilings::OBJECT_STEP_CEILING;
use crate::output_capture::Capture;
use crate::start_retry::RetryStart;
use crate::wait_progress::{Cadence, ChildWaitFacts, Ticker, child_wait_line};
use crate::bounded_push::{Patience, StopCause};
use crate::push_progress::{Clock, SAMPLE_INTERVAL};
use crate::quiet_ledger;
use std::process::{Child, Command, Output, Stdio};
use std::io::Write;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender, channel};
use crate::stopped_git_locks::LockLedger;
use crate::bounded_push::STOP_GRACE;
use crate::process_tree::{TreeMember, descendants_of, kill_members, member_of, still_running, terminate_members};
use std::thread::JoinHandle;
use std::time::{Duration, Instant};
const SHORTEST_SLEEP: Duration = Duration::from_millis(1);
pub struct WatchTerms {
pub limit: Duration,
pub cadence: Cadence,
pub notices: Box<dyn Write + Send>,
}
impl WatchTerms {
pub fn for_class(class: ChildClass) -> Self {
let interval: Cadence = Cadence::from_environment();
let limit: Duration = class.limit();
return Self {
limit,
cadence: Cadence { grace: grace_for(class, interval.interval, limit), interval: interval.interval },
notices: Box::new(std::io::stderr()),
};
}
pub fn from_environment() -> Self {
return Self::for_class(ChildClass::Object);
}
}
pub fn grace_for(class: ChildClass, interval: Duration, limit: Duration) -> Duration {
return class.notice_grace().min(interval).min(limit / 2);
}
enum Mark {
Begin(Instant),
End,
}
const STOP_POLL: Duration = Duration::from_millis(10);
struct Marks {
since: Option<Instant>,
ticker: Ticker,
printed: u32,
clock: Option<Clock>,
pid: u32,
window: Duration,
longest: Arc<AtomicU64>,
}
impl Marks {
fn apply(&mut self, mark: Mark, cadence: Cadence) {
match mark {
Mark::Begin(at) => {
self.since = Some(at);
self.ticker = Ticker::new(cadence);
self.printed = 0;
self.clock = Some(Clock::start(self.pid, Patience::Progress { stall: self.window }).without_wall_fallback());
}
Mark::End => {
self.keep_gap();
self.since = None;
self.clock = None;
}
}
}
fn keep_gap(&self) {
if let Some(clock) = self.clock.as_ref().filter(|measuring| { return measuring.measures(); }) {
let micros: u64 = u64::try_from(clock.longest_quiet().as_micros()).unwrap_or(u64::MAX);
self.longest.fetch_max(micros, Ordering::AcqRel);
}
}
}
#[cfg(unix)]
fn stop_process(pid: u32, locks: &LockLedger, incoming: &Receiver<Mark>) {
let Ok(raw) = i32::try_from(pid) else {
return;
};
let below: Vec<TreeMember> = descendants_of(raw);
let root: Option<TreeMember> = member_of(raw);
let mut tree: Vec<TreeMember> = root.iter().cloned().collect();
tree.extend(below.iter().cloned());
let identified: bool = cfg!(target_os = "linux");
let signal_root = |signal: libc::c_int| {
if root.as_ref().map_or(!identified, still_running) {
unsafe {
libc::kill(raw, signal);
}
}
};
signal_root(libc::SIGTERM);
terminate_members(below.as_slice());
let asked: Instant = Instant::now();
while asked.elapsed() < STOP_GRACE && tree.iter().any(still_running) {
std::thread::sleep(STOP_POLL);
}
signal_root(libc::SIGKILL);
kill_members(below.as_slice());
while tree.iter().any(still_running) {
match incoming.recv_timeout(STOP_POLL) {
Err(RecvTimeoutError::Disconnected) => break,
Ok(_) | Err(RecvTimeoutError::Timeout) => {}
}
}
locks.sweep(tree.as_slice());
}
#[cfg(not(unix))]
fn stop_process(_pid: u32, _locks: &LockLedger, _incoming: &Receiver<Mark>) {}
struct WatchFlags {
expired: Arc<AtomicBool>,
stoppable: Arc<AtomicBool>,
longest: Arc<AtomicU64>,
ceiling: Arc<AtomicU64>,
ceiling_hit: Arc<AtomicBool>,
}
pub struct ChildWatch {
marks: Option<Sender<Mark>>,
expired: Arc<AtomicBool>,
thread: Option<JoinHandle<()>>,
limit: Duration,
stoppable: Arc<AtomicBool>,
longest: Arc<AtomicU64>,
ledger_step: Option<String>,
ceiling: Arc<AtomicU64>,
ceiling_hit: Arc<AtomicBool>,
}
impl ChildWatch {
pub fn start(waiting_for: &str, pid: u32, terms: WatchTerms) -> Self {
return Self::start_with_locks(waiting_for, pid, terms, LockLedger::none());
}
pub fn start_with_locks(waiting_for: &str, pid: u32, terms: WatchTerms, locks: LockLedger) -> Self {
let (marks, incoming): (Sender<Mark>, Receiver<Mark>) = channel();
let expired: Arc<AtomicBool> = Arc::new(AtomicBool::new(false));
let stoppable: Arc<AtomicBool> = Arc::new(AtomicBool::new(true));
let longest: Arc<AtomicU64> = Arc::new(AtomicU64::new(0));
let ceiling: Arc<AtomicU64> = Arc::new(AtomicU64::new(u64::MAX));
let ceiling_hit: Arc<AtomicBool> = Arc::new(AtomicBool::new(false));
let flags: WatchFlags = WatchFlags {
expired: Arc::clone(&expired),
stoppable: Arc::clone(&stoppable),
longest: Arc::clone(&longest),
ceiling: Arc::clone(&ceiling),
ceiling_hit: Arc::clone(&ceiling_hit),
};
let subject: String = String::from(waiting_for);
let limit: Duration = terms.limit;
let thread: Option<JoinHandle<()>> = std::thread::Builder::new()
.name(String::from("cli-git-child-watch"))
.spawn(move || {
watch_loop(&incoming, &flags, subject.as_str(), pid, terms, &locks);
})
.ok();
return Self { marks: Some(marks), expired, thread, limit, stoppable, longest, ledger_step: None, ceiling, ceiling_hit };
}
pub fn kept_as(mut self, step: &str) -> Self {
self.ledger_step = Some(String::from(step));
return self;
}
pub fn never_stop(&self) {
self.stoppable.store(false, Ordering::Release);
}
pub fn within(self, ceiling: Duration) -> Self {
self.ceiling.store(u64::try_from(ceiling.as_micros()).unwrap_or(u64::MAX - 1), Ordering::Release);
return self;
}
pub fn ceiling_hit(&self) -> bool {
return self.ceiling_hit.load(Ordering::Acquire);
}
fn send(&self, mark: Mark) {
if let Some(marks) = &self.marks {
let _sent = marks.send(mark);
}
}
pub fn begin(&self) {
self.send(Mark::Begin(Instant::now()));
}
pub fn end(&self) {
self.send(Mark::End);
}
pub fn expired(&self) -> bool {
return self.expired.load(Ordering::Acquire);
}
pub fn limit(&self) -> Duration {
return self.limit;
}
}
impl Drop for ChildWatch {
fn drop(&mut self) {
drop(self.marks.take());
if let Some(thread) = self.thread.take() {
let _joined = thread.join();
}
if let Some(step) = self.ledger_step.take() {
let gap: Duration = Duration::from_micros(self.longest.load(Ordering::Acquire));
quiet_ledger::note(step.as_str(), gap, self.expired());
}
}
}
fn watch_loop(incoming: &Receiver<Mark>, flags: &WatchFlags, waiting_for: &str, pid: u32, mut terms: WatchTerms, locks: &LockLedger) {
let expired: &AtomicBool = flags.expired.as_ref();
let stoppable: &AtomicBool = flags.stoppable.as_ref();
let longest: Arc<AtomicU64> = Arc::clone(&flags.longest);
let mut marks: Marks = Marks { since: None, ticker: Ticker::new(terms.cadence), printed: 0, clock: None, pid, window: terms.limit, longest };
loop {
let Some(since) = marks.since else {
match incoming.recv() {
Ok(mark) => marks.apply(mark, terms.cadence),
Err(_) => return,
}
continue;
};
let waited: Duration = since.elapsed();
if marks.ticker.due(waited) {
let line: String = child_wait_line(&ChildWaitFacts {
waiting_for,
pid,
waited,
limit: terms.limit,
limit_note: " (CLI_GIT_CHILD_WAIT_SECONDS changes this)",
repeat: marks.printed > 0,
ends_when_quiet: stoppable.load(Ordering::Acquire),
});
marks.printed += 1;
let _written = writeln!(terms.notices, "{line}");
while let Ok(mark) = incoming.try_recv() {
marks.apply(mark, terms.cadence);
}
if marks.since != Some(since) {
continue;
}
}
let ceiling: Duration = Duration::from_micros(flags.ceiling.load(Ordering::Acquire));
if waited >= ceiling {
flags.ceiling_hit.store(true, Ordering::Release);
expired.store(true, Ordering::Release);
marks.keep_gap();
marks.since = None;
marks.clock = None;
stop_process(pid, locks, incoming);
continue;
}
let verdict: Option<StopCause> = marks.clock.as_mut().and_then(Clock::check);
let quiet: bool = verdict.is_some_and(|cause| { return matches!(cause, StopCause::Stalled(_) | StopCause::Deadline); });
if quiet && stoppable.load(Ordering::Acquire) {
expired.store(true, Ordering::Release);
marks.keep_gap();
marks.since = None;
marks.clock = None;
stop_process(pid, locks, incoming);
continue;
}
let until_notice: Duration = marks.ticker.next_in(since.elapsed());
let until_ceiling: Duration = Duration::from_micros(flags.ceiling.load(Ordering::Acquire)).saturating_sub(since.elapsed());
let sleep: Duration = until_notice.min(SAMPLE_INTERVAL).min(terms.limit).min(until_ceiling).max(SHORTEST_SLEEP);
match incoming.recv_timeout(sleep) {
Ok(mark) => marks.apply(mark, terms.cadence),
Err(RecvTimeoutError::Timeout) => {}
Err(RecvTimeoutError::Disconnected) => return,
}
}
}
pub fn reader_subject() -> String {
return format!("git cat-file --batch, {}", ChildClass::Object.label());
}
pub fn watched_timeout_text(waiting_for: &str, limit: Duration) -> String {
return format!(
"{waiting_for} showed no sign of work for {} seconds and was stopped ({CHILD_WAIT_VARIABLE} changes this). \
If the step belonged to a commit, the next Git command of this wrapper finishes or undoes that commit, so \
run the command again. If the step is quiet while it works, raise the window with {CHILD_WAIT_VARIABLE}=<seconds>; \
if it stalls again, check with `ps` what that process waits for.",
limit.as_secs()
);
}
pub fn output_watched(command: &mut Command, waiting_for: &str, terms: WatchTerms) -> std::io::Result<Output> {
return output_watched_stopping(command, waiting_for, waiting_for, terms, true);
}
pub fn output_watched_stopping(
command: &mut Command,
waiting_for: &str,
ledger_step: &str,
terms: WatchTerms,
stoppable: bool,
) -> std::io::Result<Output> {
return output_watched_within(command, waiting_for, ledger_step, terms, stoppable, StepBound::EndAt(OBJECT_STEP_CEILING));
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum StepBound {
EndAt(Duration),
LeaveRunning {
after: Duration,
extra: Duration,
},
Unbounded,
}
#[derive(Debug)]
pub struct CeilingReached(
pub String,
);
impl std::fmt::Display for CeilingReached {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
return formatter.write_str(self.0.as_str());
}
}
impl std::error::Error for CeilingReached {}
fn output_of(status: std::process::ExitStatus, captured: Option<&Capture>) -> Output {
let Some(files) = captured else {
return Output { status, stdout: Vec::new(), stderr: Vec::new() };
};
let shown: crate::output_capture::Captured = files.snapshot(crate::output_capture::STEP_OUTPUT_LIMIT);
let mut stderr: Vec<u8> = shown.stderr;
for note in &shown.notes {
stderr.extend_from_slice(format!("\ncli-git: {note}\n").as_bytes());
}
return Output { status, stdout: shown.stdout, stderr };
}
pub fn not_known_text(waiting_for: &str, after: Duration, extra: Duration) -> String {
return format!(
"{waiting_for} was still running {} seconds after it began (its bound of {} seconds and {} more for the write it started), so it \
was left running and not ended: it is not known whether it finished. Check what it changes with `git status` and `git log -1`, \
look at the process with `ps` and at the disk it writes to, and run the command again if it did not finish.",
after.saturating_add(extra).as_secs(),
after.as_secs(),
extra.as_secs()
);
}
#[derive(Debug)]
pub struct NotKnown(
pub String,
);
impl std::fmt::Display for NotKnown {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
return formatter.write_str(self.0.as_str());
}
}
impl std::error::Error for NotKnown {}
pub fn is_not_known(error: &std::io::Error) -> bool {
return error.get_ref().is_some_and(|inner| { return inner.is::<NotKnown>(); });
}
pub fn is_ceiling(error: &std::io::Error) -> bool {
return error.get_ref().is_some_and(|inner| { return inner.is::<CeilingReached>(); });
}
pub fn output_watched_within(
command: &mut Command,
waiting_for: &str,
ledger_step: &str,
terms: WatchTerms,
stoppable: bool,
bound: StepBound,
) -> std::io::Result<Output> {
let ceiling: Option<Duration> = match bound {
StepBound::EndAt(limit) => Some(limit),
StepBound::LeaveRunning { .. } | StepBound::Unbounded => None,
};
let captured: Option<Capture> = if matches!(bound, StepBound::EndAt(_)) { None } else { Some(Capture::new()?) };
match captured.as_ref() {
Some(files) => {
let (out, err) = files.stdio()?;
command.stdout(out).stderr(err);
}
None => {
command.stdout(Stdio::piped()).stderr(Stdio::piped());
}
}
let locks: LockLedger = LockLedger::before_start(command);
let child: Child = command.spawn_retrying()?;
let started: ChildWatch = ChildWatch::start_with_locks(waiting_for, child.id(), terms, locks).kept_as(ledger_step);
let watch: ChildWatch = match ceiling {
Some(limit) => started.within(limit),
None => started,
};
if !stoppable {
watch.never_stop();
}
watch.begin();
let output: std::io::Result<Output> = match bound {
StepBound::LeaveRunning { after, extra } => {
let (sender, receiver) = std::sync::mpsc::channel::<std::io::Result<std::process::ExitStatus>>();
let waiter = std::thread::Builder::new().name(String::from("cli-git-left-running-step")).spawn(move || {
let mut leaving: Child = child;
let _abandoned = sender.send(leaving.wait());
});
if let Err(error) = waiter {
watch.end();
return Err(std::io::Error::other(format!("{waiting_for} was started but no thread could be started to wait for it: {error}")));
}
match receiver.recv_timeout(after.saturating_add(extra)) {
Ok(finished) => finished.map(|status| { return output_of(status, captured.as_ref()); }),
Err(_timeout_or_gone) => {
watch.end();
if let Some(files) = captured.as_ref() {
files.discard();
}
return Err(std::io::Error::new(std::io::ErrorKind::TimedOut, NotKnown(not_known_text(waiting_for, after, extra))));
}
}
}
StepBound::EndAt(_) => child.wait_with_output(),
StepBound::Unbounded => {
let mut waited_for: Child = child;
waited_for.wait().map(|status| { return output_of(status, captured.as_ref()); })
}
};
watch.end();
if let Some(limit) = ceiling
&& watch.ceiling_hit()
{
return Err(std::io::Error::new(
std::io::ErrorKind::TimedOut,
CeilingReached(format!("{waiting_for} ran for {} seconds, the longest the wrapper lets one step run, and was ended", limit.as_secs())),
));
}
if watch.expired() {
return Err(std::io::Error::new(
std::io::ErrorKind::TimedOut,
watched_timeout_text(waiting_for, watch.limit()),
));
}
return output;
}
#[cfg(test)]
#[path = "child_watch_tests.rs"]
mod tests;