diff options
Diffstat (limited to 'src/core/ext/transport/chttp2/server/chttp2_server.cc')
| -rw-r--r-- | src/core/ext/transport/chttp2/server/chttp2_server.cc | 108 |
1 files changed, 58 insertions, 50 deletions
diff --git a/src/core/ext/transport/chttp2/server/chttp2_server.cc b/src/core/ext/transport/chttp2/server/chttp2_server.cc index 5669fa4090..49ee677464 100644 --- a/src/core/ext/transport/chttp2/server/chttp2_server.cc +++ b/src/core/ext/transport/chttp2/server/chttp2_server.cc @@ -69,17 +69,17 @@ typedef struct { } server_connection_state; static void server_connection_state_unref( - server_connection_state* connection_state) { + grpc_exec_ctx* exec_ctx, server_connection_state* connection_state) { if (gpr_unref(&connection_state->refs)) { if (connection_state->transport != nullptr) { - GRPC_CHTTP2_UNREF_TRANSPORT(connection_state->transport, + GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, connection_state->transport, "receive settings timeout"); } gpr_free(connection_state); } } -static void on_timeout(void* arg, grpc_error* error) { +static void on_timeout(grpc_exec_ctx* exec_ctx, void* arg, grpc_error* error) { server_connection_state* connection_state = (server_connection_state*)arg; // Note that we may be called with GRPC_ERROR_NONE when the timer fires // or with an error indicating that the timer system is being shut down. @@ -87,20 +87,22 @@ static void on_timeout(void* arg, grpc_error* error) { grpc_transport_op* op = grpc_make_transport_op(nullptr); op->disconnect_with_error = GRPC_ERROR_CREATE_FROM_STATIC_STRING( "Did not receive HTTP/2 settings before handshake timeout"); - grpc_transport_perform_op(&connection_state->transport->base, op); + grpc_transport_perform_op(exec_ctx, &connection_state->transport->base, op); } - server_connection_state_unref(connection_state); + server_connection_state_unref(exec_ctx, connection_state); } -static void on_receive_settings(void* arg, grpc_error* error) { +static void on_receive_settings(grpc_exec_ctx* exec_ctx, void* arg, + grpc_error* error) { server_connection_state* connection_state = (server_connection_state*)arg; if (error == GRPC_ERROR_NONE) { - grpc_timer_cancel(&connection_state->timer); + grpc_timer_cancel(exec_ctx, &connection_state->timer); } - server_connection_state_unref(connection_state); + server_connection_state_unref(exec_ctx, connection_state); } -static void on_handshake_done(void* arg, grpc_error* error) { +static void on_handshake_done(grpc_exec_ctx* exec_ctx, void* arg, + grpc_error* error) { grpc_handshaker_args* args = (grpc_handshaker_args*)arg; server_connection_state* connection_state = (server_connection_state*)args->user_data; @@ -115,10 +117,10 @@ static void on_handshake_done(void* arg, grpc_error* error) { // before destroying them, even if we know that there are no // pending read/write callbacks. This should be fixed, at which // point this can be removed. - grpc_endpoint_shutdown(args->endpoint, GRPC_ERROR_NONE); - grpc_endpoint_destroy(args->endpoint); - grpc_channel_args_destroy(args->args); - grpc_slice_buffer_destroy_internal(args->read_buffer); + grpc_endpoint_shutdown(exec_ctx, args->endpoint, GRPC_ERROR_NONE); + grpc_endpoint_destroy(exec_ctx, args->endpoint); + grpc_channel_args_destroy(exec_ctx, args->args); + grpc_slice_buffer_destroy_internal(exec_ctx, args->read_buffer); gpr_free(args->read_buffer); } } else { @@ -126,10 +128,10 @@ static void on_handshake_done(void* arg, grpc_error* error) { // handshaker may have handed off the connection to some external // code, so we can just clean up here without creating a transport. if (args->endpoint != nullptr) { - grpc_transport* transport = - grpc_create_chttp2_transport(args->args, args->endpoint, false); + grpc_transport* transport = grpc_create_chttp2_transport( + exec_ctx, args->args, args->endpoint, false); grpc_server_setup_transport( - connection_state->svr_state->server, transport, + exec_ctx, connection_state->svr_state->server, transport, connection_state->accepting_pollset, args->args); // Use notify_on_receive_settings callback to enforce the // handshake deadline. @@ -139,14 +141,16 @@ static void on_handshake_done(void* arg, grpc_error* error) { on_receive_settings, connection_state, grpc_schedule_on_exec_ctx); grpc_chttp2_transport_start_reading( - transport, args->read_buffer, &connection_state->on_receive_settings); - grpc_channel_args_destroy(args->args); + exec_ctx, transport, args->read_buffer, + &connection_state->on_receive_settings); + grpc_channel_args_destroy(exec_ctx, args->args); gpr_ref(&connection_state->refs); GRPC_CHTTP2_REF_TRANSPORT((grpc_chttp2_transport*)transport, "receive settings timeout"); GRPC_CLOSURE_INIT(&connection_state->on_timeout, on_timeout, connection_state, grpc_schedule_on_exec_ctx); - grpc_timer_init(&connection_state->timer, connection_state->deadline, + grpc_timer_init(exec_ctx, &connection_state->timer, + connection_state->deadline, &connection_state->on_timeout); } } @@ -154,21 +158,21 @@ static void on_handshake_done(void* arg, grpc_error* error) { &connection_state->svr_state->pending_handshake_mgrs, connection_state->handshake_mgr); gpr_mu_unlock(&connection_state->svr_state->mu); - grpc_handshake_manager_destroy(connection_state->handshake_mgr); + grpc_handshake_manager_destroy(exec_ctx, connection_state->handshake_mgr); gpr_free(connection_state->acceptor); - grpc_tcp_server_unref(connection_state->svr_state->tcp_server); - server_connection_state_unref(connection_state); + grpc_tcp_server_unref(exec_ctx, connection_state->svr_state->tcp_server); + server_connection_state_unref(exec_ctx, connection_state); } -static void on_accept(void* arg, grpc_endpoint* tcp, +static void on_accept(grpc_exec_ctx* exec_ctx, void* arg, grpc_endpoint* tcp, grpc_pollset* accepting_pollset, grpc_tcp_server_acceptor* acceptor) { server_state* state = (server_state*)arg; gpr_mu_lock(&state->mu); if (state->shutdown) { gpr_mu_unlock(&state->mu); - grpc_endpoint_shutdown(tcp, GRPC_ERROR_NONE); - grpc_endpoint_destroy(tcp); + grpc_endpoint_shutdown(exec_ctx, tcp, GRPC_ERROR_NONE); + grpc_endpoint_destroy(exec_ctx, tcp); gpr_free(acceptor); return; } @@ -184,56 +188,59 @@ static void on_accept(void* arg, grpc_endpoint* tcp, connection_state->accepting_pollset = accepting_pollset; connection_state->acceptor = acceptor; connection_state->handshake_mgr = handshake_mgr; - grpc_handshakers_add(HANDSHAKER_SERVER, state->args, + grpc_handshakers_add(exec_ctx, HANDSHAKER_SERVER, state->args, connection_state->handshake_mgr); const grpc_arg* timeout_arg = grpc_channel_args_find(state->args, GRPC_ARG_SERVER_HANDSHAKE_TIMEOUT_MS); connection_state->deadline = - grpc_core::ExecCtx::Get()->Now() + + grpc_exec_ctx_now(exec_ctx) + grpc_channel_arg_get_integer(timeout_arg, {120 * GPR_MS_PER_SEC, 1, INT_MAX}); - grpc_handshake_manager_do_handshake( - connection_state->handshake_mgr, nullptr /* interested_parties */, tcp, - state->args, connection_state->deadline, acceptor, on_handshake_done, - connection_state); + grpc_handshake_manager_do_handshake(exec_ctx, connection_state->handshake_mgr, + nullptr /* interested_parties */, tcp, + state->args, connection_state->deadline, + acceptor, on_handshake_done, + connection_state); } /* Server callback: start listening on our ports */ -static void server_start_listener(grpc_server* server, void* arg, - grpc_pollset** pollsets, +static void server_start_listener(grpc_exec_ctx* exec_ctx, grpc_server* server, + void* arg, grpc_pollset** pollsets, size_t pollset_count) { server_state* state = (server_state*)arg; gpr_mu_lock(&state->mu); state->shutdown = false; gpr_mu_unlock(&state->mu); - grpc_tcp_server_start(state->tcp_server, pollsets, pollset_count, on_accept, - state); + grpc_tcp_server_start(exec_ctx, state->tcp_server, pollsets, pollset_count, + on_accept, state); } -static void tcp_server_shutdown_complete(void* arg, grpc_error* error) { +static void tcp_server_shutdown_complete(grpc_exec_ctx* exec_ctx, void* arg, + grpc_error* error) { server_state* state = (server_state*)arg; /* ensure all threads have unlocked */ gpr_mu_lock(&state->mu); grpc_closure* destroy_done = state->server_destroy_listener_done; GPR_ASSERT(state->shutdown); grpc_handshake_manager_pending_list_shutdown_all( - state->pending_handshake_mgrs, GRPC_ERROR_REF(error)); + exec_ctx, state->pending_handshake_mgrs, GRPC_ERROR_REF(error)); gpr_mu_unlock(&state->mu); // Flush queued work before destroying handshaker factory, since that // may do a synchronous unref. - grpc_core::ExecCtx::Get()->Flush(); + grpc_exec_ctx_flush(exec_ctx); if (destroy_done != nullptr) { - destroy_done->cb(destroy_done->cb_arg, GRPC_ERROR_REF(error)); - grpc_core::ExecCtx::Get()->Flush(); + destroy_done->cb(exec_ctx, destroy_done->cb_arg, GRPC_ERROR_REF(error)); + grpc_exec_ctx_flush(exec_ctx); } - grpc_channel_args_destroy(state->args); + grpc_channel_args_destroy(exec_ctx, state->args); gpr_mu_destroy(&state->mu); gpr_free(state); } /* Server callback: destroy the tcp listener (so we don't generate further callbacks) */ -static void server_destroy_listener(grpc_server* server, void* arg, +static void server_destroy_listener(grpc_exec_ctx* exec_ctx, + grpc_server* server, void* arg, grpc_closure* destroy_done) { server_state* state = (server_state*)arg; gpr_mu_lock(&state->mu); @@ -241,11 +248,12 @@ static void server_destroy_listener(grpc_server* server, void* arg, state->server_destroy_listener_done = destroy_done; grpc_tcp_server* tcp_server = state->tcp_server; gpr_mu_unlock(&state->mu); - grpc_tcp_server_shutdown_listeners(tcp_server); - grpc_tcp_server_unref(tcp_server); + grpc_tcp_server_shutdown_listeners(exec_ctx, tcp_server); + grpc_tcp_server_unref(exec_ctx, tcp_server); } -grpc_error* grpc_chttp2_server_add_port(grpc_server* server, const char* addr, +grpc_error* grpc_chttp2_server_add_port(grpc_exec_ctx* exec_ctx, + grpc_server* server, const char* addr, grpc_channel_args* args, int* port_num) { grpc_resolved_addresses* resolved = nullptr; @@ -269,8 +277,8 @@ grpc_error* grpc_chttp2_server_add_port(grpc_server* server, const char* addr, GRPC_CLOSURE_INIT(&state->tcp_server_shutdown_complete, tcp_server_shutdown_complete, state, grpc_schedule_on_exec_ctx); - err = grpc_tcp_server_create(&state->tcp_server_shutdown_complete, args, - &tcp_server); + err = grpc_tcp_server_create(exec_ctx, &state->tcp_server_shutdown_complete, + args, &tcp_server); if (err != GRPC_ERROR_NONE) { goto error; } @@ -319,7 +327,7 @@ grpc_error* grpc_chttp2_server_add_port(grpc_server* server, const char* addr, grpc_resolved_addresses_destroy(resolved); /* Register with the server only upon success */ - grpc_server_add_listener(server, state, server_start_listener, + grpc_server_add_listener(exec_ctx, server, state, server_start_listener, server_destroy_listener); goto done; @@ -330,9 +338,9 @@ error: grpc_resolved_addresses_destroy(resolved); } if (tcp_server) { - grpc_tcp_server_unref(tcp_server); + grpc_tcp_server_unref(exec_ctx, tcp_server); } else { - grpc_channel_args_destroy(args); + grpc_channel_args_destroy(exec_ctx, args); gpr_free(state); } *port_num = 0; |