Merge branch 'bye-bye-completion-queue-pie' into we-dont-need-no-backup
diff --git a/Makefile b/Makefile
index 417b538..cd5858c 100644
--- a/Makefile
+++ b/Makefile
@@ -633,7 +633,6 @@
grpc_base64_test: $(BINDIR)/$(CONFIG)/grpc_base64_test
grpc_byte_buffer_reader_test: $(BINDIR)/$(CONFIG)/grpc_byte_buffer_reader_test
grpc_channel_stack_test: $(BINDIR)/$(CONFIG)/grpc_channel_stack_test
-grpc_completion_queue_benchmark: $(BINDIR)/$(CONFIG)/grpc_completion_queue_benchmark
grpc_completion_queue_test: $(BINDIR)/$(CONFIG)/grpc_completion_queue_test
grpc_create_jwt: $(BINDIR)/$(CONFIG)/grpc_create_jwt
grpc_credentials_test: $(BINDIR)/$(CONFIG)/grpc_credentials_test
@@ -1844,7 +1843,7 @@
tools: privatelibs $(BINDIR)/$(CONFIG)/gen_hpack_tables $(BINDIR)/$(CONFIG)/grpc_create_jwt $(BINDIR)/$(CONFIG)/grpc_fetch_oauth2 $(BINDIR)/$(CONFIG)/grpc_print_google_default_creds_token $(BINDIR)/$(CONFIG)/qps_driver $(BINDIR)/$(CONFIG)/qps_worker
-buildbenchmarks: privatelibs $(BINDIR)/$(CONFIG)/grpc_completion_queue_benchmark $(BINDIR)/$(CONFIG)/low_level_ping_pong_benchmark
+buildbenchmarks: privatelibs $(BINDIR)/$(CONFIG)/low_level_ping_pong_benchmark
benchmarks: buildbenchmarks
@@ -5714,35 +5713,6 @@
endif
-GRPC_COMPLETION_QUEUE_BENCHMARK_SRC = \
- test/core/surface/completion_queue_benchmark.c \
-
-GRPC_COMPLETION_QUEUE_BENCHMARK_OBJS = $(addprefix $(OBJDIR)/$(CONFIG)/, $(addsuffix .o, $(basename $(GRPC_COMPLETION_QUEUE_BENCHMARK_SRC))))
-ifeq ($(NO_SECURE),true)
-
-# You can't build secure targets if you don't have OpenSSL with ALPN.
-
-$(BINDIR)/$(CONFIG)/grpc_completion_queue_benchmark: openssl_dep_error
-
-else
-
-$(BINDIR)/$(CONFIG)/grpc_completion_queue_benchmark: $(GRPC_COMPLETION_QUEUE_BENCHMARK_OBJS) $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a
- $(E) "[LD] Linking $@"
- $(Q) mkdir -p `dirname $@`
- $(Q) $(LD) $(LDFLAGS) $(GRPC_COMPLETION_QUEUE_BENCHMARK_OBJS) $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a $(LDLIBS) $(LDLIBS_SECURE) -o $(BINDIR)/$(CONFIG)/grpc_completion_queue_benchmark
-
-endif
-
-$(OBJDIR)/$(CONFIG)/test/core/surface/completion_queue_benchmark.o: $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a
-deps_grpc_completion_queue_benchmark: $(GRPC_COMPLETION_QUEUE_BENCHMARK_OBJS:.o=.dep)
-
-ifneq ($(NO_SECURE),true)
-ifneq ($(NO_DEPS),true)
--include $(GRPC_COMPLETION_QUEUE_BENCHMARK_OBJS:.o=.dep)
-endif
-endif
-
-
GRPC_COMPLETION_QUEUE_TEST_SRC = \
test/core/surface/completion_queue_test.c \
diff --git a/build.json b/build.json
index ac81fff..8d256e9 100644
--- a/build.json
+++ b/build.json
@@ -1310,20 +1310,6 @@
]
},
{
- "name": "grpc_completion_queue_benchmark",
- "build": "benchmark",
- "language": "c",
- "src": [
- "test/core/surface/completion_queue_benchmark.c"
- ],
- "deps": [
- "grpc_test_util",
- "grpc",
- "gpr_test_util",
- "gpr"
- ]
- },
- {
"name": "grpc_completion_queue_test",
"build": "test",
"language": "c",
diff --git a/include/grpc/grpc.h b/include/grpc/grpc.h
index 0212b42..ce01b2d 100644
--- a/include/grpc/grpc.h
+++ b/include/grpc/grpc.h
@@ -145,14 +145,6 @@
GRPC_CALL_ERROR_INVALID_METADATA
} grpc_call_error;
-/* Result of a grpc operation */
-typedef enum grpc_op_error {
- /* everything went ok */
- GRPC_OP_OK = 0,
- /* something failed, we don't know what */
- GRPC_OP_ERROR
-} grpc_op_error;
-
/* Write Flags: */
/* Hint that the write may be buffered and need not go out on the wire
immediately. GRPC is free to buffer the message until the next non-buffered
@@ -201,22 +193,15 @@
} grpc_metadata;
typedef enum grpc_completion_type {
- GRPC_QUEUE_SHUTDOWN, /* Shutting down */
- GRPC_OP_COMPLETE, /* operation completion */
- GRPC_SERVER_SHUTDOWN, /* The server has finished shutting down */
- GRPC_COMPLETION_DO_NOT_USE /* must be last, forces users to include
- a default: case */
+ GRPC_QUEUE_SHUTDOWN, /* Shutting down */
+ GRPC_QUEUE_TIMEOUT, /* No event before timeout */
+ GRPC_OP_COMPLETE /* operation completion */
} grpc_completion_type;
typedef struct grpc_event {
grpc_completion_type type;
+ int success;
void *tag;
- grpc_call *call;
- /* Data associated with the completion type. Field names match the type of
- completion as listed in grpc_completion_type. */
- union {
- grpc_op_error op_complete;
- } data;
} grpc_event;
typedef struct {
@@ -352,26 +337,21 @@
/* Blocks until an event is available, the completion queue is being shut down,
or deadline is reached. Returns NULL on timeout, otherwise the event that
- occurred. Callers should call grpc_event_finish once they have processed
- the event.
+ occurred.
Callers must not call grpc_completion_queue_next and
grpc_completion_queue_pluck simultaneously on the same completion queue. */
-grpc_event *grpc_completion_queue_next(grpc_completion_queue *cq,
- gpr_timespec deadline);
+grpc_event grpc_completion_queue_next(grpc_completion_queue *cq,
+ gpr_timespec deadline);
/* Blocks until an event with tag 'tag' is available, the completion queue is
being shutdown or deadline is reached. Returns NULL on timeout, or a pointer
- to the event that occurred. Callers should call grpc_event_finish once they
- have processed the event.
+ to the event that occurred.
Callers must not call grpc_completion_queue_next and
grpc_completion_queue_pluck simultaneously on the same completion queue. */
-grpc_event *grpc_completion_queue_pluck(grpc_completion_queue *cq, void *tag,
- gpr_timespec deadline);
-
-/* Clean up any data owned by the event */
-void grpc_event_finish(grpc_event *event);
+grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cq, void *tag,
+ gpr_timespec deadline);
/* Begin destruction of a completion queue. Once all possible events are
drained then grpc_completion_queue_next will start to produce
@@ -508,7 +488,7 @@
Shutdown is idempotent. */
void grpc_server_shutdown(grpc_server *server);
-/* As per grpc_server_shutdown, but send a GRPC_SERVER_SHUTDOWN event when
+/* As per grpc_server_shutdown, but send a GRPC_OP_COMPLETE event when
there are no more calls being serviced.
Shutdown is idempotent, and all tags will be notified at once if multiple
grpc_server_shutdown_and_notify calls are made. */
diff --git a/src/core/surface/call.c b/src/core/surface/call.c
index 41988fa..d403eee 100644
--- a/src/core/surface/call.c
+++ b/src/core/surface/call.c
@@ -62,7 +62,7 @@
typedef struct {
grpc_ioreq_completion_func on_complete;
void *user_data;
- grpc_op_error status;
+ int success;
} completed_request;
/* See request_set in grpc_call below for a description */
@@ -74,7 +74,7 @@
typedef struct {
/* Overall status of the operation: starts OK, may degrade to
non-OK */
- grpc_op_error status;
+ int success;
/* Completion function to call at the end of the operation */
grpc_ioreq_completion_func on_complete;
void *user_data;
@@ -235,7 +235,6 @@
#define CALL_FROM_TOP_ELEM(top_elem) \
CALL_FROM_CALL_STACK(grpc_call_stack_from_top_element(top_elem))
-static void do_nothing(void *ignored, grpc_op_error also_ignored) {}
static void set_deadline_alarm(grpc_call *call, gpr_timespec deadline);
static void call_on_done_recv(void *call, int success);
static void call_on_done_send(void *call, int success);
@@ -457,7 +456,7 @@
if (completing_requests > 0) {
for (i = 0; i < completing_requests; i++) {
- completed_requests[i].on_complete(call, completed_requests[i].status,
+ completed_requests[i].on_complete(call, completed_requests[i].success,
completed_requests[i].user_data);
}
lock(call);
@@ -517,7 +516,7 @@
}
static void finish_live_ioreq_op(grpc_call *call, grpc_ioreq_op op,
- grpc_op_error status) {
+ int success) {
completed_request *cr;
gpr_uint8 master_set = call->request_set[op];
reqinfo_master *master;
@@ -525,8 +524,8 @@
/* ioreq is live: we need to do something */
master = &call->masters[master_set];
master->complete_mask |= 1u << op;
- if (status != GRPC_OP_OK) {
- master->status = status;
+ if (!success) {
+ master->success = 0;
}
if (master->complete_mask == master->need_mask) {
for (i = 0; i < GRPC_IOREQ_OP_COUNT; i++) {
@@ -537,7 +536,7 @@
switch ((grpc_ioreq_op)i) {
case GRPC_IOREQ_RECV_MESSAGE:
case GRPC_IOREQ_SEND_MESSAGE:
- if (master->status == GRPC_OP_OK) {
+ if (master->success) {
call->request_set[i] = REQSET_EMPTY;
} else {
call->write_state = WRITE_STATE_WRITE_CLOSED;
@@ -572,34 +571,32 @@
}
}
cr = &call->completed_requests[call->num_completed_requests++];
- cr->status = master->status;
+ cr->success = master->success;
cr->on_complete = master->on_complete;
cr->user_data = master->user_data;
}
}
-static void finish_ioreq_op(grpc_call *call, grpc_ioreq_op op,
- grpc_op_error status) {
+static void finish_ioreq_op(grpc_call *call, grpc_ioreq_op op, int success) {
if (is_op_live(call, op)) {
- finish_live_ioreq_op(call, op, status);
+ finish_live_ioreq_op(call, op, success);
}
}
static void call_on_done_send(void *pc, int success) {
grpc_call *call = pc;
- grpc_op_error error = success ? GRPC_OP_OK : GRPC_OP_ERROR;
lock(call);
if (call->last_send_contains & (1 << GRPC_IOREQ_SEND_INITIAL_METADATA)) {
- finish_ioreq_op(call, GRPC_IOREQ_SEND_INITIAL_METADATA, error);
+ finish_ioreq_op(call, GRPC_IOREQ_SEND_INITIAL_METADATA, success);
call->write_state = WRITE_STATE_STARTED;
}
if (call->last_send_contains & (1 << GRPC_IOREQ_SEND_MESSAGE)) {
- finish_ioreq_op(call, GRPC_IOREQ_SEND_MESSAGE, error);
+ finish_ioreq_op(call, GRPC_IOREQ_SEND_MESSAGE, success);
}
if (call->last_send_contains & (1 << GRPC_IOREQ_SEND_CLOSE)) {
- finish_ioreq_op(call, GRPC_IOREQ_SEND_TRAILING_METADATA, error);
- finish_ioreq_op(call, GRPC_IOREQ_SEND_STATUS, error);
- finish_ioreq_op(call, GRPC_IOREQ_SEND_CLOSE, GRPC_OP_OK);
+ finish_ioreq_op(call, GRPC_IOREQ_SEND_TRAILING_METADATA, success);
+ finish_ioreq_op(call, GRPC_IOREQ_SEND_STATUS, success);
+ finish_ioreq_op(call, GRPC_IOREQ_SEND_CLOSE, 1);
call->write_state = WRITE_STATE_WRITE_CLOSED;
}
if (!success) {
@@ -726,12 +723,12 @@
}
finish_read_ops(call);
} else {
- finish_ioreq_op(call, GRPC_IOREQ_RECV_MESSAGE, GRPC_OP_ERROR);
- finish_ioreq_op(call, GRPC_IOREQ_RECV_STATUS, GRPC_OP_ERROR);
- finish_ioreq_op(call, GRPC_IOREQ_RECV_CLOSE, GRPC_OP_ERROR);
- finish_ioreq_op(call, GRPC_IOREQ_RECV_TRAILING_METADATA, GRPC_OP_ERROR);
- finish_ioreq_op(call, GRPC_IOREQ_RECV_INITIAL_METADATA, GRPC_OP_ERROR);
- finish_ioreq_op(call, GRPC_IOREQ_RECV_STATUS_DETAILS, GRPC_OP_ERROR);
+ finish_ioreq_op(call, GRPC_IOREQ_RECV_MESSAGE, 0);
+ finish_ioreq_op(call, GRPC_IOREQ_RECV_STATUS, 0);
+ finish_ioreq_op(call, GRPC_IOREQ_RECV_CLOSE, 0);
+ finish_ioreq_op(call, GRPC_IOREQ_RECV_TRAILING_METADATA, 0);
+ finish_ioreq_op(call, GRPC_IOREQ_RECV_INITIAL_METADATA, 0);
+ finish_ioreq_op(call, GRPC_IOREQ_RECV_STATUS_DETAILS, 0);
}
call->recv_ops.nops = 0;
unlock(call);
@@ -894,7 +891,7 @@
(NULL == (*call->request_data[GRPC_IOREQ_RECV_MESSAGE].recv_message =
grpc_bbq_pop(&call->incoming_queue)));
if (!empty) {
- finish_live_ioreq_op(call, GRPC_IOREQ_RECV_MESSAGE, GRPC_OP_OK);
+ finish_live_ioreq_op(call, GRPC_IOREQ_RECV_MESSAGE, 1);
empty = grpc_bbq_empty(&call->incoming_queue);
}
} else {
@@ -904,19 +901,19 @@
switch (call->read_state) {
case READ_STATE_STREAM_CLOSED:
if (empty) {
- finish_ioreq_op(call, GRPC_IOREQ_RECV_CLOSE, GRPC_OP_OK);
+ finish_ioreq_op(call, GRPC_IOREQ_RECV_CLOSE, 1);
}
/* fallthrough */
case READ_STATE_READ_CLOSED:
if (empty) {
- finish_ioreq_op(call, GRPC_IOREQ_RECV_MESSAGE, GRPC_OP_OK);
+ finish_ioreq_op(call, GRPC_IOREQ_RECV_MESSAGE, 1);
}
- finish_ioreq_op(call, GRPC_IOREQ_RECV_STATUS, GRPC_OP_OK);
- finish_ioreq_op(call, GRPC_IOREQ_RECV_STATUS_DETAILS, GRPC_OP_OK);
- finish_ioreq_op(call, GRPC_IOREQ_RECV_TRAILING_METADATA, GRPC_OP_OK);
+ finish_ioreq_op(call, GRPC_IOREQ_RECV_STATUS, 1);
+ finish_ioreq_op(call, GRPC_IOREQ_RECV_STATUS_DETAILS, 1);
+ finish_ioreq_op(call, GRPC_IOREQ_RECV_TRAILING_METADATA, 1);
/* fallthrough */
case READ_STATE_GOT_INITIAL_METADATA:
- finish_ioreq_op(call, GRPC_IOREQ_RECV_INITIAL_METADATA, GRPC_OP_OK);
+ finish_ioreq_op(call, GRPC_IOREQ_RECV_INITIAL_METADATA, 1);
/* fallthrough */
case READ_STATE_INITIAL:
/* do nothing */
@@ -927,13 +924,13 @@
static void early_out_write_ops(grpc_call *call) {
switch (call->write_state) {
case WRITE_STATE_WRITE_CLOSED:
- finish_ioreq_op(call, GRPC_IOREQ_SEND_MESSAGE, GRPC_OP_ERROR);
- finish_ioreq_op(call, GRPC_IOREQ_SEND_STATUS, GRPC_OP_ERROR);
- finish_ioreq_op(call, GRPC_IOREQ_SEND_TRAILING_METADATA, GRPC_OP_ERROR);
- finish_ioreq_op(call, GRPC_IOREQ_SEND_CLOSE, GRPC_OP_OK);
+ finish_ioreq_op(call, GRPC_IOREQ_SEND_MESSAGE, 0);
+ finish_ioreq_op(call, GRPC_IOREQ_SEND_STATUS, 0);
+ finish_ioreq_op(call, GRPC_IOREQ_SEND_TRAILING_METADATA, 0);
+ finish_ioreq_op(call, GRPC_IOREQ_SEND_CLOSE, 1);
/* fallthrough */
case WRITE_STATE_STARTED:
- finish_ioreq_op(call, GRPC_IOREQ_SEND_INITIAL_METADATA, GRPC_OP_ERROR);
+ finish_ioreq_op(call, GRPC_IOREQ_SEND_INITIAL_METADATA, 0);
/* fallthrough */
case WRITE_STATE_INITIAL:
/* do nothing */
@@ -982,7 +979,7 @@
}
master = &call->masters[set];
- master->status = GRPC_OP_OK;
+ master->success = 1;
master->need_mask = have_ops;
master->complete_mask = 0;
master->on_complete = completion;
@@ -1180,8 +1177,8 @@
*(grpc_status_code *)dest = (status != GRPC_STATUS_OK);
}
-static void finish_batch(grpc_call *call, grpc_op_error result, void *tag) {
- grpc_cq_end_op(call->cq, tag, call, do_nothing, NULL, GRPC_OP_OK);
+static void finish_batch(grpc_call *call, int success, void *tag) {
+ grpc_cq_end_op(call->cq, tag, call, 1);
}
grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops,
@@ -1195,8 +1192,8 @@
GRPC_CALL_LOG_BATCH(GPR_INFO, call, ops, nops, tag);
if (nops == 0) {
- grpc_cq_begin_op(call->cq, call, GRPC_OP_COMPLETE);
- grpc_cq_end_op(call->cq, tag, call, do_nothing, NULL, GRPC_OP_OK);
+ grpc_cq_begin_op(call->cq, call);
+ grpc_cq_end_op(call->cq, tag, call, 1);
return GRPC_CALL_OK;
}
@@ -1287,7 +1284,7 @@
}
}
- grpc_cq_begin_op(call->cq, call, GRPC_OP_COMPLETE);
+ grpc_cq_begin_op(call->cq, call);
return grpc_call_start_ioreq_and_call_back(call, reqs, out, finish_batch,
tag);
diff --git a/src/core/surface/call.h b/src/core/surface/call.h
index 02378b6..60222bf 100644
--- a/src/core/surface/call.h
+++ b/src/core/surface/call.h
@@ -81,8 +81,7 @@
grpc_ioreq_data data;
} grpc_ioreq;
-typedef void (*grpc_ioreq_completion_func)(grpc_call *call,
- grpc_op_error status,
+typedef void (*grpc_ioreq_completion_func)(grpc_call *call, int success,
void *user_data);
grpc_call *grpc_call_create(grpc_channel *channel, grpc_completion_queue *cq,
diff --git a/src/core/surface/completion_queue.c b/src/core/surface/completion_queue.c
index 3e90318..48910af 100644
--- a/src/core/surface/completion_queue.c
+++ b/src/core/surface/completion_queue.c
@@ -51,8 +51,6 @@
function (on_finish) that is hidden from outside this module */
typedef struct event {
grpc_event base;
- grpc_event_finish_func on_finish;
- void *on_finish_user_data;
struct event *queue_next;
struct event *queue_prev;
struct event *bucket_next;
@@ -78,16 +76,8 @@
event *queue;
/* Fixed size chained hash table of events for pluck() */
event *buckets[NUM_TAG_BUCKETS];
-
-#ifndef NDEBUG
- /* Debug support: track which operations are in flight at any given time */
- gpr_atm pending_op_count[GRPC_COMPLETION_DO_NOT_USE];
-#endif
};
-/* Default do-nothing on_finish function */
-static void null_on_finish(void *user_data, grpc_op_error error) {}
-
grpc_completion_queue *grpc_completion_queue_create(void) {
grpc_completion_queue *cc = gpr_malloc(sizeof(grpc_completion_queue));
memset(cc, 0, sizeof(*cc));
@@ -124,15 +114,11 @@
members can be filled in.
Requires GRPC_POLLSET_MU(&cc->pollset) locked. */
static event *add_locked(grpc_completion_queue *cc, grpc_completion_type type,
- void *tag, grpc_call *call,
- grpc_event_finish_func on_finish, void *user_data) {
+ void *tag, grpc_call *call) {
event *ev = gpr_malloc(sizeof(event));
gpr_uintptr bucket = ((gpr_uintptr)tag) % NUM_TAG_BUCKETS;
ev->base.type = type;
ev->base.tag = tag;
- ev->base.call = call;
- ev->on_finish = on_finish ? on_finish : null_on_finish;
- ev->on_finish_user_data = user_data;
if (cc->queue == NULL) {
cc->queue = ev->queue_next = ev->queue_prev = ev;
} else {
@@ -152,22 +138,15 @@
return ev;
}
-void grpc_cq_begin_op(grpc_completion_queue *cc, grpc_call *call,
- grpc_completion_type type) {
+void grpc_cq_begin_op(grpc_completion_queue *cc, grpc_call *call) {
gpr_ref(&cc->refs);
if (call) GRPC_CALL_INTERNAL_REF(call, "cq");
-#ifndef NDEBUG
- gpr_atm_no_barrier_fetch_add(&cc->pending_op_count[type], 1);
-#endif
}
/* Signal the end of an operation - if this is the last waiting-to-be-queued
event, then enter shutdown mode */
static void end_op_locked(grpc_completion_queue *cc,
grpc_completion_type type) {
-#ifndef NDEBUG
- GPR_ASSERT(gpr_atm_full_fetch_add(&cc->pending_op_count[type], -1) > 0);
-#endif
if (gpr_unref(&cc->refs)) {
GPR_ASSERT(!cc->shutdown);
GPR_ASSERT(cc->shutdown_called);
@@ -176,37 +155,29 @@
}
}
-void grpc_cq_end_server_shutdown(grpc_completion_queue *cc, void *tag) {
- gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
- add_locked(cc, GRPC_SERVER_SHUTDOWN, tag, NULL, NULL, NULL);
- end_op_locked(cc, GRPC_SERVER_SHUTDOWN);
- gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
-}
-
void grpc_cq_end_op(grpc_completion_queue *cc, void *tag, grpc_call *call,
- grpc_event_finish_func on_finish, void *user_data,
- grpc_op_error error) {
+ int success) {
event *ev;
gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
- ev = add_locked(cc, GRPC_OP_COMPLETE, tag, call, on_finish, user_data);
- ev->base.data.op_complete = error;
+ ev = add_locked(cc, GRPC_OP_COMPLETE, tag, call);
+ ev->base.success = success;
end_op_locked(cc, GRPC_OP_COMPLETE);
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
+ if (call) GRPC_CALL_INTERNAL_UNREF(call, "cq", 0);
}
/* Create a GRPC_QUEUE_SHUTDOWN event without queuing it anywhere */
static event *create_shutdown_event(void) {
event *ev = gpr_malloc(sizeof(event));
ev->base.type = GRPC_QUEUE_SHUTDOWN;
- ev->base.call = NULL;
ev->base.tag = NULL;
- ev->on_finish = null_on_finish;
return ev;
}
-grpc_event *grpc_completion_queue_next(grpc_completion_queue *cc,
- gpr_timespec deadline) {
+grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
+ gpr_timespec deadline) {
event *ev = NULL;
+ grpc_event ret;
gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
for (;;) {
@@ -240,12 +211,17 @@
if (gpr_cv_wait(GRPC_POLLSET_CV(&cc->pollset),
GRPC_POLLSET_MU(&cc->pollset), deadline)) {
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
- return NULL;
+ memset(&ret, 0, sizeof(ret));
+ ret.type = GRPC_QUEUE_TIMEOUT;
+ GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret);
+ return ret;
}
}
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
- GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ev->base);
- return &ev->base;
+ ret = ev->base;
+ gpr_free(ev);
+ GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret);
+ return ret;
}
static event *pluck_event(grpc_completion_queue *cc, void *tag) {
@@ -277,9 +253,10 @@
return NULL;
}
-grpc_event *grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
- gpr_timespec deadline) {
+grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
+ gpr_timespec deadline) {
event *ev = NULL;
+ grpc_event ret;
gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
for (;;) {
@@ -296,12 +273,17 @@
if (gpr_cv_wait(GRPC_POLLSET_CV(&cc->pollset),
GRPC_POLLSET_MU(&cc->pollset), deadline)) {
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
- return NULL;
+ memset(&ret, 0, sizeof(ret));
+ ret.type = GRPC_QUEUE_TIMEOUT;
+ GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ev->base);
+ return ret;
}
}
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
+ ret = ev->base;
+ gpr_free(ev);
GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ev->base);
- return &ev->base;
+ return ret;
}
/* Shutdown simply drops a ref that we reserved at creation time; if we drop
@@ -324,30 +306,6 @@
grpc_cq_internal_unref(cc);
}
-void grpc_event_finish(grpc_event *base) {
- event *ev = (event *)base;
- ev->on_finish(ev->on_finish_user_data, GRPC_OP_OK);
- if (ev->base.call) {
- GRPC_CALL_INTERNAL_UNREF(ev->base.call, "cq", 1);
- }
- gpr_free(ev);
-}
-
-void grpc_cq_dump_pending_ops(grpc_completion_queue *cc) {
-#ifndef NDEBUG
- char tmp[GRPC_COMPLETION_DO_NOT_USE * (1 + GPR_LTOA_MIN_BUFSIZE)];
- char *p = tmp;
- int i;
-
- for (i = 0; i < GRPC_COMPLETION_DO_NOT_USE; i++) {
- *p++ = ' ';
- p += gpr_ltoa(cc->pending_op_count[i], p);
- }
-
- gpr_log(GPR_INFO, "pending ops:%s", tmp);
-#endif
-}
-
grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc) {
return &cc->pollset;
}
diff --git a/src/core/surface/completion_queue.h b/src/core/surface/completion_queue.h
index a0d7eea..7b6fad9 100644
--- a/src/core/surface/completion_queue.h
+++ b/src/core/surface/completion_queue.h
@@ -39,41 +39,20 @@
#include "src/core/iomgr/pollset.h"
#include <grpc/grpc.h>
-/* A finish func is executed whenever the event consumer calls
- grpc_event_finish */
-typedef void (*grpc_event_finish_func)(void *user_data, grpc_op_error error);
-
void grpc_cq_internal_ref(grpc_completion_queue *cc);
void grpc_cq_internal_unref(grpc_completion_queue *cc);
/* Flag that an operation is beginning: the completion channel will not finish
shutdown until a corrensponding grpc_cq_end_* call is made */
-void grpc_cq_begin_op(grpc_completion_queue *cc, grpc_call *call,
- grpc_completion_type type);
-
-/* grpc_cq_end_* functions pair with a grpc_cq_begin_op
-
- grpc_cq_end_* common arguments:
- cc - the completion channel to queue on
- tag - the user supplied operation tag
- on_finish - grpc_event_finish_func that is called during grpc_event_finish
- can be NULL to not get a callback
- user_data - user_data parameter to be passed to on_finish
-
- Other parameters match the data member of grpc_event */
+void grpc_cq_begin_op(grpc_completion_queue *cc, grpc_call *call);
/* Queue a GRPC_OP_COMPLETED operation */
void grpc_cq_end_op(grpc_completion_queue *cc, void *tag, grpc_call *call,
- grpc_event_finish_func on_finish, void *user_data,
- grpc_op_error error);
-
-void grpc_cq_end_server_shutdown(grpc_completion_queue *cc, void *tag);
+ int success);
/* disable polling for some tests */
void grpc_completion_queue_dont_poll_test_only(grpc_completion_queue *cc);
-void grpc_cq_dump_pending_ops(grpc_completion_queue *cc);
-
grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc);
void grpc_cq_hack_spin_pollset(grpc_completion_queue *cc);
diff --git a/src/core/surface/event_string.c b/src/core/surface/event_string.c
index 30bdff6..448bb11 100644
--- a/src/core/surface/event_string.c
+++ b/src/core/surface/event_string.c
@@ -40,23 +40,15 @@
static void addhdr(gpr_strvec *buf, grpc_event *ev) {
char *tmp;
- gpr_asprintf(&tmp, "tag:%p call:%p", ev->tag, (void *)ev->call);
+ gpr_asprintf(&tmp, "tag:%p", ev->tag);
gpr_strvec_add(buf, tmp);
}
-static const char *errstr(grpc_op_error err) {
- switch (err) {
- case GRPC_OP_OK:
- return "OK";
- case GRPC_OP_ERROR:
- return "ERROR";
- }
- return "UNKNOWN_UNKNOWN";
-}
+static const char *errstr(int success) { return success ? "OK" : "ERROR"; }
-static void adderr(gpr_strvec *buf, grpc_op_error err) {
+static void adderr(gpr_strvec *buf, int success) {
char *tmp;
- gpr_asprintf(&tmp, " err=%s", errstr(err));
+ gpr_asprintf(&tmp, " %s", errstr(success));
gpr_strvec_add(buf, tmp);
}
@@ -69,8 +61,8 @@
gpr_strvec_init(&buf);
switch (ev->type) {
- case GRPC_SERVER_SHUTDOWN:
- gpr_strvec_add(&buf, gpr_strdup("SERVER_SHUTDOWN"));
+ case GRPC_QUEUE_TIMEOUT:
+ gpr_strvec_add(&buf, gpr_strdup("QUEUE_TIMEOUT"));
break;
case GRPC_QUEUE_SHUTDOWN:
gpr_strvec_add(&buf, gpr_strdup("QUEUE_SHUTDOWN"));
@@ -78,11 +70,7 @@
case GRPC_OP_COMPLETE:
gpr_strvec_add(&buf, gpr_strdup("OP_COMPLETE: "));
addhdr(&buf, ev);
- adderr(&buf, ev->data.op_complete);
- break;
- case GRPC_COMPLETION_DO_NOT_USE:
- gpr_strvec_add(&buf, gpr_strdup("DO_NOT_USE (this is a bug)"));
- addhdr(&buf, ev);
+ adderr(&buf, ev->success);
break;
}
diff --git a/src/core/surface/server.c b/src/core/surface/server.c
index 96c1b7c..351ed5b 100644
--- a/src/core/surface/server.c
+++ b/src/core/surface/server.c
@@ -185,8 +185,6 @@
#define SERVER_FROM_CALL_ELEM(elem) \
(((channel_data *)(elem)->channel_data)->server)
-static void do_nothing(void *unused, grpc_op_error ignored) {}
-
static void begin_call(grpc_server *server, call_data *calld,
requested_call *rc);
static void fail_call(grpc_server *server, requested_call *rc);
@@ -535,8 +533,8 @@
if (chand->server->shutdown && chand->server->lists[ALL_CALLS] == NULL) {
for (i = 0; i < chand->server->num_shutdown_tags; i++) {
for (j = 0; j < chand->server->cq_count; j++) {
- grpc_cq_end_server_shutdown(chand->server->cqs[j],
- chand->server->shutdown_tags[i]);
+ grpc_cq_end_op(chand->server->cqs[j], chand->server->shutdown_tags[i],
+ NULL, 1);
}
}
}
@@ -809,7 +807,7 @@
gpr_mu_lock(&server->mu);
if (have_shutdown_tag) {
for (i = 0; i < server->cq_count; i++) {
- grpc_cq_begin_op(server->cqs[i], NULL, GRPC_SERVER_SHUTDOWN);
+ grpc_cq_begin_op(server->cqs[i], NULL);
}
server->shutdown_tags =
gpr_realloc(server->shutdown_tags,
@@ -859,7 +857,7 @@
if (server->lists[ALL_CALLS] == NULL) {
for (i = 0; i < server->num_shutdown_tags; i++) {
for (j = 0; j < server->cq_count; j++) {
- grpc_cq_end_server_shutdown(server->cqs[j], server->shutdown_tags[i]);
+ grpc_cq_end_op(server->cqs[j], server->shutdown_tags[i], NULL, 1);
}
}
}
@@ -1010,7 +1008,7 @@
grpc_completion_queue *cq_bound_to_call,
grpc_completion_queue *cq_for_notification, void *tag) {
requested_call rc;
- grpc_cq_begin_op(cq_for_notification, NULL, GRPC_OP_COMPLETE);
+ grpc_cq_begin_op(cq_for_notification, NULL);
rc.type = BATCH_CALL;
rc.tag = tag;
rc.cq_bound_to_call = cq_bound_to_call;
@@ -1028,7 +1026,7 @@
grpc_completion_queue *cq_for_notification, void *tag) {
requested_call rc;
registered_method *registered_method = rm;
- grpc_cq_begin_op(cq_for_notification, NULL, GRPC_OP_COMPLETE);
+ grpc_cq_begin_op(cq_for_notification, NULL);
rc.type = REGISTERED_CALL;
rc.tag = tag;
rc.cq_bound_to_call = cq_bound_to_call;
@@ -1041,10 +1039,9 @@
return queue_call_request(server, &rc);
}
-static void publish_registered_or_batch(grpc_call *call, grpc_op_error status,
+static void publish_registered_or_batch(grpc_call *call, int success,
void *tag);
-static void publish_was_not_set(grpc_call *call, grpc_op_error status,
- void *tag) {
+static void publish_was_not_set(grpc_call *call, int success, void *tag) {
abort();
}
@@ -1115,16 +1112,15 @@
rc->data.registered.initial_metadata->count = 0;
break;
}
- grpc_cq_end_op(rc->cq_for_notification, rc->tag, NULL, do_nothing, NULL,
- GRPC_OP_ERROR);
+ grpc_cq_end_op(rc->cq_for_notification, rc->tag, NULL, 0);
}
-static void publish_registered_or_batch(grpc_call *call, grpc_op_error status,
+static void publish_registered_or_batch(grpc_call *call, int success,
void *tag) {
grpc_call_element *elem =
grpc_call_stack_element(grpc_call_get_call_stack(call), 0);
call_data *calld = elem->call_data;
- grpc_cq_end_op(calld->cq_new, tag, call, do_nothing, NULL, status);
+ grpc_cq_end_op(calld->cq_new, tag, call, success);
}
const grpc_channel_args *grpc_server_get_channel_args(grpc_server *server) {
diff --git a/src/cpp/client/client_context.cc b/src/cpp/client/client_context.cc
index 70c9cb4..f38a694 100644
--- a/src/cpp/client/client_context.cc
+++ b/src/cpp/client/client_context.cc
@@ -49,15 +49,11 @@
grpc_call_destroy(call_);
}
if (cq_) {
- grpc_completion_queue_shutdown(cq_);
// Drain cq_.
- grpc_event* ev;
- grpc_completion_type t;
- do {
- ev = grpc_completion_queue_next(cq_, gpr_inf_future);
- t = ev->type;
- grpc_event_finish(ev);
- } while (t != GRPC_QUEUE_SHUTDOWN);
+ grpc_completion_queue_shutdown(cq_);
+ while (grpc_completion_queue_next(cq_, gpr_inf_future).type !=
+ GRPC_QUEUE_SHUTDOWN)
+ ;
grpc_completion_queue_destroy(cq_);
}
}
diff --git a/src/cpp/common/completion_queue.cc b/src/cpp/common/completion_queue.cc
index 2b9000e..b2dd1ac 100644
--- a/src/cpp/common/completion_queue.cc
+++ b/src/cpp/common/completion_queue.cc
@@ -48,53 +48,41 @@
void CompletionQueue::Shutdown() { grpc_completion_queue_shutdown(cq_); }
-// Helper class so we can declare a unique_ptr with grpc_event
-class EventDeleter {
- public:
- void operator()(grpc_event* ev) {
- if (ev) grpc_event_finish(ev);
- }
-};
-
CompletionQueue::NextStatus CompletionQueue::AsyncNextInternal(
void** tag, bool* ok, gpr_timespec deadline) {
- std::unique_ptr<grpc_event, EventDeleter> ev;
-
for (;;) {
- ev.reset(grpc_completion_queue_next(cq_, deadline));
- if (!ev) { /* got a NULL back because deadline passed */
- return TIMEOUT;
- }
- if (ev->type == GRPC_QUEUE_SHUTDOWN) {
- return SHUTDOWN;
- }
- auto cq_tag = static_cast<CompletionQueueTag*>(ev->tag);
- *ok = ev->data.op_complete == GRPC_OP_OK;
- *tag = cq_tag;
- if (cq_tag->FinalizeResult(tag, ok)) {
- return GOT_EVENT;
+ auto ev = grpc_completion_queue_next(cq_, deadline);
+ switch (ev.type) {
+ case GRPC_QUEUE_TIMEOUT:
+ return TIMEOUT;
+ case GRPC_QUEUE_SHUTDOWN:
+ return SHUTDOWN;
+ case GRPC_OP_COMPLETE:
+ auto cq_tag = static_cast<CompletionQueueTag*>(ev.tag);
+ *ok = ev.success != 0;
+ *tag = cq_tag;
+ if (cq_tag->FinalizeResult(tag, ok)) {
+ return GOT_EVENT;
+ }
+ break;
}
}
}
bool CompletionQueue::Pluck(CompletionQueueTag* tag) {
- std::unique_ptr<grpc_event, EventDeleter> ev;
-
- ev.reset(grpc_completion_queue_pluck(cq_, tag, gpr_inf_future));
- bool ok = ev->data.op_complete == GRPC_OP_OK;
+ auto ev = grpc_completion_queue_pluck(cq_, tag, gpr_inf_future);
+ bool ok = ev.success != 0;
void* ignored = tag;
GPR_ASSERT(tag->FinalizeResult(&ignored, &ok));
GPR_ASSERT(ignored == tag);
// Ignore mutations by FinalizeResult: Pluck returns the C API status
- return ev->data.op_complete == GRPC_OP_OK;
+ return ev.success != 0;
}
void CompletionQueue::TryPluck(CompletionQueueTag* tag) {
- std::unique_ptr<grpc_event, EventDeleter> ev;
-
- ev.reset(grpc_completion_queue_pluck(cq_, tag, gpr_time_0));
- if (!ev) return;
- bool ok = ev->data.op_complete == GRPC_OP_OK;
+ auto ev = grpc_completion_queue_pluck(cq_, tag, gpr_time_0);
+ if (ev.type == GRPC_QUEUE_TIMEOUT) return;
+ bool ok = ev.success != 0;
void* ignored = tag;
// the tag must be swallowed if using TryPluck
GPR_ASSERT(!tag->FinalizeResult(&ignored, &ok));
diff --git a/src/csharp/Grpc.Core.Tests/PInvokeTest.cs b/src/csharp/Grpc.Core.Tests/PInvokeTest.cs
index 3beffc3..26f8766 100644
--- a/src/csharp/Grpc.Core.Tests/PInvokeTest.cs
+++ b/src/csharp/Grpc.Core.Tests/PInvokeTest.cs
@@ -134,7 +134,7 @@
});
}
- private void Handler(GRPCOpError op, IntPtr ptr)
+ private void Handler(bool success, IntPtr ptr)
{
counter++;
}
diff --git a/src/csharp/Grpc.Core/Internal/AsyncCall.cs b/src/csharp/Grpc.Core/Internal/AsyncCall.cs
index 3532f73..9bb918d 100644
--- a/src/csharp/Grpc.Core/Internal/AsyncCall.cs
+++ b/src/csharp/Grpc.Core/Internal/AsyncCall.cs
@@ -274,7 +274,7 @@
/// <summary>
/// Handler for unary response completion.
/// </summary>
- private void HandleUnaryResponse(bool wasError, BatchContextSafeHandleNotOwned ctx)
+ private void HandleUnaryResponse(bool success, BatchContextSafeHandleNotOwned ctx)
{
lock (myLock)
{
@@ -284,7 +284,7 @@
ReleaseResourcesIfPossible();
}
- if (wasError)
+ if (!success)
{
unaryResponseTcs.SetException(new RpcException(new Status(StatusCode.Internal, "Internal error occured.")));
return;
@@ -307,7 +307,7 @@
/// <summary>
/// Handles receive status completion for calls with streaming response.
/// </summary>
- private void HandleFinished(bool wasError, BatchContextSafeHandleNotOwned ctx)
+ private void HandleFinished(bool success, BatchContextSafeHandleNotOwned ctx)
{
var status = ctx.GetReceivedStatus();
diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs
index fc5bee4..b4f4edb 100644
--- a/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs
+++ b/src/csharp/Grpc.Core/Internal/AsyncCallBase.cs
@@ -287,13 +287,12 @@
/// </summary>
protected CompletionCallbackDelegate CreateBatchCompletionCallback(Action<bool, BatchContextSafeHandleNotOwned> handler)
{
- return new CompletionCallbackDelegate((error, batchContextPtr) =>
+ return new CompletionCallbackDelegate((success, batchContextPtr) =>
{
try
{
var ctx = new BatchContextSafeHandleNotOwned(batchContextPtr);
- bool wasError = (error != GRPCOpError.GRPC_OP_OK);
- handler(wasError, ctx);
+ handler(success, ctx);
}
catch (Exception e)
{
@@ -305,7 +304,7 @@
/// <summary>
/// Handles send completion.
/// </summary>
- private void HandleSendFinished(bool wasError, BatchContextSafeHandleNotOwned ctx)
+ private void HandleSendFinished(bool success, BatchContextSafeHandleNotOwned ctx)
{
AsyncCompletionDelegate<object> origCompletionDelegate = null;
lock (myLock)
@@ -316,7 +315,7 @@
ReleaseResourcesIfPossible();
}
- if (wasError)
+ if (!success)
{
FireCompletion(origCompletionDelegate, null, new OperationFailedException("Send failed"));
}
@@ -329,7 +328,7 @@
/// <summary>
/// Handles halfclose completion.
/// </summary>
- private void HandleHalfclosed(bool wasError, BatchContextSafeHandleNotOwned ctx)
+ private void HandleHalfclosed(bool success, BatchContextSafeHandleNotOwned ctx)
{
AsyncCompletionDelegate<object> origCompletionDelegate = null;
lock (myLock)
@@ -341,7 +340,7 @@
ReleaseResourcesIfPossible();
}
- if (wasError)
+ if (!success)
{
FireCompletion(origCompletionDelegate, null, new OperationFailedException("Halfclose failed"));
}
@@ -354,7 +353,7 @@
/// <summary>
/// Handles streaming read completion.
/// </summary>
- private void HandleReadFinished(bool wasError, BatchContextSafeHandleNotOwned ctx)
+ private void HandleReadFinished(bool success, BatchContextSafeHandleNotOwned ctx)
{
var payload = ctx.GetReceivedMessage();
diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs
index 171d0c7..1f0335e 100644
--- a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs
+++ b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs
@@ -121,7 +121,7 @@
/// <summary>
/// Handles the server side close completion.
/// </summary>
- private void HandleFinishedServerside(bool wasError, BatchContextSafeHandleNotOwned ctx)
+ private void HandleFinishedServerside(bool success, BatchContextSafeHandleNotOwned ctx)
{
bool cancelled = ctx.GetReceivedCloseOnServerCancelled();
diff --git a/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs b/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs
index c97a3bc..491b841 100644
--- a/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs
+++ b/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs
@@ -37,7 +37,7 @@
namespace Grpc.Core.Internal
{
- internal delegate void CompletionCallbackDelegate(GRPCOpError error, IntPtr batchContextPtr);
+ internal delegate void CompletionCallbackDelegate(bool success, IntPtr batchContextPtr);
/// <summary>
/// grpc_call from <grpc/grpc.h>
diff --git a/src/csharp/Grpc.Core/Internal/Enums.cs b/src/csharp/Grpc.Core/Internal/Enums.cs
index 94a2fd1..2b4f6ca 100644
--- a/src/csharp/Grpc.Core/Internal/Enums.cs
+++ b/src/csharp/Grpc.Core/Internal/Enums.cs
@@ -70,45 +70,12 @@
internal enum GRPCCompletionType
{
/* Shutting down */
- GRPC_QUEUE_SHUTDOWN,
+ GRPC_QUEUE_SHUTDOWN,
+
+ /* No event before timeout */
+ GRPC_QUEUE_TIMEOUT,
/* operation completion */
- GRPC_OP_COMPLETE,
-
- /* A read has completed */
- GRPC_READ,
-
- /* A write has been accepted by flow control */
- GRPC_WRITE_ACCEPTED,
-
- /* writes_done or write_status has been accepted */
- GRPC_FINISH_ACCEPTED,
-
- /* The metadata array sent by server received at client */
- GRPC_CLIENT_METADATA_READ,
-
- /* An RPC has finished. The event contains status.
- * On the server this will be OK or Cancelled. */
- GRPC_FINISHED,
-
- /* A new RPC has arrived at the server */
- GRPC_SERVER_RPC_NEW,
-
- /* The server has finished shutting down */
- GRPC_SERVER_SHUTDOWN,
-
- /* must be last, forces users to include a default: case */
- GRPC_COMPLETION_DO_NOT_USE
- }
-
- /// <summary>
- /// grpc_op_error from grpc/grpc.h
- /// </summary>
- internal enum GRPCOpError
- {
- /* everything went ok */
- GRPC_OP_OK = 0,
- /* something failed, we don't know what */
- GRPC_OP_ERROR
+ GRPC_OP_COMPLETE
}
}
diff --git a/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs b/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs
index 8080643..731ea2b 100644
--- a/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs
+++ b/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs
@@ -40,7 +40,7 @@
namespace Grpc.Core.Internal
{
// TODO: we need to make sure that the delegates are not collected before invoked.
- internal delegate void ServerShutdownCallbackDelegate(IntPtr eventPtr);
+ //internal delegate void ServerShutdownCallbackDelegate(bool success);
/// <summary>
/// grpc_server from grpc/grpc.h
@@ -65,9 +65,8 @@
[DllImport("grpc_csharp_ext.dll")]
static extern void grpcsharp_server_shutdown(ServerSafeHandle server);
- // TODO: get rid of the old callback style
- [DllImport("grpc_csharp_ext.dll", EntryPoint = "grpcsharp_server_shutdown_and_notify")]
- static extern void grpcsharp_server_shutdown_and_notify_CALLBACK(ServerSafeHandle server, [MarshalAs(UnmanagedType.FunctionPtr)] ServerShutdownCallbackDelegate callback);
+ [DllImport("grpc_csharp_ext.dll")]
+ static extern void grpcsharp_server_shutdown_and_notify_callback(ServerSafeHandle server, [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback);
[DllImport("grpc_csharp_ext.dll")]
static extern void grpcsharp_server_destroy(IntPtr server);
@@ -101,9 +100,9 @@
grpcsharp_server_shutdown(this);
}
- public void ShutdownAndNotify(ServerShutdownCallbackDelegate callback)
+ public void ShutdownAndNotify(CompletionCallbackDelegate callback)
{
- grpcsharp_server_shutdown_and_notify_CALLBACK(this, callback);
+ grpcsharp_server_shutdown_and_notify_callback(this, callback);
}
public void RequestCall(CompletionQueueSafeHandle cq, CompletionCallbackDelegate callback)
diff --git a/src/csharp/Grpc.Core/Server.cs b/src/csharp/Grpc.Core/Server.cs
index 0df46bb..4a7abbb 100644
--- a/src/csharp/Grpc.Core/Server.cs
+++ b/src/csharp/Grpc.Core/Server.cs
@@ -54,7 +54,7 @@
// TODO(jtattermusch) : make sure the delegate doesn't get garbage collected while
// native callbacks are in the completion queue.
- readonly ServerShutdownCallbackDelegate serverShutdownHandler;
+ readonly CompletionCallbackDelegate serverShutdownHandler;
readonly CompletionCallbackDelegate newServerRpcHandler;
readonly ServerSafeHandle handle;
@@ -222,16 +222,13 @@
/// <summary>
/// Handles the native callback.
/// </summary>
- private void HandleNewServerRpc(GRPCOpError error, IntPtr batchContextPtr)
+ private void HandleNewServerRpc(bool success, IntPtr batchContextPtr)
{
try
{
var ctx = new BatchContextSafeHandleNotOwned(batchContextPtr);
- if (error != GRPCOpError.GRPC_OP_OK)
- {
- // TODO: handle error
- }
+ // TODO: handle error
CallSafeHandle call = ctx.GetServerRpcNewCall();
string method = ctx.GetServerRpcNewMethod();
@@ -253,8 +250,7 @@
/// <summary>
/// Handles native callback.
/// </summary>
- /// <param name="eventPtr"></param>
- private void HandleServerShutdown(IntPtr eventPtr)
+ private void HandleServerShutdown(bool success, IntPtr batchContextPtr)
{
try
{
diff --git a/src/csharp/ext/grpc_csharp_ext.c b/src/csharp/ext/grpc_csharp_ext.c
index 6ceface..cea23f0 100644
--- a/src/csharp/ext/grpc_csharp_ext.c
+++ b/src/csharp/ext/grpc_csharp_ext.c
@@ -63,8 +63,7 @@
return bb;
}
-typedef void(GPR_CALLTYPE *callback_funcptr)(grpc_op_error op_error,
- void *batch_context);
+typedef void(GPR_CALLTYPE *callback_funcptr)(gpr_int32 success, void *batch_context);
/*
* Helper to maintain lifetime of batch op inputs and store batch op outputs.
@@ -308,27 +307,18 @@
GPR_EXPORT grpc_completion_type GPR_CALLTYPE
grpcsharp_completion_queue_next_with_callback(grpc_completion_queue *cq) {
- grpc_event *ev;
+ grpc_event ev;
grpcsharp_batch_context *batch_context;
grpc_completion_type t;
- void(GPR_CALLTYPE * callback)(grpc_event *);
ev = grpc_completion_queue_next(cq, gpr_inf_future);
- t = ev->type;
- if (t == GRPC_OP_COMPLETE && ev->tag) {
+ t = ev.type;
+ if (t == GRPC_OP_COMPLETE && ev.tag) {
/* NEW API handler */
- batch_context = (grpcsharp_batch_context *)ev->tag;
- batch_context->callback(ev->data.op_complete, batch_context);
+ batch_context = (grpcsharp_batch_context *)ev.tag;
+ batch_context->callback((gpr_int32) ev.success, batch_context);
grpcsharp_batch_context_destroy(batch_context);
- } else if (ev->tag) {
- /* call the callback in ev->tag */
- /* C forbids to cast object pointers to function pointers, so
- * we cast to intptr first.
- */
- callback = (void(GPR_CALLTYPE *)(grpc_event *))(gpr_intptr)ev->tag;
- (*callback)(ev);
}
- grpc_event_finish(ev);
/* return completion type to allow some handling for events that have no
* tag - such as GRPC_QUEUE_SHUTDOWN
@@ -692,8 +682,11 @@
}
GPR_EXPORT void GPR_CALLTYPE
-grpcsharp_server_shutdown_and_notify(grpc_server *server, void *tag) {
- grpc_server_shutdown_and_notify(server, tag);
+grpcsharp_server_shutdown_and_notify_callback(grpc_server *server,
+ callback_funcptr callback) {
+ grpcsharp_batch_context *ctx = grpcsharp_batch_context_create();
+ ctx->callback = callback;
+ grpc_server_shutdown_and_notify(server, ctx);
}
GPR_EXPORT void GPR_CALLTYPE grpcsharp_server_destroy(grpc_server *server) {
@@ -797,7 +790,7 @@
/* For testing */
GPR_EXPORT void GPR_CALLTYPE
grpcsharp_test_callback(callback_funcptr callback) {
- callback(GRPC_OP_OK, NULL);
+ callback(1, NULL);
}
/* For testing */
diff --git a/src/node/ext/completion_queue_async_worker.cc b/src/node/ext/completion_queue_async_worker.cc
index 4e57121..4be208c 100644
--- a/src/node/ext/completion_queue_async_worker.cc
+++ b/src/node/ext/completion_queue_async_worker.cc
@@ -63,7 +63,7 @@
void CompletionQueueAsyncWorker::Execute() {
result = grpc_completion_queue_next(queue, gpr_inf_future);
- if (result->data.op_complete != GRPC_OP_OK) {
+ if (!result.success) {
SetErrorMessage("The batch encountered an error");
}
}
@@ -96,25 +96,21 @@
} else {
current_threads -= 1;
}
- NanCallback *callback = GetTagCallback(result->tag);
- Handle<Value> argv[] = {NanNull(), GetTagNodeValue(result->tag)};
+ NanCallback *callback = GetTagCallback(result.tag);
+ Handle<Value> argv[] = {NanNull(), GetTagNodeValue(result.tag)};
callback->Call(2, argv);
- DestroyTag(result->tag);
- grpc_event_finish(result);
- result = NULL;
+ DestroyTag(result.tag);
}
void CompletionQueueAsyncWorker::HandleErrorCallback() {
NanScope();
- NanCallback *callback = GetTagCallback(result->tag);
+ NanCallback *callback = GetTagCallback(result.tag);
Handle<Value> argv[] = {NanError(ErrorMessage())};
callback->Call(1, argv);
- DestroyTag(result->tag);
- grpc_event_finish(result);
- result = NULL;
+ DestroyTag(result.tag);
}
} // namespace node
diff --git a/src/node/ext/completion_queue_async_worker.h b/src/node/ext/completion_queue_async_worker.h
index 5d52bbb..27fedf2 100644
--- a/src/node/ext/completion_queue_async_worker.h
+++ b/src/node/ext/completion_queue_async_worker.h
@@ -70,7 +70,7 @@
void HandleErrorCallback();
private:
- grpc_event *result;
+ grpc_event result;
static grpc_completion_queue *queue;
diff --git a/src/objective-c/.gitignore b/src/objective-c/.gitignore
index ec839c0..1511066 100644
--- a/src/objective-c/.gitignore
+++ b/src/objective-c/.gitignore
@@ -16,3 +16,4 @@
*.hmap
*.ipa
*.xcuserstate
+*.DS_Store
\ No newline at end of file
diff --git a/src/objective-c/GRPCClient/private/GRPCCompletionQueue.h b/src/objective-c/GRPCClient/private/GRPCCompletionQueue.h
index 25ca9bd..603bf01 100644
--- a/src/objective-c/GRPCClient/private/GRPCCompletionQueue.h
+++ b/src/objective-c/GRPCClient/private/GRPCCompletionQueue.h
@@ -34,7 +34,7 @@
#import <Foundation/Foundation.h>
#include <grpc/grpc.h>
-typedef void(^GRPCQueueCompletionHandler)(grpc_op_error error);
+typedef void(^GRPCQueueCompletionHandler)(bool success);
// This class lets one more easily use grpc_completion_queue. To use it, pass
// the value of the unmanagedQueue property of an instance of this class to
diff --git a/src/objective-c/GRPCClient/private/GRPCCompletionQueue.m b/src/objective-c/GRPCClient/private/GRPCCompletionQueue.m
index a0a1016..40aade4 100644
--- a/src/objective-c/GRPCClient/private/GRPCCompletionQueue.m
+++ b/src/objective-c/GRPCClient/private/GRPCCompletionQueue.m
@@ -65,20 +65,17 @@
dispatch_async(gDefaultConcurrentQueue, ^{
while (YES) {
// The following call blocks until an event is available.
- grpc_event *event = grpc_completion_queue_next(unmanagedQueue, gpr_inf_future);
+ grpc_event event = grpc_completion_queue_next(unmanagedQueue, gpr_inf_future);
GRPCQueueCompletionHandler handler;
- switch (event->type) {
+ switch (event.type) {
case GRPC_OP_COMPLETE:
- handler = (__bridge_transfer GRPCQueueCompletionHandler)event->tag;
- handler(event->data.op_complete);
- grpc_event_finish(event);
+ handler = (__bridge_transfer GRPCQueueCompletionHandler)event.tag;
+ handler(event.success);
break;
case GRPC_QUEUE_SHUTDOWN:
- grpc_event_finish(event);
grpc_completion_queue_destroy(unmanagedQueue);
return;
default:
- grpc_event_finish(event);
[NSException raise:@"Unrecognized completion type" format:@""];
}
};
diff --git a/src/objective-c/GRPCClient/private/GRPCWrappedCall.m b/src/objective-c/GRPCClient/private/GRPCWrappedCall.m
index 41ec1a1..9bc4693 100644
--- a/src/objective-c/GRPCClient/private/GRPCWrappedCall.m
+++ b/src/objective-c/GRPCClient/private/GRPCWrappedCall.m
@@ -296,8 +296,8 @@
[op getOp:&ops_array[i++]];
}
grpc_call_error error = grpc_call_start_batch(_call, ops_array, nops,
- (__bridge_retained void *)(^(grpc_op_error error){
- if (error != GRPC_OP_OK) {
+ (__bridge_retained void *)(^(bool success){
+ if (!success) {
if (errorHandler) {
errorHandler();
} else {
diff --git a/src/php/ext/grpc/call.c b/src/php/ext/grpc/call.c
index b1525e9..9cc4fd7 100644
--- a/src/php/ext/grpc/call.c
+++ b/src/php/ext/grpc/call.c
@@ -61,17 +61,12 @@
/* Frees and destroys an instance of wrapped_grpc_call */
void free_wrapped_grpc_call(void *object TSRMLS_DC) {
wrapped_grpc_call *call = (wrapped_grpc_call *)object;
- grpc_event *event;
if (call->owned && call->wrapped != NULL) {
if (call->queue != NULL) {
grpc_completion_queue_shutdown(call->queue);
- event = grpc_completion_queue_next(call->queue, gpr_inf_future);
- while (event != NULL) {
- if (event->type == GRPC_QUEUE_SHUTDOWN) {
- break;
- }
- event = grpc_completion_queue_next(call->queue, gpr_inf_future);
- }
+ while (grpc_completion_queue_next(call->queue, gpr_inf_future).type !=
+ GRPC_QUEUE_SHUTDOWN)
+ ;
grpc_completion_queue_destroy(call->queue);
}
grpc_call_destroy(call->wrapped);
@@ -287,7 +282,7 @@
grpc_byte_buffer *message;
int cancelled;
grpc_call_error error;
- grpc_event *event;
+ grpc_event event;
zval *result;
char *message_str;
size_t message_len;
@@ -422,7 +417,7 @@
}
event = grpc_completion_queue_pluck(call->queue, call->wrapped,
gpr_inf_future);
- if (event->data.op_complete != GRPC_OP_OK) {
+ if (!event.success) {
zend_throw_exception(spl_ce_LogicException,
"The batch failed for some reason",
1 TSRMLS_CC);
diff --git a/src/php/ext/grpc/server.c b/src/php/ext/grpc/server.c
index 18d34ab..c2e00b1 100644
--- a/src/php/ext/grpc/server.c
+++ b/src/php/ext/grpc/server.c
@@ -61,16 +61,11 @@
/* Frees and destroys an instance of wrapped_grpc_server */
void free_wrapped_grpc_server(void *object TSRMLS_DC) {
wrapped_grpc_server *server = (wrapped_grpc_server *)object;
- grpc_event *event;
if (server->queue != NULL) {
grpc_completion_queue_shutdown(server->queue);
- event = grpc_completion_queue_next(server->queue, gpr_inf_future);
- while (event != NULL) {
- if (event->type == GRPC_QUEUE_SHUTDOWN) {
- break;
- }
- event = grpc_completion_queue_next(server->queue, gpr_inf_future);
- }
+ while (grpc_completion_queue_next(server->queue, gpr_inf_future).type !=
+ GRPC_QUEUE_SHUTDOWN)
+ ;
grpc_completion_queue_destroy(server->queue);
}
if (server->wrapped != NULL) {
@@ -142,7 +137,7 @@
grpc_call_details details;
grpc_metadata_array metadata;
zval *result;
- grpc_event *event;
+ grpc_event event;
MAKE_STD_ZVAL(result);
object_init(result);
grpc_call_details_init(&details);
@@ -156,7 +151,7 @@
goto cleanup;
}
event = grpc_completion_queue_pluck(server->queue, NULL, gpr_inf_future);
- if (event->data.op_complete != GRPC_OP_OK) {
+ if (!event.success) {
zend_throw_exception(spl_ce_LogicException,
"Failed to request a call for some reason",
1 TSRMLS_CC);
diff --git a/src/python/src/grpc/_adapter/_completion_queue.c b/src/python/src/grpc/_adapter/_completion_queue.c
index f616faf..97828e6 100644
--- a/src/python/src/grpc/_adapter/_completion_queue.c
+++ b/src/python/src/grpc/_adapter/_completion_queue.c
@@ -349,7 +349,7 @@
PyObject *deadline;
double double_deadline;
gpr_timespec deadline_timespec;
- grpc_event *c_event;
+ grpc_event c_event;
PyObject *event_args;
PyObject *event;
@@ -378,15 +378,14 @@
grpc_completion_queue_next(self->c_completion_queue, deadline_timespec);
Py_END_ALLOW_THREADS;
- if (c_event == NULL) {
- Py_RETURN_NONE;
- }
+ tag = (pygrpc_tag *)c_event.tag;
- tag = (pygrpc_tag *)c_event->tag;
-
- switch (c_event->type) {
+ switch (c_event.type) {
+ case GRPC_QUEUE_TIMEOUT:
+ Py_RETURN_NONE;
+ break;
case GRPC_QUEUE_SHUTDOWN:
- event_args = pygrpc_stop_event_args(c_event);
+ event_args = pygrpc_stop_event_args(&c_event);
break;
case GRPC_OP_COMPLETE: {
if (!tag) {
@@ -398,28 +397,27 @@
if (tag) {
pygrpc_tag_destroy(tag);
}
- grpc_event_finish(c_event);
return pygrpc_completion_queue_get(self, args);
case PYGRPC_WRITE_ACCEPTED:
- event_args = pygrpc_write_event_args(c_event);
+ event_args = pygrpc_write_event_args(&c_event);
break;
case PYGRPC_FINISH_ACCEPTED:
- event_args = pygrpc_complete_event_args(c_event);
+ event_args = pygrpc_complete_event_args(&c_event);
break;
case PYGRPC_SERVER_RPC_NEW:
- event_args = pygrpc_service_event_args(c_event);
+ event_args = pygrpc_service_event_args(&c_event);
break;
case PYGRPC_READ:
- event_args = pygrpc_read_event_args(c_event);
+ event_args = pygrpc_read_event_args(&c_event);
break;
case PYGRPC_CLIENT_METADATA_READ:
- event_args = pygrpc_metadata_event_args(c_event);
+ event_args = pygrpc_metadata_event_args(&c_event);
break;
case PYGRPC_FINISHED_CLIENT:
- event_args = pygrpc_finished_client_event_args(c_event);
+ event_args = pygrpc_finished_client_event_args(&c_event);
break;
case PYGRPC_FINISHED_SERVER:
- event_args = pygrpc_finished_server_event_args(c_event);
+ event_args = pygrpc_finished_server_event_args(&c_event);
break;
default:
PyErr_SetString(PyExc_Exception, "Unrecognized op event type!");
@@ -442,7 +440,6 @@
if (tag) {
pygrpc_tag_destroy(tag);
}
- grpc_event_finish(c_event);
return event;
}
diff --git a/src/ruby/ext/grpc/rb_call.c b/src/ruby/ext/grpc/rb_call.c
index e76bb93..c46af25 100644
--- a/src/ruby/ext/grpc/rb_call.c
+++ b/src/ruby/ext/grpc/rb_call.c
@@ -581,7 +581,7 @@
VALUE timeout, VALUE ops_hash) {
run_batch_stack st;
grpc_call *call = NULL;
- grpc_event *ev = NULL;
+ grpc_event ev;
grpc_call_error err;
VALUE result = Qnil;
TypedData_Get_Struct(self, grpc_call, &grpc_call_data_type, call);
@@ -605,15 +605,14 @@
return Qnil;
}
ev = grpc_rb_completion_queue_pluck_event(cqueue, tag, timeout);
- if (ev == NULL) {
+ if (ev.type == GRPC_QUEUE_TIMEOUT) {
grpc_run_batch_stack_cleanup(&st);
rb_raise(grpc_rb_eOutOfTime, "grpc_call_start_batch timed out");
return Qnil;
}
- if (ev->data.op_complete != GRPC_OP_OK) {
+ if (!ev.success) {
grpc_run_batch_stack_cleanup(&st);
- rb_raise(grpc_rb_eCallError, "start_batch completion failed, (code=%d)",
- ev->data.op_complete);
+ rb_raise(grpc_rb_eCallError, "start_batch completion failed");
return Qnil;
}
diff --git a/src/ruby/ext/grpc/rb_completion_queue.c b/src/ruby/ext/grpc/rb_completion_queue.c
index 3cf6c31..d83a20e 100644
--- a/src/ruby/ext/grpc/rb_completion_queue.c
+++ b/src/ruby/ext/grpc/rb_completion_queue.c
@@ -47,7 +47,7 @@
/* Used to allow grpc_completion_queue_next call to release the GIL */
typedef struct next_call_stack {
grpc_completion_queue *cq;
- grpc_event *event;
+ grpc_event event;
gpr_timespec timeout;
void *tag;
} next_call_stack;
@@ -80,7 +80,7 @@
grpc_completion_queue_shutdown(cq);
next_call.cq = cq;
- next_call.event = NULL;
+ next_call.event.type = GRPC_QUEUE_TIMEOUT;
/* TODO: the timeout should be a module level constant that defaults
* to gpr_inf_future.
*
@@ -95,16 +95,11 @@
do {
rb_thread_call_without_gvl(grpc_rb_completion_queue_next_no_gil,
(void *)&next_call, NULL, NULL);
- if (next_call.event == NULL) {
- break;
- }
- type = next_call.event->type;
+ type = next_call.event.type;
if (type != GRPC_QUEUE_SHUTDOWN) {
++drained;
rb_warning("completion queue shutdown: %d undrained events", drained);
}
- grpc_event_finish(next_call.event);
- next_call.event = NULL;
} while (type != GRPC_QUEUE_SHUTDOWN);
}
@@ -138,49 +133,19 @@
return TypedData_Wrap_Struct(cls, &grpc_rb_completion_queue_data_type, cq);
}
-/* Blocks until the next event is available, and returns the event. */
-static VALUE grpc_rb_completion_queue_next(VALUE self, VALUE timeout) {
- next_call_stack next_call;
- MEMZERO(&next_call, next_call_stack, 1);
- TypedData_Get_Struct(self, grpc_completion_queue,
- &grpc_rb_completion_queue_data_type, next_call.cq);
- next_call.timeout = grpc_rb_time_timeval(timeout, /* absolute time*/ 0);
- next_call.event = NULL;
- rb_thread_call_without_gvl(grpc_rb_completion_queue_next_no_gil,
- (void *)&next_call, NULL, NULL);
- if (next_call.event == NULL) {
- return Qnil;
- }
- return grpc_rb_new_event(next_call.event);
-}
-
/* Blocks until the next event for given tag is available, and returns the
* event. */
-VALUE grpc_rb_completion_queue_pluck(VALUE self, VALUE tag,
- VALUE timeout) {
- grpc_event *ev = grpc_rb_completion_queue_pluck_event(self, tag, timeout);
- if (ev == NULL) {
- return Qnil;
- }
- return grpc_rb_new_event(ev);
-}
-
-/* Blocks until the next event for given tag is available, and returns the
- * event. */
-grpc_event* grpc_rb_completion_queue_pluck_event(VALUE self, VALUE tag,
- VALUE timeout) {
+grpc_event grpc_rb_completion_queue_pluck_event(VALUE self, VALUE tag,
+ VALUE timeout) {
next_call_stack next_call;
MEMZERO(&next_call, next_call_stack, 1);
TypedData_Get_Struct(self, grpc_completion_queue,
&grpc_rb_completion_queue_data_type, next_call.cq);
next_call.timeout = grpc_rb_time_timeval(timeout, /* absolute time*/ 0);
next_call.tag = ROBJECT(tag);
- next_call.event = NULL;
+ next_call.event.type = GRPC_QUEUE_TIMEOUT;
rb_thread_call_without_gvl(grpc_rb_completion_queue_pluck_no_gil,
(void *)&next_call, NULL, NULL);
- if (next_call.event == NULL) {
- return NULL;
- }
return next_call.event;
}
@@ -193,14 +158,6 @@
this func, so no separate initialization step is necessary. */
rb_define_alloc_func(grpc_rb_cCompletionQueue,
grpc_rb_completion_queue_alloc);
-
- /* Add the next method that waits for the next event. */
- rb_define_method(grpc_rb_cCompletionQueue, "next",
- grpc_rb_completion_queue_next, 1);
-
- /* Add the pluck method that waits for the next event of given tag */
- rb_define_method(grpc_rb_cCompletionQueue, "pluck",
- grpc_rb_completion_queue_pluck, 2);
}
/* Gets the wrapped completion queue from the ruby wrapper */
diff --git a/src/ruby/ext/grpc/rb_completion_queue.h b/src/ruby/ext/grpc/rb_completion_queue.h
index 4d0f49a..e4d04b1 100644
--- a/src/ruby/ext/grpc/rb_completion_queue.h
+++ b/src/ruby/ext/grpc/rb_completion_queue.h
@@ -45,8 +45,8 @@
*
* This avoids having code that holds the GIL repeated at multiple sites.
*/
-grpc_event* grpc_rb_completion_queue_pluck_event(VALUE cqueue, VALUE tag,
- VALUE timeout);
+grpc_event grpc_rb_completion_queue_pluck_event(VALUE cqueue, VALUE tag,
+ VALUE timeout);
/* Initializes the CompletionQueue class. */
void Init_grpc_completion_queue();
diff --git a/src/ruby/ext/grpc/rb_server.c b/src/ruby/ext/grpc/rb_server.c
index 85ada19..0651c36 100644
--- a/src/ruby/ext/grpc/rb_server.c
+++ b/src/ruby/ext/grpc/rb_server.c
@@ -204,7 +204,7 @@
VALUE tag_new, VALUE timeout) {
grpc_rb_server *s = NULL;
grpc_call *call = NULL;
- grpc_event *ev = NULL;
+ grpc_event ev;
grpc_call_error err;
request_call_stack st;
VALUE result;
@@ -229,15 +229,13 @@
return Qnil;
}
ev = grpc_rb_completion_queue_pluck_event(cqueue, tag_new, timeout);
- if (ev == NULL) {
+ if (ev.type == GRPC_QUEUE_TIMEOUT) {
grpc_request_call_stack_cleanup(&st);
return Qnil;
}
- if (ev->data.op_complete != GRPC_OP_OK) {
+ if (!ev.success) {
grpc_request_call_stack_cleanup(&st);
- grpc_event_finish(ev);
- rb_raise(grpc_rb_eCallError, "request_call completion failed: (code=%d)",
- ev->data.op_complete);
+ rb_raise(grpc_rb_eCallError, "request_call completion failed");
return Qnil;
}
@@ -251,7 +249,6 @@
grpc_rb_md_ary_to_h(&st.md_ary),
grpc_rb_wrap_call(call),
NULL);
- grpc_event_finish(ev);
grpc_request_call_stack_cleanup(&st);
return result;
}
diff --git a/src/ruby/spec/completion_queue_spec.rb b/src/ruby/spec/completion_queue_spec.rb
index 11d4e99..886a7f2 100644
--- a/src/ruby/spec/completion_queue_spec.rb
+++ b/src/ruby/spec/completion_queue_spec.rb
@@ -39,36 +39,4 @@
expect { GRPC::Core::CompletionQueue.new }.not_to raise_error
end
end
-
- describe '#next' do
- it 'can be called without failing' do
- expect { @cq.next(3) }.not_to raise_error
- end
-
- it 'can be called with a time constant' do
- # don't use INFINITE_FUTURE, as are no events and this blocks.
- #
- # don't use INFINITE_PAST, as this fails on docker, and does not need to
- # be tested, as its not used anywhere in the ruby implementation
- a_time = GRPC::Core::TimeConsts::ZERO
- expect { @cq.next(a_time) }.not_to raise_error
- end
- end
-
- describe '#pluck' do
- it 'can be called without failing' do
- tag = Object.new
- expect { @cq.pluck(tag, 3) }.not_to raise_error
- end
-
- it 'can be called with a time constant' do
- # don't use INFINITE_FUTURE, as there no events and this blocks.
- #
- # don't use INFINITE_PAST, as this fails on docker, and does not need to
- # be tested, as its not used anywhere in the ruby implementation
- tag = Object.new
- a_time = GRPC::Core::TimeConsts::ZERO
- expect { @cq.pluck(tag, a_time) }.not_to raise_error
- end
- end
end
diff --git a/test/core/end2end/cq_verifier.c b/test/core/end2end/cq_verifier.c
index f291e73..6648025 100644
--- a/test/core/end2end/cq_verifier.c
+++ b/test/core/end2end/cq_verifier.c
@@ -45,6 +45,8 @@
#include <grpc/support/time.h>
#include <grpc/support/useful.h>
+#define ROOT_EXPECTATION 1000
+
/* a set of metadata we expect to find on an event */
typedef struct metadata {
size_t count;
@@ -60,9 +62,7 @@
struct expectation *prev;
grpc_completion_type type;
void *tag;
- union {
- grpc_op_error op_complete;
- } data;
+ int success;
} expectation;
/* the verifier itself */
@@ -75,7 +75,7 @@
cq_verifier *cq_verifier_create(grpc_completion_queue *cq) {
cq_verifier *v = gpr_malloc(sizeof(cq_verifier));
- v->expect.type = GRPC_COMPLETION_DO_NOT_USE;
+ v->expect.type = ROOT_EXPECTATION;
v->expect.tag = NULL;
v->expect.next = &v->expect;
v->expect.prev = &v->expect;
@@ -149,11 +149,9 @@
abort();
break;
case GRPC_OP_COMPLETE:
- GPR_ASSERT(e->data.op_complete == ev->data.op_complete);
+ GPR_ASSERT(e->success == ev->success);
break;
- case GRPC_SERVER_SHUTDOWN:
- break;
- case GRPC_COMPLETION_DO_NOT_USE:
+ case GRPC_QUEUE_TIMEOUT:
gpr_log(GPR_ERROR, "not implemented");
abort();
break;
@@ -165,13 +163,10 @@
switch (e->type) {
case GRPC_OP_COMPLETE:
- gpr_asprintf(&tmp, "GRPC_OP_COMPLETE result=%d", e->data.op_complete);
+ gpr_asprintf(&tmp, "GRPC_OP_COMPLETE result=%d", e->success);
gpr_strvec_add(buf, tmp);
break;
- case GRPC_SERVER_SHUTDOWN:
- gpr_strvec_add(buf, gpr_strdup("GRPC_SERVER_SHUTDOWN"));
- break;
- case GRPC_COMPLETION_DO_NOT_USE:
+ case GRPC_QUEUE_TIMEOUT:
case GRPC_QUEUE_SHUTDOWN:
gpr_log(GPR_ERROR, "not implemented");
abort();
@@ -203,7 +198,7 @@
void cq_verify(cq_verifier *v) {
gpr_timespec deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10);
- grpc_event *ev;
+ grpc_event ev;
expectation *e;
char *s;
gpr_strvec have_tags;
@@ -212,15 +207,16 @@
while (v->expect.next != &v->expect) {
ev = grpc_completion_queue_next(v->cq, deadline);
- if (!ev) {
+ if (ev.type == GRPC_QUEUE_TIMEOUT) {
fail_no_event_received(v);
+ break;
}
for (e = v->expect.next; e != &v->expect; e = e->next) {
gpr_asprintf(&s, " %p", e->tag);
gpr_strvec_add(&have_tags, s);
- if (e->tag == ev->tag) {
- verify_matches(e, ev);
+ if (e->tag == ev.tag) {
+ verify_matches(e, &ev);
e->next->prev = e->prev;
e->prev->next = e->next;
gpr_free(e);
@@ -228,7 +224,7 @@
}
}
if (e == &v->expect) {
- s = grpc_event_string(ev);
+ s = grpc_event_string(&ev);
gpr_log(GPR_ERROR, "event not found: %s", s);
gpr_free(s);
s = gpr_strvec_flatten(&have_tags, NULL);
@@ -237,8 +233,6 @@
gpr_strvec_destroy(&have_tags);
abort();
}
-
- grpc_event_finish(ev);
}
gpr_strvec_destroy(&have_tags);
@@ -246,13 +240,13 @@
void cq_verify_empty(cq_verifier *v) {
gpr_timespec deadline = gpr_time_add(gpr_now(), gpr_time_from_seconds(1));
- grpc_event *ev;
+ grpc_event ev;
GPR_ASSERT(v->expect.next == &v->expect && "expectation queue must be empty");
ev = grpc_completion_queue_next(v->cq, deadline);
- if (ev != NULL) {
- char *s = grpc_event_string(ev);
+ if (ev.type != GRPC_QUEUE_TIMEOUT) {
+ char *s = grpc_event_string(&ev);
gpr_log(GPR_ERROR, "unexpected event (expected nothing): %s", s);
gpr_free(s);
abort();
@@ -269,10 +263,6 @@
return e;
}
-void cq_expect_completion(cq_verifier *v, void *tag, grpc_op_error result) {
- add(v, GRPC_OP_COMPLETE, tag)->data.op_complete = result;
-}
-
-void cq_expect_server_shutdown(cq_verifier *v, void *tag) {
- add(v, GRPC_SERVER_SHUTDOWN, tag);
+void cq_expect_completion(cq_verifier *v, void *tag, int success) {
+ add(v, GRPC_OP_COMPLETE, tag)->success = success;
}
diff --git a/test/core/end2end/cq_verifier.h b/test/core/end2end/cq_verifier.h
index bae3c6c..1ecd4db 100644
--- a/test/core/end2end/cq_verifier.h
+++ b/test/core/end2end/cq_verifier.h
@@ -57,8 +57,7 @@
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_completion(cq_verifier *v, void *tag, grpc_op_error result);
-void cq_expect_server_shutdown(cq_verifier *v, void *tag);
+void cq_expect_completion(cq_verifier *v, void *tag, int success);
int byte_buffer_eq_string(grpc_byte_buffer *byte_buffer, const char *string);
int contains_metadata(grpc_metadata_array *array, const char *key, const char *value);
diff --git a/test/core/end2end/dualstack_socket_test.c b/test/core/end2end/dualstack_socket_test.c
index d418ea8..b54b8ec 100644
--- a/test/core/end2end/dualstack_socket_test.c
+++ b/test/core/end2end/dualstack_socket_test.c
@@ -50,15 +50,10 @@
}
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
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);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
void test_connect(const char *server_host, const char *client_host, int port,
@@ -156,7 +151,7 @@
grpc_server_request_call(server, &s, &call_details,
&request_metadata_recv, cq,
cq, tag(101)));
- cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(101), 1);
cq_verify(cqv);
op = ops;
@@ -174,8 +169,8 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
- cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
@@ -187,7 +182,7 @@
grpc_call_destroy(s);
} else {
/* Check for a failed connection. */
- cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_DEADLINE_EXCEEDED);
diff --git a/test/core/end2end/no_server_test.c b/test/core/end2end/no_server_test.c
index b292620..bba9cd1 100644
--- a/test/core/end2end/no_server_test.c
+++ b/test/core/end2end/no_server_test.c
@@ -45,8 +45,6 @@
gpr_timespec deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(2);
grpc_completion_queue *cq;
cq_verifier *cqv;
- grpc_event *ev;
- int done;
grpc_op ops[6];
grpc_op *op;
grpc_metadata_array trailing_metadata_recv;
@@ -79,17 +77,15 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_batch(call, ops, op - ops, tag(1)));
/* verify that all tags get completed */
- cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_DEADLINE_EXCEEDED);
grpc_completion_queue_shutdown(cq);
- for (done = 0; !done;) {
- ev = grpc_completion_queue_next(cq, gpr_inf_future);
- done = ev->type == GRPC_QUEUE_SHUTDOWN;
- grpc_event_finish(ev);
- }
+ while (grpc_completion_queue_next(cq, gpr_inf_future).type !=
+ GRPC_QUEUE_SHUTDOWN)
+ ;
grpc_completion_queue_destroy(cq);
grpc_call_destroy(call);
grpc_channel_destroy(chan);
diff --git a/test/core/end2end/tests/bad_hostname.c b/test/core/end2end/tests/bad_hostname.c
index d9971a6..726bada 100644
--- a/test/core/end2end/tests/bad_hostname.c
+++ b/test/core/end2end/tests/bad_hostname.c
@@ -68,14 +68,10 @@
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
- GPR_ASSERT(ev);
- type = ev->type;
- grpc_event_finish(ev);
- } while (type != GRPC_QUEUE_SHUTDOWN);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -140,7 +136,7 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1)));
- cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_UNAUTHENTICATED);
diff --git a/test/core/end2end/tests/cancel_after_accept.c b/test/core/end2end/tests/cancel_after_accept.c
index 8ae53b7..db6d089 100644
--- a/test/core/end2end/tests/cancel_after_accept.c
+++ b/test/core/end2end/tests/cancel_after_accept.c
@@ -67,14 +67,10 @@
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
- GPR_ASSERT(ev);
- type = ev->type;
- grpc_event_finish(ev);
- } while (type != GRPC_QUEUE_SHUTDOWN);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -160,7 +156,7 @@
grpc_server_request_call(f.server, &s, &call_details,
&request_metadata_recv, f.cq,
f.cq, tag(2)));
- cq_expect_completion(cqv, tag(2), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(2), 1);
cq_verify(cqv);
op = ops;
@@ -180,8 +176,8 @@
GPR_ASSERT(GRPC_CALL_OK == mode.initiate_cancel(c));
- cq_expect_completion(cqv, tag(3), GRPC_OP_OK);
- cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(3), 1);
+ cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
GPR_ASSERT(status == mode.expect_status);
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 4b7fdd6..0162df7 100644
--- a/test/core/end2end/tests/cancel_after_accept_and_writes_closed.c
+++ b/test/core/end2end/tests/cancel_after_accept_and_writes_closed.c
@@ -67,14 +67,10 @@
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
- GPR_ASSERT(ev);
- type = ev->type;
- grpc_event_finish(ev);
- } while (type != GRPC_QUEUE_SHUTDOWN);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -162,7 +158,7 @@
grpc_server_request_call(f.server, &s, &call_details,
&request_metadata_recv, f.cq,
f.cq, tag(2)));
- cq_expect_completion(cqv, tag(2), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(2), 1);
cq_verify(cqv);
op = ops;
@@ -182,8 +178,8 @@
GPR_ASSERT(GRPC_CALL_OK == mode.initiate_cancel(c));
- cq_expect_completion(cqv, tag(3), GRPC_OP_OK);
- cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(3), 1);
+ cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
GPR_ASSERT(status == mode.expect_status);
diff --git a/test/core/end2end/tests/cancel_after_invoke.c b/test/core/end2end/tests/cancel_after_invoke.c
index 7ee4350..cf12269 100644
--- a/test/core/end2end/tests/cancel_after_invoke.c
+++ b/test/core/end2end/tests/cancel_after_invoke.c
@@ -68,14 +68,10 @@
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
- GPR_ASSERT(ev);
- type = ev->type;
- grpc_event_finish(ev);
- } while (type != GRPC_QUEUE_SHUTDOWN);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -156,7 +152,7 @@
GPR_ASSERT(GRPC_CALL_OK == mode.initiate_cancel(c));
- cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
GPR_ASSERT(status == mode.expect_status);
diff --git a/test/core/end2end/tests/cancel_before_invoke.c b/test/core/end2end/tests/cancel_before_invoke.c
index 8a53e18..19d3bad 100644
--- a/test/core/end2end/tests/cancel_before_invoke.c
+++ b/test/core/end2end/tests/cancel_before_invoke.c
@@ -66,14 +66,10 @@
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
- GPR_ASSERT(ev);
- type = ev->type;
- grpc_event_finish(ev);
- } while (type != GRPC_QUEUE_SHUTDOWN);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -153,7 +149,7 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, test_ops, tag(1)));
- cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_CANCELLED);
diff --git a/test/core/end2end/tests/cancel_in_a_vacuum.c b/test/core/end2end/tests/cancel_in_a_vacuum.c
index 467a621..679b137 100644
--- a/test/core/end2end/tests/cancel_in_a_vacuum.c
+++ b/test/core/end2end/tests/cancel_in_a_vacuum.c
@@ -65,14 +65,10 @@
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
- GPR_ASSERT(ev);
- type = ev->type;
- grpc_event_finish(ev);
- } while (type != GRPC_QUEUE_SHUTDOWN);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
diff --git a/test/core/end2end/tests/census_simple_request.c b/test/core/end2end/tests/census_simple_request.c
index 4f18eab..4f6c57c 100644
--- a/test/core/end2end/tests/census_simple_request.c
+++ b/test/core/end2end/tests/census_simple_request.c
@@ -74,14 +74,10 @@
}
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, n_seconds_time(5));
- GPR_ASSERT(ev);
- type = ev->type;
- grpc_event_finish(ev);
- } while (type != GRPC_QUEUE_SHUTDOWN);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void end_test(grpc_end2end_test_fixture *f) {
@@ -141,7 +137,7 @@
grpc_server_request_call(f.server, &s, &call_details,
&request_metadata_recv, f.cq,
f.cq, tag(101)));
- cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(101), 1);
cq_verify(cqv);
op = ops;
@@ -158,8 +154,8 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
- cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
diff --git a/test/core/end2end/tests/disappearing_server.c b/test/core/end2end/tests/disappearing_server.c
index 1e5560f..9bcc79d 100644
--- a/test/core/end2end/tests/disappearing_server.c
+++ b/test/core/end2end/tests/disappearing_server.c
@@ -54,14 +54,10 @@
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
- GPR_ASSERT(ev);
- type = ev->type;
- grpc_event_finish(ev);
- } while (type != GRPC_QUEUE_SHUTDOWN);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -132,7 +128,7 @@
grpc_server_request_call(f->server, &s, &call_details,
&request_metadata_recv, f->cq,
f->cq, tag(101)));
- cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(101), 1);
cq_verify(cqv);
/* should be able to shut down the server early
@@ -153,8 +149,8 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
- cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
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 64f2f61..3efbf6a 100644
--- a/test/core/end2end/tests/early_server_shutdown_finishes_inflight_calls.c
+++ b/test/core/end2end/tests/early_server_shutdown_finishes_inflight_calls.c
@@ -66,14 +66,10 @@
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
- GPR_ASSERT(ev);
- type = ev->type;
- grpc_event_finish(ev);
- } while (type != GRPC_QUEUE_SHUTDOWN);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -147,7 +143,7 @@
grpc_server_request_call(f.server, &s, &call_details,
&request_metadata_recv, f.cq,
f.cq, tag(101)));
- cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(101), 1);
cq_verify(cqv);
op = ops;
@@ -159,8 +155,8 @@
/* shutdown and destroy the server */
shutdown_server(&f);
- cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
- cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_UNAVAILABLE);
diff --git a/test/core/end2end/tests/early_server_shutdown_finishes_tags.c b/test/core/end2end/tests/early_server_shutdown_finishes_tags.c
index b193820..c52e54b 100644
--- a/test/core/end2end/tests/early_server_shutdown_finishes_tags.c
+++ b/test/core/end2end/tests/early_server_shutdown_finishes_tags.c
@@ -66,14 +66,10 @@
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
- GPR_ASSERT(ev);
- type = ev->type;
- grpc_event_finish(ev);
- } while (type != GRPC_QUEUE_SHUTDOWN);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -116,7 +112,7 @@
&request_metadata_recv, f.cq,
f.cq, tag(101)));
grpc_server_shutdown(f.server);
- cq_expect_completion(cqv, tag(101), GRPC_OP_ERROR);
+ cq_expect_completion(cqv, tag(101), 0);
cq_verify(cqv);
GPR_ASSERT(s == NULL);
diff --git a/test/core/end2end/tests/empty_batch.c b/test/core/end2end/tests/empty_batch.c
index b3f24aa..f1c3fb2 100644
--- a/test/core/end2end/tests/empty_batch.c
+++ b/test/core/end2end/tests/empty_batch.c
@@ -68,14 +68,10 @@
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
- GPR_ASSERT(ev);
- type = ev->type;
- grpc_event_finish(ev);
- } while (type != GRPC_QUEUE_SHUTDOWN);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -111,7 +107,7 @@
GPR_ASSERT(c);
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, op, 0, tag(1)));
- cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
grpc_call_destroy(c);
diff --git a/test/core/end2end/tests/graceful_server_shutdown.c b/test/core/end2end/tests/graceful_server_shutdown.c
index b5a3906..70a0655 100644
--- a/test/core/end2end/tests/graceful_server_shutdown.c
+++ b/test/core/end2end/tests/graceful_server_shutdown.c
@@ -66,14 +66,10 @@
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
- GPR_ASSERT(ev);
- type = ev->type;
- grpc_event_finish(ev);
- } while (type != GRPC_QUEUE_SHUTDOWN);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -146,7 +142,7 @@
grpc_server_request_call(f.server, &s, &call_details,
&request_metadata_recv, f.cq,
f.cq, tag(101)));
- cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(101), 1);
cq_verify(cqv);
/* shutdown and destroy the server */
@@ -167,12 +163,12 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(102), 1);
cq_verify(cqv);
grpc_call_destroy(s);
- cq_expect_server_shutdown(cqv, tag(0xdead));
- cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(0xdead), 1);
+ cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
diff --git a/test/core/end2end/tests/invoke_large_request.c b/test/core/end2end/tests/invoke_large_request.c
index 6fbd78d..d4035d0 100644
--- a/test/core/end2end/tests/invoke_large_request.c
+++ b/test/core/end2end/tests/invoke_large_request.c
@@ -64,14 +64,10 @@
}
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, n_seconds_time(5));
- GPR_ASSERT(ev);
- type = ev->type;
- grpc_event_finish(ev);
- } while (type != GRPC_QUEUE_SHUTDOWN);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -164,7 +160,7 @@
grpc_server_request_call(f.server, &s, &call_details,
&request_metadata_recv, f.cq,
f.cq, tag(101)));
- cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(101), 1);
cq_verify(cqv);
op = ops;
@@ -187,8 +183,8 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
- cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
diff --git a/test/core/end2end/tests/max_concurrent_streams.c b/test/core/end2end/tests/max_concurrent_streams.c
index 15b3c86..57300a0 100644
--- a/test/core/end2end/tests/max_concurrent_streams.c
+++ b/test/core/end2end/tests/max_concurrent_streams.c
@@ -66,14 +66,10 @@
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
- GPR_ASSERT(ev);
- type = ev->type;
- grpc_event_finish(ev);
- } while (type != GRPC_QUEUE_SHUTDOWN);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -144,7 +140,7 @@
grpc_server_request_call(f.server, &s, &call_details,
&request_metadata_recv, f.cq,
f.cq, tag(101)));
- cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(101), 1);
cq_verify(cqv);
op = ops;
@@ -161,8 +157,8 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
- cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
@@ -194,7 +190,7 @@
int live_call;
gpr_timespec deadline;
cq_verifier *cqv;
- grpc_event *ev;
+ grpc_event ev;
grpc_call_details call_details;
grpc_metadata_array request_metadata_recv;
grpc_metadata_array initial_metadata_recv1;
@@ -297,26 +293,26 @@
got_client_start = 0;
got_server_start = 0;
+ live_call = -1;
while (!got_client_start || !got_server_start) {
ev = grpc_completion_queue_next(f.cq,
GRPC_TIMEOUT_SECONDS_TO_DEADLINE(3));
- GPR_ASSERT(ev);
- GPR_ASSERT(ev->type == GRPC_OP_COMPLETE);
- GPR_ASSERT(ev->data.op_complete == GRPC_OP_OK);
- if (ev->tag == tag(101)) {
+ GPR_ASSERT(ev.type == GRPC_OP_COMPLETE);
+ GPR_ASSERT(ev.success);
+ if (ev.tag == tag(101)) {
GPR_ASSERT(!got_server_start);
got_server_start = 1;
} else {
GPR_ASSERT(!got_client_start);
- GPR_ASSERT(ev->tag == tag(301) || ev->tag == tag(401));
+ GPR_ASSERT(ev.tag == tag(301) || ev.tag == tag(401));
/* The /alpha or /beta calls started above could be invoked (but NOT both);
* check this here */
/* We'll get tag 303 or 403, we want 300, 400 */
- live_call = ((int)(gpr_intptr)ev->tag) - 1;
+ live_call = ((int)(gpr_intptr)ev.tag) - 1;
got_client_start = 1;
}
- grpc_event_finish(ev);
}
+ GPR_ASSERT(live_call == 300 || live_call == 400);
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
@@ -333,18 +329,18 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_batch(s1, ops, op - ops, tag(102)));
- cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
- cq_expect_completion(cqv, tag(live_call + 2), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(live_call + 2), 1);
/* first request is finished, we should be able to start the second */
live_call = (live_call == 300) ? 400 : 300;
- cq_expect_completion(cqv, tag(live_call + 1), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(live_call + 1), 1);
cq_verify(cqv);
GPR_ASSERT(GRPC_CALL_OK ==
grpc_server_request_call(f.server, &s2, &call_details,
&request_metadata_recv, f.cq,
f.cq, tag(201)));
- cq_expect_completion(cqv, tag(201), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(201), 1);
cq_verify(cqv);
op = ops;
@@ -362,8 +358,8 @@
GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_batch(s2, ops, op - ops, tag(202)));
- cq_expect_completion(cqv, tag(live_call + 2), GRPC_OP_OK);
- cq_expect_completion(cqv, tag(202), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(live_call + 2), 1);
+ cq_expect_completion(cqv, tag(202), 1);
cq_verify(cqv);
cq_verifier_destroy(cqv);
diff --git a/test/core/end2end/tests/max_message_length.c b/test/core/end2end/tests/max_message_length.c
index d32930c..364d59e 100644
--- a/test/core/end2end/tests/max_message_length.c
+++ b/test/core/end2end/tests/max_message_length.c
@@ -66,14 +66,10 @@
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
- GPR_ASSERT(ev);
- type = ev->type;
- grpc_event_finish(ev);
- } while (type != GRPC_QUEUE_SHUTDOWN);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -162,7 +158,7 @@
grpc_server_request_call(f.server, &s, &call_details,
&request_metadata_recv, f.cq,
f.cq, tag(101)));
- cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(101), 1);
cq_verify(cqv);
op = ops;
@@ -171,8 +167,8 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
- cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_CANCELLED);
diff --git a/test/core/end2end/tests/no_op.c b/test/core/end2end/tests/no_op.c
index b7cd2f1..979eb06 100644
--- a/test/core/end2end/tests/no_op.c
+++ b/test/core/end2end/tests/no_op.c
@@ -64,14 +64,10 @@
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
- GPR_ASSERT(ev);
- type = ev->type;
- grpc_event_finish(ev);
- } while (type != GRPC_QUEUE_SHUTDOWN);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
diff --git a/test/core/end2end/tests/ping_pong_streaming.c b/test/core/end2end/tests/ping_pong_streaming.c
index bc9a254..c00beb7 100644
--- a/test/core/end2end/tests/ping_pong_streaming.c
+++ b/test/core/end2end/tests/ping_pong_streaming.c
@@ -66,14 +66,10 @@
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
- GPR_ASSERT(ev);
- type = ev->type;
- grpc_event_finish(ev);
- } while (type != GRPC_QUEUE_SHUTDOWN);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -152,7 +148,7 @@
grpc_server_request_call(f.server, &s, &call_details,
&request_metadata_recv, f.cq,
f.cq, tag(100)));
- cq_expect_completion(cqv, tag(100), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(100), 1);
cq_verify(cqv);
op = ops;
@@ -183,7 +179,7 @@
op++;
GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(102), 1);
cq_verify(cqv);
op = ops;
@@ -192,8 +188,8 @@
op++;
GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_batch(s, ops, op - ops, tag(103)));
- cq_expect_completion(cqv, tag(103), GRPC_OP_OK);
- cq_expect_completion(cqv, tag(2), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(103), 1);
+ cq_expect_completion(cqv, tag(2), 1);
cq_verify(cqv);
grpc_byte_buffer_destroy(request_payload);
@@ -218,10 +214,10 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(104)));
- cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
- cq_expect_completion(cqv, tag(3), GRPC_OP_OK);
- cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
- cq_expect_completion(cqv, tag(104), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(1), 1);
+ cq_expect_completion(cqv, tag(3), 1);
+ cq_expect_completion(cqv, tag(101), 1);
+ cq_expect_completion(cqv, tag(104), 1);
cq_verify(cqv);
grpc_call_destroy(c);
diff --git a/test/core/end2end/tests/registered_call.c b/test/core/end2end/tests/registered_call.c
index b2bae66..80f7d20 100644
--- a/test/core/end2end/tests/registered_call.c
+++ b/test/core/end2end/tests/registered_call.c
@@ -68,14 +68,10 @@
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
- GPR_ASSERT(ev);
- type = ev->type;
- grpc_event_finish(ev);
- } while (type != GRPC_QUEUE_SHUTDOWN);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -145,7 +141,7 @@
grpc_server_request_call(f.server, &s, &call_details,
&request_metadata_recv, f.cq,
f.cq, tag(101)));
- cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(101), 1);
cq_verify(cqv);
op = ops;
@@ -162,8 +158,8 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
- cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
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 58abf95..ce872a4 100644
--- a/test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c
+++ b/test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c
@@ -66,14 +66,10 @@
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
- GPR_ASSERT(ev);
- type = ev->type;
- grpc_event_finish(ev);
- } while (type != GRPC_QUEUE_SHUTDOWN);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -180,7 +176,7 @@
grpc_server_request_call(f.server, &s, &call_details,
&request_metadata_recv, f.cq,
f.cq, tag(101)));
- cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(101), 1);
cq_verify(cqv);
op = ops;
@@ -204,8 +200,8 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
- cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_OK);
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 d5e5cb9..b4edb73 100644
--- a/test/core/end2end/tests/request_response_with_metadata_and_payload.c
+++ b/test/core/end2end/tests/request_response_with_metadata_and_payload.c
@@ -66,14 +66,10 @@
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
- GPR_ASSERT(ev);
- type = ev->type;
- grpc_event_finish(ev);
- } while (type != GRPC_QUEUE_SHUTDOWN);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -166,7 +162,7 @@
grpc_server_request_call(f.server, &s, &call_details,
&request_metadata_recv, f.cq,
f.cq, tag(101)));
- cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(101), 1);
cq_verify(cqv);
op = ops;
@@ -190,8 +186,8 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
- cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_OK);
diff --git a/test/core/end2end/tests/request_response_with_payload.c b/test/core/end2end/tests/request_response_with_payload.c
index 558b2dd..cb50855 100644
--- a/test/core/end2end/tests/request_response_with_payload.c
+++ b/test/core/end2end/tests/request_response_with_payload.c
@@ -66,14 +66,10 @@
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
- GPR_ASSERT(ev);
- type = ev->type;
- grpc_event_finish(ev);
- } while (type != GRPC_QUEUE_SHUTDOWN);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -158,7 +154,7 @@
grpc_server_request_call(f.server, &s, &call_details,
&request_metadata_recv, f.cq,
f.cq, tag(101)));
- cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(101), 1);
cq_verify(cqv);
op = ops;
@@ -181,8 +177,8 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
- cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_OK);
diff --git a/test/core/end2end/tests/request_response_with_payload_and_call_creds.c b/test/core/end2end/tests/request_response_with_payload_and_call_creds.c
index 3759bd0..ec3f310 100644
--- a/test/core/end2end/tests/request_response_with_payload_and_call_creds.c
+++ b/test/core/end2end/tests/request_response_with_payload_and_call_creds.c
@@ -80,14 +80,10 @@
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
- GPR_ASSERT(ev);
- type = ev->type;
- grpc_event_finish(ev);
- } while (type != GRPC_QUEUE_SHUTDOWN);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -214,11 +210,11 @@
&request_metadata_recv,
f.cq, f.cq,
tag(101)));
- cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(101), 1);
cq_verify(cqv);
/* Cannot set creds on the server call object. */
- GPR_ASSERT(grpc_call_set_credentials(s, NULL) != GRPC_CALL_OK);
+ GPR_ASSERT(!grpc_call_set_credentials(s, NULL));
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
@@ -240,8 +236,8 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
- cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_OK);
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 9c0e1d2..1f1cb4c 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
@@ -66,14 +66,10 @@
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
- GPR_ASSERT(ev);
- type = ev->type;
- grpc_event_finish(ev);
- } while (type != GRPC_QUEUE_SHUTDOWN);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -168,9 +164,14 @@
GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
&call_details,
&request_metadata_recv,
+<<<<<<< HEAD
+ f.server_cq, tag(101)));
+ cq_expect_completion(v_server, tag(101), 1);
+=======
f.server_cq, f.server_cq,
tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
+>>>>>>> a468c36601dd5997580129bbd66b5ebed02521f8
cq_verify(v_server);
op = ops;
@@ -195,10 +196,10 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(v_server, tag(102), GRPC_OP_OK);
+ cq_expect_completion(v_server, tag(102), 1);
cq_verify(v_server);
- cq_expect_completion(v_client, tag(1), GRPC_OP_OK);
+ cq_expect_completion(v_client, tag(1), 1);
cq_verify(v_client);
GPR_ASSERT(status == GRPC_STATUS_OK);
diff --git a/test/core/end2end/tests/request_with_large_metadata.c b/test/core/end2end/tests/request_with_large_metadata.c
index 33228e1..a137786 100644
--- a/test/core/end2end/tests/request_with_large_metadata.c
+++ b/test/core/end2end/tests/request_with_large_metadata.c
@@ -66,14 +66,10 @@
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
- GPR_ASSERT(ev);
- type = ev->type;
- grpc_event_finish(ev);
- } while (type != GRPC_QUEUE_SHUTDOWN);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -162,7 +158,7 @@
grpc_server_request_call(f.server, &s, &call_details,
&request_metadata_recv, f.cq,
f.cq, tag(101)));
- cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(101), 1);
cq_verify(cqv);
op = ops;
@@ -182,8 +178,8 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
- cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_OK);
diff --git a/test/core/end2end/tests/request_with_payload.c b/test/core/end2end/tests/request_with_payload.c
index 13e9d59..86199c9 100644
--- a/test/core/end2end/tests/request_with_payload.c
+++ b/test/core/end2end/tests/request_with_payload.c
@@ -66,14 +66,10 @@
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
- GPR_ASSERT(ev);
- type = ev->type;
- grpc_event_finish(ev);
- } while (type != GRPC_QUEUE_SHUTDOWN);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -153,7 +149,7 @@
grpc_server_request_call(f.server, &s, &call_details,
&request_metadata_recv, f.cq,
f.cq, tag(101)));
- cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(101), 1);
cq_verify(cqv);
op = ops;
@@ -173,8 +169,8 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
- cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_OK);
diff --git a/test/core/end2end/tests/simple_delayed_request.c b/test/core/end2end/tests/simple_delayed_request.c
index 289acae..c51019c 100644
--- a/test/core/end2end/tests/simple_delayed_request.c
+++ b/test/core/end2end/tests/simple_delayed_request.c
@@ -54,14 +54,10 @@
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
- GPR_ASSERT(ev);
- type = ev->type;
- grpc_event_finish(ev);
- } while (type != GRPC_QUEUE_SHUTDOWN);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -140,7 +136,7 @@
grpc_server_request_call(f->server, &s, &call_details,
&request_metadata_recv, f->cq,
f->cq, tag(101)));
- cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(101), 1);
cq_verify(cqv);
op = ops;
@@ -157,8 +153,8 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
- cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
diff --git a/test/core/end2end/tests/simple_request.c b/test/core/end2end/tests/simple_request.c
index 8dd5a2b..be9e414 100644
--- a/test/core/end2end/tests/simple_request.c
+++ b/test/core/end2end/tests/simple_request.c
@@ -68,14 +68,10 @@
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
- GPR_ASSERT(ev);
- type = ev->type;
- grpc_event_finish(ev);
- } while (type != GRPC_QUEUE_SHUTDOWN);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -146,7 +142,7 @@
grpc_server_request_call(f.server, &s, &call_details,
&request_metadata_recv, f.cq,
f.cq, tag(101)));
- cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(101), 1);
cq_verify(cqv);
op = ops;
@@ -163,8 +159,8 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
- cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
diff --git a/test/core/end2end/tests/simple_request_with_high_initial_sequence_number.c b/test/core/end2end/tests/simple_request_with_high_initial_sequence_number.c
index 241da3f..566f971 100644
--- a/test/core/end2end/tests/simple_request_with_high_initial_sequence_number.c
+++ b/test/core/end2end/tests/simple_request_with_high_initial_sequence_number.c
@@ -68,14 +68,10 @@
static gpr_timespec five_seconds_time(void) { return n_seconds_time(5); }
static void drain_cq(grpc_completion_queue *cq) {
- grpc_event *ev;
- grpc_completion_type type;
+ grpc_event ev;
do {
ev = grpc_completion_queue_next(cq, five_seconds_time());
- GPR_ASSERT(ev);
- type = ev->type;
- grpc_event_finish(ev);
- } while (type != GRPC_QUEUE_SHUTDOWN);
+ } while (ev.type != GRPC_QUEUE_SHUTDOWN);
}
static void shutdown_server(grpc_end2end_test_fixture *f) {
@@ -146,7 +142,7 @@
grpc_server_request_call(f.server, &s, &call_details,
&request_metadata_recv, f.cq,
f.cq, tag(101)));
- cq_expect_completion(cqv, tag(101), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(101), 1);
cq_verify(cqv);
op = ops;
@@ -163,8 +159,8 @@
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(s, ops, op - ops, tag(102)));
- cq_expect_completion(cqv, tag(102), GRPC_OP_OK);
- cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(102), 1);
+ cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
GPR_ASSERT(status == GRPC_STATUS_UNIMPLEMENTED);
diff --git a/test/core/fling/client.c b/test/core/fling/client.c
index 68164b1..37d787c 100644
--- a/test/core/fling/client.c
+++ b/test/core/fling/client.c
@@ -93,7 +93,7 @@
"localhost", gpr_inf_future);
GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_batch(call, ops, op - ops, (void *)1));
- grpc_event_finish(grpc_completion_queue_next(cq, gpr_inf_future));
+ grpc_completion_queue_next(cq, gpr_inf_future);
grpc_call_destroy(call);
grpc_byte_buffer_destroy(response_payload_recv);
call = NULL;
@@ -106,7 +106,7 @@
stream_init_op.data.send_initial_metadata.count = 0;
GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_batch(call, &stream_init_op, 1, (void *)1));
- grpc_event_finish(grpc_completion_queue_next(cq, gpr_inf_future));
+ grpc_completion_queue_next(cq, gpr_inf_future);
grpc_metadata_array_init(&initial_metadata_recv);
@@ -119,7 +119,7 @@
static void step_ping_pong_stream(void) {
GPR_ASSERT(GRPC_CALL_OK ==
grpc_call_start_batch(call, stream_step_ops, 2, (void *)1));
- grpc_event_finish(grpc_completion_queue_next(cq, gpr_inf_future));
+ grpc_completion_queue_next(cq, gpr_inf_future);
grpc_byte_buffer_destroy(response_payload_recv);
}
@@ -147,7 +147,6 @@
char *fake_argv[1];
int payload_size = 1;
- int done;
int secure = 0;
char *target = "localhost:443";
gpr_cmdline *cl;
@@ -209,12 +208,9 @@
grpc_channel_destroy(channel);
grpc_completion_queue_shutdown(cq);
- done = 0;
- while (!done) {
- grpc_event *ev = grpc_completion_queue_next(cq, gpr_inf_future);
- done = (ev->type == GRPC_QUEUE_SHUTDOWN);
- grpc_event_finish(ev);
- }
+ while (grpc_completion_queue_next(cq, gpr_inf_future).type !=
+ GRPC_QUEUE_SHUTDOWN)
+ ;
grpc_completion_queue_destroy(cq);
grpc_byte_buffer_destroy(the_buffer);
gpr_slice_unref(slice);
diff --git a/test/core/fling/server.c b/test/core/fling/server.c
index 1f1f857..48304ed 100644
--- a/test/core/fling/server.c
+++ b/test/core/fling/server.c
@@ -174,7 +174,7 @@
static void sigint_handler(int x) { _exit(0); }
int main(int argc, char **argv) {
- grpc_event *ev;
+ grpc_event ev;
call_state *s;
char *addr_buf = NULL;
gpr_cmdline *cl;
@@ -239,9 +239,8 @@
}
ev = grpc_completion_queue_next(
cq, gpr_time_add(gpr_now(), gpr_time_from_micros(1000000)));
- if (!ev) continue;
- s = ev->tag;
- switch (ev->type) {
+ s = ev.tag;
+ switch (ev.type) {
case GRPC_OP_COMPLETE:
switch ((gpr_intptr)s) {
case FLING_SERVER_NEW_REQUEST:
@@ -303,10 +302,9 @@
GPR_ASSERT(shutdown_started);
shutdown_finished = 1;
break;
- default:
- GPR_ASSERT(0);
+ case GRPC_QUEUE_TIMEOUT:
+ break;
}
- grpc_event_finish(ev);
}
grpc_profiler_stop();
grpc_call_details_destroy(&call_details);
diff --git a/test/core/surface/completion_queue_benchmark.c b/test/core/surface/completion_queue_benchmark.c
deleted file mode 100644
index 81ebe15..0000000
--- a/test/core/surface/completion_queue_benchmark.c
+++ /dev/null
@@ -1,168 +0,0 @@
-/*
- *
- * Copyright 2015, Google Inc.
- * All rights reserved.
- *
- * Redistribution and use in source and binary forms, with or without
- * modification, are permitted provided that the following conditions are
- * met:
- *
- * * Redistributions of source code must retain the above copyright
- * notice, this list of conditions and the following disclaimer.
- * * Redistributions in binary form must reproduce the above
- * copyright notice, this list of conditions and the following disclaimer
- * in the documentation and/or other materials provided with the
- * distribution.
- * * Neither the name of Google Inc. nor the names of its
- * contributors may be used to endorse or promote products derived from
- * this software without specific prior written permission.
- *
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
- * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
- * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
- * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
- * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
- * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
- * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
- * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
- * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
- * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
- * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
- *
- */
-
-#include "src/core/surface/completion_queue.h"
-
-#include <math.h>
-#include <stdio.h>
-
-#include <grpc/support/alloc.h>
-#include <grpc/support/log.h>
-#include <grpc/support/thd.h>
-#include <grpc/support/time.h>
-
-typedef struct test_thread_options {
- gpr_event on_started;
- gpr_event *start;
- gpr_event on_finished;
- grpc_completion_queue *cc;
- int iterations;
-} test_thread_options;
-
-static void producer_thread(void *arg) {
- test_thread_options *opt = arg;
- int i;
-
- gpr_event_set(&opt->on_started, (void *)(gpr_intptr) 1);
- GPR_ASSERT(gpr_event_wait(opt->start, gpr_inf_future));
-
- for (i = 0; i < opt->iterations; i++) {
- grpc_cq_begin_op(opt->cc, NULL, GRPC_WRITE_ACCEPTED);
- grpc_cq_end_write_accepted(opt->cc, (void *)(gpr_intptr) 1, NULL, NULL,
- NULL, GRPC_OP_OK);
- }
-
- gpr_event_set(&opt->on_finished, (void *)(gpr_intptr) 1);
-}
-
-static void consumer_thread(void *arg) {
- test_thread_options *opt = arg;
- grpc_event *ev;
-
- gpr_event_set(&opt->on_started, (void *)(gpr_intptr) 1);
- GPR_ASSERT(gpr_event_wait(opt->start, gpr_inf_future));
-
- for (;;) {
- ev = grpc_completion_queue_next(opt->cc, gpr_inf_future);
- switch (ev->type) {
- case GRPC_WRITE_ACCEPTED:
- break;
- case GRPC_QUEUE_SHUTDOWN:
- gpr_event_set(&opt->on_finished, (void *)(gpr_intptr) 1);
- return;
- default:
- gpr_log(GPR_ERROR, "Invalid event received: %d", ev->type);
- abort();
- }
- grpc_event_finish(ev);
- }
-}
-
-double ops_per_second(int consumers, int producers, int iterations) {
- test_thread_options *options =
- gpr_malloc((producers + consumers) * sizeof(test_thread_options));
- gpr_event start = GPR_EVENT_INIT;
- grpc_completion_queue *cc = grpc_completion_queue_create();
- int i;
- gpr_timespec t_start, t_end, t_delta;
-
- /* start all threads: they will wait for phase1 */
- for (i = 0; i < producers + consumers; i++) {
- gpr_thd_id id;
- gpr_event_init(&options[i].on_started);
- gpr_event_init(&options[i].on_finished);
- options[i].start = &start;
- options[i].cc = cc;
- options[i].iterations = iterations;
- GPR_ASSERT(gpr_thd_new(&id,
- i < producers ? producer_thread : consumer_thread,
- options + i, NULL));
- gpr_event_wait(&options[i].on_started, gpr_inf_future);
- }
-
- /* start the benchmark */
- t_start = gpr_now();
- gpr_event_set(&start, (void *)(gpr_intptr) 1);
-
- /* wait for producers to finish */
- for (i = 0; i < producers; i++) {
- GPR_ASSERT(gpr_event_wait(&options[i].on_finished, gpr_inf_future));
- }
-
- /* in parallel, we shutdown the completion channel - all events should still
- be consumed */
- grpc_completion_queue_shutdown(cc);
-
- /* join all threads */
- for (i = producers; i < producers + consumers; i++) {
- GPR_ASSERT(gpr_event_wait(&options[i].on_finished, gpr_inf_future));
- }
- t_end = gpr_now();
-
- /* destroy the completion channel */
- grpc_completion_queue_destroy(cc);
-
- gpr_free(options);
-
- t_delta = gpr_time_sub(t_end, t_start);
- return (t_delta.tv_sec + 1e-9 * t_delta.tv_nsec) / (producers * iterations);
-}
-
-double ops_per_second_top(int consumers, int producers) {
- return ops_per_second(consumers, producers, 1000000 / producers);
-}
-
-int main(void) {
- const int counts[] = {1, 2, 3, 4, 5, 6, 7, 8, 12, 16, 20, 24, 32, 40, 64};
- const int ncounts = sizeof(counts) / sizeof(*counts);
- int i, j;
-
- printf("\"\",");
- for (i = 0; i < ncounts; i++) {
- int producers = counts[i];
- printf("%d%s", producers, i == ncounts - 1 ? "\n" : ",");
- }
-
- for (j = 0; j < ncounts; j++) {
- int consumers = counts[j];
- printf("%d,", consumers);
- for (i = 0; i < ncounts; i++) {
- int producers = counts[i];
- printf("%f%s", ops_per_second_top(consumers, producers),
- i == ncounts - 1 ? "\n" : ",");
- fflush(stdout);
- }
- }
-
- return 0;
-}
diff --git a/test/core/surface/completion_queue_test.c b/test/core/surface/completion_queue_test.c
index 29fb7a9..3e84eaf 100644
--- a/test/core/surface/completion_queue_test.c
+++ b/test/core/surface/completion_queue_test.c
@@ -43,10 +43,6 @@
#define LOG_TEST() gpr_log(GPR_INFO, "%s", __FUNCTION__)
-static void increment_int_on_finish(void *user_data, grpc_op_error error) {
- ++*(int *)user_data;
-}
-
static void *create_test_tag(void) {
static gpr_intptr i = 0;
return (void *)(++i);
@@ -54,12 +50,10 @@
/* helper for tests to shutdown correctly and tersely */
static void shutdown_and_destroy(grpc_completion_queue *cc) {
- grpc_event *ev;
+ grpc_event ev;
grpc_completion_queue_shutdown(cc);
ev = grpc_completion_queue_next(cc, gpr_inf_past);
- GPR_ASSERT(ev != NULL);
- GPR_ASSERT(ev->type == GRPC_QUEUE_SHUTDOWN);
- grpc_event_finish(ev);
+ GPR_ASSERT(ev.type == GRPC_QUEUE_SHUTDOWN);
grpc_completion_queue_destroy(cc);
}
@@ -75,42 +69,36 @@
LOG_TEST();
cc = grpc_completion_queue_create();
- GPR_ASSERT(grpc_completion_queue_next(cc, gpr_now()) == NULL);
+ GPR_ASSERT(grpc_completion_queue_next(cc, gpr_now()).type ==
+ GRPC_QUEUE_TIMEOUT);
shutdown_and_destroy(cc);
}
static void test_cq_end_op(void) {
- grpc_event *ev;
+ grpc_event ev;
grpc_completion_queue *cc;
- int on_finish_called = 0;
void *tag = create_test_tag();
LOG_TEST();
cc = grpc_completion_queue_create();
- grpc_cq_begin_op(cc, NULL, GRPC_OP_COMPLETE);
- grpc_cq_end_op(cc, tag, NULL, increment_int_on_finish, &on_finish_called,
- GRPC_OP_OK);
+ grpc_cq_begin_op(cc, NULL);
+ grpc_cq_end_op(cc, tag, NULL, 1);
ev = grpc_completion_queue_next(cc, gpr_inf_past);
- GPR_ASSERT(ev != NULL);
- GPR_ASSERT(ev->type == GRPC_OP_COMPLETE);
- GPR_ASSERT(ev->tag == tag);
- GPR_ASSERT(ev->data.op_complete == GRPC_OP_OK);
- GPR_ASSERT(on_finish_called == 0);
- grpc_event_finish(ev);
- GPR_ASSERT(on_finish_called == 1);
+ GPR_ASSERT(ev.type == GRPC_OP_COMPLETE);
+ GPR_ASSERT(ev.tag == tag);
+ GPR_ASSERT(ev.success);
shutdown_and_destroy(cc);
}
static void test_pluck(void) {
- grpc_event *ev;
+ grpc_event ev;
grpc_completion_queue *cc;
void *tags[128];
unsigned i, j;
- int on_finish_called = 0;
LOG_TEST();
@@ -124,34 +112,26 @@
cc = grpc_completion_queue_create();
for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) {
- grpc_cq_begin_op(cc, NULL, GRPC_OP_COMPLETE);
- grpc_cq_end_op(cc, tags[i], NULL, increment_int_on_finish,
- &on_finish_called, GRPC_OP_OK);
+ grpc_cq_begin_op(cc, NULL);
+ grpc_cq_end_op(cc, tags[i], NULL, 1);
}
for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) {
ev = grpc_completion_queue_pluck(cc, tags[i], gpr_inf_past);
- GPR_ASSERT(ev->tag == tags[i]);
- grpc_event_finish(ev);
+ GPR_ASSERT(ev.tag == tags[i]);
}
- GPR_ASSERT(on_finish_called == GPR_ARRAY_SIZE(tags));
-
for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) {
- grpc_cq_begin_op(cc, NULL, GRPC_OP_COMPLETE);
- grpc_cq_end_op(cc, tags[i], NULL, increment_int_on_finish,
- &on_finish_called, GRPC_OP_OK);
+ grpc_cq_begin_op(cc, NULL);
+ grpc_cq_end_op(cc, tags[i], NULL, 1);
}
for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) {
ev = grpc_completion_queue_pluck(cc, tags[GPR_ARRAY_SIZE(tags) - i - 1],
gpr_inf_past);
- GPR_ASSERT(ev->tag == tags[GPR_ARRAY_SIZE(tags) - i - 1]);
- grpc_event_finish(ev);
+ GPR_ASSERT(ev.tag == tags[GPR_ARRAY_SIZE(tags) - i - 1]);
}
- GPR_ASSERT(on_finish_called == 2 * GPR_ARRAY_SIZE(tags));
-
shutdown_and_destroy(cc);
}
@@ -182,7 +162,7 @@
gpr_log(GPR_INFO, "producer %d phase 1", opt->id);
for (i = 0; i < TEST_THREAD_EVENTS; i++) {
- grpc_cq_begin_op(opt->cc, NULL, GRPC_OP_COMPLETE);
+ grpc_cq_begin_op(opt->cc, NULL);
}
gpr_log(GPR_INFO, "producer %d phase 1 done", opt->id);
@@ -191,8 +171,7 @@
gpr_log(GPR_INFO, "producer %d phase 2", opt->id);
for (i = 0; i < TEST_THREAD_EVENTS; i++) {
- grpc_cq_end_op(opt->cc, (void *)(gpr_intptr)1, NULL, NULL, NULL,
- GRPC_OP_OK);
+ grpc_cq_end_op(opt->cc, (void *)(gpr_intptr)1, NULL, 1);
opt->events_triggered++;
}
@@ -202,7 +181,7 @@
static void consumer_thread(void *arg) {
test_thread_options *opt = arg;
- grpc_event *ev;
+ grpc_event ev;
gpr_log(GPR_INFO, "consumer %d started", opt->id);
gpr_event_set(&opt->on_started, (void *)(gpr_intptr) 1);
@@ -217,20 +196,17 @@
gpr_log(GPR_INFO, "consumer %d phase 2", opt->id);
for (;;) {
ev = grpc_completion_queue_next(opt->cc, ten_seconds_time());
- GPR_ASSERT(ev);
- switch (ev->type) {
+ switch (ev.type) {
case GRPC_OP_COMPLETE:
- GPR_ASSERT(ev->data.op_complete == GRPC_OP_OK);
+ GPR_ASSERT(ev.success);
opt->events_triggered++;
- grpc_event_finish(ev);
break;
case GRPC_QUEUE_SHUTDOWN:
gpr_log(GPR_INFO, "consumer %d phase 2 done", opt->id);
gpr_event_set(&opt->on_finished, (void *)(gpr_intptr) 1);
- grpc_event_finish(ev);
return;
- default:
- gpr_log(GPR_ERROR, "Invalid event received: %d", ev->type);
+ case GRPC_QUEUE_TIMEOUT:
+ gpr_log(GPR_ERROR, "Invalid timeout received");
abort();
}
}
diff --git a/test/core/surface/lame_client_test.c b/test/core/surface/lame_client_test.c
index 05e8f95..34d37f0 100644
--- a/test/core/surface/lame_client_test.c
+++ b/test/core/surface/lame_client_test.c
@@ -79,7 +79,7 @@
grpc_call_start_batch(call, ops, op - ops, tag(1)));
/* the call should immediately fail */
- cq_expect_completion(cqv, tag(1), GRPC_OP_OK);
+ cq_expect_completion(cqv, tag(1), 1);
cq_verify(cqv);
grpc_call_destroy(call);
diff --git a/vsprojects/Grpc.mak b/vsprojects/Grpc.mak
index 7823687..4c34859 100644
--- a/vsprojects/Grpc.mak
+++ b/vsprojects/Grpc.mak
@@ -322,13 +322,6 @@
grpc_channel_stack_test: grpc_channel_stack_test.exe
echo Running grpc_channel_stack_test
$(OUT_DIR)\grpc_channel_stack_test.exe
-grpc_completion_queue_benchmark.exe: build_libs $(OUT_DIR)
- echo Building grpc_completion_queue_benchmark
- $(CC) $(CFLAGS) /Fo:$(OUT_DIR)\ $(REPO_ROOT)\test\core\surface\completion_queue_benchmark.c
- $(LINK) $(LFLAGS) /OUT:"$(OUT_DIR)\grpc_completion_queue_benchmark.exe" Debug\grpc_test_util.lib Debug\grpc.lib Debug\gpr_test_util.lib Debug\gpr.lib $(LIBS) $(OUT_DIR)\completion_queue_benchmark.obj
-grpc_completion_queue_benchmark: grpc_completion_queue_benchmark.exe
- echo Running grpc_completion_queue_benchmark
- $(OUT_DIR)\grpc_completion_queue_benchmark.exe
grpc_completion_queue_test.exe: build_libs $(OUT_DIR)
echo Building grpc_completion_queue_test
$(CC) $(CFLAGS) /Fo:$(OUT_DIR)\ $(REPO_ROOT)\test\core\surface\completion_queue_test.c