SERVER-46202 Implement ShardRegistry on top of ReadThroughCache

10 files changed, 787 insertions, 377 deletions

diff --git a/src/mongo/db/s/migration_coordinator.cpp b/src/mongo/db/s/migration_coordinator.cpp
index 22102845062..d72311a3369 100644
--- a/src/mongo/db/s/migration_coordinator.cpp
+++ b/src/mongo/db/s/migration_coordinator.cpp
@@ -235,21 +235,32 @@ void MigrationCoordinator::_abortMigrationOnDonorAndRecipient(OperationContext*
         23899, 2, "Making abort decision durable", "migrationId"_attr = _migrationInfo.getId());
     migrationutil::persistAbortDecision(opCtx, _migrationInfo.getId());
 
-    LOGV2_DEBUG(
-        23900,
-        2,
-        "Bumping transaction number with lsid {lsid} and current txnNumber {currentTxnNumber} on "
-        "recipient shard {recipientShardId} for abort of collection {nss}",
-        "Bumping transaction number on recipient shard for abort",
-        "namespace"_attr = _migrationInfo.getNss(),
-        "recipientShardId"_attr = _migrationInfo.getRecipientShardId(),
-        "lsid"_attr = _migrationInfo.getLsid(),
-        "currentTxnNumber"_attr = _migrationInfo.getTxnNumber(),
-        "migrationId"_attr = _migrationInfo.getId());
-    migrationutil::advanceTransactionOnRecipient(opCtx,
-                                                 _migrationInfo.getRecipientShardId(),
-                                                 _migrationInfo.getLsid(),
-                                                 _migrationInfo.getTxnNumber());
+    try {
+        LOGV2_DEBUG(23900,
+                    2,
+                    "Bumping transaction number with lsid {lsid} and current txnNumber "
+                    "{currentTxnNumber} on "
+                    "recipient shard {recipientShardId} for abort of collection {nss}",
+                    "Bumping transaction number on recipient shard for abort",
+                    "namespace"_attr = _migrationInfo.getNss(),
+                    "recipientShardId"_attr = _migrationInfo.getRecipientShardId(),
+                    "lsid"_attr = _migrationInfo.getLsid(),
+                    "currentTxnNumber"_attr = _migrationInfo.getTxnNumber(),
+                    "migrationId"_attr = _migrationInfo.getId());
+        migrationutil::advanceTransactionOnRecipient(opCtx,
+                                                     _migrationInfo.getRecipientShardId(),
+                                                     _migrationInfo.getLsid(),
+                                                     _migrationInfo.getTxnNumber());
+    } catch (const ExceptionFor<ErrorCodes::ShardNotFound>& exShardNotFound) {
+        LOGV2_DEBUG(4620231,
+                    1,
+                    "Failed to advance transaction number on recipient shard for abort",
+                    "namespace"_attr = _migrationInfo.getNss(),
+                    "migrationId"_attr = _migrationInfo.getId(),
+                    "recipientShardId"_attr = _migrationInfo.getRecipientShardId(),
+                    "currentTxnNumber"_attr = _migrationInfo.getTxnNumber(),
+                    "error"_attr = exShardNotFound);
+    }
 
     hangBeforeSendingAbortDecision.pauseWhileSet();
 
diff --git a/src/mongo/db/s/migration_util.cpp b/src/mongo/db/s/migration_util.cpp
index f1c7cec016a..7d289fb979e 100644
--- a/src/mongo/db/s/migration_util.cpp
+++ b/src/mongo/db/s/migration_util.cpp
@@ -679,11 +679,20 @@ void markAsReadyRangeDeletionTaskOnRecipient(OperationContext* opCtx,
         opCtx, "ready remote range deletion", [&](OperationContext* newOpCtx) {
             hangInReadyRangeDeletionOnRecipientInterruptible.pauseWhileSet(newOpCtx);
 
-            sendToRecipient(
-                newOpCtx,
-                recipientId,
-                updateOp,
-                BSON(WriteConcernOptions::kWriteConcernField << WriteConcernOptions::Majority));
+            try {
+                sendToRecipient(
+                    newOpCtx,
+                    recipientId,
+                    updateOp,
+                    BSON(WriteConcernOptions::kWriteConcernField << WriteConcernOptions::Majority));
+            } catch (const ExceptionFor<ErrorCodes::ShardNotFound>& exShardNotFound) {
+                LOGV2_DEBUG(4620232,
+                            1,
+                            "Failed to mark range deletion task on recipient shard as ready",
+                            "migrationId"_attr = migrationId,
+                            "error"_attr = exShardNotFound);
+                return;
+            }
 
             if (hangInReadyRangeDeletionOnRecipientThenSimulateErrorUninterruptible.shouldFail()) {
                 hangInReadyRangeDeletionOnRecipientThenSimulateErrorUninterruptible.pauseWhileSet(
diff --git a/src/mongo/db/s/sessions_collection_config_server.cpp b/src/mongo/db/s/sessions_collection_config_server.cpp
index 95ee087e5ae..99eb2277d02 100644
--- a/src/mongo/db/s/sessions_collection_config_server.cpp
+++ b/src/mongo/db/s/sessions_collection_config_server.cpp
@@ -61,7 +61,7 @@ void SessionsCollectionConfigServer::_shardCollectionIfNeeded(OperationContext*
     uassert(ErrorCodes::ShardNotFound,
             str::stream() << "Failed to create " << NamespaceString::kLogicalSessionsNamespace
                           << ": cannot create the collection until there are shards",
-            Grid::get(opCtx)->shardRegistry()->getNumShards() != 0);
+            Grid::get(opCtx)->shardRegistry()->getNumShardsNoReload() != 0);
 
     // First, shard the sessions collection to create it.
     ConfigsvrShardCollectionRequest shardCollection;
diff --git a/src/mongo/db/s/sharding_mongod_test_fixture.cpp b/src/mongo/db/s/sharding_mongod_test_fixture.cpp
index a78fea13156..78bd0f861ae 100644
--- a/src/mongo/db/s/sharding_mongod_test_fixture.cpp
+++ b/src/mongo/db/s/sharding_mongod_test_fixture.cpp
@@ -266,10 +266,6 @@ Status ShardingMongodTestFixture::initializeGlobalShardingStateForMongodForTest(
                std::move(executorPoolPtr),
                _mockNetwork);
 
-    // NOTE: ShardRegistry::startup() is not called because it starts a task executor with a
-    // self-rescheduling task to reload the ShardRegistry over the network.
-    // grid->shardRegistry()->startup();
-
     if (grid->catalogClient()) {
         grid->catalogClient()->startup();
     }
diff --git a/src/mongo/s/client/shard_registry.cpp b/src/mongo/s/client/shard_registry.cpp
index e0a5cc13476..842da344bec 100644
--- a/src/mongo/s/client/shard_registry.cpp
+++ b/src/mongo/s/client/shard_registry.cpp
@@ -33,50 +33,41 @@
 
 #include "mongo/s/client/shard_registry.h"
 
-#include <memory>
-#include <set>
-
-#include "mongo/bson/bsonobj.h"
-#include "mongo/bson/util/bson_extract.h"
-#include "mongo/client/connection_string.h"
 #include "mongo/client/replica_set_monitor.h"
 #include "mongo/db/client.h"
 #include "mongo/db/logical_time_metadata_hook.h"
-#include "mongo/db/operation_context.h"
-#include "mongo/db/server_options.h"
-#include "mongo/executor/network_connection_hook.h"
+#include "mongo/db/vector_clock.h"
 #include "mongo/executor/network_interface_factory.h"
 #include "mongo/executor/network_interface_thread_pool.h"
-#include "mongo/executor/task_executor.h"
 #include "mongo/executor/task_executor_pool.h"
 #include "mongo/executor/thread_pool_task_executor.h"
 #include "mongo/logv2/log.h"
-#include "mongo/platform/mutex.h"
 #include "mongo/rpc/metadata/egress_metadata_hook_list.h"
 #include "mongo/s/catalog/sharding_catalog_client.h"
 #include "mongo/s/catalog/type_shard.h"
-#include "mongo/s/client/shard.h"
-#include "mongo/s/client/shard_factory.h"
 #include "mongo/s/grid.h"
-#include "mongo/util/concurrency/with_lock.h"
-#include "mongo/util/scopeguard.h"
 #include "mongo/util/str.h"
 
 namespace mongo {
 
-using executor::NetworkInterface;
-using executor::NetworkInterfaceThreadPool;
-using executor::TaskExecutor;
-using executor::TaskExecutorPool;
-using executor::ThreadPoolTaskExecutor;
-using CallbackArgs = TaskExecutor::CallbackArgs;
-using CallbackHandle = TaskExecutor::CallbackHandle;
-
-
 namespace {
+
 const Seconds kRefreshPeriod(30);
+
+/**
+ * Whether or not the actual topologyTime should be used.  When this is false, the
+ * topologyTime part of the cache's Time will stay fixed and not advance.
+ */
+bool useActualTopologyTime() {
+    return serverGlobalParams.featureCompatibility.isVersionInitialized() &&
+        serverGlobalParams.featureCompatibility.isGreaterThanOrEqualTo(
+            ServerGlobalParams::FeatureCompatibility::Version::kVersion47);
+}
+
 }  // namespace
 
+using CallbackArgs = executor::TaskExecutor::CallbackArgs;
+
 const ShardId ShardRegistry::kConfigServerShardId = ShardId("config");
 
 ShardRegistry::ShardRegistry(std::unique_ptr<ShardFactory> shardFactory,
@@ -84,154 +75,145 @@ ShardRegistry::ShardRegistry(std::unique_ptr<ShardFactory> shardFactory,
                              std::vector<ShardRemovalHook> shardRemovalHooks)
     : _shardFactory(std::move(shardFactory)),
       _initConfigServerCS(configServerCS),
-      _shardRemovalHooks(std::move(shardRemovalHooks)) {
+      _shardRemovalHooks(std::move(shardRemovalHooks)),
+      _threadPool([] {
+          ThreadPool::Options options;
+          options.poolName = "ShardRegistry";
+          options.minThreads = 0;
+          options.maxThreads = 1;
+          return options;
+      }()) {
     invariant(_initConfigServerCS.isValid());
+    _threadPool.startup();
 }
 
 ShardRegistry::~ShardRegistry() {
     shutdown();
 }
 
-void ShardRegistry::shutdown() {
-    if (_executor && !_isShutdown.load()) {
-        LOGV2_DEBUG(22723, 1, "Shutting down task executor for reloading shard registry");
-        _executor->shutdown();
-        _executor->join();
-        _isShutdown.store(true);
-    }
-}
-
-ConnectionString ShardRegistry::getConfigServerConnectionString() const {
-    return getConfigShard()->getConnString();
-}
-
-StatusWith<std::shared_ptr<Shard>> ShardRegistry::getShard(OperationContext* opCtx,
-                                                           const ShardId& shardId) {
-    // If we know about the shard, return it.
-    auto shard = getShardNoReload(shardId);
-    if (shard) {
-        return shard;
-    }
-
-    // If we can't find the shard, attempt to reload the ShardRegistry.
-    bool didReload = reload(opCtx);
-    shard = getShardNoReload(shardId);
-
-    // If we found the shard, return it.
-    if (shard) {
-        return shard;
-    }
-
-    // If we did not find the shard but performed the reload
-    // ourselves, return, because it means the shard does not exist.
-    if (didReload) {
-        return {ErrorCodes::ShardNotFound, str::stream() << "Shard " << shardId << " not found"};
-    }
-
-    // If we did not perform the reload ourselves (because there was a concurrent reload), force a
-    // reload again to ensure that we have seen data at least as up to date as our first reload.
-    reload(opCtx);
-    shard = getShardNoReload(shardId);
-
-    if (shard) {
-        return shard;
-    }
-
-    return {ErrorCodes::ShardNotFound, str::stream() << "Shard " << shardId << " not found"};
-}
-
-std::shared_ptr<Shard> ShardRegistry::getShardNoReload(const ShardId& shardId) {
-    stdx::lock_guard<Latch> lk(_mutex);
-    return _data.findShard(shardId);
-}
-
-std::shared_ptr<Shard> ShardRegistry::getShardForHostNoReload(const HostAndPort& host) {
-    stdx::lock_guard<Latch> lk(_mutex);
-    return _data.findByHostAndPort(host);
-}
+void ShardRegistry::init(ServiceContext* service) {
+    invariant(!_isInitialized.load());
 
-std::shared_ptr<Shard> ShardRegistry::getConfigShard() const {
-    stdx::lock_guard<Latch> lk(_mutex);
-    invariant(_configShard);
-    return _configShard;
-}
+    invariant(!_service);
+    _service = service;
 
-std::unique_ptr<Shard> ShardRegistry::createConnection(const ConnectionString& connStr) const {
-    return _shardFactory->createUniqueShard(ShardId("<unnamed>"), connStr);
-}
+    auto lookupFn = [this](OperationContext* opCtx,
+                           const Singleton& key,
+                           const Cache::ValueHandle& cachedData,
+                           const Time& timeInStore) {
+        return _lookup(opCtx, key, cachedData, timeInStore);
+    };
 
-std::shared_ptr<Shard> ShardRegistry::lookupRSName(const std::string& name) const {
-    stdx::lock_guard<Latch> lk(_mutex);
-    return _data.findByRSName(name);
-}
+    _cache =
+        std::make_unique<Cache>(_cacheMutex, _service, _threadPool, lookupFn, 1 /* cacheSize */);
 
-void ShardRegistry::getAllShardIdsNoReload(std::vector<ShardId>* all) const {
-    std::set<ShardId> seen;
     {
         stdx::lock_guard<Latch> lk(_mutex);
-        _data.getAllShardIds(seen);
+        _configShardData = ShardRegistryData::createWithConfigShardOnly(
+            _shardFactory->createShard(kConfigServerShardId, _initConfigServerCS));
     }
-    all->assign(seen.begin(), seen.end());
+
+    _isInitialized.store(true);
 }
 
-void ShardRegistry::getAllShardIds(OperationContext* opCtx, std::vector<ShardId>* all) {
-    getAllShardIdsNoReload(all);
-    if (all->empty()) {
-        bool didReload = reload(opCtx);
-        getAllShardIdsNoReload(all);
-        // If we didn't do the reload ourselves, we should retry to ensure
-        // that the reload is actually initiated while we're executing this
-        if (!didReload && all->empty()) {
-            reload(opCtx);
-            getAllShardIdsNoReload(all);
+ShardRegistry::Cache::LookupResult ShardRegistry::_lookup(OperationContext* opCtx,
+                                                          const Singleton& key,
+                                                          const Cache::ValueHandle& cachedData,
+                                                          const Time& timeInStore) {
+    invariant(key == _kSingleton);
+    invariant(cachedData, "ShardRegistry::_lookup called but the cache is empty");
+
+    LOGV2_DEBUG(4620250,
+                2,
+                "Starting ShardRegistry::_lookup",
+                "cachedData"_attr = cachedData->toBSON(),
+                "cachedData.getTime()"_attr = cachedData.getTime().toBSON(),
+                "timeInStore"_attr = timeInStore.toBSON());
+
+    // Check if we need to refresh from the configsvrs.  If so, then do that and get the results,
+    // otherwise (this is a lookup only to incorporate updated connection strings from the RSM),
+    // then get the equivalent values from the previously cached data.
+    auto [returnData,
+          returnTopologyTime,
+          returnForceReloadIncrement,
+          removedShards,
+          fetchedFromConfigServers] = [&]()
+        -> std::tuple<ShardRegistryData, Timestamp, Increment, ShardRegistryData::ShardMap, bool> {
+        if (timeInStore.topologyTime > cachedData.getTime().topologyTime ||
+            timeInStore.forceReloadIncrement > cachedData.getTime().forceReloadIncrement) {
+            auto [reloadedData, maxTopologyTime] =
+                ShardRegistryData::createFromCatalogClient(opCtx, _shardFactory.get());
+            if (!useActualTopologyTime()) {
+                // If not using the actual topology time, then just use the topologyTime currently
+                // in the cache, instead of the maximum topologyTime value from config.shards.  This
+                // is necessary during upgrade/downgrade when topologyTime might not be gossiped by
+                // all nodes (and so isn't being used).
+                maxTopologyTime = cachedData.getTime().topologyTime;
+            }
+
+            auto [mergedData, removedShards] =
+                ShardRegistryData::mergeExisting(*cachedData, reloadedData);
+
+            return {
+                mergedData, maxTopologyTime, timeInStore.forceReloadIncrement, removedShards, true};
+        } else {
+            return {*cachedData,
+                    cachedData.getTime().topologyTime,
+                    cachedData.getTime().forceReloadIncrement,
+                    {},
+                    false};
         }
-    }
-}
+    }();
 
-int ShardRegistry::getNumShards() const {
-    std::set<ShardId> seen;
-    {
-        stdx::lock_guard<Latch> lk(_mutex);
-        _data.getAllShardIds(seen);
-    }
-    return seen.size();
-}
+    // Always apply the latest conn strings.
+    auto [latestConnStrings, rsmIncrementForConnStrings] = _getLatestConnStrings();
 
-void ShardRegistry::updateReplSetHosts(const ConnectionString& newConnString) {
-    invariant(newConnString.type() == ConnectionString::SET ||
-              newConnString.type() == ConnectionString::CUSTOM);  // For dbtests
+    for (const auto& latestConnString : latestConnStrings) {
+        // TODO SERVER-50909: Optimise by only doing this work if the latest conn string differs.
 
-    // to prevent update config shard connection string during init
-    stdx::unique_lock<Latch> lock(_mutex);
+        auto shard = returnData.findByRSName(latestConnString.first.toString());
+        if (!shard) {
+            continue;
+        }
 
-    auto shard = _data.findByRSName(newConnString.getSetName());
-    if (!shard) {
-        return;
+        auto newData = ShardRegistryData::createFromExisting(
+            returnData, latestConnString.second, _shardFactory.get());
+        returnData = newData;
     }
 
-    auto [data, updatedShard] =
-        ShardRegistryData::createFromExisting(_data, newConnString, _shardFactory.get());
+    // Remove RSMs that are not in the catalog any more.
+    for (auto& pair : removedShards) {
+        auto& shardId = pair.first;
+        auto& shard = pair.second;
+        invariant(shard);
 
-    if (updatedShard && updatedShard->isConfig()) {
-        _configShard = updatedShard;
+        auto name = shard->getConnString().getSetName();
+        ReplicaSetMonitor::remove(name);
+        for (auto& callback : _shardRemovalHooks) {
+            // Run callbacks asynchronously.
+            // TODO SERVER-50906: Consider running these callbacks synchronously.
+            ExecutorFuture<void>(Grid::get(opCtx)->getExecutorPool()->getFixedExecutor())
+                .getAsync([=](const Status&) { callback(shardId); });
+        }
     }
 
-    _data = data;
-}
-
-void ShardRegistry::init() {
-    invariant(!_isInitialized.load());
-    {
-        stdx::unique_lock<Latch> lock(_mutex);
-        _configShard =
-            _shardFactory->createShard(ShardRegistry::kConfigServerShardId, _initConfigServerCS);
-        _data = ShardRegistryData::createWithConfigShardOnly(_configShard);
+    // The registry is "up" once there has been a successful lookup from the config servers.
+    if (fetchedFromConfigServers) {
+        _isUp.store(true);
     }
-    _isInitialized.store(true);
+
+    Time returnTime{returnTopologyTime, rsmIncrementForConnStrings, returnForceReloadIncrement};
+    LOGV2_DEBUG(4620251,
+                2,
+                "Finished ShardRegistry::_lookup",
+                "returnData"_attr = returnData.toBSON(),
+                "returnTime"_attr = returnTime);
+    return Cache::LookupResult{returnData, returnTime};
 }
 
-void ShardRegistry::startup(OperationContext* opCtx) {
-    // startup() must be called only once
+void ShardRegistry::startupPeriodicReloader(OperationContext* opCtx) {
+    invariant(_isInitialized.load());
+    // startupPeriodicReloader() must be called only once
     invariant(!_executor);
 
     auto hookList = std::make_unique<rpc::EgressMetadataHookList>();
@@ -240,16 +222,17 @@ void ShardRegistry::startup(OperationContext* opCtx) {
     // construct task executor
     auto net = executor::makeNetworkInterface("ShardRegistryUpdater", nullptr, std::move(hookList));
     auto netPtr = net.get();
-    _executor = std::make_unique<ThreadPoolTaskExecutor>(
-        std::make_unique<NetworkInterfaceThreadPool>(netPtr), std::move(net));
+    _executor = std::make_unique<executor::ThreadPoolTaskExecutor>(
+        std::make_unique<executor::NetworkInterfaceThreadPool>(netPtr), std::move(net));
     LOGV2_DEBUG(22724, 1, "Starting up task executor for periodic reloading of ShardRegistry");
     _executor->startup();
 
     auto status =
-        _executor->scheduleWork([this](const CallbackArgs& cbArgs) { _internalReload(cbArgs); });
+        _executor->scheduleWork([this](const CallbackArgs& cbArgs) { _periodicReload(cbArgs); });
 
     if (status.getStatus() == ErrorCodes::ShutdownInProgress) {
-        LOGV2_DEBUG(22725, 1, "Cant schedule Shard Registry reload. Executor shutdown in progress");
+        LOGV2_DEBUG(
+            22725, 1, "Can't schedule Shard Registry reload. Executor shutdown in progress");
         return;
     }
 
@@ -261,7 +244,27 @@ void ShardRegistry::startup(OperationContext* opCtx) {
     }
 }
 
-void ShardRegistry::_internalReload(const CallbackArgs& cbArgs) {
+void ShardRegistry::shutdownPeriodicReloader() {
+    if (_executor) {
+        LOGV2_DEBUG(22723, 1, "Shutting down task executor for reloading shard registry");
+        _executor->shutdown();
+        _executor->join();
+        _executor.reset();
+    }
+}
+
+void ShardRegistry::shutdown() {
+    shutdownPeriodicReloader();
+
+    if (!_isShutdown.load()) {
+        LOGV2_DEBUG(4620235, 1, "Shutting down shard registry");
+        _threadPool.shutdown();
+        _threadPool.join();
+        _isShutdown.store(true);
+    }
+}
+
+void ShardRegistry::_periodicReload(const CallbackArgs& cbArgs) {
     LOGV2_DEBUG(22726, 1, "Reloading shardRegistry");
     if (!cbArgs.status.isOK()) {
         LOGV2_WARNING(22734,
@@ -275,21 +278,26 @@ void ShardRegistry::_internalReload(const CallbackArgs& cbArgs) {
 
     auto opCtx = tc->makeOperationContext();
 
+    auto refreshPeriod = kRefreshPeriod;
+
     try {
         reload(opCtx.get());
     } catch (const DBException& e) {
+        if (e.code() == ErrorCodes::ReadConcernMajorityNotAvailableYet) {
+            refreshPeriod = Seconds(1);
+        }
         LOGV2(22727,
               "Error running periodic reload of shard registry caused by {error}; will retry after "
               "{shardRegistryReloadInterval}",
               "Error running periodic reload of shard registry",
               "error"_attr = redact(e),
-              "shardRegistryReloadInterval"_attr = kRefreshPeriod);
+              "shardRegistryReloadInterval"_attr = refreshPeriod);
     }
 
     // reschedule itself
     auto status =
-        _executor->scheduleWorkAt(_executor->now() + kRefreshPeriod,
-                                  [this](const CallbackArgs& cbArgs) { _internalReload(cbArgs); });
+        _executor->scheduleWorkAt(_executor->now() + refreshPeriod,
+                                  [this](const CallbackArgs& cbArgs) { _periodicReload(cbArgs); });
 
     if (status.getStatus() == ErrorCodes::ShutdownInProgress) {
         LOGV2_DEBUG(
@@ -305,85 +313,155 @@ void ShardRegistry::_internalReload(const CallbackArgs& cbArgs) {
     }
 }
 
-bool ShardRegistry::isUp() const {
-    return _isUp.load();
+ConnectionString ShardRegistry::getConfigServerConnectionString() const {
+    return getConfigShard()->getConnString();
 }
 
-bool ShardRegistry::reload(OperationContext* opCtx) {
-    stdx::unique_lock<Latch> reloadLock(_reloadMutex);
-
-    if (_reloadState == ReloadState::Reloading) {
-        // Another thread is already in the process of reloading so no need to do duplicate work.
-        // There is also an issue if multiple threads are allowed to call getAllShards()
-        // simultaneously because there is no good way to determine which of the threads has the
-        // more recent version of the data.
-        try {
-            opCtx->waitForConditionOrInterrupt(
-                _inReloadCV, reloadLock, [&] { return _reloadState != ReloadState::Reloading; });
-        } catch (const DBException& e) {
-            LOGV2_DEBUG(22729,
-                        1,
-                        "Error reloading shard registry caused by {error}",
-                        "Error reloading shard registry",
-                        "error"_attr = redact(e));
-            return false;
-        }
+std::shared_ptr<Shard> ShardRegistry::getConfigShard() const {
+    stdx::lock_guard<Latch> lk(_mutex);
+    return _configShardData.findShard(kConfigServerShardId);
+}
 
-        if (_reloadState == ReloadState::Idle) {
-            return false;
+StatusWith<std::shared_ptr<Shard>> ShardRegistry::getShard(OperationContext* opCtx,
+                                                           const ShardId& shardId) {
+    // First check if this is a config shard lookup.
+    {
+        stdx::lock_guard<Latch> lk(_mutex);
+        if (auto shard = _configShardData.findShard(shardId)) {
+            return shard;
         }
-        // else proceed to reload since an error occured on the last reload attempt.
-        invariant(_reloadState == ReloadState::Failed);
     }
 
-    _reloadState = ReloadState::Reloading;
-    reloadLock.unlock();
+    if (auto shard = _getData(opCtx)->findShard(shardId)) {
+        return shard;
+    }
 
-    auto nextReloadState = ReloadState::Failed;
+    // Reload and try again if the shard was not in the registry
+    reload(opCtx);
+    if (auto shard = _getData(opCtx)->findShard(shardId)) {
+        return shard;
+    }
 
-    auto failGuard = makeGuard([&] {
-        if (!reloadLock.owns_lock()) {
-            reloadLock.lock();
-        }
-        _reloadState = nextReloadState;
-        _inReloadCV.notify_all();
-    });
+    return {ErrorCodes::ShardNotFound, str::stream() << "Shard " << shardId << " not found"};
+}
 
-    ShardRegistryData reloadedData =
-        ShardRegistryData::createFromCatalogClient(opCtx, _shardFactory.get(), getConfigShard());
+void ShardRegistry::getAllShardIds(OperationContext* opCtx, std::vector<ShardId>* all) {
+    std::set<ShardId> seen;
+    auto data = _getData(opCtx);
+    data->getAllShardIds(seen);
+    if (seen.empty()) {
+        reload(opCtx);
+        data = _getData(opCtx);
+        data->getAllShardIds(seen);
+    }
+    all->assign(seen.begin(), seen.end());
+}
 
+int ShardRegistry::getNumShards(OperationContext* opCtx) {
+    std::set<ShardId> seen;
+    auto data = _getData(opCtx);
+    data->getAllShardIds(seen);
+    return seen.size();
+}
+
+std::pair<std::vector<ShardRegistry::LatestConnStrings::value_type>, ShardRegistry::Increment>
+ShardRegistry::_getLatestConnStrings() const {
     stdx::unique_lock<Latch> lock(_mutex);
+    return {{_latestConnStrings.begin(), _latestConnStrings.end()}, _rsmIncrement.load()};
+}
 
-    auto [mergedData, removedShards] = ShardRegistryData::mergeExisting(_data, reloadedData);
-    _data = std::move(mergedData);
+void ShardRegistry::updateReplSetHosts(const ConnectionString& newConnString) {
+    invariant(newConnString.type() == ConnectionString::SET ||
+              newConnString.type() == ConnectionString::CUSTOM);  // For dbtests
 
-    lock.unlock();
+    stdx::lock_guard<Latch> lk(_mutex);
+    if (auto shard = _configShardData.findByRSName(newConnString.getSetName())) {
+        auto newData = ShardRegistryData::createFromExisting(
+            _configShardData, newConnString, _shardFactory.get());
+        _configShardData = newData;
+
+    } else {
+        // Stash the new connection string and bump the RSM increment.
+        _latestConnStrings[newConnString.getSetName()] = newConnString;
+        auto value = _rsmIncrement.addAndFetch(1);
+        LOGV2_DEBUG(4620252,
+                    2,
+                    "ShardRegistry stashed new connection string",
+                    "newConnString"_attr = newConnString,
+                    "newRSMIncrement"_attr = value);
+    }
+
+    // Schedule a lookup, to incorporate the new connection string.
+    // TODO SERVER-50910: To avoid needing to use a separate thread to schedule the lookup, make
+    // _getData() async.
+    auto status = Grid::get(_service)->getExecutorPool()->getFixedExecutor()->scheduleWork(
+        [this](const CallbackArgs& cbArgs) {
+            ThreadClient tc("shard-registry-rsm-reload", _service);
+
+            auto opCtx = tc->makeOperationContext();
+
+            try {
+                _getData(opCtx.get());
+            } catch (const DBException& e) {
+                LOGV2(4620201,
+                      "Error running reload of ShardRegistry for RSM update, caused by {error}",
+                      "Error running reload of ShardRegistry for RSM update",
+                      "error"_attr = redact(e));
+            }
+        });
 
-    // Remove RSMs that are not in the catalog any more.
-    for (auto& pair : removedShards) {
-        auto& shardId = pair.first;
-        auto& shard = pair.second;
-        invariant(shard);
+    if (status.getStatus() == ErrorCodes::ShutdownInProgress) {
+        LOGV2_DEBUG(
+            4620202,
+            1,
+            "Can't schedule ShardRegistry reload for RSM update, executor shutdown in progress");
+        return;
+    }
 
-        auto name = shard->getConnString().getSetName();
-        ReplicaSetMonitor::remove(name);
-        for (auto& callback : _shardRemovalHooks) {
-            // Run callbacks asynchronously.
-            ExecutorFuture<void>(Grid::get(opCtx)->getExecutorPool()->getFixedExecutor())
-                .getAsync([=](const Status&) { callback(shardId); });
-        }
+    if (!status.isOK()) {
+        LOGV2_FATAL(4620203,
+                    "Error scheduling ShardRegistry reload for RSM update, caused by {error}",
+                    "Error scheduling ShardRegistry reload for RSM update",
+                    "error"_attr = redact(status.getStatus()));
+    }
+}
+
+std::unique_ptr<Shard> ShardRegistry::createConnection(const ConnectionString& connStr) const {
+    return _shardFactory->createUniqueShard(ShardId("<unnamed>"), connStr);
+}
+
+bool ShardRegistry::isUp() const {
+    return _isUp.load();
+}
+
+void ShardRegistry::toBSON(BSONObjBuilder* result) const {
+    BSONObjBuilder map;
+    BSONObjBuilder hosts;
+    BSONObjBuilder connStrings;
+    auto data = _getCachedData();
+    data->toBSON(&map, &hosts, &connStrings);
+    {
+        stdx::lock_guard<Latch> lk(_mutex);
+        _configShardData.toBSON(&map, &hosts, &connStrings);
     }
+    result->append("map", map.obj());
+    result->append("hosts", hosts.obj());
+    result->append("connStrings", connStrings.obj());
+}
+
+bool ShardRegistry::reload(OperationContext* opCtx) {
+    // Make the next acquire do a lookup.
+    auto value = _forceReloadIncrement.addAndFetch(1);
+    LOGV2_DEBUG(4620253, 2, "Forcing ShardRegistry reload", "newForceReloadIncrement"_attr = value);
+
+    // Force it to actually happen now.
+    _getData(opCtx);
 
-    nextReloadState = ReloadState::Idle;
-    // first successful reload means that registry is up
-    _isUp.store(true);
     return true;
 }
 
 void ShardRegistry::clearEntries() {
-    ShardRegistryData empty;
-    stdx::lock_guard<Latch> lk(_mutex);
-    _data = empty;
+    _cache->invalidateAll();
 }
 
 void ShardRegistry::updateReplicaSetOnConfigServer(ServiceContext* serviceContext,
@@ -393,7 +471,8 @@ void ShardRegistry::updateReplicaSetOnConfigServer(ServiceContext* serviceContex
     auto opCtx = tc->makeOperationContext();
     auto const grid = Grid::get(opCtx.get());
 
-    std::shared_ptr<Shard> s = grid->shardRegistry()->lookupRSName(connStr.getSetName());
+    std::shared_ptr<Shard> s =
+        grid->shardRegistry()->_getShardForRSNameNoReload(connStr.getSetName());
     if (!s) {
         LOGV2_DEBUG(22730,
                     1,
@@ -427,25 +506,93 @@ void ShardRegistry::updateReplicaSetOnConfigServer(ServiceContext* serviceContex
     }
 }
 
-void ShardRegistry::toBSON(BSONObjBuilder* result) const {
-    std::vector<std::shared_ptr<Shard>> shards;
+// Inserts the initial empty ShardRegistryData into the cache, if the cache is empty.
+void ShardRegistry::_initializeCacheIfNecessary() const {
+    if (!_cache->peekLatestCached(_kSingleton)) {
+        stdx::lock_guard<Latch> lk(_mutex);
+        if (!_cache->peekLatestCached(_kSingleton)) {
+            _cache->insertOrAssign(_kSingleton, {}, Date_t::now(), Time());
+        }
+    }
+}
+
+ShardRegistry::Cache::ValueHandle ShardRegistry::_getData(OperationContext* opCtx) {
+    _initializeCacheIfNecessary();
+
+    // If the forceReloadIncrement is 0, then we've never done a lookup, so we should be sure to do
+    // one now.
+    Increment uninitializedIncrement{0};
+    _forceReloadIncrement.compareAndSwap(&uninitializedIncrement, 1);
+
+    // Update the time the cache should be aiming for.
+    auto now = VectorClock::get(opCtx)->getTime();
+    // The topologyTime should be advanced to either the actual topologyTime (if it is being
+    // gossiped), or else the previously cached topologyTime value (so that this part of the cache's
+    // time doesn't advance, if topologyTime isn't being gossiped).
+    Timestamp topologyTime = useActualTopologyTime()
+        ? now.topologyTime().asTimestamp()
+        : _cache->peekLatestCached(_kSingleton).getTime().topologyTime;
+    _cache->advanceTimeInStore(
+        _kSingleton, Time(topologyTime, _rsmIncrement.load(), _forceReloadIncrement.load()));
+
+    return _cache->acquire(opCtx, _kSingleton, CacheCausalConsistency::kLatestKnown);
+}
+
+// TODO SERVER-50206: Remove usage of these non-causally consistent accessors.
+
+ShardRegistry::Cache::ValueHandle ShardRegistry::_getCachedData() const {
+    _initializeCacheIfNecessary();
+    return _cache->peekLatestCached(_kSingleton);
+}
+
+std::shared_ptr<Shard> ShardRegistry::getShardNoReload(const ShardId& shardId) const {
+    // First check if this is a config shard lookup.
     {
         stdx::lock_guard<Latch> lk(_mutex);
-        _data.getAllShards(shards);
+        if (auto shard = _configShardData.findShard(shardId)) {
+            return shard;
+        }
     }
+    auto data = _getCachedData();
+    return data->findShard(shardId);
+}
 
-    std::sort(std::begin(shards),
-              std::end(shards),
-              [](std::shared_ptr<const Shard> lhs, std::shared_ptr<const Shard> rhs) {
-                  return lhs->getId() < rhs->getId();
-              });
+std::shared_ptr<Shard> ShardRegistry::getShardForHostNoReload(const HostAndPort& host) const {
+    // First check if this is a config shard lookup.
+    {
+        stdx::lock_guard<Latch> lk(_mutex);
+        if (auto shard = _configShardData.findByHostAndPort(host)) {
+            return shard;
+        }
+    }
+    auto data = _getCachedData();
+    return data->findByHostAndPort(host);
+}
 
-    BSONObjBuilder mapBob(result->subobjStart("map"));
-    for (auto&& shard : shards) {
-        // Intentionally calling getConnString while not holding _mutex
-        // because it can take ReplicaSetMonitor::SetState::mutex if it's ShardRemote.
-        mapBob.append(shard->getId(), shard->getConnString().toString());
+void ShardRegistry::getAllShardIdsNoReload(std::vector<ShardId>* all) const {
+    std::set<ShardId> seen;
+    auto data = _getCachedData();
+    data->getAllShardIds(seen);
+    all->assign(seen.begin(), seen.end());
+}
+
+int ShardRegistry::getNumShardsNoReload() const {
+    std::set<ShardId> seen;
+    auto data = _getCachedData();
+    data->getAllShardIds(seen);
+    return seen.size();
+}
+
+std::shared_ptr<Shard> ShardRegistry::_getShardForRSNameNoReload(const std::string& name) const {
+    // First check if this is a config shard lookup.
+    {
+        stdx::lock_guard<Latch> lk(_mutex);
+        if (auto shard = _configShardData.findByRSName(name)) {
+            return shard;
+        }
     }
+    auto data = _getCachedData();
+    return data->findByRSName(name);
 }
 
 ////////////// ShardRegistryData //////////////////
@@ -456,9 +603,8 @@ ShardRegistryData ShardRegistryData::createWithConfigShardOnly(std::shared_ptr<S
     return data;
 }
 
-ShardRegistryData ShardRegistryData::createFromCatalogClient(OperationContext* opCtx,
-                                                             ShardFactory* shardFactory,
-                                                             std::shared_ptr<Shard> configShard) {
+std::pair<ShardRegistryData, Timestamp> ShardRegistryData::createFromCatalogClient(
+    OperationContext* opCtx, ShardFactory* shardFactory) {
     auto const catalogClient = Grid::get(opCtx)->catalogClient();
 
     auto readConcern = repl::ReadConcernLevel::kMajorityReadConcern;
@@ -489,6 +635,7 @@ ShardRegistryData ShardRegistryData::createFromCatalogClient(OperationContext* o
     // Do this before re-taking the mutex to avoid deadlock with the ReplicaSetMonitor updating
     // hosts for a given shard.
     std::vector<std::tuple<std::string, ConnectionString>> shardsInfo;
+    Timestamp maxTopologyTime;
     for (const auto& shardType : shards) {
         // This validation should ideally go inside the ShardType::validate call. However, doing
         // it there would prevent us from loading previously faulty shard hosts, which might have
@@ -502,11 +649,15 @@ ShardRegistryData ShardRegistryData::createFromCatalogClient(OperationContext* o
             continue;
         }
 
+        if (auto thisTopologyTime = shardType.getTopologyTime();
+            maxTopologyTime < thisTopologyTime) {
+            maxTopologyTime = thisTopologyTime;
+        }
+
         shardsInfo.push_back(std::make_tuple(shardType.getName(), shardHostStatus.getValue()));
     }
 
     ShardRegistryData data;
-    data._addShard(configShard, true);
     for (auto& shardInfo : shardsInfo) {
         if (std::get<0>(shardInfo) == "config") {
             continue;
@@ -517,7 +668,7 @@ ShardRegistryData ShardRegistryData::createFromCatalogClient(OperationContext* o
 
         data._addShard(std::move(shard), false);
     }
-    return data;
+    return {data, maxTopologyTime};
 }
 
 std::pair<ShardRegistryData, ShardRegistryData::ShardMap> ShardRegistryData::mergeExisting(
@@ -550,21 +701,20 @@ std::pair<ShardRegistryData, ShardRegistryData::ShardMap> ShardRegistryData::mer
     return {mergedData, removedShards};
 }
 
-std::pair<ShardRegistryData, std::shared_ptr<Shard>> ShardRegistryData::createFromExisting(
-    const ShardRegistryData& existingData,
-    const ConnectionString& newConnString,
-    ShardFactory* shardFactory) {
+ShardRegistryData ShardRegistryData::createFromExisting(const ShardRegistryData& existingData,
+                                                        const ConnectionString& newConnString,
+                                                        ShardFactory* shardFactory) {
     ShardRegistryData data(existingData);
 
     auto it = data._rsLookup.find(newConnString.getSetName());
     if (it == data._rsLookup.end()) {
-        return {data, nullptr};
+        return data;
     }
     invariant(it->second);
     auto updatedShard = shardFactory->createShard(it->second->getId(), newConnString);
     data._addShard(updatedShard, true);
 
-    return {data, updatedShard};
+    return data;
 }
 
 std::shared_ptr<Shard> ShardRegistryData::findByRSName(const std::string& name) const {
@@ -682,4 +832,70 @@ void ShardRegistryData::_addShard(std::shared_ptr<Shard> shard, bool useOriginal
     }
 }
 
+void ShardRegistryData::toBSON(BSONObjBuilder* map,
+                               BSONObjBuilder* hosts,
+                               BSONObjBuilder* connStrings) const {
+    std::vector<std::shared_ptr<Shard>> shards;
+    getAllShards(shards);
+
+    std::sort(std::begin(shards),
+              std::end(shards),
+              [](std::shared_ptr<const Shard> lhs, std::shared_ptr<const Shard> rhs) {
+                  return lhs->getId() < rhs->getId();
+              });
+
+    if (map) {
+        for (auto&& shard : shards) {
+            map->append(shard->getId(), shard->getConnString().toString());
+        }
+    }
+
+    if (hosts) {
+        for (const auto& hostIt : _hostLookup) {
+            hosts->append(hostIt.first.toString(), hostIt.second->getId());
+        }
+    }
+
+    if (connStrings) {
+        for (const auto& connStringIt : _connStringLookup) {
+            connStrings->append(connStringIt.first.toString(), connStringIt.second->getId());
+        }
+    }
+}
+
+void ShardRegistryData::toBSON(BSONObjBuilder* result) const {
+    std::vector<std::shared_ptr<Shard>> shards;
+    getAllShards(shards);
+
+    std::sort(std::begin(shards),
+              std::end(shards),
+              [](std::shared_ptr<const Shard> lhs, std::shared_ptr<const Shard> rhs) {
+                  return lhs->getId() < rhs->getId();
+              });
+
+    BSONObjBuilder mapBob(result->subobjStart("map"));
+    for (auto&& shard : shards) {
+        mapBob.append(shard->getId(), shard->getConnString().toString());
+    }
+    mapBob.done();
+
+    BSONObjBuilder hostsBob(result->subobjStart("hosts"));
+    for (const auto& hostIt : _hostLookup) {
+        hostsBob.append(hostIt.first.toString(), hostIt.second->getId());
+    }
+    hostsBob.done();
+
+    BSONObjBuilder connStringsBob(result->subobjStart("connStrings"));
+    for (const auto& connStringIt : _connStringLookup) {
+        connStringsBob.append(connStringIt.first.toString(), connStringIt.second->getId());
+    }
+    connStringsBob.done();
+}
+
+BSONObj ShardRegistryData::toBSON() const {
+    BSONObjBuilder bob;
+    toBSON(&bob);
+    return bob.obj();
+}
+
 }  // namespace mongo
diff --git a/src/mongo/s/client/shard_registry.h b/src/mongo/s/client/shard_registry.h
index 486788c57dd..711c8e0d34b 100644
--- a/src/mongo/s/client/shard_registry.h
+++ b/src/mongo/s/client/shard_registry.h
@@ -29,30 +29,24 @@
 
 #pragma once
 
-#include <memory>
-#include <set>
 #include <string>
 #include <vector>
 
-#include "mongo/db/jsobj.h"
+#include "mongo/bson/bsonobjbuilder.h"
+#include "mongo/client/connection_string.h"
+#include "mongo/db/operation_context.h"
+#include "mongo/db/server_options.h"
 #include "mongo/executor/task_executor.h"
 #include "mongo/platform/mutex.h"
 #include "mongo/s/client/shard.h"
-#include "mongo/stdx/condition_variable.h"
+#include "mongo/s/client/shard_factory.h"
 #include "mongo/stdx/unordered_map.h"
-#include "mongo/util/concurrency/with_lock.h"
+#include "mongo/util/concurrency/thread_pool.h"
+#include "mongo/util/net/hostandport.h"
+#include "mongo/util/read_through_cache.h"
 
 namespace mongo {
 
-class BSONObjBuilder;
-struct HostAndPort;
-class NamespaceString;
-class OperationContext;
-class ServiceContext;
-class ShardFactory;
-class Shard;
-class ShardType;
-
 class ShardRegistryData {
 public:
     using ShardMap = stdx::unordered_map<ShardId, std::shared_ptr<Shard>, ShardId::Hasher>;
@@ -67,9 +61,8 @@ public:
     /**
      * Reads shards docs from the catalog client and fills in maps.
      */
-    static ShardRegistryData createFromCatalogClient(OperationContext* opCtx,
-                                                     ShardFactory* shardFactory,
-                                                     std::shared_ptr<Shard> configShard);
+    static std::pair<ShardRegistryData, Timestamp> createFromCatalogClient(
+        OperationContext* opCtx, ShardFactory* shardFactory);
 
     /**
      * Merges alreadyCachedData and configServerData into a new ShardRegistryData.
@@ -93,10 +86,9 @@ public:
      * Create a duplicate of existingData, but additionally updates the shard for newConnString.
      * Used when notified by the RSM of a new connection string from a shard.
      */
-    static std::pair<ShardRegistryData, std::shared_ptr<Shard>> createFromExisting(
-        const ShardRegistryData& existingData,
-        const ConnectionString& newConnString,
-        ShardFactory* shardFactory);
+    static ShardRegistryData createFromExisting(const ShardRegistryData& existingData,
+                                                const ConnectionString& newConnString,
+                                                ShardFactory* shardFactory);
 
     /**
      * Returns the shard with the given shard id, connection string, or host and port.
@@ -128,6 +120,10 @@ public:
      */
     void getAllShards(std::vector<std::shared_ptr<Shard>>& result) const;
 
+    void toBSON(BSONObjBuilder* result) const;
+    void toBSON(BSONObjBuilder* map, BSONObjBuilder* hosts, BSONObjBuilder* connStrings) const;
+    BSONObj toBSON() const;
+
 private:
     /**
      * Returns the shard with the given shard id, or nullptr if no such shard.
@@ -198,70 +194,69 @@ public:
     ~ShardRegistry();
 
     /**
-     *  Starts ReplicaSetMonitor by adding a config shard.
+     * Initializes ShardRegistry with config shard. Must be called outside c-tor to avoid calls on
+     * this while its still not fully constructed.
      */
-    void startup(OperationContext* opCtx);
+    void init(ServiceContext* service);
 
     /**
-     * This is invalid to use on the config server and will hit an invariant if it is done.
-     * If the config server has need of a connection string for itself, it should get it from the
-     * replication state.
-     *
-     * Returns the connection string for the config server.
+     * Startup the periodic reloader of the ShardRegistry.
+     * Can be called only after ShardRegistry::init()
      */
-    ConnectionString getConfigServerConnectionString() const;
+    void startupPeriodicReloader(OperationContext* opCtx);
 
     /**
-     * Reloads the ShardRegistry based on the contents of the config server's config.shards
-     * collection. Returns true if this call performed a reload and false if this call only waited
-     * for another thread to perform the reload and did not actually reload. Because of this, it is
-     * possible that calling reload once may not result in the most up to date view. If strict
-     * reloading is required, the caller should call this method one more time if the first call
-     * returned false.
+     * Shutdown the periodic reloader of the ShardRegistry.
      */
-    bool reload(OperationContext* opCtx);
+    void shutdownPeriodicReloader();
 
     /**
-     * Clears all entries from the shard registry entries, which will force the registry to do a
-     * reload on next access.
+     * Shuts down the threadPool. Needs to be called explicitly because ShardRegistry is never
+     * destroyed as it's owned by the static grid object.
      */
-    void clearEntries();
+    void shutdown();
 
     /**
-     * Takes a connection string describing either a shard or config server replica set, looks
-     * up the corresponding Shard object based on the replica set name, then updates the
-     * ShardRegistry's notion of what hosts make up that shard.
+     * This is invalid to use on the config server and will hit an invariant if it is done.
+     * If the config server has need of a connection string for itself, it should get it from the
+     * replication state.
+     *
+     * Returns the connection string for the config server.
      */
-    void updateReplSetHosts(const ConnectionString& newConnString);
+    ConnectionString getConfigServerConnectionString() const;
+
+    /**
+     * Returns shared pointer to the shard object representing the config servers.
+     *
+     * The config shard is always known, so this function never blocks.
+     */
+    std::shared_ptr<Shard> getConfigShard() const;
 
     /**
      * Returns a shared pointer to the shard object with the given shard id, or ShardNotFound error
      * otherwise.
      *
      * May refresh the shard registry if there's no cached information about the shard. The shardId
-     * parameter can actually be the shard name or the HostAndPort for any
-     * server in the shard.
+     * parameter can actually be the shard name or the HostAndPort for any server in the shard.
      */
     StatusWith<std::shared_ptr<Shard>> getShard(OperationContext* opCtx, const ShardId& shardId);
 
     /**
-     * Returns a shared pointer to the shard object with the given shard id. The shardId parameter
-     * can actually be the shard name or the HostAndPort for any server in the shard. Will not
-     * refresh the shard registry or otherwise perform any network traffic. This means that if the
-     * shard was recently added it may not be found.  USE WITH CAUTION.
+     * Populates all known shard ids into the given vector.
      */
-    std::shared_ptr<Shard> getShardNoReload(const ShardId& shardId);
+    void getAllShardIds(OperationContext* opCtx, std::vector<ShardId>* all);
 
     /**
-     * Finds the Shard that the mongod listening at this HostAndPort is a member of. Will not
-     * refresh the shard registry or otherwise perform any network traffic.
+     * Returns the number of shards.
      */
-    std::shared_ptr<Shard> getShardForHostNoReload(const HostAndPort& shardHost);
+    int getNumShards(OperationContext* opCtx);
 
     /**
-     * Returns shared pointer to the shard object representing the config servers.
+     * Takes a connection string describing either a shard or config server replica set, looks
+     * up the corresponding Shard object based on the replica set name, then updates the
+     * ShardRegistry's notion of what hosts make up that shard.
      */
-    std::shared_ptr<Shard> getConfigShard() const;
+    void updateReplSetHosts(const ConnectionString& newConnString);
 
     /**
      * Instantiates a new detached shard connection, which does not appear in the list of shards
@@ -274,36 +269,28 @@ public:
     std::unique_ptr<Shard> createConnection(const ConnectionString& connStr) const;
 
     /**
-     * Lookup shard by replica set name. Returns nullptr if the name can't be found.
-     * Note: this doesn't refresh the table if the name isn't found, so it's possible that a
-     * newly added shard/Replica Set may not be found.
+     * The ShardRegistry is "up" once a successful lookup from the config servers has been
+     * completed.
      */
-    std::shared_ptr<Shard> lookupRSName(const std::string& name) const;
-
-    void getAllShardIdsNoReload(std::vector<ShardId>* all) const;
-
-    /**
-     * Like getAllShardIdsNoReload(), but does a reload internally in the case that
-     * getAllShardIdsNoReload() comes back empty
-     */
-    void getAllShardIds(OperationContext* opCtx, std::vector<ShardId>* all);
-
-    int getNumShards() const;
+    bool isUp() const;
 
     void toBSON(BSONObjBuilder* result) const;
-    bool isUp() const;
 
     /**
-     * Initializes ShardRegistry with config shard. Must be called outside c-tor to avoid calls on
-     * this while its still not fully constructed.
+     * Reloads the ShardRegistry based on the contents of the config server's config.shards
+     * collection. Returns true if this call performed a reload and false if this call only waited
+     * for another thread to perform the reload and did not actually reload. Because of this, it is
+     * possible that calling reload once may not result in the most up to date view. If strict
+     * reloading is required, the caller should call this method one more time if the first call
+     * returned false.
      */
-    void init();
+    bool reload(OperationContext* opCtx);
 
     /**
-     * Shuts down _executor. Needs to be called explicitly because ShardRegistry is never destroyed
-     * as it's owned by the static grid object.
+     * Clears all entries from the shard registry entries, which will force the registry to do a
+     * reload on next access.
      */
-    void shutdown();
+    void clearEntries();
 
     /**
      * For use in mongos which needs notifications about changes to shard replset membership to
@@ -312,8 +299,136 @@ public:
     static void updateReplicaSetOnConfigServer(ServiceContext* serviceContex,
                                                const ConnectionString& connStr) noexcept;
 
+    // TODO SERVER-50206: Remove usage of these non-causally consistent accessors.
+    //
+    // Their most important current users are dispatching requests to hosts, and processing
+    // responses from hosts.  These contexts need to know the shard that the host is associated
+    // with, but usually have no access to any associated opCtx (if there even is one), and also
+    // cannot tolerate waiting for further network activity (if the cache is stale and needs to be
+    // refreshed via _lookup()).
+
+    /**
+     * Returns a shared pointer to the shard object with the given shard id. The shardId parameter
+     * can actually be the shard name or the HostAndPort for any server in the shard. Will not
+     * refresh the shard registry or otherwise perform any network traffic. This means that if the
+     * shard was recently added it may not be found.  USE WITH CAUTION.
+     */
+    std::shared_ptr<Shard> getShardNoReload(const ShardId& shardId) const;
+
+    /**
+     * Finds the Shard that the mongod listening at this HostAndPort is a member of. Will not
+     * refresh the shard registry or otherwise perform any network traffic.
+     */
+    std::shared_ptr<Shard> getShardForHostNoReload(const HostAndPort& shardHost) const;
+
+    void getAllShardIdsNoReload(std::vector<ShardId>* all) const;
+
+    int getNumShardsNoReload() const;
+
 private:
-    void _internalReload(const executor::TaskExecutor::CallbackArgs& cbArgs);
+    /**
+     * The ShardRegistry uses the ReadThroughCache to handle refreshing itself.  The cache stores
+     * a single entry, with key of Singleton, value of ShardRegistryData, and causal-consistency
+     * time which is primarily Timestamp (based on the TopologyTime), but with additional
+     * "increment"s that are used to flag additional refresh criteria.
+     */
+
+    using Increment = int64_t;
+
+    struct Time {
+        explicit Time() {}
+
+        explicit Time(Timestamp topologyTime,
+                      Increment rsmIncrement,
+                      Increment forceReloadIncrement)
+            : topologyTime(topologyTime),
+              rsmIncrement(rsmIncrement),
+              forceReloadIncrement(forceReloadIncrement) {}
+
+        bool operator==(const Time& other) const {
+            return topologyTime == other.topologyTime && rsmIncrement == other.rsmIncrement &&
+                forceReloadIncrement == other.forceReloadIncrement;
+        }
+        bool operator!=(const Time& other) const {
+            return !(*this == other);
+        }
+        bool operator>(const Time& other) const {
+            return topologyTime > other.topologyTime || rsmIncrement > other.rsmIncrement ||
+                forceReloadIncrement > other.forceReloadIncrement;
+        }
+        bool operator>=(const Time& other) const {
+            return (*this > other) || (*this == other);
+        }
+        bool operator<(const Time& other) const {
+            return !(*this >= other);
+        }
+        bool operator<=(const Time& other) const {
+            return !(*this > other);
+        }
+
+        BSONObj toBSON() const {
+            BSONObjBuilder bob;
+            bob.append("topologyTime", topologyTime);
+            bob.append("rsmIncrement", rsmIncrement);
+            bob.append("forceReloadIncrement", forceReloadIncrement);
+            return bob.obj();
+        }
+
+        Timestamp topologyTime;
+
+        // The increments are used locally to trigger the lookup function.
+        //
+        // The rsmIncrement is used to indicate that that there are stashed RSM updates that need to
+        // be incorporated.
+        //
+        // The forceReloadIncrement is used to indicate that the latest data should be fetched from
+        // the configsvrs (ie. when the topologyTime can't be used for this, eg. in the first
+        // lookup, and in contexts like unittests where topologyTime isn't gossipped but the
+        // ShardRegistry still needs to be reloaded).  This is how reload() is able to force a
+        // refresh from the config servers - incrementing the forceReloadIncrement causes the cache
+        // to call _lookup() (rather than having reload() attempt to do a synchronous refresh).
+        Increment rsmIncrement{0};
+        Increment forceReloadIncrement{0};
+    };
+
+    enum class Singleton { Only };
+    static constexpr auto _kSingleton = Singleton::Only;
+
+    using Cache = ReadThroughCache<Singleton, ShardRegistryData, Time>;
+
+    Cache::LookupResult _lookup(OperationContext* opCtx,
+                                const Singleton& key,
+                                const Cache::ValueHandle& cachedData,
+                                const Time& timeInStore);
+
+    /**
+     * Gets a causally-consistent (ie. latest-known) copy of the ShardRegistryData, refreshing from
+     * the config servers if necessary.
+     */
+    Cache::ValueHandle _getData(OperationContext* opCtx);
+
+    /**
+     * Gets the latest-cached copy of the ShardRegistryData.  Never fetches from the config servers.
+     * Only used by the "NoReload" accessors.
+     * TODO SERVER-50206: Remove usage of this non-causally consistent accessor.
+     */
+    Cache::ValueHandle _getCachedData() const;
+
+    /**
+     * Lookup shard by replica set name. Returns nullptr if the name can't be found.
+     * Note: this doesn't refresh the table if the name isn't found, so it's possible that a
+     * newly added shard/Replica Set may not be found.
+     * TODO SERVER-50206: Remove usage of this non-causally consistent accessor.
+     */
+    std::shared_ptr<Shard> _getShardForRSNameNoReload(const std::string& name) const;
+
+    using LatestConnStrings = stdx::unordered_map<ShardId, ConnectionString, ShardId::Hasher>;
+
+    std::pair<std::vector<LatestConnStrings::value_type>, Increment> _getLatestConnStrings() const;
+
+    void _initializeCacheIfNecessary() const;
+
+    void _periodicReload(const executor::TaskExecutor::CallbackArgs& cbArgs);
 
     /**
      * Factory to create shards.  Never changed after startup so safe to access outside of _mutex.
@@ -326,24 +441,37 @@ private:
      */
     const ConnectionString _initConfigServerCS;
 
-    AtomicWord<bool> _isInitialized{false};
-
     /**
      * A list of callbacks to be called asynchronously when it has been discovered that a shard was
      * removed.
      */
     const std::vector<ShardRemovalHook> _shardRemovalHooks;
 
-    // Protects the ShardRegistryData lookup maps in _data, and _configShard.
+    // Thread pool used when looking up new values for the cache (ie. in which _lookup() runs).
+    ThreadPool _threadPool;
+
+    // Executor for periodically reloading the registry (ie. in which _periodicReload() runs).
+    std::unique_ptr<executor::TaskExecutor> _executor{};
+
+    mutable Mutex _cacheMutex = MONGO_MAKE_LATCH("ShardRegistry::_cacheMutex");
+    std::unique_ptr<Cache> _cache;
+
+    // Counters for incrementing the rsmIncrement and forceReloadIncrement fields of the Time used
+    // by the _cache.  See the comments for these fields in the Time class above for an explanation
+    // of their purpose.
+    AtomicWord<Increment> _rsmIncrement{0};
+    AtomicWord<Increment> _forceReloadIncrement{0};
+
+    // Protects _configShardData, and _latestNewConnStrings.
     mutable Mutex _mutex = MONGO_MAKE_LATCH("ShardRegistry::_mutex");
 
-    ShardRegistryData _data;
+    // Store a reference to the configShard.
+    ShardRegistryData _configShardData;
 
-    // Store a separate reference to the configShard.
-    std::shared_ptr<Shard> _configShard;
+    // The key is replset name (the type is ShardId just to take advantage of its hasher).
+    LatestConnStrings _latestConnStrings;
 
-    // Executor for reloading.
-    std::unique_ptr<executor::TaskExecutor> _executor{};
+    AtomicWord<bool> _isInitialized{false};
 
     // The ShardRegistry is "up" once there has been a successful refresh.
     AtomicWord<bool> _isUp{false};
@@ -351,17 +479,7 @@ private:
     // Set to true in shutdown call to prevent calling it twice.
     AtomicWord<bool> _isShutdown{false};
 
-    // Protects the _reloadState during startup and refresh.
-    mutable Mutex _reloadMutex = MONGO_MAKE_LATCH("ShardRegistry::_reloadMutex");
-    stdx::condition_variable _inReloadCV;
-
-    enum class ReloadState {
-        Idle,       // no other thread is loading data from config server in reload().
-        Reloading,  // another thread is loading data from the config server in reload().
-        Failed,     // last call to reload() caused an error when contacting the config server.
-    };
-
-    ReloadState _reloadState{ReloadState::Idle};
+    ServiceContext* _service{nullptr};
 };
 
 }  // namespace mongo
diff --git a/src/mongo/s/grid.cpp b/src/mongo/s/grid.cpp
index c2e3cd011ac..ea28033590e 100644
--- a/src/mongo/s/grid.cpp
+++ b/src/mongo/s/grid.cpp
@@ -90,7 +90,7 @@ void Grid::init(std::unique_ptr<ShardingCatalogClient> catalogClient,
     _executorPool = std::move(executorPool);
     _network = network;
 
-    _shardRegistry->init();
+    _shardRegistry->init(grid.owner(this));
 }
 
 bool Grid::isShardingInitialized() const {
diff --git a/src/mongo/s/sharding_initialization.cpp b/src/mongo/s/sharding_initialization.cpp
index e3baf3d6e60..2b73154e767 100644
--- a/src/mongo/s/sharding_initialization.cpp
+++ b/src/mongo/s/sharding_initialization.cpp
@@ -218,7 +218,7 @@ Status initializeGlobalShardingState(OperationContext* opCtx,
                networkPtr);
 
     // The shard registry must be started once the grid is initialized
-    grid->shardRegistry()->startup(opCtx);
+    grid->shardRegistry()->startupPeriodicReloader(opCtx);
 
     // The catalog client must be started after the shard registry has been started up
     grid->catalogClient()->startup();
diff --git a/src/mongo/util/invalidating_lru_cache.h b/src/mongo/util/invalidating_lru_cache.h
index c8ead4adecc..2852313ef22 100644
--- a/src/mongo/util/invalidating_lru_cache.h
+++ b/src/mongo/util/invalidating_lru_cache.h
@@ -67,6 +67,18 @@ struct CacheNotCausallyConsistent {
 };
 
 /**
+ * Helper for determining if a given type is CacheNotCausallyConsistent or not.
+ */
+template <typename T>
+struct isCausallyConsistentImpl : std::true_type {};
+
+template <>
+struct isCausallyConsistentImpl<CacheNotCausallyConsistent> : std::false_type {};
+
+template <class T>
+inline constexpr bool isCausallyConsistent = isCausallyConsistentImpl<T>::value;
+
+/**
  * Specifies the desired causal consistency for calls to 'get' (and 'acquire', respectively in the
  * ReadThroughCache, which is its main consumer).
  */
@@ -200,12 +212,8 @@ public:
         // doesn't support pinning items. Their only usage must be in the authorization mananager
         // for the internal authentication user.
         explicit ValueHandle(Value&& value)
-            : _value(std::make_shared<StoredValue>(nullptr,
-                                                   0,
-                                                   boost::none,
-                                                   std::move(value),
-                                                   CacheNotCausallyConsistent(),
-                                                   CacheNotCausallyConsistent())) {}
+            : _value(std::make_shared<StoredValue>(
+                  nullptr, 0, boost::none, std::move(value), Time(), Time())) {}
 
         explicit ValueHandle(Value&& value, const Time& t)
             : _value(
@@ -218,9 +226,15 @@ public:
         }
 
         bool isValid() const {
+            invariant(bool(*this));
             return _value->isValid.loadRelaxed();
         }
 
+        const Time& getTime() const {
+            invariant(bool(*this));
+            return _value->time;
+        }
+
         Value* get() {
             invariant(bool(*this));
             return &_value->value;
@@ -260,13 +274,16 @@ public:
      * was called, it will become invalidated.
      *
      * The 'time' parameter is mandatory for causally-consistent caches, but not needed otherwise
-     * (since the time never changes). Using a default of '= CacheNotCausallyConsistent()' allows
-     * non-causally-consistent users to not have to pass a second parameter, but would fail
-     * compilation if causally-consistent users forget to pass it.
+     * (since the time never changes).
      */
-    void insertOrAssign(const Key& key,
-                        Value&& value,
-                        const Time& time = CacheNotCausallyConsistent()) {
+    void insertOrAssign(const Key& key, Value&& value) {
+        MONGO_STATIC_ASSERT_MSG(
+            !isCausallyConsistent<Time>,
+            "Time must be passed to insertOrAssign on causally consistent caches");
+        insertOrAssign(key, std::move(value), Time());
+    }
+
+    void insertOrAssign(const Key& key, Value&& value, const Time& time) {
         LockGuardWithPostUnlockDestructor guard(_mutex);
         Time currentTime, currentTimeInStore;
         _invalidate(&guard, key, _cache.find(key), &currentTime, &currentTimeInStore);
@@ -307,13 +324,16 @@ public:
      * destroyed.
      *
      * The 'time' parameter is mandatory for causally-consistent caches, but not needed otherwise
-     * (since the time never changes). Using a default of '= CacheNotCausallyConsistent()' allows
-     * non-causally-consistent users to not have to pass a second parameter, but would fail
-     * compilation if causally-consistent users forget to pass it.
+     * (since the time never changes).
      */
-    ValueHandle insertOrAssignAndGet(const Key& key,
-                                     Value&& value,
-                                     const Time& time = CacheNotCausallyConsistent()) {
+    ValueHandle insertOrAssignAndGet(const Key& key, Value&& value) {
+        MONGO_STATIC_ASSERT_MSG(
+            !isCausallyConsistent<Time>,
+            "Time must be passed to insertOrAssignAndGet on causally consistent caches");
+        return insertOrAssignAndGet(key, std::move(value), Time());
+    }
+
+    ValueHandle insertOrAssignAndGet(const Key& key, Value&& value, const Time& time) {
         LockGuardWithPostUnlockDestructor guard(_mutex);
         Time currentTime, currentTimeInStore;
         _invalidate(&guard, key, _cache.find(key), &currentTime, &currentTimeInStore);
diff --git a/src/mongo/util/read_through_cache.h b/src/mongo/util/read_through_cache.h
index 72b3e7a5771..32efbc576ae 100644
--- a/src/mongo/util/read_through_cache.h
+++ b/src/mongo/util/read_through_cache.h
@@ -152,6 +152,10 @@ public:
             return _valueHandle.isValid();
         }
 
+        const Time& getTime() const {
+            return _valueHandle.getTime();
+        }
+
         Value* get() {
             return &_valueHandle->value;
         }
@@ -302,6 +306,33 @@ public:
 
     /**
      * Invalidates the given 'key' and immediately replaces it with a new value.
+     *
+     * The 'time' parameter is mandatory for causally-consistent caches, but not needed otherwise
+     * (since the time never changes).
+     */
+    void insertOrAssign(const Key& key, Value&& newValue, Date_t updateWallClockTime) {
+        stdx::lock_guard lg(_mutex);
+        if (auto it = _inProgressLookups.find(key); it != _inProgressLookups.end())
+            it->second->invalidateAndCancelCurrentLookupRound(lg);
+        _cache.insertOrAssign(key, {std::move(newValue), updateWallClockTime});
+    }
+
+    void insertOrAssign(const Key& key,
+                        Value&& newValue,
+                        Date_t updateWallClockTime,
+                        const Time& time) {
+        stdx::lock_guard lg(_mutex);
+        if (auto it = _inProgressLookups.find(key); it != _inProgressLookups.end())
+            it->second->invalidateAndCancelCurrentLookupRound(lg);
+        _cache.insertOrAssign(key, {std::move(newValue), updateWallClockTime}, time);
+    }
+
+    /**
+     * Invalidates the given 'key' and immediately replaces it with a new value, returning a handle
+     * to the new value.
+     *
+     * The 'time' parameter is mandatory for causally-consistent caches, but not needed otherwise
+     * (since the time never changes).
      */
     ValueHandle insertOrAssignAndGet(const Key& key, Value&& newValue, Date_t updateWallClockTime) {
         stdx::lock_guard lg(_mutex);
@@ -310,6 +341,16 @@ public:
         return _cache.insertOrAssignAndGet(key, {std::move(newValue), updateWallClockTime});
     }
 
+    ValueHandle insertOrAssignAndGet(const Key& key,
+                                     Value&& newValue,
+                                     Date_t updateWallClockTime,
+                                     const Time& time) {
+        stdx::lock_guard lg(_mutex);
+        if (auto it = _inProgressLookups.find(key); it != _inProgressLookups.end())
+            it->second->invalidateAndCancelCurrentLookupRound(lg);
+        return _cache.insertOrAssignAndGet(key, {std::move(newValue), updateWallClockTime}, time);
+    }
+
     /**
      * Indicates to the cache that the backing store has a newer version of 'key', corresponding to
      * 'newTime'. Subsequent calls to 'acquireAsync' with a causal consistency set to 'LatestKnown'
@@ -377,9 +418,8 @@ public:
         return _cache.getCacheInfo();
     }
 
-protected:
     /**
-     * ReadThroughCache constructor, to be called by sub-classes, which implement 'lookup'.
+     * ReadThroughCache constructor.
      *
      * The 'mutex' is for the exclusive usage of the ReadThroughCache and must not be used in any
      * way by the implementing class. Having the mutex stored by the sub-class allows latch