parallel_processor/
phase_times_monitor.rs

1use crate::{
2    memory_data_size::MemoryDataSize,
3    memory_fs::{allocator::CHUNKS_ALLOCATOR, MemoryFs},
4};
5use parking_lot::lock_api::RawRwLock;
6use parking_lot::RwLock;
7use std::time::{Duration, Instant};
8#[cfg(feature = "process-stats")]
9use {nightly_quirks::utils::NightlyUtils, parking_lot::Mutex, std::cmp::max};
10
11pub struct PhaseResult {
12    name: String,
13    time: Duration,
14}
15
16pub struct PhaseTimesMonitor {
17    timer: Option<Instant>,
18    phase: Option<(String, Instant)>,
19    results: Vec<PhaseResult>,
20}
21
22pub static PHASES_TIMES_MONITOR: RwLock<PhaseTimesMonitor> =
23    RwLock::const_new(parking_lot::RawRwLock::INIT, PhaseTimesMonitor::new());
24
25#[cfg(feature = "process-stats")]
26struct ProcessStats {
27    user_cpu_total: f64,
28    kernel_cpu_total: f64,
29    mem_total: u128,
30    mem_max: u64,
31    samples_cnt: u64,
32    disk_space: u64,
33    max_disk_space: u64,
34}
35
36#[cfg(feature = "process-stats")]
37impl ProcessStats {
38    const fn new() -> Self {
39        ProcessStats {
40            user_cpu_total: 0.0,
41            kernel_cpu_total: 0.0,
42            mem_total: 0,
43            mem_max: 0,
44            samples_cnt: 0,
45            disk_space: 0,
46            max_disk_space: 0,
47        }
48    }
49
50    fn update(
51        &mut self,
52        elapsed_time: Duration,
53        elapsed_cpu: Duration,
54        elapsed_kernel: Duration,
55        current_mem: u64,
56        disk_space: u64,
57        _max_disk_space: u64,
58    ) {
59        self.samples_cnt += 1;
60        self.user_cpu_total += elapsed_cpu.as_secs_f64() / elapsed_time.as_secs_f64();
61        self.kernel_cpu_total += elapsed_kernel.as_secs_f64() / elapsed_time.as_secs_f64();
62        self.mem_total += current_mem as u128;
63        self.mem_max = max(self.mem_max, current_mem);
64        self.disk_space = disk_space;
65        self.max_disk_space = max(self.max_disk_space, disk_space);
66    }
67
68    fn format(&self) -> String {
69        let samples_cnt = if self.samples_cnt == 0 {
70            1
71        } else {
72            self.samples_cnt
73        };
74
75        format!(
76            "(uc:{:.2} kc:{:.2} mm:{:.2} cm:{:.2} ds: {:.2} mds: {:.2})",
77            (self.user_cpu_total / (samples_cnt as f64)),
78            (self.kernel_cpu_total / (samples_cnt as f64)),
79            MemoryDataSize::from_bytes(self.mem_max as usize),
80            MemoryDataSize::from_bytes((self.mem_total / (samples_cnt as u128)) as usize),
81            MemoryDataSize::from_bytes(self.disk_space as usize),
82            MemoryDataSize::from_bytes(self.max_disk_space as usize),
83        )
84    }
85
86    fn reset(&mut self) {
87        *self = Self::new();
88    }
89}
90
91#[cfg(feature = "process-stats")]
92static GLOBAL_STATS: Mutex<ProcessStats> = NightlyUtils::new_mutex(ProcessStats::new());
93#[cfg(feature = "process-stats")]
94static PHASE_STATS: Mutex<ProcessStats> = NightlyUtils::new_mutex(ProcessStats::new());
95#[cfg(feature = "process-stats")]
96static CURRENT_STATS: Mutex<ProcessStats> = NightlyUtils::new_mutex(ProcessStats::new());
97
98impl PhaseTimesMonitor {
99    const fn new() -> PhaseTimesMonitor {
100        PhaseTimesMonitor {
101            timer: None,
102            phase: None,
103            results: Vec::new(),
104        }
105    }
106
107    pub fn init(&mut self) {
108        self.timer = Some(Instant::now());
109
110        #[cfg(feature = "process-stats")]
111        {
112            std::thread::spawn(|| {
113                let clock = Instant::now();
114
115                let mut last_stats = crate::simple_process_stats::ProcessStats::get().unwrap();
116                let mut last_clock = clock.elapsed();
117
118                loop {
119                    std::thread::sleep(Duration::from_millis(100));
120                    let stats = crate::simple_process_stats::ProcessStats::get().unwrap();
121
122                    let time_now = clock.elapsed();
123
124                    let elapsed = time_now - last_clock;
125                    let kernel_elapsed_usage = stats.cpu_time_kernel - last_stats.cpu_time_kernel;
126                    let user_elapsed_usage = stats.cpu_time_user - last_stats.cpu_time_user;
127                    let current_memory = stats.memory_usage_bytes;
128                    let disk_stats = MemoryFs::get_stats();
129
130                    GLOBAL_STATS.lock().update(
131                        elapsed,
132                        user_elapsed_usage,
133                        kernel_elapsed_usage,
134                        current_memory,
135                        disk_stats.current_disk_usage,
136                        disk_stats.max_disk_usage,
137                    );
138                    PHASE_STATS.lock().update(
139                        elapsed,
140                        user_elapsed_usage,
141                        kernel_elapsed_usage,
142                        current_memory,
143                        disk_stats.current_disk_usage,
144                        disk_stats.max_disk_usage,
145                    );
146
147                    let mut current_stats = CURRENT_STATS.lock();
148                    current_stats.update(
149                        elapsed,
150                        user_elapsed_usage,
151                        kernel_elapsed_usage,
152                        current_memory,
153                        disk_stats.current_disk_usage,
154                        disk_stats.max_disk_usage,
155                    );
156
157                    last_clock = time_now;
158                    last_stats = stats;
159                }
160            });
161        }
162    }
163
164    fn end_phase(&mut self) {
165        if let Some((name, phase_timer)) = self.phase.take() {
166            let elapsed = phase_timer.elapsed();
167            crate::log_info!(
168                "Finished {}. phase duration: {:.2?} gtime: {:.2?}{}", // memory: {:.2} {:.2}%
169                name,
170                &elapsed,
171                self.get_wallclock(),
172                Self::format_process_stats()
173            );
174            self.results.push(PhaseResult {
175                name,
176                time: elapsed,
177            })
178        }
179    }
180
181    pub fn start_phase(&mut self, name: String) {
182        self.end_phase();
183        crate::log_info!(
184            "Started {}{}{}",
185            name,
186            match () {
187                #[cfg(feature = "process-stats")]
188                () => " prev stats: ",
189                #[cfg(not(feature = "process-stats"))]
190                () => String::new(),
191            },
192            Self::format_process_stats()
193        );
194        #[cfg(feature = "process-stats")]
195        PHASE_STATS.lock().reset();
196        self.phase = Some((name, Instant::now()));
197    }
198
199    pub fn get_wallclock(&self) -> Duration {
200        self.timer
201            .as_ref()
202            .map(|t| t.elapsed())
203            .unwrap_or(Duration::from_millis(0))
204    }
205
206    pub fn get_phase_desc(&self) -> String {
207        self.phase
208            .as_ref()
209            .map(|x| x.0.clone())
210            .unwrap_or(String::new())
211    }
212
213    pub fn get_phase_timer(&self) -> Duration {
214        self.phase
215            .as_ref()
216            .map(|x| x.1.elapsed())
217            .unwrap_or(Duration::from_millis(0))
218    }
219
220    fn format_process_stats() -> String {
221        #[cfg(feature = "process-stats")]
222        {
223            let memory = crate::simple_process_stats::ProcessStats::get()
224                .unwrap()
225                .memory_usage_bytes;
226
227            let stats = format!(
228                " GL:{} PH:{} CT: {} CM: {:.2}",
229                GLOBAL_STATS.lock().format(),
230                PHASE_STATS.lock().format(),
231                CURRENT_STATS.lock().format(),
232                MemoryDataSize::from_bytes(memory as usize),
233            );
234            CURRENT_STATS.lock().reset();
235            stats
236        }
237        #[cfg(not(feature = "process-stats"))]
238        String::new()
239    }
240
241    pub fn get_formatted_counter(&self) -> String {
242        let total_mem = CHUNKS_ALLOCATOR.get_total_memory();
243        let free_mem = CHUNKS_ALLOCATOR.get_free_memory();
244
245        format!(
246            " ptime: {:.2?} gtime: {:.2?} memory: {:.2} {:.2}%{}",
247            self.phase
248                .as_ref()
249                .map(|pt| pt.1.elapsed())
250                .unwrap_or(Duration::from_millis(0)),
251            self.get_wallclock(),
252            total_mem - free_mem,
253            ((1.0 - free_mem / total_mem) * 100.0),
254            Self::format_process_stats()
255        )
256    }
257
258    pub fn get_formatted_counter_without_memory(&self) -> String {
259        format!(
260            " ptime: {:.2?} gtime: {:.2?}{}",
261            self.phase
262                .as_ref()
263                .map(|pt| pt.1.elapsed())
264                .unwrap_or(Duration::from_millis(0)),
265            self.get_wallclock(),
266            Self::format_process_stats()
267        )
268    }
269
270    pub fn print_stats(&mut self, end_message: String) {
271        self.end_phase();
272
273        crate::log_info!("{}", end_message);
274        crate::log_info!("TOTAL TIME: {:.2?}", self.get_wallclock());
275        let fs_stats = MemoryFs::get_stats();
276
277        crate::log_info!(
278            "Max virtual fs usage: {:.2}",
279            MemoryDataSize::from_bytes(fs_stats.max_files_usage as usize),
280        );
281        crate::log_info!(
282            "Max disk usage: {:.2}",
283            MemoryDataSize::from_bytes(fs_stats.max_disk_usage as usize)
284        );
285        crate::log_info!("Final stats:");
286
287        for PhaseResult { name, time } in self.results.iter() {
288            crate::log_info!("\t{} \t=> {:.2?}", name, time);
289        }
290    }
291}
parallel_processor/phase_times_monitor.rs

parallel_processor/
phase_times_monitor.rs
