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.

3 files changed, 239 insertions, 1 deletions

diff --git a/src/mongo/stdx/SConscript b/src/mongo/stdx/SConscript
new file mode 100644
index 00000000000..a51aa3423f4
--- /dev/null
+++ b/src/mongo/stdx/SConscript
@@ -0,0 +1,15 @@
+# -*- mode: python -*-
+
+Import("env")
+
+env = env.Clone()
+
+env.Benchmark(
+    target='condition_variable_bm',
+    source=[
+        'condition_variable_bm.cpp',
+    ],
+    LIBDEPS=[
+    ],
+)
+
diff --git a/src/mongo/stdx/condition_variable.h b/src/mongo/stdx/condition_variable.h
index 40ca92ba8f6..6de5dbdb66d 100644
--- a/src/mongo/stdx/condition_variable.h
+++ b/src/mongo/stdx/condition_variable.h
@@ -28,15 +28,151 @@
 
 #pragma once
 
+#include <atomic>
 #include <condition_variable>
+#include <list>
+
+#include "mongo/platform/atomic_word.h"
+#include "mongo/stdx/mutex.h"
 
 namespace mongo {
+
+/**
+ * Notifyable is a slim type meant to allow integration of special kinds of waiters for
+ * stdx::condition_variable.  Specifially, the notify() on this type will be called directly from
+ * stdx::condition_varibale::notify_(one|all).
+ *
+ * See Waitable for the stdx::condition_variable integration.
+ */
+class Notifyable {
+public:
+    virtual void notify() noexcept = 0;
+
+protected:
+    ~Notifyable() = default;
+};
+
+class Waitable;
+
 namespace stdx {
 
-using condition_variable = ::std::condition_variable;          // NOLINT
 using condition_variable_any = ::std::condition_variable_any;  // NOLINT
 using cv_status = ::std::cv_status;                            // NOLINT
 using ::std::notify_all_at_thread_exit;                        // NOLINT
 
+/**
+ * We wrap std::condition_variable to allow us to register Notifyables which can "wait" on the
+ * condvar without actually waiting on the std::condition_variable.  This allows us to possibly do
+ * productive work in those types, rather than sleeping in the os.
+ */
+class condition_variable : private std::condition_variable {  // NOLINT
+public:
+    using std::condition_variable::condition_variable;  // NOLINT
+
+    void notify_one() noexcept {
+        if (_notifyableCount.load()) {
+            stdx::lock_guard<stdx::mutex> lk(_mutex);
+
+            if (_notifyNextNotifyable(lk)) {
+                return;
+            }
+        }
+
+        std::condition_variable::notify_one();  // NOLINT
+    }
+
+    void notify_all() noexcept {
+        if (_notifyableCount.load()) {
+            stdx::lock_guard<stdx::mutex> lk(_mutex);
+
+            while (_notifyNextNotifyable(lk)) {
+            }
+        }
+
+        std::condition_variable::notify_all();  // NOLINT
+    }
+
+    using std::condition_variable::wait;           // NOLINT
+    using std::condition_variable::wait_for;       // NOLINT
+    using std::condition_variable::wait_until;     // NOLINT
+    using std::condition_variable::native_handle;  // NOLINT
+
+private:
+    friend class ::mongo::Waitable;
+
+    /**
+     * Runs the callback with the Notifyable registered on the condvar.  This ensures that for the
+     * duration of the callback execution, a notification on the condvar will trigger a notify() to
+     * the Notifyable.
+     *
+     * The scheme here is that list entries are erased from the notification list when notified (so
+     * that they don't eat multiple notify_one's).  We detect that condition by noting that our
+     * Notifyable* has been overwritten with null (in which case we should avoid a double erase).
+     *
+     * The method is private, and accessed via friendship in Waitable.
+     */
+    template <typename Callback>
+    void _runWithNotifyable(Notifyable& notifyable, Callback&& cb) noexcept {
+        static_assert(noexcept(std::forward<Callback>(cb)()),
+                      "Only noexcept functions may be invoked with _runWithNotifyable");
+
+        // We use this local pad to receive notification that we were notified, rather than timing
+        // out organically.
+        //
+        // Note that n must be guarded by _mutex after its insertion in _notifyables (so that we can
+        // detect notification in a thread-safe manner).
+        Notifyable* n = &notifyable;
+
+        auto iter = [&] {
+            stdx::lock_guard<stdx::mutex> localMutex(_mutex);
+            _notifyableCount.addAndFetch(1);
+            return _notifyables.insert(_notifyables.end(), &n);
+        }();
+
+        std::forward<Callback>(cb)();
+
+        stdx::lock_guard<stdx::mutex> localMutex(_mutex);
+        // if n is null, we were notified, and erased in _notifyNextNotifyable
+        if (n) {
+            _notifyableCount.subtractAndFetch(1);
+            _notifyables.erase(iter);
+        }
+    }
+
+    /**
+     * Notifies the next notifyable.
+     *
+     * Returns true if there was a notifyable to be notified.
+     *
+     * Note that as part of notifying, we zero out pointers allocated on the stack by
+     * _runWithNotifyable callers.  This is safe because we hold _mutex while we do so, and our
+     * erasure communicates that those waiters need not clear themselves from the notification list
+     * on wakeup.
+     */
+    bool _notifyNextNotifyable(const stdx::lock_guard<stdx::mutex>&) noexcept {
+        auto iter = _notifyables.begin();
+        if (iter == _notifyables.end()) {
+            return false;
+        }
+
+        _notifyableCount.subtractAndFetch(1);
+
+        (**iter)->notify();
+
+        // null out iter here, so that the notifyable won't remove itself from the list when it
+        // wakes up
+        **iter = nullptr;
+
+        _notifyables.erase(iter);
+
+        return true;
+    }
+
+    AtomicUInt64 _notifyableCount;
+
+    stdx::mutex _mutex;
+    std::list<Notifyable**> _notifyables;
+};
+
 }  // namespace stdx
 }  // namespace mongo
diff --git a/src/mongo/stdx/condition_variable_bm.cpp b/src/mongo/stdx/condition_variable_bm.cpp
new file mode 100644
index 00000000000..7e020d60c0e
--- /dev/null
+++ b/src/mongo/stdx/condition_variable_bm.cpp
@@ -0,0 +1,87 @@
+/**
+ *    Copyright (C) 2018 MongoDB 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/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/platform/basic.h"
+
+#include <benchmark/benchmark.h>
+
+#include "mongo/bson/inline_decls.h"
+#include "mongo/stdx/condition_variable.h"
+#include "mongo/stdx/mutex.h"
+#include "mongo/stdx/thread.h"
+
+namespace mongo {
+
+void BM_stdNotifyOne(benchmark::State& state) {
+    std::condition_variable cv;  // NOLINT
+
+    for (auto _ : state) {
+        benchmark::ClobberMemory();
+        cv.notify_one();
+    }
+}
+
+void BM_stdxNotifyOneNoNotifyables(benchmark::State& state) {
+    stdx::condition_variable cv;
+
+    for (auto _ : state) {
+        benchmark::ClobberMemory();
+        cv.notify_one();
+    }
+}
+
+volatile bool alwaysTrue = true;
+
+void BM_stdWaitWithTruePredicate(benchmark::State& state) {
+    std::condition_variable cv;  // NOLINT
+    stdx::mutex mutex;
+    stdx::unique_lock<stdx::mutex> lk(mutex);
+
+    for (auto _ : state) {
+        benchmark::ClobberMemory();
+        cv.wait(lk, [&] { return alwaysTrue; });
+    }
+}
+
+void BM_stdxWaitWithTruePredicate(benchmark::State& state) {
+    stdx::condition_variable cv;
+    stdx::mutex mutex;
+    stdx::unique_lock<stdx::mutex> lk(mutex);
+
+    for (auto _ : state) {
+        benchmark::ClobberMemory();
+        cv.wait(lk, [&] { return alwaysTrue; });
+    }
+}
+
+BENCHMARK(BM_stdNotifyOne);
+BENCHMARK(BM_stdWaitWithTruePredicate);
+BENCHMARK(BM_stdxNotifyOneNoNotifyables);
+BENCHMARK(BM_stdxWaitWithTruePredicate);
+
+}  // namespace mongo