Skip to main content

fxfs/object_store/
transaction.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::checksum::Checksum;
6use crate::filesystem::FxFilesystem;
7use crate::log::*;
8use crate::lsm_tree::types::Item;
9use crate::object_handle::INVALID_OBJECT_ID;
10use crate::object_store::allocator::{AllocatorItem, Reservation};
11use crate::object_store::object_manager::{ObjectManager, reserved_space_from_journal_usage};
12use crate::object_store::object_record::{
13    FxfsKey, FxfsKeyV40, FxfsKeyV49, ObjectItem, ObjectItemV40, ObjectItemV41, ObjectItemV43,
14    ObjectItemV46, ObjectItemV47, ObjectItemV49, ObjectItemV50, ObjectItemV55, ObjectItemV56,
15    ObjectKey, ObjectKeyData, ObjectValue, ProjectProperty,
16};
17use crate::object_store::{AttributeId, AttributeKey, ProjectId};
18use crate::serialized_types::{Migrate, Versioned, migrate_nodefault, migrate_to_version};
19use anyhow::Error;
20use either::{Either, Left, Right};
21use fprint::TypeFingerprint;
22use fuchsia_sync::Mutex;
23use futures::future::poll_fn;
24use futures::pin_mut;
25use rustc_hash::FxHashMap as HashMap;
26use scopeguard::ScopeGuard;
27use serde::{Deserialize, Serialize};
28use std::cell::UnsafeCell;
29use std::cmp::Ordering;
30use std::collections::BTreeSet;
31use std::collections::hash_map::Entry;
32use std::marker::PhantomPinned;
33use std::ops::{Deref, DerefMut, Range};
34use std::sync::Arc;
35use std::task::{Poll, Waker};
36use std::{fmt, mem};
37
38/// This allows for special handling of certain transactions such as deletes and the
39/// extension of Journal extents. For most other use cases it is appropriate to use
40/// `default()` here.
41#[derive(Clone, Copy, Default)]
42pub struct Options<'a> {
43    /// If true, don't check for low journal space.  This should be true for any transactions that
44    /// might alleviate journal space (i.e. compaction).
45    pub skip_journal_checks: bool,
46
47    /// If true, don't check if we need to roll the mutations key.
48    pub skip_key_roll: bool,
49
50    /// If true, borrow metadata space from the metadata reservation.  This setting should be set to
51    /// true for any transaction that will either not affect space usage after compaction
52    /// (e.g. setting attributes), or reduce space usage (e.g. unlinking).  Otherwise, a transaction
53    /// might fail with an out-of-space error.
54    pub borrow_metadata_space: bool,
55
56    /// If specified, a reservation to be used with the transaction.  If not set, any allocations
57    /// that are part of this transaction will have to take their chances, and will fail if there is
58    /// no free space.  The intention is that this should be used for things like the journal which
59    /// require guaranteed space.
60    pub allocator_reservation: Option<&'a Reservation>,
61}
62
63// This is the amount of space that we reserve for metadata when we are creating a new transaction.
64// A transaction should not take more than this.  This is expressed in terms of space occupied in
65// the journal; transactions must not take up more space in the journal than the number below.  The
66// amount chosen here must be large enough for the maximum possible transaction that can be created,
67// so transactions always need to be bounded which might involve splitting an operation up into
68// smaller transactions.
69pub const TRANSACTION_MAX_JOURNAL_USAGE: u64 = 24_576;
70pub const TRANSACTION_METADATA_MAX_AMOUNT: u64 =
71    reserved_space_from_journal_usage(TRANSACTION_MAX_JOURNAL_USAGE);
72
73#[must_use]
74pub struct TransactionLocks<'a>(pub WriteGuard<'a>);
75
76/// The journal consists of these records which will be replayed at mount time.  Within a
77/// transaction, these are stored as a set which allows some mutations to be deduplicated and found
78/// (and we require custom comparison functions below).  For example, we need to be able to find
79/// object size changes.
80pub type Mutation = MutationV56;
81
82#[derive(
83    Clone, Debug, Eq, Ord, PartialEq, PartialOrd, Serialize, Deserialize, TypeFingerprint, Versioned,
84)]
85#[cfg_attr(fuzz, derive(arbitrary::Arbitrary))]
86pub enum MutationV56 {
87    ObjectStore(ObjectStoreMutationV56),
88    EncryptedObjectStore(#[serde(with = "crate::zerocopy_serialization")] Box<[u8]>),
89    Allocator(AllocatorMutationV32),
90    /// Indicates the beginning of a flush. This would typically involve sealing a tree.
91    BeginFlush,
92    /// Indicates the end of a flush. This would typically involve replacing the immutable layers
93    /// with compacted ones.
94    EndFlush,
95    /// Volume has been deleted. Requires we remove it from the set of managed ObjectStore.
96    DeleteVolume,
97    UpdateBorrowed(u64),
98    UpdateMutationsKey(UpdateMutationsKey),
99    CreateInternalDir(u64),
100}
101
102#[derive(Migrate, Clone, Debug, PartialEq, Serialize, Deserialize, TypeFingerprint, Versioned)]
103#[migrate_to_version(MutationV56)]
104pub enum MutationV55 {
105    ObjectStore(ObjectStoreMutationV55),
106    EncryptedObjectStore(#[serde(with = "crate::zerocopy_serialization")] Box<[u8]>),
107    Allocator(AllocatorMutationV32),
108    BeginFlush,
109    EndFlush,
110    DeleteVolume,
111    UpdateBorrowed(u64),
112    UpdateMutationsKey(UpdateMutationsKey),
113    CreateInternalDir(u64),
114}
115
116#[derive(Migrate, Serialize, Deserialize, TypeFingerprint, Versioned)]
117#[migrate_to_version(MutationV55)]
118pub enum MutationV54 {
119    ObjectStore(ObjectStoreMutationV54),
120    EncryptedObjectStore(#[serde(with = "crate::zerocopy_serialization")] Box<[u8]>),
121    Allocator(AllocatorMutationV32),
122    BeginFlush,
123    EndFlush,
124    DeleteVolume,
125    UpdateBorrowed(u64),
126    UpdateMutationsKey(UpdateMutationsKey),
127    CreateInternalDir(u64),
128}
129
130#[derive(Migrate, Serialize, Deserialize, TypeFingerprint, Versioned)]
131#[migrate_to_version(MutationV54)]
132pub enum MutationV50 {
133    ObjectStore(ObjectStoreMutationV50),
134    EncryptedObjectStore(#[serde(with = "crate::zerocopy_serialization")] Box<[u8]>),
135    Allocator(AllocatorMutationV32),
136    BeginFlush,
137    EndFlush,
138    DeleteVolume,
139    UpdateBorrowed(u64),
140    UpdateMutationsKey(UpdateMutationsKeyV49),
141    CreateInternalDir(u64),
142}
143
144#[derive(Migrate, Serialize, Deserialize, TypeFingerprint, Versioned)]
145#[migrate_to_version(MutationV50)]
146pub enum MutationV49 {
147    ObjectStore(ObjectStoreMutationV49),
148    EncryptedObjectStore(Box<[u8]>),
149    Allocator(AllocatorMutationV32),
150    BeginFlush,
151    EndFlush,
152    DeleteVolume,
153    UpdateBorrowed(u64),
154    UpdateMutationsKey(UpdateMutationsKeyV49),
155    CreateInternalDir(u64),
156}
157
158#[derive(Migrate, Serialize, Deserialize, TypeFingerprint, Versioned)]
159#[migrate_to_version(MutationV49)]
160pub enum MutationV47 {
161    ObjectStore(ObjectStoreMutationV47),
162    EncryptedObjectStore(Box<[u8]>),
163    Allocator(AllocatorMutationV32),
164    BeginFlush,
165    EndFlush,
166    DeleteVolume,
167    UpdateBorrowed(u64),
168    UpdateMutationsKey(UpdateMutationsKeyV40),
169    CreateInternalDir(u64),
170}
171
172#[derive(Migrate, Serialize, Deserialize, TypeFingerprint, Versioned)]
173#[migrate_to_version(MutationV47)]
174pub enum MutationV46 {
175    ObjectStore(ObjectStoreMutationV46),
176    EncryptedObjectStore(Box<[u8]>),
177    Allocator(AllocatorMutationV32),
178    BeginFlush,
179    EndFlush,
180    DeleteVolume,
181    UpdateBorrowed(u64),
182    UpdateMutationsKey(UpdateMutationsKeyV40),
183    CreateInternalDir(u64),
184}
185
186#[derive(Migrate, Serialize, Deserialize, TypeFingerprint, Versioned)]
187#[migrate_to_version(MutationV46)]
188pub enum MutationV43 {
189    ObjectStore(ObjectStoreMutationV43),
190    EncryptedObjectStore(Box<[u8]>),
191    Allocator(AllocatorMutationV32),
192    BeginFlush,
193    EndFlush,
194    DeleteVolume,
195    UpdateBorrowed(u64),
196    UpdateMutationsKey(UpdateMutationsKeyV40),
197    CreateInternalDir(u64),
198}
199
200#[derive(Migrate, Serialize, Deserialize, TypeFingerprint, Versioned)]
201#[migrate_to_version(MutationV43)]
202pub enum MutationV41 {
203    ObjectStore(ObjectStoreMutationV41),
204    EncryptedObjectStore(Box<[u8]>),
205    Allocator(AllocatorMutationV32),
206    BeginFlush,
207    EndFlush,
208    DeleteVolume,
209    UpdateBorrowed(u64),
210    UpdateMutationsKey(UpdateMutationsKeyV40),
211    CreateInternalDir(u64),
212}
213
214#[derive(Migrate, Serialize, Deserialize, TypeFingerprint, Versioned)]
215#[migrate_to_version(MutationV41)]
216pub enum MutationV40 {
217    ObjectStore(ObjectStoreMutationV40),
218    EncryptedObjectStore(Box<[u8]>),
219    Allocator(AllocatorMutationV32),
220    BeginFlush,
221    EndFlush,
222    DeleteVolume,
223    UpdateBorrowed(u64),
224    UpdateMutationsKey(UpdateMutationsKeyV40),
225    CreateInternalDir(u64),
226}
227
228impl Mutation {
229    pub fn insert_object(key: ObjectKey, value: ObjectValue) -> Self {
230        Mutation::ObjectStore(ObjectStoreMutation {
231            item: Item::new(key, value),
232            op: Operation::Insert,
233        })
234    }
235
236    pub fn replace_or_insert_object(key: ObjectKey, value: ObjectValue) -> Self {
237        Mutation::ObjectStore(ObjectStoreMutation {
238            item: Item::new(key, value),
239            op: Operation::ReplaceOrInsert,
240        })
241    }
242
243    pub fn merge_object(key: ObjectKey, value: ObjectValue) -> Self {
244        Mutation::ObjectStore(ObjectStoreMutation {
245            item: Item::new(key, value),
246            op: Operation::Merge,
247        })
248    }
249
250    pub fn update_mutations_key(key: FxfsKey) -> Self {
251        Mutation::UpdateMutationsKey(key.into())
252    }
253}
254
255// We have custom comparison functions for mutations that just use the key, rather than the key and
256// value that would be used by default so that we can deduplicate and find mutations (see
257// get_object_mutation below).
258pub type ObjectStoreMutation = ObjectStoreMutationV56;
259
260#[derive(Clone, Debug, Serialize, Deserialize, TypeFingerprint)]
261#[cfg_attr(fuzz, derive(arbitrary::Arbitrary))]
262pub struct ObjectStoreMutationV56 {
263    pub item: ObjectItemV56,
264    pub op: Operation,
265}
266
267#[derive(Migrate, Clone, Debug, PartialEq, Serialize, Deserialize, TypeFingerprint, Versioned)]
268#[migrate_to_version(ObjectStoreMutationV56)]
269#[migrate_nodefault]
270pub struct ObjectStoreMutationV55 {
271    pub item: ObjectItemV55,
272    pub op: Operation,
273}
274
275#[derive(Migrate, Serialize, Deserialize, TypeFingerprint, Versioned)]
276#[migrate_to_version(ObjectStoreMutationV55)]
277#[migrate_nodefault]
278pub struct ObjectStoreMutationV54 {
279    pub item: crate::object_store::object_record::ObjectItemV54,
280    pub op: Operation,
281}
282
283#[derive(Migrate, Serialize, Deserialize, TypeFingerprint, Versioned)]
284#[migrate_to_version(ObjectStoreMutationV54)]
285#[migrate_nodefault]
286pub struct ObjectStoreMutationV50 {
287    pub item: ObjectItemV50,
288    pub op: OperationV32,
289}
290
291#[derive(Migrate, Serialize, Deserialize, TypeFingerprint)]
292#[migrate_to_version(ObjectStoreMutationV50)]
293#[migrate_nodefault]
294pub struct ObjectStoreMutationV49 {
295    pub item: ObjectItemV49,
296    pub op: OperationV32,
297}
298
299#[derive(Migrate, Serialize, Deserialize, TypeFingerprint)]
300#[migrate_to_version(ObjectStoreMutationV49)]
301#[migrate_nodefault]
302pub struct ObjectStoreMutationV47 {
303    pub item: ObjectItemV47,
304    pub op: OperationV32,
305}
306
307#[derive(Migrate, Serialize, Deserialize, TypeFingerprint)]
308#[migrate_to_version(ObjectStoreMutationV47)]
309#[migrate_nodefault]
310pub struct ObjectStoreMutationV46 {
311    pub item: ObjectItemV46,
312    pub op: OperationV32,
313}
314
315#[derive(Migrate, Serialize, Deserialize, TypeFingerprint)]
316#[migrate_to_version(ObjectStoreMutationV46)]
317#[migrate_nodefault]
318pub struct ObjectStoreMutationV43 {
319    pub item: ObjectItemV43,
320    pub op: OperationV32,
321}
322
323#[derive(Migrate, Serialize, Deserialize, TypeFingerprint)]
324#[migrate_to_version(ObjectStoreMutationV43)]
325#[migrate_nodefault]
326pub struct ObjectStoreMutationV41 {
327    pub item: ObjectItemV41,
328    pub op: OperationV32,
329}
330
331#[derive(Migrate, Serialize, Deserialize, TypeFingerprint)]
332#[migrate_nodefault]
333#[migrate_to_version(ObjectStoreMutationV41)]
334pub struct ObjectStoreMutationV40 {
335    pub item: ObjectItemV40,
336    pub op: OperationV32,
337}
338
339/// The different LSM tree operations that can be performed as part of a mutation.
340pub type Operation = OperationV32;
341
342#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, TypeFingerprint)]
343#[cfg_attr(fuzz, derive(arbitrary::Arbitrary))]
344pub enum OperationV32 {
345    Insert,
346    ReplaceOrInsert,
347    Merge,
348}
349
350impl Ord for ObjectStoreMutation {
351    fn cmp(&self, other: &Self) -> Ordering {
352        self.item.key.cmp(&other.item.key)
353    }
354}
355
356impl PartialOrd for ObjectStoreMutation {
357    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
358        Some(self.cmp(other))
359    }
360}
361
362impl PartialEq for ObjectStoreMutation {
363    fn eq(&self, other: &Self) -> bool {
364        self.item.key.eq(&other.item.key)
365    }
366}
367
368impl Eq for ObjectStoreMutation {}
369
370impl Ord for AllocatorItem {
371    fn cmp(&self, other: &Self) -> Ordering {
372        self.key.cmp(&other.key)
373    }
374}
375
376impl PartialOrd for AllocatorItem {
377    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
378        Some(self.cmp(other))
379    }
380}
381
382/// Same as std::ops::Range but with Ord and PartialOrd support, sorted first by start of the range,
383/// then by the end.
384pub type DeviceRange = DeviceRangeV32;
385
386#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize, TypeFingerprint)]
387#[cfg_attr(fuzz, derive(arbitrary::Arbitrary))]
388pub struct DeviceRangeV32(pub Range<u64>);
389
390impl Deref for DeviceRange {
391    type Target = Range<u64>;
392
393    fn deref(&self) -> &Self::Target {
394        &self.0
395    }
396}
397
398impl DerefMut for DeviceRange {
399    fn deref_mut(&mut self) -> &mut Self::Target {
400        &mut self.0
401    }
402}
403
404impl From<Range<u64>> for DeviceRange {
405    fn from(range: Range<u64>) -> Self {
406        Self(range)
407    }
408}
409
410impl Into<Range<u64>> for DeviceRange {
411    fn into(self) -> Range<u64> {
412        self.0
413    }
414}
415
416impl Ord for DeviceRange {
417    fn cmp(&self, other: &Self) -> Ordering {
418        self.start.cmp(&other.start).then(self.end.cmp(&other.end))
419    }
420}
421
422impl PartialOrd for DeviceRange {
423    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
424        Some(self.cmp(other))
425    }
426}
427
428pub type AllocatorMutation = AllocatorMutationV32;
429
430#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd, Serialize, Deserialize, TypeFingerprint)]
431#[cfg_attr(fuzz, derive(arbitrary::Arbitrary))]
432pub enum AllocatorMutationV32 {
433    Allocate {
434        device_range: DeviceRangeV32,
435        owner_object_id: u64,
436    },
437    Deallocate {
438        device_range: DeviceRangeV32,
439        owner_object_id: u64,
440    },
441    SetLimit {
442        owner_object_id: u64,
443        bytes: u64,
444    },
445    /// Marks all extents with a given owner_object_id for deletion.
446    /// Used to free space allocated to encrypted ObjectStore where we may not have the key.
447    /// Note that the actual deletion time is undefined so this should never be used where an
448    /// ObjectStore is still in use due to a high risk of corruption. Similarly, owner_object_id
449    /// should never be reused for the same reasons.
450    MarkForDeletion(u64),
451}
452
453pub type UpdateMutationsKey = UpdateMutationsKeyV49;
454
455#[derive(Clone, Debug, Serialize, Deserialize, TypeFingerprint)]
456pub struct UpdateMutationsKeyV49(pub FxfsKeyV49);
457
458#[derive(Migrate, Serialize, Deserialize, TypeFingerprint)]
459#[migrate_to_version(UpdateMutationsKeyV49)]
460pub struct UpdateMutationsKeyV40(pub FxfsKeyV40);
461
462impl From<UpdateMutationsKey> for FxfsKey {
463    fn from(outer: UpdateMutationsKey) -> Self {
464        outer.0
465    }
466}
467
468impl From<FxfsKey> for UpdateMutationsKey {
469    fn from(inner: FxfsKey) -> Self {
470        Self(inner)
471    }
472}
473
474#[cfg(fuzz)]
475impl<'a> arbitrary::Arbitrary<'a> for UpdateMutationsKey {
476    fn arbitrary(u: &mut arbitrary::Unstructured<'a>) -> arbitrary::Result<Self> {
477        Ok(UpdateMutationsKey::from(FxfsKey::arbitrary(u).unwrap()))
478    }
479}
480
481impl Ord for UpdateMutationsKey {
482    fn cmp(&self, other: &Self) -> Ordering {
483        (self as *const UpdateMutationsKey).cmp(&(other as *const _))
484    }
485}
486
487impl PartialOrd for UpdateMutationsKey {
488    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
489        Some(self.cmp(other))
490    }
491}
492
493impl Eq for UpdateMutationsKey {}
494
495impl PartialEq for UpdateMutationsKey {
496    fn eq(&self, other: &Self) -> bool {
497        std::ptr::eq(self, other)
498    }
499}
500
501/// When creating a transaction, locks typically need to be held to prevent two or more writers
502/// trying to make conflicting mutations at the same time.  LockKeys are used for this.
503/// NOTE: Ordering is important here!  The lock manager sorts the list of locks in a transaction
504/// to acquire them in a consistent order, but there is a special case for the Flush lock.
505/// The Flush lock is taken when we flush an LSM tree (e.g. an object store), and is held for
506/// several transactions.  As such, it must come first in the lock acquisition ordering, so that
507/// other transactions using the Flush lock have the same ordering as in flushing.
508#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Copy)]
509pub enum LockKey {
510    /// Used to lock flushing an object.
511    Flush {
512        object_id: u64,
513    },
514
515    /// Used to lock changes to a particular object attribute (e.g. writes).
516    ObjectAttribute {
517        store_object_id: u64,
518        object_id: u64,
519        attribute_id: AttributeId,
520    },
521
522    /// Used to lock changes to a particular object (e.g. adding a child to a directory).
523    Object {
524        store_object_id: u64,
525        object_id: u64,
526    },
527
528    ProjectId {
529        store_object_id: u64,
530        project_id: ProjectId,
531    },
532
533    /// Used to lock any truncate operations for a file.
534    Truncate {
535        store_object_id: u64,
536        object_id: u64,
537    },
538
539    /// A lock used when getting or creating the internal directory.
540    InternalDirectory {
541        store_object_id: u64,
542    },
543
544    /// Used to lock mutations key roll.
545    MutationsKeyRoll {
546        store_object_id: u64,
547    },
548
549    /// Used to serialize pre caching of keys.  The lock ordering is different for this: it is
550    /// acquired in `Filesystem::commit_transaction` and happens *after* other keys have been
551    /// promoted to write locks, but before the commit lock.
552    PreCacheKeys {
553        store_object_id: u64,
554    },
555}
556
557impl LockKey {
558    pub const fn object_attribute(
559        store_object_id: u64,
560        object_id: u64,
561        attribute_id: AttributeId,
562    ) -> Self {
563        LockKey::ObjectAttribute { store_object_id, object_id, attribute_id }
564    }
565
566    pub const fn object(store_object_id: u64, object_id: u64) -> Self {
567        LockKey::Object { store_object_id, object_id }
568    }
569
570    pub const fn flush(object_id: u64) -> Self {
571        LockKey::Flush { object_id }
572    }
573
574    pub const fn truncate(store_object_id: u64, object_id: u64) -> Self {
575        LockKey::Truncate { store_object_id, object_id }
576    }
577
578    pub const fn mutations_key_roll(store_object_id: u64) -> Self {
579        LockKey::MutationsKeyRoll { store_object_id }
580    }
581
582    pub const fn pre_cache_keys(store_object_id: u64) -> Self {
583        LockKey::PreCacheKeys { store_object_id }
584    }
585}
586
587/// A container for holding `LockKey` objects. Can store a single `LockKey` inline.
588#[derive(Clone, Debug)]
589pub enum LockKeys {
590    None,
591    Inline(LockKey),
592    Vec(Vec<LockKey>),
593}
594
595impl LockKeys {
596    pub fn with_capacity(capacity: usize) -> Self {
597        if capacity > 1 { LockKeys::Vec(Vec::with_capacity(capacity)) } else { LockKeys::None }
598    }
599
600    pub fn push(&mut self, key: LockKey) {
601        match self {
602            Self::None => *self = LockKeys::Inline(key),
603            Self::Inline(inline) => {
604                *self = LockKeys::Vec(vec![*inline, key]);
605            }
606            Self::Vec(vec) => vec.push(key),
607        }
608    }
609
610    pub fn truncate(&mut self, len: usize) {
611        match self {
612            Self::None => {}
613            Self::Inline(_) => {
614                if len == 0 {
615                    *self = Self::None;
616                }
617            }
618            Self::Vec(vec) => vec.truncate(len),
619        }
620    }
621
622    fn len(&self) -> usize {
623        match self {
624            Self::None => 0,
625            Self::Inline(_) => 1,
626            Self::Vec(vec) => vec.len(),
627        }
628    }
629
630    fn contains(&self, key: &LockKey) -> bool {
631        match self {
632            Self::None => false,
633            Self::Inline(single) => single == key,
634            Self::Vec(vec) => vec.contains(key),
635        }
636    }
637
638    fn sort_unstable(&mut self) {
639        match self {
640            Self::Vec(vec) => vec.sort_unstable(),
641            _ => {}
642        }
643    }
644
645    fn dedup(&mut self) {
646        match self {
647            Self::Vec(vec) => vec.dedup(),
648            _ => {}
649        }
650    }
651
652    fn iter(&self) -> LockKeysIter<'_> {
653        match self {
654            LockKeys::None => LockKeysIter::None,
655            LockKeys::Inline(key) => LockKeysIter::Inline(key),
656            LockKeys::Vec(keys) => LockKeysIter::Vec(keys.iter()),
657        }
658    }
659}
660
661enum LockKeysIter<'a> {
662    None,
663    Inline(&'a LockKey),
664    Vec(std::slice::Iter<'a, LockKey>),
665}
666
667impl<'a> Iterator for LockKeysIter<'a> {
668    type Item = &'a LockKey;
669    fn next(&mut self) -> Option<Self::Item> {
670        match self {
671            Self::None => None,
672            Self::Inline(inline) => {
673                let next = *inline;
674                *self = Self::None;
675                Some(next)
676            }
677            Self::Vec(vec) => vec.next(),
678        }
679    }
680}
681
682impl Default for LockKeys {
683    fn default() -> Self {
684        LockKeys::None
685    }
686}
687
688#[macro_export]
689macro_rules! lock_keys {
690    () => {
691        $crate::object_store::transaction::LockKeys::None
692    };
693    ($lock_key:expr $(,)?) => {
694        $crate::object_store::transaction::LockKeys::Inline($lock_key)
695    };
696    ($($lock_keys:expr),+ $(,)?) => {
697        $crate::object_store::transaction::LockKeys::Vec(vec![$($lock_keys),+])
698    };
699}
700pub use lock_keys;
701
702/// Mutations in a transaction can be associated with an object so that when mutations are applied,
703/// updates can be applied to in-memory structures. For example, we cache object sizes, so when a
704/// size change is applied, we can update the cached object size.
705pub trait AssociatedObject: Send + Sync {
706    fn will_apply_mutation(&self, _mutation: &Mutation, _object_id: u64, _manager: &ObjectManager) {
707    }
708}
709
710pub enum AssocObj<'a> {
711    None,
712    Borrowed(&'a dyn AssociatedObject),
713    Owned(Box<dyn AssociatedObject>),
714}
715
716impl AssocObj<'_> {
717    pub fn map<R, F: FnOnce(&dyn AssociatedObject) -> R>(&self, f: F) -> Option<R> {
718        match self {
719            AssocObj::None => None,
720            AssocObj::Borrowed(b) => Some(f(*b)),
721            AssocObj::Owned(o) => Some(f(o.as_ref())),
722        }
723    }
724}
725
726pub struct TxnMutation<'a> {
727    // This, at time of writing, is either the object ID of an object store, or the object ID of the
728    // allocator.  In the case of an object mutation, there's another object ID in the mutation
729    // record that would be for the object actually being changed.
730    pub object_id: u64,
731
732    // The actual mutation.  This gets serialized to the journal.
733    pub mutation: Mutation,
734
735    // An optional associated object for the mutation.  During replay, there will always be no
736    // associated object.
737    pub associated_object: AssocObj<'a>,
738}
739
740// We store TxnMutation in a set, and for that, we only use object_id and mutation and not the
741// associated object or checksum.
742//
743// WARNING: It is critical that `object_id` (which corresponds to the store ID for store mutations)
744// remains the primary key for sorting. The commit pipeline (`commit_transaction`) relies on
745// the mutations being sorted by `object_id` to acquire store locks in a consistent order,
746// preventing deadlocks.
747impl Ord for TxnMutation<'_> {
748    fn cmp(&self, other: &Self) -> Ordering {
749        self.object_id.cmp(&other.object_id).then_with(|| self.mutation.cmp(&other.mutation))
750    }
751}
752
753impl PartialOrd for TxnMutation<'_> {
754    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
755        Some(self.cmp(other))
756    }
757}
758
759impl PartialEq for TxnMutation<'_> {
760    fn eq(&self, other: &Self) -> bool {
761        self.object_id.eq(&other.object_id) && self.mutation.eq(&other.mutation)
762    }
763}
764
765impl Eq for TxnMutation<'_> {}
766
767impl std::fmt::Debug for TxnMutation<'_> {
768    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
769        f.debug_struct("TxnMutation")
770            .field("object_id", &self.object_id)
771            .field("mutation", &self.mutation)
772            .finish()
773    }
774}
775
776pub enum MetadataReservation {
777    // The state after a transaction has been dropped.
778    None,
779
780    // Metadata space for this transaction is being borrowed from ObjectManager's metadata
781    // reservation.
782    Borrowed,
783
784    // A metadata reservation was made when the transaction was created.
785    Reservation(Reservation),
786
787    // The metadata space is being _held_ within `allocator_reservation`.
788    Hold(u64),
789}
790
791/// A transaction groups mutation records to be committed as a group.
792pub struct Transaction<'a> {
793    fs: Arc<FxFilesystem>,
794
795    // The mutations that make up this transaction.
796    mutations: BTreeSet<TxnMutation<'a>>,
797
798    // The locks that this transaction currently holds.
799    txn_locks: LockKeys,
800
801    /// If set, an allocator reservation that should be used for allocations.
802    pub allocator_reservation: Option<&'a Reservation>,
803
804    /// The reservation for the metadata for this transaction.
805    pub metadata_reservation: MetadataReservation,
806
807    // Keep track of objects explicitly created by this transaction. No locks are required for them.
808    // Addressed by (owner_object_id, object_id).
809    new_objects: BTreeSet<(u64, u64)>,
810
811    /// Any data checksums which should be evaluated when replaying this transaction.
812    checksums: Vec<(Range<u64>, Vec<Checksum>, bool)>,
813
814    /// Set if this transaction contains data (i.e. includes any extent mutations).
815    includes_write: bool,
816}
817
818impl<'a> Transaction<'a> {
819    /// Creates a new transaction.  `txn_locks` are read locks that can be upgraded to write locks
820    /// at commit time.
821    pub async fn new(
822        fs: Arc<FxFilesystem>,
823        options: Options<'a>,
824        txn_locks: LockKeys,
825    ) -> Result<Transaction<'a>, Error> {
826        fs.add_transaction(options.skip_journal_checks).await;
827        let fs_clone = fs.clone();
828        let guard = scopeguard::guard((), |_| fs_clone.sub_transaction());
829        let (metadata_reservation, allocator_reservation, hold) =
830            fs.reservation_for_transaction(options).await?;
831
832        let txn_locks = {
833            let lock_manager = fs.lock_manager();
834            let mut write_guard = lock_manager.txn_lock(txn_locks).await;
835            std::mem::take(&mut write_guard.0.lock_keys)
836        };
837        let mut transaction = Transaction {
838            fs,
839            mutations: BTreeSet::new(),
840            txn_locks,
841            allocator_reservation: None,
842            metadata_reservation,
843            new_objects: BTreeSet::new(),
844            checksums: Vec::new(),
845            includes_write: false,
846        };
847
848        ScopeGuard::into_inner(guard);
849        hold.map(|h| h.forget()); // Transaction takes ownership from here on.
850        transaction.allocator_reservation = allocator_reservation;
851        Ok(transaction)
852    }
853
854    pub fn mutations(&self) -> &BTreeSet<TxnMutation<'a>> {
855        &self.mutations
856    }
857
858    pub fn take_mutations(&mut self) -> BTreeSet<TxnMutation<'a>> {
859        self.new_objects.clear();
860        mem::take(&mut self.mutations)
861    }
862
863    /// Adds a mutation to this transaction.  If the mutation already exists, it is replaced and the
864    /// old mutation is returned.
865    pub fn add(&mut self, object_id: u64, mutation: Mutation) -> Option<Mutation> {
866        self.add_with_object(object_id, mutation, AssocObj::None)
867    }
868
869    /// Removes a mutation that matches `mutation`.
870    pub fn remove(&mut self, object_id: u64, mutation: Mutation) {
871        let txn_mutation = TxnMutation { object_id, mutation, associated_object: AssocObj::None };
872        if self.mutations.remove(&txn_mutation) {
873            if let Mutation::ObjectStore(ObjectStoreMutation {
874                item:
875                    ObjectItem {
876                        key: ObjectKey { object_id: new_object_id, data: ObjectKeyData::Object },
877                        ..
878                    },
879                op: Operation::Insert,
880            }) = txn_mutation.mutation
881            {
882                self.new_objects.remove(&(object_id, new_object_id));
883            }
884        }
885    }
886
887    /// Adds a mutation with an associated object. If the mutation already exists, it is replaced
888    /// and the old mutation is returned.
889    pub fn add_with_object(
890        &mut self,
891        object_id: u64,
892        mutation: Mutation,
893        associated_object: AssocObj<'a>,
894    ) -> Option<Mutation> {
895        assert!(object_id != INVALID_OBJECT_ID);
896        if let Mutation::ObjectStore(ObjectStoreMutation {
897            item:
898                Item {
899                    key:
900                        ObjectKey { data: ObjectKeyData::Attribute(_, AttributeKey::Extent(_)), .. },
901                    ..
902                },
903            ..
904        }) = &mutation
905        {
906            self.includes_write = true;
907        }
908        let txn_mutation = TxnMutation { object_id, mutation, associated_object };
909        self.verify_locks(&txn_mutation);
910        self.mutations.replace(txn_mutation).map(|m| m.mutation)
911    }
912
913    pub fn add_checksum(&mut self, range: Range<u64>, checksums: Vec<Checksum>, first_write: bool) {
914        self.checksums.push((range, checksums, first_write));
915    }
916
917    pub fn includes_write(&self) -> bool {
918        self.includes_write
919    }
920
921    pub fn checksums(&self) -> &[(Range<u64>, Vec<Checksum>, bool)] {
922        &self.checksums
923    }
924
925    pub fn take_checksums(&mut self) -> Vec<(Range<u64>, Vec<Checksum>, bool)> {
926        std::mem::replace(&mut self.checksums, Vec::new())
927    }
928
929    fn verify_locks(&mut self, mutation: &TxnMutation<'_>) {
930        // It was considered to change the locks from Vec to BTreeSet since we'll now be searching
931        // through it, but given the small set that these locks usually comprise, it probably isn't
932        // worth it.
933        match mutation {
934            TxnMutation {
935                mutation:
936                    Mutation::ObjectStore {
937                        0: ObjectStoreMutation { item: ObjectItem { key, .. }, op },
938                    },
939                object_id: store_object_id,
940                ..
941            } => {
942                match &key.data {
943                    ObjectKeyData::Attribute(..) => {
944                        // TODO(https://fxbug.dev/42073914): Check lock requirements.
945                    }
946                    ObjectKeyData::Child { .. }
947                    | ObjectKeyData::EncryptedChild(_)
948                    | ObjectKeyData::EncryptedCasefoldChild(_)
949                    | ObjectKeyData::CasefoldChild { .. }
950                    | ObjectKeyData::LegacyCasefoldChild(_) => {
951                        let id = key.object_id;
952                        if !self.txn_locks.contains(&LockKey::object(*store_object_id, id))
953                            && !self.new_objects.contains(&(*store_object_id, id))
954                        {
955                            debug_assert!(
956                                false,
957                                "Not holding required lock for object {id} \
958                                in store {store_object_id}"
959                            );
960                            error!(
961                                "Not holding required lock for object {id} in store \
962                                {store_object_id}"
963                            )
964                        }
965                    }
966                    ObjectKeyData::GraveyardEntry { .. } => {
967                        // TODO(https://fxbug.dev/42073911): Check lock requirements.
968                    }
969                    ObjectKeyData::GraveyardAttributeEntry { .. } => {
970                        // TODO(https://fxbug.dev/122974): Check lock requirements.
971                    }
972                    ObjectKeyData::Keys => {
973                        let id = key.object_id;
974                        if !self.txn_locks.contains(&LockKey::object(*store_object_id, id))
975                            && !self.new_objects.contains(&(*store_object_id, id))
976                        {
977                            debug_assert!(
978                                false,
979                                "Not holding required lock for object {id} \
980                                in store {store_object_id}"
981                            );
982                            error!(
983                                "Not holding required lock for object {id} in store \
984                                {store_object_id}"
985                            )
986                        }
987                    }
988                    ObjectKeyData::Object => match op {
989                        // Insert implies the caller expects no object with which to race
990                        Operation::Insert => {
991                            self.new_objects.insert((*store_object_id, key.object_id));
992                        }
993                        Operation::Merge | Operation::ReplaceOrInsert => {
994                            let id = key.object_id;
995                            if !self.txn_locks.contains(&LockKey::object(*store_object_id, id))
996                                && !self.new_objects.contains(&(*store_object_id, id))
997                            {
998                                debug_assert!(
999                                    false,
1000                                    "Not holding required lock for object {id} \
1001                                    in store {store_object_id}"
1002                                );
1003                                error!(
1004                                    "Not holding required lock for object {id} in store \
1005                                    {store_object_id}"
1006                                )
1007                            }
1008                        }
1009                    },
1010                    ObjectKeyData::Project { project_id, property: ProjectProperty::Limit } => {
1011                        if !self.txn_locks.contains(&LockKey::ProjectId {
1012                            store_object_id: *store_object_id,
1013                            project_id: *project_id,
1014                        }) {
1015                            debug_assert!(
1016                                false,
1017                                "Not holding required lock for project limit id {project_id} \
1018                                in store {store_object_id}"
1019                            );
1020                            error!(
1021                                "Not holding required lock for project limit id {project_id} in \
1022                                store {store_object_id}"
1023                            )
1024                        }
1025                    }
1026                    ObjectKeyData::Project { property: ProjectProperty::Usage, .. } => match op {
1027                        Operation::Insert | Operation::ReplaceOrInsert => {
1028                            panic!(
1029                                "Project usage is all handled by merging deltas, no inserts or \
1030                                replacements should be used"
1031                            );
1032                        }
1033                        // Merges are all handled like atomic +/- and serialized by the tree locks.
1034                        Operation::Merge => {}
1035                    },
1036                    ObjectKeyData::ExtendedAttribute { .. } => {
1037                        let id = key.object_id;
1038                        if !self.txn_locks.contains(&LockKey::object(*store_object_id, id))
1039                            && !self.new_objects.contains(&(*store_object_id, id))
1040                        {
1041                            debug_assert!(
1042                                false,
1043                                "Not holding required lock for object {id} \
1044                                in store {store_object_id} while mutating extended attribute"
1045                            );
1046                            error!(
1047                                "Not holding required lock for object {id} in store \
1048                                {store_object_id} while mutating extended attribute"
1049                            )
1050                        }
1051                    }
1052                }
1053            }
1054            TxnMutation { mutation: Mutation::DeleteVolume, object_id, .. } => {
1055                if !self.txn_locks.contains(&LockKey::flush(*object_id)) {
1056                    debug_assert!(false, "Not holding required lock for DeleteVolume");
1057                    error!("Not holding required lock for DeleteVolume");
1058                }
1059            }
1060            _ => {}
1061        }
1062    }
1063
1064    /// Returns true if this transaction has no mutations.
1065    pub fn is_empty(&self) -> bool {
1066        self.mutations.is_empty()
1067    }
1068
1069    /// Searches for an existing object mutation within the transaction that has the given key and
1070    /// returns it if found.
1071    pub fn get_object_mutation(
1072        &self,
1073        store_object_id: u64,
1074        key: ObjectKey,
1075    ) -> Option<&ObjectStoreMutation> {
1076        if let Some(TxnMutation { mutation: Mutation::ObjectStore(mutation), .. }) =
1077            self.mutations.get(&TxnMutation {
1078                object_id: store_object_id,
1079                mutation: Mutation::insert_object(key, ObjectValue::None),
1080                associated_object: AssocObj::None,
1081            })
1082        {
1083            Some(mutation)
1084        } else {
1085            None
1086        }
1087    }
1088
1089    /// Commits a transaction.  If successful, returns the journal offset of the transaction.
1090    pub async fn commit(mut self) -> Result<u64, Error> {
1091        debug!(txn:? = &self; "Commit");
1092        self.fs.clone().commit_transaction(&mut self, |x| x).await
1093    }
1094
1095    /// Commits and then runs the callback whilst locks are held.  The callback accepts a single
1096    /// parameter which is the journal offset of the transaction.
1097    pub async fn commit_with_callback<R: Send>(
1098        mut self,
1099        f: impl FnOnce(u64) -> R + Send,
1100    ) -> Result<R, Error> {
1101        debug!(txn:? = &self; "Commit");
1102        self.fs.clone().commit_transaction(&mut self, f).await
1103    }
1104
1105    /// Commits the transaction, but allows the transaction to be used again.  The locks are not
1106    /// dropped (but transaction locks will get downgraded to read locks).
1107    pub async fn commit_and_continue(&mut self) -> Result<(), Error> {
1108        debug!(txn:? = self; "Commit");
1109        self.fs.clone().commit_transaction(self, |_| {}).await?;
1110        assert!(self.mutations.is_empty());
1111        self.fs.lock_manager().downgrade_locks(&self.txn_locks);
1112        Ok(())
1113    }
1114}
1115
1116impl Drop for Transaction<'_> {
1117    fn drop(&mut self) {
1118        // Call the filesystem implementation of drop_transaction which should, as a minimum, call
1119        // LockManager's drop_transaction to ensure the locks are released.
1120        debug!(txn:? = &self; "Drop");
1121        self.fs.clone().drop_transaction(self);
1122    }
1123}
1124
1125impl std::fmt::Debug for Transaction<'_> {
1126    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1127        f.debug_struct("Transaction")
1128            .field("mutations", &self.mutations)
1129            .field("txn_locks", &self.txn_locks)
1130            .field("reservation", &self.allocator_reservation)
1131            .finish()
1132    }
1133}
1134
1135pub enum BorrowedOrOwned<'a, T> {
1136    Borrowed(&'a T),
1137    Owned(T),
1138}
1139
1140impl<T> Deref for BorrowedOrOwned<'_, T> {
1141    type Target = T;
1142
1143    fn deref(&self) -> &Self::Target {
1144        match self {
1145            BorrowedOrOwned::Borrowed(b) => b,
1146            BorrowedOrOwned::Owned(o) => &o,
1147        }
1148    }
1149}
1150
1151impl<'a, T> From<&'a T> for BorrowedOrOwned<'a, T> {
1152    fn from(value: &'a T) -> Self {
1153        BorrowedOrOwned::Borrowed(value)
1154    }
1155}
1156
1157impl<T> From<T> for BorrowedOrOwned<'_, T> {
1158    fn from(value: T) -> Self {
1159        BorrowedOrOwned::Owned(value)
1160    }
1161}
1162
1163/// LockManager holds the locks that transactions might have taken.  A TransactionManager
1164/// implementation would typically have one of these.
1165///
1166/// Three different kinds of locks are supported.  There are read locks and write locks, which are
1167/// as one would expect.  The third kind of lock is a _transaction_ lock (which is also known as an
1168/// upgradeable read lock).  When first acquired, these block other writes (including other
1169/// transaction locks) but do not block reads.  When it is time to commit a transaction, these locks
1170/// are upgraded to full write locks (without ever dropping the lock) and then dropped after
1171/// committing (unless commit_and_continue is used).  This way, reads are only blocked for the
1172/// shortest possible time.  It follows that write locks should be used sparingly.  Locks are
1173/// granted in order with one exception: when a lock is in the initial _transaction_ lock state
1174/// (LockState::Locked), all read locks are allowed even if there are other tasks waiting for the
1175/// lock.  The reason for this is because we allow read locks to be taken by tasks that have taken a
1176/// _transaction_ lock (i.e. recursion is allowed).  In other cases, such as when a writer is
1177/// waiting and there are only readers, readers will queue up behind the writer.
1178///
1179/// To summarize:
1180///
1181/// +-------------------------+-----------------+----------------+------------------+
1182/// |                         | While read_lock | While txn_lock | While write_lock |
1183/// |                         | is held         | is held        | is held          |
1184/// +-------------------------+-----------------+----------------+------------------+
1185/// | Can acquire read_lock?  | true            | true           | false            |
1186/// +-------------------------+-----------------+----------------+------------------+
1187/// | Can acquire txn_lock?   | true            | false          | false            |
1188/// +-------------------------+-----------------+----------------+------------------+
1189/// | Can acquire write_lock? | false           | false          | false            |
1190/// +-------------------------+-----------------+----------------+------------------+
1191pub struct LockManager {
1192    locks: Mutex<Locks>,
1193}
1194
1195struct Locks {
1196    keys: HashMap<LockKey, LockEntry>,
1197}
1198
1199impl Locks {
1200    fn drop_lock(&mut self, key: LockKey, state: LockState) {
1201        if let Entry::Occupied(mut occupied) = self.keys.entry(key) {
1202            let entry = occupied.get_mut();
1203            let wake = match state {
1204                LockState::ReadLock => {
1205                    entry.read_count -= 1;
1206                    entry.read_count == 0
1207                }
1208                // drop_write_locks currently depends on us treating Locked and WriteLock the same.
1209                LockState::Locked | LockState::WriteLock => {
1210                    entry.state = LockState::ReadLock;
1211                    true
1212                }
1213            };
1214            if wake {
1215                // SAFETY: The lock in `LockManager::locks` is held.
1216                unsafe {
1217                    entry.wake();
1218                }
1219                if entry.can_remove() {
1220                    occupied.remove_entry();
1221                }
1222            }
1223        } else {
1224            unreachable!();
1225        }
1226    }
1227
1228    fn drop_read_locks(&mut self, lock_keys: LockKeys) {
1229        for lock in lock_keys.iter() {
1230            self.drop_lock(*lock, LockState::ReadLock);
1231        }
1232    }
1233
1234    fn drop_write_locks(&mut self, lock_keys: LockKeys) {
1235        for lock in lock_keys.iter() {
1236            // This is a bit hacky, but this works for locks in either the Locked or WriteLock
1237            // states.
1238            self.drop_lock(*lock, LockState::WriteLock);
1239        }
1240    }
1241
1242    // Downgrades locks from WriteLock to Locked.
1243    fn downgrade_locks(&mut self, lock_keys: &LockKeys) {
1244        for lock in lock_keys.iter() {
1245            // SAFETY: The lock in `LockManager::locks` is held.
1246            unsafe {
1247                self.keys.get_mut(lock).unwrap().downgrade_lock();
1248            }
1249        }
1250    }
1251}
1252
1253#[derive(Debug)]
1254struct LockEntry {
1255    // In the states that allow readers (ReadLock, Locked), this count can be non-zero
1256    // to indicate the number of active readers.
1257    read_count: u64,
1258
1259    // The state of the lock (see below).
1260    state: LockState,
1261
1262    // A doubly-linked list of wakers that should be woken when they have been granted the lock.
1263    // New wakers are usually chained on to tail, with the exception being the case where a lock in
1264    // state Locked is to be upgraded to WriteLock, but can't because there are readers.  It might
1265    // be possible to use intrusive-collections in the future.
1266    head: *const LockWaker,
1267    tail: *const LockWaker,
1268}
1269
1270unsafe impl Send for LockEntry {}
1271
1272// Represents a node in the waker list.  It is only safe to access the members wrapped by UnsafeCell
1273// when LockManager's `locks` member is locked.
1274struct LockWaker {
1275    // The next and previous pointers in the doubly-linked list.
1276    next: UnsafeCell<*const LockWaker>,
1277    prev: UnsafeCell<*const LockWaker>,
1278
1279    // Holds the lock key for this waker.  This is required so that we can find the associated
1280    // `LockEntry`.
1281    key: LockKey,
1282
1283    // The underlying waker that should be used to wake the task.
1284    waker: UnsafeCell<WakerState>,
1285
1286    // The target state for this waker.
1287    target_state: LockState,
1288
1289    // True if this is an upgrade.
1290    is_upgrade: bool,
1291
1292    // We need to be pinned because these form part of the linked list.
1293    _pin: PhantomPinned,
1294}
1295
1296enum WakerState {
1297    // This is the initial state before the waker has been first polled.
1298    Pending,
1299
1300    // Once polled, this contains the actual waker.
1301    Registered(Waker),
1302
1303    // The waker has been woken and has been granted the lock.
1304    Woken,
1305}
1306
1307impl WakerState {
1308    fn is_woken(&self) -> bool {
1309        matches!(self, WakerState::Woken)
1310    }
1311}
1312
1313unsafe impl Send for LockWaker {}
1314unsafe impl Sync for LockWaker {}
1315
1316impl LockWaker {
1317    // Waits for the waker to be woken.
1318    async fn wait(&self, manager: &LockManager) {
1319        // We must guard against the future being dropped.
1320        let waker_guard = scopeguard::guard((), |_| {
1321            let mut locks = manager.locks.lock();
1322            // SAFETY: We've acquired the lock.
1323            unsafe {
1324                if (*self.waker.get()).is_woken() {
1325                    // We were woken, but didn't actually run, so we must drop the lock.
1326                    if self.is_upgrade {
1327                        locks.keys.get_mut(&self.key).unwrap().downgrade_lock();
1328                    } else {
1329                        locks.drop_lock(self.key, self.target_state);
1330                    }
1331                } else {
1332                    // We haven't been woken but we've been dropped so we must remove ourself from
1333                    // the waker list.
1334                    locks.keys.get_mut(&self.key).unwrap().remove_waker(self);
1335                }
1336            }
1337        });
1338
1339        poll_fn(|cx| {
1340            let _locks = manager.locks.lock();
1341            // SAFETY: We've acquired the lock.
1342            unsafe {
1343                if (*self.waker.get()).is_woken() {
1344                    Poll::Ready(())
1345                } else {
1346                    *self.waker.get() = WakerState::Registered(cx.waker().clone());
1347                    Poll::Pending
1348                }
1349            }
1350        })
1351        .await;
1352
1353        ScopeGuard::into_inner(waker_guard);
1354    }
1355}
1356
1357#[derive(Copy, Clone, Debug, PartialEq)]
1358enum LockState {
1359    // In this state, there are only readers.
1360    ReadLock,
1361
1362    // This state is used for transactions to lock other writers (including other transactions), but
1363    // it still allows readers.
1364    Locked,
1365
1366    // A writer has exclusive access; all other readers and writers are blocked.
1367    WriteLock,
1368}
1369
1370impl LockManager {
1371    pub fn new() -> Self {
1372        LockManager { locks: Mutex::new(Locks { keys: HashMap::default() }) }
1373    }
1374
1375    /// Acquires the locks.  It is the caller's responsibility to ensure that drop_transaction is
1376    /// called when a transaction is dropped i.e. the filesystem's drop_transaction method should
1377    /// call LockManager's drop_transaction method.
1378    pub async fn txn_lock<'a>(&'a self, lock_keys: LockKeys) -> TransactionLocks<'a> {
1379        TransactionLocks(
1380            debug_assert_not_too_long!(self.lock(lock_keys, LockState::Locked)).right().unwrap(),
1381        )
1382    }
1383
1384    // `state` indicates the kind of lock required.  ReadLock means acquire a read lock.  Locked
1385    // means lock other writers, but still allow readers.  WriteLock means acquire a write lock.
1386    async fn lock<'a>(
1387        &'a self,
1388        mut lock_keys: LockKeys,
1389        target_state: LockState,
1390    ) -> Either<ReadGuard<'a>, WriteGuard<'a>> {
1391        let mut guard = match &target_state {
1392            LockState::ReadLock => Left(ReadGuard {
1393                manager: self.into(),
1394                lock_keys: LockKeys::with_capacity(lock_keys.len()),
1395            }),
1396            LockState::Locked | LockState::WriteLock => Right(WriteGuard {
1397                manager: self.into(),
1398                lock_keys: LockKeys::with_capacity(lock_keys.len()),
1399            }),
1400        };
1401        let guard_keys = match &mut guard {
1402            Left(g) => &mut g.lock_keys,
1403            Right(g) => &mut g.lock_keys,
1404        };
1405        lock_keys.sort_unstable();
1406        lock_keys.dedup();
1407        for lock in lock_keys.iter() {
1408            let lock_waker = None;
1409            pin_mut!(lock_waker);
1410            {
1411                let mut locks = self.locks.lock();
1412                match locks.keys.entry(*lock) {
1413                    Entry::Vacant(vacant) => {
1414                        vacant.insert(LockEntry {
1415                            read_count: if let LockState::ReadLock = target_state {
1416                                guard_keys.push(*lock);
1417                                1
1418                            } else {
1419                                guard_keys.push(*lock);
1420                                0
1421                            },
1422                            state: target_state,
1423                            head: std::ptr::null(),
1424                            tail: std::ptr::null(),
1425                        });
1426                    }
1427                    Entry::Occupied(mut occupied) => {
1428                        let entry = occupied.get_mut();
1429                        // SAFETY: We've acquired the lock.
1430                        if unsafe { entry.is_allowed(target_state, entry.head.is_null()) } {
1431                            if let LockState::ReadLock = target_state {
1432                                entry.read_count += 1;
1433                                guard_keys.push(*lock);
1434                            } else {
1435                                entry.state = target_state;
1436                                guard_keys.push(*lock);
1437                            }
1438                        } else {
1439                            // Initialise a waker and push it on the tail of the list.
1440                            // SAFETY: `lock_waker` isn't used prior to this point.
1441                            unsafe {
1442                                *lock_waker.as_mut().get_unchecked_mut() = Some(LockWaker {
1443                                    next: UnsafeCell::new(std::ptr::null()),
1444                                    prev: UnsafeCell::new(entry.tail),
1445                                    key: *lock,
1446                                    waker: UnsafeCell::new(WakerState::Pending),
1447                                    target_state: target_state,
1448                                    is_upgrade: false,
1449                                    _pin: PhantomPinned,
1450                                });
1451                            }
1452                            let waker = (*lock_waker).as_ref().unwrap();
1453                            if entry.tail.is_null() {
1454                                entry.head = waker;
1455                            } else {
1456                                // SAFETY: We've acquired the lock.
1457                                unsafe {
1458                                    *(*entry.tail).next.get() = waker;
1459                                }
1460                            }
1461                            entry.tail = waker;
1462                        }
1463                    }
1464                }
1465            }
1466            if let Some(waker) = &*lock_waker {
1467                waker.wait(self).await;
1468                guard_keys.push(*lock);
1469            }
1470        }
1471        guard
1472    }
1473
1474    /// This should be called by the filesystem's drop_transaction implementation.
1475    pub fn drop_transaction(&self, transaction: &mut Transaction<'_>) {
1476        let mut locks = self.locks.lock();
1477        locks.drop_write_locks(std::mem::take(&mut transaction.txn_locks));
1478    }
1479
1480    /// Prepares to commit by waiting for readers to finish.
1481    pub async fn commit_prepare(&self, transaction: &Transaction<'_>) {
1482        self.commit_prepare_keys(&transaction.txn_locks).await;
1483    }
1484
1485    async fn commit_prepare_keys(&self, lock_keys: &LockKeys) {
1486        for lock in lock_keys.iter() {
1487            let lock_waker = None;
1488            pin_mut!(lock_waker);
1489            {
1490                let mut locks = self.locks.lock();
1491                let entry = locks.keys.get_mut(lock).unwrap();
1492                assert_eq!(entry.state, LockState::Locked);
1493
1494                if entry.read_count == 0 {
1495                    entry.state = LockState::WriteLock;
1496                } else {
1497                    // Initialise a waker and push it on the head of the list.
1498                    // SAFETY: `lock_waker` isn't used prior to this point.
1499                    unsafe {
1500                        *lock_waker.as_mut().get_unchecked_mut() = Some(LockWaker {
1501                            next: UnsafeCell::new(entry.head),
1502                            prev: UnsafeCell::new(std::ptr::null()),
1503                            key: *lock,
1504                            waker: UnsafeCell::new(WakerState::Pending),
1505                            target_state: LockState::WriteLock,
1506                            is_upgrade: true,
1507                            _pin: PhantomPinned,
1508                        });
1509                    }
1510                    let waker = (*lock_waker).as_ref().unwrap();
1511                    if entry.head.is_null() {
1512                        entry.tail = (*lock_waker).as_ref().unwrap();
1513                    } else {
1514                        // SAFETY: We've acquired the lock.
1515                        unsafe {
1516                            *(*entry.head).prev.get() = waker;
1517                        }
1518                    }
1519                    entry.head = waker;
1520                }
1521            }
1522
1523            if let Some(waker) = &*lock_waker {
1524                waker.wait(self).await;
1525            }
1526        }
1527    }
1528
1529    /// Acquires a read lock for the given keys.  Read locks are only blocked whilst a transaction
1530    /// is being committed for the same locks.  They are only necessary where consistency is
1531    /// required between different mutations within a transaction.  For example, a write might
1532    /// change the size and extents for an object, in which case a read lock is required so that
1533    /// observed size and extents are seen together or not at all.
1534    pub async fn read_lock<'a>(&'a self, lock_keys: LockKeys) -> ReadGuard<'a> {
1535        debug_assert_not_too_long!(self.lock(lock_keys, LockState::ReadLock)).left().unwrap()
1536    }
1537
1538    /// Acquires a write lock for the given keys.  Write locks provide exclusive access to the
1539    /// requested lock keys.
1540    pub async fn write_lock<'a>(&'a self, lock_keys: LockKeys) -> WriteGuard<'a> {
1541        debug_assert_not_too_long!(self.lock(lock_keys, LockState::WriteLock)).right().unwrap()
1542    }
1543
1544    /// Downgrades locks from the WriteLock state to Locked state.  This will panic if the locks are
1545    /// not in the WriteLock state.
1546    pub fn downgrade_locks(&self, lock_keys: &LockKeys) {
1547        self.locks.lock().downgrade_locks(lock_keys);
1548    }
1549}
1550
1551// These unsafe functions require that `locks` in LockManager is locked.
1552impl LockEntry {
1553    unsafe fn wake(&mut self) {
1554        // If the lock's state is WriteLock, or there's nothing waiting, return early.
1555        if self.head.is_null() || self.state == LockState::WriteLock {
1556            return;
1557        }
1558
1559        let waker = unsafe { &*self.head };
1560
1561        if waker.is_upgrade {
1562            if self.read_count > 0 {
1563                return;
1564            }
1565        } else if !unsafe { self.is_allowed(waker.target_state, true) } {
1566            return;
1567        }
1568
1569        unsafe { self.pop_and_wake() };
1570
1571        // If the waker was a write lock, we can't wake any more up, but otherwise, we can keep
1572        // waking up readers.
1573        if waker.target_state == LockState::WriteLock {
1574            return;
1575        }
1576
1577        while !self.head.is_null() && unsafe { (*self.head).target_state } == LockState::ReadLock {
1578            unsafe { self.pop_and_wake() };
1579        }
1580    }
1581
1582    unsafe fn pop_and_wake(&mut self) {
1583        let waker = unsafe { &*self.head };
1584
1585        // Pop the waker.
1586        self.head = unsafe { *waker.next.get() };
1587        if self.head.is_null() {
1588            self.tail = std::ptr::null()
1589        } else {
1590            unsafe { *(*self.head).prev.get() = std::ptr::null() };
1591        }
1592
1593        // Adjust our state accordingly.
1594        if waker.target_state == LockState::ReadLock {
1595            self.read_count += 1;
1596        } else {
1597            self.state = waker.target_state;
1598        }
1599
1600        // Now wake the task.
1601        if let WakerState::Registered(waker) =
1602            std::mem::replace(unsafe { &mut *waker.waker.get() }, WakerState::Woken)
1603        {
1604            waker.wake();
1605        }
1606    }
1607
1608    fn can_remove(&self) -> bool {
1609        self.state == LockState::ReadLock && self.read_count == 0
1610    }
1611
1612    unsafe fn remove_waker(&mut self, waker: &LockWaker) {
1613        unsafe {
1614            let is_first = (*waker.prev.get()).is_null();
1615            if is_first {
1616                self.head = *waker.next.get();
1617            } else {
1618                *(**waker.prev.get()).next.get() = *waker.next.get();
1619            }
1620            if (*waker.next.get()).is_null() {
1621                self.tail = *waker.prev.get();
1622            } else {
1623                *(**waker.next.get()).prev.get() = *waker.prev.get();
1624            }
1625            if is_first {
1626                // We must call wake in case we erased a pending write lock and readers can now
1627                // proceed.
1628                self.wake();
1629            }
1630        }
1631    }
1632
1633    // Returns whether or not a lock with given `target_state` can proceed.  `is_head` should be
1634    // true if this is something at the head of the waker list (or the waker list is empty) and
1635    // false if there are other items on the waker list that are prior.
1636    unsafe fn is_allowed(&self, target_state: LockState, is_head: bool) -> bool {
1637        match self.state {
1638            LockState::ReadLock => {
1639                // Allow ReadLock and Locked so long as nothing else is waiting.
1640                (self.read_count == 0
1641                    || target_state == LockState::Locked
1642                    || target_state == LockState::ReadLock)
1643                    && is_head
1644            }
1645            LockState::Locked => {
1646                // Always allow reads unless there's an upgrade waiting.  We have to
1647                // always allow reads in this state because tasks that have locks in
1648                // the Locked state can later try and acquire ReadLock.
1649                target_state == LockState::ReadLock
1650                    && (is_head || unsafe { !(*self.head).is_upgrade })
1651            }
1652            LockState::WriteLock => false,
1653        }
1654    }
1655
1656    unsafe fn downgrade_lock(&mut self) {
1657        assert_eq!(std::mem::replace(&mut self.state, LockState::Locked), LockState::WriteLock);
1658        unsafe { self.wake() };
1659    }
1660}
1661
1662#[must_use]
1663pub struct ReadGuard<'a> {
1664    manager: LockManagerRef<'a>,
1665    lock_keys: LockKeys,
1666}
1667
1668impl ReadGuard<'_> {
1669    pub fn fs(&self) -> Option<&Arc<FxFilesystem>> {
1670        if let LockManagerRef::Owned(fs) = &self.manager { Some(fs) } else { None }
1671    }
1672
1673    pub fn into_owned(mut self, fs: Arc<FxFilesystem>) -> ReadGuard<'static> {
1674        ReadGuard {
1675            manager: LockManagerRef::Owned(fs),
1676            lock_keys: std::mem::replace(&mut self.lock_keys, LockKeys::None),
1677        }
1678    }
1679}
1680
1681impl Drop for ReadGuard<'_> {
1682    fn drop(&mut self) {
1683        let mut locks = self.manager.locks.lock();
1684        locks.drop_read_locks(std::mem::take(&mut self.lock_keys));
1685    }
1686}
1687
1688impl fmt::Debug for ReadGuard<'_> {
1689    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1690        f.debug_struct("ReadGuard")
1691            .field("manager", &(&self.manager as *const _))
1692            .field("lock_keys", &self.lock_keys)
1693            .finish()
1694    }
1695}
1696
1697#[must_use]
1698pub struct WriteGuard<'a> {
1699    manager: LockManagerRef<'a>,
1700    lock_keys: LockKeys,
1701}
1702
1703impl Drop for WriteGuard<'_> {
1704    fn drop(&mut self) {
1705        let mut locks = self.manager.locks.lock();
1706        locks.drop_write_locks(std::mem::take(&mut self.lock_keys));
1707    }
1708}
1709
1710impl fmt::Debug for WriteGuard<'_> {
1711    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1712        f.debug_struct("WriteGuard")
1713            .field("manager", &(&self.manager as *const _))
1714            .field("lock_keys", &self.lock_keys)
1715            .finish()
1716    }
1717}
1718
1719enum LockManagerRef<'a> {
1720    Borrowed(&'a LockManager),
1721    Owned(Arc<FxFilesystem>),
1722}
1723
1724impl Deref for LockManagerRef<'_> {
1725    type Target = LockManager;
1726
1727    fn deref(&self) -> &Self::Target {
1728        match self {
1729            LockManagerRef::Borrowed(m) => m,
1730            LockManagerRef::Owned(f) => f.lock_manager(),
1731        }
1732    }
1733}
1734
1735impl<'a> From<&'a LockManager> for LockManagerRef<'a> {
1736    fn from(value: &'a LockManager) -> Self {
1737        LockManagerRef::Borrowed(value)
1738    }
1739}
1740
1741#[cfg(test)]
1742mod tests {
1743    use super::{AttributeId, LockKey, LockKeys, LockManager, LockState, Mutation, Options};
1744    use crate::filesystem::FxFilesystem;
1745    use fuchsia_async as fasync;
1746    use fuchsia_sync::Mutex;
1747    use futures::channel::oneshot::channel;
1748    use futures::future::FutureExt;
1749    use futures::stream::FuturesUnordered;
1750    use futures::{StreamExt, join, pin_mut};
1751    use std::task::Poll;
1752    use std::time::Duration;
1753    use storage_device::DeviceHolder;
1754    use storage_device::fake_device::FakeDevice;
1755
1756    #[fuchsia::test]
1757    async fn test_simple() {
1758        let device = DeviceHolder::new(FakeDevice::new(4096, 1024));
1759        let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
1760        let mut t = fs
1761            .root_store()
1762            .new_transaction(lock_keys![], Options::default())
1763            .await
1764            .expect("new_transaction failed");
1765        t.add(1, Mutation::BeginFlush);
1766        assert!(!t.is_empty());
1767    }
1768
1769    #[fuchsia::test]
1770    async fn test_locks() {
1771        let device = DeviceHolder::new(FakeDevice::new(4096, 1024));
1772        let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
1773        let (send1, recv1) = channel();
1774        let (send2, recv2) = channel();
1775        let (send3, recv3) = channel();
1776        let done = Mutex::new(false);
1777        let mut futures = FuturesUnordered::new();
1778        futures.push(
1779            async {
1780                let _t = fs
1781                    .root_store()
1782                    .new_transaction(
1783                        lock_keys![LockKey::object_attribute(1, 2, AttributeId::TEST_ID)],
1784                        Options::default(),
1785                    )
1786                    .await
1787                    .expect("new_transaction failed");
1788                send1.send(()).unwrap(); // Tell the next future to continue.
1789                send3.send(()).unwrap(); // Tell the last future to continue.
1790                recv2.await.unwrap();
1791                // This is a halting problem so all we can do is sleep.
1792                fasync::Timer::new(Duration::from_millis(100)).await;
1793                assert!(!*done.lock());
1794            }
1795            .boxed(),
1796        );
1797        futures.push(
1798            async {
1799                recv1.await.unwrap();
1800                // This should not block since it is a different key.
1801                let _t = fs
1802                    .root_store()
1803                    .new_transaction(
1804                        lock_keys![LockKey::object_attribute(2, 2, AttributeId::TEST_ID)],
1805                        Options::default(),
1806                    )
1807                    .await
1808                    .expect("new_transaction failed");
1809                // Tell the first future to continue.
1810                send2.send(()).unwrap();
1811            }
1812            .boxed(),
1813        );
1814        futures.push(
1815            async {
1816                // This should block until the first future has completed.
1817                recv3.await.unwrap();
1818                let _t = fs
1819                    .root_store()
1820                    .new_transaction(
1821                        lock_keys![LockKey::object_attribute(1, 2, AttributeId::TEST_ID)],
1822                        Options::default(),
1823                    )
1824                    .await;
1825                *done.lock() = true;
1826            }
1827            .boxed(),
1828        );
1829        while let Some(()) = futures.next().await {}
1830    }
1831
1832    #[fuchsia::test]
1833    async fn test_read_lock_after_write_lock() {
1834        let device = DeviceHolder::new(FakeDevice::new(4096, 1024));
1835        let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
1836        let (send1, recv1) = channel();
1837        let (send2, recv2) = channel();
1838        let done = Mutex::new(false);
1839        join!(
1840            async {
1841                let t = fs
1842                    .root_store()
1843                    .new_transaction(
1844                        lock_keys![LockKey::object_attribute(1, 2, AttributeId::TEST_ID)],
1845                        Options::default(),
1846                    )
1847                    .await
1848                    .expect("new_transaction failed");
1849                send1.send(()).unwrap(); // Tell the next future to continue.
1850                recv2.await.unwrap();
1851                t.commit().await.expect("commit failed");
1852                *done.lock() = true;
1853            },
1854            async {
1855                recv1.await.unwrap();
1856                // Reads should not be blocked until the transaction is committed.
1857                let _guard = fs
1858                    .lock_manager()
1859                    .read_lock(lock_keys![LockKey::object_attribute(1, 2, AttributeId::TEST_ID)])
1860                    .await;
1861                // Tell the first future to continue.
1862                send2.send(()).unwrap();
1863                // It shouldn't proceed until we release our read lock, but it's a halting
1864                // problem, so sleep.
1865                fasync::Timer::new(Duration::from_millis(100)).await;
1866                assert!(!*done.lock());
1867            },
1868        );
1869    }
1870
1871    #[fuchsia::test]
1872    async fn test_write_lock_after_read_lock() {
1873        let device = DeviceHolder::new(FakeDevice::new(4096, 1024));
1874        let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
1875        let (send1, recv1) = channel();
1876        let (send2, recv2) = channel();
1877        let done = Mutex::new(false);
1878        join!(
1879            async {
1880                // Reads should not be blocked until the transaction is committed.
1881                let _guard = fs
1882                    .lock_manager()
1883                    .read_lock(lock_keys![LockKey::object_attribute(1, 2, AttributeId::TEST_ID)])
1884                    .await;
1885                // Tell the next future to continue and then wait.
1886                send1.send(()).unwrap();
1887                recv2.await.unwrap();
1888                // It shouldn't proceed until we release our read lock, but it's a halting
1889                // problem, so sleep.
1890                fasync::Timer::new(Duration::from_millis(100)).await;
1891                assert!(!*done.lock());
1892            },
1893            async {
1894                recv1.await.unwrap();
1895                let t = fs
1896                    .root_store()
1897                    .new_transaction(
1898                        lock_keys![LockKey::object_attribute(1, 2, AttributeId::TEST_ID)],
1899                        Options::default(),
1900                    )
1901                    .await
1902                    .expect("new_transaction failed");
1903                send2.send(()).unwrap(); // Tell the first future to continue;
1904                t.commit().await.expect("commit failed");
1905                *done.lock() = true;
1906            },
1907        );
1908    }
1909
1910    #[fuchsia::test]
1911    async fn test_drop_uncommitted_transaction() {
1912        let device = DeviceHolder::new(FakeDevice::new(4096, 1024));
1913        let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
1914        let key = lock_keys![LockKey::object(1, 1)];
1915
1916        // Dropping while there's a reader.
1917        {
1918            let _write_lock = fs
1919                .root_store()
1920                .new_transaction(key.clone(), Options::default())
1921                .await
1922                .expect("new_transaction failed");
1923            let _read_lock = fs.lock_manager().read_lock(key.clone()).await;
1924        }
1925        // Dropping while there's no reader.
1926        {
1927            let _write_lock = fs
1928                .root_store()
1929                .new_transaction(key.clone(), Options::default())
1930                .await
1931                .expect("new_transaction failed");
1932        }
1933        // Make sure we can take the lock again (i.e. it was actually released).
1934        fs.root_store()
1935            .new_transaction(key.clone(), Options::default())
1936            .await
1937            .expect("new_transaction failed");
1938    }
1939
1940    #[fuchsia::test]
1941    async fn test_drop_waiting_write_lock() {
1942        let manager = LockManager::new();
1943        let keys = lock_keys![LockKey::object(1, 1)];
1944        {
1945            let _guard = manager.lock(keys.clone(), LockState::ReadLock).await;
1946            if let Poll::Ready(_) =
1947                futures::poll!(manager.lock(keys.clone(), LockState::WriteLock).boxed())
1948            {
1949                assert!(false);
1950            }
1951        }
1952        let _ = manager.lock(keys, LockState::WriteLock).await;
1953    }
1954
1955    #[fuchsia::test]
1956    async fn test_write_lock_blocks_everything() {
1957        let manager = LockManager::new();
1958        let keys = lock_keys![LockKey::object(1, 1)];
1959        {
1960            let _guard = manager.lock(keys.clone(), LockState::WriteLock).await;
1961            if let Poll::Ready(_) =
1962                futures::poll!(manager.lock(keys.clone(), LockState::WriteLock).boxed())
1963            {
1964                assert!(false);
1965            }
1966            if let Poll::Ready(_) =
1967                futures::poll!(manager.lock(keys.clone(), LockState::ReadLock).boxed())
1968            {
1969                assert!(false);
1970            }
1971        }
1972        {
1973            let _guard = manager.lock(keys.clone(), LockState::WriteLock).await;
1974        }
1975        {
1976            let _guard = manager.lock(keys, LockState::ReadLock).await;
1977        }
1978    }
1979
1980    #[fuchsia::test]
1981    async fn test_downgrade_locks() {
1982        let manager = LockManager::new();
1983        let keys = lock_keys![LockKey::object(1, 1)];
1984        let _guard = manager.txn_lock(keys.clone()).await;
1985        manager.commit_prepare_keys(&keys).await;
1986
1987        // Use FuturesUnordered so that we can check that the waker is woken.
1988        let mut read_lock: FuturesUnordered<_> =
1989            std::iter::once(manager.read_lock(keys.clone())).collect();
1990
1991        // Trying to acquire a read lock now should be blocked.
1992        assert!(futures::poll!(read_lock.next()).is_pending());
1993
1994        manager.downgrade_locks(&keys);
1995
1996        // After downgrading, it should be possible to take a read lock.
1997        assert!(futures::poll!(read_lock.next()).is_ready());
1998    }
1999
2000    #[fuchsia::test]
2001    async fn test_dropped_write_lock_wakes() {
2002        let manager = LockManager::new();
2003        let keys = lock_keys![LockKey::object(1, 1)];
2004        let _guard = manager.lock(keys.clone(), LockState::ReadLock).await;
2005        let mut read_lock = FuturesUnordered::new();
2006        read_lock.push(manager.lock(keys.clone(), LockState::ReadLock));
2007
2008        {
2009            let write_lock = manager.lock(keys, LockState::WriteLock);
2010            pin_mut!(write_lock);
2011
2012            // The write lock should be blocked because of the read lock.
2013            assert!(futures::poll!(write_lock).is_pending());
2014
2015            // Another read lock should be blocked because of the write lock.
2016            assert!(futures::poll!(read_lock.next()).is_pending());
2017        }
2018
2019        // Dropping the write lock should allow the read lock to proceed.
2020        assert!(futures::poll!(read_lock.next()).is_ready());
2021    }
2022
2023    #[fuchsia::test]
2024    async fn test_drop_upgrade() {
2025        let manager = LockManager::new();
2026        let keys = lock_keys![LockKey::object(1, 1)];
2027        let _guard = manager.lock(keys.clone(), LockState::Locked).await;
2028
2029        {
2030            let commit_prepare = manager.commit_prepare_keys(&keys);
2031            pin_mut!(commit_prepare);
2032            let _read_guard = manager.lock(keys.clone(), LockState::ReadLock).await;
2033            assert!(futures::poll!(commit_prepare).is_pending());
2034
2035            // Now we test dropping read_guard which should wake commit_prepare and
2036            // then dropping commit_prepare.
2037        }
2038
2039        // We should be able to still commit_prepare.
2040        manager.commit_prepare_keys(&keys).await;
2041    }
2042
2043    #[fasync::run_singlethreaded(test)]
2044    async fn test_woken_upgrade_blocks_reads() {
2045        let manager = LockManager::new();
2046        let keys = lock_keys![LockKey::object(1, 1)];
2047        // Start with a transaction lock.
2048        let guard = manager.lock(keys.clone(), LockState::Locked).await;
2049
2050        // Take a read lock.
2051        let read1 = manager.lock(keys.clone(), LockState::ReadLock).await;
2052
2053        // Try and upgrade the transaction lock, which should not be possible because of the read.
2054        let commit_prepare = manager.commit_prepare_keys(&keys);
2055        pin_mut!(commit_prepare);
2056        assert!(futures::poll!(commit_prepare.as_mut()).is_pending());
2057
2058        // Taking another read should also be blocked.
2059        let read2 = manager.lock(keys.clone(), LockState::ReadLock);
2060        pin_mut!(read2);
2061        assert!(futures::poll!(read2.as_mut()).is_pending());
2062
2063        // Drop the first read and the upgrade should complete.
2064        std::mem::drop(read1);
2065        assert!(futures::poll!(commit_prepare).is_ready());
2066
2067        // But the second read should still be blocked.
2068        assert!(futures::poll!(read2.as_mut()).is_pending());
2069
2070        // If we drop the write lock now, the read should be unblocked.
2071        std::mem::drop(guard);
2072        assert!(futures::poll!(read2).is_ready());
2073    }
2074
2075    static LOCK_KEY_1: LockKey = LockKey::flush(1);
2076    static LOCK_KEY_2: LockKey = LockKey::flush(2);
2077    static LOCK_KEY_3: LockKey = LockKey::flush(3);
2078
2079    // The keys, storage method, and capacity must all match.
2080    fn assert_lock_keys_equal(value: &LockKeys, expected: &LockKeys) {
2081        match (value, expected) {
2082            (LockKeys::None, LockKeys::None) => {}
2083            (LockKeys::Inline(key1), LockKeys::Inline(key2)) => {
2084                if key1 != key2 {
2085                    panic!("{key1:?} != {key2:?}");
2086                }
2087            }
2088            (LockKeys::Vec(vec1), LockKeys::Vec(vec2)) => {
2089                if vec1 != vec2 {
2090                    panic!("{vec1:?} != {vec2:?}");
2091                }
2092                if vec1.capacity() != vec2.capacity() {
2093                    panic!(
2094                        "LockKeys have different capacity: {} != {}",
2095                        vec1.capacity(),
2096                        vec2.capacity()
2097                    );
2098                }
2099            }
2100            (_, _) => panic!("{value:?} != {expected:?}"),
2101        }
2102    }
2103
2104    // Only the keys must match. Storage method and capacity don't matter.
2105    fn assert_lock_keys_equivalent(value: &LockKeys, expected: &LockKeys) {
2106        let value: Vec<_> = value.iter().collect();
2107        let expected: Vec<_> = expected.iter().collect();
2108        assert_eq!(value, expected);
2109    }
2110
2111    #[test]
2112    fn test_lock_keys_macro() {
2113        assert_lock_keys_equal(&lock_keys![], &LockKeys::None);
2114        assert_lock_keys_equal(&lock_keys![LOCK_KEY_1], &LockKeys::Inline(LOCK_KEY_1));
2115        assert_lock_keys_equal(
2116            &lock_keys![LOCK_KEY_1, LOCK_KEY_2],
2117            &LockKeys::Vec(vec![LOCK_KEY_1, LOCK_KEY_2]),
2118        );
2119    }
2120
2121    #[test]
2122    fn test_lock_keys_with_capacity() {
2123        assert_lock_keys_equal(&LockKeys::with_capacity(0), &LockKeys::None);
2124        assert_lock_keys_equal(&LockKeys::with_capacity(1), &LockKeys::None);
2125        assert_lock_keys_equal(&LockKeys::with_capacity(2), &LockKeys::Vec(Vec::with_capacity(2)));
2126    }
2127
2128    #[test]
2129    fn test_lock_keys_len() {
2130        assert_eq!(lock_keys![].len(), 0);
2131        assert_eq!(lock_keys![LOCK_KEY_1].len(), 1);
2132        assert_eq!(lock_keys![LOCK_KEY_1, LOCK_KEY_2].len(), 2);
2133    }
2134
2135    #[test]
2136    fn test_lock_keys_contains() {
2137        assert_eq!(lock_keys![].contains(&LOCK_KEY_1), false);
2138        assert_eq!(lock_keys![LOCK_KEY_1].contains(&LOCK_KEY_1), true);
2139        assert_eq!(lock_keys![LOCK_KEY_1].contains(&LOCK_KEY_2), false);
2140        assert_eq!(lock_keys![LOCK_KEY_1, LOCK_KEY_2].contains(&LOCK_KEY_1), true);
2141        assert_eq!(lock_keys![LOCK_KEY_1, LOCK_KEY_2].contains(&LOCK_KEY_2), true);
2142        assert_eq!(lock_keys![LOCK_KEY_1, LOCK_KEY_2].contains(&LOCK_KEY_3), false);
2143    }
2144
2145    #[test]
2146    fn test_lock_keys_push() {
2147        let mut keys = lock_keys![];
2148        keys.push(LOCK_KEY_1);
2149        assert_lock_keys_equal(&keys, &LockKeys::Inline(LOCK_KEY_1));
2150        keys.push(LOCK_KEY_2);
2151        assert_lock_keys_equal(&keys, &LockKeys::Vec(vec![LOCK_KEY_1, LOCK_KEY_2]));
2152        keys.push(LOCK_KEY_3);
2153        assert_lock_keys_equivalent(
2154            &keys,
2155            &LockKeys::Vec(vec![LOCK_KEY_1, LOCK_KEY_2, LOCK_KEY_3]),
2156        );
2157    }
2158
2159    #[test]
2160    fn test_lock_keys_sort_unstable() {
2161        let mut keys = lock_keys![];
2162        keys.sort_unstable();
2163        assert_lock_keys_equal(&keys, &lock_keys![]);
2164
2165        let mut keys = lock_keys![LOCK_KEY_1];
2166        keys.sort_unstable();
2167        assert_lock_keys_equal(&keys, &lock_keys![LOCK_KEY_1]);
2168
2169        let mut keys = lock_keys![LOCK_KEY_2, LOCK_KEY_1];
2170        keys.sort_unstable();
2171        assert_lock_keys_equal(&keys, &lock_keys![LOCK_KEY_1, LOCK_KEY_2]);
2172    }
2173
2174    #[test]
2175    fn test_lock_keys_dedup() {
2176        let mut keys = lock_keys![];
2177        keys.dedup();
2178        assert_lock_keys_equal(&keys, &lock_keys![]);
2179
2180        let mut keys = lock_keys![LOCK_KEY_1];
2181        keys.dedup();
2182        assert_lock_keys_equal(&keys, &lock_keys![LOCK_KEY_1]);
2183
2184        let mut keys = lock_keys![LOCK_KEY_1, LOCK_KEY_1];
2185        keys.dedup();
2186        assert_lock_keys_equivalent(&keys, &lock_keys![LOCK_KEY_1]);
2187    }
2188
2189    #[test]
2190    fn test_lock_keys_truncate() {
2191        let mut keys = lock_keys![];
2192        keys.truncate(5);
2193        assert_lock_keys_equal(&keys, &lock_keys![]);
2194        keys.truncate(0);
2195        assert_lock_keys_equal(&keys, &lock_keys![]);
2196
2197        let mut keys = lock_keys![LOCK_KEY_1];
2198        keys.truncate(5);
2199        assert_lock_keys_equal(&keys, &lock_keys![LOCK_KEY_1]);
2200        keys.truncate(0);
2201        assert_lock_keys_equal(&keys, &lock_keys![]);
2202
2203        let mut keys = lock_keys![LOCK_KEY_1, LOCK_KEY_2];
2204        keys.truncate(5);
2205        assert_lock_keys_equal(&keys, &lock_keys![LOCK_KEY_1, LOCK_KEY_2]);
2206        keys.truncate(1);
2207        // Although there's only 1 key after truncate the key is not stored inline.
2208        assert_lock_keys_equivalent(&keys, &lock_keys![LOCK_KEY_1]);
2209    }
2210
2211    #[test]
2212    fn test_lock_keys_iter() {
2213        assert_eq!(lock_keys![].iter().collect::<Vec<_>>(), Vec::<&LockKey>::new());
2214
2215        assert_eq!(lock_keys![LOCK_KEY_1].iter().collect::<Vec<_>>(), vec![&LOCK_KEY_1]);
2216
2217        assert_eq!(
2218            lock_keys![LOCK_KEY_1, LOCK_KEY_2].iter().collect::<Vec<_>>(),
2219            vec![&LOCK_KEY_1, &LOCK_KEY_2]
2220        );
2221    }
2222}