<commit_before>/**
 * @file fsLogic.cc
 * @author <NAME>
 * @copyright (C) 2016 ACK CYFRONET AGH
 * @copyright This software is released under the MIT license cited in
 * 'LICENSE.txt'
 */

#include "fsLogic.h"

#include "communication/communicator.h"
#include "context.h"
#include "fslogic/virtualfs/archivematica.h"
#include "helpers/logging.h"
#include "messages/configuration.h"
#include "messages/fuse/blockSynchronizationRequest.h"
#include "messages/fuse/changeMode.h"
#include "messages/fuse/createDir.h"
#include "messages/fuse/createFile.h"
#include "messages/fuse/deleteFile.h"
#include "messages/fuse/fileAttr.h"
#include "messages/fuse/fileBlock.h"
#include "messages/fuse/fileChildren.h"
#include "messages/fuse/fileChildrenAttrs.h"
#include "messages/fuse/fileCreated.h"
#include "messages/fuse/fileLocation.h"
#include "messages/fuse/fileLocationChanged.h"
#include "messages/fuse/fileOpened.h"
#include "messages/fuse/fileRenamed.h"
#include "messages/fuse/fileRenamedEntry.h"
#include "messages/fuse/fsStats.h"
#include "messages/fuse/fsync.h"
#include "messages/fuse/getFSStats.h"
#include "messages/fuse/getFileChildren.h"
#include "messages/fuse/getFileChildrenAttrs.h"
#include "messages/fuse/getHelperParams.h"
#include "messages/fuse/getXAttr.h"
#include "messages/fuse/helperParams.h"
#include "messages/fuse/listXAttr.h"
#include "messages/fuse/makeFile.h"
#include "messages/fuse/makeLink.h"
#include "messages/fuse/makeSymLink.h"
#include "messages/fuse/openFile.h"
#include "messages/fuse/readSymLink.h"
#include "messages/fuse/release.h"
#include "messages/fuse/removeXAttr.h"
#include "messages/fuse/rename.h"
#include "messages/fuse/setXAttr.h"
#include "messages/fuse/symLink.h"
#include "messages/fuse/syncResponse.h"
#include "messages/fuse/synchronizeBlock.h"
#include "messages/fuse/synchronizeBlockAndComputeChecksum.h"
#include "messages/fuse/truncate.h"
#include "messages/fuse/updateTimes.h"
#include "messages/fuse/xattr.h"
#include "messages/fuse/xattrList.h"
#include "monitoring/monitoring.h"
#include "util/cdmi.h"
#include "util/uuid.h"
#include "util/xattrHelper.h"

#include <boost/icl/interval_set.hpp>
#include <folly/Demangle.h>
#include <folly/Enumerate.h>
#include <folly/Range.h>
#include <folly/ScopeGuard.h>
#include <folly/fibers/Baton.h>
#include <folly/fibers/FiberManager.h>
#include <folly/fibers/ForEach.h>
#include <folly/json.h>
#include <fuse/fuse_lowlevel.h>
#include <openssl/md4.h>

#include "buffering/bufferAgent.h"

#define IOTRACE_START() auto __ioTraceStart = std::chrono::system_clock::now();

#define IOTRACE_END(TraceType, optype, uuid, handleId, ...)                    \
    if (m_ioTraceLoggerEnabled)                                                \
        m_ioTraceLogger->log(TraceType(__ioTraceStart, optype,                 \
            std::chrono::duration_cast<std::chrono::microseconds>(             \
                std::chrono::system_clock::now() - __ioTraceStart),            \
            uuid, handleId, 0, ##__VA_ARGS__));

#define IOTRACE_GUARD(TraceType, optype, uuid, handleId, ...)                  \
    IOTRACE_START()                                                            \
    auto __ioTraceGuard = folly::makeGuard([&] {                               \
        IOTRACE_END(TraceType, optype, uuid, handleId, ##__VA_ARGS__)          \
    });

namespace one {
namespace client {
namespace fslogic {

using namespace std::literals;

namespace {
const std::string kAbsLinkPrefix = "<__onedata_space_id:";
} // namespace

/**
 * Filters given flags set to one of RDONLY, WRONLY or RDWR.
 * Returns RDONLY if flag value is zero.
 * @param Flags value
 */
inline helpers::Flag getOpenFlag(const helpers::FlagsSet &flagsSet)
{
    if (flagsSet.count(one::helpers::Flag::RDONLY) > 0)
        return one::helpers::Flag::RDONLY;
    if (flagsSet.count(one::helpers::Flag::WRONLY) > 0)
        return one::helpers::Flag::WRONLY;
    if (flagsSet.count(one::helpers::Flag::RDWR) > 0)
        return one::helpers::Flag::RDWR;

    return one::helpers::Flag::RDONLY;
}

constexpr auto XATTR_FILE_BLOCKS_MAP_LENGTH = 50;
constexpr auto LINEAR_PREFETCH_THRESHOLD_MATCH_RATIO = 0.9;

inline static folly::fbstring ONE_XATTR(std::string name)
{
    assert(!name.empty());
    return ONE_XATTR_PREFIX + name;
}

FsLogic::FsLogic(std::shared_ptr<Context> context,
    std::shared_ptr<messages::Configuration> configuration,
    std::unique_ptr<cache::HelpersCache> helpersCache,
    unsigned int metadataCacheSize, bool readEventsDisabled,
    bool forceFullblockRead, const std::chrono::seconds providerTimeout,
    const std::chrono::seconds directoryCacheDropAfter,
    std::function<void(folly::Function<void()>)> runInFiber)
    : m_context{context}
    , m_metadataCache{*m_context->communicator(), metadataCacheSize,
          providerTimeout, directoryCacheDropAfter, configuration->rootUuid(),
          m_context->options()->getSpaceNames(),
          m_context->options()->getSpaceIds(),
          m_context->options()->showOnlyFullReplicas(),
          m_context->options()->showHardLinkCount(),
          m_context->options()->showSpaceIds()}
    , m_helpersCache{std::move(helpersCache)}
    , m_virtualFsHelpersCache{std::make_shared<
          virtualfs::VirtualFsHelpersCache>(*this)}
    , m_readdirCache{std::make_shared<cache::ReaddirCache>(
          m_metadataCache, m_context, m_virtualFsHelpersCache, runInFiber)}
    , m_readEventsDisabled{readEventsDisabled}
    , m_forceFullblockRead{forceFullblockRead}
    , m_fsSubscriptions{m_eventManager, m_metadataCache, m_forceProxyIOCache,
          runInFiber}
    , m_nextFuseHandleId{0}
    , m_providerTimeout{providerTimeout}
    , m_storageTimeout{m_context->options()->getStorageTimeout()}
    , m_runInFiber{std::move(runInFiber)} /* clang-format off */
    , m_prefetchModeAsync{m_context->options()->getPrefetchMode() == "async"}
    , m_minPrefetchBlockSize{m_context->options()->getMinimumBlockPrefetchSize()}
    , m_linearReadPrefetchThreshold{m_context->options()
          ->getLinearReadPrefetchThreshold()}
    , m_randomReadPrefetchThreshold{m_context->options()
          ->getRandomReadPrefetchThreshold()}
    , m_randomReadPrefetchBlockThreshold{m_context->options()
          ->getRandomReadPrefetchBlockThreshold()}
    , m_randomReadPrefetchClusterWindow{m_context->options()
          ->getRandomReadPrefetchClusterWindow()}
    , m_randomReadPrefetchClusterBlockThreshold{m_context->options()
          ->getRandomReadPrefetchClusterBlockThreshold()}
    , m_randomReadPrefetchEvaluationFrequency{m_context->options()
          ->getRandomReadPrefetchEvaluationFrequency()}
    , m_randomReadPrefetchClusterWindowGrowFactor{m_context->options()
          ->getRandomReadPrefetchClusterWindowGrowFactor()}
    , m_clusterPrefetchThresholdRandom{m_context->options()
          ->isClusterPrefetchThresholdRandom()}
    , m_showOnlyFullReplicas{m_context->options()->showOnlyFullReplicas()}
    , m_showSpaceIdsNotNames{m_context->options()->showSpaceIds()}
    , m_showHardLinkCount{m_context->options()->showHardLinkCount()}
    , m_ioTraceLoggerEnabled{m_context->options()->isIOTraceLoggerEnabled()}
    , m_tagOnCreate{m_context->options()->getOnCreateTag()}
    , m_tagOnModify{m_context->options()->getOnModifyTag()}
    , m_rootUuid{configuration->rootUuid()}
/* clang-format on */
{
    m_nextFuseHandleId = 0;

    m_runInFiber([this]() {
        auto tid = std::this_thread::get_id();
        setFiberThreadId(tid);
        m_metadataCache.setFiberThreadId(tid);
        m_readdirCache->setFiberThreadId(tid);
    });

    m_eventManager.subscribe(*configuration);

    m_metadataCache.setReaddirCache(m_readdirCache);

    m_metadataCache.setVirtualFsHelpersCache(m_virtualFsHelpersCache);

    // Register Archivematica virtual fs
    if (m_context->options()->isArchivematicaModeEnabled())
        m_virtualFsHelpersCache->add("archivematica",
            std::make_shared<
                virtualfs::archivematica::ArchivematicaVirtualFsAdapter>(
                *this, "archivematica"));

    disableSpaces(configuration->disabledSpaces());

    //
    // Registration of force proxy IO cache callbacks
    //
    m_forceProxyIOCache.onAdd([this](const folly::fbstring &uuid) {
        m_fsSubscriptions.subscribeFilePermChanged(uuid);
    });

    m_forceProxyIOCache.onRemove([this](const folly::fbstring &uuid) {
        m_fsSubscriptions.unsubscribeFilePermChanged(uuid);
    });

    //
    // Registration of medatacache events callbacks
    //
    // Called when file attributes are added to the metadata cache
    m_metadataCache.onAdd([this](const folly::fbstring &uuid) {
        m_fsSubscriptions.subscribeFileAttrChanged(uuid);
        m_fsSubscriptions.subscribeFileRemoved(uuid);
        m_fsSubscriptions.subscribeFileRenamed(uuid);

        if (m_showOnlyFullReplicas)
            m_fsSubscriptions.subscribeReplicaStatusChanged(uuid);
    });

    // Called when file is opened
    m_metadataCache.onOpen([this](const folly::fbstring &uuid) {
        m_fsSubscriptions.subscribeFileAttrChanged(uuid);
        m_fsSubscriptions.subscribeFileLocationChanged(uuid);
        m_fsSubscriptions.subscribeFileRemoved(uuid);
        m_fsSubscriptions.subscribeFileRenamed(uuid);
    });

    // Called when file is closed
    m_metadataCache.onRelease([this](const folly::fbstring &uuid) {
        m_fsSubscriptions.unsubscribeFileLocationChanged(uuid);
    });

    // Called when file attributes are dropped from metadata cache
    m_metadataCache.onDropFile([this](const folly::fbstring &uuid) {
        m_fsSubscriptions.unsubscribeFileAttrChanged(uuid);
        m_fsSubscriptions.unsubscribeFileLocationChanged(uuid);
        m_fsSubscriptions.unsubscribeFileRemoved(uuid);
        m_fsSubscriptions.unsubscribeFileRenamed(uuid);
        if (m_showOnlyFullReplicas)
            m_fsSubscriptions.unsubscribeReplicaStatusChanged(uuid);
    });

    // Called when directory attributes are dropped from metadata cache
    m_metadataCache.onDropDirectory([this](const folly::fbstring &uuid) {
        m_fsSubscriptions.unsubscribeFileAttrChanged(uuid);
        m_fsSubscriptions.unsubscribeFileRemoved(uuid);
        m_fsSubscriptions.unsubscribeFileRenamed(uuid);
        if (m_showOnlyFullReplicas)
            m_fsSubscriptions.unsubscribeReplicaStatusChanged(uuid);
    });

    // Called when file is renamed
    m_metadataCache.onRename(
        [this](const folly::fbstring &oldUuid, const folly::fbstring &newUuid) {
            if (oldUuid != newUuid) {
                m_fsSubscriptions.unsubscribeFileAttrChanged(oldUuid);
                m_fsSubscriptions.unsubscribeFileRemoved(oldUuid);
                m_fsSubscriptions.unsubscribeFileRenamed(oldUuid);
                if (m_showOnlyFullReplicas)
                    m_fsSubscriptions.unsubscribeReplicaStatusChanged(oldUuid);
                m_fsSubscriptions.subscribeFileAttrChanged(newUuid);
                m_fsSubscriptions.subscribeFileRemoved(newUuid);
                m_fsSubscriptions.subscribeFileRenamed(newUuid);
                if (m_showOnlyFullReplicas)
                    m_fsSubscriptions.subscribeReplicaStatusChanged(newUuid);

                if (m_fsSubscriptions.unsubscribeFileLocationChanged(oldUuid))
                    m_fsSubscriptions.subscribeFileLocationChanged(newUuid);
            }
            m_onRename(oldUuid, newUuid);
        });

    // Called when file is removed
    m_metadataCache.onMarkDeleted(
        [this](const folly::fbstring &uuid) { m_onMarkDeleted(uuid); });

    if (m_clusterPrefetchThresholdRandom) {
        m_clusterPrefetchDistribution = std::uniform_int_distribution<int>(
            2, m_randomReadPrefetchClusterBlockThreshold);
    }

    if (m_ioTraceLoggerEnabled) {
        m_ioTraceLogger = createIOTraceLogger();
        IOTRACE_GUARD(IOTraceMount, IOTraceLogger::OpType::MOUNT,
            configuration->rootUuid(), 0,
            context->options()->getMountpoint().string());
    }

    m_runInFiber([this, directoryCacheDropAfter]() {
        pruneExpiredDirectories(directoryCacheDropAfter);
    });

    m_context->communicator()->setOnConnectionLostCallback([this]() {
        LOG_DBG(2) << "Called on connection lost callback";
        m_runInFiber([this]() { reset(); });
    });

    m_context->communicator()->setOnReconnectCallback([this]() {
        LOG_DBG(2) << "Called on reconnect callback";
        m_runInFiber([]() {});
    });

    start();
}

FsLogic::~FsLogic()
{
    m_fsSubscriptions.stop();
    m_stopped = true;

    m_directoryCachePruneBaton.post();
    m_context->communicator()->stop();
}

void FsLogic::start()
{
    // Quota initial configuration
    m_eventManager.subscribe(
        events::QuotaExceededSubscription{[=](auto events) {
            m_runInFiber([this, events = std::move(events)] {
                this->disableSpaces(events.back()->spaces());
            });
        }});

    m_stopped = false;
}

void FsLogic::reset()
{
    LOG_DBG(1) << "Resetting internal caches after connection lost...";

    // Close all files
    for (auto &fh : m_fuseFileHandles) {
        fh.second->reset();
    }
    m_fuseFileHandles.clear();
    m_openFileHandles.clear();

    // Clear metadata cache
    m_metadataCache.clear();

    // Cancel all subscriptions
    m_fsSubscriptions.unsubscribeAll();

    m_eventManager.reset();
}

constexpr int FsLogic::MAX_RETRY_COUNT; // NOLINT

struct statvfs FsLogic::statfs(const folly::fbstring &uuid)
{
    LOG_FCALL() << LOG_FARG(uuid);

    assertInFiber();

    constexpr auto kMaxNameLength = 255;
    constexpr auto kBlockSize = 4096;
    constexpr auto kFreeInodes = 10000000;

    auto emulatedFreeSpace = m_context->options()->getEmulateAvailableSpace();

    struct statvfs statinfo = {};

    if (emulatedFreeSpace > 0ULL) {
        statinfo.f_bsize = kBlockSize;
        statinfo.f_frsize = statinfo.f_bsize;
        statinfo.f_blocks = statinfo.f_bfree = statinfo.f_bavail =
            emulatedFreeSpace / statinfo.f_frsize;
        statinfo.f_files = statinfo.f_ffree = kFreeInodes;
    }
    else {
        size_t totalSize = 0;
        size_t totalFreeSize = 0;
        if (uuid != m_rootUuid) {
            // Handle statfs for a specific space or a file inside a space
            messages::fuse::GetFSStats msg{uuid.toStdString()};
            auto fsStats = communicate<messages::fuse::FSStats>(
                std::move(msg), m_providerTimeout);
            totalSize = fsStats.getTotalSize();
            totalFreeSize = fsStats.getTotalFreeSize();
        }
        else {
            // Handle statfs for entire mountpoint by summing storage size
            // on all accessible spaces
            constexpr auto kMaxStatFSSpaceCount = 1024;
            auto spaces = readdir(uuid, kMaxStatFSSpaceCount, 0);
            for (const auto &space : spaces) {
                if (space == "." || space == "..")
                    continue;

                try {
                    auto spaceAttrs = lookup(uuid, space);

                    messages::fuse::GetFSStats msg{
                        spaceAttrs->uuid().toStdString()};
                    auto fsStats = communicate<messages::fuse::FSStats>(
                        std::move(msg), m_providerTimeout);

                    totalSize += fsStats.getTotalSize();
                    totalFreeSize += fsStats.getTotalFreeSize();
                }
                catch (const std::system_error &e) {
                    if (e.code().value() == ENOENT)
                        continue;

                    throw e;
                }
            }
        }

        // block and fragment size
        statinfo.f_frsize = statinfo.f_bsize = kBlockSize;
        // size of fs in f_frsize units
        statinfo.f_blocks = std::ceil(static_cast<double>(totalSize) /
            static_cast<double>(statinfo.f_frsize));
        // free blocks for privileged and unprivileged users
        statinfo.f_bfree = statinfo.f_bavail =
            std::ceil(static_cast<double>(totalFreeSize) /
                static_cast<double>(statinfo.f_frsize));
        // free inodes for privileged and unprivileged users
        statinfo.f_ffree = statinfo.f_favail = kFreeInodes;
    }
    statinfo.f_namemax = kMaxNameLength;

    return statinfo;
}

FileAttrPtr FsLogic::lookup(
    const folly::fbstring &uuid, const folly::fbstring &name)
{
    LOG_FCALL() << LOG_FARG(uuid) << LOG_FARG(name);

    IOTRACE_START()

    assertInFiber();

    FileAttrPtr attr{};

    try {
        attr = m_metadataCache.getAttr(uuid, name);
    }
    catch (std::system_error &e) {
        if (m_metadataCache.getAttr(uuid)->isVirtual()) {
            if (e.code() == std::errc::no_such_file_or_directory) {
                // Force update of the directory contents and lookup
                // the file again
                readdir(uuid, std::numeric_limits<int>::max(), 0);
                attr = m_metadataCache.getAttr(uuid, name);
            }
        }
    }

    if (!attr)
        throw std::system_error(
            std::make_error_code(std::errc::no_such_file_or_directory));

    auto type = attr->type() == FileAttr::FileType::directory ? "d" : "f";
    auto size = attr->size();

    IOTRACE_END(IOTraceLookup, IOTraceLogger::OpType::LOOKUP, uuid, 0, name,
        attr->uuid(), type, size ? *size : 0);

    return attr;
}

FileAttrPtr FsLogic::getattr(const folly::fbstring &uuid)
{
    LOG_FCALL() << LOG_FARG(uuid);

    IOTRACE_GUARD(IOTraceGetAttr, IOTraceLogger::OpType::GETATTR, uuid, 0)

    assertInFiber();

    return m_metadataCache.getAttr(uuid);
}

std::uint64_t FsLogic::opendir(const folly::fbstring &uuid)
{
    LOG_FCALL() << LOG_FARG(uuid);

    assertInFiber();

    // Check that directory exists<commit_msg>VFS-7509 Fixed space-relative symlink getattr size<commit_after>/**
 * @file fsLogic.cc
 * @author <NAME>
 * @copyright (C) 2016 ACK CYFRONET AGH
 * @copyright This software is released under the MIT license cited in
 * 'LICENSE.txt'
 */

#include "fsLogic.h"

#include "communication/communicator.h"
#include "context.h"
#include "fslogic/virtualfs/archivematica.h"
#include "helpers/logging.h"
#include "messages/configuration.h"
#include "messages/fuse/blockSynchronizationRequest.h"
#include "messages/fuse/changeMode.h"
#include "messages/fuse/createDir.h"
#include "messages/fuse/createFile.h"
#include "messages/fuse/deleteFile.h"
#include "messages/fuse/fileAttr.h"
#include "messages/fuse/fileBlock.h"
#include "messages/fuse/fileChildren.h"
#include "messages/fuse/fileChildrenAttrs.h"
#include "messages/fuse/fileCreated.h"
#include "messages/fuse/fileLocation.h"
#include "messages/fuse/fileLocationChanged.h"
#include "messages/fuse/fileOpened.h"
#include "messages/fuse/fileRenamed.h"
#include "messages/fuse/fileRenamedEntry.h"
#include "messages/fuse/fsStats.h"
#include "messages/fuse/fsync.h"
#include "messages/fuse/getFSStats.h"
#include "messages/fuse/getFileChildren.h"
#include "messages/fuse/getFileChildrenAttrs.h"
#include "messages/fuse/getHelperParams.h"
#include "messages/fuse/getXAttr.h"
#include "messages/fuse/helperParams.h"
#include "messages/fuse/listXAttr.h"
#include "messages/fuse/makeFile.h"
#include "messages/fuse/makeLink.h"
#include "messages/fuse/makeSymLink.h"
#include "messages/fuse/openFile.h"
#include "messages/fuse/readSymLink.h"
#include "messages/fuse/release.h"
#include "messages/fuse/removeXAttr.h"
#include "messages/fuse/rename.h"
#include "messages/fuse/setXAttr.h"
#include "messages/fuse/symLink.h"
#include "messages/fuse/syncResponse.h"
#include "messages/fuse/synchronizeBlock.h"
#include "messages/fuse/synchronizeBlockAndComputeChecksum.h"
#include "messages/fuse/truncate.h"
#include "messages/fuse/updateTimes.h"
#include "messages/fuse/xattr.h"
#include "messages/fuse/xattrList.h"
#include "monitoring/monitoring.h"
#include "util/cdmi.h"
#include "util/uuid.h"
#include "util/xattrHelper.h"

#include <boost/icl/interval_set.hpp>
#include <folly/Demangle.h>
#include <folly/Enumerate.h>
#include <folly/Range.h>
#include <folly/ScopeGuard.h>
#include <folly/fibers/Baton.h>
#include <folly/fibers/FiberManager.h>
#include <folly/fibers/ForEach.h>
#include <folly/json.h>
#include <fuse/fuse_lowlevel.h>
#include <openssl/md4.h>

#include "buffering/bufferAgent.h"

#define IOTRACE_START() auto __ioTraceStart = std::chrono::system_clock::now();

#define IOTRACE_END(TraceType, optype, uuid, handleId, ...)                    \
    if (m_ioTraceLoggerEnabled)                                                \
        m_ioTraceLogger->log(TraceType(__ioTraceStart, optype,                 \
            std::chrono::duration_cast<std::chrono::microseconds>(             \
                std::chrono::system_clock::now() - __ioTraceStart),            \
            uuid, handleId, 0, ##__VA_ARGS__));

#define IOTRACE_GUARD(TraceType, optype, uuid, handleId, ...)                  \
    IOTRACE_START()                                                            \
    auto __ioTraceGuard = folly::makeGuard([&] {                               \
        IOTRACE_END(TraceType, optype, uuid, handleId, ##__VA_ARGS__)          \
    });

namespace one {
namespace client {
namespace fslogic {

using namespace std::literals;

namespace {
const std::string kAbsLinkPrefix = "<__onedata_space_id:";
} // namespace

/**
 * Filters given flags set to one of RDONLY, WRONLY or RDWR.
 * Returns RDONLY if flag value is zero.
 * @param Flags value
 */
inline helpers::Flag getOpenFlag(const helpers::FlagsSet &flagsSet)
{
    if (flagsSet.count(one::helpers::Flag::RDONLY) > 0)
        return one::helpers::Flag::RDONLY;
    if (flagsSet.count(one::helpers::Flag::WRONLY) > 0)
        return one::helpers::Flag::WRONLY;
    if (flagsSet.count(one::helpers::Flag::RDWR) > 0)
        return one::helpers::Flag::RDWR;

    return one::helpers::Flag::RDONLY;
}

constexpr auto XATTR_FILE_BLOCKS_MAP_LENGTH = 50;
constexpr auto LINEAR_PREFETCH_THRESHOLD_MATCH_RATIO = 0.9;

inline static folly::fbstring ONE_XATTR(std::string name)
{
    assert(!name.empty());
    return ONE_XATTR_PREFIX + name;
}

FsLogic::FsLogic(std::shared_ptr<Context> context,
    std::shared_ptr<messages::Configuration> configuration,
    std::unique_ptr<cache::HelpersCache> helpersCache,
    unsigned int metadataCacheSize, bool readEventsDisabled,
    bool forceFullblockRead, const std::chrono::seconds providerTimeout,
    const std::chrono::seconds directoryCacheDropAfter,
    std::function<void(folly::Function<void()>)> runInFiber)
    : m_context{context}
    , m_metadataCache{*m_context->communicator(), metadataCacheSize,
          providerTimeout, directoryCacheDropAfter, configuration->rootUuid(),
          m_context->options()->getSpaceNames(),
          m_context->options()->getSpaceIds(),
          m_context->options()->showOnlyFullReplicas(),
          m_context->options()->showHardLinkCount(),
          m_context->options()->showSpaceIds()}
    , m_helpersCache{std::move(helpersCache)}
    , m_virtualFsHelpersCache{std::make_shared<
          virtualfs::VirtualFsHelpersCache>(*this)}
    , m_readdirCache{std::make_shared<cache::ReaddirCache>(
          m_metadataCache, m_context, m_virtualFsHelpersCache, runInFiber)}
    , m_readEventsDisabled{readEventsDisabled}
    , m_forceFullblockRead{forceFullblockRead}
    , m_fsSubscriptions{m_eventManager, m_metadataCache, m_forceProxyIOCache,
          runInFiber}
    , m_nextFuseHandleId{0}
    , m_providerTimeout{providerTimeout}
    , m_storageTimeout{m_context->options()->getStorageTimeout()}
    , m_runInFiber{std::move(runInFiber)} /* clang-format off */
    , m_prefetchModeAsync{m_context->options()->getPrefetchMode() == "async"}
    , m_minPrefetchBlockSize{m_context->options()->getMinimumBlockPrefetchSize()}
    , m_linearReadPrefetchThreshold{m_context->options()
          ->getLinearReadPrefetchThreshold()}
    , m_randomReadPrefetchThreshold{m_context->options()
          ->getRandomReadPrefetchThreshold()}
    , m_randomReadPrefetchBlockThreshold{m_context->options()
          ->getRandomReadPrefetchBlockThreshold()}
    , m_randomReadPrefetchClusterWindow{m_context->options()
          ->getRandomReadPrefetchClusterWindow()}
    , m_randomReadPrefetchClusterBlockThreshold{m_context->options()
          ->getRandomReadPrefetchClusterBlockThreshold()}
    , m_randomReadPrefetchEvaluationFrequency{m_context->options()
          ->getRandomReadPrefetchEvaluationFrequency()}
    , m_randomReadPrefetchClusterWindowGrowFactor{m_context->options()
          ->getRandomReadPrefetchClusterWindowGrowFactor()}
    , m_clusterPrefetchThresholdRandom{m_context->options()
          ->isClusterPrefetchThresholdRandom()}
    , m_showOnlyFullReplicas{m_context->options()->showOnlyFullReplicas()}
    , m_showSpaceIdsNotNames{m_context->options()->showSpaceIds()}
    , m_showHardLinkCount{m_context->options()->showHardLinkCount()}
    , m_ioTraceLoggerEnabled{m_context->options()->isIOTraceLoggerEnabled()}
    , m_tagOnCreate{m_context->options()->getOnCreateTag()}
    , m_tagOnModify{m_context->options()->getOnModifyTag()}
    , m_rootUuid{configuration->rootUuid()}
/* clang-format on */
{
    m_nextFuseHandleId = 0;

    m_runInFiber([this]() {
        auto tid = std::this_thread::get_id();
        setFiberThreadId(tid);
        m_metadataCache.setFiberThreadId(tid);
        m_readdirCache->setFiberThreadId(tid);
    });

    m_eventManager.subscribe(*configuration);

    m_metadataCache.setReaddirCache(m_readdirCache);

    m_metadataCache.setVirtualFsHelpersCache(m_virtualFsHelpersCache);

    // Register Archivematica virtual fs
    if (m_context->options()->isArchivematicaModeEnabled())
        m_virtualFsHelpersCache->add("archivematica",
            std::make_shared<
                virtualfs::archivematica::ArchivematicaVirtualFsAdapter>(
                *this, "archivematica"));

    disableSpaces(configuration->disabledSpaces());

    //
    // Registration of force proxy IO cache callbacks
    //
    m_forceProxyIOCache.onAdd([this](const folly::fbstring &uuid) {
        m_fsSubscriptions.subscribeFilePermChanged(uuid);
    });

    m_forceProxyIOCache.onRemove([this](const folly::fbstring &uuid) {
        m_fsSubscriptions.unsubscribeFilePermChanged(uuid);
    });

    //
    // Registration of medatacache events callbacks
    //
    // Called when file attributes are added to the metadata cache
    m_metadataCache.onAdd([this](const folly::fbstring &uuid) {
        m_fsSubscriptions.subscribeFileAttrChanged(uuid);
        m_fsSubscriptions.subscribeFileRemoved(uuid);
        m_fsSubscriptions.subscribeFileRenamed(uuid);

        if (m_showOnlyFullReplicas)
            m_fsSubscriptions.subscribeReplicaStatusChanged(uuid);
    });

    // Called when file is opened
    m_metadataCache.onOpen([this](const folly::fbstring &uuid) {
        m_fsSubscriptions.subscribeFileAttrChanged(uuid);
        m_fsSubscriptions.subscribeFileLocationChanged(uuid);
        m_fsSubscriptions.subscribeFileRemoved(uuid);
        m_fsSubscriptions.subscribeFileRenamed(uuid);
    });

    // Called when file is closed
    m_metadataCache.onRelease([this](const folly::fbstring &uuid) {
        m_fsSubscriptions.unsubscribeFileLocationChanged(uuid);
    });

    // Called when file attributes are dropped from metadata cache
    m_metadataCache.onDropFile([this](const folly::fbstring &uuid) {
        m_fsSubscriptions.unsubscribeFileAttrChanged(uuid);
        m_fsSubscriptions.unsubscribeFileLocationChanged(uuid);
        m_fsSubscriptions.unsubscribeFileRemoved(uuid);
        m_fsSubscriptions.unsubscribeFileRenamed(uuid);
        if (m_showOnlyFullReplicas)
            m_fsSubscriptions.unsubscribeReplicaStatusChanged(uuid);
    });

    // Called when directory attributes are dropped from metadata cache
    m_metadataCache.onDropDirectory([this](const folly::fbstring &uuid) {
        m_fsSubscriptions.unsubscribeFileAttrChanged(uuid);
        m_fsSubscriptions.unsubscribeFileRemoved(uuid);
        m_fsSubscriptions.unsubscribeFileRenamed(uuid);
        if (m_showOnlyFullReplicas)
            m_fsSubscriptions.unsubscribeReplicaStatusChanged(uuid);
    });

    // Called when file is renamed
    m_metadataCache.onRename(
        [this](const folly::fbstring &oldUuid, const folly::fbstring &newUuid) {
            if (oldUuid != newUuid) {
                m_fsSubscriptions.unsubscribeFileAttrChanged(oldUuid);
                m_fsSubscriptions.unsubscribeFileRemoved(oldUuid);
                m_fsSubscriptions.unsubscribeFileRenamed(oldUuid);
                if (m_showOnlyFullReplicas)
                    m_fsSubscriptions.unsubscribeReplicaStatusChanged(oldUuid);
                m_fsSubscriptions.subscribeFileAttrChanged(newUuid);
                m_fsSubscriptions.subscribeFileRemoved(newUuid);
                m_fsSubscriptions.subscribeFileRenamed(newUuid);
                if (m_showOnlyFullReplicas)
                    m_fsSubscriptions.subscribeReplicaStatusChanged(newUuid);

                if (m_fsSubscriptions.unsubscribeFileLocationChanged(oldUuid))
                    m_fsSubscriptions.subscribeFileLocationChanged(newUuid);
            }
            m_onRename(oldUuid, newUuid);
        });

    // Called when file is removed
    m_metadataCache.onMarkDeleted(
        [this](const folly::fbstring &uuid) { m_onMarkDeleted(uuid); });

    if (m_clusterPrefetchThresholdRandom) {
        m_clusterPrefetchDistribution = std::uniform_int_distribution<int>(
            2, m_randomReadPrefetchClusterBlockThreshold);
    }

    if (m_ioTraceLoggerEnabled) {
        m_ioTraceLogger = createIOTraceLogger();
        IOTRACE_GUARD(IOTraceMount, IOTraceLogger::OpType::MOUNT,
            configuration->rootUuid(), 0,
            context->options()->getMountpoint().string());
    }

    m_runInFiber([this, directoryCacheDropAfter]() {
        pruneExpiredDirectories(directoryCacheDropAfter);
    });

    m_context->communicator()->setOnConnectionLostCallback([this]() {
        LOG_DBG(2) << "Called on connection lost callback";
        m_runInFiber([this]() { reset(); });
    });

    m_context->communicator()->setOnReconnectCallback([this]() {
        LOG_DBG(2) << "Called on reconnect callback";
        m_runInFiber([]() {});
    });

    start();
}

FsLogic::~FsLogic()
{
    m_fsSubscriptions.stop();
    m_stopped = true;

    m_directoryCachePruneBaton.post();
    m_context->communicator()->stop();
}

void FsLogic::start()
{
    // Quota initial configuration
    m_eventManager.subscribe(
        events::QuotaExceededSubscription{[=](auto events) {
            m_runInFiber([this, events = std::move(events)] {
                this->disableSpaces(events.back()->spaces());
            });
        }});

    m_stopped = false;
}

void FsLogic::reset()
{
    LOG_DBG(1) << "Resetting internal caches after connection lost...";

    // Close all files
    for (auto &fh : m_fuseFileHandles) {
        fh.second->reset();
    }
    m_fuseFileHandles.clear();
    m_openFileHandles.clear();

    // Clear metadata cache
    m_metadataCache.clear();

    // Cancel all subscriptions
    m_fsSubscriptions.unsubscribeAll();

    m_eventManager.reset();
}

constexpr int FsLogic::MAX_RETRY_COUNT; // NOLINT

struct statvfs FsLogic::statfs(const folly::fbstring &uuid)
{
    LOG_FCALL() << LOG_FARG(uuid);

    assertInFiber();

    constexpr auto kMaxNameLength = 255;
    constexpr auto kBlockSize = 4096;
    constexpr auto kFreeInodes = 10000000;

    auto emulatedFreeSpace = m_context->options()->getEmulateAvailableSpace();

    struct statvfs statinfo = {};

    if (emulatedFreeSpace > 0ULL) {
        statinfo.f_bsize = kBlockSize;
        statinfo.f_frsize = statinfo.f_bsize;
        statinfo.f_blocks = statinfo.f_bfree = statinfo.f_bavail =
            emulatedFreeSpace / statinfo.f_frsize;
        statinfo.f_files = statinfo.f_ffree = kFreeInodes;
    }
    else {
        size_t totalSize = 0;
        size_t totalFreeSize = 0;
        if (uuid != m_rootUuid) {
            // Handle statfs for a specific space or a file inside a space
            messages::fuse::GetFSStats msg{uuid.toStdString()};
            auto fsStats = communicate<messages::fuse::FSStats>(
                std::move(msg), m_providerTimeout);
            totalSize = fsStats.getTotalSize();
            totalFreeSize = fsStats.getTotalFreeSize();
        }
        else {
            // Handle statfs for entire mountpoint by summing storage size
            // on all accessible spaces
            constexpr auto kMaxStatFSSpaceCount = 1024;
            auto spaces = readdir(uuid, kMaxStatFSSpaceCount, 0);
            for (const auto &space : spaces) {
                if (space == "." || space == "..")
                    continue;

                try {
                    auto spaceAttrs = lookup(uuid, space);

                    messages::fuse::GetFSStats msg{
                        spaceAttrs->uuid().toStdString()};
                    auto fsStats = communicate<messages::fuse::FSStats>(
                        std::move(msg), m_providerTimeout);

                    totalSize += fsStats.getTotalSize();
                    totalFreeSize += fsStats.getTotalFreeSize();
                }
                catch (const std::system_error &e) {
                    if (e.code().value() == ENOENT)
                        continue;

                    throw e;
                }
            }
        }

        // block and fragment size
        statinfo.f_frsize = statinfo.f_bsize = kBlockSize;
        // size of fs in f_frsize units
        statinfo.f_blocks = std::ceil(static_cast<double>(totalSize) /
            static_cast<double>(statinfo.f_frsize));
        // free blocks for privileged and unprivileged users
        statinfo.f_bfree = statinfo.f_bavail =
            std::ceil(static_cast<double>(totalFreeSize) /
                static_cast<double>(statinfo.f_frsize));
        // free inodes for privileged and unprivileged users
        statinfo.f_ffree = statinfo.f_favail = kFreeInodes;
    }
    statinfo.f_namemax = kMaxNameLength;

    return statinfo;
}

FileAttrPtr FsLogic::lookup(
    const folly::fbstring &uuid, const folly::fbstring &name)
{
    LOG_FCALL() << LOG_FARG(uuid) << LOG_FARG(name);

    IOTRACE_START()

    assertInFiber();

    FileAttrPtr attr{};

    try {
        attr = m_metadataCache.getAttr(uuid, name);
    }
    catch (std::system_error &e) {
        if (m_metadataCache.getAttr(uuid)->isVirtual()) {
            if (e.code() == std::errc::no_such_file_or_directory) {
                // Force update of the directory contents and lookup
                // the file again
                readdir(uuid, std::numeric_limits<int>::max(), 0);
                attr = m_metadataCache.getAttr(uuid, name);
            }
        }
    }

    if (!attr)
        throw std::system_error(
            std::make_error_code(std::errc::no_such_file_or_directory));

    auto type = attr->type() == FileAttr::FileType::directory ? "d" : "f";
    auto size = attr->size();

    IOTRACE_END(IOTraceLookup, IOTraceLogger::OpType::LOOKUP, uuid, 0, name,
        attr->uuid(), type, size ? *size : 0);

    return attr;
}

FileAttrPtr FsLogic::getattr(const folly::fbstring &uuid)
{
    LOG_FCALL() << LOG_FARG(uuid);

    IOTRACE_GUARD(IOTraceGetAttr, IOTraceLogger::OpType::GETATTR, uuid, 0)

    assertInFiber();

    auto attr = m_metadataCache.getAttr(uuid);

    if (attr->type() == FileAttr::FileType::symlink) {
        // If this is a symlink, return an attr with the size set to a
        // length of a resolved symlink
        auto symlinkAttr = std::make_shared<FileAttr>(*attr);
        symlinkAttr->size(readlink(uuid).size());
        return symlinkAttr;
    }

    return attr;
}

std::uint64_t FsLogic::opendir(const folly::fbstring &uuid)
{
    LOG_FCALL() << LOG_FARG(uuid);

    assertInFiber();

    // Check that directory exists
{'fraction_non_alphanumeric': 0.11849077541961438, 'fraction_numerical': 0.0028297960882230546, 'mean_word_length': 3.2492042909348107, 'pattern_counts': {'":': 0, '<': 96, '<?xml version=': 0, '>': 216, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6922974', 'n_tokens_mistral': 11323, 'n_tokens_neox': 11049, 'n_words': 1749}