zenoh_sync/
cache.rs

1//
2// Copyright (c) 2025 ZettaScale Technology
3//
4// This program and the accompanying materials are made available under the
5// terms of the Eclipse Public License 2.0 which is available at
6// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
7// which is available at https://www.apache.org/licenses/LICENSE-2.0.
8//
9// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
10//
11// Contributors:
12//   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
13//
14
15use std::sync::{
16    atomic::{AtomicBool, Ordering},
17    Arc,
18};
19
20use arc_swap::{ArcSwap, Guard};
21
22#[derive(Debug)]
23pub struct CacheValue<T: Sized> {
24    version: usize,
25    value: T,
26}
27
28impl<T> CacheValue<T> {
29    pub fn get_ref(&self) -> &T {
30        &self.value
31    }
32}
33
34/// This is a lock-free concurrent cache.
35/// It stores only the most up-to-date value.
36pub struct Cache<T> {
37    value: ArcSwap<CacheValue<T>>,
38    is_updating: AtomicBool,
39}
40
41impl<T> std::fmt::Debug for Cache<T> {
42    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
43        f.debug_struct("Cache")
44            .field("is_updating", &self.is_updating.load(Ordering::SeqCst))
45            .finish_non_exhaustive()
46    }
47}
48
49pub type CacheValueType<T> = Guard<Arc<CacheValue<T>>>;
50
51impl<T> Cache<T> {
52    pub fn new(value: T, version: usize) -> Self {
53        Cache {
54            value: ArcSwap::new(CacheValue::<T> { version, value }.into()),
55            is_updating: AtomicBool::new(false),
56        }
57    }
58
59    fn finish_update(&self) {
60        self.is_updating.store(false, Ordering::SeqCst);
61    }
62
63    /// Tries to retrieve value for the specified version.
64    /// Returns a result either containing a cached value, or an f (which is guaranteed to be not invoked by function call in this case).
65    /// If requested version corresponds to the value currently stored in cache - the value is returned.
66    /// If requested version is older None will be returned.
67    /// If requested version is newer, the new value will be computed and stored by calling f, and then returned,
68    /// unless the value is being currently updated - in this case None will be returned.
69    /// If None is returned it is guaranteed that f was not called.
70    pub fn value(
71        &self,
72        version: usize,
73        f: impl FnOnce() -> T,
74    ) -> Result<CacheValueType<T>, impl FnOnce() -> T> {
75        let v = self.value.load();
76        match v.version.cmp(&version) {
77            std::cmp::Ordering::Equal => Ok(v),
78            std::cmp::Ordering::Greater => Err(f), //requesting too old version
79            std::cmp::Ordering::Less => {
80                // try to update
81                drop(v);
82                match self.is_updating.compare_exchange(
83                    false,
84                    true,
85                    Ordering::SeqCst,
86                    Ordering::SeqCst,
87                ) {
88                    Ok(_) => {
89                        let v = self.value.load();
90                        match v.version.cmp(&version) {
91                            std::cmp::Ordering::Equal => {
92                                // already updated by someone else to the version we need
93                                self.finish_update();
94                                Ok(v)
95                            }
96                            std::cmp::Ordering::Greater => {
97                                // already updated by someone else beyond the version we need
98                                self.finish_update();
99                                Err(f)
100                            }
101                            std::cmp::Ordering::Less => {
102                                drop(v);
103                                self.value.store(
104                                    CacheValue {
105                                        value: f(),
106                                        version,
107                                    }
108                                    .into(),
109                                );
110                                let v = self.value.load(); // is_updating set to true guarantees that nobody else will modify the value.
111                                self.finish_update();
112                                Ok(v)
113                            }
114                        }
115                    }
116                    Err(_) => Err(f),
117                }
118            }
119        }
120    }
121}
122
123#[cfg(test)]
124mod tests {
125    use std::{sync::Arc, time::Duration};
126
127    use super::Cache;
128
129    #[test]
130    fn test_cache() {
131        let cache = Cache::<String>::new("0".to_string(), 0);
132
133        assert_eq!(
134            cache
135                .value(0, || { "1".to_string() })
136                .as_ref()
137                .map(|v| v.get_ref().as_str())
138                .unwrap_or(""),
139            "0"
140        );
141        assert_eq!(
142            cache
143                .value(1, || { "1".to_string() })
144                .as_ref()
145                .map(|v| v.get_ref().as_str())
146                .unwrap_or(""),
147            "1"
148        );
149        assert!(cache.value(0, || { "2".to_string() }).is_err());
150
151        // try to get-update value from another thread
152        let cache = Arc::new(cache);
153        let cache2 = cache.clone();
154        std::thread::spawn(move || {
155            let res = cache2.value(2, || {
156                std::thread::sleep(Duration::from_secs(5));
157                "2".to_string()
158            });
159            assert_eq!(
160                res.as_ref().map(|v| v.get_ref().as_str()).unwrap_or(""),
161                "2"
162            );
163        });
164        std::thread::sleep(Duration::from_secs(1));
165        while cache.value(2, || "".to_string()).is_err() {
166            std::thread::sleep(Duration::from_secs(1));
167        }
168        assert_eq!(
169            cache
170                .value(2, || { "".to_string() })
171                .as_ref()
172                .map(|v| v.get_ref().as_str())
173                .unwrap_or(""),
174            "2"
175        );
176    }
177}
zenoh_sync/cache.rs

zenoh_sync/
cache.rs
