Ensure that DataPipeProducerDispatcher::CreateEquivalentDispatcherAndCloseImplNoLock serializes the object.

Otherwise the channel_ is only serialized later in EndSerializeAndCloseImplNoLock. In between those two calls, if the consumer is closed the channel_ can call the old and destructed DataPipeProducerDispatcher.

BUG=478251

Review URL: https://codereview.chromium.org/1403773005

Cr-Commit-Position: refs/heads/master@{#353914}

3 files changed, 106 insertions, 12 deletions

diff --git a/mojo/edk/system/data_pipe_producer_dispatcher.cc b/mojo/edk/system/data_pipe_producer_dispatcher.cc
index 3a579c8..4ae4ba8 100644
--- a/mojo/edk/system/data_pipe_producer_dispatcher.cc
+++ b/mojo/edk/system/data_pipe_producer_dispatcher.cc
@@ -81,7 +81,7 @@ DataPipeProducerDispatcher::Deserialize(
 
 DataPipeProducerDispatcher::DataPipeProducerDispatcher(
     const MojoCreateDataPipeOptions& options)
-    : options_(options), channel_(nullptr), error_(false) {
+    : options_(options), channel_(nullptr), error_(false), serialized_(false) {
 }
 
 DataPipeProducerDispatcher::~DataPipeProducerDispatcher() {
@@ -104,10 +104,12 @@ scoped_refptr<Dispatcher>
 DataPipeProducerDispatcher::CreateEquivalentDispatcherAndCloseImplNoLock() {
   lock().AssertAcquired();
 
+  SerializeInternal();
+
   scoped_refptr<DataPipeProducerDispatcher> rv = Create(options_);
-  rv->channel_ = channel_;
-  channel_ = nullptr;
-  rv->options_ = options_;
+  serialized_write_buffer_.swap(rv->serialized_write_buffer_);
+  rv->serialized_platform_handle_ = serialized_platform_handle_.Pass();
+  rv->serialized_ = true;
   return scoped_refptr<Dispatcher>(rv.get());
 }
 
@@ -256,14 +258,9 @@ void DataPipeProducerDispatcher::StartSerializeImplNoLock(
     size_t* max_size,
     size_t* max_platform_handles) {
   DCHECK(HasOneRef());  // Only one ref => no need to take the lock.
+  if (!serialized_)
+    SerializeInternal();
 
-  if (channel_) {
-    std::vector<char> serialized_read_buffer;
-    serialized_platform_handle_ = channel_->ReleaseHandle(
-        &serialized_read_buffer, &serialized_write_buffer_);
-    channel_ = nullptr;
-    CHECK(serialized_read_buffer.empty());
-  }
   DataPipe::StartSerialize(serialized_platform_handle_.is_valid(),
                            !serialized_write_buffer_.empty(), max_size,
                            max_platform_handles);
@@ -378,5 +375,18 @@ bool DataPipeProducerDispatcher::WriteDataIntoMessages(
   return true;
 }
 
+void DataPipeProducerDispatcher::SerializeInternal() {
+  // We need to stop watching handle immediately, even though not on IO thread,
+  // so that other messages aren't read after this.
+  if (channel_) {
+    std::vector<char> serialized_read_buffer;
+    serialized_platform_handle_ = channel_->ReleaseHandle(
+        &serialized_read_buffer, &serialized_write_buffer_);
+    CHECK(serialized_read_buffer.empty());
+    channel_ = nullptr;
+  }
+  serialized_ = true;
+}
+
 }  // namespace edk
 }  // namespace mojo
diff --git a/mojo/edk/system/data_pipe_producer_dispatcher.h b/mojo/edk/system/data_pipe_producer_dispatcher.h
index d8c86e4..daf038b 100644
--- a/mojo/edk/system/data_pipe_producer_dispatcher.h
+++ b/mojo/edk/system/data_pipe_producer_dispatcher.h
@@ -85,6 +85,9 @@ class MOJO_SYSTEM_IMPL_EXPORT DataPipeProducerDispatcher final
   bool InTwoPhaseWrite() const;
   bool WriteDataIntoMessages(const void* elements, uint32_t num_bytes);
 
+  // See comment in MessagePipeDispatcher for this method.
+  void SerializeInternal();
+
   MojoCreateDataPipeOptions options_;
 
   // Protected by |lock()|:
@@ -98,6 +101,7 @@ class MOJO_SYSTEM_IMPL_EXPORT DataPipeProducerDispatcher final
 
   bool error_;
 
+  bool serialized_;
   ScopedPlatformHandle serialized_platform_handle_;
   std::vector<char> serialized_write_buffer_;
   std::vector<char> two_phase_data_;
diff --git a/mojo/edk/system/data_pipe_unittest.cc b/mojo/edk/system/data_pipe_unittest.cc
index beda4592..d32f73f 100644
--- a/mojo/edk/system/data_pipe_unittest.cc
+++ b/mojo/edk/system/data_pipe_unittest.cc
@@ -13,9 +13,9 @@
 #include "mojo/edk/embedder/simple_platform_support.h"
 #include "mojo/edk/system/test_utils.h"
 #include "mojo/edk/system/waiter.h"
-#include "mojo/public/c/system/core.h"
 #include "mojo/public/c/system/data_pipe.h"
 #include "mojo/public/c/system/functions.h"
+#include "mojo/public/c/system/message_pipe.h"
 #include "mojo/public/cpp/system/macros.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
@@ -1482,6 +1482,86 @@ TEST_F(DataPipeTest, TwoPhaseMoreInvalidArguments) {
   ASSERT_EQ(1u * sizeof(int32_t), num_bytes);
 }
 
+// Test that a producer can be sent over a MP.
+TEST_F(DataPipeTest, SendProducer) {
+  const char kTestData[] = "hello world";
+  const uint32_t kTestDataSize = static_cast<uint32_t>(sizeof(kTestData));
+
+  const MojoCreateDataPipeOptions options = {
+      kSizeOfOptions,                           // |struct_size|.
+      MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE,  // |flags|.
+      1u,                                       // |element_num_bytes|.
+      1000u                                     // |capacity_num_bytes|.
+  };
+  ASSERT_EQ(MOJO_RESULT_OK, Create(&options));
+  MojoHandleSignalsState hss;
+
+  // Write some data.
+  uint32_t num_bytes = kTestDataSize;
+  ASSERT_EQ(MOJO_RESULT_OK, WriteData(kTestData, &num_bytes));
+  ASSERT_EQ(kTestDataSize, num_bytes);
+
+  // Wait for the data.
+  hss = MojoHandleSignalsState();
+  ASSERT_EQ(MOJO_RESULT_OK,
+            MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE,
+                     MOJO_DEADLINE_INDEFINITE, &hss));
+  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
+  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
+
+  // Check the data.
+  const void* read_buffer = nullptr;
+  num_bytes = 0u;
+  ASSERT_EQ(MOJO_RESULT_OK,
+            BeginReadData(&read_buffer, &num_bytes, false));
+  ASSERT_EQ(0, memcmp(read_buffer, kTestData, kTestDataSize));
+  EndReadData(num_bytes);
+
+  // Now send the producer over a MP so that it's serialized.
+  MojoHandle pipe0, pipe1;
+  ASSERT_EQ(MOJO_RESULT_OK,
+            MojoCreateMessagePipe(nullptr, &pipe0, &pipe1));
+
+  ASSERT_EQ(MOJO_RESULT_OK,
+            MojoWriteMessage(pipe0, nullptr, 0, &producer_, 1,
+                             MOJO_WRITE_MESSAGE_FLAG_NONE));
+  producer_ = MOJO_HANDLE_INVALID;
+  ASSERT_EQ(MOJO_RESULT_OK, MojoWait(pipe1, MOJO_HANDLE_SIGNAL_READABLE,
+                                     MOJO_DEADLINE_INDEFINITE, &hss));
+  uint32_t num_handles = 1;
+  ASSERT_EQ(MOJO_RESULT_OK,
+            MojoReadMessage(pipe1, nullptr, 0, &producer_, &num_handles,
+                            MOJO_READ_MESSAGE_FLAG_NONE));
+  ASSERT_EQ(num_handles, 1u);
+
+  // Write more data.
+  const char kExtraData[] = "bye world";
+  const uint32_t kExtraDataSize = static_cast<uint32_t>(sizeof(kExtraData));
+  num_bytes = kExtraDataSize;
+  ASSERT_EQ(MOJO_RESULT_OK, WriteData(kExtraData, &num_bytes));
+  ASSERT_EQ(kExtraDataSize, num_bytes);
+
+  // Wait for it.
+  hss = MojoHandleSignalsState();
+  ASSERT_EQ(MOJO_RESULT_OK,
+            MojoWait(consumer_, MOJO_HANDLE_SIGNAL_READABLE,
+                     MOJO_DEADLINE_INDEFINITE, &hss));
+  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
+  ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
+            hss.satisfiable_signals);
+
+  // Check the second write.
+  num_bytes = 0u;
+  ASSERT_EQ(MOJO_RESULT_OK,
+            BeginReadData(&read_buffer, &num_bytes, false));
+  ASSERT_EQ(0, memcmp(read_buffer, kExtraData, kExtraDataSize));
+  EndReadData(num_bytes);
+
+  ASSERT_EQ(MOJO_RESULT_OK, MojoClose(pipe0));
+  ASSERT_EQ(MOJO_RESULT_OK, MojoClose(pipe1));
+}
+
 // Ensures that if a data pipe consumer whose producer has closed is passed over
 // a message pipe, the deserialized dispatcher is also marked as having a closed
 // peer.