Fix some bugs leading to memory leaks

src/core/iomgr/iomgr.c

@@ -105,28 +105,25 @@ void grpc_iomgr_shutdown(void) {
     if (grpc_alarm_check(&exec_ctx, gpr_inf_future(GPR_CLOCK_MONOTONIC),
                          NULL)) {
       gpr_mu_unlock(&g_mu);
-      grpc_exec_ctx_finish(&exec_ctx);
+      grpc_exec_ctx_flush(&exec_ctx);
       gpr_mu_lock(&g_mu);
       continue;
     }
     if (g_root_object.next != &g_root_object) {
-      int timeout = 0;
       gpr_timespec short_deadline = gpr_time_add(
           gpr_now(GPR_CLOCK_REALTIME), gpr_time_from_millis(100, GPR_TIMESPAN));
       if (gpr_cv_wait(&g_rcv, &g_mu, short_deadline)) {
         if (gpr_time_cmp(gpr_now(GPR_CLOCK_REALTIME), shutdown_deadline) > 0) {
-          timeout = 1;
+          if (g_root_object.next != &g_root_object) {
+            gpr_log(GPR_DEBUG,
+                    "Failed to free %d iomgr objects before shutdown deadline: "
+                    "memory leaks are likely",
+                    count_objects());
+            dump_objects("LEAKED");
+          }
           break;
         }
       }
-      if (timeout && g_root_object.next != &g_root_object) {
-        gpr_log(GPR_DEBUG,
-                "Failed to free %d iomgr objects before shutdown deadline: "
-                "memory leaks are likely",
-                count_objects());
-        dump_objects("LEAKED");
-        break;
-      }
     }
   }
   gpr_mu_unlock(&g_mu);

src/core/iomgr/pollset_posix.c

@@ -169,6 +169,7 @@ void grpc_pollset_del_fd(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
 }
 
 static void finish_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset) {
+  GPR_ASSERT(grpc_closure_list_empty(pollset->idle_jobs));
   pollset->vtable->finish_shutdown(pollset);
   grpc_exec_ctx_enqueue(exec_ctx, pollset->shutdown_done, 1);
 }
@@ -236,6 +237,11 @@ done:
        * grpc_pollset_work.
        * TODO(dklempner): Can we refactor the shutdown logic to avoid this? */
       gpr_mu_lock(&pollset->mu);
+    } else if (!grpc_closure_list_empty(pollset->idle_jobs)) {
+      gpr_mu_unlock(&pollset->mu);
+      grpc_exec_ctx_enqueue_list(exec_ctx, &pollset->idle_jobs);
+      grpc_exec_ctx_flush(exec_ctx);
+      gpr_mu_lock(&pollset->mu);
     }
   }
 }
@@ -251,6 +257,9 @@ void grpc_pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
     pollset->called_shutdown = 1;
     call_shutdown = 1;
   }
+  if (!grpc_pollset_has_workers(pollset)) {
+    grpc_exec_ctx_enqueue_list(exec_ctx, &pollset->idle_jobs);
+  }
   pollset->shutdown_done = closure;
   grpc_pollset_kick(pollset, GRPC_POLLSET_KICK_BROADCAST);
   gpr_mu_unlock(&pollset->mu);
@@ -326,9 +335,7 @@ static void basic_do_promote(grpc_exec_ctx *exec_ctx, void *args, int success) {
   if (pollset->shutting_down) {
     /* We don't care about this pollset anymore. */
     if (pollset->in_flight_cbs == 0 && !pollset->called_shutdown) {
-      GPR_ASSERT(!grpc_pollset_has_workers(pollset));
+      finish_shutdown(exec_ctx, pollset);
-      pollset->called_shutdown = 1;
-      grpc_exec_ctx_enqueue(exec_ctx, pollset->shutdown_done, 1);
     }
   } else if (grpc_fd_is_orphaned(fd)) {
     /* Don't try to add it to anything, we'll drop our ref on it below */

src/core/surface/channel_connectivity.c

@@ -52,6 +52,7 @@ grpc_connectivity_state grpc_channel_check_connectivity_state(
             "grpc_channel_check_connectivity_state called on something that is "
             "not a client channel, but '%s'",
             client_channel_elem->filter->name);
+    grpc_exec_ctx_finish(&exec_ctx);
     return GRPC_CHANNEL_FATAL_FAILURE;
   }
   state = grpc_client_channel_check_connectivity_state(

src/core/surface/secure_channel_create.c

@@ -249,6 +249,7 @@ grpc_channel *grpc_secure_channel_create(grpc_credentials *creds,
   GPR_ASSERT(reserved == NULL);
   if (grpc_find_security_connector_in_args(args) != NULL) {
     gpr_log(GPR_ERROR, "Cannot set security context in channel args.");
+    grpc_exec_ctx_finish(&exec_ctx);
     return grpc_lame_client_channel_create(
         target, GRPC_STATUS_INVALID_ARGUMENT,
         "Security connector exists in channel args.");
@@ -257,6 +258,7 @@ grpc_channel *grpc_secure_channel_create(grpc_credentials *creds,
   if (grpc_credentials_create_security_connector(
           creds, target, args, NULL, &security_connector,
           &new_args_from_connector) != GRPC_SECURITY_OK) {
+    grpc_exec_ctx_finish(&exec_ctx);
     return grpc_lame_client_channel_create(
         target, GRPC_STATUS_INVALID_ARGUMENT,
         "Failed to create security connector.");
@@ -289,6 +291,7 @@ grpc_channel *grpc_secure_channel_create(grpc_credentials *creds,
   GRPC_CHANNEL_INTERNAL_REF(channel, "subchannel_factory");
   resolver = grpc_resolver_create(target, &f->base);
   if (!resolver) {
+    grpc_exec_ctx_finish(&exec_ctx);
     return NULL;
   }