Repo sync

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "aperf"
-version = "1.2.3"
+version = "1.3.0"
 edition = "2021"
 
 publish = false
@@ -32,13 +32,11 @@ clap_complete = "4.2.3"
 serde = { version = "1.0", features = ["derive"] }
 chrono = { version = "0.4", features = ["serde"] }
 rustix = { version = "0.38.28", features = ["system"] }
-serde_yaml = "0.9"
 thiserror = "1.0"
 log = "0.4.21"
 lazy_static = "1.4.0"
 ctor = "0.2.6"
 anyhow = "1.0"
-serde_urlencoded = "0.7"
 serde_json = "1.0"
 itoa = "1"
 ryu = "1"
@@ -50,29 +48,27 @@ libc = "0.2"
 flate2 = "1.0.30"
 tar = "0.4.40"
 infer = "0.13.0"
-inquire = "0.7.5"
 bincode = "1.3.3"
 inferno = "0.11.19"
-indexmap = "2.1.0"
-include_directory = "0.1.1"
-cfg-if = "1.0"
+indexmap = { version = "2.1.0", features = ["serde"] }
+include_dir = "0.7.4"
 tempfile = "3"
-serial_test = "3.1.1"
 log4rs = "1.3.0"
-tdigest = "0.2"
-csv = "1.2"
 csv-to-html = "0.5.10"
 numeric-sort = "0.1"
 regex = "1"
 rmcp = { version = "1.5", features = ["server", "transport-io"], optional = true }
 schemars = { version = "0.8", optional = true }
+exmex = "0.21"
 
 [target.'cfg(target_os = "linux")'.dependencies]
 sysinfo = "0.26.2"
 sysctl = "*"
 perf-event2 = "0.7.2"
+perf-event-open-sys2 = "5.0.6"
 procfs = "0.12.0"
 timerfd = "1.6.0"
+rlimit = "0.11"
 linux-perf-data = "0.12"
 linux-perf-event-reader = "0.10"
 object = { version = "0.36", features = ["read"] }

src/data.rs

@@ -33,7 +33,7 @@ use diskstats::{Diskstats, DiskstatsRaw};
 use efa_stat::{EfaStat, EfaStatRaw};
 use ena_stat::{EnaStat, EnaStatRaw};
 use hotline::{Hotline, HotlineRaw};
-use include_directory::{include_directory, Dir};
+use include_dir::{include_dir, Dir};
 use interrupts::{InterruptData, InterruptDataRaw};
 use java_profile::{JavaProfile, JavaProfileRaw};
 use kernel_config::KernelConfig;
@@ -74,6 +74,7 @@ pub struct CollectorParams {
     pub signal: Signal,
     pub runlog: PathBuf,
     pub pmu_config: Option<PathBuf>,
+    pub pmu_counter_mode: String,
     pub perf_frequency: u32,
     pub save_profile_events: bool,
     pub hotline_frequency: u32,
@@ -95,6 +96,7 @@ impl CollectorParams {
             signal: signal::SIGTERM,
             runlog: PathBuf::new(),
             pmu_config: None,
+            pmu_counter_mode: String::new(),
             perf_frequency: 99,
             save_profile_events: false,
             hotline_frequency: 1000,
@@ -156,6 +158,7 @@ impl DataType {
         self.collector_params.perf_frequency = param.perf_frequency;
         self.collector_params.save_profile_events = param.save_profile_events;
         self.collector_params.pmu_config = param.pmu_config.clone();
+        self.collector_params.pmu_counter_mode = param.pmu_counter_mode.clone();
         self.collector_params.interval = param.interval;
         self.collector_params.hotline_frequency = param.hotline_frequency;
         self.collector_params.num_to_report = param.num_to_report;
@@ -329,7 +332,7 @@ macro_rules! data {
 /// creates the VisualizationData object and invokes the function.
 macro_rules! report_data {
     ( $( $report_data:ident ),* ) => {
-        pub static JS_DIR: Dir<'_> = include_directory!("$JS_DIR");
+        pub static JS_DIR: Dir<'_> = include_dir!("$JS_DIR");
 
         #[derive(Clone, Debug, Deserialize, Serialize)]
         pub enum ReportData {

src/data/common/processed_data_accessor.rs

@@ -528,7 +528,13 @@ fn write_series_json_string(
         if i > 0 {
             buf.push(',');
         }
-        buf.push_str(ryu_buf.format(f64_to_fixed_2(v)));
+        if v.is_finite() {
+            buf.push_str(ryu_buf.format(f64_to_fixed_2(v)));
+        } else {
+            // ryu formats non-finite floats as bare `NaN`/`inf`/`-inf` tokens, which
+            // are invalid JSON. Emit `null` instead (matching serde_json's behavior).
+            buf.push_str("null");
+        }
     }
     buf.push(']');
 
@@ -1189,6 +1195,20 @@ mod tests {
         assert_eq!(v["is_aggregate"], true);
     }
 
+    #[test]
+    fn test_write_series_non_finite_values_are_null() {
+        let series = make_series(
+            "total",
+            vec![0, 10, 20, 30, 40],
+            vec![1.5, f64::NAN, f64::INFINITY, f64::NEG_INFINITY, 2.5],
+        );
+        let mut buf = String::new();
+        write_series_json_string(&mut buf, &series, None, None, None);
+        // The whole line must be valid JSON.
+        let v: serde_json::Value = serde_json::from_str(&buf).unwrap();
+        assert_eq!(v["values"], serde_json::json!([1.5, null, null, null, 2.5]));
+    }
+
     // ---- time_series_metric_json_string tests ----
 
     #[test]

src/data/common/time_series_data_processor.rs

@@ -42,6 +42,9 @@ pub struct TimeSeriesDataProcessor {
     time_zero: Option<TimeEnum>,
     // The current time_diff to be added to every series
     cur_time_diff: u64,
+    // Whether the first accumulative value, where there is no previous value to compute the delta,
+    // should be ignored (the delta is 0) or used (the delta is the value itself).
+    ignore_first_accumulative_value: bool,
     // Map<metric_name.series_name, number of decreasing accumulative data> - used to count the
     // number of unexpected decreases of accumulative data, which are to be logged as warning
     // at the end of collection
@@ -67,11 +70,18 @@ impl TimeSeriesDataProcessor {
             per_metric_sum_count: HashMap::new(),
             time_zero,
             cur_time_diff: 0,
+            ignore_first_accumulative_value: true,
             decreasing_accumulative_data: HashMap::new(),
             fixed_value_range: None,
         }
     }
 
+    /// Force to use the accumulative value itself, instead of 0, when there is no
+    /// previous value to compute the delta.
+    pub fn use_first_accumulative_value(&mut self) {
+        self.ignore_first_accumulative_value = false;
+    }
+
     /// Override the name of the auto-generated aggregate series
     pub fn set_aggregate_series_name(&mut self, aggregate_series_name: &'static str) {
         self.aggregate_series_name = Some(aggregate_series_name);
@@ -210,7 +220,11 @@ impl TimeSeriesDataProcessor {
                 }
                 Some(data_value - prev_value)
             }
-            None => Some(0.0),
+            None => Some(if self.ignore_first_accumulative_value {
+                0.0
+            } else {
+                data_value
+            }),
         }
     }
 

src/data/common/utils.rs

@@ -6,6 +6,8 @@ use std::fs;
 use std::fs::File;
 use std::io::{BufRead, BufReader};
 use std::path::PathBuf;
+#[cfg(target_os = "linux")]
+use {anyhow::Context, log::debug};
 
 use crate::data::common::data_formats::{Graph, GraphData};
 
@@ -31,43 +33,137 @@ pub fn get_data_name_from_type<T>() -> &'static str {
 #[cfg(target_os = "linux")]
 #[derive(Clone, Debug)]
 pub struct CpuInfo {
-    pub vendor: String,
-    pub model_name: String,
+    pub part: Option<String>,
+    pub vendor_id: Option<String>,
+    pub model_name: Option<String>,
 }
 
 #[cfg(target_os = "linux")]
 impl CpuInfo {
-    fn new() -> Self {
-        CpuInfo {
-            vendor: String::new(),
-            model_name: String::new(),
+    pub fn new() -> Result<Self> {
+        let cpu_info_file = File::open("/proc/cpuinfo")?;
+        let cpu_info_reader = BufReader::new(cpu_info_file);
+        let mut part = None;
+        let mut vendor_id = None;
+        let mut model_name = None;
+        for line in cpu_info_reader.lines() {
+            let info_line = line?;
+            if info_line.is_empty() {
+                break;
+            }
+            let key_value: Vec<&str> = info_line.split(':').collect();
+            if key_value.len() < 2 {
+                continue;
+            }
+            let key = key_value[0].trim().to_string();
+            let value = key_value[1].trim().to_string();
+            match key.as_str() {
+                "CPU part" => part = Some(value),
+                "vendor_id" => vendor_id = Some(value),
+                "model name" => model_name = Some(value),
+                _ => continue,
+            }
         }
+
+        Ok(Self {
+            part,
+            vendor_id,
+            model_name,
+        })
+    }
+
+    pub fn is_graviton(&self) -> bool {
+        self.part.is_some()
+    }
+
+    pub fn is_graviton_5(&self) -> bool {
+        self.part.as_ref().map_or(false, |part| part == "0xd84")
+    }
+
+    pub fn is_intel(&self) -> bool {
+        self.vendor_id
+            .as_ref()
+            .map_or(false, |vendor_id| vendor_id == "GenuineIntel")
+    }
+
+    pub fn is_intel_icelake(&self) -> bool {
+        self.model_name.as_ref().map_or(false, |model_name| {
+            model_name == "Intel(R) Xeon(R) Platinum 8375C CPU @ 2.90GHz"
+        })
+    }
+
+    pub fn is_intel_sapphire_rapids(&self) -> bool {
+        self.model_name.as_ref().map_or(false, |model_name| {
+            model_name == "Intel(R) Xeon(R) Platinum 8488C"
+        })
+    }
+
+    pub fn is_amd(&self) -> bool {
+        self.vendor_id
+            .as_ref()
+            .map_or(false, |vendor_id| vendor_id == "AuthenticAMD")
+    }
+
+    pub fn is_amd_genoa(&self) -> bool {
+        self.model_name
+            .as_ref()
+            .map_or(false, |model_name| model_name == "AMD EPYC 9R14")
+    }
+
+    pub fn is_amd_milan(&self) -> bool {
+        self.model_name
+            .as_ref()
+            .map_or(false, |model_name| model_name == "AMD EPYC 7R13")
     }
 }
 
 #[cfg(target_os = "linux")]
-pub fn get_cpu_info() -> Result<CpuInfo> {
-    let file = File::open("/proc/cpuinfo")?;
-    let proc_cpuinfo = BufReader::new(file);
-    let mut cpu_info = CpuInfo::new();
-    for line in proc_cpuinfo.lines() {
-        let info_line = line?;
-        if info_line.is_empty() {
-            break;
-        }
-        let key_value: Vec<&str> = info_line.split(':').collect();
-        if key_value.len() < 2 {
+/// Return the IDs of all online CPUs by parsing /sys/devices/system/cpu/online,
+/// a comma-separated list of single CPUs and inclusive ranges, e.g. "0-1,3-5,7".
+pub fn get_online_cpu_ids() -> Result<Vec<usize>> {
+    let mut ids = Vec::new();
+    let cpu_list = fs::read_to_string("/sys/devices/system/cpu/online")?;
+    let cpu_list = cpu_list.trim();
+    if cpu_list.is_empty() {
+        return Ok(ids);
+    }
+    for part in cpu_list.split(',') {
+        let part = part.trim();
+        if part.is_empty() {
             continue;
         }
-        let key = key_value[0].trim().to_string();
-        let value = key_value[1].trim().to_string();
-        match key.as_str() {
-            "vendor_id" => cpu_info.vendor = value,
-            "model name" => cpu_info.model_name = value,
-            _ => {}
+        match part.split_once('-') {
+            Some((low, high)) => {
+                let low: usize = low.trim().parse()?;
+                let high: usize = high.trim().parse()?;
+                if high < low {
+                    bail!("invalid CPU range '{part}' in cpu list '{cpu_list}'");
+                }
+                ids.extend(low..=high);
+            }
+            None => ids.push(part.parse()?),
+        }
+    }
+    ids.sort_unstable();
+    ids.dedup();
+    Ok(ids)
+}
+
+/// Check the current fd limit and raise it if the number of required fd is larger.
+#[cfg(target_os = "linux")]
+pub fn raise_fd_limit(num_required_fds: u64) -> Result<()> {
+    let (cur_fd_limit, max_fd_limit) = rlimit::Resource::NOFILE
+        .get()
+        .context("Failed to read fd limit")?;
+    if num_required_fds > cur_fd_limit {
+        if num_required_fds >= max_fd_limit {
+            bail!("The number of required fds ({num_required_fds}) is larger than the max fds ({max_fd_limit}).")
         }
+        debug!("Increasing fd limit from {cur_fd_limit} to {num_required_fds}");
+        rlimit::increase_nofile_limit(num_required_fds)
+            .with_context(|| format!("Failed to increase the fd limit to {num_required_fds}"))?;
     }
-    Ok(cpu_info)
+    Ok(())
 }
 
 pub fn no_tar_gz_file_name(path: &PathBuf) -> Option<String> {
@@ -272,6 +368,37 @@ mod utils_test {
     use std::path::PathBuf;
     use tempfile::TempDir;
 
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn test_get_online_cpu_ids() {
+        use super::get_online_cpu_ids;
+        let ids = get_online_cpu_ids().expect("should read /sys/devices/system/cpu/online");
+        // At least CPU 0 is always online.
+        assert!(!ids.is_empty(), "expected at least one online CPU");
+        assert!(ids.contains(&0), "CPU 0 should be online");
+        // Result is sorted and de-duplicated.
+        let mut sorted = ids.clone();
+        sorted.sort_unstable();
+        sorted.dedup();
+        assert_eq!(ids, sorted, "ids should be sorted and unique");
+        // Count should match sysconf(_SC_NPROCESSORS_ONLN) on a normal system.
+        let nproc = unsafe { libc::sysconf(libc::_SC_NPROCESSORS_ONLN as libc::c_int) } as usize;
+        assert_eq!(ids.len(), nproc, "online CPU count should match sysconf");
+    }
+
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn test_cpu_info() {
+        use super::CpuInfo;
+
+        let cpu_info = CpuInfo::new().expect("Should read and parse /proc/cpuinfo");
+
+        assert!(
+            cpu_info.is_graviton() || cpu_info.is_intel() || cpu_info.is_amd(),
+            "The CPU should be recognized as one of Graviton, Intel, and AMD"
+        );
+    }
+
     #[test]
     fn test_find_file_prefix_match() {
         let dir = TempDir::new().unwrap();