Release LevelDB 1.16

- Make Log::Reader not report a corruption when the last record in a log file is truncated. - Fix issue 224: variable created but not utilized. - Remove comment that referenced a removed feature.
Release LevelDB 1.15
2026-08-20 10:43:29 +08:00 · 2014-02-10 11:36:06 -08:00 · 2013-12-10 10:36:31 -08:00 · 2013-09-19 13:49:19 -07:00 · 2013-08-21 11:12:47 -07:00 · 2013-06-18 11:00:13 -07:00
46 changed files with 1103 additions and 458 deletions
--- a/.gitignore
+++ b/.gitignore
@@ -6,3 +6,4 @@ build_config.mk
 *.so.*
 *_test
 db_bench
+leveldbutil
--- a/4
+++ b/4
@@ -6,3 +6,7 @@ Google Inc.
 # Initial version authors:
 Jeffrey Dean <jeff@google.com>
 Sanjay Ghemawat <sanjay@google.com>
+
+# Partial list of contributors:
+Kevin Regan <kevin.d.regan@gmail.com>
+Johan Bilien <jobi@litl.com>
--- a/20
+++ b/20
@@ -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,6 +31,7 @@ TESTHARNESS = ./util/testharness.o $(TESTUTIL)

 TESTS = \
 	arena_test \
+	autocompact_test \
 	bloom_test \
 	c_test \
 	cache_test \
@@ -42,6 +43,8 @@ TESTS = \
 	env_test \
 	filename_test \
 	filter_block_test \
+	issue178_test \
+	issue200_test \
 	log_test \
 	memenv_test \
 	skiplist_test \
@@ -69,7 +72,7 @@ SHARED = $(SHARED1)
 else
 # Update db.h if you change these.
 SHARED_MAJOR = 1
-SHARED_MINOR = 9
+SHARED_MINOR = 16
 SHARED1 = libleveldb.$(PLATFORM_SHARED_EXT)
 SHARED2 = $(SHARED1).$(SHARED_MAJOR)
 SHARED3 = $(SHARED1).$(SHARED_MAJOR).$(SHARED_MINOR)
@@ -113,6 +116,9 @@ 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)

@@ -146,6 +152,12 @@ filename_test: db/filename_test.o $(LIBOBJECTS) $(TESTHARNESS)
 filter_block_test: table/filter_block_test.o $(LIBOBJECTS) $(TESTHARNESS)
 	$(CXX) $(LDFLAGS) table/filter_block_test.o $(LIBOBJECTS) $(TESTHARNESS) -o $@ $(LIBS)

+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)

@@ -183,14 +195,14 @@ IOSVERSION=$(shell defaults read $(PLATFORMSROOT)/iPhoneOS.platform/version CFBu
 	mkdir -p ios-x86/$(dir $@)
 	$(CXX) $(CXXFLAGS) -isysroot $(SIMULATORROOT)/SDKs/iPhoneSimulator$(IOSVERSION).sdk -arch i686 -c $< -o ios-x86/$@
 	mkdir -p ios-arm/$(dir $@)
-	$(DEVICEROOT)/usr/bin/$(CXX) $(CXXFLAGS) -isysroot $(DEVICEROOT)/SDKs/iPhoneOS$(IOSVERSION).sdk -arch armv6 -arch armv7 -c $< -o ios-arm/$@
+	xcrun -sdk iphoneos $(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 -c $< -o ios-x86/$@
 	mkdir -p ios-arm/$(dir $@)
-	$(DEVICEROOT)/usr/bin/$(CC) $(CFLAGS) -isysroot $(DEVICEROOT)/SDKs/iPhoneOS$(IOSVERSION).sdk -arch armv6 -arch armv7 -c $< -o ios-arm/$@
+	xcrun -sdk iphoneos $(CC) $(CFLAGS) -isysroot $(DEVICEROOT)/SDKs/iPhoneOS$(IOSVERSION).sdk -arch armv6 -arch armv7 -c $< -o ios-arm/$@
 	lipo ios-x86/$@ ios-arm/$@ -create -output $@

 else
--- a/24
+++ b/24
@@ -44,6 +44,10 @@ 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`
@@ -127,6 +131,16 @@ 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=
+        ;;
    *)
        echo "Unknown platform!" >&2
        exit 1
@@ -155,8 +169,10 @@ 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 <cstdatomic>.  Otherwise use port_posix.h.
-    $CXX $CXXFLAGS -std=c++0x -x c++ - -o /dev/null 2>/dev/null  <<EOF
+    $CXX $CXXFLAGS -std=c++0x -x c++ - -o $CXXOUTPUT 2>/dev/null  <<EOF
      #include <cstdatomic>
      int main() {}
 EOF
@@ -169,7 +185,7 @@ EOF

    # Test whether Snappy library is installed
    # http://code.google.com/p/snappy/
-    $CXX $CXXFLAGS -x c++ - -o /dev/null 2>/dev/null  <<EOF
+    $CXX $CXXFLAGS -x c++ - -o $CXXOUTPUT 2>/dev/null  <<EOF
      #include <snappy.h>
      int main() {}
 EOF
@@ -179,12 +195,14 @@ EOF
    fi

    # Test whether tcmalloc is available
-    $CXX $CXXFLAGS -x c++ - -o /dev/null -ltcmalloc 2>/dev/null  <<EOF
+    $CXX $CXXFLAGS -x c++ - -o $CXXOUTPUT -ltcmalloc 2>/dev/null  <<EOF
      int main() {}
 EOF
    if [ "$?" = 0 ]; then
        PLATFORM_LIBS="$PLATFORM_LIBS -ltcmalloc"
    fi
+
+    rm -f $CXXOUTPUT 2>/dev/null
 fi

 PLATFORM_CCFLAGS="$PLATFORM_CCFLAGS $COMMON_FLAGS"
--- a/db/autocompact_test.cc
+++ b/db/autocompact_test.cc
@@ -0,0 +1,118 @@
+// 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<DBImpl*>(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();
+}
--- a/db/corruption_test.cc
+++ b/db/corruption_test.cc
@@ -35,6 +35,7 @@ class CorruptionTest {
  CorruptionTest() {
    tiny_cache_ = NewLRUCache(100);
    options_.env = &env_;
+    options_.block_cache = tiny_cache_;
    dbname_ = test::TmpDir() + "/db_test";
    DestroyDB(dbname_, options_);

@@ -50,17 +51,14 @@ class CorruptionTest {
     delete tiny_cache_;
  }

-  Status TryReopen(Options* options = NULL) {
+  Status TryReopen() {
    delete db_;
    db_ = NULL;
-    Options opt = (options ? *options : options_);
-    opt.env = &env_;
-    opt.block_cache = tiny_cache_;
-    return DB::Open(opt, dbname_, &db_);
+    return DB::Open(options_, dbname_, &db_);
  }

-  void Reopen(Options* options = NULL) {
-    ASSERT_OK(TryReopen(options));
+  void Reopen() {
+    ASSERT_OK(TryReopen());
  }

  void RepairDB() {
@@ -77,7 +75,13 @@ class CorruptionTest {
      Slice key = Key(i, &key_space);
      batch.Clear();
      batch.Put(key, Value(i, &value_space));
-      ASSERT_OK(db_->Write(WriteOptions(), &batch));
+      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));
    }
  }

@@ -92,6 +96,10 @@ 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) {
@@ -123,7 +131,7 @@ class CorruptionTest {
    FileType type;
    std::string fname;
    int picked_number = -1;
-    for (int i = 0; i < filenames.size(); i++) {
+    for (size_t i = 0; i < filenames.size(); i++) {
      if (ParseFileName(filenames[i], &number, &type) &&
          type == filetype &&
          int(number) > picked_number) {  // Pick latest file
@@ -233,7 +241,23 @@ TEST(CorruptionTest, TableFile) {
  dbi->TEST_CompactRange(1, NULL, NULL);

  Corrupt(kTableFile, 100, 1);
-  Check(99, 99);
+  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<DBImpl*>(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);
 }

 TEST(CorruptionTest, TableFileIndexData) {
@@ -299,7 +323,7 @@ TEST(CorruptionTest, CompactionInputError) {
  ASSERT_EQ(1, Property("leveldb.num-files-at-level" + NumberToString(last)));

  Corrupt(kTableFile, 100, 1);
-  Check(9, 9);
+  Check(5, 9);

  // Force compactions by writing lots of values
  Build(10000);
@@ -307,32 +331,23 @@ TEST(CorruptionTest, CompactionInputError) {
 }

 TEST(CorruptionTest, CompactionInputErrorParanoid) {
-  Options options;
-  options.paranoid_checks = true;
-  options.write_buffer_size = 1048576;
-  Reopen(&options);
+  options_.paranoid_checks = true;
+  options_.write_buffer_size = 512 << 10;
+  Reopen();
  DBImpl* dbi = reinterpret_cast<DBImpl*>(db_);

-  // 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");
+  // Make multiple inputs so we need to compact.
+  for (int i = 0; i < 2; i++) {
+    Build(10);
    dbi->TEST_CompactMemTable();
+    Corrupt(kTableFile, 100, 1);
+    env_.SleepForMicroseconds(100000);
  }
+  dbi->CompactRange(NULL, NULL);

-  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;
+  // Write must fail because of corrupted table
  std::string tmp1, tmp2;
-  for (int i = 0; i < 10000 && s.ok(); i++) {
-    s = db_->Put(WriteOptions(), Key(i, &tmp1), Value(i, &tmp2));
-  }
+  Status s = db_->Put(WriteOptions(), Key(5, &tmp1), Value(5, &tmp2));
  ASSERT_TRUE(!s.ok()) << "write did not fail in corrupted paranoid db";
 }

--- a/db/db_bench.cc
+++ b/db/db_bench.cc
@@ -128,7 +128,7 @@ class RandomGenerator {
    pos_ = 0;
  }

-  Slice Generate(int len) {
+  Slice Generate(size_t len) {
    if (pos_ + len > data_.size()) {
      pos_ = 0;
      assert(len < data_.size());
@@ -139,11 +139,11 @@ class RandomGenerator {
 };

 static Slice TrimSpace(Slice s) {
-  int start = 0;
+  size_t start = 0;
  while (start < s.size() && isspace(s[start])) {
    start++;
  }
-  int limit = s.size();
+  size_t limit = s.size();
  while (limit > start && isspace(s[limit-1])) {
    limit--;
  }
@@ -399,7 +399,7 @@ class Benchmark {
    heap_counter_(0) {
    std::vector<std::string> files;
    Env::Default()->GetChildren(FLAGS_db, &files);
-    for (int i = 0; i < files.size(); i++) {
+    for (size_t i = 0; i < files.size(); i++) {
      if (Slice(files[i]).starts_with("heap-")) {
        Env::Default()->DeleteFile(std::string(FLAGS_db) + "/" + files[i]);
      }
--- a/db/db_impl.cc
+++ b/db/db_impl.cc
@@ -35,6 +35,8 @@

 namespace leveldb {

+const int kNumNonTableCacheFiles = 10;
+
 // Information kept for every waiting writer
 struct DBImpl::Writer {
  Status status;
@@ -92,9 +94,9 @@ Options SanitizeOptions(const std::string& dbname,
  Options result = src;
  result.comparator = icmp;
  result.filter_policy = (src.filter_policy != NULL) ? ipolicy : NULL;
-  ClipToRange(&result.max_open_files,            20,     50000);
-  ClipToRange(&result.write_buffer_size,         64<<10, 1<<30);
-  ClipToRange(&result.block_size,                1<<10,  4<<20);
+  ClipToRange(&result.max_open_files,    64 + kNumNonTableCacheFiles, 50000);
+  ClipToRange(&result.write_buffer_size, 64<<10,                      1<<30);
+  ClipToRange(&result.block_size,        1<<10,                       4<<20);
  if (result.info_log == NULL) {
    // Open a log file in the same directory as the db
    src.env->CreateDir(dbname);  // In case it does not exist
@@ -111,14 +113,14 @@ Options SanitizeOptions(const std::string& dbname,
  return result;
 }

-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),
+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),
      dbname_(dbname),
      db_lock_(NULL),
      shutting_down_(NULL),
@@ -128,6 +130,7 @@ DBImpl::DBImpl(const Options& 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) {
@@ -135,7 +138,7 @@ DBImpl::DBImpl(const Options& options, const std::string& dbname)
  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 - 10;
+  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_,
@@ -213,6 +216,12 @@ 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<uint64_t> live = pending_outputs_;
  versions_->AddLiveFiles(&live);
@@ -310,16 +319,24 @@ Status DBImpl::Recover(VersionEdit* edit) {
    if (!s.ok()) {
      return s;
    }
+    std::set<uint64_t> expected;
+    versions_->AddLiveFiles(&expected);
    uint64_t number;
    FileType type;
    std::vector<uint64_t> logs;
    for (size_t i = 0; i < filenames.size(); i++) {
-      if (ParseFileName(filenames[i], &number, &type)
-          && type == kLogFile
-          && ((number >= min_log) || (number == prev_log))) {
-        logs.push_back(number);
+      if (ParseFileName(filenames[i], &number, &type)) {
+        expected.erase(number);
+        if (type == kLogFile && ((number >= min_log) || (number == prev_log)))
+          logs.push_back(number);
      }
    }
+    if (!expected.empty()) {
+      char buf[50];
+      snprintf(buf, sizeof(buf), "%d missing files; e.g.",
+               static_cast<int>(expected.size()));
+      return Status::Corruption(buf, TableFileName(dbname_, *(expected.begin())));
+    }

    // Recover in the order in which the logs were generated
    std::sort(logs.begin(), logs.end());
@@ -483,7 +500,7 @@ Status DBImpl::WriteLevel0Table(MemTable* mem, VersionEdit* edit,
  return s;
 }

-Status DBImpl::CompactMemTable() {
+void DBImpl::CompactMemTable() {
  mutex_.AssertHeld();
  assert(imm_ != NULL);

@@ -511,9 +528,9 @@ Status DBImpl::CompactMemTable() {
    imm_ = NULL;
    has_imm_.Release_Store(NULL);
    DeleteObsoleteFiles();
+  } else {
+    RecordBackgroundError(s);
  }
-
-  return s;
 }

 void DBImpl::CompactRange(const Slice* begin, const Slice* end) {
@@ -556,16 +573,18 @@ void DBImpl::TEST_CompactRange(int level, const Slice* begin,const Slice* end) {
  }

  MutexLock l(&mutex_);
-  while (!manual.done) {
-    while (manual_compaction_ != NULL) {
-      bg_cv_.Wait();
-    }
-    manual_compaction_ = &manual;
-    MaybeScheduleCompaction();
-    while (manual_compaction_ == &manual) {
+  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.
      bg_cv_.Wait();
    }
  }
+  if (manual_compaction_ == &manual) {
+    // Cancel my manual compaction since we aborted early for some reason.
+    manual_compaction_ = NULL;
+  }
 }

 Status DBImpl::TEST_CompactMemTable() {
@@ -584,12 +603,22 @@ 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()) {
@@ -607,24 +636,12 @@ void DBImpl::BGWork(void* db) {
 void DBImpl::BackgroundCall() {
  MutexLock l(&mutex_);
  assert(bg_compaction_scheduled_);
-  if (!shutting_down_.Acquire_Load()) {
-    Status s = BackgroundCompaction();
-    if (s.ok()) {
-      // Success
-    } 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();
-      env_->SleepForMicroseconds(1000000);
-      mutex_.Lock();
-    }
+  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();
  }

  bg_compaction_scheduled_ = false;
@@ -635,11 +652,12 @@ void DBImpl::BackgroundCall() {
  bg_cv_.SignalAll();
 }

-Status DBImpl::BackgroundCompaction() {
+void DBImpl::BackgroundCompaction() {
  mutex_.AssertHeld();

  if (imm_ != NULL) {
-    return CompactMemTable();
+    CompactMemTable();
+    return;
  }

  Compaction* c;
@@ -673,6 +691,9 @@ Status 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<unsigned long long>(f->number),
@@ -683,6 +704,9 @@ Status DBImpl::BackgroundCompaction() {
  } else {
    CompactionState* compact = new CompactionState(c);
    status = DoCompactionWork(compact);
+    if (!status.ok()) {
+      RecordBackgroundError(status);
+    }
    CleanupCompaction(compact);
    c->ReleaseInputs();
    DeleteObsoleteFiles();
@@ -696,9 +720,6 @@ Status 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) {
@@ -714,7 +735,6 @@ Status DBImpl::BackgroundCompaction() {
    }
    manual_compaction_ = NULL;
  }
-  return status;
 }

 void DBImpl::CleanupCompaction(CompactionState* compact) {
@@ -984,6 +1004,9 @@ 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));
@@ -1010,7 +1033,8 @@ static void CleanupIteratorState(void* arg1, void* arg2) {
 }  // namespace

 Iterator* DBImpl::NewInternalIterator(const ReadOptions& options,
-                                      SequenceNumber* latest_snapshot) {
+                                      SequenceNumber* latest_snapshot,
+                                      uint32_t* seed) {
  IterState* cleanup = new IterState;
  mutex_.Lock();
  *latest_snapshot = versions_->LastSequence();
@@ -1034,13 +1058,15 @@ 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;
-  return NewInternalIterator(ReadOptions(), &ignored);
+  uint32_t ignored_seed;
+  return NewInternalIterator(ReadOptions(), &ignored, &ignored_seed);
 }

 int64_t DBImpl::TEST_MaxNextLevelOverlappingBytes() {
@@ -1097,12 +1123,21 @@ Status DBImpl::Get(const ReadOptions& options,

 Iterator* DBImpl::NewIterator(const ReadOptions& options) {
  SequenceNumber latest_snapshot;
-  Iterator* internal_iter = NewInternalIterator(options, &latest_snapshot);
+  uint32_t seed;
+  Iterator* iter = NewInternalIterator(options, &latest_snapshot, &seed);
  return NewDBIterator(
-      &dbname_, env_, user_comparator(), internal_iter,
+      this, user_comparator(), iter,
      (options.snapshot != NULL
       ? reinterpret_cast<const SnapshotImpl*>(options.snapshot)->number_
-       : latest_snapshot));
+       : latest_snapshot),
+      seed);
+}
+
+void DBImpl::RecordReadSample(Slice key) {
+  MutexLock l(&mutex_);
+  if (versions_->current()->RecordReadSample(key)) {
+    MaybeScheduleCompaction();
+  }
 }

 const Snapshot* DBImpl::GetSnapshot() {
@@ -1155,13 +1190,23 @@ 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();

@@ -1268,10 +1313,11 @@ Status DBImpl::MakeRoomForWrite(bool force) {
    } else if (imm_ != NULL) {
      // We have filled up the current memtable, but the previous
      // one is still being compacted, so we wait.
+      Log(options_.info_log, "Current memtable full; waiting...\n");
      bg_cv_.Wait();
    } else if (versions_->NumLevelFiles(0) >= config::kL0_StopWritesTrigger) {
      // There are too many level-0 files.
-      Log(options_.info_log, "waiting...\n");
+      Log(options_.info_log, "Too many L0 files; waiting...\n");
      bg_cv_.Wait();
    } else {
      // Attempt to switch to a new memtable and trigger compaction of old
--- a/db/db_impl.h
+++ b/db/db_impl.h
@@ -59,13 +59,19 @@ 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);
+                                SequenceNumber* latest_snapshot,
+                                uint32_t* seed);

  Status NewDB();

@@ -81,8 +87,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.
-  Status CompactMemTable()
-      EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+  // Errors are recorded in bg_error_.
+  void CompactMemTable() EXCLUSIVE_LOCKS_REQUIRED(mutex_);

  Status RecoverLogFile(uint64_t log_number,
                        VersionEdit* edit,
@@ -96,10 +102,12 @@ 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();
-  Status BackgroundCompaction() EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+  void  BackgroundCompaction() EXCLUSIVE_LOCKS_REQUIRED(mutex_);
  void CleanupCompaction(CompactionState* compact)
      EXCLUSIVE_LOCKS_REQUIRED(mutex_);
  Status DoCompactionWork(CompactionState* compact)
@@ -135,6 +143,7 @@ class DBImpl : public DB {
  WritableFile* logfile_;
  uint64_t logfile_number_;
  log::Writer* log_;
+  uint32_t seed_;                // For sampling.

  // Queue of writers.
  std::deque<Writer*> writers_;
--- a/db/db_iter.cc
+++ b/db/db_iter.cc
@@ -5,12 +5,14 @@
 #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 {

@@ -46,15 +48,16 @@ class DBIter: public Iterator {
    kReverse
  };

-  DBIter(const std::string* dbname, Env* env,
-         const Comparator* cmp, Iterator* iter, SequenceNumber s)
-      : dbname_(dbname),
-        env_(env),
+  DBIter(DBImpl* db, const Comparator* cmp, Iterator* iter, SequenceNumber s,
+         uint32_t seed)
+      : db_(db),
        user_comparator_(cmp),
        iter_(iter),
        sequence_(s),
        direction_(kForward),
-        valid_(false) {
+        valid_(false),
+        rnd_(seed),
+        bytes_counter_(RandomPeriod()) {
  }
  virtual ~DBIter() {
    delete iter_;
@@ -100,8 +103,12 @@ class DBIter: public Iterator {
    }
  }

-  const std::string* const dbname_;
-  Env* const env_;
+  // Pick next gap with average value of config::kReadBytesPeriod.
+  ssize_t RandomPeriod() {
+    return rnd_.Uniform(2*config::kReadBytesPeriod);
+  }
+
+  DBImpl* db_;
  const Comparator* const user_comparator_;
  Iterator* const iter_;
  SequenceNumber const sequence_;
@@ -112,13 +119,23 @@ 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) {
-  if (!ParseInternalKey(iter_->key(), 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)) {
    status_ = Status::Corruption("corrupted internal key in DBIter");
    return false;
  } else {
@@ -144,12 +161,13 @@ 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_);
  }

-  // Temporarily use saved_key_ as storage for key to skip.
-  std::string* skip = &saved_key_;
-  SaveKey(ExtractUserKey(iter_->key()), skip);
-  FindNextUserEntry(true, skip);
+  FindNextUserEntry(true, &saved_key_);
 }

 void DBIter::FindNextUserEntry(bool skipping, std::string* skip) {
@@ -288,12 +306,12 @@ void DBIter::SeekToLast() {
 }  // anonymous namespace

 Iterator* NewDBIterator(
-    const std::string* dbname,
-    Env* env,
+    DBImpl* db,
    const Comparator* user_key_comparator,
    Iterator* internal_iter,
-    const SequenceNumber& sequence) {
-  return new DBIter(dbname, env, user_key_comparator, internal_iter, sequence);
+    SequenceNumber sequence,
+    uint32_t seed) {
+  return new DBIter(db, user_key_comparator, internal_iter, sequence, seed);
 }

 }  // namespace leveldb
--- a/db/db_iter.h
+++ b/db/db_iter.h
@@ -11,15 +11,17 @@

 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(
-    const std::string* dbname,
-    Env* env,
+    DBImpl* db,
    const Comparator* user_key_comparator,
    Iterator* internal_iter,
-    const SequenceNumber& sequence);
+    SequenceNumber sequence,
+    uint32_t seed);

 }  // namespace leveldb

--- a/db/db_test.cc
+++ b/db/db_test.cc
@@ -33,8 +33,11 @@ class AtomicCounter {
 public:
  AtomicCounter() : count_(0) { }
  void Increment() {
+    IncrementBy(1);
+  }
+  void IncrementBy(int count) {
    MutexLock l(&mu_);
-    count_++;
+    count_ += count;
  }
  int Read() {
    MutexLock l(&mu_);
@@ -45,13 +48,20 @@ class AtomicCounter {
    count_ = 0;
  }
 };
+
+void DelayMilliseconds(int millis) {
+  Env::Default()->SleepForMicroseconds(millis * 1000);
+}
 }

 // Special Env used to delay background operations
 class SpecialEnv : public EnvWrapper {
 public:
-  // sstable Sync() calls are blocked while this pointer is non-NULL.
-  port::AtomicPointer delay_sstable_sync_;
+  // 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_;

  // Simulate no-space errors while this pointer is non-NULL.
  port::AtomicPointer no_space_;
@@ -68,10 +78,9 @@ class SpecialEnv : public EnvWrapper {
  bool count_random_reads_;
  AtomicCounter random_read_counter_;

-  AtomicCounter sleep_counter_;
-
  explicit SpecialEnv(Env* base) : EnvWrapper(base) {
-    delay_sstable_sync_.Release_Store(NULL);
+    delay_data_sync_.Release_Store(NULL);
+    data_sync_error_.Release_Store(NULL);
    no_space_.Release_Store(NULL);
    non_writable_.Release_Store(NULL);
    count_random_reads_ = false;
@@ -80,17 +89,17 @@ class SpecialEnv : public EnvWrapper {
  }

  Status NewWritableFile(const std::string& f, WritableFile** r) {
-    class SSTableFile : public WritableFile {
+    class DataFile : public WritableFile {
     private:
      SpecialEnv* env_;
      WritableFile* base_;

     public:
-      SSTableFile(SpecialEnv* env, WritableFile* base)
+      DataFile(SpecialEnv* env, WritableFile* base)
          : env_(env),
            base_(base) {
      }
-      ~SSTableFile() { delete base_; }
+      ~DataFile() { delete base_; }
      Status Append(const Slice& data) {
        if (env_->no_space_.Acquire_Load() != NULL) {
          // Drop writes on the floor
@@ -102,8 +111,11 @@ class SpecialEnv : public EnvWrapper {
      Status Close() { return base_->Close(); }
      Status Flush() { return base_->Flush(); }
      Status Sync() {
-        while (env_->delay_sstable_sync_.Acquire_Load() != NULL) {
-          env_->SleepForMicroseconds(100000);
+        if (env_->data_sync_error_.Acquire_Load() != NULL) {
+          return Status::IOError("simulated data sync error");
+        }
+        while (env_->delay_data_sync_.Acquire_Load() != NULL) {
+          DelayMilliseconds(100);
        }
        return base_->Sync();
      }
@@ -139,8 +151,9 @@ class SpecialEnv : public EnvWrapper {

    Status s = target()->NewWritableFile(f, r);
    if (s.ok()) {
-      if (strstr(f.c_str(), ".sst") != NULL) {
-        *r = new SSTableFile(this, *r);
+      if (strstr(f.c_str(), ".ldb") != NULL ||
+          strstr(f.c_str(), ".log") != NULL) {
+        *r = new DataFile(this, *r);
      } else if (strstr(f.c_str(), "MANIFEST") != NULL) {
        *r = new ManifestFile(this, *r);
      }
@@ -171,11 +184,6 @@ class SpecialEnv : public EnvWrapper {
    }
    return s;
  }
-
-  virtual void SleepForMicroseconds(int micros) {
-    sleep_counter_.Increment();
-    target()->SleepForMicroseconds(micros);
-  }
 };

 class DBTest {
@@ -313,7 +321,7 @@ class DBTest {
    }

    // Check reverse iteration results are the reverse of forward results
-    int matched = 0;
+    size_t matched = 0;
    for (iter->SeekToLast(); iter->Valid(); iter->Prev()) {
      ASSERT_LT(matched, forward.size());
      ASSERT_EQ(IterStatus(iter), forward[forward.size() - matched - 1]);
@@ -461,6 +469,38 @@ class DBTest {
    }
    return result;
  }
+
+  bool DeleteAnSSTFile() {
+    std::vector<std::string> filenames;
+    ASSERT_OK(env_->GetChildren(dbname_, &filenames));
+    uint64_t number;
+    FileType type;
+    for (size_t i = 0; i < filenames.size(); i++) {
+      if (ParseFileName(filenames[i], &number, &type) && type == kTableFile) {
+        ASSERT_OK(env_->DeleteFile(TableFileName(dbname_, number)));
+        return true;
+      }
+    }
+    return false;
+  }
+
+  // Returns number of files renamed.
+  int RenameLDBToSST() {
+    std::vector<std::string> 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) {
@@ -502,11 +542,11 @@ TEST(DBTest, GetFromImmutableLayer) {
    ASSERT_OK(Put("foo", "v1"));
    ASSERT_EQ("v1", Get("foo"));

-    env_->delay_sstable_sync_.Release_Store(env_);   // Block sync calls
+    env_->delay_data_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_sstable_sync_.Release_Store(NULL);   // Release sync calls
+    env_->delay_data_sync_.Release_Store(NULL);      // Release sync calls
  } while (ChangeOptions());
 }

@@ -611,7 +651,7 @@ TEST(DBTest, GetEncountersEmptyLevel) {
    }

    // Step 4: Wait for compaction to finish
-    env_->SleepForMicroseconds(1000000);
+    DelayMilliseconds(1000);

    ASSERT_EQ(NumTableFilesAtLevel(0), 0);
  } while (ChangeOptions());
@@ -1295,7 +1335,7 @@ TEST(DBTest, L0_CompactionBug_Issue44_a) {
  Reopen();
  Reopen();
  ASSERT_EQ("(a->v)", Contents());
-  env_->SleepForMicroseconds(1000000);  // Wait for compaction to finish
+  DelayMilliseconds(1000);  // Wait for compaction to finish
  ASSERT_EQ("(a->v)", Contents());
 }

@@ -1311,7 +1351,7 @@ TEST(DBTest, L0_CompactionBug_Issue44_b) {
  Put("","");
  Reopen();
  Put("","");
-  env_->SleepForMicroseconds(1000000);  // Wait for compaction to finish
+  DelayMilliseconds(1000);  // Wait for compaction to finish
  Reopen();
  Put("d","dv");
  Reopen();
@@ -1321,7 +1361,7 @@ TEST(DBTest, L0_CompactionBug_Issue44_b) {
  Delete("b");
  Reopen();
  ASSERT_EQ("(->)(c->cv)", Contents());
-  env_->SleepForMicroseconds(1000000);  // Wait for compaction to finish
+  DelayMilliseconds(1000);  // Wait for compaction to finish
  ASSERT_EQ("(->)(c->cv)", Contents());
 }

@@ -1493,17 +1533,13 @@ TEST(DBTest, NoSpace) {
  Compact("a", "z");
  const int num_files = CountFiles();
  env_->no_space_.Release_Store(env_);   // Force out-of-space errors
-  env_->sleep_counter_.Reset();
-  for (int i = 0; i < 5; i++) {
+  for (int i = 0; i < 10; 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, NonWritableFileSystem) {
@@ -1519,13 +1555,44 @@ TEST(DBTest, NonWritableFileSystem) {
    fprintf(stderr, "iter %d; errors %d\n", i, errors);
    if (!Put("foo", big).ok()) {
      errors++;
-      env_->SleepForMicroseconds(100000);
+      DelayMilliseconds(100);
    }
  }
  ASSERT_GT(errors, 0);
  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
@@ -1567,6 +1634,40 @@ TEST(DBTest, ManifestWriteError) {
  }
 }

+TEST(DBTest, MissingSSTFile) {
+  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_TRUE(DeleteAnSSTFile());
+  Options options = CurrentOptions();
+  options.paranoid_checks = true;
+  Status s = TryReopen(&options);
+  ASSERT_TRUE(!s.ok());
+  ASSERT_TRUE(s.ToString().find("issing") != std::string::npos)
+      << 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");
@@ -1598,7 +1699,7 @@ TEST(DBTest, BloomFilter) {
  dbfull()->TEST_CompactMemTable();

  // Prevent auto compactions triggered by seeks
-  env_->delay_sstable_sync_.Release_Store(env_);
+  env_->delay_data_sync_.Release_Store(env_);

  // Lookup present keys.  Should rarely read from small sstable.
  env_->random_read_counter_.Reset();
@@ -1619,7 +1720,7 @@ TEST(DBTest, BloomFilter) {
  fprintf(stderr, "%d missing => %d reads\n", N, reads);
  ASSERT_LE(reads, 3*N/100);

-  env_->delay_sstable_sync_.Release_Store(NULL);
+  env_->delay_data_sync_.Release_Store(NULL);
  Close();
  delete options.block_cache;
  delete options.filter_policy;
@@ -1679,7 +1780,7 @@ static void MTThreadBody(void* arg) {
        ASSERT_EQ(k, key);
        ASSERT_GE(w, 0);
        ASSERT_LT(w, kNumThreads);
-        ASSERT_LE(c, reinterpret_cast<uintptr_t>(
+        ASSERT_LE(static_cast<uintptr_t>(c), reinterpret_cast<uintptr_t>(
            t->state->counter[w].Acquire_Load()));
      }
    }
@@ -1711,13 +1812,13 @@ TEST(DBTest, MultiThreaded) {
    }

    // Let them run for a while
-    env_->SleepForMicroseconds(kTestSeconds * 1000000);
+    DelayMilliseconds(kTestSeconds * 1000);

    // Stop the threads and wait for them to finish
    mt.stop.Release_Store(&mt);
    for (int id = 0; id < kNumThreads; id++) {
      while (mt.thread_done[id].Acquire_Load() == NULL) {
-        env_->SleepForMicroseconds(100000);
+        DelayMilliseconds(100);
      }
    }
  } while (ChangeOptions());
--- a/db/dbformat.cc
+++ b/db/dbformat.cc
@@ -26,7 +26,7 @@ std::string ParsedInternalKey::DebugString() const {
           (unsigned long long) sequence,
           int(type));
  std::string result = "'";
-  result += user_key.ToString();
+  result += EscapeString(user_key.ToString());
  result += buf;
  return result;
 }
--- a/db/dbformat.h
+++ b/db/dbformat.h
@@ -38,6 +38,9 @@ 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;
--- a/db/filename.cc
+++ b/db/filename.cc
@@ -30,6 +30,11 @@ 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");
 }
@@ -71,7 +76,7 @@ std::string OldInfoLogFileName(const std::string& dbname) {
 //    dbname/LOG
 //    dbname/LOG.old
 //    dbname/MANIFEST-[0-9]+
-//    dbname/[0-9]+.(log|sst)
+//    dbname/[0-9]+.(log|sst|ldb)
 bool ParseFileName(const std::string& fname,
                   uint64_t* number,
                   FileType* type) {
@@ -106,7 +111,7 @@ bool ParseFileName(const std::string& fname,
    Slice suffix = rest;
    if (suffix == Slice(".log")) {
      *type = kLogFile;
-    } else if (suffix == Slice(".sst")) {
+    } else if (suffix == Slice(".sst") || suffix == Slice(".ldb")) {
      *type = kTableFile;
    } else if (suffix == Slice(".dbtmp")) {
      *type = kTempFile;
--- a/db/filename.h
+++ b/db/filename.h
@@ -37,6 +37,11 @@ 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".
--- a/db/filename_test.cc
+++ b/db/filename_test.cc
@@ -27,6 +27,7 @@ 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 },
@@ -70,7 +71,7 @@ TEST(FileNameTest, Parse) {
  for (int i = 0; i < sizeof(errors) / sizeof(errors[0]); i++) {
    std::string f = errors[i];
    ASSERT_TRUE(!ParseFileName(f, &number, &type)) << f;
-  };
+  }
 }

 TEST(FileNameTest, Construction) {
--- a/db/log_reader.cc
+++ b/db/log_reader.cc
@@ -133,7 +133,9 @@ bool Reader::ReadRecord(Slice* record, std::string* scratch) {

      case kEof:
        if (in_fragmented_record) {
-          ReportCorruption(scratch->size(), "partial record without end(3)");
+          // 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.
          scratch->clear();
        }
        return false;
@@ -193,13 +195,12 @@ unsigned int Reader::ReadPhysicalRecord(Slice* result) {
          eof_ = true;
        }
        continue;
-      } else if (buffer_.size() == 0) {
-        // End of file
-        return kEof;
      } else {
-        size_t drop_size = buffer_.size();
+        // 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.
        buffer_.clear();
-        ReportCorruption(drop_size, "truncated record at end of file");
        return kEof;
      }
    }
@@ -213,8 +214,14 @@ unsigned int Reader::ReadPhysicalRecord(Slice* result) {
    if (kHeaderSize + length > buffer_.size()) {
      size_t drop_size = buffer_.size();
      buffer_.clear();
-      ReportCorruption(drop_size, "bad record length");
-      return kBadRecord;
+      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;
    }

    if (type == kZeroType && length == 0) {
--- a/db/log_test.cc
+++ b/db/log_test.cc
@@ -351,20 +351,32 @@ TEST(LogTest, BadRecordType) {
  ASSERT_EQ("OK", MatchError("unknown record type"));
 }

-TEST(LogTest, TruncatedTrailingRecord) {
+TEST(LogTest, TruncatedTrailingRecordIsIgnored) {
  Write("foo");
  ShrinkSize(4);   // Drop all payload as well as a header byte
  ASSERT_EQ("EOF", Read());
-  ASSERT_EQ(kHeaderSize - 1, DroppedBytes());
-  ASSERT_EQ("OK", MatchError("truncated record at end of file"));
+  // Truncated last record is ignored, not treated as an error.
+  ASSERT_EQ(0, DroppedBytes());
+  ASSERT_EQ("", ReportMessage());
 }

 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(kHeaderSize + 2, DroppedBytes());
-  ASSERT_EQ("OK", MatchError("bad record length"));
+  ASSERT_EQ(0, DroppedBytes());
+  ASSERT_EQ("", ReportMessage());
 }

 TEST(LogTest, ChecksumMismatch) {
@@ -415,6 +427,24 @@ 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)
--- a/db/repair.cc
+++ b/db/repair.cc
@@ -242,62 +242,134 @@ class Repairer {
  }

  void ExtractMetaData() {
-    std::vector<TableInfo> kept;
    for (size_t i = 0; i < table_numbers_.size(); i++) {
-      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);
-      }
+      ScanTable(table_numbers_[i]);
    }
  }

-  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;
-        }
+  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);
+  }

-        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;
-        }
+  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 (!iter->status().ok()) {
-        status = iter->status();
-      }
-      delete iter;
    }
+    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,
+        (unsigned long long) t.meta.number,
        counter,
        status.ToString().c_str());
-    return status;
+
+    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);
+        tables_.push_back(t);
+      }
+    }
+    if (!s.ok()) {
+      env_->DeleteFile(copy);
+    }
  }

  Status WriteDescriptor() {
--- a/db/table_cache.cc
+++ b/db/table_cache.cc
@@ -54,6 +54,12 @@ 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);
    }
--- a/db/version_set.cc
+++ b/db/version_set.cc
@@ -289,6 +289,51 @@ 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<FileMetaData*> 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,
@@ -401,6 +446,44 @@ 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<State*>(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_;
 }
@@ -435,10 +518,13 @@ int Version::PickLevelForMemTableOutput(
      if (OverlapInLevel(level + 1, &smallest_user_key, &largest_user_key)) {
        break;
      }
-      GetOverlappingInputs(level + 2, &start, &limit, &overlaps);
-      const int64_t sum = TotalFileSize(overlaps);
-      if (sum > kMaxGrandParentOverlapBytes) {
-        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;
+        }
      }
      level++;
    }
@@ -452,6 +538,8 @@ void Version::GetOverlappingInputs(
    const InternalKey* begin,
    const InternalKey* end,
    std::vector<FileMetaData*>* inputs) {
+  assert(level >= 0);
+  assert(level < config::kNumLevels);
  inputs->clear();
  Slice user_begin, user_end;
  if (begin != NULL) {
@@ -788,12 +876,6 @@ 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();
-        }
      }
    }

@@ -801,8 +883,6 @@ 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();
@@ -1036,31 +1116,6 @@ 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++) {
@@ -1331,14 +1386,19 @@ Compaction* VersionSet::CompactRange(
  }

  // Avoid compacting too much in one shot in case the range is large.
-  const uint64_t limit = MaxFileSizeForLevel(level);
-  uint64_t total = 0;
-  for (size_t i = 0; i < inputs.size(); i++) {
-    uint64_t s = inputs[i]->file_size;
-    total += s;
-    if (total >= limit) {
-      inputs.resize(i + 1);
-      break;
+  // But we cannot do this for level-0 since level-0 files can overlap
+  // and we must not pick one file and drop another older file if the
+  // two files overlap.
+  if (level > 0) {
+    const uint64_t limit = MaxFileSizeForLevel(level);
+    uint64_t total = 0;
+    for (size_t i = 0; i < inputs.size(); i++) {
+      uint64_t s = inputs[i]->file_size;
+      total += s;
+      if (total >= limit) {
+        inputs.resize(i + 1);
+        break;
+      }
    }
  }

--- a/db/version_set.h
+++ b/db/version_set.h
@@ -78,6 +78,12 @@ 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();
@@ -114,6 +120,15 @@ 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
@@ -277,8 +292,6 @@ class VersionSet {

  void AppendVersion(Version* v);

-  bool ManifestContains(const std::string& record) const;
-
  Env* const env_;
  const std::string dbname_;
  const Options* const options_;
--- a/doc/impl.html
+++ b/doc/impl.html
@@ -11,7 +11,7 @@

 The implementation of leveldb is similar in spirit to the
 representation of a single
-<a href="http://labs.google.com/papers/bigtable.html">
+<a href="http://research.google.com/archive/bigtable.html">
 Bigtable tablet (section 5.3)</a>.
 However the organization of the files that make up the representation
 is somewhat different and is explained below.
--- a/include/leveldb/c.h
+++ b/include/leveldb/c.h
@@ -9,7 +9,6 @@
  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:
--- a/include/leveldb/db.h
+++ b/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 = 9;
+static const int kMinorVersion = 16;

 struct Options;
 struct ReadOptions;
--- a/include/leveldb/env.h
+++ b/include/leveldb/env.h
@@ -13,9 +13,9 @@
 #ifndef STORAGE_LEVELDB_INCLUDE_ENV_H_
 #define STORAGE_LEVELDB_INCLUDE_ENV_H_

-#include <cstdarg>
 #include <string>
 #include <vector>
+#include <stdarg.h>
 #include <stdint.h>
 #include "leveldb/status.h"

--- a/issues/issue178_test.cc
+++ b/issues/issue178_test.cc
@@ -0,0 +1,92 @@
+// 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 <iostream>
+#include <sstream>
+#include <cstdlib>
+
+#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();
+}
--- a/issues/issue200_test.cc
+++ b/issues/issue200_test.cc
@@ -0,0 +1,59 @@
+// 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();
+}
--- a/port/atomic_pointer.h
+++ b/port/atomic_pointer.h
@@ -50,6 +50,13 @@ 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() {
@@ -68,13 +75,6 @@ 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);
--- a/table/block.cc
+++ b/table/block.cc
@@ -16,7 +16,7 @@
 namespace leveldb {

 inline uint32_t Block::NumRestarts() const {
-  assert(size_ >= 2*sizeof(uint32_t));
+  assert(size_ >= sizeof(uint32_t));
  return DecodeFixed32(data_ + size_ - sizeof(uint32_t));
 }

@@ -27,11 +27,12 @@ Block::Block(const BlockContents& contents)
  if (size_ < sizeof(uint32_t)) {
    size_ = 0;  // Error marker
  } else {
-    restart_offset_ = size_ - (1 + NumRestarts()) * sizeof(uint32_t);
-    if (restart_offset_ > size_ - sizeof(uint32_t)) {
-      // The size is too small for NumRestarts() and therefore
-      // restart_offset_ wrapped around.
+    size_t max_restarts_allowed = (size_-sizeof(uint32_t)) / sizeof(uint32_t);
+    if (NumRestarts() > max_restarts_allowed) {
+      // The size is too small for NumRestarts()
      size_ = 0;
+    } else {
+      restart_offset_ = size_ - (1 + NumRestarts()) * sizeof(uint32_t);
    }
  }
 }
@@ -253,7 +254,7 @@ class Block::Iter : public Iterator {
 };

 Iterator* Block::NewIterator(const Comparator* cmp) {
-  if (size_ < 2*sizeof(uint32_t)) {
+  if (size_ < sizeof(uint32_t)) {
    return NewErrorIterator(Status::Corruption("bad block contents"));
  }
  const uint32_t num_restarts = NumRestarts();
--- a/table/filter_block_test.cc
+++ b/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 (int i = 0; i + 4 <= filter.size(); i += 4) {
+    for (size_t i = 0; i + 4 <= filter.size(); i += 4) {
      if (h == DecodeFixed32(filter.data() + i)) {
        return true;
      }
--- a/table/table.cc
+++ b/table/table.cc
@@ -228,7 +228,6 @@ Status Table::InternalGet(const ReadOptions& options, const Slice& k,
        !filter->KeyMayMatch(handle.offset(), k)) {
      // Not found
    } else {
-      Slice handle = iiter->value();
      Iterator* block_iter = BlockReader(this, options, iiter->value());
      block_iter->Seek(k);
      if (block_iter->Valid()) {
--- a/table/table_test.cc
+++ b/table/table_test.cc
@@ -644,6 +644,36 @@ class Harness {
  Constructor* constructor_;
 };

+// Test empty table/block.
+TEST(Harness, Empty) {
+  for (int i = 0; i < kNumTestArgs; i++) {
+    Init(kTestArgList[i]);
+    Random rnd(test::RandomSeed() + 1);
+    Test(&rnd);
+  }
+}
+
+// Special test for a block with no restart entries.  The C++ leveldb
+// code never generates such blocks, but the Java version of leveldb
+// seems to.
+TEST(Harness, ZeroRestartPointsInBlock) {
+  char data[sizeof(uint32_t)];
+  memset(data, 0, sizeof(data));
+  BlockContents contents;
+  contents.data = Slice(data, sizeof(data));
+  contents.cachable = false;
+  contents.heap_allocated = false;
+  Block block(contents);
+  Iterator* iter = block.NewIterator(BytewiseComparator());
+  iter->SeekToFirst();
+  ASSERT_TRUE(!iter->Valid());
+  iter->SeekToLast();
+  ASSERT_TRUE(!iter->Valid());
+  iter->Seek("foo");
+  ASSERT_TRUE(!iter->Valid());
+  delete iter;
+}
+
 // Test the empty key
 TEST(Harness, SimpleEmptyKey) {
  for (int i = 0; i < kNumTestArgs; i++) {
--- a/util/arena.cc
+++ b/util/arena.cc
@@ -40,7 +40,7 @@ char* Arena::AllocateFallback(size_t bytes) {
 }

 char* Arena::AllocateAligned(size_t bytes) {
-  const int align = sizeof(void*);    // We'll align to pointer size
+  const int align = (sizeof(void*) > 8) ? sizeof(void*) : 8;
  assert((align & (align-1)) == 0);   // Pointer size should be a power of 2
  size_t current_mod = reinterpret_cast<uintptr_t>(alloc_ptr_) & (align-1);
  size_t slop = (current_mod == 0 ? 0 : align - current_mod);
--- a/util/arena.h
+++ b/util/arena.h
@@ -5,9 +5,9 @@
 #ifndef STORAGE_LEVELDB_UTIL_ARENA_H_
 #define STORAGE_LEVELDB_UTIL_ARENA_H_

-#include <cstddef>
 #include <vector>
 #include <assert.h>
+#include <stddef.h>
 #include <stdint.h>

 namespace leveldb {
--- a/util/arena_test.cc
+++ b/util/arena_test.cc
@@ -40,7 +40,7 @@ TEST(ArenaTest, Simple) {
      r = arena.Allocate(s);
    }

-    for (int b = 0; b < s; b++) {
+    for (size_t 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 (int i = 0; i < allocated.size(); i++) {
+  for (size_t i = 0; i < allocated.size(); i++) {
    size_t num_bytes = allocated[i].first;
    const char* p = allocated[i].second;
-    for (int b = 0; b < num_bytes; b++) {
+    for (size_t b = 0; b < num_bytes; b++) {
      // Check the "i"th allocation for the known bit pattern
      ASSERT_EQ(int(p[b]) & 0xff, i % 256);
    }
--- a/util/bloom_test.cc
+++ b/util/bloom_test.cc
@@ -126,7 +126,8 @@ TEST(BloomTest, VaryingLengths) {
    }
    Build();

-    ASSERT_LE(FilterSize(), (length * 10 / 8) + 40) << length;
+    ASSERT_LE(FilterSize(), static_cast<size_t>((length * 10 / 8) + 40))
+        << length;

    // All added keys must match
    for (int i = 0; i < length; i++) {
--- a/util/cache.cc
+++ b/util/cache.cc
@@ -116,7 +116,6 @@ class HandleTable {
      LRUHandle* h = list_[i];
      while (h != NULL) {
        LRUHandle* next = h->next_hash;
-        Slice key = h->key();
        uint32_t hash = h->hash;
        LRUHandle** ptr = &new_list[hash & (new_length - 1)];
        h->next_hash = *ptr;
@@ -160,7 +159,6 @@ class LRUCache {
  // mutex_ protects the following state.
  port::Mutex mutex_;
  size_t usage_;
-  uint64_t last_id_;

  // Dummy head of LRU list.
  // lru.prev is newest entry, lru.next is oldest entry.
@@ -170,8 +168,7 @@ class LRUCache {
 };

 LRUCache::LRUCache()
-    : usage_(0),
-      last_id_(0) {
+    : usage_(0) {
  // Make empty circular linked list
  lru_.next = &lru_;
  lru_.prev = &lru_;
--- a/util/coding_test.cc
+++ b/util/coding_test.cc
@@ -109,16 +109,16 @@ TEST(Coding, Varint64) {
    values.push_back(power);
    values.push_back(power-1);
    values.push_back(power+1);
-  };
+  }

  std::string s;
-  for (int i = 0; i < values.size(); i++) {
+  for (size_t i = 0; i < values.size(); i++) {
    PutVarint64(&s, values[i]);
  }

  const char* p = s.data();
  const char* limit = p + s.size();
-  for (int i = 0; i < values.size(); i++) {
+  for (size_t 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 (int len = 0; len < s.size() - 1; len++) {
+  for (size_t 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 (int len = 0; len < s.size() - 1; len++) {
+  for (size_t 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);
--- a/util/env_posix.cc
+++ b/util/env_posix.cc
@@ -175,172 +175,83 @@ class PosixMmapReadableFile: public RandomAccessFile {
  }
 };

-// 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 {
+class PosixWritableFile : public WritableFile {
 private:
  std::string filename_;
-  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<char*>(ptr);
-    limit_ = base_ + map_size_;
-    dst_ = base_;
-    last_sync_ = base_;
-    return true;
-  }
+  FILE* file_;

 public:
-  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(const std::string& fname, FILE* f)
+      : filename_(fname), file_(f) { }

-
-  ~PosixMmapFile() {
-    if (fd_ >= 0) {
-      PosixMmapFile::Close();
+  ~PosixWritableFile() {
+    if (file_ != NULL) {
+      // Ignoring any potential errors
+      fclose(file_);
    }
  }

  virtual Status Append(const Slice& data) {
-    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;
+    size_t r = fwrite_unlocked(data.data(), 1, data.size(), file_);
+    if (r != data.size()) {
+      return IOError(filename_, errno);
    }
    return Status::OK();
  }

  virtual Status Close() {
-    Status s;
-    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);
-      }
+    Status result;
+    if (fclose(file_) != 0) {
+      result = IOError(filename_, errno);
    }
-
-    if (close(fd_) < 0) {
-      if (s.ok()) {
-        s = IOError(filename_, errno);
-      }
-    }
-
-    fd_ = -1;
-    base_ = NULL;
-    limit_ = NULL;
-    return s;
+    file_ = NULL;
+    return result;
  }

  virtual Status Flush() {
+    if (fflush_unlocked(file_) != 0) {
+      return IOError(filename_, errno);
+    }
    return Status::OK();
  }

-  virtual Status Sync() {
+  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 (pending_sync_) {
-      // Some unmapped data was not synced
-      pending_sync_ = false;
-      if (fdatasync(fd_) < 0) {
-        s = IOError(filename_, errno);
+    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);
      }
    }
+    return s;
+  }

-    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);
-      }
+  virtual Status Sync() {
+    // Ensure new files referred to by the manifest are in the filesystem.
+    Status s = SyncDirIfManifest();
+    if (!s.ok()) {
+      return s;
+    }
+    if (fflush_unlocked(file_) != 0 ||
+        fdatasync(fileno(file_)) != 0) {
+      s = Status::IOError(filename_, strerror(errno));
    }
-
    return s;
  }
 };
@@ -385,7 +296,7 @@ class PosixEnv : public Env {
  PosixEnv();
  virtual ~PosixEnv() {
    fprintf(stderr, "Destroying Env::Default()\n");
-    exit(1);
+    abort();
  }

  virtual Status NewSequentialFile(const std::string& fname,
@@ -431,12 +342,12 @@ class PosixEnv : public Env {
  virtual Status NewWritableFile(const std::string& fname,
                                 WritableFile** result) {
    Status s;
-    const int fd = open(fname.c_str(), O_CREAT | O_RDWR | O_TRUNC, 0644);
-    if (fd < 0) {
+    FILE* f = fopen(fname.c_str(), "w");
+    if (f == NULL) {
      *result = NULL;
      s = IOError(fname, errno);
    } else {
-      *result = new PosixMmapFile(fname, fd, page_size_);
+      *result = new PosixWritableFile(fname, f);
    }
    return s;
  }
@@ -466,7 +377,7 @@ class PosixEnv : public Env {
      result = IOError(fname, errno);
    }
    return result;
-  };
+  }

  virtual Status CreateDir(const std::string& name) {
    Status result;
@@ -474,7 +385,7 @@ class PosixEnv : public Env {
      result = IOError(name, errno);
    }
    return result;
-  };
+  }

  virtual Status DeleteDir(const std::string& name) {
    Status result;
@@ -482,7 +393,7 @@ class PosixEnv : public Env {
      result = IOError(name, errno);
    }
    return result;
-  };
+  }

  virtual Status GetFileSize(const std::string& fname, uint64_t* size) {
    Status s;
@@ -588,7 +499,7 @@ class PosixEnv : public Env {
  void PthreadCall(const char* label, int result) {
    if (result != 0) {
      fprintf(stderr, "pthread %s: %s\n", label, strerror(result));
-      exit(1);
+      abort();
    }
  }

@@ -599,7 +510,6 @@ class PosixEnv : public Env {
    return NULL;
  }

-  size_t page_size_;
  pthread_mutex_t mu_;
  pthread_cond_t bgsignal_;
  pthread_t bgthread_;
@@ -614,8 +524,7 @@ class PosixEnv : public Env {
  MmapLimiter mmap_limit_;
 };

-PosixEnv::PosixEnv() : page_size_(getpagesize()),
-                       started_bgthread_(false) {
+PosixEnv::PosixEnv() : started_bgthread_(false) {
  PthreadCall("mutex_init", pthread_mutex_init(&mu_, NULL));
  PthreadCall("cvar_init", pthread_cond_init(&bgsignal_, NULL));
 }
--- a/util/hash.cc
+++ b/util/hash.cc
@@ -6,6 +6,13 @@
 #include "util/coding.h"
 #include "util/hash.h"

+// The FALLTHROUGH_INTENDED macro can be used to annotate implicit fall-through
+// between switch labels. The real definition should be provided externally.
+// This one is a fallback version for unsupported compilers.
+#ifndef FALLTHROUGH_INTENDED
+#define FALLTHROUGH_INTENDED do { } while (0)
+#endif
+
 namespace leveldb {

 uint32_t Hash(const char* data, size_t n, uint32_t seed) {
@@ -28,10 +35,10 @@ uint32_t Hash(const char* data, size_t n, uint32_t seed) {
  switch (limit - data) {
    case 3:
      h += data[2] << 16;
-      // fall through
+      FALLTHROUGH_INTENDED;
    case 2:
      h += data[1] << 8;
-      // fall through
+      FALLTHROUGH_INTENDED;
    case 1:
      h += data[0];
      h *= m;
--- a/util/random.h
+++ b/util/random.h
@@ -16,7 +16,12 @@ class Random {
 private:
  uint32_t seed_;
 public:
-  explicit Random(uint32_t s) : seed_(s & 0x7fffffffu) { }
+  explicit Random(uint32_t s) : seed_(s & 0x7fffffffu) {
+    // Avoid bad seeds.
+    if (seed_ == 0 || seed_ == 2147483647L) {
+      seed_ = 1;
+    }
+  }
  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
--- a/util/testharness.cc
+++ b/util/testharness.cc
@@ -38,7 +38,7 @@ int RunAllTests() {

  int num = 0;
  if (tests != NULL) {
-    for (int i = 0; i < tests->size(); i++) {
+    for (size_t i = 0; i < tests->size(); i++) {
      const Test& t = (*tests)[i];
      if (matcher != NULL) {
        std::string name = t.base;
--- a/util/testutil.cc
+++ b/util/testutil.cc
@@ -32,7 +32,7 @@ std::string RandomKey(Random* rnd, int len) {


 extern Slice CompressibleString(Random* rnd, double compressed_fraction,
-                                int len, std::string* dst) {
+                                size_t len, std::string* dst) {
  int raw = static_cast<int>(len * compressed_fraction);
  if (raw < 1) raw = 1;
  std::string raw_data;
--- a/util/testutil.h
+++ b/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,
-                                int len, std::string* dst);
+                                size_t len, std::string* dst);

 // A wrapper that allows injection of errors.
 class ErrorEnv : public EnvWrapper {
Author	SHA1	Message	Date
David Grogan	269fc6ca94	Release LevelDB 1.16 - Make Log::Reader not report a corruption when the last record in a log file is truncated. - Fix issue 224: variable created but not utilized. - Remove comment that referenced a removed feature.	2014-02-10 11:36:06 -08:00
David Grogan	0cfb990d58	Release LevelDB 1.15 - switched from mmap based writing to simpler stdio based writing. Has a minor impact (0.5 microseconds) on microbenchmarks for asynchronous writes. Synchronous writes speed up from 30ms to 10ms on linux/ext4. Should be much more reliable on diverse platforms. - compaction errors now immediately put the database into a read-only mode (until it is re-opened). As a downside, a disk going out of space and then space being created will require a re-open to recover from, whereas previously that would happen automatically. On the plus side, many corruption possibilities go away. - force the DB to enter an error-state so that all future writes fail when a synchronous log write succeeds but the sync fails. - repair now regenerates sstables that exhibit problems - fix issue 218 - Use native memory barriers on OSX - fix issue 212 - QNX build is broken - fix build on iOS with xcode 5 - make tests compile and pass on windows	2013-12-10 10:36:31 -08:00
David Grogan	0b9a89f40e	Release LevelDB 1.14 Fix issues 200, 201 Also, * Fix link to bigtable paper in docs. * New sstables will have the file extension .ldb. .sst files will continue to be recognized. * When building for iOS, use xcrun to execute the compiler. This may affect issue 177.	2013-09-19 13:49:19 -07:00
David Grogan	748539c183	LevelDB 1.13 Fix issues 77, 87, 182, 190. Additionally, fix the bug described in https://groups.google.com/d/msg/leveldb/yL6h1mAOc20/vLU64RylIdMJ where a large contiguous keyspace of deleted data was not getting compacted. Also fix a bug where options.max_open_files was not getting clamped properly.	2013-08-21 11:12:47 -07:00
David Grogan	5bd76dc10d	Release leveldb 1.12 Non-functional changes only: 1) Add Kevin Regan to the AUTHORS file 2) Add standard license header to issues/issue178_test.cc	2013-06-18 11:00:13 -07:00
David Grogan	7b094f12e4	Release leveldb 1.11 Fixes issues 161 174 178 As well as the issue reported by edouarda14@gmail.com about MissingSSTFile unit test failing on windows.	2013-06-13 16:14:06 -07:00
David Grogan	28dad918f2	Release leveldb 1.10 Fixes issues 147 - thanks feniksgordonfreeman 153 156 166 Additionally, * Remove calls to exit(1). * Fix unused-variable warnings from clang. * Fix possible overflow error related to num_restart value >= (2^32/4). * Add leveldbutil to .gitignore. * Add better log messages when Write is stalled on a compaction.	2013-05-14 17:03:07 -07:00
David Grogan	514c943a8e	Make DB::Open fail if sst files are missing. Also, cleanup for Clang's -Wimplicit-fallthrough warning.	2013-02-06 18:03:32 -08:00