Last set of fixes to make the src/mojo/edk pass the page cycler.

mojo/edk/system/raw_channel.cc

 // Copyright 2014 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/raw_channel.h"
 
 #include <string.h>
 
 #include <algorithm>
 
@@ skipping 152 matching lines @@
 }
 
 // RawChannel ------------------------------------------------------------------
 
 RawChannel::RawChannel()
     : message_loop_for_io_(nullptr),
       delegate_(nullptr),
       write_ready_(false),
       write_stopped_(false),
       error_occurred_(false),
+      pending_error_(false),
       weak_ptr_factory_(this) {
   read_buffer_.reset(new ReadBuffer);
   write_buffer_.reset(new WriteBuffer());
 }
 
 RawChannel::~RawChannel() {
   DCHECK(!read_buffer_);
   DCHECK(!write_buffer_);
 
   // Only want to decrement counter if Init was called.
@@ skipping 140 matching lines @@
 
 bool RawChannel::SendQueuedMessagesNoLock() {
   DCHECK_EQ(write_buffer_->data_offset_, 0u);
 
   size_t platform_handles_written = 0;
   size_t bytes_written = 0;
   IOResult io_result = WriteNoLock(&platform_handles_written, &bytes_written);
   if (io_result == IO_PENDING)
     return true;
 
-  bool result = OnWriteCompletedNoLock(io_result, platform_handles_written,
-                                       bytes_written);
+  bool result = OnWriteCompletedInternalNoLock(
+      io_result, platform_handles_written, bytes_written);
   if (!result) {
     // Even if we're on the I/O thread, don't call |OnError()| in the nested
     // context.
+    pending_error_ = true;

[yzshen1, 2015/10/14 18:23:45]

pending_error_ is never reset to false, is that intended?

[jam, 2015/10/14 20:01:44]

yep

     message_loop_for_io_->PostTask(
         FROM_HERE,
         base::Bind(&RawChannel::LockAndCallOnError,
                    weak_ptr_factory_.GetWeakPtr(),
                    Delegate::ERROR_WRITE));
   }
 
   return result;
 }
 
@@ skipping 22 matching lines @@
       scoped_ptr<MessageInTransit> message(new MessageInTransit(
           MessageInTransit::Type::RAW_MESSAGE, message_num_bytes,
           static_cast<const char*>(serialized_write_buffer) + offset));
       write_buffer_->message_queue_.AddMessage(message.Pass());
       offset += message_num_bytes;
     }
   }
 }
 
 void RawChannel::OnReadCompleted(IOResult io_result, size_t bytes_read) {
+  base::AutoLock locker(read_lock_);
+  OnReadCompletedNoLock(io_result, bytes_read);
+}
+
+void RawChannel::OnReadCompletedNoLock(IOResult io_result, size_t bytes_read) {
   DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io_);
-
-  base::AutoLock locker(read_lock_);
-
+  read_lock_.AssertAcquired();
   // Keep reading data in a loop, and dispatch messages if enough data is
   // received. Exit the loop if any of the following happens:
   //   - one or more messages were dispatched;
   //   - the last read failed, was a partial read or would block;
   //   - |Shutdown()| was called.
   do {
     switch (io_result) {
       case IO_SUCCEEDED:
         break;
       case IO_FAILED_SHUTDOWN:
@@ skipping 39 matching lines @@
     // (2) If we didn't max out |kReadSize|, stop reading for now.
     bool schedule_for_later = did_dispatch_message || bytes_read < kReadSize;
     bytes_read = 0;
     io_result = schedule_for_later ? ScheduleRead() : Read(&bytes_read);
   } while (io_result != IO_PENDING);
 }
 
 void RawChannel::OnWriteCompleted(IOResult io_result,
                                   size_t platform_handles_written,
                                   size_t bytes_written) {
+  base::AutoLock locker(write_lock_);
+  OnWriteCompletedNoLock(io_result, platform_handles_written, bytes_written);
+}
+
+void RawChannel::OnWriteCompletedNoLock(IOResult io_result,
+                                        size_t platform_handles_written,
+                                        size_t bytes_written) {
   DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io_);
+  write_lock_.AssertAcquired();
   DCHECK_NE(io_result, IO_PENDING);
 
-  bool did_fail = false;
-  {
-    base::AutoLock locker(write_lock_);
-    did_fail = !OnWriteCompletedNoLock(io_result, platform_handles_written,
-                                       bytes_written);
-  }
-
-  if (did_fail) {
+  bool did_fail = !OnWriteCompletedInternalNoLock(
+      io_result, platform_handles_written, bytes_written);
+  if (did_fail)
     LockAndCallOnError(Delegate::ERROR_WRITE);
-    return;  // |this| may have been destroyed in |CallOnError()|.
-  }
 }
 
 void RawChannel::SerializeReadBuffer(size_t additional_bytes_read,
                                      std::vector<char>* buffer) {
   read_buffer_->num_valid_bytes_ += additional_bytes_read;
   read_buffer_->buffer_.resize(read_buffer_->num_valid_bytes_);
   read_buffer_->buffer_.swap(*buffer);
   read_buffer_->num_valid_bytes_ = 0;
 }
 
 void RawChannel::SerializeWriteBuffer(
-    std::vector<char>* buffer,
     size_t additional_bytes_written,
-    size_t additional_platform_handles_written) {
+    size_t additional_platform_handles_written,
+    std::vector<char>* buffer) {
   if (write_buffer_->IsEmpty()) {
     DCHECK_EQ(0u, additional_bytes_written);
     DCHECK_EQ(0u, additional_platform_handles_written);
     return;
   }
 
   UpdateWriteBuffer(
       additional_platform_handles_written, additional_bytes_written);
   while (!write_buffer_->message_queue_.IsEmpty()) {
     SerializePlatformHandles();
     std::vector<WriteBuffer::Buffer> buffers;
     write_buffer_no_lock()->GetBuffers(&buffers);
     for (size_t i = 0; i < buffers.size(); ++i) {
       buffer->insert(buffer->end(), buffers[i].addr,
                      buffers[i].addr + buffers[i].size);
     }
     write_buffer_->message_queue_.DiscardMessage();
   }
 }
 
 void RawChannel::EnqueueMessageNoLock(scoped_ptr<MessageInTransit> message) {
   write_lock_.AssertAcquired();
   DCHECK(HandleForDebuggingNoLock().is_valid());
   write_buffer_->message_queue_.AddMessage(message.Pass());
 }
 
 bool RawChannel::OnReadMessageForRawChannel(
     const MessageInTransit::View& message_view) {
   if (message_view.type() == MessageInTransit::Type::RAW_CHANNEL_QUIT) {
+    pending_error_ = true;
     message_loop_for_io_->PostTask(
         FROM_HERE, base::Bind(&RawChannel::LockAndCallOnError,
                               weak_ptr_factory_.GetWeakPtr(),
                               Delegate::ERROR_READ_SHUTDOWN));
     return true;
   }
 
   // No non-implementation specific |RawChannel| control messages.
   LOG(ERROR) << "Invalid control message (type " << message_view.type()
              << ")";
@@ skipping 13 matching lines @@
     case IO_PENDING:
       NOTREACHED();
       break;
   }
   return Delegate::ERROR_READ_UNKNOWN;
 }
 
 void RawChannel::CallOnError(Delegate::Error error) {
   DCHECK_EQ(base::MessageLoop::current(), message_loop_for_io_);
   read_lock_.AssertAcquired();
+  bool called_on_error_before = error_occurred_;
   error_occurred_ = true;
   if (delegate_) {
-    delegate_->OnError(error);
+    // We only want to call OnError once. The delegate may have posted a task
+    // to call Shutdown though, so we don't want to null delegate_ the first
+    // time and have two shutdown calls.
+    if (!called_on_error_before)
+      delegate_->OnError(error);
   } else {
     // We depend on delegate to delete since it could be waiting to call
     // ReleaseHandle.
     base::MessageLoop::current()->PostTask(
         FROM_HERE,
         base::Bind(&RawChannel::Shutdown, weak_ptr_factory_.GetWeakPtr()));
   }
 }
 
 void RawChannel::LockAndCallOnError(Delegate::Error error) {
   base::AutoLock locker(read_lock_);
   CallOnError(error);
 }
 
-bool RawChannel::OnWriteCompletedNoLock(IOResult io_result,
-                                        size_t platform_handles_written,
-                                        size_t bytes_written) {
+bool RawChannel::OnWriteCompletedInternalNoLock(IOResult io_result,
+                                                size_t platform_handles_written,
+                                                size_t bytes_written) {
   write_lock_.AssertAcquired();
 
   DCHECK(!write_buffer_->message_queue_.IsEmpty());
 
   if (io_result == IO_SUCCEEDED) {
     UpdateWriteBuffer(platform_handles_written, bytes_written);
     if (write_buffer_->message_queue_.IsEmpty())
       return true;
 
     // Schedule the next write.
@@ skipping 112 matching lines @@
     // Complete write.
     CHECK_EQ(write_buffer_->data_offset_, message->total_size());
     write_buffer_->message_queue_.DiscardMessage();
     write_buffer_->platform_handles_offset_ = 0;
     write_buffer_->data_offset_ = 0;
   }
 }
 
 }  // namespace edk
 }  // namespace mojo