(&self)
| 23 | |
| 24 | impl Graph { |
| 25 | pub(super) fn dmon(&self) -> JoinHandle<anyhow::Result<()>> { |
| 26 | let conns: Vec<_> = self.conns.values().cloned().collect(); |
| 27 | let ctx = self.ctx.clone(); |
| 28 | let mut first = true; // drop qps result fetched first |
| 29 | crate::rt::task::spawn(async move { |
| 30 | let wait_graph = ctx.wait().fuse(); |
| 31 | pin_mut!(wait_graph); |
| 32 | 'start: while !ctx.is_closed() { |
| 33 | if !crate::debug::QPS.enable() { |
| 34 | first = true; |
| 35 | } |
| 36 | let wait_qps = crate::debug::QPS.wait().fuse(); |
| 37 | pin_mut!(wait_qps); |
| 38 | select! { |
| 39 | _ = wait_qps => {}, |
| 40 | _ = wait_graph => break 'start, |
| 41 | } |
| 42 | |
| 43 | let args = crate::debug::QPS_args |
| 44 | .read() |
| 45 | .unwrap() |
| 46 | .iter() |
| 47 | .map(|(k, v)| (*k, v.ratio)) |
| 48 | .next(); |
| 49 | if let Some((seq_id, ratio)) = args { |
| 50 | let div_ratio = 1f32 / ratio; |
| 51 | let mut nodes = HashMap::new(); |
| 52 | |
| 53 | loop { |
| 54 | for conn in &conns { |
| 55 | let is_block = conn.get().is_almost_full(); |
| 56 | let size = conn.get().len(); |
| 57 | let tx_qps = conn.get().swap_tx_counter() as f32 * ratio; |
| 58 | let rx_qps = conn.get().swap_rx_counter() as f32 * ratio; |
| 59 | if first { |
| 60 | continue; |
| 61 | } |
| 62 | for tx in &conn.info().tx { |
| 63 | let qps = nodes.entry(tx.node_name.clone()).or_insert(NodeQps { |
| 64 | name: tx.node_name.clone(), |
| 65 | qps: Default::default(), |
| 66 | is_block: false, |
| 67 | }); |
| 68 | qps.qps |
| 69 | .insert(tx.port_name.clone(), (size, tx_qps as usize)); |
| 70 | } |
| 71 | for rx in &conn.info().rx { |
| 72 | let qps = nodes.entry(rx.node_name.clone()).or_insert(NodeQps { |
| 73 | name: rx.node_name.clone(), |
| 74 | qps: Default::default(), |
| 75 | is_block: false, |
| 76 | }); |
| 77 | qps.qps |
| 78 | .insert(rx.port_name.clone(), (size, rx_qps as usize)); |
| 79 | qps.is_block = qps.is_block || is_block; |
| 80 | } |
| 81 | } |
| 82 | if first { |
nothing calls this directly
no test coverage detected