SERVER-10026 SERVER-10461 runner registry: short-term safe yielding for non-cached runners

Author: Hari Khalsa

src/mongo/db/clientcursor.cpp

@@ -48,6 +48,7 @@ namespace mongo {
     ClientCursor::CCById ClientCursor::clientCursorsById;
     boost::recursive_mutex& ClientCursor::ccmutex( *(new boost::recursive_mutex()) );
     long long ClientCursor::numberTimedOut = 0;
+    set<Runner*> ClientCursor::nonCachedRunners;
 
     void aboutToDeleteForSharding(const StringData& ns,
                                   const Database* db,
@@ -140,6 +141,14 @@ namespace mongo {
         verify(db);
         verify(str::startsWith(ns, db->name()));
 
+        for (set<Runner*>::iterator it = nonCachedRunners.begin(); it != nonCachedRunners.end(); ++it) {
+            Runner* runner = *it;
+            const char* runnerNS = runner->getQuery().getParsed().ns();
+            if ((isDB && str::startsWith(runnerNS, ns)) || (str::equals(runnerNS, ns))) {
+                runner->kill();
+            }
+        }
+
         recursive_scoped_lock cclock(ccmutex);
         CCById::const_iterator it = clientCursorsById.begin();
         while (it != clientCursorsById.end()) {
@@ -203,6 +212,14 @@ namespace mongo {
 
         aboutToDeleteForSharding( ns, db, nsd, dl );
 
+        // Check our non-cached active runner list.
+        for (set<Runner*>::iterator it = nonCachedRunners.begin(); it != nonCachedRunners.end(); ++it) {
+            Runner* runner = *it;
+            if (0 == ns.compare(runner->getQuery().getParsed().ns())) {
+                runner->invalidate(dl);
+            }
+        }
+
         // TODO: This requires optimization.  We walk through *all* CCs and send the delete to every
         // CC open on the db we're deleting from.  We could:
         // 1. Map from ns to open runners,
@@ -295,6 +312,18 @@ namespace mongo {
         // End cursor-only
     }
 
+    void ClientCursor::registerRunner(Runner* runner) {
+        recursive_scoped_lock lock(ccmutex);
+        verify(nonCachedRunners.end() == nonCachedRunners.find(runner));
+        nonCachedRunners.insert(runner);
+    }
+
+    void ClientCursor::deregisterRunner(Runner* runner) {
+        recursive_scoped_lock lock(ccmutex);
+        verify(nonCachedRunners.end() != nonCachedRunners.find(runner));
+        nonCachedRunners.erase(runner);
+    }
+
     void yieldOrSleepFor1Microsecond() {
 #ifdef _WIN32
         SwitchToThread();

src/mongo/db/clientcursor.h

@@ -85,11 +85,23 @@ namespace mongo {
                                   const NamespaceDetails* nsd,
                                   const DiskLoc& dl);
 
+        /**
+         * Register a runner so that it can be notified of deletion/invalidation during yields.
+         * Must be called before a runner yields.  If a runner is cached (inside a ClientCursor) it
+         * MUST NOT be registered; the two are mutually exclusive.
+         */
+        static void registerRunner(Runner* runner);
+
+        /**
+         * Remove a runner from the runner registry.
+         */
+        static void deregisterRunner(Runner* runner);
+
         //
         // Yielding.
         // 
 
-        static void staticYield(int micros, const StringData& ns, Record* rec );
+        static void staticYield(int micros, const StringData& ns, Record* rec);
 
         //
         // Static methods about all ClientCursors  TODO: Document.
@@ -253,6 +265,13 @@ namespace mongo {
         typedef map<CursorId, ClientCursor*> CCById;
         static CCById clientCursorsById;
 
+        // A list of NON-CACHED runners.  Any runner that yields must be put into this map before
+        // yielding in order to be notified of invalidation and namespace deletion.  Before the
+        // runner is deleted, it must be removed from this map.
+        //
+        // TODO: This is temporary and as such is highly NOT optimized.
+        static set<Runner*> nonCachedRunners;
+
         // How many cursors have timed out?
         static long long numberTimedOut;
 

src/mongo/db/exec/stagedebug_cmd.cpp

@@ -33,7 +33,7 @@
 #include "mongo/db/matcher/expression_parser.h"
 #include "mongo/db/namespace_details.h"
 #include "mongo/db/pdfile.h"
-#include "mongo/db/query/simple_plan_runner.h"
+#include "mongo/db/query/plan_executor.h"
 
 namespace mongo {
 
@@ -96,14 +96,14 @@ namespace mongo {
             BSONObj argObj = argElt.Obj();
 
             OwnedPointerVector<MatchExpression> exprs;
-            SimplePlanRunner runner;
+            PlanExecutor runner;
             auto_ptr<PlanStage> root(parseQuery(dbname, argObj, runner.getWorkingSet(), &exprs));
             uassert(16911, "Couldn't parse plan from " + argObj.toString(), root.get());
             runner.setRoot(root.release());
 
             BSONArrayBuilder resultBuilder(result.subarrayStart("results"));
 
-            for (BSONObj obj; runner.getNext(&obj); ) {
+            for (BSONObj obj; Runner::RUNNER_ADVANCED == runner.getNext(&obj); ) {
                 resultBuilder.append(obj);
             }
 

src/mongo/db/query/cached_plan_runner.h

@@ -16,10 +16,11 @@
 
 #pragma once
 
+#include "mongo/db/parsed_query.h"
 #include "mongo/db/query/canonical_query.h"
 #include "mongo/db/query/plan_cache.h"
+#include "mongo/db/query/plan_executor.h"
 #include "mongo/db/query/runner.h"
-#include "mongo/db/query/simple_plan_runner.h"
 #include "mongo/db/query/stage_builder.h"
 
 namespace mongo {
@@ -40,53 +41,80 @@ namespace mongo {
         CachedPlanRunner(CanonicalQuery* canonicalQuery, CachedSolution* cached,
                          PlanStage* root, WorkingSet* ws)
             : _canonicalQuery(canonicalQuery), _cachedQuery(cached),
-              _runner(new SimplePlanRunner(ws, root)) { }
+              _exec(new PlanExecutor(ws, root)), _killed(false), _updatedCache(false) { }
 
-        bool getNext(BSONObj* objOut) {
-            // Use the underlying runner until it's exhausted.
-            if (_runner->getNext(objOut)) {
-                return true;
-            }
+        Runner::RunnerState getNext(BSONObj* objOut) {
+            if (_killed) { return Runner::RUNNER_DEAD; }
+
+            Runner::RunnerState state = _exec->getNext(objOut);
 
-            // We're done.  Update the cache.
-            PlanCache* cache = PlanCache::get(_canonicalQuery->ns());
+            if (Runner::RUNNER_EOF == state && !_updatedCache) {
+                // We're done.  Update the cache.
+                PlanCache* cache = PlanCache::get(_canonicalQuery->ns());
 
-            // TODO: is this a verify?
-            if (NULL == cache) { return false; }
+                // TODO: Is this an error?
+                if (NULL == cache) { return Runner::RUNNER_EOF; }
 
-            // TODO: How do we decide this?
-            bool shouldRemovePlan = false;
+                // TODO: How do we decide this?
+                bool shouldRemovePlan = false;
 
-            if (shouldRemovePlan) {
-                if (!cache->remove(*_canonicalQuery, *_cachedQuery->solution)) {
-                    warning() << "Cached plan runner couldn't remove plan from cache.  Maybe"
-                                 " somebody else did already?";
+                if (shouldRemovePlan) {
+                    if (!cache->remove(*_canonicalQuery, *_cachedQuery->solution)) {
+                        warning() << "Cached plan runner couldn't remove plan from cache.  Maybe"
+                            " somebody else did already?";
+                    }
+                    return Runner::RUNNER_EOF;
                 }
-                return false;
+
+                // We're done running.  Update cache.
+                auto_ptr<CachedSolutionFeedback> feedback(new CachedSolutionFeedback());
+                feedback->stats = _exec->getStats();
+                cache->feedback(*_canonicalQuery, *_cachedQuery->solution, feedback.release());
             }
+            return state;
+        }
 
-            // We're done running.  Update cache.
-            auto_ptr<CachedSolutionFeedback> feedback(new CachedSolutionFeedback());
-            feedback->stats = _runner->getStats();
-            cache->feedback(*_canonicalQuery, *_cachedQuery->solution, feedback.release());
-            return false;
+        virtual void saveState() {
+            if (!_killed) {
+                _exec->saveState();
+            }
         }
 
-        virtual void saveState() { _runner->saveState(); }
-        virtual void restoreState() { _runner->restoreState(); }
+        virtual void restoreState() {
+            if (!_killed) {
+                _exec->restoreState();
+            }
+        }
 
         virtual void invalidate(const DiskLoc& dl) {
-            _runner->invalidate(dl);
+            if (!_killed) {
+                _exec->invalidate(dl);
+            }
         }
 
-        virtual const CanonicalQuery& getQuery() {
-            return *_canonicalQuery;
+        virtual const CanonicalQuery& getQuery() { return *_canonicalQuery; }
+
+        virtual void kill() { _killed = true; }
+
+        virtual bool forceYield() {
+            saveState();
+            ClientCursor::registerRunner(this);
+            ClientCursor::staticYield(ClientCursor::suggestYieldMicros(), getQuery().getParsed().ns(), NULL);
+            ClientCursor::deregisterRunner(this);
+            if (!_killed) { restoreState(); }
+            return !_killed;
         }
 
     private:
         scoped_ptr<CanonicalQuery> _canonicalQuery;
         scoped_ptr<CachedSolution> _cachedQuery;
-        scoped_ptr<SimplePlanRunner> _runner;
+        scoped_ptr<PlanExecutor> _exec;
+
+        // Were we killed during a yield?
+        bool _killed;
+
+        // Have we updated the cache with our plan stats yet?
+        bool _updatedCache;
     };
 
 }  // namespace mongo

src/mongo/db/query/canonical_query.cpp

@@ -49,7 +49,11 @@ namespace mongo {
                                         CanonicalQuery** out) {
         auto_ptr<CanonicalQuery> cq(new CanonicalQuery());
 
-        cq->_pq.reset(new ParsedQuery(ns.c_str(), 0, 0, 0, query, BSONObj()));
+        // ParsedQuery saves the pointer to the NS that we provide it.  It's not going to remain
+        // valid unless we cache it ourselves.
+        cq->_ns = ns;
+
+        cq->_pq.reset(new ParsedQuery(cq->_ns.c_str(), 0, 0, 0, query, BSONObj()));
 
         StatusWithMatchExpression swme = MatchExpressionParser::parse(cq->_pq->getFilter());
         if (!swme.isOK()) { return swme.getStatus(); }

src/mongo/db/query/canonical_query.h

@@ -52,6 +52,8 @@ namespace mongo {
 
         // _root points into _pq->getFilter()
         scoped_ptr<MatchExpression> _root;
+
+        string _ns;
     };
 
 }  // namespace mongo