fxfs_platform_testing/fuchsia/
node.rs

1// Copyright 2021 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 crate::fuchsia::directory::FxDirectory;
6use crate::fuchsia::file::FxFile;
7use fuchsia_sync::Mutex;
8use futures::future::poll_fn;
9use fxfs::object_store::ObjectDescriptor;
10use fxfs_macros::ToWeakNode;
11use std::any::TypeId;
12use std::collections::BTreeMap;
13use std::collections::btree_map::Entry;
14use std::fmt;
15use std::mem::ManuallyDrop;
16use std::sync::{Arc, Weak};
17use std::task::{Poll, Waker};
18use vfs::common::IntoAny;
19
20/// FxNode is a node in the filesystem hierarchy (either a file or directory).
21pub trait FxNode: IntoAny + ToWeakNode + Send + Sync + 'static {
22    fn object_id(&self) -> u64;
23    fn parent(&self) -> Option<Arc<FxDirectory>>;
24    fn set_parent(&self, parent: Arc<FxDirectory>);
25    fn open_count_add_one(&self);
26
27    /// Atomically check if this brought the count to zero while the node is
28    /// marked for purge. If so, this *must* queue the node for tombstone the node.
29    fn open_count_sub_one(self: Arc<Self>);
30    fn object_descriptor(&self) -> ObjectDescriptor;
31
32    /// Marks the object to be purged. Queues the node for tombstone if open count is zero.
33    fn mark_to_be_purged(self: Arc<Self>) {
34        panic!("Unexpected call to mark_to_be_purged");
35    }
36
37    /// Called when the filesystem is shutting down. Implementations should break any strong
38    /// reference cycles that would prevent the node from being dropped.
39    fn terminate(&self) {}
40}
41
42struct PlaceholderInner {
43    object_id: u64,
44    waker_sequence: u64,
45    wakers: Vec<Waker>,
46}
47
48#[derive(ToWeakNode)]
49struct Placeholder(Mutex<PlaceholderInner>);
50
51impl FxNode for Placeholder {
52    fn object_id(&self) -> u64 {
53        self.0.lock().object_id
54    }
55    fn parent(&self) -> Option<Arc<FxDirectory>> {
56        unreachable!();
57    }
58    fn set_parent(&self, _parent: Arc<FxDirectory>) {
59        unreachable!();
60    }
61    fn open_count_add_one(&self) {}
62    fn open_count_sub_one(self: Arc<Self>) {}
63
64    fn object_descriptor(&self) -> ObjectDescriptor {
65        ObjectDescriptor::File
66    }
67}
68
69impl fmt::Debug for dyn FxNode {
70    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
71        f.debug_struct("FxNode")
72            .field("id", &self.object_id())
73            .field("descriptor", &self.object_descriptor())
74            .finish()
75    }
76}
77
78/// PlaceholderOwner is a reserved slot in the node cache.
79pub struct PlaceholderOwner<'a> {
80    inner: Arc<Placeholder>,
81    committed: bool,
82    cache: &'a NodeCache,
83}
84
85impl PlaceholderOwner<'_> {
86    /// Commits a node to the cache, replacing the placeholder and unblocking any waiting callers.
87    pub fn commit(mut self, node: &Arc<dyn FxNode>) {
88        let this_object_id = self.inner.object_id();
89        assert_eq!(node.object_id(), this_object_id);
90        self.committed = true;
91        self.cache.commit(node);
92    }
93}
94
95impl Drop for PlaceholderOwner<'_> {
96    fn drop(&mut self) {
97        let wakers = {
98            let mut cache_guard = (!self.committed).then(|| self.cache.0.lock());
99            let mut p = self.inner.0.lock();
100            if let Some(cache) = &mut cache_guard {
101                // If the placeholder is dropped before it was committed, remove the cache entry
102                // so that another caller blocked in NodeCache::get_or_reserve can take the slot.
103                cache.map.remove(&p.object_id);
104            }
105            std::mem::take(&mut p.wakers)
106        };
107        for waker in wakers {
108            waker.wake();
109        }
110    }
111}
112
113/// See NodeCache::get_or_reserve.
114pub enum GetResult<'a> {
115    Placeholder(PlaceholderOwner<'a>),
116    Node(Arc<dyn FxNode>),
117}
118
119impl<'a> GetResult<'a> {
120    pub fn placeholder(self) -> Option<PlaceholderOwner<'a>> {
121        match self {
122            GetResult::Placeholder(placeholder) => Some(placeholder),
123            _ => None,
124        }
125    }
126}
127
128/// WeakNode permits `upgrade_and_downcast_node` below which only tries to upgrade if we know the
129/// downcast will work.  This makes for a simpler and more performant implementation of `FileIter`
130/// which would otherwise have to jump through some hoops to drop the reference count whilst not
131/// holding locks if the downcast fails.
132pub struct WeakNode {
133    vtable: &'static WeakNodeVTable,
134    node: *const (),
135}
136
137impl WeakNode {
138    // This is unsafe because the caller must make sure `node` comes from `Weak<T>::into_raw` and
139    // provide correct implementations for the functions in `vtable`.
140    pub(crate) unsafe fn new(vtable: &'static WeakNodeVTable, node: *const ()) -> Self {
141        Self { vtable, node }
142    }
143
144    unsafe fn is_type<T: 'static>(&self) -> bool {
145        unsafe { (self.vtable.type_id)() == TypeId::of::<T>() }
146    }
147}
148
149impl Drop for WeakNode {
150    fn drop(&mut self) {
151        // SAFETY: See the implementation of `ToWeakNode` in `macros.rs`.  `node` should be a
152        // pointer that came from `Weak<T>::into_raw`.  The `drop` function needs to use
153        // `Weak::from_raw`.
154        unsafe {
155            (self.vtable.drop)(self.node);
156        }
157    }
158}
159
160// SAFETY: `node` comes from `Weak<T>::into_raw` which is safe to send across threads.
161unsafe impl Send for WeakNode {}
162
163pub(crate) struct WeakNodeVTable {
164    // This should convert the pointer back to `Weak<T>` to drop it.
165    drop: unsafe fn(*const ()),
166
167    // Returns the `TypeId::of::<T>` for `Weak<T>`.
168    type_id: unsafe fn() -> TypeId,
169
170    // Tries to upgrade the `Weak<T>` and returns `Arc<dyn FxNode>`.
171    upgrade: unsafe fn(*const ()) -> Option<Arc<dyn FxNode>>,
172}
173
174impl WeakNodeVTable {
175    pub(crate) const fn new(
176        drop: unsafe fn(*const ()),
177        type_id: unsafe fn() -> TypeId,
178        upgrade: unsafe fn(*const ()) -> Option<Arc<dyn FxNode>>,
179    ) -> Self {
180        Self { drop, type_id, upgrade }
181    }
182}
183
184/// Used to convert nodes into `WeakNode` which is stored in the cache.  This should be implemented
185/// using the `ToWeakNode` derive macro.
186pub trait ToWeakNode {
187    fn to_weak_node(self: Arc<Self>) -> WeakNode;
188}
189
190/// Upgrades and downcasts as a single step.  This won't do the upgrade if the downcast will fail,
191/// which avoids issues with dropping the reference whilst locks are held.
192fn upgrade_and_downcast_node<T: 'static>(weak_node: &WeakNode) -> Option<Arc<T>> {
193    // SAFETY: We check `T` matches before converting the pointer (which should be the result of
194    // `Weak<T>::into_raw`), back to `Weak<T>`.
195    unsafe {
196        if weak_node.is_type::<T>() {
197            ManuallyDrop::new(Weak::from_raw(weak_node.node as *const T)).upgrade()
198        } else {
199            None
200        }
201    }
202}
203
204/// Upgrades to `Arc<dyn FxNode>`.
205fn upgrade_node(weak_node: &WeakNode) -> Option<Arc<dyn FxNode>> {
206    // SAFETY: Safe if `WeakNode::new` is called correctly and the `upgrade` function is implemented
207    // correctly.  See the implementation in `macros.rs`.
208    unsafe { (weak_node.vtable.upgrade)(weak_node.node) }
209}
210
211struct NodeCacheInner {
212    map: BTreeMap<u64, WeakNode>,
213    next_waker_sequence: u64,
214}
215
216/// NodeCache is an in-memory cache of weak node references.
217pub struct NodeCache(Mutex<NodeCacheInner>);
218
219/// Iterates over all files in the cache (skipping directories).
220pub struct FileIter<'a> {
221    cache: &'a NodeCache,
222    object_id: Option<u64>,
223}
224
225impl<'a> Iterator for FileIter<'a> {
226    type Item = Arc<FxFile>;
227    fn next(&mut self) -> Option<Self::Item> {
228        let cache = self.cache.0.lock();
229        let range = match self.object_id {
230            None => cache.map.range(0..),
231            Some(oid) => cache.map.range(oid + 1..),
232        };
233        for (object_id, node) in range {
234            if let Some(file) = upgrade_and_downcast_node(node) {
235                self.object_id = Some(*object_id);
236                return Some(file);
237            }
238        }
239        None
240    }
241}
242
243impl NodeCache {
244    pub fn new() -> Self {
245        Self(Mutex::new(NodeCacheInner { map: BTreeMap::new(), next_waker_sequence: 0 }))
246    }
247
248    /// Gets a node in the cache, or reserves a placeholder in the cache to fill.
249    ///
250    /// Only the first caller will receive a placeholder result; all callers after that will block
251    /// until the placeholder is filled (or the placeholder is dropped, at which point the next
252    /// caller would get a placeholder). Callers that receive a placeholder should later commit a
253    /// node with NodeCache::commit.
254    pub async fn get_or_reserve<'a>(&'a self, object_id: u64) -> GetResult<'a> {
255        let mut waker_sequence = 0;
256        let mut waker_index = 0;
257        poll_fn(|cx| {
258            let mut this = self.0.lock();
259            if let Some(node) = this.map.get(&object_id) {
260                if let Some(node) = upgrade_node(node) {
261                    if let Ok(placeholder) = node.clone().into_any().downcast::<Placeholder>() {
262                        let mut inner = placeholder.0.lock();
263                        if inner.waker_sequence == waker_sequence {
264                            inner.wakers[waker_index] = cx.waker().clone();
265                        } else {
266                            waker_index = inner.wakers.len();
267                            waker_sequence = inner.waker_sequence;
268                            inner.wakers.push(cx.waker().clone());
269                        }
270                        return Poll::Pending;
271                    } else {
272                        return Poll::Ready(GetResult::Node(node));
273                    }
274                }
275            }
276            this.next_waker_sequence += 1;
277            let inner = Arc::new(Placeholder(Mutex::new(PlaceholderInner {
278                object_id,
279                waker_sequence: this.next_waker_sequence,
280                wakers: vec![],
281            })));
282            this.map.insert(object_id, inner.clone().to_weak_node());
283            Poll::Ready(GetResult::Placeholder(PlaceholderOwner {
284                inner,
285                committed: false,
286                cache: self,
287            }))
288        })
289        .await
290    }
291
292    /// Removes a node from the cache. Calling this on a placeholder is an error; instead, the
293    /// placeholder should simply be dropped.
294    pub fn remove(&self, node: &dyn FxNode) {
295        let mut this = self.0.lock();
296        if let Entry::Occupied(o) = this.map.entry(node.object_id()) {
297            // If this method is called when a node is being dropped, then upgrade will fail and
298            // it's possible the cache has been populated with another node, so to avoid that race,
299            // we must check that the node in the cache is the node we want to remove.
300            //
301            // Note this ugly cast in place of `std::ptr::eq(o.get().as_ptr(), node)` here is
302            // to ensure we don't compare vtable pointers, which are not strictly guaranteed to be
303            // the same across casts done in different code generation units at compilation time.
304            if o.get().node == node as *const dyn FxNode as *const () {
305                o.remove();
306            }
307        }
308    }
309
310    /// Returns the given node if present in the cache. This call should be handled with care. If it
311    /// is being used to check if there is no live node version, then there should be precautions
312    /// taken to ensure one will not be created in a race with this call or any of the resulting
313    /// actions.
314    pub fn get(&self, object_id: u64) -> Option<Arc<dyn FxNode>> {
315        let l = self.0.lock();
316        let weak_node = l.map.get(&object_id)?;
317        // Don't return placeholders as valid nodes. Panicking here, as this is a result of a race.
318        // If it happens we're likely breaking an assumption, but should not be permanent for the
319        // filesystem. Better to remount.
320        assert!(
321            unsafe { !weak_node.is_type::<Placeholder>() },
322            "Returning a placeholder indicates a race with open"
323        );
324        upgrade_node(weak_node)
325    }
326
327    /// Returns true if an entry exists in the cache for the given `object_id`.
328    ///
329    /// Note that this returns true if the node is currently open, if it is being loaded (i.e. has
330    /// a `Placeholder`), or if there is a weak entry with zero strong references because another
331    /// thread is in the middle of dropping the node and removing it from the cache.
332    pub fn contains_key(&self, object_id: u64) -> bool {
333        self.0.lock().map.contains_key(&object_id)
334    }
335
336    /// Returns an iterator over all files in the cache.
337    pub fn files(&self) -> FileIter<'_> {
338        FileIter { cache: self, object_id: None }
339    }
340
341    pub fn terminate(&self) {
342        let _drop_list = {
343            let this = self.0.lock();
344            let mut drop_list = Vec::with_capacity(this.map.len());
345            for (_, node) in &this.map {
346                if let Some(node) = upgrade_node(&node) {
347                    node.terminate();
348                    // We must drop later when we're not holding the lock.
349                    drop_list.push(node);
350                }
351            }
352            drop_list
353        };
354    }
355
356    fn commit(&self, node: &Arc<dyn FxNode>) {
357        let mut this = self.0.lock();
358        this.map.insert(node.object_id(), node.clone().to_weak_node());
359    }
360}
361
362// Wraps a node with an open count.
363pub struct OpenedNode<N: FxNode + ?Sized>(pub Arc<N>);
364
365impl<N: FxNode + ?Sized> OpenedNode<N> {
366    pub fn new(node: Arc<N>) -> Self {
367        node.open_count_add_one();
368        OpenedNode(node)
369    }
370
371    /// Downcasts to something that implements FxNode.
372    pub fn downcast<T: FxNode>(self) -> Result<OpenedNode<T>, Self> {
373        if self.is::<T>() {
374            Ok(OpenedNode(
375                self.take().into_any().downcast::<T>().unwrap_or_else(|_| unreachable!()),
376            ))
377        } else {
378            Err(self)
379        }
380    }
381
382    /// Takes the wrapped node.  The caller takes responsibility for dropping the open count.
383    pub fn take(self) -> Arc<N> {
384        let this = std::mem::ManuallyDrop::new(self);
385        unsafe { std::ptr::read(&this.0) }
386    }
387
388    /// Returns true if this is an instance of T.
389    pub fn is<T: 'static>(&self) -> bool {
390        self.0.as_ref().type_id() == TypeId::of::<T>()
391    }
392
393    /// Makes another OpenedNode, increasing the open count on the node.
394    pub fn dup(&self) -> Self {
395        self.0.open_count_add_one();
396        OpenedNode(self.0.clone())
397    }
398}
399
400impl<N: FxNode + ?Sized> Drop for OpenedNode<N> {
401    fn drop(&mut self) {
402        self.clone().open_count_sub_one();
403    }
404}
405
406impl<N: FxNode + ?Sized> std::ops::Deref for OpenedNode<N> {
407    type Target = Arc<N>;
408
409    fn deref(&self) -> &Self::Target {
410        &self.0
411    }
412}
413
414#[cfg(test)]
415mod tests {
416    use crate::fuchsia::directory::FxDirectory;
417    use crate::fuchsia::node::{FxNode, GetResult, NodeCache};
418    use fuchsia_async as fasync;
419    use fuchsia_sync::Mutex;
420    use futures::future::join_all;
421    use fxfs::object_store::ObjectDescriptor;
422    use fxfs_macros::ToWeakNode;
423    use std::sync::Arc;
424    use std::sync::atomic::{AtomicU64, Ordering};
425    use std::time::Duration;
426
427    #[derive(ToWeakNode)]
428    struct FakeNode(u64, Arc<NodeCache>);
429    impl FxNode for FakeNode {
430        fn object_id(&self) -> u64 {
431            self.0
432        }
433        fn parent(&self) -> Option<Arc<FxDirectory>> {
434            unreachable!();
435        }
436        fn set_parent(&self, _parent: Arc<FxDirectory>) {
437            unreachable!();
438        }
439        fn open_count_add_one(&self) {}
440        fn open_count_sub_one(self: Arc<Self>) {}
441
442        fn object_descriptor(&self) -> ObjectDescriptor {
443            ObjectDescriptor::Directory
444        }
445    }
446    impl Drop for FakeNode {
447        fn drop(&mut self) {
448            self.1.remove(self);
449        }
450    }
451
452    #[fuchsia::test]
453    async fn test_drop_placeholder() {
454        let cache = Arc::new(NodeCache::new());
455        let object_id = 0u64;
456        match cache.get_or_reserve(object_id).await {
457            GetResult::Node(_) => panic!("Unexpected node"),
458            GetResult::Placeholder(_) => {}
459        };
460        match cache.get_or_reserve(object_id).await {
461            GetResult::Node(_) => panic!("Unexpected node"),
462            GetResult::Placeholder(_) => {}
463        };
464    }
465
466    #[fuchsia::test]
467    async fn test_simple() {
468        let cache = Arc::new(NodeCache::new());
469        let object_id = {
470            let node = Arc::new(FakeNode(0, cache.clone()));
471            match cache.get_or_reserve(node.object_id()).await {
472                GetResult::Node(_) => panic!("Unexpected node"),
473                GetResult::Placeholder(p) => {
474                    p.commit(&(node.clone() as Arc<dyn FxNode>));
475                }
476            };
477            match cache.get_or_reserve(node.object_id()).await {
478                GetResult::Node(n) => assert_eq!(n.object_id(), node.object_id()),
479                GetResult::Placeholder(_) => panic!("No node found"),
480            };
481            node.object_id()
482        };
483        match cache.get_or_reserve(object_id).await {
484            GetResult::Node(_) => panic!("Unexpected node"),
485            GetResult::Placeholder(_) => {}
486        };
487    }
488
489    #[fuchsia::test(threads = 10)]
490    async fn test_subsequent_callers_block() {
491        let cache = Arc::new(NodeCache::new());
492        let object_id = 0u64;
493        let writes_to_cache = Arc::new(AtomicU64::new(0));
494        let reads_from_cache = Arc::new(AtomicU64::new(0));
495        let node = Arc::new(FakeNode(object_id, cache.clone()));
496        join_all((0..10).map(|_| {
497            let node = node.clone();
498            let cache = cache.clone();
499            let object_id = object_id.clone();
500            let writes_to_cache = writes_to_cache.clone();
501            let reads_from_cache = reads_from_cache.clone();
502            async move {
503                match cache.get_or_reserve(object_id).await {
504                    GetResult::Node(node) => {
505                        reads_from_cache.fetch_add(1, Ordering::SeqCst);
506                        assert_eq!(node.object_id(), object_id);
507                    }
508                    GetResult::Placeholder(p) => {
509                        writes_to_cache.fetch_add(1, Ordering::SeqCst);
510                        // Add a delay to simulate doing some work (e.g. loading from disk).
511                        fasync::Timer::new(Duration::from_millis(100)).await;
512                        p.commit(&(node as Arc<dyn FxNode>));
513                    }
514                }
515            }
516        }))
517        .await;
518        assert_eq!(writes_to_cache.load(Ordering::SeqCst), 1);
519        assert_eq!(reads_from_cache.load(Ordering::SeqCst), 9);
520    }
521
522    #[fuchsia::test(threads = 10)]
523    async fn test_multiple_nodes() {
524        const NUM_OBJECTS: usize = 5;
525        const TASKS_PER_OBJECT: usize = 4;
526
527        let cache = Arc::new(NodeCache::new());
528        let writes = Arc::new(Mutex::new(vec![0u64; NUM_OBJECTS]));
529        let reads = Arc::new(Mutex::new(vec![0u64; NUM_OBJECTS]));
530        let nodes: Vec<_> = (0..NUM_OBJECTS as u64)
531            .map(|object_id| Arc::new(FakeNode(object_id, cache.clone())))
532            .collect();
533
534        join_all((0..TASKS_PER_OBJECT).flat_map(|_| {
535            nodes.iter().cloned().map(|node| {
536                let cache = cache.clone();
537                let writes = writes.clone();
538                let reads = reads.clone();
539                async move {
540                    match cache.get_or_reserve(node.object_id()).await {
541                        GetResult::Node(result) => {
542                            assert_eq!(node.object_id(), result.object_id());
543                            reads.lock()[node.object_id() as usize] += 1;
544                        }
545                        GetResult::Placeholder(p) => {
546                            writes.lock()[node.object_id() as usize] += 1;
547                            // Add a delay to simulate doing some work (e.g. loading from disk).
548                            fasync::Timer::new(Duration::from_millis(100)).await;
549                            p.commit(&(node as Arc<dyn FxNode>));
550                        }
551                    }
552                }
553            })
554        }))
555        .await;
556        assert_eq!(*writes.lock(), vec![1u64; NUM_OBJECTS]);
557        assert_eq!(*reads.lock(), vec![TASKS_PER_OBJECT as u64 - 1; NUM_OBJECTS]);
558    }
559}
fxfs_platform_testing/fuchsia/node.rs

fxfs_platform_testing/fuchsia/
node.rs
