Add common EventEngine Closure types (#30438)

* Add common EventEngine Closure types

These have been written a few times, so let's standardize on a spot for
them.

* sanitize

* format

BUILD

@@ -2261,6 +2261,16 @@ grpc_cc_library(
     ],
 )
 
+grpc_cc_library(
+    name = "common_event_engine_closures",
+    hdrs = ["src/core/lib/event_engine/common_closures.h"],
+    external_deps = ["absl/functional:any_invocable"],
+    deps = [
+        "event_engine_base_hdrs",
+        "gpr_platform",
+    ],
+)
+
 grpc_cc_library(
     name = "iomgr_ee_timer",
     srcs = [

CMakeLists.txt

@@ -903,6 +903,7 @@ if(gRPC_BUILD_TESTS)
   if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX)
     add_dependencies(buildtests_cxx combiner_test)
   endif()
+  add_dependencies(buildtests_cxx common_closures_test)
   add_dependencies(buildtests_cxx completion_queue_threading_test)
   add_dependencies(buildtests_cxx compression_test)
   add_dependencies(buildtests_cxx concurrent_connectivity_test)
@@ -8158,6 +8159,41 @@ endif()
 endif()
 if(gRPC_BUILD_TESTS)
 
+add_executable(common_closures_test
+  test/core/event_engine/common_closures_test.cc
+  third_party/googletest/googletest/src/gtest-all.cc
+  third_party/googletest/googlemock/src/gmock-all.cc
+)
+
+target_include_directories(common_closures_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(common_closures_test
+  ${_gRPC_PROTOBUF_LIBRARIES}
+  ${_gRPC_ALLTARGETS_LIBRARIES}
+  grpc_test_util_unsecure
+)
+
+
+endif()
+if(gRPC_BUILD_TESTS)
+
 add_executable(completion_queue_threading_test
   test/core/surface/completion_queue_threading_test.cc
   third_party/googletest/googletest/src/gtest-all.cc

build_autogenerated.yaml

@@ -5186,6 +5186,16 @@ targets:
   - linux
   - posix
   - mac
+- name: common_closures_test
+  gtest: true
+  build: test
+  language: c++
+  headers:
+  - src/core/lib/event_engine/common_closures.h
+  src:
+  - test/core/event_engine/common_closures_test.cc
+  deps:
+  - grpc_test_util_unsecure
 - name: completion_queue_threading_test
   gtest: true
   build: test

src/core/lib/event_engine/common_closures.h

@@ -0,0 +1,71 @@
+// Copyright 2022 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_CORE_LIB_EVENT_ENGINE_COMMON_CLOSURES_H
+#define GRPC_CORE_LIB_EVENT_ENGINE_COMMON_CLOSURES_H
+
+#include <grpc/support/port_platform.h>
+
+#include <utility>
+
+#include "absl/functional/any_invocable.h"
+
+#include <grpc/event_engine/event_engine.h>
+
+namespace grpc_event_engine {
+namespace experimental {
+
+class AnyInvocableClosure : public EventEngine::Closure {
+ public:
+  explicit AnyInvocableClosure(absl::AnyInvocable<void()> cb)
+      : cb_(std::move(cb)) {}
+  void Run() override { cb_(); }
+
+ private:
+  absl::AnyInvocable<void()> cb_;
+};
+
+class SelfDeletingClosure : public EventEngine::Closure {
+ public:
+  // Creates a SelfDeletingClosure.
+  // The closure will be deleted after Run is called.
+  static Closure* Create(absl::AnyInvocable<void()> cb) {
+    return new SelfDeletingClosure(std::move(cb), nullptr);
+  }
+  // Creates a SelfDeletingClosure with a custom destructor.
+  static Closure* Create(absl::AnyInvocable<void()> cb,
+                         absl::AnyInvocable<void()> dest_cb) {
+    return new SelfDeletingClosure(std::move(cb), std::move(dest_cb));
+  }
+  ~SelfDeletingClosure() override {
+    if (dest_cb_) dest_cb_();
+  };
+
+  void Run() override {
+    cb_();
+    delete this;
+  }
+
+ private:
+  explicit SelfDeletingClosure(absl::AnyInvocable<void()> cb,
+                               absl::AnyInvocable<void()> dest_cb)
+      : cb_(std::move(cb)), dest_cb_(std::move(dest_cb)) {}
+  absl::AnyInvocable<void()> cb_;
+  absl::AnyInvocable<void()> dest_cb_;
+};
+
+}  // namespace experimental
+}  // namespace grpc_event_engine
+
+#endif  // GRPC_CORE_LIB_EVENT_ENGINE_COMMON_CLOSURES_H

src/core/lib/event_engine/promise.h

@@ -15,6 +15,8 @@
 #define GRPC_CORE_LIB_EVENT_ENGINE_PROMISE_H
 #include <grpc/support/port_platform.h>
 
+#include "absl/time/time.h"
+
 #include <grpc/support/log.h>
 
 #include "src/core/lib/gprpp/sync.h"
@@ -32,10 +34,13 @@ class Promise {
  public:
   // The getter will wait until the setter has been called, and will return the
   // value passed during Set.
-  T& Get() {
+  T& Get() { return WaitWithTimeout(absl::Hours(1)); }
+  // The getter will wait with timeout until the setter has been called, and
+  // will return the value passed during Set.
+  T& WaitWithTimeout(absl::Duration d) {
     grpc_core::MutexLock lock(&mu_);
     if (!set_) {
-      cv_.Wait(&mu_);
+      cv_.WaitWithTimeout(&mu_, d);
     }
     return val_;
   }

test/core/event_engine/BUILD

@@ -21,6 +21,18 @@ grpc_package(
     visibility = "tests",
 )
 
+grpc_cc_test(
+    name = "common_closures_test",
+    srcs = ["common_closures_test.cc"],
+    external_deps = ["gtest"],
+    deps = [
+        "//:common_event_engine_closures",
+        "//:gpr_platform",
+        "//:grpc_base",
+        "//test/core/util:grpc_test_util_unsecure",
+    ],
+)
+
 grpc_cc_test(
     name = "endpoint_config_test",
     srcs = ["endpoint_config_test.cc"],

test/core/event_engine/common_closures_test.cc

@@ -0,0 +1,63 @@
+// Copyright 2022 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 <grpc/support/port_platform.h>
+
+#include "src/core/lib/event_engine/common_closures.h"
+
+#include <gtest/gtest.h>
+
+#include "src/core/lib/event_engine/promise.h"
+#include "test/core/util/test_config.h"
+
+using ::grpc_event_engine::experimental::AnyInvocableClosure;
+using ::grpc_event_engine::experimental::Promise;
+using ::grpc_event_engine::experimental::SelfDeletingClosure;
+
+class AnyInvocableClosureTest : public testing::Test {};
+
+TEST_F(AnyInvocableClosureTest, CallsItsFunction) {
+  Promise<bool> promise;
+  AnyInvocableClosure closure([&promise] { promise.Set(true); });
+  closure.Run();
+  ASSERT_TRUE(promise.WaitWithTimeout(absl::Seconds(3)));
+}
+
+class SelfDeletingClosureTest : public testing::Test {};
+
+TEST_F(SelfDeletingClosureTest, CallsItsFunction) {
+  Promise<bool> promise;
+  auto* closure =
+      SelfDeletingClosure::Create([&promise] { promise.Set(true); });
+  closure->Run();
+  ASSERT_TRUE(promise.WaitWithTimeout(absl::Seconds(3)));
+  // ASAN should catch if this closure is not deleted
+}
+
+TEST_F(SelfDeletingClosureTest, CallsItsFunctionAndIsDestroyed) {
+  Promise<bool> fn_called;
+  Promise<bool> destroyed;
+  auto* closure =
+      SelfDeletingClosure::Create([&fn_called] { fn_called.Set(true); },
+                                  [&destroyed] { destroyed.Set(true); });
+  closure->Run();
+  ASSERT_TRUE(fn_called.WaitWithTimeout(absl::Seconds(3)));
+  ASSERT_TRUE(destroyed.WaitWithTimeout(absl::Seconds(3)));
+}
+
+int main(int argc, char** argv) {
+  testing::InitGoogleTest(&argc, argv);
+  auto result = RUN_ALL_TESTS();
+  return result;
+}

tools/run_tests/generated/tests.json

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