SERVER-46826 Instantiate the JournalFlusher thread for inMemory and non-durable (nojournal=true) storage engines.

Non-durable engines will not run the JournalFlusher logic periodically, but only upon request,
because checkpoints are costly. inMemory will run the JournalFlusher periodically because it
updates the repl timestamps faster.

(cherry picked from commit 4609f3ebfb178f37153bc04678176af722b0d304)

6 files changed, 89 insertions, 43 deletions

diff --git a/src/mongo/db/service_context_d_test_fixture.cpp b/src/mongo/db/service_context_d_test_fixture.cpp
index 2740a03d446..dcd3587d98a 100644
--- a/src/mongo/db/service_context_d_test_fixture.cpp
+++ b/src/mongo/db/service_context_d_test_fixture.cpp
@@ -84,7 +84,7 @@ ServiceContextMongoDTest::ServiceContextMongoDTest(std::string engine, RepairAct
     initializeStorageEngine(serviceContext,
                             StorageEngineInitFlags::kAllowNoLockFile |
                                 StorageEngineInitFlags::kSkipMetadataFile);
-    StorageControl::startStorageControls(serviceContext);
+    StorageControl::startStorageControls(serviceContext, true /*forTestOnly*/);
 
     DatabaseHolder::set(serviceContext, std::make_unique<DatabaseHolderImpl>());
     IndexAccessMethodFactory::set(serviceContext, std::make_unique<IndexAccessMethodFactoryImpl>());
diff --git a/src/mongo/db/storage/control/journal_flusher.cpp b/src/mongo/db/storage/control/journal_flusher.cpp
index c5e8629b055..7a1d532ec93 100644
--- a/src/mongo/db/storage/control/journal_flusher.cpp
+++ b/src/mongo/db/storage/control/journal_flusher.cpp
@@ -117,7 +117,8 @@ void JournalFlusher::run() {
         }
 
         // Wait until either journalCommitIntervalMs passes or an immediate journal flush is
-        // requested (or shutdown).
+        // requested (or shutdown). If _disablePeriodicFlushes is set, then the thread will not
+        // wake up until a journal flush is externally requested.
 
         auto deadline =
             Date_t::now() + Milliseconds(storageGlobalParams.journalCommitIntervalMs.load());
@@ -125,8 +126,13 @@ void JournalFlusher::run() {
         stdx::unique_lock<Latch> lk(_stateMutex);
 
         MONGO_IDLE_THREAD_BLOCK;
-        _flushJournalNowCV.wait_until(
-            lk, deadline.toSystemTimePoint(), [&] { return _flushJournalNow || _shuttingDown; });
+        if (_disablePeriodicFlushes) {
+            _flushJournalNowCV.wait(lk, [&] { return _flushJournalNow || _shuttingDown; });
+        } else {
+            _flushJournalNowCV.wait_until(lk, deadline.toSystemTimePoint(), [&] {
+                return _flushJournalNow || _shuttingDown;
+            });
+        }
 
         _flushJournalNow = false;
 
diff --git a/src/mongo/db/storage/control/journal_flusher.h b/src/mongo/db/storage/control/journal_flusher.h
index 7204a09104c..f38bd50f2e0 100644
--- a/src/mongo/db/storage/control/journal_flusher.h
+++ b/src/mongo/db/storage/control/journal_flusher.h
@@ -38,9 +38,32 @@ namespace mongo {
 
 class OperationContext;
 
+/**
+ * A periodic and signalable thread that flushes data to disk. Constructor parameter will dictate
+ * whether to periodically flush or only on signal.
+ *
+ * This thread is helpful for two reasons:
+ *  - Periodically flushing data to disk may protect users doing writes with {j: false} from losing
+ *    a great deal of their data across a server crash.
+ *  - Asynchronously grouping data flush requests reduces the total number of flushes executed,
+ *    reducing i/o load on the system and improving write performance. This thread groups both the
+ *    periodic flushes and immediate flush requests from the rest of the system.
+ *
+ * And incidentally helpful for another reason:
+ *  - waitUntilDurable() calls update the replication JournalListener, so more frequent calls may be
+ *    helpful to unblock replication related operations more quickly.
+ */
 class JournalFlusher : public BackgroundJob {
 public:
-    explicit JournalFlusher() : BackgroundJob(/*deleteSelf*/ false) {}
+    /**
+     * Setting 'disablePeriodicFlushes' to true will cause the JournalFlusher thread to only execute
+     * a data flush upon explicit request: flushes will no longer be executed periodically in
+     * addition. This is useful for storage engines that do not want frequent durability updates,
+     * like engines without a journal where the cost of durability is high (using checkpoints
+     * instead).
+     */
+    explicit JournalFlusher(bool disablePeriodicFlushes)
+        : BackgroundJob(/*deleteSelf*/ false), _disablePeriodicFlushes(disablePeriodicFlushes) {}
 
     static JournalFlusher* get(ServiceContext* serviceCtx);
     static JournalFlusher* get(OperationContext* opCtx);
@@ -50,6 +73,11 @@ public:
         return "JournalFlusher";
     }
 
+    /**
+     * Runs data flushes every 'storageGlobalParams.journalCommitIntervalMs' millis (unless
+     * '_disablePeriodicFlushes' is set) or immediately if triggerJournalFlush() or
+     * waitForJournalFlush() is called.
+     */
     void run();
 
     /**
@@ -101,6 +129,11 @@ private:
         std::make_unique<SharedPromise<void>>();
     std::unique_ptr<SharedPromise<void>> _nextSharedPromise =
         std::make_unique<SharedPromise<void>>();
+
+    // Controls whether to ignore the 'storageGlobalParams.journalCommitIntervalMs' setting. If set,
+    // data flushes will only be executed upon explicit request, no longer periodically in addition
+    // to upon request.
+    bool _disablePeriodicFlushes;
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/storage/control/storage_control.cpp b/src/mongo/db/storage/control/storage_control.cpp
index 46534da1db5..6160e0bb0f7 100644
--- a/src/mongo/db/storage/control/storage_control.cpp
+++ b/src/mongo/db/storage/control/storage_control.cpp
@@ -43,50 +43,56 @@ namespace mongo {
 
 namespace StorageControl {
 
-void startStorageControls(ServiceContext* serviceContext) {
+namespace {
+
+bool areControlsStarted = false;
+
+}  // namespace
+
+void startStorageControls(ServiceContext* serviceContext, bool forTestOnly) {
     auto storageEngine = serviceContext->getStorageEngine();
 
     // Instantiate a thread to periodically, and upon request, flush writes to disk.
-    if (!storageEngine->isEphemeral() && storageEngine->isDurable()) {
-        std::unique_ptr<JournalFlusher> journalFlusher = std::make_unique<JournalFlusher>();
-        journalFlusher->go();
-        JournalFlusher::set(serviceContext, std::move(journalFlusher));
-    }
+    //
+    // Persisted storage engines that have a journal should periodically flush the journal to disk
+    // to avoid risking much user data loss across a server crash if the user is not doing {j: true}
+    // writes often.
+    //
+    // Non-durable, i.e. no journal, storage engines should only flush upon request because
+    // waitUntilDurable() will perform a checkpoint and checkpoints are costly. Periodic flushes
+    // will be disabled and only requests will provoke a flush.
+    //
+    // Ephemeral engines are not durable -- waitUntilDurable() returns early -- but frequent updates
+    // to replication's JournalListener in the waitUntilDurable() code may help update replication
+    // timestamps more quickly.
+    //
+    // (Note: the ephemeral engine returns false for isDurable(), so we must be careful not to
+    // disable it.)
+    std::unique_ptr<JournalFlusher> journalFlusher = std::make_unique<JournalFlusher>(
+        /*disablePeriodicFlushes*/ forTestOnly ||
+        (!storageEngine->isDurable() && !storageEngine->isEphemeral()));
+    journalFlusher->go();
+    JournalFlusher::set(serviceContext, std::move(journalFlusher));
+
+    areControlsStarted = true;
 }
 
 void stopStorageControls(ServiceContext* serviceContext) {
-    auto storageEngine = serviceContext->getStorageEngine();
-
-    if (!storageEngine->isEphemeral() && storageEngine->isDurable()) {
+    if (areControlsStarted) {
         JournalFlusher::get(serviceContext)->shutdown();
     }
 }
 
 void triggerJournalFlush(ServiceContext* serviceContext) {
-    auto storageEngine = serviceContext->getStorageEngine();
-
-    if (!storageEngine->isEphemeral() && storageEngine->isDurable()) {
-        JournalFlusher::get(serviceContext)->triggerJournalFlush();
-    }
+    JournalFlusher::get(serviceContext)->triggerJournalFlush();
 }
 
 void waitForJournalFlush(OperationContext* opCtx) {
-    auto serviceContext = opCtx->getServiceContext();
-    auto storageEngine = serviceContext->getStorageEngine();
-
-    if (!storageEngine->isEphemeral() && storageEngine->isDurable()) {
-        JournalFlusher::get(serviceContext)->waitForJournalFlush();
-    } else {
-        opCtx->recoveryUnit()->waitUntilDurable(opCtx);
-    }
+    JournalFlusher::get(opCtx)->waitForJournalFlush();
 }
 
 void interruptJournalFlusherForReplStateChange(ServiceContext* serviceContext) {
-    auto storageEngine = serviceContext->getStorageEngine();
-
-    if (!storageEngine->isEphemeral() && storageEngine->isDurable()) {
-        JournalFlusher::get(serviceContext)->interruptJournalFlusherForReplStateChange();
-    }
+    JournalFlusher::get(serviceContext)->interruptJournalFlusherForReplStateChange();
 }
 
 }  // namespace StorageControl
diff --git a/src/mongo/db/storage/control/storage_control.h b/src/mongo/db/storage/control/storage_control.h
index 75a1289adbf..e54cb51469f 100644
--- a/src/mongo/db/storage/control/storage_control.h
+++ b/src/mongo/db/storage/control/storage_control.h
@@ -41,23 +41,24 @@ class ServiceContext;
 namespace StorageControl {
 
 /**
- * Responsible for initializing independent processes for replication that and interact with the
- * storage layer.
+ * Responsible for initializing independent processes for replication that interact with the storage
+ * layer.
  *
- * Instantiates the JournalFlusher to periodically, and upon request, flush writes to disk.
+ * Instantiates the JournalFlusher to flush writes to disk periodically and upon request. If
+ * 'forTestOnly' is set, then the JournalFlusher will only run upon request so as not to disrupt
+ * unit test expectations.
  *
- * Safe to call again after stopStorageControls has been called, to restart any processes that were
- * stopped.
+ * Safe to call again after stopStorageControls() has been called, to restart any processes that
+ * were stopped.
  */
-void startStorageControls(ServiceContext* serviceContext);
+void startStorageControls(ServiceContext* serviceContext, bool forTestOnly = false);
 
 /**
- * Stops the processes begun by startStorageControls.
+ * Stops the processes begun by startStorageControls().
  *
- * The OplogCapMaintainerThread is unowned and therefore ignored; the JournalFlusher is shut
- * down.
+ * The JournalFlusher is shut down.
  *
- * Safe to call multiple times.
+ * Safe to call multiple times, whether or not startStorageControls() has been called.
  */
 void stopStorageControls(ServiceContext* serviceContext);
 
diff --git a/src/mongo/dbtests/framework.cpp b/src/mongo/dbtests/framework.cpp
index 4c457fc1072..e1516434bb7 100644
--- a/src/mongo/dbtests/framework.cpp
+++ b/src/mongo/dbtests/framework.cpp
@@ -106,7 +106,7 @@ int runDbTests(int argc, char** argv) {
     globalServiceContext->setPeriodicRunner(std::move(runner));
 
     initializeStorageEngine(globalServiceContext, StorageEngineInitFlags::kNone);
-    StorageControl::startStorageControls(globalServiceContext);
+    StorageControl::startStorageControls(globalServiceContext, true /*forTestOnly*/);
     DatabaseHolder::set(globalServiceContext, std::make_unique<DatabaseHolderImpl>());
     IndexAccessMethodFactory::set(globalServiceContext,
                                   std::make_unique<IndexAccessMethodFactoryImpl>());