| // Copyright 2021 The Fuchsia Authors. All rights reserved. |
| // Use of this source code is governed by a BSD-style license that can be |
| // found in the LICENSE file. |
| |
| use { |
| crate::{ |
| crypt::Crypt, |
| debug_assert_not_too_long, |
| errors::FxfsError, |
| log::*, |
| metrics::{traits::Metric as _, traits::NumericMetric as _, UintMetric}, |
| object_store::{ |
| allocator::{Allocator, Hold, Reservation, ReservationOwner}, |
| directory::Directory, |
| graveyard::Graveyard, |
| journal::{super_block::SuperBlock, Journal, JournalCheckpoint, JournalOptions}, |
| object_manager::ObjectManager, |
| transaction::{ |
| self, AssocObj, LockKey, LockManager, MetadataReservation, Mutation, ReadGuard, |
| Transaction, TransactionHandler, TransactionLocks, WriteGuard, |
| TRANSACTION_METADATA_MAX_AMOUNT, |
| }, |
| volume::{root_volume, VOLUMES_DIRECTORY}, |
| ObjectStore, |
| }, |
| serialized_types::Version, |
| trace_duration, |
| }, |
| anyhow::{Context, Error}, |
| async_trait::async_trait, |
| fuchsia_async as fasync, |
| futures::channel::oneshot::{channel, Sender}, |
| once_cell::sync::OnceCell, |
| std::sync::{ |
| atomic::{self, AtomicBool}, |
| Arc, Mutex, |
| }, |
| storage_device::{Device, DeviceHolder}, |
| }; |
| |
| pub const MIN_BLOCK_SIZE: u64 = 4096; |
| |
| /// Holds information on an Fxfs Filesystem |
| pub struct Info { |
| pub total_bytes: u64, |
| pub used_bytes: u64, |
| } |
| |
| pub struct Options { |
| /// The metadata keys will be rolled after this many bytes. This must be large enough such that |
| /// we can't end up with more than two live keys (so it must be bigger than the maximum possible |
| /// size of unflushed journal contents). This is exposed for testing purposes. |
| pub roll_metadata_key_byte_count: u64, |
| |
| #[cfg(test)] |
| pub after_metadata_key_roll_hook: Option< |
| Box< |
| dyn Fn(Arc<dyn Filesystem>) -> futures::future::BoxFuture<'static, Result<(), Error>> |
| + Send |
| + Sync, |
| >, |
| >, |
| } |
| |
| impl Default for Options { |
| fn default() -> Self { |
| Options { |
| roll_metadata_key_byte_count: 128 * 1024 * 1024, |
| #[cfg(test)] |
| after_metadata_key_roll_hook: None, |
| } |
| } |
| } |
| |
| #[async_trait] |
| pub trait Filesystem: TransactionHandler { |
| /// Returns access to the undeyling device. |
| fn device(&self) -> Arc<dyn Device>; |
| |
| /// Returns the root store or panics if it is not available. |
| fn root_store(&self) -> Arc<ObjectStore>; |
| |
| /// Returns the allocator or panics if it is not available. |
| fn allocator(&self) -> Arc<dyn Allocator>; |
| |
| /// Returns the object manager for the filesystem. |
| fn object_manager(&self) -> &Arc<ObjectManager>; |
| |
| /// Flushes buffered data to the underlying device. |
| async fn sync(&self, options: SyncOptions<'_>) -> Result<(), Error>; |
| |
| /// Returns the filesystem block size. |
| fn block_size(&self) -> u64; |
| |
| /// Returns filesystem information. |
| fn get_info(&self) -> Info; |
| |
| /// Returns the graveyard manager (whilst there exists a graveyard per store, the manager |
| /// handles all of them). |
| fn graveyard(&self) -> &Arc<Graveyard>; |
| |
| /// Whether to enable verbose logging. |
| fn trace(&self) -> bool { |
| false |
| } |
| |
| /// Returns the filesystem options. |
| fn options(&self) -> &Options; |
| } |
| |
| /// The context in which a transaction is being applied. |
| pub struct ApplyContext<'a, 'b> { |
| /// The mode indicates whether the transaction is being replayed. |
| pub mode: ApplyMode<'a, 'b>, |
| |
| /// The transaction checkpoint for this mutation. |
| pub checkpoint: JournalCheckpoint, |
| } |
| |
| /// A transaction can be applied during replay or on a live running system (in which case a |
| /// transaction object will be available). |
| pub enum ApplyMode<'a, 'b> { |
| Replay, |
| Live(&'a Transaction<'b>), |
| } |
| |
| impl ApplyMode<'_, '_> { |
| pub fn is_replay(&self) -> bool { |
| matches!(self, ApplyMode::Replay) |
| } |
| |
| pub fn is_live(&self) -> bool { |
| matches!(self, ApplyMode::Live(_)) |
| } |
| } |
| |
| #[async_trait] |
| pub trait JournalingObject: Send + Sync { |
| /// Objects that use the journaling system to track mutations should implement this trait. This |
| /// method will get called when the transaction commits, which can either be during live |
| /// operation or during journal replay, in which case transaction will be None. Also see |
| /// ObjectManager's apply_mutation method. |
| async fn apply_mutation( |
| &self, |
| mutation: Mutation, |
| context: &ApplyContext<'_, '_>, |
| assoc_obj: AssocObj<'_>, |
| ); |
| |
| /// Called when a transaction fails to commit. |
| fn drop_mutation(&self, mutation: Mutation, transaction: &Transaction<'_>); |
| |
| /// Flushes in-memory changes to the device (to allow journal space to be freed). |
| /// |
| /// Also returns the earliest version of a struct in the filesystem. |
| async fn flush(&self) -> Result<Version, Error>; |
| |
| /// Optionally encrypts a mutation. |
| fn encrypt_mutation(&self, _mutation: &Mutation) -> Option<Mutation> { |
| None |
| } |
| } |
| |
| #[derive(Default)] |
| pub struct SyncOptions<'a> { |
| pub flush_device: bool, |
| |
| // A precondition that is evaluated whilst a lock is held that determines whether or not the |
| // sync needs to proceed. |
| pub precondition: Option<Box<dyn FnOnce() -> bool + 'a + Send>>, |
| } |
| |
| pub struct OpenFxFilesystem(Arc<FxFilesystem>); |
| |
| impl OpenFxFilesystem { |
| /// Waits for filesystem to be dropped (so callers should ensure all direct and indirect |
| /// references are dropped) and returns the device. No attempt is made at a graceful shutdown. |
| pub async fn take_device(self) -> DeviceHolder { |
| let (sender, receiver) = channel::<DeviceHolder>(); |
| self.device_sender |
| .set(sender) |
| .unwrap_or_else(|_| panic!("take_device should only be called once")); |
| std::mem::drop(self); |
| debug_assert_not_too_long!(receiver).unwrap() |
| } |
| } |
| |
| impl From<Arc<FxFilesystem>> for OpenFxFilesystem { |
| fn from(fs: Arc<FxFilesystem>) -> Self { |
| Self(fs) |
| } |
| } |
| |
| impl Drop for OpenFxFilesystem { |
| fn drop(&mut self) { |
| if !self.read_only && !self.closed.load(atomic::Ordering::SeqCst) { |
| error!("OpenFxFilesystem dropped without first being closed. Data loss may occur."); |
| } |
| } |
| } |
| |
| impl std::ops::Deref for OpenFxFilesystem { |
| type Target = Arc<FxFilesystem>; |
| |
| fn deref(&self) -> &Self::Target { |
| &self.0 |
| } |
| } |
| |
| pub struct FxFilesystem { |
| device: OnceCell<DeviceHolder>, |
| block_size: u64, |
| objects: Arc<ObjectManager>, |
| journal: Journal, |
| lock_manager: LockManager, |
| flush_task: Mutex<Option<fasync::Task<()>>>, |
| device_sender: OnceCell<Sender<DeviceHolder>>, |
| closed: AtomicBool, |
| read_only: bool, |
| trace: bool, |
| graveyard: Arc<Graveyard>, |
| completed_transactions: UintMetric, |
| options: Options, |
| } |
| |
| #[derive(Default)] |
| pub struct OpenOptions { |
| pub trace: bool, |
| pub read_only: bool, |
| pub journal_options: JournalOptions, |
| |
| /// Called immediately after creating the allocator. |
| pub on_new_allocator: Option<Box<dyn Fn(Arc<dyn Allocator>) + Send + Sync>>, |
| |
| /// Called each time a new store is registered with ObjectManager. |
| pub on_new_store: Option<Box<dyn Fn(&ObjectStore) + Send + Sync>>, |
| |
| /// Filesystem options. |
| pub filesystem_options: Options, |
| } |
| |
| impl FxFilesystem { |
| pub async fn new_empty(device: DeviceHolder) -> Result<OpenFxFilesystem, Error> { |
| let objects = Arc::new(ObjectManager::new(None)); |
| let journal = Journal::new(objects.clone(), JournalOptions::default()); |
| let block_size = std::cmp::max(device.block_size().into(), MIN_BLOCK_SIZE); |
| assert_eq!(block_size % MIN_BLOCK_SIZE, 0); |
| let filesystem = Arc::new(FxFilesystem { |
| device: OnceCell::new(), |
| block_size, |
| objects: objects.clone(), |
| journal, |
| lock_manager: LockManager::new(), |
| flush_task: Mutex::new(None), |
| device_sender: OnceCell::new(), |
| closed: AtomicBool::new(true), |
| read_only: false, |
| trace: false, |
| graveyard: Graveyard::new(objects.clone()), |
| completed_transactions: UintMetric::new("completed_transactions", 0), |
| options: Default::default(), |
| }); |
| filesystem.device.set(device).unwrap_or_else(|_| unreachable!()); |
| filesystem.journal.init_empty(filesystem.clone()).await?; |
| |
| // Start the graveyard's background reaping task. |
| filesystem.graveyard.clone().reap_async(); |
| |
| // Create the root volume directory. |
| let root_store = filesystem.root_store(); |
| let root_directory = Directory::open(&root_store, root_store.root_directory_object_id()) |
| .await |
| .context("Unable to open root volume directory")?; |
| let mut transaction = filesystem |
| .clone() |
| .new_transaction( |
| &[LockKey::object(root_store.store_object_id(), root_directory.object_id())], |
| transaction::Options::default(), |
| ) |
| .await?; |
| let volume_directory = |
| root_directory.create_child_dir(&mut transaction, VOLUMES_DIRECTORY).await?; |
| transaction.commit().await?; |
| |
| objects.set_volume_directory(volume_directory); |
| filesystem.closed.store(false, atomic::Ordering::SeqCst); |
| Ok(filesystem.into()) |
| } |
| |
| pub async fn open_with_options( |
| device: DeviceHolder, |
| options: OpenOptions, |
| ) -> Result<OpenFxFilesystem, Error> { |
| let objects = Arc::new(ObjectManager::new(options.on_new_store)); |
| let journal = Journal::new(objects.clone(), options.journal_options); |
| let block_size = std::cmp::max(device.block_size().into(), MIN_BLOCK_SIZE); |
| assert_eq!(block_size % MIN_BLOCK_SIZE, 0); |
| let filesystem = Arc::new(FxFilesystem { |
| device: OnceCell::new(), |
| block_size, |
| objects: objects.clone(), |
| journal, |
| lock_manager: LockManager::new(), |
| flush_task: Mutex::new(None), |
| device_sender: OnceCell::new(), |
| closed: AtomicBool::new(true), |
| read_only: options.read_only, |
| trace: options.trace, |
| graveyard: Graveyard::new(objects.clone()), |
| completed_transactions: UintMetric::new("completed_transactions", 0), |
| options: options.filesystem_options, |
| }); |
| if !options.read_only { |
| // See comment in JournalRecord::DidFlushDevice for why we need to flush the device |
| // before replay. |
| device.flush().await.context("Device flush failed")?; |
| } |
| filesystem.device.set(device).unwrap_or_else(|_| unreachable!()); |
| filesystem |
| .journal |
| .replay(filesystem.clone(), options.on_new_allocator) |
| .await |
| .context("Journal replay failed")?; |
| filesystem.journal.set_trace(filesystem.trace); |
| filesystem.root_store().set_trace(filesystem.trace); |
| if !options.read_only { |
| // Start the async reaper. |
| filesystem.graveyard.clone().reap_async(); |
| // Purge old entries. |
| for store in objects.unlocked_stores() { |
| filesystem |
| .graveyard |
| .initial_reap(&store, filesystem.journal.journal_file_offset()) |
| .await?; |
| } |
| } |
| filesystem.closed.store(false, atomic::Ordering::SeqCst); |
| Ok(filesystem.into()) |
| } |
| |
| pub async fn open(device: DeviceHolder) -> Result<OpenFxFilesystem, Error> { |
| Self::open_with_options(device, OpenOptions::default()).await |
| } |
| |
| pub fn root_parent_store(&self) -> Arc<ObjectStore> { |
| self.objects.root_parent_store() |
| } |
| |
| fn root_store(&self) -> Arc<ObjectStore> { |
| self.objects.root_store() |
| } |
| |
| pub async fn close(&self) -> Result<(), Error> { |
| assert_eq!(self.closed.swap(true, atomic::Ordering::SeqCst), false); |
| debug_assert_not_too_long!(self.graveyard.wait_for_reap()); |
| self.journal.stop_compactions().await; |
| let sync_status = |
| self.journal.sync(SyncOptions { flush_device: true, ..Default::default() }).await; |
| if let Err(e) = &sync_status { |
| error!(error = e.as_value(), "Failed to sync filesystem; data may be lost"); |
| } |
| self.journal.terminate(); |
| let flush_task = self.flush_task.lock().unwrap().take(); |
| if let Some(task) = flush_task { |
| debug_assert_not_too_long!(task); |
| } |
| // Regardless of whether sync succeeds, we should close the device, since otherwise we will |
| // crash instead of exiting gracefully. |
| self.device().close().await.context("Failed to close device")?; |
| sync_status.map(|_| ()) |
| } |
| |
| pub fn super_block(&self) -> SuperBlock { |
| self.journal.super_block() |
| } |
| |
| async fn reservation_for_transaction<'a>( |
| self: &Arc<Self>, |
| options: transaction::Options<'a>, |
| ) -> Result<(MetadataReservation, Option<&'a Reservation>, Option<Hold<'a>>), Error> { |
| if !options.skip_journal_checks { |
| // TODO(fxbug.dev/96073): for now, we don't allow for transactions that might be |
| // inflight but not committed. In theory, if there are a large number of them, it would |
| // be possible to run out of journal space. We should probably have an in-flight limit. |
| self.journal.check_journal_space().await?; |
| } |
| |
| // We support three options for metadata space reservation: |
| // |
| // 1. We can borrow from the filesystem's metadata reservation. This should only be |
| // be used on the understanding that eventually, potentially after a full compaction, |
| // there should be no net increase in space used. For example, unlinking an object |
| // should eventually decrease the amount of space used and setting most attributes |
| // should not result in any change. |
| // |
| // 2. A reservation is provided in which case we'll place a hold on some of it for |
| // metadata. |
| // |
| // 3. No reservation is supplied, so we try and reserve space with the allocator now, |
| // and will return NoSpace if that fails. |
| let mut hold = None; |
| let metadata_reservation = if options.borrow_metadata_space { |
| MetadataReservation::Borrowed |
| } else { |
| match options.allocator_reservation { |
| Some(reservation) => { |
| hold = Some( |
| reservation |
| .reserve(TRANSACTION_METADATA_MAX_AMOUNT) |
| .ok_or(FxfsError::NoSpace)?, |
| ); |
| MetadataReservation::Hold(TRANSACTION_METADATA_MAX_AMOUNT) |
| } |
| None => { |
| let reservation = self |
| .allocator() |
| .reserve(TRANSACTION_METADATA_MAX_AMOUNT) |
| .ok_or(FxfsError::NoSpace)?; |
| MetadataReservation::Reservation(reservation) |
| } |
| } |
| }; |
| Ok((metadata_reservation, options.allocator_reservation, hold)) |
| } |
| } |
| |
| impl Drop for FxFilesystem { |
| fn drop(&mut self) { |
| if let Some(sender) = self.device_sender.take() { |
| // We don't care if this fails to send. |
| let _ = sender.send(self.device.take().unwrap()); |
| } |
| } |
| } |
| |
| #[async_trait] |
| impl Filesystem for FxFilesystem { |
| fn device(&self) -> Arc<dyn Device> { |
| Arc::clone(self.device.get().unwrap()) |
| } |
| |
| fn root_store(&self) -> Arc<ObjectStore> { |
| self.objects.root_store() |
| } |
| |
| fn allocator(&self) -> Arc<dyn Allocator> { |
| self.objects.allocator() |
| } |
| |
| fn object_manager(&self) -> &Arc<ObjectManager> { |
| &self.objects |
| } |
| |
| async fn sync(&self, options: SyncOptions<'_>) -> Result<(), Error> { |
| self.journal.sync(options).await.map(|_| ()) |
| } |
| |
| fn block_size(&self) -> u64 { |
| self.block_size |
| } |
| |
| fn get_info(&self) -> Info { |
| Info { |
| total_bytes: self.device.get().unwrap().size(), |
| used_bytes: self.object_manager().allocator().get_used_bytes(), |
| } |
| } |
| |
| fn graveyard(&self) -> &Arc<Graveyard> { |
| &self.graveyard |
| } |
| |
| fn trace(&self) -> bool { |
| self.trace |
| } |
| |
| fn options(&self) -> &Options { |
| &self.options |
| } |
| } |
| |
| #[async_trait] |
| impl TransactionHandler for FxFilesystem { |
| async fn new_transaction<'a>( |
| self: Arc<Self>, |
| locks: &[LockKey], |
| options: transaction::Options<'a>, |
| ) -> Result<Transaction<'a>, Error> { |
| let (metadata_reservation, allocator_reservation, hold) = |
| self.reservation_for_transaction(options).await?; |
| let mut transaction = |
| Transaction::new(self, metadata_reservation, &[LockKey::Filesystem], locks).await; |
| hold.map(|h| h.forget()); // Transaction takes ownership from here on. |
| transaction.allocator_reservation = allocator_reservation; |
| Ok(transaction) |
| } |
| |
| async fn transaction_lock<'a>(&'a self, lock_keys: &[LockKey]) -> TransactionLocks<'a> { |
| let lock_manager: &LockManager = self.as_ref(); |
| TransactionLocks(debug_assert_not_too_long!(lock_manager.txn_lock(lock_keys))) |
| } |
| |
| async fn commit_transaction( |
| self: Arc<Self>, |
| transaction: &mut Transaction<'_>, |
| ) -> Result<u64, Error> { |
| trace_duration!("FxFilesystem::commit_transaction"); |
| debug_assert_not_too_long!(self.lock_manager.commit_prepare(&transaction)); |
| { |
| let mut flush_task = self.flush_task.lock().unwrap(); |
| if flush_task.is_none() { |
| let this = self.clone(); |
| *flush_task = Some(fasync::Task::spawn(async move { |
| this.journal.flush_task().await; |
| })); |
| } |
| } |
| let journal_offset = self.journal.commit(transaction).await?; |
| self.completed_transactions.add(1); |
| Ok(journal_offset) |
| } |
| |
| fn drop_transaction(&self, transaction: &mut Transaction<'_>) { |
| // If we placed a hold for metadata space, return it now. |
| if let MetadataReservation::Hold(hold_amount) = &mut transaction.metadata_reservation { |
| transaction |
| .allocator_reservation |
| .unwrap() |
| .release_reservation(std::mem::take(hold_amount)); |
| } |
| self.objects.drop_transaction(transaction); |
| self.lock_manager.drop_transaction(transaction); |
| } |
| |
| async fn read_lock<'a>(&'a self, lock_keys: &[LockKey]) -> ReadGuard<'a> { |
| debug_assert_not_too_long!(self.lock_manager.read_lock(lock_keys)) |
| } |
| |
| async fn write_lock<'a>(&'a self, lock_keys: &[LockKey]) -> WriteGuard<'a> { |
| debug_assert_not_too_long!(self.lock_manager.write_lock(lock_keys)) |
| } |
| } |
| |
| impl AsRef<LockManager> for FxFilesystem { |
| fn as_ref(&self) -> &LockManager { |
| &self.lock_manager |
| } |
| } |
| |
| /// Helper method for making a new filesystem. |
| pub async fn mkfs(device: DeviceHolder, crypt: Option<Arc<dyn Crypt>>) -> Result<(), Error> { |
| let fs = FxFilesystem::new_empty(device).await?; |
| { |
| // expect instead of propagating errors here, since otherwise we could drop |fs| before |
| // close is called, which leads to confusing and unrelated error messages. |
| let root_volume = root_volume(&fs).await.expect("Open root_volume failed"); |
| root_volume.new_volume("default", crypt).await.expect("Create volume failed"); |
| } |
| fs.close().await?; |
| Ok(()) |
| } |
| |
| #[cfg(test)] |
| mod tests { |
| use { |
| super::{Filesystem, FxFilesystem, OpenOptions, SyncOptions}, |
| crate::{ |
| fsck::fsck, |
| lsm_tree::{types::Item, Operation}, |
| object_handle::{ObjectHandle, WriteObjectHandle}, |
| object_store::{ |
| allocator::SimpleAllocator, |
| directory::replace_child, |
| directory::Directory, |
| journal::JournalOptions, |
| transaction::{Options, TransactionHandler}, |
| }, |
| }, |
| fuchsia_async as fasync, |
| futures::future::join_all, |
| std::{ |
| collections::HashMap, |
| sync::{Arc, Mutex}, |
| }, |
| storage_device::{fake_device::FakeDevice, DeviceHolder}, |
| }; |
| |
| const TEST_DEVICE_BLOCK_SIZE: u32 = 512; |
| |
| #[fasync::run(10, test)] |
| async fn test_compaction() { |
| let device = DeviceHolder::new(FakeDevice::new(8192, TEST_DEVICE_BLOCK_SIZE)); |
| |
| // If compaction is not working correctly, this test will run out of space. |
| let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed"); |
| let root_store = fs.root_store(); |
| let root_directory = Directory::open(&root_store, root_store.root_directory_object_id()) |
| .await |
| .expect("open failed"); |
| |
| let mut tasks = Vec::new(); |
| for i in 0..2 { |
| let mut transaction = fs |
| .clone() |
| .new_transaction(&[], Options::default()) |
| .await |
| .expect("new_transaction failed"); |
| let handle = root_directory |
| .create_child_file(&mut transaction, &format!("{}", i)) |
| .await |
| .expect("create_child_file failed"); |
| transaction.commit().await.expect("commit failed"); |
| tasks.push(fasync::Task::spawn(async move { |
| const TEST_DATA: &[u8] = b"hello"; |
| let mut buf = handle.allocate_buffer(TEST_DATA.len()); |
| buf.as_mut_slice().copy_from_slice(TEST_DATA); |
| for _ in 0..1500 { |
| handle.write_or_append(Some(0), buf.as_ref()).await.expect("write failed"); |
| } |
| })); |
| } |
| join_all(tasks).await; |
| fs.sync(SyncOptions::default()).await.expect("sync failed"); |
| |
| fsck(&fs, None).await.expect("fsck failed"); |
| fs.close().await.expect("Close failed"); |
| } |
| |
| #[fasync::run(10, test)] |
| async fn test_replay_is_identical() { |
| let device = DeviceHolder::new(FakeDevice::new(8192, TEST_DEVICE_BLOCK_SIZE)); |
| let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed"); |
| |
| // Reopen the store, but set reclaim size to a very large value which will effectively |
| // stop the journal from flushing and allows us to track all the mutations to the store. |
| fs.close().await.expect("close failed"); |
| let device = fs.take_device().await; |
| device.reopen(); |
| |
| struct Mutations<K, V>(Mutex<Vec<(Operation, Item<K, V>)>>); |
| |
| impl<K: Clone, V: Clone> Mutations<K, V> { |
| fn new() -> Self { |
| Mutations(Mutex::new(Vec::new())) |
| } |
| |
| fn push(&self, operation: Operation, item: &Item<K, V>) { |
| self.0.lock().unwrap().push((operation, item.clone())); |
| } |
| } |
| |
| let open_fs = |device, |
| object_mutations: Arc<Mutex<HashMap<_, _>>>, |
| allocator_mutations: Arc<Mutations<_, _>>| async { |
| FxFilesystem::open_with_options( |
| device, |
| OpenOptions { |
| journal_options: JournalOptions { |
| reclaim_size: u64::MAX, |
| ..Default::default() |
| }, |
| on_new_allocator: Some(Box::new(move |allocator| { |
| let allocator = allocator |
| .as_any() |
| .downcast::<SimpleAllocator>() |
| .unwrap_or_else(|_| panic!("Unexpected allocator")); |
| let allocator_mutations = allocator_mutations.clone(); |
| allocator.tree().set_mutation_callback(Some(Box::new(move |op, item| { |
| allocator_mutations.push(op, item) |
| }))); |
| })), |
| on_new_store: Some(Box::new(move |store| { |
| let mutations = Arc::new(Mutations::new()); |
| object_mutations |
| .lock() |
| .unwrap() |
| .insert(store.store_object_id(), mutations.clone()); |
| store.tree().set_mutation_callback(Some(Box::new(move |op, item| { |
| mutations.push(op, item) |
| }))); |
| })), |
| ..Default::default() |
| }, |
| ) |
| .await |
| .expect("open_with_options failed") |
| }; |
| |
| let allocator_mutations = Arc::new(Mutations::new()); |
| let object_mutations = Arc::new(Mutex::new(HashMap::new())); |
| let fs = open_fs(device, object_mutations.clone(), allocator_mutations.clone()).await; |
| |
| let root_store = fs.root_store(); |
| let root_directory = Directory::open(&root_store, root_store.root_directory_object_id()) |
| .await |
| .expect("open failed"); |
| |
| let mut transaction = fs |
| .clone() |
| .new_transaction(&[], Options::default()) |
| .await |
| .expect("new_transaction failed"); |
| let object = root_directory |
| .create_child_file(&mut transaction, "test") |
| .await |
| .expect("create_child_file failed"); |
| transaction.commit().await.expect("commit failed"); |
| |
| // Append some data. |
| let buf = object.allocate_buffer(10000); |
| object.write_or_append(Some(0), buf.as_ref()).await.expect("write failed"); |
| |
| // Overwrite some data. |
| object.write_or_append(Some(5000), buf.as_ref()).await.expect("write failed"); |
| |
| // Truncate. |
| (&object as &dyn WriteObjectHandle).truncate(3000).await.expect("truncate failed"); |
| |
| // Delete the object. |
| let mut transaction = fs |
| .clone() |
| .new_transaction(&[], Options::default()) |
| .await |
| .expect("new_transaction failed"); |
| |
| replace_child(&mut transaction, None, (&root_directory, "test")) |
| .await |
| .expect("replace_child failed"); |
| |
| transaction.commit().await.expect("commit failed"); |
| |
| // Finally tombstone the object. |
| root_store |
| .tombstone(object.object_id(), Options::default()) |
| .await |
| .expect("tombstone failed"); |
| |
| // Now reopen and check that replay produces the same set of mutations. |
| fs.close().await.expect("close failed"); |
| |
| let metadata_reservation_amount = fs.object_manager().metadata_reservation().amount(); |
| |
| let device = fs.take_device().await; |
| device.reopen(); |
| |
| let replayed_object_mutations = Arc::new(Mutex::new(HashMap::new())); |
| let replayed_allocator_mutations = Arc::new(Mutations::new()); |
| let fs = open_fs( |
| device, |
| replayed_object_mutations.clone(), |
| replayed_allocator_mutations.clone(), |
| ) |
| .await; |
| |
| let m1 = object_mutations.lock().unwrap(); |
| let m2 = replayed_object_mutations.lock().unwrap(); |
| assert_eq!(m1.len(), m2.len()); |
| for (store_id, mutations) in &*m1 { |
| let mutations = mutations.0.lock().unwrap(); |
| let replayed = m2.get(&store_id).expect("Found unexpected store").0.lock().unwrap(); |
| assert_eq!(mutations.len(), replayed.len()); |
| for ((op1, i1), (op2, i2)) in mutations.iter().zip(replayed.iter()) { |
| assert_eq!(op1, op2); |
| assert_eq!(i1.key, i2.key); |
| assert_eq!(i1.value, i2.value); |
| assert_eq!(i1.sequence, i2.sequence); |
| } |
| } |
| |
| let a1 = allocator_mutations.0.lock().unwrap(); |
| let a2 = replayed_allocator_mutations.0.lock().unwrap(); |
| assert_eq!(a1.len(), a2.len()); |
| for ((op1, i1), (op2, i2)) in a1.iter().zip(a2.iter()) { |
| assert_eq!(op1, op2); |
| assert_eq!(i1.key, i2.key); |
| assert_eq!(i1.value, i2.value); |
| assert_eq!(i1.sequence, i2.sequence); |
| } |
| |
| assert_eq!( |
| fs.object_manager().metadata_reservation().amount(), |
| metadata_reservation_amount |
| ); |
| } |
| } |
