SERVER-47475 Make PeriodicJob aware of cancelation at global shutdown

4 files changed, 53 insertions, 7 deletions

diff --git a/src/mongo/base/error_codes.yml b/src/mongo/base/error_codes.yml
index cc48f70d715..0d1403edc5a 100644
--- a/src/mongo/base/error_codes.yml
+++ b/src/mongo/base/error_codes.yml
@@ -354,6 +354,7 @@ error_codes:
     - {code: 307,name: RangeDeletionAbandonedBecauseTaskDocumentDoesNotExist}
     - {code: 308,name: CurrentConfigNotCommittedYet}
     - {code: 309,name: ExhaustCommandFinished}
+    - {code: 310,name: PeriodicJobIsStopped,categories: [CancelationError]}
 
     # Error codes 4000-8999 are reserved.
 
diff --git a/src/mongo/db/db.cpp b/src/mongo/db/db.cpp
index 37a58b8dd9d..1afdf7a2d75 100644
--- a/src/mongo/db/db.cpp
+++ b/src/mongo/db/db.cpp
@@ -706,11 +706,19 @@ ExitCode _initAndListen(ServiceContext* serviceContext, int listenPort) {
     // Only do this on storage engines supporting snapshot reads, which hold resources we wish to
     // release periodically in order to avoid storage cache pressure build up.
     if (storageEngine->supportsReadConcernSnapshot()) {
-        PeriodicThreadToAbortExpiredTransactions::get(serviceContext)->start();
-        // The inMemory engine is not yet used for replica or sharded transactions in production so
-        // it does not currently maintain snapshot history. It is live in testing, however.
-        if (!storageEngine->isEphemeral() || getTestCommandsEnabled()) {
-            PeriodicThreadToDecreaseSnapshotHistoryCachePressure::get(serviceContext)->start();
+        try {
+            PeriodicThreadToAbortExpiredTransactions::get(serviceContext)->start();
+            // The inMemory engine is not yet used for replica or sharded transactions in production
+            // so it does not currently maintain snapshot history. It is live in testing, however.
+            if (!storageEngine->isEphemeral() || getTestCommandsEnabled()) {
+                PeriodicThreadToDecreaseSnapshotHistoryCachePressure::get(serviceContext)->start();
+            }
+        } catch (ExceptionFor<ErrorCodes::PeriodicJobIsStopped>&) {
+            LOGV2_WARNING(4747501, "Not starting periodic jobs as shutdown is in progress");
+            // Shutdown has already started before initialization is complete. Wait for the
+            // shutdown task to complete and return.
+            MONGO_IDLE_THREAD_BLOCK;
+            return waitForShutdown();
         }
     }
 
diff --git a/src/mongo/util/periodic_runner_impl.cpp b/src/mongo/util/periodic_runner_impl.cpp
index d39df21ac02..362b304c9d9 100644
--- a/src/mongo/util/periodic_runner_impl.cpp
+++ b/src/mongo/util/periodic_runner_impl.cpp
@@ -33,9 +33,11 @@
 
 #include "mongo/util/periodic_runner_impl.h"
 
+#include "mongo/base/error_codes.h"
 #include "mongo/db/client.h"
 #include "mongo/db/service_context.h"
 #include "mongo/logv2/log.h"
+#include "mongo/util/assert_util.h"
 #include "mongo/util/clock_source.h"
 #include "mongo/util/scopeguard.h"
 
@@ -57,13 +59,15 @@ PeriodicRunnerImpl::PeriodicJobImpl::PeriodicJobImpl(PeriodicJob job,
     : _job(std::move(job)), _clockSource(source), _serviceContext(svc), _client(nullptr) {}
 
 void PeriodicRunnerImpl::PeriodicJobImpl::_run() {
-    auto [startPromise, startFuture] = makePromiseFuture<void>();
-
     {
         stdx::lock_guard lk(_mutex);
+        if (MONGO_unlikely(_execStatus == ExecutionStatus::CANCELED))
+            uasserted(ErrorCodes::PeriodicJobIsStopped,
+                      "Attempted to start an already stopped job");
         invariant(_execStatus == ExecutionStatus::NOT_SCHEDULED);
     }
 
+    auto [startPromise, startFuture] = makePromiseFuture<void>();
 
     _thread = stdx::thread([this, startPromise = std::move(startPromise)]() mutable {
         ON_BLOCK_EXIT([this] { _stopPromise.emplaceValue(); });
@@ -129,6 +133,8 @@ void PeriodicRunnerImpl::PeriodicJobImpl::start() {
 
 void PeriodicRunnerImpl::PeriodicJobImpl::pause() {
     stdx::lock_guard<Latch> lk(_mutex);
+    if (MONGO_unlikely(_execStatus == ExecutionStatus::CANCELED))
+        uasserted(ErrorCodes::PeriodicJobIsStopped, "Attempted to pause an already stopped job");
     invariant(_execStatus == PeriodicJobImpl::ExecutionStatus::RUNNING);
     _execStatus = PeriodicJobImpl::ExecutionStatus::PAUSED;
 }
@@ -136,6 +142,9 @@ void PeriodicRunnerImpl::PeriodicJobImpl::pause() {
 void PeriodicRunnerImpl::PeriodicJobImpl::resume() {
     {
         stdx::lock_guard<Latch> lk(_mutex);
+        if (MONGO_unlikely(_execStatus == ExecutionStatus::CANCELED))
+            uasserted(ErrorCodes::PeriodicJobIsStopped,
+                      "Attempted to resume an already stopped job");
         invariant(_execStatus == PeriodicJobImpl::ExecutionStatus::PAUSED);
         _execStatus = PeriodicJobImpl::ExecutionStatus::RUNNING;
     }
diff --git a/src/mongo/util/periodic_runner_impl_test.cpp b/src/mongo/util/periodic_runner_impl_test.cpp
index 22bcbe3506b..6fd5639a161 100644
--- a/src/mongo/util/periodic_runner_impl_test.cpp
+++ b/src/mongo/util/periodic_runner_impl_test.cpp
@@ -27,6 +27,9 @@
  *    it in the license file.
  */
 
+#include <boost/optional.hpp>
+
+#include "mongo/base/error_codes.h"
 #include "mongo/platform/atomic_word.h"
 #include "mongo/platform/basic.h"
 
@@ -71,6 +74,14 @@ public:
     void setUp() override {
         PeriodicRunnerImplTestNoSetup::setUp();
     }
+
+    auto makeStoppedJob() {
+        PeriodicRunner::PeriodicJob job("job", [](Client* client) {}, Seconds{1});
+        auto jobAnchor = runner().makeJob(std::move(job));
+        jobAnchor.start();
+        jobAnchor.stop();
+        return jobAnchor;
+    }
 };
 
 TEST_F(PeriodicRunnerImplTest, OneJobTest) {
@@ -476,5 +487,22 @@ TEST_F(PeriodicRunnerImplTest, StopProperlyInterruptsOpCtx) {
     tearDown();
 }
 
+TEST_F(PeriodicRunnerImplTest, ThrowsErrorOnceStopped) {
+    auto jobAnchor = makeStoppedJob();
+    ASSERT_THROWS_CODE_AND_WHAT(jobAnchor.start(),
+                                AssertionException,
+                                ErrorCodes::PeriodicJobIsStopped,
+                                "Attempted to start an already stopped job");
+    ASSERT_THROWS_CODE_AND_WHAT(jobAnchor.pause(),
+                                AssertionException,
+                                ErrorCodes::PeriodicJobIsStopped,
+                                "Attempted to pause an already stopped job");
+    ASSERT_THROWS_CODE_AND_WHAT(jobAnchor.resume(),
+                                AssertionException,
+                                ErrorCodes::PeriodicJobIsStopped,
+                                "Attempted to resume an already stopped job");
+    jobAnchor.stop();
+}
+
 }  // namespace
 }  // namespace mongo