10225501448 李度 10225101546 陈胤遒 10215501422 高宇菲
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.

1667 lines
52 KiB

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