1use 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#[derive(Clone, Copy, Default)]
42pub struct Options<'a> {
43 pub skip_journal_checks: bool,
46
47 pub skip_key_roll: bool,
49
50 pub borrow_metadata_space: bool,
55
56 pub allocator_reservation: Option<&'a Reservation>,
61}
62
63pub 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
76pub 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 BeginFlush,
92 EndFlush,
95 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
255pub 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
339pub 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
382pub 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 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#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Copy)]
509pub enum LockKey {
510 Flush {
512 object_id: u64,
513 },
514
515 ObjectAttribute {
517 store_object_id: u64,
518 object_id: u64,
519 attribute_id: AttributeId,
520 },
521
522 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 Truncate {
535 store_object_id: u64,
536 object_id: u64,
537 },
538
539 InternalDirectory {
541 store_object_id: u64,
542 },
543
544 MutationsKeyRoll {
546 store_object_id: u64,
547 },
548
549 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#[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
702pub 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 pub object_id: u64,
731
732 pub mutation: Mutation,
734
735 pub associated_object: AssocObj<'a>,
738}
739
740impl 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 None,
779
780 Borrowed,
783
784 Reservation(Reservation),
786
787 Hold(u64),
789}
790
791pub struct Transaction<'a> {
793 fs: Arc<FxFilesystem>,
794
795 mutations: BTreeSet<TxnMutation<'a>>,
797
798 txn_locks: LockKeys,
800
801 pub allocator_reservation: Option<&'a Reservation>,
803
804 pub metadata_reservation: MetadataReservation,
806
807 new_objects: BTreeSet<(u64, u64)>,
810
811 checksums: Vec<(Range<u64>, Vec<Checksum>, bool)>,
813
814 includes_write: bool,
816}
817
818impl<'a> Transaction<'a> {
819 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.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 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 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 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 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 }
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 }
969 ObjectKeyData::GraveyardAttributeEntry { .. } => {
970 }
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 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 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 pub fn is_empty(&self) -> bool {
1066 self.mutations.is_empty()
1067 }
1068
1069 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 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 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 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 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
1163pub 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 LockState::Locked | LockState::WriteLock => {
1210 entry.state = LockState::ReadLock;
1211 true
1212 }
1213 };
1214 if wake {
1215 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 self.drop_lock(*lock, LockState::WriteLock);
1239 }
1240 }
1241
1242 fn downgrade_locks(&mut self, lock_keys: &LockKeys) {
1244 for lock in lock_keys.iter() {
1245 unsafe {
1247 self.keys.get_mut(lock).unwrap().downgrade_lock();
1248 }
1249 }
1250 }
1251}
1252
1253#[derive(Debug)]
1254struct LockEntry {
1255 read_count: u64,
1258
1259 state: LockState,
1261
1262 head: *const LockWaker,
1267 tail: *const LockWaker,
1268}
1269
1270unsafe impl Send for LockEntry {}
1271
1272struct LockWaker {
1275 next: UnsafeCell<*const LockWaker>,
1277 prev: UnsafeCell<*const LockWaker>,
1278
1279 key: LockKey,
1282
1283 waker: UnsafeCell<WakerState>,
1285
1286 target_state: LockState,
1288
1289 is_upgrade: bool,
1291
1292 _pin: PhantomPinned,
1294}
1295
1296enum WakerState {
1297 Pending,
1299
1300 Registered(Waker),
1302
1303 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 async fn wait(&self, manager: &LockManager) {
1319 let waker_guard = scopeguard::guard((), |_| {
1321 let mut locks = manager.locks.lock();
1322 unsafe {
1324 if (*self.waker.get()).is_woken() {
1325 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 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 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 ReadLock,
1361
1362 Locked,
1365
1366 WriteLock,
1368}
1369
1370impl LockManager {
1371 pub fn new() -> Self {
1372 LockManager { locks: Mutex::new(Locks { keys: HashMap::default() }) }
1373 }
1374
1375 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 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 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 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 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 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 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 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 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 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 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 pub fn downgrade_locks(&self, lock_keys: &LockKeys) {
1547 self.locks.lock().downgrade_locks(lock_keys);
1548 }
1549}
1550
1551impl LockEntry {
1553 unsafe fn wake(&mut self) {
1554 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 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 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 if waker.target_state == LockState::ReadLock {
1595 self.read_count += 1;
1596 } else {
1597 self.state = waker.target_state;
1598 }
1599
1600 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 self.wake();
1629 }
1630 }
1631 }
1632
1633 unsafe fn is_allowed(&self, target_state: LockState, is_head: bool) -> bool {
1637 match self.state {
1638 LockState::ReadLock => {
1639 (self.read_count == 0
1641 || target_state == LockState::Locked
1642 || target_state == LockState::ReadLock)
1643 && is_head
1644 }
1645 LockState::Locked => {
1646 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(); send3.send(()).unwrap(); recv2.await.unwrap();
1791 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 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 send2.send(()).unwrap();
1811 }
1812 .boxed(),
1813 );
1814 futures.push(
1815 async {
1816 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(); recv2.await.unwrap();
1851 t.commit().await.expect("commit failed");
1852 *done.lock() = true;
1853 },
1854 async {
1855 recv1.await.unwrap();
1856 let _guard = fs
1858 .lock_manager()
1859 .read_lock(lock_keys![LockKey::object_attribute(1, 2, AttributeId::TEST_ID)])
1860 .await;
1861 send2.send(()).unwrap();
1863 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 let _guard = fs
1882 .lock_manager()
1883 .read_lock(lock_keys![LockKey::object_attribute(1, 2, AttributeId::TEST_ID)])
1884 .await;
1885 send1.send(()).unwrap();
1887 recv2.await.unwrap();
1888 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(); 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 {
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 {
1927 let _write_lock = fs
1928 .root_store()
1929 .new_transaction(key.clone(), Options::default())
1930 .await
1931 .expect("new_transaction failed");
1932 }
1933 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 let mut read_lock: FuturesUnordered<_> =
1989 std::iter::once(manager.read_lock(keys.clone())).collect();
1990
1991 assert!(futures::poll!(read_lock.next()).is_pending());
1993
1994 manager.downgrade_locks(&keys);
1995
1996 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 assert!(futures::poll!(write_lock).is_pending());
2014
2015 assert!(futures::poll!(read_lock.next()).is_pending());
2017 }
2018
2019 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 }
2038
2039 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 let guard = manager.lock(keys.clone(), LockState::Locked).await;
2049
2050 let read1 = manager.lock(keys.clone(), LockState::ReadLock).await;
2052
2053 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 let read2 = manager.lock(keys.clone(), LockState::ReadLock);
2060 pin_mut!(read2);
2061 assert!(futures::poll!(read2.as_mut()).is_pending());
2062
2063 std::mem::drop(read1);
2065 assert!(futures::poll!(commit_prepare).is_ready());
2066
2067 assert!(futures::poll!(read2.as_mut()).is_pending());
2069
2070 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 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 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 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}