Add tests for TimerManager (#30758)

* Add tests for TimerManager

* Automated change: Fix sanity tests

Co-authored-by: drfloob <drfloob@users.noreply.github.com>

BUILD

@@ -2376,6 +2376,7 @@ grpc_cc_library(
         "gpr",
         "gpr_codegen",
         "gpr_tls",
+        "grpc_trace",
         "posix_event_engine_timer",
         "time",
     ],

CMakeLists.txt

@@ -1206,6 +1206,7 @@ if(gRPC_BUILD_TESTS)
   endif()
   add_dependencies(buildtests_cxx time_util_test)
   add_dependencies(buildtests_cxx timeout_encoding_test)
+  add_dependencies(buildtests_cxx timer_manager_test)
   add_dependencies(buildtests_cxx timer_test)
   add_dependencies(buildtests_cxx tls_certificate_verifier_test)
   add_dependencies(buildtests_cxx tls_key_export_test)
@@ -18556,6 +18557,41 @@ target_link_libraries(timeout_encoding_test
 )
 
 
+endif()
+if(gRPC_BUILD_TESTS)
+
+add_executable(timer_manager_test
+  test/core/event_engine/posix/timer_manager_test.cc
+  third_party/googletest/googletest/src/gtest-all.cc
+  third_party/googletest/googlemock/src/gmock-all.cc
+)
+
+target_include_directories(timer_manager_test
+  PRIVATE
+    ${CMAKE_CURRENT_SOURCE_DIR}
+    ${CMAKE_CURRENT_SOURCE_DIR}/include
+    ${_gRPC_ADDRESS_SORTING_INCLUDE_DIR}
+    ${_gRPC_RE2_INCLUDE_DIR}
+    ${_gRPC_SSL_INCLUDE_DIR}
+    ${_gRPC_UPB_GENERATED_DIR}
+    ${_gRPC_UPB_GRPC_GENERATED_DIR}
+    ${_gRPC_UPB_INCLUDE_DIR}
+    ${_gRPC_XXHASH_INCLUDE_DIR}
+    ${_gRPC_ZLIB_INCLUDE_DIR}
+    third_party/googletest/googletest/include
+    third_party/googletest/googletest
+    third_party/googletest/googlemock/include
+    third_party/googletest/googlemock
+    ${_gRPC_PROTO_GENS_DIR}
+)
+
+target_link_libraries(timer_manager_test
+  ${_gRPC_PROTOBUF_LIBRARIES}
+  ${_gRPC_ALLTARGETS_LIBRARIES}
+  grpc_test_util
+)
+
+
 endif()
 if(gRPC_BUILD_TESTS)
 

build_autogenerated.yaml

@@ -10274,6 +10274,17 @@ targets:
   deps:
   - grpc_test_util
   uses_polling: false
+- name: timer_manager_test
+  gtest: true
+  build: test
+  language: c++
+  headers:
+  - src/core/lib/event_engine/common_closures.h
+  src:
+  - test/core/event_engine/posix/timer_manager_test.cc
+  deps:
+  - grpc_test_util
+  uses_polling: false
 - name: timer_test
   gtest: true
   build: test

src/core/lib/event_engine/posix_engine/timer_manager.cc

@@ -32,6 +32,7 @@
 #include <grpc/support/log.h>
 #include <grpc/support/time.h>
 
+#include "src/core/lib/debug/trace.h"
 #include "src/core/lib/gpr/tls.h"
 #include "src/core/lib/gprpp/thd.h"
 
@@ -40,6 +41,8 @@ static GPR_THREAD_LOCAL(bool) g_timer_thread;
 namespace grpc_event_engine {
 namespace posix_engine {
 
+grpc_core::DebugOnlyTraceFlag grpc_event_engine_timer_trace(false, "timer");
+
 namespace {
 class ThreadCollector {
  public:
@@ -201,7 +204,15 @@ void TimerManager::MainLoop() {
 void TimerManager::RunThread(void* arg) {
   g_timer_thread = true;
   std::unique_ptr<RunThreadArgs> thread(static_cast<RunThreadArgs*>(arg));
+  if (grpc_event_engine_timer_trace.enabled()) {
+    gpr_log(GPR_DEBUG, "TimerManager::%p starting thread::%p", thread->self,
+            &thread->thread);
+  }
   thread->self->Run(std::move(thread->thread));
+  if (grpc_event_engine_timer_trace.enabled()) {
+    gpr_log(GPR_DEBUG, "TimerManager::%p thread::%p finished", thread->self,
+            &thread->thread);
+  }
 }
 
 void TimerManager::Run(grpc_core::Thread thread) {
@@ -235,14 +246,24 @@ bool TimerManager::TimerCancel(Timer* timer) {
 }
 
 TimerManager::~TimerManager() {
+  if (grpc_event_engine_timer_trace.enabled()) {
+    gpr_log(GPR_DEBUG, "TimerManager::%p shutting down", this);
+  }
   ThreadCollector collector;
   grpc_core::MutexLock lock(&mu_);
   shutdown_ = true;
   cv_wait_.SignalAll();
   while (thread_count_ > 0) {
     cv_threadcount_.Wait(&mu_);
+    if (grpc_event_engine_timer_trace.enabled()) {
+      gpr_log(GPR_DEBUG, "TimerManager::%p waiting for %zu threads to finish",
+              this, thread_count_);
+    }
   }
   collector.Collect(std::move(completed_threads_));
+  if (grpc_event_engine_timer_trace.enabled()) {
+    gpr_log(GPR_DEBUG, "TimerManager::%p shutdown complete", this);
+  }
 }
 
 void TimerManager::Host::Kick() { timer_manager_->Kick(); }

test/core/event_engine/posix/BUILD

@@ -46,6 +46,21 @@ grpc_cc_test(
     ],
 )
 
+grpc_cc_test(
+    name = "timer_manager_test",
+    srcs = ["timer_manager_test.cc"],
+    external_deps = ["gtest"],
+    language = "C++",
+    uses_event_engine = False,
+    uses_polling = False,
+    deps = [
+        "//:common_event_engine_closures",
+        "//:exec_ctx",
+        "//:posix_event_engine_timer_manager",
+        "//test/core/util:grpc_test_util",
+    ],
+)
+
 grpc_cc_test(
     name = "event_poller_posix_test",
     srcs = ["event_poller_posix_test.cc"],

test/core/event_engine/posix/timer_manager_test.cc

@@ -0,0 +1,100 @@
+// Copyright 2022 The 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 "src/core/lib/event_engine/posix_engine/timer_manager.h"
+
+#include <string.h>
+
+#include <atomic>
+#include <cstdint>
+#include <limits>
+#include <random>
+
+#include <gtest/gtest.h>
+
+#include <grpc/grpc.h>
+
+#include "src/core/lib/event_engine/common_closures.h"
+#include "src/core/lib/event_engine/posix_engine/timer.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "test/core/util/test_config.h"
+
+namespace grpc_event_engine {
+namespace posix_engine {
+
+TEST(TimerManagerTest, StressTest) {
+  grpc_core::ExecCtx exec_ctx;
+  auto now = exec_ctx.Now();
+  auto test_deadline = now + grpc_core::Duration::Seconds(15);
+  std::vector<Timer> timers;
+  constexpr int kTimerCount = 500;
+  timers.resize(kTimerCount);
+  std::atomic_int called{0};
+  std::random_device rd;
+  std::mt19937 gen(rd());
+  std::uniform_real_distribution<> dis_millis(100, 3000);
+  {
+    TimerManager manager;
+    for (auto& timer : timers) {
+      exec_ctx.InvalidateNow();
+      manager.TimerInit(
+          &timer, now + grpc_core::Duration::Milliseconds(dis_millis(gen)),
+          experimental::SelfDeletingClosure::Create([&called]() {
+            absl::SleepFor(absl::Milliseconds(50));
+            ++called;
+          }));
+    }
+    // Wait for all callbacks to have been called
+    while (called.load(std::memory_order_relaxed) < kTimerCount) {
+      exec_ctx.InvalidateNow();
+      if (exec_ctx.Now() > test_deadline) {
+        FAIL() << "Deadline exceeded. "
+               << called.load(std::memory_order_relaxed) << "/" << kTimerCount
+               << " callbacks executed";
+      }
+      gpr_log(GPR_DEBUG, "Processed %d/%d callbacks",
+              called.load(std::memory_order_relaxed), kTimerCount);
+      absl::SleepFor(absl::Milliseconds(333));
+    }
+  }
+}
+
+TEST(TimerManagerTest, ShutDownBeforeAllCallbacksAreExecuted) {
+  // Should the internal timer_list complain in this scenario?
+  grpc_core::ExecCtx exec_ctx;
+  std::vector<Timer> timers;
+  constexpr int kTimerCount = 100;
+  timers.resize(kTimerCount);
+  std::atomic_int called{0};
+  experimental::AnyInvocableClosure closure([&called] { ++called; });
+  {
+    TimerManager manager;
+    for (auto& timer : timers) {
+      manager.TimerInit(&timer, grpc_core::Timestamp::InfFuture(), &closure);
+    }
+  }
+  ASSERT_EQ(called.load(), 0);
+}
+
+}  // namespace posix_engine
+}  // namespace grpc_event_engine
+
+int main(int argc, char** argv) {
+  grpc::testing::TestEnvironment env(&argc, argv);
+  ::testing::InitGoogleTest(&argc, argv);
+  grpc_init();
+  int ret = RUN_ALL_TESTS();
+  grpc_shutdown();
+  return ret;
+}

tools/run_tests/generated/tests.json

@@ -7047,6 +7047,30 @@
     ],
     "uses_polling": false
   },
+  {
+    "args": [],
+    "benchmark": false,
+    "ci_platforms": [
+      "linux",
+      "mac",
+      "posix",
+      "windows"
+    ],
+    "cpu_cost": 1.0,
+    "exclude_configs": [],
+    "exclude_iomgrs": [],
+    "flaky": false,
+    "gtest": true,
+    "language": "c++",
+    "name": "timer_manager_test",
+    "platforms": [
+      "linux",
+      "mac",
+      "posix",
+      "windows"
+    ],
+    "uses_polling": false
+  },
   {
     "args": [],
     "benchmark": false,