1 //
2 //
3 // Copyright 2016 gRPC authors.
4 //
5 // Licensed under the Apache License, Version 2.0 (the "License");
6 // you may not use this file except in compliance with the License.
7 // You may obtain a copy of the License at
8 //
9 // http://www.apache.org/licenses/LICENSE-2.0
10 //
11 // Unless required by applicable law or agreed to in writing, software
12 // distributed under the License is distributed on an "AS IS" BASIS,
13 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 // See the License for the specific language governing permissions and
15 // limitations under the License.
16 //
17 //
18
19 #include <grpc/support/port_platform.h>
20
21 #include "src/core/lib/transport/handshaker.h"
22
23 #include <inttypes.h>
24
25 #include <initializer_list>
26 #include <string>
27 #include <utility>
28
29 #include "absl/status/status.h"
30 #include "absl/strings/str_format.h"
31
32 #include <grpc/byte_buffer.h>
33 #include <grpc/event_engine/event_engine.h>
34 #include <grpc/grpc.h>
35 #include <grpc/slice_buffer.h>
36 #include <grpc/support/alloc.h>
37 #include <grpc/support/log.h>
38
39 #include "src/core/lib/channel/channel_args.h"
40 #include "src/core/lib/debug/trace.h"
41 #include "src/core/lib/gprpp/debug_location.h"
42 #include "src/core/lib/gprpp/status_helper.h"
43 #include "src/core/lib/iomgr/event_engine_shims/endpoint.h"
44 #include "src/core/lib/iomgr/exec_ctx.h"
45
46 namespace grpc_core {
47
48 TraceFlag grpc_handshaker_trace(false, "handshaker");
49
50 namespace {
51
52 using ::grpc_event_engine::experimental::EventEngine;
53
HandshakerArgsString(HandshakerArgs * args)54 std::string HandshakerArgsString(HandshakerArgs* args) {
55 size_t read_buffer_length =
56 args->read_buffer != nullptr ? args->read_buffer->length : 0;
57 return absl::StrFormat(
58 "{endpoint=%p, args=%s, read_buffer=%p (length=%" PRIuPTR
59 "), exit_early=%d}",
60 args->endpoint, args->args.ToString(), args->read_buffer,
61 read_buffer_length, args->exit_early);
62 }
63
64 } // namespace
65
HandshakeManager()66 HandshakeManager::HandshakeManager()
67 : RefCounted(GRPC_TRACE_FLAG_ENABLED(grpc_handshaker_trace)
68 ? "HandshakeManager"
69 : nullptr) {}
70
Add(RefCountedPtr<Handshaker> handshaker)71 void HandshakeManager::Add(RefCountedPtr<Handshaker> handshaker) {
72 MutexLock lock(&mu_);
73 if (GRPC_TRACE_FLAG_ENABLED(grpc_handshaker_trace)) {
74 gpr_log(
75 GPR_INFO,
76 "handshake_manager %p: adding handshaker %s [%p] at index %" PRIuPTR,
77 this, handshaker->name(), handshaker.get(), handshakers_.size());
78 }
79 handshakers_.push_back(std::move(handshaker));
80 }
81
~HandshakeManager()82 HandshakeManager::~HandshakeManager() { handshakers_.clear(); }
83
Shutdown(grpc_error_handle why)84 void HandshakeManager::Shutdown(grpc_error_handle why) {
85 {
86 MutexLock lock(&mu_);
87 // Shutdown the handshaker that's currently in progress, if any.
88 if (!is_shutdown_ && index_ > 0) {
89 is_shutdown_ = true;
90 handshakers_[index_ - 1]->Shutdown(why);
91 }
92 }
93 }
94
95 // Helper function to call either the next handshaker or the
96 // on_handshake_done callback.
97 // Returns true if we've scheduled the on_handshake_done callback.
CallNextHandshakerLocked(grpc_error_handle error)98 bool HandshakeManager::CallNextHandshakerLocked(grpc_error_handle error) {
99 if (GRPC_TRACE_FLAG_ENABLED(grpc_handshaker_trace)) {
100 gpr_log(GPR_INFO,
101 "handshake_manager %p: error=%s shutdown=%d index=%" PRIuPTR
102 ", args=%s",
103 this, StatusToString(error).c_str(), is_shutdown_, index_,
104 HandshakerArgsString(&args_).c_str());
105 }
106 GPR_ASSERT(index_ <= handshakers_.size());
107 // If we got an error or we've been shut down or we're exiting early or
108 // we've finished the last handshaker, invoke the on_handshake_done
109 // callback. Otherwise, call the next handshaker.
110 if (!error.ok() || is_shutdown_ || args_.exit_early ||
111 index_ == handshakers_.size()) {
112 if (error.ok() && is_shutdown_) {
113 error = GRPC_ERROR_CREATE("handshaker shutdown");
114 // It is possible that the endpoint has already been destroyed by
115 // a shutdown call while this callback was sitting on the ExecCtx
116 // with no error.
117 if (args_.endpoint != nullptr) {
118 // TODO(roth): It is currently necessary to shutdown endpoints
119 // before destroying then, even when we know that there are no
120 // pending read/write callbacks. This should be fixed, at which
121 // point this can be removed.
122 grpc_endpoint_shutdown(args_.endpoint, error);
123 grpc_endpoint_destroy(args_.endpoint);
124 args_.endpoint = nullptr;
125 args_.args = ChannelArgs();
126 grpc_slice_buffer_destroy(args_.read_buffer);
127 gpr_free(args_.read_buffer);
128 args_.read_buffer = nullptr;
129 }
130 }
131 if (GRPC_TRACE_FLAG_ENABLED(grpc_handshaker_trace)) {
132 gpr_log(GPR_INFO,
133 "handshake_manager %p: handshaking complete -- scheduling "
134 "on_handshake_done with error=%s",
135 this, StatusToString(error).c_str());
136 }
137 // Cancel deadline timer, since we're invoking the on_handshake_done
138 // callback now.
139 event_engine_->Cancel(deadline_timer_handle_);
140 ExecCtx::Run(DEBUG_LOCATION, &on_handshake_done_, error);
141 is_shutdown_ = true;
142 } else {
143 auto handshaker = handshakers_[index_];
144 if (GRPC_TRACE_FLAG_ENABLED(grpc_handshaker_trace)) {
145 gpr_log(
146 GPR_INFO,
147 "handshake_manager %p: calling handshaker %s [%p] at index %" PRIuPTR,
148 this, handshaker->name(), handshaker.get(), index_);
149 }
150 handshaker->DoHandshake(acceptor_, &call_next_handshaker_, &args_);
151 }
152 ++index_;
153 return is_shutdown_;
154 }
155
CallNextHandshakerFn(void * arg,grpc_error_handle error)156 void HandshakeManager::CallNextHandshakerFn(void* arg,
157 grpc_error_handle error) {
158 auto* mgr = static_cast<HandshakeManager*>(arg);
159 bool done;
160 {
161 MutexLock lock(&mgr->mu_);
162 done = mgr->CallNextHandshakerLocked(error);
163 }
164 // If we're invoked the final callback, we won't be coming back
165 // to this function, so we can release our reference to the
166 // handshake manager.
167 if (done) {
168 mgr->Unref();
169 }
170 }
171
DoHandshake(grpc_endpoint * endpoint,const ChannelArgs & channel_args,Timestamp deadline,grpc_tcp_server_acceptor * acceptor,grpc_iomgr_cb_func on_handshake_done,void * user_data)172 void HandshakeManager::DoHandshake(grpc_endpoint* endpoint,
173 const ChannelArgs& channel_args,
174 Timestamp deadline,
175 grpc_tcp_server_acceptor* acceptor,
176 grpc_iomgr_cb_func on_handshake_done,
177 void* user_data) {
178 bool done;
179 {
180 MutexLock lock(&mu_);
181 GPR_ASSERT(index_ == 0);
182 // Construct handshaker args. These will be passed through all
183 // handshakers and eventually be freed by the on_handshake_done callback.
184 args_.endpoint = endpoint;
185 args_.deadline = deadline;
186 args_.args = channel_args;
187 args_.user_data = user_data;
188 args_.read_buffer =
189 static_cast<grpc_slice_buffer*>(gpr_malloc(sizeof(*args_.read_buffer)));
190 grpc_slice_buffer_init(args_.read_buffer);
191 if (acceptor != nullptr && acceptor->external_connection &&
192 acceptor->pending_data != nullptr) {
193 grpc_slice_buffer_swap(args_.read_buffer,
194 &(acceptor->pending_data->data.raw.slice_buffer));
195 // TODO(vigneshbabu): For connections accepted through event engine
196 // listeners, the ownership of the byte buffer received is transferred to
197 // this callback and it is thus this callback's duty to delete it.
198 // Make this hack default once event engine is rolled out.
199 if (grpc_event_engine::experimental::grpc_is_event_engine_endpoint(
200 endpoint)) {
201 grpc_byte_buffer_destroy(acceptor->pending_data);
202 }
203 }
204 // Initialize state needed for calling handshakers.
205 acceptor_ = acceptor;
206 GRPC_CLOSURE_INIT(&call_next_handshaker_,
207 &HandshakeManager::CallNextHandshakerFn, this,
208 grpc_schedule_on_exec_ctx);
209 GRPC_CLOSURE_INIT(&on_handshake_done_, on_handshake_done, &args_,
210 grpc_schedule_on_exec_ctx);
211 // Start deadline timer, which owns a ref.
212 const Duration time_to_deadline = deadline - Timestamp::Now();
213 event_engine_ = args_.args.GetObjectRef<EventEngine>();
214 deadline_timer_handle_ =
215 event_engine_->RunAfter(time_to_deadline, [self = Ref()]() mutable {
216 ApplicationCallbackExecCtx callback_exec_ctx;
217 ExecCtx exec_ctx;
218 self->Shutdown(GRPC_ERROR_CREATE("Handshake timed out"));
219 // HandshakeManager deletion might require an active ExecCtx.
220 self.reset();
221 });
222 // Start first handshaker, which also owns a ref.
223 Ref().release();
224 done = CallNextHandshakerLocked(absl::OkStatus());
225 }
226 if (done) {
227 Unref();
228 }
229 }
230
231 } // namespace grpc_core
232
grpc_handshake_manager_add(grpc_handshake_manager * mgr,grpc_handshaker * handshaker)233 void grpc_handshake_manager_add(grpc_handshake_manager* mgr,
234 grpc_handshaker* handshaker) {
235 // This is a transition method to aid the API change for handshakers.
236 grpc_core::RefCountedPtr<grpc_core::Handshaker> refd_hs(
237 static_cast<grpc_core::Handshaker*>(handshaker));
238 mgr->Add(refd_hs);
239 }
240