radiate_core/domain/sync/
group.rs

1use std::sync::{
2    Arc, Condvar, Mutex,
3    atomic::{AtomicUsize, Ordering},
4};
5
6struct Inner {
7    counter: AtomicUsize,
8    lock: Mutex<()>,
9    cvar: Condvar,
10}
11
12#[derive(Clone)]
13pub struct WaitGroup {
14    inner: Arc<Inner>,
15    total_count: Arc<AtomicUsize>,
16}
17
18impl WaitGroup {
19    pub fn new() -> Self {
20        Self {
21            inner: Arc::new(Inner {
22                counter: AtomicUsize::new(0),
23                lock: Mutex::new(()),
24                cvar: Condvar::new(),
25            }),
26            total_count: Arc::new(AtomicUsize::new(0)),
27        }
28    }
29
30    pub fn get_count(&self) -> usize {
31        self.total_count.load(Ordering::Acquire)
32    }
33
34    pub fn guard(&self) -> WaitGuard {
35        self.inner.counter.fetch_add(1, Ordering::AcqRel);
36        self.total_count.fetch_add(1, Ordering::AcqRel);
37
38        WaitGuard { wg: self.clone() }
39    }
40
41    /// Waits until the counter reaches zero.
42    pub fn wait(&self) -> usize {
43        if self.inner.counter.load(Ordering::Acquire) == 0 {
44            return 0;
45        }
46
47        let lock = self.inner.lock.lock().unwrap();
48        let _unused = self
49            .inner
50            .cvar
51            .wait_while(lock, |_| self.inner.counter.load(Ordering::Acquire) != 0);
52
53        self.get_count()
54    }
55}
56
57impl Default for WaitGroup {
58    fn default() -> Self {
59        Self::new()
60    }
61}
62
63pub struct WaitGuard {
64    wg: WaitGroup,
65}
66
67impl Drop for WaitGuard {
68    fn drop(&mut self) {
69        if self.wg.inner.counter.fetch_sub(1, Ordering::AcqRel) == 1 {
70            let _guard = self.wg.inner.lock.lock().unwrap();
71            self.wg.inner.cvar.notify_all();
72        }
73    }
74}
radiate_core/domain/sync/group.rs

radiate_core/domain/sync/
group.rs
