SERVER-41160 Add shutdown method for CatalogCacheLoader

Author: renctan

src/mongo/db/db.cpp

@@ -978,6 +978,10 @@ void shutdownTask(const ShutdownTaskArgs& shutdownArgs) {
         validator->shutDown();
     }
 
+    if (ShardingState::get(serviceContext)->enabled()) {
+        CatalogCacheLoader::get(serviceContext).shutDown();
+    }
+
 #if __has_feature(address_sanitizer)
     // When running under address sanitizer, we get false positive leaks due to disorder around
     // the lifecycle of a connection and request. When we are running under ASAN, we try a lot

src/mongo/db/s/catalog_cache_loader_mock.cpp

@@ -82,6 +82,8 @@ void CatalogCacheLoaderMock::onStepUp() {
     MONGO_UNREACHABLE;
 }
 
+void CatalogCacheLoaderMock::shutDown() {}
+
 void CatalogCacheLoaderMock::notifyOfCollectionVersionUpdate(const NamespaceString& nss) {
     MONGO_UNREACHABLE;
 }

src/mongo/db/s/catalog_cache_loader_mock.h

@@ -52,6 +52,7 @@ class CatalogCacheLoaderMock final : public CatalogCacheLoader {
     void initializeReplicaSetRole(bool isPrimary) override;
     void onStepDown() override;
     void onStepUp() override;
+    void shutDown() override;
     void notifyOfCollectionVersionUpdate(const NamespaceString& nss) override;
     void waitForCollectionFlush(OperationContext* opCtx, const NamespaceString& nss) override;
     void waitForDatabaseFlush(OperationContext* opCtx, StringData dbName) override;

src/mongo/db/s/read_only_catalog_cache_loader.cpp

@@ -35,6 +35,10 @@ namespace mongo {
 
 using CollectionAndChangedChunks = CatalogCacheLoader::CollectionAndChangedChunks;
 
+ReadOnlyCatalogCacheLoader::~ReadOnlyCatalogCacheLoader() {
+    shutDown();
+}
+
 void ReadOnlyCatalogCacheLoader::waitForCollectionFlush(OperationContext* opCtx,
                                                         const NamespaceString& nss) {
     MONGO_UNREACHABLE;
@@ -44,6 +48,10 @@ void ReadOnlyCatalogCacheLoader::waitForDatabaseFlush(OperationContext* opCtx, S
     MONGO_UNREACHABLE;
 }
 
+void ReadOnlyCatalogCacheLoader::shutDown() {
+    _configServerLoader.shutDown();
+}
+
 std::shared_ptr<Notification<void>> ReadOnlyCatalogCacheLoader::getChunksSince(
     const NamespaceString& nss, ChunkVersion version, GetChunksSinceCallbackFn callbackFn) {
     return _configServerLoader.getChunksSince(nss, version, callbackFn);

src/mongo/db/s/read_only_catalog_cache_loader.h

@@ -40,9 +40,12 @@ namespace mongo {
  */
 class ReadOnlyCatalogCacheLoader final : public CatalogCacheLoader {
 public:
+    ~ReadOnlyCatalogCacheLoader();
+
     void initializeReplicaSetRole(bool isPrimary) override {}
     void onStepDown() override {}
     void onStepUp() override {}
+    void shutDown() override;
     void notifyOfCollectionVersionUpdate(const NamespaceString& nss) override {}
     void waitForCollectionFlush(OperationContext* opCtx, const NamespaceString& nss) override;
     void waitForDatabaseFlush(OperationContext* opCtx, StringData dbName) override;

src/mongo/db/s/shard_server_catalog_cache_loader.cpp

@@ -342,16 +342,7 @@ ShardServerCatalogCacheLoader::ShardServerCatalogCacheLoader(
 }
 
 ShardServerCatalogCacheLoader::~ShardServerCatalogCacheLoader() {
-    // Prevent further scheduling, then interrupt ongoing tasks.
-    _threadPool.shutdown();
-    {
-        stdx::lock_guard<stdx::mutex> lock(_mutex);
-        _contexts.interrupt(ErrorCodes::InterruptedAtShutdown);
-        ++_term;
-    }
-
-    _threadPool.join();
-    invariant(_contexts.isEmpty());
+    shutDown();
 }
 
 void ShardServerCatalogCacheLoader::notifyOfCollectionVersionUpdate(const NamespaceString& nss) {
@@ -384,6 +375,30 @@ void ShardServerCatalogCacheLoader::onStepUp() {
     _role = ReplicaSetRole::Primary;
 }
 
+void ShardServerCatalogCacheLoader::shutDown() {
+    {
+        stdx::lock_guard<stdx::mutex> lg(_mutex);
+        if (_inShutdown) {
+            return;
+        }
+
+        _inShutdown = true;
+    }
+
+    // Prevent further scheduling, then interrupt ongoing tasks.
+    _threadPool.shutdown();
+    {
+        stdx::lock_guard<stdx::mutex> lock(_mutex);
+        _contexts.interrupt(ErrorCodes::InterruptedAtShutdown);
+        ++_term;
+    }
+
+    _threadPool.join();
+    invariant(_contexts.isEmpty());
+
+    _configServerLoader->shutDown();
+}
+
 std::shared_ptr<Notification<void>> ShardServerCatalogCacheLoader::getChunksSince(
     const NamespaceString& nss, ChunkVersion version, GetChunksSinceCallbackFn callbackFn) {
     auto notify = std::make_shared<Notification<void>>();

src/mongo/db/s/shard_server_catalog_cache_loader.h

@@ -69,6 +69,8 @@ class ShardServerCatalogCacheLoader : public CatalogCacheLoader {
      */
     void onStepUp() override;
 
+    void shutDown() override;
+
     /**
      * Sets any notifications waiting for this version to arrive and invalidates the catalog cache's
      * chunk metadata for collection 'nss' so that the next caller provokes a refresh.
@@ -473,7 +475,7 @@ class ShardServerCatalogCacheLoader : public CatalogCacheLoader {
 
     // Loader used by the shard primary to retrieve the authoritative routing metadata from the
     // config server
-    const std::unique_ptr<CatalogCacheLoader> _configServerLoader;
+    std::unique_ptr<CatalogCacheLoader> _configServerLoader;
 
     // Thread pool used to run blocking tasks which perform disk reads and writes
     ThreadPool _threadPool;
@@ -484,6 +486,9 @@ class ShardServerCatalogCacheLoader : public CatalogCacheLoader {
     // Protects the class state below
     stdx::mutex _mutex;
 
+    // True if shutDown was called.
+    bool _inShutdown{false};
+
     // This value is bumped every time the set of currently scheduled tasks should no longer be
     // running. This includes, replica set state transitions and shutdown.
     long long _term{0};

src/mongo/db/s/shard_server_catalog_cache_loader_test.cpp

@@ -107,6 +107,7 @@ void ShardServerCatalogCacheLoaderTest::setUp() {
 }
 
 void ShardServerCatalogCacheLoaderTest::tearDown() {
+    _shardLoader->shutDown();
     _shardLoader.reset();
     ShardServerTestFixture::tearDown();
 }

src/mongo/s/catalog_cache_loader.h

@@ -108,6 +108,12 @@ class CatalogCacheLoader {
      */
     virtual void onStepUp() = 0;
 
+    /**
+     * Transitions into shut down and cleans up state. Once this transitions to shut down, should
+     * not be able to transition back to normal. Should be safe to be called more than once.
+     */
+    virtual void shutDown() = 0;
+
     /**
      * Notifies the loader that the persisted collection version for 'nss' has been updated.
      */

src/mongo/s/config_server_catalog_cache_loader.cpp

@@ -143,8 +143,7 @@ ConfigServerCatalogCacheLoader::ConfigServerCatalogCacheLoader()
 }
 
 ConfigServerCatalogCacheLoader::~ConfigServerCatalogCacheLoader() {
-    _threadPool.shutdown();
-    _threadPool.join();
+    shutDown();
 }
 
 void ConfigServerCatalogCacheLoader::initializeReplicaSetRole(bool isPrimary) {
@@ -159,6 +158,20 @@ void ConfigServerCatalogCacheLoader::onStepUp() {
     MONGO_UNREACHABLE;
 }
 
+void ConfigServerCatalogCacheLoader::shutDown() {
+    {
+        stdx::lock_guard<stdx::mutex> lg(_mutex);
+        if (_inShutdown) {
+            return;
+        }
+
+        _inShutdown = true;
+    }
+
+    _threadPool.shutdown();
+    _threadPool.join();
+}
+
 void ConfigServerCatalogCacheLoader::notifyOfCollectionVersionUpdate(const NamespaceString& nss) {
     MONGO_UNREACHABLE;
 }