michael@0: // Copyright (c) 2008 The Chromium Authors. All rights reserved.
michael@0: // Use of this source code is governed by a BSD-style license that can be
michael@0: // found in the LICENSE file.
michael@0: 
michael@0: #include "chrome/common/ipc_channel_posix.h"
michael@0: 
michael@0: #include <errno.h>
michael@0: #include <fcntl.h>
michael@0: #include <stddef.h>
michael@0: #include <unistd.h>
michael@0: #include <sys/types.h>
michael@0: #include <sys/socket.h>
michael@0: #include <sys/stat.h>
michael@0: #include <sys/un.h>
michael@0: #include <sys/uio.h>
michael@0: 
michael@0: #include <string>
michael@0: #include <map>
michael@0: 
michael@0: #include "base/command_line.h"
michael@0: #include "base/eintr_wrapper.h"
michael@0: #include "base/lock.h"
michael@0: #include "base/logging.h"
michael@0: #include "base/process_util.h"
michael@0: #include "base/scoped_ptr.h"
michael@0: #include "base/string_util.h"
michael@0: #include "base/singleton.h"
michael@0: #include "base/stats_counters.h"
michael@0: #include "chrome/common/chrome_switches.h"
michael@0: #include "chrome/common/file_descriptor_set_posix.h"
michael@0: #include "chrome/common/ipc_logging.h"
michael@0: #include "chrome/common/ipc_message_utils.h"
michael@0: #include "mozilla/ipc/ProtocolUtils.h"
michael@0: 
michael@0: #ifdef MOZ_TASK_TRACER
michael@0: #include "GeckoTaskTracerImpl.h"
michael@0: using namespace mozilla::tasktracer;
michael@0: #endif
michael@0: 
michael@0: namespace IPC {
michael@0: 
michael@0: // IPC channels on Windows use named pipes (CreateNamedPipe()) with
michael@0: // channel ids as the pipe names.  Channels on POSIX use anonymous
michael@0: // Unix domain sockets created via socketpair() as pipes.  These don't
michael@0: // quite line up.
michael@0: //
michael@0: // When creating a child subprocess, the parent side of the fork
michael@0: // arranges it such that the initial control channel ends up on the
michael@0: // magic file descriptor kClientChannelFd in the child.  Future
michael@0: // connections (file descriptors) can then be passed via that
michael@0: // connection via sendmsg().
michael@0: 
michael@0: //------------------------------------------------------------------------------
michael@0: namespace {
michael@0: 
michael@0: // The PipeMap class works around this quirk related to unit tests:
michael@0: //
michael@0: // When running as a server, we install the client socket in a
michael@0: // specific file descriptor number (@kClientChannelFd). However, we
michael@0: // also have to support the case where we are running unittests in the
michael@0: // same process.  (We do not support forking without execing.)
michael@0: //
michael@0: // Case 1: normal running
michael@0: //   The IPC server object will install a mapping in PipeMap from the
michael@0: //   name which it was given to the client pipe. When forking the client, the
michael@0: //   GetClientFileDescriptorMapping will ensure that the socket is installed in
michael@0: //   the magic slot (@kClientChannelFd). The client will search for the
michael@0: //   mapping, but it won't find any since we are in a new process. Thus the
michael@0: //   magic fd number is returned. Once the client connects, the server will
michael@0: //   close its copy of the client socket and remove the mapping.
michael@0: //
michael@0: // Case 2: unittests - client and server in the same process
michael@0: //   The IPC server will install a mapping as before. The client will search
michael@0: //   for a mapping and find out. It duplicates the file descriptor and
michael@0: //   connects. Once the client connects, the server will close the original
michael@0: //   copy of the client socket and remove the mapping. Thus, when the client
michael@0: //   object closes, it will close the only remaining copy of the client socket
michael@0: //   in the fd table and the server will see EOF on its side.
michael@0: //
michael@0: // TODO(port): a client process cannot connect to multiple IPC channels with
michael@0: // this scheme.
michael@0: 
michael@0: class PipeMap {
michael@0:  public:
michael@0:   // Lookup a given channel id. Return -1 if not found.
michael@0:   int Lookup(const std::string& channel_id) {
michael@0:     AutoLock locked(lock_);
michael@0: 
michael@0:     ChannelToFDMap::const_iterator i = map_.find(channel_id);
michael@0:     if (i == map_.end())
michael@0:       return -1;
michael@0:     return i->second;
michael@0:   }
michael@0: 
michael@0:   // Remove the mapping for the given channel id. No error is signaled if the
michael@0:   // channel_id doesn't exist
michael@0:   void Remove(const std::string& channel_id) {
michael@0:     AutoLock locked(lock_);
michael@0: 
michael@0:     ChannelToFDMap::iterator i = map_.find(channel_id);
michael@0:     if (i != map_.end())
michael@0:       map_.erase(i);
michael@0:   }
michael@0: 
michael@0:   // Insert a mapping from @channel_id to @fd. It's a fatal error to insert a
michael@0:   // mapping if one already exists for the given channel_id
michael@0:   void Insert(const std::string& channel_id, int fd) {
michael@0:     AutoLock locked(lock_);
michael@0:     DCHECK(fd != -1);
michael@0: 
michael@0:     ChannelToFDMap::const_iterator i = map_.find(channel_id);
michael@0:     CHECK(i == map_.end()) << "Creating second IPC server for '"
michael@0:                            << channel_id
michael@0:                            << "' while first still exists";
michael@0:     map_[channel_id] = fd;
michael@0:   }
michael@0: 
michael@0:  private:
michael@0:   Lock lock_;
michael@0:   typedef std::map<std::string, int> ChannelToFDMap;
michael@0:   ChannelToFDMap map_;
michael@0: };
michael@0: 
michael@0: // This is the file descriptor number that a client process expects to find its
michael@0: // IPC socket.
michael@0: static const int kClientChannelFd = 3;
michael@0: 
michael@0: // Used to map a channel name to the equivalent FD # in the client process.
michael@0: int ChannelNameToClientFD(const std::string& channel_id) {
michael@0:   // See the large block comment above PipeMap for the reasoning here.
michael@0:   const int fd = Singleton<PipeMap>()->Lookup(channel_id);
michael@0:   if (fd != -1)
michael@0:     return dup(fd);
michael@0: 
michael@0:   // If we don't find an entry, we assume that the correct value has been
michael@0:   // inserted in the magic slot.
michael@0:   return kClientChannelFd;
michael@0: }
michael@0: 
michael@0: //------------------------------------------------------------------------------
michael@0: const size_t kMaxPipeNameLength = sizeof(((sockaddr_un*)0)->sun_path);
michael@0: 
michael@0: // Creates a Fifo with the specified name ready to listen on.
michael@0: bool CreateServerFifo(const std::string& pipe_name, int* server_listen_fd) {
michael@0:   DCHECK(server_listen_fd);
michael@0:   DCHECK_GT(pipe_name.length(), 0u);
michael@0:   DCHECK_LT(pipe_name.length(), kMaxPipeNameLength);
michael@0: 
michael@0:   if (pipe_name.length() == 0 || pipe_name.length() >= kMaxPipeNameLength) {
michael@0:     return false;
michael@0:   }
michael@0: 
michael@0:   // Create socket.
michael@0:   int fd = socket(AF_UNIX, SOCK_STREAM, 0);
michael@0:   if (fd < 0) {
michael@0:     return false;
michael@0:   }
michael@0: 
michael@0:   // Make socket non-blocking
michael@0:   if (fcntl(fd, F_SETFL, O_NONBLOCK) == -1) {
michael@0:     HANDLE_EINTR(close(fd));
michael@0:     return false;
michael@0:   }
michael@0: 
michael@0:   // Delete any old FS instances.
michael@0:   unlink(pipe_name.c_str());
michael@0: 
michael@0:   // Create unix_addr structure
michael@0:   struct sockaddr_un unix_addr;
michael@0:   memset(&unix_addr, 0, sizeof(unix_addr));
michael@0:   unix_addr.sun_family = AF_UNIX;
michael@0:   snprintf(unix_addr.sun_path, kMaxPipeNameLength, "%s", pipe_name.c_str());
michael@0:   size_t unix_addr_len = offsetof(struct sockaddr_un, sun_path) +
michael@0:       strlen(unix_addr.sun_path) + 1;
michael@0: 
michael@0:   // Bind the socket.
michael@0:   if (bind(fd, reinterpret_cast<const sockaddr*>(&unix_addr),
michael@0:            unix_addr_len) != 0) {
michael@0:     HANDLE_EINTR(close(fd));
michael@0:     return false;
michael@0:   }
michael@0: 
michael@0:   // Start listening on the socket.
michael@0:   const int listen_queue_length = 1;
michael@0:   if (listen(fd, listen_queue_length) != 0) {
michael@0:     HANDLE_EINTR(close(fd));
michael@0:     return false;
michael@0:   }
michael@0: 
michael@0:   *server_listen_fd = fd;
michael@0:   return true;
michael@0: }
michael@0: 
michael@0: // Accept a connection on a fifo.
michael@0: bool ServerAcceptFifoConnection(int server_listen_fd, int* server_socket) {
michael@0:   DCHECK(server_socket);
michael@0: 
michael@0:   int accept_fd = HANDLE_EINTR(accept(server_listen_fd, NULL, 0));
michael@0:   if (accept_fd < 0)
michael@0:     return false;
michael@0:   if (fcntl(accept_fd, F_SETFL, O_NONBLOCK) == -1) {
michael@0:     HANDLE_EINTR(close(accept_fd));
michael@0:     return false;
michael@0:   }
michael@0: 
michael@0:   *server_socket = accept_fd;
michael@0:   return true;
michael@0: }
michael@0: 
michael@0: bool ClientConnectToFifo(const std::string &pipe_name, int* client_socket) {
michael@0:   DCHECK(client_socket);
michael@0:   DCHECK_LT(pipe_name.length(), kMaxPipeNameLength);
michael@0: 
michael@0:   // Create socket.
michael@0:   int fd = socket(AF_UNIX, SOCK_STREAM, 0);
michael@0:   if (fd < 0) {
michael@0:     CHROMIUM_LOG(ERROR) << "fd is invalid";
michael@0:     return false;
michael@0:   }
michael@0: 
michael@0:   // Make socket non-blocking
michael@0:   if (fcntl(fd, F_SETFL, O_NONBLOCK) == -1) {
michael@0:     CHROMIUM_LOG(ERROR) << "fcntl failed";
michael@0:     HANDLE_EINTR(close(fd));
michael@0:     return false;
michael@0:   }
michael@0: 
michael@0:   // Create server side of socket.
michael@0:   struct sockaddr_un  server_unix_addr;
michael@0:   memset(&server_unix_addr, 0, sizeof(server_unix_addr));
michael@0:   server_unix_addr.sun_family = AF_UNIX;
michael@0:   snprintf(server_unix_addr.sun_path, kMaxPipeNameLength, "%s",
michael@0:            pipe_name.c_str());
michael@0:   size_t server_unix_addr_len = offsetof(struct sockaddr_un, sun_path) +
michael@0:       strlen(server_unix_addr.sun_path) + 1;
michael@0: 
michael@0:   if (HANDLE_EINTR(connect(fd, reinterpret_cast<sockaddr*>(&server_unix_addr),
michael@0:                            server_unix_addr_len)) != 0) {
michael@0:     HANDLE_EINTR(close(fd));
michael@0:     return false;
michael@0:   }
michael@0: 
michael@0:   *client_socket = fd;
michael@0:   return true;
michael@0: }
michael@0: 
michael@0: bool SetCloseOnExec(int fd) {
michael@0:   int flags = fcntl(fd, F_GETFD);
michael@0:   if (flags == -1)
michael@0:     return false;
michael@0: 
michael@0:   flags |= FD_CLOEXEC;
michael@0:   if (fcntl(fd, F_SETFD, flags) == -1)
michael@0:     return false;
michael@0: 
michael@0:   return true;
michael@0: }
michael@0: 
michael@0: }  // namespace
michael@0: //------------------------------------------------------------------------------
michael@0: 
michael@0: Channel::ChannelImpl::ChannelImpl(const std::wstring& channel_id, Mode mode,
michael@0:                                   Listener* listener)
michael@0:     : factory_(this) {
michael@0:   Init(mode, listener);
michael@0:   uses_fifo_ = CommandLine::ForCurrentProcess()->HasSwitch(switches::kIPCUseFIFO);
michael@0: 
michael@0:   if (!CreatePipe(channel_id, mode)) {
michael@0:     // The pipe may have been closed already.
michael@0:     CHROMIUM_LOG(WARNING) << "Unable to create pipe named \"" << channel_id <<
michael@0:                              "\" in " << (mode == MODE_SERVER ? "server" : "client") <<
michael@0:                              " mode error(" << strerror(errno) << ").";
michael@0:   }
michael@0: }
michael@0: 
michael@0: Channel::ChannelImpl::ChannelImpl(int fd, Mode mode, Listener* listener)
michael@0:     : factory_(this) {
michael@0:   Init(mode, listener);
michael@0:   pipe_ = fd;
michael@0:   waiting_connect_ = (MODE_SERVER == mode);
michael@0: 
michael@0:   EnqueueHelloMessage();
michael@0: }
michael@0: 
michael@0: void Channel::ChannelImpl::Init(Mode mode, Listener* listener) {
michael@0:   mode_ = mode;
michael@0:   is_blocked_on_write_ = false;
michael@0:   message_send_bytes_written_ = 0;
michael@0:   uses_fifo_ = false;
michael@0:   server_listen_pipe_ = -1;
michael@0:   pipe_ = -1;
michael@0:   client_pipe_ = -1;
michael@0:   listener_ = listener;
michael@0:   waiting_connect_ = true;
michael@0:   processing_incoming_ = false;
michael@0:   closed_ = false;
michael@0: #if defined(OS_MACOSX)
michael@0:   last_pending_fd_id_ = 0;
michael@0: #endif
michael@0:   output_queue_length_ = 0;
michael@0: }
michael@0: 
michael@0: bool Channel::ChannelImpl::CreatePipe(const std::wstring& channel_id,
michael@0:                                       Mode mode) {
michael@0:   DCHECK(server_listen_pipe_ == -1 && pipe_ == -1);
michael@0: 
michael@0:   if (uses_fifo_) {
michael@0:     // This only happens in unit tests; see the comment above PipeMap.
michael@0:     // TODO(playmobil): We shouldn't need to create fifos on disk.
michael@0:     // TODO(playmobil): If we do, they should be in the user data directory.
michael@0:     // TODO(playmobil): Cleanup any stale fifos.
michael@0:     pipe_name_ = "/var/tmp/chrome_" + WideToASCII(channel_id);
michael@0:     if (mode == MODE_SERVER) {
michael@0:       if (!CreateServerFifo(pipe_name_, &server_listen_pipe_)) {
michael@0:         return false;
michael@0:       }
michael@0:     } else {
michael@0:       if (!ClientConnectToFifo(pipe_name_, &pipe_)) {
michael@0:         return false;
michael@0:       }
michael@0:       waiting_connect_ = false;
michael@0:     }
michael@0:   } else {
michael@0:     // socketpair()
michael@0:     pipe_name_ = WideToASCII(channel_id);
michael@0:     if (mode == MODE_SERVER) {
michael@0:       int pipe_fds[2];
michael@0:       if (socketpair(AF_UNIX, SOCK_STREAM, 0, pipe_fds) != 0) {
michael@0:         return false;
michael@0:       }
michael@0:       // Set both ends to be non-blocking.
michael@0:       if (fcntl(pipe_fds[0], F_SETFL, O_NONBLOCK) == -1 ||
michael@0:           fcntl(pipe_fds[1], F_SETFL, O_NONBLOCK) == -1) {
michael@0:         HANDLE_EINTR(close(pipe_fds[0]));
michael@0:         HANDLE_EINTR(close(pipe_fds[1]));
michael@0:         return false;
michael@0:       }
michael@0: 
michael@0:       if (!SetCloseOnExec(pipe_fds[0]) ||
michael@0:           !SetCloseOnExec(pipe_fds[1])) {
michael@0:         HANDLE_EINTR(close(pipe_fds[0]));
michael@0:         HANDLE_EINTR(close(pipe_fds[1]));
michael@0:         return false;
michael@0:       }
michael@0: 
michael@0:       pipe_ = pipe_fds[0];
michael@0:       client_pipe_ = pipe_fds[1];
michael@0: 
michael@0:       if (pipe_name_.length()) {
michael@0:         Singleton<PipeMap>()->Insert(pipe_name_, client_pipe_);
michael@0:       }
michael@0:     } else {
michael@0:       pipe_ = ChannelNameToClientFD(pipe_name_);
michael@0:       DCHECK(pipe_ > 0);
michael@0:       waiting_connect_ = false;
michael@0:     }
michael@0:   }
michael@0: 
michael@0:   // Create the Hello message to be sent when Connect is called
michael@0:   return EnqueueHelloMessage();
michael@0: }
michael@0: 
michael@0: /**
michael@0:  * Reset the file descriptor for communication with the peer.
michael@0:  */
michael@0: void Channel::ChannelImpl::ResetFileDescriptor(int fd) {
michael@0:   NS_ASSERTION(fd > 0 && fd == pipe_, "Invalid file descriptor");
michael@0: 
michael@0:   EnqueueHelloMessage();
michael@0: }
michael@0: 
michael@0: bool Channel::ChannelImpl::EnqueueHelloMessage() {
michael@0:   scoped_ptr<Message> msg(new Message(MSG_ROUTING_NONE,
michael@0:                                       HELLO_MESSAGE_TYPE,
michael@0:                                       IPC::Message::PRIORITY_NORMAL));
michael@0:   if (!msg->WriteInt(base::GetCurrentProcId())) {
michael@0:     Close();
michael@0:     return false;
michael@0:   }
michael@0: 
michael@0:   OutputQueuePush(msg.release());
michael@0:   return true;
michael@0: }
michael@0: 
michael@0: static void
michael@0: ClearAndShrink(std::string& s, size_t capacity)
michael@0: {
michael@0:   // This swap trick is the closest thing C++ has to a guaranteed way to
michael@0:   // shrink the capacity of a string.
michael@0:   std::string tmp;
michael@0:   tmp.reserve(capacity);
michael@0:   s.swap(tmp);
michael@0: }
michael@0: 
michael@0: bool Channel::ChannelImpl::Connect() {
michael@0:   if (mode_ == MODE_SERVER && uses_fifo_) {
michael@0:     if (server_listen_pipe_ == -1) {
michael@0:       return false;
michael@0:     }
michael@0:     MessageLoopForIO::current()->WatchFileDescriptor(
michael@0:         server_listen_pipe_,
michael@0:         true,
michael@0:         MessageLoopForIO::WATCH_READ,
michael@0:         &server_listen_connection_watcher_,
michael@0:         this);
michael@0:   } else {
michael@0:     if (pipe_ == -1) {
michael@0:       return false;
michael@0:     }
michael@0:     MessageLoopForIO::current()->WatchFileDescriptor(
michael@0:         pipe_,
michael@0:         true,
michael@0:         MessageLoopForIO::WATCH_READ,
michael@0:         &read_watcher_,
michael@0:         this);
michael@0:     waiting_connect_ = false;
michael@0:   }
michael@0: 
michael@0:   if (!waiting_connect_)
michael@0:     return ProcessOutgoingMessages();
michael@0:   return true;
michael@0: }
michael@0: 
michael@0: bool Channel::ChannelImpl::ProcessIncomingMessages() {
michael@0:   ssize_t bytes_read = 0;
michael@0: 
michael@0:   struct msghdr msg = {0};
michael@0:   struct iovec iov = {input_buf_, Channel::kReadBufferSize};
michael@0: 
michael@0:   msg.msg_iov = &iov;
michael@0:   msg.msg_iovlen = 1;
michael@0:   msg.msg_control = input_cmsg_buf_;
michael@0: 
michael@0:   for (;;) {
michael@0:     msg.msg_controllen = sizeof(input_cmsg_buf_);
michael@0: 
michael@0:     if (bytes_read == 0) {
michael@0:       if (pipe_ == -1)
michael@0:         return false;
michael@0: 
michael@0:       // Read from pipe.
michael@0:       // recvmsg() returns 0 if the connection has closed or EAGAIN if no data
michael@0:       // is waiting on the pipe.
michael@0:       bytes_read = HANDLE_EINTR(recvmsg(pipe_, &msg, MSG_DONTWAIT));
michael@0: 
michael@0:       if (bytes_read < 0) {
michael@0:         if (errno == EAGAIN) {
michael@0:           return true;
michael@0:         } else {
michael@0:           CHROMIUM_LOG(ERROR) << "pipe error (" << pipe_ << "): " << strerror(errno);
michael@0:           return false;
michael@0:         }
michael@0:       } else if (bytes_read == 0) {
michael@0:         // The pipe has closed...
michael@0:         Close();
michael@0:         return false;
michael@0:       }
michael@0:     }
michael@0:     DCHECK(bytes_read);
michael@0: 
michael@0:     if (client_pipe_ != -1) {
michael@0:       Singleton<PipeMap>()->Remove(pipe_name_);
michael@0:       HANDLE_EINTR(close(client_pipe_));
michael@0:       client_pipe_ = -1;
michael@0:     }
michael@0: 
michael@0:     // a pointer to an array of |num_wire_fds| file descriptors from the read
michael@0:     const int* wire_fds = NULL;
michael@0:     unsigned num_wire_fds = 0;
michael@0: 
michael@0:     // walk the list of control messages and, if we find an array of file
michael@0:     // descriptors, save a pointer to the array
michael@0: 
michael@0:     // This next if statement is to work around an OSX issue where
michael@0:     // CMSG_FIRSTHDR will return non-NULL in the case that controllen == 0.
michael@0:     // Here's a test case:
michael@0:     //
michael@0:     // int main() {
michael@0:     // struct msghdr msg;
michael@0:     //   msg.msg_control = &msg;
michael@0:     //   msg.msg_controllen = 0;
michael@0:     //   if (CMSG_FIRSTHDR(&msg))
michael@0:     //     printf("Bug found!\n");
michael@0:     // }
michael@0:     if (msg.msg_controllen > 0) {
michael@0:       // On OSX, CMSG_FIRSTHDR doesn't handle the case where controllen is 0
michael@0:       // and will return a pointer into nowhere.
michael@0:       for (struct cmsghdr* cmsg = CMSG_FIRSTHDR(&msg); cmsg;
michael@0:            cmsg = CMSG_NXTHDR(&msg, cmsg)) {
michael@0:         if (cmsg->cmsg_level == SOL_SOCKET &&
michael@0:             cmsg->cmsg_type == SCM_RIGHTS) {
michael@0:           const unsigned payload_len = cmsg->cmsg_len - CMSG_LEN(0);
michael@0:           DCHECK(payload_len % sizeof(int) == 0);
michael@0:           wire_fds = reinterpret_cast<int*>(CMSG_DATA(cmsg));
michael@0:           num_wire_fds = payload_len / 4;
michael@0: 
michael@0:           if (msg.msg_flags & MSG_CTRUNC) {
michael@0:             CHROMIUM_LOG(ERROR) << "SCM_RIGHTS message was truncated"
michael@0:                                 << " cmsg_len:" << cmsg->cmsg_len
michael@0:                                 << " fd:" << pipe_;
michael@0:             for (unsigned i = 0; i < num_wire_fds; ++i)
michael@0:               HANDLE_EINTR(close(wire_fds[i]));
michael@0:             return false;
michael@0:           }
michael@0:           break;
michael@0:         }
michael@0:       }
michael@0:     }
michael@0: 
michael@0:     // Process messages from input buffer.
michael@0:     const char *p;
michael@0:     const char *overflowp;
michael@0:     const char *end;
michael@0:     if (input_overflow_buf_.empty()) {
michael@0:       overflowp = NULL;
michael@0:       p = input_buf_;
michael@0:       end = p + bytes_read;
michael@0:     } else {
michael@0:       if (input_overflow_buf_.size() >
michael@0:          static_cast<size_t>(kMaximumMessageSize - bytes_read)) {
michael@0:         ClearAndShrink(input_overflow_buf_, Channel::kReadBufferSize);
michael@0:         CHROMIUM_LOG(ERROR) << "IPC message is too big";
michael@0:         return false;
michael@0:       }
michael@0:       input_overflow_buf_.append(input_buf_, bytes_read);
michael@0:       overflowp = p = input_overflow_buf_.data();
michael@0:       end = p + input_overflow_buf_.size();
michael@0:     }
michael@0: 
michael@0:     // A pointer to an array of |num_fds| file descriptors which includes any
michael@0:     // fds that have spilled over from a previous read.
michael@0:     const int* fds;
michael@0:     unsigned num_fds;
michael@0:     unsigned fds_i = 0;  // the index of the first unused descriptor
michael@0: 
michael@0:     if (input_overflow_fds_.empty()) {
michael@0:       fds = wire_fds;
michael@0:       num_fds = num_wire_fds;
michael@0:     } else {
michael@0:       const size_t prev_size = input_overflow_fds_.size();
michael@0:       input_overflow_fds_.resize(prev_size + num_wire_fds);
michael@0:       memcpy(&input_overflow_fds_[prev_size], wire_fds,
michael@0:              num_wire_fds * sizeof(int));
michael@0:       fds = &input_overflow_fds_[0];
michael@0:       num_fds = input_overflow_fds_.size();
michael@0:     }
michael@0: 
michael@0:     while (p < end) {
michael@0:       const char* message_tail = Message::FindNext(p, end);
michael@0:       if (message_tail) {
michael@0:         int len = static_cast<int>(message_tail - p);
michael@0:         Message m(p, len);
michael@0:         if (m.header()->num_fds) {
michael@0:           // the message has file descriptors
michael@0:           const char* error = NULL;
michael@0:           if (m.header()->num_fds > num_fds - fds_i) {
michael@0:             // the message has been completely received, but we didn't get
michael@0:             // enough file descriptors.
michael@0:             error = "Message needs unreceived descriptors";
michael@0:           }
michael@0: 
michael@0:           if (m.header()->num_fds >
michael@0:               FileDescriptorSet::MAX_DESCRIPTORS_PER_MESSAGE) {
michael@0:             // There are too many descriptors in this message
michael@0:             error = "Message requires an excessive number of descriptors";
michael@0:           }
michael@0: 
michael@0:           if (error) {
michael@0:             CHROMIUM_LOG(WARNING) << error
michael@0:                                   << " channel:" << this
michael@0:                                   << " message-type:" << m.type()
michael@0:                                   << " header()->num_fds:" << m.header()->num_fds
michael@0:                                   << " num_fds:" << num_fds
michael@0:                                   << " fds_i:" << fds_i;
michael@0:             // close the existing file descriptors so that we don't leak them
michael@0:             for (unsigned i = fds_i; i < num_fds; ++i)
michael@0:               HANDLE_EINTR(close(fds[i]));
michael@0:             input_overflow_fds_.clear();
michael@0:             // abort the connection
michael@0:             return false;
michael@0:           }
michael@0: 
michael@0: #if defined(OS_MACOSX)
michael@0:           // Send a message to the other side, indicating that we are now
michael@0:           // responsible for closing the descriptor.
michael@0:           Message *fdAck = new Message(MSG_ROUTING_NONE,
michael@0:                                        RECEIVED_FDS_MESSAGE_TYPE,
michael@0:                                        IPC::Message::PRIORITY_NORMAL);
michael@0:           DCHECK(m.fd_cookie() != 0);
michael@0:           fdAck->set_fd_cookie(m.fd_cookie());
michael@0:           OutputQueuePush(fdAck);
michael@0: #endif
michael@0: 
michael@0:           m.file_descriptor_set()->SetDescriptors(
michael@0:               &fds[fds_i], m.header()->num_fds);
michael@0:           fds_i += m.header()->num_fds;
michael@0:         }
michael@0: #ifdef IPC_MESSAGE_DEBUG_EXTRA
michael@0:         DLOG(INFO) << "received message on channel @" << this <<
michael@0:                       " with type " << m.type();
michael@0: #endif
michael@0: 
michael@0: #ifdef MOZ_TASK_TRACER
michael@0:         AutoSaveCurTraceInfo saveCurTraceInfo;
michael@0:         SetCurTraceInfo(m.header()->source_event_id,
michael@0:                         m.header()->parent_task_id,
michael@0:                         m.header()->source_event_type);
michael@0: #endif
michael@0: 
michael@0:         if (m.routing_id() == MSG_ROUTING_NONE &&
michael@0:             m.type() == HELLO_MESSAGE_TYPE) {
michael@0:           // The Hello message contains only the process id.
michael@0:           listener_->OnChannelConnected(MessageIterator(m).NextInt());
michael@0: #if defined(OS_MACOSX)
michael@0:         } else if (m.routing_id() == MSG_ROUTING_NONE &&
michael@0:                    m.type() == RECEIVED_FDS_MESSAGE_TYPE) {
michael@0:           DCHECK(m.fd_cookie() != 0);
michael@0:           CloseDescriptors(m.fd_cookie());
michael@0: #endif
michael@0:         } else {
michael@0:           listener_->OnMessageReceived(m);
michael@0:         }
michael@0:         p = message_tail;
michael@0:       } else {
michael@0:         // Last message is partial.
michael@0:         break;
michael@0:       }
michael@0:     }
michael@0:     if (end == p) {
michael@0:       ClearAndShrink(input_overflow_buf_, Channel::kReadBufferSize);
michael@0:     } else if (!overflowp) {
michael@0:       // p is from input_buf_
michael@0:       input_overflow_buf_.assign(p, end - p);
michael@0:     } else if (p > overflowp) {
michael@0:       // p is from input_overflow_buf_
michael@0:       input_overflow_buf_.erase(0, p - overflowp);
michael@0:     }
michael@0:     input_overflow_fds_ = std::vector<int>(&fds[fds_i], &fds[num_fds]);
michael@0: 
michael@0:     // When the input data buffer is empty, the overflow fds should be too. If
michael@0:     // this is not the case, we probably have a rogue renderer which is trying
michael@0:     // to fill our descriptor table.
michael@0:     if (input_overflow_buf_.empty() && !input_overflow_fds_.empty()) {
michael@0:       // We close these descriptors in Close()
michael@0:       return false;
michael@0:     }
michael@0: 
michael@0:     bytes_read = 0;  // Get more data.
michael@0:   }
michael@0: 
michael@0:   return true;
michael@0: }
michael@0: 
michael@0: bool Channel::ChannelImpl::ProcessOutgoingMessages() {
michael@0:   DCHECK(!waiting_connect_);  // Why are we trying to send messages if there's
michael@0:                               // no connection?
michael@0:   is_blocked_on_write_ = false;
michael@0: 
michael@0:   if (output_queue_.empty())
michael@0:     return true;
michael@0: 
michael@0:   if (pipe_ == -1)
michael@0:     return false;
michael@0: 
michael@0:   // Write out all the messages we can till the write blocks or there are no
michael@0:   // more outgoing messages.
michael@0:   while (!output_queue_.empty()) {
michael@0:     Message* msg = output_queue_.front();
michael@0: 
michael@0:     struct msghdr msgh = {0};
michael@0: 
michael@0:     static const int tmp = CMSG_SPACE(sizeof(
michael@0:         int[FileDescriptorSet::MAX_DESCRIPTORS_PER_MESSAGE]));
michael@0:     char buf[tmp];
michael@0: 
michael@0:     if (message_send_bytes_written_ == 0 &&
michael@0:         !msg->file_descriptor_set()->empty()) {
michael@0:       // This is the first chunk of a message which has descriptors to send
michael@0:       struct cmsghdr *cmsg;
michael@0:       const unsigned num_fds = msg->file_descriptor_set()->size();
michael@0: 
michael@0:       if (num_fds > FileDescriptorSet::MAX_DESCRIPTORS_PER_MESSAGE) {
michael@0:         CHROMIUM_LOG(FATAL) << "Too many file descriptors!";
michael@0:         // This should not be reached.
michael@0:         return false;
michael@0:       }
michael@0: 
michael@0:       msgh.msg_control = buf;
michael@0:       msgh.msg_controllen = CMSG_SPACE(sizeof(int) * num_fds);
michael@0:       cmsg = CMSG_FIRSTHDR(&msgh);
michael@0:       cmsg->cmsg_level = SOL_SOCKET;
michael@0:       cmsg->cmsg_type = SCM_RIGHTS;
michael@0:       cmsg->cmsg_len = CMSG_LEN(sizeof(int) * num_fds);
michael@0:       msg->file_descriptor_set()->GetDescriptors(
michael@0:           reinterpret_cast<int*>(CMSG_DATA(cmsg)));
michael@0:       msgh.msg_controllen = cmsg->cmsg_len;
michael@0: 
michael@0:       msg->header()->num_fds = num_fds;
michael@0: #if defined(OS_MACOSX)
michael@0:       msg->set_fd_cookie(++last_pending_fd_id_);
michael@0: #endif
michael@0:     }
michael@0: #ifdef MOZ_TASK_TRACER
michael@0:     GetCurTraceInfo(&msg->header()->source_event_id,
michael@0:                     &msg->header()->parent_task_id,
michael@0:                     &msg->header()->source_event_type);
michael@0: #endif
michael@0: 
michael@0:     size_t amt_to_write = msg->size() - message_send_bytes_written_;
michael@0:     DCHECK(amt_to_write != 0);
michael@0:     const char *out_bytes = reinterpret_cast<const char*>(msg->data()) +
michael@0:         message_send_bytes_written_;
michael@0: 
michael@0:     struct iovec iov = {const_cast<char*>(out_bytes), amt_to_write};
michael@0:     msgh.msg_iov = &iov;
michael@0:     msgh.msg_iovlen = 1;
michael@0: 
michael@0:     ssize_t bytes_written = HANDLE_EINTR(sendmsg(pipe_, &msgh, MSG_DONTWAIT));
michael@0: #if !defined(OS_MACOSX)
michael@0:     // On OSX CommitAll gets called later, once we get the RECEIVED_FDS_MESSAGE_TYPE
michael@0:     // message.
michael@0:     if (bytes_written > 0)
michael@0:       msg->file_descriptor_set()->CommitAll();
michael@0: #endif
michael@0: 
michael@0:     if (bytes_written < 0 && errno != EAGAIN) {
michael@0:       CHROMIUM_LOG(ERROR) << "pipe error: " << strerror(errno);
michael@0:       return false;
michael@0:     }
michael@0: 
michael@0:     if (static_cast<size_t>(bytes_written) != amt_to_write) {
michael@0:       if (bytes_written > 0) {
michael@0:         // If write() fails with EAGAIN then bytes_written will be -1.
michael@0:         message_send_bytes_written_ += bytes_written;
michael@0:       }
michael@0: 
michael@0:       // Tell libevent to call us back once things are unblocked.
michael@0:       is_blocked_on_write_ = true;
michael@0:       MessageLoopForIO::current()->WatchFileDescriptor(
michael@0:           pipe_,
michael@0:           false,  // One shot
michael@0:           MessageLoopForIO::WATCH_WRITE,
michael@0:           &write_watcher_,
michael@0:           this);
michael@0:       return true;
michael@0:     } else {
michael@0:       message_send_bytes_written_ = 0;
michael@0: 
michael@0: #if defined(OS_MACOSX)
michael@0:       if (!msg->file_descriptor_set()->empty())
michael@0:         pending_fds_.push_back(PendingDescriptors(msg->fd_cookie(),
michael@0:                                                   msg->file_descriptor_set()));
michael@0: #endif
michael@0: 
michael@0:       // Message sent OK!
michael@0: #ifdef IPC_MESSAGE_DEBUG_EXTRA
michael@0:       DLOG(INFO) << "sent message @" << msg << " on channel @" << this <<
michael@0:                     " with type " << msg->type();
michael@0: #endif
michael@0:       OutputQueuePop();
michael@0:       delete msg;
michael@0:     }
michael@0:   }
michael@0:   return true;
michael@0: }
michael@0: 
michael@0: bool Channel::ChannelImpl::Send(Message* message) {
michael@0: #ifdef IPC_MESSAGE_DEBUG_EXTRA
michael@0:   DLOG(INFO) << "sending message @" << message << " on channel @" << this
michael@0:              << " with type " << message->type()
michael@0:              << " (" << output_queue_.size() << " in queue)";
michael@0: #endif
michael@0: 
michael@0: #ifdef IPC_MESSAGE_LOG_ENABLED
michael@0:   Logging::current()->OnSendMessage(message, L"");
michael@0: #endif
michael@0: 
michael@0:   // If the channel has been closed, ProcessOutgoingMessages() is never going
michael@0:   // to pop anything off output_queue; output_queue will only get emptied when
michael@0:   // the channel is destructed.  We might as well delete message now, instead
michael@0:   // of waiting for the channel to be destructed.
michael@0:   if (closed_) {
michael@0:     if (mozilla::ipc::LoggingEnabled()) {
michael@0:       fprintf(stderr, "Can't send message %s, because this channel is closed.\n",
michael@0:               message->name());
michael@0:     }
michael@0:     delete message;
michael@0:     return false;
michael@0:   }
michael@0: 
michael@0:   OutputQueuePush(message);
michael@0:   if (!waiting_connect_) {
michael@0:     if (!is_blocked_on_write_) {
michael@0:       if (!ProcessOutgoingMessages())
michael@0:         return false;
michael@0:     }
michael@0:   }
michael@0: 
michael@0:   return true;
michael@0: }
michael@0: 
michael@0: void Channel::ChannelImpl::GetClientFileDescriptorMapping(int *src_fd,
michael@0:                                                           int *dest_fd) const {
michael@0:   DCHECK(mode_ == MODE_SERVER);
michael@0:   *src_fd = client_pipe_;
michael@0:   *dest_fd = kClientChannelFd;
michael@0: }
michael@0: 
michael@0: void Channel::ChannelImpl::CloseClientFileDescriptor() {
michael@0:   if (client_pipe_ != -1) {
michael@0:     Singleton<PipeMap>()->Remove(pipe_name_);
michael@0:     HANDLE_EINTR(close(client_pipe_));
michael@0:     client_pipe_ = -1;
michael@0:   }
michael@0: }
michael@0: 
michael@0: // Called by libevent when we can read from th pipe without blocking.
michael@0: void Channel::ChannelImpl::OnFileCanReadWithoutBlocking(int fd) {
michael@0:   bool send_server_hello_msg = false;
michael@0:   if (waiting_connect_ && mode_ == MODE_SERVER) {
michael@0:     // In the case of a socketpair() the server starts listening on its end
michael@0:     // of the pipe in Connect().
michael@0:     DCHECK(uses_fifo_);
michael@0: 
michael@0:     if (!ServerAcceptFifoConnection(server_listen_pipe_, &pipe_)) {
michael@0:       Close();
michael@0:     }
michael@0: 
michael@0:     // No need to watch the listening socket any longer since only one client
michael@0:     // can connect.  So unregister with libevent.
michael@0:     server_listen_connection_watcher_.StopWatchingFileDescriptor();
michael@0: 
michael@0:     // Start watching our end of the socket.
michael@0:     MessageLoopForIO::current()->WatchFileDescriptor(
michael@0:         pipe_,
michael@0:         true,
michael@0:         MessageLoopForIO::WATCH_READ,
michael@0:         &read_watcher_,
michael@0:         this);
michael@0: 
michael@0:     waiting_connect_ = false;
michael@0:     send_server_hello_msg = true;
michael@0:   }
michael@0: 
michael@0:   if (!waiting_connect_ && fd == pipe_) {
michael@0:     if (!ProcessIncomingMessages()) {
michael@0:       Close();
michael@0:       listener_->OnChannelError();
michael@0:     }
michael@0:   }
michael@0: 
michael@0:   // If we're a server and handshaking, then we want to make sure that we
michael@0:   // only send our handshake message after we've processed the client's.
michael@0:   // This gives us a chance to kill the client if the incoming handshake
michael@0:   // is invalid.
michael@0:   if (send_server_hello_msg) {
michael@0:     // This should be our first write so there's no chance we can block here...
michael@0:     DCHECK(is_blocked_on_write_ == false);
michael@0:     ProcessOutgoingMessages();
michael@0:   }
michael@0: }
michael@0: 
michael@0: #if defined(OS_MACOSX)
michael@0: void Channel::ChannelImpl::CloseDescriptors(uint32_t pending_fd_id)
michael@0: {
michael@0:   DCHECK(pending_fd_id != 0);
michael@0:   for (std::list<PendingDescriptors>::iterator i = pending_fds_.begin();
michael@0:        i != pending_fds_.end();
michael@0:        i++) {
michael@0:     if ((*i).id == pending_fd_id) {
michael@0:       (*i).fds->CommitAll();
michael@0:       pending_fds_.erase(i);
michael@0:       return;
michael@0:     }
michael@0:   }
michael@0:   DCHECK(false) << "pending_fd_id not in our list!";
michael@0: }
michael@0: #endif
michael@0: 
michael@0: void Channel::ChannelImpl::OutputQueuePush(Message* msg)
michael@0: {
michael@0:   output_queue_.push(msg);
michael@0:   output_queue_length_++;
michael@0: }
michael@0: 
michael@0: void Channel::ChannelImpl::OutputQueuePop()
michael@0: {
michael@0:   output_queue_.pop();
michael@0:   output_queue_length_--;
michael@0: }
michael@0: 
michael@0: // Called by libevent when we can write to the pipe without blocking.
michael@0: void Channel::ChannelImpl::OnFileCanWriteWithoutBlocking(int fd) {
michael@0:   if (!ProcessOutgoingMessages()) {
michael@0:     Close();
michael@0:     listener_->OnChannelError();
michael@0:   }
michael@0: }
michael@0: 
michael@0: void Channel::ChannelImpl::Close() {
michael@0:   // Close can be called multiple times, so we need to make sure we're
michael@0:   // idempotent.
michael@0: 
michael@0:   // Unregister libevent for the listening socket and close it.
michael@0:   server_listen_connection_watcher_.StopWatchingFileDescriptor();
michael@0: 
michael@0:   if (server_listen_pipe_ != -1) {
michael@0:     HANDLE_EINTR(close(server_listen_pipe_));
michael@0:     server_listen_pipe_ = -1;
michael@0:   }
michael@0: 
michael@0:   // Unregister libevent for the FIFO and close it.
michael@0:   read_watcher_.StopWatchingFileDescriptor();
michael@0:   write_watcher_.StopWatchingFileDescriptor();
michael@0:   if (pipe_ != -1) {
michael@0:     HANDLE_EINTR(close(pipe_));
michael@0:     pipe_ = -1;
michael@0:   }
michael@0:   if (client_pipe_ != -1) {
michael@0:     Singleton<PipeMap>()->Remove(pipe_name_);
michael@0:     HANDLE_EINTR(close(client_pipe_));
michael@0:     client_pipe_ = -1;
michael@0:   }
michael@0: 
michael@0:   if (uses_fifo_) {
michael@0:     // Unlink the FIFO
michael@0:     unlink(pipe_name_.c_str());
michael@0:   }
michael@0: 
michael@0:   while (!output_queue_.empty()) {
michael@0:     Message* m = output_queue_.front();
michael@0:     OutputQueuePop();
michael@0:     delete m;
michael@0:   }
michael@0: 
michael@0:   // Close any outstanding, received file descriptors
michael@0:   for (std::vector<int>::iterator
michael@0:        i = input_overflow_fds_.begin(); i != input_overflow_fds_.end(); ++i) {
michael@0:     HANDLE_EINTR(close(*i));
michael@0:   }
michael@0:   input_overflow_fds_.clear();
michael@0: 
michael@0: #if defined(OS_MACOSX)
michael@0:   for (std::list<PendingDescriptors>::iterator i = pending_fds_.begin();
michael@0:        i != pending_fds_.end();
michael@0:        i++) {
michael@0:     (*i).fds->CommitAll();
michael@0:   }
michael@0:   pending_fds_.clear();
michael@0: #endif
michael@0: 
michael@0:   closed_ = true;
michael@0: }
michael@0: 
michael@0: bool Channel::ChannelImpl::Unsound_IsClosed() const
michael@0: {
michael@0:   return closed_;
michael@0: }
michael@0: 
michael@0: uint32_t Channel::ChannelImpl::Unsound_NumQueuedMessages() const
michael@0: {
michael@0:   return output_queue_length_;
michael@0: }
michael@0: 
michael@0: //------------------------------------------------------------------------------
michael@0: // Channel's methods simply call through to ChannelImpl.
michael@0: Channel::Channel(const std::wstring& channel_id, Mode mode,
michael@0:                  Listener* listener)
michael@0:     : channel_impl_(new ChannelImpl(channel_id, mode, listener)) {
michael@0: }
michael@0: 
michael@0: Channel::Channel(int fd, Mode mode, Listener* listener)
michael@0:     : channel_impl_(new ChannelImpl(fd, mode, listener)) {
michael@0: }
michael@0: 
michael@0: Channel::~Channel() {
michael@0:   delete channel_impl_;
michael@0: }
michael@0: 
michael@0: bool Channel::Connect() {
michael@0:   return channel_impl_->Connect();
michael@0: }
michael@0: 
michael@0: void Channel::Close() {
michael@0:   channel_impl_->Close();
michael@0: }
michael@0: 
michael@0: Channel::Listener* Channel::set_listener(Listener* listener) {
michael@0:   return channel_impl_->set_listener(listener);
michael@0: }
michael@0: 
michael@0: bool Channel::Send(Message* message) {
michael@0:   return channel_impl_->Send(message);
michael@0: }
michael@0: 
michael@0: void Channel::GetClientFileDescriptorMapping(int *src_fd, int *dest_fd) const {
michael@0:   return channel_impl_->GetClientFileDescriptorMapping(src_fd, dest_fd);
michael@0: }
michael@0: 
michael@0: void Channel::ResetFileDescriptor(int fd) {
michael@0:   channel_impl_->ResetFileDescriptor(fd);
michael@0: }
michael@0: 
michael@0: int Channel::GetFileDescriptor() const {
michael@0:     return channel_impl_->GetFileDescriptor();
michael@0: }
michael@0: 
michael@0: void Channel::CloseClientFileDescriptor() {
michael@0:   channel_impl_->CloseClientFileDescriptor();
michael@0: }
michael@0: 
michael@0: bool Channel::Unsound_IsClosed() const {
michael@0:   return channel_impl_->Unsound_IsClosed();
michael@0: }
michael@0: 
michael@0: uint32_t Channel::Unsound_NumQueuedMessages() const {
michael@0:   return channel_impl_->Unsound_NumQueuedMessages();
michael@0: }
michael@0: 
michael@0: }  // namespace IPC