gpui_authoring/
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;
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#[cfg(test)]
139mod tests {
140    use super::*;
141    use crate::{AppContext, TestAppContext};
142    use futures::channel::oneshot;
143    use std::{
144        cell::Cell,
145        collections::VecDeque,
146        hash::{Hash, Hasher},
147        rc::Rc,
148        sync::{
149            Arc, Mutex,
150            atomic::{AtomicUsize, Ordering},
151        },
152    };
153
154    #[gpui::test]
155    fn cached_load_caches_success_and_error(cx: &mut TestAppContext) {
156        let successful_load =
157            cx.update(|cx| CachedLoad::new(async { Ok::<_, &'static str>(42) }, cx));
158        let failed_load =
159            cx.update(|cx| CachedLoad::new(async { Err::<i32, _>("load failed") }, cx));
160
161        assert_eq!(successful_load.get(), None);
162        assert_eq!(failed_load.get(), None);
163
164        cx.run_until_parked();
165
166        assert_eq!(successful_load.get(), Some(Ok(42)));
167        assert_eq!(successful_load.get(), Some(Ok(42)));
168        assert_eq!(failed_load.get(), Some(Err("load failed")));
169        assert_eq!(failed_load.get(), Some(Err("load failed")));
170    }
171
172    #[gpui::test]
173    fn completed_load_without_subscribers_is_cached_and_deduplicated(cx: &mut TestAppContext) {
174        let (sender, receiver) = oneshot::channel();
175        let source = TestAssetSource::new(1, [receiver]);
176
177        assert_eq!(cx.update(|cx| cx.fetch_asset::<TestAsset>(&source)), None);
178        assert_eq!(cx.update(|cx| cx.fetch_asset::<TestAsset>(&source)), None);
179        assert_eq!(source.load_count(), 1);
180
181        assert!(sender.send(Ok(42)).is_ok());
182        cx.run_until_parked();
183
184        assert_eq!(
185            cx.update(|cx| cx.fetch_asset::<TestAsset>(&source)),
186            Some(Ok(42))
187        );
188        assert_eq!(
189            cx.update(|cx| cx.fetch_asset::<TestAsset>(&source)),
190            Some(Ok(42))
191        );
192        assert_eq!(source.load_count(), 1);
193    }
194
195    #[gpui::test]
196    fn load_completion_notifies_each_observing_entity_once(cx: &mut TestAppContext) {
197        let (sender, receiver) = oneshot::channel::<i32>();
198        let load = cx.update(|cx| {
199            CachedLoad::new(
200                async move { receiver.await.expect("test sends a result") },
201                cx,
202            )
203        });
204        let first_notification_count = Rc::new(Cell::new(0));
205        let second_notification_count = Rc::new(Cell::new(0));
206        let (first_entity, second_entity) = cx.update(|cx| {
207            let first_entity = cx.new(|_| ());
208            let second_entity = cx.new(|_| ());
209            cx.observe(&first_entity, {
210                let first_notification_count = first_notification_count.clone();
211                move |_, _| first_notification_count.set(first_notification_count.get() + 1)
212            })
213            .detach();
214            cx.observe(&second_entity, {
215                let second_notification_count = second_notification_count.clone();
216                move |_, _| second_notification_count.set(second_notification_count.get() + 1)
217            })
218            .detach();
219            (first_entity, second_entity)
220        });
221
222        assert_eq!(load.use_by(first_entity.entity_id()), None);
223        assert_eq!(load.use_by(first_entity.entity_id()), None);
224        assert_eq!(load.use_by(second_entity.entity_id()), None);
225
226        assert!(sender.send(42).is_ok());
227        cx.run_until_parked();
228
229        assert_eq!(first_notification_count.get(), 1);
230        assert_eq!(second_notification_count.get(), 1);
231        assert_eq!(load.use_by(first_entity.entity_id()), Some(42));
232        assert_eq!(load.use_by(second_entity.entity_id()), Some(42));
233    }
234
235    #[gpui::test]
236    fn removing_pending_asset_cancels_it_and_replacement_ignores_stale_completion(
237        cx: &mut TestAppContext,
238    ) {
239        let (stale_sender, stale_receiver) = oneshot::channel();
240        let (replacement_sender, replacement_receiver) = oneshot::channel();
241        let source = TestAssetSource::new(2, [stale_receiver, replacement_receiver]);
242
243        assert_eq!(cx.update(|cx| cx.fetch_asset::<TestAsset>(&source)), None);
244        assert_eq!(source.load_count(), 1);
245        cx.run_until_parked();
246
247        cx.update(|cx| cx.remove_asset::<TestAsset>(&source));
248        assert!(!cx.update(|cx| cx.has_asset::<TestAsset>(&source)));
249        cx.run_until_parked();
250        assert!(stale_sender.send(Ok(1)).is_err());
251
252        assert_eq!(cx.update(|cx| cx.fetch_asset::<TestAsset>(&source)), None);
253        assert_eq!(source.load_count(), 2);
254
255        assert!(replacement_sender.send(Err("replacement failed")).is_ok());
256        cx.run_until_parked();
257
258        assert_eq!(
259            cx.update(|cx| cx.fetch_asset::<TestAsset>(&source)),
260            Some(Err("replacement failed"))
261        );
262        assert_eq!(source.load_count(), 2);
263    }
264
265    #[gpui::test]
266    fn eviction_after_worker_completion_does_not_publish_into_replacement(cx: &mut TestAppContext) {
267        let (sender, receiver) = oneshot::channel();
268        let (replacement_sender, replacement_receiver) = oneshot::channel();
269        let source = TestAssetSource::new(3, [receiver, replacement_receiver]);
270
271        assert_eq!(cx.update(|cx| cx.fetch_asset::<TestAsset>(&source)), None);
272        assert!(sender.send(Ok(1)).is_ok());
273        while source.completion_count.load(Ordering::SeqCst) == 0 {
274            assert!(cx.background_executor.tick());
275        }
276        assert_eq!(cx.update(|cx| cx.fetch_asset::<TestAsset>(&source)), None);
277
278        cx.update(|cx| {
279            cx.remove_asset::<TestAsset>(&source);
280            assert_eq!(cx.fetch_asset::<TestAsset>(&source), None);
281        });
282        cx.run_until_parked();
283        assert_eq!(cx.update(|cx| cx.fetch_asset::<TestAsset>(&source)), None);
284
285        assert!(replacement_sender.send(Ok(2)).is_ok());
286        cx.run_until_parked();
287        assert_eq!(
288            cx.update(|cx| cx.fetch_asset::<TestAsset>(&source)),
289            Some(Ok(2))
290        );
291        assert_eq!(source.load_count(), 2);
292    }
293
294    struct TestAsset;
295
296    #[derive(Clone)]
297    struct TestAssetSource {
298        id: usize,
299        load_count: Arc<AtomicUsize>,
300        completion_count: Arc<AtomicUsize>,
301        receivers: Arc<Mutex<VecDeque<oneshot::Receiver<Result<i32, &'static str>>>>>,
302    }
303
304    impl TestAssetSource {
305        fn new(
306            id: usize,
307            receivers: impl IntoIterator<Item = oneshot::Receiver<Result<i32, &'static str>>>,
308        ) -> Self {
309            Self {
310                id,
311                load_count: Arc::new(AtomicUsize::new(0)),
312                completion_count: Arc::new(AtomicUsize::new(0)),
313                receivers: Arc::new(Mutex::new(receivers.into_iter().collect())),
314            }
315        }
316
317        fn load_count(&self) -> usize {
318            self.load_count.load(Ordering::SeqCst)
319        }
320    }
321
322    impl Hash for TestAssetSource {
323        fn hash<H: Hasher>(&self, state: &mut H) {
324            self.id.hash(state);
325        }
326    }
327
328    impl Asset for TestAsset {
329        type Source = TestAssetSource;
330        type Output = Result<i32, &'static str>;
331
332        fn load(
333            source: Self::Source,
334            _cx: &mut App,
335        ) -> impl Future<Output = Self::Output> + Send + 'static {
336            source.load_count.fetch_add(1, Ordering::SeqCst);
337            let receiver = source
338                .receivers
339                .lock()
340                .expect("test receiver mutex should not be poisoned")
341                .pop_front()
342                .expect("each test load should have a receiver");
343            async move {
344                let result = match receiver.await {
345                    Ok(result) => result,
346                    Err(_) => Err("cancelled"),
347                };
348                source.completion_count.fetch_add(1, Ordering::SeqCst);
349                result
350            }
351        }
352    }
353}
gpui_authoring/asset_cache.rs

gpui_authoring/
asset_cache.rs
