fedimint_core/db/
notifications.rs

1use std::collections::hash_map::DefaultHasher;
2use std::hash::{Hash, Hasher};
3
4use bitvec::vec::BitVec;
5use tokio::sync::Notify;
6use tokio::sync::futures::Notified;
7
8/// Number of buckets used for `Notifications`.
9const NOTIFY_BUCKETS: usize = 32;
10
11/// The state of Notification.
12///
13/// This stores `NOTIFY_BUCKETS` number of `Notifies`.
14/// Each key is assigned a bucket based on its hash value.
15/// This will cause some false positives.
16#[derive(Debug)]
17pub struct Notifications {
18    buckets: Vec<Notify>,
19}
20
21impl Default for Notifications {
22    fn default() -> Self {
23        Self {
24            buckets: (0..NOTIFY_BUCKETS).map(|_| Notify::new()).collect(),
25        }
26    }
27}
28
29fn slot_index_for_hash(hash_value: u64) -> usize {
30    (hash_value % (NOTIFY_BUCKETS as u64)) as usize
31}
32
33fn slot_index_for_key<K: Hash>(key: K) -> usize {
34    let mut hasher = DefaultHasher::new();
35    key.hash(&mut hasher);
36    let hash_value = hasher.finish();
37    slot_index_for_hash(hash_value)
38}
39
40impl Notifications {
41    pub fn new() -> Self {
42        Self::default()
43    }
44
45    /// This registers for notification when called.
46    ///
47    /// Then waits for the notification when .awaited.
48    ///
49    /// NOTE: This may some false positives.
50    pub fn register<K>(&self, key: K) -> Notified
51    where
52        K: Hash,
53    {
54        self.buckets[slot_index_for_key(key)].notified()
55    }
56
57    /// Notify a key.
58    ///
59    /// All the waiters for this keys will be notified.
60    pub fn notify<K>(&self, key: K)
61    where
62        K: Hash,
63    {
64        self.buckets[slot_index_for_key(key)].notify_waiters();
65    }
66
67    /// Notifies the waiters about the notifications recorded in `NotifyQueue`.
68    pub fn submit_queue(&self, queue: &NotifyQueue) {
69        for bucket in queue.buckets.iter_ones() {
70            self.buckets[bucket].notify_waiters();
71        }
72    }
73}
74
75/// Save notifications to be sent after transaction is complete.
76#[derive(Debug)]
77pub struct NotifyQueue {
78    buckets: BitVec,
79}
80
81impl Default for NotifyQueue {
82    fn default() -> Self {
83        Self {
84            buckets: BitVec::repeat(false, NOTIFY_BUCKETS),
85        }
86    }
87}
88
89impl NotifyQueue {
90    pub fn new() -> Self {
91        Self::default()
92    }
93
94    pub fn add<K>(&mut self, key: &K)
95    where
96        K: Hash,
97    {
98        self.buckets.set(slot_index_for_key(key), true);
99    }
100}
101
102#[cfg(test)]
103mod tests;
fedimint_core/db/notifications.rs

fedimint_core/db/
notifications.rs
