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 #pragma once 10 11 #include <atomic> 12 #include <deque> 13 #include <functional> 14 #include <limits> 15 #include <set> 16 #include <string> 17 #include <utility> 18 #include <vector> 19 20 #include "db/column_family.h" 21 #include "db/compaction/compaction_iterator.h" 22 #include "db/dbformat.h" 23 #include "db/flush_scheduler.h" 24 #include "db/internal_stats.h" 25 #include "db/job_context.h" 26 #include "db/log_writer.h" 27 #include "db/memtable_list.h" 28 #include "db/range_del_aggregator.h" 29 #include "db/version_edit.h" 30 #include "db/write_controller.h" 31 #include "db/write_thread.h" 32 #include "logging/event_logger.h" 33 #include "options/cf_options.h" 34 #include "options/db_options.h" 35 #include "port/port.h" 36 #include "rocksdb/compaction_filter.h" 37 #include "rocksdb/compaction_job_stats.h" 38 #include "rocksdb/db.h" 39 #include "rocksdb/env.h" 40 #include "rocksdb/memtablerep.h" 41 #include "rocksdb/transaction_log.h" 42 #include "table/scoped_arena_iterator.h" 43 #include "util/autovector.h" 44 #include "util/stop_watch.h" 45 #include "util/thread_local.h" 46 47 namespace ROCKSDB_NAMESPACE { 48 49 class Arena; 50 class ErrorHandler; 51 class MemTable; 52 class SnapshotChecker; 53 class TableCache; 54 class Version; 55 class VersionEdit; 56 class VersionSet; 57 58 // CompactionJob is responsible for executing the compaction. Each (manual or 59 // automated) compaction corresponds to a CompactionJob object, and usually 60 // goes through the stages of `Prepare()`->`Run()`->`Install()`. CompactionJob 61 // will divide the compaction into subcompactions and execute them in parallel 62 // if needed. 63 class CompactionJob { 64 public: 65 CompactionJob(int job_id, Compaction* compaction, 66 const ImmutableDBOptions& db_options, 67 const FileOptions& file_options, VersionSet* versions, 68 const std::atomic<bool>* shutting_down, 69 const SequenceNumber preserve_deletes_seqnum, 70 LogBuffer* log_buffer, FSDirectory* db_directory, 71 FSDirectory* output_directory, Statistics* stats, 72 InstrumentedMutex* db_mutex, ErrorHandler* db_error_handler, 73 std::vector<SequenceNumber> existing_snapshots, 74 SequenceNumber earliest_write_conflict_snapshot, 75 const SnapshotChecker* snapshot_checker, 76 std::shared_ptr<Cache> table_cache, EventLogger* event_logger, 77 bool paranoid_file_checks, bool measure_io_stats, 78 const std::string& dbname, 79 CompactionJobStats* compaction_job_stats, 80 Env::Priority thread_pri, 81 const std::atomic<bool>* manual_compaction_paused = nullptr); 82 83 ~CompactionJob(); 84 85 // no copy/move 86 CompactionJob(CompactionJob&& job) = delete; 87 CompactionJob(const CompactionJob& job) = delete; 88 CompactionJob& operator=(const CompactionJob& job) = delete; 89 90 // REQUIRED: mutex held 91 // Prepare for the compaction by setting up boundaries for each subcompaction 92 void Prepare(); 93 // REQUIRED mutex not held 94 // Launch threads for each subcompaction and wait for them to finish. After 95 // that, verify table is usable and finally do bookkeeping to unify 96 // subcompaction results 97 Status Run(); 98 99 // REQUIRED: mutex held 100 // Add compaction input/output to the current version 101 Status Install(const MutableCFOptions& mutable_cf_options); 102 103 // Return the IO status io_status()104 IOStatus io_status() const { return io_status_; } 105 106 private: 107 struct SubcompactionState; 108 109 void AggregateStatistics(); 110 111 // Generates a histogram representing potential divisions of key ranges from 112 // the input. It adds the starting and/or ending keys of certain input files 113 // to the working set and then finds the approximate size of data in between 114 // each consecutive pair of slices. Then it divides these ranges into 115 // consecutive groups such that each group has a similar size. 116 void GenSubcompactionBoundaries(); 117 118 // update the thread status for starting a compaction. 119 void ReportStartedCompaction(Compaction* compaction); 120 void AllocateCompactionOutputFileNumbers(); 121 // Call compaction filter. Then iterate through input and compact the 122 // kv-pairs 123 void ProcessKeyValueCompaction(SubcompactionState* sub_compact); 124 125 Status FinishCompactionOutputFile( 126 const Status& input_status, SubcompactionState* sub_compact, 127 CompactionRangeDelAggregator* range_del_agg, 128 CompactionIterationStats* range_del_out_stats, 129 const Slice* next_table_min_key = nullptr); 130 Status InstallCompactionResults(const MutableCFOptions& mutable_cf_options); 131 void RecordCompactionIOStats(); 132 Status OpenCompactionOutputFile(SubcompactionState* sub_compact); 133 void CleanupCompaction(); 134 void UpdateCompactionJobStats( 135 const InternalStats::CompactionStats& stats) const; 136 void RecordDroppedKeys(const CompactionIterationStats& c_iter_stats, 137 CompactionJobStats* compaction_job_stats = nullptr); 138 139 void UpdateCompactionStats(); 140 void UpdateCompactionInputStatsHelper( 141 int* num_files, uint64_t* bytes_read, int input_level); 142 143 void LogCompaction(); 144 145 int job_id_; 146 147 // CompactionJob state 148 struct CompactionState; 149 CompactionState* compact_; 150 CompactionJobStats* compaction_job_stats_; 151 InternalStats::CompactionStats compaction_stats_; 152 153 // DBImpl state 154 const std::string& dbname_; 155 const ImmutableDBOptions& db_options_; 156 const FileOptions file_options_; 157 158 Env* env_; 159 FileSystem* fs_; 160 // env_option optimized for compaction table reads 161 FileOptions file_options_for_read_; 162 VersionSet* versions_; 163 const std::atomic<bool>* shutting_down_; 164 const std::atomic<bool>* manual_compaction_paused_; 165 const SequenceNumber preserve_deletes_seqnum_; 166 LogBuffer* log_buffer_; 167 FSDirectory* db_directory_; 168 FSDirectory* output_directory_; 169 Statistics* stats_; 170 InstrumentedMutex* db_mutex_; 171 ErrorHandler* db_error_handler_; 172 // If there were two snapshots with seq numbers s1 and 173 // s2 and s1 < s2, and if we find two instances of a key k1 then lies 174 // entirely within s1 and s2, then the earlier version of k1 can be safely 175 // deleted because that version is not visible in any snapshot. 176 std::vector<SequenceNumber> existing_snapshots_; 177 178 // This is the earliest snapshot that could be used for write-conflict 179 // checking by a transaction. For any user-key newer than this snapshot, we 180 // should make sure not to remove evidence that a write occurred. 181 SequenceNumber earliest_write_conflict_snapshot_; 182 183 const SnapshotChecker* const snapshot_checker_; 184 185 std::shared_ptr<Cache> table_cache_; 186 187 EventLogger* event_logger_; 188 189 // Is this compaction creating a file in the bottom most level? 190 bool bottommost_level_; 191 bool paranoid_file_checks_; 192 bool measure_io_stats_; 193 // Stores the Slices that designate the boundaries for each subcompaction 194 std::vector<Slice> boundaries_; 195 // Stores the approx size of keys covered in the range of each subcompaction 196 std::vector<uint64_t> sizes_; 197 Env::WriteLifeTimeHint write_hint_; 198 Env::Priority thread_pri_; 199 IOStatus io_status_; 200 }; 201 202 } // namespace ROCKSDB_NAMESPACE 203