Make the data pipe "cancel all state" functions cancel two-phase operations.

mojo/edk/system/data_pipe.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.h"
 
 #include <string.h>
 
 #include <algorithm>
 #include <limits>
@@ skipping 254 matching lines @@
   *data_pipe =
       incoming_endpoint->ConvertToDataPipeConsumer(revalidated_options);
   if (!*data_pipe)
     return false;
 
   return true;
 }
 
 void DataPipe::ProducerCancelAllState() {
   MutexLocker locker(&mutex_);
-  DCHECK(has_local_producer_no_lock());
-  producer_awakable_list_->CancelAll();
+  ProducerCancelAllStateNoLock();
 }
 
 void DataPipe::ProducerClose() {
   MutexLocker locker(&mutex_);
   ProducerCloseNoLock();
 }
 
 MojoResult DataPipe::ProducerSetOptions(uint32_t write_threshold_num_bytes) {
   MutexLocker locker(&mutex_);
   DCHECK(has_local_producer_no_lock());
@@ skipping 156 matching lines @@
     void* destination,
     size_t* actual_size,
     std::vector<ScopedPlatformHandle>* platform_handles) {
   MutexLocker locker(&mutex_);
   DCHECK(has_local_producer_no_lock());
   // Warning: After |ProducerEndSerialize()|, quite probably |impl_| has
   // changed.
   bool rv = impl_->ProducerEndSerialize(channel, destination, actual_size,
                                         platform_handles);
 
-  // TODO(vtl): The code below is similar to, but not quite the same as,
-  // |ProducerCloseNoLock()|.
-  DCHECK(has_local_producer_no_lock());
-  producer_awakable_list_->CancelAll();
+  ProducerCancelAllStateNoLock();
   producer_awakable_list_.reset();
-  // Not a bug, except possibly in "user" code.
-  DVLOG_IF(2, producer_in_two_phase_write_no_lock())
-      << "Producer transferred with active two-phase write";
-  producer_two_phase_max_num_bytes_written_ = 0;
   if (!has_local_consumer_no_lock())
     producer_open_ = false;
 
   return rv;
 }
 
 bool DataPipe::ProducerIsBusy() const {
   MutexLocker locker(&mutex_);
   return producer_in_two_phase_write_no_lock();
 }
 
 void DataPipe::ConsumerCancelAllState() {
   MutexLocker locker(&mutex_);
-  DCHECK(has_local_consumer_no_lock());
-  consumer_awakable_list_->CancelAll();
+  ConsumerCancelAllStateNoLock();
 }
 
 void DataPipe::ConsumerClose() {
   MutexLocker locker(&mutex_);
   ConsumerCloseNoLock();
 }
 
 MojoResult DataPipe::ConsumerSetOptions(uint32_t read_threshold_num_bytes) {
   MutexLocker locker(&mutex_);
   DCHECK(has_local_consumer_no_lock());
@@ skipping 188 matching lines @@
     void* destination,
     size_t* actual_size,
     std::vector<ScopedPlatformHandle>* platform_handles) {
   MutexLocker locker(&mutex_);
   DCHECK(has_local_consumer_no_lock());
   // Warning: After |ConsumerEndSerialize()|, quite probably |impl_| has
   // changed.
   bool rv = impl_->ConsumerEndSerialize(channel, destination, actual_size,
                                         platform_handles);
 
-  // TODO(vtl): The code below is similar to, but not quite the same as,
-  // |ConsumerCloseNoLock()|.
-  consumer_awakable_list_->CancelAll();
+  ConsumerCancelAllStateNoLock();
   consumer_awakable_list_.reset();
-  // Not a bug, except possibly in "user" code.
-  DVLOG_IF(2, consumer_in_two_phase_read_no_lock())
-      << "Consumer transferred with active two-phase read";
-  consumer_two_phase_max_num_bytes_read_ = 0;
   if (!has_local_producer_no_lock())
     consumer_open_ = false;
 
   return rv;
 }
 
 bool DataPipe::ConsumerIsBusy() const {
   MutexLocker locker(&mutex_);
   return consumer_in_two_phase_read_no_lock();
 }
@@ skipping 37 matching lines @@
   mutex_.AssertHeld();
   DCHECK(new_impl);
 
   impl_->set_owner(nullptr);
   std::unique_ptr<DataPipeImpl> rv(std::move(impl_));
   impl_ = std::move(new_impl);
   impl_->set_owner(this);
   return rv;
 }
 
+void DataPipe::ProducerCancelAllStateNoLock() {
+  DCHECK(has_local_producer_no_lock());
+  if (producer_awakable_list_)
+    producer_awakable_list_->CancelAll();
+  // Not a bug, except possibly in "user" code.
+  DVLOG_IF(2, producer_in_two_phase_write_no_lock())
+      << "Active two-phase write cancelled";
+  producer_two_phase_max_num_bytes_written_ = 0;
+}
+
+void DataPipe::ConsumerCancelAllStateNoLock() {
+  DCHECK(has_local_consumer_no_lock());
+  if (consumer_awakable_list_)
+    consumer_awakable_list_->CancelAll();
+  // Not a bug, except possibly in "user" code.
+  DVLOG_IF(2, consumer_in_two_phase_read_no_lock())
+      << "Active two-phase read cancelled";
+  consumer_two_phase_max_num_bytes_read_ = 0;
+}
+
 void DataPipe::SetProducerClosedNoLock() {
   mutex_.AssertHeld();
   DCHECK(!has_local_producer_no_lock());
   DCHECK(producer_open_);
   producer_open_ = false;
 }
 
 void DataPipe::SetConsumerClosedNoLock() {
   mutex_.AssertHeld();
   DCHECK(!has_local_consumer_no_lock());
   DCHECK(consumer_open_);
   consumer_open_ = false;
 }
 
 void DataPipe::ProducerCloseNoLock() {
   mutex_.AssertHeld();
   DCHECK(producer_open_);
   producer_open_ = false;
   if (has_local_producer_no_lock()) {
+    ProducerCancelAllStateNoLock();
     producer_awakable_list_.reset();
-    // Not a bug, except possibly in "user" code.
-    DVLOG_IF(2, producer_in_two_phase_write_no_lock())
-        << "Producer closed with active two-phase write";
-    producer_two_phase_max_num_bytes_written_ = 0;
     impl_->ProducerClose();
     AwakeConsumerAwakablesForStateChangeNoLock(
         impl_->ConsumerGetHandleSignalsState());
   }
 }
 
 void DataPipe::ConsumerCloseNoLock() {
   mutex_.AssertHeld();
   DCHECK(consumer_open_);
   consumer_open_ = false;
   if (has_local_consumer_no_lock()) {
+    ConsumerCancelAllStateNoLock();
     consumer_awakable_list_.reset();
-    // Not a bug, except possibly in "user" code.
-    DVLOG_IF(2, consumer_in_two_phase_read_no_lock())
-        << "Consumer closed with active two-phase read";
-    consumer_two_phase_max_num_bytes_read_ = 0;
     impl_->ConsumerClose();
     AwakeProducerAwakablesForStateChangeNoLock(
         impl_->ProducerGetHandleSignalsState());
   }
 }
 
 bool DataPipe::OnReadMessage(unsigned port, MessageInTransit* message) {
   MutexLocker locker(&mutex_);
   DCHECK(!has_local_producer_no_lock() || !has_local_consumer_no_lock());
 
@@ skipping 58 matching lines @@
   SetProducerClosedNoLock();
 }
 
 void DataPipe::SetConsumerClosed() {
   MutexLocker locker(&mutex_);
   SetConsumerClosedNoLock();
 }
 
 }  // namespace system
 }  // namespace mojo