text: ////////////////////////////////////////////////////////////////////////////////
  /// DISCLAIMER
  ///
  /// Copyright 2014-2017 ArangoDB GmbH, Cologne, Germany
  /// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
  ///
  /// Licensed under the Apache License, Version 2.0 (the "License");
  /// you may not use this file except in compliance with the License.
  /// You may obtain a copy of the License at
  ///
  ///     http://www.apache.org/licenses/LICENSE-2.0
  ///
  /// Unless required by applicable law or agreed to in writing, software
  /// distributed under the License is distributed on an "AS IS" BASIS,
  /// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  /// See the License for the specific language governing permissions and
  /// limitations under the License.
  ///
  /// Copyright holder is ArangoDB GmbH, Cologne, Germany
  ///
  /// @author Jan Steemann
  /// @author Jan Christoph Uhde
  ////////////////////////////////////////////////////////////////////////////////
  
  #ifndef ARANGOD_MMFILES_MMFILES_ENGINE_H
  #define ARANGOD_MMFILES_MMFILES_ENGINE_H 1
  
  #include "Basics/Common.h"
  #include "Basics/Mutex.h"
  #include "MMFiles/MMFilesCollectorCache.h"
  #include "MMFiles/MMFilesDatafile.h"
  #include "StorageEngine/StorageEngine.h"
  #include "VocBase/AccessMode.h"
  
  #include <velocypack/Builder.h>
  
  namespace arangodb {
  
  class MMFilesCleanupThread;
  class MMFilesCompactorThread;
  class MMFilesWalAccess;
  
  class MMFilesRecoveryHelper {
   public:
    virtual ~MMFilesRecoveryHelper() = default;
    virtual Result replayMarker(MMFilesMarker const& marker) const = 0;
  };
  
  /// @brief collection file structure
  struct MMFilesEngineCollectionFiles {
    std::vector<std::string> journals;
    std::vector<std::string> compactors;
    std::vector<std::string> datafiles;
    std::vector<std::string> indexes;
  };
  
  class MMFilesEngine final : public StorageEngine {
   public:
    // create the storage engine
    explicit MMFilesEngine(application_features::ApplicationServer& server);
  
    ~MMFilesEngine();
  
    // inherited from ApplicationFeature
    // ---------------------------------
  
    // add the storage engine's specific options to the global list of options
    void collectOptions(std::shared_ptr<options::ProgramOptions>) override;
  
    // validate the storage engine's specific options
    void validateOptions(std::shared_ptr<options::ProgramOptions>) override;
  
    // preparation phase for storage engine. can be used for internal setup.
    // the storage engine must not start any threads here or write any files
    void prepare() override;
  
    // initialize engine
    void start() override;
  
    // flush wal wait for collector
    void stop() override;
  
    bool supportsDfdb() const override { return true; }
  
    bool useRawDocumentPointers() override { return true; }
  
    void cleanupReplicationContexts() override {}
  
    velocypack::Builder getReplicationApplierConfiguration(TRI_vocbase_t& vocbase,
                                                           int& status) override;
    velocypack::Builder getReplicationApplierConfiguration(int& status) override;
    int removeReplicationApplierConfiguration(TRI_vocbase_t& vocbase) override;
    int removeReplicationApplierConfiguration() override;
    int saveReplicationApplierConfiguration(TRI_vocbase_t& vocbase,
                                            velocypack::Slice slice, bool doSync) override;
    int saveReplicationApplierConfiguration(arangodb::velocypack::Slice slice, bool doSync) override;
    // TODO worker-safety
    Result handleSyncKeys(DatabaseInitialSyncer& syncer, LogicalCollection& col,
                          std::string const& keysId) override;
  
    Result createLoggerState(TRI_vocbase_t* vocbase, VPackBuilder& builder) override;
    Result createTickRanges(VPackBuilder& builder) override;
    Result firstTick(uint64_t& tick) override;
    Result lastLogger(TRI_vocbase_t& vocbase,
                      std::shared_ptr<transaction::Context> transactionContext,
                      uint64_t tickStart, uint64_t tickEnd,
                      std::shared_ptr<VPackBuilder>& builderSPtr) override;
    WalAccess const* walAccess() const override;
  
    std::unique_ptr<transaction::Manager> createTransactionManager(transaction::ManagerFeature&) override;
    std::unique_ptr<transaction::ContextData> createTransactionContextData() override;
    std::unique_ptr<TransactionState> createTransactionState(
        TRI_vocbase_t& vocbase, TRI_voc_tick_t, transaction::Options const& options) override;
    std::unique_ptr<TransactionCollection> createTransactionCollection(
        TransactionState& state, TRI_voc_cid_t cid, AccessMode::Type accessType,
        int nestingLevel) override;
  
    // create storage-engine specific collection
    std::unique_ptr<PhysicalCollection> createPhysicalCollection(
        LogicalCollection& collection, velocypack::Slice const& info) override;
  
    // inventory functionality
    // -----------------------
  
    // fill the Builder object with an array of databases that were detected
    // by the storage engine. this method must sort out databases that were not
    // fully created (see "createDatabase" below). called at server start only
    void getDatabases(arangodb::velocypack::Builder& result) override;
  
    // fills the provided builder with information about the collection
    void getCollectionInfo(TRI_vocbase_t& vocbase, TRI_voc_cid_t cid,
                           arangodb::velocypack::Builder& result,
                           bool includeIndexes, TRI_voc_tick_t maxTick) override;
  
    // fill the Builder object with an array of collections (and their
    // corresponding
    // indexes) that were detected by the storage engine. called at server start
    // separately
    // for each database
    int getCollectionsAndIndexes(TRI_vocbase_t& vocbase, arangodb::velocypack::Builder& result,
                                 bool wasCleanShutdown, bool isUpgrade) override;
  
    int getViews(TRI_vocbase_t& vocbase, arangodb::velocypack::Builder& result) override;
  
    // return the path for a collection
    std::string collectionPath(TRI_vocbase_t const& vocbase, TRI_voc_cid_t id) const override {
      return collectionDirectory(vocbase.id(), id);
    }
  
    // database, collection and index management
    // -----------------------------------------
  
    // return the path for all databases
    std::string dataPath() const override {
      return _databasePath;
    }
  
    // return the path for a database
    std::string databasePath(TRI_vocbase_t const* vocbase) const override {
      return databaseDirectory(vocbase->id());
    }
  
    std::string versionFilename(TRI_voc_tick_t id) const override;
  
    void waitForSyncTick(TRI_voc_tick_t tick) override;
  
    /// @brief return a list of the currently open WAL files
    std::vector<std::string> currentWalFiles() const override;
  
    Result flushWal(bool waitForSync, bool waitForCollector, bool writeShutdownFile) override;
  
    void waitForEstimatorSync(std::chrono::milliseconds maxWaitTime) override {}
  
    virtual std::unique_ptr<TRI_vocbase_t> openDatabase(arangodb::velocypack::Slice const& parameters,
                                                        bool isUpgrade, int&) override;
    std::unique_ptr<TRI_vocbase_t> createDatabase(TRI_voc_tick_t id,
                                                  velocypack::Slice const& args,
                                                  int& status) override {
      status = TRI_ERROR_NO_ERROR;
      return createDatabaseMMFiles(id, args);
    }
    int writeCreateDatabaseMarker(TRI_voc_tick_t id, VPackSlice const& slice) override;
  
    void prepareDropDatabase(TRI_vocbase_t& vocbase, bool useWriteMarker, int& status) override;
    Result dropDatabase(TRI_vocbase_t& database) override;
    void waitUntilDeletion(TRI_voc_tick_t id, bool force, int& status) override;
  
    // current recovery state
    RecoveryState recoveryState() noexcept override;
  
    // current recovery tick
    TRI_voc_tick_t recoveryTick() noexcept override;
  
    // start compactor thread and delete files form collections marked as deleted
    void recoveryDone(TRI_vocbase_t& vocbase) override;
  
    Result persistLocalDocumentIds(TRI_vocbase_t& vocbase);
  
    /// @brief regiter a recovery helper
    /// @note not thread-safe on the assumption of static factory registration
    static arangodb::Result registerRecoveryHelper(MMFilesRecoveryHelper const& helper);
  
    /// @brief invoke visitor with each registered recovery helper
    /// @return all recovery registered helpers invoked and returned success
    static bool visitRecoveryHelpers(std::function<bool(MMFilesRecoveryHelper const&)> const& visitor);
  
   private:
    int dropDatabaseMMFiles(TRI_vocbase_t* vocbase);
    std::unique_ptr<TRI_vocbase_t> createDatabaseMMFiles(TRI_voc_tick_t id,
                                                         velocypack::Slice const& data);
  
   public:
    // asks the storage engine to create a collection as specified in the VPack
    // Slice object and persist the creation info. It is guaranteed by the server
    // that no other active collection with the same name and id exists in the
    // same
    // database when this function is called. If this operation fails somewhere in
    // the middle, the storage engine is required to fully clean up the creation
    // and throw only then, so that subsequent collection creation requests will
    // not fail.
    // the WAL entry for the collection creation will be written *after* the call
    // to "createCollection" returns
    std::string createCollection(TRI_vocbase_t& vocbase,
                                 LogicalCollection const& collection) override;
  
    // asks the storage engine to persist the collection.
    // After this call the collection is persisted over recovery.
    // This call will write wal markers.
    arangodb::Result persistCollection(TRI_vocbase_t& vocbase,
                                       LogicalCollection const& collection) override;
  
    // asks the storage engine to drop the specified collection and persist the
    // deletion info. Note that physical deletion of the collection data must not
    // be carried out by this call, as there may
    // still be readers of the collection's data.
    // This call will write the WAL entry for collection deletion
    arangodb::Result dropCollection(TRI_vocbase_t& vocbase, LogicalCollection& collection) override;
  
    // perform a physical deletion of the collection
    // After this call data of this collection is corrupted, only perform if
    // assured that no one is using the collection anymore
    void destroyCollection(TRI_vocbase_t& vocbase, LogicalCollection& collection) override;
  
    // asks the storage engine to change properties of the collection as specified
    // in
    // the VPack Slice object and persist them. If this operation fails
    // somewhere in the middle, the storage engine is required to fully revert the
    // property changes and throw only then, so that subsequent operations will
    // not fail.
    // the WAL entry for the propery change will be written *after* the call
    // to "changeCollection" returns
    void changeCollection(TRI_vocbase_t& vocbase,
                          LogicalCollection const& collection, bool doSync) override;
  
    // asks the storage engine to persist renaming of a collection
    // This will write a renameMarker if not in recovery
    arangodb::Result renameCollection(TRI_vocbase_t& vocbase, LogicalCollection const& collection,
                                      std::string const& oldName) override;
  
    // asks the storage engine to create an index as specified in the VPack
    // Slice object and persist the creation info. The database id, collection id
    // and index data are passed in the Slice object. Note that this function
    // is not responsible for inserting the individual documents into the index.
    // If this operation fails somewhere in the middle, the storage engine is
    // required
    // to fully clean up the creation and throw only then, so that subsequent
    // index
    // creation requests will not fail.
    // the WAL entry for the index creation will be written *after* the call
    // to "createIndex" returns
    void createIndex(TRI_vocbase_t& vocbase, TRI_voc_cid_t collectionId,
                     TRI_idx_iid_t id, arangodb::velocypack::Slice const& data);
  
    // asks the storage engine to drop the specified index and persist the
    // deletion
    // info. Note that physical deletion of the index must not be carried out by
    // this call,
    // as there may still be users of the index. It is recommended that this
    // operation
    // only sets a deletion flag for the index but let's an async task perform
    // the actual deletion.
    // the WAL entry for index deletion will be written *after* the call
    // to "dropIndex" returns
    void dropIndex(TRI_vocbase_t* vocbase, TRI_voc_cid_t collectionId, TRI_idx_iid_t id);
  
    void dropIndexWalMarker(TRI_vocbase_t* vocbase, TRI_voc_cid_t collectionId,
                            arangodb::velocypack::Slice const& data,
                            bool writeMarker, int&);
  
    void unloadCollection(TRI_vocbase_t& vocbase, LogicalCollection& collection) override;
  
    arangodb::Result changeView(TRI_vocbase_t& vocbase,
                                arangodb::LogicalView const&, bool doSync) override;
  
    arangodb::Result createView(TRI_vocbase_t& vocbase, TRI_voc_cid_t id,
                                arangodb::LogicalView const& view) override;
  
    void getViewProperties(TRI_vocbase_t& vocbase, LogicalView const& view,
                           VPackBuilder& builder) override;
  
    arangodb::Result dropView(TRI_vocbase_t const& vocbase, LogicalView const& view) override;
  
    void destroyView(TRI_vocbase_t const& vocbase, LogicalView const& view) noexcept override;
  
    std::string createViewDirectoryName(std::string const& basePath, TRI_voc_cid_t id);
  
    void saveViewInfo(TRI_vocbase_t const& vocbase, LogicalView const& view, bool sync) const;
  
    void signalCleanup(TRI_vocbase_t& vocbase) override;
  
    /// @brief scans a collection and locates all files
    MMFilesEngineCollectionFiles scanCollectionDirectory(std::string const& path);
  
    /// @brief remove data of expired compaction blockers
    bool cleanupCompactionBlockers(TRI_vocbase_t* vocbase);
  
    /// @brief insert a compaction blocker
    int insertCompactionBlocker(TRI_vocbase_t* vocbase, double ttl, TRI_voc_tick_t& id);
  
    /// @brief touch an existing compaction blocker
    int extendCompactionBlocker(TRI_vocbase_t* vocbase, TRI_voc_tick_t id, double ttl);
  
    /// @brief remove an existing compaction blocker
    int removeCompactionBlocker(TRI_vocbase_t* vocbase, TRI_voc_tick_t id);
  
    /// @brief a callback function that is run while it is guaranteed that there
    /// is no compaction ongoing
    void preventCompaction(TRI_vocbase_t* vocbase,
                           std::function<void(TRI_vocbase_t*)> const& callback);
  
    /// @brief a callback function that is run there is no compaction ongoing
    bool tryPreventCompaction(TRI_vocbase_t* vocbase,
                              std::function<void(TRI_vocbase_t*)> const& callback,
                              bool checkForActiveBlockers);
  
    int shutdownDatabase(TRI_vocbase_t& vocbase) override;
  
    int openCollection(TRI_vocbase_t* vocbase, LogicalCollection* collection, bool ignoreErrors);
  
    /// @brief Add engine-specific optimizer rules
    void addOptimizerRules(aql::OptimizerRulesFeature&) override;
  
    /// @brief Add engine-specific V8 functions
    void addV8Functions() override;
  
    /// @brief Add engine-specific REST handlers
    void addRestHandlers(rest::RestHandlerFactory&) override;
  
    /// @brief transfer markers into a collection
    int transferMarkers(LogicalCollection* collection, MMFilesCollectorCache*,
                        MMFilesOperationsType const&, uint64_t& numBytesTransferred);
  
    std::string viewDirectory(TRI_voc_tick_t databaseId, TRI_voc_cid_t viewId) const;
  
    virtual TRI_voc_tick_t currentTick() const override;
    virtual TRI_voc_tick_t releasedTick() const override;
    virtual void releaseTick(TRI_voc_tick_t) override;
  
    void disableCompaction();
    void enableCompaction();
    bool isCompactionDisabled() const;
  
    /// @brief whether the engine is currently running an upgrade procedure
    bool upgrading() const;
  
   private:
    velocypack::Builder getReplicationApplierConfiguration(std::string const& filename,
                                                           int& status);
    int removeReplicationApplierConfiguration(std::string const& filename);
    int saveReplicationApplierConfiguration(std::string const& filename,
                                            arangodb::velocypack::Slice, bool doSync);
  
    /// @brief: check the initial markers in a datafile
    bool checkDatafileHeader(MMFilesDatafile* datafile, std::string const& filename) const;
  
    /// @brief transfer markers into a collection, worker function
    int transferMarkersWorker(LogicalCollection* collection, MMFilesCollectorCache*,
                              MMFilesOperationsType const&, uint64_t& numBytesTransferred);
  
    /// @brief sync the active journal of a collection
    int syncJournalCollection(LogicalCollection* collection);
  
    /// @brief get the next free position for a new marker of the specified size
    char* nextFreeMarkerPosition(LogicalCollection* collection, TRI_voc_tick_t,
                                 MMFilesMarkerType, uint32_t, MMFilesCollectorCache*);
  
    /// @brief set the tick of a marker and calculate its CRC value
    void finishMarker(char const*, char*, LogicalCollection* collection,
                      TRI_voc_tick_t, MMFilesCollectorCache*);
  
    void verifyDirectories();
    std::vector<std::string> getDatabaseNames() const;
  
    /// @brief create a new database directory
    int createDatabaseDirectory(TRI_voc_tick_t id, std::string const& name);
  
    /// @brief save a parameter.json file for a database
    int saveDatabaseParameters(TRI_voc_tick_t id, std::string const& name, bool deleted);
  
    arangodb::velocypack::Builder databaseToVelocyPack(TRI_voc_tick_t databaseId,
                                                       std::string const& name,
                                                       bool deleted) const;
  
    std::string databaseDirectory(TRI_voc_tick_t databaseId) const;
    std::string databaseParametersFilename(TRI_voc_tick_t databaseId) const;
    std::string collectionDirectory(TRI_voc_tick_t databaseId, TRI_voc_cid_t collectionId) const;
    std::string collectionParametersFilename(TRI_voc_tick_t databaseId,
                                             TRI_voc_cid_t collectionId) const;
    std::string viewParametersFilename(TRI_voc_tick_t databaseId, TRI_voc_cid_t viewId) const;
    std::string indexFilename(TRI_voc_tick_t databaseId, TRI_voc_cid_t collectionId,
                              TRI_idx_iid_t indexId) const;
    std::string indexFilename(TRI_idx_iid_t indexId) const;
  
    /// @brief open an existing database. internal function
    std::unique_ptr<TRI_vocbase_t> openExistingDatabase(TRI_voc_tick_t id,
                                                        VPackSlice args,
                                                        bool wasCleanShutdown, bool isUpgrade);
  
    /// @brief note the maximum local tick
    void noteTick(TRI_voc_tick_t tick) {
      if (tick > _maxTick) {
        _maxTick = tick;
      }
    }
  
    /// @brief physically erases the database directory
    int dropDatabaseDirectory(std::string const& path);
  
    /// @brief iterate over a set of datafiles, identified by filenames
    /// note: the files will be opened and closed
    bool iterateFiles(std::vector<std::string> const& files);
  
    /// @brief find the maximum tick in a collection's journals
    bool findMaxTickInJournals(std::string const& path);
  
    /// @brief create a full directory name for a collection
    std::string createCollectionDirectoryName(std::string const& basePath, TRI_voc_cid_t cid);
  
    void registerCollectionPath(TRI_voc_tick_t databaseId, TRI_voc_cid_t id,
                                std::string const& path);
    void unregisterCollectionPath(TRI_voc_tick_t databaseId, TRI_voc_cid_t id);
    void registerViewPath(TRI_voc_tick_t databaseId, TRI_voc_cid_t id, std::string const& path);
    void unregisterViewPath(TRI_voc_tick_t databaseId, TRI_voc_cid_t id);
  
    void saveCollectionInfo(TRI_vocbase_t* vocbase, TRI_voc_cid_t id,
                            arangodb::LogicalCollection const* parameters,
                            bool forceSync) const;
  
    arangodb::velocypack::Builder loadCollectionInfo(TRI_vocbase_t* vocbase,
                                                     std::string const& path);
    arangodb::velocypack::Builder loadViewInfo(TRI_vocbase_t* vocbase, std::string const& path);
  
    // start the cleanup thread for the database
    int startCleanup(TRI_vocbase_t* vocbase);
    // stop and delete the cleanup thread for the database
    int stopCleanup(TRI_vocbase_t* vocbase);
  
    // start the compactor thread for the database
    int startCompactor(TRI_vocbase_t& vocbase);
    // signal the compactor thread to stop
    int beginShutdownCompactor(TRI_vocbase_t* vocbase);
    // stop and delete the compactor thread for the database
    int stopCompactor(TRI_vocbase_t* vocbase);
  
    // stop and delete the compactor and cleanup threads for all databases
    void stopAllThreads();
  
    /// @brief writes a drop-database marker into the log
    int writeDropMarker(TRI_voc_tick_t id, std::string const& name);
  
    /// @brief wait until everything written to the WAL got completely
    /// synced
    void waitForSyncTimeout(double maxWait);
  
   public:
    static std::string const EngineName;
    static std::string const FeatureName;
  
   private:
    std::string _basePath;
    std::string _databasePath;
    bool _isUpgrade;
    TRI_voc_tick_t _maxTick;
    /// @brief Local wal access abstraction
    std::unique_ptr<MMFilesWalAccess> _walAccess;
    std::vector<std::pair<std::string, std::string>> _deleted;
  
    arangodb::basics::ReadWriteLock mutable _releaseLock;
    TRI_voc_tick_t _releasedTick;
  
    arangodb::basics::ReadWriteLock mutable _pathsLock;
    std::unordered_map<TRI_voc_tick_t, std::unordered_map<TRI_voc_cid_t, std::string>> _collectionPaths;
    std::unordered_map<TRI_voc_tick_t, std::unordered_map<TRI_voc_cid_t, std::string>> _viewPaths;
  
    struct CompactionBlocker {
      CompactionBlocker(TRI_voc_tick_t id, double expires)
          : _id(id), _expires(expires) {}
      CompactionBlocker() = delete;
  
      TRI_voc_tick_t _id;
      double _expires;
    };
  
    // lock for compaction blockers
    arangodb::basics::ReadWriteLock mutable _compactionBlockersLock;
    // cross-database map of compaction blockers, protected by
    // _compactionBlockersLock
    std::unordered_map<TRI_vocbase_t*, std::vector<CompactionBlocker>> _compactionBlockers;
  
    // lock for threads
    arangodb::Mutex _threadsLock;
    // per-database compactor threads, protected by _threadsLock
    std::unordered_map<TRI_vocbase_t*, std::shared_ptr<MMFilesCompactorThread>> _compactorThreads;
    // per-database cleanup threads, protected by _threadsLock
    std::unordered_map<TRI_vocbase_t*, std::shared_ptr<MMFilesCleanupThread>> _cleanupThreads;
  
    // whether or not compaction is disabled (> 0) or not (== 0)
    // can be called multiple times. the last one to set this to 0 again will
    // enable compaction again
    std::atomic<uint64_t> _compactionDisabled;
  
    // whether the engine is currently running an upgrade procedure
    std::atomic<bool> _upgrading{false};
  };
  
  }  // namespace arangodb
  
  #endif
  