Mojo: Add more data pipe tests, fix waiting bug.

Oops, I forgot to wake up waiters on read/write.

R=sky@chromium.org

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

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

3 files changed, 104 insertions, 8 deletions

diff --git a/mojo/system/data_pipe.cc b/mojo/system/data_pipe.cc
index b5a0f1b..4dfabba 100644
--- a/mojo/system/data_pipe.cc
+++ b/mojo/system/data_pipe.cc
@@ -9,6 +9,7 @@
 #include <algorithm>
 #include <limits>
 
+#include "base/compiler_specific.h"
 #include "base/logging.h"
 #include "mojo/system/constants.h"
 #include "mojo/system/memory.h"
@@ -270,7 +271,9 @@ DataPipe::DataPipe(bool has_local_producer,
       consumer_waiter_list_(has_local_consumer ? new WaiterList() : NULL),
       producer_in_two_phase_write_(false),
       consumer_in_two_phase_read_(false) {
-//FIXME
+  // Check that the passed in options actually are validated.
+  MojoCreateDataPipeOptions unused ALLOW_UNUSED = { 0 };
+  DCHECK_EQ(ValidateOptions(&validated_options, &unused), MOJO_RESULT_OK);
 }
 
 DataPipe::~DataPipe() {
diff --git a/mojo/system/local_data_pipe.cc b/mojo/system/local_data_pipe.cc
index 57c95a7..a23d524 100644
--- a/mojo/system/local_data_pipe.cc
+++ b/mojo/system/local_data_pipe.cc
@@ -44,7 +44,7 @@ MojoResult LocalDataPipe::ProducerWriteDataImplNoLock(const void* elements,
   DCHECK_GT(*num_bytes, 0u);
 
   // TODO(vtl): This implementation may write less than requested, even if room
-  // is available. Fix this.
+  // is available. Fix this. Note to self: Don't forget to wake up waiters.
   void* buffer = NULL;
   uint32_t buffer_num_bytes = *num_bytes;
   MojoResult rv = ProducerBeginWriteDataImplNoLock(&buffer,
@@ -97,9 +97,15 @@ 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());
   two_phase_max_num_bytes_written_ = 0;  // For safety.
+
+  if (was_empty && num_bytes_written > 0)
+    AwakeConsumerWaitersForStateChangeNoLock();
+
   return MOJO_RESULT_OK;
 }
 
@@ -131,7 +137,7 @@ MojoResult LocalDataPipe::ConsumerReadDataImplNoLock(void* elements,
   DCHECK_GT(*num_bytes, 0u);
 
   // TODO(vtl): This implementation may write less than requested, even if room
-  // is available. Fix this.
+  // is available. Fix this. Note to self: Don't forget to wake up waiters.
   const void* buffer = NULL;
   uint32_t buffer_num_bytes = *num_bytes;
   MojoResult rv = ConsumerBeginReadDataImplNoLock(&buffer, &buffer_num_bytes,
@@ -165,13 +171,15 @@ MojoResult LocalDataPipe::ConsumerDiscardDataImplNoLock(uint32_t* num_bytes,
   if (current_num_bytes_ == 0)
     return MOJO_RESULT_NOT_FOUND;
 
+  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;
 
-  AwakeProducerWaitersForStateChangeNoLock();
-  AwakeConsumerWaitersForStateChangeNoLock();
+  if (was_full && num_bytes_to_discard > 0)
+    AwakeProducerWaitersForStateChangeNoLock();
 
   *num_bytes = static_cast<uint32_t>(num_bytes_to_discard);
   return MOJO_RESULT_OK;
@@ -211,12 +219,18 @@ 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;
   two_phase_max_num_bytes_read_ = 0;  // For safety.
+
+  if (was_full && num_bytes_read > 0)
+    AwakeProducerWaitersForStateChangeNoLock();
+
   return MOJO_RESULT_OK;
 }
 
diff --git a/mojo/system/local_data_pipe_unittest.cc b/mojo/system/local_data_pipe_unittest.cc
index a5099c8..0176322 100644
--- a/mojo/system/local_data_pipe_unittest.cc
+++ b/mojo/system/local_data_pipe_unittest.cc
@@ -7,6 +7,7 @@
 #include "base/basictypes.h"
 #include "base/memory/ref_counted.h"
 #include "mojo/system/data_pipe.h"
+#include "mojo/system/waiter.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
 namespace mojo {
@@ -174,13 +175,91 @@ TEST(LocalDataPipeTest, SimpleReadWrite) {
   EXPECT_EQ(MOJO_RESULT_OK, dp->ConsumerQueryData(&num_bytes));
   EXPECT_EQ(0u, num_bytes);
 
-  // TODO(vtl): More: discard (with/without "all or none"). More "all or none"
-  // tests.
-
   dp->ProducerClose();
   dp->ConsumerClose();
 }
 
+// TODO(vtl): BasicProducerWaiting
+
+TEST(LocalDataPipeTest, BasicConsumerWaiting) {
+  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;
+
+    // Never writable.
+    waiter.Init();
+    EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
+              dp->ConsumerAddWaiter(&waiter, MOJO_WAIT_FLAG_WRITABLE, 12));
+
+    // Not yet readable.
+    waiter.Init();
+    EXPECT_EQ(MOJO_RESULT_OK,
+              dp->ConsumerAddWaiter(&waiter, MOJO_WAIT_FLAG_READABLE, 34));
+    EXPECT_EQ(MOJO_RESULT_DEADLINE_EXCEEDED, waiter.Wait(0));
+    dp->ConsumerRemoveWaiter(&waiter);
+
+    // 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));
+
+    // Should already be readable.
+    waiter.Init();
+    EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
+              dp->ConsumerAddWaiter(&waiter, MOJO_WAIT_FLAG_READABLE, 56));
+
+    // Discard one element.
+    num_bytes = static_cast<uint32_t>(sizeof(elements[0]));
+    EXPECT_EQ(MOJO_RESULT_OK, dp->ConsumerDiscardData(&num_bytes, true));
+    EXPECT_EQ(static_cast<uint32_t>(sizeof(elements[0])), num_bytes);
+
+    // Should still be readable.
+    waiter.Init();
+    EXPECT_EQ(MOJO_RESULT_ALREADY_EXISTS,
+              dp->ConsumerAddWaiter(&waiter, MOJO_WAIT_FLAG_READABLE, 78));
+
+    // Discard another element.
+    num_bytes = static_cast<uint32_t>(sizeof(elements[0]));
+    EXPECT_EQ(MOJO_RESULT_OK, dp->ConsumerDiscardData(&num_bytes, true));
+    EXPECT_EQ(static_cast<uint32_t>(sizeof(elements[0])), num_bytes);
+
+    // Adding a waiter should now succeed.
+    waiter.Init();
+    EXPECT_EQ(MOJO_RESULT_OK,
+              dp->ConsumerAddWaiter(&waiter, MOJO_WAIT_FLAG_READABLE, 90));
+
+    // Write one element.
+    num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
+    EXPECT_EQ(MOJO_RESULT_OK,
+              dp->ProducerWriteData(elements, &num_bytes, true));
+
+    // Waiting should now succeed.
+    EXPECT_EQ(90, waiter.Wait(1000));
+    dp->ConsumerRemoveWaiter(&waiter);
+
+//FIXME
+
+//FIXME
+    dp->ProducerClose();
+    dp->ConsumerClose();
+  }
+
+}
+
+// TODO(vtl): More: discard (with/without "all or none"). More "all or none"
+// tests.
+
 // TODO(vtl): More.
 
 }  // namespace