Moving prototype from Google to GitHub

I'd started some prototyping work on this change before the move to
GitHub; this change restores things.
pull/34/head
Craig Tiller 10 years ago
parent 94e5ddece6
commit 80fa15c151
  1. 5
      include/grpc/grpc.h
  2. 10
      src/core/surface/byte_buffer.c
  3. 149
      src/core/surface/call.c
  4. 11
      test/core/end2end/cq_verifier.c
  5. 1
      test/core/end2end/cq_verifier.h
  6. 9
      test/core/end2end/dualstack_socket_test.c
  7. 213
      test/core/end2end/dualstack_socket_test.c.orig
  8. 3
      test/core/end2end/no_server_test.c
  9. 4
      test/core/end2end/tests/cancel_after_accept.c
  10. 4
      test/core/end2end/tests/cancel_after_accept_and_writes_closed.c
  11. 4
      test/core/end2end/tests/cancel_after_invoke.c
  12. 3
      test/core/end2end/tests/cancel_before_invoke.c
  13. 5
      test/core/end2end/tests/disappearing_server.c
  14. 4
      test/core/end2end/tests/early_server_shutdown_finishes_inflight_calls.c
  15. 4
      test/core/end2end/tests/invoke_large_request.c
  16. 23
      test/core/end2end/tests/max_concurrent_streams.c
  17. 3
      test/core/end2end/tests/ping_pong_streaming.c
  18. 4
      test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c
  19. 4
      test/core/end2end/tests/request_response_with_metadata_and_payload.c
  20. 4
      test/core/end2end/tests/request_response_with_payload.c
  21. 4
      test/core/end2end/tests/request_response_with_trailing_metadata_and_payload.c
  22. 4
      test/core/end2end/tests/request_with_large_metadata.c
  23. 4
      test/core/end2end/tests/request_with_payload.c
  24. 4
      test/core/end2end/tests/simple_delayed_request.c
  25. 8
      test/core/end2end/tests/simple_request.c
  26. 37
      test/core/end2end/tests/thread_stress.c
  27. 4
      test/core/end2end/tests/writes_done_hangs_with_pending_read.c
  28. 3
      test/core/surface/lame_client_test.c

@ -158,6 +158,7 @@ typedef struct grpc_byte_buffer grpc_byte_buffer;
/* Sample helpers to obtain byte buffers (these will certainly move place */ /* 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_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); size_t grpc_byte_buffer_length(grpc_byte_buffer *bb);
void grpc_byte_buffer_destroy(grpc_byte_buffer *byte_buffer); 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 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 completed (there may be other events for the call pending at this
time) */ 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_call_error grpc_call_start_invoke(grpc_call *call,
grpc_completion_queue *cq, grpc_completion_queue *cq,
void *invoke_accepted_tag, void *invoke_accepted_tag,

@ -49,6 +49,16 @@ grpc_byte_buffer *grpc_byte_buffer_create(gpr_slice *slices, size_t nslices) {
return bb; 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) { void grpc_byte_buffer_destroy(grpc_byte_buffer *bb) {
switch (bb->type) { switch (bb->type) {
case GRPC_BB_SLICE_BUFFER: case GRPC_BB_SLICE_BUFFER:

@ -173,11 +173,14 @@ struct grpc_call {
/* protects variables in this section */ /* protects variables in this section */
gpr_mu read_mu; gpr_mu read_mu;
gpr_uint8 received_start;
gpr_uint8 start_ok;
gpr_uint8 reads_done; gpr_uint8 reads_done;
gpr_uint8 received_finish; gpr_uint8 received_finish;
gpr_uint8 received_metadata; gpr_uint8 received_metadata;
gpr_uint8 have_read; gpr_uint8 have_read;
gpr_uint8 have_alarm; gpr_uint8 have_alarm;
gpr_uint8 pending_writes_done;
/* The current outstanding read message tag (only valid if have_read == 1) */ /* The current outstanding read message tag (only valid if have_read == 1) */
void *read_tag; void *read_tag;
void *metadata_tag; void *metadata_tag;
@ -189,6 +192,8 @@ struct grpc_call {
/* The current outstanding send message/context/invoke/end tag (only valid if /* The current outstanding send message/context/invoke/end tag (only valid if
have_write == 1) */ have_write == 1) */
void *write_tag; void *write_tag;
grpc_byte_buffer *pending_write;
gpr_uint32 pending_write_flags;
/* The final status of the call */ /* The final status of the call */
grpc_status_code status_code; grpc_status_code status_code;
@ -230,6 +235,9 @@ grpc_call *grpc_call_create(grpc_channel *channel,
call->status_details = NULL; call->status_details = NULL;
call->received_finish = 0; call->received_finish = 0;
call->reads_done = 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); grpc_metadata_buffer_init(&call->incoming_metadata);
gpr_ref_init(&call->internal_refcount, 1); gpr_ref_init(&call->internal_refcount, 1);
grpc_call_stack_init(channel_stack, server_transport_data, 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; 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) { static void finish_call(grpc_call *call) {
size_t count; size_t count;
grpc_metadata *elements; grpc_metadata *elements;
@ -359,6 +357,88 @@ grpc_call_error grpc_call_start_invoke(grpc_call *call,
void *invoke_accepted_tag, void *invoke_accepted_tag,
void *metadata_read_tag, void *metadata_read_tag,
void *finished_tag, gpr_uint32 flags) { 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_element *elem;
grpc_call_op op; 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 */ /* inform the completion queue of an incoming operation */
grpc_cq_begin_op(cq, call, GRPC_FINISHED); 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_CLIENT_METADATA_READ);
grpc_cq_begin_op(cq, call, GRPC_INVOKE_ACCEPTED);
gpr_mu_lock(&call->read_mu); gpr_mu_lock(&call->read_mu);
@ -401,8 +480,6 @@ grpc_call_error grpc_call_start_invoke(grpc_call *call,
if (call->received_finish) { if (call->received_finish) {
/* handle early cancellation */ /* 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, grpc_cq_end_client_metadata_read(call->cq, metadata_read_tag, call, NULL,
NULL, 0, NULL); NULL, 0, NULL);
finish_call(call); finish_call(call);
@ -412,18 +489,15 @@ grpc_call_error grpc_call_start_invoke(grpc_call *call,
return GRPC_CALL_OK; return GRPC_CALL_OK;
} }
call->write_tag = invoke_accepted_tag;
call->metadata_tag = metadata_read_tag; call->metadata_tag = metadata_read_tag;
call->have_write = 1;
gpr_mu_unlock(&call->read_mu); gpr_mu_unlock(&call->read_mu);
/* call down the filter stack */ /* call down the filter stack */
op.type = GRPC_SEND_START; op.type = GRPC_SEND_START;
op.dir = GRPC_CALL_DOWN; op.dir = GRPC_CALL_DOWN;
op.flags = flags; op.flags = flags;
op.done_cb = done_invoke; op.done_cb = call_started;
op.data.start.pollset = grpc_cq_pollset(cq); op.data.start.pollset = grpc_cq_pollset(cq);
op.user_data = call; 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; 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( void grpc_call_client_initial_metadata_complete(
grpc_call_element *surface_element) { grpc_call_element *surface_element) {
grpc_call *call = grpc_call_from_top_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); 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 /* TODO(ctiller): if flags & GRPC_WRITE_BUFFER_HINT == 0, this indicates a
flush, and that flush should be propogated down from here */ flush, and that flush should be propogated down from here */
if (byte_buffer == NULL) { if (byte_buffer == NULL) {
@ -647,6 +699,15 @@ grpc_call_error grpc_call_start_write(grpc_call *call,
call->write_tag = tag; call->write_tag = tag;
call->have_write = 1; 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.type = GRPC_SEND_MESSAGE;
op.dir = GRPC_CALL_DOWN; op.dir = GRPC_CALL_DOWN;
op.flags = flags; 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 = CALL_ELEM_FROM_CALL(call, 0);
elem->filter->call_op(elem, NULL, &op); elem->filter->call_op(elem, NULL, &op);
}
return GRPC_CALL_OK; 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->write_tag = tag;
call->have_write = 1; 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.type = GRPC_SEND_FINISH;
op.dir = GRPC_CALL_DOWN; op.dir = GRPC_CALL_DOWN;
op.flags = 0; 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 = CALL_ELEM_FROM_CALL(call, 0);
elem->filter->call_op(elem, NULL, &op); elem->filter->call_op(elem, NULL, &op);
}
return GRPC_CALL_OK; return GRPC_CALL_OK;
} }

@ -70,7 +70,6 @@ typedef struct expectation {
union { union {
grpc_op_error finish_accepted; grpc_op_error finish_accepted;
grpc_op_error write_accepted; grpc_op_error write_accepted;
grpc_op_error invoke_accepted;
struct { struct {
const char *method; const char *method;
const char *host; 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); GPR_ASSERT(e->data.write_accepted == ev->data.write_accepted);
break; break;
case GRPC_INVOKE_ACCEPTED: case GRPC_INVOKE_ACCEPTED:
GPR_ASSERT(e->data.invoke_accepted == ev->data.invoke_accepted); abort();
break; break;
case GRPC_SERVER_RPC_NEW: case GRPC_SERVER_RPC_NEW:
GPR_ASSERT(string_equivalent(e->data.server_rpc_new.method, 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", return sprintf(out, "GRPC_WRITE_ACCEPTED result=%d",
e->data.write_accepted); e->data.write_accepted);
case GRPC_INVOKE_ACCEPTED: case GRPC_INVOKE_ACCEPTED:
return sprintf(out, "GRPC_INVOKE_ACCEPTED result=%d", return sprintf(out, "GRPC_INVOKE_ACCEPTED");
e->data.invoke_accepted);
case GRPC_SERVER_RPC_NEW: case GRPC_SERVER_RPC_NEW:
timeout = gpr_time_sub(e->data.server_rpc_new.deadline, gpr_now()); 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", 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) { void cq_expect_write_accepted(cq_verifier *v, void *tag, grpc_op_error result) {
add(v, GRPC_WRITE_ACCEPTED, tag)->data.write_accepted = result; add(v, GRPC_WRITE_ACCEPTED, tag)->data.write_accepted = result;
} }

@ -56,7 +56,6 @@ void cq_verify_empty(cq_verifier *v);
Any functions taking ... expect a NULL terminated list of key/value pairs 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 (each pair using two parameter slots) of metadata that MUST be present in
the event. */ 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_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_finish_accepted(cq_verifier *v, void *tag, grpc_op_error result);
void cq_expect_read(cq_verifier *v, void *tag, gpr_slice bytes); void cq_expect_read(cq_verifier *v, void *tag, gpr_slice bytes);

@ -107,13 +107,10 @@ void test_connect(const char *server_host, const char *client_host, int port,
GPR_ASSERT(c); GPR_ASSERT(c);
GPR_ASSERT(GRPC_CALL_OK == 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) { if (expect_ok) {
/* Check for a successful request. */ /* 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_expect_finish_accepted(v_client, tag(4), GRPC_OP_OK);
cq_verify(v_client); 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); grpc_call_destroy(s);
} else { } else {
/* Check for a failed connection. */ /* 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_client_metadata_read(v_client, tag(2), NULL);
cq_expect_finished_with_status(v_client, tag(3), GRPC_STATUS_CANCELLED, cq_expect_finished_with_status(v_client, tag(3), GRPC_STATUS_CANCELLED,
NULL, NULL); NULL, NULL);
cq_expect_finish_accepted(v_client, tag(4), GRPC_OP_ERROR);
cq_verify(v_client); cq_verify(v_client);
grpc_call_destroy(c); grpc_call_destroy(c);

@ -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;
}

@ -57,10 +57,9 @@ int main(int argc, char **argv) {
/* create a call, channel to a non existant server */ /* create a call, channel to a non existant server */
chan = grpc_channel_create("nonexistant:54321", NULL); chan = grpc_channel_create("nonexistant:54321", NULL);
call = grpc_channel_create_call(chan, "/foo", "nonexistant", deadline); 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); GRPC_CALL_OK);
/* verify that all tags get completed */ /* 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_client_metadata_read(cqv, tag(2), NULL);
cq_expect_finished_with_status(cqv, tag(3), GRPC_STATUS_CANCELLED, NULL, cq_expect_finished_with_status(cqv, tag(3), GRPC_STATUS_CANCELLED, NULL,
NULL); NULL);

@ -124,9 +124,7 @@ static void test_cancel_after_accept(grpc_end2end_test_config config,
GPR_ASSERT(c); GPR_ASSERT(c);
GPR_ASSERT(GRPC_CALL_OK == GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); 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_server_request_call(f.server, tag(100))); 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", cq_expect_server_rpc_new(v_server, &s, tag(100), "/foo", "test.google.com",

@ -124,9 +124,7 @@ static void test_cancel_after_accept_and_writes_closed(
GPR_ASSERT(c); GPR_ASSERT(c);
GPR_ASSERT(GRPC_CALL_OK == GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); 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_server_request_call(f.server, tag(100))); 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", cq_expect_server_rpc_new(v_server, &s, tag(100), "/foo", "test.google.com",

@ -122,9 +122,7 @@ static void test_cancel_after_invoke(grpc_end2end_test_config config,
GPR_ASSERT(c); GPR_ASSERT(c);
GPR_ASSERT(GRPC_CALL_OK == GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); 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 == call_cancel(c)); GPR_ASSERT(GRPC_CALL_OK == call_cancel(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_cancel(c));
GPR_ASSERT(GRPC_CALL_OK == GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); grpc_call_invoke(c, f.client_cq, tag(2), tag(3), 0));
cq_expect_invoke_accepted(v_client, tag(1), GRPC_OP_ERROR);
cq_expect_client_metadata_read(v_client, tag(2), NULL); cq_expect_client_metadata_read(v_client, tag(2), NULL);
cq_expect_finished_with_status(v_client, tag(3), GRPC_STATUS_CANCELLED, NULL, cq_expect_finished_with_status(v_client, tag(3), GRPC_STATUS_CANCELLED, NULL,
NULL); NULL);

@ -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); c = grpc_channel_create_call(f->client, "/foo", "test.google.com", deadline);
GPR_ASSERT(c); 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)); 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))); GPR_ASSERT(GRPC_CALL_OK == grpc_call_writes_done(c, tag(4)));
cq_expect_finish_accepted(v_client, tag(4), GRPC_OP_OK); cq_expect_finish_accepted(v_client, tag(4), GRPC_OP_OK);

@ -115,9 +115,7 @@ static void test_early_server_shutdown_finishes_inflight_calls(
GPR_ASSERT(c); GPR_ASSERT(c);
GPR_ASSERT(GRPC_CALL_OK == GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); 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))); GPR_ASSERT(GRPC_CALL_OK == grpc_call_writes_done(c, tag(4)));
cq_expect_finish_accepted(v_client, tag(4), GRPC_OP_OK); cq_expect_finish_accepted(v_client, tag(4), GRPC_OP_OK);

@ -126,9 +126,7 @@ static void test_invoke_large_request(grpc_end2end_test_config config) {
GPR_ASSERT(c); GPR_ASSERT(c);
GPR_ASSERT(GRPC_CALL_OK == GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); 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 == GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_write(c, request_payload, tag(4), 0)); grpc_call_start_write(c, request_payload, tag(4), 0));

@ -113,9 +113,7 @@ static void simple_request_body(grpc_end2end_test_fixture f) {
GPR_ASSERT(c); GPR_ASSERT(c);
GPR_ASSERT(GRPC_CALL_OK == GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); 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))); GPR_ASSERT(GRPC_CALL_OK == grpc_call_writes_done(c, tag(4)));
cq_expect_finish_accepted(v_client, tag(4), GRPC_OP_OK); 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_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)); 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)); 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( ev = grpc_completion_queue_next(
f.client_cq, gpr_time_add(gpr_now(), gpr_time_from_seconds(10))); f.client_cq, gpr_time_add(gpr_now(), gpr_time_from_seconds(10)));
GPR_ASSERT(ev); 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); GPR_ASSERT(ev->data.invoke_accepted == GRPC_OP_OK);
/* The /alpha or /beta calls started above could be invoked (but NOT both); /* The /alpha or /beta calls started above could be invoked (but NOT both);
* check this here */ * 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; live_call_obj = live_call == 300 ? c1 : c2;
grpc_event_finish(ev); 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), cq_expect_server_rpc_new(v_server, &s1, tag(100),
live_call == 300 ? "/alpha" : "/beta", live_call == 300 ? "/alpha" : "/beta",
"test.google.com", deadline, NULL); "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); GRPC_STATUS_UNIMPLEMENTED, "xyz", NULL);
live_call = (live_call == 300) ? 400 : 300; live_call = (live_call == 300) ? 400 : 300;
live_call_obj = live_call == 300 ? c1 : c2; 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_expect_finish_accepted(v_client, tag(live_call + 3), GRPC_OP_OK);
cq_verify(v_client); cq_verify(v_client);

@ -122,8 +122,7 @@ static void test_pingpong_streaming(grpc_end2end_test_config config,
GPR_ASSERT(c); GPR_ASSERT(c);
GPR_ASSERT(GRPC_CALL_OK == GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); grpc_call_invoke(c, f.client_cq, tag(2), tag(3), 0));
cq_expect_invoke_accepted(v_client, tag(1), GRPC_OP_OK);
GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, tag(100))); GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, tag(100)));

@ -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_add_metadata(c, &meta2, 0));
GPR_ASSERT(GRPC_CALL_OK == GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); 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 == GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_write(c, request_payload, tag(4), 0)); grpc_call_start_write(c, request_payload, tag(4), 0));

@ -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_add_metadata(c, &meta2, 0));
GPR_ASSERT(GRPC_CALL_OK == GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); 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 == GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_write(c, request_payload, tag(4), 0)); grpc_call_start_write(c, request_payload, tag(4), 0));

@ -125,9 +125,7 @@ static void request_response_with_payload(grpc_end2end_test_fixture f) {
GPR_ASSERT(c); GPR_ASSERT(c);
GPR_ASSERT(GRPC_CALL_OK == GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); 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 == GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_write(c, request_payload, tag(4), 0)); grpc_call_start_write(c, request_payload, tag(4), 0));

@ -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_add_metadata(c, &meta2, 0));
GPR_ASSERT(GRPC_CALL_OK == GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); 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 == GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_write(c, request_payload, tag(4), 0)); grpc_call_start_write(c, request_payload, tag(4), 0));

@ -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_add_metadata(c, &meta, 0));
GPR_ASSERT(GRPC_CALL_OK == GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); 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);
cq_expect_server_rpc_new(v_server, &s, tag(100), "/foo", "test.google.com", cq_expect_server_rpc_new(v_server, &s, tag(100), "/foo", "test.google.com",
deadline, "key", meta.value, NULL); deadline, "key", meta.value, NULL);

@ -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_server_request_call(f.server, tag(100)));
GPR_ASSERT(GRPC_CALL_OK == GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); 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_start_write(c, payload, tag(4), 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 /* destroy byte buffer early to ensure async code keeps track of its contents

@ -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); c = grpc_channel_create_call(f->client, "/foo", "test.google.com", deadline);
GPR_ASSERT(c); 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)); 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); config.init_server(f, server_args);

@ -113,9 +113,7 @@ static void simple_request_body(grpc_end2end_test_fixture f) {
GPR_ASSERT(c); GPR_ASSERT(c);
GPR_ASSERT(GRPC_CALL_OK == GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); 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))); GPR_ASSERT(GRPC_CALL_OK == grpc_call_writes_done(c, tag(4)));
cq_expect_finish_accepted(v_client, tag(4), GRPC_OP_OK); 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(c);
GPR_ASSERT(GRPC_CALL_OK == GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); 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))); GPR_ASSERT(GRPC_CALL_OK == grpc_call_writes_done(c, tag(4)));
cq_expect_finish_accepted(v_client, tag(4), GRPC_OP_OK); cq_expect_finish_accepted(v_client, tag(4), GRPC_OP_OK);

@ -106,25 +106,31 @@ static void drain_cq(int client, grpc_completion_queue *cq) {
/* Kick off a new request - assumes g_mu taken */ /* Kick off a new request - assumes g_mu taken */
static void start_request() { static void start_request() {
gpr_slice slice = gpr_slice_malloc(100);
grpc_byte_buffer *buf;
grpc_call *call = grpc_channel_create_call( grpc_call *call = grpc_channel_create_call(
g_fixture.client, "/Foo", "test.google.com", g_test_end_time); 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++; g_active_requests++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_invoke(call, g_fixture.client_cq, GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_invoke(call, g_fixture.client_cq,
NULL, NULL, NULL, 0)); 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 /* Async client: handle sending requests, reading responses, and starting
new requests when old ones finish */ new requests when old ones finish */
static void client_thread(void *p) { static void client_thread(void *p) {
int id = (gpr_intptr)p; gpr_intptr id = (gpr_intptr)p;
grpc_event *ev; grpc_event *ev;
gpr_slice slice = gpr_slice_malloc(100);
grpc_byte_buffer *buf;
char *estr; 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 (;;) { for (;;) {
ev = grpc_completion_queue_next(g_fixture.client_cq, n_seconds_time(1)); 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_log(GPR_ERROR, "unexpected event: %s", estr);
gpr_free(estr); gpr_free(estr);
break; 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: case GRPC_READ:
break; break;
case GRPC_WRITE_ACCEPTED: case GRPC_WRITE_ACCEPTED:
@ -173,7 +171,6 @@ static void client_thread(void *p) {
gpr_mu_unlock(&g_mu); gpr_mu_unlock(&g_mu);
} }
grpc_byte_buffer_destroy(buf);
gpr_event_set(&g_client_done[id], (void *)1); 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) { static void server_thread(void *p) {
int id = (gpr_intptr)p; int id = (gpr_intptr)p;
grpc_event *ev;
gpr_slice slice = gpr_slice_malloc(100); gpr_slice slice = gpr_slice_malloc(100);
grpc_byte_buffer *buf; grpc_byte_buffer *buf;
grpc_event *ev;
char *estr; 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); buf = grpc_byte_buffer_create(&slice, 1);
gpr_slice_unref(slice); gpr_slice_unref(slice);
request_server_call();
for (;;) { for (;;) {
ev = grpc_completion_queue_next(g_fixture.server_cq, n_seconds_time(1)); ev = grpc_completion_queue_next(g_fixture.server_cq, n_seconds_time(1));
if (ev) { if (ev) {

@ -128,9 +128,7 @@ static void test_writes_done_hangs_with_pending_read(
GPR_ASSERT(c); GPR_ASSERT(c);
GPR_ASSERT(GRPC_CALL_OK == GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_invoke(c, f.client_cq, tag(1), tag(2), tag(3), 0)); 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 == GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_write(c, request_payload, tag(4), 0)); grpc_call_start_write(c, request_payload, tag(4), 0));

@ -63,10 +63,9 @@ int main(int argc, char **argv) {
/* and invoke the call */ /* and invoke the call */
GPR_ASSERT(GRPC_CALL_OK == 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 */ /* 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_client_metadata_read(cqv, tag(2), NULL);
cq_expect_finished(cqv, tag(3), NULL); cq_expect_finished(cqv, tag(3), NULL);
cq_verify(cqv); cq_verify(cqv);

Loading…
Cancel
Save