Add a Mojo EDK for Chrome that uses one OS pipe per message pipe.

mojo/edk/system/raw_channel_posix.cc

Index: mojo/edk/system/raw_channel_posix.cc
diff --git a/third_party/mojo/src/mojo/edk/system/raw_channel_posix.cc b/mojo/edk/system/raw_channel_posix.cc
similarity index 77%
copy from third_party/mojo/src/mojo/edk/system/raw_channel_posix.cc
copy to mojo/edk/system/raw_channel_posix.cc
index c2be79fec3d2d69d89cb785d5da93911c2607850..a9a1e52720467f2cfd7b483cda36d80db8c864d0 100644
--- a/third_party/mojo/src/mojo/edk/system/raw_channel_posix.cc
+++ b/mojo/edk/system/raw_channel_posix.cc
@@ -2,9 +2,10 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
-#include "third_party/mojo/src/mojo/edk/system/raw_channel.h"
+#include "mojo/edk/system/raw_channel.h"
 
 #include <errno.h>
+#include <sys/stat.h>
 #include <sys/uio.h>
 #include <unistd.h>
 
@@ -18,26 +19,23 @@
 #include "base/memory/weak_ptr.h"
 #include "base/message_loop/message_loop.h"
 #include "base/synchronization/lock.h"
+#include "mojo/edk/embedder/platform_channel_utils_posix.h"
+#include "mojo/edk/embedder/platform_handle.h"
+#include "mojo/edk/embedder/platform_handle_vector.h"
+#include "mojo/edk/system/transport_data.h"
 #include "mojo/public/cpp/system/macros.h"
-#include "third_party/mojo/src/mojo/edk/embedder/platform_channel_utils_posix.h"
-#include "third_party/mojo/src/mojo/edk/embedder/platform_handle.h"
-#include "third_party/mojo/src/mojo/edk/embedder/platform_handle_vector.h"
-#include "third_party/mojo/src/mojo/edk/system/transport_data.h"
 
 namespace mojo {
-namespace system {
+namespace edk {
 
 namespace {
 
 class RawChannelPosix final : public RawChannel,
                               public base::MessageLoopForIO::Watcher {
  public:
-  explicit RawChannelPosix(embedder::ScopedPlatformHandle handle);
+  explicit RawChannelPosix(ScopedPlatformHandle handle);
   ~RawChannelPosix() override;
 
-  // |RawChannel| public methods:
-  size_t GetSerializedPlatformHandleSize() const override;
-
  private:
   // |RawChannel| protected methods:
   // Actually override this so that we can send multiple messages with (only)
@@ -46,9 +44,13 @@ class RawChannelPosix final : public RawChannel,
   // Override this to handle those extra FD-only messages.
   bool OnReadMessageForRawChannel(
       const MessageInTransit::View& message_view) override;
+
+  ScopedPlatformHandle ReleaseHandleNoLock(
+      std::vector<char>* read_buffer_out) override;
+  PlatformHandle HandleForDebuggingNoLock() override;
   IOResult Read(size_t* bytes_read) override;
   IOResult ScheduleRead() override;
-  embedder::ScopedPlatformHandleVectorPtr GetReadPlatformHandles(
+  ScopedPlatformHandleVectorPtr GetReadPlatformHandles(
       size_t num_platform_handles,
       const void* platform_handle_table) override;
   IOResult WriteNoLock(size_t* platform_handles_written,
@@ -68,7 +70,7 @@ class RawChannelPosix final : public RawChannel,
   // Watches for |fd_| to become writable. Must be called on the I/O thread.
   void WaitToWrite();
 
-  embedder::ScopedPlatformHandle fd_;
+  ScopedPlatformHandle fd_;
 
   // The following members are only used on the I/O thread:
   scoped_ptr<base::MessageLoopForIO::FileDescriptorWatcher> read_watcher_;
@@ -76,7 +78,7 @@ class RawChannelPosix final : public RawChannel,
 
   bool pending_read_;
 
-  std::deque<embedder::PlatformHandle> read_platform_handles_;
+  std::deque<PlatformHandle> read_platform_handles_;
 
   // The following members are used on multiple threads and protected by
   // |write_lock()|:
@@ -90,7 +92,7 @@ class RawChannelPosix final : public RawChannel,
   MOJO_DISALLOW_COPY_AND_ASSIGN(RawChannelPosix);
 };
 
-RawChannelPosix::RawChannelPosix(embedder::ScopedPlatformHandle handle)
+RawChannelPosix::RawChannelPosix(ScopedPlatformHandle handle)
     : fd_(handle.Pass()),
       pending_read_(false),
       pending_write_(false),
@@ -111,37 +113,29 @@ RawChannelPosix::~RawChannelPosix() {
   DCHECK(!read_watcher_);
   DCHECK(!write_watcher_);
 
-  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;
+  CloseAllPlatformHandles(&read_platform_handles_);
 }
 
 void RawChannelPosix::EnqueueMessageNoLock(
     scoped_ptr<MessageInTransit> message) {
   if (message->transport_data()) {
-    embedder::PlatformHandleVector* const platform_handles =
+    PlatformHandleVector* const platform_handles =
         message->transport_data()->platform_handles();
     if (platform_handles &&
-        platform_handles->size() > embedder::kPlatformChannelMaxNumHandles) {
+        platform_handles->size() > kPlatformChannelMaxNumHandles) {
       // We can't attach all the FDs to a single message, so we have to "split"
       // the message. Send as many control messages as needed first with FDs
       // attached (and no data).
       size_t i = 0;
-      for (; platform_handles->size() - i >
-                 embedder::kPlatformChannelMaxNumHandles;
-           i += embedder::kPlatformChannelMaxNumHandles) {
+      for (; platform_handles->size() - i > kPlatformChannelMaxNumHandles;
+           i += kPlatformChannelMaxNumHandles) {
         scoped_ptr<MessageInTransit> fd_message(new MessageInTransit(
-            MessageInTransit::Type::RAW_CHANNEL,
-            MessageInTransit::Subtype::RAW_CHANNEL_POSIX_EXTRA_PLATFORM_HANDLES,
-            0, nullptr));
-        embedder::ScopedPlatformHandleVectorPtr fds(
-            new embedder::PlatformHandleVector(
+            MessageInTransit::Type::RAW_CHANNEL_POSIX_EXTRA_PLATFORM_HANDLES, 0,
+            nullptr));
+        ScopedPlatformHandleVectorPtr fds(
+            new PlatformHandleVector(
                 platform_handles->begin() + i,
-                platform_handles->begin() + i +
-                    embedder::kPlatformChannelMaxNumHandles));
+                platform_handles->begin() + i + kPlatformChannelMaxNumHandles));
         fd_message->SetTransportData(make_scoped_ptr(
             new TransportData(fds.Pass(), GetSerializedPlatformHandleSize())));
         RawChannel::EnqueueMessageNoLock(fd_message.Pass());
@@ -158,10 +152,8 @@ void RawChannelPosix::EnqueueMessageNoLock(
 
 bool RawChannelPosix::OnReadMessageForRawChannel(
     const MessageInTransit::View& message_view) {
-  DCHECK_EQ(message_view.type(), MessageInTransit::Type::RAW_CHANNEL);
-
-  if (message_view.subtype() ==
-      MessageInTransit::Subtype::RAW_CHANNEL_POSIX_EXTRA_PLATFORM_HANDLES) {
+  if (message_view.type() ==
+      MessageInTransit::Type::RAW_CHANNEL_POSIX_EXTRA_PLATFORM_HANDLES) {
     // We don't need to do anything. |RawChannel| won't extract the platform
     // handles, and they'll be accumulated in |Read()|.
     return true;
@@ -170,6 +162,40 @@ bool RawChannelPosix::OnReadMessageForRawChannel(
   return RawChannel::OnReadMessageForRawChannel(message_view);
 }
 
+
+ScopedPlatformHandle RawChannelPosix::ReleaseHandleNoLock(
+    std::vector<char>* read_buffer_out) {
+  std::vector<WriteBuffer::Buffer> buffers;
+  write_buffer_no_lock()->GetBuffers(&buffers);
+  if (!buffers.empty()) {
+    // TODO(jam): copy code in OnShutdownNoLock
+    NOTREACHED() << "releasing handle with pending write buffer";
+  }
+
+  NOTREACHED() << "TODO(jam) IMPLEMENT";/*
+  if (handle_.is_valid()) {
+    // SetInitialBuffer could have been called on main thread before OnInit
+    // is called on Io thread. and in meantime releasehandle called.
+    //DCHECK(read_buffer()->num_valid_bytes() == 0);
+    if (read_buffer()->num_valid_bytes()) {
+      read_buffer_out->resize(read_buffer()->num_valid_bytes());
+      memcpy(&(*read_buffer_out)[0], read_buffer()->buffer(),
+             read_buffer()->num_valid_bytes());
+      read_buffer()->Reset();
+    }
+    DCHECK(write_buffer_no_lock()->queue_size() == 0);
+    return ScopedPlatformHandle(PlatformHandle(handle_.release().handle));
+  }
+
+  return io_handler_->ReleaseHandle(read_buffer_out);
+  */
+  return fd_.Pass();
+}
+
+PlatformHandle RawChannelPosix::HandleForDebuggingNoLock() {
+  return fd_.get();
+}
+
 RawChannel::IOResult RawChannelPosix::Read(size_t* bytes_read) {
   DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io());
   DCHECK(!pending_read_);
@@ -191,19 +217,19 @@ RawChannel::IOResult RawChannelPosix::ScheduleRead() {
   return IO_PENDING;
 }
 
-embedder::ScopedPlatformHandleVectorPtr RawChannelPosix::GetReadPlatformHandles(
+ScopedPlatformHandleVectorPtr RawChannelPosix::GetReadPlatformHandles(
     size_t num_platform_handles,
     const void* /*platform_handle_table*/) {
   DCHECK_GT(num_platform_handles, 0u);
 
   if (read_platform_handles_.size() < num_platform_handles) {
-    embedder::CloseAllPlatformHandles(&read_platform_handles_);
+    CloseAllPlatformHandles(&read_platform_handles_);
     read_platform_handles_.clear();
-    return embedder::ScopedPlatformHandleVectorPtr();
+    return ScopedPlatformHandleVectorPtr();
   }
 
-  embedder::ScopedPlatformHandleVectorPtr rv(
-      new embedder::PlatformHandleVector(num_platform_handles));
+  ScopedPlatformHandleVectorPtr rv(
+      new PlatformHandleVector(num_platform_handles));
   rv->assign(read_platform_handles_.begin(),
              read_platform_handles_.begin() + num_platform_handles);
   read_platform_handles_.erase(
@@ -222,12 +248,12 @@ RawChannel::IOResult RawChannelPosix::WriteNoLock(
   size_t num_platform_handles = 0;
   ssize_t write_result;
   if (write_buffer_no_lock()->HavePlatformHandlesToSend()) {
-    embedder::PlatformHandle* platform_handles;
+    PlatformHandle* platform_handles;
     void* serialization_data;  // Actually unused.
     write_buffer_no_lock()->GetPlatformHandlesToSend(
         &num_platform_handles, &platform_handles, &serialization_data);
     DCHECK_GT(num_platform_handles, 0u);
-    DCHECK_LE(num_platform_handles, embedder::kPlatformChannelMaxNumHandles);
+    DCHECK_LE(num_platform_handles, kPlatformChannelMaxNumHandles);
     DCHECK(platform_handles);
 
     // TODO(vtl): Reduce code duplication. (This is duplicated from below.)
@@ -242,7 +268,7 @@ RawChannel::IOResult RawChannelPosix::WriteNoLock(
       iov[i].iov_len = buffers[i].size;
     }
 
-    write_result = embedder::PlatformChannelSendmsgWithHandles(
+    write_result = PlatformChannelSendmsgWithHandles(
         fd_.get(), iov, buffer_count, platform_handles, num_platform_handles);
     if (write_result >= 0) {
       for (size_t i = 0; i < num_platform_handles; i++)
@@ -254,8 +280,8 @@ RawChannel::IOResult RawChannelPosix::WriteNoLock(
     DCHECK(!buffers.empty());
 
     if (buffers.size() == 1) {
-      write_result = embedder::PlatformChannelWrite(fd_.get(), buffers[0].addr,
-                                                    buffers[0].size);
+      write_result = PlatformChannelWrite(fd_.get(), buffers[0].addr,
+                                          buffers[0].size);
     } else {
       const size_t kMaxBufferCount = 10;
       iovec iov[kMaxBufferCount];
@@ -265,8 +291,7 @@ RawChannel::IOResult RawChannelPosix::WriteNoLock(
         iov[i].iov_len = buffers[i].size;
       }
 
-      write_result =
-          embedder::PlatformChannelWritev(fd_.get(), iov, buffer_count);
+      write_result = PlatformChannelWritev(fd_.get(), iov, buffer_count);
     }
   }
 
@@ -405,22 +430,22 @@ RawChannel::IOResult RawChannelPosix::ReadImpl(size_t* bytes_read) {
   read_buffer()->GetBuffer(&buffer, &bytes_to_read);
 
   size_t old_num_platform_handles = read_platform_handles_.size();
-  ssize_t read_result = embedder::PlatformChannelRecvmsg(
+  ssize_t read_result = PlatformChannelRecvmsg(
       fd_.get(), buffer, bytes_to_read, &read_platform_handles_);
   if (read_platform_handles_.size() > old_num_platform_handles) {
     DCHECK_LE(read_platform_handles_.size() - old_num_platform_handles,
-              embedder::kPlatformChannelMaxNumHandles);
+              kPlatformChannelMaxNumHandles);
 
     // We should never accumulate more than |TransportData::kMaxPlatformHandles
-    // + embedder::kPlatformChannelMaxNumHandles| handles. (The latter part is
+    // + 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::GetMaxPlatformHandles() +
-         embedder::kPlatformChannelMaxNumHandles)) {
+         kPlatformChannelMaxNumHandles)) {
       LOG(ERROR) << "Received too many platform handles";
-      embedder::CloseAllPlatformHandles(&read_platform_handles_);
+      CloseAllPlatformHandles(&read_platform_handles_);
       read_platform_handles_.clear();
       return IO_FAILED_UNKNOWN;
     }
@@ -470,10 +495,26 @@ void RawChannelPosix::WaitToWrite() {
 
 // Static factory method declared in raw_channel.h.
 // static
-scoped_ptr<RawChannel> RawChannel::Create(
-    embedder::ScopedPlatformHandle handle) {
-  return make_scoped_ptr(new RawChannelPosix(handle.Pass()));
+RawChannel* RawChannel::Create(ScopedPlatformHandle handle) {
+  return new RawChannelPosix(handle.Pass());
+}
+
+size_t RawChannel::GetSerializedPlatformHandleSize() {
+  // We don't actually need any space on POSIX (since we just send FDs).
+  return 0;
+}
+
+bool RawChannel::IsOtherEndOf(RawChannel* other) {

[yzshen1, 2015/10/02 17:32:18]

nit: passing |this| as argument to this method will return true.

+  PlatformHandle this_handle = HandleForDebuggingNoLock();
+  PlatformHandle other_handle = other->HandleForDebuggingNoLock();
+
+  struct stat stat1, stat2;
+  if (fstat(this_handle.fd, &stat1) < 0)
+    return false;
+  if (fstat(other_handle.fd, &stat2) < 0)
+    return false;
+  return (stat1.st_dev == stat2.st_dev) && (stat1.st_ino == stat2.st_ino);
 }
 
-}  // namespace system
+}  // namespace edk
 }  // namespace mojo