diff --git a/src/leveldb/AUTHORS b/src/leveldb/AUTHORS index 2439d7a4..fc40194a 100644 --- a/src/leveldb/AUTHORS +++ b/src/leveldb/AUTHORS @@ -9,4 +9,3 @@ Sanjay Ghemawat # Partial list of contributors: Kevin Regan -Johan Bilien diff --git a/src/leveldb/Makefile b/src/leveldb/Makefile index f8903b69..38b9bf77 100644 --- a/src/leveldb/Makefile +++ b/src/leveldb/Makefile @@ -12,7 +12,7 @@ OPT ?= -O2 -DNDEBUG # (A) Production use (optimized mode) #----------------------------------------------- # detect what platform we're building on -$(shell CC="$(CC)" CXX="$(CXX)" TARGET_OS="$(TARGET_OS)" \ +$(shell CC=$(CC) CXX=$(CXX) TARGET_OS=$(TARGET_OS) \ ./build_detect_platform build_config.mk ./) # this file is generated by the previous line to set build flags and sources include build_config.mk @@ -31,7 +31,6 @@ TESTHARNESS = ./util/testharness.o $(TESTUTIL) TESTS = \ arena_test \ - autocompact_test \ bloom_test \ c_test \ cache_test \ @@ -44,7 +43,6 @@ TESTS = \ filename_test \ filter_block_test \ issue178_test \ - issue200_test \ log_test \ memenv_test \ skiplist_test \ @@ -72,7 +70,7 @@ SHARED = $(SHARED1) else # Update db.h if you change these. SHARED_MAJOR = 1 -SHARED_MINOR = 17 +SHARED_MINOR = 12 SHARED1 = libleveldb.$(PLATFORM_SHARED_EXT) SHARED2 = $(SHARED1).$(SHARED_MAJOR) SHARED3 = $(SHARED1).$(SHARED_MAJOR).$(SHARED_MINOR) @@ -116,9 +114,6 @@ leveldbutil: db/leveldb_main.o $(LIBOBJECTS) arena_test: util/arena_test.o $(LIBOBJECTS) $(TESTHARNESS) $(CXX) $(LDFLAGS) util/arena_test.o $(LIBOBJECTS) $(TESTHARNESS) -o $@ $(LIBS) -autocompact_test: db/autocompact_test.o $(LIBOBJECTS) $(TESTHARNESS) - $(CXX) $(LDFLAGS) db/autocompact_test.o $(LIBOBJECTS) $(TESTHARNESS) -o $@ $(LIBS) - bloom_test: util/bloom_test.o $(LIBOBJECTS) $(TESTHARNESS) $(CXX) $(LDFLAGS) util/bloom_test.o $(LIBOBJECTS) $(TESTHARNESS) -o $@ $(LIBS) @@ -155,9 +150,6 @@ filter_block_test: table/filter_block_test.o $(LIBOBJECTS) $(TESTHARNESS) issue178_test: issues/issue178_test.o $(LIBOBJECTS) $(TESTHARNESS) $(CXX) $(LDFLAGS) issues/issue178_test.o $(LIBOBJECTS) $(TESTHARNESS) -o $@ $(LIBS) -issue200_test: issues/issue200_test.o $(LIBOBJECTS) $(TESTHARNESS) - $(CXX) $(LDFLAGS) issues/issue200_test.o $(LIBOBJECTS) $(TESTHARNESS) -o $@ $(LIBS) - log_test: db/log_test.o $(LIBOBJECTS) $(TESTHARNESS) $(CXX) $(LDFLAGS) db/log_test.o $(LIBOBJECTS) $(TESTHARNESS) -o $@ $(LIBS) @@ -190,20 +182,19 @@ PLATFORMSROOT=/Applications/Xcode.app/Contents/Developer/Platforms SIMULATORROOT=$(PLATFORMSROOT)/iPhoneSimulator.platform/Developer DEVICEROOT=$(PLATFORMSROOT)/iPhoneOS.platform/Developer IOSVERSION=$(shell defaults read $(PLATFORMSROOT)/iPhoneOS.platform/version CFBundleShortVersionString) -IOSARCH=-arch armv6 -arch armv7 -arch armv7s -arch arm64 .cc.o: mkdir -p ios-x86/$(dir $@) - $(CXX) $(CXXFLAGS) -isysroot $(SIMULATORROOT)/SDKs/iPhoneSimulator$(IOSVERSION).sdk -arch i686 -arch x86_64 -c $< -o ios-x86/$@ + $(CXX) $(CXXFLAGS) -isysroot $(SIMULATORROOT)/SDKs/iPhoneSimulator$(IOSVERSION).sdk -arch i686 -c $< -o ios-x86/$@ mkdir -p ios-arm/$(dir $@) - xcrun -sdk iphoneos $(CXX) $(CXXFLAGS) -isysroot $(DEVICEROOT)/SDKs/iPhoneOS$(IOSVERSION).sdk $(IOSARCH) -c $< -o ios-arm/$@ + $(DEVICEROOT)/usr/bin/$(CXX) $(CXXFLAGS) -isysroot $(DEVICEROOT)/SDKs/iPhoneOS$(IOSVERSION).sdk -arch armv6 -arch armv7 -c $< -o ios-arm/$@ lipo ios-x86/$@ ios-arm/$@ -create -output $@ .c.o: mkdir -p ios-x86/$(dir $@) - $(CC) $(CFLAGS) -isysroot $(SIMULATORROOT)/SDKs/iPhoneSimulator$(IOSVERSION).sdk -arch i686 -arch x86_64 -c $< -o ios-x86/$@ + $(CC) $(CFLAGS) -isysroot $(SIMULATORROOT)/SDKs/iPhoneSimulator$(IOSVERSION).sdk -arch i686 -c $< -o ios-x86/$@ mkdir -p ios-arm/$(dir $@) - xcrun -sdk iphoneos $(CC) $(CFLAGS) -isysroot $(DEVICEROOT)/SDKs/iPhoneOS$(IOSVERSION).sdk $(IOSARCH) -c $< -o ios-arm/$@ + $(DEVICEROOT)/usr/bin/$(CC) $(CFLAGS) -isysroot $(DEVICEROOT)/SDKs/iPhoneOS$(IOSVERSION).sdk -arch armv6 -arch armv7 -c $< -o ios-arm/$@ lipo ios-x86/$@ ios-arm/$@ -create -output $@ else diff --git a/src/leveldb/build_detect_platform b/src/leveldb/build_detect_platform index 85b1ce02..a3ad057e 100755 --- a/src/leveldb/build_detect_platform +++ b/src/leveldb/build_detect_platform @@ -44,10 +44,6 @@ if test -z "$CXX"; then CXX=g++ fi -if test -z "$TMPDIR"; then - TMPDIR=/tmp -fi - # Detect OS if test -z "$TARGET_OS"; then TARGET_OS=`uname -s` @@ -137,16 +133,6 @@ case "$TARGET_OS" in # man ld: +h internal_name PLATFORM_SHARED_LDFLAGS="-shared -Wl,+h -Wl," ;; - IOS) - PLATFORM=IOS - COMMON_FLAGS="$MEMCMP_FLAG -DOS_MACOSX" - [ -z "$INSTALL_PATH" ] && INSTALL_PATH=`pwd` - PORT_FILE=port/port_posix.cc - PLATFORM_SHARED_EXT= - PLATFORM_SHARED_LDFLAGS= - PLATFORM_SHARED_CFLAGS= - PLATFORM_SHARED_VERSIONED= - ;; OS_WINDOWS_CROSSCOMPILE | NATIVE_WINDOWS) PLATFORM=OS_WINDOWS COMMON_FLAGS="-fno-builtin-memcmp -D_REENTRANT -DOS_WINDOWS -DLEVELDB_PLATFORM_WINDOWS -DWINVER=0x0500 -D__USE_MINGW_ANSI_STDIO=1" @@ -183,10 +169,8 @@ if [ "$CROSS_COMPILE" = "true" ]; then # Cross-compiling; do not try any compilation tests. true else - CXXOUTPUT="${TMPDIR}/leveldb_build_detect_platform-cxx.$$" - # If -std=c++0x works, use . Otherwise use port_posix.h. - $CXX $CXXFLAGS -std=c++0x -x c++ - -o $CXXOUTPUT 2>/dev/null </dev/null < int main() {} EOF @@ -198,14 +182,12 @@ EOF fi # Test whether tcmalloc is available - $CXX $CXXFLAGS -x c++ - -o $CXXOUTPUT -ltcmalloc 2>/dev/null </dev/null </dev/null fi PLATFORM_CCFLAGS="$PLATFORM_CCFLAGS $COMMON_FLAGS" diff --git a/src/leveldb/db/autocompact_test.cc b/src/leveldb/db/autocompact_test.cc deleted file mode 100644 index d20a2362..00000000 --- a/src/leveldb/db/autocompact_test.cc +++ /dev/null @@ -1,118 +0,0 @@ -// Copyright (c) 2013 The LevelDB Authors. All rights reserved. -// Use of this source code is governed by a BSD-style license that can be -// found in the LICENSE file. See the AUTHORS file for names of contributors. - -#include "leveldb/db.h" -#include "db/db_impl.h" -#include "leveldb/cache.h" -#include "util/testharness.h" -#include "util/testutil.h" - -namespace leveldb { - -class AutoCompactTest { - public: - std::string dbname_; - Cache* tiny_cache_; - Options options_; - DB* db_; - - AutoCompactTest() { - dbname_ = test::TmpDir() + "/autocompact_test"; - tiny_cache_ = NewLRUCache(100); - options_.block_cache = tiny_cache_; - DestroyDB(dbname_, options_); - options_.create_if_missing = true; - options_.compression = kNoCompression; - ASSERT_OK(DB::Open(options_, dbname_, &db_)); - } - - ~AutoCompactTest() { - delete db_; - DestroyDB(dbname_, Options()); - delete tiny_cache_; - } - - std::string Key(int i) { - char buf[100]; - snprintf(buf, sizeof(buf), "key%06d", i); - return std::string(buf); - } - - uint64_t Size(const Slice& start, const Slice& limit) { - Range r(start, limit); - uint64_t size; - db_->GetApproximateSizes(&r, 1, &size); - return size; - } - - void DoReads(int n); -}; - -static const int kValueSize = 200 * 1024; -static const int kTotalSize = 100 * 1024 * 1024; -static const int kCount = kTotalSize / kValueSize; - -// Read through the first n keys repeatedly and check that they get -// compacted (verified by checking the size of the key space). -void AutoCompactTest::DoReads(int n) { - std::string value(kValueSize, 'x'); - DBImpl* dbi = reinterpret_cast(db_); - - // Fill database - for (int i = 0; i < kCount; i++) { - ASSERT_OK(db_->Put(WriteOptions(), Key(i), value)); - } - ASSERT_OK(dbi->TEST_CompactMemTable()); - - // Delete everything - for (int i = 0; i < kCount; i++) { - ASSERT_OK(db_->Delete(WriteOptions(), Key(i))); - } - ASSERT_OK(dbi->TEST_CompactMemTable()); - - // Get initial measurement of the space we will be reading. - const int64_t initial_size = Size(Key(0), Key(n)); - const int64_t initial_other_size = Size(Key(n), Key(kCount)); - - // Read until size drops significantly. - std::string limit_key = Key(n); - for (int read = 0; true; read++) { - ASSERT_LT(read, 100) << "Taking too long to compact"; - Iterator* iter = db_->NewIterator(ReadOptions()); - for (iter->SeekToFirst(); - iter->Valid() && iter->key().ToString() < limit_key; - iter->Next()) { - // Drop data - } - delete iter; - // Wait a little bit to allow any triggered compactions to complete. - Env::Default()->SleepForMicroseconds(1000000); - uint64_t size = Size(Key(0), Key(n)); - fprintf(stderr, "iter %3d => %7.3f MB [other %7.3f MB]\n", - read+1, size/1048576.0, Size(Key(n), Key(kCount))/1048576.0); - if (size <= initial_size/10) { - break; - } - } - - // Verify that the size of the key space not touched by the reads - // is pretty much unchanged. - const int64_t final_other_size = Size(Key(n), Key(kCount)); - ASSERT_LE(final_other_size, initial_other_size + 1048576); - ASSERT_GE(final_other_size, initial_other_size/5 - 1048576); -} - -TEST(AutoCompactTest, ReadAll) { - DoReads(kCount); -} - -TEST(AutoCompactTest, ReadHalf) { - DoReads(kCount/2); -} - -} // namespace leveldb - -int main(int argc, char** argv) { - return leveldb::test::RunAllTests(); -} diff --git a/src/leveldb/db/corruption_test.cc b/src/leveldb/db/corruption_test.cc index 96afc689..31b2d5f4 100644 --- a/src/leveldb/db/corruption_test.cc +++ b/src/leveldb/db/corruption_test.cc @@ -35,7 +35,6 @@ class CorruptionTest { CorruptionTest() { tiny_cache_ = NewLRUCache(100); options_.env = &env_; - options_.block_cache = tiny_cache_; dbname_ = test::TmpDir() + "/db_test"; DestroyDB(dbname_, options_); @@ -51,14 +50,17 @@ class CorruptionTest { delete tiny_cache_; } - Status TryReopen() { + Status TryReopen(Options* options = NULL) { delete db_; db_ = NULL; - return DB::Open(options_, dbname_, &db_); + Options opt = (options ? *options : options_); + opt.env = &env_; + opt.block_cache = tiny_cache_; + return DB::Open(opt, dbname_, &db_); } - void Reopen() { - ASSERT_OK(TryReopen()); + void Reopen(Options* options = NULL) { + ASSERT_OK(TryReopen(options)); } void RepairDB() { @@ -75,13 +77,7 @@ class CorruptionTest { Slice key = Key(i, &key_space); batch.Clear(); batch.Put(key, Value(i, &value_space)); - WriteOptions options; - // Corrupt() doesn't work without this sync on windows; stat reports 0 for - // the file size. - if (i == n - 1) { - options.sync = true; - } - ASSERT_OK(db_->Write(options, &batch)); + ASSERT_OK(db_->Write(WriteOptions(), &batch)); } } @@ -96,10 +92,6 @@ class CorruptionTest { for (iter->SeekToFirst(); iter->Valid(); iter->Next()) { uint64_t key; Slice in(iter->key()); - if (in == "" || in == "~") { - // Ignore boundary keys. - continue; - } if (!ConsumeDecimalNumber(&in, &key) || !in.empty() || key < next_expected) { @@ -131,7 +123,7 @@ class CorruptionTest { FileType type; std::string fname; int picked_number = -1; - for (size_t i = 0; i < filenames.size(); i++) { + for (int i = 0; i < filenames.size(); i++) { if (ParseFileName(filenames[i], &number, &type) && type == filetype && int(number) > picked_number) { // Pick latest file @@ -241,23 +233,7 @@ TEST(CorruptionTest, TableFile) { dbi->TEST_CompactRange(1, NULL, NULL); Corrupt(kTableFile, 100, 1); - Check(90, 99); -} - -TEST(CorruptionTest, TableFileRepair) { - options_.block_size = 2 * kValueSize; // Limit scope of corruption - options_.paranoid_checks = true; - Reopen(); - Build(100); - DBImpl* dbi = reinterpret_cast(db_); - dbi->TEST_CompactMemTable(); - dbi->TEST_CompactRange(0, NULL, NULL); - dbi->TEST_CompactRange(1, NULL, NULL); - - Corrupt(kTableFile, 100, 1); - RepairDB(); - Reopen(); - Check(95, 99); + Check(99, 99); } TEST(CorruptionTest, TableFileIndexData) { @@ -323,7 +299,7 @@ TEST(CorruptionTest, CompactionInputError) { ASSERT_EQ(1, Property("leveldb.num-files-at-level" + NumberToString(last))); Corrupt(kTableFile, 100, 1); - Check(5, 9); + Check(9, 9); // Force compactions by writing lots of values Build(10000); @@ -331,23 +307,32 @@ TEST(CorruptionTest, CompactionInputError) { } TEST(CorruptionTest, CompactionInputErrorParanoid) { - options_.paranoid_checks = true; - options_.write_buffer_size = 512 << 10; - Reopen(); + Options options; + options.paranoid_checks = true; + options.write_buffer_size = 1048576; + Reopen(&options); DBImpl* dbi = reinterpret_cast(db_); - // Make multiple inputs so we need to compact. - for (int i = 0; i < 2; i++) { - Build(10); + // Fill levels >= 1 so memtable compaction outputs to level 1 + for (int level = 1; level < config::kNumLevels; level++) { + dbi->Put(WriteOptions(), "", "begin"); + dbi->Put(WriteOptions(), "~", "end"); dbi->TEST_CompactMemTable(); - Corrupt(kTableFile, 100, 1); - env_.SleepForMicroseconds(100000); } - dbi->CompactRange(NULL, NULL); - // Write must fail because of corrupted table + Build(10); + dbi->TEST_CompactMemTable(); + ASSERT_EQ(1, Property("leveldb.num-files-at-level0")); + + Corrupt(kTableFile, 100, 1); + Check(9, 9); + + // Write must eventually fail because of corrupted table + Status s; std::string tmp1, tmp2; - Status s = db_->Put(WriteOptions(), Key(5, &tmp1), Value(5, &tmp2)); + for (int i = 0; i < 10000 && s.ok(); i++) { + s = db_->Put(WriteOptions(), Key(i, &tmp1), Value(i, &tmp2)); + } ASSERT_TRUE(!s.ok()) << "write did not fail in corrupted paranoid db"; } diff --git a/src/leveldb/db/db_bench.cc b/src/leveldb/db/db_bench.cc index fc46d896..7abdf875 100644 --- a/src/leveldb/db/db_bench.cc +++ b/src/leveldb/db/db_bench.cc @@ -128,7 +128,7 @@ class RandomGenerator { pos_ = 0; } - Slice Generate(size_t len) { + Slice Generate(int len) { if (pos_ + len > data_.size()) { pos_ = 0; assert(len < data_.size()); @@ -139,11 +139,11 @@ class RandomGenerator { }; static Slice TrimSpace(Slice s) { - size_t start = 0; + int start = 0; while (start < s.size() && isspace(s[start])) { start++; } - size_t limit = s.size(); + int limit = s.size(); while (limit > start && isspace(s[limit-1])) { limit--; } @@ -399,7 +399,7 @@ class Benchmark { heap_counter_(0) { std::vector files; Env::Default()->GetChildren(FLAGS_db, &files); - for (size_t i = 0; i < files.size(); i++) { + for (int i = 0; i < files.size(); i++) { if (Slice(files[i]).starts_with("heap-")) { Env::Default()->DeleteFile(std::string(FLAGS_db) + "/" + files[i]); } diff --git a/src/leveldb/db/db_impl.cc b/src/leveldb/db/db_impl.cc index faf5e7d7..af02467b 100644 --- a/src/leveldb/db/db_impl.cc +++ b/src/leveldb/db/db_impl.cc @@ -113,14 +113,14 @@ Options SanitizeOptions(const std::string& dbname, return result; } -DBImpl::DBImpl(const Options& raw_options, const std::string& dbname) - : env_(raw_options.env), - internal_comparator_(raw_options.comparator), - internal_filter_policy_(raw_options.filter_policy), - options_(SanitizeOptions(dbname, &internal_comparator_, - &internal_filter_policy_, raw_options)), - owns_info_log_(options_.info_log != raw_options.info_log), - owns_cache_(options_.block_cache != raw_options.block_cache), +DBImpl::DBImpl(const Options& options, const std::string& dbname) + : env_(options.env), + internal_comparator_(options.comparator), + internal_filter_policy_(options.filter_policy), + options_(SanitizeOptions( + dbname, &internal_comparator_, &internal_filter_policy_, options)), + owns_info_log_(options_.info_log != options.info_log), + owns_cache_(options_.block_cache != options.block_cache), dbname_(dbname), db_lock_(NULL), shutting_down_(NULL), @@ -130,15 +130,15 @@ DBImpl::DBImpl(const Options& raw_options, const std::string& dbname) logfile_(NULL), logfile_number_(0), log_(NULL), - seed_(0), tmp_batch_(new WriteBatch), bg_compaction_scheduled_(false), - manual_compaction_(NULL) { + manual_compaction_(NULL), + consecutive_compaction_errors_(0) { mem_->Ref(); has_imm_.Release_Store(NULL); // Reserve ten files or so for other uses and give the rest to TableCache. - const int table_cache_size = options_.max_open_files - kNumNonTableCacheFiles; + const int table_cache_size = options.max_open_files - kNumNonTableCacheFiles; table_cache_ = new TableCache(dbname_, &options_, table_cache_size); versions_ = new VersionSet(dbname_, &options_, table_cache_, @@ -216,12 +216,6 @@ void DBImpl::MaybeIgnoreError(Status* s) const { } void DBImpl::DeleteObsoleteFiles() { - if (!bg_error_.ok()) { - // After a background error, we don't know whether a new version may - // or may not have been committed, so we cannot safely garbage collect. - return; - } - // Make a set of all of the live files std::set live = pending_outputs_; versions_->AddLiveFiles(&live); @@ -328,7 +322,7 @@ Status DBImpl::Recover(VersionEdit* edit) { if (ParseFileName(filenames[i], &number, &type)) { expected.erase(number); if (type == kLogFile && ((number >= min_log) || (number == prev_log))) - logs.push_back(number); + logs.push_back(number); } } if (!expected.empty()) { @@ -500,7 +494,7 @@ Status DBImpl::WriteLevel0Table(MemTable* mem, VersionEdit* edit, return s; } -void DBImpl::CompactMemTable() { +Status DBImpl::CompactMemTable() { mutex_.AssertHeld(); assert(imm_ != NULL); @@ -528,9 +522,9 @@ void DBImpl::CompactMemTable() { imm_ = NULL; has_imm_.Release_Store(NULL); DeleteObsoleteFiles(); - } else { - RecordBackgroundError(s); } + + return s; } void DBImpl::CompactRange(const Slice* begin, const Slice* end) { @@ -573,17 +567,15 @@ void DBImpl::TEST_CompactRange(int level, const Slice* begin,const Slice* end) { } MutexLock l(&mutex_); - while (!manual.done && !shutting_down_.Acquire_Load() && bg_error_.ok()) { - if (manual_compaction_ == NULL) { // Idle - manual_compaction_ = &manual; - MaybeScheduleCompaction(); - } else { // Running either my compaction or another compaction. + while (!manual.done) { + while (manual_compaction_ != NULL) { + bg_cv_.Wait(); + } + manual_compaction_ = &manual; + MaybeScheduleCompaction(); + while (manual_compaction_ == &manual) { bg_cv_.Wait(); } - } - if (manual_compaction_ == &manual) { - // Cancel my manual compaction since we aborted early for some reason. - manual_compaction_ = NULL; } } @@ -603,22 +595,12 @@ Status DBImpl::TEST_CompactMemTable() { return s; } -void DBImpl::RecordBackgroundError(const Status& s) { - mutex_.AssertHeld(); - if (bg_error_.ok()) { - bg_error_ = s; - bg_cv_.SignalAll(); - } -} - void DBImpl::MaybeScheduleCompaction() { mutex_.AssertHeld(); if (bg_compaction_scheduled_) { // Already scheduled } else if (shutting_down_.Acquire_Load()) { // DB is being deleted; no more background compactions - } else if (!bg_error_.ok()) { - // Already got an error; no more changes } else if (imm_ == NULL && manual_compaction_ == NULL && !versions_->NeedsCompaction()) { @@ -636,12 +618,30 @@ void DBImpl::BGWork(void* db) { void DBImpl::BackgroundCall() { MutexLock l(&mutex_); assert(bg_compaction_scheduled_); - if (shutting_down_.Acquire_Load()) { - // No more background work when shutting down. - } else if (!bg_error_.ok()) { - // No more background work after a background error. - } else { - BackgroundCompaction(); + if (!shutting_down_.Acquire_Load()) { + Status s = BackgroundCompaction(); + if (s.ok()) { + // Success + consecutive_compaction_errors_ = 0; + } else if (shutting_down_.Acquire_Load()) { + // Error most likely due to shutdown; do not wait + } else { + // Wait a little bit before retrying background compaction in + // case this is an environmental problem and we do not want to + // chew up resources for failed compactions for the duration of + // the problem. + bg_cv_.SignalAll(); // In case a waiter can proceed despite the error + Log(options_.info_log, "Waiting after background compaction error: %s", + s.ToString().c_str()); + mutex_.Unlock(); + ++consecutive_compaction_errors_; + int seconds_to_sleep = 1; + for (int i = 0; i < 3 && i < consecutive_compaction_errors_ - 1; ++i) { + seconds_to_sleep *= 2; + } + env_->SleepForMicroseconds(seconds_to_sleep * 1000000); + mutex_.Lock(); + } } bg_compaction_scheduled_ = false; @@ -652,12 +652,11 @@ void DBImpl::BackgroundCall() { bg_cv_.SignalAll(); } -void DBImpl::BackgroundCompaction() { +Status DBImpl::BackgroundCompaction() { mutex_.AssertHeld(); if (imm_ != NULL) { - CompactMemTable(); - return; + return CompactMemTable(); } Compaction* c; @@ -691,9 +690,6 @@ void DBImpl::BackgroundCompaction() { c->edit()->AddFile(c->level() + 1, f->number, f->file_size, f->smallest, f->largest); status = versions_->LogAndApply(c->edit(), &mutex_); - if (!status.ok()) { - RecordBackgroundError(status); - } VersionSet::LevelSummaryStorage tmp; Log(options_.info_log, "Moved #%lld to level-%d %lld bytes %s: %s\n", static_cast(f->number), @@ -704,9 +700,6 @@ void DBImpl::BackgroundCompaction() { } else { CompactionState* compact = new CompactionState(c); status = DoCompactionWork(compact); - if (!status.ok()) { - RecordBackgroundError(status); - } CleanupCompaction(compact); c->ReleaseInputs(); DeleteObsoleteFiles(); @@ -720,6 +713,9 @@ void DBImpl::BackgroundCompaction() { } else { Log(options_.info_log, "Compaction error: %s", status.ToString().c_str()); + if (options_.paranoid_checks && bg_error_.ok()) { + bg_error_ = status; + } } if (is_manual) { @@ -735,6 +731,7 @@ void DBImpl::BackgroundCompaction() { } manual_compaction_ = NULL; } + return status; } void DBImpl::CleanupCompaction(CompactionState* compact) { @@ -825,6 +822,9 @@ Status DBImpl::FinishCompactionOutputFile(CompactionState* compact, (unsigned long long) output_number, (unsigned long long) current_entries, (unsigned long long) current_bytes); + + // rate-limit compaction file creation with a 100ms pause + env_->SleepForMicroseconds(100000); } } return s; @@ -1004,9 +1004,6 @@ Status DBImpl::DoCompactionWork(CompactionState* compact) { if (status.ok()) { status = InstallCompactionResults(compact); } - if (!status.ok()) { - RecordBackgroundError(status); - } VersionSet::LevelSummaryStorage tmp; Log(options_.info_log, "compacted to: %s", versions_->LevelSummary(&tmp)); @@ -1033,8 +1030,7 @@ static void CleanupIteratorState(void* arg1, void* arg2) { } // namespace Iterator* DBImpl::NewInternalIterator(const ReadOptions& options, - SequenceNumber* latest_snapshot, - uint32_t* seed) { + SequenceNumber* latest_snapshot) { IterState* cleanup = new IterState; mutex_.Lock(); *latest_snapshot = versions_->LastSequence(); @@ -1058,15 +1054,13 @@ Iterator* DBImpl::NewInternalIterator(const ReadOptions& options, cleanup->version = versions_->current(); internal_iter->RegisterCleanup(CleanupIteratorState, cleanup, NULL); - *seed = ++seed_; mutex_.Unlock(); return internal_iter; } Iterator* DBImpl::TEST_NewInternalIterator() { SequenceNumber ignored; - uint32_t ignored_seed; - return NewInternalIterator(ReadOptions(), &ignored, &ignored_seed); + return NewInternalIterator(ReadOptions(), &ignored); } int64_t DBImpl::TEST_MaxNextLevelOverlappingBytes() { @@ -1123,21 +1117,12 @@ Status DBImpl::Get(const ReadOptions& options, Iterator* DBImpl::NewIterator(const ReadOptions& options) { SequenceNumber latest_snapshot; - uint32_t seed; - Iterator* iter = NewInternalIterator(options, &latest_snapshot, &seed); + Iterator* internal_iter = NewInternalIterator(options, &latest_snapshot); return NewDBIterator( - this, user_comparator(), iter, + &dbname_, env_, user_comparator(), internal_iter, (options.snapshot != NULL ? reinterpret_cast(options.snapshot)->number_ - : latest_snapshot), - seed); -} - -void DBImpl::RecordReadSample(Slice key) { - MutexLock l(&mutex_); - if (versions_->current()->RecordReadSample(key)) { - MaybeScheduleCompaction(); - } + : latest_snapshot)); } const Snapshot* DBImpl::GetSnapshot() { @@ -1190,23 +1175,13 @@ Status DBImpl::Write(const WriteOptions& options, WriteBatch* my_batch) { { mutex_.Unlock(); status = log_->AddRecord(WriteBatchInternal::Contents(updates)); - bool sync_error = false; if (status.ok() && options.sync) { status = logfile_->Sync(); - if (!status.ok()) { - sync_error = true; - } } if (status.ok()) { status = WriteBatchInternal::InsertInto(updates, mem_); } mutex_.Lock(); - if (sync_error) { - // The state of the log file is indeterminate: the log record we - // just added may or may not show up when the DB is re-opened. - // So we force the DB into a mode where all future writes fail. - RecordBackgroundError(status); - } } if (updates == tmp_batch_) tmp_batch_->Clear(); diff --git a/src/leveldb/db/db_impl.h b/src/leveldb/db/db_impl.h index cfc99816..3c8d711a 100644 --- a/src/leveldb/db/db_impl.h +++ b/src/leveldb/db/db_impl.h @@ -59,19 +59,13 @@ class DBImpl : public DB { // file at a level >= 1. int64_t TEST_MaxNextLevelOverlappingBytes(); - // Record a sample of bytes read at the specified internal key. - // Samples are taken approximately once every config::kReadBytesPeriod - // bytes. - void RecordReadSample(Slice key); - private: friend class DB; struct CompactionState; struct Writer; Iterator* NewInternalIterator(const ReadOptions&, - SequenceNumber* latest_snapshot, - uint32_t* seed); + SequenceNumber* latest_snapshot); Status NewDB(); @@ -87,8 +81,8 @@ class DBImpl : public DB { // Compact the in-memory write buffer to disk. Switches to a new // log-file/memtable and writes a new descriptor iff successful. - // Errors are recorded in bg_error_. - void CompactMemTable() EXCLUSIVE_LOCKS_REQUIRED(mutex_); + Status CompactMemTable() + EXCLUSIVE_LOCKS_REQUIRED(mutex_); Status RecoverLogFile(uint64_t log_number, VersionEdit* edit, @@ -102,12 +96,10 @@ class DBImpl : public DB { EXCLUSIVE_LOCKS_REQUIRED(mutex_); WriteBatch* BuildBatchGroup(Writer** last_writer); - void RecordBackgroundError(const Status& s); - void MaybeScheduleCompaction() EXCLUSIVE_LOCKS_REQUIRED(mutex_); static void BGWork(void* db); void BackgroundCall(); - void BackgroundCompaction() EXCLUSIVE_LOCKS_REQUIRED(mutex_); + Status BackgroundCompaction() EXCLUSIVE_LOCKS_REQUIRED(mutex_); void CleanupCompaction(CompactionState* compact) EXCLUSIVE_LOCKS_REQUIRED(mutex_); Status DoCompactionWork(CompactionState* compact) @@ -143,7 +135,6 @@ class DBImpl : public DB { WritableFile* logfile_; uint64_t logfile_number_; log::Writer* log_; - uint32_t seed_; // For sampling. // Queue of writers. std::deque writers_; @@ -172,6 +163,7 @@ class DBImpl : public DB { // Have we encountered a background error in paranoid mode? Status bg_error_; + int consecutive_compaction_errors_; // Per level compaction stats. stats_[level] stores the stats for // compactions that produced data for the specified "level". diff --git a/src/leveldb/db/db_iter.cc b/src/leveldb/db/db_iter.cc index 3b2035e9..87dca2de 100644 --- a/src/leveldb/db/db_iter.cc +++ b/src/leveldb/db/db_iter.cc @@ -5,14 +5,12 @@ #include "db/db_iter.h" #include "db/filename.h" -#include "db/db_impl.h" #include "db/dbformat.h" #include "leveldb/env.h" #include "leveldb/iterator.h" #include "port/port.h" #include "util/logging.h" #include "util/mutexlock.h" -#include "util/random.h" namespace leveldb { @@ -48,16 +46,15 @@ class DBIter: public Iterator { kReverse }; - DBIter(DBImpl* db, const Comparator* cmp, Iterator* iter, SequenceNumber s, - uint32_t seed) - : db_(db), + DBIter(const std::string* dbname, Env* env, + const Comparator* cmp, Iterator* iter, SequenceNumber s) + : dbname_(dbname), + env_(env), user_comparator_(cmp), iter_(iter), sequence_(s), direction_(kForward), - valid_(false), - rnd_(seed), - bytes_counter_(RandomPeriod()) { + valid_(false) { } virtual ~DBIter() { delete iter_; @@ -103,12 +100,8 @@ class DBIter: public Iterator { } } - // Pick next gap with average value of config::kReadBytesPeriod. - ssize_t RandomPeriod() { - return rnd_.Uniform(2*config::kReadBytesPeriod); - } - - DBImpl* db_; + const std::string* const dbname_; + Env* const env_; const Comparator* const user_comparator_; Iterator* const iter_; SequenceNumber const sequence_; @@ -119,23 +112,13 @@ class DBIter: public Iterator { Direction direction_; bool valid_; - Random rnd_; - ssize_t bytes_counter_; - // No copying allowed DBIter(const DBIter&); void operator=(const DBIter&); }; inline bool DBIter::ParseKey(ParsedInternalKey* ikey) { - Slice k = iter_->key(); - ssize_t n = k.size() + iter_->value().size(); - bytes_counter_ -= n; - while (bytes_counter_ < 0) { - bytes_counter_ += RandomPeriod(); - db_->RecordReadSample(k); - } - if (!ParseInternalKey(k, ikey)) { + if (!ParseInternalKey(iter_->key(), ikey)) { status_ = Status::Corruption("corrupted internal key in DBIter"); return false; } else { @@ -161,13 +144,12 @@ void DBIter::Next() { saved_key_.clear(); return; } - // saved_key_ already contains the key to skip past. - } else { - // Store in saved_key_ the current key so we skip it below. - SaveKey(ExtractUserKey(iter_->key()), &saved_key_); } - FindNextUserEntry(true, &saved_key_); + // Temporarily use saved_key_ as storage for key to skip. + std::string* skip = &saved_key_; + SaveKey(ExtractUserKey(iter_->key()), skip); + FindNextUserEntry(true, skip); } void DBIter::FindNextUserEntry(bool skipping, std::string* skip) { @@ -306,12 +288,12 @@ void DBIter::SeekToLast() { } // anonymous namespace Iterator* NewDBIterator( - DBImpl* db, + const std::string* dbname, + Env* env, const Comparator* user_key_comparator, Iterator* internal_iter, - SequenceNumber sequence, - uint32_t seed) { - return new DBIter(db, user_key_comparator, internal_iter, sequence, seed); + const SequenceNumber& sequence) { + return new DBIter(dbname, env, user_key_comparator, internal_iter, sequence); } } // namespace leveldb diff --git a/src/leveldb/db/db_iter.h b/src/leveldb/db/db_iter.h index 04927e93..d9e1b174 100644 --- a/src/leveldb/db/db_iter.h +++ b/src/leveldb/db/db_iter.h @@ -11,17 +11,15 @@ namespace leveldb { -class DBImpl; - // Return a new iterator that converts internal keys (yielded by // "*internal_iter") that were live at the specified "sequence" number // into appropriate user keys. extern Iterator* NewDBIterator( - DBImpl* db, + const std::string* dbname, + Env* env, const Comparator* user_key_comparator, Iterator* internal_iter, - SequenceNumber sequence, - uint32_t seed); + const SequenceNumber& sequence); } // namespace leveldb diff --git a/src/leveldb/db/db_test.cc b/src/leveldb/db/db_test.cc index 280b01c1..49aae04d 100644 --- a/src/leveldb/db/db_test.cc +++ b/src/leveldb/db/db_test.cc @@ -57,11 +57,8 @@ void DelayMilliseconds(int millis) { // Special Env used to delay background operations class SpecialEnv : public EnvWrapper { public: - // sstable/log Sync() calls are blocked while this pointer is non-NULL. - port::AtomicPointer delay_data_sync_; - - // sstable/log Sync() calls return an error. - port::AtomicPointer data_sync_error_; + // sstable Sync() calls are blocked while this pointer is non-NULL. + port::AtomicPointer delay_sstable_sync_; // Simulate no-space errors while this pointer is non-NULL. port::AtomicPointer no_space_; @@ -78,9 +75,11 @@ class SpecialEnv : public EnvWrapper { bool count_random_reads_; AtomicCounter random_read_counter_; + AtomicCounter sleep_counter_; + AtomicCounter sleep_time_counter_; + explicit SpecialEnv(Env* base) : EnvWrapper(base) { - delay_data_sync_.Release_Store(NULL); - data_sync_error_.Release_Store(NULL); + delay_sstable_sync_.Release_Store(NULL); no_space_.Release_Store(NULL); non_writable_.Release_Store(NULL); count_random_reads_ = false; @@ -89,17 +88,17 @@ class SpecialEnv : public EnvWrapper { } Status NewWritableFile(const std::string& f, WritableFile** r) { - class DataFile : public WritableFile { + class SSTableFile : public WritableFile { private: SpecialEnv* env_; WritableFile* base_; public: - DataFile(SpecialEnv* env, WritableFile* base) + SSTableFile(SpecialEnv* env, WritableFile* base) : env_(env), base_(base) { } - ~DataFile() { delete base_; } + ~SSTableFile() { delete base_; } Status Append(const Slice& data) { if (env_->no_space_.Acquire_Load() != NULL) { // Drop writes on the floor @@ -111,10 +110,7 @@ class SpecialEnv : public EnvWrapper { Status Close() { return base_->Close(); } Status Flush() { return base_->Flush(); } Status Sync() { - if (env_->data_sync_error_.Acquire_Load() != NULL) { - return Status::IOError("simulated data sync error"); - } - while (env_->delay_data_sync_.Acquire_Load() != NULL) { + while (env_->delay_sstable_sync_.Acquire_Load() != NULL) { DelayMilliseconds(100); } return base_->Sync(); @@ -151,9 +147,8 @@ class SpecialEnv : public EnvWrapper { Status s = target()->NewWritableFile(f, r); if (s.ok()) { - if (strstr(f.c_str(), ".ldb") != NULL || - strstr(f.c_str(), ".log") != NULL) { - *r = new DataFile(this, *r); + if (strstr(f.c_str(), ".sst") != NULL) { + *r = new SSTableFile(this, *r); } else if (strstr(f.c_str(), "MANIFEST") != NULL) { *r = new ManifestFile(this, *r); } @@ -184,6 +179,12 @@ class SpecialEnv : public EnvWrapper { } return s; } + + virtual void SleepForMicroseconds(int micros) { + sleep_counter_.Increment(); + sleep_time_counter_.IncrementBy(micros); + } + }; class DBTest { @@ -321,7 +322,7 @@ class DBTest { } // Check reverse iteration results are the reverse of forward results - size_t matched = 0; + int matched = 0; for (iter->SeekToLast(); iter->Valid(); iter->Prev()) { ASSERT_LT(matched, forward.size()); ASSERT_EQ(IterStatus(iter), forward[forward.size() - matched - 1]); @@ -483,24 +484,6 @@ class DBTest { } return false; } - - // Returns number of files renamed. - int RenameLDBToSST() { - std::vector filenames; - ASSERT_OK(env_->GetChildren(dbname_, &filenames)); - uint64_t number; - FileType type; - int files_renamed = 0; - for (size_t i = 0; i < filenames.size(); i++) { - if (ParseFileName(filenames[i], &number, &type) && type == kTableFile) { - const std::string from = TableFileName(dbname_, number); - const std::string to = SSTTableFileName(dbname_, number); - ASSERT_OK(env_->RenameFile(from, to)); - files_renamed++; - } - } - return files_renamed; - } }; TEST(DBTest, Empty) { @@ -542,11 +525,11 @@ TEST(DBTest, GetFromImmutableLayer) { ASSERT_OK(Put("foo", "v1")); ASSERT_EQ("v1", Get("foo")); - env_->delay_data_sync_.Release_Store(env_); // Block sync calls + env_->delay_sstable_sync_.Release_Store(env_); // Block sync calls Put("k1", std::string(100000, 'x')); // Fill memtable Put("k2", std::string(100000, 'y')); // Trigger compaction ASSERT_EQ("v1", Get("foo")); - env_->delay_data_sync_.Release_Store(NULL); // Release sync calls + env_->delay_sstable_sync_.Release_Store(NULL); // Release sync calls } while (ChangeOptions()); } @@ -1533,13 +1516,41 @@ TEST(DBTest, NoSpace) { Compact("a", "z"); const int num_files = CountFiles(); env_->no_space_.Release_Store(env_); // Force out-of-space errors - for (int i = 0; i < 10; i++) { + env_->sleep_counter_.Reset(); + for (int i = 0; i < 5; i++) { for (int level = 0; level < config::kNumLevels-1; level++) { dbfull()->TEST_CompactRange(level, NULL, NULL); } } env_->no_space_.Release_Store(NULL); ASSERT_LT(CountFiles(), num_files + 3); + + // Check that compaction attempts slept after errors + ASSERT_GE(env_->sleep_counter_.Read(), 5); +} + +TEST(DBTest, ExponentialBackoff) { + Options options = CurrentOptions(); + options.env = env_; + Reopen(&options); + + ASSERT_OK(Put("foo", "v1")); + ASSERT_EQ("v1", Get("foo")); + Compact("a", "z"); + env_->non_writable_.Release_Store(env_); // Force errors for new files + env_->sleep_counter_.Reset(); + env_->sleep_time_counter_.Reset(); + for (int i = 0; i < 5; i++) { + dbfull()->TEST_CompactRange(2, NULL, NULL); + } + env_->non_writable_.Release_Store(NULL); + + // Wait for compaction to finish + DelayMilliseconds(1000); + + ASSERT_GE(env_->sleep_counter_.Read(), 5); + ASSERT_LT(env_->sleep_counter_.Read(), 10); + ASSERT_GE(env_->sleep_time_counter_.Read(), 10e6); } TEST(DBTest, NonWritableFileSystem) { @@ -1562,37 +1573,6 @@ TEST(DBTest, NonWritableFileSystem) { env_->non_writable_.Release_Store(NULL); } -TEST(DBTest, WriteSyncError) { - // Check that log sync errors cause the DB to disallow future writes. - - // (a) Cause log sync calls to fail - Options options = CurrentOptions(); - options.env = env_; - Reopen(&options); - env_->data_sync_error_.Release_Store(env_); - - // (b) Normal write should succeed - WriteOptions w; - ASSERT_OK(db_->Put(w, "k1", "v1")); - ASSERT_EQ("v1", Get("k1")); - - // (c) Do a sync write; should fail - w.sync = true; - ASSERT_TRUE(!db_->Put(w, "k2", "v2").ok()); - ASSERT_EQ("v1", Get("k1")); - ASSERT_EQ("NOT_FOUND", Get("k2")); - - // (d) make sync behave normally - env_->data_sync_error_.Release_Store(NULL); - - // (e) Do a non-sync write; should fail - w.sync = false; - ASSERT_TRUE(!db_->Put(w, "k3", "v3").ok()); - ASSERT_EQ("v1", Get("k1")); - ASSERT_EQ("NOT_FOUND", Get("k2")); - ASSERT_EQ("NOT_FOUND", Get("k3")); -} - TEST(DBTest, ManifestWriteError) { // Test for the following problem: // (a) Compaction produces file F @@ -1652,22 +1632,6 @@ TEST(DBTest, MissingSSTFile) { << s.ToString(); } -TEST(DBTest, StillReadSST) { - ASSERT_OK(Put("foo", "bar")); - ASSERT_EQ("bar", Get("foo")); - - // Dump the memtable to disk. - dbfull()->TEST_CompactMemTable(); - ASSERT_EQ("bar", Get("foo")); - Close(); - ASSERT_GT(RenameLDBToSST(), 0); - Options options = CurrentOptions(); - options.paranoid_checks = true; - Status s = TryReopen(&options); - ASSERT_TRUE(s.ok()); - ASSERT_EQ("bar", Get("foo")); -} - TEST(DBTest, FilesDeletedAfterCompaction) { ASSERT_OK(Put("foo", "v2")); Compact("a", "z"); @@ -1699,7 +1663,7 @@ TEST(DBTest, BloomFilter) { dbfull()->TEST_CompactMemTable(); // Prevent auto compactions triggered by seeks - env_->delay_data_sync_.Release_Store(env_); + env_->delay_sstable_sync_.Release_Store(env_); // Lookup present keys. Should rarely read from small sstable. env_->random_read_counter_.Reset(); @@ -1720,7 +1684,7 @@ TEST(DBTest, BloomFilter) { fprintf(stderr, "%d missing => %d reads\n", N, reads); ASSERT_LE(reads, 3*N/100); - env_->delay_data_sync_.Release_Store(NULL); + env_->delay_sstable_sync_.Release_Store(NULL); Close(); delete options.block_cache; delete options.filter_policy; @@ -1780,7 +1744,7 @@ static void MTThreadBody(void* arg) { ASSERT_EQ(k, key); ASSERT_GE(w, 0); ASSERT_LT(w, kNumThreads); - ASSERT_LE(static_cast(c), reinterpret_cast( + ASSERT_LE(c, reinterpret_cast( t->state->counter[w].Acquire_Load())); } } diff --git a/src/leveldb/db/dbformat.h b/src/leveldb/db/dbformat.h index 5d8a032b..f7f64daf 100644 --- a/src/leveldb/db/dbformat.h +++ b/src/leveldb/db/dbformat.h @@ -38,9 +38,6 @@ static const int kL0_StopWritesTrigger = 12; // space if the same key space is being repeatedly overwritten. static const int kMaxMemCompactLevel = 2; -// Approximate gap in bytes between samples of data read during iteration. -static const int kReadBytesPeriod = 1048576; - } // namespace config class InternalKey; diff --git a/src/leveldb/db/filename.cc b/src/leveldb/db/filename.cc index da32946d..3c4d49f6 100644 --- a/src/leveldb/db/filename.cc +++ b/src/leveldb/db/filename.cc @@ -30,11 +30,6 @@ std::string LogFileName(const std::string& name, uint64_t number) { } std::string TableFileName(const std::string& name, uint64_t number) { - assert(number > 0); - return MakeFileName(name, number, "ldb"); -} - -std::string SSTTableFileName(const std::string& name, uint64_t number) { assert(number > 0); return MakeFileName(name, number, "sst"); } @@ -76,7 +71,7 @@ std::string OldInfoLogFileName(const std::string& dbname) { // dbname/LOG // dbname/LOG.old // dbname/MANIFEST-[0-9]+ -// dbname/[0-9]+.(log|sst|ldb) +// dbname/[0-9]+.(log|sst) bool ParseFileName(const std::string& fname, uint64_t* number, FileType* type) { @@ -111,7 +106,7 @@ bool ParseFileName(const std::string& fname, Slice suffix = rest; if (suffix == Slice(".log")) { *type = kLogFile; - } else if (suffix == Slice(".sst") || suffix == Slice(".ldb")) { + } else if (suffix == Slice(".sst")) { *type = kTableFile; } else if (suffix == Slice(".dbtmp")) { *type = kTempFile; diff --git a/src/leveldb/db/filename.h b/src/leveldb/db/filename.h index 87a75260..d5d09b11 100644 --- a/src/leveldb/db/filename.h +++ b/src/leveldb/db/filename.h @@ -37,11 +37,6 @@ extern std::string LogFileName(const std::string& dbname, uint64_t number); // "dbname". extern std::string TableFileName(const std::string& dbname, uint64_t number); -// Return the legacy file name for an sstable with the specified number -// in the db named by "dbname". The result will be prefixed with -// "dbname". -extern std::string SSTTableFileName(const std::string& dbname, uint64_t number); - // Return the name of the descriptor file for the db named by // "dbname" and the specified incarnation number. The result will be // prefixed with "dbname". diff --git a/src/leveldb/db/filename_test.cc b/src/leveldb/db/filename_test.cc index a32556de..5a26da47 100644 --- a/src/leveldb/db/filename_test.cc +++ b/src/leveldb/db/filename_test.cc @@ -27,7 +27,6 @@ TEST(FileNameTest, Parse) { { "100.log", 100, kLogFile }, { "0.log", 0, kLogFile }, { "0.sst", 0, kTableFile }, - { "0.ldb", 0, kTableFile }, { "CURRENT", 0, kCurrentFile }, { "LOCK", 0, kDBLockFile }, { "MANIFEST-2", 2, kDescriptorFile }, diff --git a/src/leveldb/db/log_reader.cc b/src/leveldb/db/log_reader.cc index 4919216d..b35f115a 100644 --- a/src/leveldb/db/log_reader.cc +++ b/src/leveldb/db/log_reader.cc @@ -133,9 +133,7 @@ bool Reader::ReadRecord(Slice* record, std::string* scratch) { case kEof: if (in_fragmented_record) { - // This can be caused by the writer dying immediately after - // writing a physical record but before completing the next; don't - // treat it as a corruption, just ignore the entire logical record. + ReportCorruption(scratch->size(), "partial record without end(3)"); scratch->clear(); } return false; @@ -195,12 +193,13 @@ unsigned int Reader::ReadPhysicalRecord(Slice* result) { eof_ = true; } continue; + } else if (buffer_.size() == 0) { + // End of file + return kEof; } else { - // Note that if buffer_ is non-empty, we have a truncated header at the - // end of the file, which can be caused by the writer crashing in the - // middle of writing the header. Instead of considering this an error, - // just report EOF. + size_t drop_size = buffer_.size(); buffer_.clear(); + ReportCorruption(drop_size, "truncated record at end of file"); return kEof; } } @@ -214,14 +213,8 @@ unsigned int Reader::ReadPhysicalRecord(Slice* result) { if (kHeaderSize + length > buffer_.size()) { size_t drop_size = buffer_.size(); buffer_.clear(); - if (!eof_) { - ReportCorruption(drop_size, "bad record length"); - return kBadRecord; - } - // If the end of the file has been reached without reading |length| bytes - // of payload, assume the writer died in the middle of writing the record. - // Don't report a corruption. - return kEof; + ReportCorruption(drop_size, "bad record length"); + return kBadRecord; } if (type == kZeroType && length == 0) { diff --git a/src/leveldb/db/log_test.cc b/src/leveldb/db/log_test.cc index 91d3caaf..4c5cf875 100644 --- a/src/leveldb/db/log_test.cc +++ b/src/leveldb/db/log_test.cc @@ -351,32 +351,20 @@ TEST(LogTest, BadRecordType) { ASSERT_EQ("OK", MatchError("unknown record type")); } -TEST(LogTest, TruncatedTrailingRecordIsIgnored) { +TEST(LogTest, TruncatedTrailingRecord) { Write("foo"); ShrinkSize(4); // Drop all payload as well as a header byte ASSERT_EQ("EOF", Read()); - // Truncated last record is ignored, not treated as an error. - ASSERT_EQ(0, DroppedBytes()); - ASSERT_EQ("", ReportMessage()); + ASSERT_EQ(kHeaderSize - 1, DroppedBytes()); + ASSERT_EQ("OK", MatchError("truncated record at end of file")); } TEST(LogTest, BadLength) { - const int kPayloadSize = kBlockSize - kHeaderSize; - Write(BigString("bar", kPayloadSize)); - Write("foo"); - // Least significant size byte is stored in header[4]. - IncrementByte(4, 1); - ASSERT_EQ("foo", Read()); - ASSERT_EQ(kBlockSize, DroppedBytes()); - ASSERT_EQ("OK", MatchError("bad record length")); -} - -TEST(LogTest, BadLengthAtEndIsIgnored) { Write("foo"); ShrinkSize(1); ASSERT_EQ("EOF", Read()); - ASSERT_EQ(0, DroppedBytes()); - ASSERT_EQ("", ReportMessage()); + ASSERT_EQ(kHeaderSize + 2, DroppedBytes()); + ASSERT_EQ("OK", MatchError("bad record length")); } TEST(LogTest, ChecksumMismatch) { @@ -427,24 +415,6 @@ TEST(LogTest, UnexpectedFirstType) { ASSERT_EQ("OK", MatchError("partial record without end")); } -TEST(LogTest, MissingLastIsIgnored) { - Write(BigString("bar", kBlockSize)); - // Remove the LAST block, including header. - ShrinkSize(14); - ASSERT_EQ("EOF", Read()); - ASSERT_EQ("", ReportMessage()); - ASSERT_EQ(0, DroppedBytes()); -} - -TEST(LogTest, PartialLastIsIgnored) { - Write(BigString("bar", kBlockSize)); - // Cause a bad record length in the LAST block. - ShrinkSize(1); - ASSERT_EQ("EOF", Read()); - ASSERT_EQ("", ReportMessage()); - ASSERT_EQ(0, DroppedBytes()); -} - TEST(LogTest, ErrorJoinsRecords) { // Consider two fragmented records: // first(R1) last(R1) first(R2) last(R2) diff --git a/src/leveldb/db/repair.cc b/src/leveldb/db/repair.cc index 7727fafc..022d52f3 100644 --- a/src/leveldb/db/repair.cc +++ b/src/leveldb/db/repair.cc @@ -242,134 +242,62 @@ class Repairer { } void ExtractMetaData() { + std::vector kept; for (size_t i = 0; i < table_numbers_.size(); i++) { - ScanTable(table_numbers_[i]); - } - } - - Iterator* NewTableIterator(const FileMetaData& meta) { - // Same as compaction iterators: if paranoid_checks are on, turn - // on checksum verification. - ReadOptions r; - r.verify_checksums = options_.paranoid_checks; - return table_cache_->NewIterator(r, meta.number, meta.file_size); - } - - void ScanTable(uint64_t number) { - TableInfo t; - t.meta.number = number; - std::string fname = TableFileName(dbname_, number); - Status status = env_->GetFileSize(fname, &t.meta.file_size); - if (!status.ok()) { - // Try alternate file name. - fname = SSTTableFileName(dbname_, number); - Status s2 = env_->GetFileSize(fname, &t.meta.file_size); - if (s2.ok()) { - status = Status::OK(); - } - } - if (!status.ok()) { - ArchiveFile(TableFileName(dbname_, number)); - ArchiveFile(SSTTableFileName(dbname_, number)); - Log(options_.info_log, "Table #%llu: dropped: %s", - (unsigned long long) t.meta.number, - status.ToString().c_str()); - return; - } - - // Extract metadata by scanning through table. - int counter = 0; - Iterator* iter = NewTableIterator(t.meta); - bool empty = true; - ParsedInternalKey parsed; - t.max_sequence = 0; - for (iter->SeekToFirst(); iter->Valid(); iter->Next()) { - Slice key = iter->key(); - if (!ParseInternalKey(key, &parsed)) { - Log(options_.info_log, "Table #%llu: unparsable key %s", - (unsigned long long) t.meta.number, - EscapeString(key).c_str()); - continue; - } - - counter++; - if (empty) { - empty = false; - t.meta.smallest.DecodeFrom(key); - } - t.meta.largest.DecodeFrom(key); - if (parsed.sequence > t.max_sequence) { - t.max_sequence = parsed.sequence; - } - } - if (!iter->status().ok()) { - status = iter->status(); - } - delete iter; - Log(options_.info_log, "Table #%llu: %d entries %s", - (unsigned long long) t.meta.number, - counter, - status.ToString().c_str()); - - if (status.ok()) { - tables_.push_back(t); - } else { - RepairTable(fname, t); // RepairTable archives input file. - } - } - - void RepairTable(const std::string& src, TableInfo t) { - // We will copy src contents to a new table and then rename the - // new table over the source. - - // Create builder. - std::string copy = TableFileName(dbname_, next_file_number_++); - WritableFile* file; - Status s = env_->NewWritableFile(copy, &file); - if (!s.ok()) { - return; - } - TableBuilder* builder = new TableBuilder(options_, file); - - // Copy data. - Iterator* iter = NewTableIterator(t.meta); - int counter = 0; - for (iter->SeekToFirst(); iter->Valid(); iter->Next()) { - builder->Add(iter->key(), iter->value()); - counter++; - } - delete iter; - - ArchiveFile(src); - if (counter == 0) { - builder->Abandon(); // Nothing to save - } else { - s = builder->Finish(); - if (s.ok()) { - t.meta.file_size = builder->FileSize(); - } - } - delete builder; - builder = NULL; - - if (s.ok()) { - s = file->Close(); - } - delete file; - file = NULL; - - if (counter > 0 && s.ok()) { - std::string orig = TableFileName(dbname_, t.meta.number); - s = env_->RenameFile(copy, orig); - if (s.ok()) { - Log(options_.info_log, "Table #%llu: %d entries repaired", - (unsigned long long) t.meta.number, counter); + TableInfo t; + t.meta.number = table_numbers_[i]; + Status status = ScanTable(&t); + if (!status.ok()) { + std::string fname = TableFileName(dbname_, table_numbers_[i]); + Log(options_.info_log, "Table #%llu: ignoring %s", + (unsigned long long) table_numbers_[i], + status.ToString().c_str()); + ArchiveFile(fname); + } else { tables_.push_back(t); } } - if (!s.ok()) { - env_->DeleteFile(copy); + } + + Status ScanTable(TableInfo* t) { + std::string fname = TableFileName(dbname_, t->meta.number); + int counter = 0; + Status status = env_->GetFileSize(fname, &t->meta.file_size); + if (status.ok()) { + Iterator* iter = table_cache_->NewIterator( + ReadOptions(), t->meta.number, t->meta.file_size); + bool empty = true; + ParsedInternalKey parsed; + t->max_sequence = 0; + for (iter->SeekToFirst(); iter->Valid(); iter->Next()) { + Slice key = iter->key(); + if (!ParseInternalKey(key, &parsed)) { + Log(options_.info_log, "Table #%llu: unparsable key %s", + (unsigned long long) t->meta.number, + EscapeString(key).c_str()); + continue; + } + + counter++; + if (empty) { + empty = false; + t->meta.smallest.DecodeFrom(key); + } + t->meta.largest.DecodeFrom(key); + if (parsed.sequence > t->max_sequence) { + t->max_sequence = parsed.sequence; + } + } + if (!iter->status().ok()) { + status = iter->status(); + } + delete iter; } + Log(options_.info_log, "Table #%llu: %d entries %s", + (unsigned long long) t->meta.number, + counter, + status.ToString().c_str()); + return status; } Status WriteDescriptor() { diff --git a/src/leveldb/db/table_cache.cc b/src/leveldb/db/table_cache.cc index e3d82cd3..497db270 100644 --- a/src/leveldb/db/table_cache.cc +++ b/src/leveldb/db/table_cache.cc @@ -54,12 +54,6 @@ Status TableCache::FindTable(uint64_t file_number, uint64_t file_size, RandomAccessFile* file = NULL; Table* table = NULL; s = env_->NewRandomAccessFile(fname, &file); - if (!s.ok()) { - std::string old_fname = SSTTableFileName(dbname_, file_number); - if (env_->NewRandomAccessFile(old_fname, &file).ok()) { - s = Status::OK(); - } - } if (s.ok()) { s = Table::Open(*options_, file, file_size, &table); } diff --git a/src/leveldb/db/version_set.cc b/src/leveldb/db/version_set.cc index aa83df55..4fd1ddef 100644 --- a/src/leveldb/db/version_set.cc +++ b/src/leveldb/db/version_set.cc @@ -54,6 +54,20 @@ static int64_t TotalFileSize(const std::vector& files) { return sum; } +namespace { +std::string IntSetToString(const std::set& s) { + std::string result = "{"; + for (std::set::const_iterator it = s.begin(); + it != s.end(); + ++it) { + result += (result.size() > 1) ? "," : ""; + result += NumberToString(*it); + } + result += "}"; + return result; +} +} // namespace + Version::~Version() { assert(refs_ == 0); @@ -275,51 +289,6 @@ static bool NewestFirst(FileMetaData* a, FileMetaData* b) { return a->number > b->number; } -void Version::ForEachOverlapping(Slice user_key, Slice internal_key, - void* arg, - bool (*func)(void*, int, FileMetaData*)) { - // TODO(sanjay): Change Version::Get() to use this function. - const Comparator* ucmp = vset_->icmp_.user_comparator(); - - // Search level-0 in order from newest to oldest. - std::vector tmp; - tmp.reserve(files_[0].size()); - for (uint32_t i = 0; i < files_[0].size(); i++) { - FileMetaData* f = files_[0][i]; - if (ucmp->Compare(user_key, f->smallest.user_key()) >= 0 && - ucmp->Compare(user_key, f->largest.user_key()) <= 0) { - tmp.push_back(f); - } - } - if (!tmp.empty()) { - std::sort(tmp.begin(), tmp.end(), NewestFirst); - for (uint32_t i = 0; i < tmp.size(); i++) { - if (!(*func)(arg, 0, tmp[i])) { - return; - } - } - } - - // Search other levels. - for (int level = 1; level < config::kNumLevels; level++) { - size_t num_files = files_[level].size(); - if (num_files == 0) continue; - - // Binary search to find earliest index whose largest key >= internal_key. - uint32_t index = FindFile(vset_->icmp_, files_[level], internal_key); - if (index < num_files) { - FileMetaData* f = files_[level][index]; - if (ucmp->Compare(user_key, f->smallest.user_key()) < 0) { - // All of "f" is past any data for user_key - } else { - if (!(*func)(arg, level, f)) { - return; - } - } - } - } -} - Status Version::Get(const ReadOptions& options, const LookupKey& k, std::string* value, @@ -432,44 +401,6 @@ bool Version::UpdateStats(const GetStats& stats) { return false; } -bool Version::RecordReadSample(Slice internal_key) { - ParsedInternalKey ikey; - if (!ParseInternalKey(internal_key, &ikey)) { - return false; - } - - struct State { - GetStats stats; // Holds first matching file - int matches; - - static bool Match(void* arg, int level, FileMetaData* f) { - State* state = reinterpret_cast(arg); - state->matches++; - if (state->matches == 1) { - // Remember first match. - state->stats.seek_file = f; - state->stats.seek_file_level = level; - } - // We can stop iterating once we have a second match. - return state->matches < 2; - } - }; - - State state; - state.matches = 0; - ForEachOverlapping(ikey.user_key, internal_key, &state, &State::Match); - - // Must have at least two matches since we want to merge across - // files. But what if we have a single file that contains many - // overwrites and deletions? Should we have another mechanism for - // finding such files? - if (state.matches >= 2) { - // 1MB cost is about 1 seek (see comment in Builder::Apply). - return UpdateStats(state.stats); - } - return false; -} - void Version::Ref() { ++refs_; } @@ -504,13 +435,10 @@ int Version::PickLevelForMemTableOutput( if (OverlapInLevel(level + 1, &smallest_user_key, &largest_user_key)) { break; } - if (level + 2 < config::kNumLevels) { - // Check that file does not overlap too many grandparent bytes. - GetOverlappingInputs(level + 2, &start, &limit, &overlaps); - const int64_t sum = TotalFileSize(overlaps); - if (sum > kMaxGrandParentOverlapBytes) { - break; - } + GetOverlappingInputs(level + 2, &start, &limit, &overlaps); + const int64_t sum = TotalFileSize(overlaps); + if (sum > kMaxGrandParentOverlapBytes) { + break; } level++; } @@ -524,8 +452,6 @@ void Version::GetOverlappingInputs( const InternalKey* begin, const InternalKey* end, std::vector* inputs) { - assert(level >= 0); - assert(level < config::kNumLevels); inputs->clear(); Slice user_begin, user_end; if (begin != NULL) { @@ -862,6 +788,12 @@ Status VersionSet::LogAndApply(VersionEdit* edit, port::Mutex* mu) { } if (!s.ok()) { Log(options_->info_log, "MANIFEST write: %s\n", s.ToString().c_str()); + if (ManifestContains(record)) { + Log(options_->info_log, + "MANIFEST contains log record despite error; advancing to new " + "version to prevent mismatch between in-memory and logged state"); + s = Status::OK(); + } } } @@ -869,6 +801,8 @@ Status VersionSet::LogAndApply(VersionEdit* edit, port::Mutex* mu) { // new CURRENT file that points to it. if (s.ok() && !new_manifest_file.empty()) { s = SetCurrentFile(env_, dbname_, manifest_file_number_); + // No need to double-check MANIFEST in case of error since it + // will be discarded below. } mu->Lock(); @@ -1102,6 +1036,31 @@ const char* VersionSet::LevelSummary(LevelSummaryStorage* scratch) const { return scratch->buffer; } +// Return true iff the manifest contains the specified record. +bool VersionSet::ManifestContains(const std::string& record) const { + std::string fname = DescriptorFileName(dbname_, manifest_file_number_); + Log(options_->info_log, "ManifestContains: checking %s\n", fname.c_str()); + SequentialFile* file = NULL; + Status s = env_->NewSequentialFile(fname, &file); + if (!s.ok()) { + Log(options_->info_log, "ManifestContains: %s\n", s.ToString().c_str()); + return false; + } + log::Reader reader(file, NULL, true/*checksum*/, 0); + Slice r; + std::string scratch; + bool result = false; + while (reader.ReadRecord(&r, &scratch)) { + if (r == Slice(record)) { + result = true; + break; + } + } + delete file; + Log(options_->info_log, "ManifestContains: result = %d\n", result ? 1 : 0); + return result; +} + uint64_t VersionSet::ApproximateOffsetOf(Version* v, const InternalKey& ikey) { uint64_t result = 0; for (int level = 0; level < config::kNumLevels; level++) { diff --git a/src/leveldb/db/version_set.h b/src/leveldb/db/version_set.h index 8dc14b8e..9d084fdb 100644 --- a/src/leveldb/db/version_set.h +++ b/src/leveldb/db/version_set.h @@ -78,12 +78,6 @@ class Version { // REQUIRES: lock is held bool UpdateStats(const GetStats& stats); - // Record a sample of bytes read at the specified internal key. - // Samples are taken approximately once every config::kReadBytesPeriod - // bytes. Returns true if a new compaction may need to be triggered. - // REQUIRES: lock is held - bool RecordReadSample(Slice key); - // Reference count management (so Versions do not disappear out from // under live iterators) void Ref(); @@ -120,15 +114,6 @@ class Version { class LevelFileNumIterator; Iterator* NewConcatenatingIterator(const ReadOptions&, int level) const; - // Call func(arg, level, f) for every file that overlaps user_key in - // order from newest to oldest. If an invocation of func returns - // false, makes no more calls. - // - // REQUIRES: user portion of internal_key == user_key. - void ForEachOverlapping(Slice user_key, Slice internal_key, - void* arg, - bool (*func)(void*, int, FileMetaData*)); - VersionSet* vset_; // VersionSet to which this Version belongs Version* next_; // Next version in linked list Version* prev_; // Previous version in linked list @@ -292,6 +277,8 @@ class VersionSet { void AppendVersion(Version* v); + bool ManifestContains(const std::string& record) const; + Env* const env_; const std::string dbname_; const Options* const options_; diff --git a/src/leveldb/doc/impl.html b/src/leveldb/doc/impl.html index 28817fe0..e870795d 100644 --- a/src/leveldb/doc/impl.html +++ b/src/leveldb/doc/impl.html @@ -11,7 +11,7 @@ The implementation of leveldb is similar in spirit to the representation of a single - + Bigtable tablet (section 5.3). However the organization of the files that make up the representation is somewhat different and is explained below. diff --git a/src/leveldb/include/leveldb/c.h b/src/leveldb/include/leveldb/c.h index 1048fe3b..1fa58866 100644 --- a/src/leveldb/include/leveldb/c.h +++ b/src/leveldb/include/leveldb/c.h @@ -9,6 +9,7 @@ Does not support: . getters for the option types . custom comparators that implement key shortening + . capturing post-write-snapshot . custom iter, db, env, cache implementations using just the C bindings Some conventions: diff --git a/src/leveldb/include/leveldb/db.h b/src/leveldb/include/leveldb/db.h index 40851b2a..da8b11a8 100644 --- a/src/leveldb/include/leveldb/db.h +++ b/src/leveldb/include/leveldb/db.h @@ -14,7 +14,7 @@ namespace leveldb { // Update Makefile if you change these static const int kMajorVersion = 1; -static const int kMinorVersion = 17; +static const int kMinorVersion = 12; struct Options; struct ReadOptions; diff --git a/src/leveldb/include/leveldb/env.h b/src/leveldb/include/leveldb/env.h index b2072d02..fa32289f 100644 --- a/src/leveldb/include/leveldb/env.h +++ b/src/leveldb/include/leveldb/env.h @@ -13,9 +13,9 @@ #ifndef STORAGE_LEVELDB_INCLUDE_ENV_H_ #define STORAGE_LEVELDB_INCLUDE_ENV_H_ +#include #include #include -#include #include #include "leveldb/status.h" diff --git a/src/leveldb/include/leveldb/slice.h b/src/leveldb/include/leveldb/slice.h index bc367986..74ea8fa4 100644 --- a/src/leveldb/include/leveldb/slice.h +++ b/src/leveldb/include/leveldb/slice.h @@ -94,7 +94,7 @@ inline bool operator!=(const Slice& x, const Slice& y) { } inline int Slice::compare(const Slice& b) const { - const size_t min_len = (size_ < b.size_) ? size_ : b.size_; + const int min_len = (size_ < b.size_) ? size_ : b.size_; int r = memcmp(data_, b.data_, min_len); if (r == 0) { if (size_ < b.size_) r = -1; diff --git a/src/leveldb/issues/issue178_test.cc b/src/leveldb/issues/issue178_test.cc deleted file mode 100644 index 1b1cf8bb..00000000 --- a/src/leveldb/issues/issue178_test.cc +++ /dev/null @@ -1,92 +0,0 @@ -// Copyright (c) 2013 The LevelDB Authors. All rights reserved. -// Use of this source code is governed by a BSD-style license that can be -// found in the LICENSE file. See the AUTHORS file for names of contributors. - -// Test for issue 178: a manual compaction causes deleted data to reappear. -#include -#include -#include - -#include "leveldb/db.h" -#include "leveldb/write_batch.h" -#include "util/testharness.h" - -namespace { - -const int kNumKeys = 1100000; - -std::string Key1(int i) { - char buf[100]; - snprintf(buf, sizeof(buf), "my_key_%d", i); - return buf; -} - -std::string Key2(int i) { - return Key1(i) + "_xxx"; -} - -class Issue178 { }; - -TEST(Issue178, Test) { - // Get rid of any state from an old run. - std::string dbpath = leveldb::test::TmpDir() + "/leveldb_cbug_test"; - DestroyDB(dbpath, leveldb::Options()); - - // Open database. Disable compression since it affects the creation - // of layers and the code below is trying to test against a very - // specific scenario. - leveldb::DB* db; - leveldb::Options db_options; - db_options.create_if_missing = true; - db_options.compression = leveldb::kNoCompression; - ASSERT_OK(leveldb::DB::Open(db_options, dbpath, &db)); - - // create first key range - leveldb::WriteBatch batch; - for (size_t i = 0; i < kNumKeys; i++) { - batch.Put(Key1(i), "value for range 1 key"); - } - ASSERT_OK(db->Write(leveldb::WriteOptions(), &batch)); - - // create second key range - batch.Clear(); - for (size_t i = 0; i < kNumKeys; i++) { - batch.Put(Key2(i), "value for range 2 key"); - } - ASSERT_OK(db->Write(leveldb::WriteOptions(), &batch)); - - // delete second key range - batch.Clear(); - for (size_t i = 0; i < kNumKeys; i++) { - batch.Delete(Key2(i)); - } - ASSERT_OK(db->Write(leveldb::WriteOptions(), &batch)); - - // compact database - std::string start_key = Key1(0); - std::string end_key = Key1(kNumKeys - 1); - leveldb::Slice least(start_key.data(), start_key.size()); - leveldb::Slice greatest(end_key.data(), end_key.size()); - - // commenting out the line below causes the example to work correctly - db->CompactRange(&least, &greatest); - - // count the keys - leveldb::Iterator* iter = db->NewIterator(leveldb::ReadOptions()); - size_t num_keys = 0; - for (iter->SeekToFirst(); iter->Valid(); iter->Next()) { - num_keys++; - } - delete iter; - ASSERT_EQ(kNumKeys, num_keys) << "Bad number of keys"; - - // close database - delete db; - DestroyDB(dbpath, leveldb::Options()); -} - -} // anonymous namespace - -int main(int argc, char** argv) { - return leveldb::test::RunAllTests(); -} diff --git a/src/leveldb/issues/issue200_test.cc b/src/leveldb/issues/issue200_test.cc deleted file mode 100644 index 1cec79f4..00000000 --- a/src/leveldb/issues/issue200_test.cc +++ /dev/null @@ -1,59 +0,0 @@ -// Copyright (c) 2013 The LevelDB Authors. All rights reserved. -// Use of this source code is governed by a BSD-style license that can be -// found in the LICENSE file. See the AUTHORS file for names of contributors. - -// Test for issue 200: when iterator switches direction from backward -// to forward, the current key can be yielded unexpectedly if a new -// mutation has been added just before the current key. - -#include "leveldb/db.h" -#include "util/testharness.h" - -namespace leveldb { - -class Issue200 { }; - -TEST(Issue200, Test) { - // Get rid of any state from an old run. - std::string dbpath = test::TmpDir() + "/leveldb_issue200_test"; - DestroyDB(dbpath, Options()); - - DB *db; - Options options; - options.create_if_missing = true; - ASSERT_OK(DB::Open(options, dbpath, &db)); - - WriteOptions write_options; - ASSERT_OK(db->Put(write_options, "1", "b")); - ASSERT_OK(db->Put(write_options, "2", "c")); - ASSERT_OK(db->Put(write_options, "3", "d")); - ASSERT_OK(db->Put(write_options, "4", "e")); - ASSERT_OK(db->Put(write_options, "5", "f")); - - ReadOptions read_options; - Iterator *iter = db->NewIterator(read_options); - - // Add an element that should not be reflected in the iterator. - ASSERT_OK(db->Put(write_options, "25", "cd")); - - iter->Seek("5"); - ASSERT_EQ(iter->key().ToString(), "5"); - iter->Prev(); - ASSERT_EQ(iter->key().ToString(), "4"); - iter->Prev(); - ASSERT_EQ(iter->key().ToString(), "3"); - iter->Next(); - ASSERT_EQ(iter->key().ToString(), "4"); - iter->Next(); - ASSERT_EQ(iter->key().ToString(), "5"); - - delete iter; - delete db; - DestroyDB(dbpath, options); -} - -} // namespace leveldb - -int main(int argc, char** argv) { - return leveldb::test::RunAllTests(); -} diff --git a/src/leveldb/port/atomic_pointer.h b/src/leveldb/port/atomic_pointer.h index a9866b23..e17bf435 100644 --- a/src/leveldb/port/atomic_pointer.h +++ b/src/leveldb/port/atomic_pointer.h @@ -50,13 +50,6 @@ namespace port { // http://msdn.microsoft.com/en-us/library/ms684208(v=vs.85).aspx #define LEVELDB_HAVE_MEMORY_BARRIER -// Mac OS -#elif defined(OS_MACOSX) -inline void MemoryBarrier() { - OSMemoryBarrier(); -} -#define LEVELDB_HAVE_MEMORY_BARRIER - // Gcc on x86 #elif defined(ARCH_CPU_X86_FAMILY) && defined(__GNUC__) inline void MemoryBarrier() { @@ -75,6 +68,13 @@ inline void MemoryBarrier() { } #define LEVELDB_HAVE_MEMORY_BARRIER +// Mac OS +#elif defined(OS_MACOSX) +inline void MemoryBarrier() { + OSMemoryBarrier(); +} +#define LEVELDB_HAVE_MEMORY_BARRIER + // ARM Linux #elif defined(ARCH_CPU_ARM_FAMILY) && defined(__linux__) typedef void (*LinuxKernelMemoryBarrierFunc)(void); diff --git a/src/leveldb/port/port_posix.h b/src/leveldb/port/port_posix.h index 21c845e2..f2b89bff 100644 --- a/src/leveldb/port/port_posix.h +++ b/src/leveldb/port/port_posix.h @@ -62,16 +62,12 @@ #define fflush_unlocked fflush #endif -#if defined(OS_FREEBSD) ||\ +#if defined(OS_MACOSX) || defined(OS_FREEBSD) ||\ defined(OS_OPENBSD) || defined(OS_DRAGONFLYBSD) // Use fsync() on platforms without fdatasync() #define fdatasync fsync #endif -#if defined(OS_MACOSX) -#define fdatasync(fd) fcntl(fd, F_FULLFSYNC, 0) -#endif - #if defined(OS_ANDROID) && __ANDROID_API__ < 9 // fdatasync() was only introduced in API level 9 on Android. Use fsync() // when targetting older platforms. diff --git a/src/leveldb/table/filter_block_test.cc b/src/leveldb/table/filter_block_test.cc index 8c4a4741..3a2a07cf 100644 --- a/src/leveldb/table/filter_block_test.cc +++ b/src/leveldb/table/filter_block_test.cc @@ -29,7 +29,7 @@ class TestHashFilter : public FilterPolicy { virtual bool KeyMayMatch(const Slice& key, const Slice& filter) const { uint32_t h = Hash(key.data(), key.size(), 1); - for (size_t i = 0; i + 4 <= filter.size(); i += 4) { + for (int i = 0; i + 4 <= filter.size(); i += 4) { if (h == DecodeFixed32(filter.data() + i)) { return true; } diff --git a/src/leveldb/util/arena.cc b/src/leveldb/util/arena.cc index 9367f714..9551d6a3 100644 --- a/src/leveldb/util/arena.cc +++ b/src/leveldb/util/arena.cc @@ -40,7 +40,7 @@ char* Arena::AllocateFallback(size_t bytes) { } char* Arena::AllocateAligned(size_t bytes) { - const int align = (sizeof(void*) > 8) ? sizeof(void*) : 8; + const int align = sizeof(void*); // We'll align to pointer size assert((align & (align-1)) == 0); // Pointer size should be a power of 2 size_t current_mod = reinterpret_cast(alloc_ptr_) & (align-1); size_t slop = (current_mod == 0 ? 0 : align - current_mod); diff --git a/src/leveldb/util/arena.h b/src/leveldb/util/arena.h index 73bbf1cb..8f7dde22 100644 --- a/src/leveldb/util/arena.h +++ b/src/leveldb/util/arena.h @@ -5,9 +5,9 @@ #ifndef STORAGE_LEVELDB_UTIL_ARENA_H_ #define STORAGE_LEVELDB_UTIL_ARENA_H_ +#include #include #include -#include #include namespace leveldb { diff --git a/src/leveldb/util/arena_test.cc b/src/leveldb/util/arena_test.cc index 58e870ec..63d17780 100644 --- a/src/leveldb/util/arena_test.cc +++ b/src/leveldb/util/arena_test.cc @@ -40,7 +40,7 @@ TEST(ArenaTest, Simple) { r = arena.Allocate(s); } - for (size_t b = 0; b < s; b++) { + for (int b = 0; b < s; b++) { // Fill the "i"th allocation with a known bit pattern r[b] = i % 256; } @@ -51,10 +51,10 @@ TEST(ArenaTest, Simple) { ASSERT_LE(arena.MemoryUsage(), bytes * 1.10); } } - for (size_t i = 0; i < allocated.size(); i++) { + for (int i = 0; i < allocated.size(); i++) { size_t num_bytes = allocated[i].first; const char* p = allocated[i].second; - for (size_t b = 0; b < num_bytes; b++) { + for (int b = 0; b < num_bytes; b++) { // Check the "i"th allocation for the known bit pattern ASSERT_EQ(int(p[b]) & 0xff, i % 256); } diff --git a/src/leveldb/util/bloom_test.cc b/src/leveldb/util/bloom_test.cc index 77fb1b31..0bf8e8d6 100644 --- a/src/leveldb/util/bloom_test.cc +++ b/src/leveldb/util/bloom_test.cc @@ -126,8 +126,7 @@ TEST(BloomTest, VaryingLengths) { } Build(); - ASSERT_LE(FilterSize(), static_cast((length * 10 / 8) + 40)) - << length; + ASSERT_LE(FilterSize(), (length * 10 / 8) + 40) << length; // All added keys must match for (int i = 0; i < length; i++) { diff --git a/src/leveldb/util/coding_test.cc b/src/leveldb/util/coding_test.cc index 521541ea..fb5726e3 100644 --- a/src/leveldb/util/coding_test.cc +++ b/src/leveldb/util/coding_test.cc @@ -112,13 +112,13 @@ TEST(Coding, Varint64) { } std::string s; - for (size_t i = 0; i < values.size(); i++) { + for (int i = 0; i < values.size(); i++) { PutVarint64(&s, values[i]); } const char* p = s.data(); const char* limit = p + s.size(); - for (size_t i = 0; i < values.size(); i++) { + for (int i = 0; i < values.size(); i++) { ASSERT_TRUE(p < limit); uint64_t actual; const char* start = p; @@ -143,7 +143,7 @@ TEST(Coding, Varint32Truncation) { std::string s; PutVarint32(&s, large_value); uint32_t result; - for (size_t len = 0; len < s.size() - 1; len++) { + for (int len = 0; len < s.size() - 1; len++) { ASSERT_TRUE(GetVarint32Ptr(s.data(), s.data() + len, &result) == NULL); } ASSERT_TRUE(GetVarint32Ptr(s.data(), s.data() + s.size(), &result) != NULL); @@ -162,7 +162,7 @@ TEST(Coding, Varint64Truncation) { std::string s; PutVarint64(&s, large_value); uint64_t result; - for (size_t len = 0; len < s.size() - 1; len++) { + for (int len = 0; len < s.size() - 1; len++) { ASSERT_TRUE(GetVarint64Ptr(s.data(), s.data() + len, &result) == NULL); } ASSERT_TRUE(GetVarint64Ptr(s.data(), s.data() + s.size(), &result) != NULL); diff --git a/src/leveldb/util/comparator.cc b/src/leveldb/util/comparator.cc index 4b7b5724..6cc31924 100644 --- a/src/leveldb/util/comparator.cc +++ b/src/leveldb/util/comparator.cc @@ -66,7 +66,7 @@ class BytewiseComparatorImpl : public Comparator { }; } // namespace -static port::OnceType once = LEVELDB_ONCE_INIT; +static port::OnceType once_comparator = LEVELDB_ONCE_INIT; static const Comparator* bytewise; static void InitModule() { @@ -74,7 +74,7 @@ static void InitModule() { } const Comparator* BytewiseComparator() { - port::InitOnce(&once, InitModule); + port::InitOnce(&once_comparator, InitModule); return bytewise; } diff --git a/src/leveldb/util/env_posix.cc b/src/leveldb/util/env_posix.cc index 93eadb1a..6badfdc2 100644 --- a/src/leveldb/util/env_posix.cc +++ b/src/leveldb/util/env_posix.cc @@ -176,83 +176,172 @@ class PosixMmapReadableFile: public RandomAccessFile { } }; -class PosixWritableFile : public WritableFile { +// We preallocate up to an extra megabyte and use memcpy to append new +// data to the file. This is safe since we either properly close the +// file before reading from it, or for log files, the reading code +// knows enough to skip zero suffixes. +class PosixMmapFile : public WritableFile { private: std::string filename_; - FILE* file_; + int fd_; + size_t page_size_; + size_t map_size_; // How much extra memory to map at a time + char* base_; // The mapped region + char* limit_; // Limit of the mapped region + char* dst_; // Where to write next (in range [base_,limit_]) + char* last_sync_; // Where have we synced up to + uint64_t file_offset_; // Offset of base_ in file + + // Have we done an munmap of unsynced data? + bool pending_sync_; + + // Roundup x to a multiple of y + static size_t Roundup(size_t x, size_t y) { + return ((x + y - 1) / y) * y; + } + + size_t TruncateToPageBoundary(size_t s) { + s -= (s & (page_size_ - 1)); + assert((s % page_size_) == 0); + return s; + } + + bool UnmapCurrentRegion() { + bool result = true; + if (base_ != NULL) { + if (last_sync_ < limit_) { + // Defer syncing this data until next Sync() call, if any + pending_sync_ = true; + } + if (munmap(base_, limit_ - base_) != 0) { + result = false; + } + file_offset_ += limit_ - base_; + base_ = NULL; + limit_ = NULL; + last_sync_ = NULL; + dst_ = NULL; + + // Increase the amount we map the next time, but capped at 1MB + if (map_size_ < (1<<20)) { + map_size_ *= 2; + } + } + return result; + } + + bool MapNewRegion() { + assert(base_ == NULL); + if (ftruncate(fd_, file_offset_ + map_size_) < 0) { + return false; + } + void* ptr = mmap(NULL, map_size_, PROT_READ | PROT_WRITE, MAP_SHARED, + fd_, file_offset_); + if (ptr == MAP_FAILED) { + return false; + } + base_ = reinterpret_cast(ptr); + limit_ = base_ + map_size_; + dst_ = base_; + last_sync_ = base_; + return true; + } public: - PosixWritableFile(const std::string& fname, FILE* f) - : filename_(fname), file_(f) { } + PosixMmapFile(const std::string& fname, int fd, size_t page_size) + : filename_(fname), + fd_(fd), + page_size_(page_size), + map_size_(Roundup(65536, page_size)), + base_(NULL), + limit_(NULL), + dst_(NULL), + last_sync_(NULL), + file_offset_(0), + pending_sync_(false) { + assert((page_size & (page_size - 1)) == 0); + } - ~PosixWritableFile() { - if (file_ != NULL) { - // Ignoring any potential errors - fclose(file_); + + ~PosixMmapFile() { + if (fd_ >= 0) { + PosixMmapFile::Close(); } } virtual Status Append(const Slice& data) { - size_t r = fwrite_unlocked(data.data(), 1, data.size(), file_); - if (r != data.size()) { - return IOError(filename_, errno); + const char* src = data.data(); + size_t left = data.size(); + while (left > 0) { + assert(base_ <= dst_); + assert(dst_ <= limit_); + size_t avail = limit_ - dst_; + if (avail == 0) { + if (!UnmapCurrentRegion() || + !MapNewRegion()) { + return IOError(filename_, errno); + } + } + + size_t n = (left <= avail) ? left : avail; + memcpy(dst_, src, n); + dst_ += n; + src += n; + left -= n; } return Status::OK(); } virtual Status Close() { - Status result; - if (fclose(file_) != 0) { - result = IOError(filename_, errno); - } - file_ = NULL; - return result; - } - - virtual Status Flush() { - if (fflush_unlocked(file_) != 0) { - return IOError(filename_, errno); - } - return Status::OK(); - } - - Status SyncDirIfManifest() { - const char* f = filename_.c_str(); - const char* sep = strrchr(f, '/'); - Slice basename; - std::string dir; - if (sep == NULL) { - dir = "."; - basename = f; - } else { - dir = std::string(f, sep - f); - basename = sep + 1; - } Status s; - if (basename.starts_with("MANIFEST")) { - int fd = open(dir.c_str(), O_RDONLY); - if (fd < 0) { - s = IOError(dir, errno); - } else { - if (fsync(fd) < 0) { - s = IOError(dir, errno); - } - close(fd); + size_t unused = limit_ - dst_; + if (!UnmapCurrentRegion()) { + s = IOError(filename_, errno); + } else if (unused > 0) { + // Trim the extra space at the end of the file + if (ftruncate(fd_, file_offset_ - unused) < 0) { + s = IOError(filename_, errno); } } + + if (close(fd_) < 0) { + if (s.ok()) { + s = IOError(filename_, errno); + } + } + + fd_ = -1; + base_ = NULL; + limit_ = NULL; return s; } + virtual Status Flush() { + return Status::OK(); + } + virtual Status Sync() { - // Ensure new files referred to by the manifest are in the filesystem. - Status s = SyncDirIfManifest(); - if (!s.ok()) { - return s; + Status s; + + if (pending_sync_) { + // Some unmapped data was not synced + pending_sync_ = false; + if (fdatasync(fd_) < 0) { + s = IOError(filename_, errno); + } } - if (fflush_unlocked(file_) != 0 || - fdatasync(fileno(file_)) != 0) { - s = Status::IOError(filename_, strerror(errno)); + + if (dst_ > last_sync_) { + // Find the beginnings of the pages that contain the first and last + // bytes to be synced. + size_t p1 = TruncateToPageBoundary(last_sync_ - base_); + size_t p2 = TruncateToPageBoundary(dst_ - base_ - 1); + last_sync_ = dst_; + if (msync(base_ + p1, p2 - p1 + page_size_, MS_SYNC) < 0) { + s = IOError(filename_, errno); + } } + return s; } }; @@ -343,12 +432,12 @@ class PosixEnv : public Env { virtual Status NewWritableFile(const std::string& fname, WritableFile** result) { Status s; - FILE* f = fopen(fname.c_str(), "w"); - if (f == NULL) { + const int fd = open(fname.c_str(), O_CREAT | O_RDWR | O_TRUNC, 0644); + if (fd < 0) { *result = NULL; s = IOError(fname, errno); } else { - *result = new PosixWritableFile(fname, f); + *result = new PosixMmapFile(fname, fd, page_size_); } return s; } @@ -511,6 +600,7 @@ class PosixEnv : public Env { return NULL; } + size_t page_size_; pthread_mutex_t mu_; pthread_cond_t bgsignal_; pthread_t bgthread_; @@ -525,7 +615,8 @@ class PosixEnv : public Env { MmapLimiter mmap_limit_; }; -PosixEnv::PosixEnv() : started_bgthread_(false) { +PosixEnv::PosixEnv() : page_size_(getpagesize()), + started_bgthread_(false) { PthreadCall("mutex_init", pthread_mutex_init(&mu_, NULL)); PthreadCall("cvar_init", pthread_cond_init(&bgsignal_, NULL)); } diff --git a/src/leveldb/util/random.h b/src/leveldb/util/random.h index ddd51b1c..07538242 100644 --- a/src/leveldb/util/random.h +++ b/src/leveldb/util/random.h @@ -16,12 +16,7 @@ class Random { private: uint32_t seed_; public: - explicit Random(uint32_t s) : seed_(s & 0x7fffffffu) { - // Avoid bad seeds. - if (seed_ == 0 || seed_ == 2147483647L) { - seed_ = 1; - } - } + explicit Random(uint32_t s) : seed_(s & 0x7fffffffu) { } uint32_t Next() { static const uint32_t M = 2147483647L; // 2^31-1 static const uint64_t A = 16807; // bits 14, 8, 7, 5, 2, 1, 0 diff --git a/src/leveldb/util/testharness.cc b/src/leveldb/util/testharness.cc index 402fab34..eb1bdd55 100644 --- a/src/leveldb/util/testharness.cc +++ b/src/leveldb/util/testharness.cc @@ -38,7 +38,7 @@ int RunAllTests() { int num = 0; if (tests != NULL) { - for (size_t i = 0; i < tests->size(); i++) { + for (int i = 0; i < tests->size(); i++) { const Test& t = (*tests)[i]; if (matcher != NULL) { std::string name = t.base; diff --git a/src/leveldb/util/testutil.cc b/src/leveldb/util/testutil.cc index bee56bf7..538d0951 100644 --- a/src/leveldb/util/testutil.cc +++ b/src/leveldb/util/testutil.cc @@ -32,7 +32,7 @@ std::string RandomKey(Random* rnd, int len) { extern Slice CompressibleString(Random* rnd, double compressed_fraction, - size_t len, std::string* dst) { + int len, std::string* dst) { int raw = static_cast(len * compressed_fraction); if (raw < 1) raw = 1; std::string raw_data; diff --git a/src/leveldb/util/testutil.h b/src/leveldb/util/testutil.h index adad3fc1..824e655b 100644 --- a/src/leveldb/util/testutil.h +++ b/src/leveldb/util/testutil.h @@ -24,7 +24,7 @@ extern std::string RandomKey(Random* rnd, int len); // "N*compressed_fraction" bytes and return a Slice that references // the generated data. extern Slice CompressibleString(Random* rnd, double compressed_fraction, - size_t len, std::string* dst); + int len, std::string* dst); // A wrapper that allows injection of errors. class ErrorEnv : public EnvWrapper {