Merge branch 'bye-bye-completion-queue-pie' into we-dont-need-no-backup
diff --git a/BUILD b/BUILD
index f7136e4..cfaa6bd 100644
--- a/BUILD
+++ b/BUILD
@@ -168,6 +168,8 @@
"src/core/iomgr/pollset_kick_posix.h",
"src/core/iomgr/pollset_kick_windows.h",
"src/core/iomgr/pollset_posix.h",
+ "src/core/iomgr/pollset_set_posix.h",
+ "src/core/iomgr/pollset_set_windows.h",
"src/core/iomgr/pollset_windows.h",
"src/core/iomgr/resolve_address.h",
"src/core/iomgr/sockaddr.h",
@@ -275,6 +277,8 @@
"src/core/iomgr/pollset_multipoller_with_epoll.c",
"src/core/iomgr/pollset_multipoller_with_poll_posix.c",
"src/core/iomgr/pollset_posix.c",
+ "src/core/iomgr/pollset_set_posix.c",
+ "src/core/iomgr/pollset_set_windows.c",
"src/core/iomgr/pollset_windows.c",
"src/core/iomgr/resolve_address_posix.c",
"src/core/iomgr/resolve_address_windows.c",
@@ -396,6 +400,8 @@
"src/core/iomgr/pollset_kick_posix.h",
"src/core/iomgr/pollset_kick_windows.h",
"src/core/iomgr/pollset_posix.h",
+ "src/core/iomgr/pollset_set_posix.h",
+ "src/core/iomgr/pollset_set_windows.h",
"src/core/iomgr/pollset_windows.h",
"src/core/iomgr/resolve_address.h",
"src/core/iomgr/sockaddr.h",
@@ -483,6 +489,8 @@
"src/core/iomgr/pollset_multipoller_with_epoll.c",
"src/core/iomgr/pollset_multipoller_with_poll_posix.c",
"src/core/iomgr/pollset_posix.c",
+ "src/core/iomgr/pollset_set_posix.c",
+ "src/core/iomgr/pollset_set_windows.c",
"src/core/iomgr/pollset_windows.c",
"src/core/iomgr/resolve_address_posix.c",
"src/core/iomgr/resolve_address_windows.c",
diff --git a/Makefile b/Makefile
index 93b5f8b..cd5858c 100644
--- a/Makefile
+++ b/Makefile
@@ -2459,6 +2459,8 @@
src/core/iomgr/pollset_multipoller_with_epoll.c \
src/core/iomgr/pollset_multipoller_with_poll_posix.c \
src/core/iomgr/pollset_posix.c \
+ src/core/iomgr/pollset_set_posix.c \
+ src/core/iomgr/pollset_set_windows.c \
src/core/iomgr/pollset_windows.c \
src/core/iomgr/resolve_address_posix.c \
src/core/iomgr/resolve_address_windows.c \
@@ -2708,6 +2710,8 @@
src/core/iomgr/pollset_multipoller_with_epoll.c \
src/core/iomgr/pollset_multipoller_with_poll_posix.c \
src/core/iomgr/pollset_posix.c \
+ src/core/iomgr/pollset_set_posix.c \
+ src/core/iomgr/pollset_set_windows.c \
src/core/iomgr/pollset_windows.c \
src/core/iomgr/resolve_address_posix.c \
src/core/iomgr/resolve_address_windows.c \
diff --git a/build.json b/build.json
index 28a5b4b..8d256e9 100644
--- a/build.json
+++ b/build.json
@@ -120,6 +120,8 @@
"src/core/iomgr/pollset_kick_posix.h",
"src/core/iomgr/pollset_kick_windows.h",
"src/core/iomgr/pollset_posix.h",
+ "src/core/iomgr/pollset_set_posix.h",
+ "src/core/iomgr/pollset_set_windows.h",
"src/core/iomgr/pollset_windows.h",
"src/core/iomgr/resolve_address.h",
"src/core/iomgr/sockaddr.h",
@@ -208,6 +210,8 @@
"src/core/iomgr/pollset_multipoller_with_epoll.c",
"src/core/iomgr/pollset_multipoller_with_poll_posix.c",
"src/core/iomgr/pollset_posix.c",
+ "src/core/iomgr/pollset_set_posix.c",
+ "src/core/iomgr/pollset_set_windows.c",
"src/core/iomgr/pollset_windows.c",
"src/core/iomgr/resolve_address_posix.c",
"src/core/iomgr/resolve_address_windows.c",
diff --git a/src/core/channel/client_channel.c b/src/core/channel/client_channel.c
index 78f8d06..26b6841 100644
--- a/src/core/channel/client_channel.c
+++ b/src/core/channel/client_channel.c
@@ -39,6 +39,7 @@
#include "src/core/channel/child_channel.h"
#include "src/core/channel/connected_channel.h"
#include "src/core/iomgr/iomgr.h"
+#include "src/core/iomgr/pollset_set.h"
#include "src/core/support/string.h"
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
@@ -133,6 +134,7 @@
for (i = 0, new_count = 0; i < chand->waiting_child_count; i++) {
if (chand->waiting_children[i] == calld) continue;
chand->waiting_children[new_count++] = chand->waiting_children[i];
+ abort(); /* what to do about waiting_pollsets */
}
GPR_ASSERT(new_count == chand->waiting_child_count - 1 ||
new_count == chand->waiting_child_count);
@@ -227,6 +229,7 @@
if (initiate_transport_setup) {
grpc_transport_setup_initiate(chand->transport_setup);
}
+ grpc_transport_setup_add_interested_party(chand->transport_setup, op->bind_pollset);
}
}
break;
diff --git a/src/core/channel/client_setup.c b/src/core/channel/client_setup.c
index 6d892d6..d3d31a9 100644
--- a/src/core/channel/client_setup.c
+++ b/src/core/channel/client_setup.c
@@ -61,6 +61,7 @@
struct grpc_client_setup_request {
/* pointer back to the setup object */
grpc_client_setup *setup;
+ grpc_pollset_set interested_parties;
gpr_timespec deadline;
};
@@ -68,6 +69,11 @@
return r->deadline;
}
+grpc_pollset_set *grpc_client_setup_get_interested_parties(
+ grpc_client_setup_request *r) {
+ return &r->interested_parties;
+}
+
static void destroy_setup(grpc_client_setup *s) {
gpr_mu_destroy(&s->mu);
gpr_cv_destroy(&s->cv);
@@ -76,6 +82,11 @@
gpr_free(s);
}
+static void destroy_request(grpc_client_setup_request *r) {
+ grpc_pollset_set_destroy(&r->interested_parties);
+ gpr_free(r);
+}
+
/* initiate handshaking */
static void setup_initiate(grpc_transport_setup *sp) {
grpc_client_setup *s = (grpc_client_setup *)sp;
@@ -83,6 +94,7 @@
int in_alarm = 0;
r->setup = s;
+ grpc_pollset_set_init(&r->interested_parties);
/* TODO(klempner): Actually set a deadline */
r->deadline = gpr_inf_future;
@@ -104,10 +116,24 @@
if (!in_alarm) {
s->initiate(s->user_data, r);
} else {
- gpr_free(r);
+ destroy_request(r);
}
}
+static void setup_add_interested_party(grpc_transport_setup *sp, grpc_pollset *pollset) {
+ grpc_client_setup *s = (grpc_client_setup *)sp;
+
+ gpr_mu_lock(&s->mu);
+ if (!s->active_request) {
+ gpr_mu_unlock(&s->mu);
+ return;
+ }
+
+ grpc_pollset_set_add_pollset(&s->active_request->interested_parties, pollset);
+
+ gpr_mu_unlock(&s->mu);
+}
+
/* cancel handshaking: cancel all requests, and shutdown (the caller promises
not to initiate again) */
static void setup_cancel(grpc_transport_setup *sp) {
@@ -157,6 +183,7 @@
/* vtable for transport setup */
static const grpc_transport_setup_vtable setup_vtable = {setup_initiate,
+ setup_add_interested_party,
setup_cancel};
void grpc_client_setup_create_and_attach(
@@ -196,26 +223,24 @@
}
static void backoff_alarm_done(void *arg /* grpc_client_setup */, int success) {
- grpc_client_setup *s = arg;
- grpc_client_setup_request *r = gpr_malloc(sizeof(grpc_client_setup_request));
- r->setup = s;
- /* TODO(klempner): Set this to something useful */
- r->deadline = gpr_inf_future;
+ grpc_client_setup_request *r = arg;
+ grpc_client_setup *s = r->setup;
/* Handle status cancelled? */
gpr_mu_lock(&s->mu);
- s->active_request = r;
s->in_alarm = 0;
- if (!success) {
+ if (s->active_request != NULL || !success) {
if (0 == --s->refs) {
gpr_mu_unlock(&s->mu);
destroy_setup(s);
- gpr_free(r);
+ destroy_request(r);
return;
} else {
gpr_mu_unlock(&s->mu);
+ destroy_request(r);
return;
}
}
+ s->active_request = r;
gpr_mu_unlock(&s->mu);
s->initiate(s->user_data, r);
}
@@ -234,12 +259,10 @@
if (!retry && 0 == --s->refs) {
gpr_mu_unlock(&s->mu);
destroy_setup(s);
- gpr_free(r);
+ destroy_request(r);
return;
}
- gpr_free(r);
-
if (retry) {
/* TODO(klempner): Replace these values with further consideration. 2x is
probably too aggressive of a backoff. */
@@ -248,7 +271,7 @@
gpr_timespec deadline = gpr_time_add(s->current_backoff_interval, now);
GPR_ASSERT(!s->in_alarm);
s->in_alarm = 1;
- grpc_alarm_init(&s->backoff_alarm, deadline, backoff_alarm_done, s, now);
+ grpc_alarm_init(&s->backoff_alarm, deadline, backoff_alarm_done, r, now);
s->current_backoff_interval =
gpr_time_add(s->current_backoff_interval, s->current_backoff_interval);
if (gpr_time_cmp(s->current_backoff_interval, max_backoff) > 0) {
diff --git a/src/core/channel/client_setup.h b/src/core/channel/client_setup.h
index 70137e1..2486da5 100644
--- a/src/core/channel/client_setup.h
+++ b/src/core/channel/client_setup.h
@@ -67,6 +67,8 @@
/* Get the deadline for a request passed in to initiate. Implementations should
make a best effort to honor this deadline. */
gpr_timespec grpc_client_setup_request_deadline(grpc_client_setup_request *r);
+grpc_pollset_set *grpc_client_setup_get_interested_parties(
+ grpc_client_setup_request *r);
grpc_mdctx *grpc_client_setup_get_mdctx(grpc_client_setup_request *r);
diff --git a/src/core/httpcli/httpcli.c b/src/core/httpcli/httpcli.c
index fe7ea6a..54cc992 100644
--- a/src/core/httpcli/httpcli.c
+++ b/src/core/httpcli/httpcli.c
@@ -59,6 +59,7 @@
int use_ssl;
grpc_httpcli_response_cb on_response;
void *user_data;
+ grpc_pollset_set *interested_parties;
} internal_request;
static grpc_httpcli_get_override g_get_override = NULL;
@@ -206,7 +207,7 @@
return;
}
addr = &req->addresses->addrs[req->next_address++];
- grpc_tcp_client_connect(on_connected, req, (struct sockaddr *)&addr->addr,
+ grpc_tcp_client_connect(on_connected, req, req->interested_parties, (struct sockaddr *)&addr->addr,
addr->len, req->deadline);
}
@@ -224,6 +225,7 @@
void grpc_httpcli_get(const grpc_httpcli_request *request,
gpr_timespec deadline,
+ grpc_pollset_set *interested_parties,
grpc_httpcli_response_cb on_response, void *user_data) {
internal_request *req;
if (g_get_override &&
@@ -238,6 +240,7 @@
req->user_data = user_data;
req->deadline = deadline;
req->use_ssl = request->use_ssl;
+ req->interested_parties = interested_parties;
if (req->use_ssl) {
req->host = gpr_strdup(request->host);
}
@@ -249,6 +252,7 @@
void grpc_httpcli_post(const grpc_httpcli_request *request,
const char *body_bytes, size_t body_size,
gpr_timespec deadline,
+ grpc_pollset_set *interested_parties,
grpc_httpcli_response_cb on_response, void *user_data) {
internal_request *req;
if (g_post_override && g_post_override(request, body_bytes, body_size,
@@ -264,6 +268,7 @@
req->user_data = user_data;
req->deadline = deadline;
req->use_ssl = request->use_ssl;
+ req->interested_parties = interested_parties;
if (req->use_ssl) {
req->host = gpr_strdup(request->host);
}
diff --git a/src/core/httpcli/httpcli.h b/src/core/httpcli/httpcli.h
index 255c5ed..fec435b 100644
--- a/src/core/httpcli/httpcli.h
+++ b/src/core/httpcli/httpcli.h
@@ -38,6 +38,8 @@
#include <grpc/support/time.h>
+#include "src/core/iomgr/pollset_set.h"
+
/* User agent this library reports */
#define GRPC_HTTPCLI_USER_AGENT "grpc-httpcli/0.0"
/* Maximum length of a header string of the form 'Key: Value\r\n' */
@@ -90,6 +92,7 @@
supplied pointer to pass to said call) */
void grpc_httpcli_get(const grpc_httpcli_request *request,
gpr_timespec deadline,
+ grpc_pollset_set *interested_parties,
grpc_httpcli_response_cb on_response, void *user_data);
/* Asynchronously perform a HTTP POST.
@@ -98,6 +101,7 @@
void grpc_httpcli_post(const grpc_httpcli_request *request,
const char *body_bytes, size_t body_size,
gpr_timespec deadline,
+ grpc_pollset_set *interested_parties,
grpc_httpcli_response_cb on_response, void *user_data);
/* override functions return 1 if they handled the request, 0 otherwise */
diff --git a/src/core/iomgr/fd_posix.c b/src/core/iomgr/fd_posix.c
index 9c8133d..cafe52b 100644
--- a/src/core/iomgr/fd_posix.c
+++ b/src/core/iomgr/fd_posix.c
@@ -139,7 +139,6 @@
grpc_fd *grpc_fd_create(int fd) {
grpc_fd *r = alloc_fd(fd);
grpc_iomgr_ref();
- grpc_pollset_add_fd(grpc_backup_pollset(), r);
return r;
}
diff --git a/src/core/iomgr/pollset.h b/src/core/iomgr/pollset.h
index 067af87..334e0eb 100644
--- a/src/core/iomgr/pollset.h
+++ b/src/core/iomgr/pollset.h
@@ -52,7 +52,6 @@
#include "src/core/iomgr/pollset_windows.h"
#endif
-
void grpc_pollset_init(grpc_pollset *pollset);
void grpc_pollset_shutdown(grpc_pollset *pollset,
void (*shutdown_done)(void *arg),
diff --git a/src/core/iomgr/pollset_posix.c b/src/core/iomgr/pollset_posix.c
index 4d1bcad..6f89a7a 100644
--- a/src/core/iomgr/pollset_posix.c
+++ b/src/core/iomgr/pollset_posix.c
@@ -54,31 +54,8 @@
#include <grpc/support/tls.h>
#include <grpc/support/useful.h>
-static grpc_pollset g_backup_pollset;
-static int g_shutdown_backup_poller;
-static gpr_event g_backup_poller_done;
-static gpr_event g_backup_pollset_shutdown_done;
-
GPR_TLS_DECL(g_current_thread_poller);
-static void backup_poller(void *p) {
- gpr_timespec delta = gpr_time_from_millis(100);
- gpr_timespec last_poll = gpr_now();
-
- gpr_mu_lock(&g_backup_pollset.mu);
- while (g_shutdown_backup_poller == 0) {
- gpr_timespec next_poll = gpr_time_add(last_poll, delta);
- grpc_pollset_work(&g_backup_pollset, gpr_time_add(gpr_now(), gpr_time_from_seconds(1)));
- gpr_mu_unlock(&g_backup_pollset.mu);
- gpr_sleep_until(next_poll);
- gpr_mu_lock(&g_backup_pollset.mu);
- last_poll = next_poll;
- }
- gpr_mu_unlock(&g_backup_pollset.mu);
-
- gpr_event_set(&g_backup_poller_done, (void *)1);
-}
-
void grpc_pollset_kick(grpc_pollset *p) {
if (gpr_tls_get(&g_current_thread_poller) != (gpr_intptr)p && p->counter) {
p->vtable->kick(p);
@@ -99,44 +76,14 @@
/* global state management */
-grpc_pollset *grpc_backup_pollset(void) { return &g_backup_pollset; }
-
void grpc_pollset_global_init(void) {
- gpr_thd_id id;
-
gpr_tls_init(&g_current_thread_poller);
/* Initialize kick fd state */
grpc_pollset_kick_global_init();
-
- /* initialize the backup pollset */
- grpc_pollset_init(&g_backup_pollset);
-
- /* start the backup poller thread */
- g_shutdown_backup_poller = 0;
- gpr_event_init(&g_backup_poller_done);
- gpr_event_init(&g_backup_pollset_shutdown_done);
- gpr_thd_new(&id, backup_poller, NULL, NULL);
-}
-
-static void on_backup_pollset_shutdown_done(void *arg) {
- gpr_event_set(&g_backup_pollset_shutdown_done, (void *)1);
}
void grpc_pollset_global_shutdown(void) {
- /* terminate the backup poller thread */
- gpr_mu_lock(&g_backup_pollset.mu);
- g_shutdown_backup_poller = 1;
- gpr_mu_unlock(&g_backup_pollset.mu);
- gpr_event_wait(&g_backup_poller_done, gpr_inf_future);
-
- grpc_pollset_shutdown(&g_backup_pollset, on_backup_pollset_shutdown_done,
- NULL);
- gpr_event_wait(&g_backup_pollset_shutdown_done, gpr_inf_future);
-
- /* destroy the backup pollset */
- grpc_pollset_destroy(&g_backup_pollset);
-
/* destroy the kick pipes */
grpc_pollset_kick_global_destroy();
diff --git a/src/core/iomgr/pollset_posix.h b/src/core/iomgr/pollset_posix.h
index da843f7..1a92e15 100644
--- a/src/core/iomgr/pollset_posix.h
+++ b/src/core/iomgr/pollset_posix.h
@@ -94,12 +94,6 @@
/* Call after polling has been kicked to leave the kicked state */
void grpc_kick_drain(grpc_pollset *p);
-/* All fds get added to a backup pollset to ensure that progress is made
- regardless of applications listening to events. Relying on this is slow
- however (the backup pollset only listens every 100ms or so) - so it's not
- to be relied on. */
-grpc_pollset *grpc_backup_pollset(void);
-
/* turn a pollset into a multipoller: platform specific */
void grpc_platform_become_multipoller(grpc_pollset *pollset,
struct grpc_fd **fds, size_t fd_count);
diff --git a/src/core/iomgr/pollset_set.h b/src/core/iomgr/pollset_set.h
new file mode 100644
index 0000000..70a8060
--- /dev/null
+++ b/src/core/iomgr/pollset_set.h
@@ -0,0 +1,57 @@
+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ * * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ * * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#ifndef GRPC_INTERNAL_CORE_IOMGR_POLLSET_SET_H
+#define GRPC_INTERNAL_CORE_IOMGR_POLLSET_SET_H
+
+#include "src/core/iomgr/pollset.h"
+
+/* A grpc_pollset_set is a set of pollsets that are interested in an
+ action. Adding a pollset to a pollset_set automatically adds any
+ fd's (etc) that have been registered with the set_set with that pollset.
+ Registering fd's automatically iterates all current pollsets. */
+
+#ifdef GPR_POSIX_SOCKET
+#include "src/core/iomgr/pollset_set_posix.h"
+#endif
+
+#ifdef GPR_WIN32
+#include "src/core/iomgr/pollset_set_windows.h"
+#endif
+
+void grpc_pollset_set_init(grpc_pollset_set *pollset_set);
+void grpc_pollset_set_destroy(grpc_pollset_set *pollset_set);
+void grpc_pollset_set_add_pollset(grpc_pollset_set *pollset_set, grpc_pollset *pollset);
+void grpc_pollset_set_del_pollset(grpc_pollset_set *pollset_set, grpc_pollset *pollset);
+
+#endif /* GRPC_INTERNAL_CORE_IOMGR_POLLSET_H */
diff --git a/src/core/iomgr/pollset_set_posix.c b/src/core/iomgr/pollset_set_posix.c
new file mode 100644
index 0000000..f54237f
--- /dev/null
+++ b/src/core/iomgr/pollset_set_posix.c
@@ -0,0 +1,111 @@
+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ * * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ * * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#include <grpc/support/port_platform.h>
+
+#ifdef GPR_POSIX_SOCKET
+
+#include <stdlib.h>
+#include <string.h>
+
+#include <grpc/support/alloc.h>
+#include <grpc/support/useful.h>
+
+#include "src/core/iomgr/pollset_set.h"
+
+void grpc_pollset_set_init(grpc_pollset_set *pollset_set) {
+ memset(pollset_set, 0, sizeof(*pollset_set));
+ gpr_mu_init(&pollset_set->mu);
+}
+
+void grpc_pollset_set_destroy(grpc_pollset_set *pollset_set) {
+ gpr_mu_destroy(&pollset_set->mu);
+ gpr_free(pollset_set->pollsets);
+ gpr_free(pollset_set->fds);
+}
+
+void grpc_pollset_set_add_pollset(grpc_pollset_set *pollset_set, grpc_pollset *pollset) {
+ size_t i;
+ gpr_mu_lock(&pollset_set->mu);
+ if (pollset_set->pollset_count == pollset_set->pollset_capacity) {
+ pollset_set->pollset_capacity = GPR_MAX(8, 2 * pollset_set->pollset_capacity);
+ pollset_set->pollsets = gpr_realloc(pollset_set->pollsets, pollset_set->pollset_capacity * sizeof(*pollset_set->pollsets));
+ }
+ pollset_set->pollsets[pollset_set->pollset_count++] = pollset;
+ for (i = 0; i < pollset_set->fd_count; i++) {
+ grpc_pollset_add_fd(pollset, pollset_set->fds[i]);
+ }
+ gpr_mu_unlock(&pollset_set->mu);
+}
+
+void grpc_pollset_set_del_pollset(grpc_pollset_set *pollset_set, grpc_pollset *pollset) {
+ size_t i;
+ gpr_mu_lock(&pollset_set->mu);
+ for (i = 0; i < pollset_set->pollset_count; i++) {
+ if (pollset_set->pollsets[i] == pollset) {
+ pollset_set->pollset_count--;
+ GPR_SWAP(grpc_pollset *, pollset_set->pollsets[i], pollset_set->pollsets[pollset_set->pollset_count]);
+ break;
+ }
+ }
+ gpr_mu_unlock(&pollset_set->mu);
+}
+
+void grpc_pollset_set_add_fd(grpc_pollset_set *pollset_set, grpc_fd *fd) {
+ size_t i;
+ gpr_mu_lock(&pollset_set->mu);
+ if (pollset_set->fd_count == pollset_set->fd_capacity) {
+ pollset_set->fd_capacity = GPR_MAX(8, 2 * pollset_set->fd_capacity);
+ pollset_set->fds = gpr_realloc(pollset_set->fds, pollset_set->fd_capacity * sizeof(*pollset_set->fds));
+ }
+ pollset_set->fds[pollset_set->fd_count++] = fd;
+ for (i = 0; i < pollset_set->pollset_count; i++) {
+ grpc_pollset_add_fd(pollset_set->pollsets[i], fd);
+ }
+ gpr_mu_unlock(&pollset_set->mu);
+}
+
+void grpc_pollset_set_del_fd(grpc_pollset_set *pollset_set, grpc_fd *fd) {
+ size_t i;
+ gpr_mu_lock(&pollset_set->mu);
+ for (i = 0; i < pollset_set->fd_count; i++) {
+ if (pollset_set->fds[i] == fd) {
+ pollset_set->fd_count--;
+ GPR_SWAP(grpc_fd *, pollset_set->fds[i], pollset_set->fds[pollset_set->pollset_count]);
+ break;
+ }
+ }
+ gpr_mu_unlock(&pollset_set->mu);
+}
+
+#endif /* GPR_POSIX_SOCKET */
diff --git a/src/core/iomgr/pollset_set_posix.h b/src/core/iomgr/pollset_set_posix.h
new file mode 100644
index 0000000..73c6e0e
--- /dev/null
+++ b/src/core/iomgr/pollset_set_posix.h
@@ -0,0 +1,55 @@
+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ * * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ * * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#ifndef GRPC_INTERNAL_CORE_IOMGR_POLLSET_SET_POSIX_H
+#define GRPC_INTERNAL_CORE_IOMGR_POLLSET_SET_POSIX_H
+
+#include "src/core/iomgr/fd_posix.h"
+#include "src/core/iomgr/pollset_posix.h"
+
+typedef struct grpc_pollset_set {
+ gpr_mu mu;
+
+ size_t pollset_count;
+ size_t pollset_capacity;
+ grpc_pollset **pollsets;
+
+ size_t fd_count;
+ size_t fd_capacity;
+ grpc_fd **fds;
+} grpc_pollset_set;
+
+void grpc_pollset_set_add_fd(grpc_pollset_set *pollset_set, grpc_fd *fd);
+void grpc_pollset_set_del_fd(grpc_pollset_set *pollset_set, grpc_fd *fd);
+
+#endif /* GRPC_INTERNAL_CORE_IOMGR_POLLSET_WINDOWS_H */
diff --git a/src/core/iomgr/pollset_set_windows.c b/src/core/iomgr/pollset_set_windows.c
new file mode 100644
index 0000000..45f8c84
--- /dev/null
+++ b/src/core/iomgr/pollset_set_windows.c
@@ -0,0 +1,49 @@
+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ * * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ * * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#include <grpc/support/port_platform.h>
+
+#ifdef GPR_WINSOCK_SOCKET
+
+#include "src/core/iomgr/pollset_set.h"
+
+void grpc_pollset_set_init(grpc_pollset_set *pollset_set) {
+}
+
+void grpc_pollset_set_destroy(grpc_pollset_set *pollset_set) {
+}
+
+void grpc_pollset_set_add_pollset(grpc_pollset_set *pollset_set, grpc_pollset *pollset) {
+}
+
+#endif /* GPR_WINSOCK_SOCKET */
diff --git a/src/core/iomgr/pollset_set_windows.h b/src/core/iomgr/pollset_set_windows.h
new file mode 100644
index 0000000..8ec6495
--- /dev/null
+++ b/src/core/iomgr/pollset_set_windows.h
@@ -0,0 +1,40 @@
+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ * * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ * * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#ifndef GRPC_INTERNAL_CORE_IOMGR_POLLSET_SET_WINDOWS_H
+#define GRPC_INTERNAL_CORE_IOMGR_POLLSET_SET_WINDOWS_H
+
+typedef struct grpc_pollset_set {
+} grpc_pollset_set;
+
+#endif /* GRPC_INTERNAL_CORE_IOMGR_POLLSET_WINDOWS_H */
diff --git a/src/core/iomgr/tcp_client.h b/src/core/iomgr/tcp_client.h
index 2e91497..e1fdf23 100644
--- a/src/core/iomgr/tcp_client.h
+++ b/src/core/iomgr/tcp_client.h
@@ -35,6 +35,7 @@
#define GRPC_INTERNAL_CORE_IOMGR_TCP_CLIENT_H
#include "src/core/iomgr/endpoint.h"
+#include "src/core/iomgr/pollset_set.h"
#include "src/core/iomgr/sockaddr.h"
#include <grpc/support/time.h>
@@ -42,7 +43,8 @@
cb with arg and the completed connection when done (or call cb with arg and
NULL on failure) */
void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *tcp),
- void *arg, const struct sockaddr *addr,
- int addr_len, gpr_timespec deadline);
+ void *arg, grpc_pollset_set *interested_parties,
+ const struct sockaddr *addr, int addr_len,
+ gpr_timespec deadline);
#endif /* GRPC_INTERNAL_CORE_IOMGR_TCP_CLIENT_H */
diff --git a/src/core/iomgr/tcp_client_posix.c b/src/core/iomgr/tcp_client_posix.c
index 2401fe0..2a81cf5 100644
--- a/src/core/iomgr/tcp_client_posix.c
+++ b/src/core/iomgr/tcp_client_posix.c
@@ -152,6 +152,7 @@
goto finish;
}
} else {
+ gpr_log(GPR_DEBUG, "connected");
ep = grpc_tcp_create(ac->fd, GRPC_TCP_DEFAULT_READ_SLICE_SIZE);
goto finish;
}
@@ -177,14 +178,16 @@
}
void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *ep),
- void *arg, const struct sockaddr *addr,
- int addr_len, gpr_timespec deadline) {
+ void *arg, grpc_pollset_set *interested_parties,
+ const struct sockaddr *addr, int addr_len,
+ gpr_timespec deadline) {
int fd;
grpc_dualstack_mode dsmode;
int err;
async_connect *ac;
struct sockaddr_in6 addr6_v4mapped;
struct sockaddr_in addr4_copy;
+ grpc_fd *fdobj;
/* Use dualstack sockets where available. */
if (grpc_sockaddr_to_v4mapped(addr, &addr6_v4mapped)) {
@@ -207,20 +210,25 @@
return;
}
+ gpr_log(GPR_DEBUG, "connecting fd %d", fd);
+
do {
err = connect(fd, addr, addr_len);
} while (err < 0 && errno == EINTR);
+ fdobj = grpc_fd_create(fd);
+ grpc_pollset_set_add_fd(interested_parties, fdobj);
+
if (err >= 0) {
gpr_log(GPR_DEBUG, "instant connect");
cb(arg,
- grpc_tcp_create(grpc_fd_create(fd), GRPC_TCP_DEFAULT_READ_SLICE_SIZE));
+ grpc_tcp_create(fdobj, GRPC_TCP_DEFAULT_READ_SLICE_SIZE));
return;
}
if (errno != EWOULDBLOCK && errno != EINPROGRESS) {
gpr_log(GPR_ERROR, "connect error: %s", strerror(errno));
- close(fd);
+ grpc_fd_orphan(fdobj, NULL, NULL);
cb(arg, NULL);
return;
}
@@ -228,7 +236,7 @@
ac = gpr_malloc(sizeof(async_connect));
ac->cb = cb;
ac->cb_arg = arg;
- ac->fd = grpc_fd_create(fd);
+ ac->fd = fdobj;
gpr_mu_init(&ac->mu);
ac->refs = 2;
ac->write_closure.cb = on_writable;
diff --git a/src/core/iomgr/tcp_client_windows.c b/src/core/iomgr/tcp_client_windows.c
index cf51743..f5d0cec 100644
--- a/src/core/iomgr/tcp_client_windows.c
+++ b/src/core/iomgr/tcp_client_windows.c
@@ -138,9 +138,10 @@
/* Tries to issue one async connection, then schedules both an IOCP
notification request for the connection, and one timeout alert. */
-void grpc_tcp_client_connect(void(*cb)(void *arg, grpc_endpoint *tcp),
- void *arg, const struct sockaddr *addr,
- int addr_len, gpr_timespec deadline) {
+void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *tcp),
+ void *arg, grpc_pollset_set *interested_parties,
+ const struct sockaddr *addr, int addr_len,
+ gpr_timespec deadline) {
SOCKET sock = INVALID_SOCKET;
BOOL success;
int status;
diff --git a/src/core/iomgr/tcp_server_posix.c b/src/core/iomgr/tcp_server_posix.c
index d1cd8a7..bb53865 100644
--- a/src/core/iomgr/tcp_server_posix.c
+++ b/src/core/iomgr/tcp_server_posix.c
@@ -114,6 +114,9 @@
/* shutdown callback */
void (*shutdown_complete)(void *);
void *shutdown_complete_arg;
+
+ grpc_pollset **pollsets;
+ size_t pollset_count;
};
grpc_tcp_server *grpc_tcp_server_create(void) {
@@ -272,6 +275,8 @@
/* event manager callback when reads are ready */
static void on_read(void *arg, int success) {
server_port *sp = arg;
+ grpc_fd *fdobj;
+ size_t i;
if (!success) {
goto error;
@@ -299,9 +304,16 @@
grpc_set_socket_no_sigpipe_if_possible(fd);
+ fdobj = grpc_fd_create(fd);
+ /* TODO(ctiller): revise this when we have server-side sharding
+ of channels -- we certainly should not be automatically adding every
+ incoming channel to every pollset owned by the server */
+ for (i = 0; i < sp->server->pollset_count; i++) {
+ grpc_pollset_add_fd(sp->server->pollsets[i], fdobj);
+ }
sp->server->cb(
sp->server->cb_arg,
- grpc_tcp_create(grpc_fd_create(fd), GRPC_TCP_DEFAULT_READ_SLICE_SIZE));
+ grpc_tcp_create(fdobj, GRPC_TCP_DEFAULT_READ_SLICE_SIZE));
}
abort();
@@ -436,6 +448,8 @@
GPR_ASSERT(s->active_ports == 0);
s->cb = cb;
s->cb_arg = cb_arg;
+ s->pollsets = pollsets;
+ s->pollset_count = pollset_count;
for (i = 0; i < s->nports; i++) {
for (j = 0; j < pollset_count; j++) {
grpc_pollset_add_fd(pollsets[j], s->ports[i].emfd);
diff --git a/src/core/security/credentials.c b/src/core/security/credentials.c
index f6366f0..e89dfee 100644
--- a/src/core/security/credentials.c
+++ b/src/core/security/credentials.c
@@ -470,6 +470,7 @@
from an http service. */
typedef void (*grpc_fetch_oauth2_func)(grpc_credentials_metadata_request *req,
+ grpc_pollset_set *interested_parties,
grpc_httpcli_response_cb response_cb,
gpr_timespec deadline);
@@ -479,6 +480,7 @@
grpc_mdctx *md_ctx;
grpc_mdelem *access_token_md;
gpr_timespec token_expiration;
+ grpc_pollset_set pollset_set;
grpc_fetch_oauth2_func fetch_func;
} grpc_oauth2_token_fetcher_credentials;
@@ -490,6 +492,7 @@
}
gpr_mu_destroy(&c->mu);
grpc_mdctx_unref(c->md_ctx);
+ grpc_pollset_set_destroy(&c->pollset_set);
gpr_free(c);
}
@@ -637,6 +640,7 @@
} else {
c->fetch_func(
grpc_credentials_metadata_request_create(creds, cb, user_data),
+ &c->pollset_set,
on_oauth2_token_fetcher_http_response,
gpr_time_add(gpr_now(), refresh_threshold));
}
@@ -651,6 +655,7 @@
c->md_ctx = grpc_mdctx_create();
c->token_expiration = gpr_inf_past;
c->fetch_func = fetch_func;
+ grpc_pollset_set_init(&c->pollset_set);
}
static grpc_mdctx *oauth2_token_fetcher_get_metadata_context(
@@ -670,6 +675,7 @@
static void compute_engine_fetch_oauth2(
grpc_credentials_metadata_request *metadata_req,
+ grpc_pollset_set *interested_parties,
grpc_httpcli_response_cb response_cb, gpr_timespec deadline) {
grpc_httpcli_header header = {"Metadata-Flavor", "Google"};
grpc_httpcli_request request;
@@ -678,7 +684,7 @@
request.path = GRPC_COMPUTE_ENGINE_METADATA_TOKEN_PATH;
request.hdr_count = 1;
request.hdrs = &header;
- grpc_httpcli_get(&request, deadline, response_cb, metadata_req);
+ grpc_httpcli_get(&request, deadline, interested_parties, response_cb, metadata_req);
}
grpc_credentials *grpc_compute_engine_credentials_create(void) {
@@ -714,7 +720,7 @@
static void service_account_fetch_oauth2(
grpc_credentials_metadata_request *metadata_req,
- grpc_httpcli_response_cb response_cb, gpr_timespec deadline) {
+ grpc_pollset_set *interested_parties, grpc_httpcli_response_cb response_cb, gpr_timespec deadline) {
grpc_service_account_credentials *c =
(grpc_service_account_credentials *)metadata_req->creds;
grpc_httpcli_header header = {"Content-Type",
@@ -739,7 +745,7 @@
request.hdr_count = 1;
request.hdrs = &header;
request.use_ssl = 1;
- grpc_httpcli_post(&request, body, strlen(body), deadline, response_cb,
+ grpc_httpcli_post(&request, body, strlen(body), deadline, &c->base.pollset_set, response_cb,
metadata_req);
gpr_free(body);
gpr_free(jwt);
@@ -788,6 +794,7 @@
static void refresh_token_fetch_oauth2(
grpc_credentials_metadata_request *metadata_req,
+ grpc_pollset_set *interested_parties,
grpc_httpcli_response_cb response_cb, gpr_timespec deadline) {
grpc_refresh_token_credentials *c =
(grpc_refresh_token_credentials *)metadata_req->creds;
@@ -804,7 +811,7 @@
request.hdr_count = 1;
request.hdrs = &header;
request.use_ssl = 1;
- grpc_httpcli_post(&request, body, strlen(body), deadline, response_cb,
+ grpc_httpcli_post(&request, body, strlen(body), deadline, interested_parties, response_cb,
metadata_req);
gpr_free(body);
}
diff --git a/src/core/security/credentials.h b/src/core/security/credentials.h
index 87c773e..4854641 100644
--- a/src/core/security/credentials.h
+++ b/src/core/security/credentials.h
@@ -105,7 +105,6 @@
grpc_credentials *c, const char *target, const grpc_channel_args *args,
grpc_credentials *request_metadata_creds,
grpc_channel_security_connector **sc, grpc_channel_args **new_args);
-
} grpc_credentials_vtable;
struct grpc_credentials {
diff --git a/src/core/security/google_default_credentials.c b/src/core/security/google_default_credentials.c
index 0e4b9fc..5220bae 100644
--- a/src/core/security/google_default_credentials.c
+++ b/src/core/security/google_default_credentials.c
@@ -60,8 +60,7 @@
}
typedef struct {
- gpr_cv cv;
- gpr_mu mu;
+ grpc_pollset pollset;
int is_done;
int success;
} compute_engine_detector;
@@ -82,22 +81,22 @@
}
}
}
- gpr_mu_lock(&detector->mu);
+ gpr_mu_lock(GRPC_POLLSET_MU(&detector->pollset));
detector->is_done = 1;
- gpr_mu_unlock(&detector->mu);
- gpr_cv_signal(&detector->cv);
+ grpc_pollset_kick(&detector->pollset);
+ gpr_mu_unlock(GRPC_POLLSET_MU(&detector->pollset));
}
static int is_stack_running_on_compute_engine(void) {
compute_engine_detector detector;
grpc_httpcli_request request;
+ grpc_pollset_set pollset_set;
/* The http call is local. If it takes more than one sec, it is for sure not
on compute engine. */
gpr_timespec max_detection_delay = {1, 0};
- gpr_mu_init(&detector.mu);
- gpr_cv_init(&detector.cv);
+ grpc_pollset_init(&detector.pollset);
detector.is_done = 0;
detector.success = 0;
@@ -105,19 +104,24 @@
request.host = GRPC_COMPUTE_ENGINE_DETECTION_HOST;
request.path = "/";
+ grpc_pollset_set_init(&pollset_set);
+ grpc_pollset_set_add_pollset(&pollset_set, &detector.pollset);
+
grpc_httpcli_get(&request, gpr_time_add(gpr_now(), max_detection_delay),
- on_compute_engine_detection_http_response, &detector);
+ &pollset_set, on_compute_engine_detection_http_response,
+ &detector);
/* Block until we get the response. This is not ideal but this should only be
called once for the lifetime of the process by the default credentials. */
- gpr_mu_lock(&detector.mu);
+ gpr_mu_lock(GRPC_POLLSET_MU(&detector.pollset));
while (!detector.is_done) {
- gpr_cv_wait(&detector.cv, &detector.mu, gpr_inf_future);
+ grpc_pollset_work(&detector.pollset, gpr_inf_future);
}
- gpr_mu_unlock(&detector.mu);
+ gpr_mu_unlock(GRPC_POLLSET_MU(&detector.pollset));
- gpr_mu_destroy(&detector.mu);
- gpr_cv_destroy(&detector.cv);
+ grpc_pollset_set_destroy(&pollset_set);
+ grpc_pollset_destroy(&detector.pollset);
+
return detector.success;
}
diff --git a/src/core/surface/call.c b/src/core/surface/call.c
index 50df36c..d403eee 100644
--- a/src/core/surface/call.c
+++ b/src/core/surface/call.c
@@ -588,6 +588,7 @@
lock(call);
if (call->last_send_contains & (1 << GRPC_IOREQ_SEND_INITIAL_METADATA)) {
finish_ioreq_op(call, GRPC_IOREQ_SEND_INITIAL_METADATA, success);
+ call->write_state = WRITE_STATE_STARTED;
}
if (call->last_send_contains & (1 << GRPC_IOREQ_SEND_MESSAGE)) {
finish_ioreq_op(call, GRPC_IOREQ_SEND_MESSAGE, success);
@@ -596,6 +597,10 @@
finish_ioreq_op(call, GRPC_IOREQ_SEND_TRAILING_METADATA, success);
finish_ioreq_op(call, GRPC_IOREQ_SEND_STATUS, success);
finish_ioreq_op(call, GRPC_IOREQ_SEND_CLOSE, 1);
+ call->write_state = WRITE_STATE_WRITE_CLOSED;
+ }
+ if (!success) {
+ call->write_state = WRITE_STATE_WRITE_CLOSED;
}
call->last_send_contains = 0;
call->sending = 0;
@@ -810,7 +815,6 @@
op->send_ops = &call->send_ops;
op->bind_pollset = grpc_cq_pollset(call->cq);
call->last_send_contains |= 1 << GRPC_IOREQ_SEND_INITIAL_METADATA;
- call->write_state = WRITE_STATE_STARTED;
call->send_initial_metadata_count = 0;
/* fall through intended */
case WRITE_STATE_STARTED:
@@ -826,7 +830,6 @@
op->is_last_send = 1;
op->send_ops = &call->send_ops;
call->last_send_contains |= 1 << GRPC_IOREQ_SEND_CLOSE;
- call->write_state = WRITE_STATE_WRITE_CLOSED;
if (!call->is_client) {
/* send trailing metadata */
data = call->request_data[GRPC_IOREQ_SEND_TRAILING_METADATA];
diff --git a/src/core/surface/channel_create.c b/src/core/surface/channel_create.c
index daa8d3a..b2d3a19 100644
--- a/src/core/surface/channel_create.c
+++ b/src/core/surface/channel_create.c
@@ -125,9 +125,10 @@
if (!r->resolved) return 0;
if (r->resolved_index == r->resolved->naddrs) return 0;
addr = &r->resolved->addrs[r->resolved_index++];
- grpc_tcp_client_connect(on_connect, r, (struct sockaddr *)&addr->addr,
- addr->len,
- grpc_client_setup_request_deadline(r->cs_request));
+ grpc_tcp_client_connect(
+ on_connect, r, grpc_client_setup_get_interested_parties(r->cs_request),
+ (struct sockaddr *)&addr->addr, addr->len,
+ grpc_client_setup_request_deadline(r->cs_request));
return 1;
}
diff --git a/src/core/surface/secure_channel_create.c b/src/core/surface/secure_channel_create.c
index 3e33129..73f4d51 100644
--- a/src/core/surface/secure_channel_create.c
+++ b/src/core/surface/secure_channel_create.c
@@ -140,9 +140,10 @@
if (!r->resolved) return 0;
if (r->resolved_index == r->resolved->naddrs) return 0;
addr = &r->resolved->addrs[r->resolved_index++];
- grpc_tcp_client_connect(on_connect, r, (struct sockaddr *)&addr->addr,
- addr->len,
- grpc_client_setup_request_deadline(r->cs_request));
+ grpc_tcp_client_connect(
+ on_connect, r, grpc_client_setup_get_interested_parties(r->cs_request),
+ (struct sockaddr *)&addr->addr, addr->len,
+ grpc_client_setup_request_deadline(r->cs_request));
return 1;
}
diff --git a/src/core/transport/transport.c b/src/core/transport/transport.c
index d9a1319..0bceeab 100644
--- a/src/core/transport/transport.c
+++ b/src/core/transport/transport.c
@@ -86,6 +86,10 @@
setup->vtable->initiate(setup);
}
+void grpc_transport_setup_add_interested_party(grpc_transport_setup *setup, grpc_pollset *pollset) {
+ setup->vtable->add_interested_party(setup, pollset);
+}
+
void grpc_transport_op_finish_with_failure(grpc_transport_op *op) {
if (op->send_ops) {
op->on_done_send(op->send_user_data, 0);
diff --git a/src/core/transport/transport.h b/src/core/transport/transport.h
index 7a389ea..5303cbe 100644
--- a/src/core/transport/transport.h
+++ b/src/core/transport/transport.h
@@ -37,6 +37,7 @@
#include <stddef.h>
#include "src/core/iomgr/pollset.h"
+#include "src/core/iomgr/pollset_set.h"
#include "src/core/transport/stream_op.h"
/* forward declarations */
@@ -194,6 +195,7 @@
struct grpc_transport_setup_vtable {
void (*initiate)(grpc_transport_setup *setup);
+ void (*add_interested_party)(grpc_transport_setup *setup, grpc_pollset *pollset);
void (*cancel)(grpc_transport_setup *setup);
};
@@ -210,6 +212,9 @@
This *may* be implemented as a no-op if the setup process monitors something
continuously. */
void grpc_transport_setup_initiate(grpc_transport_setup *setup);
+
+void grpc_transport_setup_add_interested_party(grpc_transport_setup *setup, grpc_pollset *pollset);
+
/* Cancel transport setup. After this returns, no new transports should be
created, and all pending transport setup callbacks should be completed.
After this call completes, setup should be considered invalid (this can be
diff --git a/test/core/end2end/dualstack_socket_test.c b/test/core/end2end/dualstack_socket_test.c
index be3c7ca..b54b8ec 100644
--- a/test/core/end2end/dualstack_socket_test.c
+++ b/test/core/end2end/dualstack_socket_test.c
@@ -62,12 +62,10 @@
char *server_hostport;
grpc_channel *client;
grpc_server *server;
- grpc_completion_queue *client_cq;
- grpc_completion_queue *server_cq;
+ grpc_completion_queue *cq;
grpc_call *c;
grpc_call *s;
- cq_verifier *v_client;
- cq_verifier *v_server;
+ cq_verifier *cqv;
gpr_timespec deadline;
int got_port;
grpc_op ops[6];
@@ -93,9 +91,9 @@
grpc_call_details_init(&call_details);
/* Create server. */
- server_cq = grpc_completion_queue_create();
+ cq = grpc_completion_queue_create();
server = grpc_server_create(NULL);
- grpc_server_register_completion_queue(server, server_cq);
+ grpc_server_register_completion_queue(server, cq);
GPR_ASSERT((got_port = grpc_server_add_http2_port(server, server_hostport)) >
0);
if (port == 0) {
@@ -104,13 +102,11 @@
GPR_ASSERT(port == got_port);
}
grpc_server_start(server);
- v_server = cq_verifier_create(server_cq);
+ cqv = cq_verifier_create(cq);
/* Create client. */
gpr_join_host_port(&client_hostport, client_host, port);
- client_cq = grpc_completion_queue_create();
client = grpc_channel_create(client_hostport, NULL);
- v_client = cq_verifier_create(client_cq);
gpr_log(GPR_INFO, "Testing with server=%s client=%s (expecting %s)",
server_hostport, client_hostport, expect_ok ? "success" : "failure");
@@ -128,7 +124,7 @@
}
/* Send a trivial request. */
- c = grpc_channel_create_call(client, client_cq, "/foo", "foo.test.google.fr",
+ c = grpc_channel_create_call(client, cq, "/foo", "foo.test.google.fr",
deadline);
GPR_ASSERT(c);
@@ -153,10 +149,10 @@
/* Check for a successful request. */
GPR_ASSERT(GRPC_CALL_OK ==
grpc_server_request_call(server, &s, &call_details,
- &request_metadata_recv, server_cq,
- server_cq, tag(101)));
- cq_expect_completion(v_server, tag(101), 1);
- cq_verify(v_server);
+ &request_metadata_recv, cq,
+ cq, tag(101)));
+ cq_expect_completion(cqv, tag(101), 1);
+ cq_verify(cqv);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
@@ -173,11 +169,9 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(v_server, tag(102), 1);
- cq_verify(v_server);
-
- cq_expect_completion(v_client, tag(1), 1);
- cq_verify(v_client);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
+ cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
GPR_ASSERT(0 == strcmp(details, "xyz"));
@@ -188,29 +182,25 @@
grpc_call_destroy(s);
} else {
/* Check for a failed connection. */
- cq_expect_completion(v_client, tag(1), 1);
- cq_verify(v_client);
+ cq_expect_completion(cqv, tag(1), 1);
+ cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_DEADLINE_EXCEEDED);
}
grpc_call_destroy(c);
- cq_verifier_destroy(v_client);
- cq_verifier_destroy(v_server);
+ cq_verifier_destroy(cqv);
/* Destroy client. */
grpc_channel_destroy(client);
- grpc_completion_queue_shutdown(client_cq);
- drain_cq(client_cq);
- grpc_completion_queue_destroy(client_cq);
/* Destroy server. */
grpc_server_shutdown(server);
grpc_server_destroy(server);
- grpc_completion_queue_shutdown(server_cq);
- drain_cq(server_cq);
- grpc_completion_queue_destroy(server_cq);
+ grpc_completion_queue_shutdown(cq);
+ drain_cq(cq);
+ grpc_completion_queue_destroy(cq);
}
int main(int argc, char **argv) {
diff --git a/test/core/end2end/end2end_tests.h b/test/core/end2end/end2end_tests.h
index a61c725..8996bac 100644
--- a/test/core/end2end/end2end_tests.h
+++ b/test/core/end2end/end2end_tests.h
@@ -44,8 +44,7 @@
#define FEATURE_MASK_SUPPORTS_PER_CALL_CREDENTIALS 4
struct grpc_end2end_test_fixture {
- grpc_completion_queue *server_cq;
- grpc_completion_queue *client_cq;
+ grpc_completion_queue *cq;
grpc_server *server;
grpc_channel *client;
void *fixture_data;
diff --git a/test/core/end2end/fixtures/chttp2_fake_security.c b/test/core/end2end/fixtures/chttp2_fake_security.c
index 5323e29..7e98d72 100644
--- a/test/core/end2end/fixtures/chttp2_fake_security.c
+++ b/test/core/end2end/fixtures/chttp2_fake_security.c
@@ -60,8 +60,7 @@
gpr_join_host_port(&ffd->localaddr, "localhost", port);
f.fixture_data = ffd;
- f.client_cq = grpc_completion_queue_create();
- f.server_cq = grpc_completion_queue_create();
+ f.cq = grpc_completion_queue_create();
return f;
}
@@ -83,7 +82,7 @@
grpc_server_destroy(f->server);
}
f->server = grpc_server_create(server_args);
- grpc_server_register_completion_queue(f->server, f->server_cq);
+ grpc_server_register_completion_queue(f->server, f->cq);
GPR_ASSERT(grpc_server_add_secure_http2_port(f->server, ffd->localaddr, server_creds));
grpc_server_credentials_release(server_creds);
grpc_server_start(f->server);
diff --git a/test/core/end2end/fixtures/chttp2_fullstack.c b/test/core/end2end/fixtures/chttp2_fullstack.c
index f92b40e..b83e227 100644
--- a/test/core/end2end/fixtures/chttp2_fullstack.c
+++ b/test/core/end2end/fixtures/chttp2_fullstack.c
@@ -65,8 +65,7 @@
gpr_join_host_port(&ffd->localaddr, "localhost", port);
f.fixture_data = ffd;
- f.client_cq = grpc_completion_queue_create();
- f.server_cq = grpc_completion_queue_create();
+ f.cq = grpc_completion_queue_create();
return f;
}
@@ -84,7 +83,7 @@
grpc_server_destroy(f->server);
}
f->server = grpc_server_create(server_args);
- grpc_server_register_completion_queue(f->server, f->server_cq);
+ grpc_server_register_completion_queue(f->server, f->cq);
GPR_ASSERT(grpc_server_add_http2_port(f->server, ffd->localaddr));
grpc_server_start(f->server);
}
diff --git a/test/core/end2end/fixtures/chttp2_fullstack_uds_posix.c b/test/core/end2end/fixtures/chttp2_fullstack_uds_posix.c
index 876782d..1c1751d 100644
--- a/test/core/end2end/fixtures/chttp2_fullstack_uds_posix.c
+++ b/test/core/end2end/fixtures/chttp2_fullstack_uds_posix.c
@@ -70,8 +70,7 @@
unique++);
f.fixture_data = ffd;
- f.client_cq = grpc_completion_queue_create();
- f.server_cq = grpc_completion_queue_create();
+ f.cq = grpc_completion_queue_create();
return f;
}
@@ -89,7 +88,7 @@
grpc_server_destroy(f->server);
}
f->server = grpc_server_create(server_args);
- grpc_server_register_completion_queue(f->server, f->server_cq);
+ grpc_server_register_completion_queue(f->server, f->cq);
GPR_ASSERT(grpc_server_add_http2_port(f->server, ffd->localaddr));
grpc_server_start(f->server);
}
diff --git a/test/core/end2end/fixtures/chttp2_simple_ssl_fullstack.c b/test/core/end2end/fixtures/chttp2_simple_ssl_fullstack.c
index 6d1b7b5..e351fb3 100644
--- a/test/core/end2end/fixtures/chttp2_simple_ssl_fullstack.c
+++ b/test/core/end2end/fixtures/chttp2_simple_ssl_fullstack.c
@@ -63,8 +63,7 @@
gpr_join_host_port(&ffd->localaddr, "localhost", port);
f.fixture_data = ffd;
- f.client_cq = grpc_completion_queue_create();
- f.server_cq = grpc_completion_queue_create();
+ f.cq = grpc_completion_queue_create();
return f;
}
@@ -86,7 +85,7 @@
grpc_server_destroy(f->server);
}
f->server = grpc_server_create(server_args);
- grpc_server_register_completion_queue(f->server, f->server_cq);
+ grpc_server_register_completion_queue(f->server, f->cq);
GPR_ASSERT(grpc_server_add_secure_http2_port(f->server, ffd->localaddr, server_creds));
grpc_server_credentials_release(server_creds);
grpc_server_start(f->server);
diff --git a/test/core/end2end/fixtures/chttp2_simple_ssl_with_oauth2_fullstack.c b/test/core/end2end/fixtures/chttp2_simple_ssl_with_oauth2_fullstack.c
index 4a15d50..8d51263 100644
--- a/test/core/end2end/fixtures/chttp2_simple_ssl_with_oauth2_fullstack.c
+++ b/test/core/end2end/fixtures/chttp2_simple_ssl_with_oauth2_fullstack.c
@@ -61,8 +61,7 @@
gpr_join_host_port(&ffd->localaddr, "localhost", port);
f.fixture_data = ffd;
- f.client_cq = grpc_completion_queue_create();
- f.server_cq = grpc_completion_queue_create();
+ f.cq = grpc_completion_queue_create();
return f;
}
@@ -84,7 +83,7 @@
grpc_server_destroy(f->server);
}
f->server = grpc_server_create(server_args);
- grpc_server_register_completion_queue(f->server, f->server_cq);
+ grpc_server_register_completion_queue(f->server, f->cq);
GPR_ASSERT(grpc_server_add_secure_http2_port(f->server, ffd->localaddr, server_creds));
grpc_server_credentials_release(server_creds);
grpc_server_start(f->server);
diff --git a/test/core/end2end/fixtures/chttp2_socket_pair.c b/test/core/end2end/fixtures/chttp2_socket_pair.c
index 43ebf7e..ec75419 100644
--- a/test/core/end2end/fixtures/chttp2_socket_pair.c
+++ b/test/core/end2end/fixtures/chttp2_socket_pair.c
@@ -94,8 +94,7 @@
grpc_end2end_test_fixture f;
memset(&f, 0, sizeof(f));
f.fixture_data = sfd;
- f.client_cq = grpc_completion_queue_create();
- f.server_cq = grpc_completion_queue_create();
+ f.cq = grpc_completion_queue_create();
*sfd = grpc_iomgr_create_endpoint_pair(65536);
@@ -118,7 +117,7 @@
grpc_endpoint_pair *sfd = f->fixture_data;
GPR_ASSERT(!f->server);
f->server = grpc_server_create_from_filters(NULL, 0, server_args);
- grpc_server_register_completion_queue(f->server, f->server_cq);
+ grpc_server_register_completion_queue(f->server, f->cq);
grpc_server_start(f->server);
grpc_create_chttp2_transport(server_setup_transport, f, server_args,
sfd->server, NULL, 0, grpc_mdctx_create(), 0);
diff --git a/test/core/end2end/fixtures/chttp2_socket_pair_one_byte_at_a_time.c b/test/core/end2end/fixtures/chttp2_socket_pair_one_byte_at_a_time.c
index 385d5a4..ac5fc40 100644
--- a/test/core/end2end/fixtures/chttp2_socket_pair_one_byte_at_a_time.c
+++ b/test/core/end2end/fixtures/chttp2_socket_pair_one_byte_at_a_time.c
@@ -94,8 +94,7 @@
grpc_end2end_test_fixture f;
memset(&f, 0, sizeof(f));
f.fixture_data = sfd;
- f.client_cq = grpc_completion_queue_create();
- f.server_cq = grpc_completion_queue_create();
+ f.cq = grpc_completion_queue_create();
*sfd = grpc_iomgr_create_endpoint_pair(1);
@@ -118,7 +117,7 @@
grpc_endpoint_pair *sfd = f->fixture_data;
GPR_ASSERT(!f->server);
f->server = grpc_server_create_from_filters(NULL, 0, server_args);
- grpc_server_register_completion_queue(f->server, f->server_cq);
+ grpc_server_register_completion_queue(f->server, f->cq);
grpc_server_start(f->server);
grpc_create_chttp2_transport(server_setup_transport, f, server_args,
sfd->server, NULL, 0, grpc_mdctx_create(), 0);
diff --git a/test/core/end2end/tests/bad_hostname.c b/test/core/end2end/tests/bad_hostname.c
index e145d2e..726bada 100644
--- a/test/core/end2end/tests/bad_hostname.c
+++ b/test/core/end2end/tests/bad_hostname.c
@@ -91,18 +91,15 @@
shutdown_server(f);
shutdown_client(f);
- grpc_completion_queue_shutdown(f->server_cq);
- drain_cq(f->server_cq);
- grpc_completion_queue_destroy(f->server_cq);
- grpc_completion_queue_shutdown(f->client_cq);
- drain_cq(f->client_cq);
- grpc_completion_queue_destroy(f->client_cq);
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
}
static void simple_request_body(grpc_end2end_test_fixture f) {
grpc_call *c;
gpr_timespec deadline = five_seconds_time();
- cq_verifier *v_client = cq_verifier_create(f.client_cq);
+ cq_verifier *cqv = cq_verifier_create(f.cq);
grpc_op ops[6];
grpc_op *op;
grpc_metadata_array initial_metadata_recv;
@@ -113,7 +110,7 @@
char *details = NULL;
size_t details_capacity = 0;
- c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
+ c = grpc_channel_create_call(f.client, f.cq, "/foo",
"slartibartfast.local", deadline);
GPR_ASSERT(c);
@@ -139,8 +136,8 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1)));
- cq_expect_completion(v_client, tag(1), 1);
- cq_verify(v_client);
+ cq_expect_completion(cqv, tag(1), 1);
+ cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_UNAUTHENTICATED);
@@ -152,7 +149,7 @@
grpc_call_destroy(c);
- cq_verifier_destroy(v_client);
+ cq_verifier_destroy(cqv);
}
static void test_invoke_simple_request(grpc_end2end_test_config config) {
diff --git a/test/core/end2end/tests/cancel_after_accept.c b/test/core/end2end/tests/cancel_after_accept.c
index 2753338..db6d089 100644
--- a/test/core/end2end/tests/cancel_after_accept.c
+++ b/test/core/end2end/tests/cancel_after_accept.c
@@ -90,12 +90,9 @@
shutdown_server(f);
shutdown_client(f);
- grpc_completion_queue_shutdown(f->server_cq);
- drain_cq(f->server_cq);
- grpc_completion_queue_destroy(f->server_cq);
- grpc_completion_queue_shutdown(f->client_cq);
- drain_cq(f->client_cq);
- grpc_completion_queue_destroy(f->client_cq);
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
}
/* Cancel after accept, no payload */
@@ -107,8 +104,7 @@
grpc_call *s;
grpc_end2end_test_fixture f = begin_test(config, __FUNCTION__, NULL, NULL);
gpr_timespec deadline = five_seconds_time();
- cq_verifier *v_client = cq_verifier_create(f.client_cq);
- cq_verifier *v_server = cq_verifier_create(f.server_cq);
+ cq_verifier *cqv = cq_verifier_create(f.cq);
grpc_metadata_array initial_metadata_recv;
grpc_metadata_array trailing_metadata_recv;
grpc_metadata_array request_metadata_recv;
@@ -126,7 +122,7 @@
grpc_byte_buffer_create(&response_payload_slice, 1);
int was_cancelled = 2;
- c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
+ c = grpc_channel_create_call(f.client, f.cq, "/foo",
"foo.test.google.fr", deadline);
GPR_ASSERT(c);
@@ -158,10 +154,10 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_server_request_call(f.server, &s, &call_details,
- &request_metadata_recv, f.server_cq,
- f.server_cq, tag(2)));
- cq_expect_completion(v_server, tag(2), 1);
- cq_verify(v_server);
+ &request_metadata_recv, f.cq,
+ f.cq, tag(2)));
+ cq_expect_completion(cqv, tag(2), 1);
+ cq_verify(cqv);
op = ops;
op->op = GRPC_OP_RECV_MESSAGE;
@@ -180,11 +176,9 @@
GPR_ASSERT(GRPC_CALL_OK == mode.initiate_cancel(c));
- cq_expect_completion(v_server, tag(3), 1);
- cq_verify(v_server);
-
- cq_expect_completion(v_client, tag(1), 1);
- cq_verify(v_client);
+ cq_expect_completion(cqv, tag(3), 1);
+ cq_expect_completion(cqv, tag(1), 1);
+ cq_verify(cqv);
GPR_ASSERT(status == mode.expect_status);
GPR_ASSERT(0 == strcmp(details, mode.expect_details));
@@ -204,8 +198,7 @@
grpc_call_destroy(c);
grpc_call_destroy(s);
- cq_verifier_destroy(v_client);
- cq_verifier_destroy(v_server);
+ cq_verifier_destroy(cqv);
end_test(&f);
config.tear_down_data(&f);
}
diff --git a/test/core/end2end/tests/cancel_after_accept_and_writes_closed.c b/test/core/end2end/tests/cancel_after_accept_and_writes_closed.c
index eaf8b60..0162df7 100644
--- a/test/core/end2end/tests/cancel_after_accept_and_writes_closed.c
+++ b/test/core/end2end/tests/cancel_after_accept_and_writes_closed.c
@@ -90,12 +90,9 @@
shutdown_server(f);
shutdown_client(f);
- grpc_completion_queue_shutdown(f->server_cq);
- drain_cq(f->server_cq);
- grpc_completion_queue_destroy(f->server_cq);
- grpc_completion_queue_shutdown(f->client_cq);
- drain_cq(f->client_cq);
- grpc_completion_queue_destroy(f->client_cq);
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
}
/* Cancel after accept with a writes closed, no payload */
@@ -107,8 +104,7 @@
grpc_call *s;
grpc_end2end_test_fixture f = begin_test(config, __FUNCTION__, NULL, NULL);
gpr_timespec deadline = five_seconds_time();
- cq_verifier *v_client = cq_verifier_create(f.client_cq);
- cq_verifier *v_server = cq_verifier_create(f.server_cq);
+ cq_verifier *cqv = cq_verifier_create(f.cq);
grpc_metadata_array initial_metadata_recv;
grpc_metadata_array trailing_metadata_recv;
grpc_metadata_array request_metadata_recv;
@@ -126,7 +122,7 @@
grpc_byte_buffer_create(&response_payload_slice, 1);
int was_cancelled = 2;
- c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
+ c = grpc_channel_create_call(f.client, f.cq, "/foo",
"foo.test.google.fr", deadline);
GPR_ASSERT(c);
@@ -160,10 +156,10 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_server_request_call(f.server, &s, &call_details,
- &request_metadata_recv, f.server_cq,
- f.server_cq, tag(2)));
- cq_expect_completion(v_server, tag(2), 1);
- cq_verify(v_server);
+ &request_metadata_recv, f.cq,
+ f.cq, tag(2)));
+ cq_expect_completion(cqv, tag(2), 1);
+ cq_verify(cqv);
op = ops;
op->op = GRPC_OP_RECV_MESSAGE;
@@ -182,11 +178,9 @@
GPR_ASSERT(GRPC_CALL_OK == mode.initiate_cancel(c));
- cq_expect_completion(v_server, tag(3), 1);
- cq_verify(v_server);
-
- cq_expect_completion(v_client, tag(1), 1);
- cq_verify(v_client);
+ cq_expect_completion(cqv, tag(3), 1);
+ cq_expect_completion(cqv, tag(1), 1);
+ cq_verify(cqv);
GPR_ASSERT(status == mode.expect_status);
GPR_ASSERT(0 == strcmp(details, mode.expect_details));
@@ -206,8 +200,7 @@
grpc_call_destroy(c);
grpc_call_destroy(s);
- cq_verifier_destroy(v_client);
- cq_verifier_destroy(v_server);
+ cq_verifier_destroy(cqv);
end_test(&f);
config.tear_down_data(&f);
}
diff --git a/test/core/end2end/tests/cancel_after_invoke.c b/test/core/end2end/tests/cancel_after_invoke.c
index 618d9e9..cf12269 100644
--- a/test/core/end2end/tests/cancel_after_invoke.c
+++ b/test/core/end2end/tests/cancel_after_invoke.c
@@ -91,12 +91,9 @@
shutdown_server(f);
shutdown_client(f);
- grpc_completion_queue_shutdown(f->server_cq);
- drain_cq(f->server_cq);
- grpc_completion_queue_destroy(f->server_cq);
- grpc_completion_queue_shutdown(f->client_cq);
- drain_cq(f->client_cq);
- grpc_completion_queue_destroy(f->client_cq);
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
}
/* Cancel after invoke, no payload */
@@ -108,7 +105,7 @@
grpc_end2end_test_fixture f =
begin_test(config, __FUNCTION__, mode, NULL, NULL);
gpr_timespec deadline = five_seconds_time();
- cq_verifier *v_client = cq_verifier_create(f.client_cq);
+ cq_verifier *cqv = cq_verifier_create(f.cq);
grpc_metadata_array initial_metadata_recv;
grpc_metadata_array trailing_metadata_recv;
grpc_metadata_array request_metadata_recv;
@@ -121,7 +118,7 @@
grpc_byte_buffer *request_payload =
grpc_byte_buffer_create(&request_payload_slice, 1);
- c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
+ c = grpc_channel_create_call(f.client, f.cq, "/foo",
"foo.test.google.fr", deadline);
GPR_ASSERT(c);
@@ -155,8 +152,8 @@
GPR_ASSERT(GRPC_CALL_OK == mode.initiate_cancel(c));
- cq_expect_completion(v_client, tag(1), 1);
- cq_verify(v_client);
+ cq_expect_completion(cqv, tag(1), 1);
+ cq_verify(cqv);
GPR_ASSERT(status == mode.expect_status);
GPR_ASSERT(0 == strcmp(details, mode.expect_details));
@@ -172,7 +169,7 @@
grpc_call_destroy(c);
- cq_verifier_destroy(v_client);
+ cq_verifier_destroy(cqv);
end_test(&f);
config.tear_down_data(&f);
}
diff --git a/test/core/end2end/tests/cancel_before_invoke.c b/test/core/end2end/tests/cancel_before_invoke.c
index 9c9d6aa..19d3bad 100644
--- a/test/core/end2end/tests/cancel_before_invoke.c
+++ b/test/core/end2end/tests/cancel_before_invoke.c
@@ -89,12 +89,9 @@
shutdown_server(f);
shutdown_client(f);
- grpc_completion_queue_shutdown(f->server_cq);
- drain_cq(f->server_cq);
- grpc_completion_queue_destroy(f->server_cq);
- grpc_completion_queue_shutdown(f->client_cq);
- drain_cq(f->client_cq);
- grpc_completion_queue_destroy(f->client_cq);
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
}
/* Cancel before invoke */
@@ -105,7 +102,7 @@
grpc_call *c;
grpc_end2end_test_fixture f = begin_test(config, __FUNCTION__, NULL, NULL);
gpr_timespec deadline = five_seconds_time();
- cq_verifier *v_client = cq_verifier_create(f.client_cq);
+ cq_verifier *cqv = cq_verifier_create(f.cq);
grpc_metadata_array initial_metadata_recv;
grpc_metadata_array trailing_metadata_recv;
grpc_metadata_array request_metadata_recv;
@@ -118,7 +115,7 @@
grpc_byte_buffer *request_payload =
grpc_byte_buffer_create(&request_payload_slice, 1);
- c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
+ c = grpc_channel_create_call(f.client, f.cq, "/foo",
"foo.test.google.fr", deadline);
GPR_ASSERT(c);
@@ -152,8 +149,8 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, test_ops, tag(1)));
- cq_expect_completion(v_client, tag(1), 1);
- cq_verify(v_client);
+ cq_expect_completion(cqv, tag(1), 1);
+ cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_CANCELLED);
@@ -168,7 +165,7 @@
grpc_call_destroy(c);
- cq_verifier_destroy(v_client);
+ cq_verifier_destroy(cqv);
end_test(&f);
config.tear_down_data(&f);
}
diff --git a/test/core/end2end/tests/cancel_in_a_vacuum.c b/test/core/end2end/tests/cancel_in_a_vacuum.c
index 1bc393a..679b137 100644
--- a/test/core/end2end/tests/cancel_in_a_vacuum.c
+++ b/test/core/end2end/tests/cancel_in_a_vacuum.c
@@ -88,12 +88,9 @@
shutdown_server(f);
shutdown_client(f);
- grpc_completion_queue_shutdown(f->server_cq);
- drain_cq(f->server_cq);
- grpc_completion_queue_destroy(f->server_cq);
- grpc_completion_queue_shutdown(f->client_cq);
- drain_cq(f->client_cq);
- grpc_completion_queue_destroy(f->client_cq);
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
}
/* Cancel and do nothing */
@@ -102,9 +99,9 @@
grpc_call *c;
grpc_end2end_test_fixture f = begin_test(config, __FUNCTION__, NULL, NULL);
gpr_timespec deadline = five_seconds_time();
- cq_verifier *v_client = cq_verifier_create(f.client_cq);
+ cq_verifier *v_client = cq_verifier_create(f.cq);
- c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
+ c = grpc_channel_create_call(f.client, f.cq, "/foo",
"foo.test.google.fr", deadline);
GPR_ASSERT(c);
diff --git a/test/core/end2end/tests/census_simple_request.c b/test/core/end2end/tests/census_simple_request.c
index 2017ab6..4f6c57c 100644
--- a/test/core/end2end/tests/census_simple_request.c
+++ b/test/core/end2end/tests/census_simple_request.c
@@ -84,12 +84,9 @@
shutdown_server(f);
shutdown_client(f);
- grpc_completion_queue_shutdown(f->server_cq);
- drain_cq(f->server_cq);
- grpc_completion_queue_destroy(f->server_cq);
- grpc_completion_queue_shutdown(f->client_cq);
- drain_cq(f->client_cq);
- grpc_completion_queue_destroy(f->client_cq);
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
}
static void *tag(gpr_intptr t) { return (void *)t; }
@@ -98,8 +95,7 @@
grpc_call *c;
grpc_call *s;
gpr_timespec deadline = n_seconds_time(5);
- cq_verifier *v_client = cq_verifier_create(f.client_cq);
- cq_verifier *v_server = cq_verifier_create(f.server_cq);
+ cq_verifier *cqv = cq_verifier_create(f.cq);
grpc_op ops[6];
grpc_op *op;
grpc_metadata_array initial_metadata_recv;
@@ -111,7 +107,7 @@
size_t details_capacity = 0;
int was_cancelled = 2;
- c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
+ c = grpc_channel_create_call(f.client, f.cq, "/foo",
"foo.test.google.fr:1234", deadline);
GPR_ASSERT(c);
@@ -139,10 +135,10 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_server_request_call(f.server, &s, &call_details,
- &request_metadata_recv, f.server_cq,
- f.server_cq, tag(101)));
- cq_expect_completion(v_server, tag(101), 1);
- cq_verify(v_server);
+ &request_metadata_recv, f.cq,
+ f.cq, tag(101)));
+ cq_expect_completion(cqv, tag(101), 1);
+ cq_verify(cqv);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
@@ -158,11 +154,9 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(v_server, tag(102), 1);
- cq_verify(v_server);
-
- cq_expect_completion(v_client, tag(1), 1);
- cq_verify(v_client);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
+ cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
GPR_ASSERT(0 == strcmp(details, "xyz"));
@@ -179,8 +173,7 @@
grpc_call_destroy(c);
grpc_call_destroy(s);
- cq_verifier_destroy(v_client);
- cq_verifier_destroy(v_server);
+ cq_verifier_destroy(cqv);
}
static void test_invoke_request_with_census(
diff --git a/test/core/end2end/tests/disappearing_server.c b/test/core/end2end/tests/disappearing_server.c
index 89fff81..9bcc79d 100644
--- a/test/core/end2end/tests/disappearing_server.c
+++ b/test/core/end2end/tests/disappearing_server.c
@@ -77,17 +77,13 @@
shutdown_server(f);
shutdown_client(f);
- grpc_completion_queue_shutdown(f->server_cq);
- drain_cq(f->server_cq);
- grpc_completion_queue_destroy(f->server_cq);
- grpc_completion_queue_shutdown(f->client_cq);
- drain_cq(f->client_cq);
- grpc_completion_queue_destroy(f->client_cq);
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
}
static void do_request_and_shutdown_server(grpc_end2end_test_fixture *f,
- cq_verifier *v_client,
- cq_verifier *v_server) {
+ cq_verifier *cqv) {
grpc_call *c;
grpc_call *s;
gpr_timespec deadline = five_seconds_time();
@@ -102,7 +98,7 @@
size_t details_capacity = 0;
int was_cancelled = 2;
- c = grpc_channel_create_call(f->client, f->client_cq, "/foo",
+ c = grpc_channel_create_call(f->client, f->cq, "/foo",
"foo.test.google.fr:1234", deadline);
GPR_ASSERT(c);
@@ -130,10 +126,10 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_server_request_call(f->server, &s, &call_details,
- &request_metadata_recv, f->server_cq,
- f->server_cq, tag(101)));
- cq_expect_completion(v_server, tag(101), 1);
- cq_verify(v_server);
+ &request_metadata_recv, f->cq,
+ f->cq, tag(101)));
+ cq_expect_completion(cqv, tag(101), 1);
+ cq_verify(cqv);
/* should be able to shut down the server early
- and still complete the request */
@@ -153,11 +149,9 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(v_server, tag(102), 1);
- cq_verify(v_server);
-
- cq_expect_completion(v_client, tag(1), 1);
- cq_verify(v_client);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
+ cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
GPR_ASSERT(0 == strcmp(details, "xyz"));
@@ -177,23 +171,21 @@
static void disappearing_server_test(grpc_end2end_test_config config) {
grpc_end2end_test_fixture f = config.create_fixture(NULL, NULL);
- cq_verifier *v_client = cq_verifier_create(f.client_cq);
- cq_verifier *v_server = cq_verifier_create(f.server_cq);
+ cq_verifier *cqv = cq_verifier_create(f.cq);
gpr_log(GPR_INFO, "%s/%s", __FUNCTION__, config.name);
config.init_client(&f, NULL);
config.init_server(&f, NULL);
- do_request_and_shutdown_server(&f, v_client, v_server);
+ do_request_and_shutdown_server(&f, cqv);
/* now destroy and recreate the server */
config.init_server(&f, NULL);
- do_request_and_shutdown_server(&f, v_client, v_server);
+ do_request_and_shutdown_server(&f, cqv);
- cq_verifier_destroy(v_client);
- cq_verifier_destroy(v_server);
+ cq_verifier_destroy(cqv);
end_test(&f);
config.tear_down_data(&f);
diff --git a/test/core/end2end/tests/early_server_shutdown_finishes_inflight_calls.c b/test/core/end2end/tests/early_server_shutdown_finishes_inflight_calls.c
index 42280a6..3efbf6a 100644
--- a/test/core/end2end/tests/early_server_shutdown_finishes_inflight_calls.c
+++ b/test/core/end2end/tests/early_server_shutdown_finishes_inflight_calls.c
@@ -89,12 +89,9 @@
shutdown_server(f);
shutdown_client(f);
- grpc_completion_queue_shutdown(f->server_cq);
- drain_cq(f->server_cq);
- grpc_completion_queue_destroy(f->server_cq);
- grpc_completion_queue_shutdown(f->client_cq);
- drain_cq(f->client_cq);
- grpc_completion_queue_destroy(f->client_cq);
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
}
static void test_early_server_shutdown_finishes_inflight_calls(
@@ -103,8 +100,7 @@
grpc_call *s;
gpr_timespec deadline = five_seconds_time();
grpc_end2end_test_fixture f = begin_test(config, __FUNCTION__, NULL, NULL);
- cq_verifier *v_client = cq_verifier_create(f.client_cq);
- cq_verifier *v_server = cq_verifier_create(f.server_cq);
+ cq_verifier *cqv = cq_verifier_create(f.cq);
grpc_op ops[6];
grpc_op *op;
grpc_metadata_array initial_metadata_recv;
@@ -116,7 +112,7 @@
size_t details_capacity = 0;
int was_cancelled = 2;
- c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
+ c = grpc_channel_create_call(f.client, f.cq, "/foo",
"foo.test.google.fr", deadline);
GPR_ASSERT(c);
@@ -145,10 +141,10 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_server_request_call(f.server, &s, &call_details,
- &request_metadata_recv, f.server_cq,
- f.server_cq, tag(101)));
- cq_expect_completion(v_server, tag(101), 1);
- cq_verify(v_server);
+ &request_metadata_recv, f.cq,
+ f.cq, tag(101)));
+ cq_expect_completion(cqv, tag(101), 1);
+ cq_verify(cqv);
op = ops;
op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
@@ -159,11 +155,9 @@
/* shutdown and destroy the server */
shutdown_server(&f);
- cq_expect_completion(v_server, tag(102), 1);
- cq_verify(v_server);
-
- cq_expect_completion(v_client, tag(1), 1);
- cq_verify(v_client);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
+ cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_UNAVAILABLE);
GPR_ASSERT(0 == strcmp(call_details.method, "/foo"));
@@ -179,8 +173,7 @@
grpc_call_destroy(c);
grpc_call_destroy(s);
- cq_verifier_destroy(v_client);
- cq_verifier_destroy(v_server);
+ cq_verifier_destroy(cqv);
end_test(&f);
config.tear_down_data(&f);
diff --git a/test/core/end2end/tests/early_server_shutdown_finishes_tags.c b/test/core/end2end/tests/early_server_shutdown_finishes_tags.c
index 857fbb3..c52e54b 100644
--- a/test/core/end2end/tests/early_server_shutdown_finishes_tags.c
+++ b/test/core/end2end/tests/early_server_shutdown_finishes_tags.c
@@ -89,18 +89,15 @@
shutdown_server(f);
shutdown_client(f);
- grpc_completion_queue_shutdown(f->server_cq);
- drain_cq(f->server_cq);
- grpc_completion_queue_destroy(f->server_cq);
- grpc_completion_queue_shutdown(f->client_cq);
- drain_cq(f->client_cq);
- grpc_completion_queue_destroy(f->client_cq);
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
}
static void test_early_server_shutdown_finishes_tags(
grpc_end2end_test_config config) {
grpc_end2end_test_fixture f = begin_test(config, __FUNCTION__, NULL, NULL);
- cq_verifier *v_server = cq_verifier_create(f.server_cq);
+ cq_verifier *cqv = cq_verifier_create(f.cq);
grpc_call *s = (void *)1;
grpc_call_details call_details;
grpc_metadata_array request_metadata_recv;
@@ -112,16 +109,16 @@
no new call */
GPR_ASSERT(GRPC_CALL_OK ==
grpc_server_request_call(f.server, &s, &call_details,
- &request_metadata_recv, f.server_cq,
- f.server_cq, tag(101)));
+ &request_metadata_recv, f.cq,
+ f.cq, tag(101)));
grpc_server_shutdown(f.server);
- cq_expect_completion(v_server, tag(101), 0);
- cq_verify(v_server);
+ cq_expect_completion(cqv, tag(101), 0);
+ cq_verify(cqv);
GPR_ASSERT(s == NULL);
end_test(&f);
config.tear_down_data(&f);
- cq_verifier_destroy(v_server);
+ cq_verifier_destroy(cqv);
}
void grpc_end2end_tests(grpc_end2end_test_config config) {
diff --git a/test/core/end2end/tests/empty_batch.c b/test/core/end2end/tests/empty_batch.c
index a57f921..f1c3fb2 100644
--- a/test/core/end2end/tests/empty_batch.c
+++ b/test/core/end2end/tests/empty_batch.c
@@ -91,31 +91,28 @@
shutdown_server(f);
shutdown_client(f);
- grpc_completion_queue_shutdown(f->server_cq);
- drain_cq(f->server_cq);
- grpc_completion_queue_destroy(f->server_cq);
- grpc_completion_queue_shutdown(f->client_cq);
- drain_cq(f->client_cq);
- grpc_completion_queue_destroy(f->client_cq);
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
}
static void empty_batch_body(grpc_end2end_test_fixture f) {
grpc_call *c;
gpr_timespec deadline = five_seconds_time();
- cq_verifier *v_client = cq_verifier_create(f.client_cq);
+ cq_verifier *cqv = cq_verifier_create(f.cq);
grpc_op *op = NULL;
- c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
+ c = grpc_channel_create_call(f.client, f.cq, "/foo",
"foo.test.google.fr", deadline);
GPR_ASSERT(c);
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, op, 0, tag(1)));
- cq_expect_completion(v_client, tag(1), 1);
- cq_verify(v_client);
+ cq_expect_completion(cqv, tag(1), 1);
+ cq_verify(cqv);
grpc_call_destroy(c);
- cq_verifier_destroy(v_client);
+ cq_verifier_destroy(cqv);
}
static void test_invoke_empty_body(grpc_end2end_test_config config) {
diff --git a/test/core/end2end/tests/graceful_server_shutdown.c b/test/core/end2end/tests/graceful_server_shutdown.c
index 9c44f6e..70a0655 100644
--- a/test/core/end2end/tests/graceful_server_shutdown.c
+++ b/test/core/end2end/tests/graceful_server_shutdown.c
@@ -88,12 +88,9 @@
shutdown_server(f);
shutdown_client(f);
- grpc_completion_queue_shutdown(f->server_cq);
- drain_cq(f->server_cq);
- grpc_completion_queue_destroy(f->server_cq);
- grpc_completion_queue_shutdown(f->client_cq);
- drain_cq(f->client_cq);
- grpc_completion_queue_destroy(f->client_cq);
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
}
static void test_early_server_shutdown_finishes_inflight_calls(
@@ -102,8 +99,7 @@
grpc_call *s;
gpr_timespec deadline = five_seconds_time();
grpc_end2end_test_fixture f = begin_test(config, __FUNCTION__, NULL, NULL);
- cq_verifier *v_client = cq_verifier_create(f.client_cq);
- cq_verifier *v_server = cq_verifier_create(f.server_cq);
+ cq_verifier *cqv = cq_verifier_create(f.cq);
grpc_op ops[6];
grpc_op *op;
grpc_metadata_array initial_metadata_recv;
@@ -115,7 +111,7 @@
size_t details_capacity = 0;
int was_cancelled = 2;
- c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
+ c = grpc_channel_create_call(f.client, f.cq, "/foo",
"foo.test.google.fr", deadline);
GPR_ASSERT(c);
@@ -144,14 +140,14 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_server_request_call(f.server, &s, &call_details,
- &request_metadata_recv, f.server_cq,
- f.server_cq, tag(101)));
- cq_expect_completion(v_server, tag(101), 1);
- cq_verify(v_server);
+ &request_metadata_recv, f.cq,
+ f.cq, tag(101)));
+ cq_expect_completion(cqv, tag(101), 1);
+ cq_verify(cqv);
/* shutdown and destroy the server */
grpc_server_shutdown_and_notify(f.server, tag(0xdead));
- cq_verify_empty(v_server);
+ cq_verify_empty(cqv);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
@@ -167,15 +163,13 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(v_server, tag(102), 1);
- cq_verify(v_server);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_verify(cqv);
grpc_call_destroy(s);
- cq_expect_completion(v_server, tag(0xdead), 1);
- cq_verify(v_server);
-
- cq_expect_completion(v_client, tag(1), 1);
- cq_verify(v_client);
+ cq_expect_completion(cqv, tag(0xdead), 1);
+ cq_expect_completion(cqv, tag(1), 1);
+ cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
GPR_ASSERT(0 == strcmp(call_details.method, "/foo"));
@@ -190,8 +184,7 @@
grpc_call_destroy(c);
- cq_verifier_destroy(v_client);
- cq_verifier_destroy(v_server);
+ cq_verifier_destroy(cqv);
end_test(&f);
config.tear_down_data(&f);
diff --git a/test/core/end2end/tests/invoke_large_request.c b/test/core/end2end/tests/invoke_large_request.c
index f369e78..d4035d0 100644
--- a/test/core/end2end/tests/invoke_large_request.c
+++ b/test/core/end2end/tests/invoke_large_request.c
@@ -87,12 +87,9 @@
shutdown_server(f);
shutdown_client(f);
- grpc_completion_queue_shutdown(f->server_cq);
- drain_cq(f->server_cq);
- grpc_completion_queue_destroy(f->server_cq);
- grpc_completion_queue_shutdown(f->client_cq);
- drain_cq(f->client_cq);
- grpc_completion_queue_destroy(f->client_cq);
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
}
static gpr_slice large_slice(void) {
@@ -113,8 +110,7 @@
grpc_byte_buffer *response_payload =
grpc_byte_buffer_create(&response_payload_slice, 1);
gpr_timespec deadline = n_seconds_time(30);
- cq_verifier *v_client = cq_verifier_create(f.client_cq);
- cq_verifier *v_server = cq_verifier_create(f.server_cq);
+ cq_verifier *cqv = cq_verifier_create(f.cq);
grpc_op ops[6];
grpc_op *op;
grpc_metadata_array initial_metadata_recv;
@@ -128,7 +124,7 @@
size_t details_capacity = 0;
int was_cancelled = 2;
- c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
+ c = grpc_channel_create_call(f.client, f.cq, "/foo",
"foo.test.google.fr", deadline);
GPR_ASSERT(c);
@@ -162,10 +158,10 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_server_request_call(f.server, &s, &call_details,
- &request_metadata_recv, f.server_cq,
- f.server_cq, tag(101)));
- cq_expect_completion(v_server, tag(101), 1);
- cq_verify(v_server);
+ &request_metadata_recv, f.cq,
+ f.cq, tag(101)));
+ cq_expect_completion(cqv, tag(101), 1);
+ cq_verify(cqv);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
@@ -187,11 +183,9 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(v_server, tag(102), 1);
- cq_verify(v_server);
-
- cq_expect_completion(v_client, tag(1), 1);
- cq_verify(v_client);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
+ cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
GPR_ASSERT(0 == strcmp(details, "xyz"));
@@ -208,8 +202,7 @@
grpc_call_destroy(c);
grpc_call_destroy(s);
- cq_verifier_destroy(v_client);
- cq_verifier_destroy(v_server);
+ cq_verifier_destroy(cqv);
grpc_byte_buffer_destroy(request_payload);
grpc_byte_buffer_destroy(response_payload);
diff --git a/test/core/end2end/tests/max_concurrent_streams.c b/test/core/end2end/tests/max_concurrent_streams.c
index 59227d9..57300a0 100644
--- a/test/core/end2end/tests/max_concurrent_streams.c
+++ b/test/core/end2end/tests/max_concurrent_streams.c
@@ -89,20 +89,16 @@
shutdown_server(f);
shutdown_client(f);
- grpc_completion_queue_shutdown(f->server_cq);
- drain_cq(f->server_cq);
- grpc_completion_queue_destroy(f->server_cq);
- grpc_completion_queue_shutdown(f->client_cq);
- drain_cq(f->client_cq);
- grpc_completion_queue_destroy(f->client_cq);
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
}
static void simple_request_body(grpc_end2end_test_fixture f) {
grpc_call *c;
grpc_call *s;
gpr_timespec deadline = five_seconds_time();
- cq_verifier *v_client = cq_verifier_create(f.client_cq);
- cq_verifier *v_server = cq_verifier_create(f.server_cq);
+ cq_verifier *cqv = cq_verifier_create(f.cq);
grpc_op ops[6];
grpc_op *op;
grpc_metadata_array initial_metadata_recv;
@@ -114,7 +110,7 @@
size_t details_capacity = 0;
int was_cancelled = 2;
- c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
+ c = grpc_channel_create_call(f.client, f.cq, "/foo",
"foo.test.google.fr:1234", deadline);
GPR_ASSERT(c);
@@ -142,10 +138,10 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_server_request_call(f.server, &s, &call_details,
- &request_metadata_recv, f.server_cq,
- f.server_cq, tag(101)));
- cq_expect_completion(v_server, tag(101), 1);
- cq_verify(v_server);
+ &request_metadata_recv, f.cq,
+ f.cq, tag(101)));
+ cq_expect_completion(cqv, tag(101), 1);
+ cq_verify(cqv);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
@@ -161,11 +157,9 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(v_server, tag(102), 1);
- cq_verify(v_server);
-
- cq_expect_completion(v_client, tag(1), 1);
- cq_verify(v_client);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
+ cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
GPR_ASSERT(0 == strcmp(details, "xyz"));
@@ -182,8 +176,7 @@
grpc_call_destroy(c);
grpc_call_destroy(s);
- cq_verifier_destroy(v_client);
- cq_verifier_destroy(v_server);
+ cq_verifier_destroy(cqv);
}
static void test_max_concurrent_streams(grpc_end2end_test_config config) {
@@ -196,8 +189,7 @@
grpc_call *s2;
int live_call;
gpr_timespec deadline;
- cq_verifier *v_client;
- cq_verifier *v_server;
+ cq_verifier *cqv;
grpc_event ev;
grpc_call_details call_details;
grpc_metadata_array request_metadata_recv;
@@ -214,6 +206,8 @@
grpc_op ops[6];
grpc_op *op;
int was_cancelled;
+ int got_client_start;
+ int got_server_start;
server_arg.key = GRPC_ARG_MAX_CONCURRENT_STREAMS;
server_arg.type = GRPC_ARG_INTEGER;
@@ -223,8 +217,7 @@
server_args.args = &server_arg;
f = begin_test(config, __FUNCTION__, NULL, &server_args);
- v_client = cq_verifier_create(f.client_cq);
- v_server = cq_verifier_create(f.server_cq);
+ cqv = cq_verifier_create(f.cq);
grpc_metadata_array_init(&request_metadata_recv);
grpc_metadata_array_init(&initial_metadata_recv1);
@@ -241,18 +234,18 @@
/* start two requests - ensuring that the second is not accepted until
the first completes */
- deadline = n_seconds_time(10);
- c1 = grpc_channel_create_call(f.client, f.client_cq, "/alpha",
+ deadline = n_seconds_time(1000);
+ c1 = grpc_channel_create_call(f.client, f.cq, "/alpha",
"foo.test.google.fr:1234", deadline);
GPR_ASSERT(c1);
- c2 = grpc_channel_create_call(f.client, f.client_cq, "/beta",
+ c2 = grpc_channel_create_call(f.client, f.cq, "/beta",
"foo.test.google.fr:1234", deadline);
GPR_ASSERT(c2);
GPR_ASSERT(GRPC_CALL_OK ==
grpc_server_request_call(f.server, &s1, &call_details,
- &request_metadata_recv, f.server_cq,
- f.server_cq, tag(101)));
+ &request_metadata_recv, f.cq,
+ f.cq, tag(101)));
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
@@ -298,18 +291,28 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_batch(c2, ops, op - ops, tag(402)));
- cq_expect_completion(v_server, tag(101), 1);
- cq_verify(v_server);
-
- ev = grpc_completion_queue_next(f.client_cq,
- GRPC_TIMEOUT_SECONDS_TO_DEADLINE(3));
- GPR_ASSERT(ev.type == GRPC_OP_COMPLETE);
- GPR_ASSERT(ev.success);
- GPR_ASSERT(ev.tag == tag(301) || ev.tag == tag(401));
- /* The /alpha or /beta calls started above could be invoked (but NOT both);
- * check this here */
- /* We'll get tag 303 or 403, we want 300, 400 */
- live_call = ((int)(gpr_intptr)ev.tag) - 1;
+ got_client_start = 0;
+ got_server_start = 0;
+ live_call = -1;
+ while (!got_client_start || !got_server_start) {
+ ev = grpc_completion_queue_next(f.cq,
+ GRPC_TIMEOUT_SECONDS_TO_DEADLINE(3));
+ GPR_ASSERT(ev.type == GRPC_OP_COMPLETE);
+ GPR_ASSERT(ev.success);
+ if (ev.tag == tag(101)) {
+ GPR_ASSERT(!got_server_start);
+ got_server_start = 1;
+ } else {
+ GPR_ASSERT(!got_client_start);
+ GPR_ASSERT(ev.tag == tag(301) || ev.tag == tag(401));
+ /* The /alpha or /beta calls started above could be invoked (but NOT both);
+ * check this here */
+ /* We'll get tag 303 or 403, we want 300, 400 */
+ live_call = ((int)(gpr_intptr)ev.tag) - 1;
+ got_client_start = 1;
+ }
+ }
+ GPR_ASSERT(live_call == 300 || live_call == 400);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
@@ -326,21 +329,19 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_batch(s1, ops, op - ops, tag(102)));
- cq_expect_completion(v_server, tag(102), 1);
- cq_verify(v_server);
-
- cq_expect_completion(v_client, tag(live_call + 2), 1);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(live_call + 2), 1);
/* first request is finished, we should be able to start the second */
live_call = (live_call == 300) ? 400 : 300;
- cq_expect_completion(v_client, tag(live_call + 1), 1);
- cq_verify(v_client);
+ cq_expect_completion(cqv, tag(live_call + 1), 1);
+ cq_verify(cqv);
GPR_ASSERT(GRPC_CALL_OK ==
grpc_server_request_call(f.server, &s2, &call_details,
- &request_metadata_recv, f.server_cq,
- f.server_cq, tag(201)));
- cq_expect_completion(v_server, tag(201), 1);
- cq_verify(v_server);
+ &request_metadata_recv, f.cq,
+ f.cq, tag(201)));
+ cq_expect_completion(cqv, tag(201), 1);
+ cq_verify(cqv);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
@@ -357,14 +358,11 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_batch(s2, ops, op - ops, tag(202)));
- cq_expect_completion(v_client, tag(live_call + 2), 1);
- cq_verify(v_client);
+ cq_expect_completion(cqv, tag(live_call + 2), 1);
+ cq_expect_completion(cqv, tag(202), 1);
+ cq_verify(cqv);
- cq_expect_completion(v_server, tag(202), 1);
- cq_verify(v_server);
-
- cq_verifier_destroy(v_client);
- cq_verifier_destroy(v_server);
+ cq_verifier_destroy(cqv);
grpc_call_destroy(c1);
grpc_call_destroy(s1);
diff --git a/test/core/end2end/tests/max_message_length.c b/test/core/end2end/tests/max_message_length.c
index 99aeb9d..364d59e 100644
--- a/test/core/end2end/tests/max_message_length.c
+++ b/test/core/end2end/tests/max_message_length.c
@@ -89,12 +89,9 @@
shutdown_server(f);
shutdown_client(f);
- grpc_completion_queue_shutdown(f->server_cq);
- drain_cq(f->server_cq);
- grpc_completion_queue_destroy(f->server_cq);
- grpc_completion_queue_shutdown(f->client_cq);
- drain_cq(f->client_cq);
- grpc_completion_queue_destroy(f->client_cq);
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
}
static void test_max_message_length(grpc_end2end_test_config config) {
@@ -103,8 +100,7 @@
grpc_channel_args server_args;
grpc_call *c;
grpc_call *s;
- cq_verifier *v_client;
- cq_verifier *v_server;
+ cq_verifier *cqv;
grpc_op ops[6];
grpc_op *op;
gpr_slice request_payload_slice = gpr_slice_from_copied_string("hello world");
@@ -127,10 +123,9 @@
server_args.args = &server_arg;
f = begin_test(config, __FUNCTION__, NULL, &server_args);
- v_client = cq_verifier_create(f.client_cq);
- v_server = cq_verifier_create(f.server_cq);
+ cqv = cq_verifier_create(f.cq);
- c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
+ c = grpc_channel_create_call(f.client, f.cq, "/foo",
"foo.test.google.fr:1234", gpr_inf_future);
GPR_ASSERT(c);
@@ -161,10 +156,10 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_server_request_call(f.server, &s, &call_details,
- &request_metadata_recv, f.server_cq,
- f.server_cq, tag(101)));
- cq_expect_completion(v_server, tag(101), 1);
- cq_verify(v_server);
+ &request_metadata_recv, f.cq,
+ f.cq, tag(101)));
+ cq_expect_completion(cqv, tag(101), 1);
+ cq_verify(cqv);
op = ops;
op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
@@ -172,11 +167,9 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(v_server, tag(102), 1);
- cq_verify(v_server);
-
- cq_expect_completion(v_client, tag(1), 1);
- cq_verify(v_client);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
+ cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_CANCELLED);
GPR_ASSERT(0 == strcmp(details, "Cancelled"));
@@ -193,8 +186,7 @@
grpc_call_destroy(c);
grpc_call_destroy(s);
- cq_verifier_destroy(v_client);
- cq_verifier_destroy(v_server);
+ cq_verifier_destroy(cqv);
end_test(&f);
config.tear_down_data(&f);
diff --git a/test/core/end2end/tests/no_op.c b/test/core/end2end/tests/no_op.c
index 000d2ff..979eb06 100644
--- a/test/core/end2end/tests/no_op.c
+++ b/test/core/end2end/tests/no_op.c
@@ -87,12 +87,9 @@
shutdown_server(f);
shutdown_client(f);
- grpc_completion_queue_shutdown(f->server_cq);
- drain_cq(f->server_cq);
- grpc_completion_queue_destroy(f->server_cq);
- grpc_completion_queue_shutdown(f->client_cq);
- drain_cq(f->client_cq);
- grpc_completion_queue_destroy(f->client_cq);
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
}
static void test_no_op(grpc_end2end_test_config config) {
diff --git a/test/core/end2end/tests/ping_pong_streaming.c b/test/core/end2end/tests/ping_pong_streaming.c
index 2bd1792..c00beb7 100644
--- a/test/core/end2end/tests/ping_pong_streaming.c
+++ b/test/core/end2end/tests/ping_pong_streaming.c
@@ -89,12 +89,9 @@
shutdown_server(f);
shutdown_client(f);
- grpc_completion_queue_shutdown(f->server_cq);
- drain_cq(f->server_cq);
- grpc_completion_queue_destroy(f->server_cq);
- grpc_completion_queue_shutdown(f->client_cq);
- drain_cq(f->client_cq);
- grpc_completion_queue_destroy(f->client_cq);
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
}
/* Client pings and server pongs. Repeat messages rounds before finishing. */
@@ -104,8 +101,7 @@
grpc_call *c;
grpc_call *s;
gpr_timespec deadline = five_seconds_time();
- cq_verifier *v_client = cq_verifier_create(f.client_cq);
- cq_verifier *v_server = cq_verifier_create(f.server_cq);
+ cq_verifier *cqv = cq_verifier_create(f.cq);
grpc_op ops[6];
grpc_op *op;
grpc_metadata_array initial_metadata_recv;
@@ -124,7 +120,7 @@
gpr_slice request_payload_slice = gpr_slice_from_copied_string("hello world");
gpr_slice response_payload_slice = gpr_slice_from_copied_string("hello you");
- c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
+ c = grpc_channel_create_call(f.client, f.cq, "/foo",
"foo.test.google.fr:1234", deadline);
GPR_ASSERT(c);
@@ -150,10 +146,10 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_server_request_call(f.server, &s, &call_details,
- &request_metadata_recv, f.server_cq,
- f.server_cq, tag(100)));
- cq_expect_completion(v_server, tag(100), 1);
- cq_verify(v_server);
+ &request_metadata_recv, f.cq,
+ f.cq, tag(100)));
+ cq_expect_completion(cqv, tag(100), 1);
+ cq_verify(cqv);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
@@ -183,8 +179,8 @@
op++;
GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(v_server, tag(102), 1);
- cq_verify(v_server);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_verify(cqv);
op = ops;
op->op = GRPC_OP_SEND_MESSAGE;
@@ -192,11 +188,9 @@
op++;
GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_batch(s, ops, op - ops, tag(103)));
- cq_expect_completion(v_server, tag(103), 1);
- cq_verify(v_server);
-
- cq_expect_completion(v_client, tag(2), 1);
- cq_verify(v_client);
+ cq_expect_completion(cqv, tag(103), 1);
+ cq_expect_completion(cqv, tag(2), 1);
+ cq_verify(cqv);
grpc_byte_buffer_destroy(request_payload);
grpc_byte_buffer_destroy(response_payload);
@@ -220,19 +214,16 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(104)));
- cq_expect_completion(v_client, tag(1), 1);
- cq_expect_completion(v_client, tag(3), 1);
- cq_verify(v_client);
-
- cq_expect_completion(v_server, tag(101), 1);
- cq_expect_completion(v_server, tag(104), 1);
- cq_verify(v_server);
+ cq_expect_completion(cqv, tag(1), 1);
+ cq_expect_completion(cqv, tag(3), 1);
+ cq_expect_completion(cqv, tag(101), 1);
+ cq_expect_completion(cqv, tag(104), 1);
+ cq_verify(cqv);
grpc_call_destroy(c);
grpc_call_destroy(s);
- cq_verifier_destroy(v_client);
- cq_verifier_destroy(v_server);
+ cq_verifier_destroy(cqv);
grpc_metadata_array_destroy(&initial_metadata_recv);
grpc_metadata_array_destroy(&trailing_metadata_recv);
diff --git a/test/core/end2end/tests/registered_call.c b/test/core/end2end/tests/registered_call.c
index 54663c3..80f7d20 100644
--- a/test/core/end2end/tests/registered_call.c
+++ b/test/core/end2end/tests/registered_call.c
@@ -91,20 +91,16 @@
shutdown_server(f);
shutdown_client(f);
- grpc_completion_queue_shutdown(f->server_cq);
- drain_cq(f->server_cq);
- grpc_completion_queue_destroy(f->server_cq);
- grpc_completion_queue_shutdown(f->client_cq);
- drain_cq(f->client_cq);
- grpc_completion_queue_destroy(f->client_cq);
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
}
static void simple_request_body(grpc_end2end_test_fixture f, void *rc) {
grpc_call *c;
grpc_call *s;
gpr_timespec deadline = five_seconds_time();
- cq_verifier *v_client = cq_verifier_create(f.client_cq);
- cq_verifier *v_server = cq_verifier_create(f.server_cq);
+ cq_verifier *cqv = cq_verifier_create(f.cq);
grpc_op ops[6];
grpc_op *op;
grpc_metadata_array initial_metadata_recv;
@@ -116,7 +112,7 @@
size_t details_capacity = 0;
int was_cancelled = 2;
- c = grpc_channel_create_registered_call(f.client, f.client_cq, rc, deadline);
+ c = grpc_channel_create_registered_call(f.client, f.cq, rc, deadline);
GPR_ASSERT(c);
grpc_metadata_array_init(&initial_metadata_recv);
@@ -143,10 +139,10 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_server_request_call(f.server, &s, &call_details,
- &request_metadata_recv, f.server_cq,
- f.server_cq, tag(101)));
- cq_expect_completion(v_server, tag(101), 1);
- cq_verify(v_server);
+ &request_metadata_recv, f.cq,
+ f.cq, tag(101)));
+ cq_expect_completion(cqv, tag(101), 1);
+ cq_verify(cqv);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
@@ -162,11 +158,9 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(v_server, tag(102), 1);
- cq_verify(v_server);
-
- cq_expect_completion(v_client, tag(1), 1);
- cq_verify(v_client);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
+ cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
GPR_ASSERT(0 == strcmp(details, "xyz"));
@@ -183,8 +177,7 @@
grpc_call_destroy(c);
grpc_call_destroy(s);
- cq_verifier_destroy(v_client);
- cq_verifier_destroy(v_server);
+ cq_verifier_destroy(cqv);
}
static void test_invoke_simple_request(grpc_end2end_test_config config) {
diff --git a/test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c b/test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c
index 709dc47..ce872a4 100644
--- a/test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c
+++ b/test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c
@@ -89,12 +89,9 @@
shutdown_server(f);
shutdown_client(f);
- grpc_completion_queue_shutdown(f->server_cq);
- drain_cq(f->server_cq);
- grpc_completion_queue_destroy(f->server_cq);
- grpc_completion_queue_shutdown(f->client_cq);
- drain_cq(f->client_cq);
- grpc_completion_queue_destroy(f->client_cq);
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
}
/* Request/response with metadata and payload.*/
@@ -128,8 +125,7 @@
16,
{{NULL, NULL, NULL}}}};
grpc_end2end_test_fixture f = begin_test(config, __FUNCTION__, NULL, NULL);
- cq_verifier *v_client = cq_verifier_create(f.client_cq);
- cq_verifier *v_server = cq_verifier_create(f.server_cq);
+ cq_verifier *cqv = cq_verifier_create(f.cq);
grpc_op ops[6];
grpc_op *op;
grpc_metadata_array initial_metadata_recv;
@@ -143,7 +139,7 @@
size_t details_capacity = 0;
int was_cancelled = 2;
- c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
+ c = grpc_channel_create_call(f.client, f.cq, "/foo",
"foo.test.google.fr", deadline);
GPR_ASSERT(c);
@@ -178,10 +174,10 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_server_request_call(f.server, &s, &call_details,
- &request_metadata_recv, f.server_cq,
- f.server_cq, tag(101)));
- cq_expect_completion(v_server, tag(101), 1);
- cq_verify(v_server);
+ &request_metadata_recv, f.cq,
+ f.cq, tag(101)));
+ cq_expect_completion(cqv, tag(101), 1);
+ cq_verify(cqv);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
@@ -204,11 +200,9 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(v_server, tag(102), 1);
- cq_verify(v_server);
-
- cq_expect_completion(v_client, tag(1), 1);
- cq_verify(v_client);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
+ cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_OK);
GPR_ASSERT(0 == strcmp(details, "xyz"));
@@ -239,8 +233,7 @@
grpc_call_destroy(c);
grpc_call_destroy(s);
- cq_verifier_destroy(v_client);
- cq_verifier_destroy(v_server);
+ cq_verifier_destroy(cqv);
grpc_byte_buffer_destroy(request_payload);
grpc_byte_buffer_destroy(response_payload);
diff --git a/test/core/end2end/tests/request_response_with_metadata_and_payload.c b/test/core/end2end/tests/request_response_with_metadata_and_payload.c
index bc32a50..b4edb73 100644
--- a/test/core/end2end/tests/request_response_with_metadata_and_payload.c
+++ b/test/core/end2end/tests/request_response_with_metadata_and_payload.c
@@ -89,12 +89,9 @@
shutdown_server(f);
shutdown_client(f);
- grpc_completion_queue_shutdown(f->server_cq);
- drain_cq(f->server_cq);
- grpc_completion_queue_destroy(f->server_cq);
- grpc_completion_queue_shutdown(f->client_cq);
- drain_cq(f->client_cq);
- grpc_completion_queue_destroy(f->client_cq);
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
}
/* Request/response with metadata and payload.*/
@@ -114,8 +111,7 @@
grpc_metadata meta_s[2] = {{"key3", "val3", 4, {{NULL, NULL, NULL}}},
{"key4", "val4", 4, {{NULL, NULL, NULL}}}};
grpc_end2end_test_fixture f = begin_test(config, __FUNCTION__, NULL, NULL);
- cq_verifier *v_client = cq_verifier_create(f.client_cq);
- cq_verifier *v_server = cq_verifier_create(f.server_cq);
+ cq_verifier *cqv = cq_verifier_create(f.cq);
grpc_op ops[6];
grpc_op *op;
grpc_metadata_array initial_metadata_recv;
@@ -129,7 +125,7 @@
size_t details_capacity = 0;
int was_cancelled = 2;
- c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
+ c = grpc_channel_create_call(f.client, f.cq, "/foo",
"foo.test.google.fr", deadline);
GPR_ASSERT(c);
@@ -164,10 +160,10 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_server_request_call(f.server, &s, &call_details,
- &request_metadata_recv, f.server_cq,
- f.server_cq, tag(101)));
- cq_expect_completion(v_server, tag(101), 1);
- cq_verify(v_server);
+ &request_metadata_recv, f.cq,
+ f.cq, tag(101)));
+ cq_expect_completion(cqv, tag(101), 1);
+ cq_verify(cqv);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
@@ -190,11 +186,9 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(v_server, tag(102), 1);
- cq_verify(v_server);
-
- cq_expect_completion(v_client, tag(1), 1);
- cq_verify(v_client);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
+ cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_OK);
GPR_ASSERT(0 == strcmp(details, "xyz"));
@@ -217,8 +211,7 @@
grpc_call_destroy(c);
grpc_call_destroy(s);
- cq_verifier_destroy(v_client);
- cq_verifier_destroy(v_server);
+ cq_verifier_destroy(cqv);
grpc_byte_buffer_destroy(request_payload);
grpc_byte_buffer_destroy(response_payload);
diff --git a/test/core/end2end/tests/request_response_with_payload.c b/test/core/end2end/tests/request_response_with_payload.c
index be0cca6..cb50855 100644
--- a/test/core/end2end/tests/request_response_with_payload.c
+++ b/test/core/end2end/tests/request_response_with_payload.c
@@ -89,12 +89,9 @@
shutdown_server(f);
shutdown_client(f);
- grpc_completion_queue_shutdown(f->server_cq);
- drain_cq(f->server_cq);
- grpc_completion_queue_destroy(f->server_cq);
- grpc_completion_queue_shutdown(f->client_cq);
- drain_cq(f->client_cq);
- grpc_completion_queue_destroy(f->client_cq);
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
}
static void request_response_with_payload(grpc_end2end_test_fixture f) {
@@ -107,8 +104,7 @@
grpc_byte_buffer *response_payload =
grpc_byte_buffer_create(&response_payload_slice, 1);
gpr_timespec deadline = five_seconds_time();
- cq_verifier *v_client = cq_verifier_create(f.client_cq);
- cq_verifier *v_server = cq_verifier_create(f.server_cq);
+ cq_verifier *cqv = cq_verifier_create(f.cq);
grpc_op ops[6];
grpc_op *op;
grpc_metadata_array initial_metadata_recv;
@@ -122,7 +118,7 @@
size_t details_capacity = 0;
int was_cancelled = 2;
- c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
+ c = grpc_channel_create_call(f.client, f.cq, "/foo",
"foo.test.google.fr", deadline);
GPR_ASSERT(c);
@@ -156,10 +152,10 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_server_request_call(f.server, &s, &call_details,
- &request_metadata_recv, f.server_cq,
- f.server_cq, tag(101)));
- cq_expect_completion(v_server, tag(101), 1);
- cq_verify(v_server);
+ &request_metadata_recv, f.cq,
+ f.cq, tag(101)));
+ cq_expect_completion(cqv, tag(101), 1);
+ cq_verify(cqv);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
@@ -181,11 +177,9 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(v_server, tag(102), 1);
- cq_verify(v_server);
-
- cq_expect_completion(v_client, tag(1), 1);
- cq_verify(v_client);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
+ cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_OK);
GPR_ASSERT(0 == strcmp(details, "xyz"));
@@ -204,8 +198,7 @@
grpc_call_destroy(c);
grpc_call_destroy(s);
- cq_verifier_destroy(v_client);
- cq_verifier_destroy(v_server);
+ cq_verifier_destroy(cqv);
grpc_byte_buffer_destroy(request_payload);
grpc_byte_buffer_destroy(response_payload);
diff --git a/test/core/end2end/tests/request_response_with_payload_and_call_creds.c b/test/core/end2end/tests/request_response_with_payload_and_call_creds.c
index 447b20d..ec3f310 100644
--- a/test/core/end2end/tests/request_response_with_payload_and_call_creds.c
+++ b/test/core/end2end/tests/request_response_with_payload_and_call_creds.c
@@ -103,12 +103,9 @@
shutdown_server(f);
shutdown_client(f);
- grpc_completion_queue_shutdown(f->server_cq);
- drain_cq(f->server_cq);
- grpc_completion_queue_destroy(f->server_cq);
- grpc_completion_queue_shutdown(f->client_cq);
- drain_cq(f->client_cq);
- grpc_completion_queue_destroy(f->client_cq);
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
}
static void test_call_creds_failure(grpc_end2end_test_config config) {
@@ -116,7 +113,7 @@
grpc_credentials *creds = NULL;
grpc_end2end_test_fixture f = begin_test(config, __FUNCTION__, NULL, NULL);
gpr_timespec deadline = five_seconds_time();
- c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
+ c = grpc_channel_create_call(f.client, f.cq, "/foo",
"foo.test.google.fr", deadline);
GPR_ASSERT(c);
@@ -143,8 +140,7 @@
gpr_timespec deadline = five_seconds_time();
grpc_end2end_test_fixture f = begin_test(config, test_name, NULL, NULL);
- cq_verifier *v_client = cq_verifier_create(f.client_cq);
- cq_verifier *v_server = cq_verifier_create(f.server_cq);
+ cq_verifier *cqv = cq_verifier_create(f.cq);
grpc_op ops[6];
grpc_op *op;
grpc_metadata_array initial_metadata_recv;
@@ -159,7 +155,7 @@
int was_cancelled = 2;
grpc_credentials *creds = NULL;
- c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
+ c = grpc_channel_create_call(f.client, f.cq, "/foo",
"foo.test.google.fr", deadline);
GPR_ASSERT(c);
creds = grpc_iam_credentials_create(iam_token, iam_selector);
@@ -212,10 +208,10 @@
GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
&call_details,
&request_metadata_recv,
- f.server_cq, f.server_cq,
+ f.cq, f.cq,
tag(101)));
- cq_expect_completion(v_server, tag(101), 1);
- cq_verify(v_server);
+ cq_expect_completion(cqv, tag(101), 1);
+ cq_verify(cqv);
/* Cannot set creds on the server call object. */
GPR_ASSERT(!grpc_call_set_credentials(s, NULL));
@@ -240,11 +236,9 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(v_server, tag(102), 1);
- cq_verify(v_server);
-
- cq_expect_completion(v_client, tag(1), 1);
- cq_verify(v_client);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
+ cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_OK);
GPR_ASSERT(0 == strcmp(details, "xyz"));
@@ -296,8 +290,7 @@
grpc_call_destroy(c);
grpc_call_destroy(s);
- cq_verifier_destroy(v_client);
- cq_verifier_destroy(v_server);
+ cq_verifier_destroy(cqv);
grpc_byte_buffer_destroy(request_payload);
grpc_byte_buffer_destroy(response_payload);
diff --git a/test/core/end2end/tests/request_with_large_metadata.c b/test/core/end2end/tests/request_with_large_metadata.c
index 08a1621..a137786 100644
--- a/test/core/end2end/tests/request_with_large_metadata.c
+++ b/test/core/end2end/tests/request_with_large_metadata.c
@@ -89,12 +89,9 @@
shutdown_server(f);
shutdown_client(f);
- grpc_completion_queue_shutdown(f->server_cq);
- drain_cq(f->server_cq);
- grpc_completion_queue_destroy(f->server_cq);
- grpc_completion_queue_shutdown(f->client_cq);
- drain_cq(f->client_cq);
- grpc_completion_queue_destroy(f->client_cq);
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
}
/* Request with a large amount of metadata.*/
@@ -107,8 +104,7 @@
gpr_timespec deadline = five_seconds_time();
grpc_metadata meta;
grpc_end2end_test_fixture f = begin_test(config, __FUNCTION__, NULL, NULL);
- cq_verifier *v_client = cq_verifier_create(f.client_cq);
- cq_verifier *v_server = cq_verifier_create(f.server_cq);
+ cq_verifier *cqv = cq_verifier_create(f.cq);
grpc_op ops[6];
grpc_op *op;
grpc_metadata_array initial_metadata_recv;
@@ -122,7 +118,7 @@
int was_cancelled = 2;
const int large_size = 64 * 1024;
- c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
+ c = grpc_channel_create_call(f.client, f.cq, "/foo",
"foo.test.google.fr", deadline);
GPR_ASSERT(c);
@@ -160,10 +156,10 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_server_request_call(f.server, &s, &call_details,
- &request_metadata_recv, f.server_cq,
- f.server_cq, tag(101)));
- cq_expect_completion(v_server, tag(101), 1);
- cq_verify(v_server);
+ &request_metadata_recv, f.cq,
+ f.cq, tag(101)));
+ cq_expect_completion(cqv, tag(101), 1);
+ cq_verify(cqv);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
@@ -182,11 +178,9 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(v_server, tag(102), 1);
- cq_verify(v_server);
-
- cq_expect_completion(v_client, tag(1), 1);
- cq_verify(v_client);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
+ cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_OK);
GPR_ASSERT(0 == strcmp(details, "xyz"));
@@ -205,8 +199,7 @@
grpc_call_destroy(c);
grpc_call_destroy(s);
- cq_verifier_destroy(v_client);
- cq_verifier_destroy(v_server);
+ cq_verifier_destroy(cqv);
grpc_byte_buffer_destroy(request_payload);
grpc_byte_buffer_destroy(request_payload_recv);
diff --git a/test/core/end2end/tests/request_with_payload.c b/test/core/end2end/tests/request_with_payload.c
index bba50b3..86199c9 100644
--- a/test/core/end2end/tests/request_with_payload.c
+++ b/test/core/end2end/tests/request_with_payload.c
@@ -89,12 +89,9 @@
shutdown_server(f);
shutdown_client(f);
- grpc_completion_queue_shutdown(f->server_cq);
- drain_cq(f->server_cq);
- grpc_completion_queue_destroy(f->server_cq);
- grpc_completion_queue_shutdown(f->client_cq);
- drain_cq(f->client_cq);
- grpc_completion_queue_destroy(f->client_cq);
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
}
/* Client sends a request with payload, server reads then returns status. */
@@ -106,8 +103,7 @@
grpc_byte_buffer_create(&request_payload_slice, 1);
gpr_timespec deadline = five_seconds_time();
grpc_end2end_test_fixture f = begin_test(config, __FUNCTION__, NULL, NULL);
- cq_verifier *v_client = cq_verifier_create(f.client_cq);
- cq_verifier *v_server = cq_verifier_create(f.server_cq);
+ cq_verifier *cqv = cq_verifier_create(f.cq);
grpc_op ops[6];
grpc_op *op;
grpc_metadata_array initial_metadata_recv;
@@ -120,7 +116,7 @@
size_t details_capacity = 0;
int was_cancelled = 2;
- c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
+ c = grpc_channel_create_call(f.client, f.cq, "/foo",
"foo.test.google.fr", deadline);
GPR_ASSERT(c);
@@ -151,10 +147,10 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_server_request_call(f.server, &s, &call_details,
- &request_metadata_recv, f.server_cq,
- f.server_cq, tag(101)));
- cq_expect_completion(v_server, tag(101), 1);
- cq_verify(v_server);
+ &request_metadata_recv, f.cq,
+ f.cq, tag(101)));
+ cq_expect_completion(cqv, tag(101), 1);
+ cq_verify(cqv);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
@@ -173,11 +169,9 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(v_server, tag(102), 1);
- cq_verify(v_server);
-
- cq_expect_completion(v_client, tag(1), 1);
- cq_verify(v_client);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
+ cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_OK);
GPR_ASSERT(0 == strcmp(details, "xyz"));
@@ -195,8 +189,7 @@
grpc_call_destroy(c);
grpc_call_destroy(s);
- cq_verifier_destroy(v_client);
- cq_verifier_destroy(v_server);
+ cq_verifier_destroy(cqv);
grpc_byte_buffer_destroy(request_payload);
grpc_byte_buffer_destroy(request_payload_recv);
diff --git a/test/core/end2end/tests/simple_delayed_request.c b/test/core/end2end/tests/simple_delayed_request.c
index d8463d4..c51019c 100644
--- a/test/core/end2end/tests/simple_delayed_request.c
+++ b/test/core/end2end/tests/simple_delayed_request.c
@@ -77,12 +77,9 @@
shutdown_server(f);
shutdown_client(f);
- grpc_completion_queue_shutdown(f->server_cq);
- drain_cq(f->server_cq);
- grpc_completion_queue_destroy(f->server_cq);
- grpc_completion_queue_shutdown(f->client_cq);
- drain_cq(f->client_cq);
- grpc_completion_queue_destroy(f->client_cq);
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
}
static void simple_delayed_request_body(grpc_end2end_test_config config,
@@ -93,8 +90,7 @@
grpc_call *c;
grpc_call *s;
gpr_timespec deadline = five_seconds_time();
- cq_verifier *v_client = cq_verifier_create(f->client_cq);
- cq_verifier *v_server = cq_verifier_create(f->server_cq);
+ cq_verifier *cqv = cq_verifier_create(f->cq);
grpc_op ops[6];
grpc_op *op;
grpc_metadata_array initial_metadata_recv;
@@ -108,7 +104,7 @@
config.init_client(f, client_args);
- c = grpc_channel_create_call(f->client, f->client_cq, "/foo",
+ c = grpc_channel_create_call(f->client, f->cq, "/foo",
"foo.test.google.fr", deadline);
GPR_ASSERT(c);
@@ -138,10 +134,10 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_server_request_call(f->server, &s, &call_details,
- &request_metadata_recv, f->server_cq,
- f->server_cq, tag(101)));
- cq_expect_completion(v_server, tag(101), 1);
- cq_verify(v_server);
+ &request_metadata_recv, f->cq,
+ f->cq, tag(101)));
+ cq_expect_completion(cqv, tag(101), 1);
+ cq_verify(cqv);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
@@ -157,11 +153,9 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(v_server, tag(102), 1);
- cq_verify(v_server);
-
- cq_expect_completion(v_client, tag(1), 1);
- cq_verify(v_client);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
+ cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
GPR_ASSERT(0 == strcmp(details, "xyz"));
@@ -178,8 +172,7 @@
grpc_call_destroy(c);
grpc_call_destroy(s);
- cq_verifier_destroy(v_client);
- cq_verifier_destroy(v_server);
+ cq_verifier_destroy(cqv);
}
static void test_simple_delayed_request_short(grpc_end2end_test_config config) {
diff --git a/test/core/end2end/tests/simple_request.c b/test/core/end2end/tests/simple_request.c
index 98e6779..be9e414 100644
--- a/test/core/end2end/tests/simple_request.c
+++ b/test/core/end2end/tests/simple_request.c
@@ -91,20 +91,16 @@
shutdown_server(f);
shutdown_client(f);
- grpc_completion_queue_shutdown(f->server_cq);
- drain_cq(f->server_cq);
- grpc_completion_queue_destroy(f->server_cq);
- grpc_completion_queue_shutdown(f->client_cq);
- drain_cq(f->client_cq);
- grpc_completion_queue_destroy(f->client_cq);
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
}
static void simple_request_body(grpc_end2end_test_fixture f) {
grpc_call *c;
grpc_call *s;
gpr_timespec deadline = five_seconds_time();
- cq_verifier *v_client = cq_verifier_create(f.client_cq);
- cq_verifier *v_server = cq_verifier_create(f.server_cq);
+ cq_verifier *cqv = cq_verifier_create(f.cq);
grpc_op ops[6];
grpc_op *op;
grpc_metadata_array initial_metadata_recv;
@@ -116,7 +112,7 @@
size_t details_capacity = 0;
int was_cancelled = 2;
- c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
+ c = grpc_channel_create_call(f.client, f.cq, "/foo",
"foo.test.google.fr:1234", deadline);
GPR_ASSERT(c);
@@ -144,10 +140,10 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_server_request_call(f.server, &s, &call_details,
- &request_metadata_recv, f.server_cq,
- f.server_cq, tag(101)));
- cq_expect_completion(v_server, tag(101), 1);
- cq_verify(v_server);
+ &request_metadata_recv, f.cq,
+ f.cq, tag(101)));
+ cq_expect_completion(cqv, tag(101), 1);
+ cq_verify(cqv);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
@@ -163,11 +159,9 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(v_server, tag(102), 1);
- cq_verify(v_server);
-
- cq_expect_completion(v_client, tag(1), 1);
- cq_verify(v_client);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
+ cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
GPR_ASSERT(0 == strcmp(details, "xyz"));
@@ -184,8 +178,7 @@
grpc_call_destroy(c);
grpc_call_destroy(s);
- cq_verifier_destroy(v_client);
- cq_verifier_destroy(v_server);
+ cq_verifier_destroy(cqv);
}
static void test_invoke_simple_request(grpc_end2end_test_config config) {
diff --git a/test/core/end2end/tests/simple_request_with_high_initial_sequence_number.c b/test/core/end2end/tests/simple_request_with_high_initial_sequence_number.c
index 3c1a19a..566f971 100644
--- a/test/core/end2end/tests/simple_request_with_high_initial_sequence_number.c
+++ b/test/core/end2end/tests/simple_request_with_high_initial_sequence_number.c
@@ -91,20 +91,16 @@
shutdown_server(f);
shutdown_client(f);
- grpc_completion_queue_shutdown(f->server_cq);
- drain_cq(f->server_cq);
- grpc_completion_queue_destroy(f->server_cq);
- grpc_completion_queue_shutdown(f->client_cq);
- drain_cq(f->client_cq);
- grpc_completion_queue_destroy(f->client_cq);
+ grpc_completion_queue_shutdown(f->cq);
+ drain_cq(f->cq);
+ grpc_completion_queue_destroy(f->cq);
}
static void simple_request_body(grpc_end2end_test_fixture f) {
grpc_call *c;
grpc_call *s;
gpr_timespec deadline = five_seconds_time();
- cq_verifier *v_client = cq_verifier_create(f.client_cq);
- cq_verifier *v_server = cq_verifier_create(f.server_cq);
+ cq_verifier *cqv = cq_verifier_create(f.cq);
grpc_op ops[6];
grpc_op *op;
grpc_metadata_array initial_metadata_recv;
@@ -116,7 +112,7 @@
size_t details_capacity = 0;
int was_cancelled = 2;
- c = grpc_channel_create_call(f.client, f.client_cq, "/foo",
+ c = grpc_channel_create_call(f.client, f.cq, "/foo",
"foo.test.google.fr:1234", deadline);
GPR_ASSERT(c);
@@ -144,10 +140,10 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_server_request_call(f.server, &s, &call_details,
- &request_metadata_recv, f.server_cq,
- f.server_cq, tag(101)));
- cq_expect_completion(v_server, tag(101), 1);
- cq_verify(v_server);
+ &request_metadata_recv, f.cq,
+ f.cq, tag(101)));
+ cq_expect_completion(cqv, tag(101), 1);
+ cq_verify(cqv);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
@@ -163,11 +159,9 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(v_server, tag(102), 1);
- cq_verify(v_server);
-
- cq_expect_completion(v_client, tag(1), 1);
- cq_verify(v_client);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
+ cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
GPR_ASSERT(0 == strcmp(details, "xyz"));
@@ -184,8 +178,7 @@
grpc_call_destroy(c);
grpc_call_destroy(s);
- cq_verifier_destroy(v_client);
- cq_verifier_destroy(v_server);
+ cq_verifier_destroy(cqv);
}
static void test_invoke_10_simple_requests(grpc_end2end_test_config config, int initial_sequence_number) {
diff --git a/test/core/httpcli/httpcli_test.c b/test/core/httpcli/httpcli_test.c
index 17a2996..7d6838f 100644
--- a/test/core/httpcli/httpcli_test.c
+++ b/test/core/httpcli/httpcli_test.c
@@ -40,7 +40,10 @@
#include <grpc/support/sync.h>
#include "test/core/util/test_config.h"
-static gpr_event g_done;
+static gpr_mu g_mu;
+static int g_done = 0;
+static grpc_pollset_set g_pollset_set;
+static grpc_pollset g_pollset;
static gpr_timespec n_seconds_time(int seconds) {
return GRPC_TIMEOUT_SECONDS_TO_DEADLINE(seconds);
@@ -50,22 +53,28 @@
GPR_ASSERT(arg == (void *)42);
GPR_ASSERT(response);
GPR_ASSERT(response->status == 200);
- gpr_event_set(&g_done, (void *)1);
+ gpr_mu_lock(&g_mu);
+ g_done = 1;
+ gpr_mu_unlock(&g_mu);
}
static void test_get(int use_ssl) {
grpc_httpcli_request req;
+ g_done = 0;
gpr_log(GPR_INFO, "running %s with use_ssl=%d.", __FUNCTION__, use_ssl);
- gpr_event_init(&g_done);
memset(&req, 0, sizeof(req));
req.host = "www.google.com";
req.path = "/";
req.use_ssl = use_ssl;
- grpc_httpcli_get(&req, n_seconds_time(15), on_finish, (void *)42);
- GPR_ASSERT(gpr_event_wait(&g_done, n_seconds_time(20)));
+ grpc_httpcli_get(&req, n_seconds_time(15), &g_pollset_set, on_finish, (void *)42);
+ gpr_mu_lock(&g_mu);
+ while (!g_done) {
+ grpc_pollset_work(&g_pollset, n_seconds_time(20));
+ }
+ gpr_mu_unlock(&g_mu);
}
/*
@@ -89,6 +98,10 @@
int main(int argc, char **argv) {
grpc_test_init(argc, argv);
grpc_iomgr_init();
+ grpc_pollset_set_init(&g_pollset_set);
+ grpc_pollset_init(&g_pollset);
+ gpr_mu_init(&g_mu);
+ grpc_pollset_set_add_pollset(&g_pollset_set, &g_pollset);
test_get(0);
test_get(1);
@@ -96,7 +109,10 @@
/* test_post(0); */
/* test_post(1); */
+ grpc_pollset_set_destroy(&g_pollset_set);
+ grpc_pollset_destroy(&g_pollset);
grpc_iomgr_shutdown();
+ gpr_mu_destroy(&g_mu);
return 0;
}
diff --git a/test/core/iomgr/tcp_client_posix_test.c b/test/core/iomgr/tcp_client_posix_test.c
index 3c4d8fe..0d80d81 100644
--- a/test/core/iomgr/tcp_client_posix_test.c
+++ b/test/core/iomgr/tcp_client_posix_test.c
@@ -45,6 +45,9 @@
#include <grpc/support/time.h>
#include "test/core/util/test_config.h"
+static grpc_pollset_set g_pollset_set;
+static grpc_pollset g_pollset;
+
static gpr_timespec test_deadline(void) {
return GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10);
}
@@ -81,7 +84,7 @@
/* connect to it */
GPR_ASSERT(getsockname(svr_fd, (struct sockaddr *)&addr, &addr_len) == 0);
- grpc_tcp_client_connect(must_succeed, &ev, (struct sockaddr *)&addr, addr_len,
+ grpc_tcp_client_connect(must_succeed, &ev, &g_pollset_set, (struct sockaddr *)&addr, addr_len,
gpr_inf_future);
/* await the connection */
@@ -107,7 +110,7 @@
addr.sin_family = AF_INET;
/* connect to a broken address */
- grpc_tcp_client_connect(must_fail, &ev, (struct sockaddr *)&addr, addr_len,
+ grpc_tcp_client_connect(must_fail, &ev, &g_pollset_set, (struct sockaddr *)&addr, addr_len,
gpr_inf_future);
/* wait for the connection callback to finish */
@@ -153,7 +156,7 @@
connect_deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(1);
- grpc_tcp_client_connect(must_fail, &ev, (struct sockaddr *)&addr, addr_len,
+ grpc_tcp_client_connect(must_fail, &ev, &g_pollset_set, (struct sockaddr *)&addr, addr_len,
connect_deadline);
/* Make sure the event doesn't trigger early */
GPR_ASSERT(!gpr_event_wait(&ev, GRPC_TIMEOUT_MILLIS_TO_DEADLINE(500)));
@@ -171,10 +174,15 @@
int main(int argc, char **argv) {
grpc_test_init(argc, argv);
grpc_iomgr_init();
+ grpc_pollset_set_init(&g_pollset_set);
+ grpc_pollset_init(&g_pollset);
+ grpc_pollset_set_add_pollset(&g_pollset_set, &g_pollset);
test_succeeds();
gpr_log(GPR_ERROR, "End of first test");
test_fails();
test_times_out();
+ grpc_pollset_set_destroy(&g_pollset_set);
+ grpc_pollset_destroy(&g_pollset);
grpc_iomgr_shutdown();
return 0;
}
diff --git a/vsprojects/grpc/grpc.vcxproj b/vsprojects/grpc/grpc.vcxproj
index e6c4745..f215cf9 100644
--- a/vsprojects/grpc/grpc.vcxproj
+++ b/vsprojects/grpc/grpc.vcxproj
@@ -199,6 +199,8 @@
<ClInclude Include="..\..\src\core\iomgr\pollset_kick_posix.h" />
<ClInclude Include="..\..\src\core\iomgr\pollset_kick_windows.h" />
<ClInclude Include="..\..\src\core\iomgr\pollset_posix.h" />
+ <ClInclude Include="..\..\src\core\iomgr\pollset_set_posix.h" />
+ <ClInclude Include="..\..\src\core\iomgr\pollset_set_windows.h" />
<ClInclude Include="..\..\src\core\iomgr\pollset_windows.h" />
<ClInclude Include="..\..\src\core\iomgr\resolve_address.h" />
<ClInclude Include="..\..\src\core\iomgr\sockaddr.h" />
@@ -356,6 +358,10 @@
</ClCompile>
<ClCompile Include="..\..\src\core\iomgr\pollset_posix.c">
</ClCompile>
+ <ClCompile Include="..\..\src\core\iomgr\pollset_set_posix.c">
+ </ClCompile>
+ <ClCompile Include="..\..\src\core\iomgr\pollset_set_windows.c">
+ </ClCompile>
<ClCompile Include="..\..\src\core\iomgr\pollset_windows.c">
</ClCompile>
<ClCompile Include="..\..\src\core\iomgr\resolve_address_posix.c">
diff --git a/vsprojects/grpc/grpc.vcxproj.filters b/vsprojects/grpc/grpc.vcxproj.filters
index ef9387d..ed19f19 100644
--- a/vsprojects/grpc/grpc.vcxproj.filters
+++ b/vsprojects/grpc/grpc.vcxproj.filters
@@ -145,6 +145,12 @@
<ClCompile Include="..\..\src\core\iomgr\pollset_posix.c">
<Filter>src\core\iomgr</Filter>
</ClCompile>
+ <ClCompile Include="..\..\src\core\iomgr\pollset_set_posix.c">
+ <Filter>src\core\iomgr</Filter>
+ </ClCompile>
+ <ClCompile Include="..\..\src\core\iomgr\pollset_set_windows.c">
+ <Filter>src\core\iomgr</Filter>
+ </ClCompile>
<ClCompile Include="..\..\src\core\iomgr\pollset_windows.c">
<Filter>src\core\iomgr</Filter>
</ClCompile>
@@ -509,6 +515,12 @@
<ClInclude Include="..\..\src\core\iomgr\pollset_posix.h">
<Filter>src\core\iomgr</Filter>
</ClInclude>
+ <ClInclude Include="..\..\src\core\iomgr\pollset_set_posix.h">
+ <Filter>src\core\iomgr</Filter>
+ </ClInclude>
+ <ClInclude Include="..\..\src\core\iomgr\pollset_set_windows.h">
+ <Filter>src\core\iomgr</Filter>
+ </ClInclude>
<ClInclude Include="..\..\src\core\iomgr\pollset_windows.h">
<Filter>src\core\iomgr</Filter>
</ClInclude>
diff --git a/vsprojects/grpc_unsecure/grpc_unsecure.vcxproj b/vsprojects/grpc_unsecure/grpc_unsecure.vcxproj
index c157225..27f0018 100644
--- a/vsprojects/grpc_unsecure/grpc_unsecure.vcxproj
+++ b/vsprojects/grpc_unsecure/grpc_unsecure.vcxproj
@@ -181,6 +181,8 @@
<ClInclude Include="..\..\src\core\iomgr\pollset_kick_posix.h" />
<ClInclude Include="..\..\src\core\iomgr\pollset_kick_windows.h" />
<ClInclude Include="..\..\src\core\iomgr\pollset_posix.h" />
+ <ClInclude Include="..\..\src\core\iomgr\pollset_set_posix.h" />
+ <ClInclude Include="..\..\src\core\iomgr\pollset_set_windows.h" />
<ClInclude Include="..\..\src\core\iomgr\pollset_windows.h" />
<ClInclude Include="..\..\src\core\iomgr\resolve_address.h" />
<ClInclude Include="..\..\src\core\iomgr\sockaddr.h" />
@@ -298,6 +300,10 @@
</ClCompile>
<ClCompile Include="..\..\src\core\iomgr\pollset_posix.c">
</ClCompile>
+ <ClCompile Include="..\..\src\core\iomgr\pollset_set_posix.c">
+ </ClCompile>
+ <ClCompile Include="..\..\src\core\iomgr\pollset_set_windows.c">
+ </ClCompile>
<ClCompile Include="..\..\src\core\iomgr\pollset_windows.c">
</ClCompile>
<ClCompile Include="..\..\src\core\iomgr\resolve_address_posix.c">
diff --git a/vsprojects/grpc_unsecure/grpc_unsecure.vcxproj.filters b/vsprojects/grpc_unsecure/grpc_unsecure.vcxproj.filters
index 0bee095..8dae709 100644
--- a/vsprojects/grpc_unsecure/grpc_unsecure.vcxproj.filters
+++ b/vsprojects/grpc_unsecure/grpc_unsecure.vcxproj.filters
@@ -85,6 +85,12 @@
<ClCompile Include="..\..\src\core\iomgr\pollset_posix.c">
<Filter>src\core\iomgr</Filter>
</ClCompile>
+ <ClCompile Include="..\..\src\core\iomgr\pollset_set_posix.c">
+ <Filter>src\core\iomgr</Filter>
+ </ClCompile>
+ <ClCompile Include="..\..\src\core\iomgr\pollset_set_windows.c">
+ <Filter>src\core\iomgr</Filter>
+ </ClCompile>
<ClCompile Include="..\..\src\core\iomgr\pollset_windows.c">
<Filter>src\core\iomgr</Filter>
</ClCompile>
@@ -398,6 +404,12 @@
<ClInclude Include="..\..\src\core\iomgr\pollset_posix.h">
<Filter>src\core\iomgr</Filter>
</ClInclude>
+ <ClInclude Include="..\..\src\core\iomgr\pollset_set_posix.h">
+ <Filter>src\core\iomgr</Filter>
+ </ClInclude>
+ <ClInclude Include="..\..\src\core\iomgr\pollset_set_windows.h">
+ <Filter>src\core\iomgr</Filter>
+ </ClInclude>
<ClInclude Include="..\..\src\core\iomgr\pollset_windows.h">
<Filter>src\core\iomgr</Filter>
</ClInclude>