src/leveldb/db/db_impl.cc

   1 // Copyright (c) 2011 The LevelDB Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file. See the AUTHORS file for names of contributors.
   4
   5 #include "db/db_impl.h"
   6
   7 #include <algorithm>
   8 #include <set>
   9 #include <string>
  10 #include <stdint.h>
  11 #include <stdio.h>
  12 #include <vector>
  13 #include "db/builder.h"
  14 #include "db/db_iter.h"
  15 #include "db/dbformat.h"
  16 #include "db/filename.h"
  17 #include "db/log_reader.h"
  18 #include "db/log_writer.h"
  19 #include "db/memtable.h"
  20 #include "db/table_cache.h"
  21 #include "db/version_set.h"
  22 #include "db/write_batch_internal.h"
  23 #include "leveldb/db.h"
  24 #include "leveldb/env.h"
  25 #include "leveldb/status.h"
  26 #include "leveldb/table.h"
  27 #include "leveldb/table_builder.h"
  28 #include "port/port.h"
  29 #include "table/block.h"
  30 #include "table/merger.h"
  31 #include "table/two_level_iterator.h"
  32 #include "util/coding.h"
  33 #include "util/logging.h"
  34 #include "util/mutexlock.h"
  35
  36 namespace leveldb {
  37
  38 const int kNumNonTableCacheFiles = 10;
  39
  40 // Information kept for every waiting writer
  41 struct DBImpl::Writer {
  42   Status status;
  43   WriteBatch* batch;
  44   bool sync;
  45   bool done;
  46   port::CondVar cv;
  47
  48   explicit Writer(port::Mutex* mu) : cv(mu) { }
  49 };
  50
  51 struct DBImpl::CompactionState {
  52   Compaction* const compaction;
  53
  54   // Sequence numbers < smallest_snapshot are not significant since we
  55   // will never have to service a snapshot below smallest_snapshot.
  56   // Therefore if we have seen a sequence number S <= smallest_snapshot,
  57   // we can drop all entries for the same key with sequence numbers < S.
  58   SequenceNumber smallest_snapshot;
  59
  60   // Files produced by compaction
  61   struct Output {
  62     uint64_t number;
  63     uint64_t file_size;
  64     InternalKey smallest, largest;
  65   };
  66   std::vector<Output> outputs;
  67
  68   // State kept for output being generated
  69   WritableFile* outfile;
  70   TableBuilder* builder;
  71
  72   uint64_t total_bytes;
  73
  74   Output* current_output() { return &outputs[outputs.size()-1]; }
  75
  76   explicit CompactionState(Compaction* c)
  77       : compaction(c),
  78         outfile(NULL),
  79         builder(NULL),
  80         total_bytes(0) {
  81   }
  82 };
  83
  84 // Fix user-supplied options to be reasonable
  85 template <class T,class V>
  86 static void ClipToRange(T* ptr, V minvalue, V maxvalue) {
  87   if (static_cast<V>(*ptr) > maxvalue) *ptr = maxvalue;
  88   if (static_cast<V>(*ptr) < minvalue) *ptr = minvalue;
  89 }
  90 Options SanitizeOptions(const std::string& dbname,
  91                         const InternalKeyComparator* icmp,
  92                         const InternalFilterPolicy* ipolicy,
  93                         const Options& src) {
  94   Options result = src;
  95   result.comparator = icmp;
  96   result.filter_policy = (src.filter_policy != NULL) ? ipolicy : NULL;
  97   ClipToRange(&result.max_open_files,    64 + kNumNonTableCacheFiles, 50000);
  98   ClipToRange(&result.write_buffer_size, 64<<10,                      1<<30);
  99   ClipToRange(&result.block_size,        1<<10,                       4<<20);
 100   if (result.info_log == NULL) {
 101     // Open a log file in the same directory as the db
 102     src.env->CreateDir(dbname);  // In case it does not exist
 103     src.env->RenameFile(InfoLogFileName(dbname), OldInfoLogFileName(dbname));
 104     Status s = src.env->NewLogger(InfoLogFileName(dbname), &result.info_log);
 105     if (!s.ok()) {
 106       // No place suitable for logging
 107       result.info_log = NULL;
 108     }
 109   }
 110   if (result.block_cache == NULL) {
 111     result.block_cache = NewLRUCache(8 << 20);
 112   }
 113   return result;
 114 }
 115
 116 DBImpl::DBImpl(const Options& options, const std::string& dbname)
 117     : env_(options.env),
 118       internal_comparator_(options.comparator),
 119       internal_filter_policy_(options.filter_policy),
 120       options_(SanitizeOptions(
 121           dbname, &internal_comparator_, &internal_filter_policy_, options)),
 122       owns_info_log_(options_.info_log != options.info_log),
 123       owns_cache_(options_.block_cache != options.block_cache),
 124       dbname_(dbname),
 125       db_lock_(NULL),
 126       shutting_down_(NULL),
 127       bg_cv_(&mutex_),
 128       mem_(new MemTable(internal_comparator_)),
 129       imm_(NULL),
 130       logfile_(NULL),
 131       logfile_number_(0),
 132       log_(NULL),
 133       tmp_batch_(new WriteBatch),
 134       bg_compaction_scheduled_(false),
 135       manual_compaction_(NULL),
 136       consecutive_compaction_errors_(0) {
 137   mem_->Ref();
 138   has_imm_.Release_Store(NULL);
 139
 140   // Reserve ten files or so for other uses and give the rest to TableCache.
 141   const int table_cache_size = options.max_open_files - kNumNonTableCacheFiles;
 142   table_cache_ = new TableCache(dbname_, &options_, table_cache_size);
 143
 144   versions_ = new VersionSet(dbname_, &options_, table_cache_,
 145                              &internal_comparator_);
 146 }
 147
 148 DBImpl::~DBImpl() {
 149   // Wait for background work to finish
 150   mutex_.Lock();
 151   shutting_down_.Release_Store(this);  // Any non-NULL value is ok
 152   while (bg_compaction_scheduled_) {
 153     bg_cv_.Wait();
 154   }
 155   mutex_.Unlock();
 156
 157   if (db_lock_ != NULL) {
 158     env_->UnlockFile(db_lock_);
 159   }
 160
 161   delete versions_;
 162   if (mem_ != NULL) mem_->Unref();
 163   if (imm_ != NULL) imm_->Unref();
 164   delete tmp_batch_;
 165   delete log_;
 166   delete logfile_;
 167   delete table_cache_;
 168
 169   if (owns_info_log_) {
 170     delete options_.info_log;
 171   }
 172   if (owns_cache_) {
 173     delete options_.block_cache;
 174   }
 175 }
 176
 177 Status DBImpl::NewDB() {
 178   VersionEdit new_db;
 179   new_db.SetComparatorName(user_comparator()->Name());
 180   new_db.SetLogNumber(0);
 181   new_db.SetNextFile(2);
 182   new_db.SetLastSequence(0);
 183
 184   const std::string manifest = DescriptorFileName(dbname_, 1);
 185   WritableFile* file;
 186   Status s = env_->NewWritableFile(manifest, &file);
 187   if (!s.ok()) {
 188     return s;
 189   }
 190   {
 191     log::Writer log(file);
 192     std::string record;
 193     new_db.EncodeTo(&record);
 194     s = log.AddRecord(record);
 195     if (s.ok()) {
 196       s = file->Close();
 197     }
 198   }
 199   delete file;
 200   if (s.ok()) {
 201     // Make "CURRENT" file that points to the new manifest file.
 202     s = SetCurrentFile(env_, dbname_, 1);
 203   } else {
 204     env_->DeleteFile(manifest);
 205   }
 206   return s;
 207 }
 208
 209 void DBImpl::MaybeIgnoreError(Status* s) const {
 210   if (s->ok() || options_.paranoid_checks) {
 211     // No change needed
 212   } else {
 213     Log(options_.info_log, "Ignoring error %s", s->ToString().c_str());
 214     *s = Status::OK();
 215   }
 216 }
 217
 218 void DBImpl::DeleteObsoleteFiles() {
 219   // Make a set of all of the live files
 220   std::set<uint64_t> live = pending_outputs_;
 221   versions_->AddLiveFiles(&live);
 222
 223   std::vector<std::string> filenames;
 224   env_->GetChildren(dbname_, &filenames); // Ignoring errors on purpose
 225   uint64_t number;
 226   FileType type;
 227   for (size_t i = 0; i < filenames.size(); i++) {
 228     if (ParseFileName(filenames[i], &number, &type)) {
 229       bool keep = true;
 230       switch (type) {
 231         case kLogFile:
 232           keep = ((number >= versions_->LogNumber()) ||
 233                   (number == versions_->PrevLogNumber()));
 234           break;
 235         case kDescriptorFile:
 236           // Keep my manifest file, and any newer incarnations'
 237           // (in case there is a race that allows other incarnations)
 238           keep = (number >= versions_->ManifestFileNumber());
 239           break;
 240         case kTableFile:
 241           keep = (live.find(number) != live.end());
 242           break;
 243         case kTempFile:
 244           // Any temp files that are currently being written to must
 245           // be recorded in pending_outputs_, which is inserted into "live"
 246           keep = (live.find(number) != live.end());
 247           break;
 248         case kCurrentFile:
 249         case kDBLockFile:
 250         case kInfoLogFile:
 251           keep = true;
 252           break;
 253       }
 254
 255       if (!keep) {
 256         if (type == kTableFile) {
 257           table_cache_->Evict(number);
 258         }
 259         Log(options_.info_log, "Delete type=%d #%lld\n",
 260             int(type),
 261             static_cast<unsigned long long>(number));
 262         env_->DeleteFile(dbname_ + "/" + filenames[i]);
 263       }
 264     }
 265   }
 266 }
 267
 268 Status DBImpl::Recover(VersionEdit* edit) {
 269   mutex_.AssertHeld();
 270
 271   // Ignore error from CreateDir since the creation of the DB is
 272   // committed only when the descriptor is created, and this directory
 273   // may already exist from a previous failed creation attempt.
 274   env_->CreateDir(dbname_);
 275   assert(db_lock_ == NULL);
 276   Status s = env_->LockFile(LockFileName(dbname_), &db_lock_);
 277   if (!s.ok()) {
 278     return s;
 279   }
 280
 281   if (!env_->FileExists(CurrentFileName(dbname_))) {
 282     if (options_.create_if_missing) {
 283       s = NewDB();
 284       if (!s.ok()) {
 285         return s;
 286       }
 287     } else {
 288       return Status::InvalidArgument(
 289           dbname_, "does not exist (create_if_missing is false)");
 290     }
 291   } else {
 292     if (options_.error_if_exists) {
 293       return Status::InvalidArgument(
 294           dbname_, "exists (error_if_exists is true)");
 295     }
 296   }
 297
 298   s = versions_->Recover();
 299   if (s.ok()) {
 300     SequenceNumber max_sequence(0);
 301
 302     // Recover from all newer log files than the ones named in the
 303     // descriptor (new log files may have been added by the previous
 304     // incarnation without registering them in the descriptor).
 305     //
 306     // Note that PrevLogNumber() is no longer used, but we pay
 307     // attention to it in case we are recovering a database
 308     // produced by an older version of leveldb.
 309     const uint64_t min_log = versions_->LogNumber();
 310     const uint64_t prev_log = versions_->PrevLogNumber();
 311     std::vector<std::string> filenames;
 312     s = env_->GetChildren(dbname_, &filenames);
 313     if (!s.ok()) {
 314       return s;
 315     }
 316     std::set<uint64_t> expected;
 317     versions_->AddLiveFiles(&expected);
 318     uint64_t number;
 319     FileType type;
 320     std::vector<uint64_t> logs;
 321     for (size_t i = 0; i < filenames.size(); i++) {
 322       if (ParseFileName(filenames[i], &number, &type)) {
 323         expected.erase(number);
 324         if (type == kLogFile && ((number >= min_log) || (number == prev_log)))
 325           logs.push_back(number);
 326       }
 327     }
 328     if (!expected.empty()) {
 329       char buf[50];
 330       snprintf(buf, sizeof(buf), "%d missing files; e.g.",
 331                static_cast<int>(expected.size()));
 332       return Status::Corruption(buf, TableFileName(dbname_, *(expected.begin())));
 333     }
 334
 335     // Recover in the order in which the logs were generated
 336     std::sort(logs.begin(), logs.end());
 337     for (size_t i = 0; i < logs.size(); i++) {
 338       s = RecoverLogFile(logs[i], edit, &max_sequence);
 339
 340       // The previous incarnation may not have written any MANIFEST
 341       // records after allocating this log number.  So we manually
 342       // update the file number allocation counter in VersionSet.
 343       versions_->MarkFileNumberUsed(logs[i]);
 344     }
 345
 346     if (s.ok()) {
 347       if (versions_->LastSequence() < max_sequence) {
 348         versions_->SetLastSequence(max_sequence);
 349       }
 350     }
 351   }
 352
 353   return s;
 354 }
 355
 356 Status DBImpl::RecoverLogFile(uint64_t log_number,
 357                               VersionEdit* edit,
 358                               SequenceNumber* max_sequence) {
 359   struct LogReporter : public log::Reader::Reporter {
 360     Env* env;
 361     Logger* info_log;
 362     const char* fname;
 363     Status* status;  // NULL if options_.paranoid_checks==false
 364     virtual void Corruption(size_t bytes, const Status& s) {
 365       Log(info_log, "%s%s: dropping %d bytes; %s",
 366           (this->status == NULL ? "(ignoring error) " : ""),
 367           fname, static_cast<int>(bytes), s.ToString().c_str());
 368       if (this->status != NULL && this->status->ok()) *this->status = s;
 369     }
 370   };
 371
 372   mutex_.AssertHeld();
 373
 374   // Open the log file
 375   std::string fname = LogFileName(dbname_, log_number);
 376   SequentialFile* file;
 377   Status status = env_->NewSequentialFile(fname, &file);
 378   if (!status.ok()) {
 379     MaybeIgnoreError(&status);
 380     return status;
 381   }
 382
 383   // Create the log reader.
 384   LogReporter reporter;
 385   reporter.env = env_;
 386   reporter.info_log = options_.info_log;
 387   reporter.fname = fname.c_str();
 388   reporter.status = (options_.paranoid_checks ? &status : NULL);
 389   // We intentially make log::Reader do checksumming even if
 390   // paranoid_checks==false so that corruptions cause entire commits
 391   // to be skipped instead of propagating bad information (like overly
 392   // large sequence numbers).
 393   log::Reader reader(file, &reporter, true/*checksum*/,
 394                      0/*initial_offset*/);
 395   Log(options_.info_log, "Recovering log #%llu",
 396       (unsigned long long) log_number);
 397
 398   // Read all the records and add to a memtable
 399   std::string scratch;
 400   Slice record;
 401   WriteBatch batch;
 402   MemTable* mem = NULL;
 403   while (reader.ReadRecord(&record, &scratch) &&
 404          status.ok()) {
 405     if (record.size() < 12) {
 406       reporter.Corruption(
 407           record.size(), Status::Corruption("log record too small"));
 408       continue;
 409     }
 410     WriteBatchInternal::SetContents(&batch, record);
 411
 412     if (mem == NULL) {
 413       mem = new MemTable(internal_comparator_);
 414       mem->Ref();
 415     }
 416     status = WriteBatchInternal::InsertInto(&batch, mem);
 417     MaybeIgnoreError(&status);
 418     if (!status.ok()) {
 419       break;
 420     }
 421     const SequenceNumber last_seq =
 422         WriteBatchInternal::Sequence(&batch) +
 423         WriteBatchInternal::Count(&batch) - 1;
 424     if (last_seq > *max_sequence) {
 425       *max_sequence = last_seq;
 426     }
 427
 428     if (mem->ApproximateMemoryUsage() > options_.write_buffer_size) {
 429       status = WriteLevel0Table(mem, edit, NULL);
 430       if (!status.ok()) {
 431         // Reflect errors immediately so that conditions like full
 432         // file-systems cause the DB::Open() to fail.
 433         break;
 434       }
 435       mem->Unref();
 436       mem = NULL;
 437     }
 438   }
 439
 440   if (status.ok() && mem != NULL) {
 441     status = WriteLevel0Table(mem, edit, NULL);
 442     // Reflect errors immediately so that conditions like full
 443     // file-systems cause the DB::Open() to fail.
 444   }
 445
 446   if (mem != NULL) mem->Unref();
 447   delete file;
 448   return status;
 449 }
 450
 451 Status DBImpl::WriteLevel0Table(MemTable* mem, VersionEdit* edit,
 452                                 Version* base) {
 453   mutex_.AssertHeld();
 454   const uint64_t start_micros = env_->NowMicros();
 455   FileMetaData meta;
 456   meta.number = versions_->NewFileNumber();
 457   pending_outputs_.insert(meta.number);
 458   Iterator* iter = mem->NewIterator();
 459   Log(options_.info_log, "Level-0 table #%llu: started",
 460       (unsigned long long) meta.number);
 461
 462   Status s;
 463   {
 464     mutex_.Unlock();
 465     s = BuildTable(dbname_, env_, options_, table_cache_, iter, &meta);
 466     mutex_.Lock();
 467   }
 468
 469   Log(options_.info_log, "Level-0 table #%llu: %lld bytes %s",
 470       (unsigned long long) meta.number,
 471       (unsigned long long) meta.file_size,
 472       s.ToString().c_str());
 473   delete iter;
 474   pending_outputs_.erase(meta.number);
 475
 476
 477   // Note that if file_size is zero, the file has been deleted and
 478   // should not be added to the manifest.
 479   int level = 0;
 480   if (s.ok() && meta.file_size > 0) {
 481     const Slice min_user_key = meta.smallest.user_key();
 482     const Slice max_user_key = meta.largest.user_key();
 483     if (base != NULL) {
 484       level = base->PickLevelForMemTableOutput(min_user_key, max_user_key);
 485     }
 486     edit->AddFile(level, meta.number, meta.file_size,
 487                   meta.smallest, meta.largest);
 488   }
 489
 490   CompactionStats stats;
 491   stats.micros = env_->NowMicros() - start_micros;
 492   stats.bytes_written = meta.file_size;
 493   stats_[level].Add(stats);
 494   return s;
 495 }
 496
 497 Status DBImpl::CompactMemTable() {
 498   mutex_.AssertHeld();
 499   assert(imm_ != NULL);
 500
 501   // Save the contents of the memtable as a new Table
 502   VersionEdit edit;
 503   Version* base = versions_->current();
 504   base->Ref();
 505   Status s = WriteLevel0Table(imm_, &edit, base);
 506   base->Unref();
 507
 508   if (s.ok() && shutting_down_.Acquire_Load()) {
 509     s = Status::IOError("Deleting DB during memtable compaction");
 510   }
 511
 512   // Replace immutable memtable with the generated Table
 513   if (s.ok()) {
 514     edit.SetPrevLogNumber(0);
 515     edit.SetLogNumber(logfile_number_);  // Earlier logs no longer needed
 516     s = versions_->LogAndApply(&edit, &mutex_);
 517   }
 518
 519   if (s.ok()) {
 520     // Commit to the new state
 521     imm_->Unref();
 522     imm_ = NULL;
 523     has_imm_.Release_Store(NULL);
 524     DeleteObsoleteFiles();
 525   }
 526
 527   return s;
 528 }
 529
 530 void DBImpl::CompactRange(const Slice* begin, const Slice* end) {
 531   int max_level_with_files = 1;
 532   {
 533     MutexLock l(&mutex_);
 534     Version* base = versions_->current();
 535     for (int level = 1; level < config::kNumLevels; level++) {
 536       if (base->OverlapInLevel(level, begin, end)) {
 537         max_level_with_files = level;
 538       }
 539     }
 540   }
 541   TEST_CompactMemTable(); // TODO(sanjay): Skip if memtable does not overlap
 542   for (int level = 0; level < max_level_with_files; level++) {
 543     TEST_CompactRange(level, begin, end);
 544   }
 545 }
 546
 547 void DBImpl::TEST_CompactRange(int level, const Slice* begin,const Slice* end) {
 548   assert(level >= 0);
 549   assert(level + 1 < config::kNumLevels);
 550
 551   InternalKey begin_storage, end_storage;
 552
 553   ManualCompaction manual;
 554   manual.level = level;
 555   manual.done = false;
 556   if (begin == NULL) {
 557     manual.begin = NULL;
 558   } else {
 559     begin_storage = InternalKey(*begin, kMaxSequenceNumber, kValueTypeForSeek);
 560     manual.begin = &begin_storage;
 561   }
 562   if (end == NULL) {
 563     manual.end = NULL;
 564   } else {
 565     end_storage = InternalKey(*end, 0, static_cast<ValueType>(0));
 566     manual.end = &end_storage;
 567   }
 568
 569   MutexLock l(&mutex_);
 570   while (!manual.done) {
 571     while (manual_compaction_ != NULL) {
 572       bg_cv_.Wait();
 573     }
 574     manual_compaction_ = &manual;
 575     MaybeScheduleCompaction();
 576     while (manual_compaction_ == &manual) {
 577       bg_cv_.Wait();
 578     }
 579   }
 580 }
 581
 582 Status DBImpl::TEST_CompactMemTable() {
 583   // NULL batch means just wait for earlier writes to be done
 584   Status s = Write(WriteOptions(), NULL);
 585   if (s.ok()) {
 586     // Wait until the compaction completes
 587     MutexLock l(&mutex_);
 588     while (imm_ != NULL && bg_error_.ok()) {
 589       bg_cv_.Wait();
 590     }
 591     if (imm_ != NULL) {
 592       s = bg_error_;
 593     }
 594   }
 595   return s;
 596 }
 597
 598 void DBImpl::MaybeScheduleCompaction() {
 599   mutex_.AssertHeld();
 600   if (bg_compaction_scheduled_) {
 601     // Already scheduled
 602   } else if (shutting_down_.Acquire_Load()) {
 603     // DB is being deleted; no more background compactions
 604   } else if (imm_ == NULL &&
 605              manual_compaction_ == NULL &&
 606              !versions_->NeedsCompaction()) {
 607     // No work to be done
 608   } else {
 609     bg_compaction_scheduled_ = true;
 610     env_->Schedule(&DBImpl::BGWork, this);
 611   }
 612 }
 613
 614 void DBImpl::BGWork(void* db) {
 615   reinterpret_cast<DBImpl*>(db)->BackgroundCall();
 616 }
 617
 618 void DBImpl::BackgroundCall() {
 619   MutexLock l(&mutex_);
 620   assert(bg_compaction_scheduled_);
 621   if (!shutting_down_.Acquire_Load()) {
 622     Status s = BackgroundCompaction();
 623     if (s.ok()) {
 624       // Success
 625       consecutive_compaction_errors_ = 0;
 626     } else if (shutting_down_.Acquire_Load()) {
 627       // Error most likely due to shutdown; do not wait
 628     } else {
 629       // Wait a little bit before retrying background compaction in
 630       // case this is an environmental problem and we do not want to
 631       // chew up resources for failed compactions for the duration of
 632       // the problem.
 633       bg_cv_.SignalAll();  // In case a waiter can proceed despite the error
 634       Log(options_.info_log, "Waiting after background compaction error: %s",
 635           s.ToString().c_str());
 636       mutex_.Unlock();
 637       ++consecutive_compaction_errors_;
 638       int seconds_to_sleep = 1;
 639       for (int i = 0; i < 3 && i < consecutive_compaction_errors_ - 1; ++i) {
 640         seconds_to_sleep *= 2;
 641       }
 642       env_->SleepForMicroseconds(seconds_to_sleep * 1000000);
 643       mutex_.Lock();
 644     }
 645   }
 646
 647   bg_compaction_scheduled_ = false;
 648
 649   // Previous compaction may have produced too many files in a level,
 650   // so reschedule another compaction if needed.
 651   MaybeScheduleCompaction();
 652   bg_cv_.SignalAll();
 653 }
 654
 655 Status DBImpl::BackgroundCompaction() {
 656   mutex_.AssertHeld();
 657
 658   if (imm_ != NULL) {
 659     return CompactMemTable();
 660   }
 661
 662   Compaction* c;
 663   bool is_manual = (manual_compaction_ != NULL);
 664   InternalKey manual_end;
 665   if (is_manual) {
 666     ManualCompaction* m = manual_compaction_;
 667     c = versions_->CompactRange(m->level, m->begin, m->end);
 668     m->done = (c == NULL);
 669     if (c != NULL) {
 670       manual_end = c->input(0, c->num_input_files(0) - 1)->largest;
 671     }
 672     Log(options_.info_log,
 673         "Manual compaction at level-%d from %s .. %s; will stop at %s\n",
 674         m->level,
 675         (m->begin ? m->begin->DebugString().c_str() : "(begin)"),
 676         (m->end ? m->end->DebugString().c_str() : "(end)"),
 677         (m->done ? "(end)" : manual_end.DebugString().c_str()));
 678   } else {
 679     c = versions_->PickCompaction();
 680   }
 681
 682   Status status;
 683   if (c == NULL) {
 684     // Nothing to do
 685   } else if (!is_manual && c->IsTrivialMove()) {
 686     // Move file to next level
 687     assert(c->num_input_files(0) == 1);
 688     FileMetaData* f = c->input(0, 0);
 689     c->edit()->DeleteFile(c->level(), f->number);
 690     c->edit()->AddFile(c->level() + 1, f->number, f->file_size,
 691                        f->smallest, f->largest);
 692     status = versions_->LogAndApply(c->edit(), &mutex_);
 693     VersionSet::LevelSummaryStorage tmp;
 694     Log(options_.info_log, "Moved #%lld to level-%d %lld bytes %s: %s\n",
 695         static_cast<unsigned long long>(f->number),
 696         c->level() + 1,
 697         static_cast<unsigned long long>(f->file_size),
 698         status.ToString().c_str(),
 699         versions_->LevelSummary(&tmp));
 700   } else {
 701     CompactionState* compact = new CompactionState(c);
 702     status = DoCompactionWork(compact);
 703     CleanupCompaction(compact);
 704     c->ReleaseInputs();
 705     DeleteObsoleteFiles();
 706   }
 707   delete c;
 708
 709   if (status.ok()) {
 710     // Done
 711   } else if (shutting_down_.Acquire_Load()) {
 712     // Ignore compaction errors found during shutting down
 713   } else {
 714     Log(options_.info_log,
 715         "Compaction error: %s", status.ToString().c_str());
 716     if (options_.paranoid_checks && bg_error_.ok()) {
 717       bg_error_ = status;
 718     }
 719   }
 720
 721   if (is_manual) {
 722     ManualCompaction* m = manual_compaction_;
 723     if (!status.ok()) {
 724       m->done = true;
 725     }
 726     if (!m->done) {
 727       // We only compacted part of the requested range.  Update *m
 728       // to the range that is left to be compacted.
 729       m->tmp_storage = manual_end;
 730       m->begin = &m->tmp_storage;
 731     }
 732     manual_compaction_ = NULL;
 733   }
 734   return status;
 735 }
 736
 737 void DBImpl::CleanupCompaction(CompactionState* compact) {
 738   mutex_.AssertHeld();
 739   if (compact->builder != NULL) {
 740     // May happen if we get a shutdown call in the middle of compaction
 741     compact->builder->Abandon();
 742     delete compact->builder;
 743   } else {
 744     assert(compact->outfile == NULL);
 745   }
 746   delete compact->outfile;
 747   for (size_t i = 0; i < compact->outputs.size(); i++) {
 748     const CompactionState::Output& out = compact->outputs[i];
 749     pending_outputs_.erase(out.number);
 750   }
 751   delete compact;
 752 }
 753
 754 Status DBImpl::OpenCompactionOutputFile(CompactionState* compact) {
 755   assert(compact != NULL);
 756   assert(compact->builder == NULL);
 757   uint64_t file_number;
 758   {
 759     mutex_.Lock();
 760     file_number = versions_->NewFileNumber();
 761     pending_outputs_.insert(file_number);
 762     CompactionState::Output out;
 763     out.number = file_number;
 764     out.smallest.Clear();
 765     out.largest.Clear();
 766     compact->outputs.push_back(out);
 767     mutex_.Unlock();
 768   }
 769
 770   // Make the output file
 771   std::string fname = TableFileName(dbname_, file_number);
 772   Status s = env_->NewWritableFile(fname, &compact->outfile);
 773   if (s.ok()) {
 774     compact->builder = new TableBuilder(options_, compact->outfile);
 775   }
 776   return s;
 777 }
 778
 779 Status DBImpl::FinishCompactionOutputFile(CompactionState* compact,
 780                                           Iterator* input) {
 781   assert(compact != NULL);
 782   assert(compact->outfile != NULL);
 783   assert(compact->builder != NULL);
 784
 785   const uint64_t output_number = compact->current_output()->number;
 786   assert(output_number != 0);
 787
 788   // Check for iterator errors
 789   Status s = input->status();
 790   const uint64_t current_entries = compact->builder->NumEntries();
 791   if (s.ok()) {
 792     s = compact->builder->Finish();
 793   } else {
 794     compact->builder->Abandon();
 795   }
 796   const uint64_t current_bytes = compact->builder->FileSize();
 797   compact->current_output()->file_size = current_bytes;
 798   compact->total_bytes += current_bytes;
 799   delete compact->builder;
 800   compact->builder = NULL;
 801
 802   // Finish and check for file errors
 803   if (s.ok()) {
 804     s = compact->outfile->Sync();
 805   }
 806   if (s.ok()) {
 807     s = compact->outfile->Close();
 808   }
 809   delete compact->outfile;
 810   compact->outfile = NULL;
 811
 812   if (s.ok() && current_entries > 0) {
 813     // Verify that the table is usable
 814     Iterator* iter = table_cache_->NewIterator(ReadOptions(),
 815                                                output_number,
 816                                                current_bytes);
 817     s = iter->status();
 818     delete iter;
 819     if (s.ok()) {
 820       Log(options_.info_log,
 821           "Generated table #%llu: %lld keys, %lld bytes",
 822           (unsigned long long) output_number,
 823           (unsigned long long) current_entries,
 824           (unsigned long long) current_bytes);
 825     }
 826   }
 827   return s;
 828 }
 829
 830
 831 Status DBImpl::InstallCompactionResults(CompactionState* compact) {
 832   mutex_.AssertHeld();
 833   Log(options_.info_log,  "Compacted %d@%d + %d@%d files => %lld bytes",
 834       compact->compaction->num_input_files(0),
 835       compact->compaction->level(),
 836       compact->compaction->num_input_files(1),
 837       compact->compaction->level() + 1,
 838       static_cast<long long>(compact->total_bytes));
 839
 840   // Add compaction outputs
 841   compact->compaction->AddInputDeletions(compact->compaction->edit());
 842   const int level = compact->compaction->level();
 843   for (size_t i = 0; i < compact->outputs.size(); i++) {
 844     const CompactionState::Output& out = compact->outputs[i];
 845     compact->compaction->edit()->AddFile(
 846         level + 1,
 847         out.number, out.file_size, out.smallest, out.largest);
 848   }
 849   return versions_->LogAndApply(compact->compaction->edit(), &mutex_);
 850 }
 851
 852 Status DBImpl::DoCompactionWork(CompactionState* compact) {
 853   const uint64_t start_micros = env_->NowMicros();
 854   int64_t imm_micros = 0;  // Micros spent doing imm_ compactions
 855
 856   Log(options_.info_log,  "Compacting %d@%d + %d@%d files",
 857       compact->compaction->num_input_files(0),
 858       compact->compaction->level(),
 859       compact->compaction->num_input_files(1),
 860       compact->compaction->level() + 1);
 861
 862   assert(versions_->NumLevelFiles(compact->compaction->level()) > 0);
 863   assert(compact->builder == NULL);
 864   assert(compact->outfile == NULL);
 865   if (snapshots_.empty()) {
 866     compact->smallest_snapshot = versions_->LastSequence();
 867   } else {
 868     compact->smallest_snapshot = snapshots_.oldest()->number_;
 869   }
 870
 871   // Release mutex while we're actually doing the compaction work
 872   mutex_.Unlock();
 873
 874   Iterator* input = versions_->MakeInputIterator(compact->compaction);
 875   input->SeekToFirst();
 876   Status status;
 877   ParsedInternalKey ikey;
 878   std::string current_user_key;
 879   bool has_current_user_key = false;
 880   SequenceNumber last_sequence_for_key = kMaxSequenceNumber;
 881   for (; input->Valid() && !shutting_down_.Acquire_Load(); ) {
 882     // Prioritize immutable compaction work
 883     if (has_imm_.NoBarrier_Load() != NULL) {
 884       const uint64_t imm_start = env_->NowMicros();
 885       mutex_.Lock();
 886       if (imm_ != NULL) {
 887         CompactMemTable();
 888         bg_cv_.SignalAll();  // Wakeup MakeRoomForWrite() if necessary
 889       }
 890       mutex_.Unlock();
 891       imm_micros += (env_->NowMicros() - imm_start);
 892     }
 893
 894     Slice key = input->key();
 895     if (compact->compaction->ShouldStopBefore(key) &&
 896         compact->builder != NULL) {
 897       status = FinishCompactionOutputFile(compact, input);
 898       if (!status.ok()) {
 899         break;
 900       }
 901     }
 902
 903     // Handle key/value, add to state, etc.
 904     bool drop = false;
 905     if (!ParseInternalKey(key, &ikey)) {
 906       // Do not hide error keys
 907       current_user_key.clear();
 908       has_current_user_key = false;
 909       last_sequence_for_key = kMaxSequenceNumber;
 910     } else {
 911       if (!has_current_user_key ||
 912           user_comparator()->Compare(ikey.user_key,
 913                                      Slice(current_user_key)) != 0) {
 914         // First occurrence of this user key
 915         current_user_key.assign(ikey.user_key.data(), ikey.user_key.size());
 916         has_current_user_key = true;
 917         last_sequence_for_key = kMaxSequenceNumber;
 918       }
 919
 920       if (last_sequence_for_key <= compact->smallest_snapshot) {
 921         // Hidden by an newer entry for same user key
 922         drop = true;    // (A)
 923       } else if (ikey.type == kTypeDeletion &&
 924                  ikey.sequence <= compact->smallest_snapshot &&
 925                  compact->compaction->IsBaseLevelForKey(ikey.user_key)) {
 926         // For this user key:
 927         // (1) there is no data in higher levels
 928         // (2) data in lower levels will have larger sequence numbers
 929         // (3) data in layers that are being compacted here and have
 930         //     smaller sequence numbers will be dropped in the next
 931         //     few iterations of this loop (by rule (A) above).
 932         // Therefore this deletion marker is obsolete and can be dropped.
 933         drop = true;
 934       }
 935
 936       last_sequence_for_key = ikey.sequence;
 937     }
 938 #if 0
 939     Log(options_.info_log,
 940         "  Compact: %s, seq %d, type: %d %d, drop: %d, is_base: %d, "
 941         "%d smallest_snapshot: %d",
 942         ikey.user_key.ToString().c_str(),
 943         (int)ikey.sequence, ikey.type, kTypeValue, drop,
 944         compact->compaction->IsBaseLevelForKey(ikey.user_key),
 945         (int)last_sequence_for_key, (int)compact->smallest_snapshot);
 946 #endif
 947
 948     if (!drop) {
 949       // Open output file if necessary
 950       if (compact->builder == NULL) {
 951         status = OpenCompactionOutputFile(compact);
 952         if (!status.ok()) {
 953           break;
 954         }
 955       }
 956       if (compact->builder->NumEntries() == 0) {
 957         compact->current_output()->smallest.DecodeFrom(key);
 958       }
 959       compact->current_output()->largest.DecodeFrom(key);
 960       compact->builder->Add(key, input->value());
 961
 962       // Close output file if it is big enough
 963       if (compact->builder->FileSize() >=
 964           compact->compaction->MaxOutputFileSize()) {
 965         status = FinishCompactionOutputFile(compact, input);
 966         if (!status.ok()) {
 967           break;
 968         }
 969       }
 970     }
 971
 972     input->Next();
 973   }
 974
 975   if (status.ok() && shutting_down_.Acquire_Load()) {
 976     status = Status::IOError("Deleting DB during compaction");
 977   }
 978   if (status.ok() && compact->builder != NULL) {
 979     status = FinishCompactionOutputFile(compact, input);
 980   }
 981   if (status.ok()) {
 982     status = input->status();
 983   }
 984   delete input;
 985   input = NULL;
 986
 987   CompactionStats stats;
 988   stats.micros = env_->NowMicros() - start_micros - imm_micros;
 989   for (int which = 0; which < 2; which++) {
 990     for (int i = 0; i < compact->compaction->num_input_files(which); i++) {
 991       stats.bytes_read += compact->compaction->input(which, i)->file_size;
 992     }
 993   }
 994   for (size_t i = 0; i < compact->outputs.size(); i++) {
 995     stats.bytes_written += compact->outputs[i].file_size;
 996   }
 997
 998   mutex_.Lock();
 999   stats_[compact->compaction->level() + 1].Add(stats);
1000
1001   if (status.ok()) {
1002     status = InstallCompactionResults(compact);
1003   }
1004   VersionSet::LevelSummaryStorage tmp;
1005   Log(options_.info_log,
1006       "compacted to: %s", versions_->LevelSummary(&tmp));
1007   return status;
1008 }
1009
1010 namespace {
1011 struct IterState {
1012   port::Mutex* mu;
1013   Version* version;
1014   MemTable* mem;
1015   MemTable* imm;
1016 };
1017
1018 static void CleanupIteratorState(void* arg1, void* arg2) {
1019   IterState* state = reinterpret_cast<IterState*>(arg1);
1020   state->mu->Lock();
1021   state->mem->Unref();
1022   if (state->imm != NULL) state->imm->Unref();
1023   state->version->Unref();
1024   state->mu->Unlock();
1025   delete state;
1026 }
1027 }  // namespace
1028
1029 Iterator* DBImpl::NewInternalIterator(const ReadOptions& options,
1030                                       SequenceNumber* latest_snapshot) {
1031   IterState* cleanup = new IterState;
1032   mutex_.Lock();
1033   *latest_snapshot = versions_->LastSequence();
1034
1035   // Collect together all needed child iterators
1036   std::vector<Iterator*> list;
1037   list.push_back(mem_->NewIterator());
1038   mem_->Ref();
1039   if (imm_ != NULL) {
1040     list.push_back(imm_->NewIterator());
1041     imm_->Ref();
1042   }
1043   versions_->current()->AddIterators(options, &list);
1044   Iterator* internal_iter =
1045       NewMergingIterator(&internal_comparator_, &list[0], list.size());
1046   versions_->current()->Ref();
1047
1048   cleanup->mu = &mutex_;
1049   cleanup->mem = mem_;
1050   cleanup->imm = imm_;
1051   cleanup->version = versions_->current();
1052   internal_iter->RegisterCleanup(CleanupIteratorState, cleanup, NULL);
1053
1054   mutex_.Unlock();
1055   return internal_iter;
1056 }
1057
1058 Iterator* DBImpl::TEST_NewInternalIterator() {
1059   SequenceNumber ignored;
1060   return NewInternalIterator(ReadOptions(), &ignored);
1061 }
1062
1063 int64_t DBImpl::TEST_MaxNextLevelOverlappingBytes() {
1064   MutexLock l(&mutex_);
1065   return versions_->MaxNextLevelOverlappingBytes();
1066 }
1067
1068 Status DBImpl::Get(const ReadOptions& options,
1069                    const Slice& key,
1070                    std::string* value) {
1071   Status s;
1072   MutexLock l(&mutex_);
1073   SequenceNumber snapshot;
1074   if (options.snapshot != NULL) {
1075     snapshot = reinterpret_cast<const SnapshotImpl*>(options.snapshot)->number_;
1076   } else {
1077     snapshot = versions_->LastSequence();
1078   }
1079
1080   MemTable* mem = mem_;
1081   MemTable* imm = imm_;
1082   Version* current = versions_->current();
1083   mem->Ref();
1084   if (imm != NULL) imm->Ref();
1085   current->Ref();
1086
1087   bool have_stat_update = false;
1088   Version::GetStats stats;
1089
1090   // Unlock while reading from files and memtables
1091   {
1092     mutex_.Unlock();
1093     // First look in the memtable, then in the immutable memtable (if any).
1094     LookupKey lkey(key, snapshot);
1095     if (mem->Get(lkey, value, &s)) {
1096       // Done
1097     } else if (imm != NULL && imm->Get(lkey, value, &s)) {
1098       // Done
1099     } else {
1100       s = current->Get(options, lkey, value, &stats);
1101       have_stat_update = true;
1102     }
1103     mutex_.Lock();
1104   }
1105
1106   if (have_stat_update && current->UpdateStats(stats)) {
1107     MaybeScheduleCompaction();
1108   }
1109   mem->Unref();
1110   if (imm != NULL) imm->Unref();
1111   current->Unref();
1112   return s;
1113 }
1114
1115 Iterator* DBImpl::NewIterator(const ReadOptions& options) {
1116   SequenceNumber latest_snapshot;
1117   Iterator* internal_iter = NewInternalIterator(options, &latest_snapshot);
1118   return NewDBIterator(
1119       &dbname_, env_, user_comparator(), internal_iter,
1120       (options.snapshot != NULL
1121        ? reinterpret_cast<const SnapshotImpl*>(options.snapshot)->number_
1122        : latest_snapshot));
1123 }
1124
1125 const Snapshot* DBImpl::GetSnapshot() {
1126   MutexLock l(&mutex_);
1127   return snapshots_.New(versions_->LastSequence());
1128 }
1129
1130 void DBImpl::ReleaseSnapshot(const Snapshot* s) {
1131   MutexLock l(&mutex_);
1132   snapshots_.Delete(reinterpret_cast<const SnapshotImpl*>(s));
1133 }
1134
1135 // Convenience methods
1136 Status DBImpl::Put(const WriteOptions& o, const Slice& key, const Slice& val) {
1137   return DB::Put(o, key, val);
1138 }
1139
1140 Status DBImpl::Delete(const WriteOptions& options, const Slice& key) {
1141   return DB::Delete(options, key);
1142 }
1143
1144 Status DBImpl::Write(const WriteOptions& options, WriteBatch* my_batch) {
1145   Writer w(&mutex_);
1146   w.batch = my_batch;
1147   w.sync = options.sync;
1148   w.done = false;
1149
1150   MutexLock l(&mutex_);
1151   writers_.push_back(&w);
1152   while (!w.done && &w != writers_.front()) {
1153     w.cv.Wait();
1154   }
1155   if (w.done) {
1156     return w.status;
1157   }
1158
1159   // May temporarily unlock and wait.
1160   Status status = MakeRoomForWrite(my_batch == NULL);
1161   uint64_t last_sequence = versions_->LastSequence();
1162   Writer* last_writer = &w;
1163   if (status.ok() && my_batch != NULL) {  // NULL batch is for compactions
1164     WriteBatch* updates = BuildBatchGroup(&last_writer);
1165     WriteBatchInternal::SetSequence(updates, last_sequence + 1);
1166     last_sequence += WriteBatchInternal::Count(updates);
1167
1168     // Add to log and apply to memtable.  We can release the lock
1169     // during this phase since &w is currently responsible for logging
1170     // and protects against concurrent loggers and concurrent writes
1171     // into mem_.
1172     {
1173       mutex_.Unlock();
1174       status = log_->AddRecord(WriteBatchInternal::Contents(updates));
1175       if (status.ok() && options.sync) {
1176         status = logfile_->Sync();
1177       }
1178       if (status.ok()) {
1179         status = WriteBatchInternal::InsertInto(updates, mem_);
1180       }
1181       mutex_.Lock();
1182     }
1183     if (updates == tmp_batch_) tmp_batch_->Clear();
1184
1185     versions_->SetLastSequence(last_sequence);
1186   }
1187
1188   while (true) {
1189     Writer* ready = writers_.front();
1190     writers_.pop_front();
1191     if (ready != &w) {
1192       ready->status = status;
1193       ready->done = true;
1194       ready->cv.Signal();
1195     }
1196     if (ready == last_writer) break;
1197   }
1198
1199   // Notify new head of write queue
1200   if (!writers_.empty()) {
1201     writers_.front()->cv.Signal();
1202   }
1203
1204   return status;
1205 }
1206
1207 // REQUIRES: Writer list must be non-empty
1208 // REQUIRES: First writer must have a non-NULL batch
1209 WriteBatch* DBImpl::BuildBatchGroup(Writer** last_writer) {
1210   assert(!writers_.empty());
1211   Writer* first = writers_.front();
1212   WriteBatch* result = first->batch;
1213   assert(result != NULL);
1214
1215   size_t size = WriteBatchInternal::ByteSize(first->batch);
1216
1217   // Allow the group to grow up to a maximum size, but if the
1218   // original write is small, limit the growth so we do not slow
1219   // down the small write too much.
1220   size_t max_size = 1 << 20;
1221   if (size <= (128<<10)) {
1222     max_size = size + (128<<10);
1223   }
1224
1225   *last_writer = first;
1226   std::deque<Writer*>::iterator iter = writers_.begin();
1227   ++iter;  // Advance past "first"
1228   for (; iter != writers_.end(); ++iter) {
1229     Writer* w = *iter;
1230     if (w->sync && !first->sync) {
1231       // Do not include a sync write into a batch handled by a non-sync write.
1232       break;
1233     }
1234
1235     if (w->batch != NULL) {
1236       size += WriteBatchInternal::ByteSize(w->batch);
1237       if (size > max_size) {
1238         // Do not make batch too big
1239         break;
1240       }
1241
1242       // Append to *reuslt
1243       if (result == first->batch) {
1244         // Switch to temporary batch instead of disturbing caller's batch
1245         result = tmp_batch_;
1246         assert(WriteBatchInternal::Count(result) == 0);
1247         WriteBatchInternal::Append(result, first->batch);
1248       }
1249       WriteBatchInternal::Append(result, w->batch);
1250     }
1251     *last_writer = w;
1252   }
1253   return result;
1254 }
1255
1256 // REQUIRES: mutex_ is held
1257 // REQUIRES: this thread is currently at the front of the writer queue
1258 Status DBImpl::MakeRoomForWrite(bool force) {
1259   mutex_.AssertHeld();
1260   assert(!writers_.empty());
1261   bool allow_delay = !force;
1262   Status s;
1263   while (true) {
1264     if (!bg_error_.ok()) {
1265       // Yield previous error
1266       s = bg_error_;
1267       break;
1268     } else if (
1269         allow_delay &&
1270         versions_->NumLevelFiles(0) >= config::kL0_SlowdownWritesTrigger) {
1271       // We are getting close to hitting a hard limit on the number of
1272       // L0 files.  Rather than delaying a single write by several
1273       // seconds when we hit the hard limit, start delaying each
1274       // individual write by 1ms to reduce latency variance.  Also,
1275       // this delay hands over some CPU to the compaction thread in
1276       // case it is sharing the same core as the writer.
1277       mutex_.Unlock();
1278       env_->SleepForMicroseconds(1000);
1279       allow_delay = false;  // Do not delay a single write more than once
1280       mutex_.Lock();
1281     } else if (!force &&
1282                (mem_->ApproximateMemoryUsage() <= options_.write_buffer_size)) {
1283       // There is room in current memtable
1284       break;
1285     } else if (imm_ != NULL) {
1286       // We have filled up the current memtable, but the previous
1287       // one is still being compacted, so we wait.
1288       Log(options_.info_log, "Current memtable full; waiting...\n");
1289       bg_cv_.Wait();
1290     } else if (versions_->NumLevelFiles(0) >= config::kL0_StopWritesTrigger) {
1291       // There are too many level-0 files.
1292       Log(options_.info_log, "Too many L0 files; waiting...\n");
1293       bg_cv_.Wait();
1294     } else {
1295       // Attempt to switch to a new memtable and trigger compaction of old
1296       assert(versions_->PrevLogNumber() == 0);
1297       uint64_t new_log_number = versions_->NewFileNumber();
1298       WritableFile* lfile = NULL;
1299       s = env_->NewWritableFile(LogFileName(dbname_, new_log_number), &lfile);
1300       if (!s.ok()) {
1301         // Avoid chewing through file number space in a tight loop.
1302         versions_->ReuseFileNumber(new_log_number);
1303         break;
1304       }
1305       delete log_;
1306       delete logfile_;
1307       logfile_ = lfile;
1308       logfile_number_ = new_log_number;
1309       log_ = new log::Writer(lfile);
1310       imm_ = mem_;
1311       has_imm_.Release_Store(imm_);
1312       mem_ = new MemTable(internal_comparator_);
1313       mem_->Ref();
1314       force = false;   // Do not force another compaction if have room
1315       MaybeScheduleCompaction();
1316     }
1317   }
1318   return s;
1319 }
1320
1321 bool DBImpl::GetProperty(const Slice& property, std::string* value) {
1322   value->clear();
1323
1324   MutexLock l(&mutex_);
1325   Slice in = property;
1326   Slice prefix("leveldb.");
1327   if (!in.starts_with(prefix)) return false;
1328   in.remove_prefix(prefix.size());
1329
1330   if (in.starts_with("num-files-at-level")) {
1331     in.remove_prefix(strlen("num-files-at-level"));
1332     uint64_t level;
1333     bool ok = ConsumeDecimalNumber(&in, &level) && in.empty();
1334     if (!ok || level >= config::kNumLevels) {
1335       return false;
1336     } else {
1337       char buf[100];
1338       snprintf(buf, sizeof(buf), "%d",
1339                versions_->NumLevelFiles(static_cast<int>(level)));
1340       *value = buf;
1341       return true;
1342     }
1343   } else if (in == "stats") {
1344     char buf[200];
1345     snprintf(buf, sizeof(buf),
1346              "                               Compactions\n"
1347              "Level  Files Size(MB) Time(sec) Read(MB) Write(MB)\n"
1348              "--------------------------------------------------\n"
1349              );
1350     value->append(buf);
1351     for (int level = 0; level < config::kNumLevels; level++) {
1352       int files = versions_->NumLevelFiles(level);
1353       if (stats_[level].micros > 0 || files > 0) {
1354         snprintf(
1355             buf, sizeof(buf),
1356             "%3d %8d %8.0f %9.0f %8.0f %9.0f\n",
1357             level,
1358             files,
1359             versions_->NumLevelBytes(level) / 1048576.0,
1360             stats_[level].micros / 1e6,
1361             stats_[level].bytes_read / 1048576.0,
1362             stats_[level].bytes_written / 1048576.0);
1363         value->append(buf);
1364       }
1365     }
1366     return true;
1367   } else if (in == "sstables") {
1368     *value = versions_->current()->DebugString();
1369     return true;
1370   }
1371
1372   return false;
1373 }
1374
1375 void DBImpl::GetApproximateSizes(
1376     const Range* range, int n,
1377     uint64_t* sizes) {
1378   // TODO(opt): better implementation
1379   Version* v;
1380   {
1381     MutexLock l(&mutex_);
1382     versions_->current()->Ref();
1383     v = versions_->current();
1384   }
1385
1386   for (int i = 0; i < n; i++) {
1387     // Convert user_key into a corresponding internal key.
1388     InternalKey k1(range[i].start, kMaxSequenceNumber, kValueTypeForSeek);
1389     InternalKey k2(range[i].limit, kMaxSequenceNumber, kValueTypeForSeek);
1390     uint64_t start = versions_->ApproximateOffsetOf(v, k1);
1391     uint64_t limit = versions_->ApproximateOffsetOf(v, k2);
1392     sizes[i] = (limit >= start ? limit - start : 0);
1393   }
1394
1395   {
1396     MutexLock l(&mutex_);
1397     v->Unref();
1398   }
1399 }
1400
1401 // Default implementations of convenience methods that subclasses of DB
1402 // can call if they wish
1403 Status DB::Put(const WriteOptions& opt, const Slice& key, const Slice& value) {
1404   WriteBatch batch;
1405   batch.Put(key, value);
1406   return Write(opt, &batch);
1407 }
1408
1409 Status DB::Delete(const WriteOptions& opt, const Slice& key) {
1410   WriteBatch batch;
1411   batch.Delete(key);
1412   return Write(opt, &batch);
1413 }
1414
1415 DB::~DB() { }
1416
1417 Status DB::Open(const Options& options, const std::string& dbname,
1418                 DB** dbptr) {
1419   *dbptr = NULL;
1420
1421   DBImpl* impl = new DBImpl(options, dbname);
1422   impl->mutex_.Lock();
1423   VersionEdit edit;
1424   Status s = impl->Recover(&edit); // Handles create_if_missing, error_if_exists
1425   if (s.ok()) {
1426     uint64_t new_log_number = impl->versions_->NewFileNumber();
1427     WritableFile* lfile;
1428     s = options.env->NewWritableFile(LogFileName(dbname, new_log_number),
1429                                      &lfile);
1430     if (s.ok()) {
1431       edit.SetLogNumber(new_log_number);
1432       impl->logfile_ = lfile;
1433       impl->logfile_number_ = new_log_number;
1434       impl->log_ = new log::Writer(lfile);
1435       s = impl->versions_->LogAndApply(&edit, &impl->mutex_);
1436     }
1437     if (s.ok()) {
1438       impl->DeleteObsoleteFiles();
1439       impl->MaybeScheduleCompaction();
1440     }
1441   }
1442   impl->mutex_.Unlock();
1443   if (s.ok()) {
1444     *dbptr = impl;
1445   } else {
1446     delete impl;
1447   }
1448   return s;
1449 }
1450
1451 Snapshot::~Snapshot() {
1452 }
1453
1454 Status DestroyDB(const std::string& dbname, const Options& options) {
1455   Env* env = options.env;
1456   std::vector<std::string> filenames;
1457   // Ignore error in case directory does not exist
1458   env->GetChildren(dbname, &filenames);
1459   if (filenames.empty()) {
1460     return Status::OK();
1461   }
1462
1463   FileLock* lock;
1464   const std::string lockname = LockFileName(dbname);
1465   Status result = env->LockFile(lockname, &lock);
1466   if (result.ok()) {
1467     uint64_t number;
1468     FileType type;
1469     for (size_t i = 0; i < filenames.size(); i++) {
1470       if (ParseFileName(filenames[i], &number, &type) &&
1471           type != kDBLockFile) {  // Lock file will be deleted at end
1472         Status del = env->DeleteFile(dbname + "/" + filenames[i]);
1473         if (result.ok() && !del.ok()) {
1474           result = del;
1475         }
1476       }
1477     }
1478     env->UnlockFile(lock);  // Ignore error since state is already gone
1479     env->DeleteFile(lockname);
1480     env->DeleteDir(dbname);  // Ignore error in case dir contains other files
1481   }
1482   return result;
1483 }
1484
1485 }  // namespace leveldb