diff options
28 files changed, 387 insertions, 148 deletions
diff --git a/include/grpc/grpc.h b/include/grpc/grpc.h index 323a944f93..f479fb8a40 100644 --- a/include/grpc/grpc.h +++ b/include/grpc/grpc.h @@ -158,6 +158,7 @@ typedef struct grpc_byte_buffer grpc_byte_buffer; /* Sample helpers to obtain byte buffers (these will certainly move place */ grpc_byte_buffer *grpc_byte_buffer_create(gpr_slice *slices, size_t nslices); +grpc_byte_buffer *grpc_byte_buffer_copy(grpc_byte_buffer *bb); size_t grpc_byte_buffer_length(grpc_byte_buffer *bb); void grpc_byte_buffer_destroy(grpc_byte_buffer *byte_buffer); @@ -319,6 +320,10 @@ grpc_call_error grpc_call_add_metadata(grpc_call *call, grpc_metadata *metadata, Produces a GRPC_FINISHED event with finished_tag when the call has been completed (there may be other events for the call pending at this time) */ +grpc_call_error grpc_call_invoke(grpc_call *call, + grpc_completion_queue *cq, + void *metadata_read_tag, + void *finished_tag, gpr_uint32 flags); grpc_call_error grpc_call_start_invoke(grpc_call *call, grpc_completion_queue *cq, void *invoke_accepted_tag, diff --git a/src/core/surface/byte_buffer.c b/src/core/surface/byte_buffer.c index 27a6c6e33d..4087622894 100644 --- a/src/core/surface/byte_buffer.c +++ b/src/core/surface/byte_buffer.c @@ -49,6 +49,16 @@ grpc_byte_buffer *grpc_byte_buffer_create(gpr_slice *slices, size_t nslices) { return bb; } +grpc_byte_buffer *grpc_byte_buffer_copy(grpc_byte_buffer *bb) { + switch (bb->type) { + case GRPC_BB_SLICE_BUFFER: + return grpc_byte_buffer_create(bb->data.slice_buffer.slices, bb->data.slice_buffer.count); + } + gpr_log(GPR_INFO, "should never get here"); + abort(); + return NULL; +} + void grpc_byte_buffer_destroy(grpc_byte_buffer *bb) { switch (bb->type) { case GRPC_BB_SLICE_BUFFER: diff --git a/src/core/surface/call.c b/src/core/surface/call.c index 9c5f5064eb..f6d93bd957 100644 --- a/src/core/surface/call.c +++ b/src/core/surface/call.c @@ -173,11 +173,14 @@ struct grpc_call { /* protects variables in this section */ gpr_mu read_mu; + gpr_uint8 received_start; + gpr_uint8 start_ok; gpr_uint8 reads_done; gpr_uint8 received_finish; gpr_uint8 received_metadata; gpr_uint8 have_read; gpr_uint8 have_alarm; + gpr_uint8 pending_writes_done; /* The current outstanding read message tag (only valid if have_read == 1) */ void *read_tag; void *metadata_tag; @@ -189,6 +192,8 @@ struct grpc_call { /* The current outstanding send message/context/invoke/end tag (only valid if have_write == 1) */ void *write_tag; + grpc_byte_buffer *pending_write; + gpr_uint32 pending_write_flags; /* The final status of the call */ grpc_status_code status_code; @@ -230,6 +235,9 @@ grpc_call *grpc_call_create(grpc_channel *channel, call->status_details = NULL; call->received_finish = 0; call->reads_done = 0; + call->received_start = 0; + call->pending_write = NULL; + call->pending_writes_done = 0; grpc_metadata_buffer_init(&call->incoming_metadata); gpr_ref_init(&call->internal_refcount, 1); grpc_call_stack_init(channel_stack, server_transport_data, @@ -330,16 +338,6 @@ grpc_call_error grpc_call_add_metadata(grpc_call *call, grpc_metadata *metadata, return GRPC_CALL_OK; } -static void done_invoke(void *user_data, grpc_op_error error) { - grpc_call *call = user_data; - void *tag = call->write_tag; - - GPR_ASSERT(call->have_write); - call->have_write = 0; - call->write_tag = INVALID_TAG; - grpc_cq_end_invoke_accepted(call->cq, tag, call, NULL, NULL, error); -} - static void finish_call(grpc_call *call) { size_t count; grpc_metadata *elements; @@ -359,6 +357,88 @@ grpc_call_error grpc_call_start_invoke(grpc_call *call, void *invoke_accepted_tag, void *metadata_read_tag, void *finished_tag, gpr_uint32 flags) { + grpc_call_error err = grpc_call_invoke(call, cq, metadata_read_tag, finished_tag, flags); + if (err == GRPC_CALL_OK) { + grpc_cq_begin_op(call->cq, call, GRPC_INVOKE_ACCEPTED); + grpc_cq_end_invoke_accepted(call->cq, invoke_accepted_tag, call, do_nothing, NULL, GRPC_OP_OK); + } + return err; +} + +static void done_write(void *user_data, grpc_op_error error) { + grpc_call *call = user_data; + void *tag = call->write_tag; + + GPR_ASSERT(call->have_write); + call->have_write = 0; + call->write_tag = INVALID_TAG; + grpc_cq_end_write_accepted(call->cq, tag, call, NULL, NULL, error); +} + +static void done_writes_done(void *user_data, grpc_op_error error) { + grpc_call *call = user_data; + void *tag = call->write_tag; + + GPR_ASSERT(call->have_write); + call->have_write = 0; + call->write_tag = INVALID_TAG; + grpc_cq_end_finish_accepted(call->cq, tag, call, NULL, NULL, error); +} + +static void call_started(void *user_data, grpc_op_error error) { + grpc_call *call = user_data; + grpc_call_element *elem; + grpc_byte_buffer *pending_write = NULL; + gpr_uint32 pending_write_flags = 0; + gpr_uint8 pending_writes_done = 0; + int ok; + grpc_call_op op; + + gpr_mu_lock(&call->read_mu); + GPR_ASSERT(!call->received_start); + call->received_start = 1; + ok = call->start_ok = (error == GRPC_OP_OK); + pending_write = call->pending_write; + pending_write_flags = call->pending_write_flags; + pending_writes_done = call->pending_writes_done; + gpr_mu_unlock(&call->read_mu); + + if (pending_write) { + if (ok) { + op.type = GRPC_SEND_MESSAGE; + op.dir = GRPC_CALL_DOWN; + op.flags = pending_write_flags; + op.done_cb = done_write; + op.user_data = call; + op.data.message = pending_write; + + elem = CALL_ELEM_FROM_CALL(call, 0); + elem->filter->call_op(elem, NULL, &op); + } else { + done_write(call, error); + } + grpc_byte_buffer_destroy(pending_write); + } + if (pending_writes_done) { + if (ok) { + op.type = GRPC_SEND_FINISH; + op.dir = GRPC_CALL_DOWN; + op.flags = 0; + op.done_cb = done_writes_done; + op.user_data = call; + + elem = CALL_ELEM_FROM_CALL(call, 0); + elem->filter->call_op(elem, NULL, &op); + } else { + done_writes_done(call, error); + } + } +} + +grpc_call_error grpc_call_invoke(grpc_call *call, + grpc_completion_queue *cq, + void *metadata_read_tag, + void *finished_tag, gpr_uint32 flags) { grpc_call_element *elem; grpc_call_op op; @@ -390,7 +470,6 @@ grpc_call_error grpc_call_start_invoke(grpc_call *call, /* inform the completion queue of an incoming operation */ grpc_cq_begin_op(cq, call, GRPC_FINISHED); grpc_cq_begin_op(cq, call, GRPC_CLIENT_METADATA_READ); - grpc_cq_begin_op(cq, call, GRPC_INVOKE_ACCEPTED); gpr_mu_lock(&call->read_mu); @@ -401,8 +480,6 @@ grpc_call_error grpc_call_start_invoke(grpc_call *call, if (call->received_finish) { /* handle early cancellation */ - grpc_cq_end_invoke_accepted(call->cq, invoke_accepted_tag, call, NULL, NULL, - GRPC_OP_ERROR); grpc_cq_end_client_metadata_read(call->cq, metadata_read_tag, call, NULL, NULL, 0, NULL); finish_call(call); @@ -412,18 +489,15 @@ grpc_call_error grpc_call_start_invoke(grpc_call *call, return GRPC_CALL_OK; } - call->write_tag = invoke_accepted_tag; call->metadata_tag = metadata_read_tag; - call->have_write = 1; - gpr_mu_unlock(&call->read_mu); /* call down the filter stack */ op.type = GRPC_SEND_START; op.dir = GRPC_CALL_DOWN; op.flags = flags; - op.done_cb = done_invoke; + op.done_cb = call_started; op.data.start.pollset = grpc_cq_pollset(cq); op.user_data = call; @@ -516,26 +590,6 @@ grpc_call_error grpc_call_accept(grpc_call *call, grpc_completion_queue *cq, return GRPC_CALL_OK; } -static void done_writes_done(void *user_data, grpc_op_error error) { - grpc_call *call = user_data; - void *tag = call->write_tag; - - GPR_ASSERT(call->have_write); - call->have_write = 0; - call->write_tag = INVALID_TAG; - grpc_cq_end_finish_accepted(call->cq, tag, call, NULL, NULL, error); -} - -static void done_write(void *user_data, grpc_op_error error) { - grpc_call *call = user_data; - void *tag = call->write_tag; - - GPR_ASSERT(call->have_write); - call->have_write = 0; - call->write_tag = INVALID_TAG; - grpc_cq_end_write_accepted(call->cq, tag, call, NULL, NULL, error); -} - void grpc_call_client_initial_metadata_complete( grpc_call_element *surface_element) { grpc_call *call = grpc_call_from_top_element(surface_element); @@ -635,8 +689,6 @@ grpc_call_error grpc_call_start_write(grpc_call *call, grpc_cq_begin_op(call->cq, call, GRPC_WRITE_ACCEPTED); - /* for now we do no buffering, so a NULL byte_buffer can have no impact - on our behavior -- succeed immediately */ /* TODO(ctiller): if flags & GRPC_WRITE_BUFFER_HINT == 0, this indicates a flush, and that flush should be propogated down from here */ if (byte_buffer == NULL) { @@ -647,6 +699,15 @@ grpc_call_error grpc_call_start_write(grpc_call *call, call->write_tag = tag; call->have_write = 1; + gpr_mu_lock(&call->read_mu); + if (!call->received_start) { + call->pending_write = grpc_byte_buffer_copy(byte_buffer); + call->pending_write_flags = flags; + + gpr_mu_unlock(&call->read_mu); + } else { + gpr_mu_unlock(&call->read_mu); + op.type = GRPC_SEND_MESSAGE; op.dir = GRPC_CALL_DOWN; op.flags = flags; @@ -656,6 +717,7 @@ grpc_call_error grpc_call_start_write(grpc_call *call, elem = CALL_ELEM_FROM_CALL(call, 0); elem->filter->call_op(elem, NULL, &op); + } return GRPC_CALL_OK; } @@ -687,6 +749,14 @@ grpc_call_error grpc_call_writes_done(grpc_call *call, void *tag) { call->write_tag = tag; call->have_write = 1; + gpr_mu_lock(&call->read_mu); + if (!call->received_start) { + call->pending_writes_done = 1; + + gpr_mu_unlock(&call->read_mu); + } else { + gpr_mu_unlock(&call->read_mu); + op.type = GRPC_SEND_FINISH; op.dir = GRPC_CALL_DOWN; op.flags = 0; @@ -695,6 +765,7 @@ grpc_call_error grpc_call_writes_done(grpc_call *call, void *tag) { elem = CALL_ELEM_FROM_CALL(call, 0); elem->filter->call_op(elem, NULL, &op); + } return GRPC_CALL_OK; } diff --git a/test/core/end2end/cq_verifier.c b/test/core/end2end/cq_verifier.c index e5b7304743..ba446c116a 100644 --- a/test/core/end2end/cq_verifier.c +++ b/test/core/end2end/cq_verifier.c @@ -70,7 +70,6 @@ typedef struct expectation { union { grpc_op_error finish_accepted; grpc_op_error write_accepted; - grpc_op_error invoke_accepted; struct { const char *method; const char *host; @@ -182,7 +181,7 @@ static void verify_matches(expectation *e, grpc_event *ev) { GPR_ASSERT(e->data.write_accepted == ev->data.write_accepted); break; case GRPC_INVOKE_ACCEPTED: - GPR_ASSERT(e->data.invoke_accepted == ev->data.invoke_accepted); + abort(); break; case GRPC_SERVER_RPC_NEW: GPR_ASSERT(string_equivalent(e->data.server_rpc_new.method, @@ -268,8 +267,7 @@ static size_t expectation_to_string(char *out, expectation *e) { return sprintf(out, "GRPC_WRITE_ACCEPTED result=%d", e->data.write_accepted); case GRPC_INVOKE_ACCEPTED: - return sprintf(out, "GRPC_INVOKE_ACCEPTED result=%d", - e->data.invoke_accepted); + return sprintf(out, "GRPC_INVOKE_ACCEPTED"); case GRPC_SERVER_RPC_NEW: timeout = gpr_time_sub(e->data.server_rpc_new.deadline, gpr_now()); return sprintf(out, "GRPC_SERVER_RPC_NEW method=%s host=%s timeout=%fsec", @@ -414,11 +412,6 @@ static metadata *metadata_from_args(va_list args) { } } -void cq_expect_invoke_accepted(cq_verifier *v, void *tag, - grpc_op_error result) { - add(v, GRPC_INVOKE_ACCEPTED, tag)->data.invoke_accepted = result; -} - void cq_expect_write_accepted(cq_verifier *v, void *tag, grpc_op_error result) { add(v, GRPC_WRITE_ACCEPTED, tag)->data.write_accepted = result; } diff --git a/test/core/end2end/cq_verifier.h b/test/core/end2end/cq_verifier.h index 9711e0067c..73d5782a33 100644 --- a/test/core/end2end/cq_verifier.h +++ b/test/core/end2end/cq_verifier.h @@ -56,7 +56,6 @@ void cq_verify_empty(cq_verifier *v); Any functions taking ... expect a NULL terminated list of key/value pairs (each pair using two parameter slots) of metadata that MUST be present in the event. */ -void cq_expect_invoke_accepted(cq_verifier *v, void *tag, grpc_op_error result); void cq_expect_write_accepted(cq_verifier *v, void *tag, grpc_op_error result); void cq_expect_finish_accepted(cq_verifier *v, void *tag, grpc_op_error result); void cq_expect_read(cq_verifier *v, void *tag, gpr_slice bytes); diff --git a/test/core/end2end/dualstack_socket_test.c b/test/core/end2end/dualstack_socket_test.c index b443caa2a6..23ffddc146 100644 --- a/test/core/end2end/dualstack_socket_test.c +++ b/test/core/end2end/dualstack_socket_test.c @@ -107,13 +107,10 @@ void test_connect(const char *server_host, const char *client_host, int port, GPR_ASSERT(c); GPR_ASSERT(GRPC_CALL_OK == - grpc_call_start_invoke(c, client_cq, tag(1), tag(2), tag(3), 0)); + grpc_call_invoke(c, client_cq, tag(2), tag(3), 0)); + GPR_ASSERT(GRPC_CALL_OK == grpc_call_writes_done(c, tag(4))); if (expect_ok) { /* Check for a successful request. */ - cq_expect_invoke_accepted(v_client, tag(1), GRPC_OP_OK); - cq_verify(v_client); - - GPR_ASSERT(GRPC_CALL_OK == grpc_call_writes_done(c, tag(4))); cq_expect_finish_accepted(v_client, tag(4), GRPC_OP_OK); cq_verify(v_client); @@ -142,10 +139,10 @@ void test_connect(const char *server_host, const char *client_host, int port, grpc_call_destroy(s); } else { /* Check for a failed connection. */ - cq_expect_invoke_accepted(v_client, tag(1), GRPC_OP_ERROR); cq_expect_client_metadata_read(v_client, tag(2), NULL); cq_expect_finished_with_status(v_client, tag(3), GRPC_STATUS_CANCELLED, NULL, NULL); + cq_expect_finish_accepted(v_client, tag(4), GRPC_OP_ERROR); cq_verify(v_client); grpc_call_destroy(c); diff --git a/test/core/end2end/dualstack_socket_test.c.orig b/test/core/end2end/dualstack_socket_test.c.orig new file mode 100644 index 0000000000..b443caa2a6 --- /dev/null +++ b/test/core/end2end/dualstack_socket_test.c.orig @@ -0,0 +1,213 @@ +/* + * + * Copyright 2014, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include "src/core/iomgr/socket_utils_posix.h" +#include <grpc/grpc.h> +#include <grpc/support/alloc.h> +#include <grpc/support/host_port.h> +#include <grpc/support/log.h> +#include "test/core/end2end/cq_verifier.h" +#include "test/core/util/port.h" +#include "test/core/util/test_config.h" + +/* This test exercises IPv4, IPv6, and dualstack sockets in various ways. */ + +static void *tag(gpr_intptr i) { return (void *)i; } + +static gpr_timespec ms_from_now(int ms) { + return gpr_time_add(gpr_now(), gpr_time_from_micros(GPR_US_PER_MS * ms)); +} + +static void drain_cq(grpc_completion_queue *cq) { + grpc_event *ev; + grpc_completion_type type; + do { + ev = grpc_completion_queue_next(cq, ms_from_now(5000)); + GPR_ASSERT(ev); + type = ev->type; + grpc_event_finish(ev); + gpr_log(GPR_INFO, "Drained event type %d", type); + } while (type != GRPC_QUEUE_SHUTDOWN); +} + +void test_connect(const char *server_host, const char *client_host, int port, + int expect_ok) { + char *client_hostport; + char *server_hostport; + grpc_channel *client; + grpc_server *server; + grpc_completion_queue *client_cq; + grpc_completion_queue *server_cq; + grpc_call *c; + grpc_call *s; + cq_verifier *v_client; + cq_verifier *v_server; + gpr_timespec deadline; + + gpr_join_host_port(&client_hostport, client_host, port); + gpr_join_host_port(&server_hostport, server_host, port); + gpr_log(GPR_INFO, "Testing with server=%s client=%s (expecting %s)", + server_hostport, client_hostport, expect_ok ? "success" : "failure"); + + /* Create server. */ + server_cq = grpc_completion_queue_create(); + server = grpc_server_create(server_cq, NULL); + GPR_ASSERT(grpc_server_add_http2_port(server, server_hostport)); + grpc_server_start(server); + gpr_free(server_hostport); + v_server = cq_verifier_create(server_cq); + + /* Create client. */ + client_cq = grpc_completion_queue_create(); + client = grpc_channel_create(client_hostport, NULL); + gpr_free(client_hostport); + v_client = cq_verifier_create(client_cq); + + if (expect_ok) { + /* Normal deadline, shouldn't be reached. */ + deadline = ms_from_now(60000); + } else { + /* Give up faster when failure is expected. + BUG: Setting this to 1000 reveals a memory leak (b/18608927). */ + deadline = ms_from_now(1500); + } + + /* Send a trivial request. */ + c = grpc_channel_create_call(client, "/foo", "test.google.com", deadline); + GPR_ASSERT(c); + + GPR_ASSERT(GRPC_CALL_OK == + grpc_call_start_invoke(c, client_cq, tag(1), tag(2), tag(3), 0)); + if (expect_ok) { + /* Check for a successful request. */ + cq_expect_invoke_accepted(v_client, tag(1), GRPC_OP_OK); + cq_verify(v_client); + + GPR_ASSERT(GRPC_CALL_OK == grpc_call_writes_done(c, tag(4))); + cq_expect_finish_accepted(v_client, tag(4), GRPC_OP_OK); + cq_verify(v_client); + + GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(server, tag(100))); + cq_expect_server_rpc_new(v_server, &s, tag(100), "/foo", "test.google.com", + deadline, NULL); + cq_verify(v_server); + + GPR_ASSERT(GRPC_CALL_OK == grpc_call_accept(s, server_cq, tag(102), 0)); + cq_expect_client_metadata_read(v_client, tag(2), NULL); + cq_verify(v_client); + + GPR_ASSERT(GRPC_CALL_OK == + grpc_call_start_write_status(s, GRPC_STATUS_UNIMPLEMENTED, "xyz", + tag(5))); + cq_expect_finished_with_status(v_client, tag(3), GRPC_STATUS_UNIMPLEMENTED, + "xyz", NULL); + cq_verify(v_client); + + cq_expect_finish_accepted(v_server, tag(5), GRPC_OP_OK); + cq_verify(v_server); + cq_expect_finished(v_server, tag(102), NULL); + cq_verify(v_server); + + grpc_call_destroy(c); + grpc_call_destroy(s); + } else { + /* Check for a failed connection. */ + cq_expect_invoke_accepted(v_client, tag(1), GRPC_OP_ERROR); + cq_expect_client_metadata_read(v_client, tag(2), NULL); + cq_expect_finished_with_status(v_client, tag(3), GRPC_STATUS_CANCELLED, + NULL, NULL); + cq_verify(v_client); + + grpc_call_destroy(c); + } + + cq_verifier_destroy(v_client); + cq_verifier_destroy(v_server); + + /* Destroy client. */ + grpc_channel_destroy(client); + grpc_completion_queue_shutdown(client_cq); + drain_cq(client_cq); + grpc_completion_queue_destroy(client_cq); + + /* Destroy server. */ + grpc_server_shutdown(server); + grpc_server_destroy(server); + grpc_completion_queue_shutdown(server_cq); + drain_cq(server_cq); + grpc_completion_queue_destroy(server_cq); +} + +int main(int argc, char **argv) { + int do_ipv6 = 1; + int i; + int port = grpc_pick_unused_port_or_die(); + + grpc_test_init(argc, argv); + grpc_init(); + + if (!grpc_ipv6_loopback_available()) { + gpr_log(GPR_INFO, "Can't bind to ::1. Skipping IPv6 tests."); + do_ipv6 = 0; + } + + for (i = 0; i <= 1; i++) { + /* For coverage, test with and without dualstack sockets. */ + grpc_forbid_dualstack_sockets_for_testing = i; + + /* :: and 0.0.0.0 are handled identically. */ + test_connect("::", "127.0.0.1", port, 1); + test_connect("::", "::ffff:127.0.0.1", port, 1); + test_connect("::", "localhost", port, 1); + test_connect("0.0.0.0", "127.0.0.1", port, 1); + test_connect("0.0.0.0", "::ffff:127.0.0.1", port, 1); + test_connect("0.0.0.0", "localhost", port, 1); + if (do_ipv6) { + test_connect("::", "::1", port, 1); + test_connect("0.0.0.0", "::1", port, 1); + } + + /* These only work when the families agree. */ + test_connect("127.0.0.1", "127.0.0.1", port, 1); + if (do_ipv6) { + test_connect("::1", "::1", port, 1); + test_connect("::1", "127.0.0.1", port, 0); + test_connect("127.0.0.1", "::1", port, 0); + } + + } + + grpc_shutdown(); + + return 0; +} diff --git a/test/core/end2end/no_server_test.c b/test/core/end2end/no_server_test.c index ba6349c109..0c251ab467 100644 --- a/test/core/end2end/no_server_test.c +++ b/test/core/end2end/no_server_test.c @@ -57,10 +57,9 @@ int main(int argc, char **argv) { /* create a call, channel to a non existant server */ chan = grpc_channel_create("nonexistant:54321", NULL); call = grpc_channel_create_call(chan, "/foo", "nonexistant", deadline); - GPR_ASSERT(grpc_call_start_invoke(call, cq, tag(1), tag(2), tag(3), 0) == + GPR_ASSERT(grpc_call_invoke(call, cq, tag(2), tag(3), 0) == GRPC_CALL_OK); /* verify that all tags get completed */ - cq_expect_invoke_accepted(cqv, tag(1), GRPC_OP_ERROR); cq_expect_client_metadata_read(cqv, tag(2), NULL); cq_expect_finished_with_status(cqv, tag(3), GRPC_STATUS_CANCELLED, NULL, NULL); diff --git a/test/core/end2end/tests/cancel_after_accept.c b/test/core/end2end/tests/cancel_after_accept.c index 3b570401ef..5ec13b937c 100644 --- a/test/core/end2end/tests/cancel_after_accept.c +++ b/test/core/end2end/tests/cancel_after_accept.c @@ -124,9 +124,7 @@ static void test_cancel_after_accept(grpc_end2end_test_config config, GPR_ASSERT(c); GPR_ASSERT(GRPC_CALL_OK == - grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); - cq_expect_invoke_accepted(v_client, tag(1), GRPC_OP_OK); - cq_verify(v_client); + grpc_call_invoke(c, f.client_cq, tag(2), tag(3), 0)); GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, tag(100))); cq_expect_server_rpc_new(v_server, &s, tag(100), "/foo", "test.google.com", diff --git a/test/core/end2end/tests/cancel_after_accept_and_writes_closed.c b/test/core/end2end/tests/cancel_after_accept_and_writes_closed.c index e20b0ec0b2..19d240471e 100644 --- a/test/core/end2end/tests/cancel_after_accept_and_writes_closed.c +++ b/test/core/end2end/tests/cancel_after_accept_and_writes_closed.c @@ -124,9 +124,7 @@ static void test_cancel_after_accept_and_writes_closed( GPR_ASSERT(c); GPR_ASSERT(GRPC_CALL_OK == - grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); - cq_expect_invoke_accepted(v_client, tag(1), GRPC_OP_OK); - cq_verify(v_client); + grpc_call_invoke(c, f.client_cq, tag(2), tag(3), 0)); GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, tag(100))); cq_expect_server_rpc_new(v_server, &s, tag(100), "/foo", "test.google.com", diff --git a/test/core/end2end/tests/cancel_after_invoke.c b/test/core/end2end/tests/cancel_after_invoke.c index 4fab9f7eff..ad0dd0c9ec 100644 --- a/test/core/end2end/tests/cancel_after_invoke.c +++ b/test/core/end2end/tests/cancel_after_invoke.c @@ -122,9 +122,7 @@ static void test_cancel_after_invoke(grpc_end2end_test_config config, GPR_ASSERT(c); GPR_ASSERT(GRPC_CALL_OK == - grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); - cq_expect_invoke_accepted(v_client, tag(1), GRPC_OP_OK); - cq_verify(v_client); + grpc_call_invoke(c, f.client_cq, tag(2), tag(3), 0)); GPR_ASSERT(GRPC_CALL_OK == call_cancel(c)); diff --git a/test/core/end2end/tests/cancel_before_invoke.c b/test/core/end2end/tests/cancel_before_invoke.c index 2d8d465b24..e523fa2f76 100644 --- a/test/core/end2end/tests/cancel_before_invoke.c +++ b/test/core/end2end/tests/cancel_before_invoke.c @@ -119,8 +119,7 @@ static void test_cancel_before_invoke(grpc_end2end_test_config config) { GPR_ASSERT(GRPC_CALL_OK == grpc_call_cancel(c)); GPR_ASSERT(GRPC_CALL_OK == - grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); - cq_expect_invoke_accepted(v_client, tag(1), GRPC_OP_ERROR); + grpc_call_invoke(c, f.client_cq, tag(2), tag(3), 0)); cq_expect_client_metadata_read(v_client, tag(2), NULL); cq_expect_finished_with_status(v_client, tag(3), GRPC_STATUS_CANCELLED, NULL, NULL); diff --git a/test/core/end2end/tests/disappearing_server.c b/test/core/end2end/tests/disappearing_server.c index c421798fc7..1cbb15f32e 100644 --- a/test/core/end2end/tests/disappearing_server.c +++ b/test/core/end2end/tests/disappearing_server.c @@ -100,11 +100,8 @@ static void do_request_and_shutdown_server(grpc_end2end_test_fixture *f, c = grpc_channel_create_call(f->client, "/foo", "test.google.com", deadline); GPR_ASSERT(c); - GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_invoke(c, f->client_cq, tag(1), + GPR_ASSERT(GRPC_CALL_OK == grpc_call_invoke(c, f->client_cq, tag(2), tag(3), 0)); - cq_expect_invoke_accepted(v_client, tag(1), GRPC_OP_OK); - - cq_verify(v_client); GPR_ASSERT(GRPC_CALL_OK == grpc_call_writes_done(c, tag(4))); cq_expect_finish_accepted(v_client, tag(4), GRPC_OP_OK); diff --git a/test/core/end2end/tests/early_server_shutdown_finishes_inflight_calls.c b/test/core/end2end/tests/early_server_shutdown_finishes_inflight_calls.c index 3855a7a3de..824d9eba88 100644 --- a/test/core/end2end/tests/early_server_shutdown_finishes_inflight_calls.c +++ b/test/core/end2end/tests/early_server_shutdown_finishes_inflight_calls.c @@ -115,9 +115,7 @@ static void test_early_server_shutdown_finishes_inflight_calls( GPR_ASSERT(c); GPR_ASSERT(GRPC_CALL_OK == - grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); - cq_expect_invoke_accepted(v_client, tag(1), GRPC_OP_OK); - cq_verify(v_client); + grpc_call_invoke(c, f.client_cq, tag(2), tag(3), 0)); GPR_ASSERT(GRPC_CALL_OK == grpc_call_writes_done(c, tag(4))); cq_expect_finish_accepted(v_client, tag(4), GRPC_OP_OK); diff --git a/test/core/end2end/tests/invoke_large_request.c b/test/core/end2end/tests/invoke_large_request.c index bad86fb9dc..bacecde01b 100644 --- a/test/core/end2end/tests/invoke_large_request.c +++ b/test/core/end2end/tests/invoke_large_request.c @@ -126,9 +126,7 @@ static void test_invoke_large_request(grpc_end2end_test_config config) { GPR_ASSERT(c); GPR_ASSERT(GRPC_CALL_OK == - grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); - cq_expect_invoke_accepted(v_client, tag(1), GRPC_OP_OK); - cq_verify(v_client); + grpc_call_invoke(c, f.client_cq, tag(2), tag(3), 0)); GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_write(c, request_payload, tag(4), 0)); diff --git a/test/core/end2end/tests/max_concurrent_streams.c b/test/core/end2end/tests/max_concurrent_streams.c index a418d1b15f..e2f30d0778 100644 --- a/test/core/end2end/tests/max_concurrent_streams.c +++ b/test/core/end2end/tests/max_concurrent_streams.c @@ -113,9 +113,7 @@ static void simple_request_body(grpc_end2end_test_fixture f) { GPR_ASSERT(c); GPR_ASSERT(GRPC_CALL_OK == - grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); - cq_expect_invoke_accepted(v_client, tag(1), GRPC_OP_OK); - cq_verify(v_client); + grpc_call_invoke(c, f.client_cq, tag(2), tag(3), 0)); GPR_ASSERT(GRPC_CALL_OK == grpc_call_writes_done(c, tag(4))); cq_expect_finish_accepted(v_client, tag(4), GRPC_OP_OK); @@ -191,14 +189,17 @@ static void test_max_concurrent_streams(grpc_end2end_test_config config) { GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, tag(100))); - GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_invoke(c1, f.client_cq, tag(300), + GPR_ASSERT(GRPC_CALL_OK == grpc_call_invoke(c1, f.client_cq, tag(301), tag(302), 0)); - GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_invoke(c2, f.client_cq, tag(400), + GPR_ASSERT(GRPC_CALL_OK == grpc_call_invoke(c2, f.client_cq, tag(401), tag(402), 0)); + GPR_ASSERT(GRPC_CALL_OK == grpc_call_writes_done(c1, tag(303))); + GPR_ASSERT(GRPC_CALL_OK == grpc_call_writes_done(c2, tag(303))); + ev = grpc_completion_queue_next( f.client_cq, gpr_time_add(gpr_now(), gpr_time_from_seconds(10))); GPR_ASSERT(ev); - GPR_ASSERT(ev->type == GRPC_INVOKE_ACCEPTED); + GPR_ASSERT(ev->type == GRPC_FINISH_ACCEPTED); GPR_ASSERT(ev->data.invoke_accepted == GRPC_OP_OK); /* The /alpha or /beta calls started above could be invoked (but NOT both); * check this here */ @@ -206,11 +207,6 @@ static void test_max_concurrent_streams(grpc_end2end_test_config config) { live_call_obj = live_call == 300 ? c1 : c2; grpc_event_finish(ev); - GPR_ASSERT(GRPC_CALL_OK == - grpc_call_writes_done(live_call_obj, tag(live_call + 3))); - cq_expect_finish_accepted(v_client, tag(live_call + 3), GRPC_OP_OK); - cq_verify(v_client); - cq_expect_server_rpc_new(v_server, &s1, tag(100), live_call == 300 ? "/alpha" : "/beta", "test.google.com", deadline, NULL); @@ -232,11 +228,6 @@ static void test_max_concurrent_streams(grpc_end2end_test_config config) { GRPC_STATUS_UNIMPLEMENTED, "xyz", NULL); live_call = (live_call == 300) ? 400 : 300; live_call_obj = live_call == 300 ? c1 : c2; - cq_expect_invoke_accepted(v_client, tag(live_call), GRPC_OP_OK); - cq_verify(v_client); - - GPR_ASSERT(GRPC_CALL_OK == - grpc_call_writes_done(live_call_obj, tag(live_call + 3))); cq_expect_finish_accepted(v_client, tag(live_call + 3), GRPC_OP_OK); cq_verify(v_client); diff --git a/test/core/end2end/tests/ping_pong_streaming.c b/test/core/end2end/tests/ping_pong_streaming.c index 02a05e8e0c..1801d34a2e 100644 --- a/test/core/end2end/tests/ping_pong_streaming.c +++ b/test/core/end2end/tests/ping_pong_streaming.c @@ -122,8 +122,7 @@ static void test_pingpong_streaming(grpc_end2end_test_config config, GPR_ASSERT(c); GPR_ASSERT(GRPC_CALL_OK == - grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); - cq_expect_invoke_accepted(v_client, tag(1), GRPC_OP_OK); + grpc_call_invoke(c, f.client_cq, tag(2), tag(3), 0)); GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, tag(100))); diff --git a/test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c b/test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c index 8a0eea8072..c6048bf96a 100644 --- a/test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c +++ b/test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c @@ -145,9 +145,7 @@ static void test_request_response_with_metadata_and_payload( GPR_ASSERT(GRPC_CALL_OK == grpc_call_add_metadata(c, &meta2, 0)); GPR_ASSERT(GRPC_CALL_OK == - grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); - cq_expect_invoke_accepted(v_client, tag(1), GRPC_OP_OK); - cq_verify(v_client); + grpc_call_invoke(c, f.client_cq, tag(2), tag(3), 0)); GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_write(c, request_payload, tag(4), 0)); diff --git a/test/core/end2end/tests/request_response_with_metadata_and_payload.c b/test/core/end2end/tests/request_response_with_metadata_and_payload.c index 77e490710c..0eebebeb00 100644 --- a/test/core/end2end/tests/request_response_with_metadata_and_payload.c +++ b/test/core/end2end/tests/request_response_with_metadata_and_payload.c @@ -136,9 +136,7 @@ static void test_request_response_with_metadata_and_payload( GPR_ASSERT(GRPC_CALL_OK == grpc_call_add_metadata(c, &meta2, 0)); GPR_ASSERT(GRPC_CALL_OK == - grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); - cq_expect_invoke_accepted(v_client, tag(1), GRPC_OP_OK); - cq_verify(v_client); + grpc_call_invoke(c, f.client_cq, tag(2), tag(3), 0)); GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_write(c, request_payload, tag(4), 0)); diff --git a/test/core/end2end/tests/request_response_with_payload.c b/test/core/end2end/tests/request_response_with_payload.c index deb61a7a8b..b2c39e8d6d 100644 --- a/test/core/end2end/tests/request_response_with_payload.c +++ b/test/core/end2end/tests/request_response_with_payload.c @@ -125,9 +125,7 @@ static void request_response_with_payload(grpc_end2end_test_fixture f) { GPR_ASSERT(c); GPR_ASSERT(GRPC_CALL_OK == - grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); - cq_expect_invoke_accepted(v_client, tag(1), GRPC_OP_OK); - cq_verify(v_client); + grpc_call_invoke(c, f.client_cq, tag(2), tag(3), 0)); GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_write(c, request_payload, tag(4), 0)); diff --git a/test/core/end2end/tests/request_response_with_trailing_metadata_and_payload.c b/test/core/end2end/tests/request_response_with_trailing_metadata_and_payload.c index 95e268441b..8fbe514bf3 100644 --- a/test/core/end2end/tests/request_response_with_trailing_metadata_and_payload.c +++ b/test/core/end2end/tests/request_response_with_trailing_metadata_and_payload.c @@ -138,9 +138,7 @@ static void test_request_response_with_metadata_and_payload( GPR_ASSERT(GRPC_CALL_OK == grpc_call_add_metadata(c, &meta2, 0)); GPR_ASSERT(GRPC_CALL_OK == - grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); - cq_expect_invoke_accepted(v_client, tag(1), GRPC_OP_OK); - cq_verify(v_client); + grpc_call_invoke(c, f.client_cq, tag(2), tag(3), 0)); GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_write(c, request_payload, tag(4), 0)); diff --git a/test/core/end2end/tests/request_with_large_metadata.c b/test/core/end2end/tests/request_with_large_metadata.c index 26b165625e..3c634a396d 100644 --- a/test/core/end2end/tests/request_with_large_metadata.c +++ b/test/core/end2end/tests/request_with_large_metadata.c @@ -128,9 +128,7 @@ static void test_request_with_large_metadata(grpc_end2end_test_config config) { GPR_ASSERT(GRPC_CALL_OK == grpc_call_add_metadata(c, &meta, 0)); GPR_ASSERT(GRPC_CALL_OK == - grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); - cq_expect_invoke_accepted(v_client, tag(1), GRPC_OP_OK); - cq_verify(v_client); + grpc_call_invoke(c, f.client_cq, tag(2), tag(3), 0)); cq_expect_server_rpc_new(v_server, &s, tag(100), "/foo", "test.google.com", deadline, "key", meta.value, NULL); diff --git a/test/core/end2end/tests/request_with_payload.c b/test/core/end2end/tests/request_with_payload.c index 602ade6f90..db10fac6f5 100644 --- a/test/core/end2end/tests/request_with_payload.c +++ b/test/core/end2end/tests/request_with_payload.c @@ -122,9 +122,7 @@ static void test_invoke_request_with_payload(grpc_end2end_test_config config) { GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, tag(100))); GPR_ASSERT(GRPC_CALL_OK == - grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); - cq_expect_invoke_accepted(v_client, tag(1), GRPC_OP_OK); - cq_verify(v_client); + grpc_call_invoke(c, f.client_cq, tag(2), tag(3), 0)); GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_write(c, payload, tag(4), 0)); /* destroy byte buffer early to ensure async code keeps track of its contents diff --git a/test/core/end2end/tests/simple_delayed_request.c b/test/core/end2end/tests/simple_delayed_request.c index b90316c32e..c1b528cc3f 100644 --- a/test/core/end2end/tests/simple_delayed_request.c +++ b/test/core/end2end/tests/simple_delayed_request.c @@ -106,10 +106,8 @@ static void simple_delayed_request_body(grpc_end2end_test_config config, c = grpc_channel_create_call(f->client, "/foo", "test.google.com", deadline); GPR_ASSERT(c); - GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_invoke(c, f->client_cq, tag(1), + GPR_ASSERT(GRPC_CALL_OK == grpc_call_invoke(c, f->client_cq, tag(2), tag(3), 0)); - gpr_sleep_until(gpr_time_add(gpr_now(), gpr_time_from_micros(delay_us))); - cq_expect_invoke_accepted(v_client, tag(1), GRPC_OP_OK); config.init_server(f, server_args); diff --git a/test/core/end2end/tests/simple_request.c b/test/core/end2end/tests/simple_request.c index 3511f276e6..58d8023090 100644 --- a/test/core/end2end/tests/simple_request.c +++ b/test/core/end2end/tests/simple_request.c @@ -113,9 +113,7 @@ static void simple_request_body(grpc_end2end_test_fixture f) { GPR_ASSERT(c); GPR_ASSERT(GRPC_CALL_OK == - grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); - cq_expect_invoke_accepted(v_client, tag(1), GRPC_OP_OK); - cq_verify(v_client); + grpc_call_invoke(c, f.client_cq, tag(2), tag(3), 0)); GPR_ASSERT(GRPC_CALL_OK == grpc_call_writes_done(c, tag(4))); cq_expect_finish_accepted(v_client, tag(4), GRPC_OP_OK); @@ -160,9 +158,7 @@ static void simple_request_body2(grpc_end2end_test_fixture f) { GPR_ASSERT(c); GPR_ASSERT(GRPC_CALL_OK == - grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); - cq_expect_invoke_accepted(v_client, tag(1), GRPC_OP_OK); - cq_verify(v_client); + grpc_call_invoke(c, f.client_cq, tag(2), tag(3), 0)); GPR_ASSERT(GRPC_CALL_OK == grpc_call_writes_done(c, tag(4))); cq_expect_finish_accepted(v_client, tag(4), GRPC_OP_OK); diff --git a/test/core/end2end/tests/thread_stress.c b/test/core/end2end/tests/thread_stress.c index 4100b0e35d..65a7930c1c 100644 --- a/test/core/end2end/tests/thread_stress.c +++ b/test/core/end2end/tests/thread_stress.c @@ -106,25 +106,31 @@ static void drain_cq(int client, grpc_completion_queue *cq) { /* Kick off a new request - assumes g_mu taken */ static void start_request() { + gpr_slice slice = gpr_slice_malloc(100); + grpc_byte_buffer *buf; grpc_call *call = grpc_channel_create_call( g_fixture.client, "/Foo", "test.google.com", g_test_end_time); + + memset(GPR_SLICE_START_PTR(slice), 1, GPR_SLICE_LENGTH(slice)); + buf = grpc_byte_buffer_create(&slice, 1); + gpr_slice_unref(slice); + g_active_requests++; GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_invoke(call, g_fixture.client_cq, NULL, NULL, NULL, 0)); + GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_read(call, NULL)); + GPR_ASSERT(GRPC_CALL_OK == + grpc_call_start_write(call, buf, NULL, 0)); + + grpc_byte_buffer_destroy(buf); } /* Async client: handle sending requests, reading responses, and starting new requests when old ones finish */ static void client_thread(void *p) { - int id = (gpr_intptr)p; + gpr_intptr id = (gpr_intptr)p; grpc_event *ev; - gpr_slice slice = gpr_slice_malloc(100); - grpc_byte_buffer *buf; char *estr; - memset(GPR_SLICE_START_PTR(slice), id, GPR_SLICE_LENGTH(slice)); - - buf = grpc_byte_buffer_create(&slice, 1); - gpr_slice_unref(slice); for (;;) { ev = grpc_completion_queue_next(g_fixture.client_cq, n_seconds_time(1)); @@ -135,14 +141,6 @@ static void client_thread(void *p) { gpr_log(GPR_ERROR, "unexpected event: %s", estr); gpr_free(estr); break; - case GRPC_INVOKE_ACCEPTED: - /* better not keep going if the invoke failed */ - if (ev->data.invoke_accepted == GRPC_OP_OK) { - GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_read(ev->call, NULL)); - GPR_ASSERT(GRPC_CALL_OK == - grpc_call_start_write(ev->call, buf, NULL, 0)); - } - break; case GRPC_READ: break; case GRPC_WRITE_ACCEPTED: @@ -173,7 +171,6 @@ static void client_thread(void *p) { gpr_mu_unlock(&g_mu); } - grpc_byte_buffer_destroy(buf); gpr_event_set(&g_client_done[id], (void *)1); } @@ -196,17 +193,17 @@ static void maybe_end_server_call(grpc_call *call, gpr_refcount *rc) { static void server_thread(void *p) { int id = (gpr_intptr)p; - grpc_event *ev; gpr_slice slice = gpr_slice_malloc(100); grpc_byte_buffer *buf; + grpc_event *ev; char *estr; - memset(GPR_SLICE_START_PTR(slice), id, GPR_SLICE_LENGTH(slice)); - - request_server_call(); + memset(GPR_SLICE_START_PTR(slice), 1, GPR_SLICE_LENGTH(slice)); buf = grpc_byte_buffer_create(&slice, 1); gpr_slice_unref(slice); + request_server_call(); + for (;;) { ev = grpc_completion_queue_next(g_fixture.server_cq, n_seconds_time(1)); if (ev) { diff --git a/test/core/end2end/tests/writes_done_hangs_with_pending_read.c b/test/core/end2end/tests/writes_done_hangs_with_pending_read.c index 2241519e89..d959111e43 100644 --- a/test/core/end2end/tests/writes_done_hangs_with_pending_read.c +++ b/test/core/end2end/tests/writes_done_hangs_with_pending_read.c @@ -128,9 +128,7 @@ static void test_writes_done_hangs_with_pending_read( GPR_ASSERT(c); GPR_ASSERT(GRPC_CALL_OK == - grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); - cq_expect_invoke_accepted(v_client, tag(1), GRPC_OP_OK); - cq_verify(v_client); + grpc_call_invoke(c, f.client_cq, tag(2), tag(3), 0)); GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_write(c, request_payload, tag(4), 0)); diff --git a/test/core/surface/lame_client_test.c b/test/core/surface/lame_client_test.c index 0520a39ea2..11d5e4a495 100644 --- a/test/core/surface/lame_client_test.c +++ b/test/core/surface/lame_client_test.c @@ -63,10 +63,9 @@ int main(int argc, char **argv) { /* and invoke the call */ GPR_ASSERT(GRPC_CALL_OK == - grpc_call_start_invoke(call, cq, tag(1), tag(2), tag(3), 0)); + grpc_call_invoke(call, cq, tag(2), tag(3), 0)); /* the call should immediately fail */ - cq_expect_invoke_accepted(cqv, tag(1), GRPC_OP_ERROR); cq_expect_client_metadata_read(cqv, tag(2), NULL); cq_expect_finished(cqv, tag(3), NULL); cq_verify(cqv); |