SERVER-39520 Use database IX lock for dropCollection

5 files changed, 96 insertions, 65 deletions

diff --git a/src/mongo/db/commands/SConscript b/src/mongo/db/commands/SConscript
index 826d195b346..882331f0073 100644
--- a/src/mongo/db/commands/SConscript
+++ b/src/mongo/db/commands/SConscript
@@ -265,6 +265,7 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/base',
         '$BUILD_DIR/mongo/db/catalog/catalog_helpers',
+        '$BUILD_DIR/mongo/db/catalog/uuid_catalog_helper',
         '$BUILD_DIR/mongo/db/catalog/index_key_validate',
         '$BUILD_DIR/mongo/db/catalog/multi_index_block',
         '$BUILD_DIR/mongo/db/command_can_run_here',
@@ -367,6 +368,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/background',
         '$BUILD_DIR/mongo/db/catalog/catalog_control',
         '$BUILD_DIR/mongo/db/catalog/catalog_helpers',
+        '$BUILD_DIR/mongo/db/catalog/uuid_catalog_helper',
         '$BUILD_DIR/mongo/db/catalog/index_key_validate',
         '$BUILD_DIR/mongo/db/cloner',
         '$BUILD_DIR/mongo/db/commands',
diff --git a/src/mongo/db/commands/dbhash.cpp b/src/mongo/db/commands/dbhash.cpp
index 6a6ab9cd1a3..db939ea91b6 100644
--- a/src/mongo/db/commands/dbhash.cpp
+++ b/src/mongo/db/commands/dbhash.cpp
@@ -33,11 +33,13 @@
 
 #include <boost/optional.hpp>
 #include <map>
+#include <set>
 #include <string>
 
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/catalog/database.h"
 #include "mongo/db/catalog/index_catalog.h"
+#include "mongo/db/catalog/uuid_catalog_helper.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/commands/test_commands_enabled.h"
 #include "mongo/db/db_raii.h"
@@ -185,7 +187,8 @@ public:
         if (opCtx->recoveryUnit()->getTimestampReadSource() ==
             RecoveryUnit::ReadSource::kProvided) {
             // However, if we are performing a read at a timestamp, then we only need to lock the
-            // database in intent mode to ensure that none of the collections get dropped.
+            // database in intent mode and then collection in intent mode as well to ensure that
+            // none of the collections get dropped.
             lockMode = LockMode::MODE_IS;
 
             // Additionally, if we are performing a read at a timestamp, then we allow oplog
@@ -196,11 +199,6 @@ public:
         }
         AutoGetDb autoDb(opCtx, ns, lockMode);
         Database* db = autoDb.getDb();
-        std::vector<NamespaceString> collections;
-        if (db) {
-            collections = UUIDCatalog::get(opCtx).getAllCollectionNamesFromDb(db->name());
-            std::sort(collections.begin(), collections.end());
-        }
 
         result.append("host", prettyHostName());
 
@@ -216,51 +214,84 @@ public:
                                                                "system.version",
                                                                "system.views"};
 
-        BSONArrayBuilder cappedCollections;
-        BSONObjBuilder collectionsByUUID;
-
-        BSONObjBuilder bb(result.subobjStart("collections"));
-        for (const auto& collNss : collections) {
-            if (collNss.size() - 1 <= dbname.size()) {
-                errmsg = str::stream() << "weird fullCollectionName [" << collNss.toString() << "]";
-                return false;
-            }
+        std::map<std::string, std::string> collectionToHashMap;
+        std::map<std::string, OptionalCollectionUUID> collectionToUUIDMap;
+        std::set<std::string> cappedCollectionSet;
+
+        bool noError = true;
+        catalog::forEachCollectionFromDb(
+            opCtx,
+            dbname,
+            MODE_IS,
+            [&](Collection* collection, CollectionCatalogEntry* catalogEntry) {
+                auto collNss = collection->ns();
+
+                if (collNss.size() - 1 <= dbname.size()) {
+                    errmsg = str::stream() << "weird fullCollectionName [" << collNss.toString()
+                                           << "]";
+                    noError = false;
+                    return false;
+                }
 
-            // Only include 'system' collections that are replicated.
-            bool isReplicatedSystemColl =
-                (replicatedSystemCollections.count(collNss.coll().toString()) > 0);
-            if (collNss.isSystem() && !isReplicatedSystemColl)
-                continue;
+                // Only include 'system' collections that are replicated.
+                bool isReplicatedSystemColl =
+                    (replicatedSystemCollections.count(collNss.coll().toString()) > 0);
+                if (collNss.isSystem() && !isReplicatedSystemColl)
+                    return true;
 
-            if (collNss.coll().startsWith("tmp.mr.")) {
-                // We skip any incremental map reduce collections as they also aren't replicated.
-                continue;
-            }
+                if (collNss.coll().startsWith("tmp.mr.")) {
+                    // We skip any incremental map reduce collections as they also aren't
+                    // replicated.
+                    return true;
+                }
 
-            if (desiredCollections.size() > 0 &&
-                desiredCollections.count(collNss.coll().toString()) == 0)
-                continue;
+                if (desiredCollections.size() > 0 &&
+                    desiredCollections.count(collNss.coll().toString()) == 0)
+                    return true;
 
-            // Don't include 'drop pending' collections.
-            if (collNss.isDropPendingNamespace())
-                continue;
+                // Don't include 'drop pending' collections.
+                if (collNss.isDropPendingNamespace())
+                    return true;
 
-            if (Collection* collection = db->getCollection(opCtx, collNss.ns())) {
                 if (collection->isCapped()) {
-                    cappedCollections.append(collNss.coll());
+                    cappedCollectionSet.insert(collNss.coll().toString());
                 }
 
                 if (OptionalCollectionUUID uuid = collection->uuid()) {
-                    uuid->appendToBuilder(&collectionsByUUID, collNss.coll());
+                    collectionToUUIDMap[collNss.coll().toString()] = uuid;
                 }
-            }
 
-            // Compute the hash for this collection.
-            std::string hash = _hashCollection(opCtx, db, collNss.toString());
+                // Compute the hash for this collection.
+                std::string hash = _hashCollection(opCtx, db, collNss.toString());
 
-            bb.append(collNss.coll(), hash);
+                collectionToHashMap[collNss.coll().toString()] = hash;
+
+                return true;
+            });
+        if (!noError)
+            return false;
+
+        BSONObjBuilder bb(result.subobjStart("collections"));
+        BSONArrayBuilder cappedCollections;
+        BSONObjBuilder collectionsByUUID;
+
+        for (auto elem : cappedCollectionSet) {
+            cappedCollections.append(elem);
+        }
+
+        for (auto entry : collectionToUUIDMap) {
+            auto collName = entry.first;
+            auto uuid = entry.second;
+            uuid->appendToBuilder(&collectionsByUUID, collName);
+        }
+
+        for (auto entry : collectionToHashMap) {
+            auto collName = entry.first;
+            auto hash = entry.second;
+            bb.append(collName, hash);
             md5_append(&globalState, (const md5_byte_t*)hash.c_str(), hash.size());
         }
+
         bb.done();
 
         result.append("capped", BSONArray(cappedCollections.done()));
@@ -284,8 +315,7 @@ private:
         NamespaceString ns(fullCollectionName);
 
         Collection* collection = db->getCollection(opCtx, ns);
-        if (!collection)
-            return "";
+        invariant(collection);
 
         boost::optional<Lock::CollectionLock> collLock;
         if (opCtx->recoveryUnit()->getTimestampReadSource() ==
@@ -294,8 +324,7 @@ private:
             // intent mode. We need to also acquire the collection lock in intent mode to ensure
             // reading from the consistent snapshot doesn't overlap with any catalog operations on
             // the collection.
-            invariant(opCtx->lockState()->isDbLockedForMode(db->name(), MODE_IS));
-            collLock.emplace(opCtx, ns, MODE_IS);
+            invariant(opCtx->lockState()->isCollectionLockedForMode(ns, MODE_IS));
 
             auto minSnapshot = collection->getMinimumVisibleSnapshot();
             auto mySnapshot = opCtx->recoveryUnit()->getPointInTimeReadTimestamp();
diff --git a/src/mongo/db/commands/list_collections.cpp b/src/mongo/db/commands/list_collections.cpp
index 8d711c43cf2..e1f20746eb4 100644
--- a/src/mongo/db/commands/list_collections.cpp
+++ b/src/mongo/db/commands/list_collections.cpp
@@ -41,6 +41,7 @@
 #include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/database.h"
 #include "mongo/db/catalog/index_catalog.h"
+#include "mongo/db/catalog/uuid_catalog_helper.h"
 #include "mongo/db/clientcursor.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/commands/list_collections_filter.h"
@@ -186,7 +187,6 @@ BSONObj buildCollectionBson(OperationContext* opCtx,
         return b.obj();
     }
 
-    Lock::CollectionLock clk(opCtx, nss, MODE_IS);
     CollectionOptions options = collection->getCatalogEntry()->getCollectionOptions(opCtx);
 
     // While the UUID is stored as a collection option, from the user's perspective it is an
@@ -311,6 +311,7 @@ public:
                             continue;
                         }
 
+                        Lock::CollectionLock clk(opCtx, nss, MODE_IS);
                         Collection* collection = db->getCollection(opCtx, nss);
                         BSONObj collBson =
                             buildCollectionBson(opCtx, collection, includePendingDrops, nameOnly);
@@ -320,25 +321,25 @@ public:
                         }
                     }
                 } else {
-                    for (auto collIt = db->begin(opCtx); collIt != db->end(opCtx); ++collIt) {
-                        auto collection = *collIt;
-                        if (!collection) {
-                            break;
-                        }
-
-                        if (authorizedCollections &&
-                            (collection->ns().coll().startsWith("system.") ||
-                             !as->isAuthorizedForAnyActionOnResource(
-                                 ResourcePattern::forExactNamespace(collection->ns())))) {
-                            continue;
-                        }
-                        BSONObj collBson =
-                            buildCollectionBson(opCtx, collection, includePendingDrops, nameOnly);
-                        if (!collBson.isEmpty()) {
-                            _addWorkingSetMember(
-                                opCtx, collBson, matcher.get(), ws.get(), root.get());
-                        }
-                    }
+                    mongo::catalog::forEachCollectionFromDb(
+                        opCtx,
+                        dbname,
+                        MODE_IS,
+                        [&](Collection* collection, CollectionCatalogEntry* catalogEntry) {
+                            if (authorizedCollections &&
+                                (collection->ns().coll().startsWith("system.") ||
+                                 !as->isAuthorizedForAnyActionOnResource(
+                                     ResourcePattern::forExactNamespace(collection->ns())))) {
+                                return true;
+                            }
+                            BSONObj collBson = buildCollectionBson(
+                                opCtx, collection, includePendingDrops, nameOnly);
+                            if (!collBson.isEmpty()) {
+                                _addWorkingSetMember(
+                                    opCtx, collBson, matcher.get(), ws.get(), root.get());
+                            }
+                            return true;
+                        });
                 }
 
                 // Skipping views is only necessary for internal cloning operations.
diff --git a/src/mongo/db/commands/list_databases.cpp b/src/mongo/db/commands/list_databases.cpp
index 46d3ab65fa8..dccec13a8b6 100644
--- a/src/mongo/db/commands/list_databases.cpp
+++ b/src/mongo/db/commands/list_databases.cpp
@@ -170,7 +170,7 @@ public:
                 b.append("sizeOnDisk", static_cast<double>(size));
 
                 b.appendBool("empty",
-                             UUIDCatalog::get(opCtx).getAllCatalogEntriesFromDb(dbname).empty());
+                             UUIDCatalog::get(opCtx).getAllCollectionUUIDsFromDb(dbname).empty());
             }
             BSONObj curDbObj = b.obj();
 
diff --git a/src/mongo/db/commands/validate.cpp b/src/mongo/db/commands/validate.cpp
index dbc5ff11323..df8abe7b222 100644
--- a/src/mongo/db/commands/validate.cpp
+++ b/src/mongo/db/commands/validate.cpp
@@ -123,7 +123,7 @@ public:
         }
 
         AutoGetDb ctx(opCtx, nss.db(), MODE_IX);
-        auto collLk = stdx::make_unique<Lock::CollectionLock>(opCtx, nss, MODE_X);
+        Lock::CollectionLock collLk(opCtx, nss, MODE_X);
         Collection* collection = ctx.getDb() ? ctx.getDb()->getCollection(opCtx, nss) : NULL;
         if (!collection) {
             if (ctx.getDb() && ViewCatalog::get(ctx.getDb())->lookup(opCtx, nss.ns())) {
@@ -163,8 +163,7 @@ public:
         const bool background = false;
 
         ValidateResults results;
-        Status status =
-            collection->validate(opCtx, level, background, std::move(collLk), &results, &result);
+        Status status = collection->validate(opCtx, level, background, &results, &result);
         if (!status.isOK()) {
             return CommandHelpers::appendCommandStatusNoThrow(result, status);
         }