package_directory/
root_dir_cache.rs

1// Copyright 2024 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use fidl_fuchsia_io as fio;
6use fuchsia_inspect as finspect;
7use futures::FutureExt as _;
8use futures::future::BoxFuture;
9use std::collections::HashMap;
10use std::collections::hash_map::Entry::{Occupied, Vacant};
11use std::sync::{Arc, Weak};
12use vfs::ObjectRequestRef;
13use vfs::directory::entry::{EntryInfo, OpenRequest};
14use vfs::directory::traversal_position::TraversalPosition;
15use vfs::execution_scope::ExecutionScope;
16
17/// `RootDirCache` is a cache of `Arc<impl Deref<Target = RootDir>>`s indexed by their hash.
18///
19/// The cache internally stores `Weak`s and drops the corresponding entry when the last `Arc` is
20/// dropped, so it is a cache of `Arc`s that are actively in use by its clients. This is useful for
21/// deduplicating the `RootDir`s used by VFS to serve package directory connections while also
22/// keeping track of which connections are open.
23///
24/// Due to `RootDir`'s VFS implementation, a package will have alive `Arc`s if there are
25/// fuchsia.io.Directory connections to the package's root directory or any sub directory.
26///
27/// This does not keep track of fuchsia.io.File connections made to a `RootDir`s files (files under
28/// meta/, files not under meta/, and the meta file) but both Blobfs and Fxblob will wait to delete
29/// blobs that have open connections or VMOs until the last one closes, so it is safe to delete
30/// packages that are not in the `RootDirCache`.
31///
32/// Clients close connections to packages by closing their end of the Zircon channel over which the
33/// fuchsia.io.Directory messages were being sent. Some time after the client end of the channel is
34/// closed, the server (usually in a different process) will be notified by the kernel, and the VFS
35/// task serving the connection will finish, dropping its `Arc`. When the last `Arc` is dropped, the
36/// strong count of the corresponding `std::sync::Weak` in the `RootDirCache` will decrement to
37/// zero. At this point the `RootDirCache` will no longer report the package as open.
38/// All this is to say that there will be some delay between a package no longer being in use and
39/// clients of `RootDirCache` finding out about that.
40#[derive(Debug, Clone)]
41pub struct RootDirCache<S> {
42    non_meta_storage: S,
43    dirs: Arc<std::sync::Mutex<HashMap<fuchsia_hash::Hash, Weak<CachedRootDir<S>>>>>,
44}
45
46impl<S: crate::NonMetaStorage + Clone> RootDirCache<S> {
47    /// Creates a `RootDirCache` that uses `non_meta_storage` as the backing for the
48    /// internally managed `crate::RootDir`s.
49    pub fn new(non_meta_storage: S) -> Self {
50        let dirs = Arc::new(std::sync::Mutex::new(HashMap::new()));
51        Self { non_meta_storage, dirs }
52    }
53
54    /// Returns an `Arc<CachedRootDir>` corresponding to `hash`.
55    /// If there is not already one in the cache, `root_dir` will be used if provided, otherwise
56    /// a new one will be created using the `non_meta_storage` provided to `Self::new`.
57    ///
58    /// If provided, `root_dir` must be backed by the same `NonMetaStorage` that `Self::new` was
59    /// called with.
60    pub async fn get_or_insert(
61        &self,
62        hash: fuchsia_hash::Hash,
63        root_dir: Option<crate::RootDir<S>>,
64    ) -> Result<Arc<CachedRootDir<S>>, crate::Error> {
65        if let Some(root_dir) = self.get(&hash) {
66            return Ok(root_dir);
67        }
68
69        let root_dir = match root_dir {
70            Some(root_dir) => root_dir,
71            None => crate::RootDir::new_raw(self.non_meta_storage.clone(), hash).await?,
72        };
73        Ok(match self.dirs.lock().expect("poisoned mutex").entry(hash) {
74            // Raced with another call to get_or_insert.
75            Occupied(mut o) => {
76                let old_root_dir = o.get_mut();
77                if let Some(old_root_dir) = old_root_dir.upgrade() {
78                    old_root_dir
79                } else {
80                    let new_root_dir = CachedRootDir::new(root_dir, &self.dirs);
81                    *old_root_dir = Arc::downgrade(&new_root_dir);
82                    new_root_dir
83                }
84            }
85            Vacant(v) => {
86                let new_root_dir = CachedRootDir::new(root_dir, &self.dirs);
87                v.insert(Arc::downgrade(&new_root_dir));
88                new_root_dir
89            }
90        })
91    }
92
93    /// Returns the `Arc<CachedRootDir>` with the given `hash`, if one exists in the cache.
94    /// Otherwise returns `None`.
95    /// Holding on to the returned `Arc` will keep the package open (as reported by
96    /// `Self::list`).
97    pub fn get(&self, hash: &fuchsia_hash::Hash) -> Option<Arc<CachedRootDir<S>>> {
98        self.dirs.lock().expect("poisoned mutex").get(hash)?.upgrade()
99    }
100
101    /// Packages with live `Arc<CachedRootDir>`s.
102    /// Holding on to the returned `Arc`s will keep the packages open.
103    pub fn list(&self) -> Vec<Arc<CachedRootDir<S>>> {
104        self.dirs.lock().expect("poisoned mutex").values().filter_map(|v| v.upgrade()).collect()
105    }
106
107    /// Returns a callback to be given to `fuchsia_inspect::Node::record_lazy_child`.
108    /// Records the package hashes and their corresponding `Arc<CachedRootDir>` strong counts.
109    pub fn record_lazy_inspect(
110        &self,
111    ) -> impl Fn() -> BoxFuture<'static, Result<finspect::Inspector, anyhow::Error>>
112    + Send
113    + Sync
114    + 'static {
115        let dirs = Arc::downgrade(&self.dirs);
116        move || {
117            let dirs = dirs.clone();
118            async move {
119                let inspector = finspect::Inspector::default();
120                if let Some(dirs) = dirs.upgrade() {
121                    let package_counts: HashMap<_, _> = {
122                        let dirs = dirs.lock().expect("poisoned mutex");
123                        dirs.iter().map(|(k, v)| (*k, v.strong_count() as u64)).collect()
124                    };
125                    let root = inspector.root();
126                    let () = package_counts.into_iter().for_each(|(pkg, count)| {
127                        root.record_child(pkg.to_string(), |n| n.record_uint("instances", count))
128                    });
129                }
130                Ok(inspector)
131            }
132            .boxed()
133        }
134    }
135}
136
137/// A wrapper around `RootDir` that notifies `RootDirCache` when dropped.
138#[derive(Debug)]
139pub struct CachedRootDir<S> {
140    root_dir: crate::RootDir<S>,
141    dirs: Weak<std::sync::Mutex<HashMap<fuchsia_hash::Hash, Weak<Self>>>>,
142}
143
144impl<S> CachedRootDir<S> {
145    #[allow(clippy::type_complexity)]
146    fn new(
147        root_dir: crate::RootDir<S>,
148        dirs: &Arc<std::sync::Mutex<HashMap<fuchsia_hash::Hash, Weak<Self>>>>,
149    ) -> Arc<Self> {
150        Arc::new(Self { root_dir, dirs: Arc::downgrade(dirs) })
151    }
152}
153
154impl<S> std::ops::Deref for CachedRootDir<S> {
155    type Target = crate::RootDir<S>;
156
157    fn deref(&self) -> &Self::Target {
158        &self.root_dir
159    }
160}
161
162impl<S: crate::NonMetaStorage> crate::root_dir::AsRootDir for CachedRootDir<S> {
163    type Storage = S;
164
165    fn as_root_dir(&self) -> &crate::RootDir<S> {
166        &self.root_dir
167    }
168}
169
170impl<S: crate::NonMetaStorage> vfs::directory::entry::DirectoryEntry for CachedRootDir<S> {
171    fn open_entry(self: Arc<Self>, request: OpenRequest<'_>) -> Result<(), zx::Status> {
172        request.open_dir(self)
173    }
174}
175
176impl<S: crate::NonMetaStorage> vfs::directory::entry::GetEntryInfo for CachedRootDir<S> {
177    fn entry_info(&self) -> EntryInfo {
178        self.root_dir.entry_info()
179    }
180}
181
182impl<S: crate::NonMetaStorage> vfs::node::Node for CachedRootDir<S> {
183    async fn get_attributes(
184        &self,
185        requested_attributes: fio::NodeAttributesQuery,
186    ) -> Result<fio::NodeAttributes2, zx::Status> {
187        self.root_dir.get_attributes(requested_attributes).await
188    }
189}
190
191impl<S: crate::NonMetaStorage> vfs::directory::entry_container::Directory for CachedRootDir<S> {
192    fn open(
193        self: Arc<Self>,
194        scope: ExecutionScope,
195        path: vfs::Path,
196        flags: fio::Flags,
197        object_request: ObjectRequestRef<'_>,
198    ) -> Result<(), zx::Status> {
199        crate::root_dir::open_impl(self, scope, path, flags, object_request)
200    }
201
202    async fn read_dirents(
203        &self,
204        pos: &TraversalPosition,
205        sink: Box<dyn vfs::directory::dirents_sink::Sink + 'static>,
206    ) -> Result<
207        (TraversalPosition, Box<dyn vfs::directory::dirents_sink::Sealed + 'static>),
208        zx::Status,
209    > {
210        self.root_dir.read_dirents(pos, sink).await
211    }
212
213    fn register_watcher(
214        self: Arc<Self>,
215        _: ExecutionScope,
216        _: fio::WatchMask,
217        _: vfs::directory::entry_container::DirectoryWatcher,
218    ) -> Result<(), zx::Status> {
219        Err(zx::Status::NOT_SUPPORTED)
220    }
221
222    // `register_watcher` is unsupported so no need to do anything here.
223    fn unregister_watcher(self: Arc<Self>, _: usize) {}
224}
225
226impl<S> Drop for CachedRootDir<S> {
227    fn drop(&mut self) {
228        let Some(dirs) = self.dirs.upgrade() else {
229            return;
230        };
231        match dirs.lock().expect("poisoned mutex").entry(self.root_dir.hash) {
232            Occupied(o) => {
233                // In case this raced with a call to get_or_insert that added a new one.
234                if o.get().strong_count() == 0 {
235                    o.remove_entry();
236                }
237            }
238            // Raced with another call to get_or_insert that already removed the entry.
239            Vacant(_) => (),
240        };
241    }
242}
243
244#[cfg(test)]
245mod tests {
246    use super::*;
247    use assert_matches::assert_matches;
248    use diagnostics_assertions::assert_data_tree;
249    use fuchsia_async as fasync;
250    use fuchsia_pkg_testing::PackageBuilder;
251    use fuchsia_pkg_testing::blobfs::Fake as FakeBlobfs;
252
253    #[fuchsia::test]
254    async fn get_or_insert_new_entry() {
255        let pkg = PackageBuilder::new("pkg-name").build().await.unwrap();
256        let (metafar_blob, _) = pkg.contents();
257        let (blobfs_fake, blobfs_client) = FakeBlobfs::new();
258        blobfs_fake.add_blob(metafar_blob.merkle, metafar_blob.contents);
259        let server = RootDirCache::new(blobfs_client);
260
261        let dir = server.get_or_insert(metafar_blob.merkle, None).await.unwrap();
262
263        assert_eq!(server.list().len(), 1);
264
265        drop(dir);
266        assert_eq!(server.list().len(), 0);
267        assert!(server.dirs.lock().expect("poisoned mutex").is_empty());
268    }
269
270    #[fuchsia::test]
271    async fn closing_package_connection_closes_package() {
272        let pkg = PackageBuilder::new("pkg-name").build().await.unwrap();
273        let (metafar_blob, _) = pkg.contents();
274        let (blobfs_fake, blobfs_client) = FakeBlobfs::new();
275        blobfs_fake.add_blob(metafar_blob.merkle, metafar_blob.contents);
276        let server = RootDirCache::new(blobfs_client);
277
278        let dir = server.get_or_insert(metafar_blob.merkle, None).await.unwrap();
279        let (proxy, server_end) = fidl::endpoints::create_proxy::<fio::DirectoryMarker>();
280        let scope = vfs::execution_scope::ExecutionScope::new();
281        vfs::directory::serve_on(dir, fio::PERM_READABLE, scope.clone(), server_end);
282        let _ = proxy
283            .get_attributes(Default::default())
284            .await
285            .expect("directory succesfully handling requests");
286        assert_eq!(server.list().len(), 1);
287
288        drop(proxy);
289        let () = scope.wait().await;
290        assert_eq!(server.list().len(), 0);
291        assert!(server.dirs.lock().expect("poisoned mutex").is_empty());
292    }
293
294    #[fuchsia::test]
295    async fn open_subdirectory_keeps_package_open() {
296        let pkg = PackageBuilder::new("pkg-name")
297            .add_resource_at("dir/file", "bloblob".as_bytes())
298            .add_resource_at("meta/dir/file", "meta-contents".as_bytes())
299            .build()
300            .await
301            .unwrap();
302        let (metafar_blob, content_blobs) = pkg.contents();
303        let (blobfs_fake, blobfs_client) = FakeBlobfs::new();
304        blobfs_fake.add_blob(metafar_blob.merkle, metafar_blob.contents);
305        for (hash, bytes) in content_blobs {
306            blobfs_fake.add_blob(hash, bytes);
307        }
308        let server = RootDirCache::new(blobfs_client);
309
310        for subdir_path in ["meta", "meta/dir", "dir"] {
311            let dir = server.get_or_insert(metafar_blob.merkle, None).await.unwrap();
312            let (proxy, server_end) = fidl::endpoints::create_proxy::<fio::DirectoryMarker>();
313            let scope = vfs::execution_scope::ExecutionScope::new();
314            vfs::directory::serve_on(dir, fio::PERM_READABLE, scope.clone(), server_end);
315            let subdir =
316                fuchsia_fs::directory::open_directory(&proxy, subdir_path, fio::PERM_READABLE)
317                    .await
318                    .unwrap();
319            assert_eq!(Arc::strong_count(&server.list()[0]), 3);
320
321            drop(proxy);
322            while Arc::strong_count(&server.list()[0]) != 2 {
323                let () = fasync::Timer::new(std::time::Duration::from_millis(10)).await;
324            }
325
326            drop(subdir);
327            let () = scope.wait().await;
328            assert_eq!(server.list().len(), 0);
329            assert!(server.dirs.lock().expect("poisoned mutex").is_empty());
330        }
331    }
332
333    #[fuchsia::test]
334    async fn get_or_insert_existing_entry() {
335        let pkg = PackageBuilder::new("pkg-name").build().await.unwrap();
336        let (metafar_blob, _) = pkg.contents();
337        let (blobfs_fake, blobfs_client) = FakeBlobfs::new();
338        blobfs_fake.add_blob(metafar_blob.merkle, metafar_blob.contents);
339        let server = RootDirCache::new(blobfs_client);
340
341        let dir0 = server.get_or_insert(metafar_blob.merkle, None).await.unwrap();
342
343        let dir1 = server.get_or_insert(metafar_blob.merkle, None).await.unwrap();
344        assert_eq!(server.list().len(), 1);
345        assert_eq!(Arc::strong_count(&server.list()[0]), 3);
346
347        drop(dir0);
348        drop(dir1);
349        assert_eq!(server.list().len(), 0);
350        assert!(server.dirs.lock().expect("poisoned mutex").is_empty());
351    }
352
353    #[fuchsia::test]
354    async fn get_or_insert_provided_root_dir() {
355        let pkg = PackageBuilder::new("pkg-name").build().await.unwrap();
356        let (metafar_blob, _) = pkg.contents();
357        let (blobfs_fake, blobfs_client) = FakeBlobfs::new();
358        blobfs_fake.add_blob(metafar_blob.merkle, metafar_blob.contents);
359        let root_dir =
360            crate::RootDir::new_raw(blobfs_client.clone(), metafar_blob.merkle).await.unwrap();
361        blobfs_fake.delete_blob(metafar_blob.merkle);
362        let server = RootDirCache::new(blobfs_client);
363
364        let dir = server.get_or_insert(metafar_blob.merkle, Some(root_dir)).await.unwrap();
365        assert_eq!(server.list().len(), 1);
366
367        drop(dir);
368        assert_eq!(server.list().len(), 0);
369        assert!(server.dirs.lock().expect("poisoned mutex").is_empty());
370    }
371
372    #[fuchsia::test]
373    async fn get_or_insert_fails_if_root_dir_creation_fails() {
374        let (_blobfs_fake, blobfs_client) = FakeBlobfs::new();
375        let server = RootDirCache::new(blobfs_client);
376
377        assert_matches!(
378            server.get_or_insert([0; 32].into(), None).await,
379            Err(crate::Error::MissingMetaFar)
380        );
381        assert!(server.dirs.lock().expect("poisoned mutex").is_empty());
382    }
383
384    #[fuchsia::test]
385    async fn get_or_insert_concurrent_race_to_insert_new_root_dir() {
386        let pkg = PackageBuilder::new("pkg-name").build().await.unwrap();
387        let (metafar_blob, _) = pkg.contents();
388        let (blobfs_fake, blobfs_client) = FakeBlobfs::new();
389        blobfs_fake.add_blob(metafar_blob.merkle, metafar_blob.contents);
390        let server = RootDirCache::new(blobfs_client);
391
392        let fut0 = server.get_or_insert(metafar_blob.merkle, None);
393
394        let fut1 = server.get_or_insert(metafar_blob.merkle, None);
395
396        let (res0, res1) = futures::future::join(fut0, fut1).await;
397        let (dir0, dir1) = (res0.unwrap(), res1.unwrap());
398
399        assert_eq!(server.list().len(), 1);
400        assert_eq!(Arc::strong_count(&server.list()[0]), 3);
401
402        drop(dir0);
403        drop(dir1);
404        assert_eq!(server.list().len(), 0);
405        assert!(server.dirs.lock().expect("poisoned mutex").is_empty());
406    }
407
408    #[fuchsia::test]
409    async fn inspect() {
410        let pkg = PackageBuilder::new("pkg-name").build().await.unwrap();
411        let (metafar_blob, _) = pkg.contents();
412        let (blobfs_fake, blobfs_client) = FakeBlobfs::new();
413        blobfs_fake.add_blob(metafar_blob.merkle, metafar_blob.contents);
414        let server = RootDirCache::new(blobfs_client);
415        let _dir = server.get_or_insert(metafar_blob.merkle, None).await.unwrap();
416
417        let inspector = finspect::Inspector::default();
418        inspector.root().record_lazy_child("open-packages", server.record_lazy_inspect());
419
420        assert_data_tree!(inspector, root: {
421            "open-packages": {
422                pkg.hash().to_string() => {
423                    "instances": 1u64,
424                },
425            }
426        });
427    }
428}
package_directory/root_dir_cache.rs

package_directory/
root_dir_cache.rs
