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

mojo/edk/system/data_pipe.cc

Index: mojo/edk/system/data_pipe.cc
diff --git a/mojo/edk/system/data_pipe.cc b/mojo/edk/system/data_pipe.cc
index 7657f644c1140bad11782a87de3b2d93ca41eeb0..d671a7625e93043f18e3a09f7aea4e76aeaf5b1d 100644
--- a/mojo/edk/system/data_pipe.cc
+++ b/mojo/edk/system/data_pipe.cc
@@ -272,8 +272,7 @@ bool DataPipe::ConsumerDeserialize(Channel* channel,
 
 void DataPipe::ProducerCancelAllState() {
   MutexLocker locker(&mutex_);
-  DCHECK(has_local_producer_no_lock());
-  producer_awakable_list_->CancelAll();
+  ProducerCancelAllStateNoLock();
 }
 
 void DataPipe::ProducerClose() {
@@ -450,15 +449,8 @@ bool DataPipe::ProducerEndSerialize(
   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;
 
@@ -472,8 +464,7 @@ bool DataPipe::ProducerIsBusy() const {
 
 void DataPipe::ConsumerCancelAllState() {
   MutexLocker locker(&mutex_);
-  DCHECK(has_local_consumer_no_lock());
-  consumer_awakable_list_->CancelAll();
+  ConsumerCancelAllStateNoLock();
 }
 
 void DataPipe::ConsumerClose() {
@@ -682,14 +673,8 @@ bool DataPipe::ConsumerEndSerialize(
   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;
 
@@ -747,6 +732,26 @@ std::unique_ptr<DataPipeImpl> DataPipe::ReplaceImplNoLock(
   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());
@@ -766,11 +771,8 @@ void DataPipe::ProducerCloseNoLock() {
   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());
@@ -782,11 +784,8 @@ void DataPipe::ConsumerCloseNoLock() {
   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());