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.

.gitignore

@@ -6,3 +6,4 @@ build_config.mk
 *.so.*
 *_test
 db_bench
+leveldbutil

Makefile

@@ -12,7 +12,7 @@ OPT ?= -O2 -DNDEBUG       # (A) Production use (optimized mode)
 #-----------------------------------------------
 
 # detect what platform we're building on
-$(shell CC=$(CC) CXX=$(CXX) TARGET_OS=$(TARGET_OS) \
+$(shell CC="$(CC)" CXX="$(CXX)" TARGET_OS="$(TARGET_OS)" \
     ./build_detect_platform build_config.mk ./)
 # this file is generated by the previous line to set build flags and sources
 include build_config.mk
@@ -50,7 +50,7 @@ TESTS = \
 	version_set_test \
 	write_batch_test
 
-PROGRAMS = db_bench $(TESTS)
+PROGRAMS = db_bench leveldbutil $(TESTS)
 BENCHMARKS = db_bench_sqlite3 db_bench_tree_db
 
 LIBRARY = libleveldb.a
@@ -69,7 +69,7 @@ SHARED = $(SHARED1)
 else
 # Update db.h if you change these.
 SHARED_MAJOR = 1
-SHARED_MINOR = 7
+SHARED_MINOR = 10
 SHARED1 = libleveldb.$(PLATFORM_SHARED_EXT)
 SHARED2 = $(SHARED1).$(SHARED_MAJOR)
 SHARED3 = $(SHARED1).$(SHARED_MAJOR).$(SHARED_MINOR)
@@ -107,6 +107,9 @@ db_bench_sqlite3: doc/bench/db_bench_sqlite3.o $(LIBOBJECTS) $(TESTUTIL)
 db_bench_tree_db: doc/bench/db_bench_tree_db.o $(LIBOBJECTS) $(TESTUTIL)
 	$(CXX) $(LDFLAGS) doc/bench/db_bench_tree_db.o $(LIBOBJECTS) $(TESTUTIL) -o $@ -lkyotocabinet $(LIBS)
 
+leveldbutil: db/leveldb_main.o $(LIBOBJECTS)
+	$(CXX) $(LDFLAGS) db/leveldb_main.o $(LIBOBJECTS) -o $@ $(LIBS)
+
 arena_test: util/arena_test.o $(LIBOBJECTS) $(TESTHARNESS)
 	$(CXX) $(LDFLAGS) util/arena_test.o $(LIBOBJECTS) $(TESTHARNESS) -o $@ $(LIBS)
 

build_detect_platform

@@ -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`
@@ -141,7 +145,8 @@ DIRS="$PREFIX/db $PREFIX/util $PREFIX/table"
 set -f # temporarily disable globbing so that our patterns aren't expanded
 PRUNE_TEST="-name *test*.cc -prune"
 PRUNE_BENCH="-name *_bench.cc -prune"
-PORTABLE_FILES=`find $DIRS $PRUNE_TEST -o $PRUNE_BENCH -o -name '*.cc' -print | sort | sed "s,^$PREFIX/,," | tr "\n" " "`
+PRUNE_TOOL="-name leveldb_main.cc -prune"
+PORTABLE_FILES=`find $DIRS $PRUNE_TEST -o $PRUNE_BENCH -o $PRUNE_TOOL -o -name '*.cc' -print | sort | sed "s,^$PREFIX/,," | tr "\n" " "`
 
 set +f # re-enable globbing
 
@@ -154,8 +159,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
@@ -168,7 +175,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
@@ -178,12 +185,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"

db/db_impl.cc

@@ -310,16 +310,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());
@@ -1268,10 +1276,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

db/db_test.cc

@@ -59,6 +59,12 @@ class SpecialEnv : public EnvWrapper {
   // Simulate non-writable file system while this pointer is non-NULL
   port::AtomicPointer non_writable_;
 
+  // Force sync of manifest files to fail while this pointer is non-NULL
+  port::AtomicPointer manifest_sync_error_;
+
+  // Force write to manifest files to fail while this pointer is non-NULL
+  port::AtomicPointer manifest_write_error_;
+
   bool count_random_reads_;
   AtomicCounter random_read_counter_;
 
@@ -69,6 +75,8 @@ class SpecialEnv : public EnvWrapper {
     no_space_.Release_Store(NULL);
     non_writable_.Release_Store(NULL);
     count_random_reads_ = false;
+    manifest_sync_error_.Release_Store(NULL);
+    manifest_write_error_.Release_Store(NULL);
   }
 
   Status NewWritableFile(const std::string& f, WritableFile** r) {
@@ -100,6 +108,30 @@ class SpecialEnv : public EnvWrapper {
         return base_->Sync();
       }
     };
+    class ManifestFile : public WritableFile {
+     private:
+      SpecialEnv* env_;
+      WritableFile* base_;
+     public:
+      ManifestFile(SpecialEnv* env, WritableFile* b) : env_(env), base_(b) { }
+      ~ManifestFile() { delete base_; }
+      Status Append(const Slice& data) {
+        if (env_->manifest_write_error_.Acquire_Load() != NULL) {
+          return Status::IOError("simulated writer error");
+        } else {
+          return base_->Append(data);
+        }
+      }
+      Status Close() { return base_->Close(); }
+      Status Flush() { return base_->Flush(); }
+      Status Sync() {
+        if (env_->manifest_sync_error_.Acquire_Load() != NULL) {
+          return Status::IOError("simulated sync error");
+        } else {
+          return base_->Sync();
+        }
+      }
+    };
 
     if (non_writable_.Acquire_Load() != NULL) {
       return Status::IOError("simulated write error");
@@ -109,6 +141,8 @@ class SpecialEnv : public EnvWrapper {
     if (s.ok()) {
       if (strstr(f.c_str(), ".sst") != NULL) {
         *r = new SSTableFile(this, *r);
+      } else if (strstr(f.c_str(), "MANIFEST") != NULL) {
+        *r = new ManifestFile(this, *r);
       }
     }
     return s;
@@ -427,6 +461,20 @@ 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;
+  }
 };
 
 TEST(DBTest, Empty) {
@@ -1492,6 +1540,64 @@ TEST(DBTest, NonWritableFileSystem) {
   env_->non_writable_.Release_Store(NULL);
 }
 
+TEST(DBTest, ManifestWriteError) {
+  // Test for the following problem:
+  // (a) Compaction produces file F
+  // (b) Log record containing F is written to MANIFEST file, but Sync() fails
+  // (c) GC deletes F
+  // (d) After reopening DB, reads fail since deleted F is named in log record
+
+  // We iterate twice.  In the second iteration, everything is the
+  // same except the log record never makes it to the MANIFEST file.
+  for (int iter = 0; iter < 2; iter++) {
+    port::AtomicPointer* error_type = (iter == 0)
+        ? &env_->manifest_sync_error_
+        : &env_->manifest_write_error_;
+
+    // Insert foo=>bar mapping
+    Options options = CurrentOptions();
+    options.env = env_;
+    options.create_if_missing = true;
+    options.error_if_exists = false;
+    DestroyAndReopen(&options);
+    ASSERT_OK(Put("foo", "bar"));
+    ASSERT_EQ("bar", Get("foo"));
+
+    // Memtable compaction (will succeed)
+    dbfull()->TEST_CompactMemTable();
+    ASSERT_EQ("bar", Get("foo"));
+    const int last = config::kMaxMemCompactLevel;
+    ASSERT_EQ(NumTableFilesAtLevel(last), 1);   // foo=>bar is now in last level
+
+    // Merging compaction (will fail)
+    error_type->Release_Store(env_);
+    dbfull()->TEST_CompactRange(last, NULL, NULL);  // Should fail
+    ASSERT_EQ("bar", Get("foo"));
+
+    // Recovery: should not lose data
+    error_type->Release_Store(NULL);
+    Reopen(&options);
+    ASSERT_EQ("bar", Get("foo"));
+  }
+}
+
+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"));
+
+  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, FilesDeletedAfterCompaction) {
   ASSERT_OK(Put("foo", "v2"));
   Compact("a", "z");

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

db/leveldb_main.cc

@@ -0,0 +1,238 @@
+// Copyright (c) 2012 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 <stdio.h>
+#include "db/dbformat.h"
+#include "db/filename.h"
+#include "db/log_reader.h"
+#include "db/version_edit.h"
+#include "db/write_batch_internal.h"
+#include "leveldb/env.h"
+#include "leveldb/iterator.h"
+#include "leveldb/options.h"
+#include "leveldb/status.h"
+#include "leveldb/table.h"
+#include "leveldb/write_batch.h"
+#include "util/logging.h"
+
+namespace leveldb {
+
+namespace {
+
+bool GuessType(const std::string& fname, FileType* type) {
+  size_t pos = fname.rfind('/');
+  std::string basename;
+  if (pos == std::string::npos) {
+    basename = fname;
+  } else {
+    basename = std::string(fname.data() + pos + 1, fname.size() - pos - 1);
+  }
+  uint64_t ignored;
+  return ParseFileName(basename, &ignored, type);
+}
+
+// Notified when log reader encounters corruption.
+class CorruptionReporter : public log::Reader::Reporter {
+ public:
+  virtual void Corruption(size_t bytes, const Status& status) {
+    printf("corruption: %d bytes; %s\n",
+            static_cast<int>(bytes),
+            status.ToString().c_str());
+  }
+};
+
+// Print contents of a log file. (*func)() is called on every record.
+bool PrintLogContents(Env* env, const std::string& fname,
+                      void (*func)(Slice)) {
+  SequentialFile* file;
+  Status s = env->NewSequentialFile(fname, &file);
+  if (!s.ok()) {
+    fprintf(stderr, "%s\n", s.ToString().c_str());
+    return false;
+  }
+  CorruptionReporter reporter;
+  log::Reader reader(file, &reporter, true, 0);
+  Slice record;
+  std::string scratch;
+  while (reader.ReadRecord(&record, &scratch)) {
+    printf("--- offset %llu; ",
+           static_cast<unsigned long long>(reader.LastRecordOffset()));
+    (*func)(record);
+  }
+  delete file;
+  return true;
+}
+
+// Called on every item found in a WriteBatch.
+class WriteBatchItemPrinter : public WriteBatch::Handler {
+ public:
+  uint64_t offset_;
+  uint64_t sequence_;
+
+  virtual void Put(const Slice& key, const Slice& value) {
+    printf("  put '%s' '%s'\n",
+           EscapeString(key).c_str(),
+           EscapeString(value).c_str());
+  }
+  virtual void Delete(const Slice& key) {
+    printf("  del '%s'\n",
+           EscapeString(key).c_str());
+  }
+};
+
+
+// Called on every log record (each one of which is a WriteBatch)
+// found in a kLogFile.
+static void WriteBatchPrinter(Slice record) {
+  if (record.size() < 12) {
+    printf("log record length %d is too small\n",
+           static_cast<int>(record.size()));
+    return;
+  }
+  WriteBatch batch;
+  WriteBatchInternal::SetContents(&batch, record);
+  printf("sequence %llu\n",
+         static_cast<unsigned long long>(WriteBatchInternal::Sequence(&batch)));
+  WriteBatchItemPrinter batch_item_printer;
+  Status s = batch.Iterate(&batch_item_printer);
+  if (!s.ok()) {
+    printf("  error: %s\n", s.ToString().c_str());
+  }
+}
+
+bool DumpLog(Env* env, const std::string& fname) {
+  return PrintLogContents(env, fname, WriteBatchPrinter);
+}
+
+// Called on every log record (each one of which is a WriteBatch)
+// found in a kDescriptorFile.
+static void VersionEditPrinter(Slice record) {
+  VersionEdit edit;
+  Status s = edit.DecodeFrom(record);
+  if (!s.ok()) {
+    printf("%s\n", s.ToString().c_str());
+    return;
+  }
+  printf("%s", edit.DebugString().c_str());
+}
+
+bool DumpDescriptor(Env* env, const std::string& fname) {
+  return PrintLogContents(env, fname, VersionEditPrinter);
+}
+
+bool DumpTable(Env* env, const std::string& fname) {
+  uint64_t file_size;
+  RandomAccessFile* file = NULL;
+  Table* table = NULL;
+  Status s = env->GetFileSize(fname, &file_size);
+  if (s.ok()) {
+    s = env->NewRandomAccessFile(fname, &file);
+  }
+  if (s.ok()) {
+    // We use the default comparator, which may or may not match the
+    // comparator used in this database. However this should not cause
+    // problems since we only use Table operations that do not require
+    // any comparisons.  In particular, we do not call Seek or Prev.
+    s = Table::Open(Options(), file, file_size, &table);
+  }
+  if (!s.ok()) {
+    fprintf(stderr, "%s\n", s.ToString().c_str());
+    delete table;
+    delete file;
+    return false;
+  }
+
+  ReadOptions ro;
+  ro.fill_cache = false;
+  Iterator* iter = table->NewIterator(ro);
+  for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
+    ParsedInternalKey key;
+    if (!ParseInternalKey(iter->key(), &key)) {
+      printf("badkey '%s' => '%s'\n",
+             EscapeString(iter->key()).c_str(),
+             EscapeString(iter->value()).c_str());
+    } else {
+      char kbuf[20];
+      const char* type;
+      if (key.type == kTypeDeletion) {
+        type = "del";
+      } else if (key.type == kTypeValue) {
+        type = "val";
+      } else {
+        snprintf(kbuf, sizeof(kbuf), "%d", static_cast<int>(key.type));
+        type = kbuf;
+      }
+      printf("'%s' @ %8llu : %s => '%s'\n",
+             EscapeString(key.user_key).c_str(),
+             static_cast<unsigned long long>(key.sequence),
+             type,
+             EscapeString(iter->value()).c_str());
+    }
+  }
+  s = iter->status();
+  if (!s.ok()) {
+    printf("iterator error: %s\n", s.ToString().c_str());
+  }
+
+  delete iter;
+  delete table;
+  delete file;
+  return true;
+}
+
+bool DumpFile(Env* env, const std::string& fname) {
+  FileType ftype;
+  if (!GuessType(fname, &ftype)) {
+    fprintf(stderr, "%s: unknown file type\n", fname.c_str());
+    return false;
+  }
+  switch (ftype) {
+    case kLogFile:         return DumpLog(env, fname);
+    case kDescriptorFile:  return DumpDescriptor(env, fname);
+    case kTableFile:       return DumpTable(env, fname);
+
+    default: {
+      fprintf(stderr, "%s: not a dump-able file type\n", fname.c_str());
+      break;
+    }
+  }
+  return false;
+}
+
+bool HandleDumpCommand(Env* env, char** files, int num) {
+  bool ok = true;
+  for (int i = 0; i < num; i++) {
+    ok &= DumpFile(env, files[i]);
+  }
+  return ok;
+}
+
+}
+}  // namespace leveldb
+
+static void Usage() {
+  fprintf(
+      stderr,
+      "Usage: leveldbutil command...\n"
+      "   dump files...         -- dump contents of specified files\n"
+      );
+}
+
+int main(int argc, char** argv) {
+  leveldb::Env* env = leveldb::Env::Default();
+  bool ok = true;
+  if (argc < 2) {
+    Usage();
+    ok = false;
+  } else {
+    std::string command = argv[1];
+    if (command == "dump") {
+      ok = leveldb::HandleDumpCommand(env, argv+2, argc-2);
+    } else {
+      Usage();
+      ok = false;
+    }
+  }
+  return (ok ? 0 : 1);
+}

db/version_set.cc

@@ -786,12 +786,23 @@ Status VersionSet::LogAndApply(VersionEdit* edit, port::Mutex* mu) {
       if (s.ok()) {
         s = descriptor_file_->Sync();
       }
+      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();
+        }
+      }
     }
 
     // If we just created a new descriptor file, install it by writing a
     // 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();
@@ -1025,6 +1036,31 @@ const char* VersionSet::LevelSummary(LevelSummaryStorage* scratch) const {
   return scratch->buffer;
 }
 
+// Return true iff the manifest contains the specified record.
+bool VersionSet::ManifestContains(const std::string& record) const {
+  std::string fname = DescriptorFileName(dbname_, manifest_file_number_);
+  Log(options_->info_log, "ManifestContains: checking %s\n", fname.c_str());
+  SequentialFile* file = NULL;
+  Status s = env_->NewSequentialFile(fname, &file);
+  if (!s.ok()) {
+    Log(options_->info_log, "ManifestContains: %s\n", s.ToString().c_str());
+    return false;
+  }
+  log::Reader reader(file, NULL, true/*checksum*/, 0);
+  Slice r;
+  std::string scratch;
+  bool result = false;
+  while (reader.ReadRecord(&r, &scratch)) {
+    if (r == Slice(record)) {
+      result = true;
+      break;
+    }
+  }
+  delete file;
+  Log(options_->info_log, "ManifestContains: result = %d\n", result ? 1 : 0);
+  return result;
+}
+
 uint64_t VersionSet::ApproximateOffsetOf(Version* v, const InternalKey& ikey) {
   uint64_t result = 0;
   for (int level = 0; level < config::kNumLevels; level++) {

db/version_set.h

@@ -277,6 +277,8 @@ class VersionSet {
 
   void AppendVersion(Version* v);
 
+  bool ManifestContains(const std::string& record) const;
+
   Env* const env_;
   const std::string dbname_;
   const Options* const options_;

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 = 7;
+static const int kMinorVersion = 10;
 
 struct Options;
 struct ReadOptions;

port/port_posix.h

@@ -26,11 +26,17 @@
   #include <sys/endian.h>
   #define PLATFORM_IS_LITTLE_ENDIAN (_BYTE_ORDER == _LITTLE_ENDIAN)
 #elif defined(OS_OPENBSD) || defined(OS_NETBSD) ||\
-      defined(OS_DRAGONFLYBSD) || defined(OS_ANDROID)
+      defined(OS_DRAGONFLYBSD)
   #include <sys/types.h>
   #include <sys/endian.h>
 #elif defined(OS_HPUX)
   #define PLATFORM_IS_LITTLE_ENDIAN false
+#elif defined(OS_ANDROID)
+  // Due to a bug in the NDK x86 <sys/endian.h> definition,
+  // _BYTE_ORDER must be used instead of __BYTE_ORDER on Android.
+  // See http://code.google.com/p/android/issues/detail?id=39824
+  #include <endian.h>
+  #define PLATFORM_IS_LITTLE_ENDIAN  (_BYTE_ORDER == _LITTLE_ENDIAN)
 #else
   #include <endian.h>
 #endif

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();

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()) {

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

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_;

util/env_posix.cc

@@ -385,7 +385,7 @@ class PosixEnv : public Env {
   PosixEnv();
   virtual ~PosixEnv() {
     fprintf(stderr, "Destroying Env::Default()\n");
-    exit(1);
+    abort();
   }
 
   virtual Status NewSequentialFile(const std::string& fname,
@@ -588,7 +588,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();
     }
   }
 

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;