diff --git a/src/data/processes.rs b/src/data/processes.rs index 323e3e81..fa316c29 100644 --- a/src/data/processes.rs +++ b/src/data/processes.rs @@ -205,11 +205,13 @@ impl ProcessData for Processes { let stat = match process_key { ProcessKey::UserSpaceTime | ProcessKey::KernelSpaceTime => { - // CPU time: cumulative value, convert to aggregate percentage + // CPU time: represents the CPU time that a process consumes, e.g. + // 3.5 means the process's execution was equivalently scheduled on + // 3 CPUs the whole time and an additional CPU for half of the time. let value_diff = current_value - prev_value; let time_diff = current_time - prev_time; let s = if time_diff > 0 { - (value_diff / (ticks_per_second as f64 * time_diff as f64)) * 100.0 + value_diff / (ticks_per_second as f64 * time_diff as f64) } else { 0.0 }; @@ -241,14 +243,28 @@ impl ProcessData for Processes { .map(|(k, v)| (k.clone(), *v)) .collect(); ranking.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(Ordering::Equal)); - let top_cpu_time_processes: Vec = + // Only retain the top 16 processes of cpu utilization. + let mut processes_to_include: Vec = ranking.into_iter().take(16).map(|(name, _)| name).collect(); - // Add top processes for each metric to time_series_data + // Also retain the metrics of the APerf process, located by the saved pid. + if let Some(aperf_pid) = params.pid { + let pid_prefix = format!("{aperf_pid}_"); + if let Some(aperf_process) = process_cpu_time_map + .keys() + .find(|name| name.starts_with(&pid_prefix)) + { + if !processes_to_include.contains(aperf_process) { + processes_to_include.push(aperf_process.clone()); + } + } + } + + // Add the selected processes for each metric to time_series_data for (process_key, process_map) in per_field_per_process_series { let mut series_vec = Vec::new(); - for process_name in &top_cpu_time_processes { + for process_name in &processes_to_include { if let Some(series) = process_map.get(process_name) { series_vec.push(series.clone()); } diff --git a/src/lib.rs b/src/lib.rs index c10d148c..123a0353 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -16,11 +16,15 @@ pub mod server; pub mod visualizer; use crate::analytics::{AnalyticalEngine, DataFindings}; +use crate::computations::Statistics; use crate::data::aperf_runlog::AperfRunlog; +use crate::data::aperf_stats::AperfStat; +use crate::data::common::data_formats::{AperfData, TimeSeriesMetric}; use crate::data::common::processed_data_accessor::ProcessedDataAccessor; +use crate::data::processes::Processes; use crate::data::TimeEnum; -use crate::visualizer::DataVisualizer; -use anyhow::Result; +use crate::visualizer::{DataVisualizer, ReportParams}; +use anyhow::{Context, Result}; use chrono::prelude::*; use data::common::utils::get_data_name_from_type; use log::{debug, error, info}; @@ -31,7 +35,6 @@ use std::path::{Path, PathBuf}; use thiserror::Error; #[cfg(target_os = "linux")] use { - crate::data::aperf_stats::AperfStat, crate::data::Data, flate2::{write::GzEncoder, Compression}, nix::poll::{poll, PollFd, PollFlags, PollTimeout}, @@ -218,11 +221,10 @@ impl PerformanceData { } pub fn collect_data_serial(&mut self) -> Result<()> { - let start = time::Instant::now(); + let start_time = time::Instant::now(); self.init_params.collection_start = Some(TimeEnum::DateTime(Utc::now())); - let mut aperf_collect_data = AperfStat::new(); - let mut current = time::Instant::now(); - let end = current + time::Duration::from_secs(self.init_params.period); + let mut aperf_stat = AperfStat::new(); + let end_time = start_time + time::Duration::from_secs(self.init_params.period); // TimerFd let mut tfd = TimerFd::new()?; @@ -244,58 +246,66 @@ impl PerformanceData { let signal_pollfd = PollFd::new(sfd.as_fd(), PollFlags::POLLIN); let mut poll_fds = [timer_pollfd, signal_pollfd]; - let mut datatype_signal = signal::SIGTERM; + let mut data_type_signal = signal::SIGTERM; - while current <= end { - aperf_collect_data.time = TimeEnum::DateTime(Utc::now()); - aperf_collect_data.data = HashMap::new(); + let mut current_time = start_time; + + while current_time <= end_time { if poll(&mut poll_fds, PollTimeout::NONE)? <= 0 { - error!("Poll error."); + error!("Failed to poll timer or signal fds"); } + if let Some(ev) = poll_fds[0].revents() { if ev.contains(PollFlags::POLLIN) { let ret = tfd.read(); if ret > 1 { error!("Missed {} interval(s)", ret - 1); } - debug!("Time elapsed: {:?}", start.elapsed()); - current += time::Duration::from_secs(ret * self.init_params.interval); - for (name, datatype) in self.collectors.iter_mut() { - if datatype.is_static { + debug!("Time elapsed: {:?}", start_time.elapsed()); + + let cur_collection_start = time::Instant::now(); + aperf_stat.time = TimeEnum::DateTime(Utc::now()); + aperf_stat.data = HashMap::new(); + + for (name, data_type) in self.collectors.iter_mut() { + if data_type.is_static { continue; } - datatype.collector_params.elapsed_time = start.elapsed().as_secs(); - - aperf_collect_data.measure( - name.clone() + "-collect", - || -> Result<()> { - datatype.collect_data()?; - Ok(()) - }, - )?; - aperf_collect_data.measure(name.clone() + "-print", || -> Result<()> { - datatype.write_to_file()?; + data_type.collector_params.elapsed_time = start_time.elapsed().as_secs(); + + aperf_stat.measure(name.clone() + "-collect", || -> Result<()> { + data_type.collect_data()?; + Ok(()) + })?; + aperf_stat.measure(name.clone() + "-print", || -> Result<()> { + data_type.write_to_file()?; Ok(()) })?; } - let data_collection_time = time::Instant::now() - current; - aperf_collect_data + let cur_collection_end = time::Instant::now(); + + let cur_collection_time = cur_collection_end - cur_collection_start; + aperf_stat .data - .insert("aperf".to_string(), data_collection_time.as_micros() as u64); - debug!("Collection time: {:?}", data_collection_time); + .insert("aperf".to_string(), cur_collection_time.as_micros() as u64); + debug!("Collection time: {:?}", cur_collection_time); + bincode::serialize_into( self.aperf_stats_handle.as_ref().unwrap(), - &aperf_collect_data, + &aperf_stat, )?; + + current_time = cur_collection_end; } } + if let Some(ev) = poll_fds[1].revents() { if ev.contains(PollFlags::POLLIN) { if let Ok(Some(siginfo)) = sfd.read_signal() { if siginfo.ssi_signo == signal::SIGINT as u32 { info!("Caught SIGINT. Exiting..."); - datatype_signal = signal::SIGINT; + data_type_signal = signal::SIGINT; } else if siginfo.ssi_signo == signal::SIGTERM as u32 { info!("Caught SIGTERM. Exiting..."); } else { @@ -306,12 +316,16 @@ impl PerformanceData { } } } + self.init_params.collection_end = Some(TimeEnum::DateTime(Utc::now())); + for (_name, datatype) in self.collectors.iter_mut() { - datatype.set_signal(datatype_signal); + datatype.set_signal(data_type_signal); datatype.finish_data_collection()?; } + tfd.set_state(TimerState::Disarmed, SetTimeFlags::Default); + Ok(()) } @@ -360,37 +374,31 @@ impl Default for PerformanceData { #[derive(Default)] pub struct VisualizationData { + per_run_report_params: HashMap, pub visualizers: HashMap, } impl VisualizationData { - pub fn new() -> Self { + pub fn new(per_run_report_params: HashMap) -> Self { VisualizationData { + per_run_report_params, visualizers: HashMap::new(), } } - pub fn init_visualizers( - &mut self, - run_name: String, - run_data_dir: PathBuf, - tmp_dir: &Path, - report_dir: &Path, - collection_start: Option, - pmu_counter_mode: String, - ) -> Result<()> { + pub fn init_visualizers(&mut self, run_name: &str) -> Result<()> { let visualizers_len = self.visualizers.len(); let mut error_count = 0; + let cur_run_report_params = + self.per_run_report_params + .get_mut(run_name) + .with_context(|| { + format!("Unrecognized run name {run_name} when initializing report data.") + })?; + for data_visualizer in self.visualizers.values_mut() { - if let Err(e) = data_visualizer.init_visualizer( - run_name.clone(), - run_data_dir.clone(), - tmp_dir, - report_dir, - collection_start, - pmu_counter_mode.clone(), - ) { + if let Err(e) = data_visualizer.init_visualizer(cur_run_report_params.clone()) { debug!("{:#?}", e); error_count += 1; } @@ -451,6 +459,111 @@ impl VisualizationData { .is_some_and(|&value| value) }) } + + /// Logics to be run after all raw data has been processed. + pub fn post_process_data(&mut self) { + // Perform the inter-data post-processing logics first. + copy_aperf_process_metrics_to_aperf_stats( + &mut self.visualizers, + &self.per_run_report_params, + ); + + // Perform the intra-data logics for each data type. + for data_visualizer in self.visualizers.values_mut() { + data_visualizer.post_process_data(); + } + } +} + +/// Extract the APerf process's metric from processes data and add them to the +/// aperf_stats data, to monitor APerf performance in the report. +fn copy_aperf_process_metrics_to_aperf_stats( + visualizers: &mut HashMap, + per_run_report_params: &HashMap, +) { + // A map from a run name to the sorted list of APerf process metrics + let mut per_run_aperf_process_metrics: HashMap> = HashMap::new(); + + let processes_data_visualizer = match visualizers.get(get_data_name_from_type::()) { + Some(processes_data_visualizer) => processes_data_visualizer, + None => return, + }; + + for (run_name, cur_run_data) in &processes_data_visualizer.processed_data.runs { + let cur_run_pid = match per_run_report_params.get(run_name) { + Some(report_params) => { + if let Some(pid) = report_params.pid { + pid + } else { + continue; + } + } + None => continue, + }; + + let cur_run_processes_data = match cur_run_data { + AperfData::TimeSeries(time_series_data) => time_series_data, + _ => continue, + }; + + let aperf_series_name = format!("{cur_run_pid}_aperf"); + + for metric_name in &cur_run_processes_data.sorted_metric_names { + // For every processes metric, locate the corresponding series for the APerf process and + // create a new dedicated metric for it. + if let Some(metric) = cur_run_processes_data.metrics.get(metric_name) { + if let Some(series) = metric + .series + .iter() + .find(|s| s.series_name == aperf_series_name) + { + let aperf_process_metric_name = format!("process_{metric_name}"); + let mut aperf_process_metric = + TimeSeriesMetric::new(aperf_process_metric_name.clone()); + let stats = Statistics::from_values(&series.values); + aperf_process_metric.value_range = + (stats.min.floor() as u64, stats.max.ceil() as u64); + aperf_process_metric.stats = stats; + aperf_process_metric.series = vec![series.clone()]; + per_run_aperf_process_metrics + .entry(run_name.clone()) + .or_default() + .push(aperf_process_metric); + } + } + } + } + + if per_run_aperf_process_metrics.is_empty() { + return; + } + + let aperf_stat_data_visualizer = + match visualizers.get_mut(get_data_name_from_type::()) { + Some(aperf_stats_visualizer) => aperf_stats_visualizer, + None => return, + }; + + for (run_name, aperf_process_metrics) in per_run_aperf_process_metrics { + if let Some(AperfData::TimeSeries(cur_run_aperf_stats_data)) = aperf_stat_data_visualizer + .processed_data + .runs + .get_mut(&run_name) + { + // The APerf process metrics should be showing upfront + let mut insert_pos = 0; + for aperf_process_metric in aperf_process_metrics { + cur_run_aperf_stats_data + .sorted_metric_names + .insert(insert_pos, aperf_process_metric.metric_name.clone()); + cur_run_aperf_stats_data.metrics.insert( + aperf_process_metric.metric_name.clone(), + aperf_process_metric, + ); + insert_pos += 1; + } + } + } } pub const GROUPED_PMU_MODE: &str = "grouped"; @@ -487,6 +600,10 @@ pub struct InitParams { /// produced by versions of aperf that did not record this. #[serde(default)] pub collection_end: Option, + /// PID of the aperf process that performed the collection. `None` + /// for archives produced by versions of aperf that did not record this. + #[serde(default)] + pub pid: Option, } impl InitParams { @@ -532,6 +649,7 @@ impl InitParams { num_to_report: 5000, collection_start: None, collection_end: None, + pid: Some(std::process::id()), } } } diff --git a/src/report.rs b/src/report.rs index 42da11fb..08f3646a 100644 --- a/src/report.rs +++ b/src/report.rs @@ -2,6 +2,7 @@ use crate::analytics::BASE_RUN_NAME; use crate::data::common::processed_data_accessor::ProcessedDataAccessor; use crate::data::common::utils::no_tar_gz_file_name; use crate::data::{TimeEnum, JS_DIR}; +use crate::visualizer::ReportParams; use crate::{data, InitParams, PDError, VisualizationData, APERF_FILE_FORMAT}; use anyhow::{Context, Result}; use chrono::Utc; @@ -41,17 +42,19 @@ impl VersionInfo { /// Stores the information of all runs to be included in the report #[derive(Default)] struct RunsInfo { - /// The list of run names (the run dir/archive file name minus the file format) + /// The path to the report directory. + report_dir: PathBuf, + /// The path to the temp working directory. + tmp_dir: PathBuf, + /// The list of run names (the run dir/archive file name minus the file format). run_names: Vec, - /// The index used to deduplicate each unique run name + /// The index used to deduplicate each unique run name. run_name_dedup_indices: HashMap, - /// The map from run names to paths to the run archives (run data tar files) + /// The map from run names to paths to the run archives (run data tar files). run_archive_paths: HashMap, - /// The map from run names to paths to the run data directory (where data are available for read and processing) - run_dir_paths: HashMap, - /// The specified start time of every run's time range + /// The specified start time of every run's time range. per_run_from_time: HashMap, - /// The specified end time of every run's time range + /// The specified end time of every run's time range. per_run_to_time: HashMap, /// Wall-clock start of `collect_data_serial`, derived from the run's /// `meta_data.bin`. Missing entry => archive was recorded by an older @@ -59,14 +62,19 @@ struct RunsInfo { per_run_start_time: HashMap, /// Wall-clock end of `collect_data_serial`. per_run_end_time: HashMap, - /// Whether the PMU counters were collected with or without groups. - /// Empty string means legacy runs before PMU config revamp. - per_run_pmu_counter_mode: HashMap, + /// Per-run report params initialized from meta_data.bin, which is a + /// serialization of InitParams, used to pass parameters between + /// record and report. + per_run_report_params: HashMap, } impl RunsInfo { - fn new() -> Self { - RunsInfo::default() + fn new(report_dir: PathBuf, tmp_dir: PathBuf) -> Self { + RunsInfo { + report_dir, + tmp_dir, + ..Default::default() + } } fn add_run( @@ -80,13 +88,20 @@ impl RunsInfo { self.run_archive_paths .insert(deduped_run_name.clone(), run_archive_path); + let mut report_params = ReportParams::new(); + report_params.report_dir = self.report_dir.clone(); + report_params.tmp_dir = self.tmp_dir.clone(); + report_params.run_name = deduped_run_name.clone(); + report_params.data_dir = run_dir_path.clone(); + // Reads meta_data.bin from a run's data directory for info about the collector. let meta_data_path = run_dir_path.join(format!("meta_data.{}", APERF_FILE_FORMAT)); if let Ok(meta_data_bytes) = fs::read(&meta_data_path) { if let Ok(meta_data) = bincode::deserialize::(&meta_data_bytes) { - self.per_run_pmu_counter_mode - .insert(deduped_run_name.clone(), meta_data.pmu_counter_mode); + report_params.pmu_counter_mode = meta_data.pmu_counter_mode; + report_params.pid = meta_data.pid; if let Some(collection_start) = meta_data.collection_start { + report_params.collection_start = Some(collection_start); self.per_run_start_time .insert(deduped_run_name.clone(), collection_start); } @@ -97,7 +112,8 @@ impl RunsInfo { } } - self.run_dir_paths.insert(deduped_run_name, run_dir_path); + self.per_run_report_params + .insert(deduped_run_name, report_params); Ok(()) } @@ -244,7 +260,7 @@ pub fn report(report: &Report, tmp_dir: &PathBuf) -> Result<()> { return Err(PDError::ReportExists(report_path_str).into()); } - let mut runs_info = RunsInfo::new(); + let mut runs_info = RunsInfo::new(report_dir_path, tmp_dir.clone()); let mut seen_run_paths: HashSet = HashSet::new(); for run in &report.run { @@ -298,7 +314,7 @@ pub fn report(report: &Report, tmp_dir: &PathBuf) -> Result<()> { runs_info.process_per_run_time_range(&report.time_range)?; - generate_report_files(report_dir_path, runs_info, tmp_dir); + generate_report_files(runs_info); Ok(()) } @@ -431,41 +447,26 @@ pub fn get_report_run_archive_paths(report_path: &PathBuf) -> Option, } impl ReportParams { @@ -34,6 +36,7 @@ impl ReportParams { data_file_path: PathBuf::new(), collection_start: None, pmu_counter_mode: String::new(), + pid: None, } } } @@ -59,18 +62,10 @@ impl DataVisualizer { } } - pub fn init_visualizer( - &mut self, - run_name: String, - run_data_dir: PathBuf, - tmp_dir: &Path, - report_dir: &Path, - collection_start: Option, - pmu_counter_mode: String, - ) -> Result<()> { - self.report_params.run_name = run_name.clone(); - self.report_params.collection_start = collection_start; - self.report_params.pmu_counter_mode = pmu_counter_mode; + pub fn init_visualizer(&mut self, report_params: ReportParams) -> Result<()> { + self.report_params = report_params; + let run_name = self.report_params.run_name.clone(); + let run_data_dir = self.report_params.data_dir.clone(); let file_path = find_file( &run_data_dir, @@ -98,9 +93,6 @@ impl DataVisualizer { Err(e) })?; let file = fs::OpenOptions::new().read(true).open(&file_path)?; - self.report_params.data_dir = run_data_dir; - self.report_params.tmp_dir = tmp_dir.to_path_buf(); - self.report_params.report_dir = report_dir.to_path_buf(); self.report_params.data_file_path = file_path; self.file_handle = Some(file); self.data_available.insert(run_name, true); diff --git a/tests/test_aperf_stats.rs b/tests/test_aperf_stats.rs index 7f3ce616..51fccb43 100644 --- a/tests/test_aperf_stats.rs +++ b/tests/test_aperf_stats.rs @@ -30,6 +30,7 @@ mod aperf_stats_tests { data_file_path: PathBuf::from(data_file_path), collection_start: None, pmu_counter_mode: String::new(), + pid: None, } } diff --git a/tests/test_java_profile.rs b/tests/test_java_profile.rs index f50ab993..0a7c1544 100644 --- a/tests/test_java_profile.rs +++ b/tests/test_java_profile.rs @@ -22,6 +22,7 @@ fn setup_test_env() -> (TempDir, PathBuf, PathBuf, ReportParams) { data_file_path: PathBuf::new(), collection_start: None, pmu_counter_mode: String::new(), + pid: None, }; (temp_dir, data_dir, report_dir, params) diff --git a/tests/test_processes.rs b/tests/test_processes.rs index a2286ec2..b28b11f8 100644 --- a/tests/test_processes.rs +++ b/tests/test_processes.rs @@ -198,7 +198,7 @@ fn test_process_processes_raw_data_complex() { [sample_idx - 1][process_name]; let delta = expected_stats.user_space_time - prev_stats.user_space_time; - (delta as f64) / (ticks_per_second as f64 * 2.0) * 100.0 + (delta as f64) / (ticks_per_second as f64 * 2.0) } } ProcessKey::KernelSpaceTime => { @@ -209,7 +209,7 @@ fn test_process_processes_raw_data_complex() { [sample_idx - 1][process_name]; let delta = expected_stats.kernel_space_time - prev_stats.kernel_space_time; - (delta as f64) / (ticks_per_second as f64 * 2.0) * 100.0 + (delta as f64) / (ticks_per_second as f64 * 2.0) } } ProcessKey::NumberThreads => expected_stats.number_threads as f64, @@ -304,7 +304,7 @@ fn test_process_processes_raw_data_simple() { let prev_stats = &expected_per_sample_per_process_stats[sample_idx - 1] [process_name]; let delta = expected_stats.user_space_time - prev_stats.user_space_time; - (delta as f64) / (ticks_per_second as f64 * 1.0) * 100.0 + (delta as f64) / (ticks_per_second as f64 * 1.0) } } ProcessKey::KernelSpaceTime => { @@ -315,7 +315,7 @@ fn test_process_processes_raw_data_simple() { [process_name]; let delta = expected_stats.kernel_space_time - prev_stats.kernel_space_time; - (delta as f64) / (ticks_per_second as f64 * 1.0) * 100.0 + (delta as f64) / (ticks_per_second as f64 * 1.0) } } ProcessKey::NumberThreads => expected_stats.number_threads as f64,