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, 229 insertions, 0 deletions

diff --git a/src/mongo/db/operation_context_test.cpp b/src/mongo/db/operation_context_test.cpp
index b779102b702..02891cfa151 100644
--- a/src/mongo/db/operation_context_test.cpp
+++ b/src/mongo/db/operation_context_test.cpp
@@ -200,6 +200,25 @@ TEST(OperationContextTest, OpCtxGroup) {
     }
 }
 
+TEST(OperationContextTest, IgnoreInterruptsWorks) {
+    auto serviceCtx = ServiceContext::make();
+    auto client = serviceCtx->makeClient("OperationContextTest");
+    auto opCtx = client->makeOperationContext();
+
+    opCtx->markKilled(ErrorCodes::BadValue);
+    ASSERT_THROWS_CODE(opCtx->checkForInterrupt(), DBException, ErrorCodes::BadValue);
+    ASSERT_EQUALS(opCtx->getKillStatus(), ErrorCodes::BadValue);
+
+    opCtx->runWithoutInterruption([&] {
+        ASSERT_OK(opCtx->checkForInterruptNoAssert());
+        ASSERT_OK(opCtx->getKillStatus());
+    });
+
+    ASSERT_THROWS_CODE(opCtx->checkForInterrupt(), DBException, ErrorCodes::BadValue);
+
+    ASSERT_EQUALS(opCtx->getKillStatus(), ErrorCodes::BadValue);
+}
+
 class OperationDeadlineTests : public unittest::Test {
 public:
     void setUp() {
@@ -210,6 +229,13 @@ public:
         client = service->makeClient("OperationDeadlineTest");
     }
 
+    void checkForInterruptForTimeout(OperationContext* opCtx) {
+        stdx::mutex m;
+        stdx::condition_variable cv;
+        stdx::unique_lock<stdx::mutex> lk(m);
+        opCtx->waitForConditionOrInterrupt(cv, lk);
+    }
+
     const std::shared_ptr<ClockSourceMock> mockClock = std::make_shared<ClockSourceMock>();
     ServiceContext::UniqueServiceContext service;
     ServiceContext::UniqueClient client;
@@ -302,6 +328,209 @@ TEST_F(OperationDeadlineTests, WaitForMaxTimeExpiredCVWithWaitUntilSet) {
             .getStatus());
 }
 
+TEST_F(OperationDeadlineTests, NestedTimeoutsTimeoutInOrder) {
+    auto opCtx = client->makeOperationContext();
+
+    opCtx->setDeadlineByDate(mockClock->now() + Milliseconds(500), ErrorCodes::MaxTimeMSExpired);
+
+    bool reachedA = false;
+    bool reachedB = false;
+    bool reachedC = false;
+
+    try {
+        opCtx->runWithDeadline(
+            mockClock->now() + Milliseconds(100), ErrorCodes::ExceededTimeLimit, [&] {
+                ASSERT_OK(opCtx->checkForInterruptNoAssert());
+
+                try {
+                    opCtx->runWithDeadline(
+                        mockClock->now() + Milliseconds(50), ErrorCodes::ExceededTimeLimit, [&] {
+                            ASSERT_OK(opCtx->checkForInterruptNoAssert());
+                            try {
+                                opCtx->runWithDeadline(mockClock->now() + Milliseconds(10),
+                                                       ErrorCodes::ExceededTimeLimit,
+                                                       [&] {
+                                                           ASSERT_OK(
+                                                               opCtx->checkForInterruptNoAssert());
+                                                           ASSERT_OK(opCtx->getKillStatus());
+                                                           mockClock->advance(Milliseconds(20));
+                                                           checkForInterruptForTimeout(opCtx.get());
+                                                           ASSERT(false);
+                                                       });
+                            } catch (const ExceptionFor<ErrorCodes::ExceededTimeLimit>&) {
+                                opCtx->checkForInterrupt();
+                                ASSERT_OK(opCtx->getKillStatus());
+                                mockClock->advance(Milliseconds(50));
+                                reachedA = true;
+                            }
+
+                            opCtx->checkForInterrupt();
+                        });
+                } catch (const ExceptionFor<ErrorCodes::ExceededTimeLimit>&) {
+                    opCtx->checkForInterrupt();
+                    ASSERT_OK(opCtx->getKillStatus());
+                    mockClock->advance(Milliseconds(50));
+                    reachedB = true;
+                }
+
+                opCtx->checkForInterrupt();
+            });
+    } catch (const ExceptionFor<ErrorCodes::ExceededTimeLimit>&) {
+        reachedC = true;
+        ASSERT_OK(opCtx->getKillStatus());
+        ASSERT_OK(opCtx->checkForInterruptNoAssert());
+    }
+
+    ASSERT(reachedA);
+    ASSERT(reachedB);
+    ASSERT(reachedC);
+
+    ASSERT_OK(opCtx->getKillStatus());
+
+    mockClock->advance(Seconds(1));
+
+    ASSERT_THROWS_CODE(opCtx->checkForInterrupt(), DBException, ErrorCodes::MaxTimeMSExpired);
+}
+
+TEST_F(OperationDeadlineTests, NestedTimeoutsThatViolateMaxTime) {
+    auto opCtx = client->makeOperationContext();
+
+    opCtx->setDeadlineByDate(mockClock->now() + Milliseconds(10), ErrorCodes::MaxTimeMSExpired);
+
+    bool reachedA = false;
+    bool reachedB = false;
+
+    try {
+        opCtx->runWithDeadline(
+            mockClock->now() + Milliseconds(100), ErrorCodes::ExceededTimeLimit, [&] {
+                ASSERT_OK(opCtx->checkForInterruptNoAssert());
+                try {
+                    opCtx->runWithDeadline(
+                        mockClock->now() + Milliseconds(100), ErrorCodes::ExceededTimeLimit, [&] {
+                            ASSERT_OK(opCtx->checkForInterruptNoAssert());
+                            ASSERT_OK(opCtx->getKillStatus());
+                            mockClock->advance(Milliseconds(50));
+                            opCtx->checkForInterrupt();
+                        });
+                } catch (const ExceptionFor<ErrorCodes::ExceededTimeLimit>&) {
+                    reachedA = true;
+                }
+
+                opCtx->checkForInterrupt();
+            });
+    } catch (const ExceptionFor<ErrorCodes::MaxTimeMSExpired>&) {
+        reachedB = true;
+    }
+
+    ASSERT(reachedA);
+    ASSERT(reachedB);
+}
+
+TEST_F(OperationDeadlineTests, NestedNonMaxTimeMSTimeoutsThatAreLargerAreIgnored) {
+    auto opCtx = client->makeOperationContext();
+
+    bool reachedA = false;
+    bool reachedB = false;
+
+    try {
+        opCtx->runWithDeadline(
+            mockClock->now() + Milliseconds(10), ErrorCodes::ExceededTimeLimit, [&] {
+                ASSERT_OK(opCtx->checkForInterruptNoAssert());
+                try {
+                    opCtx->runWithDeadline(
+                        mockClock->now() + Milliseconds(100), ErrorCodes::ExceededTimeLimit, [&] {
+                            ASSERT_OK(opCtx->checkForInterruptNoAssert());
+                            ASSERT_OK(opCtx->getKillStatus());
+                            mockClock->advance(Milliseconds(50));
+                            opCtx->checkForInterrupt();
+                        });
+                } catch (const ExceptionFor<ErrorCodes::ExceededTimeLimit>&) {
+                    reachedA = true;
+                }
+
+                opCtx->checkForInterrupt();
+            });
+    } catch (const ExceptionFor<ErrorCodes::ExceededTimeLimit>&) {
+        reachedB = true;
+    }
+
+    ASSERT(reachedA);
+    ASSERT(reachedB);
+}
+
+TEST_F(OperationDeadlineTests, DeadlineAfterIgnoreInterruptsReopens) {
+    auto opCtx = client->makeOperationContext();
+
+    bool reachedA = false;
+    bool reachedB = false;
+    bool reachedC = false;
+
+    try {
+        opCtx->runWithDeadline(
+            mockClock->now() + Milliseconds(500), ErrorCodes::ExceededTimeLimit, [&] {
+                ASSERT_OK(opCtx->checkForInterruptNoAssert());
+
+                opCtx->runWithoutInterruption([&] {
+                    try {
+                        opCtx->runWithDeadline(
+                            mockClock->now() + Seconds(1), ErrorCodes::ExceededTimeLimit, [&] {
+                                ASSERT_OK(opCtx->checkForInterruptNoAssert());
+                                ASSERT_OK(opCtx->getKillStatus());
+                                mockClock->advance(Milliseconds(750));
+                                ASSERT_OK(opCtx->checkForInterruptNoAssert());
+                                mockClock->advance(Milliseconds(500));
+                                reachedA = true;
+                                opCtx->checkForInterrupt();
+                            });
+                    } catch (const ExceptionFor<ErrorCodes::ExceededTimeLimit>&) {
+                        opCtx->checkForInterrupt();
+                        reachedB = true;
+                    }
+                });
+
+                opCtx->checkForInterrupt();
+            });
+    } catch (const ExceptionFor<ErrorCodes::ExceededTimeLimit>& ex) {
+        reachedC = true;
+    }
+
+    ASSERT(reachedA);
+    ASSERT(reachedB);
+    ASSERT(reachedC);
+}
+
+TEST_F(OperationDeadlineTests, DeadlineAfterRunWithoutInterruptSeesViolatedMaxMS) {
+    auto opCtx = client->makeOperationContext();
+
+    opCtx->setDeadlineByDate(mockClock->now() + Milliseconds(100), ErrorCodes::MaxTimeMSExpired);
+
+    ASSERT_THROWS_CODE(opCtx->runWithoutInterruption([&] {
+        opCtx->runWithDeadline(
+            mockClock->now() + Milliseconds(200), ErrorCodes::ExceededTimeLimit, [&] {
+                mockClock->advance(Milliseconds(300));
+                opCtx->checkForInterrupt();
+            });
+    }),
+                       DBException,
+                       ErrorCodes::MaxTimeMSExpired);
+}
+
+TEST_F(OperationDeadlineTests, DeadlineAfterRunWithoutInterruptDoesntSeeUnviolatedMaxMS) {
+    auto opCtx = client->makeOperationContext();
+
+    opCtx->setDeadlineByDate(mockClock->now() + Milliseconds(200), ErrorCodes::MaxTimeMSExpired);
+
+    ASSERT_THROWS_CODE(opCtx->runWithoutInterruption([&] {
+        opCtx->runWithDeadline(
+            mockClock->now() + Milliseconds(100), ErrorCodes::ExceededTimeLimit, [&] {
+                mockClock->advance(Milliseconds(150));
+                opCtx->checkForInterrupt();
+            });
+    }),
+                       DBException,
+                       ErrorCodes::ExceededTimeLimit);
+}
+
 TEST_F(OperationDeadlineTests, WaitForKilledOpCV) {
     auto opCtx = client->makeOperationContext();
     opCtx->markKilled();