[3/3] mesos git commit: Added synchronization in link logic to prevent relinking races.

[jo...@apache.org]

Added synchronization in link logic to prevent relinking races.

There is a general pattern in the `SocketManager` in which most methods
will grab the mutex and then check if the socket to manage exists in
the `SocketManager`s mapping.  If the socket does not exist, the
`SocketManager` silently returns.

This adds similar logic inside two critical sections of the `link`
codepath.  If there are multiple calls to `link` in-flight at once,
this prevents sockets from being leaked into unmanaged callback loops.

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


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

Branch: refs/heads/master
Commit: 3b27c92f9b31d9b40c86f368ebd30c97d4fb69e8
Parents: 79c72d7
Author: Joseph Wu <jo...@mesosphere.io>
Authored: Wed Sep 28 12:33:47 2016 -0700
Committer: Joseph Wu <jo...@apache.org>
Committed: Wed Sep 28 12:33:47 2016 -0700

----------------------------------------------------------------------
 3rdparty/libprocess/src/process.cpp | 43 ++++++++++++++++++++++++--------
 1 file changed, 32 insertions(+), 11 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/mesos/blob/3b27c92f/3rdparty/libprocess/src/process.cpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/src/process.cpp b/3rdparty/libprocess/src/process.cpp
index c03fec6..f1d746c 100644
--- a/3rdparty/libprocess/src/process.cpp
+++ b/3rdparty/libprocess/src/process.cpp
@@ -1505,16 +1505,26 @@ void SocketManager::link_connect(
     return;
   }
 
-  size_t size = 80 * 1024;
-  char* data = new char[size];
+  synchronized (mutex) {
+    // It is possible that a prior call to `link()` with `RECONNECT`
+    // semantics has swapped out this socket before we finished
+    // connecting. In this case, we simply stop here and allow the
+    // latest created socket to complete the link.
+    if (sockets.count(socket) <= 0) {
+      return;
+    }
 
-  socket.recv(data, size)
-    .onAny(lambda::bind(
-        &internal::ignore_recv_data,
-        lambda::_1,
-        socket,
-        data,
-        size));
+    size_t size = 80 * 1024;
+    char* data = new char[size];
+
+    socket.recv(data, size)
+      .onAny(lambda::bind(
+          &internal::ignore_recv_data,
+          lambda::_1,
+          socket,
+          data,
+          size));
+  }
 
   // In order to avoid a race condition where internal::send() is
   // called after SocketManager::link() but before the socket is
@@ -1611,11 +1621,22 @@ void SocketManager::link(
 
         // Update all the data structures that are mapped to the old
         // socket. They will now point to the new socket we are about
-        // to try to connect. The old socket should no longer have any
-        // references after the swap and should be closed.
+        // to try to connect.
         Socket existing(sockets.at(persists.at(to.address)));
         swap_implementing_socket(existing, socket.get());
 
+        // The `existing` socket could be a perfectly functional socket.
+        // In this case, the socket may be referenced in the callback
+        // loop of `internal::ignore_recv_data`. We shutdown the socket
+        // in order to interrupt this callback loop and thereby release
+        // the final socket reference. This will not result in an
+        // `ExitedEvent` because we have already removed the `existing`
+        // socket from the mapping of linkees and linkers.
+        Try<Nothing> shutdown = existing.shutdown();
+        if (shutdown.isError()) {
+          VLOG(1) << "Failed to shutdown old link: " << shutdown.error();
+        }
+
         connect = true;
       }
     }