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

Arguably more importantly, make data pipe producers not writable during
two-phase writes and consumers not readable during two-phase reads. This
allows multiple threads to simultaneously use the two-phase APIs in a
sane way; they can respond to a "busy" error by waiting. (Why not
"should wait" instead of "busy"? Because waiting is only an appropriate
response if you know that someone else is going to end the two-phase
read/write -- e.g., if a two-phase read/write is happening on another
thread.)

Also refactor the waiter-awakening stuff so that it's taken care of by
the DataPipe superclass, rather than the LocalDataPipe implementation
subclass.

Note that a two-phase write will only discard if "all-or-none" is
requested. Otherwise, as usual, it'll provide as much space as
available.

R=darin@chromium.org

Review URL: https://codereview.chromium.org/129163003

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@243920 0039d316-1c4b-4281-b951-d872f2087c98

mojo/system/data_pipe.cc

@@ -79,6 +79,7 @@ void DataPipe::ProducerClose() {
       << "Producer closed with active two-phase write";
   producer_two_phase_max_num_bytes_written_ = 0;
   ProducerCloseImplNoLock();
+  AwakeConsumerWaitersForStateChangeNoLock();
 }
 
 MojoResult DataPipe::ProducerWriteData(const void* elements,
@@ -99,7 +100,11 @@ MojoResult DataPipe::ProducerWriteData(const void* elements,
   if (*num_bytes == 0)
     return MOJO_RESULT_OK;  // Nothing to do.
 
-  return ProducerWriteDataImplNoLock(elements, num_bytes, all_or_none);
+  MojoWaitFlags old_consumer_satisfied_flags = ConsumerSatisfiedFlagsNoLock();
+  MojoResult rv = ProducerWriteDataImplNoLock(elements, num_bytes, all_or_none);
+  if (ConsumerSatisfiedFlagsNoLock() != old_consumer_satisfied_flags)
+    AwakeConsumerWaitersForStateChangeNoLock();
+  return rv;
 }
 
 MojoResult DataPipe::ProducerBeginWriteData(void** buffer,
@@ -117,7 +122,10 @@ MojoResult DataPipe::ProducerBeginWriteData(void** buffer,
                                                    all_or_none);
   if (rv != MOJO_RESULT_OK)
     return rv;
-
+  // Note: No need to awake producer waiters, even though we're going from
+  // writable to non-writable (since you can't wait on non-writability).
+  // Similarly, though this may have discarded data (in "may discard" mode),
+  // making it non-readable, there's still no need to awake consumer waiters.
   DCHECK(producer_in_two_phase_write_no_lock());
   return MOJO_RESULT_OK;
 }
@@ -131,9 +139,16 @@ MojoResult DataPipe::ProducerEndWriteData(uint32_t num_bytes_written) {
   // Note: Allow successful completion of the two-phase write even if the
   // consumer has been closed.
 
+  MojoWaitFlags old_consumer_satisfied_flags = ConsumerSatisfiedFlagsNoLock();
   MojoResult 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 ((ProducerSatisfiedFlagsNoLock() & MOJO_WAIT_FLAG_WRITABLE))
+    AwakeProducerWaitersForStateChangeNoLock();
+  if (ConsumerSatisfiedFlagsNoLock() != old_consumer_satisfied_flags)
+    AwakeConsumerWaitersForStateChangeNoLock();
   return rv;
 }
 
@@ -175,6 +190,7 @@ void DataPipe::ConsumerClose() {
       << "Consumer closed with active two-phase read";
   consumer_two_phase_max_num_bytes_read_ = 0;
   ConsumerCloseImplNoLock();
+  AwakeProducerWaitersForStateChangeNoLock();
 }
 
 MojoResult DataPipe::ConsumerReadData(void* elements,
@@ -192,7 +208,11 @@ MojoResult DataPipe::ConsumerReadData(void* elements,
   if (*num_bytes == 0)
     return MOJO_RESULT_OK;  // Nothing to do.
 
-  return ConsumerReadDataImplNoLock(elements, num_bytes, all_or_none);
+  MojoWaitFlags old_producer_satisfied_flags = ProducerSatisfiedFlagsNoLock();
+  MojoResult rv = ConsumerReadDataImplNoLock(elements, num_bytes, all_or_none);
+  if (ProducerSatisfiedFlagsNoLock() != old_producer_satisfied_flags)
+    AwakeProducerWaitersForStateChangeNoLock();
+  return rv;
 }
 
 MojoResult DataPipe::ConsumerDiscardData(uint32_t* num_bytes,
@@ -209,7 +229,11 @@ MojoResult DataPipe::ConsumerDiscardData(uint32_t* num_bytes,
   if (*num_bytes == 0)
     return MOJO_RESULT_OK;  // Nothing to do.
 
-  return ConsumerDiscardDataImplNoLock(num_bytes, all_or_none);
+  MojoWaitFlags old_producer_satisfied_flags = ProducerSatisfiedFlagsNoLock();
+  MojoResult rv = ConsumerDiscardDataImplNoLock(num_bytes, all_or_none);
+  if (ProducerSatisfiedFlagsNoLock() != old_producer_satisfied_flags)
+    AwakeProducerWaitersForStateChangeNoLock();
+  return rv;
 }
 
 MojoResult DataPipe::ConsumerQueryData(uint32_t* num_bytes) {
@@ -236,7 +260,6 @@ MojoResult DataPipe::ConsumerBeginReadData(const void** buffer,
                                                   all_or_none);
   if (rv != MOJO_RESULT_OK)
     return rv;
-
   DCHECK(consumer_in_two_phase_read_no_lock());
   return MOJO_RESULT_OK;
 }
@@ -248,9 +271,16 @@ MojoResult DataPipe::ConsumerEndReadData(uint32_t num_bytes_read) {
   if (!consumer_in_two_phase_read_no_lock())
     return MOJO_RESULT_FAILED_PRECONDITION;
 
+  MojoWaitFlags old_producer_satisfied_flags = ProducerSatisfiedFlagsNoLock();
   MojoResult 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.
+  if ((ConsumerSatisfiedFlagsNoLock() & MOJO_WAIT_FLAG_READABLE))
+    AwakeConsumerWaitersForStateChangeNoLock();
+  if (ProducerSatisfiedFlagsNoLock() != old_producer_satisfied_flags)
+    AwakeProducerWaitersForStateChangeNoLock();
   return rv;
 }
 

mojo/system/data_pipe.h

@@ -83,9 +83,6 @@ class MOJO_SYSTEM_IMPL_EXPORT DataPipe :
   friend class base::RefCountedThreadSafe<DataPipe>;
   virtual ~DataPipe();
 
-  void AwakeProducerWaitersForStateChangeNoLock();
-  void AwakeConsumerWaitersForStateChangeNoLock();
-
   virtual void ProducerCloseImplNoLock() = 0;
   // |*num_bytes| will be a nonzero multiple of |element_num_bytes_|.
   virtual MojoResult ProducerWriteDataImplNoLock(const void* elements,
@@ -97,6 +94,7 @@ class MOJO_SYSTEM_IMPL_EXPORT DataPipe :
       bool all_or_none) = 0;
   virtual MojoResult ProducerEndWriteDataImplNoLock(
       uint32_t num_bytes_written) = 0;
+  // Note: A producer should not be writable during a two-phase write.
   virtual MojoWaitFlags ProducerSatisfiedFlagsNoLock() = 0;
   virtual MojoWaitFlags ProducerSatisfiableFlagsNoLock() = 0;
 
@@ -113,6 +111,7 @@ class MOJO_SYSTEM_IMPL_EXPORT DataPipe :
                                                      uint32_t* buffer_num_bytes,
                                                      bool all_or_none) = 0;
   virtual MojoResult ConsumerEndReadDataImplNoLock(uint32_t num_bytes_read) = 0;
+  // Note: A consumer should not be writable during a two-phase read.
   virtual MojoWaitFlags ConsumerSatisfiedFlagsNoLock() = 0;
   virtual MojoWaitFlags ConsumerSatisfiableFlagsNoLock() = 0;
 
@@ -158,6 +157,9 @@ class MOJO_SYSTEM_IMPL_EXPORT DataPipe :
   }
 
  private:
+  void AwakeProducerWaitersForStateChangeNoLock();
+  void AwakeConsumerWaitersForStateChangeNoLock();
+
   bool has_local_producer_no_lock() const {
     lock_.AssertAcquired();
     return !!producer_waiter_list_.get();

mojo/system/local_data_pipe.cc

@@ -42,7 +42,6 @@ void LocalDataPipe::ProducerCloseImplNoLock() {
     DCHECK(!consumer_in_two_phase_read_no_lock());
     DestroyBufferNoLock();
   }
-  AwakeConsumerWaitersForStateChangeNoLock();
 }
 
 MojoResult LocalDataPipe::ProducerWriteDataImplNoLock(const void* elements,
@@ -61,11 +60,10 @@ MojoResult LocalDataPipe::ProducerWriteDataImplNoLock(const void* elements,
                                   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_) {
@@ -96,14 +94,8 @@ MojoResult LocalDataPipe::ProducerWriteDataImplNoLock(const void* elements,
            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;
 }
@@ -114,19 +106,33 @@ MojoResult LocalDataPipe::ProducerBeginWriteDataImplNoLock(
     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);
@@ -143,21 +149,17 @@ MojoResult LocalDataPipe::ProducerEndWriteDataImplNoLock(
     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;
 }
@@ -176,7 +178,6 @@ void LocalDataPipe::ConsumerCloseImplNoLock() {
   if (!producer_open_no_lock() || !producer_in_two_phase_write_no_lock())
     DestroyBufferNoLock();
   current_num_bytes_ = 0;
-  AwakeProducerWaitersForStateChangeNoLock();
 }
 
 MojoResult LocalDataPipe::ConsumerReadDataImplNoLock(void* elements,
@@ -211,15 +212,7 @@ MojoResult LocalDataPipe::ConsumerReadDataImplNoLock(void* elements,
            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;
 }
@@ -242,16 +235,9 @@ MojoResult LocalDataPipe::ConsumerDiscardDataImplNoLock(uint32_t* num_bytes,
                                      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;
 }
@@ -295,24 +281,15 @@ MojoResult LocalDataPipe::ConsumerEndReadDataImplNoLock(
     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;
 }
@@ -360,5 +337,12 @@ size_t LocalDataPipe::GetMaxNumBytesToReadNoLock() {
   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

mojo/system/local_data_pipe.h

@@ -65,6 +65,10 @@ class MOJO_SYSTEM_IMPL_EXPORT LocalDataPipe : public DataPipe {
   size_t GetMaxNumBytesToWriteNoLock();
   size_t GetMaxNumBytesToReadNoLock();
 
+  // Marks the given number of bytes as consumed/discarded. |num_bytes| must be
+  // greater than |current_num_bytes_|.
+  void MarkDataAsConsumedNoLock(size_t num_bytes);
+
   // The members below are protected by |DataPipe|'s |lock_|:
   scoped_ptr_malloc<char, base::ScopedPtrAlignedFree> buffer_;
   // Circular buffer.

mojo/system/local_data_pipe_unittest.cc

@@ -182,6 +182,9 @@ TEST(LocalDataPipeTest, SimpleReadWrite) {
   dp->ConsumerClose();
 }
 
+// Note: The "basic" waiting tests test that the "wait states" are correct in
+// various situations; they don't test that waiters are properly awoken on state
+// changes. (For that, we need to use multiple threads.)
 TEST(LocalDataPipeTest, BasicProducerWaiting) {
   // Note: We take advantage of the fact that for |LocalDataPipe|, capacities
   // are strict maximums. This is not guaranteed by the API.
@@ -196,110 +199,107 @@ TEST(LocalDataPipeTest, BasicProducerWaiting) {
   EXPECT_EQ(MOJO_RESULT_OK,
             DataPipe::ValidateOptions(&options, &validated_options));
 
-  {
-    scoped_refptr<LocalDataPipe> dp(new LocalDataPipe(validated_options));
-    Waiter waiter;
+  scoped_refptr<LocalDataPipe> dp(new LocalDataPipe(validated_options));
+  Waiter waiter;
 
-    // Never readable.
-    waiter.Init();
-    EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
-              dp->ProducerAddWaiter(&waiter, MOJO_WAIT_FLAG_READABLE, 12));
+  // Never readable.
+  waiter.Init();
+  EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
+            dp->ProducerAddWaiter(&waiter, MOJO_WAIT_FLAG_READABLE, 12));
 
-    // Already writable.
-    waiter.Init();
-    EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
-              dp->ProducerAddWaiter(&waiter, MOJO_WAIT_FLAG_WRITABLE, 34));
+  // Already writable.
+  waiter.Init();
+  EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
+            dp->ProducerAddWaiter(&waiter, MOJO_WAIT_FLAG_WRITABLE, 34));
 
-    // Write two elements.
-    int32_t elements[2] = { 123, 456 };
-    uint32_t num_bytes = static_cast<uint32_t>(2u * sizeof(elements[0]));
-    EXPECT_EQ(MOJO_RESULT_OK,
-              dp->ProducerWriteData(elements, &num_bytes, true));
-    EXPECT_EQ(static_cast<uint32_t>(2u * sizeof(elements[0])), num_bytes);
+  // Write two elements.
+  int32_t elements[2] = { 123, 456 };
+  uint32_t num_bytes = static_cast<uint32_t>(2u * sizeof(elements[0]));
+  EXPECT_EQ(MOJO_RESULT_OK, dp->ProducerWriteData(elements, &num_bytes, true));
+  EXPECT_EQ(static_cast<uint32_t>(2u * sizeof(elements[0])), num_bytes);
 
-    // Adding a waiter should now succeed.
-    waiter.Init();
-    EXPECT_EQ(MOJO_RESULT_OK,
-              dp->ProducerAddWaiter(&waiter, MOJO_WAIT_FLAG_WRITABLE, 56));
-    // And it shouldn't be writable yet.
-    EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0));
-    dp->ProducerRemoveWaiter(&waiter);
+  // Adding a waiter should now succeed.
+  waiter.Init();
+  EXPECT_EQ(MOJO_RESULT_OK,
+            dp->ProducerAddWaiter(&waiter, MOJO_WAIT_FLAG_WRITABLE, 56));
+  // And it shouldn't be writable yet.
+  EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0));
+  dp->ProducerRemoveWaiter(&waiter);
 
-    // Do it again.
-    waiter.Init();
-    EXPECT_EQ(MOJO_RESULT_OK,
-              dp->ProducerAddWaiter(&waiter, MOJO_WAIT_FLAG_WRITABLE, 78));
+  // Do it again.
+  waiter.Init();
+  EXPECT_EQ(MOJO_RESULT_OK,
+            dp->ProducerAddWaiter(&waiter, MOJO_WAIT_FLAG_WRITABLE, 78));
 
-    // Read one element.
-    elements[0] = -1;
-    elements[1] = -1;
-    num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
-    EXPECT_EQ(MOJO_RESULT_OK, dp->ConsumerReadData(elements, &num_bytes, true));
-    EXPECT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
-    EXPECT_EQ(123, elements[0]);
-    EXPECT_EQ(-1, elements[1]);
+  // Read one element.
+  elements[0] = -1;
+  elements[1] = -1;
+  num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
+  EXPECT_EQ(MOJO_RESULT_OK, dp->ConsumerReadData(elements, &num_bytes, true));
+  EXPECT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
+  EXPECT_EQ(123, elements[0]);
+  EXPECT_EQ(-1, elements[1]);
 
-    // Waiting should now succeed.
-    EXPECT_EQ(78, waiter.Wait(1000));
-    dp->ProducerRemoveWaiter(&waiter);
+  // Waiting should now succeed.
+  EXPECT_EQ(78, waiter.Wait(1000));
+  dp->ProducerRemoveWaiter(&waiter);
 
-    // Try writing, using a two-phase write.
-    void* buffer = NULL;
-    num_bytes = static_cast<uint32_t>(3u * sizeof(elements[0]));
-    EXPECT_EQ(MOJO_RESULT_OK,
-              dp->ProducerBeginWriteData(&buffer, &num_bytes, false));
-    EXPECT_TRUE(buffer != NULL);
-    EXPECT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
-    static_cast<int32_t*>(buffer)[0] = 789;
-    EXPECT_EQ(MOJO_RESULT_OK,
-              dp->ProducerEndWriteData(
-                  static_cast<uint32_t>(1u * sizeof(elements[0]))));
+  // Try writing, using a two-phase write.
+  void* buffer = NULL;
+  num_bytes = static_cast<uint32_t>(3u * sizeof(elements[0]));
+  EXPECT_EQ(MOJO_RESULT_OK,
+            dp->ProducerBeginWriteData(&buffer, &num_bytes, false));
+  EXPECT_TRUE(buffer != NULL);
+  EXPECT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
 
-    // Add a waiter.
-    waiter.Init();
-    EXPECT_EQ(MOJO_RESULT_OK,
-              dp->ProducerAddWaiter(&waiter, MOJO_WAIT_FLAG_WRITABLE, 90));
+  static_cast<int32_t*>(buffer)[0] = 789;
+  EXPECT_EQ(MOJO_RESULT_OK,
+            dp->ProducerEndWriteData(
+                static_cast<uint32_t>(1u * sizeof(elements[0]))));
 
-    // Read one element, using a two-phase read.
-    const void* read_buffer = NULL;
-    num_bytes = 0u;
-    EXPECT_EQ(MOJO_RESULT_OK,
-              dp->ConsumerBeginReadData(&read_buffer, &num_bytes, false));
-    EXPECT_TRUE(read_buffer != NULL);
-    // Since we only read one element (after having written three in all), the
-    // two-phase read should only allow us to read one. This checks an
-    // implementation detail!
-    EXPECT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
-    EXPECT_EQ(456, static_cast<const int32_t*>(read_buffer)[0]);
-    EXPECT_EQ(MOJO_RESULT_OK,
-              dp->ConsumerEndReadData(
-                  static_cast<uint32_t>(1u * sizeof(elements[0]))));
+  // Add a waiter.
+  waiter.Init();
+  EXPECT_EQ(MOJO_RESULT_OK,
+            dp->ProducerAddWaiter(&waiter, MOJO_WAIT_FLAG_WRITABLE, 90));
 
-    // Waiting should succeed.
-    EXPECT_EQ(90, waiter.Wait(1000));
-    dp->ProducerRemoveWaiter(&waiter);
+  // Read one element, using a two-phase read.
+  const void* read_buffer = NULL;
+  num_bytes = 0u;
+  EXPECT_EQ(MOJO_RESULT_OK,
+            dp->ConsumerBeginReadData(&read_buffer, &num_bytes, false));
+  EXPECT_TRUE(read_buffer != NULL);
+  // Since we only read one element (after having written three in all), the
+  // two-phase read should only allow us to read one. This checks an
+  // implementation detail!
+  EXPECT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
+  EXPECT_EQ(456, static_cast<const int32_t*>(read_buffer)[0]);
+  EXPECT_EQ(MOJO_RESULT_OK,
+            dp->ConsumerEndReadData(
+                static_cast<uint32_t>(1u * sizeof(elements[0]))));
 
-    // Write one element.
-    elements[0] = 123;
-    num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
-    EXPECT_EQ(MOJO_RESULT_OK,
-              dp->ProducerWriteData(elements, &num_bytes, false));
-    EXPECT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
+  // Waiting should succeed.
+  EXPECT_EQ(90, waiter.Wait(1000));
+  dp->ProducerRemoveWaiter(&waiter);
 
-    // Add a waiter.
-    waiter.Init();
-    EXPECT_EQ(MOJO_RESULT_OK,
-              dp->ProducerAddWaiter(&waiter, MOJO_WAIT_FLAG_WRITABLE, 12));
+  // Write one element.
+  elements[0] = 123;
+  num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
+  EXPECT_EQ(MOJO_RESULT_OK, dp->ProducerWriteData(elements, &num_bytes, false));
+  EXPECT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
 
-    // Close the consumer.
-    dp->ConsumerClose();
+  // Add a waiter.
+  waiter.Init();
+  EXPECT_EQ(MOJO_RESULT_OK,
+            dp->ProducerAddWaiter(&waiter, MOJO_WAIT_FLAG_WRITABLE, 12));
 
-    // It should now be never-writable.
-    EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION, waiter.Wait(1000));
-    dp->ProducerRemoveWaiter(&waiter);
+  // Close the consumer.
+  dp->ConsumerClose();
 
-    dp->ProducerClose();
-  }
+  // It should now be never-writable.
+  EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION, waiter.Wait(1000));
+  dp->ProducerRemoveWaiter(&waiter);
+
+  dp->ProducerClose();
 }
 
 TEST(LocalDataPipeTest, BasicConsumerWaiting) {
@@ -476,6 +476,195 @@ TEST(LocalDataPipeTest, BasicConsumerWaiting) {
   }
 }
 
+// Tests that data pipes aren't writable/readable during two-phase writes/reads.
+TEST(LocalDataPipeTest, BasicTwoPhaseWaiting) {
+  const MojoCreateDataPipeOptions options = {
+    kSizeOfOptions,  // |struct_size|.
+    MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_NONE,  // |flags|.
+    static_cast<uint32_t>(sizeof(int32_t)),  // |element_num_bytes|.
+    1000 * sizeof(int32_t)  // |capacity_num_bytes|.
+  };
+  MojoCreateDataPipeOptions validated_options = { 0 };
+  EXPECT_EQ(MOJO_RESULT_OK,
+            DataPipe::ValidateOptions(&options, &validated_options));
+
+  scoped_refptr<LocalDataPipe> dp(new LocalDataPipe(validated_options));
+  Waiter waiter;
+
+  // It should be writable.
+  waiter.Init();
+  EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
+            dp->ProducerAddWaiter(&waiter, MOJO_WAIT_FLAG_WRITABLE, 0));
+
+  uint32_t num_bytes = static_cast<uint32_t>(1u * sizeof(int32_t));
+  void* write_ptr = NULL;
+  EXPECT_EQ(MOJO_RESULT_OK,
+            dp->ProducerBeginWriteData(&write_ptr, &num_bytes, false));
+  EXPECT_TRUE(write_ptr != NULL);
+  EXPECT_GE(num_bytes, static_cast<uint32_t>(1u * sizeof(int32_t)));
+
+  // At this point, it shouldn't be writable.
+  waiter.Init();
+  EXPECT_EQ(MOJO_RESULT_OK,
+            dp->ProducerAddWaiter(&waiter, MOJO_WAIT_FLAG_WRITABLE, 1));
+  EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0));
+  dp->ProducerRemoveWaiter(&waiter);
+
+  // It shouldn't be readable yet either.
+  waiter.Init();
+  EXPECT_EQ(MOJO_RESULT_OK,
+            dp->ConsumerAddWaiter(&waiter, MOJO_WAIT_FLAG_READABLE, 2));
+  EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0));
+  dp->ConsumerRemoveWaiter(&waiter);
+
+  static_cast<int32_t*>(write_ptr)[0] = 123;
+  EXPECT_EQ(MOJO_RESULT_OK,
+            dp->ProducerEndWriteData(
+                static_cast<uint32_t>(1u * sizeof(int32_t))));
+
+  // It should be writable again.
+  waiter.Init();
+  EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
+            dp->ProducerAddWaiter(&waiter, MOJO_WAIT_FLAG_WRITABLE, 3));
+
+  // And readable.
+  waiter.Init();
+  EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
+            dp->ConsumerAddWaiter(&waiter, MOJO_WAIT_FLAG_READABLE, 4));
+
+  // Start another two-phase write and check that it's readable even in the
+  // middle of it.
+  num_bytes = static_cast<uint32_t>(1u * sizeof(int32_t));
+  write_ptr = NULL;
+  EXPECT_EQ(MOJO_RESULT_OK,
+            dp->ProducerBeginWriteData(&write_ptr, &num_bytes, false));
+  EXPECT_TRUE(write_ptr != NULL);
+  EXPECT_GE(num_bytes, static_cast<uint32_t>(1u * sizeof(int32_t)));
+
+  // It should be readable.
+  waiter.Init();
+  EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
+            dp->ConsumerAddWaiter(&waiter, MOJO_WAIT_FLAG_READABLE, 5));
+
+  // End the two-phase write without writing anything.
+  EXPECT_EQ(MOJO_RESULT_OK, dp->ProducerEndWriteData(0u));
+
+  // Start a two-phase read.
+  num_bytes = static_cast<uint32_t>(1u * sizeof(int32_t));
+  const void* read_ptr = NULL;
+  EXPECT_EQ(MOJO_RESULT_OK,
+            dp->ConsumerBeginReadData(&read_ptr, &num_bytes, false));
+  EXPECT_TRUE(read_ptr != NULL);
+  EXPECT_EQ(static_cast<uint32_t>(1u * sizeof(int32_t)), num_bytes);
+
+  // At this point, it should still be writable.
+  waiter.Init();
+  EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
+            dp->ProducerAddWaiter(&waiter, MOJO_WAIT_FLAG_WRITABLE, 6));
+
+  // But not readable.
+  waiter.Init();
+  EXPECT_EQ(MOJO_RESULT_OK,
+            dp->ConsumerAddWaiter(&waiter, MOJO_WAIT_FLAG_READABLE, 7));
+  EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0));
+  dp->ConsumerRemoveWaiter(&waiter);
+
+  // End the two-phase read without reading anything.
+  EXPECT_EQ(MOJO_RESULT_OK, dp->ConsumerEndReadData(0u));
+
+  // It should be readable again.
+  waiter.Init();
+  EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
+            dp->ConsumerAddWaiter(&waiter, MOJO_WAIT_FLAG_READABLE, 8));
+
+  dp->ProducerClose();
+  dp->ConsumerClose();
+}
+
+// Test that a "may discard" data pipe is writable even when it's full.
+TEST(LocalDataPipeTest, BasicMayDiscardWaiting) {
+  const MojoCreateDataPipeOptions options = {
+    kSizeOfOptions,  // |struct_size|.
+    MOJO_CREATE_DATA_PIPE_OPTIONS_FLAG_MAY_DISCARD,  // |flags|.
+    static_cast<uint32_t>(sizeof(int32_t)),  // |element_num_bytes|.
+    1 * sizeof(int32_t)  // |capacity_num_bytes|.
+  };
+  MojoCreateDataPipeOptions validated_options = { 0 };
+  EXPECT_EQ(MOJO_RESULT_OK,
+            DataPipe::ValidateOptions(&options, &validated_options));
+
+  scoped_refptr<LocalDataPipe> dp(new LocalDataPipe(validated_options));
+  Waiter waiter;
+
+  // Writable.
+  waiter.Init();
+  EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
+            dp->ProducerAddWaiter(&waiter, MOJO_WAIT_FLAG_WRITABLE, 0));
+
+  // Not readable.
+  waiter.Init();
+  EXPECT_EQ(MOJO_RESULT_OK,
+            dp->ConsumerAddWaiter(&waiter, MOJO_WAIT_FLAG_READABLE, 1));
+  EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0));
+  dp->ConsumerRemoveWaiter(&waiter);
+
+  uint32_t num_bytes = static_cast<uint32_t>(sizeof(int32_t));
+  int32_t element = 123;
+  EXPECT_EQ(MOJO_RESULT_OK,
+            dp->ProducerWriteData(&element, &num_bytes, false));
+  EXPECT_EQ(static_cast<uint32_t>(sizeof(int32_t)), num_bytes);
+
+  // Still writable (even though it's full.
+  waiter.Init();
+  EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
+            dp->ProducerAddWaiter(&waiter, MOJO_WAIT_FLAG_WRITABLE, 2));
+
+  // Now readable.
+  waiter.Init();
+  EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
+            dp->ConsumerAddWaiter(&waiter, MOJO_WAIT_FLAG_READABLE, 3));
+
+  // Overwrite that element.
+  num_bytes = static_cast<uint32_t>(sizeof(int32_t));
+  element = 456;
+  EXPECT_EQ(MOJO_RESULT_OK,
+            dp->ProducerWriteData(&element, &num_bytes, false));
+  EXPECT_EQ(static_cast<uint32_t>(sizeof(int32_t)), num_bytes);
+
+  // Still writable.
+  waiter.Init();
+  EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
+            dp->ProducerAddWaiter(&waiter, MOJO_WAIT_FLAG_WRITABLE, 4));
+
+  // And still readable.
+  waiter.Init();
+  EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
+            dp->ConsumerAddWaiter(&waiter, MOJO_WAIT_FLAG_READABLE, 5));
+
+  // Read that element.
+  num_bytes = static_cast<uint32_t>(sizeof(int32_t));
+  element = 0;
+  EXPECT_EQ(MOJO_RESULT_OK,
+            dp->ConsumerReadData(&element, &num_bytes, false));
+  EXPECT_EQ(static_cast<uint32_t>(sizeof(int32_t)), num_bytes);
+  EXPECT_EQ(456, element);
+
+  // Still writable.
+  waiter.Init();
+  EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
+            dp->ProducerAddWaiter(&waiter, MOJO_WAIT_FLAG_WRITABLE, 6));
+
+  // No longer readable.
+  waiter.Init();
+  EXPECT_EQ(MOJO_RESULT_OK,
+            dp->ConsumerAddWaiter(&waiter, MOJO_WAIT_FLAG_READABLE, 7));
+  EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0));
+  dp->ConsumerRemoveWaiter(&waiter);
+
+  dp->ProducerClose();
+  dp->ConsumerClose();
+}
+
 void Seq(int32_t start, size_t count, int32_t* out) {
   for (size_t i = 0; i < count; i++)
     out[i] = start + static_cast<int32_t>(i);
@@ -573,7 +762,81 @@ TEST(LocalDataPipeTest, MayDiscard) {
   expected_buffer[9] = 304;
   EXPECT_EQ(0, memcmp(buffer, expected_buffer, sizeof(buffer)));
 
-  // TODO(vtl): Test two-phase write when it supports "may discard".
+  // Test two-phase writes, including in all-or-none mode.
+  // Note: Again, the following depends on an implementation detail -- namely
+  // that the write pointer will point at the 5th element of the buffer (and the
+  // buffer has exactly the capacity requested).
+
+  num_bytes = 0u;
+  void* write_ptr = NULL;
+  EXPECT_EQ(MOJO_RESULT_OK,
+            dp->ProducerBeginWriteData(&write_ptr, &num_bytes, false));
+  EXPECT_TRUE(write_ptr != NULL);
+  EXPECT_EQ(6u * sizeof(int32_t), num_bytes);
+  Seq(400, 6, static_cast<int32_t*>(write_ptr));
+  EXPECT_EQ(MOJO_RESULT_OK, dp->ProducerEndWriteData(6u * sizeof(int32_t)));
+  // Internally, a circular buffer would now look like:
+  //   -, -, -, -, 400, 401, 402, 403, 404, 405
+
+  // |ProducerBeginWriteData()| ignores |*num_bytes| except in "all-or-none"
+  // mode.
+  num_bytes = 6u * sizeof(int32_t);
+  write_ptr = NULL;
+  EXPECT_EQ(MOJO_RESULT_OK,
+            dp->ProducerBeginWriteData(&write_ptr, &num_bytes, false));
+  EXPECT_EQ(4u * sizeof(int32_t), num_bytes);
+  static_cast<int32_t*>(write_ptr)[0] = 500;
+  EXPECT_EQ(MOJO_RESULT_OK, dp->ProducerEndWriteData(1u * sizeof(int32_t)));
+  // Internally, a circular buffer would now look like:
+  //   500, -, -, -, 400, 401, 402, 403, 404, 405
+
+  // Requesting a 10-element buffer in all-or-none mode fails at this point.
+  num_bytes = 10u * sizeof(int32_t);
+  write_ptr = NULL;
+  EXPECT_EQ(MOJO_RESULT_OUT_OF_RANGE,
+            dp->ProducerBeginWriteData(&write_ptr, &num_bytes, true));
+
+  // But requesting, say, a 5-element (up to 9, really) buffer should be okay.
+  // It will discard two elements.
+  num_bytes = 5u * sizeof(int32_t);
+  write_ptr = NULL;
+  EXPECT_EQ(MOJO_RESULT_OK,
+            dp->ProducerBeginWriteData(&write_ptr, &num_bytes, true));
+  EXPECT_EQ(5u * sizeof(int32_t), num_bytes);
+  // Only write 4 elements though.
+  Seq(600, 4, static_cast<int32_t*>(write_ptr));
+  EXPECT_EQ(MOJO_RESULT_OK, dp->ProducerEndWriteData(4u * sizeof(int32_t)));
+  // Internally, a circular buffer would now look like:
+  //   500, 600, 601, 602, 603, -, 402, 403, 404, 405
+
+  // Do this again. Make sure we can get a buffer all the way out to the end of
+  // the internal buffer.
+  num_bytes = 5u * sizeof(int32_t);
+  write_ptr = NULL;
+  EXPECT_EQ(MOJO_RESULT_OK,
+            dp->ProducerBeginWriteData(&write_ptr, &num_bytes, true));
+  EXPECT_EQ(5u * sizeof(int32_t), num_bytes);
+  // Only write 3 elements though.
+  Seq(700, 3, static_cast<int32_t*>(write_ptr));
+  EXPECT_EQ(MOJO_RESULT_OK, dp->ProducerEndWriteData(3u * sizeof(int32_t)));
+  // Internally, a circular buffer would now look like:
+  //   500, 600, 601, 602, 603, 700, 701, 702, -, -
+
+  // Read everything.
+  num_bytes = sizeof(buffer);
+  memset(buffer, 0xab, sizeof(buffer));
+  EXPECT_EQ(MOJO_RESULT_OK, dp->ConsumerReadData(buffer, &num_bytes, false));
+  EXPECT_EQ(8u * sizeof(int32_t), num_bytes);
+  memset(expected_buffer, 0xab, sizeof(expected_buffer));
+  expected_buffer[0] = 500;
+  expected_buffer[1] = 600;
+  expected_buffer[2] = 601;
+  expected_buffer[3] = 602;
+  expected_buffer[4] = 603;
+  expected_buffer[5] = 700;
+  expected_buffer[6] = 701;
+  expected_buffer[7] = 702;
+  EXPECT_EQ(0, memcmp(buffer, expected_buffer, sizeof(buffer)));
 
   dp->ProducerClose();
   dp->ConsumerClose();
@@ -801,6 +1064,9 @@ TEST(LocalDataPipeTest, AllOrNoneMayDiscard) {
   Seq(300, 10, expected_buffer);
   EXPECT_EQ(0, memcmp(buffer, expected_buffer, sizeof(buffer)));
 
+  // Note: All-or-none two-phase writes on a "may discard" data pipe are tested
+  // in LocalDataPipeTest.MayDiscard.
+
   dp->ProducerClose();
   dp->ConsumerClose();
 }
@@ -909,9 +1175,6 @@ TEST(LocalDataPipeTest, TwoPhaseAllOrNone) {
   dp->ConsumerClose();
 }
 
-// TODO(vtl): Test two-phase read/write with "all or none" and "may discard",
-// once that's supported.
-
 // Tests that |ProducerWriteData()| and |ConsumerReadData()| writes and reads,
 // respectively, as much as possible, even if it has to "wrap around" the
 // internal circular buffer. (Note that the two-phase write and read do not do
@@ -1168,8 +1431,6 @@ TEST(LocalDataPipeTest, CloseWriteRead) {
   }
 }
 
-// TODO(vtl): More.
-
 }  // namespace
 }  // namespace system
 }  // namespace mojo