[mesos] 01/02: Fixed an early fd close in the cgroups event notifier.

[qi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

qianzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/mesos.git

commit 610064942d4a75f16f045480ca9e3414d37f1ecc
Author: Benjamin Mahler <bm...@apache.org>
AuthorDate: Fri Oct 26 10:50:54 2018 +0800

    Fixed an early fd close in the cgroups event notifier.
    
    The cgroups event notifier was closing the eventfd while an
    `io::read()` operation may be in progress. This can lead to
    bugs where the fd gets re-used and read from a stale io::read.
    
    Review: https://reviews.apache.org/r/69123/
---
 src/linux/cgroups.cpp | 41 +++++++++++++++++++++++++++--------------
 1 file changed, 27 insertions(+), 14 deletions(-)

diff --git a/src/linux/cgroups.cpp b/src/linux/cgroups.cpp
index 95efc1d..45dd5cf 100644
--- a/src/linux/cgroups.cpp
+++ b/src/linux/cgroups.cpp
@@ -1047,7 +1047,7 @@ public:
       // uint64_t) from the event file, it indicates that an event has
       // occurred.
       reading = io::read(eventfd.get(), &data, sizeof(data));
-      reading.onAny(defer(self(), &Listener::_listen));
+      reading->onAny(defer(self(), &Listener::_listen, lambda::_1));
     }
 
     return promise.get()->future();
@@ -1069,14 +1069,23 @@ protected:
   void finalize() override
   {
     // Discard the nonblocking read.
-    reading.discard();
+    if (reading.isSome()) {
+      reading->discard();
+    }
 
-    // Unregister the eventfd if needed.
+    // Unregister the eventfd if needed. If there's a pending read,
+    // we must wait for it to finish.
     if (eventfd.isSome()) {
-      Try<Nothing> unregister = unregisterNotifier(eventfd.get());
-      if (unregister.isError()) {
-        LOG(ERROR) << "Failed to unregister eventfd: " << unregister.error();
-      }
+      int fd = eventfd.get();
+
+      reading.getOrElse(Future<size_t>(0))
+        .onAny([fd]() {
+          Try<Nothing> unregister = unregisterNotifier(fd);
+          if (unregister.isError()) {
+            LOG(ERROR) << "Failed to unregister eventfd '" << fd << "'"
+                       << ": " << unregister.error();
+          }
+      });
     }
 
     // TODO(chzhcn): Fail our promise only after 'reading' has
@@ -1089,11 +1098,15 @@ protected:
 private:
   // This function is called when the nonblocking read on the eventfd has
   // result, either because the event has happened, or an error has occurred.
-  void _listen()
+  void _listen(Future<size_t> read)
   {
     CHECK_SOME(promise);
+    CHECK_SOME(reading);
+
+    // Reset to none since we're no longer reading.
+    reading = None();
 
-    if (reading.isReady() && reading.get() == sizeof(data)) {
+    if (read.isReady() && read.get() == sizeof(data)) {
       promise.get()->set(data);
 
       // After fulfilling the promise, reset to get ready for the next one.
@@ -1101,14 +1114,14 @@ private:
       return;
     }
 
-    if (reading.isDiscarded()) {
+    if (read.isDiscarded()) {
       error = Error("Reading eventfd stopped unexpectedly");
-    } else if (reading.isFailed()) {
-      error = Error("Failed to read eventfd: " + reading.failure());
+    } else if (read.isFailed()) {
+      error = Error("Failed to read eventfd: " + read.failure());
     } else {
       error = Error("Read less than expected. Expect " +
                     stringify(sizeof(data)) + " bytes; actual " +
-                    stringify(reading.get()) + " bytes");
+                    stringify(read.get()) + " bytes");
     }
 
     // Inform failure and not listen again.
@@ -1121,7 +1134,7 @@ private:
   const Option<string> args;
 
   Option<Owned<Promise<uint64_t>>> promise;
-  Future<size_t> reading;
+  Option<Future<size_t>> reading;
   Option<Error> error;
   Option<int> eventfd;
   uint64_t data;                // The data read from the eventfd last time.