Ensure that DataPipeProducerDispatcher::CreateEquivalentDispatcherAndCloseImplNoLock serializes the…

mojo/edk/system/data_pipe_producer_dispatcher.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/edk/system/data_pipe_producer_dispatcher.h"
 
 #include "base/bind.h"
 #include "base/logging.h"
 #include "base/message_loop/message_loop.h"
 #include "mojo/edk/embedder/embedder_internal.h"
@@ skipping 63 matching lines @@
 
   if (platform_handle.is_valid()) {
     rv->Init(platform_handle.Pass(), serialized_write_buffer,
              serialized_write_buffer_size);
   }
   return rv;
 }
 
 DataPipeProducerDispatcher::DataPipeProducerDispatcher(
     const MojoCreateDataPipeOptions& options)
-    : options_(options), channel_(nullptr), error_(false) {
+    : options_(options), channel_(nullptr), error_(false), serialized_(false) {
 }
 
 DataPipeProducerDispatcher::~DataPipeProducerDispatcher() {
   // |Close()|/|CloseImplNoLock()| should have taken care of the channel.
   DCHECK(!channel_);
 }
 
 void DataPipeProducerDispatcher::CancelAllAwakablesNoLock() {
   lock().AssertAcquired();
   awakable_list_.CancelAll();
 }
 
 void DataPipeProducerDispatcher::CloseImplNoLock() {
   lock().AssertAcquired();
   internal::g_io_thread_task_runner->PostTask(
       FROM_HERE, base::Bind(&DataPipeProducerDispatcher::CloseOnIO, this));
 }
 
 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());
 }
 
 MojoResult DataPipeProducerDispatcher::WriteDataImplNoLock(
     const void* elements,
     uint32_t* num_bytes,
     MojoWriteDataFlags flags) {
   lock().AssertAcquired();
   if (InTwoPhaseWrite())
     return MOJO_RESULT_BUSY;
@@ skipping 128 matching lines @@
   lock().AssertAcquired();
   awakable_list_.Remove(awakable);
   if (signals_state)
     *signals_state = GetHandleSignalsStateImplNoLock();
 }
 
 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);
 }
 
 bool DataPipeProducerDispatcher::EndSerializeAndCloseImplNoLock(
     void* destination,
     size_t* actual_size,
     PlatformHandleVector* platform_handles) {
   DCHECK(HasOneRef());  // Only one ref => no need to take the lock.
@@ skipping 94 matching lines @@
       error_ = true;
       return false;
     }
 
     offset += message_num_bytes;
   }
 
   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