You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by be...@apache.org on 2017/08/01 21:03:09 UTC
[11/16] mesos git commit: Added a lock-free event queue.
Added a lock-free event queue.
Review: https://reviews.apache.org/r/61058
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/1a8b36e2
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/1a8b36e2
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/1a8b36e2
Branch: refs/heads/master
Commit: 1a8b36e2cf0fe339be40f17ddadf81c5fc9844c7
Parents: 535e77e
Author: Benjamin Hindman <be...@gmail.com>
Authored: Fri Jul 21 15:12:03 2017 -0700
Committer: Benjamin Hindman <be...@gmail.com>
Committed: Tue Aug 1 14:01:51 2017 -0700
----------------------------------------------------------------------
3rdparty/libprocess/configure.ac | 11 ++
3rdparty/libprocess/src/CMakeLists.txt | 4 +
3rdparty/libprocess/src/event_queue.hpp | 280 +++++++++++++++++++++++++--
3rdparty/libprocess/src/process.cpp | 28 +--
4 files changed, 295 insertions(+), 28 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/1a8b36e2/3rdparty/libprocess/configure.ac
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/configure.ac b/3rdparty/libprocess/configure.ac
index cb2cf4f..131f149 100644
--- a/3rdparty/libprocess/configure.ac
+++ b/3rdparty/libprocess/configure.ac
@@ -69,6 +69,13 @@ AC_CONFIG_FILES([3rdparty/gmock_sources.cc])
# Optional features.
###############################################################################
+# TODO(benh): Eventaully make this enabled by default.
+AC_ARG_ENABLE([lock_free_event_queue],
+ AS_HELP_STRING([--enable-lock-free-event-queue],
+ [enables the lock free event queue]),
+ [], [enable_lock_free_event_queue=no])
+
+# TODO(benh): Eventaully make this enabled by default.
AC_ARG_ENABLE([lock_free_run_queue],
AS_HELP_STRING([--enable-lock-free-run-queue],
[enables the lock free run queue]),
@@ -240,6 +247,10 @@ AC_ARG_VAR([TEST_DRIVER], [executable and arguments of a test driver])
###############################################################################
# Check if we should use the lock free run queue.
+AS_IF([test "x$enable_lock_free_event_queue" = "xyes"],
+ [AC_DEFINE([LOCK_FREE_EVENT_QUEUE])])
+
+# Check if we should use the lock free run queue.
AS_IF([test "x$enable_lock_free_run_queue" = "xyes"],
[AC_DEFINE([LOCK_FREE_RUN_QUEUE])])
http://git-wip-us.apache.org/repos/asf/mesos/blob/1a8b36e2/3rdparty/libprocess/src/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/src/CMakeLists.txt b/3rdparty/libprocess/src/CMakeLists.txt
index 8d2150e..c241e2a 100644
--- a/3rdparty/libprocess/src/CMakeLists.txt
+++ b/3rdparty/libprocess/src/CMakeLists.txt
@@ -145,3 +145,7 @@ target_link_libraries(${PROCESS_TARGET} ${PROCESS_LIBS})
if (ENABLE_LOCK_FREE_RUN_QUEUE)
target_compile_definitions(${PROCESS_TARGET} PRIVATE LOCK_FREE_RUN_QUEUE)
endif ()
+
+if (ENABLE_LOCK_FREE_EVENT_QUEUE)
+ target_compile_definitions(${PROCESS_TARGET} PRIVATE LOCK_FREE_EVENT_QUEUE)
+endif ()
http://git-wip-us.apache.org/repos/asf/mesos/blob/1a8b36e2/3rdparty/libprocess/src/event_queue.hpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/src/event_queue.hpp b/3rdparty/libprocess/src/event_queue.hpp
index 37d2359..8b4f53d 100644
--- a/3rdparty/libprocess/src/event_queue.hpp
+++ b/3rdparty/libprocess/src/event_queue.hpp
@@ -17,6 +17,10 @@
#include <mutex>
#include <string>
+#ifdef LOCK_FREE_EVENT_QUEUE
+#include <concurrentqueue.h>
+#endif // LOCK_FREE_EVENT_QUEUE
+
#include <process/event.hpp>
#include <process/http.hpp>
@@ -26,19 +30,87 @@
namespace process {
-// TODO(benh): Document that the contract/semantics that an event
-// queue is designed to be _multiple_ producers and _single_ consumer
-// and that currently that requirement is not enforced or even checked
-// at runtime.
-
+// A _multiple_ producer (MP) _single_ consumer (SC) event queue for a
+// process. Note that we don't _enforce_ the MP/SC semantics during
+// runtime but we have explicitly separated out the `Producer`
+// interface and the `Consumer` interface in order to help avoid
+// incorrect usage.
+//
+// Notable semantics:
+//
+// * Consumers _must_ call `empty()` before calling
+// `dequeue()`. Failing to do so may result in undefined behavior.
+//
+// * After a consumer calls `decomission()` they _must_ not call any
+// thing else (not even `empty()` and especially not
+// `dequeue()`). Doing so is undefined behavior.
+//
+// Notes on the lock-free implementation:
+//
+// The SC requirement is necessary for the lock-free implementation
+// because the underlying queue does not provide linearizability which
+// means events can be dequeued "out of order". Usually this is not a
+// problem, after all, in most circumstances we won't know the order
+// in which events might be enqueued in the first place. However, this
+// can be a very bad problem if a single process attempts to enqueue
+// two events in a different process AND THOSE EVENTS ARE
+// REORDERED. To ensure this will never be the case we give every
+// event a sequence number. That way an event from the same process
+// will always have a happens-before relationship with respect to the
+// events that they enqueue because they'll have distinct sequence
+// numbers.
+//
+// This makes the consumer implementation more difficult because the
+// consumer might need to "reorder" events as it reads them out. To do
+// this efficiently we require only a single consumer, which fits well
+// into the actor model because there will only ever be a single
+// thread consuming an actors events at a time.
class EventQueue
{
public:
+ EventQueue() : producer(this), consumer(this) {}
+
+ class Producer
+ {
+ public:
+ void enqueue(Event* event) { queue->enqueue(event); }
+
+ private:
+ friend class EventQueue;
+
+ Producer(EventQueue* queue) : queue(queue) {}
+
+ EventQueue* queue;
+ } producer;
+
+ class Consumer
+ {
+ public:
+ Event* dequeue() { return queue->dequeue(); }
+ bool empty() { return queue->empty(); }
+ void decomission() { queue->decomission(); }
+ template <typename T>
+ size_t count() { return queue->count<T>(); }
+ operator JSON::Array() { return queue->operator JSON::Array(); }
+
+ private:
+ friend class EventQueue;
+
+ Consumer(EventQueue* queue) : queue(queue) {}
+
+ EventQueue* queue;
+ } consumer;
+
+private:
+ friend class Producer;
+ friend class Consumer;
+
+#ifndef LOCK_FREE_EVENT_QUEUE
void enqueue(Event* event)
{
bool enqueued = false;
synchronized (mutex) {
- if (!decomissioned) {
+ if (comissioned) {
events.push_back(event);
enqueued = true;
}
@@ -51,6 +123,8 @@ public:
Event* dequeue()
{
+ Event* event = nullptr;
+
synchronized (mutex) {
if (events.size() > 0) {
Event* event = events.front();
@@ -58,11 +132,10 @@ public:
return event;
}
}
- // TODO(benh): The current semantics are that a call to dequeue
- // will always be coupled by a call to empty and only get called
- // if we're not empty which implies we should never return nullptr
- // here, should we check this?
- return nullptr;
+
+ // Semantics are the consumer _must_ call `empty()` before calling
+ // `dequeue()` which means an event must be present.
+ return CHECK_NOTNULL(event);
}
bool empty()
@@ -75,7 +148,7 @@ public:
void decomission()
{
synchronized (mutex) {
- decomissioned = true;
+ comissioned = false;
while (!events.empty()) {
Event* event = events.front();
events.pop_front();
@@ -102,17 +175,192 @@ public:
JSON::Array array;
synchronized (mutex) {
foreach (Event* event, events) {
- JSON::Object object = *event;
- array.values.push_back(object);
+ array.values.push_back(JSON::Object(*event));
}
}
return array;
}
-private:
std::mutex mutex;
std::deque<Event*> events;
- bool decomissioned = false;
+ bool comissioned = true;
+#else // LOCK_FREE_EVENT_QUEUE
+ void enqueue(Event* event)
+ {
+ Item item = {sequence.fetch_add(1), event};
+ if (comissioned.load()) {
+ queue.enqueue(std::move(item));
+ } else {
+ sequence.fetch_sub(1);
+ delete event;
+ }
+ }
+
+ Event* dequeue()
+ {
+ // NOTE: for performance reasons we don't check `comissioned` here
+ // so it's possible that we'll loop forever if a consumer called
+ // `decomission()` and then subsequently called `dequeue()`.
+ Event* event = nullptr;
+ do {
+ // Given the nature of the concurrent queue implementation it's
+ // possible that we'll need to try to dequeue multiple times
+ // until it returns an event even though we know there is an
+ // event because the semantics are that we shouldn't call
+ // `dequeue()` before calling `empty()`.
+ event = try_dequeue();
+ } while (event == nullptr);
+ return event;
+ }
+
+ bool empty()
+ {
+ // NOTE: for performance reasons we don't check `comissioned` here
+ // so it's possible that we'll return true when in fact we've been
+ // decomissioned and you shouldn't attempt to dequeue anything.
+ return (sequence.load() - next) == 0;
+ }
+
+ void decomission()
+ {
+ comissioned.store(true);
+ while (!empty()) {
+ // NOTE: we use `try_dequeue()` here because we might be racing
+ // with `enqueue()` where they've already incremented `sequence`
+ // so we think there are more items to dequeue but they aren't
+ // actually going to enqueue anything because they've since seen
+ // `comissioned` is true. We'll attempt to dequeue with
+ // `try_dequeue()` and eventually they'll decrement `sequence`
+ // and so `empty()` will return true and we'll bail.
+ Event* event = try_dequeue();
+ if (event != nullptr) {
+ delete event;
+ }
+ }
+ }
+
+ template <typename T>
+ size_t count()
+ {
+ // Try and dequeue more elements first!
+ queue.try_dequeue_bulk(std::back_inserter(items), SIZE_MAX);
+
+ return std::count_if(
+ items.begin(),
+ items.end(),
+ [](const Item& item) {
+ if (item.event != nullptr) {
+ return item.event->is<T>();
+ }
+ return false;
+ });
+ }
+
+ operator JSON::Array()
+ {
+ // Try and dequeue more elements first!
+ queue.try_dequeue_bulk(std::back_inserter(items), SIZE_MAX);
+
+ JSON::Array array;
+ foreach (const Item& item, items) {
+ if (item.event != nullptr) {
+ array.values.push_back(JSON::Object(*item.event));
+ }
+ }
+
+ return array;
+ }
+
+ struct Item
+ {
+ uint64_t sequence;
+ Event* event;
+ };
+
+ Event* try_dequeue()
+ {
+ // The general algoritm here is as follows: we bulk dequeue as
+ // many items from the concurrent queue as possible. We then look
+ // for the `next` item in the sequence hoping that it's at the
+ // beginning of `items` but because the `queue` is not
+ // linearizable it might be "out of order". If we find it out of
+ // order we effectively dequeue it but leave it in `items` so as
+ // not to incur any costly rearrangements/compactions in
+ // `items`. We'll later pop the out of order items once they get
+ // to the front.
+
+ // Start by popping any items that we effectively dequeued but
+ // didn't remove from `items` so as not to incur costly
+ // rearragements/compactions.
+ while (!items.empty() && next > items.front().sequence) {
+ items.pop_front();
+ }
+
+ // Optimistically let's hope that the next item is at the front of
+ // `item`. If so, pop the item, increment `next`, and return the
+ // event.
+ if (!items.empty() && items.front().sequence == next) {
+ Event* event = items.front().event;
+ items.pop_front();
+ next += 1;
+ return event;
+ }
+
+ size_t index = 0;
+
+ do {
+ // Now look for a potentially out of order item. If found,
+ // signifiy the item has been dequeued by nulling the event
+ // (necessary for the implementation of `count()` and `operator
+ // JSON::Array()`) and return the event.
+ for (; index < items.size(); index++) {
+ if (items[index].sequence == next) {
+ Event* event = items[index].event;
+ items[index].event = nullptr;
+ next += 1;
+ return event;
+ }
+ }
+
+ // If we can bulk dequeue more items then keep looking for the
+ // out of order event!
+ } while (queue.try_dequeue_bulk(std::back_inserter(items), SIZE_MAX) != 0);
+
+ return nullptr;
+ }
+
+ // Underlying queue of items.
+ moodycamel::ConcurrentQueue<Item> queue;
+
+ // Counter to represent the item sequence. Note that we use a
+ // unsigned 64-bit integer which means that even if we were adding
+ // one item to the queue every nanosecond we'd be able to run for
+ // 18,446,744,073,709,551,615 nanoseconds or ~585 years! ;-)
+ std::atomic<uint64_t> sequence = ATOMIC_VAR_INIT(0);
+
+ // Counter to represent the next item we expect to dequeue. Note
+ // that we don't need to make this be atomic because only a single
+ // consumer is ever reading or writing this variable!
+ uint64_t next = 0;
+
+ // Collection of bulk dequeued items that may be out of order. Note
+ // that like `next` this will only ever be read/written by a single
+ // consumer.
+ //
+ // The use of a deque was explicit because it is implemented as an
+ // array of arrays (or vector of vectors) which usually gives good
+ // performance for appending to the back and popping from the front
+ // which is exactly what we need to do. To avoid any performance
+ // issues that might be incurred we do not remove any items from the
+ // middle of the deque (see comments in `try_dequeue()` above for
+ // more details).
+ std::deque<Item> items;
+
+ // Whether or not the event queue has been decomissioned. This must
+ // be atomic as it can be read by a producer even though it's only
+ // written by a consumer.
+ std::atomic<bool> comissioned = ATOMIC_VAR_INIT(true);
+#endif // LOCK_FREE_EVENT_QUEUE
};
} // namespace process {
http://git-wip-us.apache.org/repos/asf/mesos/blob/1a8b36e2/3rdparty/libprocess/src/process.cpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/src/process.cpp b/3rdparty/libprocess/src/process.cpp
index 7395f00..d6b8db3 100644
--- a/3rdparty/libprocess/src/process.cpp
+++ b/3rdparty/libprocess/src/process.cpp
@@ -3195,8 +3195,12 @@ void ProcessManager::resume(ProcessBase* process)
while (!terminate && !blocked) {
Event* event = nullptr;
- if (!process->events->empty()) {
- event = process->events->dequeue();
+ // NOTE: the event queue requires only a _single_ consumer at a
+ // time ... this is where we act as that single consumer (and down
+ // in `ProcessManager::cleanup` which we call from here).
+
+ if (!process->events->consumer.empty()) {
+ event = process->events->consumer.dequeue();
} else {
state = ProcessBase::State::BLOCKED;
process->state.store(state);
@@ -3206,7 +3210,7 @@ void ProcessManager::resume(ProcessBase* process)
// before we set ourselves to BLOCKED since we won't have been
// added to the run queue in those circumstances so we need to
// serve those events!
- if (!process->events->empty()) {
+ if (!process->events->consumer.empty()) {
// Make sure the state is in READY! Either we need to
// explicitly do this because `ProcessBase::enqueue` saw us as
// READY (or BOTTOM) and didn't change the state or we're
@@ -3235,7 +3239,7 @@ void ProcessManager::resume(ProcessBase* process)
// Now purge all events until the terminate event.
while (!event->is<TerminateEvent>()) {
delete event;
- event = process->events->dequeue();
+ event = process->events->consumer.dequeue();
CHECK_NOTNULL(event);
}
}
@@ -3315,7 +3319,7 @@ void ProcessManager::cleanup(ProcessBase* process)
// _new_ process that gets spawned with the same PID.
process->state.store(ProcessBase::State::TERMINATING);
- process->events->decomission();
+ process->events->consumer.decomission();
// Remove help strings for all installed routes for this process.
dispatch(help, &Help::remove, process->pid.id);
@@ -3688,7 +3692,7 @@ template <>
size_t ProcessBase::eventCount<MessageEvent>()
{
CHECK_EQ(this, __process__);
- return events->count<MessageEvent>();
+ return events->consumer.count<MessageEvent>();
}
@@ -3696,7 +3700,7 @@ template <>
size_t ProcessBase::eventCount<DispatchEvent>()
{
CHECK_EQ(this, __process__);
- return events->count<DispatchEvent>();
+ return events->consumer.count<DispatchEvent>();
}
@@ -3704,7 +3708,7 @@ template <>
size_t ProcessBase::eventCount<HttpEvent>()
{
CHECK_EQ(this, __process__);
- return events->count<HttpEvent>();
+ return events->consumer.count<HttpEvent>();
}
@@ -3712,7 +3716,7 @@ template <>
size_t ProcessBase::eventCount<ExitedEvent>()
{
CHECK_EQ(this, __process__);
- return events->count<ExitedEvent>();
+ return events->consumer.count<ExitedEvent>();
}
@@ -3720,7 +3724,7 @@ template <>
size_t ProcessBase::eventCount<TerminateEvent>()
{
CHECK_EQ(this, __process__);
- return events->count<TerminateEvent>();
+ return events->consumer.count<TerminateEvent>();
}
@@ -3741,7 +3745,7 @@ void ProcessBase::enqueue(Event* event)
case State::BOTTOM:
case State::READY:
case State::BLOCKED:
- events->enqueue(event);
+ events->producer.enqueue(event);
break;
case State::TERMINATING:
delete event;
@@ -4070,7 +4074,7 @@ ProcessBase:: operator JSON::Object()
JSON::Object object;
object.values["id"] = pid.id;
- object.values["events"] = JSON::Array(*events);
+ object.values["events"] = JSON::Array(events->consumer);
return object;
}