rigtop_core/
source.rs

1//! Data sources: anything that can produce [`Snapshot`]s.
2//!
3//! Front-ends only talk to the [`DataSource`] trait. The local machine is one
4//! implementation ([`LocalSource`]); a remote host reached through
5//! `ssh host rigtop --agent` will be another, so remote monitoring needs no
6//! changes in the front-ends.
7
8use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
9
10use crate::config::Config;
11use crate::gpu::NvmlCollector;
12use crate::smart::SmartReader;
13#[cfg(target_os = "linux")]
14use crate::snapshot::device_matches_disk;
15use crate::snapshot::{PowerStats, Snapshot};
16use crate::system::SystemCollector;
17
18#[cfg(target_os = "linux")]
19use crate::linux::{
20    self, SysRoots, diskstats::DiskStatsMeter, net::NetLinkReader, rapl::RaplMeter,
21};
22
23/// Shortest sensible interval between two polls. CPU usage is computed from
24/// the difference between polls and is inaccurate below this.
25pub const MIN_POLL_INTERVAL: Duration = sysinfo::MINIMUM_CPU_UPDATE_INTERVAL;
26
27/// Errors a data source can report.
28#[derive(Debug, thiserror::Error)]
29pub enum SourceError {
30    /// Reading from the underlying channel failed.
31    #[error("I/O error: {0}")]
32    Io(#[from] std::io::Error),
33    /// The source is no longer available (e.g. the SSH connection closed).
34    #[error("source disconnected: {0}")]
35    Disconnected(String),
36    /// Sending a signal to a process failed (no such process, permission denied, ...).
37    #[error("cannot signal process {pid}: {source}")]
38    Signal {
39        /// Target process id.
40        pid: u32,
41        /// Underlying OS error.
42        source: std::io::Error,
43    },
44}
45
46/// Signals a front-end can send to a process.
47#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
48pub enum ProcessSignal {
49    /// Ask the process to exit (SIGTERM).
50    Terminate,
51    /// Force the process to exit (SIGKILL).
52    Kill,
53}
54
55/// A producer of snapshots, local or remote.
56pub trait DataSource {
57    /// Takes a new measurement.
58    ///
59    /// Rates (CPU %, throughput, power) cover the time since the previous
60    /// call, so callers should poll at a steady interval of at least
61    /// [`MIN_POLL_INTERVAL`].
62    fn poll(&mut self) -> Result<Snapshot, SourceError>;
63
64    /// Sends `signal` to process `pid` on the monitored host, with the
65    /// privileges of the user the source runs as.
66    fn signal_process(&mut self, pid: u32, signal: ProcessSignal) -> Result<(), SourceError>;
67}
68
69/// Collects snapshots of the machine rigtop runs on.
70pub struct LocalSource {
71    system: SystemCollector,
72    nvml: NvmlCollector,
73    smart: SmartReader,
74    last_poll: Option<Instant>,
75    #[cfg(target_os = "linux")]
76    roots: SysRoots,
77    #[cfg(target_os = "linux")]
78    rapl: RaplMeter,
79    #[cfg(target_os = "linux")]
80    diskstats: DiskStatsMeter,
81    #[cfg(target_os = "linux")]
82    net_links: NetLinkReader,
83    /// Directories searched for btrfs scrub status files, in order.
84    #[cfg(target_os = "linux")]
85    scrub_dirs: Vec<std::path::PathBuf>,
86}
87
88impl LocalSource {
89    /// Creates a source with the default configuration. See [`LocalSource::with_config`].
90    pub fn new() -> Self {
91        Self::with_config(&Config::default())
92    }
93
94    /// Creates a source and takes a warm-up sample, so that the first
95    /// [`DataSource::poll`] (after at least [`MIN_POLL_INTERVAL`]) already
96    /// has rates.
97    ///
98    /// Only collection-related settings (the SMART directory) are used here;
99    /// display settings such as sensor labels are applied by front-ends.
100    pub fn with_config(config: &Config) -> Self {
101        let mut source = Self {
102            system: SystemCollector::new(),
103            nvml: NvmlCollector::new(),
104            smart: SmartReader::new(&config.smart.dir),
105            last_poll: None,
106            #[cfg(target_os = "linux")]
107            roots: SysRoots::default(),
108            #[cfg(target_os = "linux")]
109            rapl: RaplMeter::new(),
110            #[cfg(target_os = "linux")]
111            diskstats: DiskStatsMeter::new(),
112            #[cfg(target_os = "linux")]
113            net_links: NetLinkReader::new(),
114            // The copy made by the root collector first, then btrfs-progs'
115            // own file in case rigtop runs as root.
116            #[cfg(target_os = "linux")]
117            scrub_dirs: vec![
118                config.btrfs.scrub_dir.clone(),
119                linux::btrfs::BTRFS_PROGS_STATUS_DIR.into(),
120            ],
121        };
122        // The warm-up result itself is meaningless (no previous sample).
123        let _ = source.collect();
124        source
125    }
126
127    /// Why NVIDIA GPUs are unavailable, if NVML failed to initialize.
128    pub fn gpu_unavailable_reason(&self) -> Option<&str> {
129        self.nvml.init_error()
130    }
131
132    fn collect(&mut self) -> Snapshot {
133        let now = Instant::now();
134        let elapsed = self
135            .last_poll
136            .map(|prev| now.saturating_duration_since(prev).as_secs_f64());
137        self.last_poll = Some(now);
138
139        self.system.refresh();
140        let processes = self.system.processes(elapsed);
141        let gpus = self.nvml.collect();
142        let gpus_w = sum_present(gpus.iter().map(|g| g.power_w));
143
144        let mut snapshot = Snapshot {
145            timestamp_ms: SystemTime::now()
146                .duration_since(UNIX_EPOCH)
147                .map(|d| d.as_millis() as u64)
148                .unwrap_or(0),
149            host: self.system.host(),
150            cpu: self.system.cpu(),
151            memory: self.system.memory(),
152            networks: self.system.networks(elapsed),
153            filesystems: self.system.filesystems(),
154            process_count: processes.len(),
155            processes,
156            gpus,
157            smart: self.smart.read(),
158            power: PowerStats {
159                gpus_w,
160                ..Default::default()
161            },
162            ..Default::default()
163        };
164
165        #[cfg(target_os = "linux")]
166        self.collect_linux(&mut snapshot, now);
167        #[cfg(not(target_os = "linux"))]
168        self.collect_portable(&mut snapshot, elapsed);
169
170        let power = &mut snapshot.power;
171        power.total_w = sum_present([power.cpu_package_w, power.gpus_w]);
172        snapshot
173    }
174
175    /// Adds what `sysinfo` offers beyond the basics on non-Linux platforms:
176    /// per-volume I/O, temperatures and (on Windows) link speeds.
177    #[cfg(not(target_os = "linux"))]
178    fn collect_portable(&mut self, snapshot: &mut Snapshot, elapsed: Option<f64>) {
179        snapshot.disks = self.system.volume_io(elapsed);
180        let (chips, cpu_temperatures) = self.system.components();
181        snapshot.sensors = chips;
182        snapshot.cpu.temperatures = cpu_temperatures;
183        #[cfg(windows)]
184        crate::windows::annotate_links(&mut snapshot.networks);
185    }
186
187    /// Adds sensors, power, per-device I/O and RAID state from `/sys` and `/proc`.
188    #[cfg(target_os = "linux")]
189    fn collect_linux(&mut self, snapshot: &mut Snapshot, now: Instant) {
190        let chips = linux::hwmon::read_chips(&self.roots);
191
192        snapshot.cpu.temperatures = chips
193            .iter()
194            .filter(|c| linux::hwmon::is_cpu_chip(&c.chip))
195            .flat_map(|c| c.chip.temperatures.iter().cloned())
196            .collect();
197        snapshot.memory.cached_bytes = linux::meminfo::read_cached_bytes(&self.roots);
198        snapshot.power.cpu_package_w = self.rapl.sample(&self.roots, now);
199        // amdgpu power is treated as the integrated GPU (Ryzen iGPU).
200        // Discrete AMD GPUs are not distinguished yet.
201        snapshot.power.igpu_w = sum_present(
202            chips
203                .iter()
204                .filter(|c| c.chip.name == "amdgpu")
205                .flat_map(|c| c.power_w.iter().map(|(_, w)| Some(*w))),
206        );
207
208        let mut disks = self.diskstats.sample(&self.roots, now);
209        for disk in &mut disks {
210            disk.temperature_c = nvme_temperature(&chips, &disk.name);
211        }
212        snapshot.disks = disks;
213
214        let mounts = linux::mounts::read_mounts(&self.roots);
215        snapshot.raid.btrfs = linux::btrfs::read_filesystems(&self.roots, &mounts);
216        for fs in &mut snapshot.raid.btrfs {
217            fs.scrub = linux::btrfs::read_scrub(&self.scrub_dirs, &fs.fsid);
218        }
219        self.net_links
220            .annotate(&self.roots, &mut snapshot.networks, now);
221        snapshot.raid.md = linux::mdstat::read_arrays(&self.roots);
222
223        snapshot.sensors = chips.into_iter().map(|c| c.chip).collect();
224    }
225}
226
227impl Default for LocalSource {
228    fn default() -> Self {
229        Self::new()
230    }
231}
232
233impl DataSource for LocalSource {
234    fn poll(&mut self) -> Result<Snapshot, SourceError> {
235        Ok(self.collect())
236    }
237
238    fn signal_process(&mut self, pid: u32, signal: ProcessSignal) -> Result<(), SourceError> {
239        send_signal(pid, signal).map_err(|source| SourceError::Signal { pid, source })
240    }
241}
242
243/// Sends a signal to a local process.
244#[cfg(unix)]
245fn send_signal(pid: u32, signal: ProcessSignal) -> std::io::Result<()> {
246    // pid 0 and negative values address process groups in kill(2); never
247    // let a bad pid turn into "signal everything".
248    let pid = libc::pid_t::try_from(pid)
249        .ok()
250        .filter(|p| *p > 0)
251        .ok_or_else(|| std::io::Error::from(std::io::ErrorKind::InvalidInput))?;
252    let sig = match signal {
253        ProcessSignal::Terminate => libc::SIGTERM,
254        ProcessSignal::Kill => libc::SIGKILL,
255    };
256    // SAFETY: kill(2) has no memory-safety preconditions.
257    if unsafe { libc::kill(pid, sig) } == 0 {
258        Ok(())
259    } else {
260        Err(std::io::Error::last_os_error())
261    }
262}
263
264#[cfg(windows)]
265use crate::windows::send_signal;
266
267/// Sends a signal to a local process (not implemented on this platform yet).
268#[cfg(not(any(unix, windows)))]
269fn send_signal(_pid: u32, _signal: ProcessSignal) -> std::io::Result<()> {
270    Err(std::io::Error::from(std::io::ErrorKind::Unsupported))
271}
272
273/// Sums the values that are present; `None` if none are.
274fn sum_present(values: impl IntoIterator<Item = Option<f32>>) -> Option<f32> {
275    values
276        .into_iter()
277        .flatten()
278        .fold(None, |acc, v| Some(acc.unwrap_or(0.0) + v))
279}
280
281/// Finds the composite temperature of the NVMe controller backing `disk`
282/// (hwmon chip key `nvme:nvme0` for disk `nvme0n1`).
283#[cfg(target_os = "linux")]
284fn nvme_temperature(chips: &[linux::hwmon::HwmonChip], disk: &str) -> Option<f32> {
285    chips.iter().find_map(|c| {
286        let controller = c.chip.key.strip_prefix("nvme:")?;
287        device_matches_disk(controller, disk).then_some(())?;
288        // The first channel is the "Composite" temperature the drive reports.
289        c.chip.temperatures.first().map(|t| t.celsius)
290    })
291}
292
293#[cfg(test)]
294mod tests {
295    use super::*;
296
297    #[cfg(unix)]
298    #[test]
299    fn signal_rejects_group_pids_and_reports_missing_process() {
300        assert_eq!(
301            send_signal(0, ProcessSignal::Terminate).unwrap_err().kind(),
302            std::io::ErrorKind::InvalidInput
303        );
304        assert_eq!(
305            send_signal(u32::MAX, ProcessSignal::Kill)
306                .unwrap_err()
307                .kind(),
308            std::io::ErrorKind::InvalidInput
309        );
310        // A child that has been reaped no longer exists: ESRCH.
311        let mut child = std::process::Command::new("true").spawn().unwrap();
312        let pid = child.id();
313        child.wait().unwrap();
314        let err = send_signal(pid, ProcessSignal::Terminate).unwrap_err();
315        assert_eq!(err.raw_os_error(), Some(libc::ESRCH));
316    }
317
318    #[cfg(unix)]
319    #[test]
320    fn signal_terminates_a_child_process() {
321        let mut child = std::process::Command::new("sleep")
322            .arg("30")
323            .spawn()
324            .unwrap();
325        send_signal(child.id(), ProcessSignal::Kill).unwrap();
326        let status = child.wait().unwrap();
327        assert!(!status.success());
328    }
329
330    #[test]
331    fn sum_present_ignores_missing_values() {
332        assert_eq!(sum_present([None, None]), None);
333        assert_eq!(sum_present([Some(1.5), None, Some(2.0)]), Some(3.5));
334    }
335
336    #[cfg(target_os = "linux")]
337    #[test]
338    fn nvme_temperature_matches_controller_exactly() {
339        use crate::linux::hwmon::HwmonChip;
340        use crate::snapshot::{SensorChip, Temperature};
341
342        let chip = |key: &str, celsius: f32| HwmonChip {
343            chip: SensorChip {
344                key: key.to_owned(),
345                name: "nvme".to_owned(),
346                temperatures: vec![Temperature {
347                    label: "Composite".to_owned(),
348                    celsius,
349                    ..Default::default()
350                }],
351                fans: Vec::new(),
352            },
353            power_w: Vec::new(),
354        };
355        let chips = vec![chip("nvme:nvme1", 40.0), chip("nvme:nvme10", 50.0)];
356        assert_eq!(nvme_temperature(&chips, "nvme10n1"), Some(50.0));
357        assert_eq!(nvme_temperature(&chips, "nvme1n1"), Some(40.0));
358        assert_eq!(nvme_temperature(&chips, "sda"), None);
359    }
360}
rigtop_core/source.rs

rigtop_core/
source.rs
