You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

db_impl.cc 46KB

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