amadan

oficina / docs Public

Clone 
git clone https://amadan.net/oficina/docs

Plain git — no account needed to clone.

Download

Download this file

1use std::sync::{Arc, Weak};
2
3use arc_swap::{ArcSwapOption, AsRaw, Guard};
4
5use crate::Origin;
6
7/// Data structure used to handle publish/subscribe callbacks of specific type. Observers perform
8/// subscriber changes in thread-safe manner, using atomic hardware intrinsics.
9pub struct Observer<F> {
10 inner: ArcSwapOption<Inner<F>>,
11}
12
13impl<F> Observer<F>
14where
15 F: 'static,
16{
17 /// Creates a new [Observer] with no active callbacks.
18 pub fn new() -> Self {
19 Observer {
20 inner: ArcSwapOption::new(None),
21 }
22 }
23
24 pub fn has_subscribers(&self) -> bool {
25 if let Some(inner) = &*self.inner.load() {
26 inner.head.load().is_some()
27 } else {
28 false
29 }
30 }
31
32 /// Cleanup already released subscriptions. Whenever a [Subscription] is dropped, the callback is released. However,
33 /// the weak reference to callback may still be kept around until it becomes touched by operations such as
34 /// [Observer::subscribe] or [Observer::callbacks].
35 ///
36 /// This method allows to perform stale callback cleanup without waiting for callbacks to be visited.
37 pub fn clean(&self) {
38 self.inner.swap(None);
39 }
40
41 fn inner(&self) -> Arc<Inner<F>> {
42 let cur = self.inner.load_full();
43 match cur {
44 Some(inner) => inner,
45 None => {
46 // inner was not initialized yet, we need to create a new one
47 let inner = Arc::new(Inner {
48 head: ArcSwapOption::new(None),
49 });
50 let old: Option<Arc<Inner<F>>> = None;
51 let prev = self.inner.compare_and_swap(&old, Some(inner.clone()));
52 // there's a slight possibility that inner was initialized twice, in that case
53 // return first swapped inner
54 Guard::into_inner(prev).unwrap_or(inner)
55 }
56 }
57 }
58
59 fn remove(mut prev: Arc<Node<F>>, id: &Origin) -> bool {
60 while let Some(next) = prev.next.load_full() {
61 if &next.uid == id {
62 prev.next.store(next.next.load_full());
63 return true;
64 }
65 prev = next;
66 }
67 false
68 }
69
70 pub fn unsubscribe(&self, id: &Origin) -> bool {
71 if let Some(inner) = &*self.inner.load() {
72 inner.remove(id)
73 } else {
74 false
75 }
76 }
77
78 /// Returns a snapshot of callbacks subscribed to this observer at the moment when this method
79 /// has been called. This snapshot can be iterated over to get access to individual callbacks
80 /// and trigger them.
81 pub fn trigger<E>(&self, mut each: E)
82 where
83 E: FnMut(&F),
84 {
85 if let Some(inner) = &*self.inner.load() {
86 let mut next = inner.head.load();
87 while let Some(node) = &*next {
88 each(&node.callback);
89 next = node.next.load();
90 }
91 }
92 }
93
94 /// Subscribes a callback parameter to a current [Observer].
95 /// Returns a subscription object which - when dropped - will unsubscribe current callback.
96 /// If the `id` was already present in the observer, current callback will be ignored.
97 pub fn subscribe_with(&self, id: Origin, callback: F) {
98 let inner = self.inner();
99 let mut node = Arc::new(Node::new(id.clone(), callback));
100 let cur = inner.head.load();
101 let head = loop {
102 {
103 // update new node next pointer to point to current head
104 // it's safe to unwrap, since until current node is successfully inserted
105 // there will be no more that a single Arc reference to it
106 let n = Arc::get_mut(&mut node).unwrap();
107 n.next.store(cur.clone());
108 }
109
110 let prev = inner.head.compare_and_swap(&*cur, Some(node.clone()));
111 let swapped = std::ptr::eq(prev.as_raw(), cur.as_raw());
112 if swapped {
113 // we successfully swapped the head, we can exit the loop
114 break node;
115 }
116 };
117 // remove all previous nodes that share the same ID
118 Self::remove(head.clone(), &id);
119 }
120}
121
122#[cfg(feature = "sync")]
123impl<F> Observer<F>
124where
125 F: Send + Sync + 'static,
126{
127 pub fn subscribe(&self, callback: F) -> Subscription {
128 let mut rng = fastrand::Rng::new();
129 let id = rng.usize(0..usize::MAX);
130 let origin = Origin::from(id);
131 self.subscribe_with(origin.clone(), callback);
132 Arc::new(Cancel {
133 id: origin,
134 inner: Arc::downgrade(&self.inner()),
135 })
136 }
137}
138
139#[cfg(not(feature = "sync"))]
140impl<F> Observer<F>
141where
142 F: 'static,
143{
144 pub fn subscribe(&self, callback: F) -> Subscription {
145 let mut rng = fastrand::Rng::new();
146 let id = rng.usize(0..usize::MAX);
147 let origin = Origin::from(id);
148 self.subscribe_with(origin.clone(), callback);
149 Arc::new(Cancel {
150 id: origin,
151 inner: Arc::downgrade(&self.inner()),
152 })
153 }
154}
155
156#[cfg(feature = "sync")]
157impl<F> Default for Observer<F>
158where
159 F: Send + Sync + 'static,
160{
161 fn default() -> Self {
162 Observer::new()
163 }
164}
165
166#[cfg(not(feature = "sync"))]
167impl<F> Default for Observer<F>
168where
169 F: 'static,
170{
171 fn default() -> Self {
172 Observer::new()
173 }
174}
175
176struct Inner<F> {
177 head: ArcSwapOption<Node<F>>,
178}
179
180impl<F> Inner<F>
181where
182 F: 'static,
183{
184 fn remove(&self, id: &Origin) -> bool {
185 while let Some(head) = self.head.load_full() {
186 if &head.uid == id {
187 // the element to remove is the head of the list
188 // we need to swap head pointer of self to the next element
189 let next = head.next.load_full();
190 let prev = self.head.compare_and_swap(&head, next);
191 if !std::ptr::eq(prev.as_raw(), Arc::as_ptr(&head)) {
192 // head changed, retry
193 continue;
194 } else {
195 return true;
196 }
197 } else {
198 // the element to remove is somewhere in the middle of the list
199 // we need to find it and repoint its predecessor's next pointer
200 // to its successor
201 return Observer::remove(head.clone(), id);
202 }
203 }
204 false
205 }
206}
207
208struct Node<T> {
209 uid: Origin,
210 callback: T,
211 next: ArcSwapOption<Node<T>>,
212}
213
214impl<F> Node<F> {
215 fn new(uid: Origin, callback: F) -> Self {
216 Node {
217 uid,
218 callback,
219 next: Default::default(),
220 }
221 }
222}
223
224#[cfg(feature = "sync")]
225struct Cancel<F>
226where
227 F: Send + Sync + 'static,
228{
229 id: Origin,
230 inner: Weak<Inner<F>>,
231}
232
233#[cfg(feature = "sync")]
234impl<F> Drop for Cancel<F>
235where
236 F: Send + Sync + 'static,
237{
238 fn drop(&mut self) {
239 if let Some(inner) = self.inner.upgrade() {
240 inner.remove(&self.id);
241 }
242 }
243}
244
245#[cfg(not(feature = "sync"))]
246struct Cancel<F>
247where
248 F: 'static,
249{
250 id: Origin,
251 inner: Weak<Inner<F>>,
252}
253
254#[cfg(not(feature = "sync"))]
255impl<F> Drop for crate::observer::Cancel<F>
256where
257 F: 'static,
258{
259 fn drop(&mut self) {
260 if let Some(inner) = self.inner.upgrade() {
261 inner.remove(&self.id);
262 }
263 }
264}
265
266/// Subscription handle returned by [Observer::subscribe] methods, which will unsubscribe corresponding
267/// callback when dropped.
268///
269/// If you need to send the Subscription handle to another thread, you may wish to enable
270/// the `sync` feature such that Subscription implements `Send+Sync`.
271#[cfg(feature = "sync")]
272pub type Subscription = Arc<dyn Drop + Send + Sync + 'static>;
273
274/// Subscription handle returned by [Observer::subscribe] methods, which will unsubscribe corresponding
275/// callback when dropped.
276///
277/// If you need to send the Subscription handle to another thread, you may wish to enable
278/// the `sync` feature such that Subscription implements `Send+Sync`.
279#[cfg(not(feature = "sync"))]
280pub type Subscription = Arc<dyn Drop + 'static>;
281
282#[cfg(test)]
283mod test {
284 use std::sync::atomic::{AtomicI32, AtomicU32, Ordering};
285 use std::sync::Arc;
286
287 use crate::observer::Observer;
288
289 #[test]
290 fn subscription() {
291 let o: Observer<Box<dyn Fn(&u32) + Send + Sync + 'static>> = Observer::new();
292 let s1_state = Arc::new(AtomicU32::new(0));
293 let s2_state = Arc::new(AtomicU32::new(0));
294
295 {
296 let a = s1_state.clone();
297 let b = s2_state.clone();
298
299 let _s1 = o.subscribe(Box::new(move |&value| {
300 a.store(value, Ordering::Release);
301 }));
302 let _s2 = o.subscribe(Box::new(move |&value| {
303 b.store(value * 2, Ordering::Release);
304 }));
305
306 o.trigger(|fun| fun(&1));
307 assert_eq!(s1_state.load(Ordering::Acquire), 1);
308 assert_eq!(s2_state.load(Ordering::Acquire), 2);
309
310 o.trigger(|fun| fun(&2));
311 assert_eq!(s1_state.load(Ordering::Acquire), 2);
312 assert_eq!(s2_state.load(Ordering::Acquire), 4);
313 }
314
315 // subscriptions were dropped, we don't expect updates to be propagated
316
317 o.trigger(|fun| fun(&3));
318 assert_eq!(s1_state.load(Ordering::Acquire), 2);
319 assert_eq!(s2_state.load(Ordering::Acquire), 4);
320 }
321
322 #[test]
323 fn subscribers_predicate() {
324 let o: Observer<Box<dyn Fn(&u32) + Send + Sync + 'static>> = Observer::new();
325 assert!(!o.has_subscribers());
326
327 let _sub = o.subscribe(Box::new(move |_| {}));
328 assert!(o.has_subscribers());
329
330 drop(_sub);
331 o.clean();
332
333 assert!(!o.has_subscribers());
334 }
335
336 #[test]
337 #[cfg(feature = "sync")]
338 fn multi_threading() {
339 let o: Observer<Box<dyn Fn(u32) + Send + Sync + 'static>> = Observer::new();
340
341 let s1_state = Arc::new(AtomicU32::new(0));
342 let a = s1_state.clone();
343 let sub1 = o.subscribe(Box::new(move |v| a.store(v, Ordering::Release)));
344
345 let s2_state = Arc::new(AtomicU32::new(0));
346 let b = s2_state.clone();
347 let sub2 = o.subscribe(Box::new(move |v| b.store(v, Ordering::Release)));
348
349 let handle = std::thread::spawn(move || {
350 o.trigger(|fun| fun(1));
351 drop(sub1);
352 drop(sub2);
353 });
354
355 handle.join().unwrap();
356
357 assert_eq!(s1_state.load(Ordering::Acquire), 1);
358 assert_eq!(s2_state.load(Ordering::Acquire), 1);
359 }
360
361 #[test]
362 fn subscribe_with_replaced_old_callback() {
363 let (tx, rx) = std::sync::mpsc::channel();
364 let o: Observer<Box<dyn Fn(u32) + Send + Sync + 'static>> = Observer::new();
365 let ta = tx.clone();
366 let _a = o.subscribe_with(
367 123.into(),
368 Box::new(move |i| ta.send(format!("a-{i}")).unwrap()),
369 );
370 o.trigger(|fun| fun(1));
371 assert_eq!(rx.try_recv().unwrap(), "a-1");
372
373 // override the callback with the same key
374 let _b = o.subscribe_with(
375 123.into(),
376 Box::new(move |i| tx.send(format!("b-{i}")).unwrap()),
377 );
378 o.trigger(|fun| fun(2));
379 assert_eq!(rx.try_recv().unwrap(), "b-2");
380 }
381
382 struct DropCounter(Arc<AtomicI32>);
383
384 impl DropCounter {
385 fn new(counter: Arc<AtomicI32>) -> Self {
386 counter.fetch_add(1, Ordering::SeqCst);
387 DropCounter(counter)
388 }
389 }
390
391 impl Drop for DropCounter {
392 fn drop(&mut self) {
393 self.0.fetch_sub(1, Ordering::SeqCst);
394 }
395 }
396
397 #[test]
398 fn drop_subscription() {
399 let counter = Arc::new(AtomicI32::new(0));
400 let o: Observer<DropCounter> = Observer::new();
401 for _ in 0..100 {
402 assert_eq!(counter.load(Ordering::SeqCst), 0);
403 let _sub = o.subscribe(DropCounter::new(counter.clone()));
404 assert_eq!(counter.load(Ordering::SeqCst), 1);
405 // drop subscription
406 }
407 }
408
409 #[test]
410 fn drop_subscription2() {
411 let counter = Arc::new(AtomicI32::new(0));
412 let o: Observer<DropCounter> = Observer::new();
413 let mut subscriptions = Vec::new();
414 for _ in 0..100 {
415 let sub = o.subscribe(DropCounter::new(counter.clone()));
416 subscriptions.push(sub);
417 }
418 assert_eq!(counter.load(Ordering::SeqCst), 100);
419 drop(subscriptions);
420 assert_eq!(counter.load(Ordering::SeqCst), 0);
421 }
422
423 #[test]
424 fn unsubscribe() {
425 let counter = Arc::new(AtomicI32::new(0));
426 let o: Observer<DropCounter> = Observer::new();
427 for i in 0..100 {
428 assert_eq!(counter.load(Ordering::SeqCst), 0);
429
430 o.subscribe_with(i.into(), DropCounter::new(counter.clone()));
431
432 assert_eq!(counter.load(Ordering::SeqCst), 1);
433
434 let unsubscribed = o.unsubscribe(&i.into());
435 assert!(unsubscribed, "unsubscribe failed for {}", i);
436 }
437 }
438
439 #[test]
440 fn unsubscribe2() {
441 let counter = Arc::new(AtomicI32::new(0));
442 let o: Observer<DropCounter> = Observer::new();
443 for i in 0..100 {
444 o.subscribe_with(i.into(), DropCounter::new(counter.clone()));
445 }
446
447 assert_eq!(counter.load(Ordering::SeqCst), 100);
448
449 for i in 0..100 {
450 let unsubscribed = o.unsubscribe(&i.into());
451 assert!(unsubscribed, "unsubscribe failed for {}", i);
452 }
453 assert_eq!(counter.load(Ordering::SeqCst), 0);
454 }
455
456 #[test]
457 fn clean() {
458 let counter = Arc::new(AtomicI32::new(0));
459 let o: Observer<DropCounter> = Observer::new();
460 let mut subscriptions = Vec::new();
461 for _ in 0..100 {
462 let sub = o.subscribe(DropCounter::new(counter.clone()));
463 subscriptions.push(sub);
464 }
465 assert_eq!(counter.load(Ordering::SeqCst), 100);
466 o.clean();
467 assert_eq!(counter.load(Ordering::SeqCst), 0);
468 }
469}
470