frozen_core/ack.rs
1//! Acknowledge primitives for durability tracking
2//!
3//! ## Example
4//!
5//! ```
6//! use frozen_core::ack::{AckTicket, Completion};
7//! use std::sync::Arc;
8//!
9//! let completion = Arc::new(Completion::default());
10//! let epoch = completion.increment_current_epoch();
11//!
12//! let ticket = AckTicket::new(epoch, completion.clone());
13//! assert!(!ticket.is_durable());
14//!
15//! completion.mark_epoch_as_durable(epoch);
16//! completion.notify_all_listeners();
17//!
18//! assert!(ticket.is_durable());
19//!
20//! let durable_epoch = ticket.wait().unwrap();
21//! assert_eq!(durable_epoch, epoch);
22//! ```
23
24use crate::error::{ErrCode, FrozenError, FrozenResult};
25use std::sync::{self, atomic};
26
27/// A monotonically increasing epoch used as identifier for tracking durability of write operations
28pub type TEpoch = u64;
29
30/// Custom interface for requesting or executing synchronization trigger
31pub trait SyncTrigger: Send + Sync {
32 /// Trigger synchronization up to the current epoch
33 fn trigger_sync(&self) -> FrozenResult<()>;
34}
35
36/// A shared durability acknowledgement state used for issuing [`AckTicket`]
37///
38/// The completion state tracks following things,
39///
40/// - The latest assigned epoch
41/// - The latest durable epoch
42/// - Waiters blocked on durability advancement
43/// - Durability errors (if any) blocking durability progress
44///
45/// ## Example
46///
47/// ```
48/// let completion = frozen_core::ack::Completion::default();
49///
50/// assert_eq!(completion.read_current_epoch(), 0);
51/// assert_eq!(completion.read_durable_epoch(), 0);
52/// ```
53#[derive(Debug)]
54pub struct Completion {
55 current_epoch: atomic::AtomicU64,
56 durable_epoch: atomic::AtomicU64,
57 error: sync::Mutex<Option<FrozenError>>,
58 wait_mutex: sync::Mutex<()>,
59 durability_condvar: sync::Condvar,
60 sync_trigger: sync::RwLock<Option<sync::Weak<dyn SyncTrigger>>>,
61}
62
63impl Default for Completion {
64 fn default() -> Self {
65 Self {
66 current_epoch: atomic::AtomicU64::new(0),
67 durable_epoch: atomic::AtomicU64::new(0),
68 error: sync::Mutex::new(None),
69 wait_mutex: sync::Mutex::new(()),
70 durability_condvar: sync::Condvar::new(),
71 sync_trigger: sync::RwLock::new(None),
72 }
73 }
74}
75
76impl Completion {
77 /// Advance current and return next durability epoch
78 ///
79 /// ## Epoch
80 ///
81 /// Epoch value is monotonically increasing and used to identify unique write operations
82 ///
83 /// ## Example
84 ///
85 /// ```
86 /// let completion = frozen_core::ack::Completion::default();
87 ///
88 /// assert_eq!(completion.increment_current_epoch(), 1);
89 /// assert_eq!(completion.increment_current_epoch(), 2);
90 /// assert_eq!(completion.increment_current_epoch(), 3);
91 /// ```
92 #[inline]
93 pub fn increment_current_epoch(&self) -> TEpoch {
94 self.current_epoch.fetch_add(1, atomic::Ordering::AcqRel).wrapping_add(1)
95 }
96
97 /// Mark given [`TEpoch`] as durable
98 ///
99 /// ## Monotonic Epoch
100 ///
101 /// Once an epoch is marked durable, all earlier epochs are implicitly understood to be durable
102 ///
103 /// ## Example
104 ///
105 /// ```
106 /// use frozen_core::ack::Completion;
107 ///
108 /// let completion = Completion::default();
109 /// completion.mark_epoch_as_durable(0x0A);
110 ///
111 /// assert_eq!(completion.read_durable_epoch(), 0x0A);
112 /// ```
113 #[inline]
114 pub fn mark_epoch_as_durable(&self, epoch: TEpoch) {
115 self.durable_epoch.fetch_max(epoch, atomic::Ordering::Release);
116 }
117
118 /// Fetch the acknowledgement error (if any)
119 ///
120 /// ## Example
121 ///
122 /// ```
123 /// use frozen_core::ack::Completion;
124 ///
125 /// let completion = Completion::default();
126 /// assert_eq!(completion.get_err(), None);
127 /// ```
128 #[inline]
129 pub fn get_err(&self) -> Option<FrozenError> {
130 let lock = self.error.lock().unwrap_or_else(|e| e.into_inner());
131 lock.clone()
132 }
133
134 /// Update
135 /// (i.e. replace) current acknowledgement error w/ a new [`FrozenError`]
136 ///
137 /// ## Example
138 ///
139 /// ```
140 /// use frozen_core::{ack::Completion, error::{FrozenError, ErrCode}};
141 ///
142 /// let completion = Completion::default();
143 /// let new_error = FrozenError::new(0x10, 0x20, ErrCode::new(0x30, "io"), "failed to read file");
144 ///
145 /// completion.set_err(new_error.clone());
146 /// assert_eq!(completion.get_err(), Some(new_error));
147 /// ```
148 #[inline]
149 pub fn set_err(&self, new_error: FrozenError) {
150 let mut lock = self.error.lock().unwrap_or_else(|e| e.into_inner());
151 *lock = Some(new_error);
152 }
153
154 /// Clear acknowledgement error by replacing the underlying error w/ `None`
155 ///
156 /// ## Example
157 ///
158 /// ```
159 /// use frozen_core::{ack::Completion, error::{FrozenError, ErrCode}};
160 ///
161 /// let completion = Completion::default();
162 /// let new_error = FrozenError::new(0x10, 0x20, ErrCode::new(0x30, "io"), "failed to read file");
163 ///
164 /// completion.set_err(new_error.clone());
165 /// assert!(completion.get_err().is_some());
166 ///
167 /// completion.del_err();
168 /// assert!(completion.get_err().is_none());
169 /// ```
170 #[inline]
171 pub fn del_err(&self) {
172 let mut lock = self.error.lock().unwrap_or_else(|e| e.into_inner());
173 *lock = None;
174 }
175
176 /// Read the latest assigned epoch
177 ///
178 /// ## Example
179 ///
180 /// ```
181 /// use frozen_core::ack::Completion;
182 ///
183 /// let completion = Completion::default();
184 /// completion.increment_current_epoch();
185 ///
186 /// assert_eq!(completion.read_current_epoch(), 1);
187 /// ```
188 #[inline]
189 pub fn read_current_epoch(&self) -> TEpoch {
190 self.current_epoch.load(atomic::Ordering::Acquire)
191 }
192
193 /// Read the latest durable epoch
194 ///
195 /// ## Example
196 ///
197 /// ```
198 /// use frozen_core::ack::Completion;
199 ///
200 /// let completion = Completion::default();
201 /// completion.mark_epoch_as_durable(0x3A);
202 ///
203 /// assert_eq!(completion.read_durable_epoch(), 0x3A);
204 /// ```
205 #[inline]
206 pub fn read_durable_epoch(&self) -> TEpoch {
207 self.durable_epoch.load(atomic::Ordering::Acquire)
208 }
209
210 /// Wake all listeners currently waiting for durability progress
211 ///
212 /// Waking listeners does not modify any durable state and is typically called after advancing
213 /// the durable epoch or setting an error
214 ///
215 /// ## Example
216 ///
217 /// ```
218 /// use frozen_core::ack::Completion;
219 ///
220 /// let completion = Completion::default();
221 /// completion.notify_all_listeners();
222 /// ```
223 #[inline]
224 pub fn notify_all_listeners(&self) {
225 self.durability_condvar.notify_all();
226 }
227
228 /// Register a [`SyncTrigger`] implementation for triggering syncs
229 pub fn set_sync_trigger(&self, trigger: sync::Weak<dyn SyncTrigger>) {
230 let mut lock = self.sync_trigger.write().unwrap_or_else(|e| e.into_inner());
231 *lock = Some(trigger);
232 }
233
234 /// Trigger synchronization using the registered [`SyncTrigger`]
235 pub fn trigger_sync(&self) -> FrozenResult<()> {
236 let trigger = {
237 let lock = self.sync_trigger.read().unwrap_or_else(|e| e.into_inner());
238 lock.as_ref().and_then(|w| w.upgrade())
239 };
240
241 if let Some(trigger) = trigger {
242 trigger.trigger_sync()
243 } else {
244 Err(FrozenError::new(
245 0,
246 0x08,
247 ErrCode::new(0x0A, "failed to sync"),
248 "no sync trigger registered or file handle closed",
249 ))
250 }
251 }
252
253 /// Blocks the current thread until the specified `epoch` becomes durable or an error is
254 /// reported
255 pub fn wait_for_epoch(&self, epoch: TEpoch) -> FrozenResult<TEpoch> {
256 if self.read_durable_epoch() >= epoch {
257 return Ok(epoch);
258 }
259
260 if let Some(err) = self.get_err() {
261 return Err(err);
262 }
263
264 let mut guard = self.wait_mutex.lock().unwrap_or_else(|e| e.into_inner());
265 loop {
266 if self.read_durable_epoch() >= epoch {
267 return Ok(epoch);
268 }
269
270 if let Some(err) = self.get_err() {
271 return Err(err);
272 }
273
274 guard = self.durability_condvar.wait(guard).unwrap_or_else(|e| e.into_inner());
275 }
276 }
277}
278
279/// Durability handle associated with a write operation
280///
281/// ## Epoch
282///
283/// Every ticket is assigned a monotonically increasing epoch to monitor durability
284///
285/// ## Durability Guarantee
286///
287/// The ticket can be queried via [`is_durable`](Self::is_durable), waited on via
288/// [`wait`](Self::wait), or forced via [`force`](Self::force)
289///
290/// Once an epoch is confirmed durable, all writes assigned to earlier epochs are also guaranteed to
291/// be durable
292///
293/// Callers that only require fire-and-forget semantics may simply discard the returned ticket
294#[derive(Debug, Clone)]
295pub struct AckTicket {
296 epoch: TEpoch,
297 completion: sync::Arc<Completion>,
298}
299
300impl AckTicket {
301 /// Construct a new [`AckTicket`] for a write operation
302 ///
303 /// ## Example
304 ///
305 /// ```
306 /// use frozen_core::ack::{AckTicket, Completion};
307 /// use std::sync::Arc;
308 ///
309 /// let completion = Arc::new(Completion::default());
310 /// let ticket = AckTicket::new(1, completion);
311 ///
312 /// assert_eq!(ticket.epoch(), 1);
313 /// ```
314 #[inline]
315 pub const fn new(epoch: TEpoch, completion: sync::Arc<Completion>) -> Self {
316 Self { epoch, completion }
317 }
318
319 /// Read assigned durability epoch for the [`AckTicket`]
320 ///
321 /// ## Example
322 ///
323 /// ```
324 /// use frozen_core::ack::{AckTicket, Completion};
325 /// use std::sync::Arc;
326 ///
327 /// let completion = Arc::new(Completion::default());
328 /// let ticket = AckTicket::new(0x4C, completion);
329 ///
330 /// assert_eq!(ticket.epoch(), 0x4C);
331 /// ```
332 #[inline(always)]
333 pub const fn epoch(&self) -> TEpoch {
334 self.epoch
335 }
336
337 /// Check if the [`AckTicket`] is already durable w/o blocking
338 ///
339 /// ## Example
340 ///
341 /// ```
342 /// use frozen_core::ack::{AckTicket, Completion};
343 /// use std::sync::Arc;
344 ///
345 /// let completion = Arc::new(Completion::default());
346 /// let ticket = AckTicket::new(10, completion.clone());
347 ///
348 /// assert!(!ticket.is_durable());
349 /// completion.mark_epoch_as_durable(10);
350 /// assert!(ticket.is_durable());
351 /// ```
352 #[inline]
353 pub fn is_durable(&self) -> bool {
354 self.completion.read_durable_epoch() >= self.epoch
355 }
356
357 /// Blocks the current thread until the [`AckTicket`] becomes durable
358 ///
359 /// ## Error Propagation
360 ///
361 /// If a durability error is reported before the epoch becomes durable, the corresponding
362 /// [`FrozenError`] is returned instead
363 ///
364 /// ## Example
365 ///
366 /// ```
367 /// use frozen_core::ack::{AckTicket, Completion};
368 /// use std::{sync::Arc, thread, time};
369 ///
370 /// let completion = Arc::new(Completion::default());
371 /// let epoch = completion.increment_current_epoch();
372 /// let ticket = AckTicket::new(epoch, completion.clone());
373 ///
374 /// thread::spawn({
375 /// let completion = completion.clone();
376 /// move || {
377 /// thread::sleep(time::Duration::from_millis(10));
378 /// completion.mark_epoch_as_durable(epoch);
379 /// completion.notify_all_listeners();
380 /// }
381 /// });
382 ///
383 /// assert_eq!(ticket.wait().unwrap(), epoch);
384 /// ```
385 #[inline]
386 pub fn wait(&self) -> FrozenResult<TEpoch> {
387 self.completion.wait_for_epoch(self.epoch)
388 }
389
390 /// Forces synchronization for this ticket and blocks until it becomes durable
391 ///
392 /// If a background sync thread is running, signals the worker to sync immediately; otherwise
393 /// executes the sync synchronously under the lock
394 ///
395 /// ## Example
396 ///
397 /// ```
398 /// use frozen_core::ack::{AckTicket, Completion};
399 /// use std::sync::Arc;
400 ///
401 /// let completion = Arc::new(Completion::default());
402 /// completion.mark_epoch_as_durable(5);
403 /// let ticket = AckTicket::new(5, completion);
404 ///
405 /// assert_eq!(ticket.force().unwrap(), 5);
406 /// ```
407 #[inline]
408 pub fn force(&self) -> FrozenResult<TEpoch> {
409 if self.is_durable() {
410 return Ok(self.epoch);
411 }
412
413 if let Some(err) = self.completion.get_err() {
414 return Err(err);
415 }
416
417 self.completion.trigger_sync()?;
418 self.completion.wait_for_epoch(self.epoch)
419 }
420
421 /// Reference to the underlying shared [`Completion`]
422 #[inline]
423 pub fn completion(&self) -> &sync::Arc<Completion> {
424 &self.completion
425 }
426}
427
428#[cfg(test)]
429mod tests {
430 use super::*;
431 use std::{sync::Arc, thread, time::Duration};
432
433 #[test]
434 fn ok_ticket_epoch_and_is_durable() {
435 let completion = Arc::new(Completion::default());
436 let ticket = AckTicket::new(0x2A, completion.clone());
437
438 assert_eq!(ticket.epoch(), 0x2A);
439 assert!(!ticket.is_durable());
440
441 completion.mark_epoch_as_durable(0x29);
442 assert!(!ticket.is_durable());
443
444 completion.mark_epoch_as_durable(0x2A);
445 assert!(ticket.is_durable());
446
447 completion.mark_epoch_as_durable(0x64);
448 assert!(ticket.is_durable());
449 }
450
451 #[test]
452 fn ok_ticket_wait_blocks_until_durable() {
453 let completion = Arc::new(Completion::default());
454 let epoch = completion.increment_current_epoch();
455 let ticket = AckTicket::new(epoch, completion.clone());
456
457 let t = thread::spawn({
458 let completion = completion.clone();
459 move || {
460 thread::sleep(Duration::from_millis(0x14));
461 completion.mark_epoch_as_durable(epoch);
462 completion.notify_all_listeners();
463 }
464 });
465
466 assert_eq!(ticket.wait().unwrap(), epoch);
467 t.join().unwrap();
468 }
469
470 #[test]
471 fn err_ticket_wait_returns_error() {
472 let completion = Arc::new(Completion::default());
473 let epoch = completion.increment_current_epoch();
474 let ticket = AckTicket::new(epoch, completion.clone());
475
476 let err = FrozenError::new(1, 2, ErrCode::new(3, "test"), "io failure");
477
478 let t = thread::spawn({
479 let completion = completion.clone();
480 let err = err.clone();
481 move || {
482 thread::sleep(Duration::from_millis(0x14));
483 completion.set_err(err);
484 completion.notify_all_listeners();
485 }
486 });
487
488 let res = ticket.wait();
489 assert!(res.is_err());
490 assert_eq!(res.unwrap_err().reason, 3);
491 t.join().unwrap();
492 }
493
494 struct MockTrigger {
495 completion: Arc<Completion>,
496 synced: atomic::AtomicBool,
497 }
498
499 impl SyncTrigger for MockTrigger {
500 fn trigger_sync(&self) -> FrozenResult<()> {
501 self.synced.store(true, atomic::Ordering::Release);
502 let curr = self.completion.read_current_epoch();
503 self.completion.mark_epoch_as_durable(curr);
504 self.completion.notify_all_listeners();
505 Ok(())
506 }
507 }
508
509 #[test]
510 fn ok_ticket_force_with_trigger() {
511 let completion = Arc::new(Completion::default());
512 let trigger = Arc::new(MockTrigger {
513 completion: completion.clone(),
514 synced: atomic::AtomicBool::new(false),
515 });
516
517 let weak = Arc::downgrade(&trigger) as sync::Weak<dyn SyncTrigger>;
518 completion.set_sync_trigger(weak);
519
520 let epoch = completion.increment_current_epoch();
521 let ticket = AckTicket::new(epoch, completion.clone());
522
523 assert!(!ticket.is_durable());
524 assert_eq!(ticket.force().unwrap(), epoch);
525 assert!(trigger.synced.load(atomic::Ordering::Acquire));
526 assert!(ticket.is_durable());
527 }
528
529 #[test]
530 fn err_ticket_force_without_trigger() {
531 let completion = Arc::new(Completion::default());
532 let epoch = completion.increment_current_epoch();
533 let ticket = AckTicket::new(epoch, completion.clone());
534
535 let res = ticket.force();
536 assert!(res.is_err());
537 }
538}