db_impl.cc 48 KB

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