Tests compile

src/core/iomgr/pollset_posix.c

@@ -140,6 +140,9 @@ void grpc_pollset_init(grpc_pollset *pollset) {
 }
 
 void grpc_pollset_add_fd(grpc_pollset *pollset, grpc_fd *fd) {
+  if (fd->workqueue->wakeup_read_fd != fd) {
+    grpc_pollset_add_fd(pollset, fd->workqueue->wakeup_read_fd);
+  }
   gpr_mu_lock(&pollset->mu);
   pollset->vtable->add_fd(pollset, fd, 1);
 /* the following (enabled only in debug) will reacquire and then release

src/core/surface/channel.c

@@ -370,3 +370,7 @@ grpc_mdstr *grpc_channel_get_message_string(grpc_channel *channel) {
 gpr_uint32 grpc_channel_get_max_message_length(grpc_channel *channel) {
   return channel->max_message_length;
 }
+
+grpc_workqueue *grpc_channel_get_workqueue(grpc_channel *channel) {
+  return channel->workqueue;
+}

test/core/bad_client/bad_client.c

@@ -88,6 +88,7 @@ void grpc_run_bad_client_test(grpc_bad_client_server_side_validator validator,
       gpr_slice_from_copied_buffer(client_payload, client_payload_length);
   gpr_slice_buffer outgoing;
   grpc_iomgr_closure done_write_closure;
+  grpc_workqueue *workqueue = grpc_workqueue_create();
 
   hex = gpr_dump(client_payload, client_payload_length,
                  GPR_DUMP_HEX | GPR_DUMP_ASCII);
@@ -101,7 +102,7 @@ void grpc_run_bad_client_test(grpc_bad_client_server_side_validator validator,
   grpc_init();
 
   /* Create endpoints */
-  sfd = grpc_iomgr_create_endpoint_pair("fixture", 65536);
+  sfd = grpc_iomgr_create_endpoint_pair("fixture", 65536, workqueue);
 
   /* Create server, completion events */
   a.server = grpc_server_create_from_filters(NULL, 0, NULL);
@@ -111,8 +112,9 @@ void grpc_run_bad_client_test(grpc_bad_client_server_side_validator validator,
   a.validator = validator;
   grpc_server_register_completion_queue(a.server, a.cq, NULL);
   grpc_server_start(a.server);
-  transport = grpc_create_chttp2_transport(NULL, sfd.server, mdctx, 0);
-  server_setup_transport(&a, transport, mdctx);
+  transport =
+      grpc_create_chttp2_transport(NULL, sfd.server, mdctx, workqueue, 0);
+  server_setup_transport(&a, transport, mdctx, workqueue);
   grpc_chttp2_transport_start_reading(transport, NULL, 0);
 
   /* Bind everything into the same pollset */
@@ -166,5 +168,6 @@ void grpc_run_bad_client_test(grpc_bad_client_server_side_validator validator,
   grpc_completion_queue_destroy(a.cq);
   gpr_slice_buffer_destroy(&outgoing);
 
+  grpc_workqueue_unref(workqueue);
   grpc_shutdown();
 }

test/core/end2end/fixtures/h2_sockpair+trace.c

@@ -54,6 +54,8 @@
 #include "test/core/util/port.h"
 #include "test/core/util/test_config.h"
 
+grpc_workqueue *g_workqueue;
+
 /* chttp2 transport that is immediately available (used for testing
    connected_channel without a client_channel */
 
@@ -63,7 +65,7 @@ static void server_setup_transport(void *ts, grpc_transport *transport,
   static grpc_channel_filter const *extra_filters[] = {
       &grpc_http_server_filter};
   grpc_server_setup_transport(f->server, transport, extra_filters,
-                              GPR_ARRAY_SIZE(extra_filters), mdctx,
+                              GPR_ARRAY_SIZE(extra_filters), mdctx, g_workqueue,
                               grpc_server_get_channel_args(f->server));
 }
 
@@ -80,8 +82,9 @@ static void client_setup_transport(void *ts, grpc_transport *transport,
                                           &grpc_compress_filter,
                                           &grpc_connected_channel_filter};
   size_t nfilters = sizeof(filters) / sizeof(*filters);
-  grpc_channel *channel = grpc_channel_create_from_filters(
-      "socketpair-target", filters, nfilters, cs->client_args, mdctx, 1);
+  grpc_channel *channel =
+      grpc_channel_create_from_filters("socketpair-target", filters, nfilters,
+                                       cs->client_args, mdctx, g_workqueue, 1);
 
   cs->f->client = channel;
 
@@ -98,7 +101,7 @@ static grpc_end2end_test_fixture chttp2_create_fixture_socketpair(
   f.fixture_data = sfd;
   f.cq = grpc_completion_queue_create(NULL);
 
-  *sfd = grpc_iomgr_create_endpoint_pair("fixture", 65536);
+  *sfd = grpc_iomgr_create_endpoint_pair("fixture", 65536, g_workqueue);
 
   return f;
 }
@@ -111,7 +114,8 @@ static void chttp2_init_client_socketpair(grpc_end2end_test_fixture *f,
   sp_client_setup cs;
   cs.client_args = client_args;
   cs.f = f;
-  transport = grpc_create_chttp2_transport(client_args, sfd->client, mdctx, 1);
+  transport = grpc_create_chttp2_transport(client_args, sfd->client, mdctx,
+                                           g_workqueue, 1);
   client_setup_transport(&cs, transport, mdctx);
   GPR_ASSERT(f->client);
   grpc_chttp2_transport_start_reading(transport, NULL, 0);
@@ -126,7 +130,8 @@ static void chttp2_init_server_socketpair(grpc_end2end_test_fixture *f,
   f->server = grpc_server_create_from_filters(NULL, 0, server_args);
   grpc_server_register_completion_queue(f->server, f->cq, NULL);
   grpc_server_start(f->server);
-  transport = grpc_create_chttp2_transport(server_args, sfd->server, mdctx, 0);
+  transport = grpc_create_chttp2_transport(server_args, sfd->server, mdctx,
+                                           g_workqueue, 0);
   server_setup_transport(f, transport, mdctx);
   grpc_chttp2_transport_start_reading(transport, NULL, 0);
 }
@@ -156,6 +161,7 @@ int main(int argc, char **argv) {
 
   grpc_test_init(argc, argv);
   grpc_init();
+  g_workqueue = grpc_workqueue_create();
 
   GPR_ASSERT(0 == grpc_tracer_set_enabled("also-doesnt-exist", 0));
   GPR_ASSERT(1 == grpc_tracer_set_enabled("http", 1));
@@ -165,6 +171,7 @@ int main(int argc, char **argv) {
     grpc_end2end_tests(configs[i]);
   }
 
+  grpc_workqueue_unref(g_workqueue);
   grpc_shutdown();
 
   return 0;

test/core/end2end/fixtures/h2_sockpair.c

@@ -53,6 +53,8 @@
 #include "test/core/util/port.h"
 #include "test/core/util/test_config.h"
 
+grpc_workqueue *g_workqueue;
+
 /* chttp2 transport that is immediately available (used for testing
    connected_channel without a client_channel */
 
@@ -62,7 +64,7 @@ static void server_setup_transport(void *ts, grpc_transport *transport,
   static grpc_channel_filter const *extra_filters[] = {
       &grpc_http_server_filter};
   grpc_server_setup_transport(f->server, transport, extra_filters,
-                              GPR_ARRAY_SIZE(extra_filters), mdctx,
+                              GPR_ARRAY_SIZE(extra_filters), mdctx, g_workqueue,
                               grpc_server_get_channel_args(f->server));
 }
 
@@ -79,8 +81,9 @@ static void client_setup_transport(void *ts, grpc_transport *transport,
                                           &grpc_compress_filter,
                                           &grpc_connected_channel_filter};
   size_t nfilters = sizeof(filters) / sizeof(*filters);
-  grpc_channel *channel = grpc_channel_create_from_filters(
-      "socketpair-target", filters, nfilters, cs->client_args, mdctx, 1);
+  grpc_channel *channel =
+      grpc_channel_create_from_filters("socketpair-target", filters, nfilters,
+                                       cs->client_args, mdctx, g_workqueue, 1);
 
   cs->f->client = channel;
 
@@ -97,7 +100,7 @@ static grpc_end2end_test_fixture chttp2_create_fixture_socketpair(
   f.fixture_data = sfd;
   f.cq = grpc_completion_queue_create(NULL);
 
-  *sfd = grpc_iomgr_create_endpoint_pair("fixture", 65536);
+  *sfd = grpc_iomgr_create_endpoint_pair("fixture", 65536, g_workqueue);
 
   return f;
 }
@@ -110,7 +113,8 @@ static void chttp2_init_client_socketpair(grpc_end2end_test_fixture *f,
   sp_client_setup cs;
   cs.client_args = client_args;
   cs.f = f;
-  transport = grpc_create_chttp2_transport(client_args, sfd->client, mdctx, 1);
+  transport = grpc_create_chttp2_transport(client_args, sfd->client, mdctx,
+                                           g_workqueue, 1);
   client_setup_transport(&cs, transport, mdctx);
   GPR_ASSERT(f->client);
   grpc_chttp2_transport_start_reading(transport, NULL, 0);
@@ -125,7 +129,8 @@ static void chttp2_init_server_socketpair(grpc_end2end_test_fixture *f,
   f->server = grpc_server_create_from_filters(NULL, 0, server_args);
   grpc_server_register_completion_queue(f->server, f->cq, NULL);
   grpc_server_start(f->server);
-  transport = grpc_create_chttp2_transport(server_args, sfd->server, mdctx, 0);
+  transport = grpc_create_chttp2_transport(server_args, sfd->server, mdctx,
+                                           g_workqueue, 0);
   server_setup_transport(f, transport, mdctx);
   grpc_chttp2_transport_start_reading(transport, NULL, 0);
 }
@@ -146,11 +151,13 @@ int main(int argc, char **argv) {
 
   grpc_test_init(argc, argv);
   grpc_init();
+  g_workqueue = grpc_workqueue_create();
 
   for (i = 0; i < sizeof(configs) / sizeof(*configs); i++) {
     grpc_end2end_tests(configs[i]);
   }
 
+  grpc_workqueue_unref(g_workqueue);
   grpc_shutdown();
 
   return 0;

test/core/end2end/fixtures/h2_sockpair_1byte.c

@@ -53,6 +53,8 @@
 #include "test/core/util/port.h"
 #include "test/core/util/test_config.h"
 
+grpc_workqueue *g_workqueue;
+
 /* chttp2 transport that is immediately available (used for testing
    connected_channel without a client_channel */
 
@@ -62,7 +64,7 @@ static void server_setup_transport(void *ts, grpc_transport *transport,
   static grpc_channel_filter const *extra_filters[] = {
       &grpc_http_server_filter};
   grpc_server_setup_transport(f->server, transport, extra_filters,
-                              GPR_ARRAY_SIZE(extra_filters), mdctx,
+                              GPR_ARRAY_SIZE(extra_filters), mdctx, g_workqueue,
                               grpc_server_get_channel_args(f->server));
 }
 
@@ -79,8 +81,9 @@ static void client_setup_transport(void *ts, grpc_transport *transport,
                                           &grpc_compress_filter,
                                           &grpc_connected_channel_filter};
   size_t nfilters = sizeof(filters) / sizeof(*filters);
-  grpc_channel *channel = grpc_channel_create_from_filters(
-      "socketpair-target", filters, nfilters, cs->client_args, mdctx, 1);
+  grpc_channel *channel =
+      grpc_channel_create_from_filters("socketpair-target", filters, nfilters,
+                                       cs->client_args, mdctx, g_workqueue, 1);
 
   cs->f->client = channel;
 
@@ -97,7 +100,7 @@ static grpc_end2end_test_fixture chttp2_create_fixture_socketpair(
   f.fixture_data = sfd;
   f.cq = grpc_completion_queue_create(NULL);
 
-  *sfd = grpc_iomgr_create_endpoint_pair("fixture", 1);
+  *sfd = grpc_iomgr_create_endpoint_pair("fixture", 1, g_workqueue);
 
   return f;
 }
@@ -110,7 +113,8 @@ static void chttp2_init_client_socketpair(grpc_end2end_test_fixture *f,
   sp_client_setup cs;
   cs.client_args = client_args;
   cs.f = f;
-  transport = grpc_create_chttp2_transport(client_args, sfd->client, mdctx, 1);
+  transport = grpc_create_chttp2_transport(client_args, sfd->client, mdctx,
+                                           g_workqueue, 1);
   client_setup_transport(&cs, transport, mdctx);
   GPR_ASSERT(f->client);
   grpc_chttp2_transport_start_reading(transport, NULL, 0);
@@ -125,7 +129,8 @@ static void chttp2_init_server_socketpair(grpc_end2end_test_fixture *f,
   f->server = grpc_server_create_from_filters(NULL, 0, server_args);
   grpc_server_register_completion_queue(f->server, f->cq, NULL);
   grpc_server_start(f->server);
-  transport = grpc_create_chttp2_transport(server_args, sfd->server, mdctx, 0);
+  transport = grpc_create_chttp2_transport(server_args, sfd->server, mdctx,
+                                           g_workqueue, 0);
   server_setup_transport(f, transport, mdctx);
   grpc_chttp2_transport_start_reading(transport, NULL, 0);
 }
@@ -146,11 +151,13 @@ int main(int argc, char **argv) {
 
   grpc_test_init(argc, argv);
   grpc_init();
+  g_workqueue = grpc_workqueue_create();
 
   for (i = 0; i < sizeof(configs) / sizeof(*configs); i++) {
     grpc_end2end_tests(configs[i]);
   }
 
+  grpc_workqueue_unref(g_workqueue);
   grpc_shutdown();
 
   return 0;

test/core/iomgr/alarm_test.c

@@ -56,7 +56,6 @@ void no_op_cb(void *arg, int success) {}
 typedef struct {
   gpr_cv cv;
   gpr_mu mu;
-  grpc_iomgr_closure *followup_closure;
   int counter;
   int done_success_ctr;
   int done_cancel_ctr;
@@ -65,10 +64,6 @@ typedef struct {
   int success;
 } alarm_arg;
 
-static void followup_cb(void *arg, int success) {
-  gpr_event_set((gpr_event *)arg, arg);
-}
-
 /* Called when an alarm expires. */
 static void alarm_cb(void *arg /* alarm_arg */, int success) {
   alarm_arg *a = arg;
@@ -83,8 +78,7 @@ static void alarm_cb(void *arg /* alarm_arg */, int success) {
   a->success = success;
   gpr_cv_signal(&a->cv);
   gpr_mu_unlock(&a->mu);
-  grpc_iomgr_closure_init(a->followup_closure, followup_cb, &a->fcb_arg);
-  grpc_iomgr_add_callback(a->followup_closure);
+  gpr_event_set((gpr_event *)arg, arg);
 }
 
 /* Test grpc_alarm add and cancel. */
@@ -110,7 +104,6 @@ static void test_grpc_alarm(void) {
   arg.done = 0;
   gpr_mu_init(&arg.mu);
   gpr_cv_init(&arg.cv);
-  arg.followup_closure = gpr_malloc(sizeof(grpc_iomgr_closure));
   gpr_event_init(&arg.fcb_arg);
 
   grpc_alarm_init(&alarm, GRPC_TIMEOUT_MILLIS_TO_DEADLINE(100), alarm_cb, &arg,
@@ -152,7 +145,6 @@ static void test_grpc_alarm(void) {
   }
   gpr_cv_destroy(&arg.cv);
   gpr_mu_destroy(&arg.mu);
-  gpr_free(arg.followup_closure);
 
   arg2.counter = 0;
   arg2.success = SUCCESS_NOT_SET;
@@ -161,7 +153,6 @@ static void test_grpc_alarm(void) {
   arg2.done = 0;
   gpr_mu_init(&arg2.mu);
   gpr_cv_init(&arg2.cv);
-  arg2.followup_closure = gpr_malloc(sizeof(grpc_iomgr_closure));
 
   gpr_event_init(&arg2.fcb_arg);
 
@@ -213,7 +204,6 @@ static void test_grpc_alarm(void) {
   }
   gpr_cv_destroy(&arg2.cv);
   gpr_mu_destroy(&arg2.mu);
-  gpr_free(arg2.followup_closure);
 
   grpc_shutdown();
 }

test/core/iomgr/endpoint_pair_test.c

@@ -43,13 +43,15 @@
 #include "test/core/iomgr/endpoint_tests.h"
 
 static grpc_pollset g_pollset;
+static grpc_workqueue *g_workqueue;
 
 static void clean_up(void) {}
 
 static grpc_endpoint_test_fixture create_fixture_endpoint_pair(
     size_t slice_size) {
   grpc_endpoint_test_fixture f;
-  grpc_endpoint_pair p = grpc_iomgr_create_endpoint_pair("test", slice_size);
+  grpc_endpoint_pair p =
+      grpc_iomgr_create_endpoint_pair("test", slice_size, g_workqueue);
 
   f.client_ep = p.client;
   f.server_ep = p.server;
@@ -69,8 +71,10 @@ int main(int argc, char **argv) {
   grpc_test_init(argc, argv);
   grpc_init();
   grpc_pollset_init(&g_pollset);
+  g_workqueue = grpc_workqueue_create();
   grpc_endpoint_tests(configs[0], &g_pollset);
   grpc_pollset_shutdown(&g_pollset, destroy_pollset, &g_pollset);
+  grpc_workqueue_unref(g_workqueue);
   grpc_shutdown();
 
   return 0;

test/core/iomgr/fd_conservation_posix_test.c

@@ -43,8 +43,11 @@ int main(int argc, char **argv) {
   int i;
   struct rlimit rlim;
   grpc_endpoint_pair p;
+  grpc_workqueue *workqueue;
+
   grpc_test_init(argc, argv);
   grpc_iomgr_init();
+  workqueue = grpc_workqueue_create();
 
   /* set max # of file descriptors to a low value, and
      verify we can create and destroy many more than this number
@@ -53,11 +56,12 @@ int main(int argc, char **argv) {
   GPR_ASSERT(0 == setrlimit(RLIMIT_NOFILE, &rlim));
 
   for (i = 0; i < 100; i++) {
-    p = grpc_iomgr_create_endpoint_pair("test", 1);
+    p = grpc_iomgr_create_endpoint_pair("test", 1, workqueue);
     grpc_endpoint_destroy(p.client);
     grpc_endpoint_destroy(p.server);
   }
 
+  grpc_workqueue_unref(workqueue);
   grpc_iomgr_shutdown();
   return 0;
 }

test/core/iomgr/fd_posix_test.c

@@ -52,6 +52,7 @@
 #include "test/core/util/test_config.h"
 
 static grpc_pollset g_pollset;
+static grpc_workqueue *g_workqueue;
 
 /* buffer size used to send and receive data.
    1024 is the minimal value to set TCP send and receive buffer. */
@@ -207,7 +208,7 @@ static void listen_cb(void *arg, /*=sv_arg*/
   fcntl(fd, F_SETFL, flags | O_NONBLOCK);
   se = gpr_malloc(sizeof(*se));
   se->sv = sv;
-  se->em_fd = grpc_fd_create(fd, "listener");
+  se->em_fd = grpc_fd_create(fd, g_workqueue, "listener");
   grpc_pollset_add_fd(&g_pollset, se->em_fd);
   se->session_read_closure.cb = session_read_cb;
   se->session_read_closure.cb_arg = se;
@@ -236,7 +237,7 @@ static int server_start(server *sv) {
   port = ntohs(sin.sin_port);
   GPR_ASSERT(listen(fd, MAX_NUM_FD) == 0);
 
-  sv->em_fd = grpc_fd_create(fd, "server");
+  sv->em_fd = grpc_fd_create(fd, g_workqueue, "server");
   grpc_pollset_add_fd(&g_pollset, sv->em_fd);
   /* Register to be interested in reading from listen_fd. */
   sv->listen_closure.cb = listen_cb;
@@ -349,7 +350,7 @@ static void client_start(client *cl, int port) {
     }
   }
 
-  cl->em_fd = grpc_fd_create(fd, "client");
+  cl->em_fd = grpc_fd_create(fd, g_workqueue, "client");
   grpc_pollset_add_fd(&g_pollset, cl->em_fd);
 
   client_session_write(cl, 1);
@@ -438,7 +439,7 @@ static void test_grpc_fd_change(void) {
   flags = fcntl(sv[1], F_GETFL, 0);
   GPR_ASSERT(fcntl(sv[1], F_SETFL, flags | O_NONBLOCK) == 0);
 
-  em_fd = grpc_fd_create(sv[0], "test_grpc_fd_change");
+  em_fd = grpc_fd_create(sv[0], g_workqueue, "test_grpc_fd_change");
   grpc_pollset_add_fd(&g_pollset, em_fd);
 
   /* Register the first callback, then make its FD readable */
@@ -490,9 +491,11 @@ int main(int argc, char **argv) {
   grpc_test_init(argc, argv);
   grpc_iomgr_init();
   grpc_pollset_init(&g_pollset);
+  g_workqueue = grpc_workqueue_create();
   test_grpc_fd();
   test_grpc_fd_change();
   grpc_pollset_shutdown(&g_pollset, destroy_pollset, &g_pollset);
+  grpc_workqueue_unref(g_workqueue);
   grpc_iomgr_shutdown();
   return 0;
 }

test/core/iomgr/tcp_client_posix_test.c

@@ -49,6 +49,7 @@
 
 static grpc_pollset_set g_pollset_set;
 static grpc_pollset g_pollset;
+static grpc_workqueue *g_workqueue;
 static int g_connections_complete = 0;
 
 static gpr_timespec test_deadline(void) {
@@ -98,7 +99,7 @@ void test_succeeds(void) {
 
   /* connect to it */
   GPR_ASSERT(getsockname(svr_fd, (struct sockaddr *)&addr, &addr_len) == 0);
-  grpc_tcp_client_connect(must_succeed, NULL, &g_pollset_set,
+  grpc_tcp_client_connect(must_succeed, NULL, &g_pollset_set, g_workqueue,
                           (struct sockaddr *)&addr, addr_len,
                           gpr_inf_future(GPR_CLOCK_REALTIME));
 
@@ -136,7 +137,7 @@ void test_fails(void) {
   gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset));
 
   /* connect to a broken address */
-  grpc_tcp_client_connect(must_fail, NULL, &g_pollset_set,
+  grpc_tcp_client_connect(must_fail, NULL, &g_pollset_set, g_workqueue,
                           (struct sockaddr *)&addr, addr_len,
                           gpr_inf_future(GPR_CLOCK_REALTIME));
 
@@ -195,7 +196,7 @@ void test_times_out(void) {
   connections_complete_before = g_connections_complete;
   gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset));
 
-  grpc_tcp_client_connect(must_fail, NULL, &g_pollset_set,
+  grpc_tcp_client_connect(must_fail, NULL, &g_pollset_set, g_workqueue,
                           (struct sockaddr *)&addr, addr_len, connect_deadline);
 
   /* Make sure the event doesn't trigger early */
@@ -239,12 +240,14 @@ int main(int argc, char **argv) {
   grpc_pollset_set_init(&g_pollset_set);
   grpc_pollset_init(&g_pollset);
   grpc_pollset_set_add_pollset(&g_pollset_set, &g_pollset);
+  g_workqueue = grpc_workqueue_create();
   test_succeeds();
   gpr_log(GPR_ERROR, "End of first test");
   test_fails();
   test_times_out();
   grpc_pollset_set_destroy(&g_pollset_set);
   grpc_pollset_shutdown(&g_pollset, destroy_pollset, &g_pollset);
+  grpc_workqueue_unref(g_workqueue);
   grpc_shutdown();
   return 0;
 }

test/core/iomgr/tcp_posix_test.c

@@ -49,6 +49,7 @@
 #include "test/core/iomgr/endpoint_tests.h"
 
 static grpc_pollset g_pollset;
+static grpc_workqueue *g_workqueue;
 
 /*
    General test notes:
@@ -184,7 +185,8 @@ static void read_test(size_t num_bytes, size_t slice_size) {
 
   create_sockets(sv);
 
-  ep = grpc_tcp_create(grpc_fd_create(sv[1], "read_test"), slice_size, "test");
+  ep = grpc_tcp_create(grpc_fd_create(sv[1], g_workqueue, "read_test"),
+                       slice_size, "test");
   grpc_endpoint_add_to_pollset(ep, &g_pollset);
 
   written_bytes = fill_socket_partial(sv[0], num_bytes);
@@ -233,8 +235,8 @@ static void large_read_test(size_t slice_size) {
 
   create_sockets(sv);
 
-  ep = grpc_tcp_create(grpc_fd_create(sv[1], "large_read_test"), slice_size,
-                       "test");
+  ep = grpc_tcp_create(grpc_fd_create(sv[1], g_workqueue, "large_read_test"),
+                       slice_size, "test");
   grpc_endpoint_add_to_pollset(ep, &g_pollset);
 
   written_bytes = fill_socket(sv[0]);
@@ -383,7 +385,7 @@ static void write_test(size_t num_bytes, size_t slice_size) {
 
   create_sockets(sv);
 
-  ep = grpc_tcp_create(grpc_fd_create(sv[1], "write_test"),
+  ep = grpc_tcp_create(grpc_fd_create(sv[1], g_workqueue, "write_test"),
                        GRPC_TCP_DEFAULT_READ_SLICE_SIZE, "test");
   grpc_endpoint_add_to_pollset(ep, &g_pollset);
 
@@ -454,10 +456,10 @@ static grpc_endpoint_test_fixture create_fixture_tcp_socketpair(
   grpc_endpoint_test_fixture f;
 
   create_sockets(sv);
-  f.client_ep = grpc_tcp_create(grpc_fd_create(sv[0], "fixture:client"),
-                                slice_size, "test");
-  f.server_ep = grpc_tcp_create(grpc_fd_create(sv[1], "fixture:server"),
-                                slice_size, "test");
+  f.client_ep = grpc_tcp_create(
+      grpc_fd_create(sv[0], g_workqueue, "fixture:client"), slice_size, "test");
+  f.server_ep = grpc_tcp_create(
+      grpc_fd_create(sv[1], g_workqueue, "fixture:server"), slice_size, "test");
   grpc_endpoint_add_to_pollset(f.client_ep, &g_pollset);
   grpc_endpoint_add_to_pollset(f.server_ep, &g_pollset);
 
@@ -473,9 +475,11 @@ static void destroy_pollset(void *p) { grpc_pollset_destroy(p); }
 int main(int argc, char **argv) {
   grpc_test_init(argc, argv);
   grpc_init();
+  g_workqueue = grpc_workqueue_create();
   grpc_pollset_init(&g_pollset);
   run_tests();
   grpc_endpoint_tests(configs[0], &g_pollset);
+  grpc_workqueue_unref(g_workqueue);
   grpc_pollset_shutdown(&g_pollset, destroy_pollset, &g_pollset);
   grpc_shutdown();
 

test/core/security/secure_endpoint_test.c

@@ -46,6 +46,7 @@
 #include "src/core/tsi/fake_transport_security.h"
 
 static grpc_pollset g_pollset;
+static grpc_workqueue *g_workqueue;
 
 static grpc_endpoint_test_fixture secure_endpoint_create_fixture_tcp_socketpair(
     size_t slice_size, gpr_slice *leftover_slices, size_t leftover_nslices) {
@@ -54,7 +55,7 @@ static grpc_endpoint_test_fixture secure_endpoint_create_fixture_tcp_socketpair(
   grpc_endpoint_test_fixture f;
   grpc_endpoint_pair tcp;
 
-  tcp = grpc_iomgr_create_endpoint_pair("fixture", slice_size);
+  tcp = grpc_iomgr_create_endpoint_pair("fixture", slice_size, g_workqueue);
   grpc_endpoint_add_to_pollset(tcp.client, &g_pollset);
   grpc_endpoint_add_to_pollset(tcp.server, &g_pollset);
 
@@ -165,9 +166,11 @@ int main(int argc, char **argv) {
   grpc_test_init(argc, argv);
 
   grpc_init();
+  g_workqueue = grpc_workqueue_create();
   grpc_pollset_init(&g_pollset);
   grpc_endpoint_tests(configs[0], &g_pollset);
   test_leftover(configs[1], 1);
+  grpc_workqueue_unref(g_workqueue);
   grpc_pollset_shutdown(&g_pollset, destroy_pollset, &g_pollset);
   grpc_shutdown();