Fix leak

1 files changed, 25 insertions, 17 deletions

diff --git a/src/core/lib/iomgr/ev_epollex_linux.cc b/src/core/lib/iomgr/ev_epollex_linux.cc
index 1e8de8910e..8b0ebc2d17 100644
--- a/src/core/lib/iomgr/ev_epollex_linux.cc
+++ b/src/core/lib/iomgr/ev_epollex_linux.cc
@@ -201,7 +201,7 @@ struct grpc_pollset_set {
 
   size_t pollset_count;
   size_t pollset_capacity;
-  pollable **pollsets;
+  grpc_pollset **pollsets;
 
   size_t fd_count;
   size_t fd_capacity;
@@ -545,6 +545,9 @@ static void pollset_global_shutdown(void) {
 
 static void pollset_maybe_finish_shutdown(grpc_exec_ctx *exec_ctx,
                                           grpc_pollset *pollset) {
+if (GRPC_TRACER_ON(grpc_polling_trace)) {
+gpr_log(GPR_DEBUG, "PS:%p (pollable:%p) maybe_finish_shutdown sc=%p (target:!NULL) rw=%p (target:NULL) cpsc=%d (target:0)", pollset, pollset->active_pollable, pollset->shutdown_closure, pollset->root_worker, pollset->containing_pollset_set_count);
+}
   if (pollset->shutdown_closure != NULL && pollset->root_worker == NULL &&
       pollset->containing_pollset_set_count == 0) {
     GRPC_CLOSURE_SCHED(exec_ctx, pollset->shutdown_closure, GRPC_ERROR_NONE);
@@ -1123,7 +1126,11 @@ static void pollset_set_unref(grpc_exec_ctx *exec_ctx, grpc_pollset_set *pss) {
   pollset_set_unref(exec_ctx, pss->parent);
   gpr_mu_destroy(&pss->mu);
   for (size_t i = 0; i < pss->pollset_count; i++) {
-    POLLABLE_UNREF(pss->pollsets[i], "pollset_set");
+    gpr_mu_lock(&pss->pollsets[i]->mu);
+    if (0==--pss->pollsets[i]->containing_pollset_set_count) {
+      pollset_maybe_finish_shutdown(exec_ctx, pss->pollsets[i]);
+    }
+    gpr_mu_unlock(&pss->pollsets[i]->mu);
   }
   for (size_t i = 0; i < pss->fd_count; i++) {
     UNREF_BY(exec_ctx, pss->fds[i], 2, "pollset_set");
@@ -1142,7 +1149,7 @@ static void pollset_set_add_fd(grpc_exec_ctx *exec_ctx, grpc_pollset_set *pss,
   static const char *err_desc = "pollset_set_add_fd";
   pss = pss_lock_adam(pss);
   for (size_t i = 0; i < pss->pollset_count; i++) {
-    append_error(&error, pollable_add_fd(pss->pollsets[i], fd), err_desc);
+    append_error(&error, pollable_add_fd(pss->pollsets[i]->active_pollable, fd), err_desc);
   }
   if (pss->fd_count == pss->fd_capacity) {
     pss->fd_capacity = GPR_MAX(pss->fd_capacity * 2, 8);
@@ -1185,8 +1192,7 @@ static void pollset_set_del_pollset(grpc_exec_ctx *exec_ctx,
   pss = pss_lock_adam(pss);
   size_t i;
   for (i = 0; i < pss->pollset_count; i++) {
-    if (pss->pollsets[i] == ps->active_pollable) {
-      POLLABLE_UNREF(pss->pollsets[i], "pollset_set");
+    if (pss->pollsets[i] == ps) {
       break;
     }
   }
@@ -1204,9 +1210,10 @@ static void pollset_set_del_pollset(grpc_exec_ctx *exec_ctx,
 }
 
 // add all fds to pollables, and output a new array of unorphaned out_fds
-static grpc_error *add_fds_to_pollables(grpc_exec_ctx *exec_ctx, grpc_fd **fds,
-                                        size_t fd_count, pollable **pollables,
-                                        size_t pollable_count,
+// assumes pollsets are multipollable
+static grpc_error *add_fds_to_pollsets(grpc_exec_ctx *exec_ctx, grpc_fd **fds,
+                                        size_t fd_count, grpc_pollset **pollsets,
+                                        size_t pollset_count,
                                         const char *err_desc, grpc_fd **out_fds,
                                         size_t *out_fd_count) {
   grpc_error *error = GRPC_ERROR_NONE;
@@ -1216,8 +1223,8 @@ static grpc_error *add_fds_to_pollables(grpc_exec_ctx *exec_ctx, grpc_fd **fds,
       gpr_mu_unlock(&fds[i]->orphan_mu);
       UNREF_BY(exec_ctx, fds[i], 2, "pollset_set");
     } else {
-      for (size_t j = 0; j < pollable_count; j++) {
-        append_error(&error, pollable_add_fd(pollables[j], fds[i]), err_desc);
+      for (size_t j = 0; j < pollset_count; j++) {
+        append_error(&error, pollable_add_fd(pollsets[j]->active_pollable, fds[i]), err_desc);
       }
       gpr_mu_unlock(&fds[i]->orphan_mu);
       out_fds[(*out_fd_count)++] = fds[i];
@@ -1246,17 +1253,18 @@ static void pollset_set_add_pollset(grpc_exec_ctx *exec_ctx,
   pss = pss_lock_adam(pss);
   size_t initial_fd_count = pss->fd_count;
   pss->fd_count = 0;
-  append_error(&error, add_fds_to_pollables(exec_ctx, pss->fds,
-                                            initial_fd_count, &pollable_obj, 1,
+  append_error(&error, add_fds_to_pollsets(exec_ctx, pss->fds,
+                                            initial_fd_count, &ps, 1,
                                             err_desc, pss->fds, &pss->fd_count),
                err_desc);
   if (pss->pollset_count == pss->pollset_capacity) {
     pss->pollset_capacity = GPR_MAX(pss->pollset_capacity * 2, 8);
-    pss->pollsets = (pollable **)gpr_realloc(
+    pss->pollsets = (grpc_pollset **)gpr_realloc(
         pss->pollsets, pss->pollset_capacity * sizeof(*pss->pollsets));
   }
-  pss->pollsets[pss->pollset_count++] = pollable_obj;
+  pss->pollsets[pss->pollset_count++] = ps;
   gpr_mu_unlock(&pss->mu);
+POLLABLE_UNREF(pollable_obj, "pollset_set");
 
   GRPC_LOG_IF_ERROR(err_desc, error);
 }
@@ -1309,18 +1317,18 @@ static void pollset_set_add_pollset_set(grpc_exec_ctx *exec_ctx,
   }
   size_t initial_a_fd_count = a->fd_count;
   a->fd_count = 0;
-  append_error(&error, add_fds_to_pollables(
+  append_error(&error, add_fds_to_pollsets(
                            exec_ctx, a->fds, initial_a_fd_count, b->pollsets,
                            b->pollset_count, "merge_a2b", a->fds, &a->fd_count),
                err_desc);
-  append_error(&error, add_fds_to_pollables(exec_ctx, b->fds, b->fd_count,
+  append_error(&error, add_fds_to_pollsets(exec_ctx, b->fds, b->fd_count,
                                             a->pollsets, a->pollset_count,
                                             "merge_b2a", a->fds, &a->fd_count),
                err_desc);
   if (a->pollset_capacity < a->pollset_count + b->pollset_count) {
     a->pollset_capacity =
         GPR_MAX(2 * a->pollset_capacity, a->pollset_count + b->pollset_count);
-    a->pollsets = (pollable **)gpr_realloc(
+    a->pollsets = (grpc_pollset **)gpr_realloc(
         a->pollsets, a->pollset_capacity * sizeof(*a->pollsets));
   }
   memcpy(a->pollsets + a->pollset_count, b->pollsets,