db_impl.cc 47 KB

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