summaryrefslogtreecommitdiff
path: root/src/mongo/s/catalog_cache.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/mongo/s/catalog_cache.cpp')
-rw-r--r--src/mongo/s/catalog_cache.cpp210
1 files changed, 103 insertions, 107 deletions
diff --git a/src/mongo/s/catalog_cache.cpp b/src/mongo/s/catalog_cache.cpp
index 60314d67ef3..fff547935bc 100644
--- a/src/mongo/s/catalog_cache.cpp
+++ b/src/mongo/s/catalog_cache.cpp
@@ -37,6 +37,8 @@
#include "mongo/s/catalog_cache.h"
+#include <fmt/format.h>
+
#include "mongo/bson/bsonobjbuilder.h"
#include "mongo/db/query/collation/collator_factory_interface.h"
#include "mongo/db/repl/optime_with.h"
@@ -57,6 +59,7 @@ namespace mongo {
namespace {
MONGO_FAIL_POINT_DEFINE(blockCollectionCacheLookup);
+MONGO_FAIL_POINT_DEFINE(blockDatabaseCacheLookup);
// How many times to try refreshing the routing info if the set of chunks loaded from the config
// server is found to be inconsistent.
@@ -68,9 +71,6 @@ const int kCollectionCacheSize = 10000;
const OperationContext::Decoration<bool> operationShouldBlockBehindCatalogCacheRefresh =
OperationContext::declareDecoration<bool>();
-const OperationContext::Decoration<bool> operationBlockedBehindCatalogCacheRefresh =
- OperationContext::declareDecoration<bool>();
-
std::shared_ptr<RoutingTableHistory> createUpdatedRoutingTableHistory(
OperationContext* opCtx,
const NamespaceString& nss,
@@ -81,28 +81,31 @@ std::shared_ptr<RoutingTableHistory> createUpdatedRoutingTableHistory(
if (isIncremental && collectionAndChunks.changedChunks.size() == 1 &&
collectionAndChunks.changedChunks[0].getVersion() == existingHistory->optRt->getVersion()) {
- invariant(collectionAndChunks.allowMigrations == existingHistory->optRt->allowMigrations(),
- str::stream() << "allowMigrations field of " << nss
- << " collection changed without changing the collection version "
- << existingHistory->optRt->getVersion().toString()
- << ". Old value: " << existingHistory->optRt->allowMigrations()
- << ", new value: " << collectionAndChunks.allowMigrations);
+ tassert(7032310,
+ fmt::format("allowMigrations field of collection '{}' changed without changing the "
+ "collection version {}. Old value: {}, new value: {}",
+ nss.toString(),
+ existingHistory->optRt->getVersion().toString(),
+ existingHistory->optRt->allowMigrations(),
+ collectionAndChunks.allowMigrations),
+ collectionAndChunks.allowMigrations == existingHistory->optRt->allowMigrations());
const auto& oldReshardingFields = existingHistory->optRt->getReshardingFields();
const auto& newReshardingFields = collectionAndChunks.reshardingFields;
- invariant(
- [&] {
- if (oldReshardingFields && newReshardingFields)
- return oldReshardingFields->toBSON().woCompare(newReshardingFields->toBSON()) ==
- 0;
- else
- return !oldReshardingFields && !newReshardingFields;
- }(),
- str::stream() << "reshardingFields field of " << nss
- << " collection changed without changing the collection version "
- << existingHistory->optRt->getVersion().toString()
- << ". Old value: " << oldReshardingFields->toBSON()
- << ", new value: " << newReshardingFields->toBSON());
+ tassert(7032311,
+ fmt::format("reshardingFields field of collection '{}' changed without changing "
+ "the collection version {}. Old value: {}, new value: {}",
+ nss.toString(),
+ existingHistory->optRt->getVersion().toString(),
+ oldReshardingFields->toBSON().toString(),
+ newReshardingFields->toBSON().toString()),
+ [&] {
+ if (oldReshardingFields && newReshardingFields)
+ return oldReshardingFields->toBSON().woCompare(
+ newReshardingFields->toBSON()) == 0;
+ else
+ return !oldReshardingFields && !newReshardingFields;
+ }());
return existingHistory->optRt;
}
@@ -114,7 +117,11 @@ std::shared_ptr<RoutingTableHistory> createUpdatedRoutingTableHistory(
return 0;
}
if (collectionAndChunks.maxChunkSizeBytes) {
- invariant(collectionAndChunks.maxChunkSizeBytes.get() > 0);
+ tassert(7032312,
+ fmt::format("Invalid maxChunkSizeBytes value {} for collection '{}'",
+ nss.toString(),
+ collectionAndChunks.maxChunkSizeBytes.get()),
+ collectionAndChunks.maxChunkSizeBytes.get() > 0);
return uint64_t(*collectionAndChunks.maxChunkSizeBytes);
}
return boost::none;
@@ -242,26 +249,51 @@ CatalogCache::CatalogCache(ServiceContext* const service, CatalogCacheLoader& ca
}
CatalogCache::~CatalogCache() {
- // The executor is used by the Database and Collection caches,
- // so it must be joined, before these caches are destroyed,
- // per the contract of ReadThroughCache.
+ // The executor is used by all the caches that correspond to the router role, so it must be
+ // joined before these caches are destroyed, per the contract of ReadThroughCache.
+ shutDownAndJoin();
+}
+
+void CatalogCache::shutDownAndJoin() {
_executor->shutdown();
_executor->join();
}
StatusWith<CachedDatabaseInfo> CatalogCache::getDatabase(OperationContext* opCtx,
- StringData dbName,
- bool allowLocks) {
- if (!allowLocks) {
- invariant(
- !opCtx->lockState() || !opCtx->lockState()->isLocked(),
+ StringData dbName) {
+ return _getDatabase(opCtx, dbName);
+}
+
+StatusWith<CachedDatabaseInfo> CatalogCache::_getDatabase(OperationContext* opCtx,
+ StringData dbName,
+ bool allowLocks) {
+ tassert(7032313,
"Do not hold a lock while refreshing the catalog cache. Doing so would potentially "
"hold the lock during a network call, and can lead to a deadlock as described in "
- "SERVER-37398.");
- }
+ "SERVER-37398.",
+ allowLocks || !opCtx->lockState() || !opCtx->lockState()->isLocked());
try {
- auto dbEntry = _databaseCache.acquire(opCtx, dbName, CacheCausalConsistency::kLatestKnown);
+ auto dbEntryFuture =
+ _databaseCache.acquireAsync(dbName, CacheCausalConsistency::kLatestKnown);
+
+ if (allowLocks) {
+ // When allowLocks is true we may be holding a lock, so we don't want to block the
+ // current thread: if the future is ready let's use it, otherwise return an error.
+ if (dbEntryFuture.isReady()) {
+ return dbEntryFuture.get(opCtx);
+ } else {
+ // This error only contains the database name and must be handled by any callers of
+ // _getDatabase with the potential for allowLocks to be true. The caller should
+ // convert this to ErrorCodes::ShardCannotRefreshDueToLocksHeld with the full
+ // namespace.
+ return Status{ShardCannotRefreshDueToLocksHeldInfo(NamespaceString(dbName)),
+ "Database info refresh did not complete"};
+ }
+ }
+
+ // From this point we can guarantee that allowLocks is false.
+ auto dbEntry = dbEntryFuture.get(opCtx);
uassert(ErrorCodes::NamespaceNotFound,
str::stream() << "database " << dbName << " not found",
dbEntry);
@@ -277,18 +309,28 @@ StatusWith<ChunkManager> CatalogCache::_getCollectionRoutingInfoAt(
const NamespaceString& nss,
boost::optional<Timestamp> atClusterTime,
bool allowLocks) {
- if (!allowLocks) {
- invariant(!opCtx->lockState() || !opCtx->lockState()->isLocked(),
- "Do not hold a lock while refreshing the catalog cache. Doing so would "
- "potentially hold "
- "the lock during a network call, and can lead to a deadlock as described in "
- "SERVER-37398.");
- }
+ tassert(7032314,
+ "Do not hold a lock while refreshing the catalog cache. Doing so would potentially "
+ "hold the lock during a network call, and can lead to a deadlock as described in "
+ "SERVER-37398.",
+ allowLocks || !opCtx->lockState() || !opCtx->lockState()->isLocked());
try {
- const auto swDbInfo = getDatabase(opCtx, nss.db(), allowLocks);
+ const auto swDbInfo = _getDatabase(opCtx, nss.db(), allowLocks);
if (!swDbInfo.isOK()) {
- if (swDbInfo == ErrorCodes::NamespaceNotFound) {
+ if (swDbInfo == ErrorCodes::ShardCannotRefreshDueToLocksHeld) {
+ // Since collection refreshes always imply database refreshes, it is ok to transform
+ // this error into a collection error rather than a database error.
+ auto dbRefreshInfo =
+ swDbInfo.getStatus().extraInfo<ShardCannotRefreshDueToLocksHeldInfo>();
+ LOGV2_DEBUG(7850500,
+ 2,
+ "Adding collection name to ShardCannotRefreshDueToLocksHeld error",
+ "dbName"_attr = dbRefreshInfo->getNss().db(),
+ "nss"_attr = nss);
+ return Status{ShardCannotRefreshDueToLocksHeldInfo(nss),
+ "Routing info refresh did not complete"};
+ } else if (swDbInfo == ErrorCodes::NamespaceNotFound) {
LOGV2_FOR_CATALOG_REFRESH(
4947103,
2,
@@ -326,9 +368,6 @@ StatusWith<ChunkManager> CatalogCache::_getCollectionRoutingInfoAt(
}
// From this point we can guarantee that allowLocks is false
-
- operationBlockedBehindCatalogCacheRefresh(opCtx) = true;
-
size_t acquireTries = 0;
Timer t;
@@ -525,37 +564,6 @@ void CatalogCache::report(BSONObjBuilder* builder) const {
_collectionCache.reportStats(&cacheStatsBuilder);
}
-void CatalogCache::checkAndRecordOperationBlockedByRefresh(OperationContext* opCtx,
- mongo::LogicalOp opType) {
- if (!isMongos() || !operationBlockedBehindCatalogCacheRefresh(opCtx)) {
- return;
- }
-
- auto& opsBlockedByRefresh = _stats.operationsBlockedByRefresh;
-
- opsBlockedByRefresh.countAllOperations.fetchAndAddRelaxed(1);
-
- switch (opType) {
- case LogicalOp::opInsert:
- opsBlockedByRefresh.countInserts.fetchAndAddRelaxed(1);
- break;
- case LogicalOp::opQuery:
- opsBlockedByRefresh.countQueries.fetchAndAddRelaxed(1);
- break;
- case LogicalOp::opUpdate:
- opsBlockedByRefresh.countUpdates.fetchAndAddRelaxed(1);
- break;
- case LogicalOp::opDelete:
- opsBlockedByRefresh.countDeletes.fetchAndAddRelaxed(1);
- break;
- case LogicalOp::opCommand:
- opsBlockedByRefresh.countCommands.fetchAndAddRelaxed(1);
- break;
- default:
- MONGO_UNREACHABLE;
- }
-}
-
void CatalogCache::invalidateDatabaseEntry_LINEARIZABLE(const StringData& dbName) {
_databaseCache.invalidateKey(dbName);
}
@@ -568,26 +576,6 @@ void CatalogCache::Stats::report(BSONObjBuilder* builder) const {
builder->append("countStaleConfigErrors", countStaleConfigErrors.load());
builder->append("totalRefreshWaitTimeMicros", totalRefreshWaitTimeMicros.load());
-
- if (isMongos()) {
- BSONObjBuilder operationsBlockedByRefreshBuilder(
- builder->subobjStart("operationsBlockedByRefresh"));
-
- operationsBlockedByRefreshBuilder.append(
- "countAllOperations", operationsBlockedByRefresh.countAllOperations.load());
- operationsBlockedByRefreshBuilder.append("countInserts",
- operationsBlockedByRefresh.countInserts.load());
- operationsBlockedByRefreshBuilder.append("countQueries",
- operationsBlockedByRefresh.countQueries.load());
- operationsBlockedByRefreshBuilder.append("countUpdates",
- operationsBlockedByRefresh.countUpdates.load());
- operationsBlockedByRefreshBuilder.append("countDeletes",
- operationsBlockedByRefresh.countDeletes.load());
- operationsBlockedByRefreshBuilder.append("countCommands",
- operationsBlockedByRefresh.countCommands.load());
-
- operationsBlockedByRefreshBuilder.done();
- }
}
CatalogCache::DatabaseCache::DatabaseCache(ServiceContext* service,
@@ -610,6 +598,10 @@ CatalogCache::DatabaseCache::LookupResult CatalogCache::DatabaseCache::_lookupDa
const std::string& dbName,
const DatabaseTypeValueHandle& previousDbType,
const ComparableDatabaseVersion& previousDbVersion) {
+ if (MONGO_unlikely(blockDatabaseCacheLookup.shouldFail())) {
+ LOGV2(8023400, "Hanging before refreshing cached database entry");
+ blockDatabaseCacheLookup.pauseWhileSet();
+ }
// TODO (SERVER-34164): Track and increment stats for database refreshes
LOGV2_FOR_CATALOG_REFRESH(24102, 2, "Refreshing cached database entry", "db"_attr = dbName);
@@ -701,7 +693,7 @@ CatalogCache::CollectionCache::LookupResult CatalogCache::CollectionCache::_look
OperationContext* opCtx,
const NamespaceString& nss,
const RoutingTableHistoryValueHandle& existingHistory,
- const ComparableChunkVersion& previousVersion) {
+ const ComparableChunkVersion& timeInStore) {
const bool isIncremental(existingHistory && existingHistory->optRt);
_updateRefreshesStats(isIncremental, true);
blockCollectionCacheLookup.pauseWhileSet(opCtx);
@@ -720,7 +712,7 @@ CatalogCache::CollectionCache::LookupResult CatalogCache::CollectionCache::_look
"Refreshing cached collection",
"namespace"_attr = nss,
"lookupSinceVersion"_attr = lookupVersion,
- "timeInStore"_attr = previousVersion);
+ "timeInStore"_attr = timeInStore);
auto collectionAndChunks = _catalogCacheLoader.getChunksSince(nss, lookupVersion).get();
@@ -742,14 +734,18 @@ CatalogCache::CollectionCache::LookupResult CatalogCache::CollectionCache::_look
const ChunkVersion newVersion = newRoutingHistory->getVersion();
newComparableVersion.setChunkVersion(newVersion);
- LOGV2_FOR_CATALOG_REFRESH(4619901,
- isIncremental || newComparableVersion != previousVersion ? 0 : 1,
- "Refreshed cached collection",
- "namespace"_attr = nss,
- "lookupSinceVersion"_attr = lookupVersion,
- "newVersion"_attr = newComparableVersion,
- "timeInStore"_attr = previousVersion,
- "duration"_attr = Milliseconds(t.millis()));
+ // The log below is logged at debug(0) (equivalent to info level) only if the new placement
+ // version is different than the one we already had (if any).
+ LOGV2_FOR_CATALOG_REFRESH(
+ 4619901,
+ (!isIncremental || newVersion != existingHistory->optRt->getVersion()) ? 0 : 1,
+ "Refreshed cached collection",
+ "namespace"_attr = nss,
+ "lookupSinceVersion"_attr = lookupVersion,
+ "newVersion"_attr = newComparableVersion,
+ "timeInStore"_attr = timeInStore,
+ "duration"_attr = Milliseconds(t.millis()));
+
_updateRefreshesStats(isIncremental, false);
return LookupResult(OptionalRoutingTableHistory(std::move(newRoutingHistory)),
generated by cgit v1.2.3 (git 2.39.1) at 2026-09-11 10:54:33 +0000