EventEngine Test Suite: Timers (#27496)

A reusable test suite for EventEngine implementations.

To exercise a custom EventEngine, simply link against :event_engine_test_suite
and provide a testing main function that sets a custom EventEngine factory:

```
#include "path/to/my_custom_event_engine.h"
#include "src/core/event_engine/test_suite/event_engine_test.h"

int main(int argc, char** argv) {
  ::testing::InitGoogleTest(&argc, argv);
  SetEventEngineFactory(
      []() { return absl::make_unique<MyCustomEventEngine>(); });
  auto result = RUN_ALL_TESTS();
  return result;
}
```

test/core/event_engine/BUILD

@@ -1,4 +1,4 @@
-# Copyright 2017 gRPC authors.
+# Copyright 2021 gRPC authors.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-load("//bazel:grpc_build_system.bzl", "grpc_cc_test", "grpc_package")
+load("//bazel:grpc_build_system.bzl", "grpc_cc_library", "grpc_cc_test", "grpc_package")
 
 licenses(["notice"])
 
@@ -28,3 +28,21 @@ grpc_cc_test(
         "//test/core/util:grpc_test_util",
     ],
 )
+
+grpc_cc_library(
+    name = "event_engine_test_suite",
+    srcs = [
+        "test_suite/event_engine_test.cc",
+        "test_suite/timer_test.cc",
+    ],
+    hdrs = ["test_suite/event_engine_test.h"],
+    external_deps = [
+        "absl/random",
+        "gtest",
+    ],
+    language = "C++",
+    deps = [
+        "//:grpc",
+        "//test/core/util:grpc_test_util",
+    ],
+)

test/core/event_engine/test_suite/README.md

@@ -0,0 +1,35 @@
+A reusable test suite for EventEngine implementations.
+
+To exercise a custom EventEngine, simply link against `:event_engine_test_suite`
+and provide a testing `main` function that sets a custom EventEngine factory:
+
+```
+#include "path/to/my_custom_event_engine.h"
+#include "src/core/event_engine/test_suite/event_engine_test.h"
+
+int main(int argc, char** argv) {
+  ::testing::InitGoogleTest(&argc, argv);
+  SetEventEngineFactory(
+      []() { return absl::make_unique<MyCustomEventEngine>(); });
+  auto result = RUN_ALL_TESTS();
+  return result;
+}
+```
+
+And add a target to the `BUILD` file:
+
+```
+grpc_cc_test(
+    name = "my_custom_event_engine_test",
+    srcs = ["test_suite/my_custom_event_engine_test.cc"],
+    external_deps = [
+        "gtest",
+    ],
+    language = "C++",
+    uses_polling = False,
+    deps = [
+        ":event_engine_test_suite",
+        "//:grpc",
+    ],
+)
+```

test/core/event_engine/test_suite/event_engine_test.cc

@@ -0,0 +1,28 @@
+// Copyright 2021 gRPC authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+#include "test/core/event_engine/test_suite/event_engine_test.h"
+
+#include <gtest/gtest.h>
+
+#include <grpc/event_engine/event_engine.h>
+
+std::function<std::unique_ptr<grpc_event_engine::experimental::EventEngine>()>*
+    g_ee_factory = nullptr;
+
+void SetEventEngineFactory(
+    std::function<
+        std::unique_ptr<grpc_event_engine::experimental::EventEngine>()>
+        factory) {
+  testing::AddGlobalTestEnvironment(new EventEngineTestEnvironment(factory));
+}

test/core/event_engine/test_suite/event_engine_test.h

@@ -0,0 +1,60 @@
+// Copyright 2021 gRPC authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef GRPC_TEST_CORE_EVENT_ENGINE_TEST_SUITE_EVENT_ENGINE_TEST_H
+#define GRPC_TEST_CORE_EVENT_ENGINE_TEST_SUITE_EVENT_ENGINE_TEST_H
+#include <gtest/gtest.h>
+
+#include <grpc/event_engine/event_engine.h>
+
+#include "src/core/lib/gprpp/sync.h"
+
+extern std::function<
+    std::unique_ptr<grpc_event_engine::experimental::EventEngine>()>*
+    g_ee_factory;
+
+// Manages the lifetime of the global EventEngine factory.
+class EventEngineTestEnvironment : public testing::Environment {
+ public:
+  explicit EventEngineTestEnvironment(
+      std::function<
+          std::unique_ptr<grpc_event_engine::experimental::EventEngine>()>
+          factory)
+      : factory_(factory) {}
+
+  void SetUp() override { g_ee_factory = &factory_; }
+
+  void TearDown() override { g_ee_factory = nullptr; }
+
+ private:
+  std::function<std::unique_ptr<grpc_event_engine::experimental::EventEngine>()>
+      factory_;
+};
+
+class EventEngineTest : public testing::Test {
+ protected:
+  std::unique_ptr<grpc_event_engine::experimental::EventEngine>
+  NewEventEngine() {
+    GPR_ASSERT(g_ee_factory != nullptr);
+    return (*g_ee_factory)();
+  }
+};
+
+// Set a custom factory for the EventEngine test suite.
+void SetEventEngineFactory(
+    std::function<
+        std::unique_ptr<grpc_event_engine::experimental::EventEngine>()>
+        factory);
+
+#endif  // GRPC_TEST_CORE_EVENT_ENGINE_TEST_SUITE_EVENT_ENGINE_TEST_H

test/core/event_engine/test_suite/timer_test.cc

@@ -0,0 +1,172 @@
+// Copyright 2021 gRPC authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include <thread>
+
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+
+#include "absl/functional/bind_front.h"
+#include "absl/random/random.h"
+#include "absl/time/time.h"
+
+#include <grpc/event_engine/event_engine.h>
+#include <grpc/support/log.h>
+
+#include "src/core/lib/gprpp/sync.h"
+#include "test/core/event_engine/test_suite/event_engine_test.h"
+
+using ::testing::ElementsAre;
+
+class EventEngineTimerTest : public EventEngineTest {
+ public:
+  void ScheduleCheckCB(absl::Time when, std::atomic<int>* call_count,
+                       std::atomic<int>* fail_count, int total_expected);
+
+ protected:
+  grpc_core::Mutex mu_;
+  grpc_core::CondVar cv_;
+  bool signaled_ ABSL_GUARDED_BY(mu_) = false;
+};
+
+TEST_F(EventEngineTimerTest, ImmediateCallbackIsExecutedQuickly) {
+  auto engine = this->NewEventEngine();
+  grpc_core::MutexLock lock(&mu_);
+  engine->RunAt(absl::Now(), [this]() {
+    grpc_core::MutexLock lock(&mu_);
+    signaled_ = true;
+    cv_.Signal();
+  });
+  cv_.WaitWithTimeout(&mu_, absl::Seconds(5));
+  ASSERT_TRUE(signaled_);
+}
+
+TEST_F(EventEngineTimerTest, SupportsCancellation) {
+  auto engine = this->NewEventEngine();
+  auto handle = engine->RunAt(absl::InfiniteFuture(), []() {});
+  ASSERT_TRUE(engine->Cancel(handle));
+}
+
+TEST_F(EventEngineTimerTest, CancelledCallbackIsNotExecuted) {
+  {
+    auto engine = this->NewEventEngine();
+    auto handle = engine->RunAt(absl::InfiniteFuture(), [this]() {
+      grpc_core::MutexLock lock(&mu_);
+      signaled_ = true;
+    });
+    ASSERT_TRUE(engine->Cancel(handle));
+  }
+  // The engine is deleted, and all closures should have been flushed
+  grpc_core::MutexLock lock(&mu_);
+  ASSERT_FALSE(signaled_);
+}
+
+TEST_F(EventEngineTimerTest, TimersRespectScheduleOrdering) {
+  // Note: this is a brittle test if the first call to `RunAt` takes longer than
+  // the second callback's wait time.
+  std::vector<uint8_t> ordered;
+  uint8_t count = 0;
+  grpc_core::MutexLock lock(&mu_);
+  {
+    auto engine = this->NewEventEngine();
+    engine->RunAt(absl::Now() + absl::Seconds(1), [&]() {
+      grpc_core::MutexLock lock(&mu_);
+      ordered.push_back(2);
+      ++count;
+      cv_.Signal();
+    });
+    engine->RunAt(absl::Now(), [&]() {
+      grpc_core::MutexLock lock(&mu_);
+      ordered.push_back(1);
+      ++count;
+      cv_.Signal();
+    });
+    // Ensure both callbacks have run. Simpler than a mutex.
+    while (count != 2) {
+      cv_.WaitWithTimeout(&mu_, absl::Microseconds(100));
+    }
+  }
+  // The engine is deleted, and all closures should have been flushed beforehand
+  ASSERT_THAT(ordered, ElementsAre(1, 2));
+}
+
+TEST_F(EventEngineTimerTest, CancellingExecutedCallbackIsNoopAndReturnsFalse) {
+  auto engine = this->NewEventEngine();
+  grpc_core::MutexLock lock(&mu_);
+  auto handle = engine->RunAt(absl::Now(), [this]() {
+    grpc_core::MutexLock lock(&mu_);
+    signaled_ = true;
+    cv_.Signal();
+  });
+  cv_.WaitWithTimeout(&mu_, absl::Seconds(10));
+  ASSERT_TRUE(signaled_);
+  // The callback has run, and now we'll try to cancel it.
+  ASSERT_FALSE(engine->Cancel(handle));
+}
+
+void EventEngineTimerTest::ScheduleCheckCB(absl::Time when,
+                                           std::atomic<int>* call_count,
+                                           std::atomic<int>* fail_count,
+                                           int total_expected) {
+  // TODO(hork): make the EventEngine the time source of truth! libuv supports
+  // millis, absl::Time reports in nanos. This generic test will be hard-coded
+  // to the lowest common denominator until EventEngines can compare relative
+  // times with supported resolution.
+  int64_t now_millis = absl::ToUnixMillis(absl::Now());
+  int64_t when_millis = absl::ToUnixMillis(when);
+  EXPECT_LE(when_millis, now_millis);
+  if (when_millis > now_millis) ++(*fail_count);
+  if (++(*call_count) == total_expected) {
+    grpc_core::MutexLock lock(&mu_);
+    signaled_ = true;
+    cv_.Signal();
+  }
+}
+
+TEST_F(EventEngineTimerTest, StressTestTimersNotCalledBeforeScheduled) {
+  auto engine = this->NewEventEngine();
+  constexpr int thread_count = 100;
+  constexpr int call_count_per_thread = 100;
+  constexpr float timeout_min_seconds = 1;
+  constexpr float timeout_max_seconds = 10;
+  std::atomic<int> call_count{0};
+  std::atomic<int> failed_call_count{0};
+  std::vector<std::thread> threads;
+  threads.reserve(thread_count);
+  for (int thread_n = 0; thread_n < thread_count; ++thread_n) {
+    threads.emplace_back([&]() {
+      absl::BitGen bitgen;
+      for (int call_n = 0; call_n < call_count_per_thread; ++call_n) {
+        absl::Time when = absl::Now() + absl::Seconds(absl::Uniform(
+                                            bitgen, timeout_min_seconds,
+                                            timeout_max_seconds));
+        engine->RunAt(
+            when, absl::bind_front(&EventEngineTimerTest::ScheduleCheckCB, this,
+                                   when, &call_count, &failed_call_count,
+                                   thread_count * call_count_per_thread));
+      }
+    });
+  }
+  for (auto& t : threads) {
+    t.join();
+  }
+  grpc_core::MutexLock lock(&mu_);
+  // to protect against spurious wakeups.
+  while (!signaled_) {
+    cv_.Wait(&mu_);
+  }
+  gpr_log(GPR_DEBUG, "failed timer count: %d of %d", failed_call_count.load(),
+          thread_count * call_count);
+  ASSERT_EQ(0, failed_call_count.load());
+}