作者: 韩晨旭 10225101440 李畅 10225102463
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.

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