rollover manifest file.

Summary:
Check in LogAndApply if the file size is more than the limit set in
Options.
Things to consider : will this be expensive?

Test Plan: make all check. Inputs on a new unit test?

Reviewers: dhruba

Reviewed By: dhruba

CC: leveldb

Differential Revision: https://reviews.facebook.net/D7701

db/db_impl.cc

@@ -290,6 +290,9 @@ void DBImpl::TEST_Destroy_DBImpl() {
   }
 }
 
+uint64_t DBImpl::TEST_Current_Manifest_FileNo() {
+  return versions_->ManifestFileNumber();
+}
 
 Status DBImpl::NewDB() {
   VersionEdit new_db(NumberLevels());

db/db_impl.h

@@ -85,6 +85,8 @@ class DBImpl : public DB {
   // Simulate a db crash, no elegant closing of database.
   void TEST_Destroy_DBImpl();
 
+  // Return the current manifest file no.
+  uint64_t TEST_Current_Manifest_FileNo();
  protected:
   Env* const env_;
   const std::string dbname_;
@@ -97,7 +99,7 @@ class DBImpl : public DB {
   }
   MemTable* GetMemTable() {
     return mem_;
-  } 
+  }
 
  private:
   friend class DB;

db/db_test.cc

@@ -210,6 +210,7 @@ class DBTest {
     kUncompressed,
     kNumLevel_3,
     kDBLogDir,
+    kManifestFileSize,
     kEnd
   };
   int option_config_;
@@ -265,6 +266,8 @@ class DBTest {
       case kDBLogDir:
         options.db_log_dir = test::TmpDir();
         break;
+      case kManifestFileSize:
+        options.max_manifest_file_size = 50; // 50 bytes
       default:
         break;
     }
@@ -1025,6 +1028,31 @@ TEST(DBTest, MinorCompactionsHappen) {
   }
 }
 
+TEST(DBTest, ManifestRollOver) {
+  Options options = CurrentOptions();
+  options.max_manifest_file_size = 10 ;  // 10 bytes
+  Reopen(&options);
+  {
+    ASSERT_OK(Put("manifest_key1", std::string(1000, '1')));
+    ASSERT_OK(Put("manifest_key2", std::string(1000, '2')));
+    ASSERT_OK(Put("manifest_key3", std::string(1000, '3')));
+    uint64_t manifest_before_fulsh =
+      dbfull()->TEST_Current_Manifest_FileNo();
+    dbfull()->Flush(FlushOptions()); // This should trigger LogAndApply.
+    uint64_t manifest_after_flush =
+      dbfull()->TEST_Current_Manifest_FileNo();
+    ASSERT_GT(manifest_after_flush, manifest_before_fulsh);
+    Reopen(&options);
+    ASSERT_GT(dbfull()->TEST_Current_Manifest_FileNo(),
+              manifest_after_flush);
+    // check if a new manifest file got inserted or not.
+    ASSERT_EQ(std::string(1000, '1'), Get("manifest_key1"));
+    ASSERT_EQ(std::string(1000, '2'), Get("manifest_key2"));
+    ASSERT_EQ(std::string(1000, '3'), Get("manifest_key3"));
+  }
+}
+
+
 TEST(DBTest, RecoverWithLargeLog) {
   {
     Options options = CurrentOptions();

db/version_set.cc

@@ -901,7 +901,8 @@ VersionSet::VersionSet(const std::string& dbname,
       dummy_versions_(this),
       current_(NULL),
       compactions_in_progress_(options_->num_levels),
-      current_version_number_(0) {
+      current_version_number_(0),
+      last_observed_manifest_size_(0) {
   compact_pointer_ = new std::string[options_->num_levels];
   Init(options_->num_levels);
   AppendVersion(new Version(this, current_version_number_++));
@@ -986,6 +987,13 @@ Status VersionSet::LogAndApply(VersionEdit* edit, port::Mutex* mu,
   std::string new_manifest_file;
   uint64_t new_manifest_file_size = 0;
   Status s;
+
+  //  No need to perform this check if a new Manifest is being created anyways.
+  if (last_observed_manifest_size_ > options_->max_manifest_file_size) {
+    new_descriptor_log = true;
+    manifest_file_number_ = NewFileNumber(); // Change manifest file no.
+  }
+
   if (descriptor_log_ == NULL || new_descriptor_log) {
     // No reason to unlock *mu here since we only hit this path in the
     // first call to LogAndApply (when opening the database).
@@ -1047,6 +1055,9 @@ Status VersionSet::LogAndApply(VersionEdit* edit, port::Mutex* mu,
     new_manifest_file_size = descriptor_file_->GetFileSize();
 
     mu->Lock();
+    // cache the manifest_file_size so that it can be used to rollover in the
+    // next call to LogAndApply
+    last_observed_manifest_size_ = new_manifest_file_size;
   }
 
   // Install the new version

db/version_set.h

@@ -443,6 +443,10 @@ class VersionSet {
   // Queue of writers to the manifest file
   std::deque<ManifestWriter*> manifest_writers_;
 
+  // Store the manifest file size when it is checked.
+  // Save us the cost of checking file size twice in LogAndApply
+  uint64_t last_observed_manifest_size_;
+
   // No copying allowed
   VersionSet(const VersionSet&);
   void operator=(const VersionSet&);

include/leveldb/options.h

@@ -308,6 +308,11 @@ struct Options {
   // exceeds rate_limit. This is ignored when <= 1.0.
   double rate_limit;
 
+  // manifest file is rolled over on reaching this limit.
+  // The older manifest file be deleted.
+  // The default value is MAX_INT so that roll-over does not take place.
+  uint64_t max_manifest_file_size;
+
   // Disable block cache. If this is set to false,
   // then no block cache should be used, and the block_cache should
   // point to a NULL object.

tools/db_stress.cc

@@ -46,7 +46,7 @@ static bool FLAGS_verbose = false;
 // (initialized to default value by "main")
 static int FLAGS_write_buffer_size = 0;
 
-// The number of in-memory memtables. 
+// The number of in-memory memtables.
 // Each memtable is of size FLAGS_write_buffer_size.
 // This is initialized to default value of 2 in "main" function.
 static int FLAGS_max_write_buffer_number = 0;
@@ -491,7 +491,7 @@ class StressTest {
     double now = FLAGS_env->NowMicros();
     fprintf(stdout, "%s Verification successful\n",
             FLAGS_env->TimeToString((uint64_t) now/1000000).c_str());
-  
+
     PrintStatistics();
   }
 
@@ -725,6 +725,7 @@ class StressTest {
     options.disable_seek_compaction = FLAGS_disable_seek_compaction;
     options.delete_obsolete_files_period_micros =
       FLAGS_delete_obsolete_files_period_micros;
+    options.max_manifest_file_size = 1024;
     Status s = DB::Open(options, FLAGS_db, &db_);
     if (!s.ok()) {
       fprintf(stderr, "open error: %s\n", s.ToString().c_str());

util/options.cc

@@ -4,6 +4,8 @@
 
 #include "leveldb/options.h"
 
+#include <limits>
+
 #include "leveldb/comparator.h"
 #include "leveldb/env.h"
 #include "leveldb/filter_policy.h"
@@ -49,6 +51,7 @@ Options::Options()
       max_background_compactions(1),
       max_log_file_size(0),
       rate_limit(0.0),
+      max_manifest_file_size(std::numeric_limits<uint64_t>::max()),
       no_block_cache(false),
       table_cache_numshardbits(4),
       compaction_filter_args(NULL),
@@ -91,6 +94,8 @@ Options::Dump(
     Log(log,"       Options.disableDataSync: %d", disableDataSync);
     Log(log,"             Options.use_fsync: %d", use_fsync);
     Log(log,"     Options.max_log_file_size: %ld", max_log_file_size);
+    Log(log,"Options.max_manifest_file_size: %ld",
+      max_manifest_file_size);
     Log(log," Options.db_stats_log_interval: %d",
         db_stats_log_interval);
     Log(log,"           Options.compression_opts.window_bits: %d",