rocketmq_runtime/task_group/
diagnostics.rs

1// Copyright 2023 The RocketMQ Rust Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Aggregate task scans and bounded detail collection.
16
17use super::{TaskDetailScan, TaskDetailScope, TaskGroup, TaskGroupDiagnostics, TaskKind, TaskKindDiagnostics};
18use std::time::Duration;
19
20impl TaskGroup {
21    /// Samples sanitized task details without computing subtree aggregates.
22    ///
23    /// At most `scan_budget` tasks and `scan_budget` descendant registrations
24    /// are examined; at most `output_budget` details are returned. Enumeration
25    /// and temporary group storage obey the same budget, including empty or
26    /// expired groups. Concurrent changes are not a globally atomic snapshot.
27    pub fn diagnostics_task_details(&self, scan_budget: usize, output_budget: usize) -> crate::RuntimeTaskDetails {
28        self.bounded_task_details(scan_budget, output_budget).into()
29    }
30
31    /// Aggregates every task of this group and its descendants.
32    ///
33    /// The work is proportional to the active tasks and groups of the subtree.
34    /// An explicit stack keeps a deep component tree off the call stack.
35    pub(crate) fn diagnostics(&self, long_running_threshold: Duration) -> TaskGroupDiagnostics {
36        let mut aggregate = TaskGroupDiagnosticsAccumulator::default();
37        let local_task_count = self.accumulate_local_diagnostics(long_running_threshold, &mut aggregate);
38        let mut pending = self.inner.registry.components_snapshot();
39        while let Some(group) = pending.pop() {
40            group.accumulate_local_diagnostics(long_running_threshold, &mut aggregate);
41            pending.extend(group.inner.registry.components_snapshot());
42        }
43        aggregate.finish(local_task_count)
44    }
45
46    /// Scans this group and its descendants for a bounded detail list.
47    ///
48    /// The scan budget bounds the work: at most `scan_budget` tasks are
49    /// examined and at most `scan_budget` descendant groups are visited, so a
50    /// tree of empty groups cannot make a bounded scan unbounded. The output
51    /// budget bounds the payload. When any budget is reached the result is
52    /// truncated and reports how many tasks were examined, so a partial list is
53    /// never presented as the complete tree.
54    pub(crate) fn bounded_task_details(&self, scan_budget: usize, output_budget: usize) -> TaskDetailScan {
55        let mut scan = TaskDetailScan::default();
56        if !self.collect_local_task_details(scan_budget, output_budget, TaskDetailScope::Local, &mut scan) {
57            return scan;
58        }
59        let mut remaining_groups = scan_budget;
60        let (mut pending, truncated) = self.inner.registry.bounded_components_snapshot(&mut remaining_groups);
61        scan.truncated |= truncated;
62        while let Some(group) = pending.pop() {
63            if scan.scanned >= scan_budget {
64                scan.truncated = true;
65                break;
66            }
67            if !group.collect_local_task_details(scan_budget, output_budget, TaskDetailScope::Subtree, &mut scan) {
68                break;
69            }
70            let (children, truncated) = group.inner.registry.bounded_components_snapshot(&mut remaining_groups);
71            scan.truncated |= truncated;
72            pending.extend(children);
73        }
74        scan.group_entries_scanned = scan_budget - remaining_groups;
75        scan
76    }
77
78    /// Adds this group's own tasks to `scan`; returns `false` once the scan
79    /// budget stops the scan.
80    fn collect_local_task_details(
81        &self,
82        scan_budget: usize,
83        output_budget: usize,
84        scope: TaskDetailScope,
85        scan: &mut TaskDetailScan,
86    ) -> bool {
87        self.inner
88            .registry
89            .collect_task_details(scan_budget, output_budget, scope, scan)
90    }
91
92    fn accumulate_local_diagnostics(
93        &self,
94        long_running_threshold: Duration,
95        aggregate: &mut TaskGroupDiagnosticsAccumulator,
96    ) -> usize {
97        aggregate.group_count = aggregate.group_count.saturating_add(1);
98        let mut local_task_count = 0;
99        for task in self.inner.registry.tasks.iter() {
100            local_task_count += 1;
101            let elapsed = task.started_at.elapsed();
102            aggregate.record_task(task.kind, elapsed, elapsed >= long_running_threshold);
103        }
104        local_task_count
105    }
106}
107
108#[derive(Debug, Default)]
109struct TaskGroupDiagnosticsAccumulator {
110    group_count: usize,
111    task_count: usize,
112    active_by_kind: [usize; TaskKind::COUNT],
113    long_running_by_kind: [usize; TaskKind::COUNT],
114    max_elapsed_by_kind: [Duration; TaskKind::COUNT],
115}
116
117impl TaskGroupDiagnosticsAccumulator {
118    fn record_task(&mut self, kind: TaskKind, elapsed: Duration, long_running: bool) {
119        let index = kind.index();
120        self.task_count = self.task_count.saturating_add(1);
121        self.active_by_kind[index] = self.active_by_kind[index].saturating_add(1);
122        if long_running {
123            self.long_running_by_kind[index] = self.long_running_by_kind[index].saturating_add(1);
124        }
125        self.max_elapsed_by_kind[index] = self.max_elapsed_by_kind[index].max(elapsed);
126    }
127
128    fn finish(self, local_task_count: usize) -> TaskGroupDiagnostics {
129        let task_kinds = TaskKind::ALL
130            .into_iter()
131            .filter_map(|kind| {
132                let index = kind.index();
133                (self.active_by_kind[index] > 0).then_some(TaskKindDiagnostics {
134                    kind,
135                    active: self.active_by_kind[index],
136                    long_running: self.long_running_by_kind[index],
137                    max_elapsed: self.max_elapsed_by_kind[index],
138                })
139            })
140            .collect();
141
142        TaskGroupDiagnostics {
143            local_task_count,
144            group_count: self.group_count,
145            task_count: self.task_count,
146            task_kinds,
147        }
148    }
149}
rocketmq_runtime/task_group/diagnostics.rs

rocketmq_runtime/task_group/
diagnostics.rs
