1 // Copyright 2024 gRPC authors.
2 //
3 // Licensed under the Apache License, Version 2.0 (the "License");
4 // you may not use this file except in compliance with the License.
5 // You may obtain a copy of the License at
6 //
7 // http://www.apache.org/licenses/LICENSE-2.0
8 //
9 // Unless required by applicable law or agreed to in writing, software
10 // distributed under the License is distributed on an "AS IS" BASIS,
11 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 // See the License for the specific language governing permissions and
13 // limitations under the License.
14
15 #include <grpc/support/port_platform.h>
16
17 #include "src/core/lib/surface/wait_for_cq_end_op.h"
18
19 #include <atomic>
20
21 #include "src/core/lib/gprpp/match.h"
22 #include "src/core/lib/promise/trace.h"
23
24 namespace grpc_core {
operator ()()25 Poll<Empty> WaitForCqEndOp::operator()() {
26 if (grpc_trace_promise_primitives.enabled()) {
27 gpr_log(GPR_INFO, "%sWaitForCqEndOp[%p] %s",
28 Activity::current()->DebugTag().c_str(), this,
29 StateString(state_).c_str());
30 }
31 if (auto* n = absl::get_if<NotStarted>(&state_)) {
32 if (n->is_closure) {
33 ExecCtx::Run(DEBUG_LOCATION, static_cast<grpc_closure*>(n->tag),
34 std::move(n->error));
35 return Empty{};
36 } else {
37 auto not_started = std::move(*n);
38 auto& started =
39 state_.emplace<Started>(GetContext<Activity>()->MakeOwningWaker());
40 grpc_cq_end_op(
41 not_started.cq, not_started.tag, std::move(not_started.error),
42 [](void* p, grpc_cq_completion*) {
43 auto started = static_cast<Started*>(p);
44 auto wakeup = std::move(started->waker);
45 started->done.store(true, std::memory_order_release);
46 wakeup.Wakeup();
47 },
48 &started, &started.completion);
49 }
50 }
51 auto& started = absl::get<Started>(state_);
52 if (started.done.load(std::memory_order_acquire)) {
53 return Empty{};
54 } else {
55 return Pending{};
56 }
57 }
58
StateString(const State & state)59 std::string WaitForCqEndOp::StateString(const State& state) {
60 return Match(
61 state,
62 [](const NotStarted& x) {
63 return absl::StrFormat(
64 "NotStarted{is_closure=%s, tag=%p, error=%s, cq=%p}",
65 x.is_closure ? "true" : "false", x.tag, x.error.ToString(), x.cq);
66 },
67 [](const Started& x) {
68 return absl::StrFormat(
69 "Started{completion=%p, done=%s}", &x.completion,
70 x.done.load(std::memory_order_relaxed) ? "true" : "false");
71 },
72 [](const Invalid&) -> std::string { return "Invalid{}"; });
73 }
74
75 } // namespace grpc_core