reifydb_runtime/sync/
waiter.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::{fmt, sync::Arc};
5
6use reifydb_value::value::duration::Duration;
7
8#[cfg(not(reifydb_single_threaded))]
9use crate::context::clock::{MockClock, TimerId};
10use crate::{
11	context::clock::{Clock, TimerWake},
12	sync::{condvar::Condvar, mutex::Mutex},
13};
14
15#[cfg(not(reifydb_single_threaded))]
16const MOCK_PARK_BACKSTOP: Duration = Duration::from_seconds_const(30);
17
18struct Signal {
19	notified: Mutex<bool>,
20	condvar: Condvar,
21}
22
23impl Signal {
24	fn new() -> Self {
25		Self {
26			notified: Mutex::new(false),
27			condvar: Condvar::new(),
28		}
29	}
30}
31
32impl TimerWake for Signal {
33	fn wake(&self) {
34		let _guard = self.notified.lock();
35		self.condvar.notify_all();
36	}
37}
38
39pub struct WaiterHandle {
40	signal: Arc<Signal>,
41	#[cfg_attr(reifydb_single_threaded, allow(dead_code))]
42	clock: Option<Clock>,
43	on_notify: Mutex<Option<Box<dyn FnOnce() + Send>>>,
44}
45
46impl fmt::Debug for WaiterHandle {
47	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
48		f.debug_struct("WaiterHandle").finish_non_exhaustive()
49	}
50}
51
52impl Default for WaiterHandle {
53	fn default() -> Self {
54		Self::new()
55	}
56}
57
58impl WaiterHandle {
59	pub fn new() -> Self {
60		Self {
61			signal: Arc::new(Signal::new()),
62			clock: None,
63			on_notify: Mutex::new(None),
64		}
65	}
66
67	pub fn on_clock(clock: Clock) -> Self {
68		Self {
69			signal: Arc::new(Signal::new()),
70			clock: Some(clock),
71			on_notify: Mutex::new(None),
72		}
73	}
74
75	pub fn with_callback(callback: Box<dyn FnOnce() + Send>) -> Self {
76		Self {
77			signal: Arc::new(Signal::new()),
78			clock: None,
79			on_notify: Mutex::new(Some(callback)),
80		}
81	}
82
83	pub fn notify(&self) {
84		let mut guard = self.signal.notified.lock();
85		*guard = true;
86		self.signal.condvar.notify_one();
87		drop(guard);
88		if let Some(callback) = self.on_notify.lock().take() {
89			callback();
90		}
91	}
92
93	pub fn wait_timeout(&self, timeout: Duration) -> bool {
94		#[cfg(not(reifydb_single_threaded))]
95		if let Some(mock) = self.clock.as_ref().and_then(Clock::as_mock) {
96			return self.wait_until_virtual_deadline(mock, timeout);
97		}
98
99		let mut guard = self.signal.notified.lock();
100		if *guard {
101			return true;
102		}
103		!self.signal.condvar.wait_for(&mut guard, timeout).timed_out()
104	}
105
106	#[cfg(not(reifydb_single_threaded))]
107	fn wait_until_virtual_deadline(&self, mock: &MockClock, timeout: Duration) -> bool {
108		let deadline = mock.now().to_nanos().saturating_add(nanos_of(timeout));
109		let _timer = TimerGuard::register(mock, deadline, self.signal.clone());
110
111		let mut guard = self.signal.notified.lock();
112		loop {
113			if *guard {
114				return true;
115			}
116			if mock.now().to_nanos() >= deadline {
117				return false;
118			}
119			if self.signal.condvar.wait_for(&mut guard, MOCK_PARK_BACKSTOP).timed_out() {
120				panic!("mock clock never advanced past the park deadline");
121			}
122		}
123	}
124}
125
126#[cfg(not(reifydb_single_threaded))]
127fn nanos_of(timeout: Duration) -> u64 {
128	timeout.to_std().as_nanos().min(u64::MAX as u128) as u64
129}
130
131#[cfg(not(reifydb_single_threaded))]
132struct TimerGuard<'a> {
133	clock: &'a MockClock,
134	id: TimerId,
135}
136
137#[cfg(not(reifydb_single_threaded))]
138impl<'a> TimerGuard<'a> {
139	fn register(clock: &'a MockClock, deadline_nanos: u64, signal: Arc<Signal>) -> Self {
140		Self {
141			clock,
142			id: clock.register_timer(deadline_nanos, signal),
143		}
144	}
145}
146
147#[cfg(not(reifydb_single_threaded))]
148impl Drop for TimerGuard<'_> {
149	fn drop(&mut self) {
150		self.clock.cancel_timer(self.id);
151	}
152}
153
154#[cfg(test)]
155mod tests {
156	use std::sync::{
157		Arc,
158		atomic::{AtomicUsize, Ordering},
159	};
160
161	use super::*;
162
163	#[test]
164	fn callback_fires_exactly_once() {
165		let count = Arc::new(AtomicUsize::new(0));
166		let c = count.clone();
167		let waiter = WaiterHandle::with_callback(Box::new(move || {
168			c.fetch_add(1, Ordering::SeqCst);
169		}));
170
171		waiter.notify();
172		waiter.notify();
173
174		assert_eq!(count.load(Ordering::SeqCst), 1, "one-shot callback must fire exactly once");
175		assert!(
176			waiter.wait_timeout(Duration::from_milliseconds(0).unwrap()),
177			"an already-notified waiter returns immediately"
178		);
179	}
180}
reifydb_runtime/sync/waiter.rs

reifydb_runtime/sync/
waiter.rs
