Mojo: DataPipe: Implement "may discard" for two-phase writes.

mojo/system/local_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.
 
 // TODO(vtl): I currently potentially overflow in doing index calculations.
 // E.g., |start_index_| and |current_num_bytes_| fit into a |uint32_t|, but
 // their sum may not. This is bad and poses a security risk. (We're currently
 // saved by the limit on capacity -- the maximum size of the buffer, checked in
 // |DataPipe::ValidateOptions()|, is currently sufficiently small.
 
@@ skipping 24 matching lines @@
   // If the consumer is still open and we still have data, we have to keep the
   // buffer around. Currently, we won't free it even if it empties later. (We
   // could do this -- requiring a check on every read -- but that seems to be
   // optimizing for the uncommon case.)
   if (!consumer_open_no_lock() || !current_num_bytes_) {
     // Note: There can only be a two-phase *read* (by the consumer) if we still
     // have data.
     DCHECK(!consumer_in_two_phase_read_no_lock());
     DestroyBufferNoLock();
   }
-  AwakeConsumerWaitersForStateChangeNoLock();
 }
 
 MojoResult LocalDataPipe::ProducerWriteDataImplNoLock(const void* elements,
                                                       uint32_t* num_bytes,
                                                       bool all_or_none) {
   DCHECK_EQ(*num_bytes % element_num_bytes(), 0u);
   DCHECK_GT(*num_bytes, 0u);
   DCHECK(consumer_open_no_lock());
 
   size_t num_bytes_to_write = 0;
   if (may_discard()) {
     if (all_or_none && *num_bytes > capacity_num_bytes())
       return MOJO_RESULT_OUT_OF_RANGE;
 
     num_bytes_to_write = std::min(static_cast<size_t>(*num_bytes),
                                   capacity_num_bytes());
     if (num_bytes_to_write > capacity_num_bytes() - current_num_bytes_) {
       // Discard as much as needed (discard oldest first).
-      size_t num_bytes_to_discard =
-          num_bytes_to_write - (capacity_num_bytes() - current_num_bytes_);
-      start_index_ += num_bytes_to_discard;
-      start_index_ %= capacity_num_bytes();
-      current_num_bytes_ -= num_bytes_to_discard;
+      MarkDataAsConsumedNoLock(
+          num_bytes_to_write - (capacity_num_bytes() - current_num_bytes_));
+      // No need to wake up write waiters, since we're definitely going to leave
+      // the buffer full.
     }
   } else {
     if (all_or_none && *num_bytes > capacity_num_bytes() - current_num_bytes_) {
       // Don't return "should wait" since you can't wait for a specified amount
       // of data.
       return MOJO_RESULT_OUT_OF_RANGE;
     }
 
     num_bytes_to_write = std::min(static_cast<size_t>(*num_bytes),
                                   capacity_num_bytes() - current_num_bytes_);
@@ skipping 10 matching lines @@
   EnsureBufferNoLock();
   memcpy(buffer_.get() + first_write_index, elements, num_bytes_to_write_first);
 
   if (num_bytes_to_write_first < num_bytes_to_write) {
     // The "second write index" is zero.
     memcpy(buffer_.get(),
            static_cast<const char*>(elements) + num_bytes_to_write_first,
            num_bytes_to_write - num_bytes_to_write_first);
   }
 
-  bool was_empty = (current_num_bytes_ == 0);
-
   current_num_bytes_ += num_bytes_to_write;
   DCHECK_LE(current_num_bytes_, capacity_num_bytes());
-
-  if (was_empty && num_bytes_to_write > 0)
-    AwakeConsumerWaitersForStateChangeNoLock();
-
   *num_bytes = static_cast<uint32_t>(num_bytes_to_write);
   return MOJO_RESULT_OK;
 }
 
 MojoResult LocalDataPipe::ProducerBeginWriteDataImplNoLock(
     void** buffer,
     uint32_t* buffer_num_bytes,
     bool all_or_none) {
   DCHECK(consumer_open_no_lock());
 
+  // The index we need to start writing at.
+  size_t write_index =
+      (start_index_ + current_num_bytes_) % capacity_num_bytes();
+
   size_t max_num_bytes_to_write = GetMaxNumBytesToWriteNoLock();
   if (all_or_none && *buffer_num_bytes > max_num_bytes_to_write) {
-    // Don't return "should wait" since you can't wait for a specified amount of
-    // data.
-    return MOJO_RESULT_OUT_OF_RANGE;
+    // In "may discard" mode, we can always write from the write index to the
+    // end of the buffer.
+    if (may_discard() &&
+        *buffer_num_bytes <= capacity_num_bytes() - write_index) {
+      // To do so, we need to discard an appropriate amount of data.
+      // We should only reach here if the start index is after the write index!
+      DCHECK_GE(start_index_, write_index);
+      DCHECK_GT(*buffer_num_bytes - max_num_bytes_to_write, 0u);
+      MarkDataAsConsumedNoLock(*buffer_num_bytes - max_num_bytes_to_write);
+      max_num_bytes_to_write = *buffer_num_bytes;
+    } else {
+      // Don't return "should wait" since you can't wait for a specified amount
+      // of data.
+      return MOJO_RESULT_OUT_OF_RANGE;
+    }
   }
 
   // Don't go into a two-phase write if there's no room.
   if (max_num_bytes_to_write == 0)
     return MOJO_RESULT_SHOULD_WAIT;
 
-  size_t write_index =
-      (start_index_ + current_num_bytes_) % capacity_num_bytes();
   EnsureBufferNoLock();
   *buffer = buffer_.get() + write_index;
   *buffer_num_bytes = static_cast<uint32_t>(max_num_bytes_to_write);
   set_producer_two_phase_max_num_bytes_written_no_lock(
       static_cast<uint32_t>(max_num_bytes_to_write));
   return MOJO_RESULT_OK;
 }
 
 MojoResult LocalDataPipe::ProducerEndWriteDataImplNoLock(
     uint32_t num_bytes_written) {
   if (num_bytes_written > producer_two_phase_max_num_bytes_written_no_lock()) {
     // Note: The two-phase write ends here even on failure.
     set_producer_two_phase_max_num_bytes_written_no_lock(0);
     return MOJO_RESULT_INVALID_ARGUMENT;
   }
 
-  bool was_empty = (current_num_bytes_ == 0);
-
   current_num_bytes_ += num_bytes_written;
   DCHECK_LE(current_num_bytes_, capacity_num_bytes());
   set_producer_two_phase_max_num_bytes_written_no_lock(0);
-
-  if (was_empty && num_bytes_written > 0)
-    AwakeConsumerWaitersForStateChangeNoLock();
-
   return MOJO_RESULT_OK;
 }
 
 MojoWaitFlags LocalDataPipe::ProducerSatisfiedFlagsNoLock() {
   MojoWaitFlags rv = MOJO_WAIT_FLAG_NONE;
-  if (consumer_open_no_lock() && current_num_bytes_ < capacity_num_bytes())
+  if (consumer_open_no_lock() &&
+      (may_discard() || current_num_bytes_ < capacity_num_bytes()) &&
+      !producer_in_two_phase_write_no_lock())
     rv |= MOJO_WAIT_FLAG_WRITABLE;
   return rv;
 }
 
 MojoWaitFlags LocalDataPipe::ProducerSatisfiableFlagsNoLock() {
   MojoWaitFlags rv = MOJO_WAIT_FLAG_NONE;
   if (consumer_open_no_lock())
     rv |= MOJO_WAIT_FLAG_WRITABLE;
   return rv;
 }
 
 void LocalDataPipe::ConsumerCloseImplNoLock() {
   // If the producer is around and in a two-phase write, we have to keep the
   // buffer around. (We then don't free it until the producer is closed. This
   // could be rectified, but again seems like optimizing for the uncommon case.)
   if (!producer_open_no_lock() || !producer_in_two_phase_write_no_lock())
     DestroyBufferNoLock();
   current_num_bytes_ = 0;
-  AwakeProducerWaitersForStateChangeNoLock();
 }
 
 MojoResult LocalDataPipe::ConsumerReadDataImplNoLock(void* elements,
                                                      uint32_t* num_bytes,
                                                      bool all_or_none) {
   DCHECK_EQ(*num_bytes % element_num_bytes(), 0u);
   DCHECK_GT(*num_bytes, 0u);
 
   if (all_or_none && *num_bytes > current_num_bytes_) {
     // Don't return "should wait" since you can't wait for a specified amount of
@@ skipping 14 matching lines @@
       std::min(num_bytes_to_read, GetMaxNumBytesToReadNoLock());
   memcpy(elements, buffer_.get() + start_index_, num_bytes_to_read_first);
 
   if (num_bytes_to_read_first < num_bytes_to_read) {
     // The "second read index" is zero.
     memcpy(static_cast<char*>(elements) + num_bytes_to_read_first,
            buffer_.get(),
            num_bytes_to_read - num_bytes_to_read_first);
   }
 
-  bool was_full = (current_num_bytes_ == capacity_num_bytes());
-
-  start_index_ += num_bytes_to_read;
-  start_index_ %= capacity_num_bytes();
-  current_num_bytes_ -= num_bytes_to_read;
-
-  if (was_full && num_bytes_to_read > 0)
-    AwakeProducerWaitersForStateChangeNoLock();
-
+  MarkDataAsConsumedNoLock(num_bytes_to_read);
   *num_bytes = static_cast<uint32_t>(num_bytes_to_read);
   return MOJO_RESULT_OK;
 }
 
 MojoResult LocalDataPipe::ConsumerDiscardDataImplNoLock(uint32_t* num_bytes,
                                                         bool all_or_none) {
   DCHECK_EQ(*num_bytes % element_num_bytes(), 0u);
   DCHECK_GT(*num_bytes, 0u);
 
   if (all_or_none && *num_bytes > current_num_bytes_) {
     // Don't return "should wait" since you can't wait for a specified amount of
     // data.
     return producer_open_no_lock() ? MOJO_RESULT_OUT_OF_RANGE :
                                      MOJO_RESULT_FAILED_PRECONDITION;
   }
 
   // Be consistent with other operations; error if no data available.
   if (current_num_bytes_ == 0) {
     return producer_open_no_lock() ? MOJO_RESULT_SHOULD_WAIT :
                                      MOJO_RESULT_FAILED_PRECONDITION;
   }
 
-  bool was_full = (current_num_bytes_ == capacity_num_bytes());
-
   size_t num_bytes_to_discard =
       std::min(static_cast<size_t>(*num_bytes), current_num_bytes_);
-  start_index_ = (start_index_ + num_bytes_to_discard) % capacity_num_bytes();
-  current_num_bytes_ -= num_bytes_to_discard;
-
-  if (was_full && num_bytes_to_discard > 0)
-    AwakeProducerWaitersForStateChangeNoLock();
-
+  MarkDataAsConsumedNoLock(num_bytes_to_discard);
   *num_bytes = static_cast<uint32_t>(num_bytes_to_discard);
   return MOJO_RESULT_OK;
 }
 
 MojoResult LocalDataPipe::ConsumerQueryDataImplNoLock(uint32_t* num_bytes) {
   // Note: This cast is safe, since the capacity fits into a |uint32_t|.
   *num_bytes = static_cast<uint32_t>(current_num_bytes_);
   return MOJO_RESULT_OK;
 }
 
@@ skipping 23 matching lines @@
 }
 
 MojoResult LocalDataPipe::ConsumerEndReadDataImplNoLock(
     uint32_t num_bytes_read) {
   if (num_bytes_read > consumer_two_phase_max_num_bytes_read_no_lock()) {
     // Note: The two-phase read ends here even on failure.
     set_consumer_two_phase_max_num_bytes_read_no_lock(0);
     return MOJO_RESULT_INVALID_ARGUMENT;
   }
 
-  bool was_full = (current_num_bytes_ == capacity_num_bytes());
-
-  start_index_ += num_bytes_read;
-  DCHECK_LE(start_index_, capacity_num_bytes());
-  start_index_ %= capacity_num_bytes();
-  DCHECK_LE(num_bytes_read, current_num_bytes_);
-  current_num_bytes_ -= num_bytes_read;
+  DCHECK_LE(start_index_ + num_bytes_read, capacity_num_bytes());
+  MarkDataAsConsumedNoLock(num_bytes_read);
   set_consumer_two_phase_max_num_bytes_read_no_lock(0);
-
-  if (was_full && num_bytes_read > 0)
-    AwakeProducerWaitersForStateChangeNoLock();
-
   return MOJO_RESULT_OK;
 }
 
 MojoWaitFlags LocalDataPipe::ConsumerSatisfiedFlagsNoLock() {
   MojoWaitFlags rv = MOJO_WAIT_FLAG_NONE;
-  if (current_num_bytes_ > 0)
+  if (current_num_bytes_ > 0 && !consumer_in_two_phase_read_no_lock())
     rv |= MOJO_WAIT_FLAG_READABLE;
   return rv;
 }
 
 MojoWaitFlags LocalDataPipe::ConsumerSatisfiableFlagsNoLock() {
   MojoWaitFlags rv = MOJO_WAIT_FLAG_NONE;
   if (current_num_bytes_ > 0 || producer_open_no_lock())
     rv |= MOJO_WAIT_FLAG_READABLE;
   return rv;
 }
@@ skipping 27 matching lines @@
   }
   return capacity_num_bytes() - next_index;
 }
 
 size_t LocalDataPipe::GetMaxNumBytesToReadNoLock() {
   if (start_index_ + current_num_bytes_ > capacity_num_bytes())
     return capacity_num_bytes() - start_index_;
   return current_num_bytes_;
 }
 
+void LocalDataPipe::MarkDataAsConsumedNoLock(size_t num_bytes) {
+  DCHECK_LE(num_bytes, current_num_bytes_);
+  start_index_ += num_bytes;
+  start_index_ %= capacity_num_bytes();
+  current_num_bytes_ -= num_bytes;
+}
+
 }  // namespace system
 }  // namespace mojo