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