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.

1631 lines
51 KiB

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