From ad4d2dde0052efbbf49d64b0843c45f0381cfeb3 Mon Sep 17 00:00:00 2001 From: Yash Tibrewal Date: Wed, 6 Dec 2017 09:05:05 -0800 Subject: Revert "All instances of exec_ctx being passed around in src/core removed" --- .../ext/transport/chttp2/server/chttp2_server.cc | 108 +++++++++++---------- .../ext/transport/chttp2/server/chttp2_server.h | 3 +- .../chttp2/server/insecure/server_chttp2.cc | 6 +- .../chttp2/server/insecure/server_chttp2_posix.cc | 17 ++-- .../chttp2/server/secure/server_secure_chttp2.cc | 11 ++- 5 files changed, 79 insertions(+), 66 deletions(-) (limited to 'src/core/ext/transport/chttp2/server') 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; diff --git a/src/core/ext/transport/chttp2/server/chttp2_server.h b/src/core/ext/transport/chttp2/server/chttp2_server.h index 7de859da42..68304fd4f7 100644 --- a/src/core/ext/transport/chttp2/server/chttp2_server.h +++ b/src/core/ext/transport/chttp2/server/chttp2_server.h @@ -25,7 +25,8 @@ /// Adds a port to \a server. Sets \a port_num to the port number. /// Takes ownership of \a args. -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); #endif /* GRPC_CORE_EXT_TRANSPORT_CHTTP2_SERVER_CHTTP2_SERVER_H */ diff --git a/src/core/ext/transport/chttp2/server/insecure/server_chttp2.cc b/src/core/ext/transport/chttp2/server/insecure/server_chttp2.cc index 52c42d056c..8984896538 100644 --- a/src/core/ext/transport/chttp2/server/insecure/server_chttp2.cc +++ b/src/core/ext/transport/chttp2/server/insecure/server_chttp2.cc @@ -26,12 +26,12 @@ #include "src/core/lib/surface/server.h" int grpc_server_add_insecure_http2_port(grpc_server* server, const char* addr) { - grpc_core::ExecCtx exec_ctx; + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; int port_num = 0; GRPC_API_TRACE("grpc_server_add_insecure_http2_port(server=%p, addr=%s)", 2, (server, addr)); grpc_error* err = grpc_chttp2_server_add_port( - server, addr, + &exec_ctx, server, addr, grpc_channel_args_copy(grpc_server_get_channel_args(server)), &port_num); if (err != GRPC_ERROR_NONE) { const char* msg = grpc_error_string(err); @@ -39,6 +39,6 @@ int grpc_server_add_insecure_http2_port(grpc_server* server, const char* addr) { GRPC_ERROR_UNREF(err); } - + grpc_exec_ctx_finish(&exec_ctx); return port_num; } diff --git a/src/core/ext/transport/chttp2/server/insecure/server_chttp2_posix.cc b/src/core/ext/transport/chttp2/server/insecure/server_chttp2_posix.cc index dafd4af6ce..3fe05ce4ef 100644 --- a/src/core/ext/transport/chttp2/server/insecure/server_chttp2_posix.cc +++ b/src/core/ext/transport/chttp2/server/insecure/server_chttp2_posix.cc @@ -38,29 +38,32 @@ void grpc_server_add_insecure_channel_from_fd(grpc_server* server, void* reserved, int fd) { GPR_ASSERT(reserved == nullptr); - grpc_core::ExecCtx exec_ctx; + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; char* name; gpr_asprintf(&name, "fd:%d", fd); - grpc_endpoint* server_endpoint = grpc_tcp_create( - grpc_fd_create(fd, name), grpc_server_get_channel_args(server), name); + grpc_endpoint* server_endpoint = + grpc_tcp_create(&exec_ctx, grpc_fd_create(fd, name), + grpc_server_get_channel_args(server), name); gpr_free(name); const grpc_channel_args* server_args = grpc_server_get_channel_args(server); grpc_transport* transport = grpc_create_chttp2_transport( - server_args, server_endpoint, false /* is_client */); + &exec_ctx, server_args, server_endpoint, false /* is_client */); grpc_pollset** pollsets; size_t num_pollsets = 0; grpc_server_get_pollsets(server, &pollsets, &num_pollsets); for (size_t i = 0; i < num_pollsets; i++) { - grpc_endpoint_add_to_pollset(server_endpoint, pollsets[i]); + grpc_endpoint_add_to_pollset(&exec_ctx, server_endpoint, pollsets[i]); } - grpc_server_setup_transport(server, transport, nullptr, server_args); - grpc_chttp2_transport_start_reading(transport, nullptr, nullptr); + grpc_server_setup_transport(&exec_ctx, server, transport, nullptr, + server_args); + grpc_chttp2_transport_start_reading(&exec_ctx, transport, nullptr, nullptr); + grpc_exec_ctx_finish(&exec_ctx); } #else // !GPR_SUPPORT_CHANNELS_FROM_FD diff --git a/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.cc b/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.cc index 723af97ff0..ac3ea40f47 100644 --- a/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.cc +++ b/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.cc @@ -36,7 +36,7 @@ int grpc_server_add_secure_http2_port(grpc_server* server, const char* addr, grpc_server_credentials* creds) { - grpc_core::ExecCtx exec_ctx; + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; grpc_error* err = GRPC_ERROR_NONE; grpc_server_security_connector* sc = nullptr; int port_num = 0; @@ -52,7 +52,8 @@ int grpc_server_add_secure_http2_port(grpc_server* server, const char* addr, "No credentials specified for secure server port (creds==NULL)"); goto done; } - status = grpc_server_credentials_create_security_connector(creds, &sc); + status = + grpc_server_credentials_create_security_connector(&exec_ctx, creds, &sc); if (status != GRPC_SECURITY_OK) { char* msg; gpr_asprintf(&msg, @@ -71,12 +72,12 @@ int grpc_server_add_secure_http2_port(grpc_server* server, const char* addr, grpc_channel_args_copy_and_add(grpc_server_get_channel_args(server), args_to_add, GPR_ARRAY_SIZE(args_to_add)); // Add server port. - err = grpc_chttp2_server_add_port(server, addr, args, &port_num); + err = grpc_chttp2_server_add_port(&exec_ctx, server, addr, args, &port_num); done: if (sc != nullptr) { - GRPC_SECURITY_CONNECTOR_UNREF(&sc->base, "server"); + GRPC_SECURITY_CONNECTOR_UNREF(&exec_ctx, &sc->base, "server"); } - + grpc_exec_ctx_finish(&exec_ctx); if (err != GRPC_ERROR_NONE) { const char* msg = grpc_error_string(err); gpr_log(GPR_ERROR, "%s", msg); -- cgit v1.2.3