cljrs_value/collections/
transient_set.rs

1use super::Snapshot;
2use crate::hash::hash_combine_unordered;
3use crate::{ClojureHash, PersistentHashSet, Value, ValueError, ValueResult};
4use std::sync::{
5    Mutex,
6    atomic::{AtomicBool, AtomicU64, Ordering},
7};
8
9#[derive(Debug)]
10pub struct TransientSet {
11    set: Mutex<PersistentHashSet>,
12    persisted: Mutex<bool>,
13    revision: AtomicU64,
14    keys_may_realize: AtomicBool,
15}
16
17impl TransientSet {
18    pub fn new() -> Self {
19        TransientSet {
20            set: Mutex::new(PersistentHashSet::empty()),
21            persisted: Mutex::new(false),
22            revision: AtomicU64::new(0),
23            keys_may_realize: AtomicBool::new(false),
24        }
25    }
26
27    pub fn new_from_set(set: &PersistentHashSet) -> TransientSet {
28        TransientSet {
29            set: Mutex::new(set.clone()),
30            persisted: Mutex::new(false),
31            revision: AtomicU64::new(0),
32            keys_may_realize: AtomicBool::new(!(set.iter().all(crate::realization::is_inert))),
33        }
34    }
35
36    fn update(
37        &self,
38        key: &Value,
39        fast: impl FnOnce(&mut PersistentHashSet),
40        operation: impl Fn(&PersistentHashSet) -> PersistentHashSet,
41    ) -> ValueResult<()> {
42        {
43            let mut collection = self.set.lock().unwrap();
44            if *self.persisted.lock().unwrap() {
45                return Err(ValueError::TransientAlreadyPersisted);
46            }
47            if !self.keys_may_realize.load(Ordering::Relaxed) && crate::realization::is_inert(key) {
48                fast(&mut collection);
49                self.revision.fetch_add(1, Ordering::Relaxed);
50                return Ok(());
51            }
52        }
53        loop {
54            let (snapshot, revision) = {
55                let set = self.set.lock().unwrap();
56                if *self.persisted.lock().unwrap() {
57                    return Err(ValueError::TransientAlreadyPersisted);
58                }
59                (
60                    Snapshot::new(set.clone()),
61                    self.revision.load(Ordering::Relaxed),
62                )
63            };
64            let candidate = operation(snapshot.get());
65            let mut set = self.set.lock().unwrap();
66            if *self.persisted.lock().unwrap() {
67                return Err(ValueError::TransientAlreadyPersisted);
68            }
69            if self.revision.load(Ordering::Relaxed) != revision {
70                continue;
71            }
72            *set = candidate;
73            if !crate::realization::is_inert(key) {
74                self.keys_may_realize.store(true, Ordering::Relaxed);
75            }
76            self.revision.fetch_add(1, Ordering::Relaxed);
77            return Ok(());
78        }
79    }
80    pub fn conj(&self, value: Value) -> ValueResult<()> {
81        self.update(
82            &value,
83            |set| {
84                set.conj_mut(value.clone());
85            },
86            |set| set.conj(value.clone()),
87        )
88    }
89    pub fn disj(&self, value: &Value) -> ValueResult<()> {
90        self.update(value, |set| set.disj_mut(value), |set| set.disj(value))
91    }
92
93    pub fn persistent(&self) -> ValueResult<PersistentHashSet> {
94        let set = self.set.lock().unwrap();
95        let mut persisted = self.persisted.lock().unwrap();
96        if *persisted {
97            return Err(ValueError::TransientAlreadyPersisted);
98        }
99        *persisted = true;
100        Ok(set.clone())
101    }
102
103    pub fn count(&self) -> usize {
104        let set = self.set.lock().unwrap();
105        set.count()
106    }
107
108    pub fn contains(&self, value: &Value) -> bool {
109        let set = self.set.lock().unwrap();
110        if !self.keys_may_realize.load(Ordering::Relaxed) && crate::realization::is_inert(value) {
111            return set.contains(value);
112        }
113        let snapshot = Snapshot::new(set.clone());
114        drop(set);
115        snapshot.get().contains(value)
116    }
117}
118
119impl Clone for TransientSet {
120    fn clone(&self) -> Self {
121        Self {
122            set: Mutex::new(self.set.lock().unwrap().clone()),
123            persisted: Mutex::new(*self.persisted.lock().unwrap()),
124            revision: AtomicU64::new(0),
125            keys_may_realize: AtomicBool::new(self.keys_may_realize.load(Ordering::Relaxed)),
126        }
127    }
128}
129
130impl ClojureHash for TransientSet {
131    fn clojure_hash(&self) -> u32 {
132        let mut hash: u32 = 0;
133        let snapshot = Snapshot::new(self.set.lock().unwrap().clone());
134        for v in snapshot.get().iter() {
135            hash = hash_combine_unordered(hash, v.clojure_hash())
136        }
137        hash
138    }
139}
140
141impl cljrs_gc::Trace for TransientSet {
142    fn trace(&self, visitor: &mut cljrs_gc::MarkVisitor) {
143        {
144            let set = self.set.lock().unwrap();
145            for v in set.iter() {
146                v.trace(visitor);
147            }
148        }
149    }
150
151    fn gc_size_extra(&self) -> usize {
152        let set = self.set.lock().unwrap();
153        set.full_size_extra()
154    }
155}
156
157impl Default for TransientSet {
158    fn default() -> Self {
159        Self::new()
160    }
161}
cljrs_value/collections/transient_set.rs

cljrs_value/collections/
transient_set.rs
