[promises] Add a bridge between callbacks and promises (#33792)

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

CMakeLists.txt

@@ -1381,6 +1381,7 @@ if(gRPC_BUILD_TESTS)
   add_dependencies(buildtests_cxx uuid_v4_test)
   add_dependencies(buildtests_cxx validation_errors_test)
   add_dependencies(buildtests_cxx varint_test)
+  add_dependencies(buildtests_cxx wait_for_callback_test)
   if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX)
     add_dependencies(buildtests_cxx wakeup_fd_posix_test)
   endif()
@@ -26088,6 +26089,46 @@ target_link_libraries(varint_test
 )
 
 
+endif()
+if(gRPC_BUILD_TESTS)
+
+add_executable(wait_for_callback_test
+  src/core/lib/promise/activity.cc
+  test/core/promise/wait_for_callback_test.cc
+  third_party/googletest/googletest/src/gtest-all.cc
+  third_party/googletest/googlemock/src/gmock-all.cc
+)
+target_compile_features(wait_for_callback_test PUBLIC cxx_std_14)
+target_include_directories(wait_for_callback_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(wait_for_callback_test
+  ${_gRPC_BASELIB_LIBRARIES}
+  ${_gRPC_PROTOBUF_LIBRARIES}
+  ${_gRPC_ZLIB_LIBRARIES}
+  ${_gRPC_ALLTARGETS_LIBRARIES}
+  absl::type_traits
+  absl::statusor
+  gpr
+)
+
+
 endif()
 if(gRPC_BUILD_TESTS)
 if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX)

build_autogenerated.yaml

@@ -16436,6 +16436,33 @@ targets:
   - gtest
   - grpc_test_util
   uses_polling: false
+- name: wait_for_callback_test
+  gtest: true
+  build: test
+  language: c++
+  headers:
+  - src/core/lib/gprpp/atomic_utils.h
+  - src/core/lib/gprpp/notification.h
+  - src/core/lib/gprpp/orphanable.h
+  - src/core/lib/gprpp/ref_counted.h
+  - src/core/lib/gprpp/ref_counted_ptr.h
+  - src/core/lib/promise/activity.h
+  - src/core/lib/promise/context.h
+  - src/core/lib/promise/detail/promise_factory.h
+  - src/core/lib/promise/detail/promise_like.h
+  - src/core/lib/promise/detail/status.h
+  - src/core/lib/promise/map.h
+  - src/core/lib/promise/poll.h
+  - src/core/lib/promise/wait_for_callback.h
+  - test/core/promise/test_wakeup_schedulers.h
+  src:
+  - src/core/lib/promise/activity.cc
+  - test/core/promise/wait_for_callback_test.cc
+  deps:
+  - absl/meta:type_traits
+  - absl/status:statusor
+  - gpr
+  uses_polling: false
 - name: wakeup_fd_posix_test
   gtest: true
   build: test

src/core/BUILD

@@ -489,6 +489,19 @@ grpc_cc_library(
     ],
 )
 
+grpc_cc_library(
+    name = "wait_for_callback",
+    hdrs = [
+        "lib/promise/wait_for_callback.h",
+    ],
+    external_deps = ["absl/base:core_headers"],
+    deps = [
+        "activity",
+        "poll",
+        "//:gpr",
+    ],
+)
+
 grpc_cc_library(
     name = "arena_promise",
     external_deps = ["absl/meta:type_traits"],

src/core/lib/promise/wait_for_callback.h

@@ -0,0 +1,69 @@
+// Copyright 2023 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_SRC_CORE_LIB_PROMISE_WAIT_FOR_CALLBACK_H
+#define GRPC_SRC_CORE_LIB_PROMISE_WAIT_FOR_CALLBACK_H
+
+#include <grpc/support/port_platform.h>
+
+#include <memory>
+#include <utility>
+
+#include "absl/base/thread_annotations.h"
+
+#include "src/core/lib/gprpp/sync.h"
+#include "src/core/lib/promise/activity.h"
+#include "src/core/lib/promise/poll.h"
+
+namespace grpc_core {
+
+// Bridge callback interfaces and promise interfaces.
+// This class helps bridge older callback interfaces with promises:
+// MakeWaitPromise() returns a promise that will wait until a callback created
+// by MakeCallback() has been invoked.
+class WaitForCallback {
+ public:
+  // Creates a promise that blocks until the callback is invoked.
+  auto MakeWaitPromise() {
+    return [state = state_]() -> Poll<Empty> {
+      MutexLock lock(&state->mutex);
+      if (state->done) return Empty{};
+      state->waker = Activity::current()->MakeNonOwningWaker();
+      return Pending{};
+    };
+  }
+
+  // Creates a callback that unblocks the promise.
+  auto MakeCallback() {
+    return [state = state_]() {
+      ReleasableMutexLock lock(&state->mutex);
+      state->done = true;
+      auto waker = std::move(state->waker);
+      lock.Release();
+      waker.Wakeup();
+    };
+  }
+
+ private:
+  struct State {
+    Mutex mutex;
+    bool done ABSL_GUARDED_BY(mutex) = false;
+    Waker waker ABSL_GUARDED_BY(mutex);
+  };
+  const std::shared_ptr<State> state_{std::make_shared<State>()};
+};
+
+}  // namespace grpc_core
+
+#endif  // GRPC_SRC_CORE_LIB_PROMISE_WAIT_FOR_CALLBACK_H

test/core/promise/BUILD

@@ -462,6 +462,25 @@ grpc_cc_test(
     ],
 )
 
+grpc_cc_test(
+    name = "wait_for_callback_test",
+    srcs = ["wait_for_callback_test.cc"],
+    external_deps = [
+        "absl/status",
+        "gtest",
+    ],
+    language = "c++",
+    tags = ["promise_test"],
+    uses_event_engine = False,
+    uses_polling = False,
+    deps = [
+        "test_wakeup_schedulers",
+        "//src/core:map",
+        "//src/core:notification",
+        "//src/core:wait_for_callback",
+    ],
+)
+
 grpc_cc_test(
     name = "event_engine_wakeup_scheduler_test",
     srcs = ["event_engine_wakeup_scheduler_test.cc"],

test/core/promise/wait_for_callback_test.cc

@@ -0,0 +1,50 @@
+// 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 "src/core/lib/promise/wait_for_callback.h"
+
+#include "absl/status/status.h"
+#include "gtest/gtest.h"
+
+#include "src/core/lib/gprpp/notification.h"
+#include "src/core/lib/promise/map.h"
+#include "test/core/promise/test_wakeup_schedulers.h"
+
+namespace grpc_core {
+
+TEST(WaitForCallbackTest, Works) {
+  WaitForCallback w4cb;
+  auto callback = w4cb.MakeCallback();
+  Notification done;
+  auto activity = MakeActivity(
+      [&w4cb]() {
+        return Map(w4cb.MakeWaitPromise(),
+                   [](Empty) { return absl::OkStatus(); });
+      },
+      InlineWakeupScheduler{},
+      [&done](const absl::Status& s) {
+        EXPECT_TRUE(s.ok());
+        done.Notify();
+      });
+  EXPECT_FALSE(done.HasBeenNotified());
+  callback();
+  done.WaitForNotification();
+}
+
+}  // namespace grpc_core
+
+int main(int argc, char** argv) {
+  ::testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}

tools/run_tests/generated/tests.json

@@ -10777,6 +10777,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": "wait_for_callback_test",
+    "platforms": [
+      "linux",
+      "mac",
+      "posix",
+      "windows"
+    ],
+    "uses_polling": false
+  },
   {
     "args": [],
     "benchmark": false,