[45/48] mesos git commit: Removed race condition from libevent based poll implementation.

[vi...@apache.org]

Removed race condition from libevent based poll implementation.

Under certains circumstances, the future returned by poll is discarded
right after the event is triggered, this causes the event callback to be
called before the discard callback which results in an abort signal
being raised by the libevent library.

Review: https://reviews.apache.org/r/43799/


Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/859b4a3d
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/859b4a3d
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/859b4a3d

Branch: refs/heads/0.27.x
Commit: 859b4a3d711d71a048d5c1e6ef07a123007fa7fb
Parents: 5f15fef
Author: Alexander Rojas <al...@mesosphere.io>
Authored: Fri Feb 26 17:17:50 2016 -0800
Committer: Michael Park <mp...@apache.org>
Committed: Fri Feb 26 18:57:17 2016 -0800

----------------------------------------------------------------------
 3rdparty/libprocess/src/libevent_poll.cpp | 40 +++++++++++++++++++++-----
 1 file changed, 33 insertions(+), 7 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/mesos/blob/859b4a3d/3rdparty/libprocess/src/libevent_poll.cpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/src/libevent_poll.cpp b/3rdparty/libprocess/src/libevent_poll.cpp
index 461624c..43a1abb 100644
--- a/3rdparty/libprocess/src/libevent_poll.cpp
+++ b/3rdparty/libprocess/src/libevent_poll.cpp
@@ -12,6 +12,8 @@
 
 #include <event2/event.h>
 
+#include <memory>
+
 #include <process/future.hpp>
 #include <process/io.hpp>
 #include <process/process.hpp> // For process::initialize.
@@ -26,7 +28,7 @@ namespace internal {
 struct Poll
 {
   Promise<short> promise;
-  event* ev;
+  std::shared_ptr<event> ev;
 };
 
 
@@ -45,14 +47,26 @@ void pollCallback(evutil_socket_t, short what, void* arg)
     poll->promise.set(events);
   }
 
-  event_free(poll->ev);
+  // Deleting the `poll` also destructs `ev` and hence triggers `event_free`,
+  // which makes the event non-pending.
   delete poll;
 }
 
 
-void pollDiscard(event* ev)
+void pollDiscard(const std::weak_ptr<event>& ev, short events)
 {
-  event_active(ev, EV_READ, 0);
+  // Discarding inside the event loop prevents `pollCallback()` from being
+  // called twice if the future is discarded.
+  run_in_event_loop([=]() {
+    std::shared_ptr<event> shared = ev.lock();
+    // If `ev` cannot be locked `pollCallback` already ran. If it was locked
+    // but not pending, `pollCallback` is scheduled to be executed.
+    if (static_cast<bool>(shared) &&
+        event_pending(shared.get(), events, NULL)) {
+      // `event_active` will trigger the `pollCallback` to be executed.
+      event_active(shared.get(), EV_READ, 0);
+    }
+  });
 }
 
 } // namespace internal {
@@ -71,15 +85,27 @@ Future<short> poll(int fd, short events)
   short what =
     ((events & io::READ) ? EV_READ : 0) | ((events & io::WRITE) ? EV_WRITE : 0);
 
-  poll->ev = event_new(base, fd, what, &internal::pollCallback, poll);
+  // Bind `event_free` to the destructor of the `ev` shared pointer
+  // guaranteeing that the event will be freed only once.
+  poll->ev.reset(
+      event_new(base, fd, what, &internal::pollCallback, poll),
+      event_free);
+
   if (poll->ev == NULL) {
     LOG(FATAL) << "Failed to poll, event_new";
   }
 
-  event_add(poll->ev, NULL);
+  // Using a `weak_ptr` prevents `ev` to become a dangling pointer if
+  // the returned future is discarded after the event is triggered.
+  // The `weak_ptr` needs to be created before `event_add` in case
+  // the event is ready and the callback is executed before creating
+  // `ev`.
+  std::weak_ptr<event> ev(poll->ev);
+
+  event_add(poll->ev.get(), NULL);
 
   return future
-    .onDiscard(lambda::bind(&internal::pollDiscard, poll->ev));
+    .onDiscard(lambda::bind(&internal::pollDiscard, ev, what));
 }
 
 } // namespace io {