Skip to content

Commit 3816630

Browse files
arkjedrzarkjedrz
committed
hmon: WIP - rework heartbeat monitor state
Rework heartbeat monitor state into two atomics.
1 parent 32fa411 commit 3816630

3 files changed

Lines changed: 275 additions & 280 deletions

File tree

‎src/health_monitoring_lib/rust/common.rs‎

Lines changed: 55 additions & 49 deletions
Original file line numberDiff line numberDiff line change
@@ -92,11 +92,11 @@ impl MonitorEvaluator for MonitorEvalHandle {
9292
}
9393
}
9494

95-
/// Get offset between HMON and monitor starting time points as [`u32`].
96-
pub(crate) fn hmon_time_offset(hmon_starting_point: Instant, monitor_starting_point: Instant) -> u32 {
95+
/// Get offset between HMON and monitor starting time points as [`u64`].
96+
pub(crate) fn hmon_time_offset(hmon_starting_point: Instant, monitor_starting_point: Instant) -> u64 {
9797
let result = hmon_starting_point.checked_duration_since(monitor_starting_point);
9898
let duration_since = result.expect("HMON starting point is earlier than monitor starting point");
99-
duration_to_u32(duration_since)
99+
duration_to_u64(duration_since)
100100
}
101101

102102
/// Get duration as [`u32`].
@@ -105,50 +105,56 @@ pub(crate) fn duration_to_u32(duration: Duration) -> u32 {
105105
u32::try_from(millis).expect("Monitor running for too long")
106106
}
107107

108-
#[cfg(all(test, not(loom)))]
109-
mod tests {
110-
use crate::common::{duration_to_u32, hmon_time_offset};
111-
use core::time::Duration;
112-
use std::time::Instant;
113-
114-
#[test]
115-
fn hmon_time_offset_valid() {
116-
let monitor_starting_point = Instant::now();
117-
let hmon_starting_point = Instant::now();
118-
let offset = hmon_time_offset(hmon_starting_point, monitor_starting_point);
119-
// Allow small offset.
120-
assert!(offset < 10);
121-
}
122-
123-
#[test]
124-
#[should_panic(expected = "HMON starting point is earlier than monitor starting point")]
125-
fn hmon_time_offset_wrong_order() {
126-
let hmon_starting_point = Instant::now();
127-
let monitor_starting_point = Instant::now();
128-
let _offset = hmon_time_offset(hmon_starting_point, monitor_starting_point);
129-
}
130-
131-
#[test]
132-
#[should_panic(expected = "Monitor running for too long")]
133-
fn hmon_time_offset_diff_too_large() {
134-
const HUNDRED_DAYS_AS_SECS: u64 = 100 * 24 * 60 * 60;
135-
let monitor_starting_point = Instant::now();
136-
let hmon_starting_point = Instant::now()
137-
.checked_add(Duration::from_secs(HUNDRED_DAYS_AS_SECS))
138-
.unwrap();
139-
let _offset = hmon_time_offset(hmon_starting_point, monitor_starting_point);
140-
}
141-
142-
#[test]
143-
fn duration_to_u32_valid() {
144-
let result = duration_to_u32(Duration::from_millis(1234));
145-
assert_eq!(result, 1234);
146-
}
147-
148-
#[test]
149-
#[should_panic(expected = "Monitor running for too long")]
150-
fn duration_to_u32_too_large() {
151-
const HUNDRED_DAYS_AS_SECS: u64 = 100 * 24 * 60 * 60;
152-
let _result = duration_to_u32(Duration::from_secs(HUNDRED_DAYS_AS_SECS));
153-
}
108+
/// Get duration as [`u64`].
109+
pub(crate) fn duration_to_u64(duration: Duration) -> u64 {
110+
let millis = duration.as_millis();
111+
u64::try_from(millis).expect("Monitor running for too long")
154112
}
113+
114+
// #[cfg(all(test, not(loom)))]
115+
// mod tests {
116+
// use crate::common::{duration_to_u32, hmon_time_offset};
117+
// use core::time::Duration;
118+
// use std::time::Instant;
119+
120+
// #[test]
121+
// fn hmon_time_offset_valid() {
122+
// let monitor_starting_point = Instant::now();
123+
// let hmon_starting_point = Instant::now();
124+
// let offset = hmon_time_offset(hmon_starting_point, monitor_starting_point);
125+
// // Allow small offset.
126+
// assert!(offset < 10);
127+
// }
128+
129+
// #[test]
130+
// #[should_panic(expected = "HMON starting point is earlier than monitor starting point")]
131+
// fn hmon_time_offset_wrong_order() {
132+
// let hmon_starting_point = Instant::now();
133+
// let monitor_starting_point = Instant::now();
134+
// let _offset = hmon_time_offset(hmon_starting_point, monitor_starting_point);
135+
// }
136+
137+
// #[test]
138+
// #[should_panic(expected = "Monitor running for too long")]
139+
// fn hmon_time_offset_diff_too_large() {
140+
// const HUNDRED_DAYS_AS_SECS: u64 = 100 * 24 * 60 * 60;
141+
// let monitor_starting_point = Instant::now();
142+
// let hmon_starting_point = Instant::now()
143+
// .checked_add(Duration::from_secs(HUNDRED_DAYS_AS_SECS))
144+
// .unwrap();
145+
// let _offset = hmon_time_offset(hmon_starting_point, monitor_starting_point);
146+
// }
147+
148+
// #[test]
149+
// fn duration_to_u32_valid() {
150+
// let result = duration_to_u32(Duration::from_millis(1234));
151+
// assert_eq!(result, 1234);
152+
// }
153+
154+
// #[test]
155+
// #[should_panic(expected = "Monitor running for too long")]
156+
// fn duration_to_u32_too_large() {
157+
// const HUNDRED_DAYS_AS_SECS: u64 = 100 * 24 * 60 * 60;
158+
// let _result = duration_to_u32(Duration::from_secs(HUNDRED_DAYS_AS_SECS));
159+
// }
160+
// }

‎src/health_monitoring_lib/rust/heartbeat/heartbeat_monitor.rs‎

Lines changed: 27 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -12,13 +12,15 @@
1212
// *******************************************************************************
1313

1414
use crate::common::{
15-
duration_to_u32, hmon_time_offset, Monitor, MonitorEvalHandle, MonitorEvaluationError, MonitorEvaluator, TimeRange,
15+
duration_to_u32, duration_to_u64, hmon_time_offset, Monitor, MonitorEvalHandle, MonitorEvaluationError,
16+
MonitorEvaluator, TimeRange,
1617
};
1718
use crate::heartbeat::heartbeat_state::{HeartbeatState, HeartbeatStateSnapshot};
1819
use crate::log::{error, warn};
1920
use crate::protected_memory::ProtectedMemoryAllocator;
2021
use crate::tag::MonitorTag;
2122
use crate::HealthMonitorError;
23+
use core::sync::atomic::{AtomicU64, Ordering};
2224
use core::time::Duration;
2325
use score_log::ScoreDebug;
2426
use std::sync::Arc;
@@ -100,30 +102,30 @@ impl Monitor for HeartbeatMonitor {
100102
}
101103
}
102104

103-
/// Time range using [`u32`].
105+
/// Time range using [`u64`].
104106
#[derive(ScoreDebug)]
105107
struct InternalRange {
106-
min: u32,
107-
max: u32,
108+
min: u64,
109+
max: u64,
108110
}
109111

110112
impl InternalRange {
111113
/// Create range using provided values.
112-
fn new(min: u32, max: u32) -> Self {
114+
fn new(min: u64, max: u64) -> Self {
113115
assert!(min <= max, "provided min is greater than provided max");
114116
Self { min, max }
115117
}
116118

117119
/// Create range with values offset by timestamp.
118-
fn offset(&self, timestamp: u32) -> Self {
120+
fn offset(&self, timestamp: u64) -> Self {
119121
Self::new(self.min + timestamp, self.max + timestamp)
120122
}
121123
}
122124

123125
impl From<TimeRange> for InternalRange {
124126
fn from(value: TimeRange) -> Self {
125-
let min = duration_to_u32(value.min);
126-
let max = duration_to_u32(value.max);
127+
let min = duration_to_u64(value.min);
128+
let max = duration_to_u64(value.max);
127129
Self::new(min, max)
128130
}
129131
}
@@ -139,6 +141,8 @@ pub(crate) struct HeartbeatMonitorInner {
139141
/// Offset is calculated during evaluation in relation to provided health monitor starting point.
140142
monitor_starting_point: Instant,
141143

144+
start_timestamp: AtomicU64,
145+
142146
/// Current heartbeat state.
143147
/// Contains data in relation to [`Self::monitor_starting_point`].
144148
heartbeat_state: HeartbeatState,
@@ -148,7 +152,7 @@ impl MonitorEvaluator for HeartbeatMonitorInner {
148152
fn evaluate(&self, hmon_starting_point: Instant, on_error: &mut dyn FnMut(&MonitorTag, MonitorEvaluationError)) {
149153
// Get current timestamp, with offset to HMON time.
150154
let offset = hmon_time_offset(hmon_starting_point, self.monitor_starting_point);
151-
let now = offset + duration_to_u32(hmon_starting_point.elapsed());
155+
let now = offset + duration_to_u64(hmon_starting_point.elapsed());
152156

153157
// Load current monitor state.
154158
let snapshot = self.heartbeat_state.snapshot();
@@ -158,15 +162,12 @@ impl MonitorEvaluator for HeartbeatMonitorInner {
158162
// It is necessary to:
159163
// - use offset as cycle starting point.
160164
// - get heartbeat snapshot in relation to zero point.
161-
let (start_timestamp, heartbeat_timestamp) = if snapshot.post_init() {
162-
let start_timestamp = snapshot.start_timestamp();
163-
let heartbeat_timestamp = start_timestamp + snapshot.heartbeat_timestamp_offset();
164-
(start_timestamp, heartbeat_timestamp)
165+
let start_timestamp = if snapshot.post_init() {
166+
self.start_timestamp.load(Ordering::Relaxed)
165167
} else {
166-
let start_timestamp = offset;
167-
let heartbeat_timestamp = snapshot.heartbeat_timestamp_offset();
168-
(start_timestamp, heartbeat_timestamp)
168+
offset
169169
};
170+
let heartbeat_timestamp = snapshot.heartbeat_timestamp();
170171

171172
// Get allowed time range as absolute values.
172173
let range = self.range.offset(start_timestamp);
@@ -208,35 +209,39 @@ impl MonitorEvaluator for HeartbeatMonitorInner {
208209
// Heartbeat in allowed state.
209210
else {
210211
// Update heartbeat monitor state with a current heartbeat as a beginning of a new cycle.
211-
let _ = self
212-
.heartbeat_state
213-
.update(|_| Some(HeartbeatStateSnapshot::new(heartbeat_timestamp)));
212+
self.start_timestamp.store(heartbeat_timestamp, Ordering::Relaxed);
213+
let _ = self.heartbeat_state.update(|_| {
214+
let mut snapshot = HeartbeatStateSnapshot::new();
215+
snapshot.set_post_init(true);
216+
Some(snapshot)
217+
});
214218
}
215219
}
216220
}
217221

218222
impl HeartbeatMonitorInner {
219223
fn new(monitor_tag: MonitorTag, range: TimeRange) -> Self {
220224
let monitor_starting_point = Instant::now();
225+
let start_timestamp = AtomicU64::new(0);
221226
let heartbeat_state_snapshot = HeartbeatStateSnapshot::default();
222227
let heartbeat_state = HeartbeatState::new(heartbeat_state_snapshot);
223228
Self {
224229
monitor_tag,
225230
range: InternalRange::from(range),
226231
monitor_starting_point,
232+
start_timestamp,
227233
heartbeat_state,
228234
}
229235
}
230236

231237
/// Provide a heartbeat.
232238
fn heartbeat(&self) {
233239
// Get current timestamp.
234-
let now = duration_to_u32(self.monitor_starting_point.elapsed());
240+
let now = duration_to_u64(self.monitor_starting_point.elapsed());
235241

236242
// Set heartbeat timestamp and update counter.
237243
let _ = self.heartbeat_state.update(|mut state| {
238-
let start_ts = state.start_timestamp();
239-
state.set_heartbeat_timestamp_offset(now - start_ts);
244+
state.set_heartbeat_timestamp(now);
240245
state.increment_counter();
241246
Some(state)
242247
});

0 commit comments

Comments
 (0)