lenso_runner/replicated/
diagnostics.rs

1use std::{
2    collections::BTreeMap,
3    sync::{
4        Arc,
5        atomic::{AtomicU64, Ordering},
6    },
7    time::Duration,
8};
9
10use lenso_app_plan::{ExecutionLaneId, ResolvedAppPlan};
11use lenso_kernel::NativeApp;
12
13#[derive(Debug)]
14pub(super) struct LaneDiagnosticsState {
15    plan: Arc<ResolvedAppPlan>,
16    lane_cpu_nanos: BTreeMap<String, AtomicU64>,
17    instance_queue_depths: BTreeMap<String, AtomicU64>,
18    total_messages: AtomicU64,
19    cross_lane_messages: AtomicU64,
20}
21
22impl LaneDiagnosticsState {
23    pub(super) fn new(plan: Arc<ResolvedAppPlan>) -> Self {
24        let lane_cpu_nanos = plan
25            .execution_lanes()
26            .iter()
27            .map(|lane| (lane.id().to_string(), AtomicU64::new(0)))
28            .collect();
29        let instance_queue_depths = plan
30            .module_instances()
31            .iter()
32            .map(|instance| (instance.instance_key().to_owned(), AtomicU64::new(0)))
33            .collect();
34        Self {
35            plan,
36            lane_cpu_nanos,
37            instance_queue_depths,
38            total_messages: AtomicU64::new(0),
39            cross_lane_messages: AtomicU64::new(0),
40        }
41    }
42
43    pub(super) fn record_invocation(
44        &self,
45        observing_lane: &ExecutionLaneId,
46        caller: &str,
47        provider: &str,
48    ) {
49        let Some(caller_lane) = self
50            .plan
51            .module_instance(caller)
52            .map(|instance| instance.execution_lane())
53        else {
54            return;
55        };
56        if caller_lane != observing_lane {
57            return;
58        }
59        let Some(provider_lane) = self
60            .plan
61            .module_instance(provider)
62            .map(|instance| instance.execution_lane())
63        else {
64            return;
65        };
66        self.total_messages.fetch_add(1, Ordering::Relaxed);
67        if caller_lane != provider_lane {
68            self.cross_lane_messages.fetch_add(1, Ordering::Relaxed);
69        }
70    }
71
72    pub(super) fn publish_lane(&self, lane: &ExecutionLaneId, app: &NativeApp, cpu_time: Duration) {
73        if let Some(cpu_nanos) = self.lane_cpu_nanos.get(lane.as_str()) {
74            cpu_nanos.store(duration_nanos(cpu_time), Ordering::Relaxed);
75        }
76        for (instance, depth) in app.instance_queue_depths() {
77            if let Some(queue_depth) = self.instance_queue_depths.get(&instance) {
78                queue_depth.store(u64::try_from(depth).unwrap_or(u64::MAX), Ordering::Relaxed);
79            }
80        }
81    }
82
83    pub(super) fn snapshot(&self) -> LaneDiagnosticsSnapshot {
84        LaneDiagnosticsSnapshot {
85            lane_cpu_time: self
86                .lane_cpu_nanos
87                .iter()
88                .map(|(lane, nanos)| {
89                    (
90                        lane.clone(),
91                        Duration::from_nanos(nanos.load(Ordering::Relaxed)),
92                    )
93                })
94                .collect(),
95            instance_queue_depths: self
96                .instance_queue_depths
97                .iter()
98                .map(|(instance, depth)| {
99                    (
100                        instance.clone(),
101                        usize::try_from(depth.load(Ordering::Relaxed)).unwrap_or(usize::MAX),
102                    )
103                })
104                .collect(),
105            total_messages: self.total_messages.load(Ordering::Relaxed),
106            cross_lane_messages: self.cross_lane_messages.load(Ordering::Relaxed),
107        }
108    }
109}
110
111fn duration_nanos(duration: Duration) -> u64 {
112    u64::try_from(duration.as_nanos()).unwrap_or(u64::MAX)
113}
114
115/// One immutable structural snapshot used to evaluate Plan placement.
116#[derive(Clone, Debug, Default, PartialEq)]
117pub struct LaneDiagnosticsSnapshot {
118    lane_cpu_time: BTreeMap<String, Duration>,
119    instance_queue_depths: BTreeMap<String, usize>,
120    total_messages: u64,
121    cross_lane_messages: u64,
122}
123
124impl LaneDiagnosticsSnapshot {
125    /// Returns CPU time consumed by each lane's owner thread.
126    pub fn lane_cpu_time(&self) -> &BTreeMap<String, Duration> {
127        &self.lane_cpu_time
128    }
129
130    /// Returns the latest bounded request queue depth for one Module Instance.
131    pub fn instance_queue_depth(&self, instance: &str) -> Option<usize> {
132        self.instance_queue_depths.get(instance).copied()
133    }
134
135    /// Returns the number of observed App request messages.
136    pub const fn total_messages(&self) -> u64 {
137        self.total_messages
138    }
139
140    /// Returns the number of observed messages whose binding crosses lanes.
141    pub const fn cross_lane_messages(&self) -> u64 {
142        self.cross_lane_messages
143    }
144
145    /// Returns the share of observed request messages whose binding crosses lanes.
146    pub fn cross_lane_message_share(&self) -> f64 {
147        if self.total_messages == 0 {
148            0.0
149        } else {
150            self.cross_lane_messages as f64 / self.total_messages as f64
151        }
152    }
153}
lenso_runner/replicated/diagnostics.rs

lenso_runner/replicated/
diagnostics.rs
