CachingFileMgr.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 <optional>
#include <ostream>

#include "DataMgr/ForeignStorage/CacheEvictionAlgorithms/LRUEvictionAlgorithm.h"
#include "FileMgr.h"
#include "Shared/File.h"
#include "Shared/SysDefinitions.h"

namespace File_Namespace {

enum class DiskCacheLevel { none, fsi, non_fsi, all };

std::ostream& operator<<(std::ostream& os, DiskCacheLevel disk_cache_level);

struct DiskCacheConfig {
  // Note the suffix UL is not portable. 'long' i a different size
  // on nix vs windows.  Hense the explicit cast to long.
  static constexpr size_t DEFAULT_MAX_SIZE{(size_t)1024 * (size_t)1024 * (size_t)1024 *
                                           (size_t)100};  // 100G default
  std::string path;
  DiskCacheLevel enabled_level = DiskCacheLevel::none;
  size_t num_reader_threads = 0;
  size_t size_limit = DEFAULT_MAX_SIZE;
  size_t page_size = DEFAULT_PAGE_SIZE;
  size_t meta_page_size = DEFAULT_METADATA_PAGE_SIZE;
  inline bool isEnabledForMutableTables() const {
    return enabled_level == DiskCacheLevel::non_fsi ||
           enabled_level == DiskCacheLevel::all;
  }
  inline bool isEnabledForFSI() const {
    return enabled_level == DiskCacheLevel::fsi || enabled_level == DiskCacheLevel::all;
  }
  inline bool isEnabled() const { return enabled_level != DiskCacheLevel::none; }
  std::string dump() const {
    std::stringstream ss;
    ss << "DiskCacheConfig(path = " << path << ", level = " << levelAsString()
       << ", threads = " << num_reader_threads << ", size limit = " << size_limit
       << ", page size = " << page_size << ")";
    return ss.str();
  }
  std::string levelAsString() const {
    switch (enabled_level) {
      case DiskCacheLevel::none:
        return "none";
      case DiskCacheLevel::fsi:
        return "fsi";
      case DiskCacheLevel::non_fsi:
        return "non_fsi";
      case DiskCacheLevel::all:
        return "all";
    }
    return "";
  }
  static std::string getDefaultPath(const std::string& base_path) {
    return base_path + "/" + shared::kDefaultDiskCacheDirName;
  }
};

inline std::string get_dir_name_for_table(int db_id, int tb_id) {
  std::stringstream file_name;
  file_name << "table_" << db_id << "_" << tb_id << "/";
  return file_name.str();
}

// Class to control access to the table-specific directories and data created inside a
// CachingFileMgr.
class TableFileMgr {
 public:
  TableFileMgr(const std::string& table_path);
  ~TableFileMgr() { close(epoch_file_); }

  void incrementEpoch();

  void writeAndSyncEpochToDisk();

  int32_t getEpoch() const;

  void removeDiskContent() const;

  size_t getReservedSpace() const;

  void deleteWrapperFile() const;

  void writeWrapperFile(const std::string& doc) const;

  bool hasWrapperFile() const;

 private:
  std::string table_path_;
  std::string epoch_file_path_;
  std::string wrapper_file_path_;
  Epoch epoch_;
  bool is_checkpointed_ = true;
  FILE* epoch_file_ = nullptr;

  mutable heavyai::shared_mutex table_mutex_;
};

// Extension of FileBuffer with restricted behaviour.
class CachingFileBuffer : public FileBuffer {
 public:
  using FileBuffer::FileBuffer;
  // The cache can only be appended to, not written, as it lets us maintain a single
  // version of the data.  This override is to make sure we don't accidentally start
  // writing to cache buffers.
  void write(int8_t* src,
             const size_t numBytes,
             const size_t offset = 0,
             const MemoryLevel srcMemoryLevel = CPU_LEVEL,
             const int32_t deviceId = -1) override {
    UNREACHABLE() << "Cache buffers support append(), but not write()";
  }
};

class CachingFileMgr : public FileMgr {
 public:
  static constexpr char WRAPPER_FILE_NAME[] = "wrapper_metadata.json";
  // We currently assign %10 of the cache to data wrapper space arbitrarily.
  // static constexpr size_t WRAPPER_SPACE_RATIO{10};
  // Portion of the CFM space reserved for metadata (metadata files and data wrappers)
  static constexpr float METADATA_SPACE_PERCENTAGE{0.1};
  // Portion of the CFM metadata space reserved for metadata files (subset of above).
  static constexpr float METADATA_FILE_SPACE_PERCENTAGE{0.01};

  static size_t getMinimumSize() {
    // Currently the minimum default size is based on the metadata file size and
    // percentage usage.
    return (DEFAULT_METADATA_PAGE_SIZE * DEFAULT_NUM_PAGES_PER_METADATA_FILE) /
           METADATA_FILE_SPACE_PERCENTAGE;
  }

  CachingFileMgr(const DiskCacheConfig& config);

  ~CachingFileMgr() override;

  // Simple getters.
  inline MgrType getMgrType() override { return CACHING_FILE_MGR; };
  inline std::string getStringMgrType() override { return ToString(CACHING_FILE_MGR); }
  inline size_t getPageSize() { return page_size_; }
  inline size_t getMaxSize() override { return max_size_; }
  inline size_t getMaxDataFiles() const { return max_num_data_files_; }
  inline size_t getMaxMetaFiles() const { return max_num_meta_files_; }
  inline size_t getMaxWrapperSize() const { return max_wrapper_space_; }
  inline size_t getDataFileSize() const { return page_size_ * num_pages_per_data_file_; }
  inline size_t getMetadataFileSize() const {
    return metadata_page_size_ * num_pages_per_metadata_file_;
  }

  size_t getNumDataFiles() const;
  size_t getNumMetaFiles() const;
  inline size_t getAvailableSpace() { return max_size_ - getAllocated(); }
  inline size_t getAvailableWrapperSpace() {
    return max_wrapper_space_ - getTableFileMgrsSize();
  }
  inline size_t getAllocated() override {
    return getFilesSize() + getTableFileMgrsSize();
  }
  size_t getMaxDataFilesSize() const;

  void removeChunkKeepMetadata(const ChunkKey& key);

  void clearForTable(int32_t db_id, int32_t tb_id);

  inline bool hasFileMgrKey() const override { return false; }

  void closeRemovePhysical() override;

  size_t getChunkSpaceReservedByTable(int32_t db_id, int32_t tb_id) const;
  size_t getMetadataSpaceReservedByTable(int32_t db_id, int32_t tb_id) const;
  size_t getTableFileMgrSpaceReserved(int32_t db_id, int32_t tb_id) const;
  size_t getSpaceReservedByTable(int32_t db_id, int32_t tb_id) const;

  std::string describeSelf() const override;

  void checkpoint(const int32_t db_id, const int32_t tb_id) override;

  int32_t epoch(int32_t db_id, int32_t tb_id) const override;

  FileBuffer* putBuffer(const ChunkKey& key,
                        AbstractBuffer* srcBuffer,
                        const size_t numBytes = 0) override;
  CachingFileBuffer* allocateBuffer(const size_t page_size,
                                    const ChunkKey& key,
                                    const size_t num_bytes = 0) override;
  CachingFileBuffer* allocateBuffer(
      const ChunkKey& key,
      const std::vector<HeaderInfo>::const_iterator& headerStartIt,
      const std::vector<HeaderInfo>::const_iterator& headerEndIt) override;

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

  inline bool failOnReadError() const override { return false; }

  void deleteBufferIfExists(const ChunkKey& key);

  size_t getNumChunksWithMetadata() const;

  size_t getNumDataChunks() const;

  std::vector<ChunkKey> getChunkKeysForPrefix(const ChunkKey& prefix) const;

  std::unique_ptr<CachingFileMgr> reconstruct() const;

  void deleteWrapperFile(int32_t db, int32_t tb);

  void writeWrapperFile(const std::string& doc, int32_t db, int32_t tb);

  bool hasWrapperFile(int32_t db_id, int32_t table_id) const;

  std::string getTableFileMgrPath(int32_t db, int32_t tb) const;

  size_t getFilesSize() const;

  size_t getTableFileMgrsSize() const;

  std::optional<FileBuffer*> getBufferIfExists(const ChunkKey& key);

  void free_page(std::pair<FileInfo*, int32_t>&& page) override;

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

  // Useful for debugging.
  std::string dumpKeysWithMetadata() const;
  std::string dumpKeysWithChunkData() const;
  std::string dumpTableQueue() const { return table_evict_alg_.dumpEvictionQueue(); }
  std::string dumpEvictionQueue() const { return chunk_evict_alg_.dumpEvictionQueue(); }
  std::string dump() const;

  // Used for unit testing
  void setMaxNumDataFiles(size_t max) { max_num_data_files_ = max; }
  void setMaxNumMetadataFiles(size_t max) { max_num_meta_files_ = max; }
  void setMaxWrapperSpace(size_t max) { max_wrapper_space_ = max; }
  std::set<ChunkKey> getKeysWithMetadata() const;
  void setDataSizeLimit(size_t max) { limit_data_size_ = max; }

  Page requestFreePage(size_t pagesize, const bool isMetadata) override;

 private:
  void incrementEpoch(int32_t db_id, int32_t tb_id);

  void init(const size_t num_reader_threads);

  void writeAndSyncEpochToDisk(int32_t db_id, int32_t tb_id);

  void readTableFileMgrs();

  FileBuffer* createBufferFromHeaders(
      const ChunkKey& key,
      const std::vector<HeaderInfo>::const_iterator& startIt,
      const std::vector<HeaderInfo>::const_iterator& endIt) override;

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

  void createTableFileMgrIfNoneExists(const int32_t db_id, const int32_t tb_id);

  void incrementAllEpochs();

  void removeTableFileMgr(int32_t db_id, int32_t tb_id);

  void removeTableBuffers(int32_t db_id, int32_t tb_id);

  void writeDirtyBuffers(int32_t db_id, int32_t tb_id);

  void touchKey(const ChunkKey& key) const;
  void removeKey(const ChunkKey& key) const;

  std::vector<ChunkKey> getKeysForTable(int32_t db_id, int32_t tb_id) const;

  FileInfo* evictMetadataPages();

  FileInfo* evictPages();

  void deleteCacheIfTooLarge();

  void setMaxSizes();

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

  ChunkKeyToChunkMap::iterator deleteBufferUnlocked(
      const ChunkKeyToChunkMap::iterator chunk_it,
      const bool purge = true) override;

  // CachingFileMgr is allowed to write in read-only mode, so override this function.
  void readOnlyCheck(const std::string& action,
                     const std::optional<std::string>& file_name = {}) const override{};

  mutable heavyai::shared_mutex table_dirs_mutex_;  // mutex for table_dirs_.
  // each table gest a separate epoch.  Uses pointers for move semantics.
  std::map<TablePair, std::unique_ptr<TableFileMgr>> table_dirs_;

  size_t max_num_data_files_;  // set based on max_size_.
  size_t max_num_meta_files_;  // set based on max_size_.
  size_t max_wrapper_space_;   // set based on max_size_.
  size_t max_size_;
  std::optional<size_t> limit_data_size_{};  // Used for testing artifically small caches.

  mutable LRUEvictionAlgorithm chunk_evict_alg_;  // last chunk touched.
  mutable LRUEvictionAlgorithm table_evict_alg_;  // last table touched.
};

}  // namespace File_Namespace