version_set.cc 77.9 KB
Newer Older
J
jorlow@chromium.org 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13
// 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/version_set.h"

#include <algorithm>
#include <stdio.h>
#include "db/filename.h"
#include "db/log_reader.h"
#include "db/log_writer.h"
#include "db/memtable.h"
#include "db/table_cache.h"
14
#include "leveldb/env.h"
15
#include "leveldb/merge_operator.h"
16
#include "leveldb/table_builder.h"
J
jorlow@chromium.org 已提交
17 18 19 20
#include "table/merger.h"
#include "table/two_level_iterator.h"
#include "util/coding.h"
#include "util/logging.h"
21
#include "util/stop_watch.h"
J
jorlow@chromium.org 已提交
22 23 24

namespace leveldb {

G
Gabor Cselle 已提交
25 26
static int64_t TotalFileSize(const std::vector<FileMetaData*>& files) {
  int64_t sum = 0;
27
  for (size_t i = 0; i < files.size() && files[i]; i++) {
G
Gabor Cselle 已提交
28 29 30 31 32
    sum += files[i]->file_size;
  }
  return sum;
}

J
jorlow@chromium.org 已提交
33 34
Version::~Version() {
  assert(refs_ == 0);
35 36 37 38 39 40

  // Remove from linked list
  prev_->next_ = next_;
  next_->prev_ = prev_;

  // Drop references to files
41
  for (int level = 0; level < vset_->NumberLevels(); level++) {
D
dgrogan@chromium.org 已提交
42
    for (size_t i = 0; i < files_[level].size(); i++) {
J
jorlow@chromium.org 已提交
43
      FileMetaData* f = files_[level][i];
44
      assert(f->refs > 0);
J
jorlow@chromium.org 已提交
45 46 47 48 49 50
      f->refs--;
      if (f->refs <= 0) {
        delete f;
      }
    }
  }
51
  delete[] files_;
J
jorlow@chromium.org 已提交
52 53
}

54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74
int FindFile(const InternalKeyComparator& icmp,
             const std::vector<FileMetaData*>& files,
             const Slice& key) {
  uint32_t left = 0;
  uint32_t right = files.size();
  while (left < right) {
    uint32_t mid = (left + right) / 2;
    const FileMetaData* f = files[mid];
    if (icmp.InternalKeyComparator::Compare(f->largest.Encode(), key) < 0) {
      // Key at "mid.largest" is < "target".  Therefore all
      // files at or before "mid" are uninteresting.
      left = mid + 1;
    } else {
      // Key at "mid.largest" is >= "target".  Therefore all files
      // after "mid" are uninteresting.
      right = mid;
    }
  }
  return right;
}

G
Gabor Cselle 已提交
75 76
static bool AfterFile(const Comparator* ucmp,
                      const Slice* user_key, const FileMetaData* f) {
A
Abhishek Kona 已提交
77 78
  // nullptr user_key occurs before all keys and is therefore never after *f
  return (user_key != nullptr &&
G
Gabor Cselle 已提交
79 80 81 82 83
          ucmp->Compare(*user_key, f->largest.user_key()) > 0);
}

static bool BeforeFile(const Comparator* ucmp,
                       const Slice* user_key, const FileMetaData* f) {
A
Abhishek Kona 已提交
84 85
  // nullptr user_key occurs after all keys and is therefore never before *f
  return (user_key != nullptr &&
G
Gabor Cselle 已提交
86 87 88
          ucmp->Compare(*user_key, f->smallest.user_key()) < 0);
}

89 90
bool SomeFileOverlapsRange(
    const InternalKeyComparator& icmp,
G
Gabor Cselle 已提交
91
    bool disjoint_sorted_files,
92
    const std::vector<FileMetaData*>& files,
G
Gabor Cselle 已提交
93 94 95 96 97
    const Slice* smallest_user_key,
    const Slice* largest_user_key) {
  const Comparator* ucmp = icmp.user_comparator();
  if (!disjoint_sorted_files) {
    // Need to check against all files
H
heyongqiang 已提交
98
    for (size_t i = 0; i < files.size(); i++) {
G
Gabor Cselle 已提交
99 100 101 102 103 104 105 106 107 108 109 110 111
      const FileMetaData* f = files[i];
      if (AfterFile(ucmp, smallest_user_key, f) ||
          BeforeFile(ucmp, largest_user_key, f)) {
        // No overlap
      } else {
        return true;  // Overlap
      }
    }
    return false;
  }

  // Binary search over file list
  uint32_t index = 0;
A
Abhishek Kona 已提交
112
  if (smallest_user_key != nullptr) {
G
Gabor Cselle 已提交
113 114 115 116 117 118 119 120 121 122 123
    // Find the earliest possible internal key for smallest_user_key
    InternalKey small(*smallest_user_key, kMaxSequenceNumber,kValueTypeForSeek);
    index = FindFile(icmp, files, small.Encode());
  }

  if (index >= files.size()) {
    // beginning of range is after all files, so no overlap.
    return false;
  }

  return !BeforeFile(ucmp, largest_user_key, files[index]);
124 125
}

J
jorlow@chromium.org 已提交
126 127 128
// An internal iterator.  For a given version/level pair, yields
// information about the files in the level.  For a given entry, key()
// is the largest key that occurs in the file, and value() is an
J
jorlow@chromium.org 已提交
129 130
// 16-byte value containing the file number and file size, both
// encoded using EncodeFixed64.
J
jorlow@chromium.org 已提交
131 132
class Version::LevelFileNumIterator : public Iterator {
 public:
133
  LevelFileNumIterator(const InternalKeyComparator& icmp,
J
jorlow@chromium.org 已提交
134
                       const std::vector<FileMetaData*>* flist)
135
      : icmp_(icmp),
J
jorlow@chromium.org 已提交
136 137 138 139 140 141 142
        flist_(flist),
        index_(flist->size()) {        // Marks as invalid
  }
  virtual bool Valid() const {
    return index_ < flist_->size();
  }
  virtual void Seek(const Slice& target) {
143
    index_ = FindFile(icmp_, *flist_, target);
J
jorlow@chromium.org 已提交
144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167
  }
  virtual void SeekToFirst() { index_ = 0; }
  virtual void SeekToLast() {
    index_ = flist_->empty() ? 0 : flist_->size() - 1;
  }
  virtual void Next() {
    assert(Valid());
    index_++;
  }
  virtual void Prev() {
    assert(Valid());
    if (index_ == 0) {
      index_ = flist_->size();  // Marks as invalid
    } else {
      index_--;
    }
  }
  Slice key() const {
    assert(Valid());
    return (*flist_)[index_]->largest.Encode();
  }
  Slice value() const {
    assert(Valid());
    EncodeFixed64(value_buf_, (*flist_)[index_]->number);
J
jorlow@chromium.org 已提交
168
    EncodeFixed64(value_buf_+8, (*flist_)[index_]->file_size);
J
jorlow@chromium.org 已提交
169 170 171 172 173 174
    return Slice(value_buf_, sizeof(value_buf_));
  }
  virtual Status status() const { return Status::OK(); }
 private:
  const InternalKeyComparator icmp_;
  const std::vector<FileMetaData*>* const flist_;
D
dgrogan@chromium.org 已提交
175
  uint32_t index_;
J
jorlow@chromium.org 已提交
176

J
jorlow@chromium.org 已提交
177 178
  // Backing store for value().  Holds the file number and size.
  mutable char value_buf_[16];
J
jorlow@chromium.org 已提交
179 180 181 182
};

static Iterator* GetFileIterator(void* arg,
                                 const ReadOptions& options,
183
                                 const EnvOptions& soptions,
184 185
                                 const Slice& file_value,
                                 bool for_compaction) {
J
jorlow@chromium.org 已提交
186
  TableCache* cache = reinterpret_cast<TableCache*>(arg);
J
jorlow@chromium.org 已提交
187
  if (file_value.size() != 16) {
J
jorlow@chromium.org 已提交
188 189 190
    return NewErrorIterator(
        Status::Corruption("FileReader invoked with unexpected value"));
  } else {
J
jorlow@chromium.org 已提交
191
    return cache->NewIterator(options,
192
                              soptions,
J
jorlow@chromium.org 已提交
193
                              DecodeFixed64(file_value.data()),
194 195 196
                              DecodeFixed64(file_value.data() + 8),
                              nullptr /* don't need reference to table*/,
                              for_compaction);
J
jorlow@chromium.org 已提交
197 198 199 200
  }
}

Iterator* Version::NewConcatenatingIterator(const ReadOptions& options,
H
Haobo Xu 已提交
201
                                            const EnvOptions& soptions,
J
jorlow@chromium.org 已提交
202 203
                                            int level) const {
  return NewTwoLevelIterator(
204
      new LevelFileNumIterator(vset_->icmp_, &files_[level]),
205
      &GetFileIterator, vset_->table_cache_, options, soptions);
J
jorlow@chromium.org 已提交
206 207 208
}

void Version::AddIterators(const ReadOptions& options,
H
Haobo Xu 已提交
209
                           const EnvOptions& soptions,
J
jorlow@chromium.org 已提交
210 211
                           std::vector<Iterator*>* iters) {
  // Merge all level zero files together since they may overlap
D
dgrogan@chromium.org 已提交
212
  for (size_t i = 0; i < files_[0].size(); i++) {
J
jorlow@chromium.org 已提交
213
    iters->push_back(
J
jorlow@chromium.org 已提交
214
        vset_->table_cache_->NewIterator(
215
            options, soptions, files_[0][i]->number, files_[0][i]->file_size));
J
jorlow@chromium.org 已提交
216 217 218 219 220
  }

  // For levels > 0, we can use a concatenating iterator that sequentially
  // walks through the non-overlapping files in the level, opening them
  // lazily.
221
  for (int level = 1; level < vset_->NumberLevels(); level++) {
J
jorlow@chromium.org 已提交
222
    if (!files_[level].empty()) {
223
      iters->push_back(NewConcatenatingIterator(options, soptions, level));
J
jorlow@chromium.org 已提交
224 225 226 227
    }
  }
}

S
Sanjay Ghemawat 已提交
228 229 230 231 232 233 234
// Callback from TableCache::Get()
namespace {
enum SaverState {
  kNotFound,
  kFound,
  kDeleted,
  kCorrupt,
235
  kMerge // value contains the current merge result (the operand)
S
Sanjay Ghemawat 已提交
236 237 238 239 240
};
struct Saver {
  SaverState state;
  const Comparator* ucmp;
  Slice user_key;
241
  bool* value_found; // Is value set correctly? Used by KeyMayExist
S
Sanjay Ghemawat 已提交
242
  std::string* value;
243 244
  const MergeOperator* merge_operator;
  Logger* logger;
245
  bool didIO;    // did we do any disk io?
S
Sanjay Ghemawat 已提交
246 247
};
}
248

249 250 251 252 253
// Called from TableCache::Get and InternalGet when file/block in which key may
// exist are not there in TableCache/BlockCache respectively. In this case we
// can't guarantee that key does not exist and are not permitted to do IO to be
// certain.Set the status=kFound and value_found=false to let the caller know
// that key may exist but is not there in memory
254 255 256
static void MarkKeyMayExist(void* arg) {
  Saver* s = reinterpret_cast<Saver*>(arg);
  s->state = kFound;
257 258 259
  if (s->value_found != nullptr) {
    *(s->value_found) = false;
  }
260 261
}

262
static bool SaveValue(void* arg, const Slice& ikey, const Slice& v, bool didIO){
S
Sanjay Ghemawat 已提交
263
  Saver* s = reinterpret_cast<Saver*>(arg);
264
  ParsedInternalKey parsed_key;
265
  // TODO: didIO and Merge?
266
  s->didIO = didIO;
S
Sanjay Ghemawat 已提交
267
  if (!ParseInternalKey(ikey, &parsed_key)) {
268
    // TODO: what about corrupt during Merge?
S
Sanjay Ghemawat 已提交
269 270 271
    s->state = kCorrupt;
  } else {
    if (s->ucmp->Compare(parsed_key.user_key, s->user_key) == 0) {
272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314
      switch (parsed_key.type) {
        case kTypeValue:
          if (kNotFound == s->state) {
            s->value->assign(v.data(), v.size());
          } else if (kMerge == s->state) {
            std::string operand;
            swap(operand, *s->value);
            s->merge_operator->Merge(s->user_key, &v, operand,
                                     s->value, s->logger);
          } else {
            assert(false);
          }
          s->state = kFound;
          return false;

        case kTypeMerge:
          if (kNotFound == s->state) {
            s->state = kMerge;
            s->value->assign(v.data(), v.size());
          } else if (kMerge == s->state) {
            std::string operand;
            swap(operand, *s->value);
            s->merge_operator->Merge(s->user_key, &v, operand,
                                     s->value, s->logger);
          } else {
            assert(false);
          }
          return true;

        case kTypeDeletion:
          if (kNotFound == s->state) {
            s->state = kDeleted;
          } else if (kMerge == s->state) {
            std::string operand;
            swap(operand, *s->value);
            s->merge_operator->Merge(s->user_key, nullptr, operand,
                                     s->value, s->logger);
            s->state = kFound;
          } else {
            assert(false);
          }
          return false;

S
Sanjay Ghemawat 已提交
315
      }
316 317
    }
  }
318 319 320 321

  // s->state could be Corrupt, merge or notfound

  return false;
322 323 324 325 326 327
}

static bool NewestFirst(FileMetaData* a, FileMetaData* b) {
  return a->number > b->number;
}

328
Version::Version(VersionSet* vset, uint64_t version_number)
329
    : vset_(vset), next_(this), prev_(this), refs_(0),
330
      files_by_size_(vset->NumberLevels()),
331
      next_file_to_compact_by_size_(vset->NumberLevels()),
A
Abhishek Kona 已提交
332
      file_to_compact_(nullptr),
333
      file_to_compact_level_(-1),
334 335 336 337
      compaction_score_(vset->NumberLevels()),
      compaction_level_(vset->NumberLevels()),
      offset_manifest_file_(0),
      version_number_(version_number) {
338 339 340
  files_ = new std::vector<FileMetaData*>[vset->NumberLevels()];
}

341 342 343 344 345
void Version::Get(const ReadOptions& options,
                  const LookupKey& k,
                  std::string* value,
                  Status *status,
                  GetStats* stats,
346
                  const Options& db_options,
347 348
                  const bool no_io,
                  bool* value_found) {
349 350 351
  Slice ikey = k.internal_key();
  Slice user_key = k.user_key();
  const Comparator* ucmp = vset_->icmp_.user_comparator();
352 353 354 355 356

  auto merge_operator = db_options.merge_operator;
  auto logger = db_options.info_log;

  assert(status->ok() || status->IsMergeInProgress());
357
  if (no_io) {
358 359
    assert(status->ok());
  }
360 361 362 363
  Saver saver;
  saver.state = status->ok()? kNotFound : kMerge;
  saver.ucmp = ucmp;
  saver.user_key = user_key;
364
  saver.value_found = value_found;
365 366 367 368
  saver.value = value;
  saver.merge_operator = merge_operator;
  saver.logger = logger.get();
  saver.didIO = false;
369

A
Abhishek Kona 已提交
370
  stats->seek_file = nullptr;
371
  stats->seek_file_level = -1;
A
Abhishek Kona 已提交
372
  FileMetaData* last_file_read = nullptr;
373
  int last_file_read_level = -1;
374 375 376 377 378 379

  // We can search level-by-level since entries never hop across
  // levels.  Therefore we are guaranteed that if we find data
  // in an smaller level, later levels are irrelevant.
  std::vector<FileMetaData*> tmp;
  FileMetaData* tmp2;
380
  for (int level = 0; level < vset_->NumberLevels(); level++) {
381 382 383 384 385 386 387 388 389
    size_t num_files = files_[level].size();
    if (num_files == 0) continue;

    // Get the list of files to search in this level
    FileMetaData* const* files = &files_[level][0];
    if (level == 0) {
      // Level-0 files may overlap each other.  Find all files that
      // overlap user_key and process them in order from newest to oldest.
      tmp.reserve(num_files);
D
dgrogan@chromium.org 已提交
390
      for (uint32_t i = 0; i < num_files; i++) {
391 392 393 394 395 396 397 398 399 400 401 402 403 404 405
        FileMetaData* f = files[i];
        if (ucmp->Compare(user_key, f->smallest.user_key()) >= 0 &&
            ucmp->Compare(user_key, f->largest.user_key()) <= 0) {
          tmp.push_back(f);
        }
      }
      if (tmp.empty()) continue;

      std::sort(tmp.begin(), tmp.end(), NewestFirst);
      files = &tmp[0];
      num_files = tmp.size();
    } else {
      // Binary search to find earliest index whose largest key >= ikey.
      uint32_t index = FindFile(vset_->icmp_, files_[level], ikey);
      if (index >= num_files) {
A
Abhishek Kona 已提交
406
        files = nullptr;
407 408 409 410 411
        num_files = 0;
      } else {
        tmp2 = files[index];
        if (ucmp->Compare(user_key, tmp2->smallest.user_key()) < 0) {
          // All of "tmp2" is past any data for user_key
A
Abhishek Kona 已提交
412
          files = nullptr;
413 414 415 416
          num_files = 0;
        } else {
          files = &tmp2;
          num_files = 1;
417
          // TODO, is level 1-n files all disjoint in user key space?
418 419 420 421 422
        }
      }
    }


423 424

    for (uint32_t i = 0; i < num_files; ++i) {
425
      FileMetaData* f = files[i];
426
      bool tableIO = false;
427
      *status = vset_->table_cache_->Get(options, f->number, f->file_size,
428
                                         ikey, &saver, SaveValue, &tableIO,
429
                                         MarkKeyMayExist, no_io);
430 431 432
      // TODO: examine the behavior for corrupted key
      if (!status->ok()) {
        return;
S
Sanjay Ghemawat 已提交
433
      }
434

A
Abhishek Kona 已提交
435
      if (last_file_read != nullptr && stats->seek_file == nullptr) {
436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452
        // We have had more than one seek for this read.  Charge the 1st file.
        stats->seek_file = last_file_read;
        stats->seek_file_level = last_file_read_level;
      }

      // If we did any IO as part of the read, then we remember it because
      // it is a possible candidate for seek-based compaction. saver.didIO
      // is true if the block had to be read in from storage and was not
      // pre-exisiting in the block cache. Also, if this file was not pre-
      // existing in the table cache and had to be freshly opened that needed
      // the index blocks to be read-in, then tableIO is true. One thing
      // to note is that the index blocks are not part of the block cache.
      if (saver.didIO || tableIO) {
        last_file_read = f;
        last_file_read_level = level;
      }

S
Sanjay Ghemawat 已提交
453 454 455 456
      switch (saver.state) {
        case kNotFound:
          break;      // Keep searching in other files
        case kFound:
457
          return;
S
Sanjay Ghemawat 已提交
458
        case kDeleted:
459 460
          *status = Status::NotFound(Slice());  // Use empty error message for speed
          return;
S
Sanjay Ghemawat 已提交
461
        case kCorrupt:
462 463 464 465
          *status = Status::Corruption("corrupted key for ", user_key);
          return;
        case kMerge:
          break;
466 467 468 469
      }
    }
  }

470 471 472 473 474 475 476 477 478 479 480 481 482 483

  if (kMerge == saver.state) {
    // merge operand is in *value and we hit the beginning of the key history
    // do a final merge of nullptr and operand;
    std::string operand;
    swap(operand, *value);
    merge_operator->Merge(user_key, nullptr, operand,
                          value, logger.get());
    *status = Status::OK();
    return;
  } else {
    *status = Status::NotFound(Slice()); // Use an empty error message for speed
    return;
  }
484 485 486 487
}

bool Version::UpdateStats(const GetStats& stats) {
  FileMetaData* f = stats.seek_file;
A
Abhishek Kona 已提交
488
  if (f != nullptr) {
489
    f->allowed_seeks--;
A
Abhishek Kona 已提交
490
    if (f->allowed_seeks <= 0 && file_to_compact_ == nullptr) {
491 492 493 494 495 496 497 498
      file_to_compact_ = f;
      file_to_compact_level_ = stats.seek_file_level;
      return true;
    }
  }
  return false;
}

J
jorlow@chromium.org 已提交
499 500 501 502 503
void Version::Ref() {
  ++refs_;
}

void Version::Unref() {
504
  assert(this != &vset_->dummy_versions_);
J
jorlow@chromium.org 已提交
505 506 507
  assert(refs_ >= 1);
  --refs_;
  if (refs_ == 0) {
508
    delete this;
J
jorlow@chromium.org 已提交
509 510 511
  }
}

512
bool Version::OverlapInLevel(int level,
G
Gabor Cselle 已提交
513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528
                             const Slice* smallest_user_key,
                             const Slice* largest_user_key) {
  return SomeFileOverlapsRange(vset_->icmp_, (level > 0), files_[level],
                               smallest_user_key, largest_user_key);
}

int Version::PickLevelForMemTableOutput(
    const Slice& smallest_user_key,
    const Slice& largest_user_key) {
  int level = 0;
  if (!OverlapInLevel(0, &smallest_user_key, &largest_user_key)) {
    // Push to next level if there is no overlap in next level,
    // and the #bytes overlapping in the level after that are limited.
    InternalKey start(smallest_user_key, kMaxSequenceNumber, kValueTypeForSeek);
    InternalKey limit(largest_user_key, 0, static_cast<ValueType>(0));
    std::vector<FileMetaData*> overlaps;
529 530
    int max_mem_compact_level = vset_->options_->max_mem_compaction_level;
    while (max_mem_compact_level > 0 && level < max_mem_compact_level) {
G
Gabor Cselle 已提交
531 532 533
      if (OverlapInLevel(level + 1, &smallest_user_key, &largest_user_key)) {
        break;
      }
534
      if (level + 2 >= vset_->NumberLevels()) {
535 536
        level++;
        break;
537
      }
G
Gabor Cselle 已提交
538 539
      GetOverlappingInputs(level + 2, &start, &limit, &overlaps);
      const int64_t sum = TotalFileSize(overlaps);
540
      if (sum > vset_->MaxGrandParentOverlapBytes(level)) {
G
Gabor Cselle 已提交
541 542 543 544 545
        break;
      }
      level++;
    }
  }
546

G
Gabor Cselle 已提交
547 548 549 550
  return level;
}

// Store in "*inputs" all files in "level" that overlap [begin,end]
A
Abhishek Kona 已提交
551
// If hint_index is specified, then it points to a file in the
552 553
// overlapping range.
// The file_index returns a pointer to any file in an overlapping range.
G
Gabor Cselle 已提交
554 555 556 557
void Version::GetOverlappingInputs(
    int level,
    const InternalKey* begin,
    const InternalKey* end,
558 559 560
    std::vector<FileMetaData*>* inputs,
    int hint_index,
    int* file_index) {
G
Gabor Cselle 已提交
561 562
  inputs->clear();
  Slice user_begin, user_end;
A
Abhishek Kona 已提交
563
  if (begin != nullptr) {
G
Gabor Cselle 已提交
564 565
    user_begin = begin->user_key();
  }
A
Abhishek Kona 已提交
566
  if (end != nullptr) {
G
Gabor Cselle 已提交
567 568
    user_end = end->user_key();
  }
569 570 571
  if (file_index) {
    *file_index = -1;
  }
G
Gabor Cselle 已提交
572
  const Comparator* user_cmp = vset_->icmp_.user_comparator();
A
Abhishek Kona 已提交
573
  if (begin != nullptr && end != nullptr && level > 0) {
574 575
    GetOverlappingInputsBinarySearch(level, user_begin, user_end, inputs,
      hint_index, file_index);
576 577
    return;
  }
H
Hans Wennborg 已提交
578 579 580 581
  for (size_t i = 0; i < files_[level].size(); ) {
    FileMetaData* f = files_[level][i++];
    const Slice file_start = f->smallest.user_key();
    const Slice file_limit = f->largest.user_key();
A
Abhishek Kona 已提交
582
    if (begin != nullptr && user_cmp->Compare(file_limit, user_begin) < 0) {
G
Gabor Cselle 已提交
583
      // "f" is completely before specified range; skip it
A
Abhishek Kona 已提交
584
    } else if (end != nullptr && user_cmp->Compare(file_start, user_end) > 0) {
G
Gabor Cselle 已提交
585 586 587
      // "f" is completely after specified range; skip it
    } else {
      inputs->push_back(f);
H
Hans Wennborg 已提交
588 589 590
      if (level == 0) {
        // Level-0 files may overlap each other.  So check if the newly
        // added file has expanded the range.  If so, restart search.
A
Abhishek Kona 已提交
591
        if (begin != nullptr && user_cmp->Compare(file_start, user_begin) < 0) {
H
Hans Wennborg 已提交
592 593 594
          user_begin = file_start;
          inputs->clear();
          i = 0;
A
Abhishek Kona 已提交
595 596
        } else if (end != nullptr
            && user_cmp->Compare(file_limit, user_end) > 0) {
H
Hans Wennborg 已提交
597 598 599 600
          user_end = file_limit;
          inputs->clear();
          i = 0;
        }
601
      } else if (file_index) {
602
        *file_index = i-1;
H
Hans Wennborg 已提交
603
      }
G
Gabor Cselle 已提交
604 605
    }
  }
606 607
}

608 609 610 611 612 613 614 615
// Store in "*inputs" all files in "level" that overlap [begin,end]
// Employ binary search to find at least one file that overlaps the
// specified range. From that file, iterate backwards and
// forwards to find all overlapping files.
void Version::GetOverlappingInputsBinarySearch(
    int level,
    const Slice& user_begin,
    const Slice& user_end,
616 617 618
    std::vector<FileMetaData*>* inputs,
    int hint_index,
    int* file_index) {
619 620 621 622 623 624
  assert(level > 0);
  int min = 0;
  int mid = 0;
  int max = files_[level].size() -1;
  bool foundOverlap = false;
  const Comparator* user_cmp = vset_->icmp_.user_comparator();
625 626 627 628 629 630 631 632 633

  // if the caller already knows the index of a file that has overlap,
  // then we can skip the binary search.
  if (hint_index != -1) {
    mid = hint_index;
    foundOverlap = true;
  }

  while (!foundOverlap && min <= max) {
634 635 636 637 638 639 640 641 642 643 644 645 646
    mid = (min + max)/2;
    FileMetaData* f = files_[level][mid];
    const Slice file_start = f->smallest.user_key();
    const Slice file_limit = f->largest.user_key();
    if (user_cmp->Compare(file_limit, user_begin) < 0) {
      min = mid + 1;
    } else if (user_cmp->Compare(user_end, file_start) < 0) {
      max = mid - 1;
    } else {
      foundOverlap = true;
      break;
    }
  }
A
Abhishek Kona 已提交
647

648 649 650 651
  // If there were no overlapping files, return immediately.
  if (!foundOverlap) {
    return;
  }
652 653 654 655
  // returns the index where an overlap is found
  if (file_index) {
    *file_index = mid;
  }
656 657
  ExtendOverlappingInputs(level, user_begin, user_end, inputs, mid);
}
A
Abhishek Kona 已提交
658

659 660 661 662 663 664 665 666 667
// Store in "*inputs" all files in "level" that overlap [begin,end]
// The midIndex specifies the index of at least one file that
// overlaps the specified range. From that file, iterate backward
// and forward to find all overlapping files.
void Version::ExtendOverlappingInputs(
    int level,
    const Slice& user_begin,
    const Slice& user_end,
    std::vector<FileMetaData*>* inputs,
668
    unsigned int midIndex) {
669 670

  const Comparator* user_cmp = vset_->icmp_.user_comparator();
671 672 673 674 675 676 677 678 679 680 681 682 683 684
#ifndef NDEBUG
  {
    // assert that the file at midIndex overlaps with the range
    assert(midIndex < files_[level].size());
    FileMetaData* f = files_[level][midIndex];
    const Slice fstart = f->smallest.user_key();
    const Slice flimit = f->largest.user_key();
    if (user_cmp->Compare(fstart, user_begin) >= 0) {
      assert(user_cmp->Compare(fstart, user_end) <= 0);
    } else {
      assert(user_cmp->Compare(flimit, user_begin) >= 0);
    }
  }
#endif
685 686
  int startIndex = midIndex + 1;
  int endIndex = midIndex;
687
  int count __attribute__((unused)) = 0;
688 689

  // check backwards from 'mid' to lower indices
690
  for (int i = midIndex; i >= 0 ; i--) {
691 692 693
    FileMetaData* f = files_[level][i];
    const Slice file_limit = f->largest.user_key();
    if (user_cmp->Compare(file_limit, user_begin) >= 0) {
694 695
      startIndex = i;
      assert((count++, true));
696 697 698 699 700
    } else {
      break;
    }
  }
  // check forward from 'mid+1' to higher indices
701
  for (unsigned int i = midIndex+1; i < files_[level].size(); i++) {
702 703 704
    FileMetaData* f = files_[level][i];
    const Slice file_start = f->smallest.user_key();
    if (user_cmp->Compare(file_start, user_end) <= 0) {
705 706
      assert((count++, true));
      endIndex = i;
707 708 709 710
    } else {
      break;
    }
  }
711 712 713 714 715
  assert(count == endIndex - startIndex + 1);

  // insert overlapping files into vector
  for (int i = startIndex; i <= endIndex; i++) {
    FileMetaData* f = files_[level][i];
716
    inputs->push_back(f);
717
  }
718 719
}

Z
Zheng Shao 已提交
720
std::string Version::DebugString(bool hex) const {
J
jorlow@chromium.org 已提交
721
  std::string r;
722
  for (int level = 0; level < vset_->NumberLevels(); level++) {
723 724 725 726 727
    // E.g.,
    //   --- level 1 ---
    //   17:123['a' .. 'd']
    //   20:43['e' .. 'g']
    r.append("--- level ");
J
jorlow@chromium.org 已提交
728
    AppendNumberTo(&r, level);
729 730
    r.append(" --- version# ");
    AppendNumberTo(&r, version_number_);
731
    r.append(" ---\n");
J
jorlow@chromium.org 已提交
732
    const std::vector<FileMetaData*>& files = files_[level];
D
dgrogan@chromium.org 已提交
733
    for (size_t i = 0; i < files.size(); i++) {
J
jorlow@chromium.org 已提交
734 735 736 737
      r.push_back(' ');
      AppendNumberTo(&r, files[i]->number);
      r.push_back(':');
      AppendNumberTo(&r, files[i]->file_size);
G
Gabor Cselle 已提交
738
      r.append("[");
Z
Zheng Shao 已提交
739
      r.append(files[i]->smallest.DebugString(hex));
G
Gabor Cselle 已提交
740
      r.append(" .. ");
Z
Zheng Shao 已提交
741
      r.append(files[i]->largest.DebugString(hex));
G
Gabor Cselle 已提交
742
      r.append("]\n");
J
jorlow@chromium.org 已提交
743 744 745 746 747
    }
  }
  return r;
}

748 749 750 751 752 753
// this is used to batch writes to the manifest file
struct VersionSet::ManifestWriter {
  Status status;
  bool done;
  port::CondVar cv;
  VersionEdit* edit;
A
Abhishek Kona 已提交
754 755

  explicit ManifestWriter(port::Mutex* mu, VersionEdit* e) :
756 757 758
             done(false), cv(mu), edit(e) {}
};

J
jorlow@chromium.org 已提交
759 760 761 762 763
// A helper class so we can efficiently apply a whole sequence
// of edits to a particular state without creating intermediate
// Versions that contain full copies of the intermediate state.
class VersionSet::Builder {
 private:
764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784
  // Helper to sort by v->files_[file_number].smallest
  struct BySmallestKey {
    const InternalKeyComparator* internal_comparator;

    bool operator()(FileMetaData* f1, FileMetaData* f2) const {
      int r = internal_comparator->Compare(f1->smallest, f2->smallest);
      if (r != 0) {
        return (r < 0);
      } else {
        // Break ties by file number
        return (f1->number < f2->number);
      }
    }
  };

  typedef std::set<FileMetaData*, BySmallestKey> FileSet;
  struct LevelState {
    std::set<uint64_t> deleted_files;
    FileSet* added_files;
  };

J
jorlow@chromium.org 已提交
785
  VersionSet* vset_;
786
  Version* base_;
787
  LevelState* levels_;
J
jorlow@chromium.org 已提交
788 789 790 791

 public:
  // Initialize a builder with the files from *base and other info from *vset
  Builder(VersionSet* vset, Version* base)
792 793 794
      : vset_(vset),
        base_(base) {
    base_->Ref();
795
    levels_ = new LevelState[vset_->NumberLevels()];
796 797
    BySmallestKey cmp;
    cmp.internal_comparator = &vset_->icmp_;
798
    for (int level = 0; level < vset_->NumberLevels(); level++) {
799
      levels_[level].added_files = new FileSet(cmp);
J
jorlow@chromium.org 已提交
800 801 802 803
    }
  }

  ~Builder() {
804
    for (int level = 0; level < vset_->NumberLevels(); level++) {
805 806 807 808 809 810 811 812
      const FileSet* added = levels_[level].added_files;
      std::vector<FileMetaData*> to_unref;
      to_unref.reserve(added->size());
      for (FileSet::const_iterator it = added->begin();
          it != added->end(); ++it) {
        to_unref.push_back(*it);
      }
      delete added;
D
dgrogan@chromium.org 已提交
813
      for (uint32_t i = 0; i < to_unref.size(); i++) {
814
        FileMetaData* f = to_unref[i];
J
jorlow@chromium.org 已提交
815 816 817 818 819 820
        f->refs--;
        if (f->refs <= 0) {
          delete f;
        }
      }
    }
821
    delete[] levels_;
822
    base_->Unref();
J
jorlow@chromium.org 已提交
823 824
  }

825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844
  void CheckConsistency(Version* v) {
#ifndef NDEBUG
    for (int level = 0; level < vset_->NumberLevels(); level++) {
      // Make sure there is no overlap in levels > 0
      if (level > 0) {
        for (uint32_t i = 1; i < v->files_[level].size(); i++) {
          const InternalKey& prev_end = v->files_[level][i-1]->largest;
          const InternalKey& this_begin = v->files_[level][i]->smallest;
          if (vset_->icmp_.Compare(prev_end, this_begin) >= 0) {
            fprintf(stderr, "overlapping ranges in same level %s vs. %s\n",
                    prev_end.DebugString().c_str(),
                    this_begin.DebugString().c_str());
            abort();
          }
        }
      }
    }
#endif
  }

845 846 847 848
  void CheckConsistencyForDeletes(
    VersionEdit* edit,
    unsigned int number,
    int level) {
849 850 851 852 853
#ifndef NDEBUG
      // a file to be deleted better exist in the previous version
      bool found = false;
      for (int l = 0; !found && l < edit->number_levels_; l++) {
        const std::vector<FileMetaData*>& base_files = base_->files_[l];
854
        for (unsigned int i = 0; i < base_files.size(); i++) {
855 856 857 858 859 860 861 862 863 864 865 866
          FileMetaData* f = base_files[i];
          if (f->number == number) {
            found =  true;
            break;
          }
        }
      }
      // if the file did not exist in the previous version, then it
      // is possibly moved from lower level to higher level in current
      // version
      for (int l = level+1; !found && l < edit->number_levels_; l++) {
        const FileSet* added = levels_[l].added_files;
D
Dhruba Borthakur 已提交
867 868 869 870 871 872 873 874 875 876 877 878 879
        for (FileSet::const_iterator added_iter = added->begin();
             added_iter != added->end(); ++added_iter) {
          FileMetaData* f = *added_iter;
          if (f->number == number) {
            found = true;
            break;
          }
        }
      }

      // maybe this file was added in a previous edit that was Applied
      if (!found) {
        const FileSet* added = levels_[level].added_files;
880 881 882 883 884 885 886 887 888 889 890 891 892
        for (FileSet::const_iterator added_iter = added->begin();
             added_iter != added->end(); ++added_iter) {
          FileMetaData* f = *added_iter;
          if (f->number == number) {
            found = true;
            break;
          }
        }
      }
      assert(found);
#endif
  }

J
jorlow@chromium.org 已提交
893 894
  // Apply all of the edits in *edit to the current state.
  void Apply(VersionEdit* edit) {
895 896
    CheckConsistency(base_);

J
jorlow@chromium.org 已提交
897
    // Update compaction pointers
D
dgrogan@chromium.org 已提交
898
    for (size_t i = 0; i < edit->compact_pointers_.size(); i++) {
J
jorlow@chromium.org 已提交
899 900 901 902 903 904 905 906 907 908 909 910
      const int level = edit->compact_pointers_[i].first;
      vset_->compact_pointer_[level] =
          edit->compact_pointers_[i].second.Encode().ToString();
    }

    // Delete files
    const VersionEdit::DeletedFileSet& del = edit->deleted_files_;
    for (VersionEdit::DeletedFileSet::const_iterator iter = del.begin();
         iter != del.end();
         ++iter) {
      const int level = iter->first;
      const uint64_t number = iter->second;
911
      levels_[level].deleted_files.insert(number);
912
      CheckConsistencyForDeletes(edit, number, level);
J
jorlow@chromium.org 已提交
913 914 915
    }

    // Add new files
D
dgrogan@chromium.org 已提交
916
    for (size_t i = 0; i < edit->new_files_.size(); i++) {
J
jorlow@chromium.org 已提交
917 918 919
      const int level = edit->new_files_[i].first;
      FileMetaData* f = new FileMetaData(edit->new_files_[i].second);
      f->refs = 1;
920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936

      // We arrange to automatically compact this file after
      // a certain number of seeks.  Let's assume:
      //   (1) One seek costs 10ms
      //   (2) Writing or reading 1MB costs 10ms (100MB/s)
      //   (3) A compaction of 1MB does 25MB of IO:
      //         1MB read from this level
      //         10-12MB read from next level (boundaries may be misaligned)
      //         10-12MB written to next level
      // This implies that 25 seeks cost the same as the compaction
      // of 1MB of data.  I.e., one seek costs approximately the
      // same as the compaction of 40KB of data.  We are a little
      // conservative and allow approximately one seek for every 16KB
      // of data before triggering a compaction.
      f->allowed_seeks = (f->file_size / 16384);
      if (f->allowed_seeks < 100) f->allowed_seeks = 100;

937 938
      levels_[level].deleted_files.erase(f->number);
      levels_[level].added_files->insert(f);
J
jorlow@chromium.org 已提交
939 940 941 942 943
    }
  }

  // Save the current state in *v.
  void SaveTo(Version* v) {
944 945
    CheckConsistency(base_);
    CheckConsistency(v);
946 947
    BySmallestKey cmp;
    cmp.internal_comparator = &vset_->icmp_;
948
    for (int level = 0; level < vset_->NumberLevels(); level++) {
949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972
      // Merge the set of added files with the set of pre-existing files.
      // Drop any deleted files.  Store the result in *v.
      const std::vector<FileMetaData*>& base_files = base_->files_[level];
      std::vector<FileMetaData*>::const_iterator base_iter = base_files.begin();
      std::vector<FileMetaData*>::const_iterator base_end = base_files.end();
      const FileSet* added = levels_[level].added_files;
      v->files_[level].reserve(base_files.size() + added->size());
      for (FileSet::const_iterator added_iter = added->begin();
           added_iter != added->end();
           ++added_iter) {
        // Add all smaller files listed in base_
        for (std::vector<FileMetaData*>::const_iterator bpos
                 = std::upper_bound(base_iter, base_end, *added_iter, cmp);
             base_iter != bpos;
             ++base_iter) {
          MaybeAddFile(v, level, *base_iter);
        }

        MaybeAddFile(v, level, *added_iter);
      }

      // Add remaining base files
      for (; base_iter != base_end; ++base_iter) {
        MaybeAddFile(v, level, *base_iter);
J
jorlow@chromium.org 已提交
973
      }
974
    }
975
    CheckConsistency(v);
976 977 978 979 980 981
  }

  void MaybeAddFile(Version* v, int level, FileMetaData* f) {
    if (levels_[level].deleted_files.count(f->number) > 0) {
      // File is deleted: do nothing
    } else {
982 983 984 985 986 987
      std::vector<FileMetaData*>* files = &v->files_[level];
      if (level > 0 && !files->empty()) {
        // Must not overlap
        assert(vset_->icmp_.Compare((*files)[files->size()-1]->largest,
                                    f->smallest) < 0);
      }
988
      f->refs++;
989
      files->push_back(f);
J
jorlow@chromium.org 已提交
990 991 992 993 994 995
    }
  }
};

VersionSet::VersionSet(const std::string& dbname,
                       const Options* options,
H
Haobo Xu 已提交
996
                       const EnvOptions& storage_options,
J
jorlow@chromium.org 已提交
997 998 999 1000 1001 1002 1003 1004 1005
                       TableCache* table_cache,
                       const InternalKeyComparator* cmp)
    : env_(options->env),
      dbname_(dbname),
      options_(options),
      table_cache_(table_cache),
      icmp_(*cmp),
      next_file_number_(2),
      manifest_file_number_(0),  // Filled by Recover()
1006 1007 1008
      last_sequence_(0),
      log_number_(0),
      prev_log_number_(0),
1009
      num_levels_(options_->num_levels),
1010
      dummy_versions_(this),
A
Abhishek Kona 已提交
1011
      current_(nullptr),
1012
      compactions_in_progress_(options_->num_levels),
A
Abhishek Kona 已提交
1013
      current_version_number_(0),
1014 1015 1016
      last_observed_manifest_size_(0),
      storage_options_(storage_options),
      storage_options_compactions_(storage_options_)  {
1017
  compact_pointer_ = new std::string[options_->num_levels];
1018
  Init(options_->num_levels);
1019
  AppendVersion(new Version(this, current_version_number_++));
J
jorlow@chromium.org 已提交
1020 1021 1022
}

VersionSet::~VersionSet() {
1023 1024
  current_->Unref();
  assert(dummy_versions_.next_ == &dummy_versions_);  // List must be empty
1025 1026 1027
  delete[] compact_pointer_;
  delete[] max_file_size_;
  delete[] level_max_bytes_;
J
jorlow@chromium.org 已提交
1028 1029
}

1030 1031 1032 1033 1034 1035 1036 1037
void VersionSet::Init(int num_levels) {
  max_file_size_ = new uint64_t[num_levels];
  level_max_bytes_ = new uint64_t[num_levels];
  int target_file_size_multiplier = options_->target_file_size_multiplier;
  int max_bytes_multiplier = options_->max_bytes_for_level_multiplier;
  for (int i = 0; i < num_levels; i++) {
    if (i > 1) {
      max_file_size_[i] = max_file_size_[i-1] * target_file_size_multiplier;
1038 1039
      level_max_bytes_[i] = level_max_bytes_[i-1] * max_bytes_multiplier *
        options_->max_bytes_for_level_multiplier_additional[i-1];
1040 1041 1042 1043 1044 1045 1046
    } else {
      max_file_size_[i] = options_->target_file_size_base;
      level_max_bytes_[i] = options_->max_bytes_for_level_base;
    }
  }
}

1047 1048 1049 1050
void VersionSet::AppendVersion(Version* v) {
  // Make "v" current
  assert(v->refs_ == 0);
  assert(v != current_);
A
Abhishek Kona 已提交
1051
  if (current_ != nullptr) {
1052
    assert(current_->refs_ > 0);
1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064
    current_->Unref();
  }
  current_ = v;
  v->Ref();

  // Append to linked list
  v->prev_ = dummy_versions_.prev_;
  v->next_ = &dummy_versions_;
  v->prev_->next_ = v;
  v->next_->prev_ = v;
}

1065 1066
Status VersionSet::LogAndApply(VersionEdit* edit, port::Mutex* mu,
    bool new_descriptor_log) {
1067
  mu->AssertHeld();
1068

1069 1070 1071 1072 1073
  // queue our request
  ManifestWriter w(mu, edit);
  manifest_writers_.push_back(&w);
  while (!w.done && &w != manifest_writers_.front()) {
    w.cv.Wait();
1074
  }
1075 1076 1077
  if (w.done) {
    return w.status;
  }
A
Abhishek Kona 已提交
1078

1079 1080 1081
  std::vector<VersionEdit*> batch_edits;
  Version* v = new Version(this, current_version_number_++);
  Builder builder(this, current_);
1082

1083 1084 1085
  // process all requests in the queue
  ManifestWriter* last_writer = &w;
  assert(!manifest_writers_.empty());
1086
  assert(manifest_writers_.front() == &w);
1087 1088 1089
  std::deque<ManifestWriter*>::iterator iter = manifest_writers_.begin();
  for (; iter != manifest_writers_.end(); ++iter) {
    last_writer = *iter;
A
Abhishek Kona 已提交
1090
    LogAndApplyHelper(&builder, v, last_writer->edit, mu);
1091
    batch_edits.push_back(last_writer->edit);
J
jorlow@chromium.org 已提交
1092
  }
1093
  builder.SaveTo(v);
J
jorlow@chromium.org 已提交
1094 1095 1096

  // Initialize new descriptor log file if necessary by creating
  // a temporary file that contains a snapshot of the current version.
1097
  std::string new_manifest_file;
1098
  uint64_t new_manifest_file_size = 0;
1099
  Status s;
A
Abhishek Kona 已提交
1100 1101

  //  No need to perform this check if a new Manifest is being created anyways.
1102 1103
  if (!descriptor_log_ ||
      last_observed_manifest_size_ > options_->max_manifest_file_size) {
A
Abhishek Kona 已提交
1104 1105 1106 1107
    new_descriptor_log = true;
    manifest_file_number_ = NewFileNumber(); // Change manifest file no.
  }

1108
  if (!descriptor_log_ || new_descriptor_log) {
1109 1110
    // No reason to unlock *mu here since we only hit this path in the
    // first call to LogAndApply (when opening the database).
1111
    assert(!descriptor_log_ || new_descriptor_log);
1112 1113
    new_manifest_file = DescriptorFileName(dbname_, manifest_file_number_);
    edit->SetNextFile(next_file_number_);
1114
    unique_ptr<WritableFile> descriptor_file;
1115 1116
    s = env_->NewWritableFile(new_manifest_file, &descriptor_file,
                              storage_options_);
1117
    if (s.ok()) {
1118 1119
      descriptor_log_.reset(new log::Writer(std::move(descriptor_file)));
      s = WriteSnapshot(descriptor_log_.get());
J
jorlow@chromium.org 已提交
1120 1121 1122
    }
  }

1123 1124
  // Unlock during expensive MANIFEST log write. New writes cannot get here
  // because &w is ensuring that all new writes get queued.
1125
  {
1126 1127 1128 1129
    // calculate the amount of data being compacted at every level
    std::vector<uint64_t> size_being_compacted(NumberLevels()-1);
    SizeBeingCompacted(size_being_compacted);

1130
    mu->Unlock();
1131

1132
    // The calls to Finalize and UpdateFilesBySize are cpu-heavy
1133
    // and is best called outside the mutex.
1134
    Finalize(v, size_being_compacted);
1135
    UpdateFilesBySize(v);
1136 1137

    // Write new record to MANIFEST log
J
jorlow@chromium.org 已提交
1138
    if (s.ok()) {
1139
      std::string record;
1140 1141 1142 1143 1144 1145 1146
      for (unsigned int i = 0; i < batch_edits.size(); i++) {
        batch_edits[i]->EncodeTo(&record);
        s = descriptor_log_->AddRecord(record);
        if (!s.ok()) {
          break;
        }
      }
1147
      if (s.ok()) {
1148
        if (options_->use_fsync) {
1149
          StopWatch sw(env_, options_->statistics, MANIFEST_FILE_SYNC_MICROS);
1150
          s = descriptor_log_->file()->Fsync();
1151
        } else {
1152
          StopWatch sw(env_, options_->statistics, MANIFEST_FILE_SYNC_MICROS);
1153
          s = descriptor_log_->file()->Sync();
1154
        }
1155
      }
1156 1157 1158 1159 1160
      if (!s.ok()) {
        Log(options_->info_log, "MANIFEST write: %s\n", s.ToString().c_str());
        if (ManifestContains(record)) {
          Log(options_->info_log,
              "MANIFEST contains log record despite error; advancing to new "
1161 1162
              "version to prevent mismatch between in-memory and logged state"
              " If paranoid is set, then the db is now in readonly mode.");
1163 1164 1165
          s = Status::OK();
        }
      }
J
jorlow@chromium.org 已提交
1166 1167
    }

1168 1169 1170 1171 1172 1173
    // If we just created a new descriptor file, install it by writing a
    // new CURRENT file that points to it.
    if (s.ok() && !new_manifest_file.empty()) {
      s = SetCurrentFile(env_, dbname_, manifest_file_number_);
    }

1174
    // find offset in manifest file where this version is stored.
1175
    new_manifest_file_size = descriptor_log_->file()->GetFileSize();
A
Abhishek Kona 已提交
1176

1177
    mu->Lock();
A
Abhishek Kona 已提交
1178 1179 1180
    // cache the manifest_file_size so that it can be used to rollover in the
    // next call to LogAndApply
    last_observed_manifest_size_ = new_manifest_file_size;
J
jorlow@chromium.org 已提交
1181 1182 1183 1184
  }

  // Install the new version
  if (s.ok()) {
1185
    v->offset_manifest_file_ = new_manifest_file_size;
1186
    AppendVersion(v);
1187 1188
    log_number_ = edit->log_number_;
    prev_log_number_ = edit->prev_log_number_;
1189

J
jorlow@chromium.org 已提交
1190
  } else {
A
Abhishek Kona 已提交
1191
    Log(options_->info_log, "Error in committing version %ld",
1192
        v->GetVersionNumber());
J
jorlow@chromium.org 已提交
1193 1194
    delete v;
    if (!new_manifest_file.empty()) {
1195
      descriptor_log_.reset();
J
jorlow@chromium.org 已提交
1196 1197 1198 1199
      env_->DeleteFile(new_manifest_file);
    }
  }

1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214
  // wake up all the waiting writers
  while (true) {
    ManifestWriter* ready = manifest_writers_.front();
    manifest_writers_.pop_front();
    if (ready != &w) {
      ready->status = s;
      ready->done = true;
      ready->cv.Signal();
    }
    if (ready == last_writer) break;
  }
  // Notify new head of write queue
  if (!manifest_writers_.empty()) {
    manifest_writers_.front()->cv.Signal();
  }
J
jorlow@chromium.org 已提交
1215 1216 1217
  return s;
}

1218 1219 1220 1221 1222 1223 1224 1225 1226 1227 1228 1229 1230 1231 1232 1233 1234 1235 1236 1237 1238
void VersionSet::LogAndApplyHelper(Builder* builder, Version* v,
  VersionEdit* edit, port::Mutex* mu) {
  mu->AssertHeld();

  if (edit->has_log_number_) {
    assert(edit->log_number_ >= log_number_);
    assert(edit->log_number_ < next_file_number_);
  } else {
    edit->SetLogNumber(log_number_);
  }

  if (!edit->has_prev_log_number_) {
    edit->SetPrevLogNumber(prev_log_number_);
  }

  edit->SetNextFile(next_file_number_);
  edit->SetLastSequence(last_sequence_);

  builder->Apply(edit);
}

1239
Status VersionSet::Recover() {
J
jorlow@chromium.org 已提交
1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257
  struct LogReporter : public log::Reader::Reporter {
    Status* status;
    virtual void Corruption(size_t bytes, const Status& s) {
      if (this->status->ok()) *this->status = s;
    }
  };

  // Read "CURRENT" file, which contains a pointer to the current manifest file
  std::string current;
  Status s = ReadFileToString(env_, CurrentFileName(dbname_), &current);
  if (!s.ok()) {
    return s;
  }
  if (current.empty() || current[current.size()-1] != '\n') {
    return Status::Corruption("CURRENT file does not end with newline");
  }
  current.resize(current.size() - 1);

H
heyongqiang 已提交
1258 1259 1260
  Log(options_->info_log, "Recovering from manifest file:%s\n",
      current.c_str());

J
jorlow@chromium.org 已提交
1261
  std::string dscname = dbname_ + "/" + current;
1262
  unique_ptr<SequentialFile> file;
1263
  s = env_->NewSequentialFile(dscname, &file, storage_options_);
J
jorlow@chromium.org 已提交
1264 1265 1266
  if (!s.ok()) {
    return s;
  }
1267 1268 1269 1270 1271
  uint64_t manifest_file_size;
  s = env_->GetFileSize(dscname, &manifest_file_size);
  if (!s.ok()) {
    return s;
  }
J
jorlow@chromium.org 已提交
1272 1273

  bool have_log_number = false;
1274
  bool have_prev_log_number = false;
J
jorlow@chromium.org 已提交
1275 1276 1277
  bool have_next_file = false;
  bool have_last_sequence = false;
  uint64_t next_file = 0;
1278 1279 1280
  uint64_t last_sequence = 0;
  uint64_t log_number = 0;
  uint64_t prev_log_number = 0;
J
jorlow@chromium.org 已提交
1281 1282 1283 1284 1285
  Builder builder(this, current_);

  {
    LogReporter reporter;
    reporter.status = &s;
1286 1287
    log::Reader reader(std::move(file), &reporter, true/*checksum*/,
                       0/*initial_offset*/);
J
jorlow@chromium.org 已提交
1288 1289 1290
    Slice record;
    std::string scratch;
    while (reader.ReadRecord(&record, &scratch) && s.ok()) {
1291
      VersionEdit edit(NumberLevels());
J
jorlow@chromium.org 已提交
1292 1293 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 1305 1306
      s = edit.DecodeFrom(record);
      if (s.ok()) {
        if (edit.has_comparator_ &&
            edit.comparator_ != icmp_.user_comparator()->Name()) {
          s = Status::InvalidArgument(
              edit.comparator_ + "does not match existing comparator ",
              icmp_.user_comparator()->Name());
        }
      }

      if (s.ok()) {
        builder.Apply(&edit);
      }

      if (edit.has_log_number_) {
1307
        log_number = edit.log_number_;
J
jorlow@chromium.org 已提交
1308 1309 1310
        have_log_number = true;
      }

1311 1312 1313 1314 1315
      if (edit.has_prev_log_number_) {
        prev_log_number = edit.prev_log_number_;
        have_prev_log_number = true;
      }

J
jorlow@chromium.org 已提交
1316 1317 1318 1319 1320 1321
      if (edit.has_next_file_number_) {
        next_file = edit.next_file_number_;
        have_next_file = true;
      }

      if (edit.has_last_sequence_) {
1322
        last_sequence = edit.last_sequence_;
J
jorlow@chromium.org 已提交
1323 1324 1325 1326
        have_last_sequence = true;
      }
    }
  }
1327
  file.reset();
J
jorlow@chromium.org 已提交
1328 1329 1330 1331 1332 1333 1334 1335 1336

  if (s.ok()) {
    if (!have_next_file) {
      s = Status::Corruption("no meta-nextfile entry in descriptor");
    } else if (!have_log_number) {
      s = Status::Corruption("no meta-lognumber entry in descriptor");
    } else if (!have_last_sequence) {
      s = Status::Corruption("no last-sequence-number entry in descriptor");
    }
1337 1338 1339 1340

    if (!have_prev_log_number) {
      prev_log_number = 0;
    }
1341 1342 1343

    MarkFileNumberUsed(prev_log_number);
    MarkFileNumberUsed(log_number);
J
jorlow@chromium.org 已提交
1344 1345 1346
  }

  if (s.ok()) {
1347
    Version* v = new Version(this, current_version_number_++);
J
jorlow@chromium.org 已提交
1348
    builder.SaveTo(v);
1349

1350
    // Install recovered version
1351 1352 1353 1354
    std::vector<uint64_t> size_being_compacted(NumberLevels()-1);
    SizeBeingCompacted(size_being_compacted);
    Finalize(v, size_being_compacted);

1355
    v->offset_manifest_file_ = manifest_file_size;
1356 1357 1358 1359 1360 1361
    AppendVersion(v);
    manifest_file_number_ = next_file;
    next_file_number_ = next_file + 1;
    last_sequence_ = last_sequence;
    log_number_ = log_number;
    prev_log_number_ = prev_log_number;
H
heyongqiang 已提交
1362

H
heyongqiang 已提交
1363
    Log(options_->info_log, "Recovered from manifest file:%s succeeded,"
1364 1365 1366
        "manifest_file_number is %ld, next_file_number is %ld, "
        "last_sequence is %ld, log_number is %ld,"
        "prev_log_number is %ld\n",
H
heyongqiang 已提交
1367 1368
        current.c_str(), manifest_file_number_, next_file_number_,
        last_sequence_, log_number_, prev_log_number_);
J
jorlow@chromium.org 已提交
1369 1370 1371 1372 1373
  }

  return s;
}

1374
Status VersionSet::DumpManifest(Options& options, std::string& dscname,
Z
Zheng Shao 已提交
1375
    bool verbose, bool hex) {
1376 1377 1378 1379 1380 1381 1382 1383
  struct LogReporter : public log::Reader::Reporter {
    Status* status;
    virtual void Corruption(size_t bytes, const Status& s) {
      if (this->status->ok()) *this->status = s;
    }
  };

  // Open the specified manifest file.
1384
  unique_ptr<SequentialFile> file;
1385
  Status s = options.env->NewSequentialFile(dscname, &file, storage_options_);
1386 1387 1388 1389 1390 1391 1392 1393 1394 1395 1396 1397
  if (!s.ok()) {
    return s;
  }

  bool have_log_number = false;
  bool have_prev_log_number = false;
  bool have_next_file = false;
  bool have_last_sequence = false;
  uint64_t next_file = 0;
  uint64_t last_sequence = 0;
  uint64_t log_number = 0;
  uint64_t prev_log_number = 0;
1398
  int count = 0;
1399 1400 1401 1402 1403
  VersionSet::Builder builder(this, current_);

  {
    LogReporter reporter;
    reporter.status = &s;
1404 1405
    log::Reader reader(std::move(file), &reporter, true/*checksum*/,
                       0/*initial_offset*/);
1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416 1417 1418 1419
    Slice record;
    std::string scratch;
    while (reader.ReadRecord(&record, &scratch) && s.ok()) {
      VersionEdit edit(NumberLevels());
      s = edit.DecodeFrom(record);
      if (s.ok()) {
        if (edit.has_comparator_ &&
            edit.comparator_ != icmp_.user_comparator()->Name()) {
          s = Status::InvalidArgument(
              edit.comparator_ + "does not match existing comparator ",
              icmp_.user_comparator()->Name());
        }
      }

1420 1421
      // Write out each individual edit
      if (verbose) {
A
Abhishek Kona 已提交
1422
        printf("*************************Edit[%d] = %s\n",
1423 1424 1425 1426
                count, edit.DebugString().c_str());
      }
      count++;

1427 1428 1429 1430 1431 1432 1433 1434 1435 1436 1437 1438 1439 1440 1441 1442 1443 1444 1445 1446 1447 1448 1449 1450 1451
      if (s.ok()) {
        builder.Apply(&edit);
      }

      if (edit.has_log_number_) {
        log_number = edit.log_number_;
        have_log_number = true;
      }

      if (edit.has_prev_log_number_) {
        prev_log_number = edit.prev_log_number_;
        have_prev_log_number = true;
      }

      if (edit.has_next_file_number_) {
        next_file = edit.next_file_number_;
        have_next_file = true;
      }

      if (edit.has_last_sequence_) {
        last_sequence = edit.last_sequence_;
        have_last_sequence = true;
      }
    }
  }
1452
  file.reset();
1453 1454 1455 1456 1457 1458 1459 1460 1461 1462 1463 1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474

  if (s.ok()) {
    if (!have_next_file) {
      s = Status::Corruption("no meta-nextfile entry in descriptor");
      printf("no meta-nextfile entry in descriptor");
    } else if (!have_log_number) {
      s = Status::Corruption("no meta-lognumber entry in descriptor");
      printf("no meta-lognumber entry in descriptor");
    } else if (!have_last_sequence) {
      printf("no last-sequence-number entry in descriptor");
      s = Status::Corruption("no last-sequence-number entry in descriptor");
    }

    if (!have_prev_log_number) {
      prev_log_number = 0;
    }

    MarkFileNumberUsed(prev_log_number);
    MarkFileNumberUsed(log_number);
  }

  if (s.ok()) {
1475
    Version* v = new Version(this, 0);
1476
    builder.SaveTo(v);
1477

1478
    // Install recovered version
1479 1480 1481 1482
    std::vector<uint64_t> size_being_compacted(NumberLevels()-1);
    SizeBeingCompacted(size_being_compacted);
    Finalize(v, size_being_compacted);

1483 1484 1485 1486 1487 1488
    AppendVersion(v);
    manifest_file_number_ = next_file;
    next_file_number_ = next_file + 1;
    last_sequence_ = last_sequence;
    log_number_ = log_number;
    prev_log_number_ = prev_log_number;
1489

1490
    printf("manifest_file_number %ld next_file_number %ld last_sequence %ld log_number %ld  prev_log_number %ld\n",
1491
           manifest_file_number_, next_file_number_,
1492
           last_sequence, log_number, prev_log_number);
Z
Zheng Shao 已提交
1493
    printf("%s \n", v->DebugString(hex).c_str());
1494
  }
1495

1496 1497 1498
  return s;
}

1499 1500 1501 1502 1503 1504
void VersionSet::MarkFileNumberUsed(uint64_t number) {
  if (next_file_number_ <= number) {
    next_file_number_ = number + 1;
  }
}

1505 1506
void VersionSet::Finalize(Version* v,
  std::vector<uint64_t>& size_being_compacted) {
J
jorlow@chromium.org 已提交
1507

1508
  double max_score = 0;
1509 1510
  int max_score_level = 0;

1511
  for (int level = 0; level < NumberLevels()-1; level++) {
1512
    double score;
J
jorlow@chromium.org 已提交
1513
    if (level == 0) {
1514 1515 1516 1517 1518 1519 1520 1521 1522 1523 1524
      // We treat level-0 specially by bounding the number of files
      // instead of number of bytes for two reasons:
      //
      // (1) With larger write-buffer sizes, it is nice not to do too
      // many level-0 compactions.
      //
      // (2) The files in level-0 are merged on every read and
      // therefore we wish to avoid too many files when the individual
      // file size is small (perhaps because of a small write-buffer
      // setting, or very high compression ratios, or lots of
      // overwrites/deletions).
1525 1526 1527 1528 1529 1530 1531 1532 1533 1534 1535 1536 1537 1538 1539 1540
      int numfiles = 0;
      for (unsigned int i = 0; i < v->files_[level].size(); i++) {
        if (!v->files_[level][i]->being_compacted) {
          numfiles++;
        }
      }

      // If we are slowing down writes, then we better compact that first
      if (numfiles >= options_->level0_stop_writes_trigger) {
        score = 1000000;
        // Log(options_->info_log, "XXX score l0 = 1000000000 max");
      } else if (numfiles >= options_->level0_slowdown_writes_trigger) {
        score = 10000;
        // Log(options_->info_log, "XXX score l0 = 1000000 medium");
      } else {
        score = numfiles /
1541
          static_cast<double>(options_->level0_file_num_compaction_trigger);
1542 1543 1544 1545
        if (score >= 1) {
          // Log(options_->info_log, "XXX score l0 = %d least", (int)score);
        }
      }
1546 1547
    } else {
      // Compute the ratio of current size to size limit.
1548
      const uint64_t level_bytes = TotalFileSize(v->files_[level]) -
1549
                                   size_being_compacted[level];
1550
      score = static_cast<double>(level_bytes) / MaxBytesForLevel(level);
1551 1552 1553
      if (score > 1) {
        // Log(options_->info_log, "XXX score l%d = %d ", level, (int)score);
      }
1554 1555
      if (max_score < score) {
        max_score = score;
1556
        max_score_level = level;
1557
      }
J
jorlow@chromium.org 已提交
1558
    }
1559 1560 1561 1562
    v->compaction_level_[level] = level;
    v->compaction_score_[level] = score;
  }

1563 1564
  // update the max compaction score in levels 1 to n-1
  v->max_compaction_score_ = max_score;
1565
  v->max_compaction_score_level_ = max_score_level;
1566

1567 1568 1569
  // sort all the levels based on their score. Higher scores get listed
  // first. Use bubble sort because the number of entries are small.
  for(int i = 0; i <  NumberLevels()-2; i++) {
1570
    for (int j = i+1; j < NumberLevels()-1; j++) {
1571 1572 1573 1574 1575 1576 1577 1578
      if (v->compaction_score_[i] < v->compaction_score_[j]) {
        double score = v->compaction_score_[i];
        int level = v->compaction_level_[i];
        v->compaction_score_[i] = v->compaction_score_[j];
        v->compaction_level_[i] = v->compaction_level_[j];
        v->compaction_score_[j] = score;
        v->compaction_level_[j] = level;
      }
J
jorlow@chromium.org 已提交
1579 1580 1581 1582
    }
  }
}

1583
// a static compator used to sort files based on their size
A
Abhishek Kona 已提交
1584
static bool compareSize(const VersionSet::Fsize& first,
1585 1586 1587 1588 1589 1590 1591 1592 1593 1594 1595 1596 1597 1598 1599 1600 1601 1602 1603 1604 1605 1606 1607 1608 1609 1610 1611 1612 1613 1614 1615 1616 1617 1618 1619 1620 1621 1622 1623
  const VersionSet::Fsize& second) {
  return (first.file->file_size > second.file->file_size);
}

// sort all files in level1 to level(n-1) based on file size
void VersionSet::UpdateFilesBySize(Version* v) {

  // No need to sort the highest level because it is never compacted.
  for (int level = 0; level < NumberLevels()-1; level++) {

    const std::vector<FileMetaData*>& files = v->files_[level];
    std::vector<int>& files_by_size = v->files_by_size_[level];
    assert(files_by_size.size() == 0);

    // populate a temp vector for sorting based on size
    std::vector<Fsize> temp(files.size());
    for (unsigned int i = 0; i < files.size(); i++) {
      temp[i].index = i;
      temp[i].file = files[i];
    }

    // sort the top number_of_files_to_sort_ based on file size
    int num = Version::number_of_files_to_sort_;
    if (num > (int)temp.size()) {
      num = temp.size();
    }
    std::partial_sort(temp.begin(),  temp.begin() + num,
                      temp.end(), compareSize);
    assert(temp.size() == files.size());

    // initialize files_by_size_
    for (unsigned int i = 0; i < temp.size(); i++) {
      files_by_size.push_back(temp[i].index);
    }
    v->next_file_to_compact_by_size_[level] = 0;
    assert(v->files_[level].size() == v->files_by_size_[level].size());
  }
}

J
jorlow@chromium.org 已提交
1624 1625 1626 1627
Status VersionSet::WriteSnapshot(log::Writer* log) {
  // TODO: Break up into multiple records to reduce memory usage on recovery?

  // Save metadata
1628
  VersionEdit edit(NumberLevels());
J
jorlow@chromium.org 已提交
1629 1630 1631
  edit.SetComparatorName(icmp_.user_comparator()->Name());

  // Save compaction pointers
1632
  for (int level = 0; level < NumberLevels(); level++) {
J
jorlow@chromium.org 已提交
1633 1634 1635 1636 1637 1638 1639 1640
    if (!compact_pointer_[level].empty()) {
      InternalKey key;
      key.DecodeFrom(compact_pointer_[level]);
      edit.SetCompactPointer(level, key);
    }
  }

  // Save files
1641
  for (int level = 0; level < NumberLevels(); level++) {
J
jorlow@chromium.org 已提交
1642
    const std::vector<FileMetaData*>& files = current_->files_[level];
D
dgrogan@chromium.org 已提交
1643
    for (size_t i = 0; i < files.size(); i++) {
J
jorlow@chromium.org 已提交
1644 1645 1646 1647 1648 1649 1650 1651 1652 1653 1654 1655
      const FileMetaData* f = files[i];
      edit.AddFile(level, f->number, f->file_size, f->smallest, f->largest);
    }
  }

  std::string record;
  edit.EncodeTo(&record);
  return log->AddRecord(record);
}

int VersionSet::NumLevelFiles(int level) const {
  assert(level >= 0);
1656
  assert(level < NumberLevels());
J
jorlow@chromium.org 已提交
1657 1658 1659
  return current_->files_[level].size();
}

1660
const char* VersionSet::LevelSummary(LevelSummaryStorage* scratch) const {
1661 1662 1663 1664 1665 1666 1667 1668 1669 1670
  int len = snprintf(scratch->buffer, sizeof(scratch->buffer), "files[");
  for (int i = 0; i < NumberLevels(); i++) {
    int sz = sizeof(scratch->buffer) - len;
    int ret = snprintf(scratch->buffer + len, sz, "%d ",
        int(current_->files_[i].size()));
    if (ret < 0 || ret >= sz)
      break;
    len += ret;
  }
  snprintf(scratch->buffer + len, sizeof(scratch->buffer) - len, "]");
1671 1672 1673
  return scratch->buffer;
}

1674 1675 1676 1677 1678 1679 1680 1681 1682 1683 1684 1685 1686 1687 1688
const char* VersionSet::LevelDataSizeSummary(
    LevelSummaryStorage* scratch) const {
  int len = snprintf(scratch->buffer, sizeof(scratch->buffer), "files_size[");
  for (int i = 0; i < NumberLevels(); i++) {
    int sz = sizeof(scratch->buffer) - len;
    int ret = snprintf(scratch->buffer + len, sz, "%ld ",
        NumLevelBytes(i));
    if (ret < 0 || ret >= sz)
      break;
    len += ret;
  }
  snprintf(scratch->buffer + len, sizeof(scratch->buffer) - len, "]");
  return scratch->buffer;
}

1689 1690 1691 1692 1693
// Opens the mainfest file and reads all records
// till it finds the record we are looking for.
bool VersionSet::ManifestContains(const std::string& record) const {
  std::string fname = DescriptorFileName(dbname_, manifest_file_number_);
  Log(options_->info_log, "ManifestContains: checking %s\n", fname.c_str());
1694
  unique_ptr<SequentialFile> file;
1695
  Status s = env_->NewSequentialFile(fname, &file, storage_options_);
1696 1697
  if (!s.ok()) {
    Log(options_->info_log, "ManifestContains: %s\n", s.ToString().c_str());
1698 1699 1700
    Log(options_->info_log,
        "ManifestContains: is unable to reopen the manifest file  %s",
        fname.c_str());
1701 1702
    return false;
  }
A
Abhishek Kona 已提交
1703
  log::Reader reader(std::move(file), nullptr, true/*checksum*/, 0);
1704 1705 1706 1707 1708 1709 1710 1711 1712 1713 1714 1715 1716 1717
  Slice r;
  std::string scratch;
  bool result = false;
  while (reader.ReadRecord(&r, &scratch)) {
    if (r == Slice(record)) {
      result = true;
      break;
    }
  }
  Log(options_->info_log, "ManifestContains: result = %d\n", result ? 1 : 0);
  return result;
}


J
jorlow@chromium.org 已提交
1718 1719
uint64_t VersionSet::ApproximateOffsetOf(Version* v, const InternalKey& ikey) {
  uint64_t result = 0;
1720
  for (int level = 0; level < NumberLevels(); level++) {
J
jorlow@chromium.org 已提交
1721
    const std::vector<FileMetaData*>& files = v->files_[level];
D
dgrogan@chromium.org 已提交
1722
    for (size_t i = 0; i < files.size(); i++) {
J
jorlow@chromium.org 已提交
1723 1724 1725 1726 1727 1728 1729 1730 1731 1732 1733 1734 1735 1736 1737 1738
      if (icmp_.Compare(files[i]->largest, ikey) <= 0) {
        // Entire file is before "ikey", so just add the file size
        result += files[i]->file_size;
      } else if (icmp_.Compare(files[i]->smallest, ikey) > 0) {
        // Entire file is after "ikey", so ignore
        if (level > 0) {
          // Files other than level 0 are sorted by meta->smallest, so
          // no further files in this level will contain data for
          // "ikey".
          break;
        }
      } else {
        // "ikey" falls in the range for this table.  Add the
        // approximate offset of "ikey" within the table.
        Table* tableptr;
        Iterator* iter = table_cache_->NewIterator(
1739 1740
            ReadOptions(), storage_options_, files[i]->number,
            files[i]->file_size, &tableptr);
A
Abhishek Kona 已提交
1741
        if (tableptr != nullptr) {
J
jorlow@chromium.org 已提交
1742 1743 1744 1745 1746 1747 1748 1749 1750
          result += tableptr->ApproximateOffsetOf(ikey.Encode());
        }
        delete iter;
      }
    }
  }
  return result;
}

1751 1752 1753
void VersionSet::AddLiveFiles(std::vector<uint64_t>* live_list) {
  // pre-calculate space requirement
  int64_t total_files = 0;
1754 1755 1756
  for (Version* v = dummy_versions_.next_;
       v != &dummy_versions_;
       v = v->next_) {
1757
    for (int level = 0; level < NumberLevels(); level++) {
1758 1759 1760 1761 1762 1763 1764 1765 1766 1767 1768 1769 1770
      total_files += v->files_[level].size();
    }
  }

  // just one time extension to the right size
  live_list->reserve(live_list->size() + total_files);

  for (Version* v = dummy_versions_.next_;
       v != &dummy_versions_;
       v = v->next_) {
    for (int level = 0; level < NumberLevels(); level++) {
      for (const auto& f : v->files_[level]) {
        live_list->push_back(f->number);
J
jorlow@chromium.org 已提交
1771 1772 1773 1774 1775
      }
    }
  }
}

1776 1777 1778 1779 1780 1781 1782 1783 1784 1785
void VersionSet::AddLiveFilesCurrentVersion(std::set<uint64_t>* live) {
  Version* v = current_;
  for (int level = 0; level < NumberLevels(); level++) {
    const std::vector<FileMetaData*>& files = v->files_[level];
    for (size_t i = 0; i < files.size(); i++) {
      live->insert(files[i]->number);
    }
  }
}

1786 1787
int64_t VersionSet::NumLevelBytes(int level) const {
  assert(level >= 0);
1788
  assert(level < NumberLevels());
1789 1790
  assert(current_);
  return TotalFileSize(current_->files_[level]);
J
jorlow@chromium.org 已提交
1791 1792 1793 1794
}

int64_t VersionSet::MaxNextLevelOverlappingBytes() {
  int64_t result = 0;
1795
  std::vector<FileMetaData*> overlaps;
1796
  for (int level = 1; level < NumberLevels() - 1; level++) {
D
dgrogan@chromium.org 已提交
1797
    for (size_t i = 0; i < current_->files_[level].size(); i++) {
1798
      const FileMetaData* f = current_->files_[level][i];
G
Gabor Cselle 已提交
1799 1800
      current_->GetOverlappingInputs(level+1, &f->smallest, &f->largest,
                                     &overlaps);
J
jorlow@chromium.org 已提交
1801
      const int64_t sum = TotalFileSize(overlaps);
1802 1803 1804 1805 1806 1807 1808 1809
      if (sum > result) {
        result = sum;
      }
    }
  }
  return result;
}

J
jorlow@chromium.org 已提交
1810 1811 1812 1813 1814 1815 1816 1817 1818
// Stores the minimal range that covers all entries in inputs in
// *smallest, *largest.
// REQUIRES: inputs is not empty
void VersionSet::GetRange(const std::vector<FileMetaData*>& inputs,
                          InternalKey* smallest,
                          InternalKey* largest) {
  assert(!inputs.empty());
  smallest->Clear();
  largest->Clear();
D
dgrogan@chromium.org 已提交
1819
  for (size_t i = 0; i < inputs.size(); i++) {
J
jorlow@chromium.org 已提交
1820 1821 1822 1823 1824 1825 1826 1827 1828 1829 1830 1831 1832 1833 1834
    FileMetaData* f = inputs[i];
    if (i == 0) {
      *smallest = f->smallest;
      *largest = f->largest;
    } else {
      if (icmp_.Compare(f->smallest, *smallest) < 0) {
        *smallest = f->smallest;
      }
      if (icmp_.Compare(f->largest, *largest) > 0) {
        *largest = f->largest;
      }
    }
  }
}

1835 1836 1837 1838 1839 1840 1841 1842 1843 1844 1845 1846
// Stores the minimal range that covers all entries in inputs1 and inputs2
// in *smallest, *largest.
// REQUIRES: inputs is not empty
void VersionSet::GetRange2(const std::vector<FileMetaData*>& inputs1,
                           const std::vector<FileMetaData*>& inputs2,
                           InternalKey* smallest,
                           InternalKey* largest) {
  std::vector<FileMetaData*> all = inputs1;
  all.insert(all.end(), inputs2.begin(), inputs2.end());
  GetRange(all, smallest, largest);
}

J
jorlow@chromium.org 已提交
1847 1848 1849 1850 1851 1852 1853 1854 1855 1856 1857 1858 1859 1860 1861
Iterator* VersionSet::MakeInputIterator(Compaction* c) {
  ReadOptions options;
  options.verify_checksums = options_->paranoid_checks;
  options.fill_cache = false;

  // Level-0 files have to be merged together.  For other levels,
  // we will make a concatenating iterator per level.
  // TODO(opt): use concatenating iterator for level-0 if there is no overlap
  const int space = (c->level() == 0 ? c->inputs_[0].size() + 1 : 2);
  Iterator** list = new Iterator*[space];
  int num = 0;
  for (int which = 0; which < 2; which++) {
    if (!c->inputs_[which].empty()) {
      if (c->level() + which == 0) {
        const std::vector<FileMetaData*>& files = c->inputs_[which];
D
dgrogan@chromium.org 已提交
1862
        for (size_t i = 0; i < files.size(); i++) {
J
jorlow@chromium.org 已提交
1863
          list[num++] = table_cache_->NewIterator(
1864
              options, storage_options_compactions_,
1865 1866
              files[i]->number, files[i]->file_size, nullptr,
              true /* for compaction */);
J
jorlow@chromium.org 已提交
1867 1868 1869 1870
        }
      } else {
        // Create concatenating iterator for the files from this level
        list[num++] = NewTwoLevelIterator(
1871
            new Version::LevelFileNumIterator(icmp_, &c->inputs_[which]),
1872 1873
            &GetFileIterator, table_cache_, options, storage_options_,
            true /* for compaction */);
J
jorlow@chromium.org 已提交
1874 1875 1876 1877 1878 1879 1880 1881 1882
      }
    }
  }
  assert(num <= space);
  Iterator* result = NewMergingIterator(&icmp_, list, num);
  delete[] list;
  return result;
}

1883
double VersionSet::MaxBytesForLevel(int level) {
A
Abhishek Kona 已提交
1884 1885
  // Note: the result for level zero is not really used since we set
  // the level-0 compaction threshold based on number of files.
1886 1887
  assert(level >= 0);
  assert(level < NumberLevels());
A
Abhishek Kona 已提交
1888
  return level_max_bytes_[level];
1889 1890 1891 1892 1893
}

uint64_t VersionSet::MaxFileSizeForLevel(int level) {
  assert(level >= 0);
  assert(level < NumberLevels());
1894
  return max_file_size_[level];
1895 1896
}

1897
int64_t VersionSet::ExpandedCompactionByteSizeLimit(int level) {
1898 1899 1900 1901 1902
  uint64_t result = MaxFileSizeForLevel(level);
  result *= options_->expanded_compaction_factor;
  return result;
}

1903
int64_t VersionSet::MaxGrandParentOverlapBytes(int level) {
1904 1905 1906 1907 1908
  uint64_t result = MaxFileSizeForLevel(level);
  result *= options_->max_grandparent_overlap_factor;
  return result;
}

A
Abhishek Kona 已提交
1909
// verify that the files listed in this compaction are present
1910 1911
// in the current version
bool VersionSet::VerifyCompactionFileConsistency(Compaction* c) {
1912
#ifndef NDEBUG
1913 1914 1915 1916 1917 1918 1919 1920 1921 1922 1923 1924 1925 1926 1927 1928 1929 1930 1931 1932 1933 1934 1935 1936 1937 1938 1939 1940 1941 1942 1943 1944 1945 1946 1947 1948 1949 1950 1951 1952
  if (c->input_version_ != current_) {
    Log(options_->info_log, "VerifyCompactionFileConsistency version mismatch");
  }

  // verify files in level
  int level = c->level();
  for (int i = 0; i < c->num_input_files(0); i++) {
    uint64_t number = c->input(0,i)->number;

    // look for this file in the current version
    bool found = false;
    for (unsigned int j = 0; j < current_->files_[level].size(); j++) {
      FileMetaData* f = current_->files_[level][j];
      if (f->number == number) {
        found = true;
        break;
      }
    }
    if (!found) {
      return false; // input files non existant in current version
    }
  }
  // verify level+1 files
  level++;
  for (int i = 0; i < c->num_input_files(1); i++) {
    uint64_t number = c->input(1,i)->number;

    // look for this file in the current version
    bool found = false;
    for (unsigned int j = 0; j < current_->files_[level].size(); j++) {
      FileMetaData* f = current_->files_[level][j];
      if (f->number == number) {
        found = true;
        break;
      }
    }
    if (!found) {
      return false; // input files non existant in current version
    }
  }
1953
#endif
1954 1955 1956 1957 1958
  return true;     // everything good
}

// Clear all files to indicate that they are not being compacted
// Delete this compaction from the list of running compactions.
1959
void VersionSet::ReleaseCompactionFiles(Compaction* c, Status status) {
1960 1961
  c->MarkFilesBeingCompacted(false);
  compactions_in_progress_[c->level()].erase(c);
1962 1963 1964
  if (!status.ok()) {
    c->ResetNextCompactionIndex();
  }
1965 1966 1967
}

// The total size of files that are currently being compacted
1968 1969 1970 1971 1972 1973 1974 1975 1976 1977 1978 1979 1980
// at at every level upto the penultimate level.
void VersionSet::SizeBeingCompacted(std::vector<uint64_t>& sizes) {
  for (int level = 0; level < NumberLevels()-1; level++) {
    uint64_t total = 0;
    for (std::set<Compaction*>::iterator it =
         compactions_in_progress_[level].begin();
         it != compactions_in_progress_[level].end();
         ++it) {
      Compaction* c = (*it);
      assert(c->level() == level);
      for (int i = 0; i < c->num_input_files(0); i++) {
        total += c->input(0,i)->file_size;
      }
1981
    }
1982
    sizes[level] = total;
1983 1984 1985
  }
}

1986
Compaction* VersionSet::PickCompactionBySize(int level, double score) {
A
Abhishek Kona 已提交
1987
  Compaction* c = nullptr;
1988 1989 1990

  // level 0 files are overlapping. So we cannot pick more
  // than one concurrent compactions at this level. This
1991
  // could be made better by looking at key-ranges that are
1992 1993
  // being compacted at level 0.
  if (level == 0 && compactions_in_progress_[level].size() == 1) {
A
Abhishek Kona 已提交
1994
    return nullptr;
1995 1996 1997 1998
  }

  assert(level >= 0);
  assert(level+1 < NumberLevels());
X
Xing Jin 已提交
1999
  c = new Compaction(level, MaxFileSizeForLevel(level+1),
2000
      MaxGrandParentOverlapBytes(level), NumberLevels());
2001
  c->score_ = score;
2002

2003 2004 2005
  // Pick the largest file in this level that is not already
  // being compacted
  std::vector<int>& file_size = current_->files_by_size_[level];
2006 2007 2008 2009 2010 2011

  // record the first file that is not yet compacted
  int nextIndex = -1;

  for (unsigned int i = current_->next_file_to_compact_by_size_[level];
       i < file_size.size(); i++) {
2012 2013 2014 2015 2016
    int index = file_size[i];
    FileMetaData* f = current_->files_[level][index];

    // check to verify files are arranged in descending size
    assert((i == file_size.size() - 1) ||
2017
           (i >= Version::number_of_files_to_sort_-1) ||
2018
          (f->file_size >= current_->files_[level][file_size[i+1]]->file_size));
2019 2020 2021 2022 2023 2024

    // do not pick a file to compact if it is being compacted
    // from n-1 level.
    if (f->being_compacted) {
      continue;
    }
2025 2026 2027 2028 2029 2030 2031 2032 2033 2034

    // remember the startIndex for the next call to PickCompaction
    if (nextIndex == -1) {
      nextIndex = i;
    }

    //if (i > Version::number_of_files_to_sort_) {
    //  Log(options_->info_log, "XXX Looking at index %d", i);
    //}

2035 2036
    // Do not pick this file if its parents at level+1 are being compacted.
    // Maybe we can avoid redoing this work in SetupOtherInputs
2037
    int parent_index = -1;
A
Abhishek Kona 已提交
2038
    if (ParentRangeInCompaction(&f->smallest, &f->largest, level,
2039
                                &parent_index)) {
2040
      continue;
2041
    }
2042
    c->inputs_[0].push_back(f);
2043
    c->base_index_ = index;
2044
    c->parent_index_ = parent_index;
2045
    break;
2046
  }
2047

2048 2049
  if (c->inputs_[0].empty()) {
    delete c;
A
Abhishek Kona 已提交
2050
    c = nullptr;
2051
  }
2052 2053 2054 2055

  // store where to start the iteration in the next call to PickCompaction
  current_->next_file_to_compact_by_size_[level] = nextIndex;

2056 2057 2058
  return c;
}

J
jorlow@chromium.org 已提交
2059
Compaction* VersionSet::PickCompaction() {
A
Abhishek Kona 已提交
2060
  Compaction* c = nullptr;
2061
  int level = -1;
2062

2063 2064
  // compute the compactions needed. It is better to do it here
  // and also in LogAndApply(), otherwise the values could be stale.
2065 2066 2067
  std::vector<uint64_t> size_being_compacted(NumberLevels()-1);
  current_->vset_->SizeBeingCompacted(size_being_compacted);
  Finalize(current_, size_being_compacted);
2068

2069 2070
  // We prefer compactions triggered by too much data in a level over
  // the compactions triggered by seeks.
2071 2072 2073
  //
  // Find the compactions by size on all levels.
  for (int i = 0; i < NumberLevels()-1; i++) {
2074 2075
    assert(i == 0 || current_->compaction_score_[i] <=
                     current_->compaction_score_[i-1]);
2076 2077
    level = current_->compaction_level_[i];
    if ((current_->compaction_score_[i] >= 1)) {
2078
      c = PickCompactionBySize(level, current_->compaction_score_[i]);
A
Abhishek Kona 已提交
2079
      if (c != nullptr) {
2080 2081 2082
        break;
      }
    }
2083 2084 2085
  }

  // Find compactions needed by seeks
2086 2087 2088
  FileMetaData* f = current_->file_to_compact_;
  if (c == nullptr && f != nullptr && !f->being_compacted) {

2089
    level = current_->file_to_compact_level_;
2090
    int parent_index = -1;
2091 2092

    // Only allow one level 0 compaction at a time.
2093
    // Do not pick this file if its parents at level+1 are being compacted.
2094
    if (level != 0 || compactions_in_progress_[0].empty()) {
2095 2096
      if(!ParentRangeInCompaction(&f->smallest, &f->largest, level,
                                  &parent_index)) {
X
Xing Jin 已提交
2097
        c = new Compaction(level, MaxFileSizeForLevel(level+1),
2098 2099 2100 2101 2102
                MaxGrandParentOverlapBytes(level), NumberLevels(), true);
        c->inputs_[0].push_back(f);
        c->parent_index_ = parent_index;
        current_->file_to_compact_ = nullptr;
      }
2103
    }
2104 2105
  }

A
Abhishek Kona 已提交
2106 2107
  if (c == nullptr) {
    return nullptr;
J
jorlow@chromium.org 已提交
2108 2109 2110 2111 2112 2113
  }

  c->input_version_ = current_;
  c->input_version_->Ref();

  // Files in level 0 may overlap each other, so pick up all overlapping ones
2114 2115
  // Two level 0 compaction won't run at the same time, so don't need to worry
  // about files on level 0 being compacted.
J
jorlow@chromium.org 已提交
2116
  if (level == 0) {
2117
    assert(compactions_in_progress_[0].empty());
2118 2119
    InternalKey smallest, largest;
    GetRange(c->inputs_[0], &smallest, &largest);
J
jorlow@chromium.org 已提交
2120 2121 2122
    // Note that the next call will discard the file we placed in
    // c->inputs_[0] earlier and replace it with an overlapping set
    // which will include the picked file.
2123
    c->inputs_[0].clear();
2124
    current_->GetOverlappingInputs(0, &smallest, &largest, &c->inputs_[0]);
2125 2126 2127 2128 2129

    // If we include more L0 files in the same compaction run it can
    // cause the 'smallest' and 'largest' key to get extended to a
    // larger range. So, re-invoke GetRange to get the new key range
    GetRange(c->inputs_[0], &smallest, &largest);
2130 2131 2132
    if (ParentRangeInCompaction(&smallest, &largest,
                                level, &c->parent_index_)) {
      delete c;
A
Abhishek Kona 已提交
2133
      return nullptr;
2134
    }
J
jorlow@chromium.org 已提交
2135 2136 2137
    assert(!c->inputs_[0].empty());
  }

2138 2139
  SetupOtherInputs(c);

2140 2141 2142 2143 2144 2145
  // mark all the files that are being compacted
  c->MarkFilesBeingCompacted(true);

  // remember this currently undergoing compaction
  compactions_in_progress_[level].insert(c);

2146 2147 2148
  return c;
}

2149
// Returns true if any one of the parent files are being compacted
2150 2151
bool VersionSet::ParentRangeInCompaction(const InternalKey* smallest,
  const InternalKey* largest, int level, int* parent_index) {
2152
  std::vector<FileMetaData*> inputs;
A
Abhishek Kona 已提交
2153

2154
  current_->GetOverlappingInputs(level+1, smallest, largest,
2155
                                 &inputs, *parent_index, parent_index);
2156 2157 2158 2159 2160 2161 2162 2163
  return FilesInCompaction(inputs);
}

// Returns true if any one of specified files are being compacted
bool VersionSet::FilesInCompaction(std::vector<FileMetaData*>& files) {
  for (unsigned int i = 0; i < files.size(); i++) {
    if (files[i]->being_compacted) {
      return true;
A
Abhishek Kona 已提交
2164
    }
2165 2166 2167 2168
  }
  return false;
}

2169 2170 2171 2172 2173
void VersionSet::SetupOtherInputs(Compaction* c) {
  const int level = c->level();
  InternalKey smallest, largest;
  GetRange(c->inputs_[0], &smallest, &largest);

2174 2175
  current_->GetOverlappingInputs(level+1, &smallest, &largest, &c->inputs_[1],
                                 c->parent_index_, &c->parent_index_);
J
jorlow@chromium.org 已提交
2176

2177 2178 2179 2180
  // Get entire range covered by compaction
  InternalKey all_start, all_limit;
  GetRange2(c->inputs_[0], c->inputs_[1], &all_start, &all_limit);

J
jorlow@chromium.org 已提交
2181 2182 2183 2184
  // See if we can grow the number of inputs in "level" without
  // changing the number of "level+1" files we pick up.
  if (!c->inputs_[1].empty()) {
    std::vector<FileMetaData*> expanded0;
2185
    current_->GetOverlappingInputs(level, &all_start, &all_limit, &expanded0,
A
Abhishek Kona 已提交
2186
                                   c->base_index_, nullptr);
2187 2188 2189
    const int64_t inputs0_size = TotalFileSize(c->inputs_[0]);
    const int64_t inputs1_size = TotalFileSize(c->inputs_[1]);
    const int64_t expanded0_size = TotalFileSize(expanded0);
2190
    int64_t limit = ExpandedCompactionByteSizeLimit(level);
2191
    if (expanded0.size() > c->inputs_[0].size() &&
2192 2193
        inputs1_size + expanded0_size < limit &&
        !FilesInCompaction(expanded0)) {
J
jorlow@chromium.org 已提交
2194 2195 2196
      InternalKey new_start, new_limit;
      GetRange(expanded0, &new_start, &new_limit);
      std::vector<FileMetaData*> expanded1;
G
Gabor Cselle 已提交
2197
      current_->GetOverlappingInputs(level+1, &new_start, &new_limit,
2198 2199
                                     &expanded1, c->parent_index_,
                                     &c->parent_index_);
A
Abhishek Kona 已提交
2200
      if (expanded1.size() == c->inputs_[1].size() &&
2201
          !FilesInCompaction(expanded1)) {
2202
        Log(options_->info_log,
2203
            "Expanding@%d %d+%d (%ld+%ld bytes) to %d+%d (%ld+%ld bytes)\n",
J
jorlow@chromium.org 已提交
2204 2205 2206
            level,
            int(c->inputs_[0].size()),
            int(c->inputs_[1].size()),
2207
            long(inputs0_size), long(inputs1_size),
J
jorlow@chromium.org 已提交
2208
            int(expanded0.size()),
2209 2210
            int(expanded1.size()),
            long(expanded0_size), long(inputs1_size));
J
jorlow@chromium.org 已提交
2211 2212 2213 2214
        smallest = new_start;
        largest = new_limit;
        c->inputs_[0] = expanded0;
        c->inputs_[1] = expanded1;
2215
        GetRange2(c->inputs_[0], c->inputs_[1], &all_start, &all_limit);
J
jorlow@chromium.org 已提交
2216 2217 2218 2219
      }
    }
  }

2220 2221
  // Compute the set of grandparent files that overlap this compaction
  // (parent == level+1; grandparent == level+2)
2222
  if (level + 2 < NumberLevels()) {
G
Gabor Cselle 已提交
2223 2224
    current_->GetOverlappingInputs(level + 2, &all_start, &all_limit,
                                   &c->grandparents_);
2225 2226
  }

J
jorlow@chromium.org 已提交
2227
  if (false) {
2228
    Log(options_->info_log, "Compacting %d '%s' .. '%s'",
J
jorlow@chromium.org 已提交
2229
        level,
G
Gabor Cselle 已提交
2230 2231
        smallest.DebugString().c_str(),
        largest.DebugString().c_str());
J
jorlow@chromium.org 已提交
2232 2233 2234 2235 2236 2237 2238
  }

  // Update the place where we will do the next compaction for this level.
  // We update this immediately instead of waiting for the VersionEdit
  // to be applied so that if the compaction fails, we will try a different
  // key range next time.
  compact_pointer_[level] = largest.Encode().ToString();
2239
  c->edit_->SetCompactPointer(level, largest);
J
jorlow@chromium.org 已提交
2240 2241 2242 2243
}

Compaction* VersionSet::CompactRange(
    int level,
G
Gabor Cselle 已提交
2244 2245
    const InternalKey* begin,
    const InternalKey* end) {
J
jorlow@chromium.org 已提交
2246
  std::vector<FileMetaData*> inputs;
G
Gabor Cselle 已提交
2247
  current_->GetOverlappingInputs(level, begin, end, &inputs);
J
jorlow@chromium.org 已提交
2248
  if (inputs.empty()) {
A
Abhishek Kona 已提交
2249
    return nullptr;
J
jorlow@chromium.org 已提交
2250 2251
  }

2252

G
Gabor Cselle 已提交
2253
  // Avoid compacting too much in one shot in case the range is large.
2254 2255 2256 2257 2258 2259
  // But we cannot do this for level-0 since level-0 files can overlap
  // and we must not pick one file and drop another older file if the
  // two files overlap.
  if (level > 0) {

    const uint64_t limit = MaxFileSizeForLevel(level) *
2260
                         options_->source_compaction_factor;
2261 2262 2263 2264 2265 2266 2267 2268
    uint64_t total = 0;
    for (size_t i = 0; i < inputs.size(); ++i) {
      uint64_t s = inputs[i]->file_size;
      total += s;
      if (total >= limit) {
        inputs.resize(i + 1);
        break;
      }
G
Gabor Cselle 已提交
2269 2270 2271
    }
  }

X
Xing Jin 已提交
2272
  Compaction* c = new Compaction(level, MaxFileSizeForLevel(level+1),
2273
    MaxGrandParentOverlapBytes(level), NumberLevels());
J
jorlow@chromium.org 已提交
2274 2275 2276
  c->input_version_ = current_;
  c->input_version_->Ref();
  c->inputs_[0] = inputs;
2277
  SetupOtherInputs(c);
2278 2279 2280 2281 2282

  // These files that are to be manaully compacted do not trample
  // upon other files because manual compactions are processed when
  // the system has a max of 1 background compaction thread.
  c->MarkFilesBeingCompacted(true);
J
jorlow@chromium.org 已提交
2283 2284 2285
  return c;
}

2286
Compaction::Compaction(int level, uint64_t target_file_size,
2287 2288
  uint64_t max_grandparent_overlap_bytes, int number_levels,
  bool seek_compaction)
J
jorlow@chromium.org 已提交
2289
    : level_(level),
2290 2291
      max_output_file_size_(target_file_size),
      maxGrandParentOverlapBytes_(max_grandparent_overlap_bytes),
A
Abhishek Kona 已提交
2292
      input_version_(nullptr),
2293
      number_levels_(number_levels),
2294
      seek_compaction_(seek_compaction),
2295
      grandparent_index_(0),
J
jorlow@chromium.org 已提交
2296
      seen_key_(false),
2297 2298
      overlapped_bytes_(0),
      base_index_(-1),
2299
      parent_index_(-1),
2300 2301
      score_(0),
      level_ptrs_(std::vector<size_t>(number_levels)) {
2302
  edit_ = new VersionEdit(number_levels_);
2303
  for (int i = 0; i < number_levels_; i++) {
J
jorlow@chromium.org 已提交
2304 2305 2306 2307 2308
    level_ptrs_[i] = 0;
  }
}

Compaction::~Compaction() {
A
Abhishek Kona 已提交
2309
  delete edit_;
A
Abhishek Kona 已提交
2310
  if (input_version_ != nullptr) {
J
jorlow@chromium.org 已提交
2311 2312 2313 2314
    input_version_->Unref();
  }
}

2315
bool Compaction::IsTrivialMove() const {
J
jorlow@chromium.org 已提交
2316
  // Avoid a move if there is lots of overlapping grandparent data.
2317 2318
  // Otherwise, the move could create a parent file that will require
  // a very expensive merge later on.
J
jorlow@chromium.org 已提交
2319 2320
  return (num_input_files(0) == 1 &&
          num_input_files(1) == 0 &&
2321
          TotalFileSize(grandparents_) <= maxGrandParentOverlapBytes_);
2322 2323
}

J
jorlow@chromium.org 已提交
2324 2325
void Compaction::AddInputDeletions(VersionEdit* edit) {
  for (int which = 0; which < 2; which++) {
D
dgrogan@chromium.org 已提交
2326
    for (size_t i = 0; i < inputs_[which].size(); i++) {
J
jorlow@chromium.org 已提交
2327 2328 2329 2330 2331 2332 2333 2334
      edit->DeleteFile(level_ + which, inputs_[which][i]->number);
    }
  }
}

bool Compaction::IsBaseLevelForKey(const Slice& user_key) {
  // Maybe use binary search to find right entry instead of linear search?
  const Comparator* user_cmp = input_version_->vset_->icmp_.user_comparator();
2335
  for (int lvl = level_ + 2; lvl < number_levels_; lvl++) {
J
jorlow@chromium.org 已提交
2336 2337 2338 2339 2340 2341 2342 2343 2344 2345 2346 2347 2348 2349 2350 2351 2352
    const std::vector<FileMetaData*>& files = input_version_->files_[lvl];
    for (; level_ptrs_[lvl] < files.size(); ) {
      FileMetaData* f = files[level_ptrs_[lvl]];
      if (user_cmp->Compare(user_key, f->largest.user_key()) <= 0) {
        // We've advanced far enough
        if (user_cmp->Compare(user_key, f->smallest.user_key()) >= 0) {
          // Key falls in this file's range, so definitely not base level
          return false;
        }
        break;
      }
      level_ptrs_[lvl]++;
    }
  }
  return true;
}

2353
bool Compaction::ShouldStopBefore(const Slice& internal_key) {
2354 2355 2356
  // Scan to find earliest grandparent file that contains key.
  const InternalKeyComparator* icmp = &input_version_->vset_->icmp_;
  while (grandparent_index_ < grandparents_.size() &&
2357 2358
      icmp->Compare(internal_key,
                    grandparents_[grandparent_index_]->largest.Encode()) > 0) {
J
jorlow@chromium.org 已提交
2359 2360 2361
    if (seen_key_) {
      overlapped_bytes_ += grandparents_[grandparent_index_]->file_size;
    }
2362 2363 2364 2365
    assert(grandparent_index_ + 1 >= grandparents_.size() ||
           icmp->Compare(grandparents_[grandparent_index_]->largest.Encode(),
                         grandparents_[grandparent_index_+1]->smallest.Encode())
                         < 0);
2366 2367
    grandparent_index_++;
  }
J
jorlow@chromium.org 已提交
2368
  seen_key_ = true;
2369

2370
  if (overlapped_bytes_ > maxGrandParentOverlapBytes_) {
J
jorlow@chromium.org 已提交
2371 2372
    // Too much overlap for current output; start new output
    overlapped_bytes_ = 0;
2373 2374 2375 2376 2377 2378
    return true;
  } else {
    return false;
  }
}

2379 2380 2381 2382 2383 2384 2385 2386 2387 2388 2389 2390
// Mark (or clear) each file that is being compacted
void Compaction::MarkFilesBeingCompacted(bool value) {
  for (int i = 0; i < 2; i++) {
    std::vector<FileMetaData*> v = inputs_[i];
    for (unsigned int j = 0; j < inputs_[i].size(); j++) {
      assert(value ? !inputs_[i][j]->being_compacted :
                      inputs_[i][j]->being_compacted);
      inputs_[i][j]->being_compacted = value;
    }
  }
}

J
jorlow@chromium.org 已提交
2391
void Compaction::ReleaseInputs() {
A
Abhishek Kona 已提交
2392
  if (input_version_ != nullptr) {
J
jorlow@chromium.org 已提交
2393
    input_version_->Unref();
A
Abhishek Kona 已提交
2394
    input_version_ = nullptr;
J
jorlow@chromium.org 已提交
2395 2396 2397
  }
}

2398
void Compaction::ResetNextCompactionIndex() {
A
Abhishek Kona 已提交
2399
  input_version_->ResetNextCompactionIndex(level_);
2400 2401
}

H
heyongqiang 已提交
2402 2403 2404 2405
static void InputSummary(std::vector<FileMetaData*>& files,
    char* output,
    int len) {
  int write = 0;
2406
  for (unsigned int i = 0; i < files.size(); i++) {
H
heyongqiang 已提交
2407
    int sz = len - write;
2408
    int ret = snprintf(output + write, sz, "%lu(%lu) ",
H
heyongqiang 已提交
2409 2410 2411 2412 2413 2414 2415 2416 2417
        files.at(i)->number,
        files.at(i)->file_size);
    if (ret < 0 || ret >= sz)
      break;
    write += ret;
  }
}

void Compaction::Summary(char* output, int len) {
2418
  int write = snprintf(output, len,
2419 2420
      "Base version %ld Base level %d, seek compaction:%d, inputs:",
      input_version_->GetVersionNumber(), level_, seek_compaction_);
H
heyongqiang 已提交
2421 2422 2423 2424
  if(write < 0 || write > len)
    return;

  char level_low_summary[100];
2425
  InputSummary(inputs_[0], level_low_summary, sizeof(level_low_summary));
H
heyongqiang 已提交
2426 2427
  char level_up_summary[100];
  if (inputs_[1].size()) {
2428
    InputSummary(inputs_[1], level_up_summary, sizeof(level_up_summary));
H
heyongqiang 已提交
2429 2430 2431 2432 2433 2434 2435 2436
  } else {
    level_up_summary[0] = '\0';
  }

  snprintf(output + write, len - write, "[%s],[%s]",
      level_low_summary, level_up_summary);
}

H
Hans Wennborg 已提交
2437
}  // namespace leveldb