Merge github.com:grpc/grpc into reuse_port

4 files changed, 55 insertions, 42 deletions

diff --git a/test/core/util/mock_endpoint.c b/test/core/util/mock_endpoint.c
index deef68ef59..ed9545e9df 100644
--- a/test/core/util/mock_endpoint.c
+++ b/test/core/util/mock_endpoint.c
@@ -51,7 +51,7 @@ static void me_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
   gpr_mu_lock(&m->mu);
   if (m->read_buffer.count > 0) {
     gpr_slice_buffer_swap(&m->read_buffer, slices);
-    grpc_exec_ctx_push(exec_ctx, cb, GRPC_ERROR_NONE, NULL);
+    grpc_exec_ctx_sched(exec_ctx, cb, GRPC_ERROR_NONE, NULL);
   } else {
     m->on_read = cb;
     m->on_read_out = slices;
@@ -65,7 +65,7 @@ static void me_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
   for (size_t i = 0; i < slices->count; i++) {
     m->on_write(slices->slices[i]);
   }
-  grpc_exec_ctx_push(exec_ctx, cb, GRPC_ERROR_NONE, NULL);
+  grpc_exec_ctx_sched(exec_ctx, cb, GRPC_ERROR_NONE, NULL);
 }
 
 static void me_add_to_pollset(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
@@ -78,8 +78,8 @@ static void me_shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) {
   grpc_mock_endpoint *m = (grpc_mock_endpoint *)ep;
   gpr_mu_lock(&m->mu);
   if (m->on_read) {
-    grpc_exec_ctx_push(exec_ctx, m->on_read,
-                       GRPC_ERROR_CREATE("Endpoint Shutdown"), NULL);
+    grpc_exec_ctx_sched(exec_ctx, m->on_read,
+                        GRPC_ERROR_CREATE("Endpoint Shutdown"), NULL);
     m->on_read = NULL;
   }
   gpr_mu_unlock(&m->mu);
@@ -116,7 +116,7 @@ void grpc_mock_endpoint_put_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
   gpr_mu_lock(&m->mu);
   if (m->on_read != NULL) {
     gpr_slice_buffer_add(m->on_read_out, slice);
-    grpc_exec_ctx_push(exec_ctx, m->on_read, GRPC_ERROR_NONE, NULL);
+    grpc_exec_ctx_sched(exec_ctx, m->on_read, GRPC_ERROR_NONE, NULL);
     m->on_read = NULL;
   } else {
     gpr_slice_buffer_add(&m->read_buffer, slice);
diff --git a/test/core/util/one_corpus_entry_fuzzer.c b/test/core/util/one_corpus_entry_fuzzer.c
index 2a23d006f1..95ae4cf706 100644
--- a/test/core/util/one_corpus_entry_fuzzer.c
+++ b/test/core/util/one_corpus_entry_fuzzer.c
@@ -31,13 +31,20 @@
  *
  */
 
+#include <stdbool.h>
+
 #include <grpc/support/log.h>
 #include "src/core/lib/iomgr/load_file.h"
 
 extern int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size);
 
+extern bool squelch;
+extern bool leak_check;
+
 int main(int argc, char **argv) {
   gpr_slice buffer;
+  squelch = false;
+  leak_check = false;
   GPR_ASSERT(
       GRPC_LOG_IF_ERROR("load_file", grpc_load_file(argv[1], 0, &buffer)));
   LLVMFuzzerTestOneInput(GPR_SLICE_START_PTR(buffer), GPR_SLICE_LENGTH(buffer));
diff --git a/test/core/util/passthru_endpoint.c b/test/core/util/passthru_endpoint.c
index 93753be251..a39f3dd66e 100644
--- a/test/core/util/passthru_endpoint.c
+++ b/test/core/util/passthru_endpoint.c
@@ -59,11 +59,11 @@ static void me_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
   half *m = (half *)ep;
   gpr_mu_lock(&m->parent->mu);
   if (m->parent->shutdown) {
-    grpc_exec_ctx_push(exec_ctx, cb, GRPC_ERROR_CREATE("Already shutdown"),
-                       NULL);
+    grpc_exec_ctx_sched(exec_ctx, cb, GRPC_ERROR_CREATE("Already shutdown"),
+                        NULL);
   } else if (m->read_buffer.count > 0) {
     gpr_slice_buffer_swap(&m->read_buffer, slices);
-    grpc_exec_ctx_push(exec_ctx, cb, GRPC_ERROR_NONE, NULL);
+    grpc_exec_ctx_sched(exec_ctx, cb, GRPC_ERROR_NONE, NULL);
   } else {
     m->on_read = cb;
     m->on_read_out = slices;
@@ -87,7 +87,7 @@ static void me_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
     for (size_t i = 0; i < slices->count; i++) {
       gpr_slice_buffer_add(m->on_read_out, gpr_slice_ref(slices->slices[i]));
     }
-    grpc_exec_ctx_push(exec_ctx, m->on_read, GRPC_ERROR_NONE, NULL);
+    grpc_exec_ctx_sched(exec_ctx, m->on_read, GRPC_ERROR_NONE, NULL);
     m->on_read = NULL;
   } else {
     for (size_t i = 0; i < slices->count; i++) {
@@ -95,7 +95,7 @@ static void me_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
     }
   }
   gpr_mu_unlock(&m->parent->mu);
-  grpc_exec_ctx_push(exec_ctx, cb, error, NULL);
+  grpc_exec_ctx_sched(exec_ctx, cb, error, NULL);
 }
 
 static void me_add_to_pollset(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
@@ -109,14 +109,14 @@ static void me_shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) {
   gpr_mu_lock(&m->parent->mu);
   m->parent->shutdown = true;
   if (m->on_read) {
-    grpc_exec_ctx_push(exec_ctx, m->on_read, GRPC_ERROR_CREATE("Shutdown"),
-                       NULL);
+    grpc_exec_ctx_sched(exec_ctx, m->on_read, GRPC_ERROR_CREATE("Shutdown"),
+                        NULL);
     m->on_read = NULL;
   }
   m = other_half(m);
   if (m->on_read) {
-    grpc_exec_ctx_push(exec_ctx, m->on_read, GRPC_ERROR_CREATE("Shutdown"),
-                       NULL);
+    grpc_exec_ctx_sched(exec_ctx, m->on_read, GRPC_ERROR_CREATE("Shutdown"),
+                        NULL);
     m->on_read = NULL;
   }
   gpr_mu_unlock(&m->parent->mu);
diff --git a/test/core/util/port_server_client.c b/test/core/util/port_server_client.c
index 167efb5369..a5c8c49650 100644
--- a/test/core/util/port_server_client.c
+++ b/test/core/util/port_server_client.c
@@ -51,14 +51,15 @@
 
 typedef struct freereq {
   gpr_mu *mu;
-  grpc_pollset *pollset;
+  grpc_polling_entity pops;
   int done;
 } freereq;
 
-static void destroy_pollset_and_shutdown(grpc_exec_ctx *exec_ctx, void *p,
-                                         grpc_error *error) {
-  grpc_pollset_destroy(p);
-  gpr_free(p);
+static void destroy_pops_and_shutdown(grpc_exec_ctx *exec_ctx, void *p,
+                                      grpc_error *error) {
+  grpc_pollset *pollset = grpc_polling_entity_pollset(p);
+  grpc_pollset_destroy(pollset);
+  gpr_free(pollset);
   grpc_shutdown();
 }
 
@@ -67,7 +68,9 @@ static void freed_port_from_server(grpc_exec_ctx *exec_ctx, void *arg,
   freereq *pr = arg;
   gpr_mu_lock(pr->mu);
   pr->done = 1;
-  GRPC_LOG_IF_ERROR("pollset_kick", grpc_pollset_kick(pr->pollset, NULL));
+  GRPC_LOG_IF_ERROR(
+      "pollset_kick",
+      grpc_pollset_kick(grpc_polling_entity_pollset(&pr->pops), NULL));
   gpr_mu_unlock(pr->mu);
 }
 
@@ -86,17 +89,17 @@ void grpc_free_port_using_server(char *server, int port) {
   memset(&req, 0, sizeof(req));
   memset(&rsp, 0, sizeof(rsp));
 
-  pr.pollset = gpr_malloc(grpc_pollset_size());
-  grpc_pollset_init(pr.pollset, &pr.mu);
-  shutdown_closure =
-      grpc_closure_create(destroy_pollset_and_shutdown, pr.pollset);
+  grpc_pollset *pollset = gpr_malloc(grpc_pollset_size());
+  grpc_pollset_init(pollset, &pr.mu);
+  pr.pops = grpc_polling_entity_create_from_pollset(pollset);
+  shutdown_closure = grpc_closure_create(destroy_pops_and_shutdown, &pr.pops);
 
   req.host = server;
   gpr_asprintf(&path, "/drop/%d", port);
   req.http.path = path;
 
   grpc_httpcli_context_init(&context);
-  grpc_httpcli_get(&exec_ctx, &context, pr.pollset, &req,
+  grpc_httpcli_get(&exec_ctx, &context, &pr.pops, &req,
                    GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10),
                    grpc_closure_create(freed_port_from_server, &pr), &rsp);
   gpr_mu_lock(pr.mu);
@@ -104,8 +107,8 @@ void grpc_free_port_using_server(char *server, int port) {
     grpc_pollset_worker *worker = NULL;
     if (!GRPC_LOG_IF_ERROR(
             "pollset_work",
-            grpc_pollset_work(&exec_ctx, pr.pollset, &worker,
-                              gpr_now(GPR_CLOCK_MONOTONIC),
+            grpc_pollset_work(&exec_ctx, grpc_polling_entity_pollset(&pr.pops),
+                              &worker, gpr_now(GPR_CLOCK_MONOTONIC),
                               GRPC_TIMEOUT_SECONDS_TO_DEADLINE(1)))) {
       pr.done = 1;
     }
@@ -114,7 +117,8 @@ void grpc_free_port_using_server(char *server, int port) {
 
   grpc_httpcli_context_destroy(&context);
   grpc_exec_ctx_finish(&exec_ctx);
-  grpc_pollset_shutdown(&exec_ctx, pr.pollset, shutdown_closure);
+  grpc_pollset_shutdown(&exec_ctx, grpc_polling_entity_pollset(&pr.pops),
+                        shutdown_closure);
   grpc_exec_ctx_finish(&exec_ctx);
   gpr_free(path);
   grpc_http_response_destroy(&rsp);
@@ -122,7 +126,7 @@ void grpc_free_port_using_server(char *server, int port) {
 
 typedef struct portreq {
   gpr_mu *mu;
-  grpc_pollset *pollset;
+  grpc_polling_entity pops;
   int port;
   int retries;
   char *server;
@@ -163,7 +167,7 @@ static void got_port_from_server(grpc_exec_ctx *exec_ctx, void *arg,
     req.http.path = "/get";
     grpc_http_response_destroy(&pr->response);
     memset(&pr->response, 0, sizeof(pr->response));
-    grpc_httpcli_get(exec_ctx, pr->ctx, pr->pollset, &req,
+    grpc_httpcli_get(exec_ctx, pr->ctx, &pr->pops, &req,
                      GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10),
                      grpc_closure_create(got_port_from_server, pr),
                      &pr->response);
@@ -178,7 +182,9 @@ static void got_port_from_server(grpc_exec_ctx *exec_ctx, void *arg,
   GPR_ASSERT(port > 1024);
   gpr_mu_lock(pr->mu);
   pr->port = port;
-  GRPC_LOG_IF_ERROR("pollset_kick", grpc_pollset_kick(pr->pollset, NULL));
+  GRPC_LOG_IF_ERROR(
+      "pollset_kick",
+      grpc_pollset_kick(grpc_polling_entity_pollset(&pr->pops), NULL));
   gpr_mu_unlock(pr->mu);
 }
 
@@ -193,10 +199,10 @@ int grpc_pick_port_using_server(char *server) {
 
   memset(&pr, 0, sizeof(pr));
   memset(&req, 0, sizeof(req));
-  pr.pollset = gpr_malloc(grpc_pollset_size());
-  grpc_pollset_init(pr.pollset, &pr.mu);
-  shutdown_closure =
-      grpc_closure_create(destroy_pollset_and_shutdown, pr.pollset);
+  grpc_pollset *pollset = gpr_malloc(grpc_pollset_size());
+  grpc_pollset_init(pollset, &pr.mu);
+  pr.pops = grpc_polling_entity_create_from_pollset(pollset);
+  shutdown_closure = grpc_closure_create(destroy_pops_and_shutdown, &pr.pops);
   pr.port = -1;
   pr.server = server;
   pr.ctx = &context;
@@ -205,18 +211,17 @@ int grpc_pick_port_using_server(char *server) {
   req.http.path = "/get";
 
   grpc_httpcli_context_init(&context);
-  grpc_httpcli_get(&exec_ctx, &context, pr.pollset, &req,
-                   GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10),
-                   grpc_closure_create(got_port_from_server, &pr),
-                   &pr.response);
+  grpc_httpcli_get(
+      &exec_ctx, &context, &pr.pops, &req, GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10),
+      grpc_closure_create(got_port_from_server, &pr), &pr.response);
   grpc_exec_ctx_finish(&exec_ctx);
   gpr_mu_lock(pr.mu);
   while (pr.port == -1) {
     grpc_pollset_worker *worker = NULL;
     if (!GRPC_LOG_IF_ERROR(
             "pollset_work",
-            grpc_pollset_work(&exec_ctx, pr.pollset, &worker,
-                              gpr_now(GPR_CLOCK_MONOTONIC),
+            grpc_pollset_work(&exec_ctx, grpc_polling_entity_pollset(&pr.pops),
+                              &worker, gpr_now(GPR_CLOCK_MONOTONIC),
                               GRPC_TIMEOUT_SECONDS_TO_DEADLINE(1)))) {
       pr.port = 0;
     }
@@ -225,7 +230,8 @@ int grpc_pick_port_using_server(char *server) {
 
   grpc_http_response_destroy(&pr.response);
   grpc_httpcli_context_destroy(&context);
-  grpc_pollset_shutdown(&exec_ctx, pr.pollset, shutdown_closure);
+  grpc_pollset_shutdown(&exec_ctx, grpc_polling_entity_pollset(&pr.pops),
+                        shutdown_closure);
   grpc_exec_ctx_finish(&exec_ctx);
 
   return pr.port;