db_impl.cc 154.8 KB
Newer Older
1 2 3 4 5
//  Copyright (c) 2013, Facebook, Inc.  All rights reserved.
//  This source code is licensed under the BSD-style license found in the
//  LICENSE file in the root directory of this source tree. An additional grant
//  of patent rights can be found in the PATENTS file in the same directory.
//
J
jorlow@chromium.org 已提交
6 7 8 9 10 11
// 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.

#include "db/db_impl.h"

L
liuhuahang 已提交
12
#ifndef __STDC_FORMAT_MACROS
I
Igor Canadi 已提交
13
#define __STDC_FORMAT_MACROS
L
liuhuahang 已提交
14 15
#endif

I
Igor Canadi 已提交
16
#include <inttypes.h>
J
jorlow@chromium.org 已提交
17
#include <algorithm>
18 19
#include <climits>
#include <cstdio>
J
jorlow@chromium.org 已提交
20
#include <set>
21
#include <stdexcept>
22 23
#include <stdint.h>
#include <string>
24
#include <unordered_set>
25
#include <unordered_map>
T
Tomislav Novak 已提交
26
#include <utility>
27
#include <vector>
28

J
jorlow@chromium.org 已提交
29
#include "db/builder.h"
I
Igor Canadi 已提交
30
#include "db/flush_job.h"
I
Igor Canadi 已提交
31
#include "db/compaction_job.h"
32
#include "db/db_iter.h"
K
kailiu 已提交
33
#include "db/dbformat.h"
34
#include "db/event_helpers.h"
J
jorlow@chromium.org 已提交
35
#include "db/filename.h"
I
Igor Canadi 已提交
36
#include "db/job_context.h"
J
jorlow@chromium.org 已提交
37 38
#include "db/log_reader.h"
#include "db/log_writer.h"
V
Venkatesh Radhakrishnan 已提交
39
#include "db/managed_iterator.h"
J
jorlow@chromium.org 已提交
40
#include "db/memtable.h"
K
kailiu 已提交
41
#include "db/memtable_list.h"
42
#include "db/merge_context.h"
43
#include "db/merge_helper.h"
J
jorlow@chromium.org 已提交
44
#include "db/table_cache.h"
K
kailiu 已提交
45
#include "db/table_properties_collector.h"
L
Lei Jin 已提交
46
#include "db/forward_iterator.h"
47
#include "db/transaction_log_impl.h"
J
jorlow@chromium.org 已提交
48
#include "db/version_set.h"
49
#include "db/writebuffer.h"
J
jorlow@chromium.org 已提交
50
#include "db/write_batch_internal.h"
A
agiardullo 已提交
51
#include "db/write_callback.h"
52
#include "port/port.h"
I
Igor Canadi 已提交
53
#include "rocksdb/cache.h"
54
#include "port/likely.h"
55 56 57 58
#include "rocksdb/compaction_filter.h"
#include "rocksdb/db.h"
#include "rocksdb/env.h"
#include "rocksdb/merge_operator.h"
I
Igor Canadi 已提交
59
#include "rocksdb/version.h"
60 61
#include "rocksdb/statistics.h"
#include "rocksdb/status.h"
S
Siying Dong 已提交
62
#include "rocksdb/table.h"
J
jorlow@chromium.org 已提交
63
#include "table/block.h"
64
#include "table/block_based_table_factory.h"
J
jorlow@chromium.org 已提交
65
#include "table/merger.h"
K
kailiu 已提交
66
#include "table/table_builder.h"
J
jorlow@chromium.org 已提交
67
#include "table/two_level_iterator.h"
68
#include "util/auto_roll_logger.h"
K
kailiu 已提交
69
#include "util/autovector.h"
70
#include "util/build_version.h"
J
jorlow@chromium.org 已提交
71
#include "util/coding.h"
I
Igor Canadi 已提交
72
#include "util/compression.h"
73
#include "util/db_info_dumper.h"
74
#include "util/file_util.h"
I
Igor Canadi 已提交
75
#include "util/hash_skiplist_rep.h"
76
#include "util/hash_linklist_rep.h"
J
jorlow@chromium.org 已提交
77
#include "util/logging.h"
H
Haobo Xu 已提交
78
#include "util/log_buffer.h"
J
jorlow@chromium.org 已提交
79
#include "util/mutexlock.h"
80
#include "util/perf_context_imp.h"
81
#include "util/iostats_context_imp.h"
82
#include "util/stop_watch.h"
83
#include "util/sync_point.h"
84
#include "util/string_util.h"
85 86
#include "util/thread_status_updater.h"
#include "util/thread_status_util.h"
V
Venkatesh Radhakrishnan 已提交
87
#include "util/xfunc.h"
J
jorlow@chromium.org 已提交
88

89
namespace rocksdb {
J
jorlow@chromium.org 已提交
90

91
const std::string kDefaultColumnFamilyName("default");
92

I
Igor Canadi 已提交
93
void DumpRocksDBBuildVersion(Logger * log);
94

S
Stanislau Hlebik 已提交
95 96
struct DBImpl::WriteContext {
  autovector<SuperVersion*> superversions_to_free_;
97
  autovector<MemTable*> memtables_to_free_;
98
  bool schedule_bg_work_ = false;
S
Stanislau Hlebik 已提交
99 100 101 102 103

  ~WriteContext() {
    for (auto& sv : superversions_to_free_) {
      delete sv;
    }
104 105 106
    for (auto& m : memtables_to_free_) {
      delete m;
    }
S
Stanislau Hlebik 已提交
107 108 109
  }
};

J
jorlow@chromium.org 已提交
110 111 112
Options SanitizeOptions(const std::string& dbname,
                        const InternalKeyComparator* icmp,
                        const Options& src) {
113
  auto db_options = SanitizeOptions(dbname, DBOptions(src));
114
  auto cf_options = SanitizeOptions(db_options, icmp, ColumnFamilyOptions(src));
115 116 117 118 119
  return Options(db_options, cf_options);
}

DBOptions SanitizeOptions(const std::string& dbname, const DBOptions& src) {
  DBOptions result = src;
L
Lei Jin 已提交
120

121 122
  // result.max_open_files means an "infinite" open files.
  if (result.max_open_files != -1) {
123
    ClipToRange(&result.max_open_files, 20, 1000000);
124
  }
125

126
  if (result.info_log == nullptr) {
K
Kai Liu 已提交
127 128
    Status s = CreateLoggerFromOptions(dbname, result.db_log_dir, src.env,
                                       result, &result.info_log);
J
jorlow@chromium.org 已提交
129 130
    if (!s.ok()) {
      // No place suitable for logging
131
      result.info_log = nullptr;
J
jorlow@chromium.org 已提交
132 133
    }
  }
134 135 136 137
  result.env->IncBackgroundThreadsIfNeeded(src.max_background_compactions,
                                           Env::Priority::LOW);
  result.env->IncBackgroundThreadsIfNeeded(src.max_background_flushes,
                                           Env::Priority::HIGH);
138

I
Igor Canadi 已提交
139
  if (result.rate_limiter.get() != nullptr) {
140 141 142 143 144
    if (result.bytes_per_sync == 0) {
      result.bytes_per_sync = 1024 * 1024;
    }
  }

145 146 147 148
  if (result.wal_dir.empty()) {
    // Use dbname as default
    result.wal_dir = dbname;
  }
149
  if (result.wal_dir.back() == '/') {
I
Igor Canadi 已提交
150
    result.wal_dir = result.wal_dir.substr(0, result.wal_dir.size() - 1);
151
  }
152

153
  if (result.db_paths.size() == 0) {
154
    result.db_paths.emplace_back(dbname, std::numeric_limits<uint64_t>::max());
155 156
  }

J
jorlow@chromium.org 已提交
157 158 159
  return result;
}

160 161
namespace {

162 163
Status SanitizeOptionsByTable(
    const DBOptions& db_opts,
164 165 166
    const std::vector<ColumnFamilyDescriptor>& column_families) {
  Status s;
  for (auto cf : column_families) {
167
    s = cf.options.table_factory->SanitizeOptions(db_opts, cf.options);
168 169 170 171 172 173 174
    if (!s.ok()) {
      return s;
    }
  }
  return Status::OK();
}

175
CompressionType GetCompressionFlush(const ImmutableCFOptions& ioptions) {
176 177 178 179 180 181
  // Compressing memtable flushes might not help unless the sequential load
  // optimization is used for leveled compaction. Otherwise the CPU and
  // latency overhead is not offset by saving much space.

  bool can_compress;

182
  if (ioptions.compaction_style == kCompactionStyleUniversal) {
183
    can_compress =
184
        (ioptions.compaction_options_universal.compression_size_percent < 0);
185 186
  } else {
    // For leveled compress when min_level_to_compress == 0.
187 188
    can_compress = ioptions.compression_per_level.empty() ||
                   ioptions.compression_per_level[0] != kNoCompression;
189 190 191
  }

  if (can_compress) {
192
    return ioptions.compression;
193 194 195 196
  } else {
    return kNoCompression;
  }
}
I
Igor Canadi 已提交
197 198

void DumpCompressionInfo(Logger* logger) {
I
Igor Canadi 已提交
199 200 201 202 203 204 205 206
  Log(InfoLogLevel::INFO_LEVEL, logger, "Compression algorithms supported:");
  Log(InfoLogLevel::INFO_LEVEL, logger, "\tSnappy supported: %d",
      Snappy_Supported());
  Log(InfoLogLevel::INFO_LEVEL, logger, "\tZlib supported: %d",
      Zlib_Supported());
  Log(InfoLogLevel::INFO_LEVEL, logger, "\tBzip supported: %d",
      BZip2_Supported());
  Log(InfoLogLevel::INFO_LEVEL, logger, "\tLZ4 supported: %d", LZ4_Supported());
I
Igor Canadi 已提交
207 208
}

209
}  // namespace
210

I
Igor Canadi 已提交
211
DBImpl::DBImpl(const DBOptions& options, const std::string& dbname)
J
jorlow@chromium.org 已提交
212
    : env_(options.env),
H
heyongqiang 已提交
213
      dbname_(dbname),
214 215
      db_options_(SanitizeOptions(dbname, options)),
      stats_(db_options_.statistics.get()),
216
      db_lock_(nullptr),
I
Igor Canadi 已提交
217
      mutex_(stats_, env_, DB_MUTEX_WAIT_MICROS, options.use_adaptive_mutex),
I
Igor Canadi 已提交
218
      shutting_down_(false),
J
jorlow@chromium.org 已提交
219
      bg_cv_(&mutex_),
220
      logfile_number_(0),
221
      log_dir_synced_(false),
I
Igor Canadi 已提交
222
      log_empty_(true),
223
      default_cf_handle_(nullptr),
I
Igor Canadi 已提交
224 225
      total_log_size_(0),
      max_total_in_memory_state_(0),
S
sdong 已提交
226
      is_snapshot_supported_(true),
227
      write_buffer_(options.db_write_buffer_size),
228 229
      unscheduled_flushes_(0),
      unscheduled_compactions_(0),
230
      bg_compaction_scheduled_(0),
231
      bg_manual_only_(0),
232
      bg_flush_scheduled_(0),
233
      manual_compaction_(nullptr),
234
      disable_delete_obsolete_files_(0),
I
Igor Canadi 已提交
235 236 237
      delete_obsolete_files_next_run_(
          options.env->NowMicros() +
          db_options_.delete_obsolete_files_period_micros),
238
      last_stats_dump_time_microsec_(0),
239
      next_job_id_(1),
240
      flush_on_destroy_(false),
241
      env_options_(db_options_),
I
Igor Canadi 已提交
242 243 244
#ifndef ROCKSDB_LITE
      wal_manager_(db_options_, env_options_),
#endif  // ROCKSDB_LITE
I
Igor Canadi 已提交
245
      event_logger_(db_options_.info_log.get()),
246
      bg_work_gate_closed_(false),
247
      refitting_level_(false),
248
      opened_successfully_(false) {
H
heyongqiang 已提交
249
  env_->GetAbsolutePath(dbname, &db_absolute_path_);
250

J
jorlow@chromium.org 已提交
251
  // Reserve ten files or so for other uses and give the rest to TableCache.
252
  // Give a large number for setting of "infinite" open files.
253 254
  const int table_cache_size = (db_options_.max_open_files == -1) ?
        4194304 : db_options_.max_open_files - 10;
255 256
  // Reserve ten files or so for other uses and give the rest to TableCache.
  table_cache_ =
257
      NewLRUCache(table_cache_size, db_options_.table_cache_numshardbits);
258

259
  versions_.reset(new VersionSet(dbname_, &db_options_, env_options_,
260 261
                                 table_cache_.get(), &write_buffer_,
                                 &write_controller_));
I
Igor Canadi 已提交
262 263
  column_family_memtables_.reset(new ColumnFamilyMemTablesImpl(
      versions_->GetColumnFamilySet(), &flush_scheduler_));
264

I
Igor Canadi 已提交
265
  DumpRocksDBBuildVersion(db_options_.info_log.get());
266 267
  DumpDBFileSummary(db_options_, dbname_);
  db_options_.Dump(db_options_.info_log.get());
I
Igor Canadi 已提交
268
  DumpCompressionInfo(db_options_.info_log.get());
269

270
  LogFlush(db_options_.info_log);
J
jorlow@chromium.org 已提交
271 272
}

273 274
// Will only lock the mutex_ and wait for completion if wait is true
void DBImpl::CancelAllBackgroundWork(bool wait) {
275
  shutting_down_.store(true, std::memory_order_release);
276 277 278 279 280
  if (!wait) {
    return;
  }
  // Wait for background work to finish
  mutex_.Lock();
281
  while (bg_compaction_scheduled_ || bg_flush_scheduled_) {
282 283 284
    bg_cv_.Wait();
  }
  mutex_.Unlock();
285 286
}

J
jorlow@chromium.org 已提交
287
DBImpl::~DBImpl() {
Y
Yueh-Hsuan Chiang 已提交
288
  EraseThreadStatusDbInfo();
289
  mutex_.Lock();
290

291 292
  if (flush_on_destroy_) {
    for (auto cfd : *versions_->GetColumnFamilySet()) {
293
      if (!cfd->IsDropped() && !cfd->mem()->IsEmpty()) {
294 295 296 297
        cfd->Ref();
        mutex_.Unlock();
        FlushMemTable(cfd, FlushOptions());
        mutex_.Lock();
298
        cfd->Unref();
299 300
      }
    }
301
    versions_->GetColumnFamilySet()->FreeDeadColumnFamilies();
302
  }
303 304 305 306 307
  // CancelAllBackgroundWork called with false means we just set the
  // shutdown marker, while holding the mutex_ here. After which we
  // do a variant of the waiting after we release the lock and unschedule work
  // (to consider: moving all the waiting into CancelAllBackgroundWork(true))
  CancelAllBackgroundWork(false);
308 309 310 311 312 313 314 315
  mutex_.Unlock();
  int compactions_unscheduled = env_->UnSchedule(this, Env::Priority::LOW);
  int flushes_unscheduled = env_->UnSchedule(this, Env::Priority::HIGH);
  mutex_.Lock();
  bg_compaction_scheduled_ -= compactions_unscheduled;
  bg_flush_scheduled_ -= flushes_unscheduled;

  // Wait for background work to finish
316
  while (bg_compaction_scheduled_ || bg_flush_scheduled_) {
317 318
    bg_cv_.Wait();
  }
I
Igor Canadi 已提交
319 320
  flush_scheduler_.Clear();

321 322 323 324 325 326 327 328 329 330 331 332 333
  while (!flush_queue_.empty()) {
    auto cfd = PopFirstFromFlushQueue();
    if (cfd->Unref()) {
      delete cfd;
    }
  }
  while (!compaction_queue_.empty()) {
    auto cfd = PopFirstFromCompactionQueue();
    if (cfd->Unref()) {
      delete cfd;
    }
  }

I
Igor Canadi 已提交
334 335 336 337 338
  if (default_cf_handle_ != nullptr) {
    // we need to delete handle outside of lock because it does its own locking
    mutex_.Unlock();
    delete default_cf_handle_;
    mutex_.Lock();
339 340
  }

I
Igor Canadi 已提交
341 342 343 344 345 346 347 348 349 350
  // Clean up obsolete files due to SuperVersion release.
  // (1) Need to delete to obsolete files before closing because RepairDB()
  // scans all existing files in the file system and builds manifest file.
  // Keeping obsolete files confuses the repair process.
  // (2) Need to check if we Open()/Recover() the DB successfully before
  // deleting because if VersionSet recover fails (may be due to corrupted
  // manifest file), it is not able to identify live files correctly. As a
  // result, all "live" files can get deleted by accident. However, corrupted
  // manifest is recoverable by RepairDB().
  if (opened_successfully_) {
351
    JobContext job_context(next_job_id_.fetch_add(1));
I
Igor Canadi 已提交
352
    FindObsoleteFiles(&job_context, true);
353 354

    mutex_.Unlock();
I
Igor Canadi 已提交
355
    // manifest number starting from 2
I
Igor Canadi 已提交
356 357 358
    job_context.manifest_file_number = 1;
    if (job_context.HaveSomethingToDelete()) {
      PurgeObsoleteFiles(job_context);
359
    }
I
Igor Canadi 已提交
360
    job_context.Clean();
361
    mutex_.Lock();
362 363
  }

364 365 366 367
  for (auto l : logs_to_free_) {
    delete l;
  }

368
  // versions need to be destroyed before table_cache since it can hold
369 370
  // references to table_cache.
  versions_.reset();
371
  mutex_.Unlock();
I
Igor Canadi 已提交
372 373 374
  if (db_lock_ != nullptr) {
    env_->UnlockFile(db_lock_);
  }
375

376
  LogFlush(db_options_.info_log);
J
jorlow@chromium.org 已提交
377 378 379
}

Status DBImpl::NewDB() {
380
  VersionEdit new_db;
381
  new_db.SetLogNumber(0);
J
jorlow@chromium.org 已提交
382 383 384
  new_db.SetNextFile(2);
  new_db.SetLastSequence(0);

385 386
  Log(InfoLogLevel::INFO_LEVEL,
      db_options_.info_log, "Creating manifest 1 \n");
J
jorlow@chromium.org 已提交
387
  const std::string manifest = DescriptorFileName(dbname_, 1);
388
  unique_ptr<WritableFile> file;
I
Igor Canadi 已提交
389
  Status s = env_->NewWritableFile(
L
Lei Jin 已提交
390
      manifest, &file, env_->OptimizeForManifestWrite(env_options_));
J
jorlow@chromium.org 已提交
391 392 393
  if (!s.ok()) {
    return s;
  }
394
  file->SetPreallocationBlockSize(db_options_.manifest_preallocation_size);
J
jorlow@chromium.org 已提交
395
  {
396
    log::Writer log(std::move(file));
J
jorlow@chromium.org 已提交
397 398 399
    std::string record;
    new_db.EncodeTo(&record);
    s = log.AddRecord(record);
S
sdong 已提交
400 401 402
    if (s.ok()) {
      s = SyncManifest(env_, &db_options_, log.file());
    }
J
jorlow@chromium.org 已提交
403 404 405
  }
  if (s.ok()) {
    // Make "CURRENT" file that points to the new manifest file.
406
    s = SetCurrentFile(env_, dbname_, 1, directories_.GetDbDir());
J
jorlow@chromium.org 已提交
407 408 409 410 411 412 413
  } else {
    env_->DeleteFile(manifest);
  }
  return s;
}

void DBImpl::MaybeIgnoreError(Status* s) const {
414
  if (s->ok() || db_options_.paranoid_checks) {
J
jorlow@chromium.org 已提交
415 416
    // No change needed
  } else {
417 418
    Log(InfoLogLevel::WARN_LEVEL,
        db_options_.info_log, "Ignoring error %s", s->ToString().c_str());
J
jorlow@chromium.org 已提交
419 420 421 422
    *s = Status::OK();
  }
}

423
const Status DBImpl::CreateArchivalDirectory() {
424 425
  if (db_options_.WAL_ttl_seconds > 0 || db_options_.WAL_size_limit_MB > 0) {
    std::string archivalPath = ArchivalDirectory(db_options_.wal_dir);
426 427 428 429 430
    return env_->CreateDirIfMissing(archivalPath);
  }
  return Status::OK();
}

431
void DBImpl::PrintStatistics() {
432
  auto dbstats = db_options_.statistics.get();
433
  if (dbstats) {
Y
Yueh-Hsuan Chiang 已提交
434
    Log(InfoLogLevel::WARN_LEVEL, db_options_.info_log,
M
Mark Callaghan 已提交
435
        "STATISTICS:\n %s",
436
        dbstats->ToString().c_str());
437 438 439
  }
}

440
void DBImpl::MaybeDumpStats() {
441
  if (db_options_.stats_dump_period_sec == 0) return;
H
Haobo Xu 已提交
442 443 444 445

  const uint64_t now_micros = env_->NowMicros();

  if (last_stats_dump_time_microsec_ +
446
      db_options_.stats_dump_period_sec * 1000000
H
Haobo Xu 已提交
447 448 449 450 451 452
      <= now_micros) {
    // Multiple threads could race in here simultaneously.
    // However, the last one will update last_stats_dump_time_microsec_
    // atomically. We could see more than one dump during one dump
    // period in rare cases.
    last_stats_dump_time_microsec_ = now_micros;
453

454
#ifndef ROCKSDB_LITE
455 456 457
    bool tmp1 = false;
    bool tmp2 = false;
    DBPropertyType cf_property_type =
458
        GetPropertyType(DB::Properties::kCFStats, &tmp1, &tmp2);
459
    DBPropertyType db_property_type =
460
        GetPropertyType(DB::Properties::kDBStats, &tmp1, &tmp2);
H
Haobo Xu 已提交
461
    std::string stats;
462
    {
463
      InstrumentedMutexLock l(&mutex_);
464
      for (auto cfd : *versions_->GetColumnFamilySet()) {
465
        cfd->internal_stats()->GetStringProperty(cf_property_type,
466 467
                                                 DB::Properties::kCFStats,
                                                 &stats);
468
      }
469
      default_cf_internal_stats_->GetStringProperty(db_property_type,
470 471
                                                    DB::Properties::kDBStats,
                                                    &stats);
472
    }
Y
Yueh-Hsuan Chiang 已提交
473
    Log(InfoLogLevel::WARN_LEVEL,
474
        db_options_.info_log, "------- DUMPING STATS -------");
Y
Yueh-Hsuan Chiang 已提交
475
    Log(InfoLogLevel::WARN_LEVEL,
476
        db_options_.info_log, "%s", stats.c_str());
477
#endif  // !ROCKSDB_LITE
478

479
    PrintStatistics();
480 481 482
  }
}

I
Igor Canadi 已提交
483
// * Returns the list of live files in 'sst_live'
I
Igor Canadi 已提交
484
// If it's doing full scan:
I
Igor Canadi 已提交
485 486
// * Returns the list of all files in the filesystem in
// 'full_scan_candidate_files'.
I
Igor Canadi 已提交
487
// Otherwise, gets obsolete files from VersionSet.
I
Igor Canadi 已提交
488 489
// no_full_scan = true -- never do the full scan using GetChildren()
// force = false -- don't force the full scan, except every
490
//  db_options_.delete_obsolete_files_period_micros
I
Igor Canadi 已提交
491
// force = true -- force the full scan
I
Igor Canadi 已提交
492
void DBImpl::FindObsoleteFiles(JobContext* job_context, bool force,
I
Igor Canadi 已提交
493
                               bool no_full_scan) {
D
Dhruba Borthakur 已提交
494 495
  mutex_.AssertHeld();

496
  // if deletion is disabled, do nothing
497
  if (disable_delete_obsolete_files_ > 0) {
498 499 500
    return;
  }

501 502 503 504 505
  bool doing_the_full_scan = false;

  // logic for figurint out if we're doing the full scan
  if (no_full_scan) {
    doing_the_full_scan = false;
506
  } else if (force || db_options_.delete_obsolete_files_period_micros == 0) {
507 508 509
    doing_the_full_scan = true;
  } else {
    const uint64_t now_micros = env_->NowMicros();
I
Igor Canadi 已提交
510
    if (delete_obsolete_files_next_run_ < now_micros) {
511
      doing_the_full_scan = true;
I
Igor Canadi 已提交
512 513
      delete_obsolete_files_next_run_ =
          now_micros + db_options_.delete_obsolete_files_period_micros;
514 515 516
    }
  }

I
Igor Canadi 已提交
517 518 519 520 521 522 523 524 525
  // don't delete files that might be currently written to from compaction
  // threads
  if (!pending_outputs_.empty()) {
    job_context->min_pending_output = *pending_outputs_.begin();
  } else {
    // delete all of them
    job_context->min_pending_output = std::numeric_limits<uint64_t>::max();
  }

526 527
  // Get obsolete files.  This function will also update the list of
  // pending files in VersionSet().
I
Igor Canadi 已提交
528 529
  versions_->GetObsoleteFiles(&job_context->sst_delete_files,
                              job_context->min_pending_output);
I
Igor Canadi 已提交
530

I
Igor Canadi 已提交
531
  // store the current filenum, lognum, etc
532
  job_context->manifest_file_number = versions_->manifest_file_number();
I
Igor Canadi 已提交
533
  job_context->pending_manifest_file_number =
534
      versions_->pending_manifest_file_number();
I
Igor Canadi 已提交
535
  job_context->log_number = versions_->MinLogNumber();
536
  job_context->prev_log_number = versions_->prev_log_number();
I
Igor Canadi 已提交
537

I
Igor Canadi 已提交
538
  versions_->AddLiveFiles(&job_context->sst_live);
539
  if (doing_the_full_scan) {
I
Igor Canadi 已提交
540 541
    for (uint32_t path_id = 0; path_id < db_options_.db_paths.size();
         path_id++) {
542 543 544
      // set of all files in the directory. We'll exclude files that are still
      // alive in the subsequent processings.
      std::vector<std::string> files;
545
      env_->GetChildren(db_options_.db_paths[path_id].path,
546
                        &files);  // Ignore errors
547
      for (std::string file : files) {
I
Igor Canadi 已提交
548
        // TODO(icanadi) clean up this mess to avoid having one-off "/" prefixes
I
Igor Canadi 已提交
549 550
        job_context->full_scan_candidate_files.emplace_back("/" + file,
                                                            path_id);
551 552
      }
    }
553 554

    //Add log files in wal_dir
555
    if (db_options_.wal_dir != dbname_) {
556
      std::vector<std::string> log_files;
557
      env_->GetChildren(db_options_.wal_dir, &log_files);  // Ignore errors
558
      for (std::string log_file : log_files) {
I
Igor Canadi 已提交
559
        job_context->full_scan_candidate_files.emplace_back(log_file, 0);
560 561
      }
    }
562
    // Add info log files in db_log_dir
563
    if (!db_options_.db_log_dir.empty() && db_options_.db_log_dir != dbname_) {
564
      std::vector<std::string> info_log_files;
565 566
      // Ignore errors
      env_->GetChildren(db_options_.db_log_dir, &info_log_files);
567
      for (std::string log_file : info_log_files) {
I
Igor Canadi 已提交
568
        job_context->full_scan_candidate_files.emplace_back(log_file, 0);
569 570
      }
    }
571
  }
572 573
}

574
namespace {
I
Igor Canadi 已提交
575 576
bool CompareCandidateFile(const JobContext::CandidateFileInfo& first,
                          const JobContext::CandidateFileInfo& second) {
577 578 579 580 581
  if (first.file_name > second.file_name) {
    return true;
  } else if (first.file_name < second.file_name) {
    return false;
  } else {
582
    return (first.path_id > second.path_id);
583 584 585 586
  }
}
};  // namespace

D
Dhruba Borthakur 已提交
587
// Diffs the files listed in filenames and those that do not
I
Igor Canadi 已提交
588
// belong to live files are posibly removed. Also, removes all the
589
// files in sst_delete_files and log_delete_files.
590
// It is not necessary to hold the mutex when invoking this method.
I
Igor Canadi 已提交
591
void DBImpl::PurgeObsoleteFiles(const JobContext& state) {
592 593
  // we'd better have sth to delete
  assert(state.HaveSomethingToDelete());
594

I
Igor Canadi 已提交
595 596 597 598
  // this checks if FindObsoleteFiles() was run before. If not, don't do
  // PurgeObsoleteFiles(). If FindObsoleteFiles() was run, we need to also
  // run PurgeObsoleteFiles(), even if disable_delete_obsolete_files_ is true
  if (state.manifest_file_number == 0) {
I
Igor Canadi 已提交
599 600
    return;
  }
601

602
  // Now, convert live list to an unordered map, WITHOUT mutex held;
603
  // set is slow.
604
  std::unordered_map<uint64_t, const FileDescriptor*> sst_live_map;
I
Igor Canadi 已提交
605
  for (const FileDescriptor& fd : state.sst_live) {
606 607
    sst_live_map[fd.GetNumber()] = &fd;
  }
I
Igor Canadi 已提交
608

I
Igor Canadi 已提交
609
  auto candidate_files = state.full_scan_candidate_files;
I
Igor Canadi 已提交
610 611 612
  candidate_files.reserve(candidate_files.size() +
                          state.sst_delete_files.size() +
                          state.log_delete_files.size());
K
kailiu 已提交
613 614
  // We may ignore the dbname when generating the file names.
  const char* kDumbDbName = "";
615
  for (auto file : state.sst_delete_files) {
616 617 618
    candidate_files.emplace_back(
        MakeTableFileName(kDumbDbName, file->fd.GetNumber()),
        file->fd.GetPathId());
619
    delete file;
I
Igor Canadi 已提交
620 621
  }

K
kailiu 已提交
622 623
  for (auto file_num : state.log_delete_files) {
    if (file_num > 0) {
624 625
      candidate_files.emplace_back(LogFileName(kDumbDbName, file_num).substr(1),
                                   0);
I
Igor Canadi 已提交
626 627
    }
  }
628

K
kailiu 已提交
629
  // dedup state.candidate_files so we don't try to delete the same
I
Igor Canadi 已提交
630
  // file twice
631
  sort(candidate_files.begin(), candidate_files.end(), CompareCandidateFile);
632 633
  candidate_files.erase(unique(candidate_files.begin(), candidate_files.end()),
                        candidate_files.end());
J
jorlow@chromium.org 已提交
634

635
  std::vector<std::string> old_info_log_files;
636
  InfoLogPrefix info_log_prefix(!db_options_.db_log_dir.empty(), dbname_);
637 638 639
  for (const auto& candidate_file : candidate_files) {
    std::string to_delete = candidate_file.file_name;
    uint32_t path_id = candidate_file.path_id;
K
kailiu 已提交
640 641 642
    uint64_t number;
    FileType type;
    // Ignore file if we cannot recognize it.
643
    if (!ParseFileName(to_delete, &number, info_log_prefix.prefix, &type)) {
K
kailiu 已提交
644 645
      continue;
    }
J
jorlow@chromium.org 已提交
646

K
kailiu 已提交
647 648 649 650 651 652 653 654
    bool keep = true;
    switch (type) {
      case kLogFile:
        keep = ((number >= state.log_number) ||
                (number == state.prev_log_number));
        break;
      case kDescriptorFile:
        // Keep my manifest file, and any newer incarnations'
655
        // (can happen during manifest roll)
K
kailiu 已提交
656 657 658
        keep = (number >= state.manifest_file_number);
        break;
      case kTableFile:
I
Igor Canadi 已提交
659 660 661 662
        // If the second condition is not there, this makes
        // DontDeletePendingOutputs fail
        keep = (sst_live_map.find(number) != sst_live_map.end()) ||
               number >= state.min_pending_output;
K
kailiu 已提交
663 664 665
        break;
      case kTempFile:
        // Any temp files that are currently being written to must
666 667 668 669
        // be recorded in pending_outputs_, which is inserted into "live".
        // Also, SetCurrentFile creates a temp file when writing out new
        // manifest, which is equal to state.pending_manifest_file_number. We
        // should not delete that file
670
        keep = (sst_live_map.find(number) != sst_live_map.end()) ||
671
               (number == state.pending_manifest_file_number);
K
kailiu 已提交
672 673 674 675
        break;
      case kInfoLogFile:
        keep = true;
        if (number != 0) {
676
          old_info_log_files.push_back(to_delete);
J
jorlow@chromium.org 已提交
677
        }
K
kailiu 已提交
678 679 680 681 682 683 684 685 686 687 688 689 690
        break;
      case kCurrentFile:
      case kDBLockFile:
      case kIdentityFile:
      case kMetaDatabase:
        keep = true;
        break;
    }

    if (keep) {
      continue;
    }

691
    std::string fname;
K
kailiu 已提交
692 693
    if (type == kTableFile) {
      // evict from cache
694
      TableCache::Evict(table_cache_.get(), number);
695
      fname = TableFileName(db_options_.db_paths, number, path_id);
696
    } else {
697 698
      fname = ((type == kLogFile) ?
          db_options_.wal_dir : dbname_) + "/" + to_delete;
K
kailiu 已提交
699
    }
700

701
#ifndef ROCKSDB_LITE
I
Igor Canadi 已提交
702
    if (type == kLogFile && (db_options_.WAL_ttl_seconds > 0 ||
703
                              db_options_.WAL_size_limit_MB > 0)) {
I
Igor Canadi 已提交
704
      wal_manager_.ArchiveWALFile(fname, number);
705 706 707 708 709
      continue;
    }
#endif  // !ROCKSDB_LITE
    auto file_deletion_status = env_->DeleteFile(fname);
    if (file_deletion_status.ok()) {
710
      Log(InfoLogLevel::DEBUG_LEVEL, db_options_.info_log,
711
          "[JOB %d] Delete %s type=%d #%" PRIu64 " -- %s\n", state.job_id,
712 713 714 715 716 717 718 719 720
          fname.c_str(), type, number,
          file_deletion_status.ToString().c_str());
    } else {
      Log(InfoLogLevel::ERROR_LEVEL, db_options_.info_log,
          "[JOB %d] Failed to delete %s type=%d #%" PRIu64 " -- %s\n",
          state.job_id, fname.c_str(), type, number,
          file_deletion_status.ToString().c_str());
    }
    if (type == kTableFile) {
721 722 723 724
      EventHelpers::LogAndNotifyTableFileDeletion(
          &event_logger_, state.job_id, number, fname,
          file_deletion_status, GetName(),
          db_options_.listeners);
J
jorlow@chromium.org 已提交
725 726
    }
  }
H
heyongqiang 已提交
727

728
  // Delete old info log files.
729
  size_t old_info_log_file_count = old_info_log_files.size();
730
  if (old_info_log_file_count >= db_options_.keep_log_file_num) {
731
    std::sort(old_info_log_files.begin(), old_info_log_files.end());
732
    size_t end = old_info_log_file_count - db_options_.keep_log_file_num;
733
    for (unsigned int i = 0; i <= end; i++) {
734
      std::string& to_delete = old_info_log_files.at(i);
735 736
      std::string full_path_to_delete = (db_options_.db_log_dir.empty() ?
           dbname_ : db_options_.db_log_dir) + "/" + to_delete;
737
      Log(InfoLogLevel::INFO_LEVEL, db_options_.info_log,
738
          "[JOB %d] Delete info log file %s\n", state.job_id,
739 740
          full_path_to_delete.c_str());
      Status s = env_->DeleteFile(full_path_to_delete);
741
      if (!s.ok()) {
742 743
        Log(InfoLogLevel::ERROR_LEVEL, db_options_.info_log,
            "[JOB %d] Delete info log file %s FAILED -- %s\n", state.job_id,
744 745
            to_delete.c_str(), s.ToString().c_str());
      }
H
heyongqiang 已提交
746 747
    }
  }
I
Igor Canadi 已提交
748 749 750
#ifndef ROCKSDB_LITE
  wal_manager_.PurgeObsoleteWALFiles();
#endif  // ROCKSDB_LITE
751
  LogFlush(db_options_.info_log);
D
Dhruba Borthakur 已提交
752 753 754 755
}

void DBImpl::DeleteObsoleteFiles() {
  mutex_.AssertHeld();
756
  JobContext job_context(next_job_id_.fetch_add(1));
I
Igor Canadi 已提交
757
  FindObsoleteFiles(&job_context, true);
758 759

  mutex_.Unlock();
I
Igor Canadi 已提交
760 761
  if (job_context.HaveSomethingToDelete()) {
    PurgeObsoleteFiles(job_context);
762
  }
I
Igor Canadi 已提交
763
  job_context.Clean();
764
  mutex_.Lock();
765 766
}

767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782
Status DBImpl::Directories::CreateAndNewDirectory(
    Env* env, const std::string& dirname,
    std::unique_ptr<Directory>* directory) const {
  // We call CreateDirIfMissing() as the directory may already exist (if we
  // are reopening a DB), when this happens we don't want creating the
  // directory to cause an error. However, we need to check if creating the
  // directory fails or else we may get an obscure message about the lock
  // file not existing. One real-world example of this occurring is if
  // env->CreateDirIfMissing() doesn't create intermediate directories, e.g.
  // when dbname_ is "dir/db" but when "dir" doesn't exist.
  Status s = env->CreateDirIfMissing(dirname);
  if (!s.ok()) {
    return s;
  }
  return env->NewDirectory(dirname, directory);
}
J
jorlow@chromium.org 已提交
783

784 785 786 787 788 789 790 791 792
Status DBImpl::Directories::SetDirectories(
    Env* env, const std::string& dbname, const std::string& wal_dir,
    const std::vector<DbPath>& data_paths) {
  Status s = CreateAndNewDirectory(env, dbname, &db_dir_);
  if (!s.ok()) {
    return s;
  }
  if (!wal_dir.empty() && dbname != wal_dir) {
    s = CreateAndNewDirectory(env, wal_dir, &wal_dir_);
793 794 795
    if (!s.ok()) {
      return s;
    }
796
  }
797

798 799 800 801 802 803 804 805
  data_dirs_.clear();
  for (auto& p : data_paths) {
    const std::string db_path = p.path;
    if (db_path == dbname) {
      data_dirs_.emplace_back(nullptr);
    } else {
      std::unique_ptr<Directory> path_directory;
      s = CreateAndNewDirectory(env, db_path, &path_directory);
806 807 808
      if (!s.ok()) {
        return s;
      }
809
      data_dirs_.emplace_back(path_directory.release());
810
    }
811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829
  }
  assert(data_dirs_.size() == data_paths.size());
  return Status::OK();
}

Directory* DBImpl::Directories::GetDataDir(size_t path_id) {
  assert(path_id < data_dirs_.size());
  Directory* ret_dir = data_dirs_[path_id].get();
  if (ret_dir == nullptr) {
    // Should use db_dir_
    return db_dir_.get();
  }
  return ret_dir;
}

Status DBImpl::Recover(
    const std::vector<ColumnFamilyDescriptor>& column_families, bool read_only,
    bool error_if_log_file_exist) {
  mutex_.AssertHeld();
830

831 832 833 834 835
  bool is_new_db = false;
  assert(db_lock_ == nullptr);
  if (!read_only) {
    Status s = directories_.SetDirectories(env_, dbname_, db_options_.wal_dir,
                                           db_options_.db_paths);
836 837 838 839
    if (!s.ok()) {
      return s;
    }

840
    s = env_->LockFile(LockFileName(dbname_), &db_lock_);
841 842 843
    if (!s.ok()) {
      return s;
    }
J
jorlow@chromium.org 已提交
844

845
    if (!env_->FileExists(CurrentFileName(dbname_))) {
846
      if (db_options_.create_if_missing) {
847
        s = NewDB();
848
        is_new_db = true;
849 850 851 852 853 854
        if (!s.ok()) {
          return s;
        }
      } else {
        return Status::InvalidArgument(
            dbname_, "does not exist (create_if_missing is false)");
J
jorlow@chromium.org 已提交
855 856
      }
    } else {
857
      if (db_options_.error_if_exists) {
858 859 860
        return Status::InvalidArgument(
            dbname_, "exists (error_if_exists is true)");
      }
J
jorlow@chromium.org 已提交
861
    }
M
Mayank Agarwal 已提交
862 863 864 865 866 867 868
    // Check for the IDENTITY file and create it if not there
    if (!env_->FileExists(IdentityFileName(dbname_))) {
      s = SetIdentityFile(env_, dbname_);
      if (!s.ok()) {
        return s;
      }
    }
J
jorlow@chromium.org 已提交
869 870
  }

871
  Status s = versions_->Recover(column_families, read_only);
872
  if (db_options_.paranoid_checks && s.ok()) {
I
Igor Canadi 已提交
873 874
    s = CheckConsistency();
  }
J
jorlow@chromium.org 已提交
875
  if (s.ok()) {
A
agiardullo 已提交
876
    SequenceNumber max_sequence(kMaxSequenceNumber);
877 878
    default_cf_handle_ = new ColumnFamilyHandleImpl(
        versions_->GetColumnFamilySet()->GetDefault(), this, &mutex_);
879
    default_cf_internal_stats_ = default_cf_handle_->cfd()->internal_stats();
880 881
    single_column_family_mode_ =
        versions_->GetColumnFamilySet()->NumberOfColumnFamilies() == 1;
882 883 884 885 886

    // Recover from all newer log files than the ones named in the
    // descriptor (new log files may have been added by the previous
    // incarnation without registering them in the descriptor).
    //
887
    // Note that prev_log_number() is no longer used, but we pay
888
    // attention to it in case we are recovering a database
889
    // produced by an older version of rocksdb.
890
    const uint64_t min_log = versions_->MinLogNumber();
891
    const uint64_t prev_log = versions_->prev_log_number();
892
    std::vector<std::string> filenames;
893
    s = env_->GetChildren(db_options_.wal_dir, &filenames);
894 895
    if (!s.ok()) {
      return s;
896
    }
K
kailiu 已提交
897

898 899
    std::vector<uint64_t> logs;
    for (size_t i = 0; i < filenames.size(); i++) {
K
kailiu 已提交
900 901
      uint64_t number;
      FileType type;
902 903 904 905 906 907 908 909 910
      if (ParseFileName(filenames[i], &number, &type) && type == kLogFile) {
        if (is_new_db) {
          return Status::Corruption(
              "While creating a new Db, wal_dir contains "
              "existing log file: ",
              filenames[i]);
        } else if ((number >= min_log) || (number == prev_log)) {
          logs.push_back(number);
        }
911
      }
J
jorlow@chromium.org 已提交
912
    }
913

H
heyongqiang 已提交
914 915 916 917 918 919
    if (logs.size() > 0 && error_if_log_file_exist) {
      return Status::Corruption(""
          "The db was opened in readonly mode with error_if_log_file_exist"
          "flag but a log file already exists");
    }

S
Stanislau Hlebik 已提交
920 921 922 923 924 925 926
    if (!logs.empty()) {
      // Recover in the order in which the logs were generated
      std::sort(logs.begin(), logs.end());
      s = RecoverLogFiles(logs, &max_sequence, read_only);
      if (!s.ok()) {
        // Clear memtables if recovery failed
        for (auto cfd : *versions_->GetColumnFamilySet()) {
A
agiardullo 已提交
927 928
          cfd->CreateNewMemtable(*cfd->GetLatestMutableCFOptions(),
                                 kMaxSequenceNumber);
S
Stanislau Hlebik 已提交
929 930
        }
      }
931
    }
L
Lei Jin 已提交
932
    SetTickerCount(stats_, SEQUENCE_NUMBER, versions_->LastSequence());
J
jorlow@chromium.org 已提交
933 934
  }

L
Lei Jin 已提交
935 936
  // Initial value
  max_total_in_memory_state_ = 0;
I
Igor Canadi 已提交
937
  for (auto cfd : *versions_->GetColumnFamilySet()) {
L
Lei Jin 已提交
938 939 940
    auto* mutable_cf_options = cfd->GetLatestMutableCFOptions();
    max_total_in_memory_state_ += mutable_cf_options->write_buffer_size *
                                  mutable_cf_options->max_write_buffer_number;
I
Igor Canadi 已提交
941 942
  }

J
jorlow@chromium.org 已提交
943 944 945
  return s;
}

S
Stanislau Hlebik 已提交
946 947 948
// REQUIRES: log_numbers are sorted in ascending order
Status DBImpl::RecoverLogFiles(const std::vector<uint64_t>& log_numbers,
                               SequenceNumber* max_sequence, bool read_only) {
J
jorlow@chromium.org 已提交
949 950
  struct LogReporter : public log::Reader::Reporter {
    Env* env;
951
    Logger* info_log;
J
jorlow@chromium.org 已提交
952
    const char* fname;
953
    Status* status;  // nullptr if db_options_.paranoid_checks==false
I
Igor Sugak 已提交
954
    virtual void Corruption(size_t bytes, const Status& s) override {
955 956
      Log(InfoLogLevel::WARN_LEVEL,
          info_log, "%s%s: dropping %d bytes; %s",
957
          (this->status == nullptr ? "(ignoring error) " : ""),
J
jorlow@chromium.org 已提交
958
          fname, static_cast<int>(bytes), s.ToString().c_str());
959
      if (this->status != nullptr && this->status->ok()) *this->status = s;
J
jorlow@chromium.org 已提交
960 961 962 963
    }
  };

  mutex_.AssertHeld();
S
Stanislau Hlebik 已提交
964
  Status status;
965
  std::unordered_map<int, VersionEdit> version_edits;
966
  // no need to refcount because iteration is under mutex
967 968
  for (auto cfd : *versions_->GetColumnFamilySet()) {
    VersionEdit edit;
969 970
    edit.SetColumnFamily(cfd->GetID());
    version_edits.insert({cfd->GetID(), edit});
971
  }
972 973 974 975 976 977 978 979 980 981 982 983
  int job_id = next_job_id_.fetch_add(1);
  {
    auto stream = event_logger_.Log();
    stream << "job" << job_id << "event"
           << "recovery_started";
    stream << "log_files";
    stream.StartArray();
    for (auto log_number : log_numbers) {
      stream << log_number;
    }
    stream.EndArray();
  }
I
Igor Canadi 已提交
984

S
Stanislau Hlebik 已提交
985 986 987 988
  for (auto log_number : log_numbers) {
    // The previous incarnation may not have written any MANIFEST
    // records after allocating this log number.  So we manually
    // update the file number allocation counter in VersionSet.
989
    versions_->MarkFileNumberUsedDuringRecovery(log_number);
S
Stanislau Hlebik 已提交
990 991 992 993
    // Open the log file
    std::string fname = LogFileName(db_options_.wal_dir, log_number);
    unique_ptr<SequentialFile> file;
    status = env_->NewSequentialFile(fname, &file, env_options_);
J
jorlow@chromium.org 已提交
994
    if (!status.ok()) {
S
Stanislau Hlebik 已提交
995 996 997 998 999 1000 1001 1002
      MaybeIgnoreError(&status);
      if (!status.ok()) {
        return status;
      } else {
        // Fail with one log file, but that's ok.
        // Try next one.
        continue;
      }
J
jorlow@chromium.org 已提交
1003 1004
    }

S
Stanislau Hlebik 已提交
1005 1006 1007 1008 1009
    // Create the log reader.
    LogReporter reporter;
    reporter.env = env_;
    reporter.info_log = db_options_.info_log.get();
    reporter.fname = fname.c_str();
1010
    reporter.status = (db_options_.paranoid_checks) ? &status : nullptr;
S
Stanislau Hlebik 已提交
1011 1012 1013 1014 1015 1016
    // We intentially make log::Reader do checksumming even if
    // paranoid_checks==false so that corruptions cause entire commits
    // to be skipped instead of propagating bad information (like overly
    // large sequence numbers).
    log::Reader reader(std::move(file), &reporter, true /*checksum*/,
                       0 /*initial_offset*/);
1017 1018
    Log(InfoLogLevel::INFO_LEVEL,
        db_options_.info_log, "Recovering log #%" PRIu64 "", log_number);
S
Stanislau Hlebik 已提交
1019 1020 1021 1022 1023

    // Read all the records and add to a memtable
    std::string scratch;
    Slice record;
    WriteBatch batch;
I
Igor Canadi 已提交
1024
    while (reader.ReadRecord(&record, &scratch) && status.ok()) {
S
Stanislau Hlebik 已提交
1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045
      if (record.size() < 12) {
        reporter.Corruption(record.size(),
                            Status::Corruption("log record too small"));
        continue;
      }
      WriteBatchInternal::SetContents(&batch, record);

      // If column family was not found, it might mean that the WAL write
      // batch references to the column family that was dropped after the
      // insert. We don't want to fail the whole write batch in that case --
      // we just ignore the update.
      // That's why we set ignore missing column families to true
      status = WriteBatchInternal::InsertInto(
          &batch, column_family_memtables_.get(), true, log_number);

      MaybeIgnoreError(&status);
      if (!status.ok()) {
        return status;
      }
      const SequenceNumber last_seq = WriteBatchInternal::Sequence(&batch) +
                                      WriteBatchInternal::Count(&batch) - 1;
A
agiardullo 已提交
1046
      if ((*max_sequence == kMaxSequenceNumber) || (last_seq > *max_sequence)) {
S
Stanislau Hlebik 已提交
1047 1048 1049 1050
        *max_sequence = last_seq;
      }

      if (!read_only) {
I
Igor Canadi 已提交
1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062
        // we can do this because this is called before client has access to the
        // DB and there is only a single thread operating on DB
        ColumnFamilyData* cfd;

        while ((cfd = flush_scheduler_.GetNextColumnFamily()) != nullptr) {
          cfd->Unref();
          // If this asserts, it means that InsertInto failed in
          // filtering updates to already-flushed column families
          assert(cfd->GetLogNumber() <= log_number);
          auto iter = version_edits.find(cfd->GetID());
          assert(iter != version_edits.end());
          VersionEdit* edit = &iter->second;
1063
          status = WriteLevel0TableForRecovery(job_id, cfd, cfd->mem(), edit);
I
Igor Canadi 已提交
1064 1065 1066 1067
          if (!status.ok()) {
            // Reflect errors immediately so that conditions like full
            // file-systems cause the DB::Open() to fail.
            return status;
1068
          }
A
agiardullo 已提交
1069 1070 1071

          cfd->CreateNewMemtable(*cfd->GetLatestMutableCFOptions(),
                                 *max_sequence);
1072
        }
J
jorlow@chromium.org 已提交
1073 1074 1075
      }
    }

1076 1077 1078 1079
    if (!status.ok()) {
      return status;
    }

I
Igor Canadi 已提交
1080
    flush_scheduler_.Clear();
A
agiardullo 已提交
1081 1082
    if ((*max_sequence != kMaxSequenceNumber) &&
        (versions_->LastSequence() < *max_sequence)) {
S
Stanislau Hlebik 已提交
1083 1084
      versions_->SetLastSequence(*max_sequence);
    }
1085 1086
  }

1087
  if (!read_only) {
1088 1089
    // no need to refcount since client still doesn't have access
    // to the DB and can not drop column families while we iterate
S
Stanislau Hlebik 已提交
1090
    auto max_log_number = log_numbers.back();
1091
    for (auto cfd : *versions_->GetColumnFamilySet()) {
1092
      auto iter = version_edits.find(cfd->GetID());
1093 1094 1095
      assert(iter != version_edits.end());
      VersionEdit* edit = &iter->second;

S
Stanislau Hlebik 已提交
1096
      if (cfd->GetLogNumber() > max_log_number) {
1097
        // Column family cfd has already flushed the data
S
Stanislau Hlebik 已提交
1098
        // from all logs. Memtable has to be empty because
1099
        // we filter the updates based on log_number
1100
        // (in WriteBatch::InsertInto)
1101 1102 1103 1104 1105 1106 1107
        assert(cfd->mem()->GetFirstSequenceNumber() == 0);
        assert(edit->NumEntries() == 0);
        continue;
      }

      // flush the final memtable (if non-empty)
      if (cfd->mem()->GetFirstSequenceNumber() != 0) {
1108
        status = WriteLevel0TableForRecovery(job_id, cfd, cfd->mem(), edit);
S
Stanislau Hlebik 已提交
1109 1110 1111 1112
        if (!status.ok()) {
          // Recovery failed
          break;
        }
A
agiardullo 已提交
1113 1114 1115

        cfd->CreateNewMemtable(*cfd->GetLatestMutableCFOptions(),
                               *max_sequence);
1116
      }
J
jorlow@chromium.org 已提交
1117

1118
      // write MANIFEST with update
S
Stanislau Hlebik 已提交
1119
      // writing log_number in the manifest means that any log file
1120 1121
      // with number strongly less than (log_number + 1) is already
      // recovered and should be ignored on next reincarnation.
S
Stanislau Hlebik 已提交
1122 1123 1124
      // Since we already recovered max_log_number, we want all logs
      // with numbers `<= max_log_number` (includes this one) to be ignored
      edit->SetLogNumber(max_log_number + 1);
1125 1126 1127
      // we must mark the next log number as used, even though it's
      // not actually used. that is because VersionSet assumes
      // VersionSet::next_file_number_ always to be strictly greater than any
I
Igor Canadi 已提交
1128
      // log number
1129
      versions_->MarkFileNumberUsedDuringRecovery(max_log_number + 1);
1130 1131
      status = versions_->LogAndApply(
          cfd, *cfd->GetLatestMutableCFOptions(), edit, &mutex_);
1132
      if (!status.ok()) {
S
Stanislau Hlebik 已提交
1133 1134
        // Recovery failed
        break;
1135 1136
      }
    }
1137
  }
I
Igor Canadi 已提交
1138

1139 1140 1141
  event_logger_.Log() << "job" << job_id << "event"
                      << "recovery_finished";

J
jorlow@chromium.org 已提交
1142 1143 1144
  return status;
}

1145 1146
Status DBImpl::WriteLevel0TableForRecovery(int job_id, ColumnFamilyData* cfd,
                                           MemTable* mem, VersionEdit* edit) {
J
jorlow@chromium.org 已提交
1147
  mutex_.AssertHeld();
1148
  const uint64_t start_micros = env_->NowMicros();
J
jorlow@chromium.org 已提交
1149
  FileMetaData meta;
1150
  meta.fd = FileDescriptor(versions_->NewFileNumber(), 0, 0);
I
Igor Canadi 已提交
1151 1152
  auto pending_outputs_inserted_elem =
      CaptureCurrentFileNumberInPendingOutputs();
1153 1154
  ReadOptions ro;
  ro.total_order_seek = true;
1155
  Arena arena;
1156
  Status s;
1157
  TableProperties table_properties;
1158
  {
1159 1160 1161 1162
    ScopedArenaIterator iter(mem->NewIterator(ro, &arena));
    const SequenceNumber newest_snapshot = snapshots_.GetNewest();
    const SequenceNumber earliest_seqno_in_memtable =
        mem->GetFirstSequenceNumber();
1163 1164 1165
    Log(InfoLogLevel::DEBUG_LEVEL, db_options_.info_log,
        "[%s] [WriteLevel0TableForRecovery]"
        " Level-0 table #%" PRIu64 ": started",
1166
        cfd->GetName().c_str(), meta.fd.GetNumber());
1167

1168 1169
    bool paranoid_file_checks =
        cfd->GetLatestMutableCFOptions()->paranoid_file_checks;
1170 1171
    {
      mutex_.Unlock();
1172
      TableFileCreationInfo info;
1173 1174
      s = BuildTable(
          dbname_, env_, *cfd->ioptions(), env_options_, cfd->table_cache(),
1175 1176
          iter.get(), &meta, cfd->internal_comparator(),
          cfd->int_tbl_prop_collector_factories(), newest_snapshot,
1177
          earliest_seqno_in_memtable, GetCompressionFlush(*cfd->ioptions()),
1178
          cfd->ioptions()->compression_opts, paranoid_file_checks, Env::IO_HIGH,
1179
          &info.table_properties);
1180
      LogFlush(db_options_.info_log);
1181 1182 1183 1184 1185 1186 1187 1188
      Log(InfoLogLevel::DEBUG_LEVEL, db_options_.info_log,
          "[%s] [WriteLevel0TableForRecovery]"
          " Level-0 table #%" PRIu64 ": %" PRIu64 " bytes %s",
          cfd->GetName().c_str(), meta.fd.GetNumber(), meta.fd.GetFileSize(),
          s.ToString().c_str());

      // output to event logger
      if (s.ok()) {
1189 1190 1191 1192 1193 1194 1195 1196 1197
        info.db_name = dbname_;
        info.cf_name = cfd->GetName();
        info.file_path = TableFileName(db_options_.db_paths,
                                       meta.fd.GetNumber(),
                                       meta.fd.GetPathId());
        info.file_size = meta.fd.GetFileSize();
        info.job_id = job_id;
        EventHelpers::LogAndNotifyTableFileCreation(
            &event_logger_, db_options_.listeners, meta.fd, info);
1198
      }
1199 1200
      mutex_.Lock();
    }
1201
  }
I
Igor Canadi 已提交
1202
  ReleaseFileNumberFromPendingOutputs(pending_outputs_inserted_elem);
1203 1204 1205 1206

  // Note that if file_size is zero, the file has been deleted and
  // should not be added to the manifest.
  int level = 0;
1207
  if (s.ok() && meta.fd.GetFileSize() > 0) {
1208 1209 1210
    edit->AddFile(level, meta.fd.GetNumber(), meta.fd.GetPathId(),
                  meta.fd.GetFileSize(), meta.smallest, meta.largest,
                  meta.smallest_seqno, meta.largest_seqno);
1211 1212
  }

L
Lei Jin 已提交
1213
  InternalStats::CompactionStats stats(1);
1214
  stats.micros = env_->NowMicros() - start_micros;
1215
  stats.bytes_written = meta.fd.GetFileSize();
M
Mark Callaghan 已提交
1216
  stats.files_out_levelnp1 = 1;
1217
  cfd->internal_stats()->AddCompactionStats(level, stats);
1218 1219
  cfd->internal_stats()->AddCFStats(
      InternalStats::BYTES_FLUSHED, meta.fd.GetFileSize());
L
Lei Jin 已提交
1220
  RecordTick(stats_, COMPACT_WRITE_BYTES, meta.fd.GetFileSize());
J
jorlow@chromium.org 已提交
1221 1222 1223
  return s;
}

1224 1225
Status DBImpl::FlushMemTableToOutputFile(
    ColumnFamilyData* cfd, const MutableCFOptions& mutable_cf_options,
I
Igor Canadi 已提交
1226
    bool* madeProgress, JobContext* job_context, LogBuffer* log_buffer) {
1227
  mutex_.AssertHeld();
1228
  assert(cfd->imm()->NumNotFlushed() != 0);
1229
  assert(cfd->imm()->IsFlushPending());
1230

I
Igor Canadi 已提交
1231 1232
  FlushJob flush_job(dbname_, cfd, db_options_, mutable_cf_options,
                     env_options_, versions_.get(), &mutex_, &shutting_down_,
I
Igor Canadi 已提交
1233
                     snapshots_.GetNewest(), job_context, log_buffer,
1234
                     directories_.GetDbDir(), directories_.GetDataDir(0U),
I
Igor Canadi 已提交
1235 1236
                     GetCompressionFlush(*cfd->ioptions()), stats_,
                     &event_logger_);
1237

1238
  uint64_t file_number;
1239 1240 1241 1242 1243 1244 1245

  // Within flush_job.Run, rocksdb may call event listener to notify
  // file creation and deletion.
  //
  // Note that flush_job.Run will unlock and lock the db_mutex,
  // and EventListener callback will be called when the db_mutex
  // is unlocked by the current thread.
1246
  Status s = flush_job.Run(&file_number);
J
jorlow@chromium.org 已提交
1247 1248

  if (s.ok()) {
I
Igor Canadi 已提交
1249
    InstallSuperVersionBackground(cfd, job_context, mutable_cf_options);
1250 1251 1252
    if (madeProgress) {
      *madeProgress = 1;
    }
S
sdong 已提交
1253
    VersionStorageInfo::LevelSummaryStorage tmp;
I
Igor Canadi 已提交
1254
    LogToBuffer(log_buffer, "[%s] Level summary: %s\n", cfd->GetName().c_str(),
S
sdong 已提交
1255
                cfd->current()->storage_info()->LevelSummary(&tmp));
1256

1257
    if (disable_delete_obsolete_files_ == 0) {
I
Igor Canadi 已提交
1258
      // add to deletion state
1259
      while (alive_log_files_.size() &&
I
Igor Canadi 已提交
1260 1261
             alive_log_files_.begin()->number < versions_->MinLogNumber()) {
        const auto& earliest = *alive_log_files_.begin();
I
Igor Canadi 已提交
1262
        job_context->log_delete_files.push_back(earliest.number);
I
Igor Canadi 已提交
1263
        total_log_size_ -= earliest.size;
1264 1265
        alive_log_files_.pop_front();
      }
1266
    }
J
jorlow@chromium.org 已提交
1267
  }
1268

1269
  if (!s.ok() && !s.IsShutdownInProgress() && db_options_.paranoid_checks &&
1270 1271 1272 1273 1274
      bg_error_.ok()) {
    // if a bad error happened (not ShutdownInProgress) and paranoid_checks is
    // true, mark DB read-only
    bg_error_ = s;
  }
1275
  RecordFlushIOStats();
1276 1277 1278
#ifndef ROCKSDB_LITE
  if (s.ok()) {
    // may temporarily unlock and lock the mutex.
1279
    NotifyOnFlushCompleted(cfd, file_number, mutable_cf_options);
1280 1281
  }
#endif  // ROCKSDB_LITE
J
jorlow@chromium.org 已提交
1282 1283 1284
  return s;
}

1285
void DBImpl::NotifyOnFlushCompleted(
1286 1287
    ColumnFamilyData* cfd, uint64_t file_number,
    const MutableCFOptions& mutable_cf_options) {
1288
#ifndef ROCKSDB_LITE
1289
  if (db_options_.listeners.size() == 0U) {
1290 1291
    return;
  }
1292 1293 1294 1295 1296 1297
  mutex_.AssertHeld();
  if (shutting_down_.load(std::memory_order_acquire)) {
    return;
  }
  bool triggered_flush_slowdown =
      (cfd->current()->storage_info()->NumLevelFiles(0) >=
1298
       mutable_cf_options.level0_slowdown_writes_trigger);
1299 1300
  bool triggered_flush_stop =
      (cfd->current()->storage_info()->NumLevelFiles(0) >=
1301
       mutable_cf_options.level0_stop_writes_trigger);
1302 1303
  // release lock while notifying events
  mutex_.Unlock();
1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314
  {
    // TODO(yhchiang): make db_paths dynamic.
    auto file_path = MakeTableFileName(db_options_.db_paths[0].path,
                                       file_number);
    for (auto listener : db_options_.listeners) {
      listener->OnFlushCompleted(
          this, cfd->GetName(), file_path,
          // Use path 0 as fulled memtables are first flushed into path 0.
          triggered_flush_slowdown, triggered_flush_stop);
    }
  }
1315 1316 1317
  mutex_.Lock();
  // no need to signal bg_cv_ as it will be signaled at the end of the
  // flush process.
1318
#endif  // ROCKSDB_LITE
1319 1320
}

1321
Status DBImpl::CompactRange(ColumnFamilyHandle* column_family,
1322
                            const Slice* begin, const Slice* end,
1323
                            bool change_level, int target_level,
1324
                            uint32_t target_path_id) {
1325
  if (target_path_id >= db_options_.db_paths.size()) {
1326 1327 1328
    return Status::InvalidArgument("Invalid target path ID");
  }

1329 1330
  auto cfh = reinterpret_cast<ColumnFamilyHandleImpl*>(column_family);
  auto cfd = cfh->cfd();
I
Igor Canadi 已提交
1331 1332

  Status s = FlushMemTable(cfd, FlushOptions());
L
Lei Jin 已提交
1333
  if (!s.ok()) {
1334
    LogFlush(db_options_.info_log);
L
Lei Jin 已提交
1335 1336 1337
    return s;
  }

I
Igor Canadi 已提交
1338
  int max_level_with_files = 0;
G
Gabor Cselle 已提交
1339
  {
1340
    InstrumentedMutexLock l(&mutex_);
I
Igor Canadi 已提交
1341
    Version* base = cfd->current();
1342 1343
    for (int level = 1; level < base->storage_info()->num_non_empty_levels();
         level++) {
S
sdong 已提交
1344
      if (base->storage_info()->OverlapInLevel(level, begin, end)) {
G
Gabor Cselle 已提交
1345 1346 1347 1348
        max_level_with_files = level;
      }
    }
  }
1349 1350 1351 1352 1353

  if (cfd->ioptions()->compaction_style == kCompactionStyleUniversal &&
      cfd->NumberLevels() > 1) {
    // Always compact all files together.
    s = RunManualCompaction(cfd, ColumnFamilyData::kCompactAllLevels,
1354 1355
                            cfd->NumberLevels() - 1, target_path_id, begin,
                            end);
1356 1357 1358 1359 1360 1361 1362 1363 1364 1365 1366
  } else {
    for (int level = 0; level <= max_level_with_files; level++) {
      // in case the compaction is unversal or if we're compacting the
      // bottom-most level, the output level will be the same as input one.
      // level 0 can never be the bottommost level (i.e. if all files are in
      // level 0, we will compact to level 1)
      if (cfd->ioptions()->compaction_style == kCompactionStyleUniversal ||
          cfd->ioptions()->compaction_style == kCompactionStyleFIFO ||
          (level == max_level_with_files && level > 0)) {
        s = RunManualCompaction(cfd, level, level, target_path_id, begin, end);
      } else {
1367 1368 1369 1370 1371 1372 1373
        int output_level = level + 1;
        if (cfd->ioptions()->compaction_style == kCompactionStyleLevel &&
            cfd->ioptions()->level_compaction_dynamic_level_bytes &&
            level == 0) {
          output_level = ColumnFamilyData::kCompactToBaseLevel;
        }
        s = RunManualCompaction(cfd, level, output_level, target_path_id, begin,
1374 1375 1376 1377 1378
                                end);
      }
      if (!s.ok()) {
        break;
      }
1379 1380
      TEST_SYNC_POINT("DBImpl::RunManualCompaction()::1");
      TEST_SYNC_POINT("DBImpl::RunManualCompaction()::2");
1381
    }
G
Gabor Cselle 已提交
1382
  }
1383 1384 1385 1386
  if (!s.ok()) {
    LogFlush(db_options_.info_log);
    return s;
  }
1387

1388
  if (change_level) {
I
Igor Canadi 已提交
1389
    s = ReFitLevel(cfd, max_level_with_files, target_level);
1390
  }
1391
  LogFlush(db_options_.info_log);
L
Lei Jin 已提交
1392

1393
  {
1394
    InstrumentedMutexLock l(&mutex_);
1395 1396 1397 1398 1399
    // an automatic compaction that has been scheduled might have been
    // preempted by the manual compactions. Need to schedule it back.
    MaybeScheduleFlushOrCompaction();
  }

L
Lei Jin 已提交
1400
  return s;
1401 1402
}

1403 1404 1405 1406 1407
Status DBImpl::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) {
I
Igor Canadi 已提交
1408 1409 1410 1411
#ifdef ROCKSDB_LITE
    // not supported in lite version
  return Status::NotSupported("Not supported in ROCKSDB LITE");
#else
1412 1413 1414 1415 1416 1417
  if (column_family == nullptr) {
    return Status::InvalidArgument("ColumnFamilyHandle must be non-null.");
  }

  auto cfd = reinterpret_cast<ColumnFamilyHandleImpl*>(column_family)->cfd();
  assert(cfd);
1418 1419 1420 1421 1422 1423 1424 1425 1426 1427 1428 1429 1430 1431 1432 1433 1434 1435 1436 1437 1438 1439 1440 1441 1442

  Status s;
  JobContext job_context(0, true);
  LogBuffer log_buffer(InfoLogLevel::INFO_LEVEL,
                       db_options_.info_log.get());

  // Perform CompactFiles
  SuperVersion* sv = GetAndRefSuperVersion(cfd);
  {
    InstrumentedMutexLock l(&mutex_);

    s = CompactFilesImpl(compact_options, cfd, sv->current,
                         input_file_names, output_level,
                         output_path_id, &job_context, &log_buffer);
  }
  ReturnAndCleanupSuperVersion(cfd, sv);

  // Find and delete obsolete files
  {
    InstrumentedMutexLock l(&mutex_);
    // If !s.ok(), this means that Compaction failed. In that case, we want
    // to delete all obsolete files we might have created and we force
    // FindObsoleteFiles(). This is because job_context does not
    // catch all created files if compaction failed.
    FindObsoleteFiles(&job_context, !s.ok());
1443
  }  // release the mutex
1444 1445 1446 1447 1448 1449 1450 1451 1452 1453

  // delete unnecessary files if any, this is done outside the mutex
  if (job_context.HaveSomethingToDelete() || !log_buffer.IsEmpty()) {
    // Have to flush the info logs before bg_compaction_scheduled_--
    // because if bg_flush_scheduled_ becomes 0 and the lock is
    // released, the deconstructor of DB can kick in and destroy all the
    // states of DB so info_log might not be available after that point.
    // It also applies to access other states that DB owns.
    log_buffer.FlushBufferToLog();
    if (job_context.HaveSomethingToDelete()) {
1454
      // no mutex is locked here.  No need to Unlock() and Lock() here.
1455 1456 1457 1458 1459
      PurgeObsoleteFiles(job_context);
    }
    job_context.Clean();
  }

1460
  return s;
I
Igor Canadi 已提交
1461
#endif  // ROCKSDB_LITE
1462 1463
}

I
Igor Canadi 已提交
1464
#ifndef ROCKSDB_LITE
1465 1466 1467
Status DBImpl::CompactFilesImpl(
    const CompactionOptions& compact_options, ColumnFamilyData* cfd,
    Version* version, const std::vector<std::string>& input_file_names,
1468 1469
    const int output_level, int output_path_id, JobContext* job_context,
    LogBuffer* log_buffer) {
1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491 1492 1493 1494 1495 1496 1497 1498 1499 1500 1501
  mutex_.AssertHeld();

  if (shutting_down_.load(std::memory_order_acquire)) {
    return Status::ShutdownInProgress();
  }

  std::unordered_set<uint64_t> input_set;
  for (auto file_name : input_file_names) {
    input_set.insert(TableFileNameToNumber(file_name));
  }

  ColumnFamilyMetaData cf_meta;
  // TODO(yhchiang): can directly use version here if none of the
  // following functions call is pluggable to external developers.
  version->GetColumnFamilyMetaData(&cf_meta);

  if (output_path_id < 0) {
    if (db_options_.db_paths.size() == 1U) {
      output_path_id = 0;
    } else {
      return Status::NotSupported(
          "Automatic output path selection is not "
          "yet supported in CompactFiles()");
    }
  }

  Status s = cfd->compaction_picker()->SanitizeCompactionInputFiles(
      &input_set, cf_meta, output_level);
  if (!s.ok()) {
    return s;
  }

1502
  std::vector<CompactionInputFiles> input_files;
1503 1504 1505 1506 1507 1508 1509 1510 1511 1512 1513 1514 1515 1516 1517 1518 1519 1520 1521 1522
  s = cfd->compaction_picker()->GetCompactionInputsFromFileNumbers(
      &input_files, &input_set, version->storage_info(), compact_options);
  if (!s.ok()) {
    return s;
  }

  for (auto inputs : input_files) {
    if (cfd->compaction_picker()->FilesInCompaction(inputs.files)) {
      return Status::Aborted(
          "Some of the necessary compaction input "
          "files are already being compacted");
    }
  }

  // At this point, CompactFiles will be run.
  bg_compaction_scheduled_++;

  unique_ptr<Compaction> c;
  assert(cfd->compaction_picker());
  c.reset(cfd->compaction_picker()->FormCompaction(
1523 1524
      compact_options, input_files, output_level, version->storage_info(),
      *cfd->GetLatestMutableCFOptions(), output_path_id));
1525 1526 1527 1528 1529 1530
  assert(c);
  c->SetInputVersion(version);
  // deletion compaction currently not allowed in CompactFiles.
  assert(!c->IsDeletionCompaction());

  auto yield_callback = [&]() {
1531 1532 1533
    return CallFlushDuringCompaction(
        c->column_family_data(), *c->mutable_cf_options(),
        job_context, log_buffer);
1534
  };
I
Igor Canadi 已提交
1535
  assert(is_snapshot_supported_ || snapshots_.empty());
1536
  CompactionJob compaction_job(
I
Igor Canadi 已提交
1537 1538 1539 1540
      job_context->job_id, c.get(), db_options_, env_options_, versions_.get(),
      &shutting_down_, log_buffer, directories_.GetDbDir(),
      directories_.GetDataDir(c->GetOutputPathId()), stats_,
      snapshots_.GetAll(), table_cache_, std::move(yield_callback),
1541 1542 1543 1544 1545 1546 1547 1548 1549 1550 1551 1552 1553 1554
      &event_logger_, c->mutable_cf_options()->paranoid_file_checks, dbname_,
      nullptr);  // Here we pass a nullptr for CompactionJobStats because
                 // CompactFiles does not trigger OnCompactionCompleted(),
                 // which is the only place where CompactionJobStats is
                 // returned.  The idea of not triggering OnCompationCompleted()
                 // is that CompactFiles runs in the caller thread, so the user
                 // should always know when it completes.  As a result, it makes
                 // less sense to notify the users something they should already
                 // know.
                 //
                 // In the future, if we would like to add CompactionJobStats
                 // support for CompactFiles, we should have CompactFiles API
                 // pass a pointer of CompactionJobStats as the out-value
                 // instead of using EventListener.
1555 1556 1557 1558 1559
  compaction_job.Prepare();

  mutex_.Unlock();
  Status status = compaction_job.Run();
  mutex_.Lock();
1560

I
Igor Canadi 已提交
1561
  compaction_job.Install(&status, *c->mutable_cf_options(), &mutex_);
1562
  if (status.ok()) {
1563
    InstallSuperVersionBackground(c->column_family_data(), job_context,
1564
                                  *c->mutable_cf_options());
1565 1566 1567 1568 1569 1570 1571 1572 1573
  }
  c->ReleaseCompactionFiles(s);
  c.reset();

  if (status.ok()) {
    // Done
  } else if (status.IsShutdownInProgress()) {
    // Ignore compaction errors found during shutting down
  } else {
1574 1575
    Log(InfoLogLevel::WARN_LEVEL, db_options_.info_log,
        "[%s] [JOB %d] Compaction error: %s",
1576
        c->column_family_data()->GetName().c_str(), job_context->job_id,
1577 1578 1579 1580 1581 1582 1583
        status.ToString().c_str());
    if (db_options_.paranoid_checks && bg_error_.ok()) {
      bg_error_ = status;
    }
  }

  bg_compaction_scheduled_--;
1584 1585 1586
  if (bg_compaction_scheduled_ == 0) {
    bg_cv_.SignalAll();
  }
1587 1588 1589

  return status;
}
I
Igor Canadi 已提交
1590
#endif  // ROCKSDB_LITE
1591

O
Ori Bernstein 已提交
1592
void DBImpl::NotifyOnCompactionCompleted(
1593 1594
    ColumnFamilyData* cfd, Compaction *c, const Status &st,
    const CompactionJobStats& compaction_job_stats,
1595
    const int job_id) {
O
Ori Bernstein 已提交
1596
#ifndef ROCKSDB_LITE
1597
  if (db_options_.listeners.size() == 0U) {
O
Ori Bernstein 已提交
1598 1599 1600 1601 1602 1603 1604 1605
    return;
  }
  mutex_.AssertHeld();
  if (shutting_down_.load(std::memory_order_acquire)) {
    return;
  }
  // release lock while notifying events
  mutex_.Unlock();
1606 1607 1608 1609
  {
    CompactionJobInfo info;
    info.cf_name = cfd->GetName();
    info.status = st;
1610 1611 1612
    info.thread_id = ThreadStatusUtil::GetThreadID();
    info.job_id = job_id;
    info.base_input_level = c->start_level();
1613
    info.output_level = c->output_level();
1614
    info.stats = compaction_job_stats;
1615 1616 1617 1618 1619 1620 1621 1622 1623 1624 1625 1626 1627 1628 1629 1630 1631 1632
    for (size_t i = 0; i < c->num_input_levels(); ++i) {
      for (const auto fmd : *c->inputs(i)) {
        info.input_files.push_back(
            TableFileName(db_options_.db_paths,
                          fmd->fd.GetNumber(),
                          fmd->fd.GetPathId()));
      }
    }
    for (const auto newf : c->edit()->GetNewFiles()) {
      info.output_files.push_back(
          TableFileName(db_options_.db_paths,
                        newf.second.fd.GetNumber(),
                        newf.second.fd.GetPathId()));
    }
    for (auto listener : db_options_.listeners) {
      listener->OnCompactionCompleted(this, info);
    }
  }
O
Ori Bernstein 已提交
1633 1634 1635 1636 1637 1638
  mutex_.Lock();
  // no need to signal bg_cv_ as it will be signaled at the end of the
  // flush process.
#endif  // ROCKSDB_LITE
}

1639
Status DBImpl::SetOptions(ColumnFamilyHandle* column_family,
1640
    const std::unordered_map<std::string, std::string>& options_map) {
I
Igor Canadi 已提交
1641 1642 1643
#ifdef ROCKSDB_LITE
  return Status::NotSupported("Not supported in ROCKSDB LITE");
#else
L
Lei Jin 已提交
1644 1645
  auto* cfd = reinterpret_cast<ColumnFamilyHandleImpl*>(column_family)->cfd();
  if (options_map.empty()) {
1646 1647
    Log(InfoLogLevel::WARN_LEVEL,
        db_options_.info_log, "SetOptions() on column family [%s], empty input",
L
Lei Jin 已提交
1648
        cfd->GetName().c_str());
1649
    return Status::InvalidArgument("empty input");
L
Lei Jin 已提交
1650 1651 1652
  }

  MutableCFOptions new_options;
1653
  Status s;
L
Lei Jin 已提交
1654
  {
1655
    InstrumentedMutexLock l(&mutex_);
1656 1657
    s = cfd->SetOptions(options_map);
    if (s.ok()) {
L
Lei Jin 已提交
1658 1659 1660 1661
      new_options = *cfd->GetLatestMutableCFOptions();
    }
  }

1662 1663
  Log(InfoLogLevel::INFO_LEVEL, db_options_.info_log,
      "SetOptions() on column family [%s], inputs:",
L
Lei Jin 已提交
1664 1665
      cfd->GetName().c_str());
  for (const auto& o : options_map) {
1666 1667
    Log(InfoLogLevel::INFO_LEVEL, db_options_.info_log,
        "%s: %s\n", o.first.c_str(), o.second.c_str());
L
Lei Jin 已提交
1668
  }
1669
  if (s.ok()) {
1670 1671
    Log(InfoLogLevel::INFO_LEVEL,
        db_options_.info_log, "[%s] SetOptions succeeded",
L
Lei Jin 已提交
1672 1673 1674
        cfd->GetName().c_str());
    new_options.Dump(db_options_.info_log.get());
  } else {
1675 1676
    Log(InfoLogLevel::WARN_LEVEL, db_options_.info_log,
        "[%s] SetOptions failed", cfd->GetName().c_str());
L
Lei Jin 已提交
1677
  }
1678
  return s;
I
Igor Canadi 已提交
1679
#endif  // ROCKSDB_LITE
1680 1681
}

1682
// return the same level if it cannot be moved
1683 1684
int DBImpl::FindMinimumEmptyLevelFitting(ColumnFamilyData* cfd,
    const MutableCFOptions& mutable_cf_options, int level) {
1685
  mutex_.AssertHeld();
S
sdong 已提交
1686
  const auto* vstorage = cfd->current()->storage_info();
1687
  int minimum_level = level;
1688
  for (int i = level - 1; i > 0; --i) {
1689
    // stop if level i is not empty
S
sdong 已提交
1690
    if (vstorage->NumLevelFiles(i) > 0) break;
1691
    // stop if level i is too small (cannot fit the level files)
1692
    if (vstorage->MaxBytesForLevel(i) < vstorage->NumLevelBytes(level)) {
1693 1694
      break;
    }
1695 1696 1697 1698 1699 1700

    minimum_level = i;
  }
  return minimum_level;
}

I
Igor Canadi 已提交
1701 1702
Status DBImpl::ReFitLevel(ColumnFamilyData* cfd, int level, int target_level) {
  assert(level < cfd->NumberLevels());
1703 1704 1705
  if (target_level >= cfd->NumberLevels()) {
    return Status::InvalidArgument("Target level exceeds number of levels");
  }
1706

I
Igor Canadi 已提交
1707
  SuperVersion* superversion_to_free = nullptr;
1708
  SuperVersion* new_superversion = new SuperVersion();
I
Igor Canadi 已提交
1709 1710

  mutex_.Lock();
1711 1712 1713

  // only allow one thread refitting
  if (refitting_level_) {
I
Igor Canadi 已提交
1714
    mutex_.Unlock();
1715 1716
    Log(InfoLogLevel::INFO_LEVEL, db_options_.info_log,
        "[ReFitLevel] another thread is refitting");
I
Igor Canadi 已提交
1717
    delete new_superversion;
L
Lei Jin 已提交
1718
    return Status::NotSupported("another thread is refitting");
1719 1720 1721 1722 1723
  }
  refitting_level_ = true;

  // wait for all background threads to stop
  bg_work_gate_closed_ = true;
1724
  while (bg_compaction_scheduled_ > 0 || bg_flush_scheduled_) {
1725 1726
    Log(InfoLogLevel::INFO_LEVEL, db_options_.info_log,
        "[RefitLevel] waiting for background threads to stop: %d %d",
1727
        bg_compaction_scheduled_, bg_flush_scheduled_);
1728 1729 1730
    bg_cv_.Wait();
  }

1731 1732
  const MutableCFOptions mutable_cf_options =
    *cfd->GetLatestMutableCFOptions();
1733
  // move to a smaller level
1734 1735
  int to_level = target_level;
  if (target_level < 0) {
1736
    to_level = FindMinimumEmptyLevelFitting(cfd, mutable_cf_options, level);
1737
  }
1738

L
Lei Jin 已提交
1739
  Status status;
1740 1741 1742 1743 1744 1745 1746 1747 1748 1749 1750 1751 1752 1753 1754
  auto* vstorage = cfd->current()->storage_info();
  if (to_level > level) {
    if (level == 0) {
      return Status::NotSupported(
          "Cannot change from level 0 to other levels.");
    }
    // Check levels are empty for a trivial move
    for (int l = level + 1; l <= to_level; l++) {
      if (vstorage->NumLevelFiles(l) > 0) {
        return Status::NotSupported(
            "Levels between source and target are not empty for a move.");
      }
    }
  }
  if (to_level != level) {
1755 1756
    Log(InfoLogLevel::DEBUG_LEVEL, db_options_.info_log,
        "[%s] Before refitting:\n%s",
1757
        cfd->GetName().c_str(), cfd->current()->DebugString().data());
1758

1759
    VersionEdit edit;
I
Igor Canadi 已提交
1760
    edit.SetColumnFamily(cfd->GetID());
1761
    for (const auto& f : vstorage->LevelFiles(level)) {
1762
      edit.DeleteFile(level, f->fd.GetNumber());
1763 1764 1765
      edit.AddFile(to_level, f->fd.GetNumber(), f->fd.GetPathId(),
                   f->fd.GetFileSize(), f->smallest, f->largest,
                   f->smallest_seqno, f->largest_seqno);
1766
    }
1767 1768
    Log(InfoLogLevel::DEBUG_LEVEL, db_options_.info_log,
        "[%s] Apply version edit:\n%s",
I
Igor Canadi 已提交
1769
        cfd->GetName().c_str(), edit.DebugString().data());
1770

1771 1772
    status = versions_->LogAndApply(cfd, mutable_cf_options, &edit, &mutex_,
                                    directories_.GetDbDir());
L
Lei Jin 已提交
1773 1774
    superversion_to_free = InstallSuperVersion(
        cfd, new_superversion, mutable_cf_options);
I
Igor Canadi 已提交
1775
    new_superversion = nullptr;
1776

1777 1778
    Log(InfoLogLevel::DEBUG_LEVEL, db_options_.info_log,
        "[%s] LogAndApply: %s\n", cfd->GetName().c_str(),
I
Igor Canadi 已提交
1779
        status.ToString().data());
1780 1781

    if (status.ok()) {
1782 1783
      Log(InfoLogLevel::DEBUG_LEVEL, db_options_.info_log,
          "[%s] After refitting:\n%s",
I
Igor Canadi 已提交
1784
          cfd->GetName().c_str(), cfd->current()->DebugString().data());
1785 1786 1787 1788 1789
    }
  }

  refitting_level_ = false;
  bg_work_gate_closed_ = false;
I
Igor Canadi 已提交
1790 1791 1792 1793

  mutex_.Unlock();
  delete superversion_to_free;
  delete new_superversion;
L
Lei Jin 已提交
1794
  return status;
G
Gabor Cselle 已提交
1795 1796
}

1797 1798 1799
int DBImpl::NumberLevels(ColumnFamilyHandle* column_family) {
  auto cfh = reinterpret_cast<ColumnFamilyHandleImpl*>(column_family);
  return cfh->cfd()->NumberLevels();
1800 1801
}

1802 1803
int DBImpl::MaxMemCompactionLevel(ColumnFamilyHandle* column_family) {
  auto cfh = reinterpret_cast<ColumnFamilyHandleImpl*>(column_family);
1804
  InstrumentedMutexLock l(&mutex_);
L
Lei Jin 已提交
1805 1806
  return cfh->cfd()->GetSuperVersion()->
      mutable_cf_options.max_mem_compaction_level;
1807 1808
}

1809 1810
int DBImpl::Level0StopWriteTrigger(ColumnFamilyHandle* column_family) {
  auto cfh = reinterpret_cast<ColumnFamilyHandleImpl*>(column_family);
1811
  InstrumentedMutexLock l(&mutex_);
L
Lei Jin 已提交
1812 1813
  return cfh->cfd()->GetSuperVersion()->
      mutable_cf_options.level0_stop_writes_trigger;
1814 1815
}

L
Lei Jin 已提交
1816
Status DBImpl::Flush(const FlushOptions& flush_options,
1817 1818
                     ColumnFamilyHandle* column_family) {
  auto cfh = reinterpret_cast<ColumnFamilyHandleImpl*>(column_family);
L
Lei Jin 已提交
1819
  return FlushMemTable(cfh->cfd(), flush_options);
H
heyongqiang 已提交
1820 1821
}

1822
SequenceNumber DBImpl::GetLatestSequenceNumber() const {
1823 1824 1825
  return versions_->LastSequence();
}

I
Igor Canadi 已提交
1826
Status DBImpl::RunManualCompaction(ColumnFamilyData* cfd, int input_level,
1827
                                   int output_level, uint32_t output_path_id,
1828 1829
                                   const Slice* begin, const Slice* end,
                                   bool disallow_trivial_move) {
1830 1831
  assert(input_level == ColumnFamilyData::kCompactAllLevels ||
         input_level >= 0);
1832

G
Gabor Cselle 已提交
1833 1834
  InternalKey begin_storage, end_storage;

H
hans@chromium.org 已提交
1835
  ManualCompaction manual;
I
Igor Canadi 已提交
1836
  manual.cfd = cfd;
1837 1838
  manual.input_level = input_level;
  manual.output_level = output_level;
1839
  manual.output_path_id = output_path_id;
G
Gabor Cselle 已提交
1840
  manual.done = false;
1841
  manual.in_progress = false;
1842
  manual.disallow_trivial_move = disallow_trivial_move;
1843 1844 1845
  // For universal compaction, we enforce every manual compaction to compact
  // all files.
  if (begin == nullptr ||
1846 1847
      cfd->ioptions()->compaction_style == kCompactionStyleUniversal ||
      cfd->ioptions()->compaction_style == kCompactionStyleFIFO) {
1848
    manual.begin = nullptr;
G
Gabor Cselle 已提交
1849
  } else {
1850
    begin_storage.SetMaxPossibleForUserKey(*begin);
G
Gabor Cselle 已提交
1851 1852
    manual.begin = &begin_storage;
  }
1853
  if (end == nullptr ||
1854 1855
      cfd->ioptions()->compaction_style == kCompactionStyleUniversal ||
      cfd->ioptions()->compaction_style == kCompactionStyleFIFO) {
1856
    manual.end = nullptr;
G
Gabor Cselle 已提交
1857
  } else {
1858
    end_storage.SetMinPossibleForUserKey(*end);
G
Gabor Cselle 已提交
1859 1860 1861
    manual.end = &end_storage;
  }

1862
  InstrumentedMutexLock l(&mutex_);
1863

1864 1865 1866 1867 1868 1869
  // When a manual compaction arrives, temporarily disable scheduling of
  // non-manual compactions and wait until the number of scheduled compaction
  // jobs drops to zero. This is needed to ensure that this manual compaction
  // can compact any range of keys/files.
  //
  // bg_manual_only_ is non-zero when at least one thread is inside
1870
  // RunManualCompaction(), i.e. during that time no other compaction will
1871 1872 1873
  // get scheduled (see MaybeScheduleFlushOrCompaction).
  //
  // Note that the following loop doesn't stop more that one thread calling
1874
  // RunManualCompaction() from getting to the second while loop below.
1875 1876 1877 1878 1879
  // However, only one of them will actually schedule compaction, while
  // others will wait on a condition variable until it completes.

  ++bg_manual_only_;
  while (bg_compaction_scheduled_ > 0) {
1880
    Log(InfoLogLevel::INFO_LEVEL, db_options_.info_log,
I
Igor Canadi 已提交
1881 1882 1883
        "[%s] Manual compaction waiting for all other scheduled background "
        "compactions to finish",
        cfd->GetName().c_str());
1884 1885
    bg_cv_.Wait();
  }
1886

1887 1888
  Log(InfoLogLevel::INFO_LEVEL, db_options_.info_log,
      "[%s] Manual compaction starting",
I
Igor Canadi 已提交
1889
      cfd->GetName().c_str());
1890

1891 1892 1893 1894
  // We don't check bg_error_ here, because if we get the error in compaction,
  // the compaction will set manual.status to bg_error_ and set manual.done to
  // true.
  while (!manual.done) {
1895 1896 1897
    assert(bg_manual_only_ > 0);
    if (manual_compaction_ != nullptr) {
      // Running either this or some other manual compaction
G
Gabor Cselle 已提交
1898
      bg_cv_.Wait();
1899 1900
    } else {
      manual_compaction_ = &manual;
1901
      bg_compaction_scheduled_++;
1902
      env_->Schedule(&DBImpl::BGWorkCompaction, this, Env::Priority::LOW, this);
G
Gabor Cselle 已提交
1903
    }
H
hans@chromium.org 已提交
1904
  }
1905

1906 1907 1908
  assert(!manual.in_progress);
  assert(bg_manual_only_ > 0);
  --bg_manual_only_;
L
Lei Jin 已提交
1909
  return manual.status;
J
jorlow@chromium.org 已提交
1910 1911
}

1912
Status DBImpl::FlushMemTable(ColumnFamilyData* cfd,
L
Lei Jin 已提交
1913
                             const FlushOptions& flush_options) {
S
Stanislau Hlebik 已提交
1914 1915 1916
  Status s;
  {
    WriteContext context;
1917
    InstrumentedMutexLock guard_lock(&mutex_);
1918

1919
    if (cfd->imm()->NumNotFlushed() == 0 && cfd->mem()->IsEmpty()) {
1920 1921 1922 1923
      // Nothing to flush
      return Status::OK();
    }

I
Igor Canadi 已提交
1924 1925
    WriteThread::Writer w(&mutex_);
    s = write_thread_.EnterWriteThread(&w, 0);
S
Stanislau Hlebik 已提交
1926 1927 1928 1929 1930
    assert(s.ok() && !w.done);  // No timeout and nobody should do our job

    // SetNewMemtableAndNewLogFile() will release and reacquire mutex
    // during execution
    s = SetNewMemtableAndNewLogFile(cfd, &context);
1931 1932
    write_thread_.ExitWriteThread(&w, &w, s);

S
Stanislau Hlebik 已提交
1933 1934
    cfd->imm()->FlushRequested();

1935 1936 1937
    // schedule flush
    SchedulePendingFlush(cfd);
    MaybeScheduleFlushOrCompaction();
S
Stanislau Hlebik 已提交
1938
  }
S
Stanislau Hlebik 已提交
1939

L
Lei Jin 已提交
1940
  if (s.ok() && flush_options.wait) {
1941
    // Wait until the compaction completes
1942
    s = WaitForFlushMemTable(cfd);
1943 1944
  }
  return s;
J
jorlow@chromium.org 已提交
1945 1946
}

1947
Status DBImpl::WaitForFlushMemTable(ColumnFamilyData* cfd) {
1948 1949
  Status s;
  // Wait until the compaction completes
1950
  InstrumentedMutexLock l(&mutex_);
1951
  while (cfd->imm()->NumNotFlushed() > 0 && bg_error_.ok()) {
1952 1953
    bg_cv_.Wait();
  }
1954
  if (!bg_error_.ok()) {
1955 1956 1957
    s = bg_error_;
  }
  return s;
H
heyongqiang 已提交
1958 1959
}

1960
void DBImpl::MaybeScheduleFlushOrCompaction() {
J
jorlow@chromium.org 已提交
1961
  mutex_.AssertHeld();
1962
  if (bg_work_gate_closed_) {
1963 1964
    // gate closed for background work
    return;
I
Igor Canadi 已提交
1965
  } else if (shutting_down_.load(std::memory_order_acquire)) {
J
jorlow@chromium.org 已提交
1966
    // DB is being deleted; no more background compactions
1967 1968
    return;
  }
1969

1970 1971 1972 1973
  while (unscheduled_flushes_ > 0 &&
         bg_flush_scheduled_ < db_options_.max_background_flushes) {
    unscheduled_flushes_--;
    bg_flush_scheduled_++;
1974
    env_->Schedule(&DBImpl::BGWorkFlush, this, Env::Priority::HIGH, this);
1975 1976
  }

1977 1978 1979 1980 1981 1982
  if (bg_manual_only_) {
    // only manual compactions are allowed to run. don't schedule automatic
    // compactions
    return;
  }

1983 1984 1985 1986 1987 1988 1989 1990 1991 1992
  if (db_options_.max_background_flushes == 0 &&
      bg_compaction_scheduled_ < db_options_.max_background_compactions &&
      unscheduled_flushes_ > 0) {
    // special case where flush is executed by compaction thread
    // (if max_background_flushes == 0).
    // Compaction thread will execute all the flushes
    unscheduled_flushes_ = 0;
    if (unscheduled_compactions_ > 0) {
      // bg compaction will execute one compaction
      unscheduled_compactions_--;
1993
    }
1994
    bg_compaction_scheduled_++;
1995
    env_->Schedule(&DBImpl::BGWorkCompaction, this, Env::Priority::LOW, this);
1996 1997 1998 1999 2000 2001
  }

  while (bg_compaction_scheduled_ < db_options_.max_background_compactions &&
         unscheduled_compactions_ > 0) {
    bg_compaction_scheduled_++;
    unscheduled_compactions_--;
2002
    env_->Schedule(&DBImpl::BGWorkCompaction, this, Env::Priority::LOW, this);
2003 2004 2005 2006 2007 2008 2009 2010 2011 2012 2013 2014 2015 2016 2017 2018 2019 2020 2021 2022 2023 2024 2025 2026 2027 2028 2029 2030 2031 2032 2033 2034 2035 2036 2037 2038 2039 2040 2041 2042 2043 2044 2045 2046 2047 2048
  }
}

void DBImpl::AddToCompactionQueue(ColumnFamilyData* cfd) {
  assert(!cfd->pending_compaction());
  cfd->Ref();
  compaction_queue_.push_back(cfd);
  cfd->set_pending_compaction(true);
}

ColumnFamilyData* DBImpl::PopFirstFromCompactionQueue() {
  assert(!compaction_queue_.empty());
  auto cfd = *compaction_queue_.begin();
  compaction_queue_.pop_front();
  assert(cfd->pending_compaction());
  cfd->set_pending_compaction(false);
  return cfd;
}

void DBImpl::AddToFlushQueue(ColumnFamilyData* cfd) {
  assert(!cfd->pending_flush());
  cfd->Ref();
  flush_queue_.push_back(cfd);
  cfd->set_pending_flush(true);
}

ColumnFamilyData* DBImpl::PopFirstFromFlushQueue() {
  assert(!flush_queue_.empty());
  auto cfd = *flush_queue_.begin();
  flush_queue_.pop_front();
  assert(cfd->pending_flush());
  cfd->set_pending_flush(false);
  return cfd;
}

void DBImpl::SchedulePendingFlush(ColumnFamilyData* cfd) {
  if (!cfd->pending_flush() && cfd->imm()->IsFlushPending()) {
    AddToFlushQueue(cfd);
    ++unscheduled_flushes_;
  }
}

void DBImpl::SchedulePendingCompaction(ColumnFamilyData* cfd) {
  if (!cfd->pending_compaction() && cfd->NeedsCompaction()) {
    AddToCompactionQueue(cfd);
    ++unscheduled_compactions_;
2049 2050 2051
  }
}

2052
void DBImpl::RecordFlushIOStats() {
I
Igor Canadi 已提交
2053
  RecordTick(stats_, FLUSH_WRITE_BYTES, IOSTATS(bytes_written));
2054 2055 2056
  IOSTATS_RESET(bytes_written);
}

2057
void DBImpl::BGWorkFlush(void* db) {
2058
  IOSTATS_SET_THREAD_POOL_ID(Env::Priority::HIGH);
2059 2060 2061 2062
  reinterpret_cast<DBImpl*>(db)->BackgroundCallFlush();
}

void DBImpl::BGWorkCompaction(void* db) {
2063
  IOSTATS_SET_THREAD_POOL_ID(Env::Priority::LOW);
2064
  TEST_SYNC_POINT("DBImpl::BGWorkCompaction");
2065 2066 2067
  reinterpret_cast<DBImpl*>(db)->BackgroundCallCompaction();
}

I
Igor Canadi 已提交
2068
Status DBImpl::BackgroundFlush(bool* madeProgress, JobContext* job_context,
H
Haobo Xu 已提交
2069
                               LogBuffer* log_buffer) {
2070
  mutex_.AssertHeld();
2071

2072 2073 2074 2075 2076 2077 2078
  Status status = bg_error_;
  if (status.ok() && shutting_down_.load(std::memory_order_acquire)) {
    status = Status::ShutdownInProgress();
  }

  if (!status.ok()) {
    return status;
2079 2080
  }

2081 2082 2083
  ColumnFamilyData* cfd = nullptr;
  while (!flush_queue_.empty()) {
    // This cfd is already referenced
I
Igor Canadi 已提交
2084
    auto first_cfd = PopFirstFromFlushQueue();
2085

I
Igor Canadi 已提交
2086
    if (first_cfd->IsDropped() || !first_cfd->imm()->IsFlushPending()) {
2087
      // can't flush this CF, try next one
I
Igor Canadi 已提交
2088 2089
      if (first_cfd->Unref()) {
        delete first_cfd;
2090 2091
      }
      continue;
2092
    }
2093 2094

    // found a flush!
I
Igor Canadi 已提交
2095
    cfd = first_cfd;
2096 2097 2098 2099 2100 2101 2102 2103 2104 2105 2106 2107 2108 2109 2110 2111 2112
    break;
  }

  if (cfd != nullptr) {
    const MutableCFOptions mutable_cf_options =
        *cfd->GetLatestMutableCFOptions();
    LogToBuffer(
        log_buffer,
        "Calling FlushMemTableToOutputFile with column "
        "family [%s], flush slots available %d, compaction slots available %d",
        cfd->GetName().c_str(),
        db_options_.max_background_flushes - bg_flush_scheduled_,
        db_options_.max_background_compactions - bg_compaction_scheduled_);
    status = FlushMemTableToOutputFile(cfd, mutable_cf_options, madeProgress,
                                       job_context, log_buffer);
    if (cfd->Unref()) {
      delete cfd;
2113
    }
J
jorlow@chromium.org 已提交
2114
  }
2115
  return status;
J
jorlow@chromium.org 已提交
2116 2117
}

2118
void DBImpl::BackgroundCallFlush() {
2119
  bool madeProgress = false;
2120
  JobContext job_context(next_job_id_.fetch_add(1), true);
2121 2122
  assert(bg_flush_scheduled_);

2123
  LogBuffer log_buffer(InfoLogLevel::INFO_LEVEL, db_options_.info_log.get());
H
Haobo Xu 已提交
2124
  {
2125
    InstrumentedMutexLock l(&mutex_);
H
Haobo Xu 已提交
2126

I
Igor Canadi 已提交
2127 2128 2129
    auto pending_outputs_inserted_elem =
        CaptureCurrentFileNumberInPendingOutputs();

2130
    Status s = BackgroundFlush(&madeProgress, &job_context, &log_buffer);
2131 2132 2133 2134 2135 2136 2137 2138 2139 2140 2141 2142 2143 2144 2145 2146 2147
    if (!s.ok() && !s.IsShutdownInProgress()) {
      // Wait a little bit before retrying background flush in
      // case this is an environmental problem and we do not want to
      // chew up resources for failed flushes for the duration of
      // the problem.
      uint64_t error_cnt =
        default_cf_internal_stats_->BumpAndGetBackgroundErrorCount();
      bg_cv_.SignalAll();  // In case a waiter can proceed despite the error
      mutex_.Unlock();
      Log(InfoLogLevel::ERROR_LEVEL, db_options_.info_log,
          "Waiting after background flush error: %s"
          "Accumulated background error counts: %" PRIu64,
          s.ToString().c_str(), error_cnt);
      log_buffer.FlushBufferToLog();
      LogFlush(db_options_.info_log);
      env_->SleepForMicroseconds(1000000);
      mutex_.Lock();
H
Haobo Xu 已提交
2148 2149
    }

I
Igor Canadi 已提交
2150 2151
    ReleaseFileNumberFromPendingOutputs(pending_outputs_inserted_elem);

2152 2153 2154 2155
    // We're just cleaning up for DB::Write()
    job_context.logs_to_free = logs_to_free_;
    logs_to_free_.clear();

2156 2157 2158
    // If flush failed, we want to delete all temporary files that we might have
    // created. Thus, we force full scan in FindObsoleteFiles()
    FindObsoleteFiles(&job_context, !s.ok() && !s.IsShutdownInProgress());
H
Haobo Xu 已提交
2159
    // delete unnecessary files if any, this is done outside the mutex
I
Igor Canadi 已提交
2160
    if (job_context.HaveSomethingToDelete() || !log_buffer.IsEmpty()) {
2161
      mutex_.Unlock();
2162 2163 2164 2165 2166
      // Have to flush the info logs before bg_flush_scheduled_--
      // because if bg_flush_scheduled_ becomes 0 and the lock is
      // released, the deconstructor of DB can kick in and destroy all the
      // states of DB so info_log might not be available after that point.
      // It also applies to access other states that DB owns.
H
Haobo Xu 已提交
2167
      log_buffer.FlushBufferToLog();
I
Igor Canadi 已提交
2168 2169
      if (job_context.HaveSomethingToDelete()) {
        PurgeObsoleteFiles(job_context);
2170
      }
I
Igor Canadi 已提交
2171
      job_context.Clean();
2172 2173
      mutex_.Lock();
    }
I
Igor Canadi 已提交
2174

H
Haobo Xu 已提交
2175
    bg_flush_scheduled_--;
2176 2177
    // See if there's more work to be done
    MaybeScheduleFlushOrCompaction();
I
Igor Canadi 已提交
2178
    RecordFlushIOStats();
H
Haobo Xu 已提交
2179
    bg_cv_.SignalAll();
I
Igor Canadi 已提交
2180 2181 2182 2183
    // IMPORTANT: there should be no code after calling SignalAll. This call may
    // signal the DB destructor that it's OK to proceed with destruction. In
    // that case, all DB variables will be dealloacated and referencing them
    // will cause trouble.
2184
  }
J
jorlow@chromium.org 已提交
2185 2186
}

2187
void DBImpl::BackgroundCallCompaction() {
2188
  bool madeProgress = false;
2189
  JobContext job_context(next_job_id_.fetch_add(1), true);
H
Haobo Xu 已提交
2190 2191

  MaybeDumpStats();
2192
  LogBuffer log_buffer(InfoLogLevel::INFO_LEVEL, db_options_.info_log.get());
2193
  {
2194
    InstrumentedMutexLock l(&mutex_);
I
Igor Canadi 已提交
2195 2196 2197 2198

    auto pending_outputs_inserted_elem =
        CaptureCurrentFileNumberInPendingOutputs();

2199
    assert(bg_compaction_scheduled_);
2200
    Status s = BackgroundCompaction(&madeProgress, &job_context, &log_buffer);
2201 2202 2203 2204 2205 2206
    if (!s.ok() && !s.IsShutdownInProgress()) {
      // Wait a little bit before retrying background compaction in
      // case this is an environmental problem and we do not want to
      // chew up resources for failed compactions for the duration of
      // the problem.
      uint64_t error_cnt =
2207
          default_cf_internal_stats_->BumpAndGetBackgroundErrorCount();
2208 2209 2210 2211 2212 2213 2214 2215 2216 2217
      bg_cv_.SignalAll();  // In case a waiter can proceed despite the error
      mutex_.Unlock();
      log_buffer.FlushBufferToLog();
      Log(InfoLogLevel::ERROR_LEVEL, db_options_.info_log,
          "Waiting after background compaction error: %s, "
          "Accumulated background error counts: %" PRIu64,
          s.ToString().c_str(), error_cnt);
      LogFlush(db_options_.info_log);
      env_->SleepForMicroseconds(1000000);
      mutex_.Lock();
2218
    }
H
Haobo Xu 已提交
2219

I
Igor Canadi 已提交
2220 2221
    ReleaseFileNumberFromPendingOutputs(pending_outputs_inserted_elem);

2222 2223 2224 2225
    // We're just cleaning up for DB::Write()
    job_context.logs_to_free = logs_to_free_;
    logs_to_free_.clear();

2226 2227 2228 2229
    // If compaction failed, we want to delete all temporary files that we might
    // have created (they might not be all recorded in job_context in case of a
    // failure). Thus, we force full scan in FindObsoleteFiles()
    FindObsoleteFiles(&job_context, !s.ok() && !s.IsShutdownInProgress());
2230 2231

    // delete unnecessary files if any, this is done outside the mutex
I
Igor Canadi 已提交
2232
    if (job_context.HaveSomethingToDelete() || !log_buffer.IsEmpty()) {
2233
      mutex_.Unlock();
2234 2235 2236 2237 2238
      // Have to flush the info logs before bg_compaction_scheduled_--
      // because if bg_flush_scheduled_ becomes 0 and the lock is
      // released, the deconstructor of DB can kick in and destroy all the
      // states of DB so info_log might not be available after that point.
      // It also applies to access other states that DB owns.
H
Haobo Xu 已提交
2239
      log_buffer.FlushBufferToLog();
I
Igor Canadi 已提交
2240 2241
      if (job_context.HaveSomethingToDelete()) {
        PurgeObsoleteFiles(job_context);
2242
      }
I
Igor Canadi 已提交
2243
      job_context.Clean();
2244 2245
      mutex_.Lock();
    }
D
Dhruba Borthakur 已提交
2246

2247
    bg_compaction_scheduled_--;
J
jorlow@chromium.org 已提交
2248

2249 2250
    versions_->GetColumnFamilySet()->FreeDeadColumnFamilies();

2251 2252
    // See if there's more work to be done
    MaybeScheduleFlushOrCompaction();
2253 2254
    if (madeProgress || bg_compaction_scheduled_ == 0 || bg_manual_only_ > 0) {
      // signal if
I
Igor Canadi 已提交
2255
      // * madeProgress -- need to wakeup DelayWrite
2256 2257 2258 2259 2260 2261
      // * bg_compaction_scheduled_ == 0 -- need to wakeup ~DBImpl
      // * bg_manual_only_ > 0 -- need to wakeup RunManualCompaction
      // If none of this is true, there is no need to signal since nobody is
      // waiting for it
      bg_cv_.SignalAll();
    }
I
Igor Canadi 已提交
2262 2263 2264 2265
    // IMPORTANT: there should be no code after calling SignalAll. This call may
    // signal the DB destructor that it's OK to proceed with destruction. In
    // that case, all DB variables will be dealloacated and referencing them
    // will cause trouble.
2266
  }
J
jorlow@chromium.org 已提交
2267 2268
}

I
Igor Canadi 已提交
2269
Status DBImpl::BackgroundCompaction(bool* madeProgress, JobContext* job_context,
2270
                                    LogBuffer* log_buffer) {
2271
  *madeProgress = false;
J
jorlow@chromium.org 已提交
2272
  mutex_.AssertHeld();
2273

2274 2275
  bool is_manual = (manual_compaction_ != nullptr) &&
                   (manual_compaction_->in_progress == false);
2276 2277
  bool trivial_move_disallowed = is_manual &&
                                 manual_compaction_->disallow_trivial_move;
2278

2279
  CompactionJobStats compaction_job_stats;
2280 2281 2282 2283 2284 2285
  Status status = bg_error_;
  if (status.ok() && shutting_down_.load(std::memory_order_acquire)) {
    status = Status::ShutdownInProgress();
  }

  if (!status.ok()) {
2286
    if (is_manual) {
2287
      manual_compaction_->status = status;
2288 2289 2290 2291
      manual_compaction_->done = true;
      manual_compaction_->in_progress = false;
      manual_compaction_ = nullptr;
    }
2292
    return status;
2293 2294
  }

2295 2296 2297
  if (is_manual) {
    // another thread cannot pick up the same work
    manual_compaction_->in_progress = true;
2298 2299 2300 2301
  } else if (manual_compaction_ != nullptr) {
    // there should be no automatic compactions running when manual compaction
    // is running
    return Status::OK();
2302 2303
  }

2304 2305 2306 2307 2308 2309 2310 2311 2312 2313 2314 2315 2316 2317
  // If there are no flush threads, then compaction thread needs to execute the
  // flushes
  if (db_options_.max_background_flushes == 0) {
    // BackgroundFlush() will only execute a single flush. We keep calling it as
    // long as there's more flushes to be done
    while (!flush_queue_.empty()) {
      LogToBuffer(
          log_buffer,
          "BackgroundCompaction calling BackgroundFlush. flush slots available "
          "%d, compaction slots available %d",
          db_options_.max_background_flushes - bg_flush_scheduled_,
          db_options_.max_background_compactions - bg_compaction_scheduled_);
      auto flush_status =
          BackgroundFlush(madeProgress, job_context, log_buffer);
2318 2319 2320 2321
      // the second condition will be false when a column family is dropped. we
      // don't want to fail compaction because of that (because it might be a
      // different column family)
      if (!flush_status.ok() && !flush_status.IsShutdownInProgress()) {
2322 2323 2324 2325 2326 2327 2328
        if (is_manual) {
          manual_compaction_->status = flush_status;
          manual_compaction_->done = true;
          manual_compaction_->in_progress = false;
          manual_compaction_ = nullptr;
        }
        return flush_status;
2329 2330 2331 2332 2333
      }
    }
  }

  unique_ptr<Compaction> c;
2334 2335
  InternalKey manual_end_storage;
  InternalKey* manual_end = &manual_end_storage;
H
hans@chromium.org 已提交
2336
  if (is_manual) {
G
Gabor Cselle 已提交
2337
    ManualCompaction* m = manual_compaction_;
2338
    assert(m->in_progress);
2339 2340 2341
    c.reset(m->cfd->CompactRange(
          *m->cfd->GetLatestMutableCFOptions(), m->input_level, m->output_level,
          m->output_path_id, m->begin, m->end, &manual_end));
2342
    if (!c) {
2343
      m->done = true;
2344 2345 2346 2347 2348 2349 2350 2351 2352 2353 2354 2355 2356 2357 2358 2359
      LogToBuffer(log_buffer,
                  "[%s] Manual compaction from level-%d from %s .. "
                  "%s; nothing to do\n",
                  m->cfd->GetName().c_str(), m->input_level,
                  (m->begin ? m->begin->DebugString().c_str() : "(begin)"),
                  (m->end ? m->end->DebugString().c_str() : "(end)"));
    } else {
      LogToBuffer(log_buffer,
                  "[%s] Manual compaction from level-%d to level-%d from %s .. "
                  "%s; will stop at %s\n",
                  m->cfd->GetName().c_str(), m->input_level, c->output_level(),
                  (m->begin ? m->begin->DebugString().c_str() : "(begin)"),
                  (m->end ? m->end->DebugString().c_str() : "(end)"),
                  ((m->done || manual_end == nullptr)
                       ? "(end)"
                       : manual_end->DebugString().c_str()));
G
Gabor Cselle 已提交
2360
    }
2361 2362 2363 2364 2365 2366 2367 2368 2369 2370 2371 2372 2373 2374 2375 2376 2377 2378 2379 2380 2381 2382 2383 2384 2385 2386 2387 2388 2389 2390 2391 2392 2393 2394 2395 2396 2397 2398 2399 2400 2401 2402 2403 2404 2405 2406 2407
  } else if (!compaction_queue_.empty()) {
    // cfd is referenced here
    auto cfd = PopFirstFromCompactionQueue();
    // We unreference here because the following code will take a Ref() on
    // this cfd if it is going to use it (Compaction class holds a
    // reference).
    // This will all happen under a mutex so we don't have to be afraid of
    // somebody else deleting it.
    if (cfd->Unref()) {
      delete cfd;
      // This was the last reference of the column family, so no need to
      // compact.
      return Status::OK();
    }

    // Pick up latest mutable CF Options and use it throughout the
    // compaction job
    // Compaction makes a copy of the latest MutableCFOptions. It should be used
    // throughout the compaction procedure to make sure consistency. It will
    // eventually be installed into SuperVersion
    auto* mutable_cf_options = cfd->GetLatestMutableCFOptions();
    if (!mutable_cf_options->disable_auto_compactions && !cfd->IsDropped()) {
      // NOTE: try to avoid unnecessary copy of MutableCFOptions if
      // compaction is not necessary. Need to make sure mutex is held
      // until we make a copy in the following code
      c.reset(cfd->PickCompaction(*mutable_cf_options, log_buffer));
      if (c != nullptr) {
        // update statistics
        MeasureTime(stats_, NUM_FILES_IN_SINGLE_COMPACTION,
                    c->inputs(0)->size());
        // There are three things that can change compaction score:
        // 1) When flush or compaction finish. This case is covered by
        // InstallSuperVersion()
        // 2) When MutableCFOptions changes. This case is also covered by
        // InstallSuperVersion(), because this is when the new options take
        // effect.
        // 3) When we Pick a new compaction, we "remove" those files being
        // compacted from the calculation, which then influences compaction
        // score. Here we check if we need the new compaction even without the
        // files that are currently being compacted. If we need another
        // compaction, we might be able to execute it in parallel, so we add it
        // to the queue and schedule a new thread.
        if (cfd->NeedsCompaction()) {
          // Yes, we need more compactions!
          AddToCompactionQueue(cfd);
          ++unscheduled_compactions_;
          MaybeScheduleFlushOrCompaction();
I
Igor Canadi 已提交
2408
        }
I
Igor Canadi 已提交
2409 2410
      }
    }
J
jorlow@chromium.org 已提交
2411 2412
  }

2413
  if (!c) {
H
hans@chromium.org 已提交
2414
    // Nothing to do
2415
    LogToBuffer(log_buffer, "Compaction nothing to do");
I
Igor Canadi 已提交
2416 2417 2418 2419 2420
  } else if (c->IsDeletionCompaction()) {
    // TODO(icanadi) Do we want to honor snapshots here? i.e. not delete old
    // file if there is alive snapshot pointing to it
    assert(c->num_input_files(1) == 0);
    assert(c->level() == 0);
2421
    assert(c->column_family_data()->ioptions()->compaction_style ==
I
Igor Canadi 已提交
2422
           kCompactionStyleFIFO);
2423 2424 2425

    compaction_job_stats.num_input_files = c->num_input_files(0);

I
Igor Canadi 已提交
2426
    for (const auto& f : *c->inputs(0)) {
2427
      c->edit()->DeleteFile(c->level(), f->fd.GetNumber());
I
Igor Canadi 已提交
2428
    }
2429 2430 2431
    status = versions_->LogAndApply(c->column_family_data(),
                                    *c->mutable_cf_options(), c->edit(),
                                    &mutex_, directories_.GetDbDir());
I
Igor Canadi 已提交
2432 2433
    InstallSuperVersionBackground(c->column_family_data(), job_context,
                                  *c->mutable_cf_options());
I
Igor Canadi 已提交
2434 2435 2436 2437
    LogToBuffer(log_buffer, "[%s] Deleted %d files\n",
                c->column_family_data()->GetName().c_str(),
                c->num_input_files(0));
    *madeProgress = true;
2438
  } else if (!trivial_move_disallowed && c->IsTrivialMove()) {
2439
    TEST_SYNC_POINT("DBImpl::BackgroundCompaction:TrivialMove");
2440 2441 2442 2443 2444
    // Instrument for event update
    // TODO(yhchiang): add op details for showing trivial-move.
    ThreadStatusUtil::SetColumnFamily(c->column_family_data());
    ThreadStatusUtil::SetThreadOperation(ThreadStatus::OP_COMPACTION);

2445 2446
    compaction_job_stats.num_input_files = c->num_input_files(0);

2447 2448 2449 2450 2451 2452 2453 2454 2455 2456 2457 2458 2459 2460 2461 2462 2463
    // Move files to next level
    int32_t moved_files = 0;
    int64_t moved_bytes = 0;
    for (size_t i = 0; i < c->num_input_files(0); i++) {
      FileMetaData* f = c->input(0, i);
      c->edit()->DeleteFile(c->level(), f->fd.GetNumber());
      c->edit()->AddFile(c->level() + 1, f->fd.GetNumber(), f->fd.GetPathId(),
                         f->fd.GetFileSize(), f->smallest, f->largest,
                         f->smallest_seqno, f->largest_seqno);

      LogToBuffer(log_buffer,
                  "[%s] Moving #%" PRIu64 " to level-%d %" PRIu64 " bytes\n",
                  c->column_family_data()->GetName().c_str(), f->fd.GetNumber(),
                  c->level() + 1, f->fd.GetFileSize());
      ++moved_files;
      moved_bytes += f->fd.GetFileSize();
    }
2464
    status = versions_->LogAndApply(c->column_family_data(),
2465 2466
                                    *c->mutable_cf_options(), c->edit(),
                                    &mutex_, directories_.GetDbDir());
2467
    // Use latest MutableCFOptions
I
Igor Canadi 已提交
2468 2469
    InstallSuperVersionBackground(c->column_family_data(), job_context,
                                  *c->mutable_cf_options());
2470

S
sdong 已提交
2471
    VersionStorageInfo::LevelSummaryStorage tmp;
2472 2473
    c->column_family_data()->internal_stats()->IncBytesMoved(c->level() + 1,
                                                             moved_bytes);
2474 2475 2476 2477
    {
      event_logger_.LogToBuffer(log_buffer)
          << "job" << job_context->job_id << "event"
          << "trivial_move"
2478 2479
          << "destination_level" << c->level() + 1 << "files" << moved_files
          << "total_files_size" << moved_bytes;
2480
    }
2481 2482
    LogToBuffer(
        log_buffer,
2483 2484 2485
        "[%s] Moved #%d files to level-%d %" PRIu64 " bytes %s: %s\n",
        c->column_family_data()->GetName().c_str(), moved_files, c->level() + 1,
        moved_bytes, status.ToString().c_str(),
2486
        c->column_family_data()->current()->storage_info()->LevelSummary(&tmp));
2487
    *madeProgress = true;
2488 2489 2490

    // Clear Instrument
    ThreadStatusUtil::ResetThreadStatus();
J
jorlow@chromium.org 已提交
2491
  } else {
2492
    TEST_SYNC_POINT("DBImpl::BackgroundCompaction:NonTrivial");
I
Igor Canadi 已提交
2493 2494 2495 2496 2497
    auto yield_callback = [&]() {
      return CallFlushDuringCompaction(c->column_family_data(),
                                       *c->mutable_cf_options(), job_context,
                                       log_buffer);
    };
I
Igor Canadi 已提交
2498
    assert(is_snapshot_supported_ || snapshots_.empty());
2499
    CompactionJob compaction_job(
I
Igor Canadi 已提交
2500 2501 2502 2503
        job_context->job_id, c.get(), db_options_, env_options_,
        versions_.get(), &shutting_down_, log_buffer, directories_.GetDbDir(),
        directories_.GetDataDir(c->GetOutputPathId()), stats_,
        snapshots_.GetAll(), table_cache_, std::move(yield_callback),
2504
        &event_logger_, c->mutable_cf_options()->paranoid_file_checks,
2505
        dbname_, &compaction_job_stats);
I
Igor Canadi 已提交
2506
    compaction_job.Prepare();
2507

I
Igor Canadi 已提交
2508 2509 2510
    mutex_.Unlock();
    status = compaction_job.Run();
    mutex_.Lock();
2511

I
Igor Canadi 已提交
2512
    compaction_job.Install(&status, *c->mutable_cf_options(), &mutex_);
I
Igor Canadi 已提交
2513 2514 2515 2516
    if (status.ok()) {
      InstallSuperVersionBackground(c->column_family_data(), job_context,
                                    *c->mutable_cf_options());
    }
O
Ori Bernstein 已提交
2517 2518 2519
    *madeProgress = true;
  }
  if (c != nullptr) {
2520 2521 2522
    NotifyOnCompactionCompleted(
        c->column_family_data(), c.get(), status,
        compaction_job_stats, job_context->job_id);
I
Igor Canadi 已提交
2523
    c->ReleaseCompactionFiles(status);
2524
    *madeProgress = true;
J
jorlow@chromium.org 已提交
2525
  }
2526
  // this will unref its input_version and column_family_data
2527
  c.reset();
J
jorlow@chromium.org 已提交
2528 2529 2530

  if (status.ok()) {
    // Done
F
Feng Zhu 已提交
2531
  } else if (status.IsShutdownInProgress()) {
J
jorlow@chromium.org 已提交
2532 2533
    // Ignore compaction errors found during shutting down
  } else {
2534
    Log(InfoLogLevel::WARN_LEVEL, db_options_.info_log, "Compaction error: %s",
2535
        status.ToString().c_str());
2536
    if (db_options_.paranoid_checks && bg_error_.ok()) {
J
jorlow@chromium.org 已提交
2537 2538 2539
      bg_error_ = status;
    }
  }
H
hans@chromium.org 已提交
2540 2541

  if (is_manual) {
G
Gabor Cselle 已提交
2542
    ManualCompaction* m = manual_compaction_;
2543
    if (!status.ok()) {
L
Lei Jin 已提交
2544
      m->status = status;
2545 2546
      m->done = true;
    }
2547 2548 2549 2550 2551 2552 2553 2554 2555
    // For universal compaction:
    //   Because universal compaction always happens at level 0, so one
    //   compaction will pick up all overlapped files. No files will be
    //   filtered out due to size limit and left for a successive compaction.
    //   So we can safely conclude the current compaction.
    //
    //   Also note that, if we don't stop here, then the current compaction
    //   writes a new file back to level 0, which will be used in successive
    //   compaction. Hence the manual compaction will never finish.
2556 2557 2558 2559 2560
    //
    // Stop the compaction if manual_end points to nullptr -- this means
    // that we compacted the whole range. manual_end should always point
    // to nullptr in case of universal compaction
    if (manual_end == nullptr) {
2561 2562
      m->done = true;
    }
G
Gabor Cselle 已提交
2563 2564 2565
    if (!m->done) {
      // We only compacted part of the requested range.  Update *m
      // to the range that is left to be compacted.
I
Igor Canadi 已提交
2566
      // Universal and FIFO compactions should always compact the whole range
S
sdong 已提交
2567 2568 2569
      assert(m->cfd->ioptions()->compaction_style !=
                 kCompactionStyleUniversal ||
             m->cfd->ioptions()->num_levels > 1);
2570
      assert(m->cfd->ioptions()->compaction_style != kCompactionStyleFIFO);
2571
      m->tmp_storage = *manual_end;
G
Gabor Cselle 已提交
2572 2573
      m->begin = &m->tmp_storage;
    }
2574
    m->in_progress = false; // not being processed anymore
2575
    manual_compaction_ = nullptr;
H
hans@chromium.org 已提交
2576
  }
2577
  return status;
J
jorlow@chromium.org 已提交
2578 2579
}

I
Igor Canadi 已提交
2580 2581 2582
uint64_t DBImpl::CallFlushDuringCompaction(
    ColumnFamilyData* cfd, const MutableCFOptions& mutable_cf_options,
    JobContext* job_context, LogBuffer* log_buffer) {
2583
  if (db_options_.max_background_flushes > 0) {
2584 2585 2586
    // flush thread will take care of this
    return 0;
  }
I
Igor Canadi 已提交
2587
  if (cfd->imm()->imm_flush_needed.load(std::memory_order_relaxed)) {
I
Igor Canadi 已提交
2588 2589 2590 2591
    const uint64_t imm_start = env_->NowMicros();
    mutex_.Lock();
    if (cfd->imm()->IsFlushPending()) {
      cfd->Ref();
I
Igor Canadi 已提交
2592 2593
      FlushMemTableToOutputFile(cfd, mutable_cf_options, nullptr, job_context,
                                log_buffer);
I
Igor Canadi 已提交
2594
      cfd->Unref();
I
Igor Canadi 已提交
2595
      bg_cv_.SignalAll();  // Wakeup DelayWrite() if necessary
I
Igor Canadi 已提交
2596 2597 2598 2599 2600 2601 2602 2603
    }
    mutex_.Unlock();
    log_buffer->FlushBufferToLog();
    return env_->NowMicros() - imm_start;
  }
  return 0;
}

2604 2605
namespace {
struct IterState {
2606
  IterState(DBImpl* _db, InstrumentedMutex* _mu, SuperVersion* _super_version)
I
Igor Canadi 已提交
2607
      : db(_db), mu(_mu), super_version(_super_version) {}
2608 2609

  DBImpl* db;
2610
  InstrumentedMutex* mu;
2611
  SuperVersion* super_version;
2612 2613 2614 2615
};

static void CleanupIteratorState(void* arg1, void* arg2) {
  IterState* state = reinterpret_cast<IterState*>(arg1);
2616

2617
  if (state->super_version->Unref()) {
2618 2619 2620
    // Job id == 0 means that this is not our background process, but rather
    // user thread
    JobContext job_context(0);
2621

2622 2623
    state->mu->Lock();
    state->super_version->Cleanup();
I
Igor Canadi 已提交
2624
    state->db->FindObsoleteFiles(&job_context, false, true);
2625 2626 2627
    state->mu->Unlock();

    delete state->super_version;
I
Igor Canadi 已提交
2628 2629
    if (job_context.HaveSomethingToDelete()) {
      state->db->PurgeObsoleteFiles(job_context);
2630
    }
I
Igor Canadi 已提交
2631
    job_context.Clean();
I
Igor Canadi 已提交
2632
  }
T
Tomislav Novak 已提交
2633

2634 2635
  delete state;
}
H
Hans Wennborg 已提交
2636
}  // namespace
2637

L
Lei Jin 已提交
2638
Iterator* DBImpl::NewInternalIterator(const ReadOptions& read_options,
2639
                                      ColumnFamilyData* cfd,
2640 2641 2642
                                      SuperVersion* super_version,
                                      Arena* arena) {
  Iterator* internal_iter;
2643 2644 2645 2646 2647
  assert(arena != nullptr);
  // Need to create internal iterator from the arena.
  MergeIteratorBuilder merge_iter_builder(&cfd->internal_comparator(), arena);
  // Collect iterator for mutable mem
  merge_iter_builder.AddIterator(
L
Lei Jin 已提交
2648
      super_version->mem->NewIterator(read_options, arena));
2649
  // Collect all needed child iterators for immutable memtables
L
Lei Jin 已提交
2650
  super_version->imm->AddIterators(read_options, &merge_iter_builder);
2651
  // Collect iterators for files in L0 - Ln
L
Lei Jin 已提交
2652
  super_version->current->AddIterators(read_options, env_options_,
2653 2654
                                       &merge_iter_builder);
  internal_iter = merge_iter_builder.Finish();
2655
  IterState* cleanup = new IterState(this, &mutex_, super_version);
2656
  internal_iter->RegisterCleanup(CleanupIteratorState, cleanup, nullptr);
J
jorlow@chromium.org 已提交
2657 2658 2659 2660

  return internal_iter;
}

2661 2662 2663 2664
ColumnFamilyHandle* DBImpl::DefaultColumnFamily() const {
  return default_cf_handle_;
}

L
Lei Jin 已提交
2665
Status DBImpl::Get(const ReadOptions& read_options,
2666
                   ColumnFamilyHandle* column_family, const Slice& key,
J
jorlow@chromium.org 已提交
2667
                   std::string* value) {
L
Lei Jin 已提交
2668
  return GetImpl(read_options, column_family, key, value);
2669 2670
}

I
Igor Canadi 已提交
2671 2672
// JobContext gets created and destructed outside of the lock --
// we
I
Igor Canadi 已提交
2673 2674
// use this convinently to:
// * malloc one SuperVersion() outside of the lock -- new_superversion
2675
// * delete SuperVersion()s outside of the lock -- superversions_to_free
I
Igor Canadi 已提交
2676
//
I
Igor Canadi 已提交
2677 2678 2679 2680
// However, if InstallSuperVersion() gets called twice with the same
// job_context, we can't reuse the SuperVersion() that got
// malloced
// because
I
Igor Canadi 已提交
2681 2682 2683
// first call already used it. In that rare case, we take a hit and create a
// new SuperVersion() inside of the mutex. We do similar thing
// for superversion_to_free
I
Igor Canadi 已提交
2684 2685
void DBImpl::InstallSuperVersionBackground(
    ColumnFamilyData* cfd, JobContext* job_context,
2686
    const MutableCFOptions& mutable_cf_options) {
2687
  mutex_.AssertHeld();
I
Igor Canadi 已提交
2688 2689 2690 2691
  SuperVersion* old_superversion = InstallSuperVersion(
      cfd, job_context->new_superversion, mutable_cf_options);
  job_context->new_superversion = nullptr;
  job_context->superversions_to_free.push_back(old_superversion);
I
Igor Canadi 已提交
2692 2693
}

L
Lei Jin 已提交
2694 2695
SuperVersion* DBImpl::InstallSuperVersion(
    ColumnFamilyData* cfd, SuperVersion* new_sv,
2696
    const MutableCFOptions& mutable_cf_options, bool dont_schedule_bg_work) {
L
Lei Jin 已提交
2697
  mutex_.AssertHeld();
2698 2699 2700 2701 2702 2703 2704 2705 2706

  // Update max_total_in_memory_state_
  size_t old_memtable_size = 0;
  auto* old_sv = cfd->GetSuperVersion();
  if (old_sv) {
    old_memtable_size = old_sv->mutable_cf_options.write_buffer_size *
                        old_sv->mutable_cf_options.max_write_buffer_number;
  }

L
Lei Jin 已提交
2707 2708 2709
  auto* old = cfd->InstallSuperVersion(
      new_sv ? new_sv : new SuperVersion(), &mutex_, mutable_cf_options);

2710 2711 2712 2713 2714 2715 2716 2717 2718
  // Whenever we install new SuperVersion, we might need to issue new flushes or
  // compactions. dont_schedule_bg_work is true when scheduling from write
  // thread and we don't want to add additional overhead. Callers promise to
  // call SchedulePendingFlush() and MaybeScheduleFlushOrCompaction() eventually
  if (!dont_schedule_bg_work) {
    SchedulePendingFlush(cfd);
    SchedulePendingCompaction(cfd);
    MaybeScheduleFlushOrCompaction();
  }
L
Lei Jin 已提交
2719 2720 2721 2722 2723 2724 2725 2726 2727 2728

  // Update max_total_in_memory_state_
  max_total_in_memory_state_ =
      max_total_in_memory_state_ - old_memtable_size +
      mutable_cf_options.write_buffer_size *
      mutable_cf_options.max_write_buffer_number;
  return old;
}

Status DBImpl::GetImpl(const ReadOptions& read_options,
2729 2730
                       ColumnFamilyHandle* column_family, const Slice& key,
                       std::string* value, bool* value_found) {
L
Lei Jin 已提交
2731
  StopWatch sw(env_, stats_, DB_GET);
2732
  PERF_TIMER_GUARD(get_snapshot_time);
2733

2734 2735 2736
  auto cfh = reinterpret_cast<ColumnFamilyHandleImpl*>(column_family);
  auto cfd = cfh->cfd();

2737
  SequenceNumber snapshot;
L
Lei Jin 已提交
2738 2739 2740
  if (read_options.snapshot != nullptr) {
    snapshot = reinterpret_cast<const SnapshotImpl*>(
        read_options.snapshot)->number_;
2741 2742
  } else {
    snapshot = versions_->LastSequence();
J
jorlow@chromium.org 已提交
2743
  }
2744
  // Acquire SuperVersion
2745
  SuperVersion* sv = GetAndRefSuperVersion(cfd);
2746
  // Prepare to store a list of merge operations if merge occurs.
2747
  MergeContext merge_context;
2748

2749
  Status s;
2750
  // First look in the memtable, then in the immutable memtable (if any).
2751
  // s is both in/out. When in, s could either be OK or MergeInProgress.
2752
  // merge_operands will contain the sequence of merges in the latter case.
2753
  LookupKey lkey(key, snapshot);
L
Lei Jin 已提交
2754
  PERF_TIMER_STOP(get_snapshot_time);
2755

2756
  if (sv->mem->Get(lkey, value, &s, &merge_context)) {
2757
    // Done
L
Lei Jin 已提交
2758
    RecordTick(stats_, MEMTABLE_HIT);
2759
  } else if (sv->imm->Get(lkey, value, &s, &merge_context)) {
2760
    // Done
L
Lei Jin 已提交
2761
    RecordTick(stats_, MEMTABLE_HIT);
2762
  } else {
2763
    PERF_TIMER_GUARD(get_from_output_files_time);
L
Lei Jin 已提交
2764 2765
    sv->current->Get(read_options, lkey, value, &s, &merge_context,
                     value_found);
L
Lei Jin 已提交
2766
    RecordTick(stats_, MEMTABLE_MISS);
2767
  }
2768

2769 2770
  {
    PERF_TIMER_GUARD(get_post_process_time);
2771

2772
    ReturnAndCleanupSuperVersion(cfd, sv);
2773

2774 2775 2776
    RecordTick(stats_, NUMBER_KEYS_READ);
    RecordTick(stats_, BYTES_READ, value->size());
  }
2777
  return s;
J
jorlow@chromium.org 已提交
2778 2779
}

2780
std::vector<Status> DBImpl::MultiGet(
L
Lei Jin 已提交
2781
    const ReadOptions& read_options,
2782
    const std::vector<ColumnFamilyHandle*>& column_family,
2783
    const std::vector<Slice>& keys, std::vector<std::string>* values) {
2784

L
Lei Jin 已提交
2785
  StopWatch sw(env_, stats_, DB_MULTIGET);
2786
  PERF_TIMER_GUARD(get_snapshot_time);
K
Kai Liu 已提交
2787

2788
  SequenceNumber snapshot;
2789

2790
  struct MultiGetColumnFamilyData {
I
Igor Canadi 已提交
2791
    ColumnFamilyData* cfd;
2792 2793 2794 2795 2796
    SuperVersion* super_version;
  };
  std::unordered_map<uint32_t, MultiGetColumnFamilyData*> multiget_cf_data;
  // fill up and allocate outside of mutex
  for (auto cf : column_family) {
2797 2798 2799 2800 2801 2802
    auto cfh = reinterpret_cast<ColumnFamilyHandleImpl*>(cf);
    auto cfd = cfh->cfd();
    if (multiget_cf_data.find(cfd->GetID()) == multiget_cf_data.end()) {
      auto mgcfd = new MultiGetColumnFamilyData();
      mgcfd->cfd = cfd;
      multiget_cf_data.insert({cfd->GetID(), mgcfd});
2803 2804 2805
    }
  }

2806
  mutex_.Lock();
L
Lei Jin 已提交
2807 2808 2809
  if (read_options.snapshot != nullptr) {
    snapshot = reinterpret_cast<const SnapshotImpl*>(
        read_options.snapshot)->number_;
2810 2811 2812
  } else {
    snapshot = versions_->LastSequence();
  }
2813
  for (auto mgd_iter : multiget_cf_data) {
2814 2815
    mgd_iter.second->super_version =
        mgd_iter.second->cfd->GetSuperVersion()->Ref();
2816
  }
2817
  mutex_.Unlock();
2818

2819 2820
  // Contain a list of merge operations if merge occurs.
  MergeContext merge_context;
2821

2822
  // Note: this always resizes the values array
2823 2824 2825
  size_t num_keys = keys.size();
  std::vector<Status> stat_list(num_keys);
  values->resize(num_keys);
2826 2827

  // Keep track of bytes that we read for statistics-recording later
2828
  uint64_t bytes_read = 0;
L
Lei Jin 已提交
2829
  PERF_TIMER_STOP(get_snapshot_time);
2830 2831 2832 2833

  // For each of the given keys, apply the entire "get" process as follows:
  // First look in the memtable, then in the immutable memtable (if any).
  // s is both in/out. When in, s could either be OK or MergeInProgress.
2834
  // merge_operands will contain the sequence of merges in the latter case.
2835
  for (size_t i = 0; i < num_keys; ++i) {
2836
    merge_context.Clear();
2837
    Status& s = stat_list[i];
2838 2839 2840
    std::string* value = &(*values)[i];

    LookupKey lkey(keys[i], snapshot);
2841 2842
    auto cfh = reinterpret_cast<ColumnFamilyHandleImpl*>(column_family[i]);
    auto mgd_iter = multiget_cf_data.find(cfh->cfd()->GetID());
2843 2844 2845
    assert(mgd_iter != multiget_cf_data.end());
    auto mgd = mgd_iter->second;
    auto super_version = mgd->super_version;
2846
    if (super_version->mem->Get(lkey, value, &s, &merge_context)) {
2847
      // Done
2848
    } else if (super_version->imm->Get(lkey, value, &s, &merge_context)) {
2849 2850
      // Done
    } else {
2851
      PERF_TIMER_GUARD(get_from_output_files_time);
L
Lei Jin 已提交
2852 2853
      super_version->current->Get(read_options, lkey, value, &s,
                                  &merge_context);
2854 2855 2856
    }

    if (s.ok()) {
2857
      bytes_read += value->size();
2858 2859 2860 2861
    }
  }

  // Post processing (decrement reference counts and record statistics)
2862
  PERF_TIMER_GUARD(get_post_process_time);
2863 2864
  autovector<SuperVersion*> superversions_to_delete;

I
Igor Canadi 已提交
2865
  // TODO(icanadi) do we need lock here or just around Cleanup()?
2866 2867 2868 2869 2870 2871
  mutex_.Lock();
  for (auto mgd_iter : multiget_cf_data) {
    auto mgd = mgd_iter.second;
    if (mgd->super_version->Unref()) {
      mgd->super_version->Cleanup();
      superversions_to_delete.push_back(mgd->super_version);
2872 2873
    }
  }
2874 2875 2876 2877 2878 2879 2880
  mutex_.Unlock();

  for (auto td : superversions_to_delete) {
    delete td;
  }
  for (auto mgd : multiget_cf_data) {
    delete mgd.second;
2881
  }
2882

L
Lei Jin 已提交
2883 2884 2885
  RecordTick(stats_, NUMBER_MULTIGET_CALLS);
  RecordTick(stats_, NUMBER_MULTIGET_KEYS_READ, num_keys);
  RecordTick(stats_, NUMBER_MULTIGET_BYTES_READ, bytes_read);
L
Lei Jin 已提交
2886
  PERF_TIMER_STOP(get_post_process_time);
2887

2888
  return stat_list;
2889 2890
}

L
Lei Jin 已提交
2891
Status DBImpl::CreateColumnFamily(const ColumnFamilyOptions& cf_options,
2892
                                  const std::string& column_family_name,
2893
                                  ColumnFamilyHandle** handle) {
Y
Yueh-Hsuan Chiang 已提交
2894
  Status s;
I
Igor Canadi 已提交
2895
  *handle = nullptr;
Y
Yueh-Hsuan Chiang 已提交
2896
  {
2897
    InstrumentedMutexLock l(&mutex_);
I
Igor Canadi 已提交
2898

Y
Yueh-Hsuan Chiang 已提交
2899 2900 2901 2902 2903 2904 2905 2906 2907 2908 2909 2910 2911 2912
    if (versions_->GetColumnFamilySet()->GetColumnFamily(column_family_name) !=
        nullptr) {
      return Status::InvalidArgument("Column family already exists");
    }
    VersionEdit edit;
    edit.AddColumnFamily(column_family_name);
    uint32_t new_id = versions_->GetColumnFamilySet()->GetNextColumnFamilyID();
    edit.SetColumnFamily(new_id);
    edit.SetLogNumber(logfile_number_);
    edit.SetComparatorName(cf_options.comparator->Name());

    // LogAndApply will both write the creation in MANIFEST and create
    // ColumnFamilyData object
    Options opt(db_options_, cf_options);
I
Igor Canadi 已提交
2913 2914 2915 2916 2917 2918 2919 2920
    {  // write thread
      WriteThread::Writer w(&mutex_);
      s = write_thread_.EnterWriteThread(&w, 0);
      assert(s.ok() && !w.done);  // No timeout and nobody should do our job
      // LogAndApply will both write the creation in MANIFEST and create
      // ColumnFamilyData object
      s = versions_->LogAndApply(
          nullptr, MutableCFOptions(opt, ImmutableCFOptions(opt)), &edit,
2921
          &mutex_, directories_.GetDbDir(), false, &cf_options);
I
Igor Canadi 已提交
2922 2923
      write_thread_.ExitWriteThread(&w, &w, s);
    }
Y
Yueh-Hsuan Chiang 已提交
2924 2925 2926 2927 2928 2929 2930
    if (s.ok()) {
      single_column_family_mode_ = false;
      auto* cfd =
          versions_->GetColumnFamilySet()->GetColumnFamily(column_family_name);
      assert(cfd != nullptr);
      delete InstallSuperVersion(
          cfd, nullptr, *cfd->GetLatestMutableCFOptions());
S
sdong 已提交
2931 2932 2933 2934 2935

      if (!cfd->mem()->IsSnapshotSupported()) {
        is_snapshot_supported_ = false;
      }

Y
Yueh-Hsuan Chiang 已提交
2936 2937 2938 2939 2940 2941 2942 2943 2944
      *handle = new ColumnFamilyHandleImpl(cfd, this, &mutex_);
      Log(InfoLogLevel::INFO_LEVEL, db_options_.info_log,
          "Created column family [%s] (ID %u)",
          column_family_name.c_str(), (unsigned)cfd->GetID());
    } else {
      Log(InfoLogLevel::ERROR_LEVEL, db_options_.info_log,
          "Creating column family [%s] FAILED -- %s",
          column_family_name.c_str(), s.ToString().c_str());
    }
2945
  }  // InstrumentedMutexLock l(&mutex_)
Y
Yueh-Hsuan Chiang 已提交
2946 2947

  // this is outside the mutex
2948
  if (s.ok()) {
Y
Yueh-Hsuan Chiang 已提交
2949 2950
    NewThreadStatusCfInfo(
        reinterpret_cast<ColumnFamilyHandleImpl*>(*handle)->cfd());
2951
  }
2952
  return s;
2953 2954
}

2955 2956 2957 2958
Status DBImpl::DropColumnFamily(ColumnFamilyHandle* column_family) {
  auto cfh = reinterpret_cast<ColumnFamilyHandleImpl*>(column_family);
  auto cfd = cfh->cfd();
  if (cfd->GetID() == 0) {
2959 2960
    return Status::InvalidArgument("Can't drop default column family");
  }
2961

S
sdong 已提交
2962 2963
  bool cf_support_snapshot = cfd->mem()->IsSnapshotSupported();

I
Igor Canadi 已提交
2964 2965
  VersionEdit edit;
  edit.DropColumnFamily();
2966 2967
  edit.SetColumnFamily(cfd->GetID());

2968
  Status s;
2969
  {
2970
    InstrumentedMutexLock l(&mutex_);
2971 2972 2973 2974
    if (cfd->IsDropped()) {
      s = Status::InvalidArgument("Column family already dropped!\n");
    }
    if (s.ok()) {
2975
      // we drop column family from a single write thread
I
Igor Canadi 已提交
2976 2977
      WriteThread::Writer w(&mutex_);
      s = write_thread_.EnterWriteThread(&w, 0);
2978
      assert(s.ok() && !w.done);  // No timeout and nobody should do our job
2979 2980
      s = versions_->LogAndApply(cfd, *cfd->GetLatestMutableCFOptions(),
                                 &edit, &mutex_);
I
Igor Canadi 已提交
2981
      write_thread_.ExitWriteThread(&w, &w, s);
2982
    }
S
sdong 已提交
2983 2984 2985 2986 2987 2988

    if (!cf_support_snapshot) {
      // Dropped Column Family doesn't support snapshot. Need to recalculate
      // is_snapshot_supported_.
      bool new_is_snapshot_supported = true;
      for (auto c : *versions_->GetColumnFamilySet()) {
2989
        if (!c->IsDropped() && !c->mem()->IsSnapshotSupported()) {
S
sdong 已提交
2990 2991 2992 2993 2994 2995
          new_is_snapshot_supported = false;
          break;
        }
      }
      is_snapshot_supported_ = new_is_snapshot_supported;
    }
2996
  }
2997

2998
  if (s.ok()) {
Y
Yueh-Hsuan Chiang 已提交
2999 3000 3001 3002
    // Note that here we erase the associated cf_info of the to-be-dropped
    // cfd before its ref-count goes to zero to avoid having to erase cf_info
    // later inside db_mutex.
    EraseThreadStatusCfInfo(cfd);
I
Igor Canadi 已提交
3003
    assert(cfd->IsDropped());
L
Lei Jin 已提交
3004 3005 3006
    auto* mutable_cf_options = cfd->GetLatestMutableCFOptions();
    max_total_in_memory_state_ -= mutable_cf_options->write_buffer_size *
                                  mutable_cf_options->max_write_buffer_number;
3007 3008
    Log(InfoLogLevel::INFO_LEVEL, db_options_.info_log,
        "Dropped column family with id %u\n",
3009
        cfd->GetID());
3010
  } else {
3011
    Log(InfoLogLevel::ERROR_LEVEL, db_options_.info_log,
3012
        "Dropping column family with id %u FAILED -- %s\n",
3013 3014 3015
        cfd->GetID(), s.ToString().c_str());
  }

3016
  return s;
3017 3018
}

L
Lei Jin 已提交
3019
bool DBImpl::KeyMayExist(const ReadOptions& read_options,
3020 3021
                         ColumnFamilyHandle* column_family, const Slice& key,
                         std::string* value, bool* value_found) {
3022
  if (value_found != nullptr) {
K
Kai Liu 已提交
3023 3024
    // falsify later if key-may-exist but can't fetch value
    *value_found = true;
3025
  }
L
Lei Jin 已提交
3026
  ReadOptions roptions = read_options;
3027
  roptions.read_tier = kBlockCacheTier; // read from block cache only
3028
  auto s = GetImpl(roptions, column_family, key, value, value_found);
K
Kai Liu 已提交
3029

3030
  // If block_cache is enabled and the index block of the table didn't
K
Kai Liu 已提交
3031 3032 3033
  // not present in block_cache, the return value will be Status::Incomplete.
  // In this case, key may still exist in the table.
  return s.ok() || s.IsIncomplete();
3034 3035
}

3036
Iterator* DBImpl::NewIterator(const ReadOptions& read_options,
3037 3038 3039
                              ColumnFamilyHandle* column_family) {
  auto cfh = reinterpret_cast<ColumnFamilyHandleImpl*>(column_family);
  auto cfd = cfh->cfd();
T
Tomislav Novak 已提交
3040

V
Venkatesh Radhakrishnan 已提交
3041 3042 3043 3044 3045 3046 3047 3048 3049 3050 3051 3052 3053 3054 3055 3056 3057 3058
  XFUNC_TEST("", "managed_new", managed_new1, xf_manage_new,
             reinterpret_cast<DBImpl*>(this),
             const_cast<ReadOptions*>(&read_options), is_snapshot_supported_);
  if (read_options.managed) {
#ifdef ROCKSDB_LITE
    // not supported in lite version
    return NewErrorIterator(Status::InvalidArgument(
        "Managed Iterators not supported in RocksDBLite."));
#else
    if ((read_options.tailing) || (read_options.snapshot != nullptr) ||
        (is_snapshot_supported_)) {
      return new ManagedIterator(this, read_options, cfd);
    }
    // Managed iter not supported
    return NewErrorIterator(Status::InvalidArgument(
        "Managed Iterators not supported without snapshots."));
#endif
  } else if (read_options.tailing) {
I
Igor Canadi 已提交
3059 3060 3061 3062
#ifdef ROCKSDB_LITE
    // not supported in lite version
    return nullptr;
#else
3063 3064
    SuperVersion* sv = cfd->GetReferencedSuperVersion(&mutex_);
    auto iter = new ForwardIterator(this, read_options, cfd, sv);
3065
    return NewDBIterator(env_, *cfd->ioptions(), cfd->user_comparator(), iter,
3066 3067 3068
        kMaxSequenceNumber,
        sv->mutable_cf_options.max_sequential_skip_in_iterations,
        read_options.iterate_upper_bound);
I
Igor Canadi 已提交
3069
#endif
T
Tomislav Novak 已提交
3070
  } else {
3071
    SequenceNumber latest_snapshot = versions_->LastSequence();
3072
    SuperVersion* sv = cfd->GetReferencedSuperVersion(&mutex_);
3073

I
Igor Canadi 已提交
3074
    auto snapshot =
3075 3076 3077
        read_options.snapshot != nullptr
            ? reinterpret_cast<const SnapshotImpl*>(
                read_options.snapshot)->number_
I
Igor Canadi 已提交
3078
            : latest_snapshot;
T
Tomislav Novak 已提交
3079

3080 3081 3082 3083 3084 3085 3086 3087 3088 3089 3090 3091 3092 3093 3094 3095 3096 3097 3098 3099 3100 3101 3102 3103 3104 3105 3106 3107 3108 3109 3110 3111 3112 3113 3114 3115 3116 3117 3118 3119 3120 3121 3122
    // Try to generate a DB iterator tree in continuous memory area to be
    // cache friendly. Here is an example of result:
    // +-------------------------------+
    // |                               |
    // | ArenaWrappedDBIter            |
    // |  +                            |
    // |  +---> Inner Iterator   ------------+
    // |  |                            |     |
    // |  |    +-- -- -- -- -- -- -- --+     |
    // |  +--- | Arena                 |     |
    // |       |                       |     |
    // |          Allocated Memory:    |     |
    // |       |   +-------------------+     |
    // |       |   | DBIter            | <---+
    // |           |  +                |
    // |       |   |  +-> iter_  ------------+
    // |       |   |                   |     |
    // |       |   +-------------------+     |
    // |       |   | MergingIterator   | <---+
    // |           |  +                |
    // |       |   |  +->child iter1  ------------+
    // |       |   |  |                |          |
    // |           |  +->child iter2  ----------+ |
    // |       |   |  |                |        | |
    // |       |   |  +->child iter3  --------+ | |
    // |           |                   |      | | |
    // |       |   +-------------------+      | | |
    // |       |   | Iterator1         | <--------+
    // |       |   +-------------------+      | |
    // |       |   | Iterator2         | <------+
    // |       |   +-------------------+      |
    // |       |   | Iterator3         | <----+
    // |       |   +-------------------+
    // |       |                       |
    // +-------+-----------------------+
    //
    // ArenaWrappedDBIter inlines an arena area where all the iterartor in the
    // the iterator tree is allocated in the order of being accessed when
    // querying.
    // Laying out the iterators in the order of being accessed makes it more
    // likely that any iterator pointer is close to the iterator it points to so
    // that they are likely to be in the same cache line and/or page.
    ArenaWrappedDBIter* db_iter = NewArenaWrappedDbIterator(
3123
        env_, *cfd->ioptions(), cfd->user_comparator(),
3124
        snapshot, sv->mutable_cf_options.max_sequential_skip_in_iterations,
3125
        read_options.iterate_upper_bound);
3126

3127
    Iterator* internal_iter =
3128
        NewInternalIterator(read_options, cfd, sv, db_iter->GetArena());
3129 3130 3131 3132
    db_iter->SetIterUnderDBIter(internal_iter);

    return db_iter;
  }
3133 3134
  // To stop compiler from complaining
  return nullptr;
J
jorlow@chromium.org 已提交
3135 3136
}

3137
Status DBImpl::NewIterators(
3138
    const ReadOptions& read_options,
I
Igor Canadi 已提交
3139
    const std::vector<ColumnFamilyHandle*>& column_families,
3140
    std::vector<Iterator*>* iterators) {
I
Igor Canadi 已提交
3141 3142
  iterators->clear();
  iterators->reserve(column_families.size());
V
Venkatesh Radhakrishnan 已提交
3143 3144 3145 3146 3147 3148 3149 3150 3151 3152 3153 3154 3155 3156 3157 3158 3159 3160 3161 3162
  XFUNC_TEST("", "managed_new", managed_new1, xf_manage_new,
             reinterpret_cast<DBImpl*>(this),
             const_cast<ReadOptions*>(&read_options), is_snapshot_supported_);
  if (read_options.managed) {
#ifdef ROCKSDB_LITE
    return Status::InvalidArgument(
        "Managed interator not supported in RocksDB lite");
#else
    if ((!read_options.tailing) && (read_options.snapshot == nullptr) &&
        (!is_snapshot_supported_)) {
      return Status::InvalidArgument(
          "Managed interator not supported without snapshots");
    }
    for (auto cfh : column_families) {
      auto cfd = reinterpret_cast<ColumnFamilyHandleImpl*>(cfh)->cfd();
      auto iter = new ManagedIterator(this, read_options, cfd);
      iterators->push_back(iter);
    }
#endif
  } else if (read_options.tailing) {
I
Igor Canadi 已提交
3163 3164 3165 3166
#ifdef ROCKSDB_LITE
    return Status::InvalidArgument(
        "Tailing interator not supported in RocksDB lite");
#else
I
Igor Canadi 已提交
3167 3168
    for (auto cfh : column_families) {
      auto cfd = reinterpret_cast<ColumnFamilyHandleImpl*>(cfh)->cfd();
3169 3170
      SuperVersion* sv = cfd->GetReferencedSuperVersion(&mutex_);
      auto iter = new ForwardIterator(this, read_options, cfd, sv);
L
Lei Jin 已提交
3171
      iterators->push_back(
3172
          NewDBIterator(env_, *cfd->ioptions(), cfd->user_comparator(), iter,
3173 3174
              kMaxSequenceNumber,
              sv->mutable_cf_options.max_sequential_skip_in_iterations));
I
Igor Canadi 已提交
3175
    }
I
Igor Canadi 已提交
3176
#endif
I
Igor Canadi 已提交
3177
  } else {
3178 3179
    SequenceNumber latest_snapshot = versions_->LastSequence();

I
Igor Canadi 已提交
3180
    for (size_t i = 0; i < column_families.size(); ++i) {
3181 3182 3183
      auto* cfd = reinterpret_cast<ColumnFamilyHandleImpl*>(
          column_families[i])->cfd();
      SuperVersion* sv = cfd->GetReferencedSuperVersion(&mutex_);
I
Igor Canadi 已提交
3184 3185

      auto snapshot =
3186 3187 3188
          read_options.snapshot != nullptr
              ? reinterpret_cast<const SnapshotImpl*>(
                  read_options.snapshot)->number_
I
Igor Canadi 已提交
3189 3190
              : latest_snapshot;

3191
      ArenaWrappedDBIter* db_iter = NewArenaWrappedDbIterator(
3192
          env_, *cfd->ioptions(), cfd->user_comparator(), snapshot,
3193
          sv->mutable_cf_options.max_sequential_skip_in_iterations);
3194
      Iterator* internal_iter = NewInternalIterator(
3195
          read_options, cfd, sv, db_iter->GetArena());
3196 3197
      db_iter->SetIterUnderDBIter(internal_iter);
      iterators->push_back(db_iter);
I
Igor Canadi 已提交
3198 3199 3200 3201
    }
  }

  return Status::OK();
3202 3203
}

J
jorlow@chromium.org 已提交
3204
const Snapshot* DBImpl::GetSnapshot() {
3205 3206 3207
  int64_t unix_time = 0;
  env_->GetCurrentTime(&unix_time);  // Ignore error

3208
  InstrumentedMutexLock l(&mutex_);
3209
  // returns null if the underlying memtable does not support snapshot.
S
sdong 已提交
3210
  if (!is_snapshot_supported_) return nullptr;
3211
  return snapshots_.New(versions_->LastSequence(), unix_time);
J
jorlow@chromium.org 已提交
3212 3213 3214
}

void DBImpl::ReleaseSnapshot(const Snapshot* s) {
3215
  InstrumentedMutexLock l(&mutex_);
3216
  snapshots_.Delete(reinterpret_cast<const SnapshotImpl*>(s));
J
jorlow@chromium.org 已提交
3217 3218 3219
}

// Convenience methods
3220 3221
Status DBImpl::Put(const WriteOptions& o, ColumnFamilyHandle* column_family,
                   const Slice& key, const Slice& val) {
3222
  return DB::Put(o, column_family, key, val);
J
jorlow@chromium.org 已提交
3223 3224
}

3225 3226 3227
Status DBImpl::Merge(const WriteOptions& o, ColumnFamilyHandle* column_family,
                     const Slice& key, const Slice& val) {
  auto cfh = reinterpret_cast<ColumnFamilyHandleImpl*>(column_family);
3228
  if (!cfh->cfd()->ioptions()->merge_operator) {
3229 3230
    return Status::NotSupported("Provide a merge_operator when opening DB");
  } else {
3231
    return DB::Merge(o, column_family, key, val);
3232 3233 3234
  }
}

L
Lei Jin 已提交
3235
Status DBImpl::Delete(const WriteOptions& write_options,
3236
                      ColumnFamilyHandle* column_family, const Slice& key) {
L
Lei Jin 已提交
3237
  return DB::Delete(write_options, column_family, key);
J
jorlow@chromium.org 已提交
3238 3239
}

L
Lei Jin 已提交
3240
Status DBImpl::Write(const WriteOptions& write_options, WriteBatch* my_batch) {
A
agiardullo 已提交
3241 3242 3243
  return WriteImpl(write_options, my_batch, nullptr);
}

A
agiardullo 已提交
3244
#ifndef ROCKSDB_LITE
A
agiardullo 已提交
3245 3246 3247 3248 3249
Status DBImpl::WriteWithCallback(const WriteOptions& write_options,
                                 WriteBatch* my_batch,
                                 WriteCallback* callback) {
  return WriteImpl(write_options, my_batch, callback);
}
A
agiardullo 已提交
3250
#endif  // ROCKSDB_LITE
A
agiardullo 已提交
3251 3252 3253

Status DBImpl::WriteImpl(const WriteOptions& write_options,
                         WriteBatch* my_batch, WriteCallback* callback) {
S
Stanislau Hlebik 已提交
3254 3255 3256
  if (my_batch == nullptr) {
    return Status::Corruption("Batch is nullptr!");
  }
A
agiardullo 已提交
3257 3258 3259 3260 3261 3262 3263 3264 3265 3266 3267 3268

  Status status;
  bool xfunc_attempted_write = false;
  XFUNC_TEST("transaction", "transaction_xftest_write_impl",
             xf_transaction_write1, xf_transaction_write, write_options,
             db_options_, my_batch, callback, this, &status,
             &xfunc_attempted_write);
  if (xfunc_attempted_write) {
    // Test already did the write
    return status;
  }

3269
  PERF_TIMER_GUARD(write_pre_and_post_process_time);
I
Igor Canadi 已提交
3270
  WriteThread::Writer w(&mutex_);
S
Stanislau Hlebik 已提交
3271
  w.batch = my_batch;
L
Lei Jin 已提交
3272 3273
  w.sync = write_options.sync;
  w.disableWAL = write_options.disableWAL;
S
Stanislau Hlebik 已提交
3274 3275
  w.in_batch_group = false;
  w.done = false;
A
agiardullo 已提交
3276
  w.has_callback = (callback != nullptr) ? true : false;
L
Lei Jin 已提交
3277
  w.timeout_hint_us = write_options.timeout_hint_us;
S
Stanislau Hlebik 已提交
3278 3279

  uint64_t expiration_time = 0;
I
Igor Canadi 已提交
3280
  bool has_timeout = false;
S
Stanislau Hlebik 已提交
3281
  if (w.timeout_hint_us == 0) {
I
Igor Canadi 已提交
3282
    w.timeout_hint_us = WriteThread::kNoTimeOut;
S
Stanislau Hlebik 已提交
3283 3284
  } else {
    expiration_time = env_->NowMicros() + w.timeout_hint_us;
I
Igor Canadi 已提交
3285
    has_timeout = true;
S
Stanislau Hlebik 已提交
3286 3287
  }

L
Lei Jin 已提交
3288
  if (!write_options.disableWAL) {
S
Stanislau Hlebik 已提交
3289 3290 3291
    RecordTick(stats_, WRITE_WITH_WAL);
  }

3292 3293
  StopWatch write_sw(env_, db_options_.statistics.get(), DB_WRITE);

S
Stanislau Hlebik 已提交
3294 3295
  WriteContext context;
  mutex_.Lock();
I
Igor Canadi 已提交
3296 3297 3298 3299 3300

  if (!write_options.disableWAL) {
    default_cf_internal_stats_->AddDBStats(InternalStats::WRITE_WITH_WAL, 1);
  }

A
agiardullo 已提交
3301
  status = write_thread_.EnterWriteThread(&w, expiration_time);
S
Stanislau Hlebik 已提交
3302 3303
  assert(status.ok() || status.IsTimedOut());
  if (status.IsTimedOut()) {
3304
    mutex_.Unlock();
L
Lei Jin 已提交
3305
    RecordTick(stats_, WRITE_TIMEDOUT);
3306
    return Status::TimedOut();
3307
  }
S
Stanislau Hlebik 已提交
3308 3309 3310 3311 3312 3313 3314 3315 3316 3317
  if (w.done) {  // write was done by someone else
    default_cf_internal_stats_->AddDBStats(InternalStats::WRITE_DONE_BY_OTHER,
                                           1);
    mutex_.Unlock();
    RecordTick(stats_, WRITE_DONE_BY_OTHER);
    return w.status;
  }

  RecordTick(stats_, WRITE_DONE_BY_SELF);
  default_cf_internal_stats_->AddDBStats(InternalStats::WRITE_DONE_BY_SELF, 1);
3318

3319 3320 3321 3322
  // Once reaches this point, the current writer "w" will try to do its write
  // job.  It may also pick up some of the remaining writers in the "writers_"
  // when it finds suitable, and finish them in the same write batch.
  // This is how a write job could be done by the other writer.
3323 3324 3325
  assert(!single_column_family_mode_ ||
         versions_->GetColumnFamilySet()->NumberOfColumnFamilies() == 1);

3326
  uint64_t max_total_wal_size = (db_options_.max_total_wal_size == 0)
3327
                                    ? 4 * max_total_in_memory_state_
3328
                                    : db_options_.max_total_wal_size;
3329
  if (UNLIKELY(!single_column_family_mode_) &&
3330
      alive_log_files_.begin()->getting_flushed == false &&
I
Igor Canadi 已提交
3331
      total_log_size_ > max_total_wal_size) {
I
Igor Canadi 已提交
3332
    uint64_t flush_column_family_if_log_file = alive_log_files_.begin()->number;
I
Igor Canadi 已提交
3333
    alive_log_files_.begin()->getting_flushed = true;
3334
    Log(InfoLogLevel::INFO_LEVEL, db_options_.info_log,
3335 3336 3337
        "Flushing all column families with data in WAL number %" PRIu64
        ". Total log size is %" PRIu64 " while max_total_wal_size is %" PRIu64,
        flush_column_family_if_log_file, total_log_size_, max_total_wal_size);
I
Igor Canadi 已提交
3338 3339
    // no need to refcount because drop is happening in write thread, so can't
    // happen while we're in the write thread
3340
    for (auto cfd : *versions_->GetColumnFamilySet()) {
3341 3342 3343
      if (cfd->IsDropped()) {
        continue;
      }
I
Igor Canadi 已提交
3344
      if (cfd->GetLogNumber() <= flush_column_family_if_log_file) {
S
Stanislau Hlebik 已提交
3345
        status = SetNewMemtableAndNewLogFile(cfd, &context);
I
Igor Canadi 已提交
3346 3347 3348
        if (!status.ok()) {
          break;
        }
S
Stanislau Hlebik 已提交
3349
        cfd->imm()->FlushRequested();
3350 3351
        SchedulePendingFlush(cfd);
        context.schedule_bg_work_ = true;
3352
      }
3353
    }
3354 3355 3356 3357 3358 3359 3360 3361
  } else if (UNLIKELY(write_buffer_.ShouldFlush())) {
    Log(InfoLogLevel::INFO_LEVEL, db_options_.info_log,
        "Flushing all column families. Write buffer is using %" PRIu64
        " bytes out of a total of %" PRIu64 ".",
        write_buffer_.memory_usage(), write_buffer_.buffer_size());
    // no need to refcount because drop is happening in write thread, so can't
    // happen while we're in the write thread
    for (auto cfd : *versions_->GetColumnFamilySet()) {
3362 3363 3364
      if (cfd->IsDropped()) {
        continue;
      }
3365 3366 3367 3368 3369 3370
      if (!cfd->mem()->IsEmpty()) {
        status = SetNewMemtableAndNewLogFile(cfd, &context);
        if (!status.ok()) {
          break;
        }
        cfd->imm()->FlushRequested();
3371 3372
        SchedulePendingFlush(cfd);
        context.schedule_bg_work_ = true;
3373 3374 3375
      }
    }
    MaybeScheduleFlushOrCompaction();
I
Igor Canadi 已提交
3376 3377 3378 3379 3380 3381 3382 3383 3384 3385
  }

  if (UNLIKELY(status.ok() && !bg_error_.ok())) {
    status = bg_error_;
  }

  if (UNLIKELY(status.ok() && !flush_scheduler_.Empty())) {
    status = ScheduleFlushes(&context);
  }

I
Igor Canadi 已提交
3386 3387
  if (UNLIKELY(status.ok() && (write_controller_.IsStopped() ||
                               write_controller_.GetDelay() > 0))) {
3388 3389 3390 3391 3392
    // If writer is stopped, we need to get it going,
    // so schedule flushes/compactions
    if (context.schedule_bg_work_) {
      MaybeScheduleFlushOrCompaction();
    }
3393 3394
    PERF_TIMER_STOP(write_pre_and_post_process_time);
    PERF_TIMER_GUARD(write_delay_time);
I
Igor Canadi 已提交
3395
    status = DelayWrite(expiration_time);
3396
    PERF_TIMER_START(write_pre_and_post_process_time);
I
Igor Canadi 已提交
3397 3398 3399 3400 3401
  }

  if (UNLIKELY(status.ok() && has_timeout &&
               env_->NowMicros() > expiration_time)) {
    status = Status::TimedOut();
3402
  }
3403

D
dgrogan@chromium.org 已提交
3404
  uint64_t last_sequence = versions_->LastSequence();
I
Igor Canadi 已提交
3405
  WriteThread::Writer* last_writer = &w;
A
agiardullo 已提交
3406 3407
  autovector<WriteBatch*> write_batch_group;

S
Stanislau Hlebik 已提交
3408
  if (status.ok()) {
I
Igor Canadi 已提交
3409
    write_thread_.BuildBatchGroup(&last_writer, &write_batch_group);
3410

3411 3412 3413
    // Add to log and apply to memtable.  We can release the lock
    // during this phase since &w is currently responsible for logging
    // and protects against concurrent loggers and concurrent writes
3414
    // into memtables
A
agiardullo 已提交
3415 3416 3417 3418 3419 3420 3421 3422 3423 3424 3425 3426 3427 3428 3429

    mutex_.Unlock();

    if (callback != nullptr) {
      // If this write has a validation callback, check to see if this write
      // is able to be written.  Must be called on the write thread.
      status = callback->Callback(this);
    }
  } else {
    mutex_.Unlock();
  }

  // At this point the mutex is unlocked

  if (status.ok()) {
3430 3431 3432 3433 3434 3435 3436 3437 3438 3439
      WriteBatch* updates = nullptr;
      if (write_batch_group.size() == 1) {
        updates = write_batch_group[0];
      } else {
        updates = &tmp_batch_;
        for (size_t i = 0; i < write_batch_group.size(); ++i) {
          WriteBatchInternal::Append(updates, write_batch_group[i]);
        }
      }

3440 3441 3442 3443
      const SequenceNumber current_sequence = last_sequence + 1;
      WriteBatchInternal::SetSequence(updates, current_sequence);
      int my_batch_count = WriteBatchInternal::Count(updates);
      last_sequence += my_batch_count;
3444
      const uint64_t batch_size = WriteBatchInternal::ByteSize(updates);
3445
      // Record statistics
L
Lei Jin 已提交
3446
      RecordTick(stats_, NUMBER_KEYS_WRITTEN, my_batch_count);
S
sdong 已提交
3447
      RecordTick(stats_, BYTES_WRITTEN, batch_size);
L
Lei Jin 已提交
3448
      if (write_options.disableWAL) {
3449
        flush_on_destroy_ = true;
3450
      }
L
Lei Jin 已提交
3451
      PERF_TIMER_STOP(write_pre_and_post_process_time);
3452

3453
      uint64_t log_size = 0;
L
Lei Jin 已提交
3454
      if (!write_options.disableWAL) {
3455
        PERF_TIMER_GUARD(write_wal_time);
3456 3457
        Slice log_entry = WriteBatchInternal::Contents(updates);
        status = log_->AddRecord(log_entry);
I
Igor Canadi 已提交
3458 3459
        total_log_size_ += log_entry.size();
        alive_log_files_.back().AddSize(log_entry.size());
3460
        log_empty_ = false;
3461
        log_size = log_entry.size();
L
Lei Jin 已提交
3462
        RecordTick(stats_, WAL_FILE_BYTES, log_size);
L
Lei Jin 已提交
3463
        if (status.ok() && write_options.sync) {
I
Igor Canadi 已提交
3464
          RecordTick(stats_, WAL_FILE_SYNCED);
3465
          StopWatch sw(env_, stats_, WAL_FILE_SYNC_MICROS);
3466
          if (db_options_.use_fsync) {
3467
            status = log_->file()->Fsync();
3468
          } else {
3469
            status = log_->file()->Sync();
3470
          }
3471
          if (status.ok() && !log_dir_synced_) {
3472 3473 3474 3475 3476
            // We only sync WAL directory the first time WAL syncing is
            // requested, so that in case users never turn on WAL sync,
            // we can avoid the disk I/O in the write code path.
            status = directories_.GetWalDir()->Fsync();
          }
3477
          log_dir_synced_ = true;
H
heyongqiang 已提交
3478
        }
3479 3480
      }
      if (status.ok()) {
3481
        PERF_TIMER_GUARD(write_memtable_time);
Y
Yueh-Hsuan Chiang 已提交
3482

3483
        status = WriteBatchInternal::InsertInto(
I
Igor Canadi 已提交
3484
            updates, column_family_memtables_.get(),
L
Lei Jin 已提交
3485
            write_options.ignore_missing_column_families, 0, this, false);
Y
Yueh-Hsuan Chiang 已提交
3486 3487 3488 3489 3490 3491 3492 3493
        // A non-OK status here indicates iteration failure (either in-memory
        // writebatch corruption (very bad), or the client specified invalid
        // column family).  This will later on trigger bg_error_.
        //
        // Note that existing logic was not sound. Any partial failure writing
        // into the memtable would result in a state that some write ops might
        // have succeeded in memtable but Status reports error for all writes.

L
Lei Jin 已提交
3494
        SetTickerCount(stats_, SEQUENCE_NUMBER, last_sequence);
3495
      }
L
Lei Jin 已提交
3496
      PERF_TIMER_START(write_pre_and_post_process_time);
3497 3498 3499
      if (updates == &tmp_batch_) {
        tmp_batch_.Clear();
      }
3500
      mutex_.Lock();
A
agiardullo 已提交
3501

3502 3503 3504
      // internal stats
      default_cf_internal_stats_->AddDBStats(
          InternalStats::BYTES_WRITTEN, batch_size);
S
sdong 已提交
3505 3506
      default_cf_internal_stats_->AddDBStats(InternalStats::NUMBER_KEYS_WRITTEN,
                                             my_batch_count);
L
Lei Jin 已提交
3507
      if (!write_options.disableWAL) {
3508 3509 3510 3511 3512
        default_cf_internal_stats_->AddDBStats(
            InternalStats::WAL_FILE_SYNCED, 1);
        default_cf_internal_stats_->AddDBStats(
            InternalStats::WAL_FILE_BYTES, log_size);
      }
3513
      if (status.ok()) {
3514
        versions_->SetLastSequence(last_sequence);
3515
      }
A
agiardullo 已提交
3516 3517 3518
  } else {
    // Operation failed.  Make sure sure mutex is held for cleanup code below.
    mutex_.Lock();
J
jorlow@chromium.org 已提交
3519
    }
A
agiardullo 已提交
3520 3521 3522

    if (db_options_.paranoid_checks && !status.ok() && !status.IsTimedOut() &&
        !status.IsBusy() && bg_error_.ok()) {
I
Igor Canadi 已提交
3523 3524
    bg_error_ = status; // stop compaction & fail any further writes
  }
3525

A
agiardullo 已提交
3526
  mutex_.AssertHeld();
I
Igor Canadi 已提交
3527
  write_thread_.ExitWriteThread(&w, last_writer, status);
3528 3529 3530 3531

  if (context.schedule_bg_work_) {
    MaybeScheduleFlushOrCompaction();
  }
I
Igor Canadi 已提交
3532
  mutex_.Unlock();
3533

3534
  if (status.IsTimedOut()) {
L
Lei Jin 已提交
3535
    RecordTick(stats_, WRITE_TIMEDOUT);
3536 3537
  }

J
jorlow@chromium.org 已提交
3538 3539 3540
  return status;
}

3541 3542
// REQUIRES: mutex_ is held
// REQUIRES: this thread is currently at the front of the writer queue
I
Igor Canadi 已提交
3543
Status DBImpl::DelayWrite(uint64_t expiration_time) {
S
sdong 已提交
3544 3545
  uint64_t time_delayed = 0;
  bool delayed = false;
3546
  bool timed_out = false;
S
sdong 已提交
3547 3548 3549 3550 3551 3552 3553 3554
  {
    StopWatch sw(env_, stats_, WRITE_STALL, &time_delayed);
    bool has_timeout = (expiration_time > 0);
    auto delay = write_controller_.GetDelay();
    if (write_controller_.IsStopped() == false && delay > 0) {
      mutex_.Unlock();
      delayed = true;
      // hopefully we don't have to sleep more than 2 billion microseconds
S
sdong 已提交
3555
      TEST_SYNC_POINT("DBImpl::DelayWrite:Sleep");
S
sdong 已提交
3556 3557 3558
      env_->SleepForMicroseconds(static_cast<int>(delay));
      mutex_.Lock();
    }
3559

S
sdong 已提交
3560 3561 3562
    while (bg_error_.ok() && write_controller_.IsStopped()) {
      delayed = true;
      if (has_timeout) {
S
sdong 已提交
3563
        TEST_SYNC_POINT("DBImpl::DelayWrite:TimedWait");
S
sdong 已提交
3564 3565
        bg_cv_.TimedWait(expiration_time);
        if (env_->NowMicros() > expiration_time) {
3566 3567
          timed_out = true;
          break;
S
sdong 已提交
3568 3569 3570
        }
      } else {
        bg_cv_.Wait();
3571 3572 3573
      }
    }
  }
S
sdong 已提交
3574 3575 3576 3577 3578
  if (delayed) {
    default_cf_internal_stats_->AddDBStats(InternalStats::WRITE_STALL_MICROS,
                                           time_delayed);
    RecordTick(stats_, STALL_MICROS, time_delayed);
  }
I
Igor Canadi 已提交
3579

3580 3581 3582 3583
  if (timed_out) {
    return Status::TimedOut();
  }

I
Igor Canadi 已提交
3584
  return bg_error_;
3585 3586
}

I
Igor Canadi 已提交
3587 3588 3589 3590
Status DBImpl::ScheduleFlushes(WriteContext* context) {
  ColumnFamilyData* cfd;
  while ((cfd = flush_scheduler_.GetNextColumnFamily()) != nullptr) {
    auto status = SetNewMemtableAndNewLogFile(cfd, context);
3591 3592
    SchedulePendingFlush(cfd);
    context->schedule_bg_work_ = true;
I
Igor Canadi 已提交
3593 3594 3595 3596 3597
    if (cfd->Unref()) {
      delete cfd;
    }
    if (!status.ok()) {
      return status;
S
Stanislau Hlebik 已提交
3598 3599
    }
  }
I
Igor Canadi 已提交
3600
  return Status::OK();
S
Stanislau Hlebik 已提交
3601 3602 3603 3604 3605 3606 3607 3608 3609 3610 3611 3612 3613
}

// REQUIRES: mutex_ is held
// REQUIRES: this thread is currently at the front of the writer queue
Status DBImpl::SetNewMemtableAndNewLogFile(ColumnFamilyData* cfd,
                                           WriteContext* context) {
  mutex_.AssertHeld();
  unique_ptr<WritableFile> lfile;
  log::Writer* new_log = nullptr;
  MemTable* new_mem = nullptr;

  // Attempt to switch to a new memtable and trigger flush of old.
  // Do this without holding the dbmutex lock.
3614
  assert(versions_->prev_log_number() == 0);
S
Stanislau Hlebik 已提交
3615 3616 3617 3618
  bool creating_new_log = !log_empty_;
  uint64_t new_log_number =
      creating_new_log ? versions_->NewFileNumber() : logfile_number_;
  SuperVersion* new_superversion = nullptr;
3619
  const MutableCFOptions mutable_cf_options = *cfd->GetLatestMutableCFOptions();
S
Stanislau Hlebik 已提交
3620 3621 3622 3623
  mutex_.Unlock();
  Status s;
  {
    if (creating_new_log) {
3624
      s = env_->NewWritableFile(
3625
          LogFileName(db_options_.wal_dir, new_log_number), &lfile,
3626
          env_->OptimizeForLogWrite(env_options_, db_options_));
S
Stanislau Hlebik 已提交
3627 3628 3629
      if (s.ok()) {
        // Our final size should be less than write_buffer_size
        // (compression, etc) but err on the side of caution.
L
Lei Jin 已提交
3630 3631
        lfile->SetPreallocationBlockSize(
            1.1 * mutable_cf_options.write_buffer_size);
S
Stanislau Hlebik 已提交
3632
        new_log = new log::Writer(std::move(lfile));
3633
        log_dir_synced_ = false;
I
Igor Canadi 已提交
3634
      }
S
Stanislau Hlebik 已提交
3635 3636 3637
    }

    if (s.ok()) {
A
agiardullo 已提交
3638 3639
      SequenceNumber seq = versions_->LastSequence();
      new_mem = cfd->ConstructNewMemtable(mutable_cf_options, seq);
S
Stanislau Hlebik 已提交
3640 3641 3642
      new_superversion = new SuperVersion();
    }
  }
3643
  Log(InfoLogLevel::DEBUG_LEVEL, db_options_.info_log,
I
Igor Canadi 已提交
3644 3645
      "[%s] New memtable created with log file: #%" PRIu64 "\n",
      cfd->GetName().c_str(), new_log_number);
S
Stanislau Hlebik 已提交
3646 3647 3648 3649 3650 3651 3652 3653 3654 3655 3656
  mutex_.Lock();
  if (!s.ok()) {
    // how do we fail if we're not creating new log?
    assert(creating_new_log);
    assert(!new_mem);
    assert(!new_log);
    return s;
  }
  if (creating_new_log) {
    logfile_number_ = new_log_number;
    assert(new_log != nullptr);
3657
    logs_to_free_.push_back(log_.release());
S
Stanislau Hlebik 已提交
3658 3659 3660
    log_.reset(new_log);
    log_empty_ = true;
    alive_log_files_.push_back(LogFileNumberSize(logfile_number_));
I
Igor Canadi 已提交
3661
    for (auto loop_cfd : *versions_->GetColumnFamilySet()) {
S
Stanislau Hlebik 已提交
3662 3663 3664 3665
      // all this is just optimization to delete logs that
      // are no longer needed -- if CF is empty, that means it
      // doesn't need that particular log to stay alive, so we just
      // advance the log number. no need to persist this in the manifest
I
Igor Canadi 已提交
3666
      if (loop_cfd->mem()->GetFirstSequenceNumber() == 0 &&
3667
          loop_cfd->imm()->NumNotFlushed() == 0) {
I
Igor Canadi 已提交
3668
        loop_cfd->SetLogNumber(logfile_number_);
3669
      }
3670 3671
    }
  }
S
Stanislau Hlebik 已提交
3672
  cfd->mem()->SetNextLogNumber(logfile_number_);
3673
  cfd->imm()->Add(cfd->mem(), &context->memtables_to_free_);
S
Stanislau Hlebik 已提交
3674 3675 3676
  new_mem->Ref();
  cfd->SetMemtable(new_mem);
  context->superversions_to_free_.push_back(
3677
      InstallSuperVersion(cfd, new_superversion, mutable_cf_options, true));
3678 3679 3680
  return s;
}

I
Igor Canadi 已提交
3681
#ifndef ROCKSDB_LITE
I
Igor Canadi 已提交
3682 3683 3684 3685 3686
Status DBImpl::GetPropertiesOfAllTables(ColumnFamilyHandle* column_family,
                                        TablePropertiesCollection* props) {
  auto cfh = reinterpret_cast<ColumnFamilyHandleImpl*>(column_family);
  auto cfd = cfh->cfd();

3687 3688
  // Increment the ref count
  mutex_.Lock();
I
Igor Canadi 已提交
3689
  auto version = cfd->current();
3690 3691 3692 3693 3694 3695 3696 3697 3698 3699 3700 3701
  version->Ref();
  mutex_.Unlock();

  auto s = version->GetPropertiesOfAllTables(props);

  // Decrement the ref count
  mutex_.Lock();
  version->Unref();
  mutex_.Unlock();

  return s;
}
I
Igor Canadi 已提交
3702
#endif  // ROCKSDB_LITE
3703

I
Igor Canadi 已提交
3704 3705 3706 3707
const std::string& DBImpl::GetName() const {
  return dbname_;
}

3708 3709 3710 3711
Env* DBImpl::GetEnv() const {
  return env_;
}

3712 3713
const Options& DBImpl::GetOptions(ColumnFamilyHandle* column_family) const {
  auto cfh = reinterpret_cast<ColumnFamilyHandleImpl*>(column_family);
I
Igor Canadi 已提交
3714
  return *cfh->cfd()->options();
I
Igor Canadi 已提交
3715 3716
}

3717 3718
const DBOptions& DBImpl::GetDBOptions() const { return db_options_; }

3719
bool DBImpl::GetProperty(ColumnFamilyHandle* column_family,
3720
                         const Slice& property, std::string* value) {
3721 3722
  bool is_int_property = false;
  bool need_out_of_mutex = false;
3723 3724 3725
  DBPropertyType property_type =
      GetPropertyType(property, &is_int_property, &need_out_of_mutex);

3726
  value->clear();
3727 3728 3729 3730 3731
  if (is_int_property) {
    uint64_t int_value;
    bool ret_value = GetIntPropertyInternal(column_family, property_type,
                                            need_out_of_mutex, &int_value);
    if (ret_value) {
3732
      *value = ToString(int_value);
3733 3734 3735 3736 3737
    }
    return ret_value;
  } else {
    auto cfh = reinterpret_cast<ColumnFamilyHandleImpl*>(column_family);
    auto cfd = cfh->cfd();
3738
    InstrumentedMutexLock l(&mutex_);
3739 3740 3741
    return cfd->internal_stats()->GetStringProperty(property_type, property,
                                                    value);
  }
J
jorlow@chromium.org 已提交
3742 3743
}

3744 3745
bool DBImpl::GetIntProperty(ColumnFamilyHandle* column_family,
                            const Slice& property, uint64_t* value) {
3746 3747
  bool is_int_property = false;
  bool need_out_of_mutex = false;
3748 3749 3750 3751 3752 3753 3754 3755 3756 3757 3758 3759
  DBPropertyType property_type =
      GetPropertyType(property, &is_int_property, &need_out_of_mutex);
  if (!is_int_property) {
    return false;
  }
  return GetIntPropertyInternal(column_family, property_type, need_out_of_mutex,
                                value);
}

bool DBImpl::GetIntPropertyInternal(ColumnFamilyHandle* column_family,
                                    DBPropertyType property_type,
                                    bool need_out_of_mutex, uint64_t* value) {
3760 3761
  auto cfh = reinterpret_cast<ColumnFamilyHandleImpl*>(column_family);
  auto cfd = cfh->cfd();
3762 3763

  if (!need_out_of_mutex) {
3764
    InstrumentedMutexLock l(&mutex_);
3765
    return cfd->internal_stats()->GetIntProperty(property_type, value, this);
3766 3767 3768 3769 3770 3771 3772 3773 3774 3775 3776 3777 3778 3779
  } else {
    SuperVersion* sv = GetAndRefSuperVersion(cfd);

    bool ret = cfd->internal_stats()->GetIntPropertyOutOfMutex(
        property_type, sv->current, value);

    ReturnAndCleanupSuperVersion(cfd, sv);

    return ret;
  }
}

SuperVersion* DBImpl::GetAndRefSuperVersion(ColumnFamilyData* cfd) {
  // TODO(ljin): consider using GetReferencedSuperVersion() directly
I
Igor Canadi 已提交
3780
  return cfd->GetThreadLocalSuperVersion(&mutex_);
3781 3782
}

A
agiardullo 已提交
3783 3784 3785 3786 3787 3788 3789 3790 3791 3792 3793 3794
// REQUIRED: this function should only be called on the write thread or if the
// mutex is held.
SuperVersion* DBImpl::GetAndRefSuperVersion(uint32_t column_family_id) {
  auto column_family_set = versions_->GetColumnFamilySet();
  auto cfd = column_family_set->GetColumnFamily(column_family_id);
  if (!cfd) {
    return nullptr;
  }

  return GetAndRefSuperVersion(cfd);
}

3795 3796
void DBImpl::ReturnAndCleanupSuperVersion(ColumnFamilyData* cfd,
                                          SuperVersion* sv) {
I
Igor Canadi 已提交
3797
  bool unref_sv = !cfd->ReturnThreadLocalSuperVersion(sv);
3798 3799 3800 3801 3802

  if (unref_sv) {
    // Release SuperVersion
    if (sv->Unref()) {
      {
3803
        InstrumentedMutexLock l(&mutex_);
3804 3805 3806 3807 3808 3809 3810
        sv->Cleanup();
      }
      delete sv;
      RecordTick(stats_, NUMBER_SUPERVERSION_CLEANUPS);
    }
    RecordTick(stats_, NUMBER_SUPERVERSION_RELEASES);
  }
3811 3812
}

A
agiardullo 已提交
3813 3814 3815 3816 3817 3818 3819 3820 3821 3822 3823 3824 3825 3826 3827 3828 3829 3830 3831 3832 3833 3834 3835 3836
// REQUIRED: this function should only be called on the write thread.
void DBImpl::ReturnAndCleanupSuperVersion(uint32_t column_family_id,
                                          SuperVersion* sv) {
  auto column_family_set = versions_->GetColumnFamilySet();
  auto cfd = column_family_set->GetColumnFamily(column_family_id);

  // If SuperVersion is held, and we successfully fetched a cfd using
  // GetAndRefSuperVersion(), it must still exist.
  assert(cfd != nullptr);
  ReturnAndCleanupSuperVersion(cfd, sv);
}

// REQUIRED: this function should only be called on the write thread or if the
// mutex is held.
ColumnFamilyHandle* DBImpl::GetColumnFamilyHandle(uint32_t column_family_id) {
  ColumnFamilyMemTables* cf_memtables = column_family_memtables_.get();

  if (!cf_memtables->Seek(column_family_id)) {
    return nullptr;
  }

  return cf_memtables->GetColumnFamilyHandle();
}

3837
void DBImpl::GetApproximateSizes(ColumnFamilyHandle* column_family,
3838
                                 const Range* range, int n, uint64_t* sizes) {
J
jorlow@chromium.org 已提交
3839
  Version* v;
3840 3841
  auto cfh = reinterpret_cast<ColumnFamilyHandleImpl*>(column_family);
  auto cfd = cfh->cfd();
J
jorlow@chromium.org 已提交
3842
  {
3843
    InstrumentedMutexLock l(&mutex_);
3844
    v = cfd->current();
3845
    v->Ref();
J
jorlow@chromium.org 已提交
3846 3847 3848 3849
  }

  for (int i = 0; i < n; i++) {
    // Convert user_key into a corresponding internal key.
3850 3851 3852
    InternalKey k1(range[i].start, kMaxSequenceNumber, kValueTypeForSeek);
    InternalKey k2(range[i].limit, kMaxSequenceNumber, kValueTypeForSeek);
    sizes[i] = versions_->ApproximateSize(v, k1.Encode(), k2.Encode());
J
jorlow@chromium.org 已提交
3853 3854 3855
  }

  {
3856
    InstrumentedMutexLock l(&mutex_);
J
jorlow@chromium.org 已提交
3857 3858 3859 3860
    v->Unref();
  }
}

I
Igor Canadi 已提交
3861 3862 3863 3864 3865 3866 3867 3868 3869 3870 3871 3872 3873 3874 3875 3876
std::list<uint64_t>::iterator
DBImpl::CaptureCurrentFileNumberInPendingOutputs() {
  // We need to remember the iterator of our insert, because after the
  // background job is done, we need to remove that element from
  // pending_outputs_.
  pending_outputs_.push_back(versions_->current_next_file_number());
  auto pending_outputs_inserted_elem = pending_outputs_.end();
  --pending_outputs_inserted_elem;
  return pending_outputs_inserted_elem;
}

void DBImpl::ReleaseFileNumberFromPendingOutputs(
    std::list<uint64_t>::iterator v) {
  pending_outputs_.erase(v);
}

I
Igor Canadi 已提交
3877 3878 3879 3880 3881
#ifndef ROCKSDB_LITE
Status DBImpl::GetUpdatesSince(
    SequenceNumber seq, unique_ptr<TransactionLogIterator>* iter,
    const TransactionLogIterator::ReadOptions& read_options) {

L
Lei Jin 已提交
3882
  RecordTick(stats_, GET_UPDATES_SINCE_CALLS);
I
Igor Canadi 已提交
3883 3884 3885
  if (seq > versions_->LastSequence()) {
    return Status::NotFound("Requested sequence not yet written in the db");
  }
I
Igor Canadi 已提交
3886
  return wal_manager_.GetUpdatesSince(seq, iter, read_options, versions_.get());
I
Igor Canadi 已提交
3887 3888
}

3889 3890 3891
Status DBImpl::DeleteFile(std::string name) {
  uint64_t number;
  FileType type;
3892 3893 3894
  WalFileType log_type;
  if (!ParseFileName(name, &number, &type, &log_type) ||
      (type != kTableFile && type != kLogFile)) {
3895 3896
    Log(InfoLogLevel::ERROR_LEVEL, db_options_.info_log,
        "DeleteFile %s failed.\n", name.c_str());
3897 3898 3899
    return Status::InvalidArgument("Invalid file name");
  }

3900 3901 3902 3903
  Status status;
  if (type == kLogFile) {
    // Only allow deleting archived log files
    if (log_type != kArchivedLogFile) {
3904 3905
      Log(InfoLogLevel::ERROR_LEVEL, db_options_.info_log,
          "DeleteFile %s failed - not archived log.\n",
3906
          name.c_str());
3907 3908
      return Status::NotSupported("Delete only supported for archived logs");
    }
3909
    status = env_->DeleteFile(db_options_.wal_dir + "/" + name.c_str());
3910
    if (!status.ok()) {
3911 3912
      Log(InfoLogLevel::ERROR_LEVEL, db_options_.info_log,
          "DeleteFile %s failed -- %s.\n",
3913
          name.c_str(), status.ToString().c_str());
3914 3915 3916 3917
    }
    return status;
  }

3918
  int level;
I
Igor Canadi 已提交
3919
  FileMetaData* metadata;
3920
  ColumnFamilyData* cfd;
3921
  VersionEdit edit;
3922
  JobContext job_context(next_job_id_.fetch_add(1), true);
D
Dhruba Borthakur 已提交
3923
  {
3924
    InstrumentedMutexLock l(&mutex_);
3925
    status = versions_->GetMetadataForFile(number, &level, &metadata, &cfd);
D
Dhruba Borthakur 已提交
3926
    if (!status.ok()) {
3927 3928
      Log(InfoLogLevel::WARN_LEVEL, db_options_.info_log,
          "DeleteFile %s failed. File not found\n", name.c_str());
I
Igor Canadi 已提交
3929
      job_context.Clean();
D
Dhruba Borthakur 已提交
3930 3931
      return Status::InvalidArgument("File not found");
    }
3932
    assert(level < cfd->NumberLevels());
3933

D
Dhruba Borthakur 已提交
3934
    // If the file is being compacted no need to delete.
3935
    if (metadata->being_compacted) {
3936
      Log(InfoLogLevel::INFO_LEVEL, db_options_.info_log,
3937
          "DeleteFile %s Skipped. File about to be compacted\n", name.c_str());
I
Igor Canadi 已提交
3938
      job_context.Clean();
D
Dhruba Borthakur 已提交
3939
      return Status::OK();
3940 3941
    }

D
Dhruba Borthakur 已提交
3942 3943 3944
    // Only the files in the last level can be deleted externally.
    // This is to make sure that any deletion tombstones are not
    // lost. Check that the level passed is the last level.
S
sdong 已提交
3945
    auto* vstoreage = cfd->current()->storage_info();
I
Igor Canadi 已提交
3946
    for (int i = level + 1; i < cfd->NumberLevels(); i++) {
S
sdong 已提交
3947
      if (vstoreage->NumLevelFiles(i) != 0) {
3948
        Log(InfoLogLevel::WARN_LEVEL, db_options_.info_log,
3949
            "DeleteFile %s FAILED. File not in last level\n", name.c_str());
I
Igor Canadi 已提交
3950
        job_context.Clean();
D
Dhruba Borthakur 已提交
3951 3952 3953
        return Status::InvalidArgument("File not in last level");
      }
    }
3954
    // if level == 0, it has to be the oldest file
S
sdong 已提交
3955 3956
    if (level == 0 &&
        vstoreage->LevelFiles(0).back()->fd.GetNumber() != number) {
3957 3958
      Log(InfoLogLevel::WARN_LEVEL, db_options_.info_log,
          "DeleteFile %s failed ---"
L
Lei Jin 已提交
3959
          " target file in level 0 must be the oldest.", name.c_str());
I
Igor Canadi 已提交
3960
      job_context.Clean();
3961 3962 3963
      return Status::InvalidArgument("File in level 0, but not oldest");
    }
    edit.SetColumnFamily(cfd->GetID());
D
Dhruba Borthakur 已提交
3964
    edit.DeleteFile(level, number);
3965
    status = versions_->LogAndApply(cfd, *cfd->GetLatestMutableCFOptions(),
3966
                                    &edit, &mutex_, directories_.GetDbDir());
I
Igor Canadi 已提交
3967
    if (status.ok()) {
I
Igor Canadi 已提交
3968 3969
      InstallSuperVersionBackground(cfd, &job_context,
                                    *cfd->GetLatestMutableCFOptions());
I
Igor Canadi 已提交
3970
    }
I
Igor Canadi 已提交
3971 3972
    FindObsoleteFiles(&job_context, false);
  }  // lock released here
3973

3974
  LogFlush(db_options_.info_log);
I
Igor Canadi 已提交
3975
  // remove files outside the db-lock
I
Igor Canadi 已提交
3976
  if (job_context.HaveSomethingToDelete()) {
3977
    // Call PurgeObsoleteFiles() without holding mutex.
I
Igor Canadi 已提交
3978
    PurgeObsoleteFiles(job_context);
3979
  }
I
Igor Canadi 已提交
3980
  job_context.Clean();
3981 3982 3983
  return status;
}

3984
void DBImpl::GetLiveFilesMetaData(std::vector<LiveFileMetaData>* metadata) {
3985
  InstrumentedMutexLock l(&mutex_);
3986
  versions_->GetLiveFilesMetaData(metadata);
3987
}
3988 3989 3990 3991 3992 3993 3994 3995 3996 3997 3998

void DBImpl::GetColumnFamilyMetaData(
    ColumnFamilyHandle* column_family,
    ColumnFamilyMetaData* cf_meta) {
  assert(column_family);
  auto* cfd = reinterpret_cast<ColumnFamilyHandleImpl*>(column_family)->cfd();
  auto* sv = GetAndRefSuperVersion(cfd);
  sv->current->GetColumnFamilyMetaData(cf_meta);
  ReturnAndCleanupSuperVersion(cfd, sv);
}

I
Igor Canadi 已提交
3999
#endif  // ROCKSDB_LITE
4000

I
Igor Canadi 已提交
4001 4002 4003 4004 4005 4006 4007
Status DBImpl::CheckConsistency() {
  mutex_.AssertHeld();
  std::vector<LiveFileMetaData> metadata;
  versions_->GetLiveFilesMetaData(&metadata);

  std::string corruption_messages;
  for (const auto& md : metadata) {
4008 4009
    // md.name has a leading "/".
    std::string file_path = md.db_path + md.name;
4010

I
Igor Canadi 已提交
4011 4012 4013 4014 4015 4016
    uint64_t fsize = 0;
    Status s = env_->GetFileSize(file_path, &fsize);
    if (!s.ok()) {
      corruption_messages +=
          "Can't access " + md.name + ": " + s.ToString() + "\n";
    } else if (fsize != md.size) {
4017
      corruption_messages += "Sst file size mismatch: " + file_path +
I
Igor Canadi 已提交
4018
                             ". Size recorded in manifest " +
4019 4020
                             ToString(md.size) + ", actual size " +
                             ToString(fsize) + "\n";
I
Igor Canadi 已提交
4021 4022 4023 4024 4025 4026 4027 4028 4029
    }
  }
  if (corruption_messages.size() == 0) {
    return Status::OK();
  } else {
    return Status::Corruption(corruption_messages);
  }
}

4030
Status DBImpl::GetDbIdentity(std::string& identity) const {
4031 4032 4033 4034 4035 4036 4037 4038 4039 4040 4041 4042 4043 4044
  std::string idfilename = IdentityFileName(dbname_);
  unique_ptr<SequentialFile> idfile;
  const EnvOptions soptions;
  Status s = env_->NewSequentialFile(idfilename, &idfile, soptions);
  if (!s.ok()) {
    return s;
  }
  uint64_t file_size;
  s = env_->GetFileSize(idfilename, &file_size);
  if (!s.ok()) {
    return s;
  }
  char buffer[file_size];
  Slice id;
4045
  s = idfile->Read(static_cast<size_t>(file_size), &id, buffer);
4046 4047 4048 4049 4050 4051 4052 4053 4054 4055 4056
  if (!s.ok()) {
    return s;
  }
  identity.assign(id.ToString());
  // If last character is '\n' remove it from identity
  if (identity.size() > 0 && identity.back() == '\n') {
    identity.pop_back();
  }
  return s;
}

J
jorlow@chromium.org 已提交
4057 4058
// Default implementations of convenience methods that subclasses of DB
// can call if they wish
4059
Status DB::Put(const WriteOptions& opt, ColumnFamilyHandle* column_family,
4060
               const Slice& key, const Slice& value) {
4061 4062 4063 4064
  // Pre-allocate size of write batch conservatively.
  // 8 bytes are taken by header, 4 bytes for count, 1 byte for type,
  // and we allocate 11 extra bytes for key length, as well as value length.
  WriteBatch batch(key.size() + value.size() + 24);
4065
  batch.Put(column_family, key, value);
J
jorlow@chromium.org 已提交
4066 4067 4068
  return Write(opt, &batch);
}

4069 4070
Status DB::Delete(const WriteOptions& opt, ColumnFamilyHandle* column_family,
                  const Slice& key) {
J
jorlow@chromium.org 已提交
4071
  WriteBatch batch;
4072
  batch.Delete(column_family, key);
J
jorlow@chromium.org 已提交
4073 4074 4075
  return Write(opt, &batch);
}

4076 4077
Status DB::Merge(const WriteOptions& opt, ColumnFamilyHandle* column_family,
                 const Slice& key, const Slice& value) {
4078
  WriteBatch batch;
4079
  batch.Merge(column_family, key, value);
4080 4081 4082
  return Write(opt, &batch);
}

4083
// Default implementation -- returns not supported status
L
Lei Jin 已提交
4084
Status DB::CreateColumnFamily(const ColumnFamilyOptions& cf_options,
4085
                              const std::string& column_family_name,
4086
                              ColumnFamilyHandle** handle) {
4087
  return Status::NotSupported("");
4088
}
4089
Status DB::DropColumnFamily(ColumnFamilyHandle* column_family) {
4090
  return Status::NotSupported("");
4091 4092
}

J
jorlow@chromium.org 已提交
4093 4094
DB::~DB() { }

J
Jim Paton 已提交
4095
Status DB::Open(const Options& options, const std::string& dbname, DB** dbptr) {
4096 4097 4098 4099
  DBOptions db_options(options);
  ColumnFamilyOptions cf_options(options);
  std::vector<ColumnFamilyDescriptor> column_families;
  column_families.push_back(
4100
      ColumnFamilyDescriptor(kDefaultColumnFamilyName, cf_options));
4101
  std::vector<ColumnFamilyHandle*> handles;
4102
  Status s = DB::Open(db_options, dbname, column_families, &handles, dbptr);
4103 4104 4105 4106 4107 4108 4109
  if (s.ok()) {
    assert(handles.size() == 1);
    // i can delete the handle since DBImpl is always holding a reference to
    // default column family
    delete handles[0];
  }
  return s;
4110 4111
}

4112 4113
Status DB::Open(const DBOptions& db_options, const std::string& dbname,
                const std::vector<ColumnFamilyDescriptor>& column_families,
4114
                std::vector<ColumnFamilyHandle*>* handles, DB** dbptr) {
4115
  Status s = SanitizeOptionsByTable(db_options, column_families);
4116 4117 4118
  if (!s.ok()) {
    return s;
  }
4119

4120
  if (db_options.db_paths.size() > 1) {
4121
    for (auto& cfd : column_families) {
4122 4123
      if ((cfd.options.compaction_style != kCompactionStyleUniversal) &&
          (cfd.options.compaction_style != kCompactionStyleLevel)) {
4124 4125
        return Status::NotSupported(
            "More than one DB paths are only supported in "
4126
            "universal and level compaction styles. ");
4127 4128 4129 4130 4131
      }
    }

    if (db_options.db_paths.size() > 4) {
      return Status::NotSupported(
4132
          "More than four DB paths are not supported yet. ");
4133
    }
4134 4135
  }

4136
  *dbptr = nullptr;
4137
  handles->clear();
J
jorlow@chromium.org 已提交
4138

I
Igor Canadi 已提交
4139 4140 4141 4142
  size_t max_write_buffer_size = 0;
  for (auto cf : column_families) {
    max_write_buffer_size =
        std::max(max_write_buffer_size, cf.options.write_buffer_size);
4143
  }
4144

I
Igor Canadi 已提交
4145
  DBImpl* impl = new DBImpl(db_options, dbname);
4146
  s = impl->env_->CreateDirIfMissing(impl->db_options_.wal_dir);
4147
  if (s.ok()) {
4148
    for (auto db_path : impl->db_options_.db_paths) {
4149
      s = impl->env_->CreateDirIfMissing(db_path.path);
4150 4151 4152 4153 4154 4155
      if (!s.ok()) {
        break;
      }
    }
  }

4156 4157 4158 4159 4160 4161
  if (!s.ok()) {
    delete impl;
    return s;
  }

  s = impl->CreateArchivalDirectory();
4162 4163 4164 4165
  if (!s.ok()) {
    delete impl;
    return s;
  }
J
jorlow@chromium.org 已提交
4166
  impl->mutex_.Lock();
4167 4168
  // Handles create_if_missing, error_if_exists
  s = impl->Recover(column_families);
J
jorlow@chromium.org 已提交
4169
  if (s.ok()) {
4170
    uint64_t new_log_number = impl->versions_->NewFileNumber();
4171
    unique_ptr<WritableFile> lfile;
4172
    EnvOptions soptions(db_options);
4173 4174
    s = impl->db_options_.env->NewWritableFile(
        LogFileName(impl->db_options_.wal_dir, new_log_number), &lfile,
4175 4176
        impl->db_options_.env->OptimizeForLogWrite(soptions,
                                                   impl->db_options_));
J
jorlow@chromium.org 已提交
4177
    if (s.ok()) {
4178
      lfile->SetPreallocationBlockSize(1.1 * max_write_buffer_size);
4179
      impl->logfile_number_ = new_log_number;
4180
      impl->log_.reset(new log::Writer(std::move(lfile)));
I
Igor Canadi 已提交
4181

4182 4183
      // set column family handles
      for (auto cf : column_families) {
I
Igor Canadi 已提交
4184 4185
        auto cfd =
            impl->versions_->GetColumnFamilySet()->GetColumnFamily(cf.name);
I
Igor Canadi 已提交
4186 4187 4188
        if (cfd != nullptr) {
          handles->push_back(
              new ColumnFamilyHandleImpl(cfd, impl, &impl->mutex_));
4189
          impl->NewThreadStatusCfInfo(cfd);
I
Igor Canadi 已提交
4190 4191 4192 4193 4194 4195 4196 4197 4198 4199 4200 4201 4202 4203 4204 4205
        } else {
          if (db_options.create_missing_column_families) {
            // missing column family, create it
            ColumnFamilyHandle* handle;
            impl->mutex_.Unlock();
            s = impl->CreateColumnFamily(cf.options, cf.name, &handle);
            impl->mutex_.Lock();
            if (s.ok()) {
              handles->push_back(handle);
            } else {
              break;
            }
          } else {
            s = Status::InvalidArgument("Column family not found: ", cf.name);
            break;
          }
I
Igor Canadi 已提交
4206
        }
4207
      }
I
Igor Canadi 已提交
4208 4209
    }
    if (s.ok()) {
4210
      for (auto cfd : *impl->versions_->GetColumnFamilySet()) {
L
Lei Jin 已提交
4211 4212
        delete impl->InstallSuperVersion(
            cfd, nullptr, *cfd->GetLatestMutableCFOptions());
4213
      }
I
Igor Canadi 已提交
4214 4215
      impl->alive_log_files_.push_back(
          DBImpl::LogFileNumberSize(impl->logfile_number_));
J
jorlow@chromium.org 已提交
4216
      impl->DeleteObsoleteFiles();
4217
      s = impl->directories_.GetDbDir()->Fsync();
J
jorlow@chromium.org 已提交
4218 4219
    }
  }
4220

I
Igor Canadi 已提交
4221 4222
  if (s.ok()) {
    for (auto cfd : *impl->versions_->GetColumnFamilySet()) {
S
sdong 已提交
4223
      if (cfd->ioptions()->compaction_style == kCompactionStyleFIFO) {
S
sdong 已提交
4224
        auto* vstorage = cfd->current()->storage_info();
4225
        for (int i = 1; i < vstorage->num_levels(); ++i) {
S
sdong 已提交
4226
          int num_files = vstorage->NumLevelFiles(i);
I
Igor Canadi 已提交
4227
          if (num_files > 0) {
I
Igor Canadi 已提交
4228 4229
            s = Status::InvalidArgument(
                "Not all files are at level 0. Cannot "
S
sdong 已提交
4230
                "open with FIFO compaction style.");
I
Igor Canadi 已提交
4231 4232 4233 4234
            break;
          }
        }
      }
S
sdong 已提交
4235 4236 4237
      if (!cfd->mem()->IsSnapshotSupported()) {
        impl->is_snapshot_supported_ = false;
      }
4238
      if (cfd->ioptions()->merge_operator != nullptr &&
4239 4240 4241 4242 4243
          !cfd->mem()->IsMergeOperatorSupported()) {
        s = Status::InvalidArgument(
            "The memtable of column family %s does not support merge operator "
            "its options.merge_operator is non-null", cfd->GetName().c_str());
      }
I
Igor Canadi 已提交
4244
      if (!s.ok()) {
4245 4246 4247 4248 4249
        break;
      }
    }
  }

4250 4251
  impl->mutex_.Unlock();

J
jorlow@chromium.org 已提交
4252
  if (s.ok()) {
4253
    impl->opened_successfully_ = true;
S
sdong 已提交
4254 4255
    Log(InfoLogLevel::INFO_LEVEL, impl->db_options_.info_log, "DB pointer %p",
        impl);
J
jorlow@chromium.org 已提交
4256 4257
    *dbptr = impl;
  } else {
Y
Yueh-Hsuan Chiang 已提交
4258
    for (auto* h : *handles) {
4259 4260
      delete h;
    }
4261
    handles->clear();
J
jorlow@chromium.org 已提交
4262 4263 4264 4265 4266
    delete impl;
  }
  return s;
}

4267 4268 4269
Status DB::ListColumnFamilies(const DBOptions& db_options,
                              const std::string& name,
                              std::vector<std::string>* column_families) {
I
Igor Canadi 已提交
4270
  return VersionSet::ListColumnFamilies(column_families, name, db_options.env);
4271 4272
}

4273 4274 4275
Snapshot::~Snapshot() {
}

J
jorlow@chromium.org 已提交
4276
Status DestroyDB(const std::string& dbname, const Options& options) {
4277
  const InternalKeyComparator comparator(options.comparator);
4278
  const Options& soptions(SanitizeOptions(dbname, &comparator, options));
4279
  Env* env = soptions.env;
J
jorlow@chromium.org 已提交
4280
  std::vector<std::string> filenames;
4281

J
jorlow@chromium.org 已提交
4282 4283
  // Ignore error in case directory does not exist
  env->GetChildren(dbname, &filenames);
4284

J
jorlow@chromium.org 已提交
4285
  FileLock* lock;
4286 4287
  const std::string lockname = LockFileName(dbname);
  Status result = env->LockFile(lockname, &lock);
J
jorlow@chromium.org 已提交
4288 4289 4290
  if (result.ok()) {
    uint64_t number;
    FileType type;
4291
    InfoLogPrefix info_log_prefix(!options.db_log_dir.empty(), dbname);
D
dgrogan@chromium.org 已提交
4292
    for (size_t i = 0; i < filenames.size(); i++) {
4293
      if (ParseFileName(filenames[i], &number, info_log_prefix.prefix, &type) &&
4294
          type != kDBLockFile) {  // Lock file will be deleted at end
K
Kosie van der Merwe 已提交
4295 4296 4297 4298 4299 4300
        Status del;
        if (type == kMetaDatabase) {
          del = DestroyDB(dbname + "/" + filenames[i], options);
        } else {
          del = env->DeleteFile(dbname + "/" + filenames[i]);
        }
J
jorlow@chromium.org 已提交
4301 4302 4303 4304 4305
        if (result.ok() && !del.ok()) {
          result = del;
        }
      }
    }
4306

4307 4308
    for (auto& db_path : options.db_paths) {
      env->GetChildren(db_path.path, &filenames);
4309 4310 4311
      for (size_t i = 0; i < filenames.size(); i++) {
        if (ParseFileName(filenames[i], &number, &type) &&
            type == kTableFile) {  // Lock file will be deleted at end
4312
          Status del = env->DeleteFile(db_path.path + "/" + filenames[i]);
4313 4314 4315 4316 4317 4318 4319
          if (result.ok() && !del.ok()) {
            result = del;
          }
        }
      }
    }

I
Igor Canadi 已提交
4320 4321 4322 4323 4324 4325 4326 4327 4328 4329 4330 4331 4332 4333 4334 4335 4336 4337
    std::vector<std::string> walDirFiles;
    std::string archivedir = ArchivalDirectory(dbname);
    if (dbname != soptions.wal_dir) {
      env->GetChildren(soptions.wal_dir, &walDirFiles);
      archivedir = ArchivalDirectory(soptions.wal_dir);
    }

    // Delete log files in the WAL dir
    for (const auto& file : walDirFiles) {
      if (ParseFileName(file, &number, &type) && type == kLogFile) {
        Status del = env->DeleteFile(soptions.wal_dir + "/" + file);
        if (result.ok() && !del.ok()) {
          result = del;
        }
      }
    }

    std::vector<std::string> archiveFiles;
4338
    env->GetChildren(archivedir, &archiveFiles);
4339 4340
    // Delete archival files.
    for (size_t i = 0; i < archiveFiles.size(); ++i) {
4341 4342
      if (ParseFileName(archiveFiles[i], &number, &type) &&
          type == kLogFile) {
4343
        Status del = env->DeleteFile(archivedir + "/" + archiveFiles[i]);
4344 4345 4346 4347 4348
        if (result.ok() && !del.ok()) {
          result = del;
        }
      }
    }
4349
    // ignore case where no archival directory is present.
4350
    env->DeleteDir(archivedir);
4351

J
jorlow@chromium.org 已提交
4352
    env->UnlockFile(lock);  // Ignore error since state is already gone
4353
    env->DeleteFile(lockname);
J
jorlow@chromium.org 已提交
4354
    env->DeleteDir(dbname);  // Ignore error in case dir contains other files
4355
    env->DeleteDir(soptions.wal_dir);
J
jorlow@chromium.org 已提交
4356 4357 4358 4359
  }
  return result;
}

Y
Yueh-Hsuan Chiang 已提交
4360
#if ROCKSDB_USING_THREAD_STATUS
4361

Y
Yueh-Hsuan Chiang 已提交
4362 4363
void DBImpl::NewThreadStatusCfInfo(
    ColumnFamilyData* cfd) const {
4364
  if (db_options_.enable_thread_tracking) {
4365
    ThreadStatusUtil::NewColumnFamilyInfo(this, cfd);
4366
  }
Y
Yueh-Hsuan Chiang 已提交
4367 4368 4369 4370
}

void DBImpl::EraseThreadStatusCfInfo(
    ColumnFamilyData* cfd) const {
4371
  if (db_options_.enable_thread_tracking) {
4372
    ThreadStatusUtil::EraseColumnFamilyInfo(cfd);
4373
  }
Y
Yueh-Hsuan Chiang 已提交
4374 4375 4376
}

void DBImpl::EraseThreadStatusDbInfo() const {
4377
  if (db_options_.enable_thread_tracking) {
4378
    ThreadStatusUtil::EraseDatabaseInfo(this);
4379
  }
Y
Yueh-Hsuan Chiang 已提交
4380 4381 4382 4383 4384 4385 4386 4387 4388 4389 4390 4391 4392 4393 4394
}

#else
void DBImpl::NewThreadStatusCfInfo(
    ColumnFamilyData* cfd) const {
}

void DBImpl::EraseThreadStatusCfInfo(
    ColumnFamilyData* cfd) const {
}

void DBImpl::EraseThreadStatusDbInfo() const {
}
#endif  // ROCKSDB_USING_THREAD_STATUS

4395 4396
//
// A global method that can dump out the build version
I
Igor Canadi 已提交
4397
void DumpRocksDBBuildVersion(Logger * log) {
I
Igor Canadi 已提交
4398
#if !defined(IOS_CROSS_COMPILE)
I
Igor Canadi 已提交
4399 4400
  // if we compile with Xcode, we don't run build_detect_vesion, so we don't
  // generate util/build_version.cc
4401
  Warn(log,
4402
      "RocksDB version: %d.%d.%d\n", ROCKSDB_MAJOR, ROCKSDB_MINOR,
I
Igor Canadi 已提交
4403
      ROCKSDB_PATCH);
4404 4405
  Warn(log, "Git sha %s", rocksdb_build_git_sha);
  Warn(log, "Compile date %s", rocksdb_build_compile_date);
I
Igor Canadi 已提交
4406
#endif
4407 4408
}

A
agiardullo 已提交
4409 4410 4411 4412 4413 4414 4415 4416 4417 4418 4419 4420 4421 4422 4423 4424 4425 4426 4427 4428 4429 4430 4431 4432 4433 4434 4435 4436 4437 4438 4439 4440 4441 4442 4443 4444 4445 4446 4447 4448 4449 4450 4451 4452 4453 4454 4455 4456 4457 4458 4459 4460 4461 4462 4463 4464 4465 4466 4467 4468 4469 4470 4471 4472 4473 4474 4475 4476 4477 4478 4479 4480 4481 4482 4483 4484 4485 4486 4487
#ifndef ROCKSDB_LITE
SequenceNumber DBImpl::GetEarliestMemTableSequenceNumber(SuperVersion* sv,
                                                         bool include_history) {
  // Find the earliest sequence number that we know we can rely on reading
  // from the memtable without needing to check sst files.
  SequenceNumber earliest_seq =
      sv->imm->GetEarliestSequenceNumber(include_history);
  if (earliest_seq == kMaxSequenceNumber) {
    earliest_seq = sv->mem->GetEarliestSequenceNumber();
  }
  assert(sv->mem->GetEarliestSequenceNumber() >= earliest_seq);

  return earliest_seq;
}
#endif  // ROCKSDB_LITE

#ifndef ROCKSDB_LITE
Status DBImpl::GetLatestSequenceForKeyFromMemtable(SuperVersion* sv,
                                                   const Slice& key,
                                                   SequenceNumber* seq) {
  Status s;
  std::string value;
  MergeContext merge_context;

  SequenceNumber current_seq = versions_->LastSequence();
  LookupKey lkey(key, current_seq);

  *seq = kMaxSequenceNumber;

  // Check if there is a record for this key in the latest memtable
  sv->mem->Get(lkey, &value, &s, &merge_context, seq);

  if (!(s.ok() || s.IsNotFound() || s.IsMergeInProgress())) {
    // unexpected error reading memtable.
    Log(InfoLogLevel::ERROR_LEVEL, db_options_.info_log,
        "Unexpected status returned from MemTable::Get: %s\n",
        s.ToString().c_str());

    return s;
  }

  if (*seq != kMaxSequenceNumber) {
    // Found a sequence number, no need to check immutable memtables
    return Status::OK();
  }

  // Check if there is a record for this key in the immutable memtables
  sv->imm->Get(lkey, &value, &s, &merge_context, seq);

  if (!(s.ok() || s.IsNotFound() || s.IsMergeInProgress())) {
    // unexpected error reading memtable.
    Log(InfoLogLevel::ERROR_LEVEL, db_options_.info_log,
        "Unexpected status returned from MemTableList::Get: %s\n",
        s.ToString().c_str());

    return s;
  }

  if (*seq != kMaxSequenceNumber) {
    // Found a sequence number, no need to check memtable history
    return Status::OK();
  }

  // Check if there is a record for this key in the immutable memtables
  sv->imm->GetFromHistory(lkey, &value, &s, &merge_context, seq);

  if (!(s.ok() || s.IsNotFound() || s.IsMergeInProgress())) {
    // unexpected error reading memtable.
    Log(InfoLogLevel::ERROR_LEVEL, db_options_.info_log,
        "Unexpected status returned from MemTableList::GetFromHistory: %s\n",
        s.ToString().c_str());

    return s;
  }

  return Status::OK();
}
#endif  // ROCKSDB_LITE

4488
}  // namespace rocksdb