1 // Copyright 2021 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/promise/activity.h"
18
19 #include <stddef.h>
20
21 #include <initializer_list>
22 #include <vector>
23
24 #include "absl/strings/str_cat.h"
25 #include "absl/strings/str_format.h"
26 #include "absl/strings/str_join.h"
27
28 #include "src/core/lib/gprpp/atomic_utils.h"
29 #include "src/core/lib/gprpp/crash.h"
30
31 namespace grpc_core {
32
33 ///////////////////////////////////////////////////////////////////////////////
34 // GLOBALS
35
36 thread_local Activity* Activity::g_current_activity_{nullptr};
37
38 namespace promise_detail {
39
40 ///////////////////////////////////////////////////////////////////////////////
41 // HELPER TYPES
42
ActivityDebugTag(WakeupMask) const43 std::string Unwakeable::ActivityDebugTag(WakeupMask) const {
44 return "<unknown>";
45 }
46
47 // Weak handle to an Activity.
48 // Handle can persist while Activity goes away.
49 class FreestandingActivity::Handle final : public Wakeable {
50 public:
Handle(FreestandingActivity * activity)51 explicit Handle(FreestandingActivity* activity) : activity_(activity) {}
52
53 // Ref the Handle (not the activity).
Ref()54 void Ref() { refs_.fetch_add(1, std::memory_order_relaxed); }
55
56 // Activity is going away... drop its reference and sever the connection back.
DropActivity()57 void DropActivity() ABSL_LOCKS_EXCLUDED(mu_) {
58 mu_.Lock();
59 GPR_ASSERT(activity_ != nullptr);
60 activity_ = nullptr;
61 mu_.Unlock();
62 Unref();
63 }
64
65 // Activity needs to wake up (if it still exists!) - wake it up, and drop the
66 // ref that was kept for this handle.
Wakeup(WakeupMask)67 void Wakeup(WakeupMask) override ABSL_LOCKS_EXCLUDED(mu_) {
68 mu_.Lock();
69 // Note that activity refcount can drop to zero, but we could win the lock
70 // against DropActivity, so we need to only increase activities refcount if
71 // it is non-zero.
72 if (activity_ && activity_->RefIfNonzero()) {
73 FreestandingActivity* activity = activity_;
74 mu_.Unlock();
75 // Activity still exists and we have a reference: wake it up, which will
76 // drop the ref.
77 activity->Wakeup(0);
78 } else {
79 // Could not get the activity - it's either gone or going. No need to wake
80 // it up!
81 mu_.Unlock();
82 }
83 // Drop the ref to the handle (we have one ref = one wakeup semantics).
84 Unref();
85 }
86
WakeupAsync(WakeupMask)87 void WakeupAsync(WakeupMask) override ABSL_LOCKS_EXCLUDED(mu_) {
88 Crash("not implemented");
89 }
90
Drop(WakeupMask)91 void Drop(WakeupMask) override { Unref(); }
92
ActivityDebugTag(WakeupMask) const93 std::string ActivityDebugTag(WakeupMask) const override {
94 MutexLock lock(&mu_);
95 return activity_ == nullptr ? "<unknown>" : activity_->DebugTag();
96 }
97
98 private:
99 // Unref the Handle (not the activity).
Unref()100 void Unref() {
101 if (1 == refs_.fetch_sub(1, std::memory_order_acq_rel)) {
102 delete this;
103 }
104 }
105
106 // Two initial refs: one for the waiter that caused instantiation, one for the
107 // activity.
108 std::atomic<size_t> refs_{2};
109 mutable Mutex mu_ ABSL_ACQUIRED_AFTER(activity_->mu_);
110 FreestandingActivity* activity_ ABSL_GUARDED_BY(mu_);
111 };
112
113 ///////////////////////////////////////////////////////////////////////////////
114 // ACTIVITY IMPLEMENTATION
115
RefIfNonzero()116 bool FreestandingActivity::RefIfNonzero() { return IncrementIfNonzero(&refs_); }
117
RefHandle()118 FreestandingActivity::Handle* FreestandingActivity::RefHandle() {
119 if (handle_ == nullptr) {
120 // No handle created yet - construct it and return it.
121 handle_ = new Handle(this);
122 return handle_;
123 } else {
124 // Already had to create a handle, ref & return it.
125 handle_->Ref();
126 return handle_;
127 }
128 }
129
DropHandle()130 void FreestandingActivity::DropHandle() {
131 handle_->DropActivity();
132 handle_ = nullptr;
133 }
134
MakeNonOwningWaker()135 Waker FreestandingActivity::MakeNonOwningWaker() {
136 mu_.AssertHeld();
137 return Waker(RefHandle(), 0);
138 }
139
140 } // namespace promise_detail
141
DebugTag() const142 std::string Activity::DebugTag() const {
143 return absl::StrFormat("ACTIVITY[%p]", this);
144 }
145
146 ///////////////////////////////////////////////////////////////////////////////
147 // INTRA ACTIVITY WAKER IMPLEMENTATION
148
DebugString() const149 std::string IntraActivityWaiter::DebugString() const {
150 std::vector<int> bits;
151 for (size_t i = 0; i < 8 * sizeof(WakeupMask); i++) {
152 if (wakeups_ & (1 << i)) bits.push_back(i);
153 }
154 return absl::StrCat("{", absl::StrJoin(bits, ","), "}");
155 }
156
157 } // namespace grpc_core
158