gpui/
asset_cache.rs

1use crate::{App, EntityId, SharedString, SharedUri, Task};
2use collections::FxHashSet;
3use futures::{Future, TryFutureExt};
4
5use std::cell::RefCell;
6use std::fmt::Debug;
7use std::hash::{BuildHasher, Hash};
8use std::marker::PhantomData;
9use std::mem;
10use std::path::{Path, PathBuf};
11use std::rc::Rc;
12use std::sync::Arc;
13
14pub(crate) struct CachedLoad<T> {
15    state: Rc<RefCell<CachedLoadState<T>>>,
16    // Keep the owner outside the state so publishing a result cannot cancel its own task.
17    _task: Task<()>,
18}
19
20enum CachedLoadState<T> {
21    Loading(FxHashSet<EntityId>),
22    Loaded(T),
23}
24
25impl<T: Clone + Send + 'static> CachedLoad<T> {
26    pub(crate) fn new(future: impl Future<Output = T> + Send + 'static, cx: &App) -> Self {
27        let state = Rc::new(RefCell::new(CachedLoadState::Loading(FxHashSet::default())));
28        let task = cx.background_executor().spawn(future);
29        let task = cx.spawn({
30            let state = Rc::downgrade(&state);
31            async move |cx| {
32                let result = task.await;
33                let Some(state) = state.upgrade() else {
34                    return;
35                };
36                let previous =
37                    mem::replace(&mut *state.borrow_mut(), CachedLoadState::Loaded(result));
38                let CachedLoadState::Loading(views) = previous else {
39                    unreachable!("a cached load completes only once");
40                };
41                cx.update(|cx| {
42                    for view in views {
43                        cx.notify(view);
44                    }
45                });
46            }
47        });
48        Self { state, _task: task }
49    }
50
51    pub(crate) fn get(&self) -> Option<T> {
52        match &*self.state.borrow() {
53            CachedLoadState::Loading(_) => None,
54            CachedLoadState::Loaded(result) => Some(result.clone()),
55        }
56    }
57
58    pub(crate) fn use_by(&self, view: EntityId) -> Option<T> {
59        match &mut *self.state.borrow_mut() {
60            CachedLoadState::Loading(views) => {
61                views.insert(view);
62                None
63            }
64            CachedLoadState::Loaded(result) => Some(result.clone()),
65        }
66    }
67}
68
69/// An enum representing
70#[derive(Debug, PartialEq, Eq, Hash, Clone)]
71pub enum Resource {
72    /// This resource is at a given URI
73    Uri(SharedUri),
74    /// This resource is at a given path in the file system
75    Path(Arc<Path>),
76    /// This resource is embedded in the application binary
77    Embedded(SharedString),
78}
79
80impl From<SharedUri> for Resource {
81    fn from(value: SharedUri) -> Self {
82        Self::Uri(value)
83    }
84}
85
86impl From<PathBuf> for Resource {
87    fn from(value: PathBuf) -> Self {
88        Self::Path(value.into())
89    }
90}
91
92impl From<Arc<Path>> for Resource {
93    fn from(value: Arc<Path>) -> Self {
94        Self::Path(value)
95    }
96}
97
98/// A trait for asynchronous asset loading.
99pub trait Asset: 'static {
100    /// The source of the asset.
101    type Source: Clone + Hash + Send;
102
103    /// The loaded asset
104    type Output: Clone + Send;
105
106    /// Load the asset asynchronously
107    fn load(
108        source: Self::Source,
109        cx: &mut App,
110    ) -> impl Future<Output = Self::Output> + Send + 'static;
111}
112
113/// An asset Loader which logs the [`Err`] variant of a [`Result`] during loading
114pub enum AssetLogger<T> {
115    #[doc(hidden)]
116    _Phantom(PhantomData<T>, &'static dyn crate::seal::Sealed),
117}
118
119impl<T, R, E> Asset for AssetLogger<T>
120where
121    T: Asset<Output = Result<R, E>>,
122    R: Clone + Send,
123    E: Clone + Send + Debug,
124{
125    type Source = T::Source;
126
127    type Output = T::Output;
128
129    fn load(
130        source: Self::Source,
131        cx: &mut App,
132    ) -> impl Future<Output = Self::Output> + Send + 'static {
133        let load = T::load(source, cx);
134        load.inspect_err(|e| log::error!("Failed to load asset: {:?}", e))
135    }
136}
137
138/// Use a quick, non-cryptographically secure hash function to get an identifier from data
139pub fn hash<T: Hash>(data: &T) -> u64 {
140    collections::FxBuildHasher.hash_one(data)
141}
142
143#[cfg(test)]
144mod tests {
145    use super::*;
146    use crate::{AppContext, TestAppContext};
147    use futures::channel::oneshot;
148    use std::{
149        cell::Cell,
150        collections::VecDeque,
151        hash::{Hash, Hasher},
152        rc::Rc,
153        sync::{
154            Arc, Mutex,
155            atomic::{AtomicUsize, Ordering},
156        },
157    };
158
159    #[gpui::test]
160    fn cached_load_caches_success_and_error(cx: &mut TestAppContext) {
161        let successful_load =
162            cx.update(|cx| CachedLoad::new(async { Ok::<_, &'static str>(42) }, cx));
163        let failed_load =
164            cx.update(|cx| CachedLoad::new(async { Err::<i32, _>("load failed") }, cx));
165
166        assert_eq!(successful_load.get(), None);
167        assert_eq!(failed_load.get(), None);
168
169        cx.run_until_parked();
170
171        assert_eq!(successful_load.get(), Some(Ok(42)));
172        assert_eq!(successful_load.get(), Some(Ok(42)));
173        assert_eq!(failed_load.get(), Some(Err("load failed")));
174        assert_eq!(failed_load.get(), Some(Err("load failed")));
175    }
176
177    #[gpui::test]
178    fn completed_load_without_subscribers_is_cached_and_deduplicated(cx: &mut TestAppContext) {
179        let (sender, receiver) = oneshot::channel();
180        let source = TestAssetSource::new(1, [receiver]);
181
182        assert_eq!(cx.update(|cx| cx.fetch_asset::<TestAsset>(&source)), None);
183        assert_eq!(cx.update(|cx| cx.fetch_asset::<TestAsset>(&source)), None);
184        assert_eq!(source.load_count(), 1);
185
186        assert!(sender.send(Ok(42)).is_ok());
187        cx.run_until_parked();
188
189        assert_eq!(
190            cx.update(|cx| cx.fetch_asset::<TestAsset>(&source)),
191            Some(Ok(42))
192        );
193        assert_eq!(
194            cx.update(|cx| cx.fetch_asset::<TestAsset>(&source)),
195            Some(Ok(42))
196        );
197        assert_eq!(source.load_count(), 1);
198    }
199
200    #[gpui::test]
201    fn load_completion_notifies_each_observing_entity_once(cx: &mut TestAppContext) {
202        let (sender, receiver) = oneshot::channel::<i32>();
203        let load = cx.update(|cx| {
204            CachedLoad::new(
205                async move { receiver.await.expect("test sends a result") },
206                cx,
207            )
208        });
209        let first_notification_count = Rc::new(Cell::new(0));
210        let second_notification_count = Rc::new(Cell::new(0));
211        let (first_entity, second_entity) = cx.update(|cx| {
212            let first_entity = cx.new(|_| ());
213            let second_entity = cx.new(|_| ());
214            cx.observe(&first_entity, {
215                let first_notification_count = first_notification_count.clone();
216                move |_, _| first_notification_count.set(first_notification_count.get() + 1)
217            })
218            .detach();
219            cx.observe(&second_entity, {
220                let second_notification_count = second_notification_count.clone();
221                move |_, _| second_notification_count.set(second_notification_count.get() + 1)
222            })
223            .detach();
224            (first_entity, second_entity)
225        });
226
227        assert_eq!(load.use_by(first_entity.entity_id()), None);
228        assert_eq!(load.use_by(first_entity.entity_id()), None);
229        assert_eq!(load.use_by(second_entity.entity_id()), None);
230
231        assert!(sender.send(42).is_ok());
232        cx.run_until_parked();
233
234        assert_eq!(first_notification_count.get(), 1);
235        assert_eq!(second_notification_count.get(), 1);
236        assert_eq!(load.use_by(first_entity.entity_id()), Some(42));
237        assert_eq!(load.use_by(second_entity.entity_id()), Some(42));
238    }
239
240    #[gpui::test]
241    fn removing_pending_asset_cancels_it_and_replacement_ignores_stale_completion(
242        cx: &mut TestAppContext,
243    ) {
244        let (stale_sender, stale_receiver) = oneshot::channel();
245        let (replacement_sender, replacement_receiver) = oneshot::channel();
246        let source = TestAssetSource::new(2, [stale_receiver, replacement_receiver]);
247
248        assert_eq!(cx.update(|cx| cx.fetch_asset::<TestAsset>(&source)), None);
249        assert_eq!(source.load_count(), 1);
250        cx.run_until_parked();
251
252        cx.update(|cx| cx.remove_asset::<TestAsset>(&source));
253        assert!(!cx.update(|cx| cx.has_asset::<TestAsset>(&source)));
254        cx.run_until_parked();
255        assert!(stale_sender.send(Ok(1)).is_err());
256
257        assert_eq!(cx.update(|cx| cx.fetch_asset::<TestAsset>(&source)), None);
258        assert_eq!(source.load_count(), 2);
259
260        assert!(replacement_sender.send(Err("replacement failed")).is_ok());
261        cx.run_until_parked();
262
263        assert_eq!(
264            cx.update(|cx| cx.fetch_asset::<TestAsset>(&source)),
265            Some(Err("replacement failed"))
266        );
267        assert_eq!(source.load_count(), 2);
268    }
269
270    #[gpui::test]
271    fn eviction_after_worker_completion_does_not_publish_into_replacement(cx: &mut TestAppContext) {
272        let (sender, receiver) = oneshot::channel();
273        let (replacement_sender, replacement_receiver) = oneshot::channel();
274        let source = TestAssetSource::new(3, [receiver, replacement_receiver]);
275
276        assert_eq!(cx.update(|cx| cx.fetch_asset::<TestAsset>(&source)), None);
277        assert!(sender.send(Ok(1)).is_ok());
278        while source.completion_count.load(Ordering::SeqCst) == 0 {
279            assert!(cx.background_executor.tick());
280        }
281        assert_eq!(cx.update(|cx| cx.fetch_asset::<TestAsset>(&source)), None);
282
283        cx.update(|cx| {
284            cx.remove_asset::<TestAsset>(&source);
285            assert_eq!(cx.fetch_asset::<TestAsset>(&source), None);
286        });
287        cx.run_until_parked();
288        assert_eq!(cx.update(|cx| cx.fetch_asset::<TestAsset>(&source)), None);
289
290        assert!(replacement_sender.send(Ok(2)).is_ok());
291        cx.run_until_parked();
292        assert_eq!(
293            cx.update(|cx| cx.fetch_asset::<TestAsset>(&source)),
294            Some(Ok(2))
295        );
296        assert_eq!(source.load_count(), 2);
297    }
298
299    struct TestAsset;
300
301    #[derive(Clone)]
302    struct TestAssetSource {
303        id: usize,
304        load_count: Arc<AtomicUsize>,
305        completion_count: Arc<AtomicUsize>,
306        receivers: Arc<Mutex<VecDeque<oneshot::Receiver<Result<i32, &'static str>>>>>,
307    }
308
309    impl TestAssetSource {
310        fn new(
311            id: usize,
312            receivers: impl IntoIterator<Item = oneshot::Receiver<Result<i32, &'static str>>>,
313        ) -> Self {
314            Self {
315                id,
316                load_count: Arc::new(AtomicUsize::new(0)),
317                completion_count: Arc::new(AtomicUsize::new(0)),
318                receivers: Arc::new(Mutex::new(receivers.into_iter().collect())),
319            }
320        }
321
322        fn load_count(&self) -> usize {
323            self.load_count.load(Ordering::SeqCst)
324        }
325    }
326
327    impl Hash for TestAssetSource {
328        fn hash<H: Hasher>(&self, state: &mut H) {
329            self.id.hash(state);
330        }
331    }
332
333    impl Asset for TestAsset {
334        type Source = TestAssetSource;
335        type Output = Result<i32, &'static str>;
336
337        fn load(
338            source: Self::Source,
339            _cx: &mut App,
340        ) -> impl Future<Output = Self::Output> + Send + 'static {
341            source.load_count.fetch_add(1, Ordering::SeqCst);
342            let receiver = source
343                .receivers
344                .lock()
345                .expect("test receiver mutex should not be poisoned")
346                .pop_front()
347                .expect("each test load should have a receiver");
348            async move {
349                let result = match receiver.await {
350                    Ok(result) => result,
351                    Err(_) => Err("cancelled"),
352                };
353                source.completion_count.fetch_add(1, Ordering::SeqCst);
354                result
355            }
356        }
357    }
358}
gpui/asset_cache.rs

gpui/
asset_cache.rs
