作者: 谢瑞阳 10225101483 徐翔宇 10225101535
Du kan inte välja fler än 25 ämnen Ämnen måste starta med en bokstav eller siffra, kan innehålla bindestreck ('-') och vara max 35 tecken långa.

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