Use SOCK_SEQPACKET for synchronous IPC.

This is a copy of https://codereview.chromium.org/11738003 by mnissler@chromium.org

SOCK_DGRAM fails in case the other end of the connection dies before
sending a reply. This causes recvmsg() calls on the socket to hang,
which results in stuck processes sticking around after running tests.

BUG=chromium:166528
TEST=No more stuck --type=zygote processes in browser_tests and content_browsertests.
Review URL: https://codereview.chromium.org/11823024

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@177638 0039d316-1c4b-4281-b951-d872f2087c98

5 files changed, 108 insertions, 25 deletions

diff --git a/base/base.gyp b/base/base.gyp
index 67b122b..64a580f 100644
--- a/base/base.gyp
+++ b/base/base.gyp
@@ -534,6 +534,7 @@
         'pickle_unittest.cc',
         'platform_file_unittest.cc',
         'posix/file_descriptor_shuffle_unittest.cc',
+        'posix/unix_domain_socket_linux_unittest.cc',
         'pr_time_unittest.cc',
         'prefs/overlay_user_pref_store_unittest.cc',
         'prefs/pref_value_map_unittest.cc',
diff --git a/base/base.gypi b/base/base.gypi
index 42a1ffa..ac8c913 100644
--- a/base/base.gypi
+++ b/base/base.gypi
@@ -327,8 +327,8 @@
           'posix/eintr_wrapper.h',
           'posix/global_descriptors.cc',
           'posix/global_descriptors.h',
-          'posix/unix_domain_socket.cc',
-          'posix/unix_domain_socket.h',
+          'posix/unix_domain_socket_linux.cc',
+          'posix/unix_domain_socket_linux.h',
           'process.h',
           'process_info.h',
           'process_info_mac.cc',
@@ -605,7 +605,7 @@
                'native_library_posix.cc',
                'path_service.cc',
                'platform_file_posix.cc',
-               'posix/unix_domain_socket.cc',
+               'posix/unix_domain_socket_linux.cc',
                'process_posix.cc',
                'process_util.cc',
                'process_util_posix.cc',
@@ -630,6 +630,7 @@
             'sources/': [
               ['include', '^files/file_path_watcher_linux\\.cc$'],
               ['include', '^process_util_linux\\.cc$'],
+              ['include', '^posix/unix_domain_socket_linux\\.cc$'],
               ['include', '^sys_info_linux\\.cc$'],
               ['include', '^sys_string_conversions_posix\\.cc$'],
               ['include', '^worker_pool_linux\\.cc$'],
diff --git a/base/posix/unix_domain_socket.cc b/base/posix/unix_domain_socket_linux.cc
index 5bfcef1..603459f 100644
--- a/base/posix/unix_domain_socket.cc
+++ b/base/posix/unix_domain_socket_linux.cc
@@ -2,12 +2,12 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
-#include "base/posix/unix_domain_socket.h"
+#include "base/posix/unix_domain_socket_linux.h"
 
 #include <errno.h>
-#include <unistd.h>
-#include <sys/uio.h>
 #include <sys/socket.h>
+#include <sys/uio.h>
+#include <unistd.h>
 
 #include "base/logging.h"
 #include "base/pickle.h"
@@ -21,9 +21,8 @@ bool UnixDomainSocket::SendMsg(int fd,
                                const void* buf,
                                size_t length,
                                const std::vector<int>& fds) {
-  struct msghdr msg;
-  memset(&msg, 0, sizeof(msg));
-  struct iovec iov = {const_cast<void*>(buf), length};
+  struct msghdr msg = {};
+  struct iovec iov = { const_cast<void*>(buf), length };
   msg.msg_iov = &iov;
   msg.msg_iovlen = 1;
 
@@ -32,7 +31,7 @@ bool UnixDomainSocket::SendMsg(int fd,
     const unsigned control_len = CMSG_SPACE(sizeof(int) * fds.size());
     control_buffer = new char[control_len];
 
-    struct cmsghdr *cmsg;
+    struct cmsghdr* cmsg;
     msg.msg_control = control_buffer;
     msg.msg_controllen = control_len;
     cmsg = CMSG_FIRSTHDR(&msg);
@@ -43,13 +42,11 @@ bool UnixDomainSocket::SendMsg(int fd,
     msg.msg_controllen = cmsg->cmsg_len;
   }
 
-  // When available, take advantage of MSG_NOSIGNAL to avoid
-  // a SIGPIPE if the other end breaks the connection.
-#if defined(MSG_NOSIGNAL)
+  // Avoid a SIGPIPE if the other end breaks the connection.
+  // Due to a bug in the Linux kernel (net/unix/af_unix.c) MSG_NOSIGNAL isn't
+  // regarded for SOCK_SEQPACKET in the AF_UNIX domain, but it is mandated by
+  // POSIX.
   const int flags = MSG_NOSIGNAL;
-#else
-  const int flags = 0;
-#endif
   const ssize_t r = HANDLE_EINTR(sendmsg(fd, &msg, flags));
   const bool ret = static_cast<ssize_t>(length) == r;
   delete[] control_buffer;
@@ -63,9 +60,8 @@ ssize_t UnixDomainSocket::RecvMsg(int fd,
                                   std::vector<int>* fds) {
   fds->clear();
 
-  struct msghdr msg;
-  memset(&msg, 0, sizeof(msg));
-  struct iovec iov = {buf, length};
+  struct msghdr msg = {};
+  struct iovec iov = { buf, length };
   msg.msg_iov = &iov;
   msg.msg_iovlen = 1;
 
@@ -117,7 +113,7 @@ ssize_t UnixDomainSocket::SendRecvMsg(int fd,
 
   // This socketpair is only used for the IPC and is cleaned up before
   // returning.
-  if (socketpair(AF_UNIX, SOCK_DGRAM, 0, fds) == -1)
+  if (socketpair(AF_UNIX, SOCK_SEQPACKET, 0, fds) == -1)
     return -1;
 
   std::vector<int> fd_vector;
@@ -130,6 +126,8 @@ ssize_t UnixDomainSocket::SendRecvMsg(int fd,
   close(fds[1]);
 
   fd_vector.clear();
+  // When porting to OSX keep in mind it doesn't support MSG_NOSIGNAL, so the
+  // sender might get a SIGPIPE.
   const ssize_t reply_len = RecvMsg(fds[0], reply, max_reply_len, &fd_vector);
   close(fds[0]);
   if (reply_len == -1)
diff --git a/base/posix/unix_domain_socket.h b/base/posix/unix_domain_socket_linux.h
index cb2a0b8..b8ba460 100644
--- a/base/posix/unix_domain_socket.h
+++ b/base/posix/unix_domain_socket_linux.h
@@ -2,8 +2,8 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
-#ifndef BASE_POSIX_UNIX_DOMAIN_SOCKET_H_
-#define BASE_POSIX_UNIX_DOMAIN_SOCKET_H_
+#ifndef BASE_POSIX_UNIX_DOMAIN_SOCKET_LINUX_H_
+#define BASE_POSIX_UNIX_DOMAIN_SOCKET_LINUX_H_
 
 #include <stdint.h>
 #include <sys/types.h>
@@ -33,8 +33,10 @@ class BASE_EXPORT UnixDomainSocket {
                          std::vector<int>* fds);
 
   // Perform a sendmsg/recvmsg pair.
-  //   1. This process creates a UNIX DGRAM socketpair.
-  //   2. This proces writes a request to |fd| with an SCM_RIGHTS control
+  //   1. This process creates a UNIX SEQPACKET socketpair. Using
+  //      connection-oriented sockets (SEQPACKET or STREAM) is critical here,
+  //      because if one of the ends closes the other one must be notified.
+  //   2. This process writes a request to |fd| with an SCM_RIGHTS control
   //      message containing on end of the fresh socket pair.
   //   3. This process blocks reading from the other end of the fresh
   //      socketpair.
@@ -55,4 +57,4 @@ class BASE_EXPORT UnixDomainSocket {
                              const Pickle& request);
 };
 
-#endif  // BASE_POSIX_UNIX_DOMAIN_SOCKET_POSIX_H_
+#endif  // BASE_POSIX_UNIX_DOMAIN_SOCKET_LINUX_H_
diff --git a/base/posix/unix_domain_socket_linux_unittest.cc b/base/posix/unix_domain_socket_linux_unittest.cc
new file mode 100644
index 0000000..1343555
--- /dev/null
+++ b/base/posix/unix_domain_socket_linux_unittest.cc
@@ -0,0 +1,81 @@
+// Copyright (c) 2013 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include <sys/socket.h>
+#include <sys/types.h>
+#include <unistd.h>
+
+#include "base/bind.h"
+#include "base/bind_helpers.h"
+#include "base/file_util.h"
+#include "base/pickle.h"
+#include "base/posix/unix_domain_socket_linux.h"
+#include "base/synchronization/waitable_event.h"
+#include "base/threading/thread.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace base {
+
+namespace {
+
+TEST(UnixDomainSocketTest, SendRecvMsgAbortOnReplyFDClose) {
+  Thread message_thread("UnixDomainSocketTest");
+  ASSERT_TRUE(message_thread.Start());
+
+  int fds[2];
+  ASSERT_EQ(0, socketpair(AF_UNIX, SOCK_SEQPACKET, 0, fds));
+  file_util::ScopedFD scoped_fd0(&fds[0]);
+  file_util::ScopedFD scoped_fd1(&fds[1]);
+
+  // Have the thread send a synchronous message via the socket.
+  Pickle request;
+  message_thread.message_loop()->PostTask(
+      FROM_HERE,
+      Bind(IgnoreResult(&UnixDomainSocket::SendRecvMsg),
+           fds[1], static_cast<uint8_t*>(NULL), 0U, static_cast<int*>(NULL),
+           request));
+
+  // Receive the message.
+  std::vector<int> message_fds;
+  uint8_t buffer[16];
+  ASSERT_EQ(static_cast<int>(request.size()),
+            UnixDomainSocket::RecvMsg(fds[0], buffer, sizeof(buffer),
+                                      &message_fds));
+  ASSERT_EQ(1U, message_fds.size());
+
+  // Close the reply FD.
+  ASSERT_EQ(0, HANDLE_EINTR(close(message_fds.front())));
+
+  // Check that the thread didn't get blocked.
+  WaitableEvent event(false, false);
+  message_thread.message_loop()->PostTask(
+      FROM_HERE,
+      Bind(&WaitableEvent::Signal, Unretained(&event)));
+  ASSERT_TRUE(event.TimedWait(TimeDelta::FromMilliseconds(5000)));
+}
+
+TEST(UnixDomainSocketTest, SendRecvMsgAvoidsSIGPIPE) {
+  // Make sure SIGPIPE isn't being ignored.
+  struct sigaction act = {}, oldact;
+  act.sa_handler = SIG_DFL;
+  ASSERT_EQ(0, sigaction(SIGPIPE, &act, &oldact));
+  int fds[2];
+  ASSERT_EQ(0, socketpair(AF_UNIX, SOCK_SEQPACKET, 0, fds));
+  file_util::ScopedFD scoped_fd1(&fds[1]);
+  ASSERT_EQ(0, HANDLE_EINTR(close(fds[0])));
+
+  // Have the thread send a synchronous message via the socket. Unless the
+  // message is sent with MSG_NOSIGNAL, this shall result in SIGPIPE.
+  Pickle request;
+  ASSERT_EQ(-1,
+      UnixDomainSocket::SendRecvMsg(fds[1], static_cast<uint8_t*>(NULL),
+                                    0U, static_cast<int*>(NULL), request));
+  ASSERT_EQ(EPIPE, errno);
+  // Restore the SIGPIPE handler.
+  ASSERT_EQ(0, sigaction(SIGPIPE, &oldact, NULL));
+}
+
+}  // namespace
+
+}  // namespace base