SERVER-19464 Have DSMergeCursors correctly cleanup _cursors in dispose()

This involves killing the cursor and returning the connection to the pool.
This commit also ensures that DSSort calls dispose on its source as needed.

(cherry picked from commit 5c20356061c19b7343cb9aab7398238581f0b600)

2 files changed, 16 insertions, 2 deletions

diff --git a/src/mongo/db/pipeline/document_source_merge_cursors.cpp b/src/mongo/db/pipeline/document_source_merge_cursors.cpp
index 60020a7a12a..0ce8b2a79ea 100644
--- a/src/mongo/db/pipeline/document_source_merge_cursors.cpp
+++ b/src/mongo/db/pipeline/document_source_merge_cursors.cpp
@@ -164,6 +164,13 @@ boost::optional<Document> DocumentSourceMergeCursors::getNext() {
 }
 
 void DocumentSourceMergeCursors::dispose() {
+    // Note it is an error to call done() on a connection before consuming the response from a
+    // request. Therefore it is an error to call dispose() if there are any outstanding connections
+    // which have not received a reply.
+    for (auto&& cursorAndConn : _cursors) {
+        cursorAndConn->cursor.kill();
+        cursorAndConn->connection.done();
+    }
     _cursors.clear();
     _currentCursor = _cursors.end();
 }
diff --git a/src/mongo/db/pipeline/document_source_sort.cpp b/src/mongo/db/pipeline/document_source_sort.cpp
index b49c2c320e1..e60d70aabad 100644
--- a/src/mongo/db/pipeline/document_source_sort.cpp
+++ b/src/mongo/db/pipeline/document_source_sort.cpp
@@ -59,8 +59,13 @@ boost::optional<Document> DocumentSourceSort::getNext() {
     if (!populated)
         populate();
 
-    if (!_output || !_output->more())
+    if (!_output || !_output->more()) {
+        // Need to be sure connections are marked as done so they can be returned to the connection
+        // pool. This only needs to happen in the _mergingPresorted case, but it doesn't hurt to
+        // always do it.
+        dispose();
         return boost::none;
+    }
 
     return _output->next().second;
 }
@@ -86,7 +91,9 @@ void DocumentSourceSort::serializeToArray(vector<Value>& array, bool explain) co
 
 void DocumentSourceSort::dispose() {
     _output.reset();
-    pSource->dispose();
+    if (pSource) {
+        pSource->dispose();
+    }
 }
 
 DocumentSourceSort::DocumentSourceSort(const intrusive_ptr<ExpressionContext>& pExpCtx)