SERVER-25694 Eliminate race in PlanExecutor cleanup.

Ensures that a collection lock is held in at least MODE_IS while
deregistering a PlanExecutor from the cursor manager. Introduces new
PlanExecutor::dispose() and ClientCursor::dispose() methods that must be
called before destruction of those classes, and ensures they are called
before destruction. These calls will thread an OperationContext all the
way through to DocumentSource::dispose() for each stage in a Pipeline,
which will give DocumentSourceCursor a chance to acquire locks and
deregister its PlanExecutor if necessary.

1 files changed, 43 insertions, 96 deletions

diff --git a/src/mongo/dbtests/executor_registry.cpp b/src/mongo/dbtests/executor_registry.cpp
index cfe2b68efb6..46ec9c1e3a0 100644
--- a/src/mongo/dbtests/executor_registry.cpp
+++ b/src/mongo/dbtests/executor_registry.cpp
@@ -27,8 +27,8 @@
  */
 
 /**
- * This file tests PlanExecutor forced yielding, ClientCursor::registerExecutor, and
- * ClientCursor::deregisterExecutor.
+ * This file tests that a manually yielded PlanExecutor will still receive invalidations and be
+ * marked as killed by events like collection drops.
  */
 
 #include "mongo/platform/basic.h"
@@ -68,7 +68,7 @@ public:
     /**
      * Return a plan executor that is going over the collection in nss.ns().
      */
-    PlanExecutor* getCollscan() {
+    std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> getCollscan() {
         unique_ptr<WorkingSet> ws(new WorkingSet());
         CollectionScanParams params;
         params.collection = collection();
@@ -91,22 +91,7 @@ public:
                                                          _ctx->db()->getCollection(&_opCtx, nss),
                                                          PlanExecutor::YIELD_MANUAL);
         ASSERT_OK(statusWithPlanExecutor.getStatus());
-        return statusWithPlanExecutor.getValue().release();
-    }
-
-    void registerExecutor(PlanExecutor* exec) {
-        WriteUnitOfWork wuow(&_opCtx);
-        _ctx->db()->getOrCreateCollection(&_opCtx, nss)->getCursorManager()->registerExecutor(exec);
-        wuow.commit();
-    }
-
-    void deregisterExecutor(PlanExecutor* exec) {
-        WriteUnitOfWork wuow(&_opCtx);
-        _ctx->db()
-            ->getOrCreateCollection(&_opCtx, nss)
-            ->getCursorManager()
-            ->deregisterExecutor(exec);
-        wuow.commit();
+        return std::move(statusWithPlanExecutor.getValue());
     }
 
     int N() {
@@ -125,7 +110,7 @@ public:
 };
 
 
-// Test that a registered runner receives invalidation notifications.
+// Test that a registered PlanExecutor receives invalidation notifications.
 class ExecutorRegistryDiskLocInvalid : public ExecutorRegistryBase {
 public:
     void run() {
@@ -133,18 +118,17 @@ public:
             return;
         }
 
-        unique_ptr<PlanExecutor> run(getCollscan());
+        auto exec = getCollscan();
         BSONObj obj;
 
         // Read some of it.
         for (int i = 0; i < 10; ++i) {
-            ASSERT_EQUALS(PlanExecutor::ADVANCED, run->getNext(&obj, NULL));
+            ASSERT_EQUALS(PlanExecutor::ADVANCED, exec->getNext(&obj, NULL));
             ASSERT_EQUALS(i, obj["foo"].numberInt());
         }
 
         // Register it.
-        run->saveState();
-        registerExecutor(run.get());
+        exec->saveState();
         // At this point it's safe to yield.  forceYield would do that.  Let's now simulate some
         // stuff going on in the yield.
 
@@ -154,174 +138,137 @@ public:
 
         // At this point, we're done yielding.  We recover our lock.
 
-        // Unregister the runner.
-        deregisterExecutor(run.get());
-
         // And clean up anything that happened before.
-        run->restoreState();
+        exec->restoreState();
 
-        // Make sure that the runner moved forward over the deleted data.  We don't see foo==10
+        // Make sure that the PlanExecutor moved forward over the deleted data.  We don't see
+        // foo==10
         // or foo==11.
         for (int i = 12; i < N(); ++i) {
-            ASSERT_EQUALS(PlanExecutor::ADVANCED, run->getNext(&obj, NULL));
+            ASSERT_EQUALS(PlanExecutor::ADVANCED, exec->getNext(&obj, NULL));
             ASSERT_EQUALS(i, obj["foo"].numberInt());
         }
 
-        ASSERT_EQUALS(PlanExecutor::IS_EOF, run->getNext(&obj, NULL));
+        ASSERT_EQUALS(PlanExecutor::IS_EOF, exec->getNext(&obj, NULL));
     }
 };
 
-// Test that registered runners are killed when their collection is dropped.
+// Test that registered PlanExecutors are killed when their collection is dropped.
 class ExecutorRegistryDropCollection : public ExecutorRegistryBase {
 public:
     void run() {
-        unique_ptr<PlanExecutor> run(getCollscan());
+        auto exec = getCollscan();
         BSONObj obj;
 
         // Read some of it.
         for (int i = 0; i < 10; ++i) {
-            ASSERT_EQUALS(PlanExecutor::ADVANCED, run->getNext(&obj, NULL));
+            ASSERT_EQUALS(PlanExecutor::ADVANCED, exec->getNext(&obj, NULL));
             ASSERT_EQUALS(i, obj["foo"].numberInt());
         }
 
         // Save state and register.
-        run->saveState();
-        registerExecutor(run.get());
+        exec->saveState();
 
         // Drop a collection that's not ours.
         _client.dropCollection("unittests.someboguscollection");
 
         // Unregister and restore state.
-        deregisterExecutor(run.get());
-        run->restoreState();
+        exec->restoreState();
 
-        ASSERT_EQUALS(PlanExecutor::ADVANCED, run->getNext(&obj, NULL));
+        ASSERT_EQUALS(PlanExecutor::ADVANCED, exec->getNext(&obj, NULL));
         ASSERT_EQUALS(10, obj["foo"].numberInt());
 
-        // Save state and register.
-        run->saveState();
-        registerExecutor(run.get());
+        exec->saveState();
 
-        // Drop our collection.
         _client.dropCollection(nss.ns());
 
-        // Unregister and restore state.
-        deregisterExecutor(run.get());
-        run->restoreState();
+        exec->restoreState();
 
         // PlanExecutor was killed.
-        ASSERT_EQUALS(PlanExecutor::DEAD, run->getNext(&obj, NULL));
+        ASSERT_EQUALS(PlanExecutor::DEAD, exec->getNext(&obj, NULL));
     }
 };
 
-// Test that registered runners are killed when all indices are dropped on the collection.
+// Test that registered PlanExecutors are killed when all indices are dropped on the collection.
 class ExecutorRegistryDropAllIndices : public ExecutorRegistryBase {
 public:
     void run() {
-        unique_ptr<PlanExecutor> run(getCollscan());
+        auto exec = getCollscan();
         BSONObj obj;
 
         ASSERT_OK(dbtests::createIndex(&_opCtx, nss.ns(), BSON("foo" << 1)));
 
         // Read some of it.
         for (int i = 0; i < 10; ++i) {
-            ASSERT_EQUALS(PlanExecutor::ADVANCED, run->getNext(&obj, NULL));
+            ASSERT_EQUALS(PlanExecutor::ADVANCED, exec->getNext(&obj, NULL));
             ASSERT_EQUALS(i, obj["foo"].numberInt());
         }
 
-        // Save state and register.
-        run->saveState();
-        registerExecutor(run.get());
-
-        // Drop all indices.
+        exec->saveState();
         _client.dropIndexes(nss.ns());
-
-        // Unregister and restore state.
-        deregisterExecutor(run.get());
-        run->restoreState();
-
-        // PlanExecutor was killed.
-        ASSERT_EQUALS(PlanExecutor::DEAD, run->getNext(&obj, NULL));
+        exec->restoreState();
+        ASSERT_EQUALS(PlanExecutor::DEAD, exec->getNext(&obj, NULL));
     }
 };
 
-// Test that registered runners are killed when an index is dropped on the collection.
+// Test that registered PlanExecutors are killed when an index is dropped on the collection.
 class ExecutorRegistryDropOneIndex : public ExecutorRegistryBase {
 public:
     void run() {
-        unique_ptr<PlanExecutor> run(getCollscan());
+        auto exec = getCollscan();
         BSONObj obj;
 
         ASSERT_OK(dbtests::createIndex(&_opCtx, nss.ns(), BSON("foo" << 1)));
 
         // Read some of it.
         for (int i = 0; i < 10; ++i) {
-            ASSERT_EQUALS(PlanExecutor::ADVANCED, run->getNext(&obj, NULL));
+            ASSERT_EQUALS(PlanExecutor::ADVANCED, exec->getNext(&obj, NULL));
             ASSERT_EQUALS(i, obj["foo"].numberInt());
         }
 
-        // Save state and register.
-        run->saveState();
-        registerExecutor(run.get());
-
-        // Drop a specific index.
+        exec->saveState();
         _client.dropIndex(nss.ns(), BSON("foo" << 1));
-
-        // Unregister and restore state.
-        deregisterExecutor(run.get());
-        run->restoreState();
-
-        // PlanExecutor was killed.
-        ASSERT_EQUALS(PlanExecutor::DEAD, run->getNext(&obj, NULL));
+        exec->restoreState();
+        ASSERT_EQUALS(PlanExecutor::DEAD, exec->getNext(&obj, NULL));
     }
 };
 
-// Test that registered runners are killed when their database is dropped.
+// Test that registered PlanExecutors are killed when their database is dropped.
 class ExecutorRegistryDropDatabase : public ExecutorRegistryBase {
 public:
     void run() {
-        unique_ptr<PlanExecutor> run(getCollscan());
+        auto exec = getCollscan();
         BSONObj obj;
 
         // Read some of it.
         for (int i = 0; i < 10; ++i) {
-            ASSERT_EQUALS(PlanExecutor::ADVANCED, run->getNext(&obj, NULL));
+            ASSERT_EQUALS(PlanExecutor::ADVANCED, exec->getNext(&obj, NULL));
             ASSERT_EQUALS(i, obj["foo"].numberInt());
         }
 
-        // Save state and register.
-        run->saveState();
-        registerExecutor(run.get());
+        exec->saveState();
 
         // Drop a DB that's not ours.  We can't have a lock at all to do this as dropping a DB
         // requires a "global write lock."
         _ctx.reset();
         _client.dropDatabase("somesillydb");
         _ctx.reset(new OldClientWriteContext(&_opCtx, nss.ns()));
+        exec->restoreState();
 
-        // Unregister and restore state.
-        deregisterExecutor(run.get());
-        run->restoreState();
-
-        ASSERT_EQUALS(PlanExecutor::ADVANCED, run->getNext(&obj, NULL));
+        ASSERT_EQUALS(PlanExecutor::ADVANCED, exec->getNext(&obj, NULL));
         ASSERT_EQUALS(10, obj["foo"].numberInt());
 
-        // Save state and register.
-        run->saveState();
-        registerExecutor(run.get());
+        exec->saveState();
 
         // Drop our DB.  Once again, must give up the lock.
         _ctx.reset();
         _client.dropDatabase("unittests");
         _ctx.reset(new OldClientWriteContext(&_opCtx, nss.ns()));
-
-        // Unregister and restore state.
-        deregisterExecutor(run.get());
-        run->restoreState();
+        exec->restoreState();
         _ctx.reset();
 
         // PlanExecutor was killed.
-        ASSERT_EQUALS(PlanExecutor::DEAD, run->getNext(&obj, NULL));
+        ASSERT_EQUALS(PlanExecutor::DEAD, exec->getNext(&obj, NULL));
     }
 };