SERVER-2212 Ability for CurOp objects to be "timer-interruptible"

- New public methods introduced to CurOp for requesting op
  interruption after a given time period.
- KillCurrentOp objects now additionally tasked with interrupting
  CurOp objects that have exceeded their time limit, during the
  "interrupt check".

5 files changed, 259 insertions, 0 deletions

diff --git a/src/mongo/SConscript b/src/mongo/SConscript
index 33344198140..d8c2c75bf9b 100644
--- a/src/mongo/SConscript
+++ b/src/mongo/SConscript
@@ -188,6 +188,11 @@ env.CppUnitTest('sock_test', ['util/net/sock_test.cpp'],
                 LIBDEPS=['mongocommon'],
                 NO_CRUTCH=True)
 
+env.CppUnitTest('curop_test',
+                ['db/curop_test.cpp'],
+                LIBDEPS=['serveronly', 'coredb', 'coreserver'],
+                NO_CRUTCH=True)
+
 env.StaticLibrary( 'mongohasher', [ "db/hasher.cpp" ] )
 
 env.StaticLibrary('synchronization', [ 'util/concurrency/synchronization.cpp' ])
diff --git a/src/mongo/db/curop.cpp b/src/mongo/db/curop.cpp
index 8544e47ab26..41f6df82604 100644
--- a/src/mongo/db/curop.cpp
+++ b/src/mongo/db/curop.cpp
@@ -48,6 +48,7 @@ namespace mongo {
         _command = false;
         _dbprofile = 0;
         _end = 0;
+        _maxTimeTracker.reset();
         _message = "";
         _progressMeter.finished();
         _killPending.store(0);
@@ -256,6 +257,24 @@ namespace mongo {
         }
     }
 
+    void CurOp::setMaxTimeMicros(uint64_t maxTimeMicros) {
+        if (maxTimeMicros == 0) {
+            // 0 is "allow to run indefinitely".
+            return;
+        }
+
+        // Note that calling startTime() will set CurOp::_start if it hasn't been set yet.
+        _maxTimeTracker.setTimeLimit(startTime(), maxTimeMicros);
+    }
+
+    bool CurOp::maxTimeHasExpired() {
+        return _maxTimeTracker.checkTimeLimit();
+    }
+
+    uint64_t CurOp::getRemainingMaxTimeMicros() const {
+        return _maxTimeTracker.getRemainingMicros();
+    }
+
     AtomicUInt CurOp::_nextOpNum;
 
     static Counter64 returnedCounter;
@@ -297,4 +316,76 @@ namespace mongo {
         if ( fastmod )
             fastmodCounter.increment();
     }
+
+    CurOp::MaxTimeTracker::MaxTimeTracker() {
+        reset();
+    }
+
+    void CurOp::MaxTimeTracker::reset() {
+        _enabled = false;
+        _targetEpochMicros = 0;
+        _approxTargetServerMillis = 0;
+    }
+
+    void CurOp::MaxTimeTracker::setTimeLimit(uint64_t startEpochMicros, uint64_t durationMicros) {
+        dassert(durationMicros != 0);
+
+        _enabled = true;
+
+        _targetEpochMicros = startEpochMicros + durationMicros;
+
+        uint64_t now = curTimeMicros64();
+        // If our accurate time source thinks time is not up yet, calculate the next target for
+        // our approximate time source.
+        if (_targetEpochMicros > now) {
+            _approxTargetServerMillis = Listener::getElapsedTimeMillis() +
+                                        static_cast<int64_t>((_targetEpochMicros - now) / 1000);
+        }
+        // Otherwise, set our approximate time source target such that it thinks time is already
+        // up.
+        else {
+            _approxTargetServerMillis = Listener::getElapsedTimeMillis();
+        }
+    }
+
+    bool CurOp::MaxTimeTracker::checkTimeLimit() {
+        if (!_enabled) {
+            return false;
+        }
+
+        // Does our approximate time source think time is not up yet?  If so, return early.
+        if (_approxTargetServerMillis > Listener::getElapsedTimeMillis()) {
+            return false;
+        }
+
+        uint64_t now = curTimeMicros64();
+        // Does our accurate time source think time is not up yet?  If so, readjust the target for
+        // our approximate time source and return early.
+        if (_targetEpochMicros > now) {
+            _approxTargetServerMillis = Listener::getElapsedTimeMillis() +
+                                        static_cast<int64_t>((_targetEpochMicros - now) / 1000);
+            return false;
+        }
+
+        // Otherwise, time is up.
+        return true;
+    }
+
+    uint64_t CurOp::MaxTimeTracker::getRemainingMicros() const {
+        if (!_enabled) {
+            // 0 is "allow to run indefinitely".
+            return 0;
+        }
+
+        // Does our accurate time source think time is up?  If so, claim there is 1 microsecond
+        // left for this operation.
+        uint64_t now = curTimeMicros64();
+        if (_targetEpochMicros <= now) {
+            return 1;
+        }
+
+        // Otherwise, calculate remaining time.
+        return _targetEpochMicros - now;
+    }
+
 }
diff --git a/src/mongo/db/curop.h b/src/mongo/db/curop.h
index 0370801b11e..3f0fdf4379c 100644
--- a/src/mongo/db/curop.h
+++ b/src/mongo/db/curop.h
@@ -200,6 +200,34 @@ namespace mongo {
         int getOp() const { return _op; }
 
         //
+        // Methods for controlling CurOp "max time".
+        //
+
+        /**
+         * Sets the amount of time operation this should be allowed to run, units of microseconds.
+         * The special value 0 is "allow to run indefinitely".
+         */
+        void setMaxTimeMicros(uint64_t maxTimeMicros);
+
+        /**
+         * Checks whether this operation has been running longer than its time limit.  Returns
+         * false if not, or if the operation has no time limit.
+         *
+         * Note that KillCurrentOp objects are responsible for interrupting CurOp objects that
+         * have exceeded their allotted time; CurOp objects do not interrupt themselves.
+         */
+        bool maxTimeHasExpired();
+
+        /**
+         * Returns the number of microseconds remaining for this operation's time limit, or the
+         * special value 0 if the operation has no time limit.
+         *
+         * Calling this method is more expensive than calling its sibling "maxTimeHasExpired()",
+         * since an accurate measure of remaining time needs to be calculated.
+         */
+        uint64_t getRemainingMaxTimeMicros() const;
+
+        //
         // Methods for getting/setting elapsed time.
         //
 
@@ -299,5 +327,51 @@ namespace mongo {
         // a writebacklisten for example will block for 30s 
         // so this should be 30000 in that case
         long long _expectedLatencyMs; 
+
+        /** Nested class that implements a time limit ($maxTimeMS) for a CurOp object. */
+        class MaxTimeTracker {
+            MONGO_DISALLOW_COPYING(MaxTimeTracker);
+        public:
+            /** Newly-constructed MaxTimeTracker objects have the time limit disabled. */
+            MaxTimeTracker();
+
+            /** Disables the time limit. */
+            void reset();
+
+            /**
+             * Enables the time limit to be "durationMicros" microseconds from "startEpochMicros"
+             * (units of microseconds since the epoch).
+             *
+             * "durationMicros" must be nonzero.
+             */
+            void setTimeLimit(uint64_t startEpochMicros, uint64_t durationMicros);
+
+            /**
+             * Checks whether the time limit has been hit.  Returns false if not, or if the time
+             * limit is disabled.
+             */
+            bool checkTimeLimit();
+
+            /**
+             * Returns the number of microseconds remaining for the time limit, or the special
+             * value 0 if the time limit is disabled.
+             *
+             * Calling this method is more expensive than calling its sibling "checkInterval()",
+             * since an accurate measure of remaining time needs to be calculated.
+             */
+            uint64_t getRemainingMicros() const;
+        private:
+            // Whether or not this operation is subject to a time limit.
+            bool _enabled;
+
+            // Point in time at which the time limit is hit.  Units of microseconds since the
+            // epoch.
+            uint64_t _targetEpochMicros;
+
+            // Approximate point in time at which the time limit is hit.   Units of milliseconds
+            // since the server process was started.
+            int64_t _approxTargetServerMillis;
+        } _maxTimeTracker;
+
     };
 }
diff --git a/src/mongo/db/curop_test.cpp b/src/mongo/db/curop_test.cpp
new file mode 100644
index 00000000000..8435b7b88d2
--- /dev/null
+++ b/src/mongo/db/curop_test.cpp
@@ -0,0 +1,80 @@
+/**
+ *    Copyright (C) 2013 10gen Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include <boost/thread/thread.hpp>
+
+#include "mongo/base/init.h"
+#include "mongo/db/curop.h"
+#include "mongo/unittest/unittest.h"
+
+namespace mongo {
+
+    CmdLine cmdLine; // needed to satisfy reference in curop.h (and elsewhere)
+
+    namespace {
+
+        const long long intervalLong = 2000 * 1000; // 2s in micros
+        const long long intervalShort = 10 * 1000; // 10ms in micros
+
+        //
+        // Before executing the TimeHasExpired suite, spawn a dummy listener thread to be the
+        // process time tracker (the tests rely on Listener::_timeTracker being available).
+        //
+
+        class TestListener : public Listener {
+        public:
+            TestListener() : Listener("test", "", 0) {} // port 0 => any available high port
+            virtual void acceptedMP(MessagingPort *mp) {}
+        };
+
+        void timeTrackerSetup() {
+            TestListener listener;
+            listener.setAsTimeTracker();
+            listener.initAndListen();
+        }
+
+        MONGO_INITIALIZER(CurOpTest)(InitializerContext* context) {
+            boost::thread t(timeTrackerSetup);
+
+            // Wait for listener thread to start tracking time.
+            while (Listener::getElapsedTimeMillis() == 0) {
+                sleepmillis(10);
+            }
+
+            return Status::OK();
+        }
+
+        // Long operation + short timeout => time should expire.
+        TEST(TimeHasExpired, PosSimple) {
+            CurOp curOp(NULL);
+            curOp.setMaxTimeMicros(intervalShort);
+            curOp.ensureStarted();
+            sleepmicros(intervalLong);
+            ASSERT_TRUE(curOp.maxTimeHasExpired());
+        }
+
+        // Short operation + long timeout => time should not expire.
+        TEST(TimeHasExpired, NegSimple) {
+            CurOp curOp(NULL);
+            curOp.setMaxTimeMicros(intervalLong);
+            curOp.ensureStarted();
+            sleepmicros(intervalShort);
+            ASSERT_FALSE(curOp.maxTimeHasExpired());
+        }
+
+    } // namespace
+
+} // namespace mongo
diff --git a/src/mongo/db/kill_current_op.cpp b/src/mongo/db/kill_current_op.cpp
index a938489ecc9..12901a345a2 100644
--- a/src/mongo/db/kill_current_op.cpp
+++ b/src/mongo/db/kill_current_op.cpp
@@ -114,6 +114,11 @@ namespace mongo {
         if (_globalKill) {
             uasserted(11600, "interrupted at shutdown");
         }
+        if (c.curop()->maxTimeHasExpired()) {
+            c.curop()->kill();
+            notifyAllWaiters();
+            uasserted(16986, "operation exceeded time limit");
+        }
         if (c.curop()->killPending()) {
             notifyAllWaiters();
             uasserted(11601, "operation was interrupted");
@@ -126,6 +131,10 @@ namespace mongo {
         if (_globalKill) {
             return "interrupted at shutdown";
         }
+        if (c.curop()->maxTimeHasExpired()) {
+            c.curop()->kill();
+            return "exceeded time limit";
+        }
         if (c.curop()->killPending()) {
             return "interrupted";
         }