Mojo: DataPipe: Implement "may discard" mode for simple writes.

(Still need to implement for two-phase writes.)

R=darin@chromium.org

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

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

mojo/system/data_pipe.cc

@@ -37,6 +37,9 @@ MojoResult DataPipe::ValidateOptions(
     return MOJO_RESULT_INVALID_ARGUMENT;
   out_options->struct_size = static_cast<uint32_t>(sizeof(*out_options));
 
+  // All flags are okay (unrecognized flags will be ignored).
+  out_options->flags = in_options->flags;
+
   if (in_options->element_num_bytes == 0)
     return MOJO_RESULT_INVALID_ARGUMENT;
   out_options->element_num_bytes = in_options->element_num_bytes;

mojo/system/local_data_pipe.cc

@@ -51,13 +51,30 @@ MojoResult LocalDataPipe::ProducerWriteDataImplNoLock(const void* elements,
   DCHECK_EQ(*num_bytes % element_num_bytes(), 0u);
   DCHECK_GT(*num_bytes, 0u);
 
-  // TODO(vtl): Consider this return value.
-  if (all_or_none && *num_bytes > capacity_num_bytes() - current_num_bytes_)
-    return MOJO_RESULT_OUT_OF_RANGE;
-
-  size_t num_bytes_to_write =
-      std::min(static_cast<size_t>(*num_bytes),
-               capacity_num_bytes() - current_num_bytes_);
+  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;
+    }
+  } else {
+    if (all_or_none && *num_bytes > capacity_num_bytes() - current_num_bytes_) {
+      return (*num_bytes > capacity_num_bytes()) ? MOJO_RESULT_OUT_OF_RANGE :
+                                                   MOJO_RESULT_SHOULD_WAIT;
+    }
+
+    num_bytes_to_write = std::min(static_cast<size_t>(*num_bytes),
+                                  capacity_num_bytes() - current_num_bytes_);
+  }
   if (num_bytes_to_write == 0)
     return MOJO_RESULT_SHOULD_WAIT;
 

mojo/system/local_data_pipe_unittest.cc

@@ -476,8 +476,110 @@ TEST(LocalDataPipeTest, BasicConsumerWaiting) {
   }
 }
 
-// TODO(vtl): More: discard (with/without "all or none"). More "all or none"
-// tests.
+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);
+}
+
+TEST(LocalDataPipeTest, MayDiscard) {
+  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|.
+    10 * 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));
+
+  int32_t buffer[100] = { 0 };
+  uint32_t num_bytes = 0;
+
+  num_bytes = 20u * sizeof(int32_t);
+  Seq(0, arraysize(buffer), buffer);
+  // Try writing more than capacity. (This test relies on the implementation
+  // enforcing the capacity strictly.)
+  EXPECT_EQ(MOJO_RESULT_OK, dp->ProducerWriteData(buffer, &num_bytes, false));
+  EXPECT_EQ(10u * sizeof(int32_t), num_bytes);
+
+  // Read half of what we wrote.
+  num_bytes = 5u * sizeof(int32_t);
+  memset(buffer, 0xab, sizeof(buffer));
+  EXPECT_EQ(MOJO_RESULT_OK, dp->ConsumerReadData(buffer, &num_bytes, false));
+  EXPECT_EQ(5u * sizeof(int32_t), num_bytes);
+  int32_t expected_buffer[100];
+  memset(expected_buffer, 0xab, sizeof(expected_buffer));
+  Seq(0, 5u, expected_buffer);
+  EXPECT_EQ(0, memcmp(buffer, expected_buffer, sizeof(buffer)));
+  // Internally, a circular buffer would now look like:
+  //   -, -, -, -, -, 5, 6, 7, 8, 9
+
+  // Write a bit more than the space that's available.
+  num_bytes = 8u * sizeof(int32_t);
+  Seq(100, arraysize(buffer), buffer);
+  EXPECT_EQ(MOJO_RESULT_OK, dp->ProducerWriteData(buffer, &num_bytes, false));
+  EXPECT_EQ(8u * sizeof(int32_t), num_bytes);
+  // Internally, a circular buffer would now look like:
+  //   100, 101, 102, 103, 104, 105, 106, 107, 8, 9
+
+  // Read half of what's available.
+  num_bytes = 5u * sizeof(int32_t);
+  memset(buffer, 0xab, sizeof(buffer));
+  EXPECT_EQ(MOJO_RESULT_OK, dp->ConsumerReadData(buffer, &num_bytes, false));
+  EXPECT_EQ(5u * sizeof(int32_t), num_bytes);
+  memset(expected_buffer, 0xab, sizeof(expected_buffer));
+  expected_buffer[0] = 8;
+  expected_buffer[1] = 9;
+  expected_buffer[2] = 100;
+  expected_buffer[3] = 101;
+  expected_buffer[4] = 102;
+  EXPECT_EQ(0, memcmp(buffer, expected_buffer, sizeof(buffer)));
+  // Internally, a circular buffer would now look like:
+  //   -, -, -, 103, 104, 105, 106, 107, -, -
+
+  // Write one integer.
+  num_bytes = 1u * sizeof(int32_t);
+  Seq(200, arraysize(buffer), buffer);
+  EXPECT_EQ(MOJO_RESULT_OK, dp->ProducerWriteData(buffer, &num_bytes, false));
+  EXPECT_EQ(1u * sizeof(int32_t), num_bytes);
+  // Internally, a circular buffer would now look like:
+  //   -, -, -, 103, 104, 105, 106, 107, 200, -
+
+  // Write five more.
+  num_bytes = 5u * sizeof(int32_t);
+  Seq(300, arraysize(buffer), buffer);
+  EXPECT_EQ(MOJO_RESULT_OK, dp->ProducerWriteData(buffer, &num_bytes, false));
+  EXPECT_EQ(5u * sizeof(int32_t), num_bytes);
+  // Internally, a circular buffer would now look like:
+  //   301, 302, 303, 304, 104, 105, 106, 107, 200, 300
+
+  // Read it all.
+  num_bytes = sizeof(buffer);
+  memset(buffer, 0xab, sizeof(buffer));
+  EXPECT_EQ(MOJO_RESULT_OK, dp->ConsumerReadData(buffer, &num_bytes, false));
+  EXPECT_EQ(10u * sizeof(int32_t), num_bytes);
+  memset(expected_buffer, 0xab, sizeof(expected_buffer));
+  expected_buffer[0] = 104;
+  expected_buffer[1] = 105;
+  expected_buffer[2] = 106;
+  expected_buffer[3] = 107;
+  expected_buffer[4] = 200;
+  expected_buffer[5] = 300;
+  expected_buffer[6] = 301;
+  expected_buffer[7] = 302;
+  expected_buffer[8] = 303;
+  expected_buffer[9] = 304;
+  EXPECT_EQ(0, memcmp(buffer, expected_buffer, sizeof(buffer)));
+
+  // TODO(vtl): Test two-phase write when it supports "may discard".
+
+  dp->ProducerClose();
+  dp->ConsumerClose();
+}
+
+// TODO(vtl): More "all or none" tests (without and with "may discard").
 
 // Tests that |ProducerWriteData()| and |ConsumerReadData()| writes and reads,
 // respectively, as much as possible, even if it has to "wrap around" the