SERVER-22541 Manage aggregation cursors on global cursor manager.

Moves registration of aggregation cursors to the global cursor manager.
This simplifies the logic for acquiring locks and resolving view
namespaces within the getMore and killCursors commands.

42 files changed, 782 insertions, 723 deletions

diff --git a/src/mongo/db/catalog/collection.cpp b/src/mongo/db/catalog/collection.cpp
index a86944badf3..de7d36b205d 100644
--- a/src/mongo/db/catalog/collection.cpp
+++ b/src/mongo/db/catalog/collection.cpp
@@ -229,7 +229,7 @@ Collection::Collection(OperationContext* opCtx,
           parseValidationAction(_details->getCollectionOptions(opCtx).validationAction))),
       _validationLevel(uassertStatusOK(
           parseValidationLevel(_details->getCollectionOptions(opCtx).validationLevel))),
-      _cursorManager(fullNS),
+      _cursorManager(_ns),
       _cappedNotifier(_recordStore->isCapped() ? new CappedInsertNotifier() : nullptr),
       _mustTakeCappedLockOnInsert(isCapped() && !_ns.isSystemDotProfile() && !_ns.isOplog()) {
 
diff --git a/src/mongo/db/catalog/cursor_manager.cpp b/src/mongo/db/catalog/cursor_manager.cpp
index b2079e9e03e..25f0be7edec 100644
--- a/src/mongo/db/catalog/cursor_manager.cpp
+++ b/src/mongo/db/catalog/cursor_manager.cpp
@@ -50,42 +50,23 @@
 
 namespace mongo {
 
-using std::string;
 using std::vector;
 
 namespace {
-unsigned idFromCursorId(CursorId id) {
+uint32_t idFromCursorId(CursorId id) {
     uint64_t x = static_cast<uint64_t>(id);
     x = x >> 32;
-    return static_cast<unsigned>(x);
+    return static_cast<uint32_t>(x);
 }
 
-CursorId cursorIdFromParts(unsigned collection, unsigned cursor) {
-    CursorId x = static_cast<CursorId>(collection) << 32;
+CursorId cursorIdFromParts(uint32_t collectionIdentifier, uint32_t cursor) {
+    // The leading two bits of a non-global CursorId should be 0.
+    invariant((collectionIdentifier & (0b11 << 30)) == 0);
+    CursorId x = static_cast<CursorId>(collectionIdentifier) << 32;
     x |= cursor;
     return x;
 }
-
-class IdWorkTest : public StartupTest {
-public:
-    void _run(unsigned a, unsigned b) {
-        CursorId x = cursorIdFromParts(a, b);
-        invariant(a == idFromCursorId(x));
-        CursorId y = cursorIdFromParts(a, b + 1);
-        invariant(x != y);
-    }
-
-    void run() {
-        _run(123, 456);
-        _run(0xdeadbeef, 0xcafecafe);
-        _run(0, 0);
-        _run(99999999, 999);
-        _run(0xFFFFFFFF, 1);
-        _run(0xFFFFFFFF, 0);
-        _run(0xFFFFFFFF, 0xFFFFFFFF);
-    }
-} idWorkTest;
-}
+}  // namespace
 
 class GlobalCursorIdCache {
 public:
@@ -93,16 +74,16 @@ public:
     ~GlobalCursorIdCache();
 
     /**
-     * this gets called when a CursorManager gets created
-     * @return the id the CursorManager should use when generating
-     * cursor ids
+     * Returns a unique 32-bit identifier to be used as the first 32 bits of all cursor ids for a
+     * new CursorManager.
      */
-    unsigned created(const std::string& ns);
+    uint32_t registerCursorManager(const NamespaceString& nss);
 
     /**
-     * called by CursorManager when its going away
+     * Must be called when a CursorManager is deleted. 'id' must be the identifier returned by
+     * registerCursorManager().
      */
-    void destroyed(unsigned id, const std::string& ns);
+    void deregisterCursorManager(uint32_t id, const NamespaceString& nss);
 
     /**
      * works globally
@@ -118,8 +99,8 @@ public:
 private:
     SimpleMutex _mutex;
 
-    typedef unordered_map<unsigned, string> Map;
-    Map _idToNS;
+    typedef unordered_map<unsigned, NamespaceString> Map;
+    Map _idToNss;
     unsigned _nextId;
 
     std::unique_ptr<SecureRandom> _secureRandom;
@@ -137,7 +118,7 @@ MONGO_INITIALIZER(GlobalCursorIdCache)(InitializerContext* context) {
 
 MONGO_INITIALIZER_WITH_PREREQUISITES(GlobalCursorManager, ("GlobalCursorIdCache"))
 (InitializerContext* context) {
-    globalCursorManager.reset(new CursorManager(""));
+    globalCursorManager.reset(new CursorManager({}));
     return Status::OK();
 }
 
@@ -152,56 +133,57 @@ int64_t GlobalCursorIdCache::nextSeed() {
     return _secureRandom->nextInt64();
 }
 
-unsigned GlobalCursorIdCache::created(const std::string& ns) {
-    static const unsigned MAX_IDS = 1000 * 1000 * 1000;
+uint32_t GlobalCursorIdCache::registerCursorManager(const NamespaceString& nss) {
+    static const uint32_t kMaxIds = 1000 * 1000 * 1000;
+    static_assert((kMaxIds & (0b11 << 30)) == 0,
+                  "the first two bits of a collection identifier must always be zeroes");
 
     stdx::lock_guard<SimpleMutex> lk(_mutex);
 
-    fassert(17359, _idToNS.size() < MAX_IDS);
+    fassert(17359, _idToNss.size() < kMaxIds);
 
-    for (unsigned i = 0; i <= MAX_IDS; i++) {
-        unsigned id = ++_nextId;
+    for (uint32_t i = 0; i <= kMaxIds; i++) {
+        uint32_t id = ++_nextId;
         if (id == 0)
             continue;
-        if (_idToNS.count(id) > 0)
+        if (_idToNss.count(id) > 0)
             continue;
-        _idToNS[id] = ns;
+        _idToNss[id] = nss;
         return id;
     }
 
-    invariant(false);
+    MONGO_UNREACHABLE;
 }
 
-void GlobalCursorIdCache::destroyed(unsigned id, const std::string& ns) {
+void GlobalCursorIdCache::deregisterCursorManager(uint32_t id, const NamespaceString& nss) {
     stdx::lock_guard<SimpleMutex> lk(_mutex);
-    invariant(ns == _idToNS[id]);
-    _idToNS.erase(id);
+    invariant(nss == _idToNss[id]);
+    _idToNss.erase(id);
 }
 
 bool GlobalCursorIdCache::eraseCursor(OperationContext* opCtx, CursorId id, bool checkAuth) {
     // Figure out what the namespace of this cursor is.
-    std::string ns;
-    if (globalCursorManager->ownsCursorId(id)) {
-        auto pin = globalCursorManager.get()->pinCursor(id);
+    NamespaceString nss;
+    if (CursorManager::isGloballyManagedCursor(id)) {
+        auto pin = globalCursorManager->pinCursor(id);
         if (!pin.isOK()) {
             invariant(pin == ErrorCodes::CursorNotFound);
             // No such cursor.  TODO: Consider writing to audit log here (even though we don't
             // have a namespace).
             return false;
         }
-        ns = pin.getValue().getCursor()->ns();
+        nss = pin.getValue().getCursor()->nss();
     } else {
         stdx::lock_guard<SimpleMutex> lk(_mutex);
-        unsigned nsid = idFromCursorId(id);
-        Map::const_iterator it = _idToNS.find(nsid);
-        if (it == _idToNS.end()) {
+        uint32_t nsid = idFromCursorId(id);
+        Map::const_iterator it = _idToNss.find(nsid);
+        if (it == _idToNss.end()) {
             // No namespace corresponding to this cursor id prefix.  TODO: Consider writing to
             // audit log here (even though we don't have a namespace).
             return false;
         }
-        ns = it->second;
+        nss = it->second;
     }
-    const NamespaceString nss(ns);
     invariant(nss.isValid());
 
     // Check if we are authorized to erase this cursor.
@@ -215,7 +197,7 @@ bool GlobalCursorIdCache::eraseCursor(OperationContext* opCtx, CursorId id, bool
     }
 
     // If this cursor is owned by the global cursor manager, ask it to erase the cursor for us.
-    if (globalCursorManager->ownsCursorId(id)) {
+    if (CursorManager::isGloballyManagedCursor(id)) {
         Status eraseStatus = globalCursorManager->eraseCursor(opCtx, id, checkAuth);
         massert(28697,
                 eraseStatus.reason(),
@@ -250,16 +232,11 @@ std::size_t GlobalCursorIdCache::timeoutCursors(OperationContext* opCtx, int mil
     totalTimedOut += globalCursorManager->timeoutCursors(millisSinceLastCall);
 
     // Compute the set of collection names that we have to time out cursors for.
-    vector<string> todo;
+    vector<NamespaceString> todo;
     {
         stdx::lock_guard<SimpleMutex> lk(_mutex);
-        for (Map::const_iterator i = _idToNS.begin(); i != _idToNS.end(); ++i) {
-            if (globalCursorManager->ownsCursorId(cursorIdFromParts(i->first, 0))) {
-                // Skip the global cursor manager, since we handle it above (and it's not
-                // associated with a collection).
-                continue;
-            }
-            todo.push_back(i->second);
+        for (auto&& entry : _idToNss) {
+            todo.push_back(entry.second);
         }
     }
 
@@ -314,14 +291,19 @@ bool CursorManager::eraseCursorGlobal(OperationContext* opCtx, CursorId id) {
 // --------------------------
 
 
-CursorManager::CursorManager(StringData ns) : _nss(ns) {
-    _collectionCacheRuntimeId = globalCursorIdCache->created(_nss.ns());
+CursorManager::CursorManager(NamespaceString nss) : _nss(std::move(nss)) {
+    if (!isGlobalManager()) {
+        // Generate a unique id for this collection.
+        _collectionCacheRuntimeId = globalCursorIdCache->registerCursorManager(_nss);
+    }
     _random.reset(new PseudoRandom(globalCursorIdCache->nextSeed()));
 }
 
 CursorManager::~CursorManager() {
     invalidateAll(true, "collection going away");
-    globalCursorIdCache->destroyed(_collectionCacheRuntimeId, _nss.ns());
+    if (!isGlobalManager()) {
+        globalCursorIdCache->deregisterCursorManager(_collectionCacheRuntimeId, _nss);
+    }
 }
 
 void CursorManager::invalidateAll(bool collectionGoingAway, const std::string& reason) {
@@ -371,11 +353,8 @@ void CursorManager::invalidateAll(bool collectionGoingAway, const std::string& r
                     continue;
                 }
 
-                if (cc->_isPinned || cc->isAggCursor()) {
-                    // Pinned cursors need to stay alive, so we leave them around.  Aggregation
-                    // cursors also can stay alive (since they don't have their lifetime bound to
-                    // the underlying collection).  However, if they have an associated executor, we
-                    // need to kill it, because it's now invalid.
+                if (cc->_isPinned) {
+                    // Pinned cursors need to stay alive, so we leave them around.
                     if (cc->getExecutor())
                         cc->getExecutor()->kill(reason);
                     newMap.insert(*i);
@@ -484,10 +463,6 @@ void CursorManager::unpin(ClientCursor* cursor) {
     cursor->_isPinned = false;
 }
 
-bool CursorManager::ownsCursorId(CursorId cursorId) const {
-    return _collectionCacheRuntimeId == idFromCursorId(cursorId);
-}
-
 void CursorManager::getCursorIds(std::set<CursorId>* openCursors) const {
     stdx::lock_guard<SimpleMutex> lk(_mutex);
 
@@ -504,19 +479,32 @@ size_t CursorManager::numCursors() const {
 
 CursorId CursorManager::_allocateCursorId_inlock() {
     for (int i = 0; i < 10000; i++) {
-        unsigned mypart = static_cast<unsigned>(_random->nextInt32());
-        CursorId id = cursorIdFromParts(_collectionCacheRuntimeId, mypart);
+        // The leading two bits of a CursorId are used to determine if the cursor is registered on
+        // the global cursor manager.
+        CursorId id;
+        if (isGlobalManager()) {
+            // This is the global cursor manager, so generate a random number and make sure the
+            // first two bits are 01.
+            uint64_t mask = 0x3FFFFFFFFFFFFFFF;
+            uint64_t bitToSet = 1ULL << 62;
+            id = ((_random->nextInt64() & mask) | bitToSet);
+        } else {
+            // The first 2 bits are 0, the next 30 bits are the collection identifier, the next 32
+            // bits are random.
+            uint32_t myPart = static_cast<uint32_t>(_random->nextInt32());
+            id = cursorIdFromParts(_collectionCacheRuntimeId, myPart);
+        }
         if (_cursors.count(id) == 0)
             return id;
     }
     fassertFailed(17360);
 }
 
-ClientCursorPin CursorManager::registerCursor(const ClientCursorParams& cursorParams) {
+ClientCursorPin CursorManager::registerCursor(ClientCursorParams&& cursorParams) {
     stdx::lock_guard<SimpleMutex> lk(_mutex);
     CursorId cursorId = _allocateCursorId_inlock();
     std::unique_ptr<ClientCursor, ClientCursor::Deleter> clientCursor(
-        new ClientCursor(cursorParams, this, cursorId));
+        new ClientCursor(std::move(cursorParams), this, cursorId));
     return _registerCursor_inlock(std::move(clientCursor));
 }
 
diff --git a/src/mongo/db/catalog/cursor_manager.h b/src/mongo/db/catalog/cursor_manager.h
index ad4289d4f38..7e4da59b626 100644
--- a/src/mongo/db/catalog/cursor_manager.h
+++ b/src/mongo/db/catalog/cursor_manager.h
@@ -73,7 +73,7 @@ class PlanExecutor;
  */
 class CursorManager {
 public:
-    CursorManager(StringData ns);
+    CursorManager(NamespaceString nss);
 
     /**
      * Destroys the CursorManager. Managed cursors which are not pinned are destroyed. Ownership of
@@ -124,7 +124,7 @@ public:
      * Constructs a new ClientCursor according to the given 'cursorParams'. The cursor is atomically
      * registered with the manager and returned in pinned state.
      */
-    ClientCursorPin registerCursor(const ClientCursorParams& cursorParams);
+    ClientCursorPin registerCursor(ClientCursorParams&& cursorParams);
 
     /**
      * Constructs and pins a special ClientCursor used to track sharding state for the given
@@ -153,15 +153,6 @@ public:
      */
     Status eraseCursor(OperationContext* opCtx, CursorId id, bool shouldAudit);
 
-    /**
-     * Returns true if the space of cursor ids that cursor manager is responsible for includes
-     * the given cursor id.  Otherwise, returns false.
-     *
-     * The return value of this method does not indicate any information about whether or not a
-     * cursor actually exists with the given cursor id.
-     */
-    bool ownsCursorId(CursorId cursorId) const;
-
     void getCursorIds(std::set<CursorId>* openCursors) const;
 
     /**
@@ -172,6 +163,17 @@ public:
 
     static CursorManager* getGlobalCursorManager();
 
+    /**
+     * Returns true if this CursorId would be registered with the global CursorManager. Note that if
+     * this method returns true it does not imply the cursor exists.
+     */
+    static bool isGloballyManagedCursor(CursorId cursorId) {
+        // The first two bits are 01 for globally managed cursors, and 00 for cursors owned by a
+        // collection. The leading bit is always 0 so that CursorIds do not appear as negative.
+        const long long mask = static_cast<long long>(0b11) << 62;
+        return (cursorId & mask) == (static_cast<long long>(0b01) << 62);
+    }
+
     static int eraseCursorGlobalIfAuthorized(OperationContext* opCtx, int n, const char* ids);
 
     static bool eraseCursorGlobalIfAuthorized(OperationContext* opCtx, CursorId id);
@@ -196,8 +198,12 @@ private:
 
     void unpin(ClientCursor* cursor);
 
+    bool isGlobalManager() const {
+        return _nss.isEmpty();
+    }
+
     NamespaceString _nss;
-    unsigned _collectionCacheRuntimeId;
+    uint32_t _collectionCacheRuntimeId;
     std::unique_ptr<PseudoRandom> _random;
 
     mutable SimpleMutex _mutex;
diff --git a/src/mongo/db/clientcursor.cpp b/src/mongo/db/clientcursor.cpp
index 708a2bc38f7..70003c2da2b 100644
--- a/src/mongo/db/clientcursor.cpp
+++ b/src/mongo/db/clientcursor.cpp
@@ -77,17 +77,16 @@ long long ClientCursor::totalOpen() {
     return cursorStatsOpen.get();
 }
 
-ClientCursor::ClientCursor(const ClientCursorParams& params,
+ClientCursor::ClientCursor(ClientCursorParams&& params,
                            CursorManager* cursorManager,
                            CursorId cursorId)
     : _cursorid(cursorId),
-      _ns(params.ns),
+      _nss(std::move(params.nss)),
       _isReadCommitted(params.isReadCommitted),
       _cursorManager(cursorManager),
-      _query(params.query),
-      _queryOptions(params.qopts),
-      _isAggCursor(params.isAggCursor) {
-    _exec.reset(params.exec);
+      _originatingCommand(params.originatingCommandObj),
+      _queryOptions(params.queryOptions),
+      _exec(std::move(params.exec)) {
     init();
 }
 
@@ -95,7 +94,7 @@ ClientCursor::ClientCursor(const Collection* collection,
                            CursorManager* cursorManager,
                            CursorId cursorId)
     : _cursorid(cursorId),
-      _ns(collection->ns().ns()),
+      _nss(collection->ns()),
       _cursorManager(cursorManager),
       _queryOptions(QueryOption_NoCursorTimeout) {
     init();
@@ -106,10 +105,9 @@ void ClientCursor::init() {
 
     cursorStatsOpen.increment();
 
-    if (_queryOptions & QueryOption_NoCursorTimeout) {
+    if (isNoTimeout()) {
         // cursors normally timeout after an inactivity period to prevent excess memory use
         // setting this prevents timeout of the cursor in question.
-        _isNoTimeout = true;
         cursorStatsOpenNoTimeout.increment();
     }
 }
@@ -120,7 +118,7 @@ ClientCursor::~ClientCursor() {
     invariant(!_cursorManager);
 
     cursorStatsOpen.decrement();
-    if (_isNoTimeout) {
+    if (isNoTimeout()) {
         cursorStatsOpenNoTimeout.decrement();
     }
 }
@@ -138,7 +136,7 @@ void ClientCursor::kill() {
 
 bool ClientCursor::shouldTimeout(int millis) {
     _idleAgeMillis += millis;
-    if (_isNoTimeout || _isPinned) {
+    if (isNoTimeout() || _isPinned) {
         return false;
     }
     return _idleAgeMillis > cursorTimeoutMillis.load();
@@ -152,7 +150,7 @@ void ClientCursor::updateSlaveLocation(OperationContext* opCtx) {
     if (_slaveReadTill.isNull())
         return;
 
-    verify(str::startsWith(_ns.c_str(), "local.oplog."));
+    verify(_nss.isOplog());
 
     Client* c = opCtx->getClient();
     verify(c);
@@ -221,7 +219,7 @@ void ClientCursorPin::release() {
 
     if (!_cursor->_cursorManager) {
         // The ClientCursor was killed while we had it.  Therefore, it is our responsibility to
-        // kill it.
+        // delete it.
         deleteUnderlying();
     } else {
         // Unpin the cursor under the collection cursor manager lock.
diff --git a/src/mongo/db/clientcursor.h b/src/mongo/db/clientcursor.h
index 44ae4600ac6..91e7a0d325c 100644
--- a/src/mongo/db/clientcursor.h
+++ b/src/mongo/db/clientcursor.h
@@ -28,6 +28,7 @@
 
 #pragma once
 
+#include "mongo/client/dbclientinterface.h"
 #include "mongo/db/cursor_id.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/db/query/plan_executor.h"
@@ -42,30 +43,30 @@ class CursorManager;
 class RecoveryUnit;
 
 /**
- * Parameters used for constructing a ClientCursor. ClientCursors cannot be constructed in
- * isolation, but rather must be constructed and managed using a CursorManager. See cursor_manager.h
- * for more details.
+ * Parameters used for constructing a ClientCursor. Makes an owned copy of 'originatingCommandObj'
+ * to be used across getMores.
+ *
+ * ClientCursors cannot be constructed in isolation, but rather must be
+ * constructed and managed using a CursorManager. See cursor_manager.h for more details.
  */
 struct ClientCursorParams {
-    ClientCursorParams(PlanExecutor* exec,
-                       std::string ns,
+    ClientCursorParams(std::unique_ptr<PlanExecutor> planExecutor,
+                       NamespaceString nss,
                        bool isReadCommitted,
-                       int qopts = 0,
-                       const BSONObj query = BSONObj(),
-                       bool isAggCursor = false)
-        : exec(exec),
-          ns(std::move(ns)),
+                       BSONObj originatingCommandObj)
+        : exec(std::move(planExecutor)),
+          nss(std::move(nss)),
           isReadCommitted(isReadCommitted),
-          qopts(qopts),
-          query(query),
-          isAggCursor(isAggCursor) {}
+          queryOptions(exec->getCanonicalQuery()
+                           ? exec->getCanonicalQuery()->getQueryRequest().getOptions()
+                           : 0),
+          originatingCommandObj(originatingCommandObj.getOwned()) {}
 
-    PlanExecutor* exec = nullptr;
-    const std::string ns;
+    std::unique_ptr<PlanExecutor> exec;
+    const NamespaceString nss;
     bool isReadCommitted = false;
-    int qopts = 0;
-    const BSONObj query = BSONObj();
-    bool isAggCursor = false;
+    int queryOptions = 0;
+    BSONObj originatingCommandObj;
 };
 
 /**
@@ -92,18 +93,14 @@ public:
         return _cursorid;
     }
 
-    std::string ns() const {
-        return _ns;
+    const NamespaceString& nss() const {
+        return _nss;
     }
 
     bool isReadCommitted() const {
         return _isReadCommitted;
     }
 
-    bool isAggCursor() const {
-        return _isAggCursor;
-    }
-
     PlanExecutor* getExecutor() const {
         return _exec.get();
     }
@@ -112,8 +109,8 @@ public:
         return _queryOptions;
     }
 
-    const BSONObj& getQuery() const {
-        return _query;
+    const BSONObj& getOriginatingCommandObj() const {
+        return _originatingCommand;
     }
 
     /**
@@ -223,7 +220,7 @@ private:
      * Constructs a ClientCursor. Since cursors must come into being registered and pinned, this is
      * private. See cursor_manager.h for more details.
      */
-    ClientCursor(const ClientCursorParams& params, CursorManager* cursorManager, CursorId cursorId);
+    ClientCursor(ClientCursorParams&& params, CursorManager* cursorManager, CursorId cursorId);
 
     /**
      * Constructs a special ClientCursor used to track sharding state for the given collection.
@@ -246,11 +243,15 @@ private:
      */
     void kill();
 
+    bool isNoTimeout() const {
+        return (_queryOptions & QueryOption_NoCursorTimeout);
+    }
+
     // The ID of the ClientCursor. A value of 0 is used to mean that no cursor id has been assigned.
     CursorId _cursorid = 0;
 
     // The namespace we're operating on.
-    std::string _ns;
+    const NamespaceString _nss;
 
     const bool _isReadCommitted = false;
 
@@ -265,21 +266,11 @@ private:
     // Tracks the number of results returned by this cursor so far.
     long long _pos = 0;
 
-    // If this cursor was created by a find operation, '_query' holds the query predicate for
-    // the find. If this cursor was created by a command (e.g. the aggregate command), then
-    // '_query' holds the command specification received from the client.
-    BSONObj _query;
+    // Holds an owned copy of the command specification received from the client.
+    const BSONObj _originatingCommand;
 
     // See the QueryOptions enum in dbclientinterface.h.
-    int _queryOptions = 0;
-
-    // Is this ClientCursor backed by an aggregation pipeline?
-    //
-    // Agg executors differ from others in that they manage their own locking internally and
-    // should not be killed or destroyed when the underlying collection is deleted.
-    //
-    // Note: This should *not* be set for the internal cursor used as input to an aggregation.
-    const bool _isAggCursor = false;
+    const int _queryOptions = 0;
 
     // While a cursor is being used by a client, it is marked as "pinned". See ClientCursorPin
     // below.
@@ -287,10 +278,6 @@ private:
     // Cursors always come into existence in a pinned state.
     bool _isPinned = true;
 
-    // Is the "no timeout" flag set on this cursor?  If false, this cursor may be automatically
-    // deleted after an interval of inactivity.
-    bool _isNoTimeout = false;
-
     // The replication position only used in master-slave.
     Timestamp _slaveReadTill;
 
diff --git a/src/mongo/db/commands/find_and_modify.cpp b/src/mongo/db/commands/find_and_modify.cpp
index a2deb9db62a..cd66033e427 100644
--- a/src/mongo/db/commands/find_and_modify.cpp
+++ b/src/mongo/db/commands/find_and_modify.cpp
@@ -203,13 +203,11 @@ Status checkCanAcceptWritesForDatabase(OperationContext* opCtx, const NamespaceS
 
 void recordStatsForTopCommand(OperationContext* opCtx) {
     auto curOp = CurOp::get(opCtx);
-    const int writeLocked = 1;
-
     Top::get(opCtx->getClient()->getServiceContext())
         .record(opCtx,
                 curOp->getNS(),
                 curOp->getLogicalOp(),
-                writeLocked,
+                Top::LockType::WriteLocked,
                 curOp->elapsedMicros(),
                 curOp->isCommand(),
                 curOp->getReadWriteType());
diff --git a/src/mongo/db/commands/find_cmd.cpp b/src/mongo/db/commands/find_cmd.cpp
index d43e991950c..60ae2e2eba4 100644
--- a/src/mongo/db/commands/find_cmd.cpp
+++ b/src/mongo/db/commands/find_cmd.cpp
@@ -399,11 +399,10 @@ public:
             // Create a ClientCursor containing this plan executor and register it with the cursor
             // manager.
             ClientCursorPin pinnedCursor = collection->getCursorManager()->registerCursor(
-                {exec.release(),
-                 nss.ns(),
+                {std::move(exec),
+                 nss,
                  opCtx->recoveryUnit()->isReadingFromMajorityCommittedSnapshot(),
-                 originalQR.getOptions(),
-                 cmdObj.getOwned()});
+                 cmdObj});
             cursorId = pinnedCursor.getCursor()->cursorid();
 
             invariant(!exec);
diff --git a/src/mongo/db/commands/getmore_cmd.cpp b/src/mongo/db/commands/getmore_cmd.cpp
index 78c7d822404..448177a8068 100644
--- a/src/mongo/db/commands/getmore_cmd.cpp
+++ b/src/mongo/db/commands/getmore_cmd.cpp
@@ -53,6 +53,7 @@
 #include "mongo/db/s/operation_sharding_state.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/stats/counters.h"
+#include "mongo/db/stats/top.h"
 #include "mongo/s/chunk_version.h"
 #include "mongo/stdx/memory.h"
 #include "mongo/util/fail_point_service.h"
@@ -154,9 +155,6 @@ public:
         auto curOp = CurOp::get(opCtx);
         curOp->debug().cursorid = request.cursorid;
 
-        // Disable shard version checking - getmore commands are always unversioned
-        OperationShardingState::get(opCtx).setShardVersion(request.nss, ChunkVersion::IGNORED());
-
         // Validate term before acquiring locks, if provided.
         if (request.term) {
             auto replCoord = repl::ReplicationCoordinator::get(opCtx);
@@ -167,67 +165,47 @@ public:
             }
         }
 
-        // Depending on the type of cursor being operated on, we hold locks for the whole
-        // getMore, or none of the getMore, or part of the getMore.  The three cases in detail:
+        // Cursors come in one of two flavors:
+        // - Cursors owned by the collection cursor manager, such as those generated via the find
+        //   command. For these cursors, we hold the appropriate collection lock for the duration of
+        //   the getMore using AutoGetCollectionForRead. This will automatically update the CurOp
+        //   object appropriately and record execution time via Top upon completion.
+        // - Cursors owned by the global cursor manager, such as those generated via the aggregate
+        //   command. These cursors either hold no collection state or manage their collection state
+        //   internally, so we acquire no locks. In this case we use the AutoStatsTracker object to
+        //   update the CurOp object appropriately and record execution time via Top upon
+        //   completion.
         //
-        // 1) Normal cursor: we lock with "ctx" and hold it for the whole getMore.
-        // 2) Cursor owned by global cursor manager: we don't lock anything.  These cursors
-        //    don't own any collection state. These cursors are generated either by the
-        //    listCollections or listIndexes commands, as these special cursor-generating commands
-        //    operate over catalog data rather than targeting the data within a collection.
-        // 3) Agg cursor: we lock with "ctx", then release, then relock with "unpinDBLock" and
-        //    "unpinCollLock".  This is because agg cursors handle locking internally (hence the
-        //    release), but the pin and unpin of the cursor must occur under the collection
-        //    lock. We don't use our AutoGetCollectionForRead "ctx" to relock, because
-        //    AutoGetCollectionForRead checks the sharding version (and we want the relock for
-        //    the unpin to succeed even if the sharding version has changed).
+        // Thus, only one of 'readLock' and 'statsTracker' will be populated as we populate
+        // 'cursorManager'.
         //
         // Note that we declare our locks before our ClientCursorPin, in order to ensure that
-        // the pin's destructor is called before the lock destructors (so that the unpin occurs
-        // under the lock).
-        std::unique_ptr<AutoGetCollectionForReadCommand> ctx;
-        std::unique_ptr<Lock::DBLock> unpinDBLock;
-        std::unique_ptr<Lock::CollectionLock> unpinCollLock;
-
+        // the pin's destructor is called before the lock's destructor (if there is one) so that the
+        // cursor cleanup can occur under the lock.
+        boost::optional<AutoGetCollectionForReadCommand> readLock;
+        boost::optional<AutoStatsTracker> statsTracker;
         CursorManager* cursorManager;
-        if (request.nss.isListIndexesCursorNS() || request.nss.isListCollectionsCursorNS()) {
+
+        if (CursorManager::isGloballyManagedCursor(request.cursorid)) {
             cursorManager = CursorManager::getGlobalCursorManager();
+
+            if (boost::optional<NamespaceString> nssForCurOp =
+                    request.nss.isGloballyManagedNamespace()
+                    ? request.nss.getTargetNSForGloballyManagedNamespace()
+                    : request.nss) {
+                const boost::optional<int> dbProfilingLevel = boost::none;
+                statsTracker.emplace(
+                    opCtx, *nssForCurOp, Top::LockType::NotLocked, dbProfilingLevel);
+            }
         } else {
-            ctx = stdx::make_unique<AutoGetCollectionOrViewForReadCommand>(opCtx, request.nss);
-            auto viewCtx = static_cast<AutoGetCollectionOrViewForReadCommand*>(ctx.get());
-            Collection* collection = ctx->getCollection();
+            // getMore commands are always unversioned, so prevent AutoGetCollectionForRead from
+            // checking the shard version.
+            OperationShardingState::get(opCtx).setShardVersion(request.nss,
+                                                               ChunkVersion::IGNORED());
+
+            readLock.emplace(opCtx, request.nss);
+            Collection* collection = readLock->getCollection();
             if (!collection) {
-                // Rewrite a getMore on a view to a getMore on the original underlying collection.
-                // If the view no longer exists, or has been rewritten, the cursor id will be
-                // unknown, resulting in an appropriate error.
-                if (viewCtx->getView()) {
-                    auto resolved =
-                        viewCtx->getDb()->getViewCatalog()->resolveView(opCtx, request.nss);
-                    if (!resolved.isOK()) {
-                        return appendCommandStatus(result, resolved.getStatus());
-                    }
-                    viewCtx->releaseLocksForView();
-
-                    // Only one shardversion can be set at a time for an operation, so unset it
-                    // here to allow setting it on the underlying namespace.
-                    OperationShardingState::get(opCtx).unsetShardVersion(request.nss);
-
-                    GetMoreRequest newRequest(resolved.getValue().getNamespace(),
-                                              request.cursorid,
-                                              request.batchSize,
-                                              request.awaitDataTimeout,
-                                              request.term,
-                                              request.lastKnownCommittedOpTime);
-
-                    bool retVal = runParsed(opCtx, origNss, newRequest, cmdObj, errmsg, result);
-                    {
-                        // Set the namespace of the curop back to the view namespace so ctx records
-                        // stats on this view namespace on destruction.
-                        stdx::lock_guard<Client> lk(*opCtx->getClient());
-                        curOp->setNS_inlock(origNss.ns());
-                    }
-                    return retVal;
-                }
                 return appendCommandStatus(result,
                                            Status(ErrorCodes::OperationFailed,
                                                   "collection dropped between getMore calls"));
@@ -243,23 +221,24 @@ public:
 
         ClientCursor* cursor = ccPin.getValue().getCursor();
 
-        // If the fail point is enabled, busy wait until it is disabled. We unlock and re-acquire
-        // the locks periodically in order to avoid deadlock (see SERVER-21997 for details).
+        // If the fail point is enabled, busy wait until it is disabled.
         while (MONGO_FAIL_POINT(keepCursorPinnedDuringGetMore)) {
-            invariant(ctx);
-            invariant(!unpinDBLock);
-            invariant(!unpinCollLock);
-            sleepFor(Milliseconds(10));
-            ctx.reset();
-            ctx = stdx::make_unique<AutoGetCollectionForReadCommand>(opCtx, request.nss);
+            if (readLock) {
+                // We unlock and re-acquire the locks periodically in order to avoid deadlock (see
+                // SERVER-21997 for details).
+                sleepFor(Milliseconds(10));
+                readLock.reset();
+                readLock.emplace(opCtx, request.nss);
+            }
         }
 
-        if (request.nss.ns() != cursor->ns()) {
+        if (request.nss != cursor->nss()) {
             return appendCommandStatus(
                 result,
                 Status(ErrorCodes::Unauthorized,
                        str::stream() << "Requested getMore on namespace '" << request.nss.ns()
-                                     << "', but cursor belongs to a different namespace"));
+                                     << "', but cursor belongs to a different namespace "
+                                     << cursor->nss().ns()));
         }
 
         if (request.nss.isOplog() && MONGO_FAIL_POINT(rsStopGetMoreCmd)) {
@@ -274,9 +253,8 @@ public:
         if (isCursorAwaitData(cursor)) {
             invariant(isCursorTailable(cursor));
 
-            if (cursor->isAggCursor()) {
-                Status status(ErrorCodes::BadValue,
-                              "awaitData cannot be set on an aggregation cursor");
+            if (CursorManager::isGloballyManagedCursor(request.cursorid)) {
+                Status status(ErrorCodes::BadValue, "awaitData cannot be set on this cursor");
                 return appendCommandStatus(result, status);
             }
         }
@@ -288,8 +266,7 @@ public:
         }
 
         // On early return, get rid of the cursor.
-        ScopeGuard cursorFreer =
-            MakeGuard(&GetMoreCmd::cleanupCursor, opCtx, &ccPin.getValue(), request);
+        ScopeGuard cursorFreer = MakeGuard(&ClientCursorPin::deleteUnderlying, &ccPin.getValue());
 
         if (cursor->isReadCommitted())
             uassertStatusOK(opCtx->recoveryUnit()->setReadFromMajorityCommittedSnapshot());
@@ -318,11 +295,6 @@ public:
         }
         opCtx->checkForInterrupt();  // May trigger maxTimeAlwaysTimeOut fail point.
 
-        if (cursor->isAggCursor()) {
-            // Agg cursors handle their own locking internally.
-            ctx.reset();  // unlocks
-        }
-
         PlanExecutor* exec = cursor->getExecutor();
         exec->reattachToOperationContext(opCtx);
         exec->restoreState();
@@ -334,7 +306,7 @@ public:
 
             // Ensure that the original query or command object is available in the slow query log,
             // profiler and currentOp.
-            auto originatingCommand = cursor->getQuery();
+            auto originatingCommand = cursor->getOriginatingCommandObj();
             if (!originatingCommand.isEmpty()) {
                 curOp->setOriginatingCommand_inlock(originatingCommand);
             }
@@ -343,12 +315,12 @@ public:
         uint64_t notifierVersion = 0;
         std::shared_ptr<CappedInsertNotifier> notifier;
         if (isCursorAwaitData(cursor)) {
-            invariant(ctx->getCollection()->isCapped());
+            invariant(readLock->getCollection()->isCapped());
             // Retrieve the notifier which we will wait on until new data arrives. We make sure
             // to do this in the lock because once we drop the lock it is possible for the
             // collection to become invalid. The notifier itself will outlive the collection if
             // the collection is dropped, as we keep a shared_ptr to it.
-            notifier = ctx->getCollection()->getCappedInsertNotifier();
+            notifier = readLock->getCollection()->getCappedInsertNotifier();
 
             // Must get the version before we call generateBatch in case a write comes in after
             // that call and before we call wait on the notifier.
@@ -386,11 +358,11 @@ public:
                 // to do this in the lock because once we drop the lock it is possible for the
                 // collection to become invalid. The notifier itself will outlive the collection if
                 // the collection is dropped, as we keep a shared_ptr to it.
-                auto notifier = ctx->getCollection()->getCappedInsertNotifier();
+                auto notifier = readLock->getCollection()->getCappedInsertNotifier();
 
                 // Save the PlanExecutor and drop our locks.
                 exec->saveState();
-                ctx.reset();
+                readLock.reset();
 
                 // Block waiting for data.
                 const auto timeout = opCtx->getRemainingMaxTimeMicros();
@@ -402,7 +374,7 @@ public:
                 // CappedInsertNotifier.
                 curOp->setExpectedLatencyMs(durationCount<Milliseconds>(timeout));
 
-                ctx.reset(new AutoGetCollectionForReadCommand(opCtx, request.nss));
+                readLock.emplace(opCtx, request.nss);
                 exec->restoreState();
 
                 // We woke up because either the timed_wait expired, or there was more data. Either
@@ -420,11 +392,11 @@ public:
         postExecutionStats.totalDocsExamined -= preExecutionStats.totalDocsExamined;
         curOp->debug().setPlanSummaryMetrics(postExecutionStats);
 
-        // We do not report 'execStats' for aggregation, both in the original request and
-        // subsequent getMore. The reason for this is that aggregation's source PlanExecutor
-        // could be destroyed before we know whether we need execStats and we do not want to
-        // generate for all operations due to cost.
-        if (!cursor->isAggCursor() && curOp->shouldDBProfile()) {
+        // We do not report 'execStats' for aggregation or other globally managed cursors, both in
+        // the original request and subsequent getMore. It would be useful to have this information
+        // for an aggregation, but the source PlanExecutor could be destroyed before we know whether
+        // we need execStats and we do not want to generate for all operations due to cost.
+        if (!CursorManager::isGloballyManagedCursor(request.cursorid) && curOp->shouldDBProfile()) {
             BSONObjBuilder execStatsBob;
             Explain::getWinningPlanStats(exec, &execStatsBob);
             curOp->debug().execStats = execStatsBob.obj();
@@ -448,9 +420,7 @@ public:
             curOp->debug().cursorExhausted = true;
         }
 
-        // Respond with the originally requested namespace, even if this is a getMore over a view
-        // that was resolved to a different backing namespace.
-        nextBatch.done(respondWithId, origNss.ns());
+        nextBatch.done(respondWithId, request.nss.ns());
 
         // Ensure log and profiler include the number of results returned in this getMore's response
         // batch.
@@ -458,15 +428,6 @@ public:
 
         if (respondWithId) {
             cursorFreer.Dismiss();
-
-            // If we are operating on an aggregation cursor, then we dropped our collection lock
-            // earlier and need to reacquire it in order to clean up our ClientCursorPin.
-            if (cursor->isAggCursor()) {
-                invariant(NULL == ctx.get());
-                unpinDBLock.reset(new Lock::DBLock(opCtx, request.nss.db(), MODE_IS));
-                unpinCollLock.reset(
-                    new Lock::CollectionLock(opCtx->lockState(), request.nss.ns(), MODE_IS));
-            }
         }
 
         return true;
@@ -554,27 +515,6 @@ public:
         return Status::OK();
     }
 
-    /**
-     * Called via a ScopeGuard on early return in order to ensure that the ClientCursor gets
-     * cleaned up properly.
-     */
-    static void cleanupCursor(OperationContext* opCtx,
-                              ClientCursorPin* ccPin,
-                              const GetMoreRequest& request) {
-        ClientCursor* cursor = ccPin->getCursor();
-
-        std::unique_ptr<Lock::DBLock> unpinDBLock;
-        std::unique_ptr<Lock::CollectionLock> unpinCollLock;
-
-        if (cursor->isAggCursor()) {
-            unpinDBLock.reset(new Lock::DBLock(opCtx, request.nss.db(), MODE_IS));
-            unpinCollLock.reset(
-                new Lock::CollectionLock(opCtx->lockState(), request.nss.ns(), MODE_IS));
-        }
-
-        ccPin->deleteUnderlying();
-    }
-
 } getMoreCmd;
 
 }  // namespace mongo
diff --git a/src/mongo/db/commands/killcursors_cmd.cpp b/src/mongo/db/commands/killcursors_cmd.cpp
index c1526d990b5..a39f0dfaa42 100644
--- a/src/mongo/db/commands/killcursors_cmd.cpp
+++ b/src/mongo/db/commands/killcursors_cmd.cpp
@@ -35,6 +35,8 @@
 #include "mongo/db/curop.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/query/killcursors_request.h"
+#include "mongo/db/stats/top.h"
+#include "mongo/util/scopeguard.h"
 
 namespace mongo {
 
@@ -48,34 +50,51 @@ private:
     Status _killCursor(OperationContext* opCtx,
                        const NamespaceString& nss,
                        CursorId cursorId) final {
-        std::unique_ptr<AutoGetCollectionOrViewForReadCommand> ctx;
-
+        // Cursors come in one of two flavors:
+        // - Cursors owned by the collection cursor manager, such as those generated via the find
+        //   command. For these cursors, we hold the appropriate collection lock for the duration of
+        //   the getMore using AutoGetCollectionForRead. This will automatically update the CurOp
+        //   object appropriately and record execution time via Top upon completion.
+        // - Cursors owned by the global cursor manager, such as those generated via the aggregate
+        //   command. These cursors either hold no collection state or manage their collection state
+        //   internally, so we acquire no locks. In this case we use the AutoStatsTracker object to
+        //   update the CurOp object appropriately and record execution time via Top upon
+        //   completion.
+        //
+        // Thus, exactly one of 'readLock' and 'statsTracker' will be populated as we populate
+        // 'cursorManager'.
+        boost::optional<AutoGetCollectionForReadCommand> readLock;
+        boost::optional<AutoStatsTracker> statsTracker;
         CursorManager* cursorManager;
-        if (nss.isListIndexesCursorNS() || nss.isListCollectionsCursorNS()) {
-            // listCollections and listIndexes are special cursor-generating commands whose cursors
-            // are managed globally, as they operate over catalog data rather than targeting the
-            // data within a collection.
+
+        if (CursorManager::isGloballyManagedCursor(cursorId)) {
             cursorManager = CursorManager::getGlobalCursorManager();
-        } else {
-            ctx = stdx::make_unique<AutoGetCollectionOrViewForReadCommand>(opCtx, nss);
-            Collection* collection = ctx->getCollection();
-            ViewDefinition* view = ctx->getView();
-            if (view) {
-                Database* db = ctx->getDb();
-                auto resolved = db->getViewCatalog()->resolveView(opCtx, nss);
-                if (!resolved.isOK()) {
-                    return resolved.getStatus();
-                }
-                ctx->releaseLocksForView();
-                Status status = _killCursor(opCtx, resolved.getValue().getNamespace(), cursorId);
-                {
-                    // Set the namespace of the curop back to the view namespace so ctx records
-                    // stats on this view namespace on destruction.
-                    stdx::lock_guard<Client> lk(*opCtx->getClient());
-                    CurOp::get(opCtx)->setNS_inlock(nss.ns());
+
+            if (auto nssForCurOp = nss.isGloballyManagedNamespace()
+                    ? nss.getTargetNSForGloballyManagedNamespace()
+                    : nss) {
+                const boost::optional<int> dbProfilingLevel = boost::none;
+                statsTracker.emplace(
+                    opCtx, *nssForCurOp, Top::LockType::NotLocked, dbProfilingLevel);
+            }
+
+            // Make sure the namespace of the cursor matches the namespace passed to the killCursors
+            // command so we can be sure we checked the correct privileges.
+            auto ccPin = cursorManager->pinCursor(cursorId);
+            if (ccPin.isOK()) {
+                auto cursorNs = ccPin.getValue().getCursor()->nss();
+                if (cursorNs != nss) {
+                    return Status{ErrorCodes::Unauthorized,
+                                  str::stream() << "issued killCursors on namespace '" << nss.ns()
+                                                << "', but cursor with id "
+                                                << cursorId
+                                                << " belongs to a different namespace: "
+                                                << cursorNs.ns()};
                 }
-                return status;
             }
+        } else {
+            readLock.emplace(opCtx, nss);
+            Collection* collection = readLock->getCollection();
             if (!collection) {
                 return {ErrorCodes::CursorNotFound,
                         str::stream() << "collection does not exist: " << nss.ns()};
diff --git a/src/mongo/db/commands/list_collections.cpp b/src/mongo/db/commands/list_collections.cpp
index da72422b4eb..a636ed51d17 100644
--- a/src/mongo/db/commands/list_collections.cpp
+++ b/src/mongo/db/commands/list_collections.cpp
@@ -298,7 +298,7 @@ public:
         const NamespaceString cursorNss = NamespaceString::makeListCollectionsNSS(dbname);
 
         auto statusWithPlanExecutor = PlanExecutor::make(
-            opCtx, std::move(ws), std::move(root), cursorNss.ns(), PlanExecutor::YIELD_MANUAL);
+            opCtx, std::move(ws), std::move(root), cursorNss, PlanExecutor::YIELD_MANUAL);
         if (!statusWithPlanExecutor.isOK()) {
             return appendCommandStatus(result, statusWithPlanExecutor.getStatus());
         }
@@ -328,9 +328,10 @@ public:
             exec->saveState();
             exec->detachFromOperationContext();
             auto pinnedCursor = CursorManager::getGlobalCursorManager()->registerCursor(
-                {exec.release(),
-                 cursorNss.ns(),
-                 opCtx->recoveryUnit()->isReadingFromMajorityCommittedSnapshot()});
+                {std::move(exec),
+                 cursorNss,
+                 opCtx->recoveryUnit()->isReadingFromMajorityCommittedSnapshot(),
+                 jsobj});
             cursorId = pinnedCursor.getCursor()->cursorid();
         }
 
diff --git a/src/mongo/db/commands/list_indexes.cpp b/src/mongo/db/commands/list_indexes.cpp
index 73e3d19289e..7ea9b23ed0d 100644
--- a/src/mongo/db/commands/list_indexes.cpp
+++ b/src/mongo/db/commands/list_indexes.cpp
@@ -198,7 +198,7 @@ public:
         dassert(ns == cursorNss.getTargetNSForListIndexes());
 
         auto statusWithPlanExecutor = PlanExecutor::make(
-            opCtx, std::move(ws), std::move(root), cursorNss.ns(), PlanExecutor::YIELD_MANUAL);
+            opCtx, std::move(ws), std::move(root), cursorNss, PlanExecutor::YIELD_MANUAL);
         if (!statusWithPlanExecutor.isOK()) {
             return appendCommandStatus(result, statusWithPlanExecutor.getStatus());
         }
@@ -228,9 +228,10 @@ public:
             exec->saveState();
             exec->detachFromOperationContext();
             auto pinnedCursor = CursorManager::getGlobalCursorManager()->registerCursor(
-                {exec.release(),
-                 cursorNss.ns(),
-                 opCtx->recoveryUnit()->isReadingFromMajorityCommittedSnapshot()});
+                {std::move(exec),
+                 cursorNss,
+                 opCtx->recoveryUnit()->isReadingFromMajorityCommittedSnapshot(),
+                 cmdObj});
             cursorId = pinnedCursor.getCursor()->cursorid();
         }
 
diff --git a/src/mongo/db/commands/parallel_collection_scan.cpp b/src/mongo/db/commands/parallel_collection_scan.cpp
index 8ff2e100a58..c46cb6d8c5b 100644
--- a/src/mongo/db/commands/parallel_collection_scan.cpp
+++ b/src/mongo/db/commands/parallel_collection_scan.cpp
@@ -148,11 +148,12 @@ public:
                 exec->saveState();
                 exec->detachFromOperationContext();
 
-                // Create and regiter a new ClientCursor.
+                // Create and register a new ClientCursor.
                 auto pinnedCursor = collection->getCursorManager()->registerCursor(
-                    {exec.release(),
-                     ns.ns(),
-                     opCtx->recoveryUnit()->isReadingFromMajorityCommittedSnapshot()});
+                    {std::move(exec),
+                     ns,
+                     opCtx->recoveryUnit()->isReadingFromMajorityCommittedSnapshot(),
+                     cmdObj});
                 pinnedCursor.getCursor()->setLeftoverMaxTimeMicros(
                     opCtx->getRemainingMaxTimeMicros());
 
diff --git a/src/mongo/db/commands/repair_cursor.cpp b/src/mongo/db/commands/repair_cursor.cpp
index ac6155cc394..69ded443ec6 100644
--- a/src/mongo/db/commands/repair_cursor.cpp
+++ b/src/mongo/db/commands/repair_cursor.cpp
@@ -107,9 +107,10 @@ public:
         exec->detachFromOperationContext();
 
         auto pinnedCursor = collection->getCursorManager()->registerCursor(
-            {exec.release(),
-             ns.ns(),
-             opCtx->recoveryUnit()->isReadingFromMajorityCommittedSnapshot()});
+            {std::move(exec),
+             ns,
+             opCtx->recoveryUnit()->isReadingFromMajorityCommittedSnapshot(),
+             cmdObj});
 
         appendCursorResponseObject(
             pinnedCursor.getCursor()->cursorid(), ns.ns(), BSONArray(), &result);
diff --git a/src/mongo/db/commands/run_aggregate.cpp b/src/mongo/db/commands/run_aggregate.cpp
index 35164560c16..f0ef162e9af 100644
--- a/src/mongo/db/commands/run_aggregate.cpp
+++ b/src/mongo/db/commands/run_aggregate.cpp
@@ -60,6 +60,7 @@
 #include "mongo/db/views/view_sharding_check.h"
 #include "mongo/stdx/memory.h"
 #include "mongo/util/log.h"
+#include "mongo/util/scopeguard.h"
 #include "mongo/util/string_map.h"
 
 namespace mongo {
@@ -76,19 +77,15 @@ namespace {
 /**
  * Returns true if we need to keep a ClientCursor saved for this pipeline (for future getMore
  * requests). Otherwise, returns false. The passed 'nsForCursor' is only used to determine the
- * namespace used in the returned cursor. In the case of views, this can be different from that
- * in 'request'.
+ * namespace used in the returned cursor, which will be registered with the global cursor manager,
+ * and thus will be different from that in 'request'.
  */
 bool handleCursorCommand(OperationContext* opCtx,
-                         const string& nsForCursor,
+                         const NamespaceString& nsForCursor,
                          ClientCursor* cursor,
-                         PlanExecutor* exec,
                          const AggregationRequest& request,
                          BSONObjBuilder& result) {
-    if (cursor) {
-        invariant(cursor->getExecutor() == exec);
-        invariant(cursor->isAggCursor());
-    }
+    invariant(cursor);
 
     long long batchSize = request.getBatchSize();
 
@@ -99,45 +96,29 @@ bool handleCursorCommand(OperationContext* opCtx,
         // The initial getNext() on a PipelineProxyStage may be very expensive so we don't
         // do it when batchSize is 0 since that indicates a desire for a fast return.
         PlanExecutor::ExecState state;
-        if ((state = exec->getNext(&next, NULL)) == PlanExecutor::IS_EOF) {
+        if ((state = cursor->getExecutor()->getNext(&next, nullptr)) == PlanExecutor::IS_EOF) {
             // make it an obvious error to use cursor or executor after this point
-            cursor = NULL;
-            exec = NULL;
+            cursor = nullptr;
             break;
         }
 
-        uassert(34426,
-                "Plan executor error during aggregation: " + WorkingSetCommon::toStatusString(next),
-                PlanExecutor::ADVANCED == state);
+        if (PlanExecutor::ADVANCED != state) {
+            auto status = WorkingSetCommon::getMemberObjectStatus(next);
+            uasserted(status.code(),
+                      "PlanExecutor error during aggregation: " +
+                          WorkingSetCommon::toStatusString(next));
+        }
 
         // If adding this object will cause us to exceed the message size limit, then we stash it
         // for later.
         if (!FindCommon::haveSpaceForNext(next, objCount, resultsArray.len())) {
-            exec->enqueue(next);
+            cursor->getExecutor()->enqueue(next);
             break;
         }
 
         resultsArray.append(next);
     }
 
-    // NOTE: exec->isEOF() can have side effects such as writing by $out. However, it should
-    // be relatively quick since if there was no cursor then the input is empty. Also, this
-    // violates the contract for batchSize==0. Sharding requires a cursor to be returned in that
-    // case. This is ok for now however, since you can't have a sharded collection that doesn't
-    // exist.
-    const bool canReturnMoreBatches = cursor;
-    if (!canReturnMoreBatches && exec && !exec->isEOF()) {
-        // msgasserting since this shouldn't be possible to trigger from today's aggregation
-        // language. The wording assumes that the only reason cursor would be null is if the
-        // collection doesn't exist.
-        msgasserted(
-            17391,
-            str::stream() << "Aggregation has more results than fit in initial batch, but can't "
-                          << "create cursor since collection "
-                          << nsForCursor
-                          << " doesn't exist");
-    }
-
     if (cursor) {
         // If a time limit was set on the pipeline, remaining time is "rolled over" to the
         // cursor (for use by future getmore ops).
@@ -147,14 +128,14 @@ bool handleCursorCommand(OperationContext* opCtx,
 
         // Cursor needs to be in a saved state while we yield locks for getmore. State
         // will be restored in getMore().
-        exec->saveState();
-        exec->detachFromOperationContext();
+        cursor->getExecutor()->saveState();
+        cursor->getExecutor()->detachFromOperationContext();
     } else {
         CurOp::get(opCtx)->debug().cursorExhausted = true;
     }
 
-    const long long cursorId = cursor ? cursor->cursorid() : 0LL;
-    appendCursorResponseObject(cursorId, nsForCursor, resultsArray.arr(), &result);
+    const CursorId cursorId = cursor ? cursor->cursorid() : 0LL;
+    appendCursorResponseObject(cursorId, nsForCursor.ns(), resultsArray.arr(), &result);
 
     return static_cast<bool>(cursor);
 }
@@ -296,7 +277,6 @@ Status runAggregate(OperationContext* opCtx,
         : uassertStatusOK(CollatorFactoryInterface::get(opCtx->getServiceContext())
                               ->makeFromBSON(request.getCollation()));
 
-    boost::optional<ClientCursorPin> pin;  // either this OR the exec will be non-null
     unique_ptr<PlanExecutor> exec;
     boost::intrusive_ptr<ExpressionContext> expCtx;
     boost::intrusive_ptr<Pipeline> pipeline;
@@ -304,11 +284,6 @@ Status runAggregate(OperationContext* opCtx,
     {
         // This will throw if the sharding version for this connection is out of date. If the
         // namespace is a view, the lock will be released before re-running the aggregation.
-        // Otherwise, the lock must be held continuously from now until we have we created both
-        // the output ClientCursor and the input executor. This ensures that both are using the
-        // same sharding version that we synchronize on here. This is also why we always need to
-        // create a ClientCursor even when we aren't outputting to a cursor. See the comment on
-        // ShardFilterStage for more details.
         AutoGetCollectionOrViewForReadCommand ctx(opCtx, nss);
         Collection* collection = ctx.getCollection();
 
@@ -416,17 +391,15 @@ Status runAggregate(OperationContext* opCtx,
         // it to the front of the pipeline if needed.
         PipelineD::prepareCursorSource(collection, &request, pipeline);
 
-        // Create the PlanExecutor which returns results from the pipeline. The WorkingSet
-        // ('ws') and the PipelineProxyStage ('proxy') will be owned by the created
-        // PlanExecutor.
         auto ws = make_unique<WorkingSet>();
         auto proxy = make_unique<PipelineProxyStage>(opCtx, pipeline, ws.get());
 
-        auto statusWithPlanExecutor = (NULL == collection)
-            ? PlanExecutor::make(
-                  opCtx, std::move(ws), std::move(proxy), nss.ns(), PlanExecutor::YIELD_MANUAL)
-            : PlanExecutor::make(
-                  opCtx, std::move(ws), std::move(proxy), collection, PlanExecutor::YIELD_MANUAL);
+        // This PlanExecutor will simply forward requests to the Pipeline, so does not need to
+        // yield or to be registered with any collection's CursorManager to receive invalidations.
+        // The Pipeline may contain PlanExecutors which *are* yielding PlanExecutors and which *are*
+        // registered with their respective collection's CursorManager
+        auto statusWithPlanExecutor = PlanExecutor::make(
+            opCtx, std::move(ws), std::move(proxy), nss, PlanExecutor::YIELD_MANUAL);
         invariant(statusWithPlanExecutor.isOK());
         exec = std::move(statusWithPlanExecutor.getValue());
 
@@ -435,75 +408,39 @@ Status runAggregate(OperationContext* opCtx,
             stdx::lock_guard<Client> lk(*opCtx->getClient());
             curOp->setPlanSummary_inlock(std::move(planSummary));
         }
-
-        if (collection) {
-            const bool isAggCursor = true;  // enable special locking behavior
-            pin.emplace(collection->getCursorManager()->registerCursor(
-                {exec.release(),
-                 nss.ns(),
-                 opCtx->recoveryUnit()->isReadingFromMajorityCommittedSnapshot(),
-                 0,
-                 cmdObj.getOwned(),
-                 isAggCursor}));
-            // Don't add any code between here and the start of the try block.
-        }
-
-        // At this point, it is safe to release the collection lock.
-        // - In the case where we have a collection: we will need to reacquire the
-        //   collection lock later when cleaning up our ClientCursorPin.
-        // - In the case where we don't have a collection: our PlanExecutor won't be
-        //   registered, so it will be safe to clean it up outside the lock.
-        invariant(!exec || !collection);
     }
 
-    try {
-        // Unless set to true, the ClientCursor created above will be deleted on block exit.
-        bool keepCursor = false;
-
-        // If both explain and cursor are specified, explain wins.
-        if (expCtx->explain) {
-            result << "stages" << Value(pipeline->writeExplainOps(*expCtx->explain));
-        } else {
-            // Cursor must be specified, if explain is not.
-            keepCursor = handleCursorCommand(opCtx,
-                                             origNss.ns(),
-                                             pin ? pin->getCursor() : nullptr,
-                                             pin ? pin->getCursor()->getExecutor() : exec.get(),
-                                             request,
-                                             result);
-        }
-
-        if (!expCtx->explain) {
-            PlanSummaryStats stats;
-            Explain::getSummaryStats(pin ? *pin->getCursor()->getExecutor() : *exec.get(), &stats);
-            curOp->debug().setPlanSummaryMetrics(stats);
-            curOp->debug().nreturned = stats.nReturned;
+    // Having released the collection lock, we can now create a cursor that returns results from the
+    // pipeline. This cursor owns no collection state, and thus we register it with the global
+    // cursor manager. The global cursor manager does not deliver invalidations or kill
+    // notifications; the underlying PlanExecutor(s) used by the pipeline will be receiving
+    // invalidations and kill notifications themselves, not the cursor we create here.
+    auto pin = CursorManager::getGlobalCursorManager()->registerCursor(
+        {std::move(exec),
+         origNss,
+         opCtx->recoveryUnit()->isReadingFromMajorityCommittedSnapshot(),
+         cmdObj});
+    ScopeGuard cursorFreer = MakeGuard(&ClientCursorPin::deleteUnderlying, &pin);
+
+    // If both explain and cursor are specified, explain wins.
+    if (expCtx->explain) {
+        result << "stages" << Value(pipeline->writeExplainOps(*expCtx->explain));
+    } else {
+        // Cursor must be specified, if explain is not.
+        const bool keepCursor =
+            handleCursorCommand(opCtx, origNss, pin.getCursor(), request, result);
+        if (keepCursor) {
+            cursorFreer.Dismiss();
         }
+    }
 
-        // Clean up our ClientCursorPin, if needed.  We must reacquire the collection lock
-        // in order to do so.
-        if (pin) {
-            // We acquire locks here with DBLock and CollectionLock instead of using
-            // AutoGetCollectionForRead.  AutoGetCollectionForRead will throw if the
-            // sharding version is out of date, and we don't care if the sharding version
-            // has changed.
-            Lock::DBLock dbLock(opCtx, nss.db(), MODE_IS);
-            Lock::CollectionLock collLock(opCtx->lockState(), nss.ns(), MODE_IS);
-            if (keepCursor) {
-                pin->release();
-            } else {
-                pin->deleteUnderlying();
-            }
-        }
-    } catch (...) {
-        // On our way out of scope, we clean up our ClientCursorPin if needed.
-        if (pin) {
-            Lock::DBLock dbLock(opCtx, nss.db(), MODE_IS);
-            Lock::CollectionLock collLock(opCtx->lockState(), nss.ns(), MODE_IS);
-            pin->deleteUnderlying();
-        }
-        throw;
+    if (!expCtx->explain) {
+        PlanSummaryStats stats;
+        Explain::getSummaryStats(*(pin.getCursor()->getExecutor()), &stats);
+        curOp->debug().setPlanSummaryMetrics(stats);
+        curOp->debug().nreturned = stats.nReturned;
     }
+
     // Any code that needs the cursor pinned must be inside the try block, above.
     return Status::OK();
 }
diff --git a/src/mongo/db/curop.cpp b/src/mongo/db/curop.cpp
index 51016d486ce..2af0907170d 100644
--- a/src/mongo/db/curop.cpp
+++ b/src/mongo/db/curop.cpp
@@ -266,10 +266,12 @@ void CurOp::ensureStarted() {
     }
 }
 
-void CurOp::enter_inlock(const char* ns, int dbProfileLevel) {
+void CurOp::enter_inlock(const char* ns, boost::optional<int> dbProfileLevel) {
     ensureStarted();
     _ns = ns;
-    raiseDbProfileLevel(dbProfileLevel);
+    if (dbProfileLevel) {
+        raiseDbProfileLevel(*dbProfileLevel);
+    }
 }
 
 void CurOp::raiseDbProfileLevel(int dbProfileLevel) {
diff --git a/src/mongo/db/curop.h b/src/mongo/db/curop.h
index 4e0ef0d2d7b..158513bf6fe 100644
--- a/src/mongo/db/curop.h
+++ b/src/mongo/db/curop.h
@@ -192,7 +192,7 @@ public:
         return _originatingCommand;
     }
 
-    void enter_inlock(const char* ns, int dbProfileLevel);
+    void enter_inlock(const char* ns, boost::optional<int> dbProfileLevel);
 
     /**
      * Sets the type of the current network operation.
diff --git a/src/mongo/db/db_raii.cpp b/src/mongo/db/db_raii.cpp
index 84b6584b540..8001b658a09 100644
--- a/src/mongo/db/db_raii.cpp
+++ b/src/mongo/db/db_raii.cpp
@@ -75,6 +75,35 @@ AutoGetOrCreateDb::AutoGetOrCreateDb(OperationContext* opCtx, StringData ns, Loc
     }
 }
 
+AutoStatsTracker::AutoStatsTracker(OperationContext* opCtx,
+                                   const NamespaceString& nss,
+                                   Top::LockType lockType,
+                                   boost::optional<int> dbProfilingLevel)
+    : _opCtx(opCtx), _lockType(lockType) {
+    if (!dbProfilingLevel) {
+        // No profiling level was determined, attempt to read the profiling level from the Database
+        // object.
+        AutoGetDb autoDb(_opCtx, nss.db(), MODE_IS);
+        if (autoDb.getDb()) {
+            dbProfilingLevel = autoDb.getDb()->getProfilingLevel();
+        }
+    }
+    stdx::lock_guard<Client> clientLock(*_opCtx->getClient());
+    CurOp::get(_opCtx)->enter_inlock(nss.ns().c_str(), dbProfilingLevel);
+}
+
+AutoStatsTracker::~AutoStatsTracker() {
+    auto curOp = CurOp::get(_opCtx);
+    Top::get(_opCtx->getServiceContext())
+        .record(_opCtx,
+                curOp->getNS(),
+                curOp->getLogicalOp(),
+                _lockType,
+                _timer.micros(),
+                curOp->isCommand(),
+                curOp->getReadWriteType());
+}
+
 AutoGetCollectionForRead::AutoGetCollectionForRead(OperationContext* opCtx,
                                                    const NamespaceString& nss,
                                                    AutoGetCollection::ViewMode viewMode) {
@@ -84,19 +113,6 @@ AutoGetCollectionForRead::AutoGetCollectionForRead(OperationContext* opCtx,
     _ensureMajorityCommittedSnapshotIsValid(nss, opCtx);
 }
 
-AutoGetCollectionForReadCommand::~AutoGetCollectionForReadCommand() {
-    // Report time spent in read lock
-    auto currentOp = CurOp::get(_opCtx);
-    Top::get(_opCtx->getClient()->getServiceContext())
-        .record(_opCtx,
-                currentOp->getNS(),
-                currentOp->getLogicalOp(),
-                -1,  // "read locked"
-                _timer.micros(),
-                currentOp->isCommand(),
-                currentOp->getReadWriteType());
-}
-
 void AutoGetCollectionForRead::_ensureMajorityCommittedSnapshotIsValid(const NamespaceString& nss,
                                                                        OperationContext* opCtx) {
     while (true) {
@@ -134,25 +150,15 @@ void AutoGetCollectionForRead::_ensureMajorityCommittedSnapshotIsValid(const Nam
 }
 
 AutoGetCollectionForReadCommand::AutoGetCollectionForReadCommand(
-    OperationContext* opCtx, const NamespaceString& nss, AutoGetCollection::ViewMode viewMode)
-    : _opCtx(opCtx) {
-    {
-        _autoCollForRead.emplace(opCtx, nss, viewMode);
-
-        auto curOp = CurOp::get(_opCtx);
-        stdx::lock_guard<Client> lk(*_opCtx->getClient());
-
-        // TODO: OldClientContext legacy, needs to be removed
-        curOp->ensureStarted();
-        curOp->setNS_inlock(nss.ns());
-
-        // At this point, we are locked in shared mode for the database by the DB lock in the
-        // constructor, so it is safe to load the DB pointer.
-        if (_autoCollForRead->getDb()) {
-            // TODO: OldClientContext legacy, needs to be removed
-            curOp->enter_inlock(nss.ns().c_str(), _autoCollForRead->getDb()->getProfilingLevel());
-        }
-    }
+    OperationContext* opCtx, const NamespaceString& nss, AutoGetCollection::ViewMode viewMode) {
+
+    _autoCollForRead.emplace(opCtx, nss, viewMode);
+    const int doNotChangeProfilingLevel = 0;
+    _statsTracker.emplace(opCtx,
+                          nss,
+                          Top::LockType::ReadLocked,
+                          _autoCollForRead->getDb() ? _autoCollForRead->getDb()->getProfilingLevel()
+                                                    : doNotChangeProfilingLevel);
 
     // We have both the DB and collection locked, which is the prerequisite to do a stable shard
     // version check, but we'd like to do the check after we have a satisfactory snapshot.
@@ -231,7 +237,8 @@ OldClientContext::~OldClientContext() {
         .record(_opCtx,
                 currentOp->getNS(),
                 currentOp->getLogicalOp(),
-                _opCtx->lockState()->isWriteLocked() ? 1 : -1,
+                _opCtx->lockState()->isWriteLocked() ? Top::LockType::WriteLocked
+                                                     : Top::LockType::ReadLocked,
                 _timer.micros(),
                 currentOp->isCommand(),
                 currentOp->getReadWriteType());
diff --git a/src/mongo/db/db_raii.h b/src/mongo/db/db_raii.h
index 8060d06f000..42444775bad 100644
--- a/src/mongo/db/db_raii.h
+++ b/src/mongo/db/db_raii.h
@@ -35,6 +35,7 @@
 #include "mongo/db/concurrency/d_concurrency.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/operation_context.h"
+#include "mongo/db/stats/top.h"
 #include "mongo/db/views/view.h"
 #include "mongo/util/timer.h"
 
@@ -170,6 +171,36 @@ private:
 };
 
 /**
+ * RAII-style class which automatically tracks the operation namespace in CurrentOp and records the
+ * operation via Top upon destruction.
+ */
+class AutoStatsTracker {
+    MONGO_DISALLOW_COPYING(AutoStatsTracker);
+
+public:
+    /**
+     * Sets the namespace of the CurOp object associated with 'opCtx' to be 'nss' and starts the
+     * CurOp timer. 'lockType' describes which type of lock is held by this operation, and will be
+     * used for reporting via Top. If 'dbProfilingLevel' is not given, this constructor will acquire
+     * and then drop a database lock in order to determine the database's profiling level.
+     */
+    AutoStatsTracker(OperationContext* opCtx,
+                     const NamespaceString& nss,
+                     Top::LockType lockType,
+                     boost::optional<int> dbProfilingLevel);
+
+    /**
+     * Records stats about the current operation via Top.
+     */
+    ~AutoStatsTracker();
+
+private:
+    const Timer _timer;
+    OperationContext* _opCtx;
+    Top::LockType _lockType;
+};
+
+/**
  * RAII-style class, which would acquire the appropriate hierarchy of locks for obtaining
  * a particular collection and would retrieve a reference to the collection. In addition, this
  * utility will ensure that the read will be performed against an appropriately committed snapshot
@@ -236,8 +267,6 @@ public:
         : AutoGetCollectionForReadCommand(
               opCtx, nss, AutoGetCollection::ViewMode::kViewsForbidden) {}
 
-    ~AutoGetCollectionForReadCommand();
-
     Database* getDb() const {
         return _autoCollForRead->getDb();
     }
@@ -246,21 +275,18 @@ public:
         return _autoCollForRead->getCollection();
     }
 
-private:
-    OperationContext* const _opCtx;
-    const Timer _timer;
-
 protected:
     AutoGetCollectionForReadCommand(OperationContext* opCtx,
                                     const NamespaceString& nss,
                                     AutoGetCollection::ViewMode viewMode);
 
-    /**
-     * This protected section must come after the private section because
-     * AutoGetCollectionOrViewForRead needs access to _autoColl, but _autoColl must be initialized
-     * after _transaction.
-     */
+    // '_autoCollForRead' may need to be reset by AutoGetCollectionOrViewForReadCommand, so needs to
+    // be a boost::optional.
     boost::optional<AutoGetCollectionForRead> _autoCollForRead;
+
+    // This needs to be initialized after 'autoCollForRead', since we need to consult the Database
+    // object to get the profiling level. Thus, it needs to be a boost::optional.
+    boost::optional<AutoStatsTracker> _statsTracker;
 };
 
 /**
diff --git a/src/mongo/db/exec/pipeline_proxy.h b/src/mongo/db/exec/pipeline_proxy.h
index 9482f9ba565..fd5b99c4bae 100644
--- a/src/mongo/db/exec/pipeline_proxy.h
+++ b/src/mongo/db/exec/pipeline_proxy.h
@@ -37,6 +37,7 @@
 #include "mongo/db/pipeline/pipeline.h"
 #include "mongo/db/query/plan_summary_stats.h"
 #include "mongo/db/record_id.h"
+#include "mongo/util/assert_util.h"
 
 namespace mongo {
 
@@ -64,13 +65,18 @@ public:
 
     // Not used.
     SpecificStats* getSpecificStats() const final {
-        return NULL;
+        MONGO_UNREACHABLE;
+    }
+
+    void doInvalidate(OperationContext* txn, const RecordId& rid, InvalidationType type) final {
+        // A PlanExecutor with a PipelineProxyStage should be registered with the global cursor
+        // manager, so should not receive invalidations.
+        MONGO_UNREACHABLE;
     }
 
     std::string getPlanSummaryStr() const;
     void getPlanSummaryStats(PlanSummaryStats* statsOut) const;
 
-    // Not used.
     StageType stageType() const final {
         return STAGE_PIPELINE_PROXY;
     }
diff --git a/src/mongo/db/namespace_string.cpp b/src/mongo/db/namespace_string.cpp
index 6ee4d989ceb..0a453546a61 100644
--- a/src/mongo/db/namespace_string.cpp
+++ b/src/mongo/db/namespace_string.cpp
@@ -137,6 +137,17 @@ NamespaceString NamespaceString::getTargetNSForListIndexes() const {
     return NamespaceString(db(), coll().substr(listIndexesCursorNSPrefix.size()));
 }
 
+boost::optional<NamespaceString> NamespaceString::getTargetNSForGloballyManagedNamespace() const {
+    // Globally managed namespaces are of the form '$cmd.commandName.<targetNs>' or simply
+    // '$cmd.commandName'.
+    dassert(isGloballyManagedNamespace());
+    const size_t indexOfNextDot = coll().find('.', 5);
+    if (indexOfNextDot == std::string::npos) {
+        return boost::none;
+    }
+    return NamespaceString{db(), coll().substr(indexOfNextDot + 1)};
+}
+
 string NamespaceString::escapeDbName(const StringData dbname) {
     std::string escapedDbName;
 
diff --git a/src/mongo/db/namespace_string.h b/src/mongo/db/namespace_string.h
index c623dbbac90..58f26d76b2c 100644
--- a/src/mongo/db/namespace_string.h
+++ b/src/mongo/db/namespace_string.h
@@ -31,6 +31,7 @@
 #pragma once
 
 #include <algorithm>
+#include <boost/optional.hpp>
 #include <iosfwd>
 #include <string>
 
@@ -93,13 +94,13 @@ public:
     NamespaceString(StringData dbName, StringData collectionName);
 
     /**
-     * Contructs a NamespaceString representing a listCollections namespace. The format for this
+     * Constructs a NamespaceString representing a listCollections namespace. The format for this
      * namespace is "<dbName>.$cmd.listCollections".
      */
     static NamespaceString makeListCollectionsNSS(StringData dbName);
 
     /**
-     * Contructs a NamespaceString representing a listIndexes namespace. The format for this
+     * Constructs a NamespaceString representing a listIndexes namespace. The format for this
      * namespace is "<dbName>.$cmd.listIndexes.<collectionName>".
      */
     static NamespaceString makeListIndexesNSS(StringData dbName, StringData collectionName);
@@ -202,10 +203,25 @@ public:
     bool isVirtualized() const {
         return virtualized(_ns);
     }
+
+    /**
+     * Returns true if cursors for this namespace are registered with the global cursor manager.
+     */
+    bool isGloballyManagedNamespace() const {
+        return coll().startsWith("$cmd."_sd);
+    }
+
     bool isListCollectionsCursorNS() const;
     bool isListIndexesCursorNS() const;
 
     /**
+     * Given a NamespaceString for which isGloballyManagedNamespace() returns true, returns the
+     * namespace the command targets, or boost::none for commands like 'listCollections' which
+     * do not target a collection.
+     */
+    boost::optional<NamespaceString> getTargetNSForGloballyManagedNamespace() const;
+
+    /**
      * Given a NamespaceString for which isListIndexesCursorNS() returns true, returns the
      * NamespaceString for the collection that the "listIndexes" targets.
      */
diff --git a/src/mongo/db/namespace_string_test.cpp b/src/mongo/db/namespace_string_test.cpp
index 69884bd5ca9..3e4b489b508 100644
--- a/src/mongo/db/namespace_string_test.cpp
+++ b/src/mongo/db/namespace_string_test.cpp
@@ -173,6 +173,41 @@ TEST(NamespaceStringTest, ListIndexesCursorNS) {
     ASSERT(!NamespaceString("test.$cmd.listCollections.foo").isListIndexesCursorNS());
 }
 
+TEST(NamespaceStringTest, IsGloballyManagedNamespace) {
+    ASSERT_TRUE(NamespaceString{"test.$cmd.aggregate.foo"}.isGloballyManagedNamespace());
+    ASSERT_TRUE(NamespaceString{"test.$cmd.listIndexes.foo"}.isGloballyManagedNamespace());
+    ASSERT_TRUE(NamespaceString{"test.$cmd.otherCommand.foo"}.isGloballyManagedNamespace());
+    ASSERT_TRUE(NamespaceString{"test.$cmd.listCollections"}.isGloballyManagedNamespace());
+    ASSERT_TRUE(NamespaceString{"test.$cmd.otherCommand"}.isGloballyManagedNamespace());
+    ASSERT_TRUE(NamespaceString{"test.$cmd.aggregate"}.isGloballyManagedNamespace());
+    ASSERT_TRUE(NamespaceString{"test.$cmd.listIndexes"}.isGloballyManagedNamespace());
+
+    ASSERT_FALSE(NamespaceString{"test.foo"}.isGloballyManagedNamespace());
+    ASSERT_FALSE(NamespaceString{"test.$cmd"}.isGloballyManagedNamespace());
+
+    ASSERT_FALSE(NamespaceString{"$cmd.aggregate.foo"}.isGloballyManagedNamespace());
+    ASSERT_FALSE(NamespaceString{"$cmd.listCollections"}.isGloballyManagedNamespace());
+}
+
+TEST(NamespaceStringTest, GetTargetNSForGloballyManagedNamespace) {
+    ASSERT_EQ(
+        (NamespaceString{"test", "foo"}),
+        NamespaceString{"test.$cmd.aggregate.foo"}.getTargetNSForGloballyManagedNamespace().get());
+    ASSERT_EQ((NamespaceString{"test", "foo"}),
+              NamespaceString{"test.$cmd.listIndexes.foo"}
+                  .getTargetNSForGloballyManagedNamespace()
+                  .get());
+    ASSERT_EQ((NamespaceString{"test", "foo"}),
+              NamespaceString{"test.$cmd.otherCommand.foo"}
+                  .getTargetNSForGloballyManagedNamespace()
+                  .get());
+
+    ASSERT_FALSE(
+        NamespaceString{"test.$cmd.listCollections"}.getTargetNSForGloballyManagedNamespace());
+    ASSERT_FALSE(
+        NamespaceString{"test.$cmd.otherCommand"}.getTargetNSForGloballyManagedNamespace());
+}
+
 TEST(NamespaceStringTest, CollectionComponentValidNames) {
     ASSERT(NamespaceString::validCollectionComponent("a.b"));
     ASSERT(NamespaceString::validCollectionComponent("a.b"));
diff --git a/src/mongo/db/ops/write_ops_exec.cpp b/src/mongo/db/ops/write_ops_exec.cpp
index 60f0b7c79b6..a29a42e4a52 100644
--- a/src/mongo/db/ops/write_ops_exec.cpp
+++ b/src/mongo/db/ops/write_ops_exec.cpp
@@ -94,7 +94,7 @@ void finishCurOp(OperationContext* opCtx, CurOp* curOp) {
             .record(opCtx,
                     curOp->getNS(),
                     curOp->getLogicalOp(),
-                    1,  // "write locked"
+                    Top::LockType::WriteLocked,
                     curOp->totalTimeMicros(),
                     curOp->isCommand(),
                     curOp->getReadWriteType());
@@ -412,7 +412,7 @@ WriteResult performInserts(OperationContext* opCtx, const InsertOp& wholeOp) {
             .record(opCtx,
                     wholeOp.ns.ns(),
                     LogicalOp::opInsert,
-                    1 /* write locked*/,
+                    Top::LockType::WriteLocked,
                     curOp.totalTimeMicros(),
                     curOp.isCommand(),
                     curOp.getReadWriteType());
diff --git a/src/mongo/db/pipeline/SConscript b/src/mongo/db/pipeline/SConscript
index e243adda24a..ee26ddc62fd 100644
--- a/src/mongo/db/pipeline/SConscript
+++ b/src/mongo/db/pipeline/SConscript
@@ -473,5 +473,6 @@ env.Library(
         '$BUILD_DIR/mongo/db/index/index_access_methods',
         '$BUILD_DIR/mongo/db/matcher/expressions_mongod_only',
         '$BUILD_DIR/mongo/db/stats/serveronly',
+        '$BUILD_DIR/mongo/db/clientcursor',
     ],
 )
diff --git a/src/mongo/db/pipeline/document_source_cursor.cpp b/src/mongo/db/pipeline/document_source_cursor.cpp
index ae62da0a978..edc49e4923e 100644
--- a/src/mongo/db/pipeline/document_source_cursor.cpp
+++ b/src/mongo/db/pipeline/document_source_cursor.cpp
@@ -31,12 +31,11 @@
 #include "mongo/db/pipeline/document_source_cursor.h"
 
 #include "mongo/db/catalog/collection.h"
+#include "mongo/db/curop.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/exec/working_set_common.h"
 #include "mongo/db/pipeline/document.h"
 #include "mongo/db/query/explain.h"
-#include "mongo/db/query/find_common.h"
-#include "mongo/db/server_parameters.h"
 #include "mongo/db/storage/storage_options.h"
 #include "mongo/util/scopeguard.h"
 
@@ -66,20 +65,27 @@ DocumentSource::GetNextResult DocumentSourceCursor::getNext() {
 }
 
 void DocumentSourceCursor::dispose() {
-    _exec.reset();
     _currentBatch.clear();
+    if (!_exec) {
+        return;
+    }
+
+    // We must hold a collection lock to destroy a PlanExecutor to ensure the CursorManager will
+    // still exist and can be safely told to deregister the PlanExecutor.
+    invariant(pExpCtx->opCtx);
+    AutoGetCollection autoColl(pExpCtx->opCtx, _exec->nss(), MODE_IS);
+    _exec.reset();
+    _rangePreserver.release();
 }
 
 void DocumentSourceCursor::loadBatch() {
     if (!_exec) {
+        // No more documents.
         dispose();
         return;
     }
 
-    // We have already validated the sharding version when we constructed the PlanExecutor
-    // so we shouldn't check it again.
-    AutoGetCollectionForRead autoColl(pExpCtx->opCtx, _nss);
-
+    AutoGetCollectionForRead autoColl(pExpCtx->opCtx, _exec->nss());
     _exec->restoreState();
 
     int memUsageBytes = 0;
@@ -88,7 +94,7 @@ void DocumentSourceCursor::loadBatch() {
     {
         ON_BLOCK_EXIT([this] { recordPlanSummaryStats(); });
 
-        while ((state = _exec->getNext(&obj, NULL)) == PlanExecutor::ADVANCED) {
+        while ((state = _exec->getNext(&obj, nullptr)) == PlanExecutor::ADVANCED) {
             if (_shouldProduceEmptyDocs) {
                 _currentBatch.push_back(Document());
             } else if (_dependencies) {
@@ -114,27 +120,28 @@ void DocumentSourceCursor::loadBatch() {
         }
     }
 
-    // If we got here, there won't be any more documents, so destroy the executor. Can't use
-    // dispose since we want to keep the _currentBatch.
+    // If we got here, there won't be any more documents, so destroy our PlanExecutor. Note we can't
+    // use dispose() since we want to keep the current batch.
     _exec.reset();
-
-    uassert(16028,
-            str::stream() << "collection or index disappeared when cursor yielded: "
-                          << WorkingSetCommon::toStatusString(obj),
-            state != PlanExecutor::DEAD);
-
-    uassert(17285,
-            str::stream() << "cursor encountered an error: "
-                          << WorkingSetCommon::toStatusString(obj),
-            state != PlanExecutor::FAILURE);
-
-    massert(17286,
-            str::stream() << "Unexpected return from PlanExecutor::getNext: " << state,
-            state == PlanExecutor::IS_EOF || state == PlanExecutor::ADVANCED);
-}
-
-long long DocumentSourceCursor::getLimit() const {
-    return _limit ? _limit->getLimit() : -1;
+    _rangePreserver.release();
+
+    switch (state) {
+        case PlanExecutor::ADVANCED:
+        case PlanExecutor::IS_EOF:
+            return;  // We've reached our limit or exhausted the cursor.
+        case PlanExecutor::DEAD: {
+            uasserted(ErrorCodes::QueryPlanKilled,
+                      str::stream() << "collection or index disappeared when cursor yielded: "
+                                    << WorkingSetCommon::toStatusString(obj));
+        }
+        case PlanExecutor::FAILURE: {
+            uasserted(17285,
+                      str::stream() << "cursor encountered an error: "
+                                    << WorkingSetCommon::toStatusString(obj));
+        }
+        default:
+            MONGO_UNREACHABLE;
+    }
 }
 
 Pipeline::SourceContainer::iterator DocumentSourceCursor::doOptimizeAt(
@@ -156,12 +163,6 @@ Pipeline::SourceContainer::iterator DocumentSourceCursor::doOptimizeAt(
     return std::next(itr);
 }
 
-
-void DocumentSourceCursor::recordPlanSummaryStr() {
-    invariant(_exec);
-    _planSummary = Explain::getPlanSummary(_exec.get());
-}
-
 void DocumentSourceCursor::recordPlanSummaryStats() {
     invariant(_exec);
     // Aggregation handles in-memory sort outside of the query sub-system. Given that we need to
@@ -175,20 +176,17 @@ void DocumentSourceCursor::recordPlanSummaryStats() {
 }
 
 Value DocumentSourceCursor::serialize(boost::optional<ExplainOptions::Verbosity> explain) const {
-    // we never parse a documentSourceCursor, so we only serialize for explain
+    // We never parse a DocumentSourceCursor, so we only serialize for explain.
     if (!explain)
         return Value();
 
     // Get planner-level explain info from the underlying PlanExecutor.
+    invariant(_exec);
     BSONObjBuilder explainBuilder;
     {
-        AutoGetCollectionForRead autoColl(pExpCtx->opCtx, _nss);
-
-        massert(17392, "No _exec. Were we disposed before explained?", _exec);
-
+        AutoGetCollectionForRead autoColl(pExpCtx->opCtx, _exec->nss());
         _exec->restoreState();
         Explain::explainStages(_exec.get(), autoColl.getCollection(), *explain, &explainBuilder);
-
         _exec->saveState();
     }
 
@@ -229,44 +227,28 @@ void DocumentSourceCursor::reattachToOperationContext(OperationContext* opCtx) {
 }
 
 DocumentSourceCursor::DocumentSourceCursor(Collection* collection,
-                                           const string& ns,
                                            std::unique_ptr<PlanExecutor> exec,
                                            const intrusive_ptr<ExpressionContext>& pCtx)
     : DocumentSource(pCtx),
       _docsAddedToBatches(0),
-      _nss(ns),
+      _rangePreserver(collection),
       _exec(std::move(exec)),
       _outputSorts(_exec->getOutputSorts()) {
-    recordPlanSummaryStr();
 
-    // We record execution metrics here to allow for capture of indexes used prior to execution.
+    _planSummary = Explain::getPlanSummary(_exec.get());
     recordPlanSummaryStats();
+
     if (collection) {
-        collection->infoCache()->notifyOfQuery(pCtx->opCtx, _planSummaryStats.indexesUsed);
+        collection->infoCache()->notifyOfQuery(pExpCtx->opCtx, _planSummaryStats.indexesUsed);
     }
 }
 
 intrusive_ptr<DocumentSourceCursor> DocumentSourceCursor::create(
     Collection* collection,
-    const string& ns,
     std::unique_ptr<PlanExecutor> exec,
     const intrusive_ptr<ExpressionContext>& pExpCtx) {
     intrusive_ptr<DocumentSourceCursor> source(
-        new DocumentSourceCursor(collection, ns, std::move(exec), pExpCtx));
+        new DocumentSourceCursor(collection, std::move(exec), pExpCtx));
     return source;
 }
-
-void DocumentSourceCursor::setProjection(const BSONObj& projection,
-                                         const boost::optional<ParsedDeps>& deps) {
-    _projection = projection;
-    _dependencies = deps;
-}
-
-const std::string& DocumentSourceCursor::getPlanSummaryStr() const {
-    return _planSummary;
-}
-
-const PlanSummaryStats& DocumentSourceCursor::getPlanSummaryStats() const {
-    return _planSummaryStats;
-}
 }
diff --git a/src/mongo/db/pipeline/document_source_cursor.h b/src/mongo/db/pipeline/document_source_cursor.h
index 3faeea86de6..348174b656a 100644
--- a/src/mongo/db/pipeline/document_source_cursor.h
+++ b/src/mongo/db/pipeline/document_source_cursor.h
@@ -34,16 +34,14 @@
 #include "mongo/db/pipeline/document_source.h"
 #include "mongo/db/pipeline/document_source_limit.h"
 #include "mongo/db/query/plan_summary_stats.h"
+#include "mongo/db/range_preserver.h"
 
 namespace mongo {
 
 class PlanExecutor;
 
 /**
- * Constructs and returns Documents from the BSONObj objects produced by a supplied
- * PlanExecutor.
- *
- * An object of this type may only be used by one thread, see SERVER-6123.
+ * Constructs and returns Documents from the BSONObj objects produced by a supplied PlanExecutor.
  */
 class DocumentSourceCursor final : public DocumentSource {
 public:
@@ -69,14 +67,11 @@ public:
     void reattachToOperationContext(OperationContext* opCtx) final;
 
     /**
-     * Create a document source based on a passed-in PlanExecutor.
-     *
-     * This is usually put at the beginning of a chain of document sources
-     * in order to fetch data from the database.
+     * Create a document source based on a passed-in PlanExecutor. 'exec' must be a yielding
+     * PlanExecutor, and must be registered with the associated collection's CursorManager.
      */
     static boost::intrusive_ptr<DocumentSourceCursor> create(
         Collection* collection,
-        const std::string& ns,
         std::unique_ptr<PlanExecutor> exec,
         const boost::intrusive_ptr<ExpressionContext>& pExpCtx);
 
@@ -116,10 +111,17 @@ public:
      * @param projection The projection that has been passed down to the query system.
      * @param deps The output of DepsTracker::toParsedDeps.
      */
-    void setProjection(const BSONObj& projection, const boost::optional<ParsedDeps>& deps);
+    void setProjection(const BSONObj& projection, const boost::optional<ParsedDeps>& deps) {
+        _projection = projection;
+        _dependencies = deps;
+    }
 
-    /// returns -1 for no limit
-    long long getLimit() const;
+    /**
+     * Returns the limit associated with this cursor, or -1 if there is no limit.
+     */
+    long long getLimit() const {
+        return _limit ? _limit->getLimit() : -1;
+    }
 
     /**
      * If subsequent sources need no information from the cursor, the cursor can simply output empty
@@ -129,20 +131,21 @@ public:
         _shouldProduceEmptyDocs = true;
     }
 
-    const std::string& getPlanSummaryStr() const;
+    const std::string& getPlanSummaryStr() const {
+        return _planSummary;
+    }
 
-    const PlanSummaryStats& getPlanSummaryStats() const;
+    const PlanSummaryStats& getPlanSummaryStats() const {
+        return _planSummaryStats;
+    }
 
 private:
     DocumentSourceCursor(Collection* collection,
-                         const std::string& ns,
                          std::unique_ptr<PlanExecutor> exec,
                          const boost::intrusive_ptr<ExpressionContext>& pExpCtx);
 
     void loadBatch();
 
-    void recordPlanSummaryStr();
-
     void recordPlanSummaryStats();
 
     std::deque<Document> _currentBatch;
@@ -156,7 +159,7 @@ private:
     boost::intrusive_ptr<DocumentSourceLimit> _limit;
     long long _docsAddedToBatches;  // for _limit enforcement
 
-    const NamespaceString _nss;
+    RangePreserver _rangePreserver;
     std::unique_ptr<PlanExecutor> _exec;
     BSONObjSet _outputSorts;
     std::string _planSummary;
diff --git a/src/mongo/db/pipeline/pipeline_d.cpp b/src/mongo/db/pipeline/pipeline_d.cpp
index 262deeb140b..c37c1301198 100644
--- a/src/mongo/db/pipeline/pipeline_d.cpp
+++ b/src/mongo/db/pipeline/pipeline_d.cpp
@@ -579,15 +579,12 @@ void PipelineD::addCursorSource(Collection* collection,
                                 const BSONObj& queryObj,
                                 const BSONObj& sortObj,
                                 const BSONObj& projectionObj) {
-    // Get the full "namespace" name.
-    const string& fullName = expCtx->ns.ns();
-
     // DocumentSourceCursor expects a yielding PlanExecutor that has had its state saved.
     exec->saveState();
 
     // Put the PlanExecutor into a DocumentSourceCursor and add it to the front of the pipeline.
     intrusive_ptr<DocumentSourceCursor> pSource =
-        DocumentSourceCursor::create(collection, fullName, std::move(exec), expCtx);
+        DocumentSourceCursor::create(collection, std::move(exec), expCtx);
 
     // Note the query, sort, and projection for explain.
     pSource->setQuery(queryObj);
diff --git a/src/mongo/db/query/explain.cpp b/src/mongo/db/query/explain.cpp
index eb43caf5047..b243336655b 100644
--- a/src/mongo/db/query/explain.cpp
+++ b/src/mongo/db/query/explain.cpp
@@ -596,7 +596,7 @@ void Explain::generatePlannerInfo(PlanExecutor* exec,
     BSONObjBuilder plannerBob(out->subobjStart("queryPlanner"));
 
     plannerBob.append("plannerVersion", QueryPlanner::kPlannerVersion);
-    plannerBob.append("namespace", exec->ns());
+    plannerBob.append("namespace", exec->nss().ns());
 
     // Find whether there is an index filter set for the query shape. The 'indexFilterSet'
     // field will always be false in the case of EOF or idhack plans.
diff --git a/src/mongo/db/query/find.cpp b/src/mongo/db/query/find.cpp
index 44f69823096..62a08389e3f 100644
--- a/src/mongo/db/query/find.cpp
+++ b/src/mongo/db/query/find.cpp
@@ -55,6 +55,7 @@
 #include "mongo/db/server_options.h"
 #include "mongo/db/server_parameters.h"
 #include "mongo/db/service_context.h"
+#include "mongo/db/stats/top.h"
 #include "mongo/db/storage/storage_options.h"
 #include "mongo/s/chunk_version.h"
 #include "mongo/s/stale_exception.h"
@@ -62,10 +63,10 @@
 #include "mongo/util/fail_point_service.h"
 #include "mongo/util/log.h"
 #include "mongo/util/mongoutils/str.h"
+#include "mongo/util/scopeguard.h"
 
 namespace mongo {
 
-using std::endl;
 using std::unique_ptr;
 using stdx::make_unique;
 
@@ -238,47 +239,45 @@ Message getMore(OperationContext* opCtx,
 
     const NamespaceString nss(ns);
 
-    // Depending on the type of cursor being operated on, we hold locks for the whole getMore,
-    // or none of the getMore, or part of the getMore.  The three cases in detail:
+    // Cursors come in one of two flavors:
+    // - Cursors owned by the collection cursor manager, such as those generated via the find
+    //   command. For these cursors, we hold the appropriate collection lock for the duration of
+    //   the getMore using AutoGetCollectionForRead. This will automatically update the CurOp
+    //   object appropriately and record execution time via Top upon completion.
+    // - Cursors owned by the global cursor manager, such as those generated via the aggregate
+    //   command. These cursors either hold no collection state or manage their collection state
+    //   internally, so we acquire no locks. In this case we use the AutoStatsTracker object to
+    //   update the CurOp object appropriately and record execution time via Top upon
+    //   completion.
     //
-    // 1) Normal cursor: we lock with "ctx" and hold it for the whole getMore.
-    // 2) Cursor owned by global cursor manager: we don't lock anything.  These cursors don't own
-    //    any collection state. These cursors are generated either by the listCollections or
-    //    listIndexes commands, as these special cursor-generating commands operate over catalog
-    //    data rather than targeting the data within a collection.
-    // 3) Agg cursor: we lock with "ctx", then release, then relock with "unpinDBLock" and
-    //    "unpinCollLock".  This is because agg cursors handle locking internally (hence the
-    //    release), but the pin and unpin of the cursor must occur under the collection lock.
-    //    We don't use our AutoGetCollectionForRead "ctx" to relock, because
-    //    AutoGetCollectionForRead checks the sharding version (and we want the relock for the
-    //    unpin to succeed even if the sharding version has changed).
-    //
-    // Note that we declare our locks before our ClientCursorPin, in order to ensure that the
-    // pin's destructor is called before the lock destructors (so that the unpin occurs under
-    // the lock).
-    unique_ptr<AutoGetCollectionForReadCommand> ctx;
-    unique_ptr<Lock::DBLock> unpinDBLock;
-    unique_ptr<Lock::CollectionLock> unpinCollLock;
-
+    // Thus, only one of 'readLock' and 'statsTracker' will be populated as we populate
+    // 'cursorManager'.
+    boost::optional<AutoGetCollectionForReadCommand> readLock;
+    boost::optional<AutoStatsTracker> statsTracker;
     CursorManager* cursorManager;
-    if (nss.isListIndexesCursorNS() || nss.isListCollectionsCursorNS()) {
-        // List collections and list indexes are special cursor-generating commands whose
-        // cursors are managed globally, as they operate over catalog data rather than targeting
-        // the data within a collection.
+
+    if (CursorManager::isGloballyManagedCursor(cursorid)) {
         cursorManager = CursorManager::getGlobalCursorManager();
-    } else {
-        ctx = stdx::make_unique<AutoGetCollectionOrViewForReadCommand>(opCtx, nss);
-        auto viewCtx = static_cast<AutoGetCollectionOrViewForReadCommand*>(ctx.get());
-        if (viewCtx->getView()) {
-            uasserted(
+
+        if (boost::optional<NamespaceString> nssForCurOp = nss.isGloballyManagedNamespace()
+                ? nss.getTargetNSForGloballyManagedNamespace()
+                : nss) {
+            AutoGetDb autoDb(opCtx, nssForCurOp->db(), MODE_IS);
+            const auto profilingLevel = autoDb.getDb()
+                ? boost::optional<int>{autoDb.getDb()->getProfilingLevel()}
+                : boost::none;
+            statsTracker.emplace(opCtx, *nssForCurOp, Top::LockType::NotLocked, profilingLevel);
+            uassert(
                 ErrorCodes::CommandNotSupportedOnView,
-                str::stream() << "Namespace " << nss.ns()
+                str::stream() << "Namespace " << nssForCurOp->ns()
                               << " is a view. OP_GET_MORE operations are not supported on views. "
                               << "Only clients which support the getMore command can be used to "
-                                 "query views.");
+                                 "query views.",
+                !autoDb.getDb()->getViewCatalog()->lookup(opCtx, nssForCurOp->ns()));
         }
-
-        Collection* collection = ctx->getCollection();
+    } else {
+        readLock.emplace(opCtx, nss);
+        Collection* collection = readLock->getCollection();
         uassert(17356, "collection dropped between getMore calls", collection);
         cursorManager = collection->getCursorManager();
     }
@@ -323,8 +322,8 @@ Message getMore(OperationContext* opCtx,
                 str::stream() << "Requested getMore on namespace " << ns << ", but cursor "
                               << cursorid
                               << " belongs to namespace "
-                              << cc->ns(),
-                ns == cc->ns());
+                              << cc->nss().ns(),
+                nss == cc->nss());
         *isCursorAuthorized = true;
 
         if (cc->isReadCommitted())
@@ -345,11 +344,6 @@ Message getMore(OperationContext* opCtx,
 
         cc->updateSlaveLocation(opCtx);
 
-        if (cc->isAggCursor()) {
-            // Agg cursors handle their own locking internally.
-            ctx.reset();  // unlocks
-        }
-
         // If we're replaying the oplog, we save the last time that we read.
         Timestamp slaveReadTill;
 
@@ -359,12 +353,12 @@ Message getMore(OperationContext* opCtx,
         uint64_t notifierVersion = 0;
         std::shared_ptr<CappedInsertNotifier> notifier;
         if (isCursorAwaitData(cc)) {
-            invariant(ctx->getCollection()->isCapped());
+            invariant(readLock->getCollection()->isCapped());
             // Retrieve the notifier which we will wait on until new data arrives. We make sure
             // to do this in the lock because once we drop the lock it is possible for the
             // collection to become invalid. The notifier itself will outlive the collection if
             // the collection is dropped, as we keep a shared_ptr to it.
-            notifier = ctx->getCollection()->getCappedInsertNotifier();
+            notifier = readLock->getCollection()->getCappedInsertNotifier();
 
             // Must get the version before we call generateBatch in case a write comes in after
             // that call and before we call wait on the notifier.
@@ -384,7 +378,7 @@ Message getMore(OperationContext* opCtx,
             // and currentOp. Upconvert _query to resemble a getMore command, and set the original
             // command or upconverted legacy query in the originatingCommand field.
             curOp.setQuery_inlock(upconvertGetMoreEntry(nss, cursorid, ntoreturn));
-            curOp.setOriginatingCommand_inlock(cc->getQuery());
+            curOp.setOriginatingCommand_inlock(cc->getOriginatingCommandObj());
         }
 
         PlanExecutor::ExecState state;
@@ -402,7 +396,7 @@ Message getMore(OperationContext* opCtx,
         if (isCursorAwaitData(cc) && state == PlanExecutor::IS_EOF && numResults == 0) {
             // Save the PlanExecutor and drop our locks.
             exec->saveState();
-            ctx.reset();
+            readLock.reset();
 
             // Block waiting for data for up to 1 second.
             Seconds timeout(1);
@@ -414,7 +408,7 @@ Message getMore(OperationContext* opCtx,
             curOp.setExpectedLatencyMs(durationCount<Milliseconds>(timeout));
 
             // Reacquiring locks.
-            ctx = make_unique<AutoGetCollectionForReadCommand>(opCtx, nss);
+            readLock.emplace(opCtx, nss);
             exec->restoreState();
 
             // We woke up because either the timed_wait expired, or there was more data. Either
@@ -428,44 +422,28 @@ Message getMore(OperationContext* opCtx,
         postExecutionStats.totalDocsExamined -= preExecutionStats.totalDocsExamined;
         curOp.debug().setPlanSummaryMetrics(postExecutionStats);
 
-        // We do not report 'execStats' for aggregation, both in the original request and
-        // subsequent getMore. The reason for this is that aggregation's source PlanExecutor
-        // could be destroyed before we know whether we need execStats and we do not want to
-        // generate for all operations due to cost.
-        if (!cc->isAggCursor() && curOp.shouldDBProfile()) {
+        // We do not report 'execStats' for aggregation or other globally managed cursors, both in
+        // the original request and subsequent getMore. It would be useful to have this information
+        // for an aggregation, but the source PlanExecutor could be destroyed before we know whether
+        // we need execStats and we do not want to generate for all operations due to cost.
+        if (!CursorManager::isGloballyManagedCursor(cursorid) && curOp.shouldDBProfile()) {
             BSONObjBuilder execStatsBob;
             Explain::getWinningPlanStats(exec, &execStatsBob);
             curOp.debug().execStats = execStatsBob.obj();
         }
 
-        // We have to do this before re-acquiring locks in the agg case because
-        // shouldSaveCursorGetMore() can make a network call for agg cursors.
-        //
-        // TODO: Getting rid of PlanExecutor::isEOF() in favor of PlanExecutor::IS_EOF would mean
-        // that this network operation is no longer necessary.
-        const bool shouldSaveCursor = shouldSaveCursorGetMore(state, exec, isCursorTailable(cc));
-
-        // In order to deregister a cursor, we need to be holding the DB + collection lock and
-        // if the cursor is aggregation, we release these locks.
-        if (cc->isAggCursor()) {
-            invariant(NULL == ctx.get());
-            unpinDBLock = make_unique<Lock::DBLock>(opCtx, nss.db(), MODE_IS);
-            unpinCollLock =
-                make_unique<Lock::CollectionLock>(opCtx->lockState(), nss.ns(), MODE_IS);
-        }
-
         // Our two possible ClientCursorPin cleanup paths are:
         // 1) If the cursor is not going to be saved, we call deleteUnderlying() on the pin.
-        // 2) If the cursor is going to be saved, we simply let the pin go out of scope.  In
-        //    this case, the pin's destructor will be invoked, which will call release() on the
-        //    pin.  Because our ClientCursorPin is declared after our lock is declared, this
-        //    will happen under the lock.
-        if (!shouldSaveCursor) {
+        // 2) If the cursor is going to be saved, we simply let the pin go out of scope. In this
+        //    case, the pin's destructor will be invoked, which will call release() on the pin.
+        //    Because our ClientCursorPin is declared after our lock is declared, this will happen
+        //    under the lock if any locking was necessary.
+        if (!shouldSaveCursorGetMore(state, exec, isCursorTailable(cc))) {
             ccPin.getValue().deleteUnderlying();
 
             // cc is now invalid, as is the executor
             cursorid = 0;
-            cc = NULL;
+            cc = nullptr;
             curOp.debug().cursorExhausted = true;
 
             LOG(5) << "getMore NOT saving client cursor, ended with state "
@@ -673,10 +651,9 @@ std::string runQuery(OperationContext* opCtx,
 
         // Allocate a new ClientCursor and register it with the cursor manager.
         ClientCursorPin pinnedCursor = collection->getCursorManager()->registerCursor(
-            {exec.release(),
-             nss.ns(),
+            {std::move(exec),
+             nss,
              opCtx->recoveryUnit()->isReadingFromMajorityCommittedSnapshot(),
-             qr.getOptions(),
              upconvertQueryEntry(q.query, qr.nss(), q.ntoreturn, q.ntoskip)});
         ccId = pinnedCursor.getCursor()->cursorid();
 
diff --git a/src/mongo/db/query/get_executor.cpp b/src/mongo/db/query/get_executor.cpp
index 9aa85a7d81a..be3673369a7 100644
--- a/src/mongo/db/query/get_executor.cpp
+++ b/src/mongo/db/query/get_executor.cpp
@@ -739,8 +739,7 @@ StatusWith<unique_ptr<PlanExecutor>> getExecutorDelete(OperationContext* opCtx,
                    << " Using EOF stage: " << redact(unparsedQuery);
             auto deleteStage = make_unique<DeleteStage>(
                 opCtx, deleteStageParams, ws.get(), nullptr, new EOFStage(opCtx));
-            return PlanExecutor::make(
-                opCtx, std::move(ws), std::move(deleteStage), nss.ns(), policy);
+            return PlanExecutor::make(opCtx, std::move(ws), std::move(deleteStage), nss, policy);
         }
 
         const IndexDescriptor* descriptor = collection->getIndexCatalog()->findIdIndex(opCtx);
@@ -906,7 +905,7 @@ StatusWith<unique_ptr<PlanExecutor>> getExecutorUpdate(OperationContext* opCtx,
             auto updateStage = make_unique<UpdateStage>(
                 opCtx, updateStageParams, ws.get(), collection, new EOFStage(opCtx));
             return PlanExecutor::make(
-                opCtx, std::move(ws), std::move(updateStage), nsString.ns(), policy);
+                opCtx, std::move(ws), std::move(updateStage), nsString, policy);
         }
 
         const IndexDescriptor* descriptor = collection->getIndexCatalog()->findIdIndex(opCtx);
@@ -1004,8 +1003,7 @@ StatusWith<unique_ptr<PlanExecutor>> getExecutorGroup(OperationContext* opCtx,
         unique_ptr<PlanStage> root =
             make_unique<GroupStage>(opCtx, request, ws.get(), new EOFStage(opCtx));
 
-        return PlanExecutor::make(
-            opCtx, std::move(ws), std::move(root), request.ns.ns(), yieldPolicy);
+        return PlanExecutor::make(opCtx, std::move(ws), std::move(root), request.ns, yieldPolicy);
     }
 
     const NamespaceString nss(request.ns);
@@ -1263,7 +1261,7 @@ StatusWith<unique_ptr<PlanExecutor>> getExecutorCount(OperationContext* opCtx,
         unique_ptr<PlanStage> root = make_unique<CountStage>(
             opCtx, collection, std::move(params), ws.get(), new EOFStage(opCtx));
         return PlanExecutor::make(
-            opCtx, std::move(ws), std::move(root), request.getNs().ns(), yieldPolicy);
+            opCtx, std::move(ws), std::move(root), request.getNs(), yieldPolicy);
     }
 
     // If the query is empty, then we can determine the count by just asking the collection
@@ -1280,7 +1278,7 @@ StatusWith<unique_ptr<PlanExecutor>> getExecutorCount(OperationContext* opCtx,
         unique_ptr<PlanStage> root =
             make_unique<CountStage>(opCtx, collection, std::move(params), ws.get(), nullptr);
         return PlanExecutor::make(
-            opCtx, std::move(ws), std::move(root), request.getNs().ns(), yieldPolicy);
+            opCtx, std::move(ws), std::move(root), request.getNs(), yieldPolicy);
     }
 
     const size_t plannerOptions = QueryPlannerParams::IS_COUNT;
diff --git a/src/mongo/db/query/internal_plans.cpp b/src/mongo/db/query/internal_plans.cpp
index ade228a0223..f7c3e96bce1 100644
--- a/src/mongo/db/query/internal_plans.cpp
+++ b/src/mongo/db/query/internal_plans.cpp
@@ -52,8 +52,8 @@ std::unique_ptr<PlanExecutor> InternalPlanner::collectionScan(OperationContext*
     if (NULL == collection) {
         auto eof = stdx::make_unique<EOFStage>(opCtx);
         // Takes ownership of 'ws' and 'eof'.
-        auto statusWithPlanExecutor =
-            PlanExecutor::make(opCtx, std::move(ws), std::move(eof), ns.toString(), yieldPolicy);
+        auto statusWithPlanExecutor = PlanExecutor::make(
+            opCtx, std::move(ws), std::move(eof), NamespaceString(ns), yieldPolicy);
         invariant(statusWithPlanExecutor.isOK());
         return std::move(statusWithPlanExecutor.getValue());
     }
diff --git a/src/mongo/db/query/plan_executor.cpp b/src/mongo/db/query/plan_executor.cpp
index dd3f66164f4..8e79fe1a2c5 100644
--- a/src/mongo/db/query/plan_executor.cpp
+++ b/src/mongo/db/query/plan_executor.cpp
@@ -91,17 +91,23 @@ StatusWith<unique_ptr<PlanExecutor>> PlanExecutor::make(OperationContext* opCtx,
                                                         const Collection* collection,
                                                         YieldPolicy yieldPolicy) {
     return PlanExecutor::make(
-        opCtx, std::move(ws), std::move(rt), nullptr, nullptr, collection, "", yieldPolicy);
+        opCtx, std::move(ws), std::move(rt), nullptr, nullptr, collection, {}, yieldPolicy);
 }
 
 // static
 StatusWith<unique_ptr<PlanExecutor>> PlanExecutor::make(OperationContext* opCtx,
                                                         unique_ptr<WorkingSet> ws,
                                                         unique_ptr<PlanStage> rt,
-                                                        const string& ns,
+                                                        NamespaceString nss,
                                                         YieldPolicy yieldPolicy) {
-    return PlanExecutor::make(
-        opCtx, std::move(ws), std::move(rt), nullptr, nullptr, nullptr, ns, yieldPolicy);
+    return PlanExecutor::make(opCtx,
+                              std::move(ws),
+                              std::move(rt),
+                              nullptr,
+                              nullptr,
+                              nullptr,
+                              std::move(nss),
+                              yieldPolicy);
 }
 
 // static
@@ -112,7 +118,7 @@ StatusWith<unique_ptr<PlanExecutor>> PlanExecutor::make(OperationContext* opCtx,
                                                         const Collection* collection,
                                                         YieldPolicy yieldPolicy) {
     return PlanExecutor::make(
-        opCtx, std::move(ws), std::move(rt), nullptr, std::move(cq), collection, "", yieldPolicy);
+        opCtx, std::move(ws), std::move(rt), nullptr, std::move(cq), collection, {}, yieldPolicy);
 }
 
 // static
@@ -129,7 +135,7 @@ StatusWith<unique_ptr<PlanExecutor>> PlanExecutor::make(OperationContext* opCtx,
                               std::move(qs),
                               std::move(cq),
                               collection,
-                              "",
+                              {},
                               yieldPolicy);
 }
 
@@ -140,10 +146,15 @@ StatusWith<unique_ptr<PlanExecutor>> PlanExecutor::make(OperationContext* opCtx,
                                                         unique_ptr<QuerySolution> qs,
                                                         unique_ptr<CanonicalQuery> cq,
                                                         const Collection* collection,
-                                                        const string& ns,
+                                                        NamespaceString nss,
                                                         YieldPolicy yieldPolicy) {
-    unique_ptr<PlanExecutor> exec(new PlanExecutor(
-        opCtx, std::move(ws), std::move(rt), std::move(qs), std::move(cq), collection, ns));
+    unique_ptr<PlanExecutor> exec(new PlanExecutor(opCtx,
+                                                   std::move(ws),
+                                                   std::move(rt),
+                                                   std::move(qs),
+                                                   std::move(cq),
+                                                   collection,
+                                                   std::move(nss)));
 
     // Perform plan selection, if necessary.
     Status status = exec->pickBestPlan(yieldPolicy, collection);
@@ -160,25 +171,24 @@ PlanExecutor::PlanExecutor(OperationContext* opCtx,
                            unique_ptr<QuerySolution> qs,
                            unique_ptr<CanonicalQuery> cq,
                            const Collection* collection,
-                           const string& ns)
+                           NamespaceString nss)
     : _opCtx(opCtx),
       _cq(std::move(cq)),
       _workingSet(std::move(ws)),
       _qs(std::move(qs)),
       _root(std::move(rt)),
-      _ns(ns),
+      _nss(std::move(nss)),
       _yieldPolicy(new PlanYieldPolicy(this, YIELD_MANUAL)) {
-    // We may still need to initialize _ns from either collection or _cq.
-    if (!_ns.empty()) {
-        // We already have an _ns set, so there's nothing more to do.
-        return;
+    // We may still need to initialize _nss from either collection or _cq.
+    if (!_nss.isEmpty()) {
+        return;  // We already have an _nss set, so there's nothing more to do.
     }
 
     if (collection) {
-        _ns = collection->ns().ns();
+        _nss = collection->ns();
     } else {
         invariant(_cq);
-        _ns = _cq->getQueryRequest().ns();
+        _nss = _cq->getQueryRequest().nss();
     }
 }
 
@@ -362,7 +372,7 @@ PlanExecutor::ExecState PlanExecutor::getNextImpl(Snapshotted<BSONObj>* objOut,
     MONGO_FAIL_POINT_BLOCK(planExecutorAlwaysDead, customKill) {
         const BSONObj& data = customKill.getData();
         BSONElement customKillNS = data["namespace"];
-        if (!customKillNS || _ns == customKillNS.str()) {
+        if (!customKillNS || _nss.ns() == customKillNS.str()) {
             deregisterExec();
             kill("hit planExecutorAlwaysDead fail point");
         }
@@ -469,7 +479,8 @@ PlanExecutor::ExecState PlanExecutor::getNextImpl(Snapshotted<BSONObj>* objOut,
                     throw WriteConflictException();
                 CurOp::get(_opCtx)->debug().writeConflicts++;
                 writeConflictsInARow++;
-                WriteConflictException::logAndBackoff(writeConflictsInARow, "plan execution", _ns);
+                WriteConflictException::logAndBackoff(
+                    writeConflictsInARow, "plan execution", _nss.ns());
 
             } else {
                 WorkingSetMember* member = _workingSet->get(id);
@@ -546,10 +557,6 @@ Status PlanExecutor::executePlan() {
     return Status::OK();
 }
 
-const string& PlanExecutor::ns() {
-    return _ns;
-}
-
 void PlanExecutor::setYieldPolicy(YieldPolicy policy,
                                   const Collection* collection,
                                   bool registerExecutor) {
diff --git a/src/mongo/db/query/plan_executor.h b/src/mongo/db/query/plan_executor.h
index 1e97963f67b..325d7642a91 100644
--- a/src/mongo/db/query/plan_executor.h
+++ b/src/mongo/db/query/plan_executor.h
@@ -153,7 +153,7 @@ public:
     static StatusWith<std::unique_ptr<PlanExecutor>> make(OperationContext* opCtx,
                                                           std::unique_ptr<WorkingSet> ws,
                                                           std::unique_ptr<PlanStage> rt,
-                                                          const std::string& ns,
+                                                          NamespaceString nss,
                                                           YieldPolicy yieldPolicy);
 
     /**
@@ -203,7 +203,9 @@ public:
     /**
      * Return the NS that the query is running over.
      */
-    const std::string& ns();
+    const NamespaceString& nss() const {
+        return _nss;
+    }
 
     /**
      * Return the OperationContext that the plan is currently executing within.
@@ -408,7 +410,7 @@ private:
                  std::unique_ptr<QuerySolution> qs,
                  std::unique_ptr<CanonicalQuery> cq,
                  const Collection* collection,
-                 const std::string& ns);
+                 NamespaceString nss);
 
     /**
      * Public factory methods delegate to this private factory to do their work.
@@ -419,7 +421,7 @@ private:
                                                           std::unique_ptr<QuerySolution> qs,
                                                           std::unique_ptr<CanonicalQuery> cq,
                                                           const Collection* collection,
-                                                          const std::string& ns,
+                                                          NamespaceString nss,
                                                           YieldPolicy yieldPolicy);
 
     /**
@@ -462,7 +464,7 @@ private:
     std::unique_ptr<ScopedExecutorRegistration> _safety;
 
     // What namespace are we operating over?
-    std::string _ns;
+    NamespaceString _nss;
 
     // This is used to handle automatic yielding when allowed by the YieldPolicy. Never NULL.
     // TODO make this a non-pointer member. This requires some header shuffling so that this
diff --git a/src/mongo/db/query/plan_yield_policy.cpp b/src/mongo/db/query/plan_yield_policy.cpp
index d84db2b225d..f315daec285 100644
--- a/src/mongo/db/query/plan_yield_policy.cpp
+++ b/src/mongo/db/query/plan_yield_policy.cpp
@@ -108,14 +108,14 @@ bool PlanYieldPolicy::yield(RecordFetcher* fetcher) {
                 opCtx->recoveryUnit()->abandonSnapshot();
             } else {
                 // Release and reacquire locks.
-                QueryYield::yieldAllLocks(opCtx, fetcher, _planYielding->ns());
+                QueryYield::yieldAllLocks(opCtx, fetcher, _planYielding->nss());
             }
 
             return _planYielding->restoreStateWithoutRetrying();
         } catch (const WriteConflictException& wce) {
             CurOp::get(opCtx)->debug().writeConflicts++;
             WriteConflictException::logAndBackoff(
-                attempt, "plan execution restoreState", _planYielding->ns());
+                attempt, "plan execution restoreState", _planYielding->nss().ns());
             // retry
         }
     }
diff --git a/src/mongo/db/query/query_yield.cpp b/src/mongo/db/query/query_yield.cpp
index 5f43cf819ec..37ecc4bf528 100644
--- a/src/mongo/db/query/query_yield.cpp
+++ b/src/mongo/db/query/query_yield.cpp
@@ -47,7 +47,7 @@ MONGO_FP_DECLARE(setYieldAllLocksWait);
 // static
 void QueryYield::yieldAllLocks(OperationContext* opCtx,
                                RecordFetcher* fetcher,
-                               const std::string& planExecNS) {
+                               const NamespaceString& planExecNS) {
     // Things have to happen here in a specific order:
     //   1) Tell the RecordFetcher to do any setup which needs to happen inside locks
     //   2) Release lock mgr locks
diff --git a/src/mongo/db/query/query_yield.h b/src/mongo/db/query/query_yield.h
index a42e29800c9..7d98b299484 100644
--- a/src/mongo/db/query/query_yield.h
+++ b/src/mongo/db/query/query_yield.h
@@ -28,7 +28,7 @@
 
 #pragma once
 
-#include <string>
+#include "mongo/db/namespace_string.h"
 
 namespace mongo {
 
@@ -50,7 +50,7 @@ public:
      */
     static void yieldAllLocks(OperationContext* opCtx,
                               RecordFetcher* fetcher,
-                              const std::string& planExecNS);
+                              const NamespaceString& planExecNS);
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/range_preserver.h b/src/mongo/db/range_preserver.h
index 55e290d52d1..35b7694234e 100644
--- a/src/mongo/db/range_preserver.h
+++ b/src/mongo/db/range_preserver.h
@@ -54,9 +54,15 @@ public:
         }
     }
 
-    ~RangePreserver() {
-        if (_pin)
+    void release() {
+        if (_pin) {
             _pin->deleteUnderlying();
+            _pin.reset();
+        }
+    }
+
+    ~RangePreserver() {
+        release();
     }
 
 private:
diff --git a/src/mongo/db/stats/top.cpp b/src/mongo/db/stats/top.cpp
index 8a8358030b8..45680f7db15 100644
--- a/src/mongo/db/stats/top.cpp
+++ b/src/mongo/db/stats/top.cpp
@@ -75,7 +75,7 @@ Top& Top::get(ServiceContext* service) {
 void Top::record(OperationContext* opCtx,
                  StringData ns,
                  LogicalOp logicalOp,
-                 int lockType,
+                 LockType lockType,
                  long long micros,
                  bool command,
                  Command::ReadWriteType readWriteType) {
@@ -97,7 +97,7 @@ void Top::record(OperationContext* opCtx,
 void Top::_record(OperationContext* opCtx,
                   CollectionData& c,
                   LogicalOp logicalOp,
-                  int lockType,
+                  LockType lockType,
                   long long micros,
                   Command::ReadWriteType readWriteType) {
 
@@ -105,9 +105,9 @@ void Top::_record(OperationContext* opCtx,
 
     c.total.inc(micros);
 
-    if (lockType > 0)
+    if (lockType == LockType::WriteLocked)
         c.writeLock.inc(micros);
-    else if (lockType < 0)
+    else if (lockType == LockType::ReadLocked)
         c.readLock.inc(micros);
 
     switch (logicalOp) {
diff --git a/src/mongo/db/stats/top.h b/src/mongo/db/stats/top.h
index 1f08e4784a5..463482199fe 100644
--- a/src/mongo/db/stats/top.h
+++ b/src/mongo/db/stats/top.h
@@ -84,13 +84,19 @@ public:
         OperationLatencyHistogram opLatencyHistogram;
     };
 
+    enum class LockType {
+        ReadLocked,
+        WriteLocked,
+        NotLocked,
+    };
+
     typedef StringMap<CollectionData> UsageMap;
 
 public:
     void record(OperationContext* opCtx,
                 StringData ns,
                 LogicalOp logicalOp,
-                int lockType,
+                LockType lockType,
                 long long micros,
                 bool command,
                 Command::ReadWriteType readWriteType);
@@ -126,7 +132,7 @@ private:
     void _record(OperationContext* opCtx,
                  CollectionData& c,
                  LogicalOp logicalOp,
-                 int lockType,
+                 LockType lockType,
                  long long micros,
                  Command::ReadWriteType readWriteType);
 
diff --git a/src/mongo/dbtests/documentsourcetests.cpp b/src/mongo/dbtests/documentsourcetests.cpp
index 58df14b77b4..a6588ffc6b2 100644
--- a/src/mongo/dbtests/documentsourcetests.cpp
+++ b/src/mongo/dbtests/documentsourcetests.cpp
@@ -108,10 +108,7 @@ protected:
             getExecutor(&_opCtx, ctx.getCollection(), std::move(cq), PlanExecutor::YIELD_MANUAL));
 
         exec->saveState();
-        exec->registerExec(ctx.getCollection());
-
-        _source =
-            DocumentSourceCursor::create(ctx.getCollection(), nss.ns(), std::move(exec), _ctx);
+        _source = DocumentSourceCursor::create(ctx.getCollection(), std::move(exec), _ctx);
     }
 
     intrusive_ptr<ExpressionContextForTest> ctx() {
diff --git a/src/mongo/dbtests/query_plan_executor.cpp b/src/mongo/dbtests/query_plan_executor.cpp
index 5fbc726b51a..b2e15787c0d 100644
--- a/src/mongo/dbtests/query_plan_executor.cpp
+++ b/src/mongo/dbtests/query_plan_executor.cpp
@@ -91,13 +91,10 @@ public:
     }
 
     /**
-     * Given a match expression, represented as the BSON object 'filterObj',
-     * create a PlanExecutor capable of executing a simple collection
-     * scan.
-     *
-     * The caller takes ownership of the returned PlanExecutor*.
+     * Given a match expression, represented as the BSON object 'filterObj', create a PlanExecutor
+     * capable of executing a simple collection scan.
      */
-    PlanExecutor* makeCollScanExec(Collection* coll, BSONObj& filterObj) {
+    unique_ptr<PlanExecutor> makeCollScanExec(Collection* coll, BSONObj& filterObj) {
         CollectionScanParams csparams;
         csparams.collection = coll;
         csparams.direction = CollectionScanParams::FORWARD;
@@ -124,7 +121,7 @@ public:
                                                          coll,
                                                          PlanExecutor::YIELD_MANUAL);
         ASSERT_OK(statusWithPlanExecutor.getStatus());
-        return statusWithPlanExecutor.getValue().release();
+        return std::move(statusWithPlanExecutor.getValue());
     }
 
     /**
@@ -302,8 +299,7 @@ public:
         // Wrap the "inner" plan executor in a DocumentSourceCursor and add it as the first source
         // in the pipeline.
         innerExec->saveState();
-        auto cursorSource =
-            DocumentSourceCursor::create(collection, nss.ns(), std::move(innerExec), expCtx);
+        auto cursorSource = DocumentSourceCursor::create(collection, std::move(innerExec), expCtx);
         auto pipeline = assertGet(Pipeline::create({cursorSource}, expCtx));
 
         // Create the output PlanExecutor that pulls results from the pipeline.
@@ -323,7 +319,7 @@ public:
 
         // Verify that the aggregation pipeline returns an error because its "inner" plan executor
         // has been killed due to the collection being dropped.
-        ASSERT_THROWS_CODE(pipeline->getNext(), UserException, 16028);
+        ASSERT_THROWS_CODE(pipeline->getNext(), UserException, ErrorCodes::QueryPlanKilled);
 
         // Verify that the "outer" plan executor has been killed due to the collection being
         // dropped.
@@ -450,10 +446,10 @@ public:
         BSONObj filterObj = fromjson("{_id: {$gt: 0}, b: {$gt: 0}}");
 
         Collection* coll = ctx.getCollection();
-        PlanExecutor* exec = makeCollScanExec(coll, filterObj);
+        auto exec = makeCollScanExec(coll, filterObj);
 
         // Make a client cursor from the plan executor.
-        coll->getCursorManager()->registerCursor({exec, nss.ns(), false, 0, BSONObj()});
+        coll->getCursorManager()->registerCursor({std::move(exec), nss, false, BSONObj()});
 
         // There should be one cursor before invalidation,
         // and zero cursors after invalidation.
@@ -476,11 +472,11 @@ public:
         Collection* collection = ctx.getCollection();
 
         BSONObj filterObj = fromjson("{_id: {$gt: 0}, b: {$gt: 0}}");
-        PlanExecutor* exec = makeCollScanExec(collection, filterObj);
+        auto exec = makeCollScanExec(collection, filterObj);
 
         // Make a client cursor from the plan executor.
-        auto ccPin =
-            collection->getCursorManager()->registerCursor({exec, nss.ns(), false, 0, BSONObj()});
+        auto ccPin = collection->getCursorManager()->registerCursor(
+            {std::move(exec), nss, false, BSONObj()});
 
         // If the cursor is pinned, it sticks around, even after invalidation.
         ASSERT_EQUALS(1U, numCursors());
@@ -490,7 +486,7 @@ public:
 
         // The invalidation should have killed the plan executor.
         BSONObj objOut;
-        ASSERT_EQUALS(PlanExecutor::DEAD, exec->getNext(&objOut, NULL));
+        ASSERT_EQUALS(PlanExecutor::DEAD, ccPin.getCursor()->getExecutor()->getNext(&objOut, NULL));
         ASSERT(WorkingSetCommon::isValidStatusMemberObject(objOut));
         const Status status = WorkingSetCommon::getMemberObjectStatus(objOut);
         ASSERT(status.reason().find(invalidateReason) != string::npos);
@@ -519,10 +515,11 @@ public:
             Collection* collection = ctx.getCollection();
 
             BSONObj filterObj = fromjson("{_id: {$gt: 0}, b: {$gt: 0}}");
-            PlanExecutor* exec = makeCollScanExec(collection, filterObj);
+            auto exec = makeCollScanExec(collection, filterObj);
 
             // Make a client cursor from the plan executor.
-            collection->getCursorManager()->registerCursor({exec, nss.ns(), false, 0, BSONObj()});
+            collection->getCursorManager()->registerCursor(
+                {std::move(exec), nss, false, BSONObj()});
         }
 
         // There should be one cursor before timeout,
diff --git a/src/mongo/dbtests/querytests.cpp b/src/mongo/dbtests/querytests.cpp
index 8171cb96cbe..d705399e27c 100644
--- a/src/mongo/dbtests/querytests.cpp
+++ b/src/mongo/dbtests/querytests.cpp
@@ -30,6 +30,7 @@
 
 #include "mongo/platform/basic.h"
 
+#include <boost/optional.hpp>
 #include <iostream>
 
 #include "mongo/client/dbclientcursor.h"
@@ -39,6 +40,7 @@
 #include "mongo/db/db_raii.h"
 #include "mongo/db/dbdirectclient.h"
 #include "mongo/db/dbhelpers.h"
+#include "mongo/db/exec/queued_data_stage.h"
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/json.h"
 #include "mongo/db/lasterror.h"
@@ -244,8 +246,8 @@ protected:
         return !_client.getPrevError().getField("err").isNull();
     }
 
-    const ServiceContext::UniqueOperationContext _txnPtr = cc().makeOperationContext();
-    OperationContext& _opCtx = *_txnPtr;
+    const ServiceContext::UniqueOperationContext _opCtxPtr = cc().makeOperationContext();
+    OperationContext& _opCtx = *_opCtxPtr;
     DBDirectClient _client;
 };
 
@@ -1751,6 +1753,107 @@ public:
     }
 };
 
+class CursorManagerIsGloballyManagedCursorShouldReturnFalseIfLeadingBitsAreZeroes {
+public:
+    void run() {
+        ASSERT_FALSE(CursorManager::isGloballyManagedCursor(0x0000000000000000));
+        ASSERT_FALSE(CursorManager::isGloballyManagedCursor(0x000000000FFFFFFF));
+        ASSERT_FALSE(CursorManager::isGloballyManagedCursor(0x000000007FFFFFFF));
+        ASSERT_FALSE(CursorManager::isGloballyManagedCursor(0x0FFFFFFFFFFFFFFF));
+        ASSERT_FALSE(CursorManager::isGloballyManagedCursor(0x3FFFFFFFFFFFFFFF));
+        ASSERT_FALSE(CursorManager::isGloballyManagedCursor(0x3dedbeefdeadbeef));
+    }
+};
+
+class CursorManagerIsGloballyManagedCursorShouldReturnTrueIfLeadingBitsAreZeroAndOne {
+public:
+    void run() {
+        ASSERT_TRUE(CursorManager::isGloballyManagedCursor(0x4FFFFFFFFFFFFFFF));
+        ASSERT_TRUE(CursorManager::isGloballyManagedCursor(0x5FFFFFFFFFFFFFFF));
+        ASSERT_TRUE(CursorManager::isGloballyManagedCursor(0x6FFFFFFFFFFFFFFF));
+        ASSERT_TRUE(CursorManager::isGloballyManagedCursor(0x7FFFFFFFFFFFFFFF));
+        ASSERT_TRUE(CursorManager::isGloballyManagedCursor(0x4000000000000000));
+        ASSERT_TRUE(CursorManager::isGloballyManagedCursor(0x4dedbeefdeadbeef));
+    }
+};
+
+class CursorManagerIsGloballyManagedCursorShouldReturnFalseIfLeadingBitIsAOne {
+public:
+    void run() {
+        ASSERT_FALSE(CursorManager::isGloballyManagedCursor(~0LL));
+        ASSERT_FALSE(CursorManager::isGloballyManagedCursor(0xFFFFFFFFFFFFFFFF));
+        ASSERT_FALSE(CursorManager::isGloballyManagedCursor(0x8FFFFFFFFFFFFFFF));
+        ASSERT_FALSE(CursorManager::isGloballyManagedCursor(0x8dedbeefdeadbeef));
+    }
+};
+
+class CursorManagerTest {
+public:
+    std::unique_ptr<PlanExecutor> makeFakePlanExecutor(OperationContext* opCtx) {
+        auto workingSet = stdx::make_unique<WorkingSet>();
+        auto queuedDataStage = stdx::make_unique<QueuedDataStage>(opCtx, workingSet.get());
+        return unittest::assertGet(PlanExecutor::make(opCtx,
+                                                      std::move(workingSet),
+                                                      std::move(queuedDataStage),
+                                                      NamespaceString{"test.collection"},
+                                                      PlanExecutor::YieldPolicy::YIELD_MANUAL));
+    }
+};
+
+class GlobalCursorManagerShouldReportOwnershipOfCursorsItCreated : public CursorManagerTest {
+public:
+    void run() {
+        auto opCtx = cc().makeOperationContext();
+        for (int i = 0; i < 1000; i++) {
+            auto exec = makeFakePlanExecutor(opCtx.get());
+            auto cursorPin = CursorManager::getGlobalCursorManager()->registerCursor(
+                {std::move(exec), NamespaceString{"test.collection"}, false, BSONObj()});
+            ASSERT_TRUE(CursorManager::isGloballyManagedCursor(cursorPin.getCursor()->cursorid()));
+        }
+    }
+};
+
+class CursorsFromCollectionCursorManagerShouldNotReportBeingManagedByGlobalCursorManager
+    : public CursorManagerTest {
+public:
+    void run() {
+        CursorManager testManager(NamespaceString{"test.collection"});
+        auto opCtx = cc().makeOperationContext();
+        for (int i = 0; i < 1000; i++) {
+            auto exec = makeFakePlanExecutor(opCtx.get());
+            auto cursorPin = testManager.registerCursor(
+                {std::move(exec), NamespaceString{"test.collection"}, false, BSONObj()});
+            ASSERT_FALSE(CursorManager::isGloballyManagedCursor(cursorPin.getCursor()->cursorid()));
+        }
+    }
+};
+
+class AllCursorsFromCollectionCursorManagerShouldContainIdentical32BitPrefixes
+    : public CursorManagerTest {
+public:
+    void run() {
+        CursorManager testManager(NamespaceString{"test.collection"});
+        auto opCtx = cc().makeOperationContext();
+        boost::optional<uint32_t> prefix;
+        for (int i = 0; i < 1000; i++) {
+            auto exec = makeFakePlanExecutor(opCtx.get());
+            auto cursorPin = testManager.registerCursor(
+                {std::move(exec), NamespaceString{"test.collection"}, false, BSONObj()});
+            auto cursorId = cursorPin.getCursor()->cursorid();
+            if (prefix) {
+                ASSERT_EQ(*prefix, extractLeading32Bits(cursorId));
+            } else {
+                prefix = extractLeading32Bits(cursorId);
+            }
+        }
+    }
+
+private:
+    uint32_t extractLeading32Bits(CursorId cursorId) {
+        return static_cast<uint32_t>((cursorId & 0xFFFFFFFF00000000) >> 32);
+    }
+};
+
 class All : public Suite {
 public:
     All() : Suite("query") {}
@@ -1808,10 +1911,14 @@ public:
         add<QueryCursorTimeout>();
         add<QueryReadsAll>();
         add<KillPinnedCursor>();
-
         add<queryobjecttests::names1>();
-
         add<OrderingTest>();
+        add<CursorManagerIsGloballyManagedCursorShouldReturnFalseIfLeadingBitsAreZeroes>();
+        add<CursorManagerIsGloballyManagedCursorShouldReturnTrueIfLeadingBitsAreZeroAndOne>();
+        add<CursorManagerIsGloballyManagedCursorShouldReturnFalseIfLeadingBitIsAOne>();
+        add<GlobalCursorManagerShouldReportOwnershipOfCursorsItCreated>();
+        add<CursorsFromCollectionCursorManagerShouldNotReportBeingManagedByGlobalCursorManager>();
+        add<AllCursorsFromCollectionCursorManagerShouldContainIdentical32BitPrefixes>();
     }
 };