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

gpui_core/
asset_cache.rs
