rocketmq_runtime/blocking/
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
15use std::sync::Arc;
16use std::sync::Weak;
17use std::time::Duration;
18use std::time::Instant;
19
20use dashmap::DashMap;
21use parking_lot::Mutex;
22use serde::Serialize;
23
24use super::BlockingKind;
25use super::BlockingLane;
26use super::BlockingTaskId;
27use super::BlockingTaskState;
28
29/// The task table of one blocking executor.
30pub(crate) type BlockingTaskTable = DashMap<BlockingTaskId, BlockingTaskMeta>;
31
32/// Task tables of the isolated executors bound to one task-group tree.
33///
34/// An isolated executor has its own budget and task table outside the managed
35/// lanes. Registering the table keeps its work visible to the shutdown report
36/// of the owner that holds the tree. Every running closure holds its table, so
37/// a closure that outlives all of its executor handles is still reported.
38#[derive(Debug, Default)]
39pub(crate) struct IsolatedBlockingTables {
40    tables: Mutex<Vec<Weak<BlockingTaskTable>>>,
41}
42
43impl IsolatedBlockingTables {
44    pub(crate) fn register(&self, table: &Arc<BlockingTaskTable>) {
45        let mut tables = self.tables.lock();
46        // A released table can no longer hold work. Pruning on registration
47        // bounds the list by the live tables plus those released since then.
48        tables.retain(|table| table.strong_count() > 0);
49        tables.push(Arc::downgrade(table));
50    }
51
52    /// Returns the number of closures still running and a snapshot of every
53    /// queued or running task across the live tables.
54    pub(crate) fn report(&self) -> (usize, Vec<BlockingTaskSnapshot>) {
55        let tables = self.tables.lock().iter().filter_map(Weak::upgrade).collect::<Vec<_>>();
56        let mut still_running = 0;
57        let mut tasks = Vec::new();
58        for table in tables {
59            for entry in table.iter() {
60                let task = entry.value();
61                if matches!(
62                    task.state,
63                    BlockingTaskState::Running | BlockingTaskState::TimedOutStillRunning
64                ) {
65                    still_running += 1;
66                }
67                tasks.push(task.snapshot());
68            }
69        }
70        (still_running, tasks)
71    }
72}
73
74#[derive(Debug, Clone)]
75pub(crate) struct BlockingTaskMeta {
76    pub id: BlockingTaskId,
77    pub name: Arc<str>,
78    pub kind: BlockingKind,
79    pub state: BlockingTaskState,
80    pub queued_at: Instant,
81    pub started_at: Option<Instant>,
82}
83
84impl BlockingTaskMeta {
85    pub(crate) fn snapshot(&self) -> BlockingTaskSnapshot {
86        let elapsed = self.started_at.unwrap_or(self.queued_at).elapsed();
87        BlockingTaskSnapshot {
88            id: self.id,
89            name: self.name.to_string(),
90            kind: self.kind,
91            state: self.state,
92            elapsed,
93        }
94    }
95}
96
97/// Internal, name-free aggregates collected without materializing task details.
98pub(crate) struct BlockingExecutorAggregate {
99    pub lane: BlockingLane,
100    pub max_concurrency: usize,
101    pub max_queue_depth: usize,
102    pub queued: usize,
103    pub running: usize,
104    pub timed_out_still_running: usize,
105    pub blocking_still_running: usize,
106    pub task_kinds: [(BlockingKind, usize, Duration); 3],
107}
108
109impl BlockingExecutorAggregate {
110    pub(crate) fn new(lane: BlockingLane, max_concurrency: usize, max_queue_depth: usize) -> Self {
111        Self {
112            lane,
113            max_concurrency,
114            max_queue_depth,
115            queued: 0,
116            running: 0,
117            timed_out_still_running: 0,
118            blocking_still_running: 0,
119            task_kinds: [
120                (BlockingKind::ShortIo, 0, Duration::ZERO),
121                (BlockingKind::CpuBound, 0, Duration::ZERO),
122                (BlockingKind::LongRunning, 0, Duration::ZERO),
123            ],
124        }
125    }
126
127    pub(crate) fn record_kind(&mut self, kind: BlockingKind, elapsed: Duration) {
128        let index = match kind {
129            BlockingKind::ShortIo => 0,
130            BlockingKind::CpuBound => 1,
131            BlockingKind::LongRunning => 2,
132        };
133        self.task_kinds[index].1 = self.task_kinds[index].1.saturating_add(1);
134        self.task_kinds[index].2 = self.task_kinds[index].2.max(elapsed);
135    }
136}
137
138impl From<BlockingExecutorSnapshot> for BlockingExecutorAggregate {
139    fn from(snapshot: BlockingExecutorSnapshot) -> Self {
140        let mut aggregate = Self::new(snapshot.lane, snapshot.max_concurrency, snapshot.max_queue_depth);
141        aggregate.queued = snapshot.queued;
142        aggregate.running = snapshot.running;
143        aggregate.timed_out_still_running = snapshot.timed_out_still_running;
144        aggregate.blocking_still_running = snapshot.blocking_still_running;
145        for task in snapshot.tasks {
146            aggregate.record_kind(task.kind, task.elapsed);
147        }
148        aggregate
149    }
150}
151
152/// Represents blocking executor snapshot.
153#[derive(Debug, Clone, Serialize)]
154pub struct BlockingExecutorSnapshot {
155    /// The name value.
156    pub name: String,
157    /// The lane value.
158    pub lane: BlockingLane,
159    /// The max concurrency value.
160    pub max_concurrency: usize,
161    /// The max queue depth value.
162    pub max_queue_depth: usize,
163    /// The global capacity value.
164    pub global_capacity: usize,
165    /// The global running value.
166    pub global_running: usize,
167    /// The global available value.
168    pub global_available: usize,
169    /// The lane reserved value.
170    pub lane_reserved: usize,
171    /// The lane running value.
172    pub lane_running: usize,
173    /// The lane borrowed value.
174    pub lane_borrowed: usize,
175    /// The queued value.
176    pub queued: usize,
177    /// The running value.
178    pub running: usize,
179    /// The timed out still running value.
180    pub timed_out_still_running: usize,
181    /// The blocking still running value.
182    pub blocking_still_running: usize,
183    /// The rejected value.
184    pub rejected: u64,
185    #[serde(with = "duration_millis")]
186    /// The oldest queue wait value.
187    pub oldest_queue_wait: Duration,
188    /// The tasks value.
189    pub tasks: Vec<BlockingTaskSnapshot>,
190}
191
192/// Represents blocking task snapshot.
193#[derive(Debug, Clone, Serialize)]
194pub struct BlockingTaskSnapshot {
195    /// The id identifier.
196    pub id: BlockingTaskId,
197    /// The name value.
198    pub name: String,
199    /// The kind value.
200    pub kind: BlockingKind,
201    /// The state value.
202    pub state: BlockingTaskState,
203    #[serde(with = "duration_millis")]
204    /// The elapsed value.
205    pub elapsed: Duration,
206}
207
208mod duration_millis {
209    use std::time::Duration;
210
211    use serde::Serializer;
212
213    pub fn serialize<S>(duration: &Duration, serializer: S) -> Result<S::Ok, S::Error>
214    where
215        S: Serializer,
216    {
217        serializer.serialize_u64(duration.as_millis() as u64)
218    }
219}
rocketmq_runtime/blocking/diagnostics.rs

rocketmq_runtime/blocking/
diagnostics.rs
