git_policy_cli/process/

thread_quiet.rs

//! What:
//!  Whether one thread of this process is at work, judged with the signs the push meter reads for a child process, and the
//!  clock that says when the thread has shown none of them for a window.
//! Why:
//!  The Markdown linter runs on a thread of this process and the gateway call of the standards review blocks one, so neither
//!  has a child process whose tree `push_progress.rs` could sample. Linux keeps the same counters for each thread under
//!  `/proc/self/task/<tid>/`: the bytes it read and wrote (`io`), and its processor ticks and state (`stat`). The thread being
//!  judged is never the thread that reads these files, so the reading does not count as the thread's work.
//!  Where `/proc/self/task/<tid>/io` or `stat` cannot be read nothing is measured and the thread is never judged quiet.
//!
//! In TS you'd write (pseudocode):
//! ```ts
//! const clock = new ThreadClock(tid, windowMs); while (!done) { if (clock.quiet()) abandon(); await sleep(SAMPLE_MS); }
//! ```

/// The parsing of a `stat` line and the state letters that count as work.
use crate::push_progress::{SAMPLE_INTERVAL, is_working, parse_activity};
/// The connected sockets of the machine, and those a process holds.
use crate::tcp_sockets::{ESTABLISHED, established_sockets, socket_inodes_of};
/// What each socket had carried when it was last seen.
use std::collections::HashMap;
/// Shared flags between the thread that judges and the one that waits.
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
/// Time spans and the monotonic clock.
use std::time::{Duration, Instant};

/// What:
///  The kernel's number for the calling thread, which names its files under `/proc/self/task/`.
/// Why:
///   `std` does not expose it; the system call takes no arguments and cannot fail.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// const tid = threadId();
/// ```
#[cfg(target_os = "linux")]
pub fn current_thread_id() -> Option<i32> {
    // SAFETY: `gettid` takes no arguments, touches no memory and cannot fail.
    let raw: libc::c_long = unsafe { libc::syscall(libc::SYS_gettid) };
    return i32::try_from(raw).ok();
}

/// Only Linux names a thread by a number that `/proc` understands; elsewhere nothing is measured.
// UNTESTED ON WINDOWS: no gate compiles or runs this branch (the crate does not build for Windows; README, Requirements).
#[cfg(not(target_os = "linux"))]
pub fn current_thread_id() -> Option<i32> {
    return None;
}

/// The bytes a thread has read and written, from its `io` file, or nothing where the file cannot be read.
#[cfg(target_os = "linux")]
fn thread_bytes(tid: i32) -> Option<u64> {
    let text: String = std::fs::read_to_string(format!("/proc/self/task/{tid}/io")).ok()?;
    let mut read: Option<u64> = None;
    let mut written: Option<u64> = None;
    for line in text.lines() {
        if let Some(value) = line.strip_prefix("rchar: ") {
            read = value.trim().parse::<u64>().ok();
        } else if let Some(value) = line.strip_prefix("wchar: ") {
            written = value.trim().parse::<u64>().ok();
        }
    }
    return Some(read?.saturating_add(written?));
}

/// Only Linux counts a thread's bytes; elsewhere nothing is measured.
// UNTESTED ON WINDOWS: no gate compiles or runs this branch (the crate does not build for Windows; README, Requirements).
#[cfg(not(target_os = "linux"))]
fn thread_bytes(_tid: i32) -> Option<u64> {
    return None;
}

/// The processor ticks of a thread and whether it is able to run or waiting for the disk, or nothing where `stat` cannot be read.
#[cfg(target_os = "linux")]
fn thread_activity(tid: i32) -> Option<(u64, bool)> {
    let text: String = std::fs::read_to_string(format!("/proc/self/task/{tid}/stat")).ok()?;
    let (state, ticks): (char, u64) = parse_activity(text.as_str())?;
    return Some((ticks, is_working(state)));
}

/// Only Linux reads what a thread is doing; elsewhere nothing is.
// UNTESTED ON WINDOWS: no gate compiles or runs this branch (the crate does not build for Windows; README, Requirements).
#[cfg(not(target_os = "linux"))]
fn thread_activity(_tid: i32) -> Option<(u64, bool)> {
    return None;
}

/// What:
///  What one thread had done when it was last looked at, and the look that says whether it did more since.
/// Why:
///   Bytes and ticks only grow, so a change since the last look is a sign of work; a thread in a state able to run or waiting
///  for the disk is at work now. The connected sockets of the process count too, when asked: an HTTP client moves its bytes
///  with `send` and `recv`, which the thread's `rchar` and `wchar` do not see.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// class ThreadSigns { tid: number; look(): boolean }
/// ```
#[derive(Debug)]
pub struct ThreadSigns {
    /// The thread's number.
    tid: i32,
    /// The bytes and processor ticks at the last look.
    before: Option<(u64, u64)>,
    /// Whether the process's connected sockets count as the thread's work.
    count_sockets: bool,
    /// The bytes each connected socket had carried at the last look, by inode.
    sockets: HashMap<u64, u64>,
    /// Whether anything about the thread could be read at the last look.
    measured: bool,
}

impl ThreadSigns {
    /// Signs of the thread `tid`; `count_sockets` adds the process's connected sockets.
    pub fn new(tid: i32, count_sockets: bool) -> Self {
        return Self { tid, before: None, count_sockets, sockets: HashMap::new(), measured: false };
    }

    /// Whether the last look could read the thread at all, so a quiet thread is quiet and not unreadable.
    pub fn measured(&self) -> bool {
        return self.measured;
    }

    /// The bytes the process's connected sockets carried since the last look, or zero where they cannot be listed.
    fn socket_bytes_moved(&mut self) -> u64 {
        let held: Vec<u64> = socket_inodes_of(i32::try_from(std::process::id()).unwrap_or(i32::MAX));
        let Some(connected) = established_sockets() else {
            return 0;
        };
        let mut moved: u64 = 0;
        for socket in connected {
            if socket.state != ESTABLISHED || !held.contains(&socket.inode) {
                continue;
            }
            let total: u64 = socket.bytes_acked.saturating_add(socket.bytes_received);
            let seen: u64 = self.sockets.insert(socket.inode, total).unwrap_or(total);
            moved = moved.saturating_add(total.saturating_sub(seen));
        }
        return moved;
    }

    /// Look at the thread once: true when it moved bytes or used processor time since the last look, or is at work now.
    ///  The first look is a sign of work, because there is nothing to compare with.
    pub fn look(&mut self) -> bool {
        let bytes: Option<u64> = thread_bytes(self.tid);
        let activity: Option<(u64, bool)> = thread_activity(self.tid);
        self.measured = bytes.is_some() || activity.is_some();
        if !self.measured {
            return false;
        }
        let now: (u64, u64) = (bytes.unwrap_or(0), activity.map_or(0, |(ticks, _)| { return ticks; }));
        let working: bool = activity.is_some_and(|(_, at_work)| { return at_work; });
        let changed: bool = self.before.is_none_or(|(old_bytes, old_ticks)| { return now.0 > old_bytes || now.1 > old_ticks; });
        self.before = Some(now);
        let carried: bool = self.count_sockets && self.socket_bytes_moved() > 0;
        return changed || working || carried;
    }
}

/// What:
///  The clock of one thread: it looks at the thread's signs at the sampling interval and says when none has shown for a window.
/// Why:
///   The waiting thread asks every few milliseconds; the counters are read only every `SAMPLE_INTERVAL`, as for a child process.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// class ThreadClock { quiet(): boolean; longestQuiet(): number }
/// ```
#[derive(Debug)]
pub struct ThreadClock {
    /// What the thread has done.
    signs: ThreadSigns,
    /// How long the thread may show no sign of work.
    window: Duration,
    /// When a sign of work was last seen (the start, before the first).
    last_progress: Instant,
    /// When the counters are next read.
    next_sample: Instant,
    /// The longest stretch between two looks that found a sign of work.
    longest_quiet: Duration,
}

impl ThreadClock {
    /// Start the clock of the thread `tid` now; `count_sockets` adds the process's connected sockets to its signs.
    pub fn start(tid: i32, window: Duration, count_sockets: bool) -> Self {
        let now: Instant = Instant::now();
        return Self { signs: ThreadSigns::new(tid, count_sockets), window, last_progress: now, next_sample: now, longest_quiet: Duration::ZERO };
    }

    /// Read the signs when a sample is due, and say whether the thread has now been quiet for the whole window.
    ///  A thread that cannot be read is never judged quiet.
    pub fn quiet(&mut self) -> bool {
        let now: Instant = Instant::now();
        if now >= self.next_sample {
            if self.signs.look() {
                self.longest_quiet = self.longest_quiet.max(now.saturating_duration_since(self.last_progress));
                self.last_progress = now;
            }
            self.next_sample = now + SAMPLE_INTERVAL;
        }
        return self.signs.measured() && now.saturating_duration_since(self.last_progress) >= self.window;
    }

    /// The longest stretch with no sign of work so far, the one still running included.
    pub fn longest_quiet(&self) -> Duration {
        return self.longest_quiet.max(Instant::now().saturating_duration_since(self.last_progress));
    }
}

/// What:
///  Run `job` on this thread while another thread measures how quiet this thread gets, and keep the longest gap in the ledger.
/// Why:
///   A call that ends itself (the gateway call ends on its transport's own idle timeouts) still needs its quiet gap recorded.
///  Nothing here ends the job.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// const value = await measureQuiet('gateway call', () => post(body));
/// ```
pub fn measured_quiet<T>(step: &str, job: impl FnOnce() -> T) -> T {
    let Some(tid) = current_thread_id() else {
        return job();
    };
    let done: Arc<AtomicBool> = Arc::new(AtomicBool::new(false));
    let finished: Arc<AtomicBool> = Arc::clone(&done);
    let sampler: Option<std::thread::JoinHandle<Duration>> = std::thread::Builder::new()
        .name(String::from("cli-git-quiet-measure"))
        .spawn(move || {
            let mut clock: ThreadClock = ThreadClock::start(tid, Duration::MAX, true);
            while !finished.load(Ordering::Acquire) {
                // The result is unused: this clock only measures, and its window is never reached.
                let _quiet = clock.quiet();
                std::thread::sleep(MEASURE_PAUSE);
            }
            return clock.longest_quiet();
        })
        .ok();
    let value: T = job();
    done.store(true, Ordering::Release);
    if let Some(handle) = sampler
        && let Ok(gap) = handle.join()
    {
        crate::quiet_ledger::note(step, gap, false);
    }
    return value;
}

/// How long the measuring thread sleeps between two looks at its clock.
const MEASURE_PAUSE: Duration = Duration::from_millis(20);

/// The thread's signs are controlled in the test module.
#[cfg(test)]
#[path = "thread_quiet_tests.rs"]
mod tests;