Removing instances of exec_ctx being passed around in functions in

src/core. exec_ctx is now a thread_local pointer of type ExecCtx instead of
grpc_exec_ctx which is initialized whenever ExecCtx is instantiated. ExecCtx
also keeps track of the previous exec_ctx so that nesting of exec_ctx is
allowed. This means that there is only one exec_ctx being used at any
time. Also, grpc_exec_ctx_finish is called in the destructor of the
object, and the previous exec_ctx is restored to avoid breaking current
functionality. The code still explicitly calls grpc_exec_ctx_finish
because removing all such instances causes the code to break.

1 files changed, 27 insertions, 36 deletions

diff --git a/src/core/lib/iomgr/tcp_server_windows.cc b/src/core/lib/iomgr/tcp_server_windows.cc
index f198aaaa5b..4dfb17229a 100644
--- a/src/core/lib/iomgr/tcp_server_windows.cc
+++ b/src/core/lib/iomgr/tcp_server_windows.cc
@@ -94,8 +94,7 @@ struct grpc_tcp_server {
 
 /* Public function. Allocates the proper data structures to hold a
    grpc_tcp_server. */
-grpc_error *grpc_tcp_server_create(grpc_exec_ctx *exec_ctx,
-                                   grpc_closure *shutdown_complete,
+grpc_error *grpc_tcp_server_create(grpc_closure *shutdown_complete,
                                    const grpc_channel_args *args,
                                    grpc_tcp_server **server) {
   grpc_tcp_server *s = (grpc_tcp_server *)gpr_malloc(sizeof(grpc_tcp_server));
@@ -114,8 +113,7 @@ grpc_error *grpc_tcp_server_create(grpc_exec_ctx *exec_ctx,
   return GRPC_ERROR_NONE;
 }
 
-static void destroy_server(grpc_exec_ctx *exec_ctx, void *arg,
-                           grpc_error *error) {
+static void destroy_server(void *arg, grpc_error *error) {
   grpc_tcp_server *s = (grpc_tcp_server *)arg;
 
   /* Now that the accepts have been aborted, we can destroy the sockets.
@@ -128,19 +126,18 @@ static void destroy_server(grpc_exec_ctx *exec_ctx, void *arg,
     grpc_winsocket_destroy(sp->socket);
     gpr_free(sp);
   }
-  grpc_channel_args_destroy(exec_ctx, s->channel_args);
+  grpc_channel_args_destroy(s->channel_args);
   gpr_free(s);
 }
 
-static void finish_shutdown_locked(grpc_exec_ctx *exec_ctx,
-                                   grpc_tcp_server *s) {
+static void finish_shutdown_locked(grpc_tcp_server *s) {
   if (s->shutdown_complete != NULL) {
-    GRPC_CLOSURE_SCHED(exec_ctx, s->shutdown_complete, GRPC_ERROR_NONE);
+    GRPC_CLOSURE_SCHED(s->shutdown_complete, GRPC_ERROR_NONE);
   }
 
-  GRPC_CLOSURE_SCHED(exec_ctx, GRPC_CLOSURE_CREATE(destroy_server, s,
-                                                   grpc_schedule_on_exec_ctx),
-                     GRPC_ERROR_NONE);
+  GRPC_CLOSURE_SCHED(
+      GRPC_CLOSURE_CREATE(destroy_server, s, grpc_schedule_on_exec_ctx),
+      GRPC_ERROR_NONE);
 }
 
 grpc_tcp_server *grpc_tcp_server_ref(grpc_tcp_server *s) {
@@ -156,14 +153,14 @@ void grpc_tcp_server_shutdown_starting_add(grpc_tcp_server *s,
   gpr_mu_unlock(&s->mu);
 }
 
-static void tcp_server_destroy(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) {
+static void tcp_server_destroy(grpc_tcp_server *s) {
   grpc_tcp_listener *sp;
   gpr_mu_lock(&s->mu);
 
   /* First, shutdown all fd's. This will queue abortion calls for all
      of the pending accepts due to the normal operation mechanism. */
   if (s->active_ports == 0) {
-    finish_shutdown_locked(exec_ctx, s);
+    finish_shutdown_locked(s);
   } else {
     for (sp = s->head; sp; sp = sp->next) {
       sp->shutting_down = 1;
@@ -173,13 +170,13 @@ static void tcp_server_destroy(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) {
   gpr_mu_unlock(&s->mu);
 }
 
-void grpc_tcp_server_unref(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) {
+void grpc_tcp_server_unref(grpc_tcp_server *s) {
   if (gpr_unref(&s->refs)) {
-    grpc_tcp_server_shutdown_listeners(exec_ctx, s);
+    grpc_tcp_server_shutdown_listeners(s);
     gpr_mu_lock(&s->mu);
-    GRPC_CLOSURE_LIST_SCHED(exec_ctx, &s->shutdown_starting);
+    GRPC_CLOSURE_LIST_SCHED(&s->shutdown_starting);
     gpr_mu_unlock(&s->mu);
-    tcp_server_destroy(exec_ctx, s);
+    tcp_server_destroy(s);
   }
 }
 
@@ -233,19 +230,17 @@ failure:
   return error;
 }
 
-static void decrement_active_ports_and_notify_locked(grpc_exec_ctx *exec_ctx,
-                                                     grpc_tcp_listener *sp) {
+static void decrement_active_ports_and_notify_locked(grpc_tcp_listener *sp) {
   sp->shutting_down = 0;
   GPR_ASSERT(sp->server->active_ports > 0);
   if (0 == --sp->server->active_ports) {
-    finish_shutdown_locked(exec_ctx, sp->server);
+    finish_shutdown_locked(sp->server);
   }
 }
 
 /* In order to do an async accept, we need to create a socket first which
    will be the one assigned to the new incoming connection. */
-static grpc_error *start_accept_locked(grpc_exec_ctx *exec_ctx,
-                                       grpc_tcp_listener *port) {
+static grpc_error *start_accept_locked(grpc_tcp_listener *port) {
   SOCKET sock = INVALID_SOCKET;
   BOOL success;
   DWORD addrlen = sizeof(struct sockaddr_in6) + 16;
@@ -284,7 +279,7 @@ static grpc_error *start_accept_locked(grpc_exec_ctx *exec_ctx,
   /* We're ready to do the accept. Calling grpc_socket_notify_on_read may
      immediately process an accept that happened in the meantime. */
   port->new_socket = sock;
-  grpc_socket_notify_on_read(exec_ctx, port->socket, &port->on_accept);
+  grpc_socket_notify_on_read(port->socket, &port->on_accept);
   port->outstanding_calls++;
   return error;
 
@@ -295,7 +290,7 @@ failure:
 }
 
 /* Event manager callback when reads are ready. */
-static void on_accept(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
+static void on_accept(void *arg, grpc_error *error) {
   grpc_tcp_listener *sp = (grpc_tcp_listener *)arg;
   SOCKET sock = sp->new_socket;
   grpc_winsocket_callback_info *info = &sp->socket->read_info;
@@ -357,7 +352,7 @@ static void on_accept(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
         gpr_free(utf8_message);
       }
       gpr_asprintf(&fd_name, "tcp_server:%s", peer_name_string);
-      ep = grpc_tcp_create(exec_ctx, grpc_winsocket_create(sock, fd_name),
+      ep = grpc_tcp_create(grpc_winsocket_create(sock, fd_name),
                            sp->server->channel_args, peer_name_string);
       gpr_free(fd_name);
       gpr_free(peer_name_string);
@@ -375,17 +370,15 @@ static void on_accept(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
     acceptor->from_server = sp->server;
     acceptor->port_index = sp->port_index;
     acceptor->fd_index = 0;
-    sp->server->on_accept_cb(exec_ctx, sp->server->on_accept_cb_arg, ep, NULL,
-                             acceptor);
+    sp->server->on_accept_cb(sp->server->on_accept_cb_arg, ep, NULL, acceptor);
   }
   /* As we were notified from the IOCP of one and exactly one accept,
      the former socked we created has now either been destroy or assigned
      to the new connection. We need to create a new one for the next
      connection. */
-  GPR_ASSERT(
-      GRPC_LOG_IF_ERROR("start_accept", start_accept_locked(exec_ctx, sp)));
+  GPR_ASSERT(GRPC_LOG_IF_ERROR("start_accept", start_accept_locked(sp)));
   if (0 == --sp->outstanding_calls) {
-    decrement_active_ports_and_notify_locked(exec_ctx, sp);
+    decrement_active_ports_and_notify_locked(sp);
   }
   gpr_mu_unlock(&sp->server->mu);
 }
@@ -522,8 +515,8 @@ done:
   return error;
 }
 
-void grpc_tcp_server_start(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s,
-                           grpc_pollset **pollset, size_t pollset_count,
+void grpc_tcp_server_start(grpc_tcp_server *s, grpc_pollset **pollset,
+                           size_t pollset_count,
                            grpc_tcp_server_cb on_accept_cb,
                            void *on_accept_cb_arg) {
   grpc_tcp_listener *sp;
@@ -534,14 +527,12 @@ void grpc_tcp_server_start(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s,
   s->on_accept_cb = on_accept_cb;
   s->on_accept_cb_arg = on_accept_cb_arg;
   for (sp = s->head; sp; sp = sp->next) {
-    GPR_ASSERT(
-        GRPC_LOG_IF_ERROR("start_accept", start_accept_locked(exec_ctx, sp)));
+    GPR_ASSERT(GRPC_LOG_IF_ERROR("start_accept", start_accept_locked(sp)));
     s->active_ports++;
   }
   gpr_mu_unlock(&s->mu);
 }
 
-void grpc_tcp_server_shutdown_listeners(grpc_exec_ctx *exec_ctx,
-                                        grpc_tcp_server *s) {}
+void grpc_tcp_server_shutdown_listeners(grpc_tcp_server *s) {}
 
 #endif /* GRPC_WINSOCK_SOCKET */