SERVER-4844

8 files changed, 57 insertions, 18 deletions

diff --git a/src/mongo/SConscript b/src/mongo/SConscript
index 1e7d274a7f8..0d9a53db546 100644
--- a/src/mongo/SConscript
+++ b/src/mongo/SConscript
@@ -64,15 +64,15 @@ commonFiles = [ "pch.cpp",
                 "util/concurrency/vars.cpp",
                 "util/concurrency/task.cpp",
                 "util/debug_util.cpp",
-                 "util/concurrency/thread_pool.cpp",
+                "util/concurrency/thread_pool.cpp",
                 "util/password.cpp",
                 "util/signal_handlers.cpp",
-                 "util/concurrency/rwlockimpl.cpp",
+                "util/concurrency/rwlockimpl.cpp",
                 "util/histogram.cpp",
                 "util/concurrency/spin_lock.cpp",
                 "util/text.cpp",
                 "util/stringutils.cpp",
-                 "util/concurrency/synchronization.cpp",
+                "util/concurrency/synchronization.cpp",
                 "util/net/sock.cpp",
                 "util/net/httpclient.cpp",
                 "util/net/message.cpp",
@@ -198,6 +198,7 @@ if has_option( "asio" ):
 
 # mongod files - also files used in tools. present in dbtests, but not in mongos and not in client libs.
 serverOnlyFiles = [ "db/curop.cpp",
+                    "db/interrupt_status_mongod.cpp",
                     "db/d_globals.cpp",
                     "db/pagefault.cpp",
                     "util/compress.cpp",
@@ -284,6 +285,7 @@ env.Library( "coreshard", [ "s/config.cpp",
                             "s/shardkey.cpp"] )
 
 shardServerFiles = [
+    "s/interrupt_status_mongos.cpp",
     "s/strategy.cpp",
     "s/strategy_shard.cpp",
     "s/strategy_single.cpp",
diff --git a/src/mongo/db/commands/pipeline_command.cpp b/src/mongo/db/commands/pipeline_command.cpp
index 4165bfe8586..295b3b2b770 100755
--- a/src/mongo/db/commands/pipeline_command.cpp
+++ b/src/mongo/db/commands/pipeline_command.cpp
@@ -19,6 +19,7 @@
 #include "db/commands/pipeline.h"

 #include "db/commands/pipeline_d.h"

 #include "db/cursor.h"

+#include "db/interrupt_status_mongod.h"

 #include "db/pdfile.h"

 #include "db/pipeline/accumulator.h"

 #include "db/pipeline/document.h"

@@ -72,7 +73,8 @@ namespace mongo {
                               int options, string &errmsg,

                               BSONObjBuilder &result, bool fromRepl) {

 

-        intrusive_ptr<ExpressionContext> pCtx(ExpressionContext::create());

+        intrusive_ptr<ExpressionContext> pCtx(

+            ExpressionContext::create(&InterruptStatusMongod::status));

 

         /* try to parse the command; if this fails, then we didn't run */

         intrusive_ptr<Pipeline> pPipeline(

@@ -123,7 +125,8 @@ namespace mongo {
         }

 

         /* on the shard servers, create the local pipeline */

-        intrusive_ptr<ExpressionContext> pShardCtx(ExpressionContext::create());

+        intrusive_ptr<ExpressionContext> pShardCtx(

+            ExpressionContext::create(&InterruptStatusMongod::status));

         intrusive_ptr<Pipeline> pShardPipeline(

             Pipeline::parseCommand(errmsg, shardBson, pShardCtx));

         if (!pShardPipeline.get()) {

diff --git a/src/mongo/db/pipeline/document_source.h b/src/mongo/db/pipeline/document_source.h
index 0a3614198e7..f3fba64922a 100755
--- a/src/mongo/db/pipeline/document_source.h
+++ b/src/mongo/db/pipeline/document_source.h
@@ -953,8 +953,9 @@ namespace mongo {
         virtual ~DocumentSourceLimit();

         virtual bool eof();

         virtual bool advance();

-        virtual const char *getSourceName() const;

         virtual intrusive_ptr<Document> getCurrent();

+        virtual const char *getSourceName() const;

+        virtual bool coalesce(const intrusive_ptr<DocumentSource> &pNextSource);

 

         /**

           Create a new limiting DocumentSource.

@@ -1003,8 +1004,9 @@ namespace mongo {
         virtual ~DocumentSourceSkip();

         virtual bool eof();

         virtual bool advance();

-        virtual const char *getSourceName() const;

         virtual intrusive_ptr<Document> getCurrent();

+        virtual const char *getSourceName() const;

+        virtual bool coalesce(const intrusive_ptr<DocumentSource> &pNextSource);

 

         /**

           Create a new skipping DocumentSource.

diff --git a/src/mongo/db/pipeline/document_source_limit.cpp b/src/mongo/db/pipeline/document_source_limit.cpp
index 74fefbd0424..7b04a6c02ce 100644
--- a/src/mongo/db/pipeline/document_source_limit.cpp
+++ b/src/mongo/db/pipeline/document_source_limit.cpp
@@ -41,6 +41,21 @@ namespace mongo {
         return limitName;
     }
 
+    bool DocumentSourceLimit::coalesce(
+        const intrusive_ptr<DocumentSource> &pNextSource) {
+        DocumentSourceLimit *pLimit =
+            dynamic_cast<DocumentSourceLimit *>(pNextSource.get());
+
+        /* if it's not another $skip, we can't coalesce */
+        if (!pLimit)
+            return false;
+
+        /* we need to limit by the minimum of the two limits */
+        if (pLimit->limit < limit)
+            limit = pLimit->limit;
+        return true;
+    }
+
     bool DocumentSourceLimit::eof() {
         return pSource->eof() || count >= limit;
     }
diff --git a/src/mongo/db/pipeline/document_source_skip.cpp b/src/mongo/db/pipeline/document_source_skip.cpp
index f0b2d3d44d9..605f85e7f92 100644
--- a/src/mongo/db/pipeline/document_source_skip.cpp
+++ b/src/mongo/db/pipeline/document_source_skip.cpp
@@ -42,6 +42,20 @@ namespace mongo {
         return skipName;
     }
 
+    bool DocumentSourceSkip::coalesce(
+        const intrusive_ptr<DocumentSource> &pNextSource) {
+        DocumentSourceSkip *pSkip =
+            dynamic_cast<DocumentSourceSkip *>(pNextSource.get());
+
+        /* if it's not another $skip, we can't coalesce */
+        if (!pSkip)
+            return false;
+
+        /* we need to skip over the sum of the two consecutive $skips */
+        skip += pSkip->skip;
+        return true;
+    }
+
     void DocumentSourceSkip::skipper() {
         if (count == 0) {
             while (!pSource->eof() && count++ < skip) {
diff --git a/src/mongo/db/pipeline/expression_context.cpp b/src/mongo/db/pipeline/expression_context.cpp
index cfde1670bfb..b56b243f25f 100755
--- a/src/mongo/db/pipeline/expression_context.cpp
+++ b/src/mongo/db/pipeline/expression_context.cpp
@@ -16,7 +16,7 @@
 

 #include "pch.h"

 

-#include "db/curop.h"

+#include "db/interrupt_status.h"

 #include "db/pipeline/expression_context.h"

 

 namespace mongo {

@@ -24,10 +24,11 @@ namespace mongo {
     ExpressionContext::~ExpressionContext() {

     }

 

-    inline ExpressionContext::ExpressionContext():

+    inline ExpressionContext::ExpressionContext(InterruptStatus *pS):

         inShard(false),

         inRouter(false),

-        intCheckCounter(1) {

+        intCheckCounter(1),

+        pStatus(pS) {

     }

 

     void ExpressionContext::checkForInterrupt() {

@@ -35,14 +36,12 @@ namespace mongo {
           Only really check periodically; the check gets a mutex, and could

           be expensive, at least in relative terms.

         */

-#ifdef MONGO_LATER_SERVER_4844

         if ((++intCheckCounter % 128) == 0)

-            killCurrentOp.checkForInterrupt();

-#endif /* MONGO_LATER_SERVER_4844 */

+            pStatus->checkForInterrupt();

     }

 

-    ExpressionContext *ExpressionContext::create() {

-        return new ExpressionContext();

+    ExpressionContext *ExpressionContext::create(InterruptStatus *pStatus) {

+        return new ExpressionContext(pStatus);

     }

 

 }

diff --git a/src/mongo/db/pipeline/expression_context.h b/src/mongo/db/pipeline/expression_context.h
index 0f4de1461ae..893826af222 100755
--- a/src/mongo/db/pipeline/expression_context.h
+++ b/src/mongo/db/pipeline/expression_context.h
@@ -22,6 +22,8 @@
 

 namespace mongo {

 

+    class InterruptStatus;

+

     class ExpressionContext :

         public IntrusiveCounterUnsigned {

     public:

@@ -40,14 +42,15 @@ namespace mongo {
          */

         void checkForInterrupt();

 

-        static ExpressionContext *create();

+        static ExpressionContext *create(InterruptStatus *pStatus);

 

     private:

-        ExpressionContext();

+        ExpressionContext(InterruptStatus *pStatus);

         

         bool inShard;

         bool inRouter;

         unsigned intCheckCounter; // interrupt check counter

+        InterruptStatus *const pStatus;

     };

 }

 

diff --git a/src/mongo/s/commands_public.cpp b/src/mongo/s/commands_public.cpp
index 491bc9b0575..702f7997486 100644
--- a/src/mongo/s/commands_public.cpp
+++ b/src/mongo/s/commands_public.cpp
@@ -26,6 +26,7 @@
 #include "../db/pipeline/document_source.h"
 #include "../db/pipeline/expression_context.h"
 #include "../db/queryutil.h"
+#include "s/interrupt_status_mongos.h"
 #include "../scripting/engine.h"
 #include "../util/timer.h"
 
@@ -1394,7 +1395,7 @@ namespace mongo {
             //const string shardedOutputCollection = getTmpName( collection );
 
             intrusive_ptr<ExpressionContext> pExpCtx(
-                ExpressionContext::create());
+                ExpressionContext::create(&InterruptStatusMongos::status));
             pExpCtx->setInRouter(true);
 
             /* parse the pipeline specification */