Mojo: Plumb wait flags state through DataPipe::Awake...WaitersForStateChangeNoLock().

mojo/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/system/data_pipe.h"
 
 #include <string.h>
 
 #include <algorithm>
 #include <limits>
@@ skipping 72 matching lines @@
   base::AutoLock locker(lock_);
   DCHECK(producer_open_);
   producer_open_ = false;
   DCHECK(has_local_producer_no_lock());
   producer_waiter_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;
   ProducerCloseImplNoLock();
-  AwakeConsumerWaitersForStateChangeNoLock();
+  AwakeConsumerWaitersForStateChangeNoLock(ConsumerGetWaitFlagsStateNoLock());
 }
 
 MojoResult DataPipe::ProducerWriteData(const void* elements,
                                        uint32_t* num_bytes,
                                        bool all_or_none) {
   base::AutoLock locker(lock_);
   DCHECK(has_local_producer_no_lock());
 
   if (producer_in_two_phase_write_no_lock())
     return MOJO_RESULT_BUSY;
   if (!consumer_open_no_lock())
     return MOJO_RESULT_FAILED_PRECONDITION;
 
   // Returning "busy" takes priority over "invalid argument".
   if (*num_bytes % element_num_bytes_ != 0)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   if (*num_bytes == 0)
     return MOJO_RESULT_OK;  // Nothing to do.
 
   WaitFlagsState old_consumer_state = ConsumerGetWaitFlagsStateNoLock();
   MojoResult rv = ProducerWriteDataImplNoLock(elements, num_bytes, all_or_none);
   WaitFlagsState new_consumer_state = ConsumerGetWaitFlagsStateNoLock();
   if (!new_consumer_state.equals(old_consumer_state))
-    AwakeConsumerWaitersForStateChangeNoLock();
+    AwakeConsumerWaitersForStateChangeNoLock(new_consumer_state);
   return rv;
 }
 
 MojoResult DataPipe::ProducerBeginWriteData(void** buffer,
                                             uint32_t* buffer_num_bytes,
                                             bool all_or_none) {
   base::AutoLock locker(lock_);
   DCHECK(has_local_producer_no_lock());
 
   if (producer_in_two_phase_write_no_lock())
@@ skipping 31 matching lines @@
       num_bytes_written % element_num_bytes_ != 0) {
     rv = MOJO_RESULT_INVALID_ARGUMENT;
     producer_two_phase_max_num_bytes_written_ = 0;
   } else {
     rv = ProducerEndWriteDataImplNoLock(num_bytes_written);
   }
   // Two-phase write ended even on failure.
   DCHECK(!producer_in_two_phase_write_no_lock());
   // If we're now writable, we *became* writable (since we weren't writable
   // during the two-phase write), so awake producer waiters.
-  if (ProducerGetWaitFlagsStateNoLock().satisfies(MOJO_WAIT_FLAG_WRITABLE))
-    AwakeProducerWaitersForStateChangeNoLock();
+  WaitFlagsState new_producer_state = ProducerGetWaitFlagsStateNoLock();
+  if (new_producer_state.satisfies(MOJO_WAIT_FLAG_WRITABLE))
+    AwakeProducerWaitersForStateChangeNoLock(new_producer_state);
   WaitFlagsState new_consumer_state = ConsumerGetWaitFlagsStateNoLock();
   if (!new_consumer_state.equals(old_consumer_state))
-    AwakeConsumerWaitersForStateChangeNoLock();
+    AwakeConsumerWaitersForStateChangeNoLock(new_consumer_state);
   return rv;
 }
 
 MojoResult DataPipe::ProducerAddWaiter(Waiter* waiter,
                                        MojoWaitFlags flags,
                                        MojoResult wake_result) {
   base::AutoLock locker(lock_);
   DCHECK(has_local_producer_no_lock());
 
   WaitFlagsState producer_state = ProducerGetWaitFlagsStateNoLock();
@@ skipping 27 matching lines @@
   base::AutoLock locker(lock_);
   DCHECK(consumer_open_);
   consumer_open_ = false;
   DCHECK(has_local_consumer_no_lock());
   consumer_waiter_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;
   ConsumerCloseImplNoLock();
-  AwakeProducerWaitersForStateChangeNoLock();
+  AwakeProducerWaitersForStateChangeNoLock(ProducerGetWaitFlagsStateNoLock());
 }
 
 MojoResult DataPipe::ConsumerReadData(void* elements,
                                       uint32_t* num_bytes,
                                       bool all_or_none) {
   base::AutoLock locker(lock_);
   DCHECK(has_local_consumer_no_lock());
 
   if (consumer_in_two_phase_read_no_lock())
     return MOJO_RESULT_BUSY;
 
   if (*num_bytes % element_num_bytes_ != 0)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   if (*num_bytes == 0)
     return MOJO_RESULT_OK;  // Nothing to do.
 
   WaitFlagsState old_producer_state = ProducerGetWaitFlagsStateNoLock();
   MojoResult rv = ConsumerReadDataImplNoLock(elements, num_bytes, all_or_none);
   WaitFlagsState new_producer_state = ProducerGetWaitFlagsStateNoLock();
   if (!new_producer_state.equals(old_producer_state))
-    AwakeProducerWaitersForStateChangeNoLock();
+    AwakeProducerWaitersForStateChangeNoLock(new_producer_state);
   return rv;
 }
 
 MojoResult DataPipe::ConsumerDiscardData(uint32_t* num_bytes,
                                          bool all_or_none) {
   base::AutoLock locker(lock_);
   DCHECK(has_local_consumer_no_lock());
 
   if (consumer_in_two_phase_read_no_lock())
     return MOJO_RESULT_BUSY;
 
   if (*num_bytes % element_num_bytes_ != 0)
     return MOJO_RESULT_INVALID_ARGUMENT;
 
   if (*num_bytes == 0)
     return MOJO_RESULT_OK;  // Nothing to do.
 
   WaitFlagsState old_producer_state = ProducerGetWaitFlagsStateNoLock();
   MojoResult rv = ConsumerDiscardDataImplNoLock(num_bytes, all_or_none);
   WaitFlagsState new_producer_state = ProducerGetWaitFlagsStateNoLock();
   if (!new_producer_state.equals(old_producer_state))
-    AwakeProducerWaitersForStateChangeNoLock();
+    AwakeProducerWaitersForStateChangeNoLock(new_producer_state);
   return rv;
 }
 
 MojoResult DataPipe::ConsumerQueryData(uint32_t* num_bytes) {
   base::AutoLock locker(lock_);
   DCHECK(has_local_consumer_no_lock());
 
   if (consumer_in_two_phase_read_no_lock())
     return MOJO_RESULT_BUSY;
 
@@ skipping 36 matching lines @@
     consumer_two_phase_max_num_bytes_read_ = 0;
   } else {
     rv = ConsumerEndReadDataImplNoLock(num_bytes_read);
   }
   // Two-phase read ended even on failure.
   DCHECK(!consumer_in_two_phase_read_no_lock());
   // If we're now readable, we *became* readable (since we weren't readable
   // during the two-phase read), so awake consumer waiters.
   WaitFlagsState new_consumer_state = ConsumerGetWaitFlagsStateNoLock();
   if (new_consumer_state.satisfies(MOJO_WAIT_FLAG_READABLE))
-    AwakeConsumerWaitersForStateChangeNoLock();
+    AwakeConsumerWaitersForStateChangeNoLock(new_consumer_state);
   WaitFlagsState new_producer_state = ProducerGetWaitFlagsStateNoLock();
   if (!new_producer_state.equals(old_producer_state))
-    AwakeProducerWaitersForStateChangeNoLock();
+    AwakeProducerWaitersForStateChangeNoLock(new_producer_state);
   return rv;
 }
 
 MojoResult DataPipe::ConsumerAddWaiter(Waiter* waiter,
                                        MojoWaitFlags flags,
                                        MojoResult wake_result) {
   base::AutoLock locker(lock_);
   DCHECK(has_local_consumer_no_lock());
 
   WaitFlagsState consumer_state = ConsumerGetWaitFlagsStateNoLock();
@@ skipping 36 matching lines @@
   DCHECK_EQ(ValidateCreateOptions(&validated_options, &unused), MOJO_RESULT_OK);
 }
 
 DataPipe::~DataPipe() {
   DCHECK(!producer_open_);
   DCHECK(!consumer_open_);
   DCHECK(!producer_waiter_list_);
   DCHECK(!consumer_waiter_list_);
 }
 
-void DataPipe::AwakeProducerWaitersForStateChangeNoLock() {
+void DataPipe::AwakeProducerWaitersForStateChangeNoLock(
+    const WaitFlagsState& new_producer_state) {
   lock_.AssertAcquired();
   if (!has_local_producer_no_lock())
     return;
-  producer_waiter_list_->AwakeWaitersForStateChange(
-      ProducerGetWaitFlagsStateNoLock());
+  producer_waiter_list_->AwakeWaitersForStateChange(new_producer_state);
 }
 
-void DataPipe::AwakeConsumerWaitersForStateChangeNoLock() {
+void DataPipe::AwakeConsumerWaitersForStateChangeNoLock(
+    const WaitFlagsState& new_consumer_state) {
   lock_.AssertAcquired();
   if (!has_local_consumer_no_lock())
     return;
-  consumer_waiter_list_->AwakeWaitersForStateChange(
-      ConsumerGetWaitFlagsStateNoLock());
+  consumer_waiter_list_->AwakeWaitersForStateChange(new_consumer_state);
 }
 
 }  // namespace system
 }  // namespace mojo