SERVER-35679 General Interruption Facility

Add support for a generalized interruptibility facility in the server.

This offers a generalized interruptibility facility, trialed in
Future<T> and ProducerConsumerQueue<T>.

It offers 3 major concepts:

Notifyable:  A type which can notified off-thread, causing a wake up
from some kind of blocking wait

Waitable: A type which is Notifyable, and also can perform work while in
a ready-to-receive notification state.  static methods offer support for
running underneath condition_variable::wait's.  The chief implementer is
the transport layer baton type

Interruptible: A type which can wait on condition variables, and offers:
  - deadlines.  This means the type integrates some sort of clock source
  - interruptibility.  This means the type offers a way of noticing
    that it should no longer run via status or exception

Additionally, Interruptible's offer special scoped guards which
offer
  - Exemption from interruption in a region defined by the lifetime
    of a guard object
  - Subsidiary deadlines which can trigger recursively, offering
    specialized timeout and status return support.

The series of virtual types allows us to slice the interface between
opCtx and future such that opctx can use future and future can use
opctx.  Additionally, cutting out more functionality allows us to flow a
noop interruptibility type which unifies our waiting behind a common
api.

1 files changed, 19 insertions, 51 deletions

diff --git a/src/mongo/db/operation_context.cpp b/src/mongo/db/operation_context.cpp
index 94a5ea312d2..873972a8c30 100644
--- a/src/mongo/db/operation_context.cpp
+++ b/src/mongo/db/operation_context.cpp
@@ -88,7 +88,9 @@ void OperationContext::setDeadlineAndMaxTime(Date_t when,
                                              ErrorCodes::Error timeoutError) {
     invariant(!getClient()->isInDirectClient());
     invariant(ErrorCodes::isExceededTimeLimitError(timeoutError));
-    uassert(40120, "Illegal attempt to change operation deadline", !hasDeadline());
+    uassert(40120,
+            "Illegal attempt to change operation deadline",
+            _hasArtificialDeadline || !hasDeadline());
     _deadline = when;
     _maxTime = maxTime;
     _timeoutError = timeoutError;
@@ -165,10 +167,6 @@ Microseconds OperationContext::getRemainingMaxTimeMicros() const {
     return _maxTime - getElapsedTime();
 }
 
-void OperationContext::checkForInterrupt() {
-    uassertStatusOK(checkForInterruptNoAssert());
-}
-
 namespace {
 
 // Helper function for checkForInterrupt fail point.  Decides whether the operation currently
@@ -190,7 +188,7 @@ bool opShouldFail(Client* client, const BSONObj& failPointInfo) {
 
 }  // namespace
 
-Status OperationContext::checkForInterruptNoAssert() {
+Status OperationContext::checkForInterruptNoAssert() noexcept {
     // TODO: Remove the MONGO_likely(getClient()) once all operation contexts are constructed with
     // clients.
     if (MONGO_likely(getClient() && getServiceContext()) &&
@@ -199,10 +197,16 @@ Status OperationContext::checkForInterruptNoAssert() {
     }
 
     if (hasDeadlineExpired()) {
-        markKilled(_timeoutError);
+        if (!_hasArtificialDeadline) {
+            markKilled(_timeoutError);
+        }
         return Status(_timeoutError, "operation exceeded time limit");
     }
 
+    if (_ignoreInterrupts) {
+        return Status::OK();
+    }
+
     MONGO_FAIL_POINT_BLOCK(checkForInterruptFail, scopedFailPoint) {
         if (opShouldFail(getClient(), scopedFailPoint.getData())) {
             log() << "set pending kill on op " << getOpID() << ", for checkForInterruptFail";
@@ -218,41 +222,6 @@ Status OperationContext::checkForInterruptNoAssert() {
     return Status::OK();
 }
 
-void OperationContext::sleepUntil(Date_t deadline) {
-    stdx::mutex m;
-    stdx::condition_variable cv;
-    stdx::unique_lock<stdx::mutex> lk(m);
-    invariant(!waitForConditionOrInterruptUntil(cv, lk, deadline, [] { return false; }));
-}
-
-void OperationContext::sleepFor(Milliseconds duration) {
-    stdx::mutex m;
-    stdx::condition_variable cv;
-    stdx::unique_lock<stdx::mutex> lk(m);
-    invariant(!waitForConditionOrInterruptFor(cv, lk, duration, [] { return false; }));
-}
-
-void OperationContext::waitForConditionOrInterrupt(stdx::condition_variable& cv,
-                                                   stdx::unique_lock<stdx::mutex>& m) {
-    uassertStatusOK(waitForConditionOrInterruptNoAssert(cv, m));
-}
-
-Status OperationContext::waitForConditionOrInterruptNoAssert(
-    stdx::condition_variable& cv, stdx::unique_lock<stdx::mutex>& m) noexcept {
-    auto status = waitForConditionOrInterruptNoAssertUntil(cv, m, Date_t::max());
-    if (!status.isOK()) {
-        return status.getStatus();
-    }
-    invariant(status.getValue() == stdx::cv_status::no_timeout);
-    return status.getStatus();
-}
-
-stdx::cv_status OperationContext::waitForConditionOrInterruptUntil(
-    stdx::condition_variable& cv, stdx::unique_lock<stdx::mutex>& m, Date_t deadline) {
-
-    return uassertStatusOK(waitForConditionOrInterruptNoAssertUntil(cv, m, deadline));
-}
-
 // Theory of operation for waitForConditionOrInterruptNoAssertUntil and markKilled:
 //
 // An operation indicates to potential killers that it is waiting on a condition variable by setting
@@ -312,14 +281,15 @@ StatusWith<stdx::cv_status> OperationContext::waitForConditionOrInterruptNoAsser
 
     const auto waitStatus = [&] {
         if (Date_t::max() == deadline) {
-            cv.wait(m);
+            Waitable::wait(_baton.get(), getServiceContext()->getPreciseClockSource(), cv, m);
             return stdx::cv_status::no_timeout;
         }
-        return getServiceContext()->getPreciseClockSource()->waitForConditionUntil(cv, m, deadline);
+        return getServiceContext()->getPreciseClockSource()->waitForConditionUntil(
+            cv, m, deadline, _baton.get());
     }();
 
     // Continue waiting on cv until no other thread is attempting to kill this one.
-    cv.wait(m, [this] {
+    Waitable::wait(_baton.get(), getServiceContext()->getPreciseClockSource(), cv, m, [this] {
         stdx::lock_guard<Client> clientLock(*getClient());
         if (0 == _numKillers) {
             _waitMutex = nullptr;
@@ -338,9 +308,12 @@ StatusWith<stdx::cv_status> OperationContext::waitForConditionOrInterruptNoAsser
         // is slightly ahead of the FastClock used in checkForInterrupt. In this case,
         // we treat the operation as though it has exceeded its time limit, just as if the
         // FastClock and system clock had agreed.
-        markKilled(_timeoutError);
+        if (!_hasArtificialDeadline) {
+            markKilled(_timeoutError);
+        }
         return Status(_timeoutError, "operation exceeded time limit");
     }
+
     return waitStatus;
 }
 
@@ -361,11 +334,6 @@ void OperationContext::markKilled(ErrorCodes::Error killCode) {
         invariant(_waitCV);
         _waitCV->notify_all();
     }
-
-    // If we have a baton, we need to wake it up.  The baton itself will check for interruption
-    if (_baton) {
-        _baton->schedule([] {});
-    }
 }
 
 void OperationContext::setLogicalSessionId(LogicalSessionId lsid) {