Introduce and use the option to disable stall notifications structures

Summary:
and code. Removing this helps with insert performance.
Closes https://github.com/facebook/rocksdb/pull/3830

Differential Revision: D7921030

Pulled By: siying

fbshipit-source-id: 84e80d50a7ef96f5441c51c9a0d089c50217cce2

CMakeLists.txt

@@ -301,6 +301,12 @@ if(WITH_TBB)
   list(APPEND THIRDPARTY_LIBS ${TBB_LIBRARIES})
 endif()
 
+# Stall notifications eat some performance from inserts
+option(DISABLE_STALL_NOTIF "Build with stall notifications" OFF)
+if(DISABLE_STALL_NOTIF)
+  add_definitions(-DROCKSDB_DISABLE_STALL_NOTIFICATION)
+endif()
+
 # Used to run CI build and tests so we can run faster
 set(OPTIMIZE_DEBUG_DEFAULT 0)        # Debug build is unoptimized by default use -DOPTDBG=1 to optimize
 

db/db_test.cc

@@ -5442,7 +5442,7 @@ TEST_F(DBTest, HardLimit) {
   sleeping_task_low.WaitUntilDone();
 }
 
-#ifndef ROCKSDB_LITE
+#if !defined(ROCKSDB_LITE) && !defined(ROCKSDB_DISABLE_STALL_NOTIFICATION)
 class WriteStallListener : public EventListener {
  public:
   WriteStallListener() : cond_(&mutex_), 
@@ -5664,7 +5664,7 @@ TEST_F(DBTest, LastWriteBufferDelay) {
   sleeping_task.WakeUp();
   sleeping_task.WaitUntilDone();
 }
-#endif  // ROCKSDB_LITE
+#endif  // !defined(ROCKSDB_LITE) && !defined(ROCKSDB_DISABLE_STALL_NOTIFICATION)
 
 TEST_F(DBTest, FailWhenCompressionNotSupportedTest) {
   CompressionType compressions[] = {kZlibCompression, kBZip2Compression,

db/job_context.h

@@ -27,7 +27,9 @@ struct SuperVersionContext {
   };
 
   autovector<SuperVersion*> superversions_to_free;
+#ifndef ROCKSDB_DISABLE_STALL_NOTIFICATION
   autovector<WriteStallNotification> write_stall_notifications;
+#endif
   unique_ptr<SuperVersion> new_superversion;  // if nullptr no new superversion
 
   explicit SuperVersionContext(bool create_superversion = false)
@@ -38,14 +40,18 @@ struct SuperVersionContext {
   }
 
   inline bool HaveSomethingToDelete() const {
-    return superversions_to_free.size() > 0 ||
-           write_stall_notifications.size() > 0;
+#ifndef ROCKSDB_DISABLE_STALL_NOTIFICATION
+    return !superversions_to_free.empty() ||
+           !write_stall_notifications.empty();
+#else
+    return !superversions_to_free.empty();
+#endif
   }
 
   void PushWriteStallNotification(
       WriteStallCondition old_cond, WriteStallCondition new_cond,
       const std::string& name, const ImmutableCFOptions* ioptions) {
-#ifndef ROCKSDB_LITE
+#if !defined(ROCKSDB_LITE) && !defined(ROCKSDB_DISABLE_STALL_NOTIFICATION)
     WriteStallNotification notif;
     notif.write_stall_info.cf_name = name;
     notif.write_stall_info.condition.prev = old_cond;
@@ -57,14 +63,14 @@ struct SuperVersionContext {
     (void)new_cond;
     (void)name;
     (void)ioptions;
-#endif  // !ROCKSDB_LITE
+#endif  // !defined(ROCKSDB_LITE) && !defined(ROCKSDB_DISABLE_STALL_NOTIFICATION)
   }
 
   void Clean() {
-#ifndef ROCKSDB_LITE
+#if !defined(ROCKSDB_LITE) && !defined(ROCKSDB_DISABLE_STALL_NOTIFICATION)
     // notify listeners on changed write stall conditions
     for (auto& notif : write_stall_notifications) {
-      for (auto listener : notif.immutable_cf_options->listeners) {
+      for (auto& listener : notif.immutable_cf_options->listeners) {
         listener->OnStallConditionsChanged(notif.write_stall_info);
       }
     }
@@ -78,8 +84,10 @@ struct SuperVersionContext {
   }
 
   ~SuperVersionContext() {
-    assert(write_stall_notifications.size() == 0);
-    assert(superversions_to_free.size() == 0);
+#ifndef ROCKSDB_DISABLE_STALL_NOTIFICATION
+    assert(write_stall_notifications.empty());
+#endif
+    assert(superversions_to_free.empty());
   }
 };
 
@@ -99,7 +107,7 @@ struct JobContext {
     std::string file_name;
     std::string file_path;
     CandidateFileInfo(std::string name, std::string path)
-        : file_name(std::move(name)), file_path(path) {}
+        : file_name(std::move(name)), file_path(std::move(path)) {}
     bool operator==(const CandidateFileInfo& other) const {
       return file_name == other.file_name &&
              file_path == other.file_path;