ceph/src/rocksdb/db/compaction/compaction_picker_fifo.cc

   1 //  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
   2 //  This source code is licensed under both the GPLv2 (found in the
   3 //  COPYING file in the root directory) and Apache 2.0 License
   4 //  (found in the LICENSE.Apache file in the root directory).
   5 //
   6 // Copyright (c) 2011 The LevelDB Authors. All rights reserved.
   7 // Use of this source code is governed by a BSD-style license that can be
   8 // found in the LICENSE file. See the AUTHORS file for names of contributors.
   9
  10 #include "db/compaction/compaction_picker_fifo.h"
  11 #ifndef ROCKSDB_LITE
  12
  13 #include <cinttypes>
  14 #include <string>
  15 #include <vector>
  16 #include "db/column_family.h"
  17 #include "logging/log_buffer.h"
  18 #include "util/string_util.h"
  19
  20 namespace ROCKSDB_NAMESPACE {
  21 namespace {
  22 uint64_t GetTotalFilesSize(const std::vector<FileMetaData*>& files) {
  23   uint64_t total_size = 0;
  24   for (const auto& f : files) {
  25     total_size += f->fd.file_size;
  26   }
  27   return total_size;
  28 }
  29 }  // anonymous namespace
  30
  31 bool FIFOCompactionPicker::NeedsCompaction(
  32     const VersionStorageInfo* vstorage) const {
  33   const int kLevel0 = 0;
  34   return vstorage->CompactionScore(kLevel0) >= 1;
  35 }
  36
  37 Compaction* FIFOCompactionPicker::PickTTLCompaction(
  38     const std::string& cf_name, const MutableCFOptions& mutable_cf_options,
  39     const MutableDBOptions& mutable_db_options, VersionStorageInfo* vstorage,
  40     LogBuffer* log_buffer) {
  41   assert(mutable_cf_options.ttl > 0);
  42
  43   const int kLevel0 = 0;
  44   const std::vector<FileMetaData*>& level_files = vstorage->LevelFiles(kLevel0);
  45   uint64_t total_size = GetTotalFilesSize(level_files);
  46
  47   int64_t _current_time;
  48   auto status = ioptions_.env->GetCurrentTime(&_current_time);
  49   if (!status.ok()) {
  50     ROCKS_LOG_BUFFER(log_buffer,
  51                      "[%s] FIFO compaction: Couldn't get current time: %s. "
  52                      "Not doing compactions based on TTL. ",
  53                      cf_name.c_str(), status.ToString().c_str());
  54     return nullptr;
  55   }
  56   const uint64_t current_time = static_cast<uint64_t>(_current_time);
  57
  58   if (!level0_compactions_in_progress_.empty()) {
  59     ROCKS_LOG_BUFFER(
  60         log_buffer,
  61         "[%s] FIFO compaction: Already executing compaction. No need "
  62         "to run parallel compactions since compactions are very fast",
  63         cf_name.c_str());
  64     return nullptr;
  65   }
  66
  67   std::vector<CompactionInputFiles> inputs;
  68   inputs.emplace_back();
  69   inputs[0].level = 0;
  70
  71   // avoid underflow
  72   if (current_time > mutable_cf_options.ttl) {
  73     for (auto ritr = level_files.rbegin(); ritr != level_files.rend(); ++ritr) {
  74       FileMetaData* f = *ritr;
  75       assert(f);
  76       if (f->fd.table_reader && f->fd.table_reader->GetTableProperties()) {
  77         uint64_t creation_time =
  78             f->fd.table_reader->GetTableProperties()->creation_time;
  79         if (creation_time == 0 ||
  80             creation_time >= (current_time - mutable_cf_options.ttl)) {
  81           break;
  82         }
  83       }
  84       total_size -= f->compensated_file_size;
  85       inputs[0].files.push_back(f);
  86     }
  87   }
  88
  89   // Return a nullptr and proceed to size-based FIFO compaction if:
  90   // 1. there are no files older than ttl OR
  91   // 2. there are a few files older than ttl, but deleting them will not bring
  92   //    the total size to be less than max_table_files_size threshold.
  93   if (inputs[0].files.empty() ||
  94       total_size >
  95           mutable_cf_options.compaction_options_fifo.max_table_files_size) {
  96     return nullptr;
  97   }
  98
  99   for (const auto& f : inputs[0].files) {
 100     uint64_t creation_time = 0;
 101     assert(f);
 102     if (f->fd.table_reader && f->fd.table_reader->GetTableProperties()) {
 103       creation_time = f->fd.table_reader->GetTableProperties()->creation_time;
 104     }
 105     ROCKS_LOG_BUFFER(log_buffer,
 106                      "[%s] FIFO compaction: picking file %" PRIu64
 107                      " with creation time %" PRIu64 " for deletion",
 108                      cf_name.c_str(), f->fd.GetNumber(), creation_time);
 109   }
 110
 111   Compaction* c = new Compaction(
 112       vstorage, ioptions_, mutable_cf_options, mutable_db_options,
 113       std::move(inputs), 0, 0, 0, 0, kNoCompression,
 114       mutable_cf_options.compression_opts,
 115       /* max_subcompactions */ 0, {}, /* is manual */ false,
 116       vstorage->CompactionScore(0),
 117       /* is deletion compaction */ true, CompactionReason::kFIFOTtl);
 118   return c;
 119 }
 120
 121 Compaction* FIFOCompactionPicker::PickSizeCompaction(
 122     const std::string& cf_name, const MutableCFOptions& mutable_cf_options,
 123     const MutableDBOptions& mutable_db_options, VersionStorageInfo* vstorage,
 124     LogBuffer* log_buffer) {
 125   const int kLevel0 = 0;
 126   const std::vector<FileMetaData*>& level_files = vstorage->LevelFiles(kLevel0);
 127   uint64_t total_size = GetTotalFilesSize(level_files);
 128
 129   if (total_size <=
 130           mutable_cf_options.compaction_options_fifo.max_table_files_size ||
 131       level_files.size() == 0) {
 132     // total size not exceeded
 133     if (mutable_cf_options.compaction_options_fifo.allow_compaction &&
 134         level_files.size() > 0) {
 135       CompactionInputFiles comp_inputs;
 136       // try to prevent same files from being compacted multiple times, which
 137       // could produce large files that may never TTL-expire. Achieve this by
 138       // disallowing compactions with files larger than memtable (inflate its
 139       // size by 10% to account for uncompressed L0 files that may have size
 140       // slightly greater than memtable size limit).
 141       size_t max_compact_bytes_per_del_file =
 142           static_cast<size_t>(MultiplyCheckOverflow(
 143               static_cast<uint64_t>(mutable_cf_options.write_buffer_size),
 144               1.1));
 145       if (FindIntraL0Compaction(
 146               level_files,
 147               mutable_cf_options
 148                   .level0_file_num_compaction_trigger /* min_files_to_compact */
 149               ,
 150               max_compact_bytes_per_del_file,
 151               mutable_cf_options.max_compaction_bytes, &comp_inputs)) {
 152         Compaction* c = new Compaction(
 153             vstorage, ioptions_, mutable_cf_options, mutable_db_options,
 154             {comp_inputs}, 0, 16 * 1024 * 1024 /* output file size limit */,
 155             0 /* max compaction bytes, not applicable */,
 156             0 /* output path ID */, mutable_cf_options.compression,
 157             mutable_cf_options.compression_opts, 0 /* max_subcompactions */, {},
 158             /* is manual */ false, vstorage->CompactionScore(0),
 159             /* is deletion compaction */ false,
 160             CompactionReason::kFIFOReduceNumFiles);
 161         return c;
 162       }
 163     }
 164
 165     ROCKS_LOG_BUFFER(
 166         log_buffer,
 167         "[%s] FIFO compaction: nothing to do. Total size %" PRIu64
 168         ", max size %" PRIu64 "\n",
 169         cf_name.c_str(), total_size,
 170         mutable_cf_options.compaction_options_fifo.max_table_files_size);
 171     return nullptr;
 172   }
 173
 174   if (!level0_compactions_in_progress_.empty()) {
 175     ROCKS_LOG_BUFFER(
 176         log_buffer,
 177         "[%s] FIFO compaction: Already executing compaction. No need "
 178         "to run parallel compactions since compactions are very fast",
 179         cf_name.c_str());
 180     return nullptr;
 181   }
 182
 183   std::vector<CompactionInputFiles> inputs;
 184   inputs.emplace_back();
 185   inputs[0].level = 0;
 186
 187   for (auto ritr = level_files.rbegin(); ritr != level_files.rend(); ++ritr) {
 188     auto f = *ritr;
 189     total_size -= f->compensated_file_size;
 190     inputs[0].files.push_back(f);
 191     char tmp_fsize[16];
 192     AppendHumanBytes(f->fd.GetFileSize(), tmp_fsize, sizeof(tmp_fsize));
 193     ROCKS_LOG_BUFFER(log_buffer,
 194                      "[%s] FIFO compaction: picking file %" PRIu64
 195                      " with size %s for deletion",
 196                      cf_name.c_str(), f->fd.GetNumber(), tmp_fsize);
 197     if (total_size <=
 198         mutable_cf_options.compaction_options_fifo.max_table_files_size) {
 199       break;
 200     }
 201   }
 202
 203   Compaction* c = new Compaction(
 204       vstorage, ioptions_, mutable_cf_options, mutable_db_options,
 205       std::move(inputs), 0, 0, 0, 0, kNoCompression,
 206       mutable_cf_options.compression_opts,
 207       /* max_subcompactions */ 0, {}, /* is manual */ false,
 208       vstorage->CompactionScore(0),
 209       /* is deletion compaction */ true, CompactionReason::kFIFOMaxSize);
 210   return c;
 211 }
 212
 213 Compaction* FIFOCompactionPicker::PickCompaction(
 214     const std::string& cf_name, const MutableCFOptions& mutable_cf_options,
 215     const MutableDBOptions& mutable_db_options, VersionStorageInfo* vstorage,
 216     LogBuffer* log_buffer, SequenceNumber /*earliest_memtable_seqno*/) {
 217   assert(vstorage->num_levels() == 1);
 218
 219   Compaction* c = nullptr;
 220   if (mutable_cf_options.ttl > 0) {
 221     c = PickTTLCompaction(cf_name, mutable_cf_options, mutable_db_options,
 222                           vstorage, log_buffer);
 223   }
 224   if (c == nullptr) {
 225     c = PickSizeCompaction(cf_name, mutable_cf_options, mutable_db_options,
 226                            vstorage, log_buffer);
 227   }
 228   RegisterCompaction(c);
 229   return c;
 230 }
 231
 232 Compaction* FIFOCompactionPicker::CompactRange(
 233     const std::string& cf_name, const MutableCFOptions& mutable_cf_options,
 234     const MutableDBOptions& mutable_db_options, VersionStorageInfo* vstorage,
 235     int input_level, int output_level,
 236     const CompactRangeOptions& /*compact_range_options*/,
 237     const InternalKey* /*begin*/, const InternalKey* /*end*/,
 238     InternalKey** compaction_end, bool* /*manual_conflict*/,
 239     uint64_t /*max_file_num_to_ignore*/) {
 240 #ifdef NDEBUG
 241   (void)input_level;
 242   (void)output_level;
 243 #endif
 244   assert(input_level == 0);
 245   assert(output_level == 0);
 246   *compaction_end = nullptr;
 247   LogBuffer log_buffer(InfoLogLevel::INFO_LEVEL, ioptions_.info_log);
 248   Compaction* c = PickCompaction(cf_name, mutable_cf_options,
 249                                  mutable_db_options, vstorage, &log_buffer);
 250   log_buffer.FlushBufferToLog();
 251   return c;
 252 }
 253
 254 }  // namespace ROCKSDB_NAMESPACE
 255 #endif  // !ROCKSDB_LITE