Enable LevelDb to create a new log file if current log file is too large.

Summary: Enable LevelDb to create a new log file if current log file is too large.

Test Plan:
Write a script and manually check the generated info LOG.

Task ID: 1803577

Blame Rev:

Reviewers: dhruba, heyongqiang

Reviewed By: heyongqiang

CC: zshao

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

db/db_impl.cc

@@ -34,11 +34,40 @@
 #include "util/logging.h"
 #include "util/mutexlock.h"
 #include "util/build_version.h"
+#include "util/auto_split_logger.h"
 
 namespace leveldb {
 
 void dumpLeveldbBuildVersion(Logger * log);
 
+static Status NewLogger(const std::string& dbname,
+                        const std::string& db_log_dir,
+                        Env* env,
+                        size_t max_log_file_size,
+                        Logger** logger) {
+  std::string db_absolute_path;
+  env->GetAbsolutePath(dbname, &db_absolute_path);
+
+  if (max_log_file_size > 0) { // need to auto split the log file?
+    AutoSplitLogger<Logger>* auto_split_logger =
+      new AutoSplitLogger<Logger>(env, dbname, db_log_dir, max_log_file_size);
+    Status s = auto_split_logger->GetStatus();
+    if (!s.ok()) {
+      delete auto_split_logger;
+    } else {
+      *logger = auto_split_logger;
+    }
+    return s;
+  } else {
+    // Open a log file in the same directory as the db
+    env->CreateDir(dbname);  // In case it does not exist
+    std::string fname = InfoLogFileName(dbname, db_absolute_path, db_log_dir);
+    env->RenameFile(fname, OldInfoLogFileName(dbname, env->NowMicros(),
+                                              db_absolute_path, db_log_dir));
+    return env->NewLogger(fname, logger);
+  }
+}
+
 // Information kept for every waiting writer
 struct DBImpl::Writer {
   Status status;
@@ -116,16 +145,9 @@ Options SanitizeOptions(const std::string& dbname,
   ClipToRange(&result.max_open_files,            20,     50000);
   ClipToRange(&result.write_buffer_size,         64<<10, 1<<30);
   ClipToRange(&result.block_size,                1<<10,  4<<20);
-  std::string db_absolute_path;
-  src.env->GetAbsolutePath(dbname, &db_absolute_path);
   if (result.info_log == NULL) {
-    // Open a log file in the same directory as the db
+    Status s = NewLogger(dbname, result.db_log_dir, src.env,
-    src.env->CreateDir(dbname);  // In case it does not exist
+                         result.max_log_file_size, &result.info_log);
-    src.env->RenameFile(InfoLogFileName(dbname, db_absolute_path,
-        result.db_log_dir), OldInfoLogFileName(dbname,src.env->NowMicros(),
-            db_absolute_path, result.db_log_dir));
-    Status s = src.env->NewLogger(InfoLogFileName(dbname, db_absolute_path,
-        result.db_log_dir), &result.info_log);
     if (!s.ok()) {
       // No place suitable for logging
       result.info_log = NULL;

include/leveldb/env.h

@@ -223,9 +223,9 @@ class WritableFile {
   virtual Status Flush() = 0;
   virtual Status Sync() = 0; // sync data
 
-  /* 
+  /*
    * Sync data and/or metadata as well.
-   * By default, sync only metadata. 
+   * By default, sync only metadata.
    * Override this method for environments where we need to sync
    * metadata as well.
    */
@@ -249,11 +249,15 @@ class WritableFile {
 // An interface for writing log messages.
 class Logger {
  public:
+  enum { DO_NOT_SUPPORT_GET_LOG_FILE_SIZE = -1 };
   Logger() { }
   virtual ~Logger();
 
   // Write an entry to the log file with the specified format.
   virtual void Logv(const char* format, va_list ap) = 0;
+  virtual size_t GetLogFileSize() const {
+    return DO_NOT_SUPPORT_GET_LOG_FILE_SIZE;
+  }
 
  private:
   // No copying allowed

include/leveldb/options.h

@@ -246,6 +246,13 @@ struct Options {
   // every compaction run.
   uint64_t delete_obsolete_files_period_micros;
 
+  // Specify the maximal size of the info log file. If the log file
+  // is larger than `max_log_file_size`, a new info log file will
+  // be created.
+  // If max_log_file_size == 0, all logs will be written to one
+  // log file.
+  size_t max_log_file_size;
+
   // Create an Options object with default values for all fields.
   Options();
 

util/options.cc

@@ -42,6 +42,7 @@ Options::Options()
       db_stats_log_interval(1800),
       db_log_dir(""),
       disable_seek_compaction(false),
+      max_log_file_size(0),
       delete_obsolete_files_period_micros(0) {
 }
 
@@ -67,7 +68,8 @@ Options::Dump(
     Log(log,"            Options.num_levels: %d", num_levels);
     Log(log,"       Options.disableDataSync: %d", disableDataSync);
     Log(log,"             Options.use_fsync: %d", use_fsync);
-    Log(log," Options.db_stats_log_interval: %d", 
+    Log(log,"     Options.max_log_file_size: %d", max_log_file_size);
+    Log(log," Options.db_stats_log_interval: %d",
         db_stats_log_interval);
     Log(log,"     Options.level0_file_num_compaction_trigger: %d",
         level0_file_num_compaction_trigger);

util/posix_logger.h

@@ -20,8 +20,11 @@ class PosixLogger : public Logger {
  private:
   FILE* file_;
   uint64_t (*gettid_)();  // Return the thread id for the current thread
+
+  size_t log_size_;
  public:
-  PosixLogger(FILE* f, uint64_t (*gettid)()) : file_(f), gettid_(gettid) { }
+  PosixLogger(FILE* f, uint64_t (*gettid)()) :
+    file_(f), gettid_(gettid), log_size_(0) { }
   virtual ~PosixLogger() {
     fclose(file_);
   }
@@ -85,12 +88,17 @@ class PosixLogger : public Logger {
       assert(p <= limit);
       fwrite(base, 1, p - base, file_);
       fflush(file_);
+      log_size_ += (p - base);
+
       if (base != buffer) {
         delete[] base;
       }
       break;
     }
   }
+  size_t GetLogFileSize() const {
+    return log_size_;
+  }
 };
 
 }  // namespace leveldb