SERVER-27065 cleanup ClientCursor, ClientCursorPin, and CursorManager

 - Makes cursors come into existence pinned. This fixes a
   race condition in which a cursor could time out in between
   being constructed/retrieved and being pinned.
 - Reduces the public interface of ClientCursor. In
   particular, makes ClientCursor's constructor and
   destructor private.
 - Cleans up header file comments in order to more clearly
   indicate expected usage.

1 files changed, 67 insertions, 69 deletions

diff --git a/src/mongo/db/clientcursor.cpp b/src/mongo/db/clientcursor.cpp
index 4cc2dd643a4..67147f9dd75 100644
--- a/src/mongo/db/clientcursor.cpp
+++ b/src/mongo/db/clientcursor.cpp
@@ -77,43 +77,34 @@ long long ClientCursor::totalOpen() {
     return cursorStatsOpen.get();
 }
 
-ClientCursor::ClientCursor(CursorManager* cursorManager,
-                           PlanExecutor* exec,
-                           const std::string& ns,
-                           bool isReadCommitted,
-                           int qopts,
-                           const BSONObj query,
-                           bool isAggCursor)
-    : _ns(ns),
-      _isReadCommitted(isReadCommitted),
+ClientCursor::ClientCursor(const ClientCursorParams& params,
+                           CursorManager* cursorManager,
+                           CursorId cursorId)
+    : _cursorid(cursorId),
+      _ns(params.ns),
+      _isReadCommitted(params.isReadCommitted),
       _cursorManager(cursorManager),
-      _countedYet(false),
-      _isAggCursor(isAggCursor) {
-    _exec.reset(exec);
-    _query = query;
-    _queryOptions = qopts;
+      _query(params.query),
+      _queryOptions(params.qopts),
+      _isAggCursor(params.isAggCursor) {
+    _exec.reset(params.exec);
     init();
 }
 
-ClientCursor::ClientCursor(const Collection* collection)
-    : _ns(collection->ns().ns()),
-      _isReadCommitted(false),
-      _cursorManager(collection->getCursorManager()),
-      _countedYet(false),
-      _queryOptions(QueryOption_NoCursorTimeout),
-      _isAggCursor(false) {
+ClientCursor::ClientCursor(const Collection* collection,
+                           CursorManager* cursorManager,
+                           CursorId cursorId)
+    : _cursorid(cursorId),
+      _ns(collection->ns().ns()),
+      _cursorManager(cursorManager),
+      _queryOptions(QueryOption_NoCursorTimeout) {
     init();
 }
 
 void ClientCursor::init() {
     invariant(_cursorManager);
 
-    _isPinned = false;
-    _isNoTimeout = false;
-
-    _idleAgeMillis = 0;
-    _leftoverMaxTimeMicros = Microseconds::max();
-    _pos = 0;
+    cursorStatsOpen.increment();
 
     if (_queryOptions & QueryOption_NoCursorTimeout) {
         // cursors normally timeout after an inactivity period to prevent excess memory use
@@ -121,45 +112,24 @@ void ClientCursor::init() {
         _isNoTimeout = true;
         cursorStatsOpenNoTimeout.increment();
     }
-
-    _cursorid = _cursorManager->registerCursor(this);
-
-    cursorStatsOpen.increment();
-    _countedYet = true;
 }
 
 ClientCursor::~ClientCursor() {
-    if (_pos == -2) {
-        // defensive: destructor called twice
-        wassert(false);
-        return;
-    }
-
-    invariant(!_isPinned);  // Must call unsetPinned() before invoking destructor.
+    // Cursors must be unpinned and deregistered from their cursor manager before being deleted.
+    invariant(!_isPinned);
+    invariant(!_cursorManager);
 
-    if (_countedYet) {
-        _countedYet = false;
-        cursorStatsOpen.decrement();
-        if (_isNoTimeout)
-            cursorStatsOpenNoTimeout.decrement();
+    cursorStatsOpen.decrement();
+    if (_isNoTimeout) {
+        cursorStatsOpenNoTimeout.decrement();
     }
-
-    if (_cursorManager) {
-        // this could be null if kill() was killed
-        _cursorManager->deregisterCursor(this);
-    }
-
-    // defensive:
-    _cursorManager = NULL;
-    _pos = -2;
-    _isNoTimeout = false;
 }
 
 void ClientCursor::kill() {
     if (_exec.get())
         _exec->kill("cursor killed");
 
-    _cursorManager = NULL;
+    _cursorManager = nullptr;
 }
 
 //
@@ -174,8 +144,8 @@ bool ClientCursor::shouldTimeout(int millis) {
     return _idleAgeMillis > cursorTimeoutMillis;
 }
 
-void ClientCursor::setIdleTime(int millis) {
-    _idleAgeMillis = millis;
+void ClientCursor::resetIdleTime() {
+    _idleAgeMillis = 0;
 }
 
 void ClientCursor::updateSlaveLocation(OperationContext* txn) {
@@ -197,9 +167,37 @@ void ClientCursor::updateSlaveLocation(OperationContext* txn) {
 // Pin methods
 //
 
-ClientCursorPin::ClientCursorPin(CursorManager* cursorManager, long long cursorid) : _cursor(NULL) {
+ClientCursorPin::ClientCursorPin(ClientCursor* cursor) : _cursor(cursor) {
+    invariant(_cursor);
+    invariant(_cursor->_isPinned);
+    invariant(_cursor->_cursorManager);
     cursorStatsOpenPinned.increment();
-    _cursor = cursorManager->find(cursorid, true);
+}
+
+ClientCursorPin::ClientCursorPin(ClientCursorPin&& other) : _cursor(other._cursor) {
+    // The pinned cursor is being transferred to us from another pin. The 'other' pin must have a
+    // pinned cursor.
+    invariant(other._cursor);
+    invariant(other._cursor->_isPinned);
+
+    // Be sure to set the 'other' pin's cursor to null in order to transfer ownership to ourself.
+    _cursor = other._cursor;
+    other._cursor = nullptr;
+}
+
+ClientCursorPin& ClientCursorPin::operator=(ClientCursorPin&& other) {
+    // The pinned cursor is being transferred to us from another pin. The 'other' pin must have a
+    // pinned cursor, and we must not have a cursor.
+    invariant(!_cursor);
+    invariant(other._cursor);
+    invariant(other._cursor->_isPinned);
+
+    // Copy the cursor pointer to ourselves, but also be sure to set the 'other' pin's cursor to
+    // null so that it no longer has the cursor pinned.
+    // Be sure to set the 'other' pin's cursor to null in order to transfer ownership to ourself.
+    _cursor = other._cursor;
+    other._cursor = nullptr;
+    return *this;
 }
 
 ClientCursorPin::~ClientCursorPin() {
@@ -211,23 +209,23 @@ void ClientCursorPin::release() {
     if (!_cursor)
         return;
 
-    invariant(_cursor->isPinned());
+    invariant(_cursor->_isPinned);
 
-    if (_cursor->cursorManager() == NULL) {
+    if (!_cursor->_cursorManager) {
         // The ClientCursor was killed while we had it.  Therefore, it is our responsibility to
         // kill it.
         deleteUnderlying();
     } else {
         // Unpin the cursor under the collection cursor manager lock.
-        _cursor->cursorManager()->unpin(_cursor);
+        _cursor->_cursorManager->unpin(_cursor);
     }
 
-    _cursor = NULL;
+    _cursor = nullptr;
 }
 
 void ClientCursorPin::deleteUnderlying() {
     invariant(_cursor);
-    invariant(_cursor->isPinned());
+    invariant(_cursor->_isPinned);
     // Note the following subtleties of this method's implementation:
     // - We must unpin the cursor before destruction, since it is an error to destroy a pinned
     //   cursor.
@@ -235,16 +233,16 @@ void ClientCursorPin::deleteUnderlying() {
     //   error to unpin a registered cursor without holding the cursor manager lock (note that
     //   we can't simply unpin with the cursor manager lock here, since we need to guarantee
     //   exclusive ownership of the cursor when we are deleting it).
-    if (_cursor->cursorManager()) {
-        _cursor->cursorManager()->deregisterCursor(_cursor);
+    if (_cursor->_cursorManager) {
+        _cursor->_cursorManager->deregisterCursor(_cursor);
         _cursor->kill();
     }
-    _cursor->unsetPinned();
+    _cursor->_isPinned = false;
     delete _cursor;
-    _cursor = NULL;
+    _cursor = nullptr;
 }
 
-ClientCursor* ClientCursorPin::c() const {
+ClientCursor* ClientCursorPin::getCursor() const {
     return _cursor;
 }