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 = u64::try_from(mock.now().to_nanos())
109			.expect("mock clock is never before the Unix epoch")
110			.saturating_add(nanos_of(timeout));
111		let _timer = TimerGuard::register(mock, deadline, self.signal.clone());
112
113		let mut guard = self.signal.notified.lock();
114		loop {
115			if *guard {
116				return true;
117			}
118			if u64::try_from(mock.now().to_nanos()).expect("mock clock is never before the Unix epoch")
119				>= deadline
120			{
121				return false;
122			}
123			if self.signal.condvar.wait_for(&mut guard, MOCK_PARK_BACKSTOP).timed_out() {
124				panic!("mock clock never advanced past the park deadline");
125			}
126		}
127	}
128}
129
130#[cfg(not(reifydb_single_threaded))]
131fn nanos_of(timeout: Duration) -> u64 {
132	timeout.to_std().as_nanos().min(u64::MAX as u128) as u64
133}
134
135#[cfg(not(reifydb_single_threaded))]
136struct TimerGuard<'a> {
137	clock: &'a MockClock,
138	id: TimerId,
139}
140
141#[cfg(not(reifydb_single_threaded))]
142impl<'a> TimerGuard<'a> {
143	fn register(clock: &'a MockClock, deadline_nanos: u64, signal: Arc<Signal>) -> Self {
144		Self {
145			clock,
146			id: clock.register_timer(deadline_nanos, signal),
147		}
148	}
149}
150
151#[cfg(not(reifydb_single_threaded))]
152impl Drop for TimerGuard<'_> {
153	fn drop(&mut self) {
154		self.clock.cancel_timer(self.id);
155	}
156}
157
158#[cfg(test)]
159mod tests {
160	use std::sync::{
161		Arc,
162		atomic::{AtomicUsize, Ordering},
163	};
164
165	use super::*;
166
167	#[test]
168	fn callback_fires_exactly_once() {
169		let count = Arc::new(AtomicUsize::new(0));
170		let c = count.clone();
171		let waiter = WaiterHandle::with_callback(Box::new(move || {
172			c.fetch_add(1, Ordering::SeqCst);
173		}));
174
175		waiter.notify();
176		waiter.notify();
177
178		assert_eq!(count.load(Ordering::SeqCst), 1, "one-shot callback must fire exactly once");
179		assert!(
180			waiter.wait_timeout(Duration::from_milliseconds(0).unwrap()),
181			"an already-notified waiter returns immediately"
182		);
183	}
184}
reifydb_runtime/sync/waiter.rs

reifydb_runtime/sync/
waiter.rs
