src/storage/fxfs/src/object_store/filesystem.rs

// 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::{
        device::Device,
        object_store::{
            allocator::Allocator,
            constants::INVALID_OBJECT_ID,
            journal::{Journal, JournalCheckpoint},
            transaction::{
                AssociatedObject, LockKey, LockManager, Mutation, ReadGuard, Transaction,
                TransactionHandler, TxnMutation,
            },
            ObjectStore,
        },
    },
    anyhow::Error,
    async_trait::async_trait,
    std::{
        collections::HashMap,
        sync::{Arc, RwLock},
    },
};

#[async_trait]
pub trait Filesystem: TransactionHandler {
    /// Informs the journaling system that a new store has been created so that when a transaction
    /// is committed or replayed, mutations can be routed to the correct store.
    fn register_store(&self, store: &Arc<ObjectStore>);

    /// Informs the journaling system that the given object ID is about to flush in-memory data.  If
    /// successful, all mutations pertinent to this object can be discarded, but any mutations that
    /// follow will still be kept.
    fn begin_object_sync(&self, object_id: u64) -> ObjectSync;

    /// 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>;
}

pub struct ObjectManager {
    objects: RwLock<Objects>,
}

// We currently maintain strong references to all stores that have been opened, but there's no
// currently no mechanism for releasing stores that aren't being used.
struct Objects {
    stores: HashMap<u64, Arc<ObjectStore>>,
    root_parent_store_object_id: u64,
    root_store_object_id: u64,
    allocator_object_id: u64,
    allocator: Option<Arc<dyn Allocator>>,

    // Records dependencies on the journal for objects i.e. an entry for object ID 1, would mean it
    // has a dependency on journal records from that offset.
    journal_file_checkpoints: HashMap<u64, JournalCheckpoint>,
}

impl ObjectManager {
    pub fn new() -> ObjectManager {
        ObjectManager {
            objects: RwLock::new(Objects {
                stores: HashMap::new(),
                root_parent_store_object_id: INVALID_OBJECT_ID,
                root_store_object_id: INVALID_OBJECT_ID,
                allocator_object_id: INVALID_OBJECT_ID,
                allocator: None,
                journal_file_checkpoints: HashMap::new(),
            }),
        }
    }

    pub fn root_parent_store(&self) -> Arc<ObjectStore> {
        let objects = self.objects.read().unwrap();
        objects.stores.get(&objects.root_parent_store_object_id).unwrap().clone()
    }

    pub fn set_root_parent_store_object_id(&self, object_id: u64) {
        let mut objects = self.objects.write().unwrap();
        assert!(objects.stores.contains_key(&object_id));
        objects.root_parent_store_object_id = object_id;
    }

    pub fn register_store(&self, store: &Arc<ObjectStore>) {
        let mut objects = self.objects.write().unwrap();
        assert_ne!(store.store_object_id(), objects.allocator_object_id);
        assert!(objects.stores.insert(store.store_object_id(), store.clone()).is_none());
    }

    pub fn store(&self, store_object_id: u64) -> Option<Arc<ObjectStore>> {
        self.objects.read().unwrap().stores.get(&store_object_id).cloned()
    }

    pub fn set_root_store_object_id(&self, object_id: u64) {
        let mut objects = self.objects.write().unwrap();
        assert!(objects.stores.contains_key(&object_id));
        objects.root_store_object_id = object_id;
    }

    pub fn root_store(&self) -> Arc<ObjectStore> {
        let objects = self.objects.read().unwrap();
        objects.stores.get(&objects.root_store_object_id).unwrap().clone()
    }

    pub fn set_allocator(&self, allocator: Arc<dyn Allocator>) {
        let mut objects = self.objects.write().unwrap();
        assert!(!objects.stores.contains_key(&allocator.object_id()));
        objects.allocator_object_id = allocator.object_id();
        objects.allocator = Some(allocator.clone());
    }

    pub fn allocator(&self) -> Arc<dyn Allocator> {
        self.objects.read().unwrap().allocator.clone().unwrap()
    }

    /// The journaling system should call this when a mutation needs to be applied. |replay|
    /// indicates whether this is for replay.  |checkpoint| indicates the location in the journal
    /// file for this mutation and is used to keep track of each object's dependencies on the
    /// journal.
    pub async fn apply_mutation(
        &self,
        object_id: u64,
        mutation: Mutation,
        replay: bool,
        checkpoint: &JournalCheckpoint,
        object: Option<AssociatedObject<'_>>,
    ) {
        {
            let mut objects = self.objects.write().unwrap();
            objects.journal_file_checkpoints.entry(object_id).or_insert_with(|| checkpoint.clone());
            if object_id == objects.allocator_object_id {
                Some(objects.allocator.clone().unwrap().as_mutations())
            } else {
                objects.stores.get(&object_id).map(|x| x.clone() as Arc<dyn Mutations>)
            }
        }
        .unwrap_or_else(|| self.root_store().lazy_open_store(object_id))
        .apply_mutation(mutation, replay, object)
        .await;
    }

    // Drops a transaction.  This is called automatically when a transaction is dropped.  If the
    // transaction has been committed, it should contain no mutations and so nothing will get rolled
    // back.  For each mutation, drop_mutation is called to allow for roll back (e.g. the allocator
    // will unreserve allocations).
    pub fn drop_transaction(&self, transaction: &mut Transaction<'_>) {
        for TxnMutation { object_id, mutation, .. } in std::mem::take(&mut transaction.mutations) {
            {
                let objects = self.objects.read().unwrap();
                if object_id == objects.allocator_object_id {
                    Some(objects.allocator.clone().unwrap().as_mutations())
                } else {
                    objects.stores.get(&object_id).map(|x| x.clone() as Arc<dyn Mutations>)
                }
            }
            .map(|o| o.drop_mutation(mutation));
        }
    }

    /// Returns the journal file offsets that each object depends on and the checkpoint for the
    /// minimum offset.
    pub fn journal_file_offsets(&self) -> (HashMap<u64, u64>, Option<JournalCheckpoint>) {
        let objects = self.objects.read().unwrap();
        let mut min_checkpoint = None;
        let mut offsets = HashMap::new();
        for (&object_id, checkpoint) in &objects.journal_file_checkpoints {
            match &mut min_checkpoint {
                None => min_checkpoint = Some(checkpoint),
                Some(ref mut min_checkpoint) => {
                    if checkpoint.file_offset < min_checkpoint.file_offset {
                        *min_checkpoint = checkpoint;
                    }
                }
            }
            offsets.insert(object_id, checkpoint.file_offset);
        }
        (offsets, min_checkpoint.cloned())
    }

    pub fn begin_object_sync(self: &Arc<Self>, object_id: u64) -> ObjectSync {
        let old_journal_file_checkpoint =
            self.objects.write().unwrap().journal_file_checkpoints.remove(&object_id);
        ObjectSync { object_manager: self.clone(), object_id, old_journal_file_checkpoint }
    }
}

/// ObjectSync is used by objects to indicate some kind of event such that if successful, existing
/// mutation records are no longer required from the journal.  For example, for object stores, it is
/// used when the in-memory layer is persisted since once that is done the records in the journal
/// are no longer required.  Clients must make sure to call the commit function upon success; the
/// default is to roll back.
#[must_use]
pub struct ObjectSync {
    object_manager: Arc<ObjectManager>,
    object_id: u64,
    old_journal_file_checkpoint: Option<JournalCheckpoint>,
}

impl ObjectSync {
    pub fn needs_sync(&self) -> bool {
        self.old_journal_file_checkpoint.is_some()
    }

    pub fn commit(mut self) {
        self.old_journal_file_checkpoint = None;
    }
}

impl Drop for ObjectSync {
    fn drop(&mut self) {
        if let Some(checkpoint) = self.old_journal_file_checkpoint.take() {
            self.object_manager
                .objects
                .write()
                .unwrap()
                .journal_file_checkpoints
                .insert(self.object_id, checkpoint);
        }
    }
}

#[async_trait]
pub trait Mutations: 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 |replay| will be true.  Also see
    /// ObjectManager's apply_mutation method.
    async fn apply_mutation(
        &self,
        mutation: Mutation,
        replay: bool,
        object: Option<AssociatedObject<'_>>,
    );

    /// Called when a transaction fails to commit.
    fn drop_mutation(&self, mutation: Mutation);
}

#[derive(Default)]
pub struct SyncOptions {}

pub struct FxFilesystem {
    device: Arc<dyn Device>,
    objects: Arc<ObjectManager>,
    journal: Journal,
    lock_manager: LockManager,
}

impl FxFilesystem {
    pub async fn new_empty(device: Arc<dyn Device>) -> Result<Arc<FxFilesystem>, Error> {
        let objects = Arc::new(ObjectManager::new());
        let journal = Journal::new(objects.clone());
        let filesystem = Arc::new(FxFilesystem {
            device: device.clone(),
            objects: objects.clone(),
            journal,
            lock_manager: LockManager::new(),
        });
        filesystem.journal.init_empty(filesystem.clone()).await?;
        Ok(filesystem)
    }

    pub async fn open(device: Arc<dyn Device>) -> Result<Arc<FxFilesystem>, Error> {
        let objects = Arc::new(ObjectManager::new());
        let journal = Journal::new(objects.clone());
        let filesystem = Arc::new(FxFilesystem {
            device: device.clone(),
            objects: objects.clone(),
            journal,
            lock_manager: LockManager::new(),
        });
        filesystem.journal.replay(filesystem.clone()).await?;
        Ok(filesystem)
    }

    pub fn root_parent_store(&self) -> Arc<ObjectStore> {
        self.objects.root_parent_store()
    }

    pub fn root_store(&self) -> Arc<ObjectStore> {
        self.objects.root_store()
    }

    pub fn store(&self, object_id: u64) -> Option<Arc<ObjectStore>> {
        self.objects.store(object_id)
    }

    pub async fn sync(&self, options: SyncOptions) -> Result<(), Error> {
        self.journal.sync(options).await
    }

    pub fn volume_info_object_id(&self) -> u64 {
        self.journal.volume_info_object_id()
    }

    pub fn set_volume_info_object_id(&self, object_id: u64) {
        self.journal.set_volume_info_object_id(object_id);
    }

    pub async fn close(&self) -> Result<(), Error> {
        // Regardless of whether sync succeeds, we should close the device, since otherwise we will
        // crash instead of exiting gracefully.
        let sync_status = self.journal.sync(SyncOptions::default()).await;
        if sync_status.is_err() {
            log::error!("Failed to sync filesystem; data may be lost: {:?}", sync_status);
        }
        self.device.close().await.expect("Failed to close device");
        sync_status
    }
}

#[async_trait]
impl Filesystem for FxFilesystem {
    fn register_store(&self, store: &Arc<ObjectStore>) {
        self.objects.register_store(store);
    }

    fn begin_object_sync(&self, object_id: u64) -> ObjectSync {
        self.objects.begin_object_sync(object_id)
    }

    fn device(&self) -> Arc<dyn Device> {
        self.device.clone()
    }

    fn root_store(&self) -> Arc<ObjectStore> {
        self.objects.root_store()
    }

    fn allocator(&self) -> Arc<dyn Allocator> {
        self.objects.allocator()
    }
}

#[async_trait]
impl TransactionHandler for FxFilesystem {
    async fn new_transaction<'a>(
        self: Arc<Self>,
        locks: &[LockKey],
    ) -> Result<Transaction<'a>, Error> {
        let mut locks: Vec<_> = locks.iter().cloned().collect();
        locks.sort_unstable();
        self.lock_manager.lock(&locks).await;
        Ok(Transaction::new(self, locks))
    }

    async fn commit_transaction(&self, transaction: Transaction<'_>) {
        self.lock_manager.commit_prepare(&transaction).await;
        self.journal.commit(transaction).await;
    }

    fn drop_transaction(&self, transaction: &mut Transaction<'_>) {
        self.objects.drop_transaction(transaction);
        self.lock_manager.drop_transaction(transaction);
    }

    async fn read_lock<'a>(&'a self, lock_keys: &[LockKey]) -> ReadGuard<'a> {
        self.lock_manager.read_lock(lock_keys).await
    }
}

// TODO(csuter): How do we ensure sync prior to drop?