1use crate::errors::FxfsError;
6use crate::filesystem::{ApplyContext, ApplyMode, FlushReason, JournalingObject};
7use crate::log::*;
8use crate::metrics;
9use crate::object_handle::INVALID_OBJECT_ID;
10use crate::object_store::allocator::{Allocator, Reservation};
11use crate::object_store::directory::Directory;
12use crate::object_store::journal::{self, JournalCheckpoint};
13use crate::object_store::transaction::{
14 AssocObj, AssociatedObject, MetadataReservation, Mutation, ObjectMutationIterator, Transaction,
15 TxnMutation,
16};
17use crate::object_store::tree_cache::TreeCache;
18use crate::object_store::volume::{VOLUMES_DIRECTORY, list_volumes};
19use crate::object_store::{ObjectDescriptor, ObjectStore};
20use crate::round::round_div;
21use crate::serialized_types::{LATEST_VERSION, Version};
22use anyhow::{Context, Error, anyhow, bail, ensure};
23use fuchsia_inspect::{Property as _, UintProperty};
24use fuchsia_sync::RwLock;
25use futures::FutureExt as _;
26use rustc_hash::FxHashMap as HashMap;
27use std::collections::hash_map::Entry;
28use std::num::Saturating;
29use std::sync::{Arc, OnceLock};
30
31pub const fn reserved_space_from_journal_usage(journal_usage: u64) -> u64 {
41 journal_usage * 4
42}
43
44pub struct ObjectManager {
46 inner: RwLock<Inner>,
47 metadata_reservation: OnceLock<Reservation>,
48 volume_directory: OnceLock<Directory<ObjectStore>>,
49 on_new_store: Option<Box<dyn Fn(&ObjectStore) + Send + Sync>>,
50}
51
52#[derive(Debug)]
55enum Checkpoints {
56 Current(JournalCheckpoint),
57 Old(JournalCheckpoint),
58 Both(JournalCheckpoint, JournalCheckpoint),
59}
60
61impl Checkpoints {
62 fn earliest(&self) -> &JournalCheckpoint {
64 match self {
65 Checkpoints::Old(x) | Checkpoints::Both(x, _) | Checkpoints::Current(x) => x,
66 }
67 }
68}
69
70struct Inner {
73 stores: HashMap<u64, Arc<ObjectStore>>,
74 root_parent_store_object_id: u64,
75 root_store_object_id: u64,
76 allocator_object_id: u64,
77 allocator: Option<Arc<Allocator>>,
78
79 journal_checkpoints: HashMap<u64, Checkpoints>,
82
83 reservations: HashMap<u64, u64>,
89
90 last_end_offset: u64,
93
94 borrowed_metadata_space: u64,
97
98 max_transaction_size: (u64, UintProperty),
100
101 reserved_space: u64,
103}
104
105impl Inner {
106 fn earliest_journal_offset(&self) -> Option<u64> {
107 self.journal_checkpoints.values().map(|c| c.earliest().file_offset).min()
108 }
109
110 fn required_reservation(&self) -> u64 {
113 self.reservations.values().max().unwrap_or(&0)
115
116 + self.earliest_journal_offset()
119 .map(|min| reserved_space_from_journal_usage(self.last_end_offset - min))
120 .unwrap_or(0)
121
122 + self.reserved_space
124 }
125
126 fn journaling_object(&self, object_id: u64) -> Option<Arc<dyn JournalingObject>> {
127 if object_id == self.allocator_object_id {
128 Some(self.allocator.clone().unwrap() as Arc<dyn JournalingObject>)
129 } else {
130 self.stores.get(&object_id).map(|x| x.clone() as Arc<dyn JournalingObject>)
131 }
132 }
133}
134
135impl ObjectManager {
136 pub fn new(on_new_store: Option<Box<dyn Fn(&ObjectStore) + Send + Sync>>) -> ObjectManager {
137 ObjectManager {
138 inner: RwLock::new(Inner {
139 stores: HashMap::default(),
140 root_parent_store_object_id: INVALID_OBJECT_ID,
141 root_store_object_id: INVALID_OBJECT_ID,
142 allocator_object_id: INVALID_OBJECT_ID,
143 allocator: None,
144 journal_checkpoints: HashMap::default(),
145 reservations: HashMap::default(),
146 last_end_offset: 0,
147 borrowed_metadata_space: 0,
148 max_transaction_size: (0, metrics::detail().create_uint("max_transaction_size", 0)),
149 reserved_space: journal::RESERVED_SPACE,
150 }),
151 metadata_reservation: OnceLock::new(),
152 volume_directory: OnceLock::new(),
153 on_new_store,
154 }
155 }
156
157 pub fn required_reservation(&self) -> u64 {
158 self.inner.read().required_reservation()
159 }
160
161 pub fn root_parent_store_object_id(&self) -> u64 {
162 self.inner.read().root_parent_store_object_id
163 }
164
165 pub fn root_parent_store(&self) -> Arc<ObjectStore> {
166 let inner = self.inner.read();
167 inner.stores.get(&inner.root_parent_store_object_id).unwrap().clone()
168 }
169
170 pub fn set_root_parent_store(&self, store: Arc<ObjectStore>) {
171 if let Some(on_new_store) = &self.on_new_store {
172 on_new_store(&store);
173 }
174 let mut inner = self.inner.write();
175 let store_id = store.store_object_id();
176 inner.stores.insert(store_id, store);
177 inner.root_parent_store_object_id = store_id;
178 }
179
180 pub fn root_store_object_id(&self) -> u64 {
181 self.inner.read().root_store_object_id
182 }
183
184 pub fn root_store(&self) -> Arc<ObjectStore> {
185 let inner = self.inner.read();
186 inner.stores.get(&inner.root_store_object_id).unwrap().clone()
187 }
188
189 pub fn set_root_store(&self, store: Arc<ObjectStore>) {
190 if let Some(on_new_store) = &self.on_new_store {
191 on_new_store(&store);
192 }
193 let mut inner = self.inner.write();
194 let store_id = store.store_object_id();
195 inner.stores.insert(store_id, store);
196 inner.root_store_object_id = store_id;
197 }
198
199 pub fn is_system_store(&self, store_id: u64) -> bool {
200 let inner = self.inner.read();
201 store_id == inner.root_store_object_id || store_id == inner.root_parent_store_object_id
202 }
203
204 pub fn store(&self, store_object_id: u64) -> Option<Arc<ObjectStore>> {
206 self.inner.read().stores.get(&store_object_id).cloned()
207 }
208
209 pub fn compaction_bytes_written(&self) -> u64 {
212 let inner = self.inner.read();
213 let mut total = 0;
214 if let Some(allocator) = &inner.allocator {
215 total += allocator.tree().compaction_bytes_written();
216 }
217 for store in inner.stores.values() {
218 total += store.tree().compaction_bytes_written();
219 }
220 total
221 }
222
223 pub async fn on_replay_complete(&self) -> Result<(), Error> {
226 let root_store = self.root_store();
227
228 let root_directory = Directory::open(&root_store, root_store.root_directory_object_id())
229 .await
230 .context("Unable to open root volume directory")?;
231
232 match root_directory.lookup(VOLUMES_DIRECTORY).await? {
233 None => bail!("Root directory not found"),
234 Some((object_id, ObjectDescriptor::Directory, _)) => {
235 let volume_directory = Directory::open(&root_store, object_id)
236 .await
237 .context("Unable to open volumes directory")?;
238 self.volume_directory.set(volume_directory).unwrap();
239 }
240 _ => {
241 bail!(
242 anyhow!(FxfsError::Inconsistent)
243 .context("Unexpected type for volumes directory")
244 )
245 }
246 }
247
248 let object_ids = list_volumes(self.volume_directory.get().unwrap())
249 .await
250 .context("Failed to list volumes")?;
251
252 for store_id in object_ids {
253 self.open_store(&root_store, store_id).await?;
254 }
255
256 self.init_metadata_reservation()
259 .context("Insufficient free space for metadata reservation.")?;
260
261 Ok(())
262 }
263
264 pub fn volume_directory(&self) -> &Directory<ObjectStore> {
265 self.volume_directory.get().unwrap()
266 }
267
268 pub fn set_volume_directory(&self, volume_directory: Directory<ObjectStore>) {
269 self.volume_directory.set(volume_directory).unwrap();
270 }
271
272 pub fn add_store(&self, store: Arc<ObjectStore>) {
273 if let Some(on_new_store) = &self.on_new_store {
274 on_new_store(&store);
275 }
276 let mut inner = self.inner.write();
277 let store_object_id = store.store_object_id();
278 assert_ne!(store_object_id, inner.root_parent_store_object_id);
279 assert_ne!(store_object_id, inner.root_store_object_id);
280 assert_ne!(store_object_id, inner.allocator_object_id);
281 inner.stores.insert(store_object_id, store);
282 }
283
284 pub fn forget_store(&self, store_object_id: u64) {
285 let mut inner = self.inner.write();
286 assert_ne!(store_object_id, inner.allocator_object_id);
287 inner.stores.remove(&store_object_id);
288 inner.reservations.remove(&store_object_id);
289 }
290
291 pub fn set_allocator(&self, allocator: Arc<Allocator>) {
292 let mut inner = self.inner.write();
293 assert!(!inner.stores.contains_key(&allocator.object_id()));
294 inner.allocator_object_id = allocator.object_id();
295 inner.allocator = Some(allocator);
296 }
297
298 pub fn allocator(&self) -> Arc<Allocator> {
299 self.inner.read().allocator.clone().unwrap()
300 }
301
302 pub fn apply_mutation(
304 &self,
305 object_id: u64,
306 mutation: Mutation,
307 context: &ApplyContext<'_, '_>,
308 associated_object: AssocObj<'_>,
309 ) -> Result<(), Error> {
310 debug!(oid = object_id, mutation:?; "applying mutation");
311 let object = {
312 let mut inner = self.inner.write();
313 match mutation {
314 Mutation::BeginFlush => {
315 if let Some(entry) = inner.journal_checkpoints.get_mut(&object_id) {
316 match entry {
317 Checkpoints::Current(x) | Checkpoints::Both(x, _) => {
318 *entry = Checkpoints::Old(x.clone());
319 }
320 _ => {}
321 }
322 }
323 }
324 Mutation::EndFlush => {
325 if let Entry::Occupied(mut o) = inner.journal_checkpoints.entry(object_id) {
326 let entry = o.get_mut();
327 match entry {
328 Checkpoints::Old(_) => {
329 o.remove();
330 }
331 Checkpoints::Both(_, x) => {
332 *entry = Checkpoints::Current(x.clone());
333 }
334 _ => {}
335 }
336 }
337 }
338 Mutation::DeleteVolume => {
339 if let Some(store) = inner.stores.remove(&object_id) {
340 store.mark_deleted();
341 }
342 inner.reservations.remove(&object_id);
343 inner.journal_checkpoints.remove(&object_id);
344 return Ok(());
345 }
346 _ => {
347 if object_id != inner.root_parent_store_object_id {
348 inner
349 .journal_checkpoints
350 .entry(object_id)
351 .and_modify(|entry| {
352 if let Checkpoints::Old(x) = entry {
353 *entry =
354 Checkpoints::Both(x.clone(), context.checkpoint.clone());
355 }
356 })
357 .or_insert_with(|| Checkpoints::Current(context.checkpoint.clone()));
358 }
359 }
360 }
361 inner.journaling_object(object_id).unwrap()
362 };
363 associated_object.map(|o| o.will_apply_mutation(&mutation, object_id, self));
364 object.apply_mutation(mutation, context, associated_object)
365 }
366
367 pub async fn replay_mutations(
372 &self,
373 mutations: Vec<(u64, Mutation)>,
374 journal_offsets: &HashMap<u64, u64>,
375 context: &ApplyContext<'_, '_>,
376 end_offset: u64,
377 ) -> Result<(), Error> {
378 debug!(checkpoint = context.checkpoint.file_offset; "REPLAY");
379 let txn_size = {
380 let mut inner = self.inner.write();
381 if end_offset > inner.last_end_offset {
382 Some(end_offset - std::mem::replace(&mut inner.last_end_offset, end_offset))
383 } else {
384 None
385 }
386 };
387
388 let allocator_object_id = self.inner.read().allocator_object_id;
389
390 for (object_id, mutation) in mutations {
391 if let Mutation::UpdateBorrowed(borrowed) = mutation {
392 if let Some(txn_size) = txn_size {
393 self.inner.write().borrowed_metadata_space = borrowed
394 .checked_add(reserved_space_from_journal_usage(txn_size))
395 .ok_or(FxfsError::Inconsistent)?;
396 }
397 continue;
398 }
399
400 if let Some(&offset) = journal_offsets.get(&object_id) {
402 if context.checkpoint.file_offset < offset {
403 continue;
404 }
405 }
406
407 if object_id != allocator_object_id {
409 self.open_store(&self.root_store(), object_id).await?;
410 }
411
412 self.apply_mutation(object_id, mutation, context, AssocObj::None)?;
413 }
414 Ok(())
415 }
416
417 async fn open_store(&self, parent: &Arc<ObjectStore>, object_id: u64) -> Result<(), Error> {
419 if self.inner.read().stores.contains_key(&object_id) {
420 return Ok(());
421 }
422 let store = ObjectStore::open(parent, object_id, Some(Box::new(TreeCache::new())))
423 .await
424 .with_context(|| format!("Failed to open store {object_id}"))?;
425 if let Some(on_new_store) = &self.on_new_store {
426 on_new_store(&store);
427 }
428 assert!(self.inner.write().stores.insert(object_id, store).is_none());
429 Ok(())
430 }
431
432 pub fn apply_transaction(
436 &self,
437 transaction: &mut Transaction<'_>,
438 checkpoint: &JournalCheckpoint,
439 ) -> Result<Option<Mutation>, Error> {
440 let old_amount = self.metadata_reservation().amount();
442 let old_required = self.inner.read().required_reservation();
443
444 debug!(checkpoint = checkpoint.file_offset; "BEGIN TXN");
445 let mutations = transaction.take_mutations();
446 let context =
447 ApplyContext { mode: ApplyMode::Live(transaction), checkpoint: checkpoint.clone() };
448 for TxnMutation { object_id, mutation, associated_object, .. } in mutations {
449 self.apply_mutation(object_id, mutation, &context, associated_object)?;
450 }
451 debug!("END TXN");
452
453 Ok(if let MetadataReservation::Borrowed = transaction.metadata_reservation {
454 let new_amount = self.metadata_reservation().amount();
459 let mut inner = self.inner.write();
460 let new_required = inner.required_reservation();
461 let add = old_amount + new_required;
462 let sub = new_amount + old_required;
463 if add >= sub {
464 inner.borrowed_metadata_space += add - sub;
465 } else {
466 inner.borrowed_metadata_space =
467 inner.borrowed_metadata_space.saturating_sub(sub - add);
468 }
469 Some(Mutation::UpdateBorrowed(inner.borrowed_metadata_space))
470 } else {
471 debug_assert_eq!(self.metadata_reservation().amount(), old_amount);
474 debug_assert_eq!(self.inner.read().required_reservation(), old_required);
475 None
476 })
477 }
478
479 pub fn did_commit_transaction(
482 &self,
483 transaction: &mut Transaction<'_>,
484 _checkpoint: &JournalCheckpoint,
485 end_offset: u64,
486 ) {
487 let reservation = self.metadata_reservation();
488 let mut inner = self.inner.write();
489 let journal_usage = end_offset - std::mem::replace(&mut inner.last_end_offset, end_offset);
490
491 if journal_usage > inner.max_transaction_size.0 {
492 inner.max_transaction_size.0 = journal_usage;
493 inner.max_transaction_size.1.set(journal_usage);
494 }
495
496 let txn_space = reserved_space_from_journal_usage(journal_usage);
497 match &mut transaction.metadata_reservation {
498 MetadataReservation::None => unreachable!(),
499 MetadataReservation::Borrowed => {
500 inner.borrowed_metadata_space += txn_space;
503
504 let to_give_back = (reservation.amount() + inner.borrowed_metadata_space)
507 .saturating_sub(inner.required_reservation());
508 if to_give_back > 0 {
509 reservation.give_back(to_give_back);
510 }
511 }
512 MetadataReservation::Hold(hold_amount) => {
513 let txn_reservation = transaction.allocator_reservation.unwrap();
515 assert_ne!(
516 txn_reservation as *const _, reservation as *const _,
517 "MetadataReservation::Borrowed should be used."
518 );
519 txn_reservation.commit(txn_space);
520 if txn_reservation.owner_object_id() != reservation.owner_object_id() {
521 assert_eq!(
522 reservation.owner_object_id(),
523 None,
524 "Should not be mixing attributed owners."
525 );
526 inner
527 .allocator
528 .as_ref()
529 .unwrap()
530 .disown_reservation(txn_reservation.owner_object_id(), txn_space);
531 }
532 if let Some(amount) = hold_amount.checked_sub(txn_space) {
533 *hold_amount = amount;
534 } else {
535 panic!("Transaction was larger than metadata reservation");
536 }
537 reservation.add(txn_space);
538 }
539 MetadataReservation::Reservation(txn_reservation) => {
540 txn_reservation.move_to(reservation, txn_space);
542 }
543 }
544 debug_assert_eq!(
546 reservation.amount() + inner.borrowed_metadata_space,
547 inner.required_reservation(),
548 "txn_space: {}, reservation_amount: {}, borrowed: {}, required: {}",
549 txn_space,
550 reservation.amount(),
551 inner.borrowed_metadata_space,
552 inner.required_reservation(),
553 );
554 }
555
556 pub fn drop_transaction(&self, transaction: &mut Transaction<'_>) {
561 for TxnMutation { object_id, mutation, .. } in transaction.take_mutations() {
562 self.journaling_object(object_id).map(|o| o.drop_mutation(mutation, transaction));
563 }
564 }
565
566 pub fn journal_file_offsets(&self) -> (HashMap<u64, u64>, Option<JournalCheckpoint>) {
569 let inner = self.inner.read();
570 let mut min_checkpoint = None;
571 let mut offsets = HashMap::default();
572 for (&object_id, checkpoint) in &inner.journal_checkpoints {
573 let checkpoint = checkpoint.earliest();
574 match &mut min_checkpoint {
575 None => min_checkpoint = Some(checkpoint),
576 Some(min_checkpoint) => {
577 if checkpoint.file_offset < min_checkpoint.file_offset {
578 *min_checkpoint = checkpoint;
579 }
580 }
581 }
582 offsets.insert(object_id, checkpoint.file_offset);
583 }
584 (offsets, min_checkpoint.cloned())
585 }
586
587 pub fn journal_checkpoint(&self, object_id: u64) -> Option<JournalCheckpoint> {
590 self.inner
591 .read()
592 .journal_checkpoints
593 .get(&object_id)
594 .map(|checkpoints| checkpoints.earliest().clone())
595 }
596
597 pub fn needs_flush(&self, object_id: u64) -> bool {
600 self.inner.read().journal_checkpoints.contains_key(&object_id)
601 }
602
603 pub async fn flush(&self, reason: FlushReason) -> Result<Version, Error> {
606 let objects = {
607 let inner = self.inner.read();
608 let mut object_ids = inner.journal_checkpoints.keys().cloned().collect::<Vec<_>>();
609 object_ids.sort_unstable();
613 object_ids
614 .iter()
615 .rev()
616 .filter_map(|oid| inner.journaling_object(*oid).map(|obj| (*oid, obj)))
617 .collect::<Vec<_>>()
618 };
619
620 let mut earliest_version: Version = LATEST_VERSION;
622 for (object_id, object) in objects {
623 let object_earliest_version = object
624 .flush(reason)
625 .await
626 .with_context(|| format!("Failed to flush oid {object_id}"))?;
627 if object_earliest_version < earliest_version {
628 earliest_version = object_earliest_version;
629 }
630 }
631
632 Ok(earliest_version)
633 }
634
635 pub fn journaling_object(&self, object_id: u64) -> Option<Arc<dyn JournalingObject>> {
636 self.inner.read().journaling_object(object_id)
637 }
638
639 pub fn init_metadata_reservation(&self) -> Result<(), Error> {
640 if self.root_parent_store().filesystem().options().read_only {
641 return Ok(());
643 }
644 let inner = self.inner.read();
645 let required = inner.required_reservation();
646 ensure!(required >= inner.borrowed_metadata_space, FxfsError::Inconsistent);
647 let allocator = inner.allocator.as_ref().cloned().unwrap();
648 self.metadata_reservation
649 .set(
650 allocator
651 .clone()
652 .reserve(None, inner.required_reservation() - inner.borrowed_metadata_space)
653 .with_context(|| {
654 format!(
655 "Failed to reserve {} - {} = {} bytes, free={}, \
656 owner_bytes={}",
657 inner.required_reservation(),
658 inner.borrowed_metadata_space,
659 inner.required_reservation() - inner.borrowed_metadata_space,
660 Saturating(allocator.get_disk_bytes()) - allocator.get_used_bytes(),
661 allocator.owner_bytes_debug(),
662 )
663 })?,
664 )
665 .unwrap();
666 Ok(())
667 }
668
669 pub fn metadata_reservation(&self) -> &Reservation {
670 self.metadata_reservation.get().unwrap()
671 }
672
673 pub fn is_metadata_reservation(&self, reservation: &Reservation) -> bool {
674 self.metadata_reservation.get().is_some_and(|r| std::ptr::eq(r, reservation))
675 }
676
677 pub fn update_reservation(&self, object_id: u64, amount: u64) {
678 self.inner.write().reservations.insert(object_id, amount);
679 }
680
681 pub fn reservation(&self, object_id: u64) -> Option<u64> {
682 self.inner.read().reservations.get(&object_id).cloned()
683 }
684
685 pub fn max_store_reservation(&self) -> u64 {
687 self.inner.read().reservations.values().max().cloned().unwrap_or(0)
688 }
689
690 pub fn set_reserved_space(&self, amount: u64) {
691 self.inner.write().reserved_space = amount;
692 }
693
694 pub fn last_end_offset(&self) -> u64 {
695 self.inner.read().last_end_offset
696 }
697
698 pub fn set_last_end_offset(&self, v: u64) {
699 self.inner.write().last_end_offset = v;
700 }
701
702 pub fn borrowed_metadata_space(&self) -> u64 {
703 self.inner.read().borrowed_metadata_space
704 }
705
706 pub fn set_borrowed_metadata_space(&self, v: u64) {
707 self.inner.write().borrowed_metadata_space = v;
708 }
709
710 pub fn write_mutations(
711 &self,
712 object_id: u64,
713 mutations: ObjectMutationIterator<'_, '_>,
714 writer: journal::Writer<'_>,
715 ) {
716 self.journaling_object(object_id).unwrap().write_mutations(mutations, writer);
717 }
718
719 pub fn unlocked_stores(&self) -> Vec<Arc<ObjectStore>> {
720 let inner = self.inner.read();
721 let mut stores = Vec::new();
722 for store in inner.stores.values() {
723 if !store.is_locked() {
724 stores.push(store.clone());
725 }
726 }
727 stores
728 }
729
730 pub fn track_statistics(self: &Arc<Self>, parent: &fuchsia_inspect::Node, name: &str) {
733 let this = Arc::downgrade(self);
734 parent.record_lazy_child(name, move || {
735 let this_clone = this.clone();
736 async move {
737 let inspector = fuchsia_inspect::Inspector::default();
738 if let Some(this) = this_clone.upgrade() {
739 let (required, borrowed, earliest_checkpoint) = {
740 let inner = this.inner.read();
742 (
743 inner.required_reservation(),
744 inner.borrowed_metadata_space,
745 inner.earliest_journal_offset(),
746 )
747 };
748 let root = inspector.root();
749 if let Some(reservation) = this.metadata_reservation.get() {
750 root.record_uint("metadata_reservation", reservation.amount());
751 }
752 root.record_uint("required_reservation", required);
753 root.record_uint("borrowed_reservation", borrowed);
754 if let Some(earliest_checkpoint) = earliest_checkpoint {
755 root.record_uint("earliest_checkpoint", earliest_checkpoint);
756 }
757
758 if let Some(x) = round_div(100 * borrowed, required) {
760 root.record_uint("borrowed_to_required_reservation_percent", x);
761 }
762 }
763 Ok(inspector)
764 }
765 .boxed()
766 });
767 }
768
769 pub fn needs_borrow_for_journal(&self, checkpoint: u64) -> bool {
774 checkpoint.checked_sub(self.inner.read().last_end_offset).unwrap() > 256
775 }
776}
777
778pub struct ReservationUpdate(u64);
783
784impl ReservationUpdate {
785 pub fn new(amount: u64) -> Self {
786 Self(amount)
787 }
788}
789
790impl AssociatedObject for ReservationUpdate {
791 fn will_apply_mutation(&self, _mutation: &Mutation, object_id: u64, manager: &ObjectManager) {
792 manager.update_reservation(object_id, self.0);
793 }
794}