Change Core to use the new completion_queue_create API

reviewable/pr9972/r1
Sree Kuchibhotla 8 years ago
parent 7a4e5b427c
commit 321881d07f
  1. 8
      include/grpc/grpc.h
  2. 27
      src/core/lib/surface/completion_queue.c
  3. 3
      src/core/lib/surface/completion_queue.h
  4. 18
      src/core/lib/surface/server.c
  5. 10
      test/core/bad_client/bad_client.c
  6. 3
      test/core/bad_ssl/bad_ssl_test.c
  7. 22
      test/core/bad_ssl/server_common.c
  8. 24
      test/core/client_channel/lb_policies_test.c
  9. 2
      test/core/client_channel/set_initial_connect_string_test.c
  10. 2
      test/core/end2end/bad_server_response_test.c
  11. 2
      test/core/end2end/connection_refused_test.c
  12. 12
      test/core/end2end/dualstack_socket_test.c
  13. 1
      test/core/end2end/end2end_tests.h
  14. 11
      test/core/end2end/fixtures/h2_census.c
  15. 11
      test/core/end2end/fixtures/h2_compress.c
  16. 4
      test/core/end2end/fixtures/h2_fakesec.c
  17. 4
      test/core/end2end/fixtures/h2_fd.c
  18. 11
      test/core/end2end/fixtures/h2_full+pipe.c
  19. 11
      test/core/end2end/fixtures/h2_full+trace.c
  20. 11
      test/core/end2end/fixtures/h2_full.c
  21. 11
      test/core/end2end/fixtures/h2_http_proxy.c
  22. 4
      test/core/end2end/fixtures/h2_load_reporting.c
  23. 4
      test/core/end2end/fixtures/h2_oauth2.c
  24. 13
      test/core/end2end/fixtures/h2_proxy.c
  25. 4
      test/core/end2end/fixtures/h2_sockpair+trace.c
  26. 4
      test/core/end2end/fixtures/h2_sockpair.c
  27. 4
      test/core/end2end/fixtures/h2_sockpair_1byte.c
  28. 4
      test/core/end2end/fixtures/h2_ssl.c
  29. 32
      test/core/end2end/fixtures/h2_ssl_cert.c
  30. 4
      test/core/end2end/fixtures/h2_ssl_proxy.c
  31. 11
      test/core/end2end/fixtures/h2_uds.c
  32. 2
      test/core/end2end/fixtures/proxy.c
  33. 28
      test/core/end2end/fuzzers/api_fuzzer.c
  34. 3
      test/core/end2end/fuzzers/client_fuzzer.c
  35. 3
      test/core/end2end/fuzzers/server_fuzzer.c
  36. 2
      test/core/end2end/goaway_server_test.c
  37. 11
      test/core/end2end/invalid_call_argument_test.c
  38. 14
      test/core/end2end/multiple_server_queues_test.c
  39. 2
      test/core/end2end/no_server_test.c
  40. 8
      test/core/end2end/tests/authority_not_supported.c
  41. 8
      test/core/end2end/tests/bad_hostname.c
  42. 8
      test/core/end2end/tests/binary_metadata.c
  43. 5
      test/core/end2end/tests/call_creds.c
  44. 8
      test/core/end2end/tests/cancel_after_accept.c
  45. 8
      test/core/end2end/tests/cancel_after_client_done.c
  46. 8
      test/core/end2end/tests/cancel_after_invoke.c
  47. 8
      test/core/end2end/tests/cancel_before_invoke.c
  48. 8
      test/core/end2end/tests/cancel_in_a_vacuum.c
  49. 8
      test/core/end2end/tests/cancel_with_status.c
  50. 8
      test/core/end2end/tests/compressed_payload.c
  51. 3
      test/core/end2end/tests/connectivity.c
  52. 8
      test/core/end2end/tests/default_host.c
  53. 3
      test/core/end2end/tests/disappearing_server.c
  54. 8
      test/core/end2end/tests/empty_batch.c
  55. 8
      test/core/end2end/tests/filter_call_init_fails.c
  56. 8
      test/core/end2end/tests/filter_causes_close.c
  57. 8
      test/core/end2end/tests/filter_latency.c
  58. 2
      test/core/end2end/tests/graceful_server_shutdown.c
  59. 8
      test/core/end2end/tests/high_initial_seqno.c
  60. 8
      test/core/end2end/tests/hpack_size.c
  61. 8
      test/core/end2end/tests/idempotent_request.c
  62. 8
      test/core/end2end/tests/invoke_large_request.c
  63. 8
      test/core/end2end/tests/large_metadata.c
  64. 8
      test/core/end2end/tests/load_reporting_hook.c
  65. 8
      test/core/end2end/tests/max_concurrent_streams.c
  66. 8
      test/core/end2end/tests/max_message_length.c
  67. 8
      test/core/end2end/tests/negative_deadline.c
  68. 8
      test/core/end2end/tests/network_status_change.c
  69. 8
      test/core/end2end/tests/no_logging.c
  70. 8
      test/core/end2end/tests/no_op.c
  71. 8
      test/core/end2end/tests/payload.c
  72. 3
      test/core/end2end/tests/ping.c
  73. 8
      test/core/end2end/tests/ping_pong_streaming.c
  74. 8
      test/core/end2end/tests/registered_call.c
  75. 8
      test/core/end2end/tests/request_with_flags.c
  76. 8
      test/core/end2end/tests/request_with_payload.c
  77. 8
      test/core/end2end/tests/resource_quota_server.c
  78. 8
      test/core/end2end/tests/server_finishes_request.c
  79. 2
      test/core/end2end/tests/shutdown_finishes_calls.c
  80. 2
      test/core/end2end/tests/shutdown_finishes_tags.c
  81. 8
      test/core/end2end/tests/simple_cacheable_request.c
  82. 8
      test/core/end2end/tests/simple_delayed_request.c
  83. 8
      test/core/end2end/tests/simple_metadata.c
  84. 8
      test/core/end2end/tests/simple_request.c
  85. 8
      test/core/end2end/tests/streaming_error_response.c
  86. 8
      test/core/end2end/tests/trailing_metadata.c
  87. 8
      test/core/end2end/tests/write_buffering.c
  88. 8
      test/core/end2end/tests/write_buffering_at_end.c
  89. 2
      test/core/fling/client.c
  90. 23
      test/core/fling/server.c
  91. 4
      test/core/handshake/client_ssl.c
  92. 3
      test/core/handshake/server_ssl.c
  93. 7
      test/core/memory_usage/client.c
  94. 21
      test/core/memory_usage/server.c
  95. 2
      test/core/surface/alarm_test.c
  96. 192
      test/core/surface/completion_queue_test.c
  97. 5
      test/core/surface/concurrent_connectivity_test.c
  98. 2
      test/core/surface/lame_client_test.c
  99. 6
      test/core/surface/sequential_connectivity_test.c
  100. 3
      test/core/surface/server_chttp2_test.c
  101. Some files were not shown because too many files have changed in this diff Show More

@ -127,14 +127,8 @@ typedef enum {
NON_POLLING NON_POLLING
} grpc_cq_polling_type; } grpc_cq_polling_type;
/** Create a completion queue.
WARNING: This API is deprecated and will soon be deleted and replaced with
completion_queue_create_ex() */
GRPCAPI grpc_completion_queue *grpc_completion_queue_create(void *reserved);
/** Create a completion queue */ /** Create a completion queue */
GRPCAPI grpc_completion_queue *grpc_completion_queue_create_ex( GRPCAPI grpc_completion_queue *grpc_completion_queue_create(
grpc_cq_completion_type completion_type, grpc_cq_polling_type polling_type, grpc_cq_completion_type completion_type, grpc_cq_polling_type polling_type,
void *reserved); void *reserved);

@ -115,7 +115,7 @@ int grpc_cq_event_timeout_trace;
static void on_pollset_shutdown_done(grpc_exec_ctx *exec_ctx, void *cc, static void on_pollset_shutdown_done(grpc_exec_ctx *exec_ctx, void *cc,
grpc_error *error); grpc_error *error);
grpc_completion_queue *grpc_completion_queue_create_ex( grpc_completion_queue *grpc_completion_queue_create(
grpc_cq_completion_type completion_type, grpc_cq_polling_type polling_type, grpc_cq_completion_type completion_type, grpc_cq_polling_type polling_type,
void *reserved) { void *reserved) {
grpc_completion_queue *cc; grpc_completion_queue *cc;
@ -132,6 +132,9 @@ grpc_completion_queue *grpc_completion_queue_create_ex(
cc->outstanding_tag_capacity = 0; cc->outstanding_tag_capacity = 0;
#endif #endif
cc->completion_type = completion_type;
cc->polling_type = polling_type;
/* Initial ref is dropped by grpc_completion_queue_shutdown */ /* Initial ref is dropped by grpc_completion_queue_shutdown */
gpr_ref_init(&cc->pending_events, 1); gpr_ref_init(&cc->pending_events, 1);
/* One for destroy(), one for pollset_shutdown */ /* One for destroy(), one for pollset_shutdown */
@ -155,8 +158,12 @@ grpc_completion_queue *grpc_completion_queue_create_ex(
return cc; return cc;
} }
grpc_completion_queue *grpc_completion_queue_create(void *reserved) { grpc_cq_completion_type grpc_get_cq_completion_type(grpc_completion_queue *cc) {
return grpc_completion_queue_create_ex(0, DEFAULT_POLLING, reserved); return cc->completion_type;
}
grpc_cq_polling_type grpc_get_cq_polling_type(grpc_completion_queue *cc) {
return cc->polling_type;
} }
#ifdef GRPC_CQ_REF_COUNT_DEBUG #ifdef GRPC_CQ_REF_COUNT_DEBUG
@ -359,6 +366,13 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
grpc_pollset_worker *worker = NULL; grpc_pollset_worker *worker = NULL;
gpr_timespec now; gpr_timespec now;
if (cc->completion_type != GRPC_CQ_NEXT) {
gpr_log(GPR_ERROR,
"grpc_completion_queue_next() cannot be called on this completion "
"queue since its completion type is not GRPC_CQ_NEXT");
abort();
}
GPR_TIMER_BEGIN("grpc_completion_queue_next", 0); GPR_TIMER_BEGIN("grpc_completion_queue_next", 0);
GRPC_API_TRACE( GRPC_API_TRACE(
@ -529,6 +543,13 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
GPR_TIMER_BEGIN("grpc_completion_queue_pluck", 0); GPR_TIMER_BEGIN("grpc_completion_queue_pluck", 0);
if (cc->completion_type != GRPC_CQ_PLUCK) {
gpr_log(GPR_ERROR,
"grpc_completion_queue_pluck() cannot be called on this completion "
"queue since its completion type is not GRPC_CQ_PLUCK");
abort();
}
if (grpc_cq_pluck_trace) { if (grpc_cq_pluck_trace) {
GRPC_API_TRACE( GRPC_API_TRACE(
"grpc_completion_queue_pluck(" "grpc_completion_queue_pluck("

@ -99,4 +99,7 @@ bool grpc_cq_is_non_listening_server_cq(grpc_completion_queue *cc);
void grpc_cq_mark_server_cq(grpc_completion_queue *cc); void grpc_cq_mark_server_cq(grpc_completion_queue *cc);
int grpc_cq_is_server_cq(grpc_completion_queue *cc); int grpc_cq_is_server_cq(grpc_completion_queue *cc);
grpc_cq_completion_type grpc_get_cq_completion_type(grpc_completion_queue *cc);
grpc_cq_polling_type grpc_get_cq_polling_type(grpc_completion_queue *cc);
#endif /* GRPC_CORE_LIB_SURFACE_COMPLETION_QUEUE_H */ #endif /* GRPC_CORE_LIB_SURFACE_COMPLETION_QUEUE_H */

@ -1001,6 +1001,14 @@ void grpc_server_register_completion_queue(grpc_server *server,
GRPC_API_TRACE( GRPC_API_TRACE(
"grpc_server_register_completion_queue(server=%p, cq=%p, reserved=%p)", 3, "grpc_server_register_completion_queue(server=%p, cq=%p, reserved=%p)", 3,
(server, cq, reserved)); (server, cq, reserved));
if (grpc_get_cq_completion_type(cq) != GRPC_CQ_NEXT) {
gpr_log(
GPR_ERROR,
"Server completion queues must have a completion type of GRPC_CQ_NEXT");
abort();
}
register_completion_queue(server, cq, false, reserved); register_completion_queue(server, cq, false, reserved);
} }
@ -1423,8 +1431,9 @@ grpc_call_error grpc_server_request_call(
"grpc_server_request_call(" "grpc_server_request_call("
"server=%p, call=%p, details=%p, initial_metadata=%p, " "server=%p, call=%p, details=%p, initial_metadata=%p, "
"cq_bound_to_call=%p, cq_for_notification=%p, tag=%p)", "cq_bound_to_call=%p, cq_for_notification=%p, tag=%p)",
7, (server, call, details, initial_metadata, cq_bound_to_call, 7,
cq_for_notification, tag)); (server, call, details, initial_metadata, cq_bound_to_call,
cq_for_notification, tag));
size_t cq_idx; size_t cq_idx;
for (cq_idx = 0; cq_idx < server->cq_count; cq_idx++) { for (cq_idx = 0; cq_idx < server->cq_count; cq_idx++) {
if (server->cqs[cq_idx] == cq_for_notification) { if (server->cqs[cq_idx] == cq_for_notification) {
@ -1466,8 +1475,9 @@ grpc_call_error grpc_server_request_registered_call(
"server=%p, rmp=%p, call=%p, deadline=%p, initial_metadata=%p, " "server=%p, rmp=%p, call=%p, deadline=%p, initial_metadata=%p, "
"optional_payload=%p, cq_bound_to_call=%p, cq_for_notification=%p, " "optional_payload=%p, cq_bound_to_call=%p, cq_for_notification=%p, "
"tag=%p)", "tag=%p)",
9, (server, rmp, call, deadline, initial_metadata, optional_payload, 9,
cq_bound_to_call, cq_for_notification, tag)); (server, rmp, call, deadline, initial_metadata, optional_payload,
cq_bound_to_call, cq_for_notification, tag));
size_t cq_idx; size_t cq_idx;
for (cq_idx = 0; cq_idx < server->cq_count; cq_idx++) { for (cq_idx = 0; cq_idx < server->cq_count; cq_idx++) {

@ -104,6 +104,7 @@ void grpc_run_bad_client_test(
grpc_slice_buffer outgoing; grpc_slice_buffer outgoing;
grpc_closure done_write_closure; grpc_closure done_write_closure;
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
grpc_completion_queue *shutdown_cq;
hex = gpr_dump(client_payload, client_payload_length, hex = gpr_dump(client_payload, client_payload_length,
GPR_DUMP_HEX | GPR_DUMP_ASCII); GPR_DUMP_HEX | GPR_DUMP_ASCII);
@ -124,7 +125,7 @@ void grpc_run_bad_client_test(
/* Create server, completion events */ /* Create server, completion events */
a.server = grpc_server_create(NULL, NULL); a.server = grpc_server_create(NULL, NULL);
a.cq = grpc_completion_queue_create(NULL); a.cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
gpr_event_init(&a.done_thd); gpr_event_init(&a.done_thd);
gpr_event_init(&a.done_write); gpr_event_init(&a.done_write);
a.validator = server_validator; a.validator = server_validator;
@ -195,10 +196,13 @@ void grpc_run_bad_client_test(
grpc_endpoint_destroy(&exec_ctx, sfd.client); grpc_endpoint_destroy(&exec_ctx, sfd.client);
grpc_exec_ctx_finish(&exec_ctx); grpc_exec_ctx_finish(&exec_ctx);
} }
grpc_server_shutdown_and_notify(a.server, a.cq, NULL);
shutdown_cq = grpc_completion_queue_create(GRPC_CQ_PLUCK, NON_POLLING, NULL);
grpc_server_shutdown_and_notify(a.server, shutdown_cq, NULL);
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(
a.cq, NULL, grpc_timeout_seconds_to_deadline(1), NULL) shutdown_cq, NULL, grpc_timeout_seconds_to_deadline(1), NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_completion_queue_destroy(shutdown_cq);
grpc_server_destroy(a.server); grpc_server_destroy(a.server);
grpc_completion_queue_destroy(a.cq); grpc_completion_queue_destroy(a.cq);
grpc_slice_buffer_destroy_internal(&exec_ctx, &outgoing); grpc_slice_buffer_destroy_internal(&exec_ctx, &outgoing);

@ -61,7 +61,8 @@ static void run_test(const char *target, size_t nops) {
grpc_status_code status; grpc_status_code status;
grpc_call_error error; grpc_call_error error;
gpr_timespec deadline = grpc_timeout_seconds_to_deadline(5); gpr_timespec deadline = grpc_timeout_seconds_to_deadline(5);
grpc_completion_queue *cq = grpc_completion_queue_create(NULL); grpc_completion_queue *cq =
grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
cq_verifier *cqv = cq_verifier_create(cq); cq_verifier *cqv = cq_verifier_create(cq);
grpc_op ops[6]; grpc_op ops[6];

@ -66,7 +66,10 @@ void bad_ssl_run(grpc_server *server) {
grpc_call *s = NULL; grpc_call *s = NULL;
grpc_call_details call_details; grpc_call_details call_details;
grpc_metadata_array request_metadata_recv; grpc_metadata_array request_metadata_recv;
grpc_completion_queue *cq = grpc_completion_queue_create(NULL);
grpc_completion_queue *cq =
grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
grpc_completion_queue *shutdown_cq;
grpc_call_details_init(&call_details); grpc_call_details_init(&call_details);
grpc_metadata_array_init(&request_metadata_recv); grpc_metadata_array_init(&request_metadata_recv);
@ -82,16 +85,21 @@ void bad_ssl_run(grpc_server *server) {
while (!shutdown_finished) { while (!shutdown_finished) {
if (got_sigint && !shutdown_started) { if (got_sigint && !shutdown_started) {
gpr_log(GPR_INFO, "Shutting down due to SIGINT"); gpr_log(GPR_INFO, "Shutting down due to SIGINT");
grpc_server_shutdown_and_notify(server, cq, NULL); shutdown_cq =
GPR_ASSERT(grpc_completion_queue_pluck( grpc_completion_queue_create(GRPC_CQ_PLUCK, NON_POLLING, NULL);
cq, NULL, grpc_timeout_seconds_to_deadline(5), NULL) grpc_server_shutdown_and_notify(server, shutdown_cq, NULL);
.type == GRPC_OP_COMPLETE); GPR_ASSERT(
grpc_completion_queue_pluck(shutdown_cq, NULL,
grpc_timeout_seconds_to_deadline(5), NULL)
.type == GRPC_OP_COMPLETE);
grpc_completion_queue_destroy(shutdown_cq);
grpc_completion_queue_shutdown(cq); grpc_completion_queue_shutdown(cq);
shutdown_started = 1; shutdown_started = 1;
} }
ev = grpc_completion_queue_next( ev = grpc_completion_queue_next(
cq, gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), cq,
gpr_time_from_micros(1000000, GPR_TIMESPAN)), gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
gpr_time_from_micros(1000000, GPR_TIMESPAN)),
NULL); NULL);
switch (ev.type) { switch (ev.type) {
case GRPC_OP_COMPLETE: case GRPC_OP_COMPLETE:

@ -59,14 +59,15 @@ typedef struct servers_fixture {
grpc_server **servers; grpc_server **servers;
grpc_call **server_calls; grpc_call **server_calls;
grpc_completion_queue *cq; grpc_completion_queue *cq;
grpc_completion_queue *shutdown_cq;
char **servers_hostports; char **servers_hostports;
grpc_metadata_array *request_metadata_recv; grpc_metadata_array *request_metadata_recv;
} servers_fixture; } servers_fixture;
typedef struct request_sequences { typedef struct request_sequences {
size_t n; /* number of iterations */ size_t n; /* number of iterations */
int *connections; /* indexed by the interation number, value is the index of int *connections; /* indexed by the interation number, value is the index of
the server it connected to or -1 if none */ the server it connected to or -1 if none */
int *connectivity_states; /* indexed by the interation number, value is the int *connectivity_states; /* indexed by the interation number, value is the
client connectivity state */ client connectivity state */
} request_sequences; } request_sequences;
@ -146,10 +147,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void kill_server(const servers_fixture *f, size_t i) { static void kill_server(const servers_fixture *f, size_t i) {
gpr_log(GPR_INFO, "KILLING SERVER %" PRIuPTR, i); gpr_log(GPR_INFO, "KILLING SERVER %" PRIuPTR, i);
GPR_ASSERT(f->servers[i] != NULL); GPR_ASSERT(f->servers[i] != NULL);
grpc_server_shutdown_and_notify(f->servers[i], f->cq, tag(10000)); grpc_server_shutdown_and_notify(f->servers[i], f->shutdown_cq, tag(10000));
GPR_ASSERT( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(10000),
grpc_completion_queue_pluck(f->cq, tag(10000), n_millis_time(5000), NULL) n_millis_time(5000), NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->servers[i]); grpc_server_destroy(f->servers[i]);
f->servers[i] = NULL; f->servers[i] = NULL;
} }
@ -196,7 +197,9 @@ static servers_fixture *setup_servers(const char *server_host,
/* Create servers. */ /* Create servers. */
f->servers = gpr_malloc(sizeof(grpc_server *) * num_servers); f->servers = gpr_malloc(sizeof(grpc_server *) * num_servers);
f->servers_hostports = gpr_malloc(sizeof(char *) * num_servers); f->servers_hostports = gpr_malloc(sizeof(char *) * num_servers);
f->cq = grpc_completion_queue_create(NULL); f->cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
f->shutdown_cq =
grpc_completion_queue_create(GRPC_CQ_PLUCK, NON_POLLING, NULL);
for (i = 0; i < num_servers; i++) { for (i = 0; i < num_servers; i++) {
grpc_metadata_array_init(&f->request_metadata_recv[i]); grpc_metadata_array_init(&f->request_metadata_recv[i]);
gpr_join_host_port(&f->servers_hostports[i], server_host, gpr_join_host_port(&f->servers_hostports[i], server_host,
@ -212,8 +215,8 @@ static void teardown_servers(servers_fixture *f) {
/* Destroy server. */ /* Destroy server. */
for (i = 0; i < f->num_servers; i++) { for (i = 0; i < f->num_servers; i++) {
if (f->servers[i] == NULL) continue; if (f->servers[i] == NULL) continue;
grpc_server_shutdown_and_notify(f->servers[i], f->cq, tag(10000)); grpc_server_shutdown_and_notify(f->servers[i], f->shutdown_cq, tag(10000));
GPR_ASSERT(grpc_completion_queue_pluck(f->cq, tag(10000), GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(10000),
n_millis_time(5000), NULL) n_millis_time(5000), NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->servers[i]); grpc_server_destroy(f->servers[i]);
@ -221,6 +224,7 @@ static void teardown_servers(servers_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
gpr_free(f->servers); gpr_free(f->servers);

@ -130,7 +130,7 @@ static gpr_timespec n_sec_deadline(int seconds) {
} }
static void start_rpc(int use_creds, int target_port) { static void start_rpc(int use_creds, int target_port) {
state.cq = grpc_completion_queue_create(NULL); state.cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
if (use_creds) { if (use_creds) {
state.creds = grpc_fake_transport_security_credentials_create(); state.creds = grpc_fake_transport_security_credentials_create();
} else { } else {

@ -178,7 +178,7 @@ static void start_rpc(int target_port, grpc_status_code expected_status,
cq_verifier *cqv; cq_verifier *cqv;
grpc_slice details; grpc_slice details;
state.cq = grpc_completion_queue_create(NULL); state.cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
cqv = cq_verifier_create(state.cq); cqv = cq_verifier_create(state.cq);
gpr_join_host_port(&state.target, "127.0.0.1", target_port); gpr_join_host_port(&state.target, "127.0.0.1", target_port);
state.channel = grpc_insecure_channel_create(state.target, NULL, NULL); state.channel = grpc_insecure_channel_create(state.target, NULL, NULL);

@ -69,7 +69,7 @@ static void run_test(bool wait_for_ready, bool use_service_config) {
grpc_metadata_array_init(&trailing_metadata_recv); grpc_metadata_array_init(&trailing_metadata_recv);
cq = grpc_completion_queue_create(NULL); cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
cqv = cq_verifier_create(cq); cqv = cq_verifier_create(cq);
/* if using service config, create channel args */ /* if using service config, create channel args */

@ -76,6 +76,7 @@ void test_connect(const char *server_host, const char *client_host, int port,
grpc_channel *client; grpc_channel *client;
grpc_server *server; grpc_server *server;
grpc_completion_queue *cq; grpc_completion_queue *cq;
grpc_completion_queue *shutdown_cq;
grpc_call *c; grpc_call *c;
grpc_call *s; grpc_call *s;
cq_verifier *cqv; cq_verifier *cqv;
@ -107,7 +108,7 @@ void test_connect(const char *server_host, const char *client_host, int port,
grpc_call_details_init(&call_details); grpc_call_details_init(&call_details);
/* Create server. */ /* Create server. */
cq = grpc_completion_queue_create(NULL); cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
server = grpc_server_create(NULL, NULL); server = grpc_server_create(NULL, NULL);
grpc_server_register_completion_queue(server, cq, NULL); grpc_server_register_completion_queue(server, cq, NULL);
GPR_ASSERT((got_port = grpc_server_add_insecure_http2_port( GPR_ASSERT((got_port = grpc_server_add_insecure_http2_port(
@ -259,11 +260,14 @@ void test_connect(const char *server_host, const char *client_host, int port,
grpc_channel_destroy(client); grpc_channel_destroy(client);
/* Destroy server. */ /* Destroy server. */
grpc_server_shutdown_and_notify(server, cq, tag(1000)); shutdown_cq = grpc_completion_queue_create(GRPC_CQ_PLUCK, NON_POLLING, NULL);
GPR_ASSERT(grpc_completion_queue_pluck( grpc_server_shutdown_and_notify(server, shutdown_cq, tag(1000));
cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) GPR_ASSERT(grpc_completion_queue_pluck(shutdown_cq, tag(1000),
grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(server); grpc_server_destroy(server);
grpc_completion_queue_destroy(shutdown_cq);
grpc_completion_queue_shutdown(cq); grpc_completion_queue_shutdown(cq);
drain_cq(cq); drain_cq(cq);
grpc_completion_queue_destroy(cq); grpc_completion_queue_destroy(cq);

@ -50,6 +50,7 @@ typedef struct grpc_end2end_test_config grpc_end2end_test_config;
struct grpc_end2end_test_fixture { struct grpc_end2end_test_fixture {
grpc_completion_queue *cq; grpc_completion_queue *cq;
grpc_completion_queue *shutdown_cq;
grpc_server *server; grpc_server *server;
grpc_channel *client; grpc_channel *client;
void *fixture_data; void *fixture_data;

@ -65,7 +65,9 @@ static grpc_end2end_test_fixture chttp2_create_fixture_fullstack(
gpr_join_host_port(&ffd->localaddr, "localhost", port); gpr_join_host_port(&ffd->localaddr, "localhost", port);
f.fixture_data = ffd; f.fixture_data = ffd;
f.cq = grpc_completion_queue_create(NULL); f.cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
f.shutdown_cq =
grpc_completion_queue_create(GRPC_CQ_PLUCK, NON_POLLING, NULL);
return f; return f;
} }
@ -119,9 +121,10 @@ void chttp2_tear_down_fullstack(grpc_end2end_test_fixture *f) {
/* All test configurations */ /* All test configurations */
static grpc_end2end_test_config configs[] = { static grpc_end2end_test_config configs[] = {
{"chttp2/fullstack+census", FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION | {"chttp2/fullstack+census",
FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL | FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION |
FEATURE_MASK_SUPPORTS_AUTHORITY_HEADER, FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL |
FEATURE_MASK_SUPPORTS_AUTHORITY_HEADER,
chttp2_create_fixture_fullstack, chttp2_init_client_fullstack, chttp2_create_fixture_fullstack, chttp2_init_client_fullstack,
chttp2_init_server_fullstack, chttp2_tear_down_fullstack}, chttp2_init_server_fullstack, chttp2_tear_down_fullstack},
}; };

@ -69,7 +69,9 @@ static grpc_end2end_test_fixture chttp2_create_fixture_fullstack_compression(
memset(&f, 0, sizeof(f)); memset(&f, 0, sizeof(f));
f.fixture_data = ffd; f.fixture_data = ffd;
f.cq = grpc_completion_queue_create(NULL); f.cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
f.shutdown_cq =
grpc_completion_queue_create(GRPC_CQ_PLUCK, NON_POLLING, NULL);
return f; return f;
} }
@ -119,9 +121,10 @@ void chttp2_tear_down_fullstack_compression(grpc_end2end_test_fixture *f) {
/* All test configurations */ /* All test configurations */
static grpc_end2end_test_config configs[] = { static grpc_end2end_test_config configs[] = {
{"chttp2/fullstack_compression", FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION | {"chttp2/fullstack_compression",
FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL | FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION |
FEATURE_MASK_SUPPORTS_AUTHORITY_HEADER, FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL |
FEATURE_MASK_SUPPORTS_AUTHORITY_HEADER,
chttp2_create_fixture_fullstack_compression, chttp2_create_fixture_fullstack_compression,
chttp2_init_client_fullstack_compression, chttp2_init_client_fullstack_compression,
chttp2_init_server_fullstack_compression, chttp2_init_server_fullstack_compression,

@ -60,7 +60,9 @@ static grpc_end2end_test_fixture chttp2_create_fixture_secure_fullstack(
gpr_join_host_port(&ffd->localaddr, "localhost", port); gpr_join_host_port(&ffd->localaddr, "localhost", port);
f.fixture_data = ffd; f.fixture_data = ffd;
f.cq = grpc_completion_queue_create(NULL); f.cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
f.shutdown_cq =
grpc_completion_queue_create(GRPC_CQ_PLUCK, NON_POLLING, NULL);
return f; return f;
} }

@ -70,7 +70,9 @@ static grpc_end2end_test_fixture chttp2_create_fixture_socketpair(
grpc_end2end_test_fixture f; grpc_end2end_test_fixture f;
memset(&f, 0, sizeof(f)); memset(&f, 0, sizeof(f));
f.fixture_data = fixture_data; f.fixture_data = fixture_data;
f.cq = grpc_completion_queue_create(NULL); f.cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
f.shutdown_cq =
grpc_completion_queue_create(GRPC_CQ_PLUCK, NON_POLLING, NULL);
create_sockets(fixture_data->fd_pair); create_sockets(fixture_data->fd_pair);

@ -70,7 +70,9 @@ static grpc_end2end_test_fixture chttp2_create_fixture_fullstack(
gpr_join_host_port(&ffd->localaddr, "localhost", port); gpr_join_host_port(&ffd->localaddr, "localhost", port);
f.fixture_data = ffd; f.fixture_data = ffd;
f.cq = grpc_completion_queue_create(NULL); f.cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
f.shutdown_cq =
grpc_completion_queue_create(GRPC_CQ_PLUCK, NON_POLLING, NULL);
return f; return f;
} }
@ -102,9 +104,10 @@ void chttp2_tear_down_fullstack(grpc_end2end_test_fixture *f) {
/* All test configurations */ /* All test configurations */
static grpc_end2end_test_config configs[] = { static grpc_end2end_test_config configs[] = {
{"chttp2/fullstack", FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION | {"chttp2/fullstack",
FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL | FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION |
FEATURE_MASK_SUPPORTS_AUTHORITY_HEADER, FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL |
FEATURE_MASK_SUPPORTS_AUTHORITY_HEADER,
chttp2_create_fixture_fullstack, chttp2_init_client_fullstack, chttp2_create_fixture_fullstack, chttp2_init_client_fullstack,
chttp2_init_server_fullstack, chttp2_tear_down_fullstack}, chttp2_init_server_fullstack, chttp2_tear_down_fullstack},
}; };

@ -70,7 +70,9 @@ static grpc_end2end_test_fixture chttp2_create_fixture_fullstack(
gpr_join_host_port(&ffd->localaddr, "localhost", port); gpr_join_host_port(&ffd->localaddr, "localhost", port);
f.fixture_data = ffd; f.fixture_data = ffd;
f.cq = grpc_completion_queue_create(NULL); f.cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
f.shutdown_cq =
grpc_completion_queue_create(GRPC_CQ_PLUCK, NON_POLLING, NULL);
return f; return f;
} }
@ -102,9 +104,10 @@ void chttp2_tear_down_fullstack(grpc_end2end_test_fixture *f) {
/* All test configurations */ /* All test configurations */
static grpc_end2end_test_config configs[] = { static grpc_end2end_test_config configs[] = {
{"chttp2/fullstack", FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION | {"chttp2/fullstack",
FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL | FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION |
FEATURE_MASK_SUPPORTS_AUTHORITY_HEADER, FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL |
FEATURE_MASK_SUPPORTS_AUTHORITY_HEADER,
chttp2_create_fixture_fullstack, chttp2_init_client_fullstack, chttp2_create_fixture_fullstack, chttp2_init_client_fullstack,
chttp2_init_server_fullstack, chttp2_tear_down_fullstack}, chttp2_init_server_fullstack, chttp2_tear_down_fullstack},
}; };

@ -64,7 +64,9 @@ static grpc_end2end_test_fixture chttp2_create_fixture_fullstack(
gpr_join_host_port(&ffd->localaddr, "localhost", port); gpr_join_host_port(&ffd->localaddr, "localhost", port);
f.fixture_data = ffd; f.fixture_data = ffd;
f.cq = grpc_completion_queue_create(NULL); f.cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
f.shutdown_cq =
grpc_completion_queue_create(GRPC_CQ_PLUCK, NON_POLLING, NULL);
return f; return f;
} }
@ -96,9 +98,10 @@ void chttp2_tear_down_fullstack(grpc_end2end_test_fixture *f) {
/* All test configurations */ /* All test configurations */
static grpc_end2end_test_config configs[] = { static grpc_end2end_test_config configs[] = {
{"chttp2/fullstack", FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION | {"chttp2/fullstack",
FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL | FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION |
FEATURE_MASK_SUPPORTS_AUTHORITY_HEADER, FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL |
FEATURE_MASK_SUPPORTS_AUTHORITY_HEADER,
chttp2_create_fixture_fullstack, chttp2_init_client_fullstack, chttp2_create_fixture_fullstack, chttp2_init_client_fullstack,
chttp2_init_server_fullstack, chttp2_tear_down_fullstack}, chttp2_init_server_fullstack, chttp2_tear_down_fullstack},
}; };

@ -69,7 +69,9 @@ static grpc_end2end_test_fixture chttp2_create_fixture_fullstack(
ffd->proxy = grpc_end2end_http_proxy_create(); ffd->proxy = grpc_end2end_http_proxy_create();
f.fixture_data = ffd; f.fixture_data = ffd;
f.cq = grpc_completion_queue_create(NULL); f.cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
f.shutdown_cq =
grpc_completion_queue_create(GRPC_CQ_PLUCK, NON_POLLING, NULL);
return f; return f;
} }
@ -107,9 +109,10 @@ void chttp2_tear_down_fullstack(grpc_end2end_test_fixture *f) {
/* All test configurations */ /* All test configurations */
static grpc_end2end_test_config configs[] = { static grpc_end2end_test_config configs[] = {
{"chttp2/fullstack", FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION | {"chttp2/fullstack",
FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL | FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION |
FEATURE_MASK_SUPPORTS_AUTHORITY_HEADER, FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL |
FEATURE_MASK_SUPPORTS_AUTHORITY_HEADER,
chttp2_create_fixture_fullstack, chttp2_init_client_fullstack, chttp2_create_fixture_fullstack, chttp2_init_client_fullstack,
chttp2_init_server_fullstack, chttp2_tear_down_fullstack}, chttp2_init_server_fullstack, chttp2_tear_down_fullstack},
}; };

@ -67,7 +67,9 @@ static grpc_end2end_test_fixture chttp2_create_fixture_load_reporting(
gpr_join_host_port(&ffd->localaddr, "localhost", port); gpr_join_host_port(&ffd->localaddr, "localhost", port);
f.fixture_data = ffd; f.fixture_data = ffd;
f.cq = grpc_completion_queue_create(NULL); f.cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
f.shutdown_cq =
grpc_completion_queue_create(GRPC_CQ_PLUCK, NON_POLLING, NULL);
return f; return f;
} }

@ -113,7 +113,9 @@ static grpc_end2end_test_fixture chttp2_create_fixture_secure_fullstack(
gpr_join_host_port(&ffd->localaddr, "localhost", port); gpr_join_host_port(&ffd->localaddr, "localhost", port);
f.fixture_data = ffd; f.fixture_data = ffd;
f.cq = grpc_completion_queue_create(NULL); f.cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
f.shutdown_cq =
grpc_completion_queue_create(GRPC_CQ_PLUCK, NON_POLLING, NULL);
return f; return f;
} }

@ -79,7 +79,9 @@ static grpc_end2end_test_fixture chttp2_create_fixture_fullstack(
ffd->proxy = grpc_end2end_proxy_create(&proxy_def, client_args, server_args); ffd->proxy = grpc_end2end_proxy_create(&proxy_def, client_args, server_args);
f.fixture_data = ffd; f.fixture_data = ffd;
f.cq = grpc_completion_queue_create(NULL); f.cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
f.shutdown_cq =
grpc_completion_queue_create(GRPC_CQ_PLUCK, NON_POLLING, NULL);
return f; return f;
} }
@ -113,10 +115,11 @@ void chttp2_tear_down_fullstack(grpc_end2end_test_fixture *f) {
/* All test configurations */ /* All test configurations */
static grpc_end2end_test_config configs[] = { static grpc_end2end_test_config configs[] = {
{"chttp2/fullstack+proxy", FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION | {"chttp2/fullstack+proxy",
FEATURE_MASK_SUPPORTS_REQUEST_PROXYING | FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION |
FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL | FEATURE_MASK_SUPPORTS_REQUEST_PROXYING |
FEATURE_MASK_SUPPORTS_AUTHORITY_HEADER, FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL |
FEATURE_MASK_SUPPORTS_AUTHORITY_HEADER,
chttp2_create_fixture_fullstack, chttp2_init_client_fullstack, chttp2_create_fixture_fullstack, chttp2_init_client_fullstack,
chttp2_init_server_fullstack, chttp2_tear_down_fullstack}, chttp2_init_server_fullstack, chttp2_tear_down_fullstack},
}; };

@ -94,7 +94,9 @@ static grpc_end2end_test_fixture chttp2_create_fixture_socketpair(
grpc_end2end_test_fixture f; grpc_end2end_test_fixture f;
memset(&f, 0, sizeof(f)); memset(&f, 0, sizeof(f));
f.fixture_data = sfd; f.fixture_data = sfd;
f.cq = grpc_completion_queue_create(NULL); f.cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
f.shutdown_cq =
grpc_completion_queue_create(GRPC_CQ_PLUCK, NON_POLLING, NULL);
grpc_resource_quota *resource_quota = grpc_resource_quota_create("fixture"); grpc_resource_quota *resource_quota = grpc_resource_quota_create("fixture");
*sfd = grpc_iomgr_create_endpoint_pair("fixture", resource_quota, 65536); *sfd = grpc_iomgr_create_endpoint_pair("fixture", resource_quota, 65536);

@ -88,7 +88,9 @@ static grpc_end2end_test_fixture chttp2_create_fixture_socketpair(
grpc_end2end_test_fixture f; grpc_end2end_test_fixture f;
memset(&f, 0, sizeof(f)); memset(&f, 0, sizeof(f));
f.fixture_data = sfd; f.fixture_data = sfd;
f.cq = grpc_completion_queue_create(NULL); f.cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
f.shutdown_cq =
grpc_completion_queue_create(GRPC_CQ_PLUCK, NON_POLLING, NULL);
grpc_resource_quota *resource_quota = grpc_resource_quota_create("fixture"); grpc_resource_quota *resource_quota = grpc_resource_quota_create("fixture");
*sfd = grpc_iomgr_create_endpoint_pair("fixture", resource_quota, 65536); *sfd = grpc_iomgr_create_endpoint_pair("fixture", resource_quota, 65536);

@ -88,7 +88,9 @@ static grpc_end2end_test_fixture chttp2_create_fixture_socketpair(
grpc_end2end_test_fixture f; grpc_end2end_test_fixture f;
memset(&f, 0, sizeof(f)); memset(&f, 0, sizeof(f));
f.fixture_data = sfd; f.fixture_data = sfd;
f.cq = grpc_completion_queue_create(NULL); f.cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
f.shutdown_cq =
grpc_completion_queue_create(GRPC_CQ_PLUCK, NON_POLLING, NULL);
grpc_resource_quota *resource_quota = grpc_resource_quota_create("fixture"); grpc_resource_quota *resource_quota = grpc_resource_quota_create("fixture");
*sfd = grpc_iomgr_create_endpoint_pair("fixture", resource_quota, 1); *sfd = grpc_iomgr_create_endpoint_pair("fixture", resource_quota, 1);

@ -64,7 +64,9 @@ static grpc_end2end_test_fixture chttp2_create_fixture_secure_fullstack(
gpr_join_host_port(&ffd->localaddr, "localhost", port); gpr_join_host_port(&ffd->localaddr, "localhost", port);
f.fixture_data = ffd; f.fixture_data = ffd;
f.cq = grpc_completion_queue_create(NULL); f.cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
f.shutdown_cq =
grpc_completion_queue_create(GRPC_CQ_PLUCK, NON_POLLING, NULL);
return f; return f;
} }

@ -67,7 +67,9 @@ static grpc_end2end_test_fixture chttp2_create_fixture_secure_fullstack(
gpr_join_host_port(&ffd->localaddr, "localhost", port); gpr_join_host_port(&ffd->localaddr, "localhost", port);
f.fixture_data = ffd; f.fixture_data = ffd;
f.cq = grpc_completion_queue_create(NULL); f.cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
f.shutdown_cq =
grpc_completion_queue_create(GRPC_CQ_PLUCK, NON_POLLING, NULL);
return f; return f;
} }
@ -203,15 +205,17 @@ CLIENT_INIT(BAD_CERT_PAIR)
typedef enum { SUCCESS, FAIL } test_result; typedef enum { SUCCESS, FAIL } test_result;
#define SSL_TEST(request_type, cert_type, result) \ #define SSL_TEST(request_type, cert_type, result) \
{ \ { \
{TEST_NAME(request_type, cert_type, result), \ {TEST_NAME(request_type, cert_type, result), \
FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION | \ FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION | \
FEATURE_MASK_SUPPORTS_PER_CALL_CREDENTIALS | \ FEATURE_MASK_SUPPORTS_PER_CALL_CREDENTIALS | \
FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL, \ FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL, \
chttp2_create_fixture_secure_fullstack, CLIENT_INIT_NAME(cert_type), \ chttp2_create_fixture_secure_fullstack, \
SERVER_INIT_NAME(request_type), chttp2_tear_down_secure_fullstack}, \ CLIENT_INIT_NAME(cert_type), \
result \ SERVER_INIT_NAME(request_type), \
chttp2_tear_down_secure_fullstack}, \
result \
} }
/* All test configurations */ /* All test configurations */
@ -289,9 +293,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -310,6 +315,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
static void simple_request_body(grpc_end2end_test_fixture f, static void simple_request_body(grpc_end2end_test_fixture f,

@ -100,7 +100,9 @@ static grpc_end2end_test_fixture chttp2_create_fixture_secure_fullstack(
ffd->proxy = grpc_end2end_proxy_create(&proxy_def, client_args, server_args); ffd->proxy = grpc_end2end_proxy_create(&proxy_def, client_args, server_args);
f.fixture_data = ffd; f.fixture_data = ffd;
f.cq = grpc_completion_queue_create(NULL); f.cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
f.shutdown_cq =
grpc_completion_queue_create(GRPC_CQ_PLUCK, NON_POLLING, NULL);
return f; return f;
} }

@ -70,7 +70,9 @@ static grpc_end2end_test_fixture chttp2_create_fixture_fullstack(
unique++); unique++);
f.fixture_data = ffd; f.fixture_data = ffd;
f.cq = grpc_completion_queue_create(NULL); f.cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
f.shutdown_cq =
grpc_completion_queue_create(GRPC_CQ_PLUCK, NON_POLLING, NULL);
return f; return f;
} }
@ -101,9 +103,10 @@ void chttp2_tear_down_fullstack(grpc_end2end_test_fixture *f) {
/* All test configurations */ /* All test configurations */
static grpc_end2end_test_config configs[] = { static grpc_end2end_test_config configs[] = {
{"chttp2/fullstack_uds", FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION | {"chttp2/fullstack_uds",
FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL | FEATURE_MASK_SUPPORTS_DELAYED_CONNECTION |
FEATURE_MASK_SUPPORTS_AUTHORITY_HEADER, FEATURE_MASK_SUPPORTS_CLIENT_CHANNEL |
FEATURE_MASK_SUPPORTS_AUTHORITY_HEADER,
chttp2_create_fixture_fullstack, chttp2_init_client_fullstack, chttp2_create_fixture_fullstack, chttp2_init_client_fullstack,
chttp2_init_server_fullstack, chttp2_tear_down_fullstack}, chttp2_init_server_fullstack, chttp2_tear_down_fullstack},
}; };

@ -104,7 +104,7 @@ grpc_end2end_proxy *grpc_end2end_proxy_create(const grpc_end2end_proxy_def *def,
gpr_log(GPR_DEBUG, "PROXY ADDR:%s BACKEND:%s", proxy->proxy_port, gpr_log(GPR_DEBUG, "PROXY ADDR:%s BACKEND:%s", proxy->proxy_port,
proxy->server_port); proxy->server_port);
proxy->cq = grpc_completion_queue_create(NULL); proxy->cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
proxy->server = def->create_server(proxy->proxy_port, server_args); proxy->server = def->create_server(proxy->proxy_port, server_args);
proxy->client = def->create_client(proxy->server_port, client_args); proxy->client = def->create_client(proxy->server_port, client_args);

@ -314,8 +314,9 @@ static grpc_call_credentials *read_call_creds(input_stream *inp) {
cred_artifact_ctx ctx = CRED_ARTIFACT_CTX_INIT; cred_artifact_ctx ctx = CRED_ARTIFACT_CTX_INIT;
const char *access_token = read_cred_artifact(&ctx, inp, NULL, 0); const char *access_token = read_cred_artifact(&ctx, inp, NULL, 0);
grpc_call_credentials *out = grpc_call_credentials *out =
access_token == NULL ? NULL : grpc_access_token_credentials_create( access_token == NULL
access_token, NULL); ? NULL
: grpc_access_token_credentials_create(access_token, NULL);
cred_artifact_ctx_finish(&ctx); cred_artifact_ctx_finish(&ctx);
return out; return out;
} }
@ -409,8 +410,9 @@ void my_resolve_address(grpc_exec_ctx *exec_ctx, const char *addr,
r->on_done = on_done; r->on_done = on_done;
r->addrs = addresses; r->addrs = addresses;
grpc_timer_init( grpc_timer_init(
exec_ctx, &r->timer, gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC), exec_ctx, &r->timer,
gpr_time_from_seconds(1, GPR_TIMESPAN)), gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC),
gpr_time_from_seconds(1, GPR_TIMESPAN)),
grpc_closure_create(finish_resolve, r, grpc_schedule_on_exec_ctx), grpc_closure_create(finish_resolve, r, grpc_schedule_on_exec_ctx),
gpr_now(GPR_CLOCK_MONOTONIC)); gpr_now(GPR_CLOCK_MONOTONIC));
} }
@ -471,8 +473,9 @@ static void sched_connect(grpc_exec_ctx *exec_ctx, grpc_closure *closure,
fc->ep = ep; fc->ep = ep;
fc->deadline = deadline; fc->deadline = deadline;
grpc_timer_init( grpc_timer_init(
exec_ctx, &fc->timer, gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC), exec_ctx, &fc->timer,
gpr_time_from_millis(1, GPR_TIMESPAN)), gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC),
gpr_time_from_millis(1, GPR_TIMESPAN)),
grpc_closure_create(do_connect, fc, grpc_schedule_on_exec_ctx), grpc_closure_create(do_connect, fc, grpc_schedule_on_exec_ctx),
gpr_now(GPR_CLOCK_MONOTONIC)); gpr_now(GPR_CLOCK_MONOTONIC));
} }
@ -735,7 +738,8 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
g_active_call = new_call(NULL, ROOT); g_active_call = new_call(NULL, ROOT);
g_resource_quota = grpc_resource_quota_create("api_fuzzer"); g_resource_quota = grpc_resource_quota_create("api_fuzzer");
grpc_completion_queue *cq = grpc_completion_queue_create(NULL); grpc_completion_queue *cq =
grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
while (!is_eof(&inp) || g_channel != NULL || g_server != NULL || while (!is_eof(&inp) || g_channel != NULL || g_server != NULL ||
pending_channel_watches > 0 || pending_pings > 0 || pending_channel_watches > 0 || pending_pings > 0 ||
@ -748,8 +752,9 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
if (g_server != NULL) { if (g_server != NULL) {
if (!server_shutdown) { if (!server_shutdown) {
grpc_server_shutdown_and_notify( grpc_server_shutdown_and_notify(
g_server, cq, create_validator(assert_success_and_decrement, g_server, cq,
&pending_server_shutdowns)); create_validator(assert_success_and_decrement,
&pending_server_shutdowns));
server_shutdown = true; server_shutdown = true;
pending_server_shutdowns++; pending_server_shutdowns++;
} else if (pending_server_shutdowns == 0) { } else if (pending_server_shutdowns == 0) {
@ -854,8 +859,9 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
case 5: { case 5: {
if (g_server != NULL) { if (g_server != NULL) {
grpc_server_shutdown_and_notify( grpc_server_shutdown_and_notify(
g_server, cq, create_validator(assert_success_and_decrement, g_server, cq,
&pending_server_shutdowns)); create_validator(assert_success_and_decrement,
&pending_server_shutdowns));
pending_server_shutdowns++; pending_server_shutdowns++;
server_shutdown = true; server_shutdown = true;
} else { } else {

@ -65,7 +65,8 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
grpc_mock_endpoint_create(discard_write, resource_quota); grpc_mock_endpoint_create(discard_write, resource_quota);
grpc_resource_quota_unref_internal(&exec_ctx, resource_quota); grpc_resource_quota_unref_internal(&exec_ctx, resource_quota);
grpc_completion_queue *cq = grpc_completion_queue_create(NULL); grpc_completion_queue *cq =
grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
grpc_transport *transport = grpc_transport *transport =
grpc_create_chttp2_transport(&exec_ctx, NULL, mock_endpoint, 1); grpc_create_chttp2_transport(&exec_ctx, NULL, mock_endpoint, 1);
grpc_chttp2_transport_start_reading(&exec_ctx, transport, NULL); grpc_chttp2_transport_start_reading(&exec_ctx, transport, NULL);

@ -67,7 +67,8 @@ int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
grpc_slice_from_copied_buffer((const char *)data, size)); grpc_slice_from_copied_buffer((const char *)data, size));
grpc_server *server = grpc_server_create(NULL, NULL); grpc_server *server = grpc_server_create(NULL, NULL);
grpc_completion_queue *cq = grpc_completion_queue_create(NULL); grpc_completion_queue *cq =
grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
grpc_server_register_completion_queue(server, cq, NULL); grpc_server_register_completion_queue(server, cq, NULL);
// TODO(ctiller): add registered methods (one for POST, one for PUT) // TODO(ctiller): add registered methods (one for POST, one for PUT)
// void *registered_method = // void *registered_method =

@ -121,7 +121,7 @@ int main(int argc, char **argv) {
grpc_metadata_array_init(&request_metadata2); grpc_metadata_array_init(&request_metadata2);
grpc_call_details_init(&request_details2); grpc_call_details_init(&request_details2);
cq = grpc_completion_queue_create(NULL); cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
cqv = cq_verifier_create(cq); cqv = cq_verifier_create(cq);
/* reserve two ports */ /* reserve two ports */

@ -73,7 +73,8 @@ static void prepare_test(int is_client) {
grpc_metadata_array_init(&g_state.initial_metadata_recv); grpc_metadata_array_init(&g_state.initial_metadata_recv);
grpc_metadata_array_init(&g_state.trailing_metadata_recv); grpc_metadata_array_init(&g_state.trailing_metadata_recv);
g_state.deadline = grpc_timeout_seconds_to_deadline(2); g_state.deadline = grpc_timeout_seconds_to_deadline(2);
g_state.cq = grpc_completion_queue_create(NULL); g_state.cq =
grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
g_state.cqv = cq_verifier_create(g_state.cq); g_state.cqv = cq_verifier_create(g_state.cq);
g_state.details = grpc_empty_slice(); g_state.details = grpc_empty_slice();
memset(g_state.ops, 0, sizeof(g_state.ops)); memset(g_state.ops, 0, sizeof(g_state.ops));
@ -123,6 +124,7 @@ static void prepare_test(int is_client) {
} }
static void cleanup_test() { static void cleanup_test() {
grpc_completion_queue *shutdown_cq;
grpc_call_destroy(g_state.call); grpc_call_destroy(g_state.call);
cq_verifier_destroy(g_state.cqv); cq_verifier_destroy(g_state.cqv);
grpc_channel_destroy(g_state.chan); grpc_channel_destroy(g_state.chan);
@ -131,12 +133,15 @@ static void cleanup_test() {
grpc_metadata_array_destroy(&g_state.trailing_metadata_recv); grpc_metadata_array_destroy(&g_state.trailing_metadata_recv);
if (!g_state.is_client) { if (!g_state.is_client) {
shutdown_cq =
grpc_completion_queue_create(GRPC_CQ_PLUCK, NON_POLLING, NULL);
grpc_call_destroy(g_state.server_call); grpc_call_destroy(g_state.server_call);
grpc_server_shutdown_and_notify(g_state.server, g_state.cq, tag(1000)); grpc_server_shutdown_and_notify(g_state.server, shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck(g_state.cq, tag(1000), GPR_ASSERT(grpc_completion_queue_pluck(shutdown_cq, tag(1000),
grpc_timeout_seconds_to_deadline(5), grpc_timeout_seconds_to_deadline(5),
NULL) NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_completion_queue_destroy(shutdown_cq);
grpc_server_destroy(g_state.server); grpc_server_destroy(g_state.server);
grpc_call_details_destroy(&g_state.call_details); grpc_call_details_destroy(&g_state.call_details);
grpc_metadata_array_destroy(&g_state.server_initial_metadata_recv); grpc_metadata_array_destroy(&g_state.server_initial_metadata_recv);

@ -37,27 +37,37 @@
int main(int argc, char **argv) { int main(int argc, char **argv) {
grpc_completion_queue *cq1; grpc_completion_queue *cq1;
grpc_completion_queue *cq2; grpc_completion_queue *cq2;
grpc_completion_queue *cq3;
grpc_server *server; grpc_server *server;
grpc_test_init(argc, argv); grpc_test_init(argc, argv);
grpc_init(); grpc_init();
cq1 = grpc_completion_queue_create(NULL); cq1 = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
cq2 = grpc_completion_queue_create(NULL); cq2 = grpc_completion_queue_create(GRPC_CQ_NEXT, NON_LISTENING, NULL);
cq3 = grpc_completion_queue_create(GRPC_CQ_NEXT, NON_POLLING, NULL);
server = grpc_server_create(NULL, NULL); server = grpc_server_create(NULL, NULL);
grpc_server_register_completion_queue(server, cq1, NULL); grpc_server_register_completion_queue(server, cq1, NULL);
grpc_server_add_insecure_http2_port(server, "[::]:0"); grpc_server_add_insecure_http2_port(server, "[::]:0");
grpc_server_register_completion_queue(server, cq2, NULL); grpc_server_register_completion_queue(server, cq2, NULL);
grpc_server_register_completion_queue(server, cq3, NULL);
grpc_server_start(server); grpc_server_start(server);
grpc_server_shutdown_and_notify(server, cq2, NULL); grpc_server_shutdown_and_notify(server, cq2, NULL);
grpc_completion_queue_next(cq2, gpr_inf_future(GPR_CLOCK_REALTIME), grpc_completion_queue_next(cq2, gpr_inf_future(GPR_CLOCK_REALTIME),
NULL); /* cue queue hang */ NULL); /* cue queue hang */
grpc_completion_queue_shutdown(cq1); grpc_completion_queue_shutdown(cq1);
grpc_completion_queue_shutdown(cq2); grpc_completion_queue_shutdown(cq2);
grpc_completion_queue_shutdown(cq3);
grpc_completion_queue_next(cq1, gpr_inf_future(GPR_CLOCK_REALTIME), NULL); grpc_completion_queue_next(cq1, gpr_inf_future(GPR_CLOCK_REALTIME), NULL);
grpc_completion_queue_next(cq2, gpr_inf_future(GPR_CLOCK_REALTIME), NULL); grpc_completion_queue_next(cq2, gpr_inf_future(GPR_CLOCK_REALTIME), NULL);
grpc_completion_queue_next(cq3, gpr_inf_future(GPR_CLOCK_REALTIME), NULL);
grpc_server_destroy(server); grpc_server_destroy(server);
grpc_completion_queue_destroy(cq1); grpc_completion_queue_destroy(cq1);
grpc_completion_queue_destroy(cq2); grpc_completion_queue_destroy(cq2);
grpc_completion_queue_destroy(cq3);
grpc_shutdown(); grpc_shutdown();
return 0; return 0;
} }

@ -59,7 +59,7 @@ int main(int argc, char **argv) {
grpc_metadata_array_init(&trailing_metadata_recv); grpc_metadata_array_init(&trailing_metadata_recv);
cq = grpc_completion_queue_create(NULL); cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
cqv = cq_verifier_create(cq); cqv = cq_verifier_create(cq);
/* create a call, channel to a non existant server */ /* create a call, channel to a non existant server */

@ -72,9 +72,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -93,6 +94,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
/* Request/response with metadata and payload.*/ /* Request/response with metadata and payload.*/

@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
static void simple_request_body(grpc_end2end_test_fixture f) { static void simple_request_body(grpc_end2end_test_fixture f) {

@ -72,9 +72,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -93,6 +94,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
/* Request/response with metadata and payload.*/ /* Request/response with metadata and payload.*/

@ -90,9 +90,9 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) f->shutdown_cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -111,6 +111,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
static void print_auth_context(int is_client, const grpc_auth_context *ctx) { static void print_auth_context(int is_client, const grpc_auth_context *ctx) {

@ -79,9 +79,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -100,6 +101,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
/* Cancel after accept, no payload */ /* Cancel after accept, no payload */

@ -73,9 +73,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -94,6 +95,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
/* Cancel after accept with a writes closed, no payload */ /* Cancel after accept with a writes closed, no payload */

@ -75,9 +75,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -96,6 +97,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
/* Cancel after invoke, no payload */ /* Cancel after invoke, no payload */

@ -72,9 +72,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -93,6 +94,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
/* Cancel before invoke */ /* Cancel before invoke */

@ -73,9 +73,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -94,6 +95,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
/* Cancel and do nothing */ /* Cancel and do nothing */

@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
static void simple_request_body(grpc_end2end_test_config config, static void simple_request_body(grpc_end2end_test_config config,

@ -80,9 +80,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -101,6 +102,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
static void request_for_disabled_algorithm( static void request_for_disabled_algorithm(

@ -171,6 +171,9 @@ static void test_connectivity(grpc_end2end_test_config config) {
grpc_channel_destroy(f.client); grpc_channel_destroy(f.client);
grpc_completion_queue_shutdown(f.cq); grpc_completion_queue_shutdown(f.cq);
grpc_completion_queue_destroy(f.cq); grpc_completion_queue_destroy(f.cq);
/* shutdown_cq is not used in this test */
grpc_completion_queue_destroy(f.shutdown_cq);
config.tear_down_data(&f); config.tear_down_data(&f);
cq_verifier_destroy(cqv); cq_verifier_destroy(cqv);

@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
static void simple_request_body(grpc_end2end_test_fixture f) { static void simple_request_body(grpc_end2end_test_fixture f) {

@ -77,6 +77,9 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
/* Note: shutdown_cq was unused in this test */
grpc_completion_queue_destroy(f->shutdown_cq);
} }
static void do_request_and_shutdown_server(grpc_end2end_test_config config, static void do_request_and_shutdown_server(grpc_end2end_test_config config,

@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
static void empty_batch_body(grpc_end2end_test_config config, static void empty_batch_body(grpc_end2end_test_config config,

@ -80,9 +80,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -101,6 +102,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
// Simple request via a server filter that always fails to initialize // Simple request via a server filter that always fails to initialize

@ -77,9 +77,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -98,6 +99,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
/* Simple request via a server filter that always closes the stream.*/ /* Simple request via a server filter that always closes the stream.*/

@ -84,9 +84,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -105,6 +106,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
// Simple request via a server filter that saves the reported latency value. // Simple request via a server filter that saves the reported latency value.

@ -89,6 +89,8 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
/* Note: shutdown_cq is not used in this test */
grpc_completion_queue_destroy(f->shutdown_cq);
} }
static void test_early_server_shutdown_finishes_inflight_calls( static void test_early_server_shutdown_finishes_inflight_calls(

@ -76,9 +76,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -97,6 +98,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
static void simple_request_body(grpc_end2end_test_config config, static void simple_request_body(grpc_end2end_test_config config,

@ -216,9 +216,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -237,6 +238,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
static void simple_request_body(grpc_end2end_test_config config, static void simple_request_body(grpc_end2end_test_config config,

@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
static void simple_request_body(grpc_end2end_test_config config, static void simple_request_body(grpc_end2end_test_config config,

@ -71,9 +71,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -92,6 +93,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
static grpc_slice large_slice(void) { static grpc_slice large_slice(void) {

@ -72,9 +72,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -93,6 +94,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
// Request with a large amount of metadata. // Request with a large amount of metadata.

@ -99,9 +99,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -120,6 +121,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
static void request_response_with_payload( static void request_response_with_payload(

@ -72,9 +72,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -93,6 +94,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
static void simple_request_body(grpc_end2end_test_config config, static void simple_request_body(grpc_end2end_test_config config,

@ -81,9 +81,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -102,6 +103,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
// Test with request larger than the limit. // Test with request larger than the limit.

@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
static void simple_request_body(grpc_end2end_test_config config, static void simple_request_body(grpc_end2end_test_config config,

@ -75,9 +75,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -96,6 +97,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
/* Client sends a request with payload, server reads then returns status. */ /* Client sends a request with payload, server reads then returns status. */

@ -102,9 +102,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -123,6 +124,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
static void simple_request_body(grpc_end2end_test_config config, static void simple_request_body(grpc_end2end_test_config config,

@ -72,9 +72,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -93,6 +94,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
static void test_no_op(grpc_end2end_test_config config) { static void test_no_op(grpc_end2end_test_config config) {

@ -72,9 +72,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -93,6 +94,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
/* Creates and returns a grpc_slice containing random alphanumeric characters. /* Creates and returns a grpc_slice containing random alphanumeric characters.

@ -95,6 +95,9 @@ static void test_ping(grpc_end2end_test_config config) {
grpc_channel_destroy(f.client); grpc_channel_destroy(f.client);
grpc_completion_queue_shutdown(f.cq); grpc_completion_queue_shutdown(f.cq);
grpc_completion_queue_destroy(f.cq); grpc_completion_queue_destroy(f.cq);
/* f.shutdown_cq is not used in this test */
grpc_completion_queue_destroy(f.shutdown_cq);
config.tear_down_data(&f); config.tear_down_data(&f);
cq_verifier_destroy(cqv); cq_verifier_destroy(cqv);

@ -72,9 +72,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -93,6 +94,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
/* Client pings and server pongs. Repeat messages rounds before finishing. */ /* Client pings and server pongs. Repeat messages rounds before finishing. */

@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
static void simple_request_body(grpc_end2end_test_config config, static void simple_request_body(grpc_end2end_test_config config,

@ -73,9 +73,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -94,6 +95,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
static void test_invoke_request_with_flags( static void test_invoke_request_with_flags(

@ -72,9 +72,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -93,6 +94,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
/* Client sends a request with payload, server reads then returns status. */ /* Client sends a request with payload, server reads then returns status. */

@ -72,9 +72,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -93,6 +94,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
/* Creates and returns a grpc_slice containing random alphanumeric characters. /* Creates and returns a grpc_slice containing random alphanumeric characters.

@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
static void simple_request_body(grpc_end2end_test_config config, static void simple_request_body(grpc_end2end_test_config config,

@ -82,6 +82,8 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
/* f->shutdown_cq is not used in this test */
grpc_completion_queue_destroy(f->shutdown_cq);
} }
static void test_early_server_shutdown_finishes_inflight_calls( static void test_early_server_shutdown_finishes_inflight_calls(

@ -82,6 +82,8 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
/* f->shutdown_cq is not used in this test */
grpc_completion_queue_destroy(f->shutdown_cq);
} }
static void test_early_server_shutdown_finishes_tags( static void test_early_server_shutdown_finishes_tags(

@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
/* Request/response with metadata and payload.*/ /* Request/response with metadata and payload.*/

@ -60,9 +60,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -81,6 +82,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
static void simple_delayed_request_body(grpc_end2end_test_config config, static void simple_delayed_request_body(grpc_end2end_test_config config,

@ -72,9 +72,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -93,6 +94,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
/* Request/response with metadata and payload.*/ /* Request/response with metadata and payload.*/

@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
static void simple_request_body(grpc_end2end_test_config config, static void simple_request_body(grpc_end2end_test_config config,

@ -74,9 +74,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -95,6 +96,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
/* Client sends a request with payload, server reads then returns status. */ /* Client sends a request with payload, server reads then returns status. */

@ -72,9 +72,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -93,6 +94,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
/* Request/response with metadata and payload.*/ /* Request/response with metadata and payload.*/

@ -72,9 +72,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -93,6 +94,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
/* Client sends a request with payload, server reads then returns status. */ /* Client sends a request with payload, server reads then returns status. */

@ -72,9 +72,10 @@ static void drain_cq(grpc_completion_queue *cq) {
static void shutdown_server(grpc_end2end_test_fixture *f) { static void shutdown_server(grpc_end2end_test_fixture *f) {
if (!f->server) return; if (!f->server) return;
grpc_server_shutdown_and_notify(f->server, f->cq, tag(1000)); grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
f->cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_timeout_seconds_to_deadline(5),
NULL)
.type == GRPC_OP_COMPLETE); .type == GRPC_OP_COMPLETE);
grpc_server_destroy(f->server); grpc_server_destroy(f->server);
f->server = NULL; f->server = NULL;
@ -93,6 +94,7 @@ static void end_test(grpc_end2end_test_fixture *f) {
grpc_completion_queue_shutdown(f->cq); grpc_completion_queue_shutdown(f->cq);
drain_cq(f->cq); drain_cq(f->cq);
grpc_completion_queue_destroy(f->cq); grpc_completion_queue_destroy(f->cq);
grpc_completion_queue_destroy(f->shutdown_cq);
} }
/* Client sends a request with payload, server reads then returns status. */ /* Client sends a request with payload, server reads then returns status. */

@ -208,7 +208,7 @@ int main(int argc, char **argv) {
} }
channel = grpc_insecure_channel_create(target, NULL, NULL); channel = grpc_insecure_channel_create(target, NULL, NULL);
cq = grpc_completion_queue_create(NULL); cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
the_buffer = grpc_raw_byte_buffer_create(&slice, (size_t)payload_size); the_buffer = grpc_raw_byte_buffer_create(&slice, (size_t)payload_size);
histogram = gpr_histogram_create(0.01, 60e9); histogram = gpr_histogram_create(0.01, 60e9);

@ -185,6 +185,7 @@ int main(int argc, char **argv) {
call_state *s; call_state *s;
char *addr_buf = NULL; char *addr_buf = NULL;
gpr_cmdline *cl; gpr_cmdline *cl;
grpc_completion_queue *shutdown_cq;
int shutdown_started = 0; int shutdown_started = 0;
int shutdown_finished = 0; int shutdown_finished = 0;
@ -214,7 +215,7 @@ int main(int argc, char **argv) {
} }
gpr_log(GPR_INFO, "creating server on: %s", addr); gpr_log(GPR_INFO, "creating server on: %s", addr);
cq = grpc_completion_queue_create(NULL); cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
if (secure) { if (secure) {
grpc_ssl_pem_key_cert_pair pem_key_cert_pair = {test_server1_key, grpc_ssl_pem_key_cert_pair pem_key_cert_pair = {test_server1_key,
test_server1_cert}; test_server1_cert};
@ -242,16 +243,24 @@ int main(int argc, char **argv) {
while (!shutdown_finished) { while (!shutdown_finished) {
if (got_sigint && !shutdown_started) { if (got_sigint && !shutdown_started) {
gpr_log(GPR_INFO, "Shutting down due to SIGINT"); gpr_log(GPR_INFO, "Shutting down due to SIGINT");
grpc_server_shutdown_and_notify(server, cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( shutdown_cq =
cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_completion_queue_create(GRPC_CQ_PLUCK, NON_POLLING, NULL);
.type == GRPC_OP_COMPLETE); grpc_server_shutdown_and_notify(server, shutdown_cq, tag(1000));
GPR_ASSERT(
grpc_completion_queue_pluck(shutdown_cq, tag(1000),
grpc_timeout_seconds_to_deadline(5), NULL)
.type == GRPC_OP_COMPLETE);
grpc_completion_queue_destroy(shutdown_cq);
grpc_completion_queue_shutdown(cq); grpc_completion_queue_shutdown(cq);
shutdown_started = 1; shutdown_started = 1;
} }
ev = grpc_completion_queue_next( ev = grpc_completion_queue_next(
cq, gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), cq,
gpr_time_from_micros(1000000, GPR_TIMESPAN)), gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
gpr_time_from_micros(1000000, GPR_TIMESPAN)),
NULL); NULL);
s = ev.tag; s = ev.tag;
switch (ev.type) { switch (ev.type) {

@ -289,7 +289,9 @@ static bool client_ssl_test(char *server_alpn_preferred) {
// completed and we know that the client's ALPN list satisfied the server. // completed and we know that the client's ALPN list satisfied the server.
int retries = 10; int retries = 10;
grpc_connectivity_state state = GRPC_CHANNEL_IDLE; grpc_connectivity_state state = GRPC_CHANNEL_IDLE;
grpc_completion_queue *cq = grpc_completion_queue_create(NULL); grpc_completion_queue *cq =
grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
while (state != GRPC_CHANNEL_READY && retries-- > 0) { while (state != GRPC_CHANNEL_READY && retries-- > 0) {
grpc_channel_watch_connectivity_state( grpc_channel_watch_connectivity_state(
channel, state, grpc_timeout_seconds_to_deadline(3), cq, NULL); channel, state, grpc_timeout_seconds_to_deadline(3), cq, NULL);

@ -104,7 +104,8 @@ static void server_thread(void *arg) {
GPR_ASSERT(grpc_server_add_secure_http2_port(server, addr, ssl_creds)); GPR_ASSERT(grpc_server_add_secure_http2_port(server, addr, ssl_creds));
free(addr); free(addr);
grpc_completion_queue *cq = grpc_completion_queue_create(NULL); grpc_completion_queue *cq =
grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
grpc_server_register_completion_queue(server, cq, NULL); grpc_server_register_completion_queue(server, cq, NULL);
grpc_server_start(server); grpc_server_start(server);

@ -222,7 +222,7 @@ int main(int argc, char **argv) {
calls[k].details = grpc_empty_slice(); calls[k].details = grpc_empty_slice();
} }
cq = grpc_completion_queue_create(NULL); cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
struct grpc_memory_counters client_channel_start = struct grpc_memory_counters client_channel_start =
grpc_memory_counters_snapshot(); grpc_memory_counters_snapshot();
@ -260,8 +260,9 @@ int main(int argc, char **argv) {
do { do {
event = grpc_completion_queue_next( event = grpc_completion_queue_next(
cq, gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), cq,
gpr_time_from_micros(10000, GPR_TIMESPAN)), gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
gpr_time_from_micros(10000, GPR_TIMESPAN)),
NULL); NULL);
} while (event.type != GRPC_QUEUE_TIMEOUT); } while (event.type != GRPC_QUEUE_TIMEOUT);

@ -161,6 +161,7 @@ int main(int argc, char **argv) {
grpc_event ev; grpc_event ev;
char *addr_buf = NULL; char *addr_buf = NULL;
gpr_cmdline *cl; gpr_cmdline *cl;
grpc_completion_queue *shutdown_cq;
int shutdown_started = 0; int shutdown_started = 0;
int shutdown_finished = 0; int shutdown_finished = 0;
@ -188,7 +189,7 @@ int main(int argc, char **argv) {
} }
gpr_log(GPR_INFO, "creating server on: %s", addr); gpr_log(GPR_INFO, "creating server on: %s", addr);
cq = grpc_completion_queue_create(NULL); cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
struct grpc_memory_counters before_server_create = struct grpc_memory_counters before_server_create =
grpc_memory_counters_snapshot(); grpc_memory_counters_snapshot();
@ -230,16 +231,22 @@ int main(int argc, char **argv) {
while (!shutdown_finished) { while (!shutdown_finished) {
if (got_sigint && !shutdown_started) { if (got_sigint && !shutdown_started) {
gpr_log(GPR_INFO, "Shutting down due to SIGINT"); gpr_log(GPR_INFO, "Shutting down due to SIGINT");
grpc_server_shutdown_and_notify(server, cq, tag(1000));
GPR_ASSERT(grpc_completion_queue_pluck( shutdown_cq =
cq, tag(1000), grpc_timeout_seconds_to_deadline(5), NULL) grpc_completion_queue_create(GRPC_CQ_PLUCK, NON_POLLING, NULL);
.type == GRPC_OP_COMPLETE); grpc_server_shutdown_and_notify(server, shutdown_cq, tag(1000));
GPR_ASSERT(
grpc_completion_queue_pluck(shutdown_cq, tag(1000),
grpc_timeout_seconds_to_deadline(5), NULL)
.type == GRPC_OP_COMPLETE);
grpc_completion_queue_destroy(shutdown_cq);
grpc_completion_queue_shutdown(cq); grpc_completion_queue_shutdown(cq);
shutdown_started = 1; shutdown_started = 1;
} }
ev = grpc_completion_queue_next( ev = grpc_completion_queue_next(
cq, gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), cq,
gpr_time_from_micros(1000000, GPR_TIMESPAN)), gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
gpr_time_from_micros(1000000, GPR_TIMESPAN)),
NULL); NULL);
fling_call *s = ev.tag; fling_call *s = ev.tag;
switch (ev.type) { switch (ev.type) {

@ -58,7 +58,7 @@ static void test_alarm(void) {
grpc_completion_queue *cc; grpc_completion_queue *cc;
LOG_TEST("test_alarm"); LOG_TEST("test_alarm");
cc = grpc_completion_queue_create(NULL); cc = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
{ {
/* regular expiry */ /* regular expiry */
grpc_event ev; grpc_event ev;

@ -51,24 +51,62 @@ static void *create_test_tag(void) {
static void shutdown_and_destroy(grpc_completion_queue *cc) { static void shutdown_and_destroy(grpc_completion_queue *cc) {
grpc_event ev; grpc_event ev;
grpc_completion_queue_shutdown(cc); grpc_completion_queue_shutdown(cc);
ev = grpc_completion_queue_next(cc, gpr_inf_past(GPR_CLOCK_REALTIME), NULL);
switch (grpc_get_cq_completion_type(cc)) {
case GRPC_CQ_NEXT: {
ev = grpc_completion_queue_next(cc, gpr_inf_past(GPR_CLOCK_REALTIME),
NULL);
break;
}
case GRPC_CQ_PLUCK: {
ev = grpc_completion_queue_pluck(cc, create_test_tag(),
gpr_inf_past(GPR_CLOCK_REALTIME), NULL);
break;
}
default: {
gpr_log(GPR_ERROR, "Unknown completion type");
break;
}
}
GPR_ASSERT(ev.type == GRPC_QUEUE_SHUTDOWN); GPR_ASSERT(ev.type == GRPC_QUEUE_SHUTDOWN);
grpc_completion_queue_destroy(cc); grpc_completion_queue_destroy(cc);
} }
/* ensure we can create and destroy a completion channel */ /* ensure we can create and destroy a completion channel */
static void test_no_op(void) { static void test_no_op(void) {
grpc_cq_completion_type completion_types[] = {GRPC_CQ_NEXT, GRPC_CQ_PLUCK};
grpc_cq_polling_type polling_types[] = {DEFAULT_POLLING, NON_LISTENING,
NON_POLLING};
LOG_TEST("test_no_op"); LOG_TEST("test_no_op");
shutdown_and_destroy(grpc_completion_queue_create(NULL));
for (size_t i = 0; i < GPR_ARRAY_SIZE(completion_types); i++) {
for (size_t j = 0; j < GPR_ARRAY_SIZE(polling_types); j++) {
shutdown_and_destroy(grpc_completion_queue_create(
completion_types[i], polling_types[j], NULL));
}
}
} }
static void test_pollset_conversion(void) { static void test_pollset_conversion(void) {
grpc_completion_queue *cq = grpc_completion_queue_create(NULL); grpc_cq_completion_type completion_types[] = {GRPC_CQ_NEXT, GRPC_CQ_PLUCK};
GPR_ASSERT(grpc_cq_from_pollset(grpc_cq_pollset(cq)) == cq); grpc_cq_polling_type polling_types[] = {DEFAULT_POLLING, NON_LISTENING};
shutdown_and_destroy(cq); grpc_completion_queue *cq;
LOG_TEST("test_pollset_conversion");
for (size_t i = 0; i < GPR_ARRAY_SIZE(completion_types); i++) {
for (size_t j = 0; j < GPR_ARRAY_SIZE(polling_types); j++) {
cq = grpc_completion_queue_create(completion_types[i], polling_types[j],
NULL);
GPR_ASSERT(grpc_cq_from_pollset(grpc_cq_pollset(cq)) == cq);
shutdown_and_destroy(cq);
}
}
} }
static void test_wait_empty(void) { static void test_wait_empty(void) {
grpc_cq_polling_type polling_types[] = {DEFAULT_POLLING, NON_LISTENING};
grpc_completion_queue *cc; grpc_completion_queue *cc;
grpc_event event; grpc_event event;
@ -87,50 +125,66 @@ static void test_cq_end_op(void) {
grpc_event ev; grpc_event ev;
grpc_completion_queue *cc; grpc_completion_queue *cc;
grpc_cq_completion completion; grpc_cq_completion completion;
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; grpc_cq_polling_type polling_types[] = {DEFAULT_POLLING, NON_LISTENING,
NON_POLLING};
grpc_exec_ctx init_exec_ctx = GRPC_EXEC_CTX_INIT;
grpc_exec_ctx exec_ctx;
void *tag = create_test_tag(); void *tag = create_test_tag();
LOG_TEST("test_cq_end_op"); LOG_TEST("test_cq_end_op");
cc = grpc_completion_queue_create(NULL); for (size_t i = 0; i < GPR_ARRAY_SIZE(polling_types); i++) {
exec_ctx = init_exec_ctx; // Reset exec_ctx
cc = grpc_completion_queue_create(GRPC_CQ_NEXT, polling_types[i], NULL);
grpc_cq_begin_op(cc, tag); grpc_cq_begin_op(cc, tag);
grpc_cq_end_op(&exec_ctx, cc, tag, GRPC_ERROR_NONE, do_nothing_end_completion, grpc_cq_end_op(&exec_ctx, cc, tag, GRPC_ERROR_NONE,
NULL, &completion); do_nothing_end_completion, NULL, &completion);
ev = grpc_completion_queue_next(cc, gpr_inf_past(GPR_CLOCK_REALTIME), NULL); ev = grpc_completion_queue_next(cc, gpr_inf_past(GPR_CLOCK_REALTIME), NULL);
GPR_ASSERT(ev.type == GRPC_OP_COMPLETE); GPR_ASSERT(ev.type == GRPC_OP_COMPLETE);
GPR_ASSERT(ev.tag == tag); GPR_ASSERT(ev.tag == tag);
GPR_ASSERT(ev.success); GPR_ASSERT(ev.success);
shutdown_and_destroy(cc); shutdown_and_destroy(cc);
grpc_exec_ctx_finish(&exec_ctx); grpc_exec_ctx_finish(&exec_ctx);
}
} }
static void test_shutdown_then_next_polling(void) { static void test_shutdown_then_next_polling(void) {
grpc_cq_polling_type polling_types[] = {DEFAULT_POLLING, NON_LISTENING,
NON_POLLING};
grpc_completion_queue *cc; grpc_completion_queue *cc;
grpc_event event; grpc_event event;
LOG_TEST("test_shutdown_then_next_polling"); LOG_TEST("test_shutdown_then_next_polling");
cc = grpc_completion_queue_create(NULL); for (size_t i = 0; i < GPR_ARRAY_SIZE(polling_types); i++) {
grpc_completion_queue_shutdown(cc); cc = grpc_completion_queue_create(GRPC_CQ_NEXT, polling_types[i], NULL);
event = grpc_completion_queue_shutdown(cc);
grpc_completion_queue_next(cc, gpr_inf_past(GPR_CLOCK_REALTIME), NULL); event =
GPR_ASSERT(event.type == GRPC_QUEUE_SHUTDOWN); grpc_completion_queue_next(cc, gpr_inf_past(GPR_CLOCK_REALTIME), NULL);
grpc_completion_queue_destroy(cc); GPR_ASSERT(event.type == GRPC_QUEUE_SHUTDOWN);
grpc_completion_queue_destroy(cc);
}
} }
static void test_shutdown_then_next_with_timeout(void) { static void test_shutdown_then_next_with_timeout(void) {
grpc_cq_polling_type polling_types[] = {DEFAULT_POLLING, NON_LISTENING,
NON_POLLING};
grpc_completion_queue *cc; grpc_completion_queue *cc;
grpc_event event; grpc_event event;
LOG_TEST("test_shutdown_then_next_with_timeout"); LOG_TEST("test_shutdown_then_next_with_timeout");
cc = grpc_completion_queue_create(NULL); for (size_t i = 0; i < GPR_ARRAY_SIZE(polling_types); i++) {
grpc_completion_queue_shutdown(cc); cc = grpc_completion_queue_create(GRPC_CQ_NEXT, polling_types[i], NULL);
event =
grpc_completion_queue_next(cc, gpr_inf_future(GPR_CLOCK_REALTIME), NULL); grpc_completion_queue_shutdown(cc);
GPR_ASSERT(event.type == GRPC_QUEUE_SHUTDOWN); event = grpc_completion_queue_next(cc, gpr_inf_future(GPR_CLOCK_REALTIME),
grpc_completion_queue_destroy(cc); NULL);
GPR_ASSERT(event.type == GRPC_QUEUE_SHUTDOWN);
grpc_completion_queue_destroy(cc);
}
} }
static void test_pluck(void) { static void test_pluck(void) {
@ -138,7 +192,10 @@ static void test_pluck(void) {
grpc_completion_queue *cc; grpc_completion_queue *cc;
void *tags[128]; void *tags[128];
grpc_cq_completion completions[GPR_ARRAY_SIZE(tags)]; grpc_cq_completion completions[GPR_ARRAY_SIZE(tags)];
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; grpc_cq_polling_type polling_types[] = {DEFAULT_POLLING, NON_LISTENING,
NON_POLLING};
grpc_exec_ctx init_exec_ctx = GRPC_EXEC_CTX_INIT;
grpc_exec_ctx exec_ctx;
unsigned i, j; unsigned i, j;
LOG_TEST("test_pluck"); LOG_TEST("test_pluck");
@ -150,47 +207,66 @@ static void test_pluck(void) {
} }
} }
cc = grpc_completion_queue_create(NULL); for (size_t pidx = 0; pidx < GPR_ARRAY_SIZE(polling_types); pidx++) {
exec_ctx = init_exec_ctx; // reset exec_ctx
cc = grpc_completion_queue_create(GRPC_CQ_PLUCK, polling_types[pidx], NULL);
for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) { for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) {
grpc_cq_begin_op(cc, tags[i]); grpc_cq_begin_op(cc, tags[i]);
grpc_cq_end_op(&exec_ctx, cc, tags[i], GRPC_ERROR_NONE, grpc_cq_end_op(&exec_ctx, cc, tags[i], GRPC_ERROR_NONE,
do_nothing_end_completion, NULL, &completions[i]); do_nothing_end_completion, NULL, &completions[i]);
} }
for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) { for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) {
ev = grpc_completion_queue_pluck(cc, tags[i], ev = grpc_completion_queue_pluck(cc, tags[i],
gpr_inf_past(GPR_CLOCK_REALTIME), NULL); gpr_inf_past(GPR_CLOCK_REALTIME), NULL);
GPR_ASSERT(ev.tag == tags[i]); GPR_ASSERT(ev.tag == tags[i]);
} }
for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) { for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) {
grpc_cq_begin_op(cc, tags[i]); grpc_cq_begin_op(cc, tags[i]);
grpc_cq_end_op(&exec_ctx, cc, tags[i], GRPC_ERROR_NONE, grpc_cq_end_op(&exec_ctx, cc, tags[i], GRPC_ERROR_NONE,
do_nothing_end_completion, NULL, &completions[i]); do_nothing_end_completion, NULL, &completions[i]);
} }
for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) { for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) {
ev = grpc_completion_queue_pluck(cc, tags[GPR_ARRAY_SIZE(tags) - i - 1], ev = grpc_completion_queue_pluck(cc, tags[GPR_ARRAY_SIZE(tags) - i - 1],
gpr_inf_past(GPR_CLOCK_REALTIME), NULL); gpr_inf_past(GPR_CLOCK_REALTIME), NULL);
GPR_ASSERT(ev.tag == tags[GPR_ARRAY_SIZE(tags) - i - 1]); GPR_ASSERT(ev.tag == tags[GPR_ARRAY_SIZE(tags) - i - 1]);
} }
shutdown_and_destroy(cc); shutdown_and_destroy(cc);
grpc_exec_ctx_finish(&exec_ctx); grpc_exec_ctx_finish(&exec_ctx);
}
} }
static void test_pluck_after_shutdown(void) { static void test_pluck_after_shutdown(void) {
grpc_cq_polling_type polling_types[] = {DEFAULT_POLLING, NON_LISTENING,
NON_POLLING};
grpc_event ev; grpc_event ev;
grpc_completion_queue *cc; grpc_completion_queue *cc;
LOG_TEST("test_pluck_after_shutdown"); LOG_TEST("test_pluck_after_shutdown");
cc = grpc_completion_queue_create(NULL);
grpc_completion_queue_shutdown(cc); for (size_t i = 0; i < GPR_ARRAY_SIZE(polling_types); i++) {
ev = grpc_completion_queue_pluck(cc, NULL, gpr_inf_future(GPR_CLOCK_REALTIME), cc = grpc_completion_queue_create(GRPC_CQ_PLUCK, polling_types[i], NULL);
NULL); grpc_completion_queue_shutdown(cc);
GPR_ASSERT(ev.type == GRPC_QUEUE_SHUTDOWN); ev = grpc_completion_queue_pluck(cc, NULL,
grpc_completion_queue_destroy(cc); gpr_inf_future(GPR_CLOCK_REALTIME), NULL);
GPR_ASSERT(ev.type == GRPC_QUEUE_SHUTDOWN);
grpc_completion_queue_destroy(cc);
}
}
struct thread_state {
grpc_completion_queue *cc;
void *tag;
};
static void pluck_one(void *arg) {
struct thread_state *state = arg;
grpc_completion_queue_pluck(state->cc, state->tag,
gpr_inf_future(GPR_CLOCK_REALTIME), NULL);
} }
int main(int argc, char **argv) { int main(int argc, char **argv) {

@ -66,7 +66,8 @@ static int detag(void *p) { return (int)(uintptr_t)p; }
void create_loop_destroy(void *addr) { void create_loop_destroy(void *addr) {
for (int i = 0; i < NUM_OUTER_LOOPS; ++i) { for (int i = 0; i < NUM_OUTER_LOOPS; ++i) {
grpc_completion_queue *cq = grpc_completion_queue_create(NULL); grpc_completion_queue *cq =
grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
grpc_channel *chan = grpc_insecure_channel_create((char *)addr, NULL, NULL); grpc_channel *chan = grpc_insecure_channel_create((char *)addr, NULL, NULL);
for (int j = 0; j < NUM_INNER_LOOPS; ++j) { for (int j = 0; j < NUM_INNER_LOOPS; ++j) {
@ -195,7 +196,7 @@ int main(int argc, char **argv) {
gpr_asprintf(&args.addr, "localhost:%d", port); gpr_asprintf(&args.addr, "localhost:%d", port);
args.server = grpc_server_create(NULL, NULL); args.server = grpc_server_create(NULL, NULL);
grpc_server_add_insecure_http2_port(args.server, args.addr); grpc_server_add_insecure_http2_port(args.server, args.addr);
args.cq = grpc_completion_queue_create(NULL); args.cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
grpc_server_register_completion_queue(args.server, args.cq, NULL); grpc_server_register_completion_queue(args.server, args.cq, NULL);
grpc_server_start(args.server); grpc_server_start(args.server);
gpr_thd_new(&server, server_thread, &args, &options); gpr_thd_new(&server, server_thread, &args, &options);

@ -108,7 +108,7 @@ int main(int argc, char **argv) {
GPR_ASSERT(GRPC_CHANNEL_SHUTDOWN == GPR_ASSERT(GRPC_CHANNEL_SHUTDOWN ==
grpc_channel_check_connectivity_state(chan, 0)); grpc_channel_check_connectivity_state(chan, 0));
cq = grpc_completion_queue_create(NULL); cq = grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
grpc_slice host = grpc_slice_from_static_string("anywhere"); grpc_slice host = grpc_slice_from_static_string("anywhere");
call = grpc_channel_create_call(chan, NULL, GRPC_PROPAGATE_DEFAULTS, cq, call = grpc_channel_create_call(chan, NULL, GRPC_PROPAGATE_DEFAULTS, cq,

@ -76,7 +76,8 @@ static void run_test(const test_fixture *fixture) {
grpc_server *server = grpc_server_create(NULL, NULL); grpc_server *server = grpc_server_create(NULL, NULL);
fixture->add_server_port(server, addr); fixture->add_server_port(server, addr);
grpc_completion_queue *server_cq = grpc_completion_queue_create(NULL); grpc_completion_queue *server_cq =
grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
grpc_server_register_completion_queue(server, server_cq, NULL); grpc_server_register_completion_queue(server, server_cq, NULL);
grpc_server_start(server); grpc_server_start(server);
@ -86,7 +87,8 @@ static void run_test(const test_fixture *fixture) {
gpr_thd_options_set_joinable(&thdopt); gpr_thd_options_set_joinable(&thdopt);
gpr_thd_new(&server_thread, server_thread_func, &sta, &thdopt); gpr_thd_new(&server_thread, server_thread_func, &sta, &thdopt);
grpc_completion_queue *cq = grpc_completion_queue_create(NULL); grpc_completion_queue *cq =
grpc_completion_queue_create(GRPC_CQ_NEXT, DEFAULT_POLLING, NULL);
grpc_channel *channels[NUM_CONNECTIONS]; grpc_channel *channels[NUM_CONNECTIONS];
for (size_t i = 0; i < NUM_CONNECTIONS; i++) { for (size_t i = 0; i < NUM_CONNECTIONS; i++) {
channels[i] = fixture->create_channel(addr); channels[i] = fixture->create_channel(addr);

@ -60,7 +60,8 @@ void test_add_same_port_twice() {
int port = grpc_pick_unused_port_or_die(); int port = grpc_pick_unused_port_or_die();
char *addr = NULL; char *addr = NULL;
grpc_completion_queue *cq = grpc_completion_queue_create(NULL); grpc_completion_queue *cq =
grpc_completion_queue_create(GRPC_CQ_PLUCK, NON_POLLING, NULL);
grpc_server *server = grpc_server_create(&args, NULL); grpc_server *server = grpc_server_create(&args, NULL);
grpc_server_credentials *fake_creds = grpc_server_credentials *fake_creds =
grpc_fake_transport_security_server_credentials_create(); grpc_fake_transport_security_server_credentials_create();

Some files were not shown because too many files have changed in this diff Show More

Loading…
Cancel
Save