FileMgr.h

Go to the documentation of this file.

/*
 * Copyright 2022 HEAVY.AI, Inc.
 *
 * 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.
 */

#pragma once

#include <future>
#include <iostream>
#include <map>
#include <mutex>
#include <set>
#include <vector>

#include "DataMgr/AbstractBuffer.h"
#include "DataMgr/AbstractBufferMgr.h"
#include "DataMgr/FileMgr/Epoch.h"
#include "DataMgr/FileMgr/FileBuffer.h"
#include "DataMgr/FileMgr/FileInfo.h"
#include "DataMgr/FileMgr/Page.h"
#include "Fragmenter/FragmentDefaultValues.h"
#include "Shared/heavyai_shared_mutex.h"

using namespace Data_Namespace;

namespace boost {
namespace filesystem {
class directory_iterator;
}
}  // namespace boost

namespace File_Namespace {
class GlobalFileMgr;  // forward declaration
using PageSizeFileMMap = std::multimap<size_t, int32_t>;

using Chunk = FileBuffer;

using ChunkKeyToChunkMap = std::map<ChunkKey, FileBuffer*>;
using TablePair = std::pair<const int32_t, const int32_t>;

struct FileMetadata {
  int32_t file_id;
  std::string file_path;
  size_t page_size;
  size_t file_size;
  size_t num_pages;
  bool is_data_file;
};

struct StorageStats {
  int32_t epoch{0};
  int32_t epoch_floor{0};
  uint32_t metadata_file_count{0};
  uint64_t total_metadata_file_size{0};
  uint64_t total_metadata_page_count{0};
  std::optional<uint64_t> total_free_metadata_page_count{};
  uint32_t data_file_count{0};
  uint64_t total_data_file_size{0};
  uint64_t total_data_page_count{0};
  std::optional<uint64_t> total_free_data_page_count{};
  std::optional<uint32_t> fragment_count{};

  StorageStats() = default;
  StorageStats(const StorageStats& storage_stats) = default;
  virtual ~StorageStats() = default;
};

struct OpenFilesResult {
  std::vector<HeaderInfo> header_infos;
  int32_t max_file_id;
  std::string compaction_status_file_name;
};

// Page header size is serialized/deserialized as an int.
using PageHeaderSizeType = int32_t;

struct PageMapping {
  PageMapping() {}

  PageMapping(int32_t source_file_id,
              size_t source_page_num,
              PageHeaderSizeType source_page_header_size,
              int32_t destination_file_id,
              size_t destination_page_num)
      : source_file_id(source_file_id)
      , source_page_num(source_page_num)
      , source_page_header_size(source_page_header_size)
      , destination_file_id(destination_file_id)
      , destination_page_num(destination_page_num) {}

  int32_t source_file_id;
  size_t source_page_num;
  PageHeaderSizeType source_page_header_size;
  int32_t destination_file_id;
  size_t destination_page_num;
};

class FileMgr : public AbstractBufferMgr {  // implements
  friend class GlobalFileMgr;

 public:
  FileMgr(const int32_t device_id,
          GlobalFileMgr* gfm,
          const TablePair file_mgr_key,
          const int32_t max_rollback_epochs = -1,
          const size_t num_reader_threads = 0,
          const int32_t epoch = -1);

  // used only to initialize enough to drop or to get basic metadata
  FileMgr(const int32_t device_id,
          GlobalFileMgr* gfm,
          const TablePair file_mgr_key,
          const bool run_core_init);

  FileMgr(GlobalFileMgr* gfm, std::string basePath);

  ~FileMgr() override;

  StorageStats getStorageStats() const;
  FileBuffer* createBuffer(const ChunkKey& key,
                           size_t pageSize = 0,
                           const size_t numBytes = 0) override;

  bool isBufferOnDevice(const ChunkKey& key) override;
  // Purge == true means delete the data chunks -
  // can't undelete and revert to previous
  // state - reclaims disk space for chunk
  void deleteBuffer(const ChunkKey& key, const bool purge = true) override;
  void deleteBuffersWithPrefix(const ChunkKey& keyPrefix,
                               const bool purge = true) override;

  FileBuffer* getBuffer(const ChunkKey& key, const size_t numBytes = 0) override;

  void fetchBuffer(const ChunkKey& key,
                   AbstractBuffer* destBuffer,
                   const size_t numBytes) override;

  FileBuffer* putBuffer(const ChunkKey& key,
                        AbstractBuffer* d,
                        const size_t numBytes = 0) override;

  // Buffer API
  AbstractBuffer* alloc(const size_t numBytes) override;
  void free(AbstractBuffer* buffer) override;
  virtual Page requestFreePage(size_t pagesize, const bool isMetadata);

  inline MgrType getMgrType() override { return FILE_MGR; };
  inline std::string getStringMgrType() override { return ToString(FILE_MGR); }
  inline std::string printSlabs() override { return "Not Implemented"; }
  inline size_t getMaxSize() override { return 0; }
  inline size_t getInUseSize() override { return 0; }
  inline size_t getAllocated() override { return 0; }
  inline bool isAllocationCapped() override { return false; }

  inline FileInfo* getFileInfoForFileId(const int32_t fileId) const {
    return files_.at(fileId);
  }

  FileMetadata getMetadataForFile(
      const boost::filesystem::directory_iterator& fileIterator) const;

  void init(const size_t num_reader_threads, const int32_t epochOverride);
  void init(const std::string& dataPathToConvertFrom, const int32_t epochOverride);

  void copyPage(Page& srcPage,
                FileMgr* destFileMgr,
                Page& destPage,
                const size_t reservedHeaderSize,
                const size_t numBytes,
                const size_t offset);

  void requestFreePages(size_t npages,
                        size_t pagesize,
                        std::vector<Page>& pages,
                        const bool isMetadata);

  void getChunkMetadataVecForKeyPrefix(ChunkMetadataVector& chunkMetadataVec,
                                       const ChunkKey& keyPrefix) override;

  bool hasChunkMetadataForKeyPrefix(const ChunkKey& keyPrefix);

  void checkpoint() override;
  void checkpoint(const int32_t db_id, const int32_t tb_id) override {
    LOG(FATAL) << "Operation not supported, api checkpoint() should be used instead";
  }
  inline virtual int32_t epoch(int32_t db_id, int32_t tb_id) const { return epoch(); }

  inline int32_t epochFloor() const { return static_cast<int32_t>(epoch_.floor()); }

  inline int32_t incrementEpoch() {
    int32_t newEpoch = epoch_.increment();
    epochIsCheckpointed_ = false;
    // We test for error here instead of in Epoch::increment so we can log FileMgr
    // metadata
    if (newEpoch > Epoch::max_allowable_epoch()) {
      LOG(FATAL) << "Epoch for table (" << fileMgrKey_.first << ", " << fileMgrKey_.second
                 << ") greater than maximum allowed value of "
                 << Epoch::max_allowable_epoch() << ".";
    }
    return newEpoch;
  }

  inline int32_t lastCheckpointedEpoch() const {
    return epoch() - (epochIsCheckpointed_ ? 0 : 1);
  }

  inline void resetEpochFloor() { epoch_.floor(epoch_.ceiling()); }

  inline int32_t maxRollbackEpochs() { return maxRollbackEpochs_; }

  inline size_t getNumReaderThreads() { return num_reader_threads_; }

  FILE* getFileForFileId(const int32_t fileId);

  size_t getNumChunks() override;
  size_t getNumUsedMetadataPagesForChunkKey(const ChunkKey& chunkKey) const;

  ChunkKeyToChunkMap chunkIndex_;  
                                   // #TM Not sure if we need this below
  int32_t getDBVersion() const;
  bool getDBConvert() const;
  void createTopLevelMetadata();  // create metadata shared by all tables of all DBs
  inline std::string getFileMgrBasePath() const { return fileMgrBasePath_; }
  virtual void closeRemovePhysical();

  void removeTableRelatedDS(const int32_t db_id, const int32_t table_id) override;

  virtual void free_page(std::pair<FileInfo*, int32_t>&& page);
  inline virtual bool hasFileMgrKey() const { return true; }
  const TablePair get_fileMgrKey() const { return fileMgrKey_; }

  boost::filesystem::path getFilePath(const std::string& file_name) const;

  // Visible for use in unit tests.
  void writePageMappingsToStatusFile(const std::vector<PageMapping>& page_mappings);

  // Visible for use in unit tests.
  void renameCompactionStatusFile(const char* const from_status,
                                  const char* const to_status);

  void compactFiles();

  virtual bool updatePageIfDeleted(FileInfo* file_info,
                                   ChunkKey& chunk_key,
                                   int32_t contingent,
                                   int32_t page_epoch,
                                   int32_t page_num);

  inline virtual bool failOnReadError() const { return true; }

  inline size_t getPageSize() const { return page_size_; }
  inline size_t getMetadataPageSize() const { return metadata_page_size_; }

  // Used to describe the manager in logging and error messages.
  virtual std::string describeSelf() const;

  static constexpr size_t DEFAULT_NUM_PAGES_PER_DATA_FILE{256};
  static constexpr size_t DEFAULT_NUM_PAGES_PER_METADATA_FILE{4096};

  // Name of files that indicate the different statuses/phases of data compaction.
  static constexpr char const* COPY_PAGES_STATUS{"pending_data_compaction_0"};
  static constexpr char const* UPDATE_PAGE_VISIBILITY_STATUS{"pending_data_compaction_1"};
  static constexpr char const* DELETE_EMPTY_FILES_STATUS{"pending_data_compaction_2"};

  // Methods that enable override of number of pages per data/metadata file
  // for use in unit tests.
  static void setNumPagesPerDataFile(size_t num_pages);
  static void setNumPagesPerMetadataFile(size_t num_pages);

  static void renameAndSymlinkLegacyFiles(const std::string& table_data_dir);

  static constexpr char LEGACY_EPOCH_FILENAME[] = "epoch";
  static constexpr char EPOCH_FILENAME[] = "epoch_metadata";
  static constexpr char DB_META_FILENAME[] = "dbmeta";
  static constexpr char FILE_MGR_VERSION_FILENAME[] = "filemgr_version";
  static constexpr int32_t INVALID_VERSION = -1;

 protected:
  // Used to initialize CachingFileMgr.
  FileMgr(const size_t defaultPageSize, const size_t defaultMetadataPageSize);

  int32_t maxRollbackEpochs_;
  std::string fileMgrBasePath_;  
  std::map<int32_t, FileInfo*>
      files_;                   
  PageSizeFileMMap fileIndex_;  
  size_t num_reader_threads_;   
  unsigned nextFileId_;         
  int32_t db_version_;          
  int32_t fileMgrVersion_;
  const int32_t latestFileMgrVersion_{2};
  FILE* DBMetaFile_ = nullptr;  
  std::mutex getPageMutex_;
  mutable heavyai::shared_mutex chunkIndexMutex_;
  mutable heavyai::shared_mutex files_rw_mutex_;

  mutable heavyai::shared_mutex mutex_free_page_;
  std::vector<std::pair<FileInfo*, int32_t>> free_pages_;
  bool isFullyInitted_{false};

  static size_t num_pages_per_data_file_;
  static size_t num_pages_per_metadata_file_;

  FileInfo* createFile(const size_t pageSize, const size_t numPages);
  FileInfo* openExistingFile(const std::string& path,
                             const int32_t fileId,
                             const size_t pageSize,
                             const size_t numPages,
                             std::vector<HeaderInfo>& headerVec);
  void createEpochFile(const std::string& epochFileName);
  int32_t openAndReadLegacyEpochFile(const std::string& epochFileName);
  void openAndReadEpochFile(const std::string& epochFileName);
  void writeAndSyncEpochToDisk();
  void setEpoch(const int32_t newEpoch);  // resets current value of epoch at startup
  int32_t readVersionFromDisk(const std::string& versionFileName) const;
  void writeAndSyncVersionToDisk(const std::string& versionFileName,
                                 const int32_t version);
  void processFileFutures(std::vector<std::future<std::vector<HeaderInfo>>>& file_futures,
                          std::vector<HeaderInfo>& headerVec);
  virtual FileBuffer* createBufferUnlocked(const ChunkKey& key,
                                           size_t pageSize = 0,
                                           const size_t numBytes = 0);
  virtual FileBuffer* createBufferFromHeaders(
      const ChunkKey& key,
      const std::vector<HeaderInfo>::const_iterator& headerStartIt,
      const std::vector<HeaderInfo>::const_iterator& headerEndIt);

  // Migration functions
  void migrateToLatestFileMgrVersion();
  void migrateEpochFileV0();
  void migrateLegacyFilesV1();

  OpenFilesResult openFiles();

  void clearFileInfos();

  // Data compaction methods
  void copySourcePageForCompaction(const Page& source_page,
                                   FileInfo* destination_file_info,
                                   std::vector<PageMapping>& page_mappings,
                                   std::set<Page>& touched_pages);
  int32_t copyPageWithoutHeaderSize(const Page& source_page,
                                    const Page& destination_page);
  void sortAndCopyFilePagesForCompaction(size_t page_size,
                                         std::vector<PageMapping>& page_mappings,
                                         std::set<Page>& touched_pages);
  void updateMappedPagesVisibility(const std::vector<PageMapping>& page_mappings);
  void deleteEmptyFiles();
  void resumeFileCompaction(const std::string& status_file_name);
  std::vector<PageMapping> readPageMappingsFromStatusFile();

  // For testing purposes only
  FileMgr(const int epoch);

  void closePhysicalUnlocked();
  void syncFilesToDisk();
  void freePages();
  void initializeNumThreads(size_t num_reader_threads = 0);
  virtual FileBuffer* allocateBuffer(const size_t page_size,
                                     const ChunkKey& key,
                                     const size_t num_bytes = 0);
  virtual FileBuffer* allocateBuffer(
      const ChunkKey& key,
      const std::vector<HeaderInfo>::const_iterator& headerStartIt,
      const std::vector<HeaderInfo>::const_iterator& headerEndIt);
  virtual ChunkKeyToChunkMap::iterator deleteBufferUnlocked(
      const ChunkKeyToChunkMap::iterator chunk_it,
      const bool purge = true);
  virtual FileBuffer* getBufferUnlocked(const ChunkKey& key,
                                        const size_t numBytes = 0) const;

 private:
  void rollOffOldData(const int32_t epochCeiling, const bool shouldCheckpoint);
  void freePagesBeforeEpoch(const int32_t min_epoch);
  void freePagesBeforeEpochUnlocked(const int32_t min_epoch,
                                    const ChunkKeyToChunkMap::iterator lower_bound,
                                    const ChunkKeyToChunkMap::iterator upper_bound);
  FileBuffer* getOrCreateBuffer(const ChunkKey& key);
  bool coreInit();
  inline int32_t epoch() const { return static_cast<int32_t>(epoch_.ceiling()); }
  void writeDirtyBuffers();

  void setDataAndMetadataFileStats(StorageStats& storage_stats) const;
  uint32_t getFragmentCount() const;

  GlobalFileMgr* gfm_;  
  TablePair fileMgrKey_;

  Epoch epoch_;
  bool epochIsCheckpointed_ = true;
  FILE* epochFile_ = nullptr;

 protected:
  // gfm_ needs to be defined before the page_size_ because it may be used to
  // initialize it.  However, we also want it to remain private, as the CachingFileMgr
  // should not be allowed to access it (it will be null). page_size_ should be
  // be protected.
  const size_t page_size_;
  const size_t metadata_page_size_;
};

}  // namespace File_Namespace