db_impl.h 58.9 KB
Newer Older
1
//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
S
Siying Dong 已提交
2 3 4
//  This source code is licensed under both the GPLv2 (found in the
//  COPYING file in the root directory) and Apache 2.0 License
//  (found in the LICENSE.Apache file in the root directory).
5
//
J
jorlow@chromium.org 已提交
6 7 8
// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file. See the AUTHORS file for names of contributors.
9
#pragma once
K
Kai Liu 已提交
10

H
Haobo Xu 已提交
11
#include <atomic>
12
#include <deque>
13
#include <functional>
14
#include <limits>
I
Igor Canadi 已提交
15
#include <list>
16
#include <map>
17
#include <set>
I
Igor Canadi 已提交
18
#include <string>
19 20
#include <utility>
#include <vector>
K
kailiu 已提交
21

22
#include "db/column_family.h"
23
#include "db/compaction_job.h"
24
#include "db/dbformat.h"
25
#include "db/external_sst_file_ingestion_job.h"
26
#include "db/flush_job.h"
27 28
#include "db/flush_scheduler.h"
#include "db/internal_stats.h"
A
agiardullo 已提交
29
#include "db/log_writer.h"
30
#include "db/pre_release_callback.h"
31
#include "db/read_callback.h"
Y
Yi Wu 已提交
32
#include "db/snapshot_checker.h"
A
agiardullo 已提交
33
#include "db/snapshot_impl.h"
34
#include "db/version_edit.h"
I
Igor Canadi 已提交
35
#include "db/wal_manager.h"
36 37
#include "db/write_controller.h"
#include "db/write_thread.h"
K
Kai Liu 已提交
38
#include "memtable_list.h"
39 40
#include "monitoring/instrumented_mutex.h"
#include "options/db_options.h"
K
Kai Liu 已提交
41
#include "port/port.h"
42 43 44
#include "rocksdb/db.h"
#include "rocksdb/env.h"
#include "rocksdb/memtablerep.h"
S
Siying Dong 已提交
45
#include "rocksdb/status.h"
46
#include "rocksdb/transaction_log.h"
47
#include "rocksdb/write_buffer_manager.h"
S
sdong 已提交
48
#include "table/scoped_arena_iterator.h"
49
#include "util/autovector.h"
I
Igor Canadi 已提交
50 51
#include "util/event_logger.h"
#include "util/hash.h"
52 53
#include "util/stop_watch.h"
#include "util/thread_local.h"
54

55
namespace rocksdb {
J
jorlow@chromium.org 已提交
56

Y
Yi Wu 已提交
57
class Arena;
Y
Yi Wu 已提交
58
class ArenaWrappedDBIter;
J
jorlow@chromium.org 已提交
59 60 61 62 63
class MemTable;
class TableCache;
class Version;
class VersionEdit;
class VersionSet;
A
agiardullo 已提交
64
class WriteCallback;
I
Igor Canadi 已提交
65
struct JobContext;
66
struct ExternalSstFileInfo;
67
struct MemTableInfo;
J
jorlow@chromium.org 已提交
68 69 70

class DBImpl : public DB {
 public:
71 72
  DBImpl(const DBOptions& options, const std::string& dbname,
         const bool seq_per_batch = false);
J
jorlow@chromium.org 已提交
73 74 75
  virtual ~DBImpl();

  // Implementations of the DB interface
76 77
  using DB::Put;
  virtual Status Put(const WriteOptions& options,
78
                     ColumnFamilyHandle* column_family, const Slice& key,
I
Igor Sugak 已提交
79
                     const Slice& value) override;
80 81
  using DB::Merge;
  virtual Status Merge(const WriteOptions& options,
82
                       ColumnFamilyHandle* column_family, const Slice& key,
I
Igor Sugak 已提交
83
                       const Slice& value) override;
84 85
  using DB::Delete;
  virtual Status Delete(const WriteOptions& options,
I
Igor Sugak 已提交
86 87
                        ColumnFamilyHandle* column_family,
                        const Slice& key) override;
A
Andres Noetzli 已提交
88 89 90 91
  using DB::SingleDelete;
  virtual Status SingleDelete(const WriteOptions& options,
                              ColumnFamilyHandle* column_family,
                              const Slice& key) override;
92
  using DB::Write;
I
Igor Sugak 已提交
93 94
  virtual Status Write(const WriteOptions& options,
                       WriteBatch* updates) override;
A
agiardullo 已提交
95

96
  using DB::Get;
J
jorlow@chromium.org 已提交
97
  virtual Status Get(const ReadOptions& options,
98
                     ColumnFamilyHandle* column_family, const Slice& key,
M
Maysam Yabandeh 已提交
99
                     PinnableSlice* value) override;
100 101 102 103 104 105 106 107

  // Function that Get and KeyMayExist call with no_io true or false
  // Note: 'value_found' from KeyMayExist propagates here
  Status GetImpl(const ReadOptions& options, ColumnFamilyHandle* column_family,
                 const Slice& key, PinnableSlice* value,
                 bool* value_found = nullptr, ReadCallback* callback = nullptr,
                 bool* is_blob_index = nullptr);

108 109 110
  using DB::MultiGet;
  virtual std::vector<Status> MultiGet(
      const ReadOptions& options,
111
      const std::vector<ColumnFamilyHandle*>& column_family,
I
Igor Sugak 已提交
112 113
      const std::vector<Slice>& keys,
      std::vector<std::string>* values) override;
114

Y
Yi Wu 已提交
115
  virtual Status CreateColumnFamily(const ColumnFamilyOptions& cf_options,
116
                                    const std::string& column_family,
I
Igor Sugak 已提交
117
                                    ColumnFamilyHandle** handle) override;
Y
Yi Wu 已提交
118 119 120 121 122 123 124
  virtual Status CreateColumnFamilies(
      const ColumnFamilyOptions& cf_options,
      const std::vector<std::string>& column_family_names,
      std::vector<ColumnFamilyHandle*>* handles) override;
  virtual Status CreateColumnFamilies(
      const std::vector<ColumnFamilyDescriptor>& column_families,
      std::vector<ColumnFamilyHandle*>* handles) override;
I
Igor Sugak 已提交
125
  virtual Status DropColumnFamily(ColumnFamilyHandle* column_family) override;
Y
Yi Wu 已提交
126 127
  virtual Status DropColumnFamilies(
      const std::vector<ColumnFamilyHandle*>& column_families) override;
128

129 130 131 132
  // Returns false if key doesn't exist in the database and true if it may.
  // If value_found is not passed in as null, then return the value if found in
  // memory. On return, if value was found, then value_found will be set to true
  // , otherwise false.
133
  using DB::KeyMayExist;
134
  virtual bool KeyMayExist(const ReadOptions& options,
135
                           ColumnFamilyHandle* column_family, const Slice& key,
I
Igor Sugak 已提交
136 137
                           std::string* value,
                           bool* value_found = nullptr) override;
Y
Yi Wu 已提交
138

139 140
  using DB::NewIterator;
  virtual Iterator* NewIterator(const ReadOptions& options,
I
Igor Sugak 已提交
141
                                ColumnFamilyHandle* column_family) override;
142 143
  virtual Status NewIterators(
      const ReadOptions& options,
I
Igor Canadi 已提交
144
      const std::vector<ColumnFamilyHandle*>& column_families,
I
Igor Sugak 已提交
145
      std::vector<Iterator*>* iterators) override;
Y
Yi Wu 已提交
146 147 148
  ArenaWrappedDBIter* NewIteratorImpl(const ReadOptions& options,
                                      ColumnFamilyData* cfd,
                                      SequenceNumber snapshot,
Y
Yi Wu 已提交
149
                                      ReadCallback* read_callback,
150 151
                                      bool allow_blob = false,
                                      bool allow_refresh = true);
Y
Yi Wu 已提交
152

I
Igor Sugak 已提交
153 154
  virtual const Snapshot* GetSnapshot() override;
  virtual void ReleaseSnapshot(const Snapshot* snapshot) override;
155
  using DB::GetProperty;
156
  virtual bool GetProperty(ColumnFamilyHandle* column_family,
I
Igor Sugak 已提交
157
                           const Slice& property, std::string* value) override;
158
  using DB::GetMapProperty;
159 160 161
  virtual bool GetMapProperty(
      ColumnFamilyHandle* column_family, const Slice& property,
      std::map<std::string, std::string>* value) override;
162 163 164
  using DB::GetIntProperty;
  virtual bool GetIntProperty(ColumnFamilyHandle* column_family,
                              const Slice& property, uint64_t* value) override;
165 166 167
  using DB::GetAggregatedIntProperty;
  virtual bool GetAggregatedIntProperty(const Slice& property,
                                        uint64_t* aggregated_value) override;
168
  using DB::GetApproximateSizes;
169
  virtual void GetApproximateSizes(ColumnFamilyHandle* column_family,
170
                                   const Range* range, int n, uint64_t* sizes,
171
                                   uint8_t include_flags
172
                                   = INCLUDE_FILES) override;
173 174 175 176 177
  using DB::GetApproximateMemTableStats;
  virtual void GetApproximateMemTableStats(ColumnFamilyHandle* column_family,
                                           const Range& range,
                                           uint64_t* const count,
                                           uint64_t* const size) override;
178
  using DB::CompactRange;
179 180 181
  virtual Status CompactRange(const CompactRangeOptions& options,
                              ColumnFamilyHandle* column_family,
                              const Slice* begin, const Slice* end) override;
182

183
  using DB::CompactFiles;
I
Igor Sugak 已提交
184 185 186 187 188
  virtual Status CompactFiles(const CompactionOptions& compact_options,
                              ColumnFamilyHandle* column_family,
                              const std::vector<std::string>& input_file_names,
                              const int output_level,
                              const int output_path_id = -1) override;
189

190 191 192
  virtual Status PauseBackgroundWork() override;
  virtual Status ContinueBackgroundWork() override;

193 194 195
  virtual Status EnableAutoCompaction(
      const std::vector<ColumnFamilyHandle*>& column_family_handles) override;

196
  using DB::SetOptions;
I
Igor Sugak 已提交
197 198 199
  Status SetOptions(
      ColumnFamilyHandle* column_family,
      const std::unordered_map<std::string, std::string>& options_map) override;
200

201 202 203
  virtual Status SetDBOptions(
      const std::unordered_map<std::string, std::string>& options_map) override;

204
  using DB::NumberLevels;
I
Igor Sugak 已提交
205
  virtual int NumberLevels(ColumnFamilyHandle* column_family) override;
206
  using DB::MaxMemCompactionLevel;
I
Igor Sugak 已提交
207
  virtual int MaxMemCompactionLevel(ColumnFamilyHandle* column_family) override;
208
  using DB::Level0StopWriteTrigger;
I
Igor Sugak 已提交
209 210 211 212
  virtual int Level0StopWriteTrigger(
      ColumnFamilyHandle* column_family) override;
  virtual const std::string& GetName() const override;
  virtual Env* GetEnv() const override;
213
  using DB::GetOptions;
214
  virtual Options GetOptions(ColumnFamilyHandle* column_family) const override;
215
  using DB::GetDBOptions;
216
  virtual DBOptions GetDBOptions() const override;
217 218
  using DB::Flush;
  virtual Status Flush(const FlushOptions& options,
I
Igor Sugak 已提交
219
                       ColumnFamilyHandle* column_family) override;
220
  virtual Status FlushWAL(bool sync) override;
221
  virtual Status SyncWAL() override;
I
Igor Canadi 已提交
222

I
Igor Sugak 已提交
223
  virtual SequenceNumber GetLatestSequenceNumber() const override;
224 225
  virtual void SetLastPublishedSequence(SequenceNumber seq);
  // Returns LastSequence in last_seq_same_as_publish_seq_
226 227 228
  // mode and LastAllocatedSequence otherwise. This is useful when visiblility
  // depends also on data written to the WAL but not to the memtable.
  SequenceNumber TEST_GetLastVisibleSequence() const;
I
Igor Canadi 已提交
229

230 231
  virtual bool SetPreserveDeletesSequenceNumber(SequenceNumber seqnum) override;

I
Igor Canadi 已提交
232
#ifndef ROCKSDB_LITE
S
Siying Dong 已提交
233 234
  using DB::ResetStats;
  virtual Status ResetStats() override;
I
Igor Sugak 已提交
235 236
  virtual Status DisableFileDeletions() override;
  virtual Status EnableFileDeletions(bool force) override;
237
  virtual int IsFileDeletionsEnabled() const;
I
Igor Canadi 已提交
238
  // All the returned filenames start with "/"
239
  virtual Status GetLiveFiles(std::vector<std::string>&,
240
                              uint64_t* manifest_file_size,
I
Igor Sugak 已提交
241 242
                              bool flush_memtable = true) override;
  virtual Status GetSortedWalFiles(VectorLogPtr& files) override;
I
Igor Canadi 已提交
243

244 245 246
  virtual Status GetUpdatesSince(
      SequenceNumber seq_number, unique_ptr<TransactionLogIterator>* iter,
      const TransactionLogIterator::ReadOptions&
I
Igor Sugak 已提交
247 248
          read_options = TransactionLogIterator::ReadOptions()) override;
  virtual Status DeleteFile(std::string name) override;
249 250 251
  Status DeleteFilesInRanges(ColumnFamilyHandle* column_family,
                             const RangePtr* ranges, size_t n,
                             bool include_end = true);
252

I
Igor Sugak 已提交
253 254
  virtual void GetLiveFilesMetaData(
      std::vector<LiveFileMetaData>* metadata) override;
255 256 257 258 259 260 261 262 263

  // Obtains the meta data of the specified column family of the DB.
  // Status::NotFound() will be returned if the current DB does not have
  // any column family match the specified name.
  // TODO(yhchiang): output parameter is placed in the end in this codebase.
  virtual void GetColumnFamilyMetaData(
      ColumnFamilyHandle* column_family,
      ColumnFamilyMetaData* metadata) override;

264
  Status SuggestCompactRange(ColumnFamilyHandle* column_family,
S
Siying Dong 已提交
265
                             const Slice* begin, const Slice* end) override;
266

S
Siying Dong 已提交
267 268
  Status PromoteL0(ColumnFamilyHandle* column_family,
                   int target_level) override;
269

A
agiardullo 已提交
270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289
  // Similar to Write() but will call the callback once on the single write
  // thread to determine whether it is safe to perform the write.
  virtual Status WriteWithCallback(const WriteOptions& write_options,
                                   WriteBatch* my_batch,
                                   WriteCallback* callback);

  // Returns the sequence number that is guaranteed to be smaller than or equal
  // to the sequence number of any key that could be inserted into the current
  // memtables. It can then be assumed that any write with a larger(or equal)
  // sequence number will be present in this memtable or a later memtable.
  //
  // If the earliest sequence number could not be determined,
  // kMaxSequenceNumber will be returned.
  //
  // If include_history=true, will also search Memtables in MemTableList
  // History.
  SequenceNumber GetEarliestMemTableSequenceNumber(SuperVersion* sv,
                                                   bool include_history);

  // For a given key, check to see if there are any records for this key
290 291 292 293
  // in the memtables, including memtable history.  If cache_only is false,
  // SST files will also be checked.
  //
  // If a key is found, *found_record_for_key will be set to true and
294
  // *seq will be set to the stored sequence number for the latest
295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310
  // operation on this key or kMaxSequenceNumber if unknown.
  // If no key is found, *found_record_for_key will be set to false.
  //
  // Note: If cache_only=false, it is possible for *seq to be set to 0 if
  // the sequence number has been cleared from the record.  If the caller is
  // holding an active db snapshot, we know the missing sequence must be less
  // than the snapshot's sequence number (sequence numbers are only cleared
  // when there are no earlier active snapshots).
  //
  // If NotFound is returned and found_record_for_key is set to false, then no
  // record for this key was found.  If the caller is holding an active db
  // snapshot, we know that no key could have existing after this snapshot
  // (since we do not compact keys that have an earlier snapshot).
  //
  // Returns OK or NotFound on success,
  // other status on unexpected error.
311
  // TODO(andrewkr): this API need to be aware of range deletion operations
312 313
  Status GetLatestSequenceForKey(SuperVersion* sv, const Slice& key,
                                 bool cache_only, SequenceNumber* seq,
314 315
                                 bool* found_record_for_key,
                                 bool* is_blob_index = nullptr);
A
agiardullo 已提交
316

317 318 319 320 321
  using DB::IngestExternalFile;
  virtual Status IngestExternalFile(
      ColumnFamilyHandle* column_family,
      const std::vector<std::string>& external_files,
      const IngestExternalFileOptions& ingestion_options) override;
322

A
Aaron G 已提交
323 324
  virtual Status VerifyChecksum() override;

I
Igor Canadi 已提交
325
#endif  // ROCKSDB_LITE
326

327 328 329 330 331 332
  // Similar to GetSnapshot(), but also lets the db know that this snapshot
  // will be used for transaction write-conflict checking.  The DB can then
  // make sure not to compact any keys that would prevent a write-conflict from
  // being detected.
  const Snapshot* GetSnapshotForWriteConflictBoundary();

I
Igor Canadi 已提交
333 334 335 336
  // checks if all live files exist on file system and that their file sizes
  // match to our in-memory records
  virtual Status CheckConsistency();

337
  virtual Status GetDbIdentity(std::string& identity) const override;
338

I
Igor Canadi 已提交
339
  Status RunManualCompaction(ColumnFamilyData* cfd, int input_level,
340
                             int output_level, uint32_t output_path_id,
341
                             const Slice* begin, const Slice* end,
342
                             bool exclusive,
343
                             bool disallow_trivial_move = false);
344

345 346 347
  // Return an internal iterator over the current state of the database.
  // The keys of this iterator are internal keys (see format.h).
  // The returned iterator should be deleted when no longer needed.
S
sdong 已提交
348
  InternalIterator* NewInternalIterator(
A
Andrew Kryczka 已提交
349 350
      Arena* arena, RangeDelAggregator* range_del_agg,
      ColumnFamilyHandle* column_family = nullptr);
351

352
#ifndef NDEBUG
J
jorlow@chromium.org 已提交
353
  // Extra methods (for testing) that are not in the public DB interface
I
Igor Canadi 已提交
354
  // Implemented in db_impl_debug.cc
J
jorlow@chromium.org 已提交
355

356
  // Compact any files in the named level that overlap [*begin, *end]
357
  Status TEST_CompactRange(int level, const Slice* begin, const Slice* end,
358 359
                           ColumnFamilyHandle* column_family = nullptr,
                           bool disallow_trivial_move = false);
J
jorlow@chromium.org 已提交
360

361
  void TEST_SwitchWAL();
362 363 364 365 366 367 368 369 370

  bool TEST_UnableToFlushOldestLog() {
    return unable_to_flush_oldest_log_;
  }

  bool TEST_IsLogGettingFlushed() {
    return alive_log_files_.begin()->getting_flushed;
  }

Y
Yi Wu 已提交
371 372
  Status TEST_SwitchMemtable(ColumnFamilyData* cfd = nullptr);

373
  // Force current memtable contents to be flushed.
374 375
  Status TEST_FlushMemTable(bool wait = true,
                            ColumnFamilyHandle* cfh = nullptr);
J
jorlow@chromium.org 已提交
376

377
  // Wait for memtable compaction
378
  Status TEST_WaitForFlushMemTable(ColumnFamilyHandle* column_family = nullptr);
379 380

  // Wait for any compaction
381 382 383
  // We add a bool parameter to wait for unscheduledCompactions_ == 0, but this
  // is only for the special test of CancelledCompactions
  Status TEST_WaitForCompact(bool waitUnscheduled = false);
384

385 386
  // Return the maximum overlapping data (in bytes) at next level for any
  // file at a level >= 1.
387 388
  int64_t TEST_MaxNextLevelOverlappingBytes(ColumnFamilyHandle* column_family =
                                                nullptr);
389

A
Abhishek Kona 已提交
390 391
  // Return the current manifest file no.
  uint64_t TEST_Current_Manifest_FileNo();
392

A
Andrew Kryczka 已提交
393 394 395
  // Returns the number that'll be assigned to the next file that's created.
  uint64_t TEST_Current_Next_FileNo();

396
  // get total level0 file size. Only for testing.
397
  uint64_t TEST_GetLevel0TotalSize();
398

399 400
  void TEST_GetFilesMetaData(ColumnFamilyHandle* column_family,
                             std::vector<std::vector<FileMetaData>>* metadata);
401

402 403 404 405 406 407 408 409 410 411
  void TEST_LockMutex();

  void TEST_UnlockMutex();

  // REQUIRES: mutex locked
  void* TEST_BeginWrite();

  // REQUIRES: mutex locked
  // pass the pointer that you got from TEST_BeginWrite()
  void TEST_EndWrite(void* w);
412

413
  uint64_t TEST_MaxTotalInMemoryState() const {
414 415
    return max_total_in_memory_state_;
  }
416

417 418
  size_t TEST_LogsToFreeSize();

419 420
  uint64_t TEST_LogfileNumber();

S
sdong 已提交
421 422
  uint64_t TEST_total_log_size() const { return total_log_size_; }

423 424 425 426
  // Returns column family name to ImmutableCFOptions map.
  Status TEST_GetAllImmutableCFOptions(
      std::unordered_map<std::string, const ImmutableCFOptions*>* iopts_map);

H
hyunwoo 已提交
427
  // Return the lastest MutableCFOptions of a column family
A
Aaron Gao 已提交
428
  Status TEST_GetLatestMutableCFOptions(ColumnFamilyHandle* column_family,
Y
yizhu.sun 已提交
429
                                        MutableCFOptions* mutable_cf_options);
A
Aaron Gao 已提交
430

431 432
  Cache* TEST_table_cache() { return table_cache_.get(); }

433
  WriteController& TEST_write_controler() { return write_controller_; }
434

435 436
  uint64_t TEST_FindMinLogContainingOutstandingPrep();
  uint64_t TEST_FindMinPrepLogReferencedByMemTable();
437 438
  size_t TEST_PreparedSectionCompletedSize();
  size_t TEST_LogsWithPrepSize();
439

440
  int TEST_BGCompactionsAllowed() const;
441
  int TEST_BGFlushesAllowed() const;
442

443
#endif  // NDEBUG
I
Igor Canadi 已提交
444

445 446 447 448 449 450 451 452 453 454 455
  struct BGJobLimits {
    int max_flushes;
    int max_compactions;
  };
  // Returns maximum background flushes and compactions allowed to be scheduled
  BGJobLimits GetBGJobLimits() const;
  // Need a static version that can be called during SanitizeOptions().
  static BGJobLimits GetBGJobLimits(int max_background_flushes,
                                    int max_background_compactions,
                                    int max_background_jobs,
                                    bool parallelize_compactions);
456

457 458 459 460
  // move logs pending closing from job_context to the DB queue and
  // schedule a purge
  void ScheduleBgLogWriterClose(JobContext* job_context);

461 462
  uint64_t MinLogNumberToKeep();

I
Igor Canadi 已提交
463
  // Returns the list of live files in 'live' and the list
K
kailiu 已提交
464
  // of all files in the filesystem in 'candidate_files'.
I
Igor Canadi 已提交
465
  // If force == false and the last call was less than
466
  // db_options_.delete_obsolete_files_period_micros microseconds ago,
I
Igor Canadi 已提交
467 468
  // it will not fill up the job_context
  void FindObsoleteFiles(JobContext* job_context, bool force,
I
Igor Canadi 已提交
469 470 471
                         bool no_full_scan = false);

  // Diffs the files listed in filenames and those that do not
472
  // belong to live files are possibly removed. Also, removes all the
I
Igor Canadi 已提交
473 474
  // files in sst_delete_files and log_delete_files.
  // It is not necessary to hold the mutex when invoking this method.
475 476
  // If FindObsoleteFiles() was run, we need to also run
  // PurgeObsoleteFiles(), even if disable_delete_obsolete_files_ is true
477 478 479 480
  void PurgeObsoleteFiles(const JobContext& background_contet,
                          bool schedule_only = false);

  void SchedulePurge();
I
Igor Canadi 已提交
481

I
Igor Sugak 已提交
482
  ColumnFamilyHandle* DefaultColumnFamily() const override;
483

484 485
  const SnapshotList& snapshots() const { return snapshots_; }

486 487 488 489
  const ImmutableDBOptions& immutable_db_options() const {
    return immutable_db_options_;
  }

490
  void CancelAllBackgroundWork(bool wait);
491

A
agiardullo 已提交
492 493 494 495 496 497 498 499 500 501 502
  // Find Super version and reference it. Based on options, it might return
  // the thread local cached one.
  // Call ReturnAndCleanupSuperVersion() when it is no longer needed.
  SuperVersion* GetAndRefSuperVersion(ColumnFamilyData* cfd);

  // Similar to the previous function but looks up based on a column family id.
  // nullptr will be returned if this column family no longer exists.
  // REQUIRED: this function should only be called on the write thread or if the
  // mutex is held.
  SuperVersion* GetAndRefSuperVersion(uint32_t column_family_id);

503 504 505
  // Un-reference the super version and clean it up if it is the last reference.
  void CleanupSuperVersion(SuperVersion* sv);

A
agiardullo 已提交
506 507 508 509 510 511 512 513 514 515 516 517 518 519
  // Un-reference the super version and return it to thread local cache if
  // needed. If it is the last reference of the super version. Clean it up
  // after un-referencing it.
  void ReturnAndCleanupSuperVersion(ColumnFamilyData* cfd, SuperVersion* sv);

  // Similar to the previous function but looks up based on a column family id.
  // nullptr will be returned if this column family no longer exists.
  // REQUIRED: this function should only be called on the write thread.
  void ReturnAndCleanupSuperVersion(uint32_t colun_family_id, SuperVersion* sv);

  // REQUIRED: this function should only be called on the write thread or if the
  // mutex is held.  Return value only valid until next call to this function or
  // mutex is released.
  ColumnFamilyHandle* GetColumnFamilyHandle(uint32_t column_family_id);
A
agiardullo 已提交
520

A
Anirban Rahut 已提交
521 522 523
  // Same as above, should called without mutex held and not on write thread.
  ColumnFamilyHandle* GetColumnFamilyHandleUnlocked(uint32_t column_family_id);

524 525 526 527 528 529 530 531 532 533 534 535 536 537
  // Returns the number of currently running flushes.
  // REQUIREMENT: mutex_ must be held when calling this function.
  int num_running_flushes() {
    mutex_.AssertHeld();
    return num_running_flushes_;
  }

  // Returns the number of currently running compactions.
  // REQUIREMENT: mutex_ must be held when calling this function.
  int num_running_compactions() {
    mutex_.AssertHeld();
    return num_running_compactions_;
  }

538 539
  const WriteController& write_controller() { return write_controller_; }

S
Siying Dong 已提交
540 541 542 543 544 545
  InternalIterator* NewInternalIterator(const ReadOptions&,
                                        ColumnFamilyData* cfd,
                                        SuperVersion* super_version,
                                        Arena* arena,
                                        RangeDelAggregator* range_del_agg);

546 547 548 549 550 551 552
  // hollow transactions shell used for recovery.
  // these will then be passed to TransactionDB so that
  // locks can be reacquired before writing can resume.
  struct RecoveredTransaction {
    uint64_t log_number_;
    std::string name_;
    WriteBatch* batch_;
M
Maysam Yabandeh 已提交
553 554
    // The seq number of the first key in the batch
    SequenceNumber seq_;
555 556 557 558
    // Number of sub-batched. A new sub-batch is created if we txn attempts to
    // inserts a duplicate key,seq to memtable. This is currently used in
    // WritePrparedTxn
    size_t batch_cnt_;
559
    explicit RecoveredTransaction(const uint64_t log, const std::string& name,
560 561 562 563 564 565 566
                                  WriteBatch* batch, SequenceNumber seq,
                                  size_t batch_cnt)
        : log_number_(log),
          name_(name),
          batch_(batch),
          seq_(seq),
          batch_cnt_(batch_cnt) {}
567 568 569 570

    ~RecoveredTransaction() { delete batch_; }
  };

571
  bool allow_2pc() const { return immutable_db_options_.allow_2pc; }
572

R
Reid Horuff 已提交
573 574 575 576 577
  std::unordered_map<std::string, RecoveredTransaction*>
  recovered_transactions() {
    return recovered_transactions_;
  }

578 579 580 581 582 583 584 585 586 587
  RecoveredTransaction* GetRecoveredTransaction(const std::string& name) {
    auto it = recovered_transactions_.find(name);
    if (it == recovered_transactions_.end()) {
      return nullptr;
    } else {
      return it->second;
    }
  }

  void InsertRecoveredTransaction(const uint64_t log, const std::string& name,
588 589
                                  WriteBatch* batch, SequenceNumber seq,
                                  size_t batch_cnt) {
M
Maysam Yabandeh 已提交
590
    recovered_transactions_[name] =
591
        new RecoveredTransaction(log, name, batch, seq, batch_cnt);
592 593 594 595 596 597 598 599 600 601 602 603
    MarkLogAsContainingPrepSection(log);
  }

  void DeleteRecoveredTransaction(const std::string& name) {
    auto it = recovered_transactions_.find(name);
    assert(it != recovered_transactions_.end());
    auto* trx = it->second;
    recovered_transactions_.erase(it);
    MarkLogAsHavingPrepSectionFlushed(trx->log_number_);
    delete trx;
  }

R
Reid Horuff 已提交
604 605 606 607 608 609 610 611
  void DeleteAllRecoveredTransactions() {
    for (auto it = recovered_transactions_.begin();
         it != recovered_transactions_.end(); it++) {
      delete it->second;
    }
    recovered_transactions_.clear();
  }

612 613
  void MarkLogAsHavingPrepSectionFlushed(uint64_t log);
  void MarkLogAsContainingPrepSection(uint64_t log);
614 615 616
  void AddToLogsToFreeQueue(log::Writer* log_writer) {
    logs_to_free_queue_.push_back(log_writer);
  }
Y
Yi Wu 已提交
617 618 619

  void SetSnapshotChecker(SnapshotChecker* snapshot_checker);

S
Siying Dong 已提交
620
  InstrumentedMutex* mutex() { return &mutex_; }
621

622 623
  Status NewDB();

624 625 626 627 628 629
  // This is to be used only by internal rocksdb classes.
  static Status Open(const DBOptions& db_options, const std::string& name,
                     const std::vector<ColumnFamilyDescriptor>& column_families,
                     std::vector<ColumnFamilyHandle*>* handles, DB** dbptr,
                     const bool seq_per_batch);

630 631
  virtual Status Close() override;

632
 protected:
H
heyongqiang 已提交
633 634
  Env* const env_;
  const std::string dbname_;
635
  unique_ptr<VersionSet> versions_;
636 637
  // Flag to check whether we allocated and own the info log file
  bool own_info_log_;
638
  const DBOptions initial_db_options_;
639 640
  const ImmutableDBOptions immutable_db_options_;
  MutableDBOptions mutable_db_options_;
L
Lei Jin 已提交
641
  Statistics* stats_;
642 643
  std::unordered_map<std::string, RecoveredTransaction*>
      recovered_transactions_;
H
heyongqiang 已提交
644

645
  // Except in DB::Open(), WriteOptionsFile can only be called when:
Y
Yi Wu 已提交
646 647 648 649
  // Persist options to options file.
  // If need_mutex_lock = false, the method will lock DB mutex.
  // If need_enter_write_thread = false, the method will enter write thread.
  Status WriteOptionsFile(bool need_mutex_lock, bool need_enter_write_thread);
650 651 652 653

  // The following two functions can only be called when:
  // 1. WriteThread::Writer::EnterUnbatched() is used.
  // 2. db_mutex is NOT held
654 655 656
  Status RenameTempFileToOptionsFile(const std::string& file_name);
  Status DeleteObsoleteOptionsFiles();

Y
Yi Wu 已提交
657 658 659 660
  void NotifyOnFlushBegin(ColumnFamilyData* cfd, FileMetaData* file_meta,
                          const MutableCFOptions& mutable_cf_options,
                          int job_id, TableProperties prop);

661
  void NotifyOnFlushCompleted(ColumnFamilyData* cfd, FileMetaData* file_meta,
662
                              const MutableCFOptions& mutable_cf_options,
663
                              int job_id, TableProperties prop);
664

O
Ori Bernstein 已提交
665
  void NotifyOnCompactionCompleted(ColumnFamilyData* cfd,
666 667
                                   Compaction *c, const Status &st,
                                   const CompactionJobStats& job_stats,
668
                                   int job_id);
W
Wanning Jiang 已提交
669
  void NotifyOnMemTableSealed(ColumnFamilyData* cfd,
670
                              const MemTableInfo& mem_table_info);
O
Ori Bernstein 已提交
671

672
#ifndef ROCKSDB_LITE
673 674
  void NotifyOnExternalFileIngested(
      ColumnFamilyData* cfd, const ExternalSstFileIngestionJob& ingestion_job);
675
#endif  // !ROCKSDB_LITE
676

Y
Yueh-Hsuan Chiang 已提交
677 678 679 680 681 682
  void NewThreadStatusCfInfo(ColumnFamilyData* cfd) const;

  void EraseThreadStatusCfInfo(ColumnFamilyData* cfd) const;

  void EraseThreadStatusDbInfo() const;

683 684 685 686 687 688
  // If disable_memtable is set the application logic must guarantee that the
  // batch will still be skipped from memtable during the recovery. In
  // WriteCommitted it is guarnateed since disable_memtable is used for prepare
  // batch which will be written to memtable later during the commit, and in
  // WritePrepared it is guaranteed since it will be used only for WAL markers
  // which will never be written to memtable.
689 690 691
  // batch_cnt is expected to be non-zero in seq_per_batch mode and indicates
  // the number of sub-patches. A sub-patch is a subset of the write batch that
  // does not have duplicate keys.
A
agiardullo 已提交
692
  Status WriteImpl(const WriteOptions& options, WriteBatch* updates,
693 694
                   WriteCallback* callback = nullptr,
                   uint64_t* log_used = nullptr, uint64_t log_ref = 0,
695
                   bool disable_memtable = false, uint64_t* seq_used = nullptr,
696
                   size_t batch_cnt = 0,
697
                   PreReleaseCallback* pre_release_callback = nullptr);
698

699 700 701
  Status PipelinedWriteImpl(const WriteOptions& options, WriteBatch* updates,
                            WriteCallback* callback = nullptr,
                            uint64_t* log_used = nullptr, uint64_t log_ref = 0,
702 703
                            bool disable_memtable = false,
                            uint64_t* seq_used = nullptr);
704

705 706 707
  // batch_cnt is expected to be non-zero in seq_per_batch mode and indicates
  // the number of sub-patches. A sub-patch is a subset of the write batch that
  // does not have duplicate keys.
708 709
  Status WriteImplWALOnly(const WriteOptions& options, WriteBatch* updates,
                          WriteCallback* callback = nullptr,
710
                          uint64_t* log_used = nullptr, uint64_t log_ref = 0,
711
                          uint64_t* seq_used = nullptr, size_t batch_cnt = 0,
712
                          PreReleaseCallback* pre_release_callback = nullptr);
713

714 715
  uint64_t FindMinLogContainingOutstandingPrep();
  uint64_t FindMinPrepLogReferencedByMemTable();
716 717 718
  // write cached_recoverable_state_ to memtable if it is not empty
  // The writer must be the leader in write_thread_ and holding mutex_
  Status WriteRecoverableState();
A
agiardullo 已提交
719

720 721 722
  // Actual implementation of Close()
  Status CloseImpl();

J
jorlow@chromium.org 已提交
723 724
 private:
  friend class DB;
725
  friend class InternalStats;
726 727 728
  friend class PessimisticTransaction;
  friend class WriteCommittedTxn;
  friend class WritePreparedTxn;
M
Maysam Yabandeh 已提交
729
  friend class WritePreparedTxnDB;
730
  friend class WriteBatchWithIndex;
I
Igor Canadi 已提交
731
#ifndef ROCKSDB_LITE
L
Lei Jin 已提交
732
  friend class ForwardIterator;
I
Igor Canadi 已提交
733
#endif
734
  friend struct SuperVersion;
L
Lei Jin 已提交
735
  friend class CompactedDBImpl;
A
agiardullo 已提交
736
#ifndef NDEBUG
Y
Yi Wu 已提交
737
  friend class DBTest2_ReadCallbackTest_Test;
738
  friend class WriteCallbackTest_WriteWithCallbackTest_Test;
A
agiardullo 已提交
739
  friend class XFTransactionWriteHandler;
Y
Yi Wu 已提交
740
  friend class DBBlobIndexTest;
A
agiardullo 已提交
741
#endif
742
  struct CompactionState;
743

744
  struct WriteContext {
745
    SuperVersionContext superversion_context;
746 747
    autovector<MemTable*> memtables_to_free_;

748 749 750
    explicit WriteContext(bool create_superversion = false)
        : superversion_context(create_superversion) {}

751
    ~WriteContext() {
752
      superversion_context.Clean();
753 754 755 756 757
      for (auto& m : memtables_to_free_) {
        delete m;
      }
    }
  };
J
jorlow@chromium.org 已提交
758

759
  struct PrepickedCompaction;
760 761
  struct PurgeFileInfo;

J
jorlow@chromium.org 已提交
762 763 764
  // Recover the descriptor from persistent storage.  May do a significant
  // amount of work to recover recently logged updates.  Any changes to
  // be made to the descriptor are added to *edit.
765
  Status Recover(const std::vector<ColumnFamilyDescriptor>& column_families,
766 767
                 bool read_only = false, bool error_if_log_file_exist = false,
                 bool error_if_data_exists_in_logs = false);
J
jorlow@chromium.org 已提交
768 769 770

  void MaybeIgnoreError(Status* s) const;

771 772
  const Status CreateArchivalDirectory();

Y
Yi Wu 已提交
773 774 775 776 777 778
  Status CreateColumnFamilyImpl(const ColumnFamilyOptions& cf_options,
                                const std::string& cf_name,
                                ColumnFamilyHandle** handle);

  Status DropColumnFamilyImpl(ColumnFamilyHandle* column_family);

J
jorlow@chromium.org 已提交
779 780
  // Delete any unneeded files and stale in-memory entries.
  void DeleteObsoleteFiles();
781
  // Delete obsolete files and log status and information of file deletion
S
Siying Dong 已提交
782 783
  void DeleteObsoleteFileImpl(int job_id, const std::string& fname,
                              FileType type, uint64_t number, uint32_t path_id);
J
jorlow@chromium.org 已提交
784

I
Igor Canadi 已提交
785 786
  // Background process needs to call
  //     auto x = CaptureCurrentFileNumberInPendingOutputs()
787
  //     auto file_num = versions_->NewFileNumber();
I
Igor Canadi 已提交
788 789
  //     <do something>
  //     ReleaseFileNumberFromPendingOutputs(x)
790 791
  // This will protect any file with number `file_num` or greater from being
  // deleted while <do something> is running.
I
Igor Canadi 已提交
792 793 794 795 796 797 798 799 800 801 802 803
  // -----------
  // This function will capture current file number and append it to
  // pending_outputs_. This will prevent any background process to delete any
  // file created after this point.
  std::list<uint64_t>::iterator CaptureCurrentFileNumberInPendingOutputs();
  // This function should be called with the result of
  // CaptureCurrentFileNumberInPendingOutputs(). It then marks that any file
  // created between the calls CaptureCurrentFileNumberInPendingOutputs() and
  // ReleaseFileNumberFromPendingOutputs() can now be deleted (if it's not live
  // and blocked by any other pending_outputs_ calls)
  void ReleaseFileNumberFromPendingOutputs(std::list<uint64_t>::iterator v);

804 805
  Status SyncClosedLogs(JobContext* job_context);

806
  // Flush the in-memory write buffer to storage.  Switches to a new
J
jorlow@chromium.org 已提交
807
  // log-file/memtable and writes a new descriptor iff successful.
I
Igor Canadi 已提交
808 809 810 811
  Status FlushMemTableToOutputFile(ColumnFamilyData* cfd,
                                   const MutableCFOptions& mutable_cf_options,
                                   bool* madeProgress, JobContext* job_context,
                                   LogBuffer* log_buffer);
J
jorlow@chromium.org 已提交
812

S
Stanislau Hlebik 已提交
813 814
  // REQUIRES: log_numbers are sorted in ascending order
  Status RecoverLogFiles(const std::vector<uint64_t>& log_numbers,
815
                         SequenceNumber* next_sequence, bool read_only);
J
jorlow@chromium.org 已提交
816

817
  // The following two methods are used to flush a memtable to
Y
Yueh-Hsuan Chiang 已提交
818
  // storage. The first one is used at database RecoveryTime (when the
819 820 821
  // database is opened) and is heavyweight because it holds the mutex
  // for the entire period. The second method WriteLevel0Table supports
  // concurrent flush memtables to storage.
822 823
  Status WriteLevel0TableForRecovery(int job_id, ColumnFamilyData* cfd,
                                     MemTable* mem, VersionEdit* edit);
S
sdong 已提交
824 825 826

  // num_bytes: for slowdown case, delay time is calculated based on
  //            `num_bytes` going through.
M
Maysam Yabandeh 已提交
827
  Status DelayWrite(uint64_t num_bytes, const WriteOptions& write_options);
828

829 830 831
  Status ThrottleLowPriWritesIfNeeded(const WriteOptions& write_options,
                                      WriteBatch* my_batch);

I
Igor Canadi 已提交
832
  Status ScheduleFlushes(WriteContext* context);
833

I
Igor Canadi 已提交
834
  Status SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context);
S
Stanislau Hlebik 已提交
835

H
heyongqiang 已提交
836
  // Force current memtable contents to be flushed.
837
  Status FlushMemTable(ColumnFamilyData* cfd, const FlushOptions& options,
838
                       FlushReason flush_reason, bool writes_stopped = false);
H
heyongqiang 已提交
839

840 841 842 843 844 845
  // Wait for memtable flushed.
  // If flush_memtable_id is non-null, wait until the memtable with the ID
  // gets flush. Otherwise, wait until the column family don't have any
  // memtable pending flush.
  Status WaitForFlushMemTable(ColumnFamilyData* cfd,
                              const uint64_t* flush_memtable_id = nullptr);
H
heyongqiang 已提交
846

847
  // REQUIRES: mutex locked
848
  Status SwitchWAL(WriteContext* write_context);
849 850 851 852 853

  // REQUIRES: mutex locked
  Status HandleWriteBufferFull(WriteContext* write_context);

  // REQUIRES: mutex locked
854 855
  Status PreprocessWrite(const WriteOptions& write_options, bool* need_log_sync,
                         WriteContext* write_context);
856

857
  WriteBatch* MergeBatch(const WriteThread::WriteGroup& write_group,
858 859
                         WriteBatch* tmp_batch, size_t* write_with_wal,
                         WriteBatch** to_be_cached_state);
860 861 862 863

  Status WriteToWAL(const WriteBatch& merged_batch, log::Writer* log_writer,
                    uint64_t* log_used, uint64_t* log_size);

864
  Status WriteToWAL(const WriteThread::WriteGroup& write_group,
865 866 867 868 869 870
                    log::Writer* log_writer, uint64_t* log_used,
                    bool need_log_sync, bool need_log_dir_sync,
                    SequenceNumber sequence);

  Status ConcurrentWriteToWAL(const WriteThread::WriteGroup& write_group,
                              uint64_t* log_used, SequenceNumber* last_sequence,
871
                              size_t seq_inc);
872

873
  // Used by WriteImpl to update bg_error_ if paranoid check is enabled.
874
  void WriteCallbackStatusCheck(const Status& status);
875 876 877

  // Used by WriteImpl to update bg_error_ in case of memtable insert error.
  void MemTableInsertStatusCheck(const Status& memtable_insert_status);
Y
Yi Wu 已提交
878

I
Igor Canadi 已提交
879
#ifndef ROCKSDB_LITE
880

881 882 883 884 885 886
  Status CompactFilesImpl(const CompactionOptions& compact_options,
                          ColumnFamilyData* cfd, Version* version,
                          const std::vector<std::string>& input_file_names,
                          const int output_level, int output_path_id,
                          JobContext* job_context, LogBuffer* log_buffer);

887 888 889
  // Wait for current IngestExternalFile() calls to finish.
  // REQUIRES: mutex_ held
  void WaitForIngestFile();
890

891
#else
892
  // IngestExternalFile is not supported in ROCKSDB_LITE so this function
893
  // will be no-op
894
  void WaitForIngestFile() {}
I
Igor Canadi 已提交
895
#endif  // ROCKSDB_LITE
896 897 898

  ColumnFamilyData* GetColumnFamilyDataByName(const std::string& cf_name);

899
  void MaybeScheduleFlushOrCompaction();
900
  void SchedulePendingFlush(ColumnFamilyData* cfd, FlushReason flush_reason);
901
  void SchedulePendingCompaction(ColumnFamilyData* cfd);
902 903
  void SchedulePendingPurge(std::string fname, FileType type, uint64_t number,
                            uint32_t path_id, int job_id);
904
  static void BGWorkCompaction(void* arg);
905 906 907
  // Runs a pre-chosen universal compaction involving bottom level in a
  // separate, bottom-pri thread pool.
  static void BGWorkBottomCompaction(void* arg);
908
  static void BGWorkFlush(void* db);
909
  static void BGWorkPurge(void* arg);
910
  static void UnscheduleCallback(void* arg);
911 912
  void BackgroundCallCompaction(PrepickedCompaction* prepicked_compaction,
                                Env::Priority bg_thread_pri);
913
  void BackgroundCallFlush();
914
  void BackgroundCallPurge();
I
Igor Canadi 已提交
915
  Status BackgroundCompaction(bool* madeProgress, JobContext* job_context,
916 917
                              LogBuffer* log_buffer,
                              PrepickedCompaction* prepicked_compaction);
I
Igor Canadi 已提交
918
  Status BackgroundFlush(bool* madeProgress, JobContext* job_context,
H
Haobo Xu 已提交
919
                         LogBuffer* log_buffer);
J
jorlow@chromium.org 已提交
920

921 922
  void PrintStatistics();

923
  // dump rocksdb.stats to LOG
924 925
  void MaybeDumpStats();

926 927
  // Return the minimum empty level that could hold the total data in the
  // input level. Return the input level, if such level could not be found.
928 929
  int FindMinimumEmptyLevelFitting(ColumnFamilyData* cfd,
      const MutableCFOptions& mutable_cf_options, int level);
930

931 932 933
  // Move the files in the input level to the target level.
  // If target_level < 0, automatically calculate the minimum level that could
  // hold the data set.
I
Igor Canadi 已提交
934
  Status ReFitLevel(ColumnFamilyData* cfd, int level, int target_level = -1);
935

936 937 938
  // helper functions for adding and removing from flush & compaction queues
  void AddToCompactionQueue(ColumnFamilyData* cfd);
  ColumnFamilyData* PopFirstFromCompactionQueue();
939
  void AddToFlushQueue(ColumnFamilyData* cfd, FlushReason flush_reason);
940 941
  ColumnFamilyData* PopFirstFromFlushQueue();

942 943 944
  // helper function to call after some of the logs_ were synced
  void MarkLogsSynced(uint64_t up_to, bool synced_dir, const Status& status);

945 946
  const Snapshot* GetSnapshotImpl(bool is_write_conflict_boundary);

947 948
  uint64_t GetMaxTotalWalSize() const;

949
  Status CloseHelper();
950

J
jorlow@chromium.org 已提交
951
  // table_cache_ provides its own synchronization
I
Igor Canadi 已提交
952
  std::shared_ptr<Cache> table_cache_;
J
jorlow@chromium.org 已提交
953

954
  // Lock over the persistent DB state.  Non-nullptr iff successfully acquired.
J
jorlow@chromium.org 已提交
955 956
  FileLock* db_lock_;

957
  // In addition to mutex_, log_write_mutex_ protected writes to logs_ and
958
  // logfile_number_. With two_write_queues it also protects alive_log_files_,
959 960
  // and log_empty_. Refer to the definition of each variable below for more
  // details.
961
  InstrumentedMutex log_write_mutex_;
J
jorlow@chromium.org 已提交
962
  // State below is protected by mutex_
963
  // With two_write_queues enabled, some of the variables that accessed during
964 965 966
  // WriteToWAL need different synchronization: log_empty_, alive_log_files_,
  // logs_, logfile_number_. Refer to the definition of each variable below for
  // more description.
967
  mutable InstrumentedMutex mutex_;
968

I
Igor Canadi 已提交
969
  std::atomic<bool> shutting_down_;
970 971
  // This condition variable is signaled on these conditions:
  // * whenever bg_compaction_scheduled_ goes down to 0
972
  // * if AnyManualCompaction, whenever a compaction finishes, even if it hasn't
973 974
  // made any progress
  // * whenever a compaction made any progress
975 976 977
  // * whenever bg_flush_scheduled_ or bg_purge_scheduled_ value decreases
  // (i.e. whenever a flush is done, even if it didn't make any progress)
  // * whenever there is an error in background purge, flush or compaction
978
  // * whenever num_running_ingest_file_ goes to 0.
979 980
  // * whenever pending_purge_obsolete_files_ goes to 0.
  // * whenever disable_delete_obsolete_files_ goes to 0.
981 982
  // * whenever SetOptions successfully updates options.
  // * whenever a column family is dropped.
983
  InstrumentedCondVar bg_cv_;
984 985 986 987
  // Writes are protected by locking both mutex_ and log_write_mutex_, and reads
  // must be under either mutex_ or log_write_mutex_. Since after ::Open,
  // logfile_number_ is currently updated only in write_thread_, it can be read
  // from the same write_thread_ without any locks.
988
  uint64_t logfile_number_;
S
Sage Weil 已提交
989
  std::deque<uint64_t>
990
      log_recycle_files_;  // a list of log files that we can recycle
991
  bool log_dir_synced_;
992
  // Without two_write_queues, read and writes to log_empty_ are protected by
993 994
  // mutex_. Since it is currently updated/read only in write_thread_, it can be
  // accessed from the same write_thread_ without any locks. With
995
  // two_write_queues writes, where it can be updated in different threads,
996 997
  // read and writes are protected by log_write_mutex_ instead. This is to avoid
  // expesnive mutex_ lock during WAL write, which update log_empty_.
I
Igor Canadi 已提交
998
  bool log_empty_;
999
  ColumnFamilyHandleImpl* default_cf_handle_;
1000
  InternalStats* default_cf_internal_stats_;
1001
  unique_ptr<ColumnFamilyMemTablesImpl> column_family_memtables_;
I
Igor Canadi 已提交
1002 1003
  struct LogFileNumberSize {
    explicit LogFileNumberSize(uint64_t _number)
1004
        : number(_number) {}
I
Igor Canadi 已提交
1005 1006
    void AddSize(uint64_t new_size) { size += new_size; }
    uint64_t number;
1007 1008 1009 1010
    uint64_t size = 0;
    bool getting_flushed = false;
  };
  struct LogWriterNumber {
1011 1012 1013 1014 1015 1016 1017 1018 1019
    // pass ownership of _writer
    LogWriterNumber(uint64_t _number, log::Writer* _writer)
        : number(_number), writer(_writer) {}

    log::Writer* ReleaseWriter() {
      auto* w = writer;
      writer = nullptr;
      return w;
    }
1020 1021
    Status ClearWriter() {
      Status s = writer->WriteBuffer();
1022 1023
      delete writer;
      writer = nullptr;
1024
      return s;
1025 1026
    }

1027
    uint64_t number;
1028 1029 1030
    // Visual Studio doesn't support deque's member to be noncopyable because
    // of a unique_ptr as a member.
    log::Writer* writer;  // own
1031 1032
    // true for some prefix of logs_
    bool getting_synced = false;
I
Igor Canadi 已提交
1033
  };
1034
  // Without two_write_queues, read and writes to alive_log_files_ are
1035 1036
  // protected by mutex_. However since back() is never popped, and push_back()
  // is done only from write_thread_, the same thread can access the item
1037
  // reffered by back() without mutex_. With two_write_queues_, writes
1038 1039
  // are protected by locking both mutex_ and log_write_mutex_, and reads must
  // be under either mutex_ or log_write_mutex_.
I
Igor Canadi 已提交
1040
  std::deque<LogFileNumberSize> alive_log_files_;
1041 1042
  // Log files that aren't fully synced, and the current log file.
  // Synchronization:
1043 1044 1045 1046 1047
  //  - push_back() is done from write_thread_ with locked mutex_ and
  //  log_write_mutex_
  //  - pop_front() is done from any thread with locked mutex_ and
  //  log_write_mutex_
  //  - reads are done with either locked mutex_ or log_write_mutex_
1048
  //  - back() and items with getting_synced=true are not popped,
1049 1050 1051 1052 1053
  //  - The same thread that sets getting_synced=true will reset it.
  //  - it follows that the object referred by back() can be safely read from
  //  the write_thread_ without using mutex
  //  - it follows that the items with getting_synced=true can be safely read
  //  from the same thread that has set getting_synced=true
1054 1055 1056
  std::deque<LogWriterNumber> logs_;
  // Signaled when getting_synced becomes false for some of the logs_.
  InstrumentedCondVar log_sync_cv_;
1057 1058 1059 1060 1061
  // This is the app-level state that is written to the WAL but will be used
  // only during recovery. Using this feature enables not writing the state to
  // memtable on normal writes and hence improving the throughput. Each new
  // write of the state will replace the previous state entirely even if the
  // keys in the two consecuitive states do not overlap.
1062
  // It is protected by log_write_mutex_ when two_write_queues_ is enabled.
1063 1064 1065
  // Otherwise only the heaad of write_thread_ can access it.
  WriteBatch cached_recoverable_state_;
  std::atomic<bool> cached_recoverable_state_empty_ = {true};
A
Aaron Gao 已提交
1066
  std::atomic<uint64_t> total_log_size_;
I
Igor Canadi 已提交
1067 1068 1069
  // only used for dynamically adjusting max_total_wal_size. it is a sum of
  // [write_buffer_size * max_write_buffer_number] over all column families
  uint64_t max_total_in_memory_state_;
1070 1071 1072
  // If true, we have only one (default) column family. We use this to optimize
  // some code-paths
  bool single_column_family_mode_;
1073 1074 1075
  // If this is non-empty, we need to delete these log files in background
  // threads. Protected by db mutex.
  autovector<log::Writer*> logs_to_free_;
I
Igor Canadi 已提交
1076

S
sdong 已提交
1077 1078
  bool is_snapshot_supported_;

1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106
  // Class to maintain directories for all database paths other than main one.
  class Directories {
   public:
    Status SetDirectories(Env* env, const std::string& dbname,
                          const std::string& wal_dir,
                          const std::vector<DbPath>& data_paths);

    Directory* GetDataDir(size_t path_id);

    Directory* GetWalDir() {
      if (wal_dir_) {
        return wal_dir_.get();
      }
      return db_dir_.get();
    }

    Directory* GetDbDir() { return db_dir_.get(); }

   private:
    std::unique_ptr<Directory> db_dir_;
    std::vector<std::unique_ptr<Directory>> data_dirs_;
    std::unique_ptr<Directory> wal_dir_;

    Status CreateAndNewDirectory(Env* env, const std::string& dirname,
                                 std::unique_ptr<Directory>* directory) const;
  };

  Directories directories_;
1107

1108
  WriteBufferManager* write_buffer_manager_;
1109

I
Igor Canadi 已提交
1110
  WriteThread write_thread_;
1111
  WriteBatch tmp_batch_;
1112 1113 1114
  // The write thread when the writers have no memtable write. This will be used
  // in 2PC to batch the prepares separately from the serial commit.
  WriteThread nonmem_write_thread_;
1115

1116
  WriteController write_controller_;
S
sdong 已提交
1117

1118 1119
  unique_ptr<RateLimiter> low_pri_write_rate_limiter_;

S
sdong 已提交
1120 1121
  // Size of the last batch group. In slowdown mode, next write needs to
  // sleep if it uses up the quota.
1122 1123
  // Note: This is to protect memtable and compaction. If the batch only writes
  // to the WAL its size need not to be included in this.
S
sdong 已提交
1124 1125
  uint64_t last_batch_group_size_;

I
Igor Canadi 已提交
1126
  FlushScheduler flush_scheduler_;
1127

J
jorlow@chromium.org 已提交
1128 1129
  SnapshotList snapshots_;

I
Igor Canadi 已提交
1130 1131 1132 1133 1134 1135 1136 1137 1138 1139
  // For each background job, pending_outputs_ keeps the current file number at
  // the time that background job started.
  // FindObsoleteFiles()/PurgeObsoleteFiles() never deletes any file that has
  // number bigger than any of the file number in pending_outputs_. Since file
  // numbers grow monotonically, this also means that pending_outputs_ is always
  // sorted. After a background job is done executing, its file number is
  // deleted from pending_outputs_, which allows PurgeObsoleteFiles() to clean
  // it up.
  // State is protected with db mutex.
  std::list<uint64_t> pending_outputs_;
J
jorlow@chromium.org 已提交
1140

1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153
  // PurgeFileInfo is a structure to hold information of files to be deleted in
  // purge_queue_
  struct PurgeFileInfo {
    std::string fname;
    FileType type;
    uint64_t number;
    uint32_t path_id;
    int job_id;
    PurgeFileInfo(std::string fn, FileType t, uint64_t num, uint32_t pid,
                  int jid)
        : fname(fn), type(t), number(num), path_id(pid), job_id(jid) {}
  };

1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177
  // flush_queue_ and compaction_queue_ hold column families that we need to
  // flush and compact, respectively.
  // A column family is inserted into flush_queue_ when it satisfies condition
  // cfd->imm()->IsFlushPending()
  // A column family is inserted into compaction_queue_ when it satisfied
  // condition cfd->NeedsCompaction()
  // Column families in this list are all Ref()-erenced
  // TODO(icanadi) Provide some kind of ReferencedColumnFamily class that will
  // do RAII on ColumnFamilyData
  // Column families are in this queue when they need to be flushed or
  // compacted. Consumers of these queues are flush and compaction threads. When
  // column family is put on this queue, we increase unscheduled_flushes_ and
  // unscheduled_compactions_. When these variables are bigger than zero, that
  // means we need to schedule background threads for compaction and thread.
  // Once the background threads are scheduled, we decrease unscheduled_flushes_
  // and unscheduled_compactions_. That way we keep track of number of
  // compaction and flush threads we need to schedule. This scheduling is done
  // in MaybeScheduleFlushOrCompaction()
  // invariant(column family present in flush_queue_ <==>
  // ColumnFamilyData::pending_flush_ == true)
  std::deque<ColumnFamilyData*> flush_queue_;
  // invariant(column family present in compaction_queue_ <==>
  // ColumnFamilyData::pending_compaction_ == true)
  std::deque<ColumnFamilyData*> compaction_queue_;
1178 1179 1180

  // A queue to store filenames of the files to be purged
  std::deque<PurgeFileInfo> purge_queue_;
1181 1182 1183

  // A queue to store log writers to close
  std::deque<log::Writer*> logs_to_free_queue_;
1184 1185
  int unscheduled_flushes_;
  int unscheduled_compactions_;
1186

1187 1188 1189 1190
  // count how many background compactions are running or have been scheduled in
  // the BOTTOM pool
  int bg_bottom_compaction_scheduled_;

1191
  // count how many background compactions are running or have been scheduled
1192
  int bg_compaction_scheduled_;
J
jorlow@chromium.org 已提交
1193

1194 1195 1196
  // stores the number of compactions are currently running
  int num_running_compactions_;

1197 1198 1199
  // number of background memtable flush jobs, submitted to the HIGH pool
  int bg_flush_scheduled_;

1200 1201 1202
  // stores the number of flushes are currently running
  int num_running_flushes_;

1203 1204 1205
  // number of background obsolete file purge jobs, submitted to the HIGH pool
  int bg_purge_scheduled_;

H
hans@chromium.org 已提交
1206
  // Information for a manual compaction
1207
  struct ManualCompactionState {
I
Igor Canadi 已提交
1208
    ColumnFamilyData* cfd;
1209 1210
    int input_level;
    int output_level;
1211
    uint32_t output_path_id;
L
Lei Jin 已提交
1212
    Status status;
1213
    bool done;
1214
    bool in_progress;             // compaction request being processed?
1215 1216 1217
    bool incomplete;              // only part of requested range compacted
    bool exclusive;               // current behavior of only one manual
    bool disallow_trivial_move;   // Force actual compaction to run
1218 1219
    const InternalKey* begin;     // nullptr means beginning of key range
    const InternalKey* end;       // nullptr means end of key range
1220
    InternalKey* manual_end;      // how far we are compacting
1221
    InternalKey tmp_storage;      // Used to keep track of compaction progress
1222
    InternalKey tmp_storage1;     // Used to keep track of compaction progress
1223 1224 1225
  };
  struct PrepickedCompaction {
    // background compaction takes ownership of `compaction`.
1226
    Compaction* compaction;
1227 1228 1229
    // caller retains ownership of `manual_compaction_state` as it is reused
    // across background compactions.
    ManualCompactionState* manual_compaction_state;  // nullptr if non-manual
1230
  };
1231
  std::deque<ManualCompactionState*> manual_compaction_dequeue_;
1232 1233

  struct CompactionArg {
1234
    // caller retains ownership of `db`.
1235
    DBImpl* db;
1236 1237
    // background compaction takes ownership of `prepicked_compaction`.
    PrepickedCompaction* prepicked_compaction;
H
hans@chromium.org 已提交
1238
  };
J
jorlow@chromium.org 已提交
1239 1240 1241 1242

  // Have we encountered a background error in paranoid mode?
  Status bg_error_;

1243
  // shall we disable deletion of obsolete files
1244 1245 1246 1247 1248 1249
  // if 0 the deletion is enabled.
  // if non-zero, files will not be getting deleted
  // This enables two different threads to call
  // EnableFileDeletions() and DisableFileDeletions()
  // without any synchronization
  int disable_delete_obsolete_files_;
1250

1251 1252 1253 1254
  // Number of times FindObsoleteFiles has found deletable files and the
  // corresponding call to PurgeObsoleteFiles has not yet finished.
  int pending_purge_obsolete_files_;

1255 1256 1257
  // last time when DeleteObsoleteFiles with full scan was executed. Originaly
  // initialized with startup time.
  uint64_t delete_obsolete_files_last_run_;
1258

1259
  // last time stats were dumped to LOG
H
Haobo Xu 已提交
1260
  std::atomic<uint64_t> last_stats_dump_time_microsec_;
1261

1262 1263 1264 1265
  // Each flush or compaction gets its own job id. this counter makes sure
  // they're unique
  std::atomic<int> next_job_id_;

1266 1267 1268
  // A flag indicating whether the current rocksdb database has any
  // data that is not yet persisted into either WAL or SST file.
  // Used when disableWAL is true.
1269
  std::atomic<bool> has_unpersisted_data_;
1270 1271 1272 1273 1274 1275 1276 1277 1278

  // if an attempt was made to flush all column families that
  // the oldest log depends on but uncommited data in the oldest
  // log prevents the log from being released.
  // We must attempt to free the dependent memtables again
  // at a later time after the transaction in the oldest
  // log is fully commited.
  bool unable_to_flush_oldest_log_;

H
heyongqiang 已提交
1279
  static const int KEEP_LOG_FILE_NUM = 1000;
D
Dmitri Smirnov 已提交
1280
  // MSVC version 1800 still does not have constexpr for ::max()
1281
  static const uint64_t kNoTimeOut = port::kMaxUint64;
D
Dmitri Smirnov 已提交
1282

H
heyongqiang 已提交
1283
  std::string db_absolute_path_;
H
heyongqiang 已提交
1284

1285
  // The options to access storage files
L
Lei Jin 已提交
1286
  const EnvOptions env_options_;
1287

1288 1289 1290
  // Additonal options for compaction and flush
  EnvOptions env_options_for_compaction_;

1291
  // Number of running IngestExternalFile() calls.
1292
  // REQUIRES: mutex held
1293
  int num_running_ingest_file_;
1294

I
Igor Canadi 已提交
1295 1296 1297 1298
#ifndef ROCKSDB_LITE
  WalManager wal_manager_;
#endif  // ROCKSDB_LITE

I
Igor Canadi 已提交
1299 1300 1301
  // Unified interface for logging events
  EventLogger event_logger_;

1302
  // A value of > 0 temporarily disables scheduling of background work
1303
  int bg_work_paused_;
1304

1305 1306 1307
  // A value of > 0 temporarily disables scheduling of background compaction
  int bg_compaction_paused_;

1308 1309 1310
  // Guard against multiple concurrent refitting
  bool refitting_level_;

1311 1312 1313
  // Indicate DB was opened successfully
  bool opened_successfully_;

1314 1315 1316
  // REQUIRES: logs_with_prep_mutex_ held
  //
  // sorted list of log numbers still containing prepared data.
1317 1318
  // this is used by FindObsoleteFiles to determine which
  // flushed logs we must keep around because they still
1319 1320 1321 1322 1323 1324 1325
  // contain prepared data which has not been committed or rolled back
  struct LogCnt {
    uint64_t log;  // the log number
    uint64_t cnt;  // number of prepared sections in the log
  };
  std::vector<LogCnt> logs_with_prep_;
  std::mutex logs_with_prep_mutex_;
1326

1327 1328 1329
  // REQUIRES: prepared_section_completed_mutex_ held
  //
  // to be used in conjunction with logs_with_prep_.
1330
  // once a transaction with data in log L is committed or rolled back
1331 1332 1333
  // rather than updating logs_with_prep_ directly we keep track of that
  // in prepared_section_completed_ which maps LOG -> instance_count. This helps
  // avoiding contention between a commit thread and the prepare threads.
1334
  //
Y
yizhu.sun 已提交
1335
  // when trying to determine the minimum log still active we first
1336 1337 1338
  // consult logs_with_prep_. while that root value maps to
  // an equal value in prepared_section_completed_ we erase the log from
  // both logs_with_prep_ and prepared_section_completed_.
1339
  std::unordered_map<uint64_t, uint64_t> prepared_section_completed_;
1340
  std::mutex prepared_section_completed_mutex_;
1341

Y
Yi Wu 已提交
1342 1343 1344 1345
  // Callback for compaction to check if a key is visible to a snapshot.
  // REQUIRES: mutex held
  std::unique_ptr<SnapshotChecker> snapshot_checker_;

J
jorlow@chromium.org 已提交
1346 1347 1348 1349
  // No copying allowed
  DBImpl(const DBImpl&);
  void operator=(const DBImpl&);

I
Igor Canadi 已提交
1350
  // Background threads call this function, which is just a wrapper around
I
Igor Canadi 已提交
1351
  // the InstallSuperVersion() function. Background threads carry
1352
  // sv_context which can have new_superversion already
I
Igor Canadi 已提交
1353
  // allocated.
1354 1355 1356
  // All ColumnFamily state changes go through this function. Here we analyze
  // the new state and we schedule background work if we detect that the new
  // state needs flush or compaction.
1357 1358
  void InstallSuperVersionAndScheduleWork(
      ColumnFamilyData* cfd, SuperVersionContext* sv_context,
I
Igor Canadi 已提交
1359
      const MutableCFOptions& mutable_cf_options);
I
Igor Canadi 已提交
1360

I
Igor Canadi 已提交
1361
#ifndef ROCKSDB_LITE
I
Igor Canadi 已提交
1362 1363 1364
  using DB::GetPropertiesOfAllTables;
  virtual Status GetPropertiesOfAllTables(ColumnFamilyHandle* column_family,
                                          TablePropertiesCollection* props)
1365
      override;
1366
  virtual Status GetPropertiesOfTablesInRange(
1367
      ColumnFamilyHandle* column_family, const Range* range, std::size_t n,
1368 1369
      TablePropertiesCollection* props) override;

I
Igor Canadi 已提交
1370
#endif  // ROCKSDB_LITE
1371

1372
  bool GetIntPropertyInternal(ColumnFamilyData* cfd,
1373 1374
                              const DBPropertyInfo& property_info,
                              bool is_locked, uint64_t* value);
1375 1376 1377

  bool HasPendingManualCompaction();
  bool HasExclusiveManualCompaction();
1378 1379 1380
  void AddManualCompaction(ManualCompactionState* m);
  void RemoveManualCompaction(ManualCompactionState* m);
  bool ShouldntRunManualCompaction(ManualCompactionState* m);
1381
  bool HaveManualCompaction(ColumnFamilyData* cfd);
1382
  bool MCOverlap(ManualCompactionState* m, ManualCompactionState* m1);
1383

1384
  size_t GetWalPreallocateBlockSize(uint64_t write_buffer_size) const;
S
Stream  
Shaohua Li 已提交
1385 1386 1387
  Env::WriteLifeTimeHint CalculateWALWriteHint() {
    return Env::WLTH_SHORT;
  }
1388

1389 1390
  // When set, we use a separate queue for writes that dont write to memtable.
  // In 2PC these are the writes at Prepare phase.
1391
  const bool two_write_queues_;
1392
  const bool manual_wal_flush_;
1393 1394 1395 1396 1397 1398
  // Increase the sequence number after writing each batch, whether memtable is
  // disabled for that or not. Otherwise the sequence number is increased after
  // writing each key into memtable. This implies that when disable_memtable is
  // set, the seq is not increased at all.
  //
  // Default: false
1399
  const bool seq_per_batch_;
1400 1401 1402
  // LastSequence also indicates last published sequence visibile to the
  // readers. Otherwise LastPublishedSequence should be used.
  const bool last_seq_same_as_publish_seq_;
1403 1404 1405
  // It indicates that a customized gc algorithm must be used for
  // flush/compaction and if it is not provided vis SnapshotChecker, we should
  // disable gc to be safe.
1406
  const bool use_custom_gc_;
1407 1408 1409 1410 1411 1412 1413

  // Clients must periodically call SetPreserveDeletesSequenceNumber()
  // to advance this seqnum. Default value is 0 which means ALL deletes are
  // preserved. Note that this has no effect if DBOptions.preserve_deletes
  // is set to false.
  std::atomic<SequenceNumber> preserve_deletes_seqnum_;
  const bool preserve_deletes_;
1414 1415 1416

  // Flag to check whether Close() has been called on this DB
  bool closed_;
J
jorlow@chromium.org 已提交
1417 1418 1419 1420
};

extern Options SanitizeOptions(const std::string& db,
                               const Options& src);
1421

1422
extern DBOptions SanitizeOptions(const std::string& db, const DBOptions& src);
S
Siying Dong 已提交
1423

S
Siying Dong 已提交
1424 1425 1426
extern CompressionType GetCompressionFlush(
    const ImmutableCFOptions& ioptions,
    const MutableCFOptions& mutable_cf_options);
1427

M
miguelportilla 已提交
1428 1429 1430 1431 1432 1433 1434
// Fix user-supplied options to be reasonable
template <class T, class V>
static void ClipToRange(T* ptr, V minvalue, V maxvalue) {
  if (static_cast<V>(*ptr) > maxvalue) *ptr = maxvalue;
  if (static_cast<V>(*ptr) < minvalue) *ptr = minvalue;
}

1435
}  // namespace rocksdb