Mojo: Make PlatformChannelRecvmsg() append handles to a deque instead.

mojo/system/raw_channel_posix.cc

 // Copyright 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 "mojo/system/raw_channel.h"
 
 #include <errno.h>
 #include <sys/uio.h>
 #include <unistd.h>
 
 #include <algorithm>
+#include <deque>
 
 #include "base/basictypes.h"
 #include "base/bind.h"
 #include "base/compiler_specific.h"
 #include "base/location.h"
 #include "base/logging.h"
 #include "base/memory/scoped_ptr.h"
 #include "base/memory/weak_ptr.h"
 #include "base/message_loop/message_loop.h"
 #include "base/synchronization/lock.h"
@@ skipping 39 matching lines @@
   void WaitToWrite();
 
   embedder::ScopedPlatformHandle fd_;
 
   // The following members are only used on the I/O thread:
   scoped_ptr<base::MessageLoopForIO::FileDescriptorWatcher> read_watcher_;
   scoped_ptr<base::MessageLoopForIO::FileDescriptorWatcher> write_watcher_;
 
   bool pending_read_;
 
-  embedder::ScopedPlatformHandleVectorPtr read_platform_handles_;
+  std::deque<embedder::PlatformHandle> read_platform_handles_;
 
   // The following members are used on multiple threads and protected by
   // |write_lock()|:
   bool pending_write_;
 
   // This is used for posting tasks from write threads to the I/O thread. It
   // must only be accessed under |write_lock_|. The weak pointers it produces
   // are only used/invalidated on the I/O thread.
   base::WeakPtrFactory<RawChannelPosix> weak_ptr_factory_;
 
@@ skipping 13 matching lines @@
   DCHECK(!pending_write_);
 
   // No need to take the |write_lock()| here -- if there are still weak pointers
   // outstanding, then we're hosed anyway (since we wouldn't be able to
   // invalidate them cleanly, since we might not be on the I/O thread).
   DCHECK(!weak_ptr_factory_.HasWeakPtrs());
 
   // These must have been shut down/destroyed on the I/O thread.
   DCHECK(!read_watcher_.get());
   DCHECK(!write_watcher_.get());
+
+  embedder::CloseAllPlatformHandles(&read_platform_handles_);
 }
 
 size_t RawChannelPosix::GetSerializedPlatformHandleSize() const {
   // We don't actually need any space on POSIX (since we just send FDs).
   return 0;
 }
 
 RawChannel::IOResult RawChannelPosix::Read(size_t* bytes_read) {
   DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io());
   DCHECK(!pending_read_);
 
   char* buffer = NULL;
   size_t bytes_to_read = 0;
   read_buffer()->GetBuffer(&buffer, &bytes_to_read);
 
-  size_t old_num_platform_handles =
-      read_platform_handles_ ? read_platform_handles_->size() : 0;
+  size_t old_num_platform_handles = read_platform_handles_.size();
   ssize_t read_result =
       embedder::PlatformChannelRecvmsg(fd_.get(),
                                        buffer,
                                        bytes_to_read,
                                        &read_platform_handles_);
-  if (read_platform_handles_ &&
-      read_platform_handles_->size() > old_num_platform_handles) {
+  if (read_platform_handles_.size() > old_num_platform_handles) {
+    DCHECK_LE(read_platform_handles_.size() - old_num_platform_handles,
+              embedder::kPlatformChannelMaxNumHandles);
+
     if (read_result != 1) {
       LOG(WARNING) << "Invalid control message with platform handles";
       return IO_FAILED;
     }
 
-    if (read_platform_handles_->size() > TransportData::kMaxPlatformHandles) {
+    // We should never accumulate more than |TransportData::kMaxPlatformHandles
+    // + embedder::kPlatformChannelMaxNumHandles| handles. (The latter part is
+    // possible because we could have accumulated all the handles for a message,
+    // then received the message data plus the first set of handles for the next
+    // message in the subsequent |recvmsg()|.)
+    if (read_platform_handles_.size() > (TransportData::kMaxPlatformHandles +
+            embedder::kPlatformChannelMaxNumHandles)) {
       LOG(WARNING) << "Received too many platform handles";
-      read_platform_handles_.reset();
+      embedder::CloseAllPlatformHandles(&read_platform_handles_);
+      read_platform_handles_.clear();
       return IO_FAILED;
     }
 
     // This wasn't a data read (platform handles just get stashed).
     return ScheduleRead();
   }
 
   if (read_result > 0) {
     *bytes_read = static_cast<size_t>(read_result);
     return IO_SUCCEEDED;
@@ skipping 19 matching lines @@
   pending_read_ = true;
 
   return IO_PENDING;
 }
 
 embedder::ScopedPlatformHandleVectorPtr RawChannelPosix::GetReadPlatformHandles(
     size_t num_platform_handles,
     const void* /*platform_handle_table*/) {
   DCHECK_GT(num_platform_handles, 0u);
 
-  if (!read_platform_handles_ ||
-      read_platform_handles_->size() != num_platform_handles) {
-    read_platform_handles_.reset();
+  if (read_platform_handles_.size() < num_platform_handles) {
+    embedder::CloseAllPlatformHandles(&read_platform_handles_);
+    read_platform_handles_.clear();
     return embedder::ScopedPlatformHandleVectorPtr();
   }
 
-  return read_platform_handles_.Pass();
+  embedder::ScopedPlatformHandleVectorPtr rv(
+      new embedder::PlatformHandleVector(num_platform_handles));
+  rv->assign(read_platform_handles_.begin(),
+             read_platform_handles_.begin() + num_platform_handles);
+  read_platform_handles_.erase(
+      read_platform_handles_.begin(),
+      read_platform_handles_.begin() + num_platform_handles);
+  return rv.Pass();
 }
 
 RawChannel::IOResult RawChannelPosix::WriteNoLock(
     size_t* platform_handles_written,
     size_t* bytes_written) {
   write_lock().AssertAcquired();
 
   DCHECK(!pending_write_);
 
   if (write_buffer_no_lock()->HavePlatformHandlesToSend()) {
@@ skipping 192 matching lines @@
 
 // Static factory method declared in raw_channel.h.
 // static
 scoped_ptr<RawChannel> RawChannel::Create(
     embedder::ScopedPlatformHandle handle) {
   return scoped_ptr<RawChannel>(new RawChannelPosix(handle.Pass()));
 }
 
 }  // namespace system
 }  // namespace mojo