mojo: do not dispatch SimpleWatcher callback twice upon ...FAILED_PRE…

…CONDITION

We used to fire DataPipeDrainer::Client::OnDataComplete() twice
because underlying SimpleWatcher fired its client callback twice when
MOJO_RESULT_FAILED_PRECONDITION occurs after the remote end of data pipe
is closed.

Change-Id: I241aa6648dec79c2cf60e9c32de70a52d59d3412
Reviewed-on: https://chromium-review.googlesource.com/958094
Reviewed-by: Ken Rockot <rockot@chromium.org>
Commit-Queue: Andrey Kosyakov <caseq@chromium.org>
Cr-Commit-Position: refs/heads/master@{#542353}

mojo/common/BUILD.gn

@@ -89,6 +89,7 @@ test("mojo_common_unittests") {
 
   sources = [
     "common_custom_types_unittest.cc",
+    "data_pipe_drainer_unittest.cc",
     "struct_traits_unittest.cc",
   ]
 }

mojo/common/data_pipe_drainer_unittest.cc

@@ -0,0 +1,65 @@
+// Copyright 2018 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/common/data_pipe_drainer.h"
+#include "base/callback.h"
+#include "base/message_loop/message_loop.h"
+#include "base/run_loop.h"
+#include "base/values.h"
+#include "mojo/public/cpp/bindings/binding.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace mojo {
+namespace common {
+namespace test {
+
+template <typename Functor>
+base::RepeatingClosure BindLambda(Functor callable) {
+  return base::BindRepeating([](Functor callable) { callable(); }, callable);
+}
+
+class DataPipeDrainerTest : public testing::Test,
+                            public DataPipeDrainer::Client {
+ protected:
+  DataPipeDrainerTest() {
+    DataPipe pipe;
+    drainer_ = std::make_unique<DataPipeDrainer>(
+        this, std::move(pipe.consumer_handle));
+    producer_handle_ = std::move(pipe.producer_handle);
+  }
+
+  ScopedDataPipeProducerHandle producer_handle_;
+  base::RepeatingClosure completion_callback_;
+
+  void OnDataAvailable(const void* data, size_t num_bytes) override {
+    data_.append(static_cast<const char*>(data), num_bytes);
+  }
+
+  void OnDataComplete() override { completion_callback_.Run(); }
+
+  base::MessageLoop message_loop_;
+  std::string data_;
+  std::unique_ptr<DataPipeDrainer> drainer_;
+
+  DISALLOW_COPY_AND_ASSIGN(DataPipeDrainerTest);
+};
+
+TEST_F(DataPipeDrainerTest, TestCompleteIsCalledOnce) {
+  bool had_data_complete = false;
+
+  completion_callback_ = BindLambda([&had_data_complete]() {
+    EXPECT_FALSE(had_data_complete);
+    had_data_complete = true;
+  });
+  uint32_t size = 5;
+  EXPECT_EQ(MOJO_RESULT_OK, producer_handle_->WriteData(
+                                "hello", &size, MOJO_WRITE_DATA_FLAG_NONE));
+  base::RunLoop().RunUntilIdle();
+  producer_handle_.reset();
+  base::RunLoop().RunUntilIdle();
+}
+
+}  // namespace test
+}  // namespace common
+}  // namespace mojo

mojo/public/cpp/system/simple_watcher.cc

@@ -274,13 +274,10 @@ void SimpleWatcher::OnHandleReady(int watch_id,
     if (!weak_self)
       return;
 
-    if (unsatisfiable_)
-      return;
-
     // Prevent |MOJO_RESULT_FAILED_PRECONDITION| task spam by only notifying
     // at most once in AUTOMATIC arming mode.
     if (result == MOJO_RESULT_FAILED_PRECONDITION)
-      unsatisfiable_ = true;
+      return;
 
     if (arming_policy_ == ArmingPolicy::AUTOMATIC && IsWatching())
       ArmOrNotify();

mojo/public/cpp/system/simple_watcher.h

@@ -229,10 +229,6 @@ class MOJO_CPP_SYSTEM_EXPORT SimpleWatcher {
   // The callback to call when the handle is signaled.
   ReadyCallbackWithState callback_;
 
-  // Tracks if the SimpleWatcher has already notified of unsatisfiability. This
-  // is used to prevent redundant notifications in AUTOMATIC mode.
-  bool unsatisfiable_ = false;
-
   // Tag used to ID memory allocations that originated from notifications in
   // this watcher.
   const char* heap_profiler_tag_ = nullptr;

mojo/public/cpp/system/tests/simple_watcher_unittest.cc

@@ -13,6 +13,7 @@
 #include "base/run_loop.h"
 #include "base/threading/thread_task_runner_handle.h"
 #include "mojo/public/c/system/types.h"
+#include "mojo/public/cpp/system/data_pipe.h"
 #include "mojo/public/cpp/system/message_pipe.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
@@ -80,6 +81,39 @@ TEST_F(SimpleWatcherTest, WatchUnsatisfiable) {
   EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION, b_watcher.Arm());
 }
 
+TEST_F(SimpleWatcherTest, WatchFailedPreconditionNoSpam) {
+  DataPipe pipe;
+  bool had_failed_precondition = false;
+
+  SimpleWatcher watcher(FROM_HERE, SimpleWatcher::ArmingPolicy::AUTOMATIC);
+  MojoResult rc =
+      watcher.Watch(pipe.consumer_handle.get(), MOJO_HANDLE_SIGNAL_READABLE,
+                    OnReady([&](MojoResult result) {
+                      EXPECT_FALSE(had_failed_precondition);
+                      switch (result) {
+                        case MOJO_RESULT_OK:
+                          const void* begin;
+                          uint32_t num_bytes;
+                          pipe.consumer_handle->BeginReadData(
+                              &begin, &num_bytes, MOJO_READ_DATA_FLAG_NONE);
+                          pipe.consumer_handle->EndReadData(num_bytes);
+                          break;
+                        case MOJO_RESULT_FAILED_PRECONDITION:
+                          had_failed_precondition = true;
+                          break;
+                      }
+                    }));
+  EXPECT_EQ(MOJO_RESULT_OK, rc);
+
+  uint32_t size = 5;
+  EXPECT_EQ(MOJO_RESULT_OK, pipe.producer_handle->WriteData(
+                                "hello", &size, MOJO_WRITE_DATA_FLAG_NONE));
+  base::RunLoop().RunUntilIdle();
+  pipe.producer_handle.reset();
+  base::RunLoop().RunUntilIdle();
+  EXPECT_TRUE(had_failed_precondition);
+}
+
 TEST_F(SimpleWatcherTest, WatchInvalidHandle) {
   ScopedMessagePipeHandle a, b;
   CreateMessagePipe(nullptr, &a, &b);