Fix race conditions in auto-rolling logger

Summary:
For GetLogFileSize() and Flush(), they previously did not follow the
synchronization pattern for accessing logger_. This meant ResetLogger() could
cause logger_ destruction while the unsynchronized functions were accessing it,
causing a segfault.

Also made the mutex instance variable mutable so we can preserve
GetLogFileSize()'s const-ness.

Test Plan:
new test case, it's quite ugly because both threads need to access
one of the functions with SyncPoints (PosixLogger::Flush()), and also special
handling is needed to prevent the mutex and sync points from conflicting.

Reviewers: kradhakrishnan, IslamAbdelRahman, sdong

Reviewed By: sdong

Subscribers: dhruba, leveldb

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

db/auto_roll_logger.cc

@@ -12,7 +12,9 @@ namespace rocksdb {
 
 // -- AutoRollLogger
 Status AutoRollLogger::ResetLogger() {
+  TEST_SYNC_POINT("AutoRollLogger::ResetLogger:BeforeNewLogger");
   status_ = env_->NewLogger(log_fname_, &logger_);
+  TEST_SYNC_POINT("AutoRollLogger::ResetLogger:AfterNewLogger");
 
   if (!status_.ok()) {
     return status_;
@@ -75,6 +77,8 @@ void AutoRollLogger::Logv(const char* format, va_list ap) {
     if ((kLogFileTimeToRoll > 0 && LogExpired()) ||
         (kMaxLogFileSize > 0 && logger_->GetLogFileSize() >= kMaxLogFileSize)) {
       RollLogFile();
+      TEST_SYNC_POINT_CALLBACK("AutoRollLogger::Logv:BeforeResetLogger",
+                               logger_.get());
       Status s = ResetLogger();
       if (!s.ok()) {
         // can't really log the error if creating a new LOG file failed

db/auto_roll_logger.h

@@ -13,6 +13,7 @@
 #include "db/filename.h"
 #include "port/port.h"
 #include "port/util_logger.h"
+#include "util/sync_point.h"
 
 namespace rocksdb {
 
@@ -53,11 +54,26 @@ class AutoRollLogger : public Logger {
     return status_;
   }
 
-  size_t GetLogFileSize() const override { return logger_->GetLogFileSize(); }
+  size_t GetLogFileSize() const override {
+    std::shared_ptr<Logger> logger;
+    {
+      MutexLock l(&mutex_);
+      // pin down the current logger_ instance before releasing the mutex.
+      logger = logger_;
+    }
+    return logger->GetLogFileSize();
+  }
 
   void Flush() override {
-    if (logger_) {
+    std::shared_ptr<Logger> logger;
-      logger_->Flush();
+    {
+      MutexLock l(&mutex_);
+      // pin down the current logger_ instance before releasing the mutex.
+      logger = logger_;
+    }
+    TEST_SYNC_POINT_CALLBACK("AutoRollLogger::Flush:PinnedLogger", nullptr);
+    if (logger) {
+      logger->Flush();
     }
   }
 
@@ -101,7 +117,7 @@ class AutoRollLogger : public Logger {
   uint64_t ctime_;
   uint64_t cached_now_access_count;
   uint64_t call_NowMicros_every_N_records_;
-  port::Mutex mutex_;
+  mutable port::Mutex mutex_;
 };
 
 // Facade to craete logger automatically

db/auto_roll_logger_test.cc

@@ -4,6 +4,7 @@
 //  of patent rights can be found in the PATENTS file in the same directory.
 //
 #include <string>
+#include <thread>
 #include <vector>
 #include <cmath>
 #include <iostream>
@@ -11,6 +12,7 @@
 #include <iterator>
 #include <algorithm>
 #include "db/auto_roll_logger.h"
+#include "util/sync_point.h"
 #include "util/testharness.h"
 #include "rocksdb/db.h"
 #include <sys/stat.h>
@@ -260,7 +262,56 @@ TEST_F(AutoRollLoggerTest, CreateLoggerFromOptions) {
       auto_roll_logger, options.log_file_time_to_roll,
       kSampleMessage + ":CreateLoggerFromOptions - both");
 }
-#endif
+
+TEST_F(AutoRollLoggerTest, LogFlushWhileRolling) {
+  DBOptions options;
+  shared_ptr<Logger> logger;
+
+  InitTestDb();
+  options.max_log_file_size = 1024 * 5;
+  ASSERT_OK(CreateLoggerFromOptions(kTestDir, options, &logger));
+  AutoRollLogger* auto_roll_logger =
+      dynamic_cast<AutoRollLogger*>(logger.get());
+  ASSERT_TRUE(auto_roll_logger);
+
+  // The test is split into two parts, with the below callback happening between
+  // them:
+  // (1) Before ResetLogger() is reached, the log rolling test code occasionally
+  //    invokes PosixLogger::Flush(). For this part, dependencies should not be
+  //    enforced.
+  // (2) After ResetLogger() has begun, any calls to PosixLogger::Flush() will
+  //    be from threads other than the log rolling thread. We want to only
+  //    enforce dependencies for this part.
+  rocksdb::SyncPoint::GetInstance()->SetCallBack(
+      "AutoRollLogger::Logv:BeforeResetLogger", [&](void* arg) {
+        rocksdb::SyncPoint::GetInstance()->LoadDependency({
+            {"PosixLogger::Flush:1",
+             "AutoRollLogger::ResetLogger:BeforeNewLogger"},
+            {"AutoRollLogger::ResetLogger:AfterNewLogger",
+             "PosixLogger::Flush:2"},
+        });
+      });
+  std::thread flush_thread;
+  // Additionally, to exercise the edge case, we need to ensure the old logger
+  // is used. For this, we pause after pinning the logger until dependencies
+  // have probably been loaded.
+  rocksdb::SyncPoint::GetInstance()->SetCallBack(
+      "AutoRollLogger::Flush:PinnedLogger", [&](void* arg) {
+        if (std::this_thread::get_id() == flush_thread.get_id()) {
+          sleep(2);
+        }
+      });
+
+  rocksdb::SyncPoint::GetInstance()->EnableProcessing();
+  flush_thread = std::thread([&]() { auto_roll_logger->Flush(); });
+  sleep(1);
+  RollLogFileBySizeTest(auto_roll_logger, options.max_log_file_size,
+                        kSampleMessage + ":LogFlushWhileRolling");
+  flush_thread.join();
+  rocksdb::SyncPoint::GetInstance()->DisableProcessing();
+}
+
+#endif  // OS_WIN
 
 TEST_F(AutoRollLoggerTest, InfoLogLevel) {
   InitTestDb();

util/posix_logger.h

@@ -25,6 +25,7 @@
 
 #include "rocksdb/env.h"
 #include "util/iostats_context_imp.h"
+#include "util/sync_point.h"
 #include <atomic>
 
 namespace rocksdb {
@@ -56,6 +57,8 @@ class PosixLogger : public Logger {
     fclose(file_);
   }
   virtual void Flush() override {
+    TEST_SYNC_POINT("PosixLogger::Flush:1");
+    TEST_SYNC_POINT("PosixLogger::Flush:2");
     if (flush_pending_) {
       flush_pending_ = false;
       fflush(file_);