作者: 谢瑞阳 10225101483 徐翔宇 10225101535
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

1120 lines
34 KiB

  1. // Copyright (c) 2011 The LevelDB Authors. All rights reserved.
  2. // Use of this source code is governed by a BSD-style license that can be
  3. // found in the LICENSE file. See the AUTHORS file for names of contributors.
  4. #include "db/version_set.h"
  5. #include <algorithm>
  6. #include <stdio.h>
  7. #include "db/filename.h"
  8. #include "db/log_reader.h"
  9. #include "db/log_writer.h"
  10. #include "db/memtable.h"
  11. #include "db/table_cache.h"
  12. #include "leveldb/env.h"
  13. #include "leveldb/table_builder.h"
  14. #include "table/merger.h"
  15. #include "table/two_level_iterator.h"
  16. #include "util/coding.h"
  17. #include "util/logging.h"
  18. namespace leveldb {
  19. static const int kTargetFileSize = 2 * 1048576;
  20. // Maximum bytes of overlaps in grandparent (i.e., level+2) before we
  21. // stop building a single file in a level->level+1 compaction.
  22. static const int64_t kMaxGrandParentOverlapBytes = 10 * kTargetFileSize;
  23. static double MaxBytesForLevel(int level) {
  24. // Note: the result for level zero is not really used since we set
  25. // the level-0 compaction threshold based on number of files.
  26. double result = 10 * 1048576.0; // Result for both level-0 and level-1
  27. while (level > 1) {
  28. result *= 10;
  29. level--;
  30. }
  31. return result;
  32. }
  33. static uint64_t MaxFileSizeForLevel(int level) {
  34. return kTargetFileSize; // We could vary per level to reduce number of files?
  35. }
  36. namespace {
  37. std::string IntSetToString(const std::set<uint64_t>& s) {
  38. std::string result = "{";
  39. for (std::set<uint64_t>::const_iterator it = s.begin();
  40. it != s.end();
  41. ++it) {
  42. result += (result.size() > 1) ? "," : "";
  43. result += NumberToString(*it);
  44. }
  45. result += "}";
  46. return result;
  47. }
  48. }
  49. Version::~Version() {
  50. assert(refs_ == 0);
  51. for (int level = 0; level < config::kNumLevels; level++) {
  52. for (int i = 0; i < files_[level].size(); i++) {
  53. FileMetaData* f = files_[level][i];
  54. assert(f->refs >= 0);
  55. f->refs--;
  56. if (f->refs <= 0) {
  57. delete f;
  58. }
  59. }
  60. }
  61. delete cleanup_mem_;
  62. }
  63. // An internal iterator. For a given version/level pair, yields
  64. // information about the files in the level. For a given entry, key()
  65. // is the largest key that occurs in the file, and value() is an
  66. // 16-byte value containing the file number and file size, both
  67. // encoded using EncodeFixed64.
  68. class Version::LevelFileNumIterator : public Iterator {
  69. public:
  70. LevelFileNumIterator(const Version* version,
  71. const std::vector<FileMetaData*>* flist)
  72. : icmp_(version->vset_->icmp_.user_comparator()),
  73. flist_(flist),
  74. index_(flist->size()) { // Marks as invalid
  75. }
  76. virtual bool Valid() const {
  77. return index_ < flist_->size();
  78. }
  79. virtual void Seek(const Slice& target) {
  80. uint32_t left = 0;
  81. uint32_t right = flist_->size() - 1;
  82. while (left < right) {
  83. uint32_t mid = (left + right) / 2;
  84. int cmp = icmp_.Compare((*flist_)[mid]->largest.Encode(), target);
  85. if (cmp < 0) {
  86. // Key at "mid.largest" is < than "target". Therefore all
  87. // files at or before "mid" are uninteresting.
  88. left = mid + 1;
  89. } else {
  90. // Key at "mid.largest" is >= "target". Therefore all files
  91. // after "mid" are uninteresting.
  92. right = mid;
  93. }
  94. }
  95. index_ = left;
  96. }
  97. virtual void SeekToFirst() { index_ = 0; }
  98. virtual void SeekToLast() {
  99. index_ = flist_->empty() ? 0 : flist_->size() - 1;
  100. }
  101. virtual void Next() {
  102. assert(Valid());
  103. index_++;
  104. }
  105. virtual void Prev() {
  106. assert(Valid());
  107. if (index_ == 0) {
  108. index_ = flist_->size(); // Marks as invalid
  109. } else {
  110. index_--;
  111. }
  112. }
  113. Slice key() const {
  114. assert(Valid());
  115. return (*flist_)[index_]->largest.Encode();
  116. }
  117. Slice value() const {
  118. assert(Valid());
  119. EncodeFixed64(value_buf_, (*flist_)[index_]->number);
  120. EncodeFixed64(value_buf_+8, (*flist_)[index_]->file_size);
  121. return Slice(value_buf_, sizeof(value_buf_));
  122. }
  123. virtual Status status() const { return Status::OK(); }
  124. private:
  125. const InternalKeyComparator icmp_;
  126. const std::vector<FileMetaData*>* const flist_;
  127. int index_;
  128. // Backing store for value(). Holds the file number and size.
  129. mutable char value_buf_[16];
  130. };
  131. static Iterator* GetFileIterator(void* arg,
  132. const ReadOptions& options,
  133. const Slice& file_value) {
  134. TableCache* cache = reinterpret_cast<TableCache*>(arg);
  135. if (file_value.size() != 16) {
  136. return NewErrorIterator(
  137. Status::Corruption("FileReader invoked with unexpected value"));
  138. } else {
  139. return cache->NewIterator(options,
  140. DecodeFixed64(file_value.data()),
  141. DecodeFixed64(file_value.data() + 8));
  142. }
  143. }
  144. Iterator* Version::NewConcatenatingIterator(const ReadOptions& options,
  145. int level) const {
  146. return NewTwoLevelIterator(
  147. new LevelFileNumIterator(this, &files_[level]),
  148. &GetFileIterator, vset_->table_cache_, options);
  149. }
  150. void Version::AddIterators(const ReadOptions& options,
  151. std::vector<Iterator*>* iters) {
  152. // Merge all level zero files together since they may overlap
  153. for (int i = 0; i < files_[0].size(); i++) {
  154. iters->push_back(
  155. vset_->table_cache_->NewIterator(
  156. options, files_[0][i]->number, files_[0][i]->file_size));
  157. }
  158. // For levels > 0, we can use a concatenating iterator that sequentially
  159. // walks through the non-overlapping files in the level, opening them
  160. // lazily.
  161. for (int level = 1; level < config::kNumLevels; level++) {
  162. if (!files_[level].empty()) {
  163. iters->push_back(NewConcatenatingIterator(options, level));
  164. }
  165. }
  166. }
  167. void Version::Ref() {
  168. ++refs_;
  169. }
  170. void Version::Unref() {
  171. assert(refs_ >= 1);
  172. --refs_;
  173. if (refs_ == 0) {
  174. vset_->MaybeDeleteOldVersions();
  175. // TODO: try to delete obsolete files
  176. }
  177. }
  178. std::string Version::DebugString() const {
  179. std::string r;
  180. for (int level = 0; level < config::kNumLevels; level++) {
  181. // E.g., level 1: 17:123['a' .. 'd'] 20:43['e' .. 'g']
  182. r.append("level ");
  183. AppendNumberTo(&r, level);
  184. r.push_back(':');
  185. const std::vector<FileMetaData*>& files = files_[level];
  186. for (int i = 0; i < files.size(); i++) {
  187. r.push_back(' ');
  188. AppendNumberTo(&r, files[i]->number);
  189. r.push_back(':');
  190. AppendNumberTo(&r, files[i]->file_size);
  191. r.append("['");
  192. AppendEscapedStringTo(&r, files[i]->smallest.Encode());
  193. r.append("' .. '");
  194. AppendEscapedStringTo(&r, files[i]->largest.Encode());
  195. r.append("']");
  196. }
  197. r.push_back('\n');
  198. }
  199. return r;
  200. }
  201. // A helper class so we can efficiently apply a whole sequence
  202. // of edits to a particular state without creating intermediate
  203. // Versions that contain full copies of the intermediate state.
  204. class VersionSet::Builder {
  205. private:
  206. typedef std::map<uint64_t, FileMetaData*> FileMap;
  207. VersionSet* vset_;
  208. FileMap files_[config::kNumLevels];
  209. public:
  210. // Initialize a builder with the files from *base and other info from *vset
  211. Builder(VersionSet* vset, Version* base)
  212. : vset_(vset) {
  213. for (int level = 0; level < config::kNumLevels; level++) {
  214. const std::vector<FileMetaData*>& files = base->files_[level];
  215. for (int i = 0; i < files.size(); i++) {
  216. FileMetaData* f = files[i];
  217. f->refs++;
  218. files_[level].insert(std::make_pair(f->number, f));
  219. }
  220. }
  221. }
  222. ~Builder() {
  223. for (int level = 0; level < config::kNumLevels; level++) {
  224. const FileMap& fmap = files_[level];
  225. for (FileMap::const_iterator iter = fmap.begin();
  226. iter != fmap.end();
  227. ++iter) {
  228. FileMetaData* f = iter->second;
  229. f->refs--;
  230. if (f->refs <= 0) {
  231. delete f;
  232. }
  233. }
  234. }
  235. }
  236. // Apply all of the edits in *edit to the current state.
  237. void Apply(VersionEdit* edit) {
  238. // Update compaction pointers
  239. for (int i = 0; i < edit->compact_pointers_.size(); i++) {
  240. const int level = edit->compact_pointers_[i].first;
  241. vset_->compact_pointer_[level] =
  242. edit->compact_pointers_[i].second.Encode().ToString();
  243. }
  244. // Delete files
  245. const VersionEdit::DeletedFileSet& del = edit->deleted_files_;
  246. for (VersionEdit::DeletedFileSet::const_iterator iter = del.begin();
  247. iter != del.end();
  248. ++iter) {
  249. const int level = iter->first;
  250. const uint64_t number = iter->second;
  251. FileMap::iterator fiter = files_[level].find(number);
  252. assert(fiter != files_[level].end()); // Sanity check for debug mode
  253. if (fiter != files_[level].end()) {
  254. FileMetaData* f = fiter->second;
  255. f->refs--;
  256. if (f->refs <= 0) {
  257. delete f;
  258. }
  259. files_[level].erase(fiter);
  260. }
  261. }
  262. // Add new files
  263. for (int i = 0; i < edit->new_files_.size(); i++) {
  264. const int level = edit->new_files_[i].first;
  265. FileMetaData* f = new FileMetaData(edit->new_files_[i].second);
  266. f->refs = 1;
  267. assert(files_[level].count(f->number) == 0);
  268. files_[level].insert(std::make_pair(f->number, f));
  269. }
  270. // Add large value refs
  271. for (int i = 0; i < edit->large_refs_added_.size(); i++) {
  272. const VersionEdit::Large& l = edit->large_refs_added_[i];
  273. vset_->RegisterLargeValueRef(l.large_ref, l.fnum, l.internal_key);
  274. }
  275. }
  276. // Save the current state in *v.
  277. void SaveTo(Version* v) {
  278. for (int level = 0; level < config::kNumLevels; level++) {
  279. const FileMap& fmap = files_[level];
  280. for (FileMap::const_iterator iter = fmap.begin();
  281. iter != fmap.end();
  282. ++iter) {
  283. FileMetaData* f = iter->second;
  284. f->refs++;
  285. v->files_[level].push_back(f);
  286. }
  287. }
  288. }
  289. };
  290. VersionSet::VersionSet(const std::string& dbname,
  291. const Options* options,
  292. TableCache* table_cache,
  293. const InternalKeyComparator* cmp)
  294. : env_(options->env),
  295. dbname_(dbname),
  296. options_(options),
  297. table_cache_(table_cache),
  298. icmp_(*cmp),
  299. next_file_number_(2),
  300. manifest_file_number_(0), // Filled by Recover()
  301. last_sequence_(0),
  302. log_number_(0),
  303. prev_log_number_(0),
  304. descriptor_file_(NULL),
  305. descriptor_log_(NULL),
  306. current_(new Version(this)),
  307. oldest_(current_) {
  308. }
  309. VersionSet::~VersionSet() {
  310. for (Version* v = oldest_; v != NULL; ) {
  311. Version* next = v->next_;
  312. assert(v->refs_ == 0);
  313. delete v;
  314. v = next;
  315. }
  316. delete descriptor_log_;
  317. delete descriptor_file_;
  318. }
  319. Status VersionSet::LogAndApply(VersionEdit* edit, MemTable* cleanup_mem) {
  320. if (edit->has_log_number_) {
  321. assert(edit->log_number_ >= log_number_);
  322. assert(edit->log_number_ < next_file_number_);
  323. } else {
  324. edit->SetLogNumber(log_number_);
  325. }
  326. if (!edit->has_prev_log_number_) {
  327. edit->SetPrevLogNumber(prev_log_number_);
  328. }
  329. edit->SetNextFile(next_file_number_);
  330. edit->SetLastSequence(last_sequence_);
  331. Version* v = new Version(this);
  332. {
  333. Builder builder(this, current_);
  334. builder.Apply(edit);
  335. builder.SaveTo(v);
  336. }
  337. std::string new_manifest_file;
  338. Status s = Finalize(v);
  339. // Initialize new descriptor log file if necessary by creating
  340. // a temporary file that contains a snapshot of the current version.
  341. if (s.ok()) {
  342. if (descriptor_log_ == NULL) {
  343. assert(descriptor_file_ == NULL);
  344. new_manifest_file = DescriptorFileName(dbname_, manifest_file_number_);
  345. edit->SetNextFile(next_file_number_);
  346. s = env_->NewWritableFile(new_manifest_file, &descriptor_file_);
  347. if (s.ok()) {
  348. descriptor_log_ = new log::Writer(descriptor_file_);
  349. s = WriteSnapshot(descriptor_log_);
  350. }
  351. }
  352. }
  353. // Write new record to MANIFEST log
  354. if (s.ok()) {
  355. std::string record;
  356. edit->EncodeTo(&record);
  357. s = descriptor_log_->AddRecord(record);
  358. if (s.ok()) {
  359. s = descriptor_file_->Sync();
  360. }
  361. }
  362. // If we just created a new descriptor file, install it by writing a
  363. // new CURRENT file that points to it.
  364. if (s.ok() && !new_manifest_file.empty()) {
  365. s = SetCurrentFile(env_, dbname_, manifest_file_number_);
  366. }
  367. // Install the new version
  368. if (s.ok()) {
  369. assert(current_->next_ == NULL);
  370. assert(current_->cleanup_mem_ == NULL);
  371. current_->cleanup_mem_ = cleanup_mem;
  372. v->next_ = NULL;
  373. current_->next_ = v;
  374. current_ = v;
  375. log_number_ = edit->log_number_;
  376. prev_log_number_ = edit->prev_log_number_;
  377. } else {
  378. delete v;
  379. if (!new_manifest_file.empty()) {
  380. delete descriptor_log_;
  381. delete descriptor_file_;
  382. descriptor_log_ = NULL;
  383. descriptor_file_ = NULL;
  384. env_->DeleteFile(new_manifest_file);
  385. }
  386. }
  387. return s;
  388. }
  389. Status VersionSet::Recover() {
  390. struct LogReporter : public log::Reader::Reporter {
  391. Status* status;
  392. virtual void Corruption(size_t bytes, const Status& s) {
  393. if (this->status->ok()) *this->status = s;
  394. }
  395. };
  396. // Read "CURRENT" file, which contains a pointer to the current manifest file
  397. std::string current;
  398. Status s = ReadFileToString(env_, CurrentFileName(dbname_), &current);
  399. if (!s.ok()) {
  400. return s;
  401. }
  402. if (current.empty() || current[current.size()-1] != '\n') {
  403. return Status::Corruption("CURRENT file does not end with newline");
  404. }
  405. current.resize(current.size() - 1);
  406. std::string dscname = dbname_ + "/" + current;
  407. SequentialFile* file;
  408. s = env_->NewSequentialFile(dscname, &file);
  409. if (!s.ok()) {
  410. return s;
  411. }
  412. bool have_log_number = false;
  413. bool have_prev_log_number = false;
  414. bool have_next_file = false;
  415. bool have_last_sequence = false;
  416. uint64_t next_file = 0;
  417. uint64_t last_sequence = 0;
  418. uint64_t log_number = 0;
  419. uint64_t prev_log_number = 0;
  420. Builder builder(this, current_);
  421. {
  422. LogReporter reporter;
  423. reporter.status = &s;
  424. log::Reader reader(file, &reporter, true/*checksum*/);
  425. Slice record;
  426. std::string scratch;
  427. while (reader.ReadRecord(&record, &scratch) && s.ok()) {
  428. VersionEdit edit;
  429. s = edit.DecodeFrom(record);
  430. if (s.ok()) {
  431. if (edit.has_comparator_ &&
  432. edit.comparator_ != icmp_.user_comparator()->Name()) {
  433. s = Status::InvalidArgument(
  434. edit.comparator_ + "does not match existing comparator ",
  435. icmp_.user_comparator()->Name());
  436. }
  437. }
  438. if (s.ok()) {
  439. builder.Apply(&edit);
  440. }
  441. if (edit.has_log_number_) {
  442. log_number = edit.log_number_;
  443. have_log_number = true;
  444. }
  445. if (edit.has_prev_log_number_) {
  446. prev_log_number = edit.prev_log_number_;
  447. have_prev_log_number = true;
  448. }
  449. if (edit.has_next_file_number_) {
  450. next_file = edit.next_file_number_;
  451. have_next_file = true;
  452. }
  453. if (edit.has_last_sequence_) {
  454. last_sequence = edit.last_sequence_;
  455. have_last_sequence = true;
  456. }
  457. }
  458. }
  459. delete file;
  460. file = NULL;
  461. if (s.ok()) {
  462. if (!have_next_file) {
  463. s = Status::Corruption("no meta-nextfile entry in descriptor");
  464. } else if (!have_log_number) {
  465. s = Status::Corruption("no meta-lognumber entry in descriptor");
  466. } else if (!have_last_sequence) {
  467. s = Status::Corruption("no last-sequence-number entry in descriptor");
  468. }
  469. if (!have_prev_log_number) {
  470. prev_log_number = 0;
  471. }
  472. }
  473. if (s.ok()) {
  474. Version* v = new Version(this);
  475. builder.SaveTo(v);
  476. s = Finalize(v);
  477. if (!s.ok()) {
  478. delete v;
  479. } else {
  480. // Install recovered version
  481. v->next_ = NULL;
  482. current_->next_ = v;
  483. current_ = v;
  484. manifest_file_number_ = next_file;
  485. next_file_number_ = next_file + 1;
  486. last_sequence_ = last_sequence;
  487. log_number_ = log_number;
  488. prev_log_number_ = prev_log_number;
  489. }
  490. }
  491. return s;
  492. }
  493. static int64_t TotalFileSize(const std::vector<FileMetaData*>& files) {
  494. int64_t sum = 0;
  495. for (int i = 0; i < files.size(); i++) {
  496. sum += files[i]->file_size;
  497. }
  498. return sum;
  499. }
  500. Status VersionSet::Finalize(Version* v) {
  501. // Precomputed best level for next compaction
  502. int best_level = -1;
  503. double best_score = -1;
  504. Status s;
  505. for (int level = 0; s.ok() && level < config::kNumLevels-1; level++) {
  506. s = SortLevel(v, level);
  507. double score;
  508. if (level == 0) {
  509. // We treat level-0 specially by bounding the number of files
  510. // instead of number of bytes for two reasons:
  511. //
  512. // (1) With larger write-buffer sizes, it is nice not to do too
  513. // many level-0 compactions.
  514. //
  515. // (2) The files in level-0 are merged on every read and
  516. // therefore we wish to avoid too many files when the individual
  517. // file size is small (perhaps because of a small write-buffer
  518. // setting, or very high compression ratios, or lots of
  519. // overwrites/deletions).
  520. score = v->files_[level].size() / 4.0;
  521. } else {
  522. // Compute the ratio of current size to size limit.
  523. const uint64_t level_bytes = TotalFileSize(v->files_[level]);
  524. score = static_cast<double>(level_bytes) / MaxBytesForLevel(level);
  525. }
  526. if (score > best_score) {
  527. best_level = level;
  528. best_score = score;
  529. }
  530. }
  531. v->compaction_level_ = best_level;
  532. v->compaction_score_ = best_score;
  533. return s;
  534. }
  535. Status VersionSet::WriteSnapshot(log::Writer* log) {
  536. // TODO: Break up into multiple records to reduce memory usage on recovery?
  537. // Save metadata
  538. VersionEdit edit;
  539. edit.SetComparatorName(icmp_.user_comparator()->Name());
  540. // Save compaction pointers
  541. for (int level = 0; level < config::kNumLevels; level++) {
  542. if (!compact_pointer_[level].empty()) {
  543. InternalKey key;
  544. key.DecodeFrom(compact_pointer_[level]);
  545. edit.SetCompactPointer(level, key);
  546. }
  547. }
  548. // Save files
  549. for (int level = 0; level < config::kNumLevels; level++) {
  550. const std::vector<FileMetaData*>& files = current_->files_[level];
  551. for (int i = 0; i < files.size(); i++) {
  552. const FileMetaData* f = files[i];
  553. edit.AddFile(level, f->number, f->file_size, f->smallest, f->largest);
  554. }
  555. }
  556. // Save large value refs
  557. for (LargeValueMap::const_iterator it = large_value_refs_.begin();
  558. it != large_value_refs_.end();
  559. ++it) {
  560. const LargeValueRef& ref = it->first;
  561. const LargeReferencesSet& pointers = it->second;
  562. for (LargeReferencesSet::const_iterator j = pointers.begin();
  563. j != pointers.end();
  564. ++j) {
  565. edit.AddLargeValueRef(ref, j->first, j->second);
  566. }
  567. }
  568. std::string record;
  569. edit.EncodeTo(&record);
  570. return log->AddRecord(record);
  571. }
  572. // Helper to sort by tables_[file_number].smallest
  573. struct VersionSet::BySmallestKey {
  574. const InternalKeyComparator* internal_comparator;
  575. bool operator()(FileMetaData* f1, FileMetaData* f2) const {
  576. return internal_comparator->Compare(f1->smallest, f2->smallest) < 0;
  577. }
  578. };
  579. Status VersionSet::SortLevel(Version* v, uint64_t level) {
  580. Status result;
  581. BySmallestKey cmp;
  582. cmp.internal_comparator = &icmp_;
  583. std::sort(v->files_[level].begin(), v->files_[level].end(), cmp);
  584. if (result.ok() && level > 0) {
  585. // There should be no overlap
  586. for (int i = 1; i < v->files_[level].size(); i++) {
  587. const InternalKey& prev_end = v->files_[level][i-1]->largest;
  588. const InternalKey& this_begin = v->files_[level][i]->smallest;
  589. if (icmp_.Compare(prev_end, this_begin) >= 0) {
  590. result = Status::Corruption(
  591. "overlapping ranges in same level",
  592. (EscapeString(prev_end.Encode()) + " vs. " +
  593. EscapeString(this_begin.Encode())));
  594. break;
  595. }
  596. }
  597. }
  598. return result;
  599. }
  600. int VersionSet::NumLevelFiles(int level) const {
  601. assert(level >= 0);
  602. assert(level < config::kNumLevels);
  603. return current_->files_[level].size();
  604. }
  605. uint64_t VersionSet::ApproximateOffsetOf(Version* v, const InternalKey& ikey) {
  606. uint64_t result = 0;
  607. for (int level = 0; level < config::kNumLevels; level++) {
  608. const std::vector<FileMetaData*>& files = v->files_[level];
  609. for (int i = 0; i < files.size(); i++) {
  610. if (icmp_.Compare(files[i]->largest, ikey) <= 0) {
  611. // Entire file is before "ikey", so just add the file size
  612. result += files[i]->file_size;
  613. } else if (icmp_.Compare(files[i]->smallest, ikey) > 0) {
  614. // Entire file is after "ikey", so ignore
  615. if (level > 0) {
  616. // Files other than level 0 are sorted by meta->smallest, so
  617. // no further files in this level will contain data for
  618. // "ikey".
  619. break;
  620. }
  621. } else {
  622. // "ikey" falls in the range for this table. Add the
  623. // approximate offset of "ikey" within the table.
  624. Table* tableptr;
  625. Iterator* iter = table_cache_->NewIterator(
  626. ReadOptions(), files[i]->number, files[i]->file_size, &tableptr);
  627. if (tableptr != NULL) {
  628. result += tableptr->ApproximateOffsetOf(ikey.Encode());
  629. }
  630. delete iter;
  631. }
  632. }
  633. }
  634. // Add in large value files which are references from internal keys
  635. // stored in the table files
  636. //
  637. // TODO(opt): this is O(# large values in db). If this becomes too slow,
  638. // we could store an auxiliary data structure indexed by internal key
  639. for (LargeValueMap::const_iterator it = large_value_refs_.begin();
  640. it != large_value_refs_.end();
  641. ++it) {
  642. const LargeValueRef& lref = it->first;
  643. for (LargeReferencesSet::const_iterator it2 = it->second.begin();
  644. it2 != it->second.end();
  645. ++it2) {
  646. if (icmp_.Compare(it2->second, ikey.Encode()) <= 0) {
  647. // Internal key for large value is before our key of interest
  648. result += lref.ValueSize();
  649. }
  650. }
  651. }
  652. return result;
  653. }
  654. bool VersionSet::RegisterLargeValueRef(const LargeValueRef& large_ref,
  655. uint64_t fnum,
  656. const InternalKey& internal_key) {
  657. LargeReferencesSet* refs = &large_value_refs_[large_ref];
  658. bool is_first = refs->empty();
  659. refs->insert(make_pair(fnum, internal_key.Encode().ToString()));
  660. return is_first;
  661. }
  662. void VersionSet::CleanupLargeValueRefs(const std::set<uint64_t>& live_tables) {
  663. for (LargeValueMap::iterator it = large_value_refs_.begin();
  664. it != large_value_refs_.end();
  665. ) {
  666. LargeReferencesSet* refs = &it->second;
  667. for (LargeReferencesSet::iterator ref_it = refs->begin();
  668. ref_it != refs->end();
  669. ) {
  670. if (ref_it->first != log_number_ && // Not in log file
  671. ref_it->first != prev_log_number_ && // Not in prev log
  672. live_tables.count(ref_it->first) == 0) { // Not in a live table
  673. // No longer live: erase
  674. LargeReferencesSet::iterator to_erase = ref_it;
  675. ++ref_it;
  676. refs->erase(to_erase);
  677. } else {
  678. // Still live: leave this reference alone
  679. ++ref_it;
  680. }
  681. }
  682. if (refs->empty()) {
  683. // No longer any live references to this large value: remove from
  684. // large_value_refs
  685. Log(env_, options_->info_log, "large value is dead: '%s'",
  686. LargeValueRefToFilenameString(it->first).c_str());
  687. LargeValueMap::iterator to_erase = it;
  688. ++it;
  689. large_value_refs_.erase(to_erase);
  690. } else {
  691. ++it;
  692. }
  693. }
  694. }
  695. bool VersionSet::LargeValueIsLive(const LargeValueRef& large_ref) {
  696. LargeValueMap::iterator it = large_value_refs_.find(large_ref);
  697. if (it == large_value_refs_.end()) {
  698. return false;
  699. } else {
  700. assert(!it->second.empty());
  701. return true;
  702. }
  703. }
  704. void VersionSet::MaybeDeleteOldVersions() {
  705. // Note: it is important to delete versions in order since a newer
  706. // version with zero refs may be holding a pointer to a memtable
  707. // that is used by somebody who has a ref on an older version.
  708. while (oldest_ != current_ && oldest_->refs_ == 0) {
  709. Version* next = oldest_->next_;
  710. delete oldest_;
  711. oldest_ = next;
  712. }
  713. }
  714. void VersionSet::AddLiveFiles(std::set<uint64_t>* live) {
  715. for (Version* v = oldest_; v != NULL; v = v->next_) {
  716. for (int level = 0; level < config::kNumLevels; level++) {
  717. const std::vector<FileMetaData*>& files = v->files_[level];
  718. for (int i = 0; i < files.size(); i++) {
  719. live->insert(files[i]->number);
  720. }
  721. }
  722. }
  723. }
  724. int64_t VersionSet::NumLevelBytes(int level) const {
  725. assert(level >= 0);
  726. assert(level < config::kNumLevels);
  727. return TotalFileSize(current_->files_[level]);
  728. }
  729. int64_t VersionSet::MaxNextLevelOverlappingBytes() {
  730. int64_t result = 0;
  731. std::vector<FileMetaData*> overlaps;
  732. for (int level = 0; level < config::kNumLevels - 1; level++) {
  733. for (int i = 0; i < current_->files_[level].size(); i++) {
  734. const FileMetaData* f = current_->files_[level][i];
  735. GetOverlappingInputs(level+1, f->smallest, f->largest, &overlaps);
  736. const int64_t sum = TotalFileSize(overlaps);
  737. if (sum > result) {
  738. result = sum;
  739. }
  740. }
  741. }
  742. return result;
  743. }
  744. // Store in "*inputs" all files in "level" that overlap [begin,end]
  745. void VersionSet::GetOverlappingInputs(
  746. int level,
  747. const InternalKey& begin,
  748. const InternalKey& end,
  749. std::vector<FileMetaData*>* inputs) {
  750. inputs->clear();
  751. Slice user_begin = begin.user_key();
  752. Slice user_end = end.user_key();
  753. const Comparator* user_cmp = icmp_.user_comparator();
  754. for (int i = 0; i < current_->files_[level].size(); i++) {
  755. FileMetaData* f = current_->files_[level][i];
  756. if (user_cmp->Compare(f->largest.user_key(), user_begin) < 0 ||
  757. user_cmp->Compare(f->smallest.user_key(), user_end) > 0) {
  758. // Either completely before or after range; skip it
  759. } else {
  760. inputs->push_back(f);
  761. }
  762. }
  763. }
  764. // Stores the minimal range that covers all entries in inputs in
  765. // *smallest, *largest.
  766. // REQUIRES: inputs is not empty
  767. void VersionSet::GetRange(const std::vector<FileMetaData*>& inputs,
  768. InternalKey* smallest,
  769. InternalKey* largest) {
  770. assert(!inputs.empty());
  771. smallest->Clear();
  772. largest->Clear();
  773. for (int i = 0; i < inputs.size(); i++) {
  774. FileMetaData* f = inputs[i];
  775. if (i == 0) {
  776. *smallest = f->smallest;
  777. *largest = f->largest;
  778. } else {
  779. if (icmp_.Compare(f->smallest, *smallest) < 0) {
  780. *smallest = f->smallest;
  781. }
  782. if (icmp_.Compare(f->largest, *largest) > 0) {
  783. *largest = f->largest;
  784. }
  785. }
  786. }
  787. }
  788. // Stores the minimal range that covers all entries in inputs1 and inputs2
  789. // in *smallest, *largest.
  790. // REQUIRES: inputs is not empty
  791. void VersionSet::GetRange2(const std::vector<FileMetaData*>& inputs1,
  792. const std::vector<FileMetaData*>& inputs2,
  793. InternalKey* smallest,
  794. InternalKey* largest) {
  795. std::vector<FileMetaData*> all = inputs1;
  796. all.insert(all.end(), inputs2.begin(), inputs2.end());
  797. GetRange(all, smallest, largest);
  798. }
  799. Iterator* VersionSet::MakeInputIterator(Compaction* c) {
  800. ReadOptions options;
  801. options.verify_checksums = options_->paranoid_checks;
  802. options.fill_cache = false;
  803. // Level-0 files have to be merged together. For other levels,
  804. // we will make a concatenating iterator per level.
  805. // TODO(opt): use concatenating iterator for level-0 if there is no overlap
  806. const int space = (c->level() == 0 ? c->inputs_[0].size() + 1 : 2);
  807. Iterator** list = new Iterator*[space];
  808. int num = 0;
  809. for (int which = 0; which < 2; which++) {
  810. if (!c->inputs_[which].empty()) {
  811. if (c->level() + which == 0) {
  812. const std::vector<FileMetaData*>& files = c->inputs_[which];
  813. for (int i = 0; i < files.size(); i++) {
  814. list[num++] = table_cache_->NewIterator(
  815. options, files[i]->number, files[i]->file_size);
  816. }
  817. } else {
  818. // Create concatenating iterator for the files from this level
  819. list[num++] = NewTwoLevelIterator(
  820. new Version::LevelFileNumIterator(
  821. c->input_version_, &c->inputs_[which]),
  822. &GetFileIterator, table_cache_, options);
  823. }
  824. }
  825. }
  826. assert(num <= space);
  827. Iterator* result = NewMergingIterator(&icmp_, list, num);
  828. delete[] list;
  829. return result;
  830. }
  831. Compaction* VersionSet::PickCompaction() {
  832. if (!NeedsCompaction()) {
  833. return NULL;
  834. }
  835. const int level = current_->compaction_level_;
  836. assert(level >= 0);
  837. assert(level+1 < config::kNumLevels);
  838. Compaction* c = new Compaction(level);
  839. c->input_version_ = current_;
  840. c->input_version_->Ref();
  841. // Pick the first file that comes after compact_pointer_[level]
  842. for (int i = 0; i < current_->files_[level].size(); i++) {
  843. FileMetaData* f = current_->files_[level][i];
  844. if (compact_pointer_[level].empty() ||
  845. icmp_.Compare(f->largest.Encode(), compact_pointer_[level]) > 0) {
  846. c->inputs_[0].push_back(f);
  847. break;
  848. }
  849. }
  850. if (c->inputs_[0].empty()) {
  851. // Wrap-around to the beginning of the key space
  852. c->inputs_[0].push_back(current_->files_[level][0]);
  853. }
  854. // Files in level 0 may overlap each other, so pick up all overlapping ones
  855. if (level == 0) {
  856. InternalKey smallest, largest;
  857. GetRange(c->inputs_[0], &smallest, &largest);
  858. // Note that the next call will discard the file we placed in
  859. // c->inputs_[0] earlier and replace it with an overlapping set
  860. // which will include the picked file.
  861. GetOverlappingInputs(0, smallest, largest, &c->inputs_[0]);
  862. assert(!c->inputs_[0].empty());
  863. }
  864. SetupOtherInputs(c);
  865. return c;
  866. }
  867. void VersionSet::SetupOtherInputs(Compaction* c) {
  868. const int level = c->level();
  869. InternalKey smallest, largest;
  870. GetRange(c->inputs_[0], &smallest, &largest);
  871. GetOverlappingInputs(level+1, smallest, largest, &c->inputs_[1]);
  872. // Get entire range covered by compaction
  873. InternalKey all_start, all_limit;
  874. GetRange2(c->inputs_[0], c->inputs_[1], &all_start, &all_limit);
  875. // See if we can grow the number of inputs in "level" without
  876. // changing the number of "level+1" files we pick up.
  877. if (!c->inputs_[1].empty()) {
  878. std::vector<FileMetaData*> expanded0;
  879. GetOverlappingInputs(level, all_start, all_limit, &expanded0);
  880. if (expanded0.size() > c->inputs_[0].size()) {
  881. InternalKey new_start, new_limit;
  882. GetRange(expanded0, &new_start, &new_limit);
  883. std::vector<FileMetaData*> expanded1;
  884. GetOverlappingInputs(level+1, new_start, new_limit, &expanded1);
  885. if (expanded1.size() == c->inputs_[1].size()) {
  886. Log(env_, options_->info_log,
  887. "Expanding@%d %d+%d to %d+%d\n",
  888. level,
  889. int(c->inputs_[0].size()),
  890. int(c->inputs_[1].size()),
  891. int(expanded0.size()),
  892. int(expanded1.size()));
  893. smallest = new_start;
  894. largest = new_limit;
  895. c->inputs_[0] = expanded0;
  896. c->inputs_[1] = expanded1;
  897. GetRange2(c->inputs_[0], c->inputs_[1], &all_start, &all_limit);
  898. }
  899. }
  900. }
  901. // Compute the set of grandparent files that overlap this compaction
  902. // (parent == level+1; grandparent == level+2)
  903. if (level + 2 < config::kNumLevels) {
  904. GetOverlappingInputs(level + 2, all_start, all_limit, &c->grandparents_);
  905. }
  906. if (false) {
  907. Log(env_, options_->info_log, "Compacting %d '%s' .. '%s'",
  908. level,
  909. EscapeString(smallest.Encode()).c_str(),
  910. EscapeString(largest.Encode()).c_str());
  911. }
  912. // Update the place where we will do the next compaction for this level.
  913. // We update this immediately instead of waiting for the VersionEdit
  914. // to be applied so that if the compaction fails, we will try a different
  915. // key range next time.
  916. compact_pointer_[level] = largest.Encode().ToString();
  917. c->edit_.SetCompactPointer(level, largest);
  918. }
  919. Compaction* VersionSet::CompactRange(
  920. int level,
  921. const InternalKey& begin,
  922. const InternalKey& end) {
  923. std::vector<FileMetaData*> inputs;
  924. GetOverlappingInputs(level, begin, end, &inputs);
  925. if (inputs.empty()) {
  926. return NULL;
  927. }
  928. Compaction* c = new Compaction(level);
  929. c->input_version_ = current_;
  930. c->input_version_->Ref();
  931. c->inputs_[0] = inputs;
  932. SetupOtherInputs(c);
  933. return c;
  934. }
  935. Compaction::Compaction(int level)
  936. : level_(level),
  937. max_output_file_size_(MaxFileSizeForLevel(level)),
  938. input_version_(NULL),
  939. grandparent_index_(0),
  940. seen_key_(false),
  941. overlapped_bytes_(0) {
  942. for (int i = 0; i < config::kNumLevels; i++) {
  943. level_ptrs_[i] = 0;
  944. }
  945. }
  946. Compaction::~Compaction() {
  947. if (input_version_ != NULL) {
  948. input_version_->Unref();
  949. }
  950. }
  951. bool Compaction::IsTrivialMove() const {
  952. // Avoid a move if there is lots of overlapping grandparent data.
  953. // Otherwise, the move could create a parent file that will require
  954. // a very expensive merge later on.
  955. return (num_input_files(0) == 1 &&
  956. num_input_files(1) == 0 &&
  957. TotalFileSize(grandparents_) <= kMaxGrandParentOverlapBytes);
  958. }
  959. void Compaction::AddInputDeletions(VersionEdit* edit) {
  960. for (int which = 0; which < 2; which++) {
  961. for (int i = 0; i < inputs_[which].size(); i++) {
  962. edit->DeleteFile(level_ + which, inputs_[which][i]->number);
  963. }
  964. }
  965. }
  966. bool Compaction::IsBaseLevelForKey(const Slice& user_key) {
  967. // Maybe use binary search to find right entry instead of linear search?
  968. const Comparator* user_cmp = input_version_->vset_->icmp_.user_comparator();
  969. for (int lvl = level_ + 2; lvl < config::kNumLevels; lvl++) {
  970. const std::vector<FileMetaData*>& files = input_version_->files_[lvl];
  971. for (; level_ptrs_[lvl] < files.size(); ) {
  972. FileMetaData* f = files[level_ptrs_[lvl]];
  973. if (user_cmp->Compare(user_key, f->largest.user_key()) <= 0) {
  974. // We've advanced far enough
  975. if (user_cmp->Compare(user_key, f->smallest.user_key()) >= 0) {
  976. // Key falls in this file's range, so definitely not base level
  977. return false;
  978. }
  979. break;
  980. }
  981. level_ptrs_[lvl]++;
  982. }
  983. }
  984. return true;
  985. }
  986. bool Compaction::ShouldStopBefore(const InternalKey& key) {
  987. // Scan to find earliest grandparent file that contains key.
  988. const InternalKeyComparator* icmp = &input_version_->vset_->icmp_;
  989. while (grandparent_index_ < grandparents_.size() &&
  990. icmp->Compare(key, grandparents_[grandparent_index_]->largest) > 0) {
  991. if (seen_key_) {
  992. overlapped_bytes_ += grandparents_[grandparent_index_]->file_size;
  993. }
  994. grandparent_index_++;
  995. }
  996. seen_key_ = true;
  997. if (overlapped_bytes_ > kMaxGrandParentOverlapBytes) {
  998. // Too much overlap for current output; start new output
  999. overlapped_bytes_ = 0;
  1000. return true;
  1001. } else {
  1002. return false;
  1003. }
  1004. }
  1005. void Compaction::ReleaseInputs() {
  1006. if (input_version_ != NULL) {
  1007. input_version_->Unref();
  1008. input_version_ = NULL;
  1009. }
  1010. }
  1011. }