Mojo: (Theoretically) implement the read side of platform handle passing.

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 "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"
 #include "mojo/embedder/platform_channel_utils_posix.h"
 #include "mojo/embedder/platform_handle.h"
+#include "mojo/embedder/platform_handle_vector.h"
+#include "mojo/system/transport_data.h"
 
 namespace mojo {
 namespace system {
 
 namespace {
 
 class RawChannelPosix : public RawChannel,
                         public base::MessageLoopForIO::Watcher {
  public:
   RawChannelPosix(embedder::ScopedPlatformHandle handle);
   virtual ~RawChannelPosix();
 
   // |RawChannel| public methods:
   virtual size_t GetSerializedPlatformHandleSize() const OVERRIDE;
 
  private:
   // |RawChannel| protected methods:
   virtual IOResult Read(size_t* bytes_read) OVERRIDE;
   virtual IOResult ScheduleRead() OVERRIDE;
+  virtual scoped_ptr<embedder::PlatformHandleVector> GetReadPlatformHandles(
+      size_t num_platform_handles,
+      const void* platform_handle_table) OVERRIDE;
   virtual IOResult WriteNoLock(size_t* platform_handles_written,
                                size_t* bytes_written) OVERRIDE;
   virtual IOResult ScheduleWriteNoLock() OVERRIDE;
   virtual bool OnInit() OVERRIDE;
   virtual void OnShutdownNoLock(
       scoped_ptr<ReadBuffer> read_buffer,
       scoped_ptr<WriteBuffer> write_buffer) OVERRIDE;
 
   // |base::MessageLoopForIO::Watcher| implementation:
   virtual void OnFileCanReadWithoutBlocking(int fd) OVERRIDE;
   virtual void OnFileCanWriteWithoutBlocking(int fd) OVERRIDE;
 
   // Watches for |fd_| to become writable. Must be called on the I/O thread.
   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_;
 
+  scoped_ptr<embedder::PlatformHandleVector> 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_;
 
   DISALLOW_COPY_AND_ASSIGN(RawChannelPosix);
@@ skipping 27 matching lines @@
 }
 
 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);
 
-  scoped_ptr<embedder::PlatformHandleVector> handles;
-  ssize_t read_result = embedder::PlatformChannelRecvmsg(fd_.get(),
-                                                         buffer,
-                                                         bytes_to_read,
-                                                         &handles);
+  size_t old_num_platform_handles =
+      read_platform_handles_ ? read_platform_handles_->size() : 0;
+  ssize_t read_result =
+      embedder::PlatformChannelRecvmsg(fd_.get(),
+                                       buffer,
+                                       bytes_to_read,
+                                       &read_platform_handles_);
   if (read_result > 0) {
     *bytes_read = static_cast<size_t>(read_result);
     return IO_SUCCEEDED;
   }
 
-  if (handles) {
+  if (read_platform_handles_ &&
+      read_platform_handles_->size() > old_num_platform_handles) {
     if (read_result != 1) {
-      LOG(WARNING) << "Invalid control message with handles";
+      LOG(WARNING) << "Invalid control message with platform handles";
       return IO_FAILED;
     }
 
-    // TODO(vtl): Implement this ("buffer" received handles). For now, just drop
-    // them on the floor. (Discard this message entirely.)
-    NOTIMPLEMENTED();
-    for (size_t i = 0; i < handles->size(); i++)
-      (*handles)[i].CloseIfNecessary();
-    return ScheduleRead();
+    if (read_platform_handles_->size() > TransportData::kMaxPlatformHandles) {
+      LOG(WARNING) << "Received too many platform handles";
+      read_platform_handles_.reset();

[yzshen1, 2014/05/11 07:16:59]

The handles need to be closed? (here and line 176)

[viettrungluu, 2014/05/12 17:33:28]

Done.

+      return IO_FAILED;
+    }
   }
 
   // |read_result == 0| means "end of file".
   if (read_result == 0 || (errno != EAGAIN && errno != EWOULDBLOCK)) {
     PLOG_IF(ERROR, read_result != 0) << "recvmsg";
 
     // Make sure that |OnFileCanReadWithoutBlocking()| won't be called again.
     read_watcher_.reset();
 
     return IO_FAILED;
   }
 
   return ScheduleRead();
 }
 
 RawChannel::IOResult RawChannelPosix::ScheduleRead() {
   DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io());
   DCHECK(!pending_read_);
 
   pending_read_ = true;
 
   return IO_PENDING;
 }
 
+scoped_ptr<embedder::PlatformHandleVector>
+    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();
+    return scoped_ptr<embedder::PlatformHandleVector>();
+  }
+
+  return read_platform_handles_.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()) {
     size_t num_platform_handles;
     embedder::PlatformHandle* platform_handles;
@@ skipping 190 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