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 <string.h>
22
23 #include "absl/strings/str_format.h"
24
25 #include <grpc/impl/codegen/slice.h>
26 #include <grpc/support/alloc.h>
27 #include <grpc/support/log.h>
28 #include <grpc/support/string_util.h>
29
30 #include "src/core/lib/channel/channel_args.h"
31 #include "src/core/lib/channel/handshaker.h"
32 #include "src/core/lib/debug/trace.h"
33 #include "src/core/lib/iomgr/timer.h"
34 #include "src/core/lib/slice/slice_internal.h"
35
36 namespace grpc_core {
37
38 TraceFlag grpc_handshaker_trace(false, "handshaker");
39
40 namespace {
41
HandshakerArgsString(HandshakerArgs * args)42 std::string HandshakerArgsString(HandshakerArgs* args) {
43 size_t num_args = args->args != nullptr ? args->args->num_args : 0;
44 size_t read_buffer_length =
45 args->read_buffer != nullptr ? args->read_buffer->length : 0;
46 return absl::StrFormat(
47 "{endpoint=%p, args=%p {size=%" PRIuPTR
48 ": %s}, read_buffer=%p (length=%" PRIuPTR "), exit_early=%d}",
49 args->endpoint, args->args, num_args,
50 grpc_channel_args_string(args->args), args->read_buffer,
51 read_buffer_length, args->exit_early);
52 }
53
54 } // namespace
55
HandshakeManager()56 HandshakeManager::HandshakeManager() { gpr_mu_init(&mu_); }
57
58 /// Add \a mgr to the server side list of all pending handshake managers, the
59 /// list starts with \a *head.
60 // Not thread-safe. Caller needs to synchronize.
AddToPendingMgrList(HandshakeManager ** head)61 void HandshakeManager::AddToPendingMgrList(HandshakeManager** head) {
62 GPR_ASSERT(prev_ == nullptr);
63 GPR_ASSERT(next_ == nullptr);
64 next_ = *head;
65 if (*head) {
66 (*head)->prev_ = this;
67 }
68 *head = this;
69 }
70
71 /// Remove \a mgr from the server side list of all pending handshake managers.
72 // Not thread-safe. Caller needs to synchronize.
RemoveFromPendingMgrList(HandshakeManager ** head)73 void HandshakeManager::RemoveFromPendingMgrList(HandshakeManager** head) {
74 if (next_ != nullptr) {
75 next_->prev_ = prev_;
76 }
77 if (prev_ != nullptr) {
78 prev_->next_ = next_;
79 } else {
80 GPR_ASSERT(*head == this);
81 *head = next_;
82 }
83 }
84
85 /// Shutdown all pending handshake managers starting at head on the server
86 /// side. Not thread-safe. Caller needs to synchronize.
ShutdownAllPending(grpc_error * why)87 void HandshakeManager::ShutdownAllPending(grpc_error* why) {
88 auto* head = this;
89 while (head != nullptr) {
90 head->Shutdown(GRPC_ERROR_REF(why));
91 head = head->next_;
92 }
93 GRPC_ERROR_UNREF(why);
94 }
95
Add(RefCountedPtr<Handshaker> handshaker)96 void HandshakeManager::Add(RefCountedPtr<Handshaker> handshaker) {
97 if (GRPC_TRACE_FLAG_ENABLED(grpc_handshaker_trace)) {
98 gpr_log(
99 GPR_INFO,
100 "handshake_manager %p: adding handshaker %s [%p] at index %" PRIuPTR,
101 this, handshaker->name(), handshaker.get(), handshakers_.size());
102 }
103 MutexLock lock(&mu_);
104 handshakers_.push_back(std::move(handshaker));
105 }
106
~HandshakeManager()107 HandshakeManager::~HandshakeManager() {
108 handshakers_.clear();
109 gpr_mu_destroy(&mu_);
110 }
111
Shutdown(grpc_error * why)112 void HandshakeManager::Shutdown(grpc_error* why) {
113 {
114 MutexLock lock(&mu_);
115 // Shutdown the handshaker that's currently in progress, if any.
116 if (!is_shutdown_ && index_ > 0) {
117 is_shutdown_ = true;
118 handshakers_[index_ - 1]->Shutdown(GRPC_ERROR_REF(why));
119 }
120 }
121 GRPC_ERROR_UNREF(why);
122 }
123
124 // Helper function to call either the next handshaker or the
125 // on_handshake_done callback.
126 // Returns true if we've scheduled the on_handshake_done callback.
CallNextHandshakerLocked(grpc_error * error)127 bool HandshakeManager::CallNextHandshakerLocked(grpc_error* error) {
128 if (GRPC_TRACE_FLAG_ENABLED(grpc_handshaker_trace)) {
129 gpr_log(GPR_INFO,
130 "handshake_manager %p: error=%s shutdown=%d index=%" PRIuPTR
131 ", args=%s",
132 this, grpc_error_string(error), is_shutdown_, index_,
133 HandshakerArgsString(&args_).c_str());
134 }
135 GPR_ASSERT(index_ <= handshakers_.size());
136 // If we got an error or we've been shut down or we're exiting early or
137 // we've finished the last handshaker, invoke the on_handshake_done
138 // callback. Otherwise, call the next handshaker.
139 if (error != GRPC_ERROR_NONE || is_shutdown_ || args_.exit_early ||
140 index_ == handshakers_.size()) {
141 if (error == GRPC_ERROR_NONE && is_shutdown_) {
142 error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("handshaker shutdown");
143 // It is possible that the endpoint has already been destroyed by
144 // a shutdown call while this callback was sitting on the ExecCtx
145 // with no error.
146 if (args_.endpoint != nullptr) {
147 // TODO(roth): It is currently necessary to shutdown endpoints
148 // before destroying then, even when we know that there are no
149 // pending read/write callbacks. This should be fixed, at which
150 // point this can be removed.
151 grpc_endpoint_shutdown(args_.endpoint, GRPC_ERROR_REF(error));
152 grpc_endpoint_destroy(args_.endpoint);
153 args_.endpoint = nullptr;
154 grpc_channel_args_destroy(args_.args);
155 args_.args = nullptr;
156 grpc_slice_buffer_destroy_internal(args_.read_buffer);
157 gpr_free(args_.read_buffer);
158 args_.read_buffer = nullptr;
159 }
160 }
161 if (GRPC_TRACE_FLAG_ENABLED(grpc_handshaker_trace)) {
162 gpr_log(GPR_INFO,
163 "handshake_manager %p: handshaking complete -- scheduling "
164 "on_handshake_done with error=%s",
165 this, grpc_error_string(error));
166 }
167 // Cancel deadline timer, since we're invoking the on_handshake_done
168 // callback now.
169 grpc_timer_cancel(&deadline_timer_);
170 ExecCtx::Run(DEBUG_LOCATION, &on_handshake_done_, error);
171 is_shutdown_ = true;
172 } else {
173 auto handshaker = handshakers_[index_];
174 if (GRPC_TRACE_FLAG_ENABLED(grpc_handshaker_trace)) {
175 gpr_log(
176 GPR_INFO,
177 "handshake_manager %p: calling handshaker %s [%p] at index %" PRIuPTR,
178 this, handshaker->name(), handshaker.get(), index_);
179 }
180 handshaker->DoHandshake(acceptor_, &call_next_handshaker_, &args_);
181 }
182 ++index_;
183 return is_shutdown_;
184 }
185
CallNextHandshakerFn(void * arg,grpc_error * error)186 void HandshakeManager::CallNextHandshakerFn(void* arg, grpc_error* error) {
187 auto* mgr = static_cast<HandshakeManager*>(arg);
188 bool done;
189 {
190 MutexLock lock(&mgr->mu_);
191 done = mgr->CallNextHandshakerLocked(GRPC_ERROR_REF(error));
192 }
193 // If we're invoked the final callback, we won't be coming back
194 // to this function, so we can release our reference to the
195 // handshake manager.
196 if (done) {
197 mgr->Unref();
198 }
199 }
200
OnTimeoutFn(void * arg,grpc_error * error)201 void HandshakeManager::OnTimeoutFn(void* arg, grpc_error* error) {
202 auto* mgr = static_cast<HandshakeManager*>(arg);
203 if (error == GRPC_ERROR_NONE) { // Timer fired, rather than being cancelled
204 mgr->Shutdown(GRPC_ERROR_CREATE_FROM_STATIC_STRING("Handshake timed out"));
205 }
206 mgr->Unref();
207 }
208
DoHandshake(grpc_endpoint * endpoint,const grpc_channel_args * channel_args,grpc_millis deadline,grpc_tcp_server_acceptor * acceptor,grpc_iomgr_cb_func on_handshake_done,void * user_data)209 void HandshakeManager::DoHandshake(grpc_endpoint* endpoint,
210 const grpc_channel_args* channel_args,
211 grpc_millis deadline,
212 grpc_tcp_server_acceptor* acceptor,
213 grpc_iomgr_cb_func on_handshake_done,
214 void* user_data) {
215 bool done;
216 {
217 MutexLock lock(&mu_);
218 GPR_ASSERT(index_ == 0);
219 GPR_ASSERT(!is_shutdown_);
220 // Construct handshaker args. These will be passed through all
221 // handshakers and eventually be freed by the on_handshake_done callback.
222 args_.endpoint = endpoint;
223 args_.args = grpc_channel_args_copy(channel_args);
224 args_.user_data = user_data;
225 args_.read_buffer =
226 static_cast<grpc_slice_buffer*>(gpr_malloc(sizeof(*args_.read_buffer)));
227 grpc_slice_buffer_init(args_.read_buffer);
228 if (acceptor != nullptr && acceptor->external_connection &&
229 acceptor->pending_data != nullptr) {
230 grpc_slice_buffer_swap(args_.read_buffer,
231 &(acceptor->pending_data->data.raw.slice_buffer));
232 }
233 // Initialize state needed for calling handshakers.
234 acceptor_ = acceptor;
235 GRPC_CLOSURE_INIT(&call_next_handshaker_,
236 &HandshakeManager::CallNextHandshakerFn, this,
237 grpc_schedule_on_exec_ctx);
238 GRPC_CLOSURE_INIT(&on_handshake_done_, on_handshake_done, &args_,
239 grpc_schedule_on_exec_ctx);
240 // Start deadline timer, which owns a ref.
241 Ref().release();
242 GRPC_CLOSURE_INIT(&on_timeout_, &HandshakeManager::OnTimeoutFn, this,
243 grpc_schedule_on_exec_ctx);
244 grpc_timer_init(&deadline_timer_, deadline, &on_timeout_);
245 // Start first handshaker, which also owns a ref.
246 Ref().release();
247 done = CallNextHandshakerLocked(GRPC_ERROR_NONE);
248 }
249 if (done) {
250 Unref();
251 }
252 }
253
254 } // namespace grpc_core
255
grpc_handshake_manager_add(grpc_handshake_manager * mgr,grpc_handshaker * handshaker)256 void grpc_handshake_manager_add(grpc_handshake_manager* mgr,
257 grpc_handshaker* handshaker) {
258 // This is a transition method to aid the API change for handshakers.
259 grpc_core::RefCountedPtr<grpc_core::Handshaker> refd_hs(
260 static_cast<grpc_core::Handshaker*>(handshaker));
261 mgr->Add(refd_hs);
262 }
263