3 * Copyright 2015 gRPC authors.
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
9 * http://www.apache.org/licenses/LICENSE-2.0
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.
19 #include <grpc/support/port_platform.h>
21 #include "src/core/ext/transport/chttp2/server/chttp2_server.h"
28 #include "absl/strings/str_cat.h"
29 #include "absl/strings/str_format.h"
31 #include <grpc/grpc.h>
32 #include <grpc/impl/codegen/grpc_types.h>
33 #include <grpc/support/alloc.h>
34 #include <grpc/support/log.h>
35 #include <grpc/support/sync.h>
37 #include "src/core/ext/filters/http/server/http_server_filter.h"
38 #include "src/core/ext/transport/chttp2/transport/chttp2_transport.h"
39 #include "src/core/ext/transport/chttp2/transport/internal.h"
40 #include "src/core/lib/channel/channel_args.h"
41 #include "src/core/lib/channel/handshaker.h"
42 #include "src/core/lib/channel/handshaker_registry.h"
43 #include "src/core/lib/gprpp/ref_counted.h"
44 #include "src/core/lib/gprpp/ref_counted_ptr.h"
45 #include "src/core/lib/iomgr/endpoint.h"
46 #include "src/core/lib/iomgr/resolve_address.h"
47 #include "src/core/lib/iomgr/resource_quota.h"
48 #include "src/core/lib/iomgr/tcp_server.h"
49 #include "src/core/lib/slice/slice_internal.h"
50 #include "src/core/lib/surface/api_trace.h"
51 #include "src/core/lib/surface/server.h"
56 class Chttp2ServerListener : public Server::ListenerInterface {
58 static grpc_error* Create(Server* server, const char* addr,
59 grpc_channel_args* args, int* port_num);
61 static grpc_error* CreateWithAcceptor(Server* server, const char* name,
62 grpc_channel_args* args);
64 // Do not instantiate directly. Use one of the factory methods above.
65 Chttp2ServerListener(Server* server, grpc_channel_args* args);
66 ~Chttp2ServerListener();
68 void Start(Server* server,
69 const std::vector<grpc_pollset*>* pollsets) override;
71 channelz::ListenSocketNode* channelz_listen_socket_node() const override {
72 return channelz_listen_socket_.get();
75 void SetOnDestroyDone(grpc_closure* on_destroy_done) override;
77 void Orphan() override;
80 class ConnectionState : public RefCounted<ConnectionState> {
82 ConnectionState(Chttp2ServerListener* listener,
83 grpc_pollset* accepting_pollset,
84 grpc_tcp_server_acceptor* acceptor,
85 RefCountedPtr<HandshakeManager> handshake_mgr,
86 grpc_channel_args* args, grpc_endpoint* endpoint);
91 static void OnTimeout(void* arg, grpc_error* error);
92 static void OnReceiveSettings(void* arg, grpc_error* error);
93 static void OnHandshakeDone(void* arg, grpc_error* error);
95 Chttp2ServerListener* const listener_;
96 grpc_pollset* const accepting_pollset_;
97 grpc_tcp_server_acceptor* const acceptor_;
98 RefCountedPtr<HandshakeManager> handshake_mgr_;
99 // State for enforcing handshake timeout on receiving HTTP/2 settings.
100 grpc_chttp2_transport* transport_ = nullptr;
101 grpc_millis deadline_;
103 grpc_closure on_timeout_;
104 grpc_closure on_receive_settings_;
105 grpc_pollset_set* const interested_parties_;
108 static void OnAccept(void* arg, grpc_endpoint* tcp,
109 grpc_pollset* accepting_pollset,
110 grpc_tcp_server_acceptor* acceptor);
112 RefCountedPtr<HandshakeManager> CreateHandshakeManager();
114 static void TcpServerShutdownComplete(void* arg, grpc_error* error);
116 static void DestroyListener(Server* /*server*/, void* arg,
117 grpc_closure* destroy_done);
119 Server* const server_;
120 grpc_channel_args* const args_;
121 grpc_tcp_server* tcp_server_;
123 bool shutdown_ = true;
124 grpc_closure tcp_server_shutdown_complete_;
125 grpc_closure* on_destroy_done_ = nullptr;
126 HandshakeManager* pending_handshake_mgrs_ = nullptr;
127 RefCountedPtr<channelz::ListenSocketNode> channelz_listen_socket_;
131 // Chttp2ServerListener::ConnectionState
134 grpc_millis GetConnectionDeadline(const grpc_channel_args* args) {
136 grpc_channel_args_find_integer(args, GRPC_ARG_SERVER_HANDSHAKE_TIMEOUT_MS,
137 {120 * GPR_MS_PER_SEC, 1, INT_MAX});
138 return ExecCtx::Get()->Now() + timeout_ms;
141 Chttp2ServerListener::ConnectionState::ConnectionState(
142 Chttp2ServerListener* listener, grpc_pollset* accepting_pollset,
143 grpc_tcp_server_acceptor* acceptor,
144 RefCountedPtr<HandshakeManager> handshake_mgr, grpc_channel_args* args,
145 grpc_endpoint* endpoint)
146 : listener_(listener),
147 accepting_pollset_(accepting_pollset),
149 handshake_mgr_(std::move(handshake_mgr)),
150 deadline_(GetConnectionDeadline(args)),
151 interested_parties_(grpc_pollset_set_create()) {
152 grpc_pollset_set_add_pollset(interested_parties_, accepting_pollset_);
153 HandshakerRegistry::AddHandshakers(HANDSHAKER_SERVER, args,
154 interested_parties_, handshake_mgr_.get());
155 handshake_mgr_->DoHandshake(endpoint, args, deadline_, acceptor_,
156 OnHandshakeDone, this);
159 Chttp2ServerListener::ConnectionState::~ConnectionState() {
160 if (transport_ != nullptr) {
161 GRPC_CHTTP2_UNREF_TRANSPORT(transport_, "receive settings timeout");
163 grpc_pollset_set_del_pollset(interested_parties_, accepting_pollset_);
164 grpc_pollset_set_destroy(interested_parties_);
167 void Chttp2ServerListener::ConnectionState::OnTimeout(void* arg,
169 ConnectionState* self = static_cast<ConnectionState*>(arg);
170 // Note that we may be called with GRPC_ERROR_NONE when the timer fires
171 // or with an error indicating that the timer system is being shut down.
172 if (error != GRPC_ERROR_CANCELLED) {
173 grpc_transport_op* op = grpc_make_transport_op(nullptr);
174 op->disconnect_with_error = GRPC_ERROR_CREATE_FROM_STATIC_STRING(
175 "Did not receive HTTP/2 settings before handshake timeout");
176 grpc_transport_perform_op(&self->transport_->base, op);
181 void Chttp2ServerListener::ConnectionState::OnReceiveSettings(
182 void* arg, grpc_error* error) {
183 ConnectionState* self = static_cast<ConnectionState*>(arg);
184 if (error == GRPC_ERROR_NONE) {
185 grpc_timer_cancel(&self->timer_);
190 void Chttp2ServerListener::ConnectionState::OnHandshakeDone(void* arg,
192 auto* args = static_cast<HandshakerArgs*>(arg);
193 ConnectionState* self = static_cast<ConnectionState*>(args->user_data);
195 MutexLock lock(&self->listener_->mu_);
196 grpc_resource_user* resource_user =
197 self->listener_->server_->default_resource_user();
198 if (error != GRPC_ERROR_NONE || self->listener_->shutdown_) {
199 const char* error_str = grpc_error_string(error);
200 gpr_log(GPR_DEBUG, "Handshaking failed: %s", error_str);
201 if (resource_user != nullptr) {
202 grpc_resource_user_free(resource_user,
203 GRPC_RESOURCE_QUOTA_CHANNEL_SIZE);
205 if (error == GRPC_ERROR_NONE && args->endpoint != nullptr) {
206 // We were shut down after handshaking completed successfully, so
207 // destroy the endpoint here.
208 // TODO(ctiller): It is currently necessary to shutdown endpoints
209 // before destroying them, even if we know that there are no
210 // pending read/write callbacks. This should be fixed, at which
211 // point this can be removed.
212 grpc_endpoint_shutdown(args->endpoint, GRPC_ERROR_NONE);
213 grpc_endpoint_destroy(args->endpoint);
214 grpc_channel_args_destroy(args->args);
215 grpc_slice_buffer_destroy_internal(args->read_buffer);
216 gpr_free(args->read_buffer);
219 // If the handshaking succeeded but there is no endpoint, then the
220 // handshaker may have handed off the connection to some external
221 // code, so we can just clean up here without creating a transport.
222 if (args->endpoint != nullptr) {
223 grpc_transport* transport = grpc_create_chttp2_transport(
224 args->args, args->endpoint, false, resource_user);
225 self->listener_->server_->SetupTransport(
226 transport, self->accepting_pollset_, args->args,
227 grpc_chttp2_transport_get_socket_node(transport), resource_user);
228 // Use notify_on_receive_settings callback to enforce the
229 // handshake deadline.
230 // Note: The reinterpret_cast<>s here are safe, because
231 // grpc_chttp2_transport is a C-style extension of
232 // grpc_transport, so this is morally equivalent of a
233 // static_cast<> to a derived class.
234 // TODO(roth): Change to static_cast<> when we C++-ify the
236 self->transport_ = reinterpret_cast<grpc_chttp2_transport*>(transport);
237 self->Ref().release(); // Held by OnReceiveSettings().
238 GRPC_CLOSURE_INIT(&self->on_receive_settings_, OnReceiveSettings, self,
239 grpc_schedule_on_exec_ctx);
240 grpc_chttp2_transport_start_reading(transport, args->read_buffer,
241 &self->on_receive_settings_);
242 grpc_channel_args_destroy(args->args);
243 self->Ref().release(); // Held by OnTimeout().
244 GRPC_CHTTP2_REF_TRANSPORT(
245 reinterpret_cast<grpc_chttp2_transport*>(transport),
246 "receive settings timeout");
247 GRPC_CLOSURE_INIT(&self->on_timeout_, OnTimeout, self,
248 grpc_schedule_on_exec_ctx);
249 grpc_timer_init(&self->timer_, self->deadline_, &self->on_timeout_);
251 if (resource_user != nullptr) {
252 grpc_resource_user_free(resource_user,
253 GRPC_RESOURCE_QUOTA_CHANNEL_SIZE);
257 self->handshake_mgr_->RemoveFromPendingMgrList(
258 &self->listener_->pending_handshake_mgrs_);
260 self->handshake_mgr_.reset();
261 gpr_free(self->acceptor_);
262 grpc_tcp_server_unref(self->listener_->tcp_server_);
267 // Chttp2ServerListener
270 grpc_error* Chttp2ServerListener::Create(Server* server, const char* addr,
271 grpc_channel_args* args,
273 std::vector<grpc_error*> error_list;
274 grpc_resolved_addresses* resolved = nullptr;
275 Chttp2ServerListener* listener = nullptr;
276 // The bulk of this method is inside of a lambda to make cleanup
277 // easier without using goto.
278 grpc_error* error = [&]() {
280 /* resolve address */
281 grpc_error* error = grpc_blocking_resolve_address(addr, "https", &resolved);
282 if (error != GRPC_ERROR_NONE) return error;
283 // Create Chttp2ServerListener.
284 listener = new Chttp2ServerListener(server, args);
285 error = grpc_tcp_server_create(&listener->tcp_server_shutdown_complete_,
286 args, &listener->tcp_server_);
287 if (error != GRPC_ERROR_NONE) return error;
288 for (size_t i = 0; i < resolved->naddrs; i++) {
290 error = grpc_tcp_server_add_port(listener->tcp_server_,
291 &resolved->addrs[i], &port_temp);
292 if (error != GRPC_ERROR_NONE) {
293 error_list.push_back(error);
295 if (*port_num == -1) {
296 *port_num = port_temp;
298 GPR_ASSERT(*port_num == port_temp);
302 if (error_list.size() == resolved->naddrs) {
304 absl::StrFormat("No address added out of total %" PRIuPTR " resolved",
306 return GRPC_ERROR_CREATE_REFERENCING_FROM_COPIED_STRING(
307 msg.c_str(), error_list.data(), error_list.size());
308 } else if (!error_list.empty()) {
309 std::string msg = absl::StrFormat(
310 "Only %" PRIuPTR " addresses added out of total %" PRIuPTR
312 resolved->naddrs - error_list.size(), resolved->naddrs);
313 error = GRPC_ERROR_CREATE_REFERENCING_FROM_COPIED_STRING(
314 msg.c_str(), error_list.data(), error_list.size());
315 gpr_log(GPR_INFO, "WARNING: %s", grpc_error_string(error));
316 GRPC_ERROR_UNREF(error);
317 /* we managed to bind some addresses: continue */
319 // Create channelz node.
320 if (grpc_channel_args_find_bool(args, GRPC_ARG_ENABLE_CHANNELZ,
321 GRPC_ENABLE_CHANNELZ_DEFAULT)) {
322 listener->channelz_listen_socket_ =
323 MakeRefCounted<channelz::ListenSocketNode>(
324 addr, absl::StrFormat("chttp2 listener %s", addr));
326 /* Register with the server only upon success */
327 server->AddListener(OrphanablePtr<Server::ListenerInterface>(listener));
328 return GRPC_ERROR_NONE;
330 if (resolved != nullptr) {
331 grpc_resolved_addresses_destroy(resolved);
333 if (error != GRPC_ERROR_NONE) {
334 if (listener != nullptr) {
335 if (listener->tcp_server_ != nullptr) {
336 grpc_tcp_server_unref(listener->tcp_server_);
341 grpc_channel_args_destroy(args);
345 for (grpc_error* error : error_list) {
346 GRPC_ERROR_UNREF(error);
351 grpc_error* Chttp2ServerListener::CreateWithAcceptor(Server* server,
353 grpc_channel_args* args) {
354 Chttp2ServerListener* listener = new Chttp2ServerListener(server, args);
355 grpc_error* error = grpc_tcp_server_create(
356 &listener->tcp_server_shutdown_complete_, args, &listener->tcp_server_);
357 if (error != GRPC_ERROR_NONE) {
361 // TODO(yangg) channelz
362 TcpServerFdHandler** arg_val =
363 grpc_channel_args_find_pointer<TcpServerFdHandler*>(args, name);
364 *arg_val = grpc_tcp_server_create_fd_handler(listener->tcp_server_);
365 server->AddListener(OrphanablePtr<Server::ListenerInterface>(listener));
366 return GRPC_ERROR_NONE;
369 Chttp2ServerListener::Chttp2ServerListener(Server* server,
370 grpc_channel_args* args)
371 : server_(server), args_(args) {
372 GRPC_CLOSURE_INIT(&tcp_server_shutdown_complete_, TcpServerShutdownComplete,
373 this, grpc_schedule_on_exec_ctx);
376 Chttp2ServerListener::~Chttp2ServerListener() {
377 grpc_channel_args_destroy(args_);
380 /* Server callback: start listening on our ports */
381 void Chttp2ServerListener::Start(Server* /*server*/,
382 const std::vector<grpc_pollset*>* pollsets) {
384 MutexLock lock(&mu_);
387 grpc_tcp_server_start(tcp_server_, pollsets, OnAccept, this);
390 void Chttp2ServerListener::SetOnDestroyDone(grpc_closure* on_destroy_done) {
391 MutexLock lock(&mu_);
392 on_destroy_done_ = on_destroy_done;
395 RefCountedPtr<HandshakeManager> Chttp2ServerListener::CreateHandshakeManager() {
396 MutexLock lock(&mu_);
397 if (shutdown_) return nullptr;
398 grpc_resource_user* resource_user = server_->default_resource_user();
399 if (resource_user != nullptr &&
400 !grpc_resource_user_safe_alloc(resource_user,
401 GRPC_RESOURCE_QUOTA_CHANNEL_SIZE)) {
403 "Memory quota exhausted, rejecting connection, no handshaking.");
406 auto handshake_mgr = MakeRefCounted<HandshakeManager>();
407 handshake_mgr->AddToPendingMgrList(&pending_handshake_mgrs_);
408 grpc_tcp_server_ref(tcp_server_); // Ref held by ConnectionState.
409 return handshake_mgr;
412 void Chttp2ServerListener::OnAccept(void* arg, grpc_endpoint* tcp,
413 grpc_pollset* accepting_pollset,
414 grpc_tcp_server_acceptor* acceptor) {
415 Chttp2ServerListener* self = static_cast<Chttp2ServerListener*>(arg);
416 RefCountedPtr<HandshakeManager> handshake_mgr =
417 self->CreateHandshakeManager();
418 if (handshake_mgr == nullptr) {
419 grpc_endpoint_shutdown(tcp, GRPC_ERROR_NONE);
420 grpc_endpoint_destroy(tcp);
424 // Deletes itself when done.
425 new ConnectionState(self, accepting_pollset, acceptor,
426 std::move(handshake_mgr), self->args_, tcp);
429 void Chttp2ServerListener::TcpServerShutdownComplete(void* arg,
431 Chttp2ServerListener* self = static_cast<Chttp2ServerListener*>(arg);
432 /* ensure all threads have unlocked */
433 grpc_closure* destroy_done = nullptr;
435 MutexLock lock(&self->mu_);
436 destroy_done = self->on_destroy_done_;
437 GPR_ASSERT(self->shutdown_);
438 if (self->pending_handshake_mgrs_ != nullptr) {
439 self->pending_handshake_mgrs_->ShutdownAllPending(GRPC_ERROR_REF(error));
441 self->channelz_listen_socket_.reset();
443 // Flush queued work before destroying handshaker factory, since that
444 // may do a synchronous unref.
445 ExecCtx::Get()->Flush();
446 if (destroy_done != nullptr) {
447 ExecCtx::Run(DEBUG_LOCATION, destroy_done, GRPC_ERROR_REF(error));
448 ExecCtx::Get()->Flush();
453 /* Server callback: destroy the tcp listener (so we don't generate further
455 void Chttp2ServerListener::Orphan() {
456 grpc_tcp_server* tcp_server;
458 MutexLock lock(&mu_);
460 tcp_server = tcp_server_;
462 grpc_tcp_server_shutdown_listeners(tcp_server);
463 grpc_tcp_server_unref(tcp_server);
469 // Chttp2ServerAddPort()
472 grpc_error* Chttp2ServerAddPort(Server* server, const char* addr,
473 grpc_channel_args* args, int* port_num) {
474 if (strncmp(addr, "external:", 9) == 0) {
475 return grpc_core::Chttp2ServerListener::CreateWithAcceptor(server, addr,
478 return grpc_core::Chttp2ServerListener::Create(server, addr, args, port_num);
481 } // namespace grpc_core