From 83f88d90b9c61c94994ed00d03a6fa469359d559 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Tue, 21 Apr 2015 16:02:05 -0700 Subject: stuff --- src/core/transport/chttp2_transport.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'src/core/transport/chttp2_transport.c') diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index e32ee284e0..86bd5d2098 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -503,7 +503,7 @@ static void init_transport(transport *t, grpc_transport_setup_callback setup, gpr_mu_lock(&t->mu); t->calling_back = 1; - ref_transport(t); + ref_transport(t); /* matches unref at end of this function */ gpr_mu_unlock(&t->mu); sr = setup(arg, &t->base, t->metadata_context); @@ -515,7 +515,7 @@ static void init_transport(transport *t, grpc_transport_setup_callback setup, if (t->destroying) gpr_cv_signal(&t->cv); unlock(t); - ref_transport(t); + ref_transport(t); /* matches unref inside recv_data */ recv_data(t, slices, nslices, GRPC_ENDPOINT_CB_OK); unref_transport(t); -- cgit v1.2.3 From be18b8db301b800edef382389749632e4099afaf Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Wed, 22 Apr 2015 14:00:47 -0700 Subject: Beginning transport work --- src/core/surface/channel_create.c | 4 +- src/core/surface/client.c | 29 +------ src/core/surface/lame_client.c | 19 +---- src/core/surface/server.c | 124 +++++++++++++++-------------- src/core/surface/server_chttp2.c | 4 +- src/core/transport/chttp2/stream_encoder.c | 10 --- src/core/transport/chttp2_transport.c | 88 ++++++++++++++------ src/core/transport/transport.h | 3 + src/core/transport/transport_impl.h | 12 +-- 9 files changed, 138 insertions(+), 155 deletions(-) (limited to 'src/core/transport/chttp2_transport.c') diff --git a/src/core/surface/channel_create.c b/src/core/surface/channel_create.c index 3104b1d00d..5f3e9bec0c 100644 --- a/src/core/surface/channel_create.c +++ b/src/core/surface/channel_create.c @@ -44,7 +44,6 @@ #include "src/core/channel/client_setup.h" #include "src/core/channel/connected_channel.h" #include "src/core/channel/http_client_filter.h" -#include "src/core/channel/http_filter.h" #include "src/core/iomgr/endpoint.h" #include "src/core/iomgr/resolve_address.h" #include "src/core/iomgr/tcp_client.h" @@ -176,8 +175,7 @@ static void done_setup(void *sp) { static grpc_transport_setup_result complete_setup(void *channel_stack, grpc_transport *transport, grpc_mdctx *mdctx) { - static grpc_channel_filter const *extra_filters[] = {&grpc_http_client_filter, - &grpc_http_filter}; + static grpc_channel_filter const *extra_filters[] = {&grpc_http_client_filter}; return grpc_client_channel_transport_setup_complete( channel_stack, transport, extra_filters, GPR_ARRAY_SIZE(extra_filters), mdctx); diff --git a/src/core/surface/client.c b/src/core/surface/client.c index 2f898ff7d7..4669c59ee4 100644 --- a/src/core/surface/client.c +++ b/src/core/surface/client.c @@ -43,32 +43,9 @@ typedef struct { void *unused; } call_data; typedef struct { void *unused; } channel_data; -static void call_op(grpc_call_element *elem, grpc_call_element *from_elem, - grpc_call_op *op) { +static void client_start_transport_op(grpc_call_element *elem, grpc_transport_op *op) { GRPC_CALL_LOG_OP(GPR_INFO, elem, op); - - switch (op->type) { - case GRPC_RECV_METADATA: - grpc_call_recv_metadata(elem, &op->data.metadata); - break; - case GRPC_RECV_MESSAGE: - grpc_call_recv_message(elem, op->data.message); - op->done_cb(op->user_data, GRPC_OP_OK); - break; - case GRPC_RECV_HALF_CLOSE: - grpc_call_read_closed(elem); - break; - case GRPC_RECV_FINISH: - grpc_call_stream_closed(elem); - break; - case GRPC_RECV_SYNTHETIC_STATUS: - grpc_call_recv_synthetic_status(elem, op->data.synthetic_status.status, - op->data.synthetic_status.message); - break; - default: - GPR_ASSERT(op->dir == GRPC_CALL_DOWN); - grpc_call_next_op(elem, op); - } + grpc_call_next_op(elem, op); } static void channel_op(grpc_channel_element *elem, @@ -104,6 +81,6 @@ static void init_channel_elem(grpc_channel_element *elem, static void destroy_channel_elem(grpc_channel_element *elem) {} const grpc_channel_filter grpc_client_surface_filter = { - call_op, channel_op, sizeof(call_data), init_call_elem, destroy_call_elem, + client_start_transport_op, channel_op, sizeof(call_data), init_call_elem, destroy_call_elem, sizeof(channel_data), init_channel_elem, destroy_channel_elem, "client", }; diff --git a/src/core/surface/lame_client.c b/src/core/surface/lame_client.c index 78170806f1..4e9eb808d7 100644 --- a/src/core/surface/lame_client.c +++ b/src/core/surface/lame_client.c @@ -46,22 +46,9 @@ typedef struct { void *unused; } call_data; typedef struct { void *unused; } channel_data; -static void call_op(grpc_call_element *elem, grpc_call_element *from_elem, - grpc_call_op *op) { +static void lame_start_transport_op(grpc_call_element *elem, grpc_transport_op *op) { GRPC_CALL_LOG_OP(GPR_INFO, elem, op); - - switch (op->type) { - case GRPC_SEND_METADATA: - grpc_metadata_batch_destroy(&op->data.metadata); - grpc_call_recv_synthetic_status(elem, GRPC_STATUS_UNKNOWN, - "Rpc sent on a lame channel."); - grpc_call_stream_closed(elem); - break; - default: - break; - } - - op->done_cb(op->user_data, GRPC_OP_ERROR); + grpc_transport_op_finish_with_failure(op); } static void channel_op(grpc_channel_element *elem, @@ -93,7 +80,7 @@ static void init_channel_elem(grpc_channel_element *elem, static void destroy_channel_elem(grpc_channel_element *elem) {} static const grpc_channel_filter lame_filter = { - call_op, channel_op, sizeof(call_data), init_call_elem, destroy_call_elem, + lame_start_transport_op, channel_op, sizeof(call_data), init_call_elem, destroy_call_elem, sizeof(channel_data), init_channel_elem, destroy_channel_elem, "lame-client", }; diff --git a/src/core/surface/server.c b/src/core/surface/server.c index e771929870..82d1323a4b 100644 --- a/src/core/surface/server.c +++ b/src/core/surface/server.c @@ -173,13 +173,19 @@ struct call_data { grpc_call *call; call_state state; - gpr_timespec deadline; grpc_mdstr *path; grpc_mdstr *host; + gpr_timespec deadline; + int got_initial_metadata; legacy_data *legacy; grpc_completion_queue *cq_new; + grpc_stream_op_buffer *recv_ops; + grpc_stream_state *recv_state; + void (*on_done_recv)(void *user_data, int success); + void *recv_user_data; + call_data **root[CALL_LIST_COUNT]; call_link links[CALL_LIST_COUNT]; }; @@ -371,46 +377,6 @@ static void kill_zombie(void *elem, int success) { grpc_call_destroy(grpc_call_from_top_element(elem)); } -static void stream_closed(grpc_call_element *elem) { - call_data *calld = elem->call_data; - channel_data *chand = elem->channel_data; - gpr_mu_lock(&chand->server->mu); - switch (calld->state) { - case ACTIVATED: - break; - case PENDING: - call_list_remove(calld, PENDING_START); - /* fallthrough intended */ - case NOT_STARTED: - calld->state = ZOMBIED; - grpc_iomgr_add_callback(kill_zombie, elem); - break; - case ZOMBIED: - break; - } - gpr_mu_unlock(&chand->server->mu); - grpc_call_stream_closed(elem); -} - -static void read_closed(grpc_call_element *elem) { - call_data *calld = elem->call_data; - channel_data *chand = elem->channel_data; - gpr_mu_lock(&chand->server->mu); - switch (calld->state) { - case ACTIVATED: - case PENDING: - grpc_call_read_closed(elem); - break; - case NOT_STARTED: - calld->state = ZOMBIED; - grpc_iomgr_add_callback(kill_zombie, elem); - break; - case ZOMBIED: - break; - } - gpr_mu_unlock(&chand->server->mu); -} - static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) { grpc_call_element *elem = user_data; channel_data *chand = elem->channel_data; @@ -425,33 +391,69 @@ static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) { return md; } -static void call_op(grpc_call_element *elem, grpc_call_element *from_elemn, - grpc_call_op *op) { +static void server_on_recv(void *ptr, int success) { + grpc_call_element *elem = ptr; call_data *calld = elem->call_data; - GRPC_CALL_LOG_OP(GPR_INFO, elem, op); - switch (op->type) { - case GRPC_RECV_METADATA: + channel_data *chand = elem->channel_data; + + if (success && !calld->got_initial_metadata) { + size_t i; + size_t nops = calld->recv_ops->nops; + grpc_stream_op *ops = calld->recv_ops->ops; + for (i = 0; i < nops; i++) { + grpc_stream_op *op = &ops[i]; + if (op->type != GRPC_OP_METADATA) continue; grpc_metadata_batch_filter(&op->data.metadata, server_filter, elem); - if (grpc_call_recv_metadata(elem, &op->data.metadata)) { + if (0 != gpr_time_cmp(op->data.metadata.deadline, gpr_inf_future)) { calld->deadline = op->data.metadata.deadline; - start_new_rpc(elem); } + calld->got_initial_metadata = 1; + start_new_rpc(elem); break; - case GRPC_RECV_MESSAGE: - grpc_call_recv_message(elem, op->data.message); - op->done_cb(op->user_data, GRPC_OP_OK); - break; - case GRPC_RECV_HALF_CLOSE: - read_closed(elem); - break; - case GRPC_RECV_FINISH: - stream_closed(elem); + } + } + + switch (*calld->recv_state) { + case GRPC_STREAM_OPEN: break; + case GRPC_STREAM_SEND_CLOSED: break; + case GRPC_STREAM_RECV_CLOSED: + gpr_mu_lock(&chand->server->mu); + if (calld->state == NOT_STARTED) { + calld->state = ZOMBIED; + grpc_iomgr_add_callback(kill_zombie, elem); + } + gpr_mu_unlock(&chand->server->mu); break; - default: - GPR_ASSERT(op->dir == GRPC_CALL_DOWN); - grpc_call_next_op(elem, op); + case GRPC_STREAM_CLOSED: + gpr_mu_lock(&chand->server->mu); + if (calld->state == NOT_STARTED) { + calld->state = ZOMBIED; + grpc_iomgr_add_callback(kill_zombie, elem); + } else if (calld->state == PENDING) { + call_list_remove(calld, PENDING_START); + } + gpr_mu_unlock(&chand->server->mu); break; } + + calld->on_done_recv(calld->recv_user_data, success); +} + +static void server_start_transport_op(grpc_call_element *elem, grpc_transport_op *op) { + call_data *calld = elem->call_data; + GRPC_CALL_LOG_OP(GPR_INFO, elem, op); + + if (op->recv_ops) { + /* substitute our callback for the higher callback */ + calld->recv_ops = op->recv_ops; + calld->recv_state = op->recv_state; + calld->on_done_recv = op->on_done_recv; + calld->recv_user_data = op->recv_user_data; + op->on_done_recv = server_on_recv; + op->recv_user_data = elem; + } + + grpc_call_next_op(elem, op); } static void channel_op(grpc_channel_element *elem, @@ -592,7 +594,7 @@ static void destroy_channel_elem(grpc_channel_element *elem) { } static const grpc_channel_filter server_surface_filter = { - call_op, channel_op, sizeof(call_data), init_call_elem, destroy_call_elem, + server_start_transport_op, channel_op, sizeof(call_data), init_call_elem, destroy_call_elem, sizeof(channel_data), init_channel_elem, destroy_channel_elem, "server", }; diff --git a/src/core/surface/server_chttp2.c b/src/core/surface/server_chttp2.c index f3b9219f8b..ebde5095a9 100644 --- a/src/core/surface/server_chttp2.c +++ b/src/core/surface/server_chttp2.c @@ -33,7 +33,6 @@ #include -#include "src/core/channel/http_filter.h" #include "src/core/channel/http_server_filter.h" #include "src/core/iomgr/resolve_address.h" #include "src/core/iomgr/tcp_server.h" @@ -46,8 +45,7 @@ static grpc_transport_setup_result setup_transport(void *server, grpc_transport *transport, grpc_mdctx *mdctx) { - static grpc_channel_filter const *extra_filters[] = {&grpc_http_server_filter, - &grpc_http_filter}; + static grpc_channel_filter const *extra_filters[] = {&grpc_http_server_filter}; return grpc_server_setup_transport(server, transport, extra_filters, GPR_ARRAY_SIZE(extra_filters), mdctx); } diff --git a/src/core/transport/chttp2/stream_encoder.c b/src/core/transport/chttp2/stream_encoder.c index 5ca31d6bc7..203de99314 100644 --- a/src/core/transport/chttp2/stream_encoder.c +++ b/src/core/transport/chttp2/stream_encoder.c @@ -481,12 +481,6 @@ gpr_uint32 grpc_chttp2_preencode(grpc_stream_op *inops, size_t *inops_count, break; case GRPC_OP_METADATA: grpc_metadata_batch_assert_ok(&op->data.metadata); - case GRPC_OP_FLOW_CTL_CB: - /* these just get copied as they don't impact the number of flow - controlled bytes */ - grpc_sopb_append(outops, op, 1); - curop++; - break; case GRPC_OP_BEGIN_MESSAGE: /* begin op: for now we just convert the op to a slice and fall through - this lets us reuse the slice framing code below */ @@ -567,10 +561,6 @@ void grpc_chttp2_encode(grpc_stream_op *ops, size_t ops_count, int eof, GPR_ERROR, "These stream ops should be filtered out by grpc_chttp2_preencode"); abort(); - case GRPC_OP_FLOW_CTL_CB: - op->data.flow_ctl_cb.cb(op->data.flow_ctl_cb.arg, GRPC_OP_OK); - curop++; - break; case GRPC_OP_METADATA: /* Encode a metadata batch; store the returned values, representing a metadata element that needs to be unreffed back into the metadata diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index 86bd5d2098..9aee7ca4f1 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -91,10 +91,6 @@ typedef enum { /* streams that are waiting to start because there are too many concurrent streams on the connection */ WAITING_FOR_CONCURRENCY, - /* streams that want to callback the application */ - PENDING_CALLBACKS, - /* streams that *ARE* calling back to the application */ - EXECUTING_CALLBACKS, STREAM_LIST_COUNT /* must be last */ } stream_list_id; @@ -182,6 +178,18 @@ typedef struct { gpr_slice debug; } pending_goaway; +typedef struct { + void (*cb)(void *user_data, int success); + void *user_data; + int status; +} op_closure; + +typedef struct { + op_closure *callbacks; + size_t count; + size_t capacity; +} op_closure_array; + struct transport { grpc_transport base; /* must be first */ const grpc_transport_callbacks *cb; @@ -202,6 +210,10 @@ struct transport { gpr_uint8 closed; error_state error_state; + /* queued callbacks */ + op_closure_array pending_callbacks; + op_closure_array executing_callbacks; + /* stream indexing */ gpr_uint32 next_stream_id; gpr_uint32 last_incoming_stream_id; @@ -289,6 +301,9 @@ struct stream { gpr_uint8 allow_window_updates; gpr_uint8 published_close; + op_closure send_done_closure; + op_closure recv_done_closure; + stream_link links[STREAM_LIST_COUNT]; gpr_uint8 included[STREAM_LIST_COUNT]; @@ -416,6 +431,8 @@ static void init_transport(transport *t, grpc_transport_setup_callback setup, GPR_ASSERT(strlen(CLIENT_CONNECT_STRING) == CLIENT_CONNECT_STRLEN); + memset(t, 0, sizeof(*t)); + t->base.vtable = &vtable; t->ep = ep; /* one ref is for destroy, the other for when ep becomes NULL */ @@ -427,27 +444,16 @@ static void init_transport(transport *t, grpc_transport_setup_callback setup, t->str_grpc_timeout = grpc_mdstr_from_string(t->metadata_context, "grpc-timeout"); t->reading = 1; - t->writing = 0; t->error_state = ERROR_STATE_NONE; t->next_stream_id = is_client ? 1 : 2; - t->last_incoming_stream_id = 0; - t->destroying = 0; - t->closed = 0; t->is_client = is_client; t->outgoing_window = DEFAULT_WINDOW; t->incoming_window = DEFAULT_WINDOW; t->connection_window_target = DEFAULT_CONNECTION_WINDOW_TARGET; t->deframe_state = is_client ? DTS_FH_0 : DTS_CLIENT_PREFIX_0; - t->expect_continuation_stream_id = 0; - t->pings = NULL; - t->ping_count = 0; - t->ping_capacity = 0; t->ping_counter = gpr_now().tv_nsec; grpc_chttp2_hpack_compressor_init(&t->hpack_compressor, mdctx); grpc_chttp2_goaway_parser_init(&t->goaway_parser); - t->pending_goaways = NULL; - t->num_pending_goaways = 0; - t->cap_pending_goaways = 0; gpr_slice_buffer_init(&t->outbuf); gpr_slice_buffer_init(&t->qbuf); grpc_sopb_init(&t->nuke_later_sopb); @@ -462,7 +468,6 @@ static void init_transport(transport *t, grpc_transport_setup_callback setup, needed. TODO(ctiller): tune this */ grpc_chttp2_stream_map_init(&t->stream_map, 8); - memset(&t->lists, 0, sizeof(t->lists)); /* copy in initial settings to all setting sets */ for (i = 0; i < NUM_SETTING_SETS; i++) { @@ -708,8 +713,6 @@ static void stream_list_add_tail(transport *t, stream *s, stream_list_id id) { } static void stream_list_join(transport *t, stream *s, stream_list_id id) { - if (id == PENDING_CALLBACKS) - GPR_ASSERT(t->cb != NULL || t->error_state == ERROR_STATE_NONE); if (s->included[id]) { return; } @@ -988,6 +991,32 @@ static void maybe_start_some_streams(transport *t) { } } +static void perform_op(grpc_transport *gt, grpc_stream *gs, grpc_transport_op *op) { + transport *t = (transport *)gt; + stream *s = (stream *)gs; + + lock(t); + + if (op->send_ops) { + abort(); + } + + if (op->recv_ops) { + abort(); + } + + if (op->bind_pollset) { + abort(); + } + + if (op->cancel_with_status) { + abort(); + } + + unlock(t); +} + +#if 0 static void send_batch(grpc_transport *gt, grpc_stream *gs, grpc_stream_op *ops, size_t ops_count, int is_last) { transport *t = (transport *)gt; @@ -1016,6 +1045,7 @@ static void send_batch(grpc_transport *gt, grpc_stream *gs, grpc_stream_op *ops, unlock(t); } +#endif static void abort_stream(grpc_transport *gt, grpc_stream *gs, grpc_status_code status) { @@ -1831,6 +1861,12 @@ static grpc_stream_state compute_state(gpr_uint8 write_closed, } static int prepare_callbacks(transport *t) { + op_closure_array temp = t->pending_callbacks; + t->pending_callbacks = t->executing_callbacks; + t->executing_callbacks = temp; + return t->executing_callbacks.count > 0; + +#if 0 stream *s; int n = 0; while ((s = stream_list_remove_head(t, PENDING_CALLBACKS))) { @@ -1855,16 +1891,16 @@ static int prepare_callbacks(transport *t) { } } return n; +#endif } static void run_callbacks(transport *t, const grpc_transport_callbacks *cb) { - stream *s; - while ((s = stream_list_remove_head(t, EXECUTING_CALLBACKS))) { - size_t nops = s->callback_sopb.nops; - s->callback_sopb.nops = 0; - cb->recv_batch(t->cb_user_data, &t->base, (grpc_stream *)s, - s->callback_sopb.ops, nops, s->callback_state); + size_t i; + for (i = 0; i < t->executing_callbacks.count; i++) { + op_closure c = t->executing_callbacks.callbacks[i]; + c.cb(c.user_data, c.status); } + t->executing_callbacks.count = 0; } static void call_cb_closed(transport *t, const grpc_transport_callbacks *cb) { @@ -1885,8 +1921,8 @@ static void add_to_pollset(grpc_transport *gt, grpc_pollset *pollset) { */ static const grpc_transport_vtable vtable = { - sizeof(stream), init_stream, send_batch, set_allow_window_updates, - add_to_pollset, destroy_stream, abort_stream, goaway, close_transport, + sizeof(stream), init_stream, perform_op, + add_to_pollset, destroy_stream, goaway, close_transport, send_ping, destroy_transport}; void grpc_create_chttp2_transport(grpc_transport_setup_callback setup, diff --git a/src/core/transport/transport.h b/src/core/transport/transport.h index 264245d351..d0007680e3 100644 --- a/src/core/transport/transport.h +++ b/src/core/transport/transport.h @@ -132,6 +132,9 @@ typedef struct grpc_transport_op { void grpc_transport_op_finish_with_failure(grpc_transport_op *op); +/* TODO(ctiller): remove this */ +void grpc_transport_add_to_pollset(grpc_transport *transport, grpc_pollset *pollset); + char *grpc_transport_op_string(grpc_transport_op *op); /* Send a batch of operations on a transport diff --git a/src/core/transport/transport_impl.h b/src/core/transport/transport_impl.h index ac275c7560..ef79ac0741 100644 --- a/src/core/transport/transport_impl.h +++ b/src/core/transport/transport_impl.h @@ -46,12 +46,8 @@ typedef struct grpc_transport_vtable { const void *server_data); /* implementation of grpc_transport_send_batch */ - void (*send_batch)(grpc_transport *self, grpc_stream *stream, - grpc_stream_op *ops, size_t ops_count, int is_last); - - /* implementation of grpc_transport_set_allow_window_updates */ - void (*set_allow_window_updates)(grpc_transport *self, grpc_stream *stream, - int allow); + void (*perform_op)(grpc_transport *self, grpc_stream *stream, + grpc_transport_op *op); /* implementation of grpc_transport_add_to_pollset */ void (*add_to_pollset)(grpc_transport *self, grpc_pollset *pollset); @@ -59,10 +55,6 @@ typedef struct grpc_transport_vtable { /* implementation of grpc_transport_destroy_stream */ void (*destroy_stream)(grpc_transport *self, grpc_stream *stream); - /* implementation of grpc_transport_abort_stream */ - void (*abort_stream)(grpc_transport *self, grpc_stream *stream, - grpc_status_code status); - /* implementation of grpc_transport_goaway */ void (*goaway)(grpc_transport *self, grpc_status_code status, gpr_slice debug_data); -- cgit v1.2.3 From c079c111cb953a8f2671be86bcfb6e77b5805e08 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Wed, 22 Apr 2015 15:23:39 -0700 Subject: Transport compiles --- src/core/transport/chttp2_transport.c | 269 ++++++++++++++++++---------------- 1 file changed, 145 insertions(+), 124 deletions(-) (limited to 'src/core/transport/chttp2_transport.c') diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index 9aee7ca4f1..fed3088789 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -91,6 +91,9 @@ typedef enum { /* streams that are waiting to start because there are too many concurrent streams on the connection */ WAITING_FOR_CONCURRENCY, + /* streams that have finished reading: we wait until unlock to coalesce + all changes into one callback */ + FINISHED_READ_OP, STREAM_LIST_COUNT /* must be last */ } stream_list_id; @@ -137,6 +140,12 @@ typedef enum { DTS_FRAME } deframe_transport_state; +typedef enum { + WRITE_STATE_OPEN, + WRITE_STATE_QUEUED_CLOSE, + WRITE_STATE_SENT_CLOSE +} WRITE_STATE; + typedef struct { stream *head; stream *tail; @@ -181,7 +190,7 @@ typedef struct { typedef struct { void (*cb)(void *user_data, int success); void *user_data; - int status; + int success; } op_closure; typedef struct { @@ -293,12 +302,10 @@ struct stream { /* when the application requests writes be closed, the write_closed is 'queued'; when the close is flow controlled into the send path, we are 'sending' it; when the write has been performed it is 'sent' */ - gpr_uint8 queued_write_closed; - gpr_uint8 sending_write_closed; - gpr_uint8 sent_write_closed; + WRITE_STATE write_state; + gpr_uint8 send_closed; gpr_uint8 read_closed; gpr_uint8 cancelled; - gpr_uint8 allow_window_updates; gpr_uint8 published_close; op_closure send_done_closure; @@ -314,7 +321,10 @@ struct stream { gpr_timespec incoming_deadline; /* sops from application */ - grpc_stream_op_buffer outgoing_sopb; + grpc_stream_op_buffer *outgoing_sopb; + grpc_stream_op_buffer *incoming_sopb; + grpc_stream_state *publish_state; + grpc_stream_state published_state; /* sops that have passed flow control to be written */ grpc_stream_op_buffer writing_sopb; @@ -363,6 +373,13 @@ static void become_skip_parser(transport *t); static void recv_data(void *tp, gpr_slice *slices, size_t nslices, grpc_endpoint_cb_status error); +static void schedule_cb(transport *t, op_closure closure, int success); +static void maybe_finish_read(transport *t, stream *s); +static void maybe_join_window_updates(transport *t, stream *s); +static void finish_reads(transport *t); +static void add_to_pollset_locked(transport *t, grpc_pollset *pollset); + + /* * CONSTRUCTION/DESTRUCTION/REFCOUNTING */ @@ -582,6 +599,8 @@ static int init_stream(grpc_transport *gt, grpc_stream *gs, transport *t = (transport *)gt; stream *s = (stream *)gs; + memset(s, 0, sizeof(*s)); + ref_transport(t); if (!server_data) { @@ -597,20 +616,7 @@ static int init_stream(grpc_transport *gt, grpc_stream *gs, t->settings[PEER_SETTINGS][GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]; s->incoming_window = t->settings[SENT_SETTINGS][GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]; - s->queued_write_closed = 0; - s->sending_write_closed = 0; - s->sent_write_closed = 0; - s->read_closed = 0; - s->cancelled = 0; - s->allow_window_updates = 0; - s->published_close = 0; - s->incoming_metadata_count = 0; - s->incoming_metadata_capacity = 0; - s->incoming_metadata = NULL; s->incoming_deadline = gpr_inf_future; - memset(&s->links, 0, sizeof(s->links)); - memset(&s->included, 0, sizeof(s->included)); - grpc_sopb_init(&s->outgoing_sopb); grpc_sopb_init(&s->writing_sopb); grpc_sopb_init(&s->callback_sopb); grpc_chttp2_data_parser_init(&s->parser); @@ -647,7 +653,7 @@ static void destroy_stream(grpc_transport *gt, grpc_stream *gs) { gpr_mu_unlock(&t->mu); - grpc_sopb_destroy(&s->outgoing_sopb); + GPR_ASSERT(s->outgoing_sopb == NULL); grpc_sopb_destroy(&s->writing_sopb); grpc_sopb_destroy(&s->callback_sopb); grpc_chttp2_data_parser_destroy(&s->parser); @@ -765,6 +771,8 @@ static void unlock(transport *t) { finalize_cancellations(t); } + finish_reads(t); + /* gather any callbacks that need to be made */ if (!t->calling_back && cb) { perform_callbacks = prepare_callbacks(t); @@ -868,22 +876,23 @@ static int prepare_write(transport *t) { while (t->outgoing_window && (s = stream_list_remove_head(t, WRITABLE)) && s->outgoing_window > 0) { window_delta = grpc_chttp2_preencode( - s->outgoing_sopb.ops, &s->outgoing_sopb.nops, + s->outgoing_sopb->ops, &s->outgoing_sopb->nops, GPR_MIN(t->outgoing_window, s->outgoing_window), &s->writing_sopb); t->outgoing_window -= window_delta; s->outgoing_window -= window_delta; - s->sending_write_closed = - s->queued_write_closed && s->outgoing_sopb.nops == 0; - if (s->writing_sopb.nops > 0 || s->sending_write_closed) { + if (s->write_state == WRITE_STATE_QUEUED_CLOSE && s->outgoing_sopb->nops == 0) { + s->send_closed = 1; + } + if (s->writing_sopb.nops > 0 || s->send_closed) { stream_list_join(t, s, WRITING); } - /* if there are still writes to do and the stream still has window - available, then schedule a further write */ - if (s->outgoing_sopb.nops > 0 && s->outgoing_window > 0) { - GPR_ASSERT(!t->outgoing_window); - stream_list_add_tail(t, s, WRITABLE); + /* we should either exhaust window or have no ops left, but not both */ + GPR_ASSERT(s->outgoing_sopb->nops == 0 || s->outgoing_window <= 0); + if (s->outgoing_sopb->nops == 0) { + s->outgoing_sopb = NULL; + schedule_cb(t, s->send_done_closure, 1); } } @@ -915,10 +924,10 @@ static void finalize_outbuf(transport *t) { while ((s = stream_list_remove_head(t, WRITING))) { grpc_chttp2_encode(s->writing_sopb.ops, s->writing_sopb.nops, - s->sending_write_closed, s->id, &t->hpack_compressor, + s->send_closed, s->id, &t->hpack_compressor, &t->outbuf); s->writing_sopb.nops = 0; - if (s->sending_write_closed) { + if (s->send_closed) { stream_list_join(t, s, WRITTEN_CLOSED); } } @@ -932,8 +941,10 @@ static void finish_write_common(transport *t, int success) { drop_connection(t); } while ((s = stream_list_remove_head(t, WRITTEN_CLOSED))) { - s->sent_write_closed = 1; - if (!s->cancelled) stream_list_join(t, s, PENDING_CALLBACKS); + s->write_state = WRITE_STATE_SENT_CLOSE; + if (!s->cancelled) { + maybe_finish_read(t, s); + } } t->outbuf.count = 0; t->outbuf.length = 0; @@ -998,66 +1009,53 @@ static void perform_op(grpc_transport *gt, grpc_stream *gs, grpc_transport_op *o lock(t); if (op->send_ops) { - abort(); + GPR_ASSERT(s->outgoing_sopb == NULL); + s->send_done_closure.cb = op->on_done_send; + s->send_done_closure.user_data = op->send_user_data; + if (!s->cancelled) { + s->outgoing_sopb = op->send_ops; + if (op->is_last_send && s->write_state == WRITE_STATE_OPEN) { + s->write_state = WRITE_STATE_QUEUED_CLOSE; + } + if (s->id == 0) { + stream_list_join(t, s, WAITING_FOR_CONCURRENCY); + maybe_start_some_streams(t); + } else if (s->outgoing_window > 0) { + stream_list_join(t, s, WRITABLE); + } + } else { + schedule_nuke_sopb(t, op->send_ops); + schedule_cb(t, s->send_done_closure, 0); + } } if (op->recv_ops) { - abort(); + GPR_ASSERT(s->incoming_sopb == NULL); + s->recv_done_closure.cb = op->on_done_recv; + s->recv_done_closure.user_data = op->recv_user_data; + if (!s->cancelled) { + s->incoming_sopb = op->recv_ops; + s->incoming_sopb->nops = 0; + s->publish_state = op->recv_state; + maybe_finish_read(t, s); + maybe_join_window_updates(t, s); + } else { + schedule_cb(t, s->recv_done_closure, 0); + } } if (op->bind_pollset) { - abort(); + add_to_pollset_locked(t, op->bind_pollset); } - if (op->cancel_with_status) { - abort(); + if (op->cancel_with_status != GRPC_STATUS_OK) { + cancel_stream(t, s, op->cancel_with_status, grpc_chttp2_grpc_status_to_http2_error(op->cancel_with_status), + 1); } unlock(t); } -#if 0 -static void send_batch(grpc_transport *gt, grpc_stream *gs, grpc_stream_op *ops, - size_t ops_count, int is_last) { - transport *t = (transport *)gt; - stream *s = (stream *)gs; - - lock(t); - - if (is_last) { - s->queued_write_closed = 1; - } - if (!s->cancelled) { - grpc_sopb_append(&s->outgoing_sopb, ops, ops_count); - if (s->id == 0) { - stream_list_join(t, s, WAITING_FOR_CONCURRENCY); - maybe_start_some_streams(t); - } else { - stream_list_join(t, s, WRITABLE); - } - } else { - grpc_sopb_append(&t->nuke_later_sopb, ops, ops_count); - } - if (is_last && s->outgoing_sopb.nops == 0 && s->read_closed && - !s->published_close) { - stream_list_join(t, s, PENDING_CALLBACKS); - } - - unlock(t); -} -#endif - -static void abort_stream(grpc_transport *gt, grpc_stream *gs, - grpc_status_code status) { - transport *t = (transport *)gt; - stream *s = (stream *)gs; - - lock(t); - cancel_stream(t, s, status, grpc_chttp2_grpc_status_to_http2_error(status), - 1); - unlock(t); -} - static void send_ping(grpc_transport *gt, void (*cb)(void *user_data), void *user_data) { transport *t = (transport *)gt; @@ -1093,8 +1091,8 @@ static void finalize_cancellations(transport *t) { while ((s = stream_list_remove_head(t, CANCELLED))) { s->read_closed = 1; - s->sent_write_closed = 1; - stream_list_join(t, s, PENDING_CALLBACKS); + s->write_state = WRITE_STATE_SENT_CLOSE; + maybe_finish_read(t, s); } } @@ -1118,12 +1116,15 @@ static void cancel_stream_inner(transport *t, stream *s, gpr_uint32 id, if (s) { /* clear out any unreported input & output: nobody cares anymore */ - had_outgoing = s->outgoing_sopb.nops != 0; + had_outgoing = s->outgoing_sopb && s->outgoing_sopb->nops != 0; schedule_nuke_sopb(t, &s->parser.incoming_sopb); - schedule_nuke_sopb(t, &s->outgoing_sopb); + if (s->outgoing_sopb) { + schedule_nuke_sopb(t, s->outgoing_sopb); + schedule_cb(t, s->send_done_closure, 0); + } if (s->cancelled) { send_rst = 0; - } else if (!s->read_closed || !s->sent_write_closed || had_outgoing) { + } else if (!s->read_closed || s->write_state != WRITE_STATE_SENT_CLOSE || had_outgoing) { s->cancelled = 1; stream_list_join(t, s, CANCELLED); @@ -1141,7 +1142,7 @@ static void cancel_stream_inner(transport *t, stream *s, gpr_uint32 id, break; } - stream_list_join(t, s, PENDING_CALLBACKS); + maybe_finish_read(t, s); } } if (!id) send_rst = 0; @@ -1180,8 +1181,14 @@ static void drop_connection(transport *t) { end_all_the_calls(t); } +static void maybe_finish_read(transport *t, stream *s) { + if (s->incoming_sopb) { + stream_list_join(t, s, FINISHED_READ_OP); + } +} + static void maybe_join_window_updates(transport *t, stream *s) { - if (s->allow_window_updates && + if (s->incoming_sopb != NULL && s->incoming_window < t->settings[LOCAL_SETTINGS] [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE] * @@ -1190,6 +1197,7 @@ static void maybe_join_window_updates(transport *t, stream *s) { } } +#if 0 static void set_allow_window_updates(grpc_transport *tp, grpc_stream *sp, int allow) { transport *t = (transport *)tp; @@ -1204,6 +1212,7 @@ static void set_allow_window_updates(grpc_transport *tp, grpc_stream *sp, } unlock(t); } +#endif static grpc_chttp2_parse_error update_incoming_window(transport *t, stream *s) { if (t->incoming_frame_size > t->incoming_window) { @@ -1301,7 +1310,6 @@ static void on_header(void *tp, grpc_mdelem *md) { grpc_mdstr_as_c_string(md->key), grpc_mdstr_as_c_string(md->value))); - stream_list_join(t, s, PENDING_CALLBACKS); if (md->key == t->str_grpc_timeout) { gpr_timespec *cached_timeout = grpc_mdelem_get_user_data(md, free_timeout); if (!cached_timeout) { @@ -1320,6 +1328,7 @@ static void on_header(void *tp, grpc_mdelem *md) { } else { add_incoming_metadata(t, s, md); } + maybe_finish_read(t, s); } static int init_header_frame_parser(transport *t, int is_continuation) { @@ -1531,14 +1540,14 @@ static int parse_frame_slice(transport *t, gpr_slice slice, int is_last) { case GRPC_CHTTP2_PARSE_OK: if (st.end_of_stream) { t->incoming_stream->read_closed = 1; - stream_list_join(t, t->incoming_stream, PENDING_CALLBACKS); + maybe_finish_read(t, t->incoming_stream); } if (st.need_flush_reads) { - stream_list_join(t, t->incoming_stream, PENDING_CALLBACKS); + maybe_finish_read(t, t->incoming_stream); } if (st.metadata_boundary) { add_metadata_batch(t, t->incoming_stream); - stream_list_join(t, t->incoming_stream, PENDING_CALLBACKS); + maybe_finish_read(t, t->incoming_stream); } if (st.ack_settings) { gpr_slice_buffer_add(&t->qbuf, grpc_chttp2_settings_ack_create()); @@ -1579,7 +1588,7 @@ static int parse_frame_slice(transport *t, gpr_slice slice, int is_last) { int was_window_empty = s->outgoing_window <= 0; s->outgoing_window += st.initial_window_update; if (was_window_empty && s->outgoing_window > 0 && - s->outgoing_sopb.nops > 0) { + s->outgoing_sopb && s->outgoing_sopb->nops > 0) { stream_list_join(t, s, WRITABLE); } } @@ -1598,7 +1607,7 @@ static int parse_frame_slice(transport *t, gpr_slice slice, int is_last) { s->outgoing_window += st.window_update; /* if this window update makes outgoing ops writable again, flag that */ - if (was_window_empty && s->outgoing_sopb.nops) { + if (was_window_empty && s->outgoing_sopb && s->outgoing_sopb->nops > 0) { stream_list_join(t, s, WRITABLE); } } @@ -1860,45 +1869,49 @@ static grpc_stream_state compute_state(gpr_uint8 write_closed, return GRPC_STREAM_OPEN; } +static void finish_reads(transport *t) { + stream *s; + + while ((s = stream_list_remove_head(t, FINISHED_READ_OP)) != NULL) { + int publish = 0; + GPR_ASSERT(s->incoming_sopb); + *s->publish_state = compute_state(s->write_state == WRITE_STATE_SENT_CLOSE, + s->read_closed); + if (*s->publish_state != s->published_state) { + s->published_state = *s->publish_state; + publish = 1; + } + if (s->parser.incoming_sopb.nops > 0) { + grpc_sopb_swap(s->incoming_sopb, &s->parser.incoming_sopb); + publish = 1; + } + if (publish) { + schedule_cb(t, s->recv_done_closure, 1); + } + } +} + +static void schedule_cb(transport *t, op_closure closure, int success) { + if (t->pending_callbacks.capacity == t->pending_callbacks.count) { + t->pending_callbacks.capacity = GPR_MAX(t->pending_callbacks.capacity * 2, 8); + t->pending_callbacks.callbacks = gpr_realloc(t->pending_callbacks.callbacks, t->pending_callbacks.capacity * sizeof(*t->pending_callbacks.callbacks)); + } + closure.success = success; + t->pending_callbacks.callbacks[t->pending_callbacks.count++] = closure; +} + static int prepare_callbacks(transport *t) { op_closure_array temp = t->pending_callbacks; t->pending_callbacks = t->executing_callbacks; t->executing_callbacks = temp; return t->executing_callbacks.count > 0; - -#if 0 - stream *s; - int n = 0; - while ((s = stream_list_remove_head(t, PENDING_CALLBACKS))) { - int execute = 1; - - s->callback_state = compute_state(s->sent_write_closed, s->read_closed); - if (s->callback_state == GRPC_STREAM_CLOSED) { - remove_from_stream_map(t, s); - if (s->published_close) { - execute = 0; - } else if (s->incoming_metadata_count) { - add_metadata_batch(t, s); - } - s->published_close = 1; - } - - grpc_sopb_swap(&s->parser.incoming_sopb, &s->callback_sopb); - - if (execute) { - stream_list_add_tail(t, s, EXECUTING_CALLBACKS); - n = 1; - } - } - return n; -#endif } static void run_callbacks(transport *t, const grpc_transport_callbacks *cb) { size_t i; for (i = 0; i < t->executing_callbacks.count; i++) { op_closure c = t->executing_callbacks.callbacks[i]; - c.cb(c.user_data, c.status); + c.cb(c.user_data, c.success); } t->executing_callbacks.count = 0; } @@ -1907,12 +1920,20 @@ static void call_cb_closed(transport *t, const grpc_transport_callbacks *cb) { cb->closed(t->cb_user_data, &t->base); } -static void add_to_pollset(grpc_transport *gt, grpc_pollset *pollset) { - transport *t = (transport *)gt; - lock(t); +/* + * POLLSET STUFF + */ + +static void add_to_pollset_locked(transport *t, grpc_pollset *pollset) { if (t->ep) { grpc_endpoint_add_to_pollset(t->ep, pollset); } +} + +static void add_to_pollset(grpc_transport *gt, grpc_pollset *pollset) { + transport *t = (transport *)gt; + lock(t); + add_to_pollset_locked(t, pollset); unlock(t); } -- cgit v1.2.3 From fbf5be26a3ab98115eec118a9858b000dd9fc045 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Wed, 22 Apr 2015 16:17:09 -0700 Subject: One test compiles --- src/core/surface/call.c | 36 +------- src/core/surface/channel.c | 4 + src/core/transport/chttp2_transport.c | 5 +- src/core/transport/stream_op.c | 14 +-- src/core/transport/stream_op.h | 6 +- src/core/transport/transport.c | 16 +--- src/core/transport/transport_op_string.c | 115 +++++++++++++----------- test/core/end2end/fixtures/chttp2_socket_pair.c | 6 +- 8 files changed, 80 insertions(+), 122 deletions(-) (limited to 'src/core/transport/chttp2_transport.c') diff --git a/src/core/surface/call.c b/src/core/surface/call.c index 7fcf6e2b04..18be81308d 100644 --- a/src/core/surface/call.c +++ b/src/core/surface/call.c @@ -283,6 +283,8 @@ grpc_call *grpc_call_create(grpc_channel *channel, grpc_completion_queue *cq, call->send_deadline = send_deadline; grpc_channel_internal_ref(channel); call->metadata_context = grpc_channel_get_metadata_context(channel); + grpc_sopb_init(&call->send_ops); + grpc_sopb_init(&call->recv_ops); /* one ref is dropped in response to destroy, the other in stream_closed */ gpr_ref_init(&call->internal_refcount, 2); @@ -330,6 +332,8 @@ static void destroy_call(void *call, int ignored_success) { destroy_legacy_state(c->legacy_state); } grpc_bbq_destroy(&c->incoming_queue); + grpc_sopb_destroy(&c->send_ops); + grpc_sopb_destroy(&c->recv_ops); gpr_free(c); } @@ -1091,41 +1095,9 @@ static void recv_metadata(grpc_call *call, grpc_metadata_batch *md) { grpc_mdctx_unlock(mdctx); } -#if 0 -void grpc_call_read_closed(grpc_call_element *elem) { - set_read_state(CALL_FROM_TOP_ELEM(elem), READ_STATE_READ_CLOSED); -} - -void grpc_call_stream_closed(grpc_call_element *elem) { - grpc_call *call = CALL_FROM_TOP_ELEM(elem); - set_read_state(call, READ_STATE_STREAM_CLOSED); - grpc_call_internal_unref(call, 0); -} - -void grpc_call_recv_message(grpc_call_element *elem, - grpc_byte_buffer *byte_buffer) { - grpc_call *call = CALL_FROM_TOP_ELEM(elem); - lock(call); - grpc_bbq_push(&call->incoming_queue, byte_buffer); - finish_read_ops(call); - unlock(call); -} - -void grpc_call_recv_synthetic_status(grpc_call_element *elem, - grpc_status_code status, - const char *message) { - grpc_call *call = CALL_FROM_TOP_ELEM(elem); - lock(call); - set_status_code(call, STATUS_FROM_CORE, status); - set_status_details(call, STATUS_FROM_CORE, - grpc_mdstr_from_string(call->metadata_context, message)); - unlock(call); -} - grpc_call_stack *grpc_call_get_call_stack(grpc_call *call) { return CALL_STACK_FROM_CALL(call); } -#endif /* * BATCH API IMPLEMENTATION diff --git a/src/core/surface/channel.c b/src/core/surface/channel.c index f1d71afaf2..de2f354c78 100644 --- a/src/core/surface/channel.c +++ b/src/core/surface/channel.c @@ -242,3 +242,7 @@ grpc_mdstr *grpc_channel_get_status_string(grpc_channel *channel) { grpc_mdstr *grpc_channel_get_message_string(grpc_channel *channel) { return channel->grpc_message_string; } + +gpr_uint32 grpc_channel_get_max_message_length(grpc_channel *channel) { + return channel->max_message_length; +} \ No newline at end of file diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index fed3088789..9c2af560c1 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -1503,8 +1503,6 @@ static int is_window_update_legal(gpr_int64 window_update, gpr_int64 window) { return window + window_update < MAX_WINDOW; } -static void free_md(void *p, grpc_op_error result) { gpr_free(p); } - static void add_metadata_batch(transport *t, stream *s) { grpc_metadata_batch b; size_t i; @@ -1522,8 +1520,7 @@ static void add_metadata_batch(transport *t, stream *s) { s->incoming_metadata[s->incoming_metadata_count - 1].next = NULL; grpc_sopb_add_metadata(&s->parser.incoming_sopb, b); - grpc_sopb_add_flow_ctl_cb(&s->parser.incoming_sopb, free_md, - s->incoming_metadata); + /* TODO(ctiller): don't leak incoming_metadata */ /* reset */ s->incoming_deadline = gpr_inf_future; diff --git a/src/core/transport/stream_op.c b/src/core/transport/stream_op.c index 882c078d51..ea22b0e1c8 100644 --- a/src/core/transport/stream_op.c +++ b/src/core/transport/stream_op.c @@ -81,9 +81,6 @@ void grpc_stream_ops_unref_owned_objects(grpc_stream_op *ops, size_t nops) { case GRPC_OP_METADATA: grpc_metadata_batch_destroy(&ops[i].data.metadata); break; - case GRPC_OP_FLOW_CTL_CB: - ops[i].data.flow_ctl_cb.cb(ops[i].data.flow_ctl_cb.arg, GRPC_OP_ERROR); - break; case GRPC_NO_OP: case GRPC_OP_BEGIN_MESSAGE: break; @@ -119,6 +116,7 @@ static grpc_stream_op *add(grpc_stream_op_buffer *sopb) { assert_contained_metadata_ok(sopb->ops, sopb->nops); + GPR_ASSERT(sopb->nops <= sopb->capacity); if (sopb->nops == sopb->capacity) { expandto(sopb, GROW(sopb->capacity)); } @@ -158,16 +156,6 @@ void grpc_sopb_add_slice(grpc_stream_op_buffer *sopb, gpr_slice slice) { assert_contained_metadata_ok(sopb->ops, sopb->nops); } -void grpc_sopb_add_flow_ctl_cb(grpc_stream_op_buffer *sopb, - void (*cb)(void *arg, grpc_op_error error), - void *arg) { - grpc_stream_op *op = add(sopb); - op->type = GRPC_OP_FLOW_CTL_CB; - op->data.flow_ctl_cb.cb = cb; - op->data.flow_ctl_cb.arg = arg; - assert_contained_metadata_ok(sopb->ops, sopb->nops); -} - void grpc_sopb_append(grpc_stream_op_buffer *sopb, grpc_stream_op *ops, size_t nops) { size_t orig_nops = sopb->nops; diff --git a/src/core/transport/stream_op.h b/src/core/transport/stream_op.h index c3901bf608..f5de64d583 100644 --- a/src/core/transport/stream_op.h +++ b/src/core/transport/stream_op.h @@ -154,12 +154,10 @@ void grpc_sopb_add_begin_message(grpc_stream_op_buffer *sopb, gpr_uint32 length, void grpc_sopb_add_metadata(grpc_stream_op_buffer *sopb, grpc_metadata_batch metadata); /* Append a GRPC_SLICE to a buffer - does not ref/unref the slice */ void grpc_sopb_add_slice(grpc_stream_op_buffer *sopb, gpr_slice slice); -/* Append a GRPC_OP_FLOW_CTL_CB to a buffer */ -void grpc_sopb_add_flow_ctl_cb(grpc_stream_op_buffer *sopb, - void (*cb)(void *arg, grpc_op_error error), - void *arg); /* Append a buffer to a buffer - does not ref/unref any internal objects */ void grpc_sopb_append(grpc_stream_op_buffer *sopb, grpc_stream_op *ops, size_t nops); +char *grpc_sopb_string(grpc_stream_op_buffer *sopb); + #endif /* GRPC_INTERNAL_CORE_TRANSPORT_STREAM_OP_H */ diff --git a/src/core/transport/transport.c b/src/core/transport/transport.c index ef0020dc58..35195348e7 100644 --- a/src/core/transport/transport.c +++ b/src/core/transport/transport.c @@ -56,14 +56,9 @@ int grpc_transport_init_stream(grpc_transport *transport, grpc_stream *stream, return transport->vtable->init_stream(transport, stream, server_data); } -void grpc_transport_send_batch(grpc_transport *transport, grpc_stream *stream, - grpc_stream_op *ops, size_t nops, int is_last) { - transport->vtable->send_batch(transport, stream, ops, nops, is_last); -} - -void grpc_transport_set_allow_window_updates(grpc_transport *transport, - grpc_stream *stream, int allow) { - transport->vtable->set_allow_window_updates(transport, stream, allow); +void grpc_transport_perform_op(grpc_transport *transport, grpc_stream *stream, + grpc_transport_op *op) { + transport->vtable->perform_op(transport, stream, op); } void grpc_transport_add_to_pollset(grpc_transport *transport, @@ -76,11 +71,6 @@ void grpc_transport_destroy_stream(grpc_transport *transport, transport->vtable->destroy_stream(transport, stream); } -void grpc_transport_abort_stream(grpc_transport *transport, grpc_stream *stream, - grpc_status_code status) { - transport->vtable->abort_stream(transport, stream, status); -} - void grpc_transport_ping(grpc_transport *transport, void (*cb)(void *user_data), void *user_data) { transport->vtable->ping(transport, cb, user_data); diff --git a/src/core/transport/transport_op_string.c b/src/core/transport/transport_op_string.c index 5f7e1be268..e886690234 100644 --- a/src/core/transport/transport_op_string.c +++ b/src/core/transport/transport_op_string.c @@ -66,65 +66,76 @@ static void put_metadata_list(gpr_strvec *b, grpc_metadata_batch md) { } } -char *grpc_call_op_string(grpc_call_op *op) { +char *grpc_sopb_string(grpc_stream_op_buffer *sopb) { + char *out; + char *tmp; + size_t i; + gpr_strvec b; + gpr_strvec_init(&b); + + for (i = 0; i < sopb->nops; i++) { + grpc_stream_op *op = &sopb->ops[i]; + if (i) gpr_strvec_add(&b, gpr_strdup(", ")); + switch (op->type) { + case GRPC_NO_OP: + gpr_strvec_add(&b, gpr_strdup("NO_OP")); + break; + case GRPC_OP_BEGIN_MESSAGE: + gpr_asprintf(&tmp, "BEGIN_MESSAGE:%d", op->data.begin_message.length); + gpr_strvec_add(&b, tmp); + break; + case GRPC_OP_SLICE: + gpr_asprintf(&tmp, "SLICE:%d", GPR_SLICE_LENGTH(op->data.slice)); + break; + case GRPC_OP_METADATA: + put_metadata_list(&b, op->data.metadata); + break; + } + } + + out = gpr_strvec_flatten(&b, NULL); + gpr_strvec_destroy(&b); + + return out; +} + +char *grpc_transport_op_string(grpc_transport_op *op) { char *tmp; char *out; + int first = 1; gpr_strvec b; gpr_strvec_init(&b); - switch (op->dir) { - case GRPC_CALL_DOWN: - gpr_strvec_add(&b, gpr_strdup(">")); - break; - case GRPC_CALL_UP: - gpr_strvec_add(&b, gpr_strdup("<")); - break; + if (op->send_ops) { + if (!first) gpr_strvec_add(&b, gpr_strdup(" ")); + first = 0; + gpr_strvec_add(&b, gpr_strdup("SEND")); + if (op->is_last_send) { + gpr_strvec_add(&b, gpr_strdup("_LAST")); + } + gpr_strvec_add(&b, gpr_strdup("[")); + gpr_strvec_add(&b, grpc_sopb_string(op->send_ops)); + gpr_strvec_add(&b, gpr_strdup("]")); } - switch (op->type) { - case GRPC_SEND_METADATA: - gpr_strvec_add(&b, gpr_strdup("SEND_METADATA")); - put_metadata_list(&b, op->data.metadata); - break; - case GRPC_SEND_MESSAGE: - gpr_strvec_add(&b, gpr_strdup("SEND_MESSAGE")); - break; - case GRPC_SEND_PREFORMATTED_MESSAGE: - gpr_strvec_add(&b, gpr_strdup("SEND_PREFORMATTED_MESSAGE")); - break; - case GRPC_SEND_FINISH: - gpr_strvec_add(&b, gpr_strdup("SEND_FINISH")); - break; - case GRPC_REQUEST_DATA: - gpr_strvec_add(&b, gpr_strdup("REQUEST_DATA")); - break; - case GRPC_RECV_METADATA: - gpr_strvec_add(&b, gpr_strdup("RECV_METADATA")); - put_metadata_list(&b, op->data.metadata); - break; - case GRPC_RECV_MESSAGE: - gpr_strvec_add(&b, gpr_strdup("RECV_MESSAGE")); - break; - case GRPC_RECV_HALF_CLOSE: - gpr_strvec_add(&b, gpr_strdup("RECV_HALF_CLOSE")); - break; - case GRPC_RECV_FINISH: - gpr_strvec_add(&b, gpr_strdup("RECV_FINISH")); - break; - case GRPC_RECV_SYNTHETIC_STATUS: - gpr_asprintf(&tmp, "RECV_SYNTHETIC_STATUS status=%d message='%s'", - op->data.synthetic_status.status, - op->data.synthetic_status.message); - gpr_strvec_add(&b, tmp); - break; - case GRPC_CANCEL_OP: - gpr_strvec_add(&b, gpr_strdup("CANCEL_OP")); - break; + + if (op->recv_ops) { + if (!first) gpr_strvec_add(&b, gpr_strdup(" ")); + first = 0; + gpr_strvec_add(&b, gpr_strdup("RECV")); } - gpr_asprintf(&tmp, " flags=0x%08x", op->flags); - gpr_strvec_add(&b, tmp); + if (op->bind_pollset) { - gpr_strvec_add(&b, gpr_strdup("bind_pollset")); + if (!first) gpr_strvec_add(&b, gpr_strdup(" ")); + first = 0; + gpr_strvec_add(&b, gpr_strdup("BIND")); + } + + if (op->cancel_with_status != GRPC_STATUS_OK) { + if (!first) gpr_strvec_add(&b, gpr_strdup(" ")); + first = 0; + gpr_asprintf(&tmp, "CANCEL:%d", op->cancel_with_status); + gpr_strvec_add(&b, tmp); } out = gpr_strvec_flatten(&b, NULL); @@ -134,8 +145,8 @@ char *grpc_call_op_string(grpc_call_op *op) { } void grpc_call_log_op(char *file, int line, gpr_log_severity severity, - grpc_call_element *elem, grpc_call_op *op) { - char *str = grpc_call_op_string(op); + grpc_call_element *elem, grpc_transport_op *op) { + char *str = grpc_transport_op_string(op); gpr_log(file, line, severity, "OP[%s:%p]: %s", elem->filter->name, elem, str); gpr_free(str); } diff --git a/test/core/end2end/fixtures/chttp2_socket_pair.c b/test/core/end2end/fixtures/chttp2_socket_pair.c index 1225f7db0c..acc0a69708 100644 --- a/test/core/end2end/fixtures/chttp2_socket_pair.c +++ b/test/core/end2end/fixtures/chttp2_socket_pair.c @@ -37,7 +37,6 @@ #include "src/core/channel/client_channel.h" #include "src/core/channel/connected_channel.h" -#include "src/core/channel/http_filter.h" #include "src/core/channel/http_client_filter.h" #include "src/core/channel/http_server_filter.h" #include "src/core/iomgr/endpoint_pair.h" @@ -60,8 +59,7 @@ static grpc_transport_setup_result server_setup_transport( void *ts, grpc_transport *transport, grpc_mdctx *mdctx) { grpc_end2end_test_fixture *f = ts; - static grpc_channel_filter const *extra_filters[] = {&grpc_http_server_filter, - &grpc_http_filter}; + static grpc_channel_filter const *extra_filters[] = {&grpc_http_server_filter}; return grpc_server_setup_transport(f->server, transport, extra_filters, GPR_ARRAY_SIZE(extra_filters), mdctx); } @@ -76,7 +74,7 @@ static grpc_transport_setup_result client_setup_transport( sp_client_setup *cs = ts; const grpc_channel_filter *filters[] = { - &grpc_client_surface_filter, &grpc_http_client_filter, &grpc_http_filter, + &grpc_client_surface_filter, &grpc_http_client_filter, &grpc_connected_channel_filter}; size_t nfilters = sizeof(filters) / sizeof(*filters); grpc_channel *channel = grpc_channel_create_from_filters( -- cgit v1.2.3 From 3f2c2214b718ee61ddfd52efe922e652535aa537 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Thu, 23 Apr 2015 07:56:33 -0700 Subject: Fiddling with an initial op --- src/core/channel/census_filter.c | 30 +++++++++++++++++------------ src/core/channel/channel_stack.c | 3 ++- src/core/channel/channel_stack.h | 4 +++- src/core/channel/child_channel.c | 6 +++--- src/core/channel/child_channel.h | 2 +- src/core/channel/client_channel.c | 3 ++- src/core/channel/connected_channel.c | 7 ++++--- src/core/transport/chttp2_transport.c | 1 + src/core/transport/transport.h | 36 +++++++++++++++++------------------ 9 files changed, 52 insertions(+), 40 deletions(-) (limited to 'src/core/transport/chttp2_transport.c') diff --git a/src/core/channel/census_filter.c b/src/core/channel/census_filter.c index 3e0fc39fc9..47461f7f2b 100644 --- a/src/core/channel/census_filter.c +++ b/src/core/channel/census_filter.c @@ -82,15 +82,18 @@ static void extract_and_annotate_method_tag(grpc_stream_op_buffer* sopb, call_da } } -static void client_start_transport_op(grpc_call_element* elem, grpc_transport_op* op) { +static void client_mutate_op(grpc_call_element *elem, grpc_transport_op *op) { call_data* calld = elem->call_data; channel_data* chand = elem->channel_data; - GPR_ASSERT(calld != NULL); - GPR_ASSERT(chand != NULL); - GPR_ASSERT((calld->op_id.upper != 0) || (calld->op_id.lower != 0)); if (op->send_ops) { extract_and_annotate_method_tag(op->send_ops, calld, chand); } +} + +static void client_start_transport_op(grpc_call_element* elem, grpc_transport_op* op) { + call_data* calld = elem->call_data; + GPR_ASSERT((calld->op_id.upper != 0) || (calld->op_id.lower != 0)); + client_mutate_op(elem, op); grpc_call_next_op(elem, op); } @@ -104,12 +107,8 @@ static void server_on_done_recv(void *ptr, int success) { calld->on_done_recv(calld->recv_user_data, success); } -static void server_start_transport_op(grpc_call_element* elem, grpc_transport_op* op) { +static void server_mutate_op(grpc_call_element *elem, grpc_transport_op *op) { call_data* calld = elem->call_data; - channel_data* chand = elem->channel_data; - GPR_ASSERT(calld != NULL); - GPR_ASSERT(chand != NULL); - GPR_ASSERT((calld->op_id.upper != 0) || (calld->op_id.lower != 0)); if (op->recv_ops) { /* substitute our callback for the op callback */ calld->recv_ops = op->recv_ops; @@ -118,7 +117,12 @@ static void server_start_transport_op(grpc_call_element* elem, grpc_transport_op op->on_done_recv = server_on_done_recv; op->recv_user_data = elem; } - /* Always pass control up or down the stack depending on op->dir */ +} + +static void server_start_transport_op(grpc_call_element* elem, grpc_transport_op* op) { + call_data* calld = elem->call_data; + GPR_ASSERT((calld->op_id.upper != 0) || (calld->op_id.lower != 0)); + server_mutate_op(elem, op); grpc_call_next_op(elem, op); } @@ -136,12 +140,13 @@ static void channel_op(grpc_channel_element* elem, } static void client_init_call_elem(grpc_call_element* elem, - const void* server_transport_data) { + const void* server_transport_data, grpc_transport_op *initial_op) { call_data* d = elem->call_data; GPR_ASSERT(d != NULL); init_rpc_stats(&d->stats); d->start_ts = gpr_now(); d->op_id = census_tracing_start_op(); + if (initial_op) client_mutate_op(elem, initial_op); } static void client_destroy_call_elem(grpc_call_element* elem) { @@ -152,12 +157,13 @@ static void client_destroy_call_elem(grpc_call_element* elem) { } static void server_init_call_elem(grpc_call_element* elem, - const void* server_transport_data) { + const void* server_transport_data, grpc_transport_op *initial_op) { call_data* d = elem->call_data; GPR_ASSERT(d != NULL); init_rpc_stats(&d->stats); d->start_ts = gpr_now(); d->op_id = census_tracing_start_op(); + if (initial_op) server_mutate_op(elem, initial_op); } static void server_destroy_call_elem(grpc_call_element* elem) { diff --git a/src/core/channel/channel_stack.c b/src/core/channel/channel_stack.c index c121e27005..022100e8bd 100644 --- a/src/core/channel/channel_stack.c +++ b/src/core/channel/channel_stack.c @@ -148,6 +148,7 @@ void grpc_channel_stack_destroy(grpc_channel_stack *stack) { void grpc_call_stack_init(grpc_channel_stack *channel_stack, const void *transport_server_data, + grpc_transport_op *initial_op, grpc_call_stack *call_stack) { grpc_channel_element *channel_elems = CHANNEL_ELEMS_FROM_STACK(channel_stack); size_t count = channel_stack->count; @@ -165,7 +166,7 @@ void grpc_call_stack_init(grpc_channel_stack *channel_stack, call_elems[i].filter = channel_elems[i].filter; call_elems[i].channel_data = channel_elems[i].channel_data; call_elems[i].call_data = user_data; - call_elems[i].filter->init_call_elem(&call_elems[i], transport_server_data); + call_elems[i].filter->init_call_elem(&call_elems[i], transport_server_data, initial_op); user_data += ROUND_UP_TO_ALIGNMENT_SIZE(call_elems[i].filter->sizeof_call_data); } diff --git a/src/core/channel/channel_stack.h b/src/core/channel/channel_stack.h index 75897ff651..94b12639fc 100644 --- a/src/core/channel/channel_stack.h +++ b/src/core/channel/channel_stack.h @@ -121,7 +121,8 @@ typedef struct { transport and is on the server. Most filters want to ignore this argument.*/ void (*init_call_elem)(grpc_call_element *elem, - const void *server_transport_data); + const void *server_transport_data, + grpc_transport_op *initial_op); /* Destroy per call data. The filter does not need to do any chaining */ void (*destroy_call_elem)(grpc_call_element *elem); @@ -200,6 +201,7 @@ void grpc_channel_stack_destroy(grpc_channel_stack *stack); server. */ void grpc_call_stack_init(grpc_channel_stack *channel_stack, const void *transport_server_data, + grpc_transport_op *initial_op, grpc_call_stack *call_stack); /* Destroy a call stack */ void grpc_call_stack_destroy(grpc_call_stack *stack); diff --git a/src/core/channel/child_channel.c b/src/core/channel/child_channel.c index 244417384a..817a2a8c70 100644 --- a/src/core/channel/child_channel.c +++ b/src/core/channel/child_channel.c @@ -121,7 +121,7 @@ static void lb_channel_op(grpc_channel_element *elem, /* Constructor for call_data */ static void lb_init_call_elem(grpc_call_element *elem, - const void *server_transport_data) {} + const void *server_transport_data, grpc_transport_op *initial_op) {} /* Destructor for call_data */ static void lb_destroy_call_elem(grpc_call_element *elem) {} @@ -261,13 +261,13 @@ void grpc_child_channel_handle_op(grpc_child_channel *channel, } grpc_child_call *grpc_child_channel_create_call(grpc_child_channel *channel, - grpc_call_element *parent) { + grpc_call_element *parent, grpc_transport_op *initial_op) { grpc_call_stack *stk = gpr_malloc((channel)->call_stack_size); grpc_call_element *lbelem; lb_call_data *lbcalld; lb_channel_data *lbchand; - grpc_call_stack_init(channel, NULL, stk); + grpc_call_stack_init(channel, NULL, initial_op, stk); lbelem = LINK_BACK_ELEM_FROM_CALL(stk); lbchand = lbelem->channel_data; lbcalld = lbelem->call_data; diff --git a/src/core/channel/child_channel.h b/src/core/channel/child_channel.h index 38695402ab..264a8bbb82 100644 --- a/src/core/channel/child_channel.h +++ b/src/core/channel/child_channel.h @@ -57,7 +57,7 @@ void grpc_child_channel_destroy(grpc_child_channel *channel, int wait_for_callbacks); grpc_child_call *grpc_child_channel_create_call(grpc_child_channel *channel, - grpc_call_element *parent); + grpc_call_element *parent, grpc_transport_op *initial_op); grpc_call_element *grpc_child_call_get_top_element(grpc_child_call *call); void grpc_child_call_destroy(grpc_child_call *call); diff --git a/src/core/channel/client_channel.c b/src/core/channel/client_channel.c index 6ad50cb944..e6b0f7bba8 100644 --- a/src/core/channel/client_channel.c +++ b/src/core/channel/client_channel.c @@ -105,7 +105,8 @@ static int prepare_activate(grpc_call_element *elem, calld->state = CALL_ACTIVE; /* create a child call */ - calld->s.active.child_call = grpc_child_channel_create_call(on_child, elem); + /* TODO(ctiller): pass the waiting op down here */ + calld->s.active.child_call = grpc_child_channel_create_call(on_child, elem, NULL); return 1; } diff --git a/src/core/channel/connected_channel.c b/src/core/channel/connected_channel.c index 9e2d92ffbc..5a5a849907 100644 --- a/src/core/channel/connected_channel.c +++ b/src/core/channel/connected_channel.c @@ -95,15 +95,16 @@ static void channel_op(grpc_channel_element *elem, /* Constructor for call_data */ static void init_call_elem(grpc_call_element *elem, - const void *server_transport_data) { + const void *server_transport_data, + grpc_transport_op *initial_op) { call_data *calld = elem->call_data; channel_data *chand = elem->channel_data; int r; GPR_ASSERT(elem->filter == &grpc_connected_channel_filter); - r = grpc_transport_init_stream(chand->transport, + r = grpc_transport_1chand->transport, TRANSPORT_STREAM_FROM_CALL_DATA(calld), - server_transport_data); + server_transport_data, initial_op); GPR_ASSERT(r == 0); } diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index 9c2af560c1..acdc98b86b 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -607,6 +607,7 @@ static int init_stream(grpc_transport *gt, grpc_stream *gs, lock(t); s->id = 0; } else { + /* already locked */ s->id = (gpr_uint32)(gpr_uintptr)server_data; t->incoming_stream = s; grpc_chttp2_stream_map_add(&t->stream_map, s->id, s); diff --git a/src/core/transport/transport.h b/src/core/transport/transport.h index d0007680e3..a51e01d3c9 100644 --- a/src/core/transport/transport.h +++ b/src/core/transport/transport.h @@ -60,6 +60,23 @@ typedef enum grpc_stream_state { GRPC_STREAM_CLOSED } grpc_stream_state; +/* Transport op: a set of operations to perform on a transport */ +typedef struct grpc_transport_op { + grpc_stream_op_buffer *send_ops; + int is_last_send; + void (*on_done_send)(void *user_data, int success); + void *send_user_data; + + grpc_stream_op_buffer *recv_ops; + grpc_stream_state *recv_state; + void (*on_done_recv)(void *user_data, int success); + void *recv_user_data; + + grpc_pollset *bind_pollset; + + grpc_status_code cancel_with_status; +} grpc_transport_op; + /* Callbacks made from the transport to the upper layers of grpc. */ struct grpc_transport_callbacks { /* Initialize a new stream on behalf of the transport. @@ -98,7 +115,7 @@ size_t grpc_transport_stream_size(grpc_transport *transport); server_data - either NULL for a client initiated stream, or a pointer supplied from the accept_stream callback function */ int grpc_transport_init_stream(grpc_transport *transport, grpc_stream *stream, - const void *server_data); + const void *server_data, grpc_transport_op *initial_op); /* Destroy transport data for a stream. @@ -113,23 +130,6 @@ int grpc_transport_init_stream(grpc_transport *transport, grpc_stream *stream, void grpc_transport_destroy_stream(grpc_transport *transport, grpc_stream *stream); -/* Transport op: a set of operations to perform on a transport */ -typedef struct grpc_transport_op { - grpc_stream_op_buffer *send_ops; - int is_last_send; - void (*on_done_send)(void *user_data, int success); - void *send_user_data; - - grpc_stream_op_buffer *recv_ops; - grpc_stream_state *recv_state; - void (*on_done_recv)(void *user_data, int success); - void *recv_user_data; - - grpc_pollset *bind_pollset; - - grpc_status_code cancel_with_status; -} grpc_transport_op; - void grpc_transport_op_finish_with_failure(grpc_transport_op *op); /* TODO(ctiller): remove this */ -- cgit v1.2.3 From 50d9db534c202a2a473d7b5d54f105174ec7f727 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Thu, 23 Apr 2015 10:52:14 -0700 Subject: Flesh out initial_op --- src/core/channel/client_channel.c | 5 ++- src/core/channel/connected_channel.c | 2 +- src/core/channel/http_client_filter.c | 12 ++++--- src/core/channel/http_server_filter.c | 59 +++++------------------------------ src/core/channel/noop_filter.c | 22 ++++++++----- src/core/surface/call.c | 2 +- src/core/surface/client.c | 2 +- src/core/surface/lame_client.c | 6 +++- src/core/surface/server.c | 11 +++++-- src/core/transport/chttp2_transport.c | 20 +++++++----- src/core/transport/transport.c | 4 +-- src/core/transport/transport_impl.h | 2 +- 12 files changed, 65 insertions(+), 82 deletions(-) (limited to 'src/core/transport/chttp2_transport.c') diff --git a/src/core/channel/client_channel.c b/src/core/channel/client_channel.c index e6b0f7bba8..77c4951038 100644 --- a/src/core/channel/client_channel.c +++ b/src/core/channel/client_channel.c @@ -325,9 +325,12 @@ static void channel_op(grpc_channel_element *elem, /* Constructor for call_data */ static void init_call_elem(grpc_call_element *elem, - const void *server_transport_data) { + const void *server_transport_data, grpc_transport_op *initial_op) { call_data *calld = elem->call_data; + /* TODO(ctiller): is there something useful we can do here? */ + GPR_ASSERT(initial_op == NULL); + GPR_ASSERT(elem->filter == &grpc_client_channel_filter); GPR_ASSERT(server_transport_data == NULL); calld->elem = elem; diff --git a/src/core/channel/connected_channel.c b/src/core/channel/connected_channel.c index 5a5a849907..9b7db6124a 100644 --- a/src/core/channel/connected_channel.c +++ b/src/core/channel/connected_channel.c @@ -102,7 +102,7 @@ static void init_call_elem(grpc_call_element *elem, int r; GPR_ASSERT(elem->filter == &grpc_connected_channel_filter); - r = grpc_transport_1chand->transport, + r = grpc_transport_init_stream(chand->transport, TRANSPORT_STREAM_FROM_CALL_DATA(calld), server_transport_data, initial_op); GPR_ASSERT(r == 0); diff --git a/src/core/channel/http_client_filter.c b/src/core/channel/http_client_filter.c index 7287408401..45a7436361 100644 --- a/src/core/channel/http_client_filter.c +++ b/src/core/channel/http_client_filter.c @@ -87,13 +87,11 @@ static void hc_on_recv(void *user_data, int success) { calld->on_done_recv(calld->recv_user_data, success); } -static void hc_start_transport_op(grpc_call_element *elem, grpc_transport_op *op) { +static void hc_mutate_op(grpc_call_element *elem, grpc_transport_op *op) { /* grab pointers to our data from the call element */ call_data *calld = elem->call_data; channel_data *channeld = elem->channel_data; size_t i; - GRPC_CALL_LOG_OP(GPR_INFO, elem, op); - if (op->send_ops && !calld->sent_initial_metadata) { size_t nops = op->send_ops->nops; grpc_stream_op *ops = op->send_ops->ops; @@ -123,7 +121,11 @@ static void hc_start_transport_op(grpc_call_element *elem, grpc_transport_op *op op->on_done_recv = hc_on_recv; op->recv_user_data = elem; } +} +static void hc_start_transport_op(grpc_call_element *elem, grpc_transport_op *op) { + GRPC_CALL_LOG_OP(GPR_INFO, elem, op); + hc_mutate_op(elem, op); grpc_call_next_op(elem, op); } @@ -146,10 +148,12 @@ static void channel_op(grpc_channel_element *elem, /* Constructor for call_data */ static void init_call_elem(grpc_call_element *elem, - const void *server_transport_data) { + const void *server_transport_data, + grpc_transport_op *initial_op) { call_data *calld = elem->call_data; calld->sent_initial_metadata = 0; calld->got_initial_metadata = 0; + if (initial_op) hc_mutate_op(elem, initial_op); } /* Destructor for call_data */ diff --git a/src/core/channel/http_server_filter.c b/src/core/channel/http_server_filter.c index 31ed8a3393..e5174acd68 100644 --- a/src/core/channel/http_server_filter.c +++ b/src/core/channel/http_server_filter.c @@ -38,12 +38,6 @@ #include #include -typedef struct { - grpc_mdelem *path; - grpc_mdelem *content_type; - grpc_byte_buffer *content; -} gettable; - typedef struct call_data { gpr_uint8 got_initial_metadata; gpr_uint8 seen_path; @@ -73,9 +67,6 @@ typedef struct channel_data { grpc_mdstr *host_key; grpc_mdctx *mdctx; - - size_t gettable_count; - gettable *gettables; } channel_data; /* used to silence 'variable not used' warnings */ @@ -187,12 +178,11 @@ static void hs_on_recv(void *user_data, int success) { calld->on_done_recv(calld->recv_user_data, success); } -static void hs_start_transport_op(grpc_call_element *elem, grpc_transport_op *op) { +static void hs_mutate_op(grpc_call_element *elem, grpc_transport_op *op) { /* grab pointers to our data from the call element */ call_data *calld = elem->call_data; channel_data *channeld = elem->channel_data; size_t i; - GRPC_CALL_LOG_OP(GPR_INFO, elem, op); if (op->send_ops && !calld->sent_status) { size_t nops = op->send_ops->nops; @@ -215,7 +205,11 @@ static void hs_start_transport_op(grpc_call_element *elem, grpc_transport_op *op op->on_done_recv = hs_on_recv; op->recv_user_data = elem; } +} +static void hs_start_transport_op(grpc_call_element *elem, grpc_transport_op *op) { + GRPC_CALL_LOG_OP(GPR_INFO, elem, op); + hs_mutate_op(elem, op); grpc_call_next_op(elem, op); } @@ -238,15 +232,12 @@ static void channel_op(grpc_channel_element *elem, /* Constructor for call_data */ static void init_call_elem(grpc_call_element *elem, - const void *server_transport_data) { + const void *server_transport_data, grpc_transport_op *initial_op) { /* grab pointers to our data from the call element */ call_data *calld = elem->call_data; - channel_data *channeld = elem->channel_data; - - ignore_unused(channeld); - /* initialize members */ memset(calld, 0, sizeof(*calld)); + if (initial_op) hs_mutate_op(elem, initial_op); } /* Destructor for call_data */ @@ -256,9 +247,6 @@ static void destroy_call_elem(grpc_call_element *elem) {} static void init_channel_elem(grpc_channel_element *elem, const grpc_channel_args *args, grpc_mdctx *mdctx, int is_first, int is_last) { - size_t i; - size_t gettable_capacity = 0; - /* grab pointers to our data from the channel element */ channel_data *channeld = elem->channel_data; @@ -284,46 +272,13 @@ static void init_channel_elem(grpc_channel_element *elem, grpc_mdelem_from_strings(mdctx, "content-type", "application/grpc"); channeld->mdctx = mdctx; - - /* initialize http download support */ - channeld->gettable_count = 0; - channeld->gettables = NULL; - for (i = 0; i < args->num_args; i++) { - if (0 == strcmp(args->args[i].key, GRPC_ARG_SERVE_OVER_HTTP)) { - gettable *g; - gpr_slice slice; - grpc_http_server_page *p = args->args[i].value.pointer.p; - if (channeld->gettable_count == gettable_capacity) { - gettable_capacity = - GPR_MAX(gettable_capacity * 3 / 2, gettable_capacity + 1); - channeld->gettables = gpr_realloc(channeld->gettables, - gettable_capacity * sizeof(gettable)); - } - g = &channeld->gettables[channeld->gettable_count++]; - g->path = grpc_mdelem_from_strings(mdctx, ":path", p->path); - g->content_type = - grpc_mdelem_from_strings(mdctx, "content-type", p->content_type); - slice = gpr_slice_from_copied_string(p->content); - g->content = grpc_byte_buffer_create(&slice, 1); - gpr_slice_unref(slice); - } - } } /* Destructor for channel data */ static void destroy_channel_elem(grpc_channel_element *elem) { - size_t i; - /* grab pointers to our data from the channel element */ channel_data *channeld = elem->channel_data; - for (i = 0; i < channeld->gettable_count; i++) { - grpc_mdelem_unref(channeld->gettables[i].path); - grpc_mdelem_unref(channeld->gettables[i].content_type); - grpc_byte_buffer_destroy(channeld->gettables[i].content); - } - gpr_free(channeld->gettables); - grpc_mdelem_unref(channeld->te_trailers); grpc_mdelem_unref(channeld->status_ok); grpc_mdelem_unref(channeld->status_not_found); diff --git a/src/core/channel/noop_filter.c b/src/core/channel/noop_filter.c index 403b60901b..62e57ce285 100644 --- a/src/core/channel/noop_filter.c +++ b/src/core/channel/noop_filter.c @@ -45,12 +45,7 @@ typedef struct channel_data { /* used to silence 'variable not used' warnings */ static void ignore_unused(void *ignored) {} -/* Called either: - - in response to an API call (or similar) from above, to send something - - a network event (or similar) from below, to receive something - op contains type and call direction information, in addition to the data - that is being sent or received. */ -static void noop_start_transport_op(grpc_call_element *elem, grpc_transport_op *op) { +static void noop_mutate_op(grpc_call_element *elem, grpc_transport_op *op) { /* grab pointers to our data from the call element */ call_data *calld = elem->call_data; channel_data *channeld = elem->channel_data; @@ -58,6 +53,17 @@ static void noop_start_transport_op(grpc_call_element *elem, grpc_transport_op * ignore_unused(calld); ignore_unused(channeld); + /* do nothing */ +} + +/* Called either: + - in response to an API call (or similar) from above, to send something + - a network event (or similar) from below, to receive something + op contains type and call direction information, in addition to the data + that is being sent or received. */ +static void noop_start_transport_op(grpc_call_element *elem, grpc_transport_op *op) { + noop_mutate_op(elem, op); + /* pass control down the stack */ grpc_call_next_op(elem, op); } @@ -81,13 +87,15 @@ static void channel_op(grpc_channel_element *elem, /* Constructor for call_data */ static void init_call_elem(grpc_call_element *elem, - const void *server_transport_data) { + const void *server_transport_data, grpc_transport_op *initial_op) { /* grab pointers to our data from the call element */ call_data *calld = elem->call_data; channel_data *channeld = elem->channel_data; /* initialize members */ calld->unused = channeld->unused; + + if (initial_op) noop_mutate_op(elem, initial_op); } /* Destructor for call_data */ diff --git a/src/core/surface/call.c b/src/core/surface/call.c index 18be81308d..c39e6cf3a4 100644 --- a/src/core/surface/call.c +++ b/src/core/surface/call.c @@ -288,7 +288,7 @@ grpc_call *grpc_call_create(grpc_channel *channel, grpc_completion_queue *cq, /* one ref is dropped in response to destroy, the other in stream_closed */ gpr_ref_init(&call->internal_refcount, 2); - grpc_call_stack_init(channel_stack, server_transport_data, + grpc_call_stack_init(channel_stack, server_transport_data, NULL, CALL_STACK_FROM_CALL(call)); if (gpr_time_cmp(send_deadline, gpr_inf_future) != 0) { set_deadline_alarm(call, send_deadline); diff --git a/src/core/surface/client.c b/src/core/surface/client.c index 4669c59ee4..7eb99895f7 100644 --- a/src/core/surface/client.c +++ b/src/core/surface/client.c @@ -67,7 +67,7 @@ static void channel_op(grpc_channel_element *elem, } static void init_call_elem(grpc_call_element *elem, - const void *transport_server_data) {} + const void *transport_server_data, grpc_transport_op *initial_op) {} static void destroy_call_elem(grpc_call_element *elem) {} diff --git a/src/core/surface/lame_client.c b/src/core/surface/lame_client.c index 4e9eb808d7..f95e2a16c2 100644 --- a/src/core/surface/lame_client.c +++ b/src/core/surface/lame_client.c @@ -66,7 +66,11 @@ static void channel_op(grpc_channel_element *elem, } static void init_call_elem(grpc_call_element *elem, - const void *transport_server_data) {} + const void *transport_server_data, grpc_transport_op *initial_op) { + if (initial_op) { + grpc_transport_op_finish_with_failure(initial_op); + } +} static void destroy_call_elem(grpc_call_element *elem) {} diff --git a/src/core/surface/server.c b/src/core/surface/server.c index 82d1323a4b..e9d6f86734 100644 --- a/src/core/surface/server.c +++ b/src/core/surface/server.c @@ -439,9 +439,8 @@ static void server_on_recv(void *ptr, int success) { calld->on_done_recv(calld->recv_user_data, success); } -static void server_start_transport_op(grpc_call_element *elem, grpc_transport_op *op) { +static void server_mutate_op(grpc_call_element *elem, grpc_transport_op *op) { call_data *calld = elem->call_data; - GRPC_CALL_LOG_OP(GPR_INFO, elem, op); if (op->recv_ops) { /* substitute our callback for the higher callback */ @@ -452,7 +451,11 @@ static void server_start_transport_op(grpc_call_element *elem, grpc_transport_op op->on_done_recv = server_on_recv; op->recv_user_data = elem; } +} +static void server_start_transport_op(grpc_call_element *elem, grpc_transport_op *op) { + GRPC_CALL_LOG_OP(GPR_INFO, elem, op); + server_mutate_op(elem, op); grpc_call_next_op(elem, op); } @@ -504,7 +507,7 @@ static void shutdown_channel(channel_data *chand) { } static void init_call_elem(grpc_call_element *elem, - const void *server_transport_data) { + const void *server_transport_data, grpc_transport_op *initial_op) { call_data *calld = elem->call_data; channel_data *chand = elem->channel_data; memset(calld, 0, sizeof(call_data)); @@ -516,6 +519,8 @@ static void init_call_elem(grpc_call_element *elem, gpr_mu_unlock(&chand->server->mu); server_ref(chand->server); + + server_mutate_op(elem, initial_op); } static void destroy_call_elem(grpc_call_element *elem) { diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index acdc98b86b..d7156142fb 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -378,7 +378,7 @@ static void maybe_finish_read(transport *t, stream *s); static void maybe_join_window_updates(transport *t, stream *s); static void finish_reads(transport *t); static void add_to_pollset_locked(transport *t, grpc_pollset *pollset); - +static void perform_op_locked(transport *t, stream *s, grpc_transport_op *op); /* * CONSTRUCTION/DESTRUCTION/REFCOUNTING @@ -595,7 +595,7 @@ static void goaway(grpc_transport *gt, grpc_status_code status, } static int init_stream(grpc_transport *gt, grpc_stream *gs, - const void *server_data) { + const void *server_data, grpc_transport_op *initial_op) { transport *t = (transport *)gt; stream *s = (stream *)gs; @@ -622,6 +622,8 @@ static int init_stream(grpc_transport *gt, grpc_stream *gs, grpc_sopb_init(&s->callback_sopb); grpc_chttp2_data_parser_init(&s->parser); + if (initial_op) perform_op_locked(t, s, initial_op); + if (!server_data) { unlock(t); } @@ -1003,12 +1005,7 @@ static void maybe_start_some_streams(transport *t) { } } -static void perform_op(grpc_transport *gt, grpc_stream *gs, grpc_transport_op *op) { - transport *t = (transport *)gt; - stream *s = (stream *)gs; - - lock(t); - +static void perform_op_locked(transport *t, stream *s, grpc_transport_op *op) { if (op->send_ops) { GPR_ASSERT(s->outgoing_sopb == NULL); s->send_done_closure.cb = op->on_done_send; @@ -1053,7 +1050,14 @@ static void perform_op(grpc_transport *gt, grpc_stream *gs, grpc_transport_op *o cancel_stream(t, s, op->cancel_with_status, grpc_chttp2_grpc_status_to_http2_error(op->cancel_with_status), 1); } +} +static void perform_op(grpc_transport *gt, grpc_stream *gs, grpc_transport_op *op) { + transport *t = (transport *)gt; + stream *s = (stream *)gs; + + lock(t); + perform_op_locked(t, s, op); unlock(t); } diff --git a/src/core/transport/transport.c b/src/core/transport/transport.c index 35195348e7..ab2f7c3470 100644 --- a/src/core/transport/transport.c +++ b/src/core/transport/transport.c @@ -52,8 +52,8 @@ void grpc_transport_destroy(grpc_transport *transport) { } int grpc_transport_init_stream(grpc_transport *transport, grpc_stream *stream, - const void *server_data) { - return transport->vtable->init_stream(transport, stream, server_data); + const void *server_data, grpc_transport_op *initial_op) { + return transport->vtable->init_stream(transport, stream, server_data, initial_op); } void grpc_transport_perform_op(grpc_transport *transport, grpc_stream *stream, diff --git a/src/core/transport/transport_impl.h b/src/core/transport/transport_impl.h index ef79ac0741..75d8d51e20 100644 --- a/src/core/transport/transport_impl.h +++ b/src/core/transport/transport_impl.h @@ -43,7 +43,7 @@ typedef struct grpc_transport_vtable { /* implementation of grpc_transport_init_stream */ int (*init_stream)(grpc_transport *self, grpc_stream *stream, - const void *server_data); + const void *server_data, grpc_transport_op *initial_op); /* implementation of grpc_transport_send_batch */ void (*perform_op)(grpc_transport *self, grpc_stream *stream, -- cgit v1.2.3 From 06aeea7e9440cccd019c910cb68e306063d28632 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Thu, 23 Apr 2015 10:54:45 -0700 Subject: clang-format --- src/core/channel/census_filter.c | 50 ++++++++++++++----------- src/core/channel/channel_stack.c | 3 +- src/core/channel/child_channel.c | 20 +++++----- src/core/channel/child_channel.h | 5 ++- src/core/channel/client_channel.c | 20 +++++----- src/core/channel/connected_channel.c | 18 ++++----- src/core/channel/http_client_filter.c | 9 +++-- src/core/channel/http_server_filter.c | 12 +++--- src/core/channel/noop_filter.c | 12 +++--- src/core/surface/channel.h | 2 +- src/core/surface/channel_create.c | 3 +- src/core/surface/client.c | 11 ++++-- src/core/surface/lame_client.c | 12 +++--- src/core/surface/server.c | 17 ++++++--- src/core/surface/server_chttp2.c | 3 +- src/core/transport/chttp2/stream_encoder.c | 8 ++-- src/core/transport/chttp2_transport.c | 48 ++++++++++++++---------- src/core/transport/stream_op.h | 29 +++++++------- src/core/transport/transport.c | 6 ++- src/core/transport/transport.h | 11 ++++-- src/core/transport/transport_impl.h | 2 +- test/core/end2end/fixtures/chttp2_socket_pair.c | 9 +++-- 22 files changed, 177 insertions(+), 133 deletions(-) (limited to 'src/core/transport/chttp2_transport.c') diff --git a/src/core/channel/census_filter.c b/src/core/channel/census_filter.c index 47461f7f2b..7e393a01a6 100644 --- a/src/core/channel/census_filter.c +++ b/src/core/channel/census_filter.c @@ -51,9 +51,9 @@ typedef struct call_data { gpr_timespec start_ts; /* recv callback */ - grpc_stream_op_buffer *recv_ops; - void (*on_done_recv)(void *user_data, int success); - void *recv_user_data; + grpc_stream_op_buffer* recv_ops; + void (*on_done_recv)(void* user_data, int success); + void* recv_user_data; } call_data; typedef struct channel_data { @@ -65,24 +65,26 @@ static void init_rpc_stats(census_rpc_stats* stats) { stats->cnt = 1; } -static void extract_and_annotate_method_tag(grpc_stream_op_buffer* sopb, call_data* calld, +static void extract_and_annotate_method_tag(grpc_stream_op_buffer* sopb, + call_data* calld, channel_data* chand) { grpc_linked_mdelem* m; size_t i; for (i = 0; i < sopb->nops; i++) { - grpc_stream_op * op = &sopb->ops[i]; + grpc_stream_op* op = &sopb->ops[i]; if (op->type != GRPC_OP_METADATA) continue; for (m = op->data.metadata.list.head; m != NULL; m = m->next) { if (m->md->key == chand->path_str) { - gpr_log(GPR_DEBUG, "%s", (const char*)GPR_SLICE_START_PTR(m->md->value->slice)); - census_add_method_tag( - calld->op_id, (const char*)GPR_SLICE_START_PTR(m->md->value->slice)); + gpr_log(GPR_DEBUG, "%s", + (const char*)GPR_SLICE_START_PTR(m->md->value->slice)); + census_add_method_tag(calld->op_id, (const char*)GPR_SLICE_START_PTR( + m->md->value->slice)); } } } } -static void client_mutate_op(grpc_call_element *elem, grpc_transport_op *op) { +static void client_mutate_op(grpc_call_element* elem, grpc_transport_op* op) { call_data* calld = elem->call_data; channel_data* chand = elem->channel_data; if (op->send_ops) { @@ -90,15 +92,16 @@ static void client_mutate_op(grpc_call_element *elem, grpc_transport_op *op) { } } -static void client_start_transport_op(grpc_call_element* elem, grpc_transport_op* op) { +static void client_start_transport_op(grpc_call_element* elem, + grpc_transport_op* op) { call_data* calld = elem->call_data; GPR_ASSERT((calld->op_id.upper != 0) || (calld->op_id.lower != 0)); client_mutate_op(elem, op); grpc_call_next_op(elem, op); } -static void server_on_done_recv(void *ptr, int success) { - grpc_call_element *elem = ptr; +static void server_on_done_recv(void* ptr, int success) { + grpc_call_element* elem = ptr; call_data* calld = elem->call_data; channel_data* chand = elem->channel_data; if (success) { @@ -107,7 +110,7 @@ static void server_on_done_recv(void *ptr, int success) { calld->on_done_recv(calld->recv_user_data, success); } -static void server_mutate_op(grpc_call_element *elem, grpc_transport_op *op) { +static void server_mutate_op(grpc_call_element* elem, grpc_transport_op* op) { call_data* calld = elem->call_data; if (op->recv_ops) { /* substitute our callback for the op callback */ @@ -119,7 +122,8 @@ static void server_mutate_op(grpc_call_element *elem, grpc_transport_op *op) { } } -static void server_start_transport_op(grpc_call_element* elem, grpc_transport_op* op) { +static void server_start_transport_op(grpc_call_element* elem, + grpc_transport_op* op) { call_data* calld = elem->call_data; GPR_ASSERT((calld->op_id.upper != 0) || (calld->op_id.lower != 0)); server_mutate_op(elem, op); @@ -140,7 +144,8 @@ static void channel_op(grpc_channel_element* elem, } static void client_init_call_elem(grpc_call_element* elem, - const void* server_transport_data, grpc_transport_op *initial_op) { + const void* server_transport_data, + grpc_transport_op* initial_op) { call_data* d = elem->call_data; GPR_ASSERT(d != NULL); init_rpc_stats(&d->stats); @@ -157,7 +162,8 @@ static void client_destroy_call_elem(grpc_call_element* elem) { } static void server_init_call_elem(grpc_call_element* elem, - const void* server_transport_data, grpc_transport_op *initial_op) { + const void* server_transport_data, + grpc_transport_op* initial_op) { call_data* d = elem->call_data; GPR_ASSERT(d != NULL); init_rpc_stats(&d->stats); @@ -194,11 +200,11 @@ static void destroy_channel_elem(grpc_channel_element* elem) { } const grpc_channel_filter grpc_client_census_filter = { - client_start_transport_op, channel_op, sizeof(call_data), client_init_call_elem, - client_destroy_call_elem, sizeof(channel_data), init_channel_elem, - destroy_channel_elem, "census-client"}; + client_start_transport_op, channel_op, sizeof(call_data), + client_init_call_elem, client_destroy_call_elem, sizeof(channel_data), + init_channel_elem, destroy_channel_elem, "census-client"}; const grpc_channel_filter grpc_server_census_filter = { - server_start_transport_op, channel_op, sizeof(call_data), server_init_call_elem, - server_destroy_call_elem, sizeof(channel_data), init_channel_elem, - destroy_channel_elem, "census-server"}; + server_start_transport_op, channel_op, sizeof(call_data), + server_init_call_elem, server_destroy_call_elem, sizeof(channel_data), + init_channel_elem, destroy_channel_elem, "census-server"}; diff --git a/src/core/channel/channel_stack.c b/src/core/channel/channel_stack.c index 022100e8bd..311f4f08ce 100644 --- a/src/core/channel/channel_stack.c +++ b/src/core/channel/channel_stack.c @@ -166,7 +166,8 @@ void grpc_call_stack_init(grpc_channel_stack *channel_stack, call_elems[i].filter = channel_elems[i].filter; call_elems[i].channel_data = channel_elems[i].channel_data; call_elems[i].call_data = user_data; - call_elems[i].filter->init_call_elem(&call_elems[i], transport_server_data, initial_op); + call_elems[i].filter->init_call_elem(&call_elems[i], transport_server_data, + initial_op); user_data += ROUND_UP_TO_ALIGNMENT_SIZE(call_elems[i].filter->sizeof_call_data); } diff --git a/src/core/channel/child_channel.c b/src/core/channel/child_channel.c index 817a2a8c70..a2f3c54290 100644 --- a/src/core/channel/child_channel.c +++ b/src/core/channel/child_channel.c @@ -60,11 +60,10 @@ typedef struct { gpr_uint8 sent_farewell; } lb_channel_data; -typedef struct { - grpc_child_channel *channel; -} lb_call_data; +typedef struct { grpc_child_channel *channel; } lb_call_data; -static void lb_start_transport_op(grpc_call_element *elem, grpc_transport_op *op) { +static void lb_start_transport_op(grpc_call_element *elem, + grpc_transport_op *op) { grpc_call_next_op(elem, op); } @@ -121,7 +120,8 @@ static void lb_channel_op(grpc_channel_element *elem, /* Constructor for call_data */ static void lb_init_call_elem(grpc_call_element *elem, - const void *server_transport_data, grpc_transport_op *initial_op) {} + const void *server_transport_data, + grpc_transport_op *initial_op) {} /* Destructor for call_data */ static void lb_destroy_call_elem(grpc_call_element *elem) {} @@ -154,9 +154,10 @@ static void lb_destroy_channel_elem(grpc_channel_element *elem) { } const grpc_channel_filter grpc_child_channel_top_filter = { - lb_start_transport_op, lb_channel_op, sizeof(lb_call_data), - lb_init_call_elem, lb_destroy_call_elem, sizeof(lb_channel_data), - lb_init_channel_elem, lb_destroy_channel_elem, "child-channel", }; + lb_start_transport_op, lb_channel_op, sizeof(lb_call_data), + lb_init_call_elem, lb_destroy_call_elem, sizeof(lb_channel_data), + lb_init_channel_elem, lb_destroy_channel_elem, "child-channel", +}; /* grpc_child_channel proper */ @@ -261,7 +262,8 @@ void grpc_child_channel_handle_op(grpc_child_channel *channel, } grpc_child_call *grpc_child_channel_create_call(grpc_child_channel *channel, - grpc_call_element *parent, grpc_transport_op *initial_op) { + grpc_call_element *parent, + grpc_transport_op *initial_op) { grpc_call_stack *stk = gpr_malloc((channel)->call_stack_size); grpc_call_element *lbelem; lb_call_data *lbcalld; diff --git a/src/core/channel/child_channel.h b/src/core/channel/child_channel.h index 264a8bbb82..556a1c731c 100644 --- a/src/core/channel/child_channel.h +++ b/src/core/channel/child_channel.h @@ -57,8 +57,9 @@ void grpc_child_channel_destroy(grpc_child_channel *channel, int wait_for_callbacks); grpc_child_call *grpc_child_channel_create_call(grpc_child_channel *channel, - grpc_call_element *parent, grpc_transport_op *initial_op); + grpc_call_element *parent, + grpc_transport_op *initial_op); grpc_call_element *grpc_child_call_get_top_element(grpc_child_call *call); void grpc_child_call_destroy(grpc_child_call *call); -#endif /* GRPC_INTERNAL_CORE_CHANNEL_CHILD_CHANNEL_H */ +#endif /* GRPC_INTERNAL_CORE_CHANNEL_CHILD_CHANNEL_H */ diff --git a/src/core/channel/client_channel.c b/src/core/channel/client_channel.c index 77c4951038..642822a267 100644 --- a/src/core/channel/client_channel.c +++ b/src/core/channel/client_channel.c @@ -106,7 +106,8 @@ static int prepare_activate(grpc_call_element *elem, /* create a child call */ /* TODO(ctiller): pass the waiting op down here */ - calld->s.active.child_call = grpc_child_channel_create_call(on_child, elem, NULL); + calld->s.active.child_call = + grpc_child_channel_create_call(on_child, elem, NULL); return 1; } @@ -184,9 +185,7 @@ static void remove_waiting_child(channel_data *chand, call_data *calld) { chand->waiting_child_count = new_count; } -static void send_up_cancelled_ops(grpc_call_element *elem) { - abort(); -} +static void send_up_cancelled_ops(grpc_call_element *elem) { abort(); } static void cancel_rpc(grpc_call_element *elem, grpc_transport_op *op) { call_data *calld = elem->call_data; @@ -219,8 +218,8 @@ static void cancel_rpc(grpc_call_element *elem, grpc_transport_op *op) { abort(); } -static void cc_start_transport_op(grpc_call_element *elem, - grpc_transport_op *op) { +static void cc_start_transport_op(grpc_call_element *elem, + grpc_transport_op *op) { call_data *calld = elem->call_data; GPR_ASSERT(elem->filter == &grpc_client_channel_filter); GRPC_CALL_LOG_OP(GPR_INFO, elem, op); @@ -325,7 +324,8 @@ static void channel_op(grpc_channel_element *elem, /* Constructor for call_data */ static void init_call_elem(grpc_call_element *elem, - const void *server_transport_data, grpc_transport_op *initial_op) { + const void *server_transport_data, + grpc_transport_op *initial_op) { call_data *calld = elem->call_data; /* TODO(ctiller): is there something useful we can do here? */ @@ -392,9 +392,9 @@ static void destroy_channel_elem(grpc_channel_element *elem) { } const grpc_channel_filter grpc_client_channel_filter = { - cc_start_transport_op, channel_op, sizeof(call_data), init_call_elem, destroy_call_elem, - sizeof(channel_data), init_channel_elem, destroy_channel_elem, - "client-channel", + cc_start_transport_op, channel_op, sizeof(call_data), init_call_elem, + destroy_call_elem, sizeof(channel_data), init_channel_elem, + destroy_channel_elem, "client-channel", }; grpc_transport_setup_result grpc_client_channel_transport_setup_complete( diff --git a/src/core/channel/connected_channel.c b/src/core/channel/connected_channel.c index 9b7db6124a..14dda88698 100644 --- a/src/core/channel/connected_channel.c +++ b/src/core/channel/connected_channel.c @@ -50,9 +50,7 @@ typedef struct connected_channel_channel_data { grpc_transport *transport; } channel_data; -typedef struct connected_channel_call_data { - void *unused; -} call_data; +typedef struct connected_channel_call_data { void *unused; } call_data; /* We perform a small hack to locate transport data alongside the connected channel data in call allocations, to allow everything to be pulled in minimal @@ -63,13 +61,15 @@ typedef struct connected_channel_call_data { /* Intercept a call operation and either push it directly up or translate it into transport stream operations */ -static void con_start_transport_op(grpc_call_element *elem, grpc_transport_op *op) { +static void con_start_transport_op(grpc_call_element *elem, + grpc_transport_op *op) { call_data *calld = elem->call_data; channel_data *chand = elem->channel_data; GPR_ASSERT(elem->filter == &grpc_connected_channel_filter); GRPC_CALL_LOG_OP(GPR_INFO, elem, op); - grpc_transport_perform_op(chand->transport, TRANSPORT_STREAM_FROM_CALL_DATA(calld), op); + grpc_transport_perform_op(chand->transport, + TRANSPORT_STREAM_FROM_CALL_DATA(calld), op); } /* Currently we assume all channel operations should just be pushed up. */ @@ -136,8 +136,9 @@ static void destroy_channel_elem(grpc_channel_element *elem) { } const grpc_channel_filter grpc_connected_channel_filter = { - con_start_transport_op, channel_op, sizeof(call_data), init_call_elem, destroy_call_elem, - sizeof(channel_data), init_channel_elem, destroy_channel_elem, "connected", + con_start_transport_op, channel_op, sizeof(call_data), init_call_elem, + destroy_call_elem, sizeof(channel_data), init_channel_elem, + destroy_channel_elem, "connected", }; /* Transport callback to accept a new stream... calls up to handle it */ @@ -189,8 +190,7 @@ static void transport_closed(void *user_data, grpc_transport *transport) { } const grpc_transport_callbacks connected_channel_transport_callbacks = { - accept_stream, - transport_goaway, transport_closed, + accept_stream, transport_goaway, transport_closed, }; grpc_transport_setup_result grpc_connected_channel_bind_transport( diff --git a/src/core/channel/http_client_filter.c b/src/core/channel/http_client_filter.c index 45a7436361..9805f325a6 100644 --- a/src/core/channel/http_client_filter.c +++ b/src/core/channel/http_client_filter.c @@ -123,7 +123,8 @@ static void hc_mutate_op(grpc_call_element *elem, grpc_transport_op *op) { } } -static void hc_start_transport_op(grpc_call_element *elem, grpc_transport_op *op) { +static void hc_start_transport_op(grpc_call_element *elem, + grpc_transport_op *op) { GRPC_CALL_LOG_OP(GPR_INFO, elem, op); hc_mutate_op(elem, op); grpc_call_next_op(elem, op); @@ -215,6 +216,6 @@ static void destroy_channel_elem(grpc_channel_element *elem) { } const grpc_channel_filter grpc_http_client_filter = { - hc_start_transport_op, channel_op, sizeof(call_data), init_call_elem, destroy_call_elem, - sizeof(channel_data), init_channel_elem, destroy_channel_elem, - "http-client"}; + hc_start_transport_op, channel_op, sizeof(call_data), init_call_elem, + destroy_call_elem, sizeof(channel_data), init_channel_elem, + destroy_channel_elem, "http-client"}; diff --git a/src/core/channel/http_server_filter.c b/src/core/channel/http_server_filter.c index e5174acd68..1f64df68e3 100644 --- a/src/core/channel/http_server_filter.c +++ b/src/core/channel/http_server_filter.c @@ -207,7 +207,8 @@ static void hs_mutate_op(grpc_call_element *elem, grpc_transport_op *op) { } } -static void hs_start_transport_op(grpc_call_element *elem, grpc_transport_op *op) { +static void hs_start_transport_op(grpc_call_element *elem, + grpc_transport_op *op) { GRPC_CALL_LOG_OP(GPR_INFO, elem, op); hs_mutate_op(elem, op); grpc_call_next_op(elem, op); @@ -232,7 +233,8 @@ static void channel_op(grpc_channel_element *elem, /* Constructor for call_data */ static void init_call_elem(grpc_call_element *elem, - const void *server_transport_data, grpc_transport_op *initial_op) { + const void *server_transport_data, + grpc_transport_op *initial_op) { /* grab pointers to our data from the call element */ call_data *calld = elem->call_data; /* initialize members */ @@ -293,6 +295,6 @@ static void destroy_channel_elem(grpc_channel_element *elem) { } const grpc_channel_filter grpc_http_server_filter = { - hs_start_transport_op, channel_op, sizeof(call_data), init_call_elem, destroy_call_elem, - sizeof(channel_data), init_channel_elem, destroy_channel_elem, - "http-server"}; + hs_start_transport_op, channel_op, sizeof(call_data), init_call_elem, + destroy_call_elem, sizeof(channel_data), init_channel_elem, + destroy_channel_elem, "http-server"}; diff --git a/src/core/channel/noop_filter.c b/src/core/channel/noop_filter.c index 62e57ce285..1d2be716d7 100644 --- a/src/core/channel/noop_filter.c +++ b/src/core/channel/noop_filter.c @@ -61,7 +61,8 @@ static void noop_mutate_op(grpc_call_element *elem, grpc_transport_op *op) { - a network event (or similar) from below, to receive something op contains type and call direction information, in addition to the data that is being sent or received. */ -static void noop_start_transport_op(grpc_call_element *elem, grpc_transport_op *op) { +static void noop_start_transport_op(grpc_call_element *elem, + grpc_transport_op *op) { noop_mutate_op(elem, op); /* pass control down the stack */ @@ -87,7 +88,8 @@ static void channel_op(grpc_channel_element *elem, /* Constructor for call_data */ static void init_call_elem(grpc_call_element *elem, - const void *server_transport_data, grpc_transport_op *initial_op) { + const void *server_transport_data, + grpc_transport_op *initial_op) { /* grab pointers to our data from the call element */ call_data *calld = elem->call_data; channel_data *channeld = elem->channel_data; @@ -134,6 +136,6 @@ static void destroy_channel_elem(grpc_channel_element *elem) { } const grpc_channel_filter grpc_no_op_filter = { - noop_start_transport_op, channel_op, sizeof(call_data), - init_call_elem, destroy_call_elem, sizeof(channel_data), - init_channel_elem, destroy_channel_elem, "no-op"}; + noop_start_transport_op, channel_op, sizeof(call_data), init_call_elem, + destroy_call_elem, sizeof(channel_data), init_channel_elem, + destroy_channel_elem, "no-op"}; diff --git a/src/core/surface/channel.h b/src/core/surface/channel.h index 05d57a905b..388be35711 100644 --- a/src/core/surface/channel.h +++ b/src/core/surface/channel.h @@ -51,4 +51,4 @@ void grpc_client_channel_closed(grpc_channel_element *elem); void grpc_channel_internal_ref(grpc_channel *channel); void grpc_channel_internal_unref(grpc_channel *channel); -#endif /* GRPC_INTERNAL_CORE_SURFACE_CHANNEL_H */ +#endif /* GRPC_INTERNAL_CORE_SURFACE_CHANNEL_H */ diff --git a/src/core/surface/channel_create.c b/src/core/surface/channel_create.c index 5f3e9bec0c..daa8d3a7c6 100644 --- a/src/core/surface/channel_create.c +++ b/src/core/surface/channel_create.c @@ -175,7 +175,8 @@ static void done_setup(void *sp) { static grpc_transport_setup_result complete_setup(void *channel_stack, grpc_transport *transport, grpc_mdctx *mdctx) { - static grpc_channel_filter const *extra_filters[] = {&grpc_http_client_filter}; + static grpc_channel_filter const *extra_filters[] = { + &grpc_http_client_filter}; return grpc_client_channel_transport_setup_complete( channel_stack, transport, extra_filters, GPR_ARRAY_SIZE(extra_filters), mdctx); diff --git a/src/core/surface/client.c b/src/core/surface/client.c index 7eb99895f7..8ac4dd1e0e 100644 --- a/src/core/surface/client.c +++ b/src/core/surface/client.c @@ -43,7 +43,8 @@ typedef struct { void *unused; } call_data; typedef struct { void *unused; } channel_data; -static void client_start_transport_op(grpc_call_element *elem, grpc_transport_op *op) { +static void client_start_transport_op(grpc_call_element *elem, + grpc_transport_op *op) { GRPC_CALL_LOG_OP(GPR_INFO, elem, op); grpc_call_next_op(elem, op); } @@ -67,7 +68,8 @@ static void channel_op(grpc_channel_element *elem, } static void init_call_elem(grpc_call_element *elem, - const void *transport_server_data, grpc_transport_op *initial_op) {} + const void *transport_server_data, + grpc_transport_op *initial_op) {} static void destroy_call_elem(grpc_call_element *elem) {} @@ -81,6 +83,7 @@ static void init_channel_elem(grpc_channel_element *elem, static void destroy_channel_elem(grpc_channel_element *elem) {} const grpc_channel_filter grpc_client_surface_filter = { - client_start_transport_op, channel_op, sizeof(call_data), init_call_elem, destroy_call_elem, - sizeof(channel_data), init_channel_elem, destroy_channel_elem, "client", + client_start_transport_op, channel_op, sizeof(call_data), init_call_elem, + destroy_call_elem, sizeof(channel_data), init_channel_elem, + destroy_channel_elem, "client", }; diff --git a/src/core/surface/lame_client.c b/src/core/surface/lame_client.c index f95e2a16c2..c93222344d 100644 --- a/src/core/surface/lame_client.c +++ b/src/core/surface/lame_client.c @@ -46,7 +46,8 @@ typedef struct { void *unused; } call_data; typedef struct { void *unused; } channel_data; -static void lame_start_transport_op(grpc_call_element *elem, grpc_transport_op *op) { +static void lame_start_transport_op(grpc_call_element *elem, + grpc_transport_op *op) { GRPC_CALL_LOG_OP(GPR_INFO, elem, op); grpc_transport_op_finish_with_failure(op); } @@ -66,7 +67,8 @@ static void channel_op(grpc_channel_element *elem, } static void init_call_elem(grpc_call_element *elem, - const void *transport_server_data, grpc_transport_op *initial_op) { + const void *transport_server_data, + grpc_transport_op *initial_op) { if (initial_op) { grpc_transport_op_finish_with_failure(initial_op); } @@ -84,9 +86,9 @@ static void init_channel_elem(grpc_channel_element *elem, static void destroy_channel_elem(grpc_channel_element *elem) {} static const grpc_channel_filter lame_filter = { - lame_start_transport_op, channel_op, sizeof(call_data), init_call_elem, destroy_call_elem, - sizeof(channel_data), init_channel_elem, destroy_channel_elem, - "lame-client", + lame_start_transport_op, channel_op, sizeof(call_data), init_call_elem, + destroy_call_elem, sizeof(channel_data), init_channel_elem, + destroy_channel_elem, "lame-client", }; grpc_channel *grpc_lame_client_channel_create(void) { diff --git a/src/core/surface/server.c b/src/core/surface/server.c index e9d6f86734..202fafca25 100644 --- a/src/core/surface/server.c +++ b/src/core/surface/server.c @@ -414,8 +414,10 @@ static void server_on_recv(void *ptr, int success) { } switch (*calld->recv_state) { - case GRPC_STREAM_OPEN: break; - case GRPC_STREAM_SEND_CLOSED: break; + case GRPC_STREAM_OPEN: + break; + case GRPC_STREAM_SEND_CLOSED: + break; case GRPC_STREAM_RECV_CLOSED: gpr_mu_lock(&chand->server->mu); if (calld->state == NOT_STARTED) { @@ -453,7 +455,8 @@ static void server_mutate_op(grpc_call_element *elem, grpc_transport_op *op) { } } -static void server_start_transport_op(grpc_call_element *elem, grpc_transport_op *op) { +static void server_start_transport_op(grpc_call_element *elem, + grpc_transport_op *op) { GRPC_CALL_LOG_OP(GPR_INFO, elem, op); server_mutate_op(elem, op); grpc_call_next_op(elem, op); @@ -507,7 +510,8 @@ static void shutdown_channel(channel_data *chand) { } static void init_call_elem(grpc_call_element *elem, - const void *server_transport_data, grpc_transport_op *initial_op) { + const void *server_transport_data, + grpc_transport_op *initial_op) { call_data *calld = elem->call_data; channel_data *chand = elem->channel_data; memset(calld, 0, sizeof(call_data)); @@ -599,8 +603,9 @@ static void destroy_channel_elem(grpc_channel_element *elem) { } static const grpc_channel_filter server_surface_filter = { - server_start_transport_op, channel_op, sizeof(call_data), init_call_elem, destroy_call_elem, - sizeof(channel_data), init_channel_elem, destroy_channel_elem, "server", + server_start_transport_op, channel_op, sizeof(call_data), init_call_elem, + destroy_call_elem, sizeof(channel_data), init_channel_elem, + destroy_channel_elem, "server", }; static void addcq(grpc_server *server, grpc_completion_queue *cq) { diff --git a/src/core/surface/server_chttp2.c b/src/core/surface/server_chttp2.c index ebde5095a9..7b5c2f227b 100644 --- a/src/core/surface/server_chttp2.c +++ b/src/core/surface/server_chttp2.c @@ -45,7 +45,8 @@ static grpc_transport_setup_result setup_transport(void *server, grpc_transport *transport, grpc_mdctx *mdctx) { - static grpc_channel_filter const *extra_filters[] = {&grpc_http_server_filter}; + static grpc_channel_filter const *extra_filters[] = { + &grpc_http_server_filter}; return grpc_server_setup_transport(server, transport, extra_filters, GPR_ARRAY_SIZE(extra_filters), mdctx); } diff --git a/src/core/transport/chttp2/stream_encoder.c b/src/core/transport/chttp2/stream_encoder.c index 9b62aefd0b..cf1e66bf8b 100644 --- a/src/core/transport/chttp2/stream_encoder.c +++ b/src/core/transport/chttp2/stream_encoder.c @@ -481,10 +481,10 @@ gpr_uint32 grpc_chttp2_preencode(grpc_stream_op *inops, size_t *inops_count, break; case GRPC_OP_METADATA: grpc_metadata_batch_assert_ok(&op->data.metadata); - /* these just get copied as they don't impact the number of flow - controlled bytes */ - grpc_sopb_append(outops, op, 1); - curop++; + /* these just get copied as they don't impact the number of flow + controlled bytes */ + grpc_sopb_append(outops, op, 1); + curop++; break; case GRPC_OP_BEGIN_MESSAGE: /* begin op: for now we just convert the op to a slice and fall diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index d7156142fb..ebc715f3b8 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -525,7 +525,7 @@ static void init_transport(transport *t, grpc_transport_setup_callback setup, gpr_mu_lock(&t->mu); t->calling_back = 1; - ref_transport(t); /* matches unref at end of this function */ + ref_transport(t); /* matches unref at end of this function */ gpr_mu_unlock(&t->mu); sr = setup(arg, &t->base, t->metadata_context); @@ -537,7 +537,7 @@ static void init_transport(transport *t, grpc_transport_setup_callback setup, if (t->destroying) gpr_cv_signal(&t->cv); unlock(t); - ref_transport(t); /* matches unref inside recv_data */ + ref_transport(t); /* matches unref inside recv_data */ recv_data(t, slices, nslices, GRPC_ENDPOINT_CB_OK); unref_transport(t); @@ -884,7 +884,8 @@ static int prepare_write(transport *t) { t->outgoing_window -= window_delta; s->outgoing_window -= window_delta; - if (s->write_state == WRITE_STATE_QUEUED_CLOSE && s->outgoing_sopb->nops == 0) { + if (s->write_state == WRITE_STATE_QUEUED_CLOSE && + s->outgoing_sopb->nops == 0) { s->send_closed = 1; } if (s->writing_sopb.nops > 0 || s->send_closed) { @@ -927,8 +928,7 @@ static void finalize_outbuf(transport *t) { while ((s = stream_list_remove_head(t, WRITING))) { grpc_chttp2_encode(s->writing_sopb.ops, s->writing_sopb.nops, - s->send_closed, s->id, &t->hpack_compressor, - &t->outbuf); + s->send_closed, s->id, &t->hpack_compressor, &t->outbuf); s->writing_sopb.nops = 0; if (s->send_closed) { stream_list_join(t, s, WRITTEN_CLOSED); @@ -1047,12 +1047,14 @@ static void perform_op_locked(transport *t, stream *s, grpc_transport_op *op) { } if (op->cancel_with_status != GRPC_STATUS_OK) { - cancel_stream(t, s, op->cancel_with_status, grpc_chttp2_grpc_status_to_http2_error(op->cancel_with_status), - 1); + cancel_stream( + t, s, op->cancel_with_status, + grpc_chttp2_grpc_status_to_http2_error(op->cancel_with_status), 1); } } -static void perform_op(grpc_transport *gt, grpc_stream *gs, grpc_transport_op *op) { +static void perform_op(grpc_transport *gt, grpc_stream *gs, + grpc_transport_op *op) { transport *t = (transport *)gt; stream *s = (stream *)gs; @@ -1129,7 +1131,8 @@ static void cancel_stream_inner(transport *t, stream *s, gpr_uint32 id, } if (s->cancelled) { send_rst = 0; - } else if (!s->read_closed || s->write_state != WRITE_STATE_SENT_CLOSE || had_outgoing) { + } else if (!s->read_closed || s->write_state != WRITE_STATE_SENT_CLOSE || + had_outgoing) { s->cancelled = 1; stream_list_join(t, s, CANCELLED); @@ -1586,11 +1589,11 @@ static int parse_frame_slice(transport *t, gpr_slice slice, int is_last) { } if (st.initial_window_update) { for (i = 0; i < t->stream_map.count; i++) { - stream *s = (stream*)(t->stream_map.values[i]); + stream *s = (stream *)(t->stream_map.values[i]); int was_window_empty = s->outgoing_window <= 0; s->outgoing_window += st.initial_window_update; - if (was_window_empty && s->outgoing_window > 0 && - s->outgoing_sopb && s->outgoing_sopb->nops > 0) { + if (was_window_empty && s->outgoing_window > 0 && s->outgoing_sopb && + s->outgoing_sopb->nops > 0) { stream_list_join(t, s, WRITABLE); } } @@ -1609,7 +1612,8 @@ static int parse_frame_slice(transport *t, gpr_slice slice, int is_last) { s->outgoing_window += st.window_update; /* if this window update makes outgoing ops writable again, flag that */ - if (was_window_empty && s->outgoing_sopb && s->outgoing_sopb->nops > 0) { + if (was_window_empty && s->outgoing_sopb && + s->outgoing_sopb->nops > 0) { stream_list_join(t, s, WRITABLE); } } @@ -1877,8 +1881,8 @@ static void finish_reads(transport *t) { while ((s = stream_list_remove_head(t, FINISHED_READ_OP)) != NULL) { int publish = 0; GPR_ASSERT(s->incoming_sopb); - *s->publish_state = compute_state(s->write_state == WRITE_STATE_SENT_CLOSE, - s->read_closed); + *s->publish_state = + compute_state(s->write_state == WRITE_STATE_SENT_CLOSE, s->read_closed); if (*s->publish_state != s->published_state) { s->published_state = *s->publish_state; publish = 1; @@ -1895,8 +1899,12 @@ static void finish_reads(transport *t) { static void schedule_cb(transport *t, op_closure closure, int success) { if (t->pending_callbacks.capacity == t->pending_callbacks.count) { - t->pending_callbacks.capacity = GPR_MAX(t->pending_callbacks.capacity * 2, 8); - t->pending_callbacks.callbacks = gpr_realloc(t->pending_callbacks.callbacks, t->pending_callbacks.capacity * sizeof(*t->pending_callbacks.callbacks)); + t->pending_callbacks.capacity = + GPR_MAX(t->pending_callbacks.capacity * 2, 8); + t->pending_callbacks.callbacks = + gpr_realloc(t->pending_callbacks.callbacks, + t->pending_callbacks.capacity * + sizeof(*t->pending_callbacks.callbacks)); } closure.success = success; t->pending_callbacks.callbacks[t->pending_callbacks.count++] = closure; @@ -1944,9 +1952,9 @@ static void add_to_pollset(grpc_transport *gt, grpc_pollset *pollset) { */ static const grpc_transport_vtable vtable = { - sizeof(stream), init_stream, perform_op, - add_to_pollset, destroy_stream, goaway, close_transport, - send_ping, destroy_transport}; + sizeof(stream), init_stream, perform_op, + add_to_pollset, destroy_stream, goaway, + close_transport, send_ping, destroy_transport}; void grpc_create_chttp2_transport(grpc_transport_setup_callback setup, void *arg, diff --git a/src/core/transport/stream_op.h b/src/core/transport/stream_op.h index f5de64d583..95497a3cc8 100644 --- a/src/core/transport/stream_op.h +++ b/src/core/transport/stream_op.h @@ -86,29 +86,31 @@ typedef struct grpc_metadata_batch { void grpc_metadata_batch_init(grpc_metadata_batch *comd); void grpc_metadata_batch_destroy(grpc_metadata_batch *comd); void grpc_metadata_batch_merge(grpc_metadata_batch *target, - grpc_metadata_batch *add); + grpc_metadata_batch *add); void grpc_metadata_batch_link_head(grpc_metadata_batch *comd, - grpc_linked_mdelem *storage); + grpc_linked_mdelem *storage); void grpc_metadata_batch_link_tail(grpc_metadata_batch *comd, - grpc_linked_mdelem *storage); + grpc_linked_mdelem *storage); void grpc_metadata_batch_add_head(grpc_metadata_batch *comd, - grpc_linked_mdelem *storage, - grpc_mdelem *elem_to_add); + grpc_linked_mdelem *storage, + grpc_mdelem *elem_to_add); void grpc_metadata_batch_add_tail(grpc_metadata_batch *comd, - grpc_linked_mdelem *storage, - grpc_mdelem *elem_to_add); + grpc_linked_mdelem *storage, + grpc_mdelem *elem_to_add); void grpc_metadata_batch_filter(grpc_metadata_batch *comd, - grpc_mdelem *(*filter)(void *user_data, - grpc_mdelem *elem), - void *user_data); + grpc_mdelem *(*filter)(void *user_data, + grpc_mdelem *elem), + void *user_data); #ifndef NDEBUG void grpc_metadata_batch_assert_ok(grpc_metadata_batch *comd); #else -#define grpc_metadata_batch_assert_ok(comd) do {} while (0) +#define grpc_metadata_batch_assert_ok(comd) \ + do { \ + } while (0) #endif /* Represents a single operation performed on a stream/transport */ @@ -151,7 +153,8 @@ void grpc_sopb_add_no_op(grpc_stream_op_buffer *sopb); /* Append a GRPC_OP_BEGIN to a buffer */ void grpc_sopb_add_begin_message(grpc_stream_op_buffer *sopb, gpr_uint32 length, gpr_uint32 flags); -void grpc_sopb_add_metadata(grpc_stream_op_buffer *sopb, grpc_metadata_batch metadata); +void grpc_sopb_add_metadata(grpc_stream_op_buffer *sopb, + grpc_metadata_batch metadata); /* Append a GRPC_SLICE to a buffer - does not ref/unref the slice */ void grpc_sopb_add_slice(grpc_stream_op_buffer *sopb, gpr_slice slice); /* Append a buffer to a buffer - does not ref/unref any internal objects */ @@ -160,4 +163,4 @@ void grpc_sopb_append(grpc_stream_op_buffer *sopb, grpc_stream_op *ops, char *grpc_sopb_string(grpc_stream_op_buffer *sopb); -#endif /* GRPC_INTERNAL_CORE_TRANSPORT_STREAM_OP_H */ +#endif /* GRPC_INTERNAL_CORE_TRANSPORT_STREAM_OP_H */ diff --git a/src/core/transport/transport.c b/src/core/transport/transport.c index ab2f7c3470..724cb0ea68 100644 --- a/src/core/transport/transport.c +++ b/src/core/transport/transport.c @@ -52,8 +52,10 @@ void grpc_transport_destroy(grpc_transport *transport) { } int grpc_transport_init_stream(grpc_transport *transport, grpc_stream *stream, - const void *server_data, grpc_transport_op *initial_op) { - return transport->vtable->init_stream(transport, stream, server_data, initial_op); + const void *server_data, + grpc_transport_op *initial_op) { + return transport->vtable->init_stream(transport, stream, server_data, + initial_op); } void grpc_transport_perform_op(grpc_transport *transport, grpc_stream *stream, diff --git a/src/core/transport/transport.h b/src/core/transport/transport.h index a51e01d3c9..5036dfc2de 100644 --- a/src/core/transport/transport.h +++ b/src/core/transport/transport.h @@ -95,7 +95,8 @@ struct grpc_transport_callbacks { void (*accept_stream)(void *user_data, grpc_transport *transport, const void *server_data); - void (*goaway)(void *user_data, grpc_transport *transport, grpc_status_code status, gpr_slice debug); + void (*goaway)(void *user_data, grpc_transport *transport, + grpc_status_code status, gpr_slice debug); /* The transport has been closed */ void (*closed)(void *user_data, grpc_transport *transport); @@ -115,7 +116,8 @@ size_t grpc_transport_stream_size(grpc_transport *transport); server_data - either NULL for a client initiated stream, or a pointer supplied from the accept_stream callback function */ int grpc_transport_init_stream(grpc_transport *transport, grpc_stream *stream, - const void *server_data, grpc_transport_op *initial_op); + const void *server_data, + grpc_transport_op *initial_op); /* Destroy transport data for a stream. @@ -133,7 +135,8 @@ void grpc_transport_destroy_stream(grpc_transport *transport, void grpc_transport_op_finish_with_failure(grpc_transport_op *op); /* TODO(ctiller): remove this */ -void grpc_transport_add_to_pollset(grpc_transport *transport, grpc_pollset *pollset); +void grpc_transport_add_to_pollset(grpc_transport *transport, + grpc_pollset *pollset); char *grpc_transport_op_string(grpc_transport_op *op); @@ -205,4 +208,4 @@ void grpc_transport_setup_initiate(grpc_transport_setup *setup); used as a destruction call by setup). */ void grpc_transport_setup_cancel(grpc_transport_setup *setup); -#endif /* GRPC_INTERNAL_CORE_TRANSPORT_TRANSPORT_H */ +#endif /* GRPC_INTERNAL_CORE_TRANSPORT_TRANSPORT_H */ diff --git a/src/core/transport/transport_impl.h b/src/core/transport/transport_impl.h index 75d8d51e20..479e15338f 100644 --- a/src/core/transport/transport_impl.h +++ b/src/core/transport/transport_impl.h @@ -76,4 +76,4 @@ struct grpc_transport { const grpc_transport_vtable *vtable; }; -#endif /* GRPC_INTERNAL_CORE_TRANSPORT_TRANSPORT_IMPL_H */ +#endif /* GRPC_INTERNAL_CORE_TRANSPORT_TRANSPORT_IMPL_H */ diff --git a/test/core/end2end/fixtures/chttp2_socket_pair.c b/test/core/end2end/fixtures/chttp2_socket_pair.c index acc0a69708..d19ceb178b 100644 --- a/test/core/end2end/fixtures/chttp2_socket_pair.c +++ b/test/core/end2end/fixtures/chttp2_socket_pair.c @@ -59,7 +59,8 @@ static grpc_transport_setup_result server_setup_transport( void *ts, grpc_transport *transport, grpc_mdctx *mdctx) { grpc_end2end_test_fixture *f = ts; - static grpc_channel_filter const *extra_filters[] = {&grpc_http_server_filter}; + static grpc_channel_filter const *extra_filters[] = { + &grpc_http_server_filter}; return grpc_server_setup_transport(f->server, transport, extra_filters, GPR_ARRAY_SIZE(extra_filters), mdctx); } @@ -73,9 +74,9 @@ static grpc_transport_setup_result client_setup_transport( void *ts, grpc_transport *transport, grpc_mdctx *mdctx) { sp_client_setup *cs = ts; - const grpc_channel_filter *filters[] = { - &grpc_client_surface_filter, &grpc_http_client_filter, - &grpc_connected_channel_filter}; + const grpc_channel_filter *filters[] = {&grpc_client_surface_filter, + &grpc_http_client_filter, + &grpc_connected_channel_filter}; size_t nfilters = sizeof(filters) / sizeof(*filters); grpc_channel *channel = grpc_channel_create_from_filters( filters, nfilters, cs->client_args, mdctx, 1); -- cgit v1.2.3 From 7e8489ae402d8d551654b3d75a2e8933506ee386 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Thu, 23 Apr 2015 12:41:16 -0700 Subject: Fix double-set of alarm --- src/core/surface/call.c | 19 +++++++++++++++++-- src/core/transport/chttp2_transport.c | 18 +----------------- 2 files changed, 18 insertions(+), 19 deletions(-) (limited to 'src/core/transport/chttp2_transport.c') diff --git a/src/core/surface/call.c b/src/core/surface/call.c index c39e6cf3a4..83c8f4c779 100644 --- a/src/core/surface/call.c +++ b/src/core/surface/call.c @@ -260,6 +260,8 @@ grpc_call *grpc_call_create(grpc_channel *channel, grpc_completion_queue *cq, size_t add_initial_metadata_count, gpr_timespec send_deadline) { size_t i; + grpc_transport_op initial_op; + grpc_transport_op *initial_op_ptr = NULL; grpc_channel_stack *channel_stack = grpc_channel_get_channel_stack(channel); grpc_call *call = gpr_malloc(sizeof(grpc_call) + channel_stack->call_stack_size); @@ -288,7 +290,19 @@ grpc_call *grpc_call_create(grpc_channel *channel, grpc_completion_queue *cq, /* one ref is dropped in response to destroy, the other in stream_closed */ gpr_ref_init(&call->internal_refcount, 2); - grpc_call_stack_init(channel_stack, server_transport_data, NULL, + /* server hack: start reads immediately so we can get initial metadata. + TODO(ctiller): figure out a cleaner solution */ + if (!call->is_client) { + memset(&initial_op, 0, sizeof(initial_op)); + initial_op.recv_ops = &call->recv_ops; + initial_op.recv_state = &call->recv_state; + initial_op.on_done_recv = call_on_done_recv; + initial_op.recv_user_data = call; + call->receiving = 1; + grpc_call_internal_ref(call); + initial_op_ptr = &initial_op; + } + grpc_call_stack_init(channel_stack, server_transport_data, initial_op_ptr, CALL_STACK_FROM_CALL(call)); if (gpr_time_cmp(send_deadline, gpr_inf_future) != 0) { set_deadline_alarm(call, send_deadline); @@ -793,7 +807,6 @@ static int fill_send_ops(grpc_call *call, grpc_transport_op *op) { mdb.list = chain_metadata_from_app(call, data.send_metadata.count, data.send_metadata.metadata); mdb.garbage.head = mdb.garbage.tail = NULL; - mdb.deadline = call->send_deadline; /* send status */ /* TODO(ctiller): cache common status values */ data = call->request_data[GRPC_IOREQ_SEND_STATUS]; @@ -814,6 +827,7 @@ static int fill_send_ops(grpc_call *call, grpc_transport_op *op) { grpc_mdstr_from_string(call->metadata_context, data.send_status.details))); } + grpc_sopb_add_metadata(&call->send_ops, mdb); } } break; @@ -1011,6 +1025,7 @@ static void call_alarm(void *arg, int success) { static void set_deadline_alarm(grpc_call *call, gpr_timespec deadline) { if (call->have_alarm) { gpr_log(GPR_ERROR, "Attempt to set deadline alarm twice"); + return; } grpc_call_internal_ref(call); call->have_alarm = 1; diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index ebc715f3b8..2261b087a2 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -1205,23 +1205,6 @@ static void maybe_join_window_updates(transport *t, stream *s) { } } -#if 0 -static void set_allow_window_updates(grpc_transport *tp, grpc_stream *sp, - int allow) { - transport *t = (transport *)tp; - stream *s = (stream *)sp; - - lock(t); - s->allow_window_updates = allow; - if (allow) { - maybe_join_window_updates(t, s); - } else { - stream_list_remove(t, s, WINDOW_UPDATE); - } - unlock(t); -} -#endif - static grpc_chttp2_parse_error update_incoming_window(transport *t, stream *s) { if (t->incoming_frame_size > t->incoming_window) { gpr_log(GPR_ERROR, "frame of size %d overflows incoming window of %d", @@ -1892,6 +1875,7 @@ static void finish_reads(transport *t) { publish = 1; } if (publish) { + s->incoming_sopb = NULL; schedule_cb(t, s->recv_done_closure, 1); } } -- cgit v1.2.3 From e889314fa2ff4d755c400a481831cc0c39a4c893 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Thu, 23 Apr 2015 16:02:01 -0700 Subject: Allow round-robin again in flow control --- src/core/transport/chttp2_transport.c | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) (limited to 'src/core/transport/chttp2_transport.c') diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index 2261b087a2..3d276b1629 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -893,10 +893,11 @@ static int prepare_write(transport *t) { } /* we should either exhaust window or have no ops left, but not both */ - GPR_ASSERT(s->outgoing_sopb->nops == 0 || s->outgoing_window <= 0); if (s->outgoing_sopb->nops == 0) { s->outgoing_sopb = NULL; schedule_cb(t, s->send_done_closure, 1); + } else if (s->outgoing_window) { + stream_list_add_tail(t, s, WRITABLE); } } -- cgit v1.2.3 From 7abc8d21494596d56358b5d8c6c953bee40bd85a Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Thu, 23 Apr 2015 16:43:55 -0700 Subject: Add missing clear --- src/core/transport/chttp2_transport.c | 1 + 1 file changed, 1 insertion(+) (limited to 'src/core/transport/chttp2_transport.c') diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index 3d276b1629..7a7c2bdfd4 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -1128,6 +1128,7 @@ static void cancel_stream_inner(transport *t, stream *s, gpr_uint32 id, schedule_nuke_sopb(t, &s->parser.incoming_sopb); if (s->outgoing_sopb) { schedule_nuke_sopb(t, s->outgoing_sopb); + s->outgoing_sopb = NULL; schedule_cb(t, s->send_done_closure, 0); } if (s->cancelled) { -- cgit v1.2.3 From 7d4a96a58b9770ad1cb41492f2afb49d7f3d9fd2 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Fri, 24 Apr 2015 07:54:07 -0700 Subject: Publish cancellation metadata --- src/core/transport/chttp2_transport.c | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) (limited to 'src/core/transport/chttp2_transport.c') diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index 7a7c2bdfd4..a02fb93677 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -379,6 +379,7 @@ static void maybe_join_window_updates(transport *t, stream *s); static void finish_reads(transport *t); static void add_to_pollset_locked(transport *t, grpc_pollset *pollset); static void perform_op_locked(transport *t, stream *s, grpc_transport_op *op); +static void add_metadata_batch(transport *t, stream *s); /* * CONSTRUCTION/DESTRUCTION/REFCOUNTING @@ -1151,7 +1152,7 @@ static void cancel_stream_inner(transport *t, stream *s, gpr_uint32 id, default: break; } - + add_metadata_batch(t, s); maybe_finish_read(t, s); } } -- cgit v1.2.3 From c1f7560ac27b6db4106115e5308f1a9124378a60 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Fri, 24 Apr 2015 11:44:53 -0700 Subject: Stream mapping fixes --- src/core/surface/call.c | 9 ++++++--- src/core/transport/chttp2_transport.c | 12 ++++++++++-- 2 files changed, 16 insertions(+), 5 deletions(-) (limited to 'src/core/transport/chttp2_transport.c') diff --git a/src/core/surface/call.c b/src/core/surface/call.c index dbf78f2cfe..2f514465fc 100644 --- a/src/core/surface/call.c +++ b/src/core/surface/call.c @@ -687,7 +687,7 @@ static int add_slice_to_message(grpc_call *call, gpr_slice slice) { static void call_on_done_recv(void *pc, int success) { grpc_call *call = pc; size_t i; - int unref = 0; + int unref_due_to_connection_close = 0; lock(call); call->receiving = 0; if (success) { @@ -714,7 +714,7 @@ static void call_on_done_recv(void *pc, int success) { if (call->recv_state == GRPC_STREAM_CLOSED) { GPR_ASSERT(call->read_state <= READ_STATE_STREAM_CLOSED); call->read_state = READ_STATE_STREAM_CLOSED; - unref = 1; + unref_due_to_connection_close = 1; } finish_read_ops(call); } else { @@ -725,9 +725,11 @@ static void call_on_done_recv(void *pc, int success) { finish_ioreq_op(call, GRPC_IOREQ_RECV_INITIAL_METADATA, GRPC_OP_ERROR); finish_ioreq_op(call, GRPC_IOREQ_RECV_STATUS_DETAILS, GRPC_OP_ERROR); } + call->recv_ops.nops = 0; unlock(call); - if (unref) { + grpc_call_internal_unref(call, 0); + if (unref_due_to_connection_close) { grpc_call_internal_unref(call, 0); } } @@ -798,6 +800,7 @@ static int fill_send_ops(grpc_call *call, grpc_transport_op *op) { op->bind_pollset = grpc_cq_pollset(call->cq); call->last_send_contains |= 1 << GRPC_IOREQ_SEND_INITIAL_METADATA; call->write_state = WRITE_STATE_STARTED; + call->send_initial_metadata_count = 0; /* fall through intended */ case WRITE_STATE_STARTED: if (is_op_live(call, GRPC_IOREQ_SEND_MESSAGE)) { diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index a02fb93677..237def41aa 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -306,7 +306,6 @@ struct stream { gpr_uint8 send_closed; gpr_uint8 read_closed; gpr_uint8 cancelled; - gpr_uint8 published_close; op_closure send_done_closure; op_closure recv_done_closure; @@ -731,6 +730,8 @@ static void stream_list_join(transport *t, stream *s, stream_list_id id) { static void remove_from_stream_map(transport *t, stream *s) { if (s->id == 0) return; + IF_TRACING(gpr_log(GPR_DEBUG, "HTTP:%s: Removing stream %d", t->is_client? "CLI" : "SVR", + s->id)); if (grpc_chttp2_stream_map_delete(&t->stream_map, s->id)) { maybe_start_some_streams(t); } @@ -999,6 +1000,8 @@ static void maybe_start_some_streams(transport *t) { stream *s = stream_list_remove_head(t, WAITING_FOR_CONCURRENCY); if (!s) break; + IF_TRACING(gpr_log(GPR_DEBUG, "HTTP:%s: Allocating new stream %p to id %d", t->is_client? "CLI" : "SVR", s, t->next_stream_id)); + GPR_ASSERT(s->id == 0); s->id = t->next_stream_id; t->next_stream_id += 2; @@ -1018,6 +1021,7 @@ static void perform_op_locked(transport *t, stream *s, grpc_transport_op *op) { s->write_state = WRITE_STATE_QUEUED_CLOSE; } if (s->id == 0) { + IF_TRACING(gpr_log(GPR_DEBUG, "HTTP:%s: New stream %p waiting for concurrency", t->is_client? "CLI" : "SVR", s)); stream_list_join(t, s, WAITING_FOR_CONCURRENCY); maybe_start_some_streams(t); } else if (s->outgoing_window > 0) { @@ -1300,7 +1304,8 @@ static void on_header(void *tp, grpc_mdelem *md) { GPR_ASSERT(s); - IF_TRACING(gpr_log(GPR_INFO, "HTTP:%d:HDR: %s: %s", s->id, + IF_TRACING(gpr_log(GPR_INFO, "HTTP:%d:%s:HDR: %s: %s", s->id, + t->is_client? "CLI" : "SVR", grpc_mdstr_as_c_string(md->key), grpc_mdstr_as_c_string(md->value))); @@ -1872,6 +1877,9 @@ static void finish_reads(transport *t) { if (*s->publish_state != s->published_state) { s->published_state = *s->publish_state; publish = 1; + if (s->published_state == GRPC_STREAM_CLOSED) { + remove_from_stream_map(t, s); + } } if (s->parser.incoming_sopb.nops > 0) { grpc_sopb_swap(s->incoming_sopb, &s->parser.incoming_sopb); -- cgit v1.2.3 From 2ea37fd2ce9046ccf2a0b89ba43c93d8fe80408a Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Fri, 24 Apr 2015 13:03:49 -0700 Subject: Bug fixing --- src/core/channel/channel_stack.h | 3 -- src/core/security/auth.c | 14 +++---- src/core/surface/call.c | 1 - src/core/transport/chttp2_transport.c | 65 +++++++++++++++++--------------- src/core/transport/transport.c | 9 +++++ src/core/transport/transport.h | 3 ++ src/core/transport/transport_op_string.c | 4 ++ 7 files changed, 56 insertions(+), 43 deletions(-) (limited to 'src/core/transport/chttp2_transport.c') diff --git a/src/core/channel/channel_stack.h b/src/core/channel/channel_stack.h index 94b12639fc..de0e4e4518 100644 --- a/src/core/channel/channel_stack.h +++ b/src/core/channel/channel_stack.h @@ -222,9 +222,6 @@ void grpc_call_log_op(char *file, int line, gpr_log_severity severity, grpc_call_element *elem, grpc_transport_op *op); void grpc_call_element_send_cancel(grpc_call_element *cur_elem); -void grpc_call_element_recv_status(grpc_call_element *cur_elem, - grpc_status_code status, - const char *message); extern int grpc_trace_channel; diff --git a/src/core/security/auth.c b/src/core/security/auth.c index 4dbc25675b..b6a002d43c 100644 --- a/src/core/security/auth.c +++ b/src/core/security/auth.c @@ -67,11 +67,6 @@ typedef struct { grpc_mdstr *status_key; } channel_data; -static void bubbleup_error(grpc_call_element *elem, const char *error_msg) { - grpc_call_element_recv_status(elem, GRPC_STATUS_UNAUTHENTICATED, error_msg); - grpc_call_element_send_cancel(elem); -} - static void on_credentials_metadata(void *user_data, grpc_mdelem **md_elems, size_t num_md, grpc_credentials_status status) { @@ -141,6 +136,7 @@ static void send_security_metadata(grpc_call_element *elem, grpc_transport_op *o static void on_host_checked(void *user_data, grpc_security_status status) { grpc_call_element *elem = (grpc_call_element *)user_data; call_data *calld = elem->call_data; + channel_data *chand = elem->channel_data; if (status == GRPC_SECURITY_OK) { send_security_metadata(elem, &calld->op); @@ -148,9 +144,9 @@ static void on_host_checked(void *user_data, grpc_security_status status) { char *error_msg; gpr_asprintf(&error_msg, "Invalid host %s set in :authority metadata.", grpc_mdstr_as_c_string(calld->host)); - bubbleup_error(elem, error_msg); + grpc_transport_op_add_cancellation(&calld->op, GRPC_STATUS_UNAUTHENTICATED, grpc_mdstr_from_string(chand->md_ctx, error_msg)); gpr_free(error_msg); - grpc_transport_op_finish_with_failure(&calld->op); + grpc_call_next_op(elem, &calld->op); } } @@ -199,9 +195,9 @@ static void auth_start_transport_op(grpc_call_element *elem, gpr_asprintf(&error_msg, "Invalid host %s set in :authority metadata.", call_host); - bubbleup_error(elem, error_msg); + grpc_transport_op_add_cancellation(&calld->op, GRPC_STATUS_UNAUTHENTICATED, grpc_mdstr_from_string(channeld->md_ctx, error_msg)); gpr_free(error_msg); - grpc_transport_op_finish_with_failure(&calld->op); + grpc_call_next_op(elem, &calld->op); } return; /* early exit */ } diff --git a/src/core/surface/call.c b/src/core/surface/call.c index 2f514465fc..8eee67bb83 100644 --- a/src/core/surface/call.c +++ b/src/core/surface/call.c @@ -424,7 +424,6 @@ static void unlock(grpc_call *call) { memset(&op, 0, sizeof(op)); if (!call->receiving && - (call->write_state >= WRITE_STATE_STARTED || !call->is_client) && need_more_data(call)) { op.recv_ops = &call->recv_ops; op.recv_state = &call->recv_state; diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index 237def41aa..7b50e285d0 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -361,7 +361,8 @@ static void cancel_stream_id(transport *t, gpr_uint32 id, grpc_chttp2_error_code error_code, int send_rst); static void cancel_stream(transport *t, stream *s, grpc_status_code local_status, - grpc_chttp2_error_code error_code, int send_rst); + grpc_chttp2_error_code error_code, + grpc_mdstr *optional_message, int send_rst); static void finalize_cancellations(transport *t); static stream *lookup_stream(transport *t, gpr_uint32 id); static void remove_from_stream_map(transport *t, stream *s); @@ -1011,6 +1012,12 @@ static void maybe_start_some_streams(transport *t) { } static void perform_op_locked(transport *t, stream *s, grpc_transport_op *op) { + if (op->cancel_with_status != GRPC_STATUS_OK) { + cancel_stream( + t, s, op->cancel_with_status, + grpc_chttp2_grpc_status_to_http2_error(op->cancel_with_status), op->cancel_message, 1); + } + if (op->send_ops) { GPR_ASSERT(s->outgoing_sopb == NULL); s->send_done_closure.cb = op->on_done_send; @@ -1037,26 +1044,16 @@ static void perform_op_locked(transport *t, stream *s, grpc_transport_op *op) { GPR_ASSERT(s->incoming_sopb == NULL); s->recv_done_closure.cb = op->on_done_recv; s->recv_done_closure.user_data = op->recv_user_data; - if (!s->cancelled) { - s->incoming_sopb = op->recv_ops; - s->incoming_sopb->nops = 0; - s->publish_state = op->recv_state; - maybe_finish_read(t, s); - maybe_join_window_updates(t, s); - } else { - schedule_cb(t, s->recv_done_closure, 0); - } + s->incoming_sopb = op->recv_ops; + s->incoming_sopb->nops = 0; + s->publish_state = op->recv_state; + maybe_finish_read(t, s); + maybe_join_window_updates(t, s); } if (op->bind_pollset) { add_to_pollset_locked(t, op->bind_pollset); } - - if (op->cancel_with_status != GRPC_STATUS_OK) { - cancel_stream( - t, s, op->cancel_with_status, - grpc_chttp2_grpc_status_to_http2_error(op->cancel_with_status), 1); - } } static void perform_op(grpc_transport *gt, grpc_stream *gs, @@ -1123,6 +1120,7 @@ static void add_incoming_metadata(transport *t, stream *s, grpc_mdelem *elem) { static void cancel_stream_inner(transport *t, stream *s, gpr_uint32 id, grpc_status_code local_status, grpc_chttp2_error_code error_code, + grpc_mdstr *optional_message, int send_rst) { int had_outgoing; char buffer[GPR_LTOA_MIN_BUFSIZE]; @@ -1147,14 +1145,18 @@ static void cancel_stream_inner(transport *t, stream *s, gpr_uint32 id, add_incoming_metadata( t, s, grpc_mdelem_from_strings(t->metadata_context, "grpc-status", buffer)); - switch (local_status) { - case GRPC_STATUS_CANCELLED: - add_incoming_metadata( - t, s, grpc_mdelem_from_strings(t->metadata_context, - "grpc-message", "Cancelled")); - break; - default: - break; + if (!optional_message) { + switch (local_status) { + case GRPC_STATUS_CANCELLED: + add_incoming_metadata( + t, s, grpc_mdelem_from_strings(t->metadata_context, + "grpc-message", "Cancelled")); + break; + default: + break; + } + } else { + add_incoming_metadata(t, s, grpc_mdelem_from_metadata_strings(t->metadata_context, grpc_mdstr_from_string(t->metadata_context, "grpc-message"), grpc_mdstr_ref(optional_message))); } add_metadata_batch(t, s); maybe_finish_read(t, s); @@ -1165,24 +1167,27 @@ static void cancel_stream_inner(transport *t, stream *s, gpr_uint32 id, gpr_slice_buffer_add(&t->qbuf, grpc_chttp2_rst_stream_create(id, error_code)); } + if (optional_message) { + grpc_mdstr_unref(optional_message); + } } static void cancel_stream_id(transport *t, gpr_uint32 id, grpc_status_code local_status, grpc_chttp2_error_code error_code, int send_rst) { cancel_stream_inner(t, lookup_stream(t, id), id, local_status, error_code, - send_rst); + NULL, send_rst); } static void cancel_stream(transport *t, stream *s, grpc_status_code local_status, - grpc_chttp2_error_code error_code, int send_rst) { - cancel_stream_inner(t, s, s->id, local_status, error_code, send_rst); + grpc_chttp2_error_code error_code, grpc_mdstr *optional_message, int send_rst) { + cancel_stream_inner(t, s, s->id, local_status, error_code, optional_message, send_rst); } static void cancel_stream_cb(void *user_data, gpr_uint32 id, void *stream) { cancel_stream(user_data, stream, GRPC_STATUS_UNAVAILABLE, - GRPC_CHTTP2_INTERNAL_ERROR, 0); + GRPC_CHTTP2_INTERNAL_ERROR, NULL, 0); } static void end_all_the_calls(transport *t) { @@ -1285,7 +1290,7 @@ static int init_data_frame_parser(transport *t) { case GRPC_CHTTP2_STREAM_ERROR: cancel_stream(t, s, grpc_chttp2_http2_error_to_grpc_status( GRPC_CHTTP2_INTERNAL_ERROR), - GRPC_CHTTP2_INTERNAL_ERROR, 1); + GRPC_CHTTP2_INTERNAL_ERROR, NULL, 1); return init_skip_frame(t, 0); case GRPC_CHTTP2_CONNECTION_ERROR: drop_connection(t); @@ -1598,7 +1603,7 @@ static int parse_frame_slice(transport *t, gpr_slice slice, int is_last) { if (!is_window_update_legal(st.window_update, s->outgoing_window)) { cancel_stream(t, s, grpc_chttp2_http2_error_to_grpc_status( GRPC_CHTTP2_FLOW_CONTROL_ERROR), - GRPC_CHTTP2_FLOW_CONTROL_ERROR, 1); + GRPC_CHTTP2_FLOW_CONTROL_ERROR, NULL, 1); } else { s->outgoing_window += st.window_update; /* if this window update makes outgoing ops writable again, diff --git a/src/core/transport/transport.c b/src/core/transport/transport.c index 987dd4c918..cc9392177f 100644 --- a/src/core/transport/transport.c +++ b/src/core/transport/transport.c @@ -94,3 +94,12 @@ void grpc_transport_op_finish_with_failure(grpc_transport_op *op) { op->on_done_recv(op->recv_user_data, 0); } } + +void grpc_transport_op_add_cancellation(grpc_transport_op *op, grpc_status_code status, grpc_mdstr *message) { + if (op->cancel_with_status == GRPC_STATUS_OK) { + op->cancel_with_status = status; + op->cancel_message = message; + } else if (message) { + grpc_mdstr_unref(message); + } +} diff --git a/src/core/transport/transport.h b/src/core/transport/transport.h index 5036dfc2de..7c4bed1863 100644 --- a/src/core/transport/transport.h +++ b/src/core/transport/transport.h @@ -75,6 +75,7 @@ typedef struct grpc_transport_op { grpc_pollset *bind_pollset; grpc_status_code cancel_with_status; + grpc_mdstr *cancel_message; } grpc_transport_op; /* Callbacks made from the transport to the upper layers of grpc. */ @@ -134,6 +135,8 @@ void grpc_transport_destroy_stream(grpc_transport *transport, void grpc_transport_op_finish_with_failure(grpc_transport_op *op); +void grpc_transport_op_add_cancellation(grpc_transport_op *op, grpc_status_code status, grpc_mdstr *message); + /* TODO(ctiller): remove this */ void grpc_transport_add_to_pollset(grpc_transport *transport, grpc_pollset *pollset); diff --git a/src/core/transport/transport_op_string.c b/src/core/transport/transport_op_string.c index 54f501f898..b9283b7abf 100644 --- a/src/core/transport/transport_op_string.c +++ b/src/core/transport/transport_op_string.c @@ -139,6 +139,10 @@ char *grpc_transport_op_string(grpc_transport_op *op) { first = 0; gpr_asprintf(&tmp, "CANCEL:%d", op->cancel_with_status); gpr_strvec_add(&b, tmp); + if (op->cancel_message) { + gpr_asprintf(&tmp, ";msg='%s'", grpc_mdstr_as_c_string(op->cancel_message)); + gpr_strvec_add(&b, tmp); + } } out = gpr_strvec_flatten(&b, NULL); -- cgit v1.2.3 From c52779ff61ca5d79d9ecc5b68976a08933078af4 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Fri, 24 Apr 2015 13:19:48 -0700 Subject: Add missing remove list --- src/core/transport/chttp2_transport.c | 1 + 1 file changed, 1 insertion(+) (limited to 'src/core/transport/chttp2_transport.c') diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index 7b50e285d0..7c3f40e3b9 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -1132,6 +1132,7 @@ static void cancel_stream_inner(transport *t, stream *s, gpr_uint32 id, if (s->outgoing_sopb) { schedule_nuke_sopb(t, s->outgoing_sopb); s->outgoing_sopb = NULL; + stream_list_remove(t, s, WRITABLE); schedule_cb(t, s->send_done_closure, 0); } if (s->cancelled) { -- cgit v1.2.3 From 1a727fde47d56b42703aedf7e672b1f4e9c7d1c2 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Fri, 24 Apr 2015 13:21:22 -0700 Subject: clang-format --- src/core/channel/client_channel.c | 29 +- src/core/security/auth.c | 26 +- src/core/security/server_secure_chttp2.c | 3 +- src/core/surface/call.c | 3 +- src/core/surface/secure_channel_create.c | 2 +- src/core/transport/chttp2_transport.c | 39 ++- src/core/transport/transport.c | 4 +- src/core/transport/transport.h | 4 +- src/core/transport/transport_op_string.c | 3 +- src/csharp/Grpc.Core/Internal/CallSafeHandle.cs | 357 ++++++++++++--------- src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs | 163 +++++----- src/csharp/Grpc.Core/Server.cs | 335 +++++++++---------- src/csharp/Grpc.Examples.MathServer/MathServer.cs | 36 +-- .../Grpc.IntegrationTesting.csproj | 3 +- src/ruby/ext/grpc/rb_call.c | 122 ++++--- test/core/channel/channel_stack_test.c | 11 +- .../chttp2_socket_pair_one_byte_at_a_time.c | 9 +- 17 files changed, 596 insertions(+), 553 deletions(-) (limited to 'src/core/transport/chttp2_transport.c') diff --git a/src/core/channel/client_channel.c b/src/core/channel/client_channel.c index b9a489e0cc..78f8d06d89 100644 --- a/src/core/channel/client_channel.c +++ b/src/core/channel/client_channel.c @@ -139,7 +139,8 @@ static void remove_waiting_child(channel_data *chand, call_data *calld) { chand->waiting_child_count = new_count; } -static void handle_op_after_cancellation(grpc_call_element *elem, grpc_transport_op *op) { +static void handle_op_after_cancellation(grpc_call_element *elem, + grpc_transport_op *op) { call_data *calld = elem->call_data; channel_data *chand = elem->channel_data; if (op->send_ops) { @@ -149,10 +150,10 @@ static void handle_op_after_cancellation(grpc_call_element *elem, grpc_transport char status[GPR_LTOA_MIN_BUFSIZE]; grpc_metadata_batch mdb; gpr_ltoa(GRPC_STATUS_CANCELLED, status); - calld->s.cancelled.status.md = grpc_mdelem_from_strings(chand->mdctx, - "grpc-status", status); - calld->s.cancelled.details.md = grpc_mdelem_from_strings(chand->mdctx, - "grpc-message", "Cancelled"); + calld->s.cancelled.status.md = + grpc_mdelem_from_strings(chand->mdctx, "grpc-status", status); + calld->s.cancelled.details.md = + grpc_mdelem_from_strings(chand->mdctx, "grpc-message", "Cancelled"); calld->s.cancelled.status.prev = calld->s.cancelled.details.next = NULL; calld->s.cancelled.status.next = &calld->s.cancelled.details; calld->s.cancelled.details.prev = &calld->s.cancelled.status; @@ -199,8 +200,10 @@ static void cc_start_transport_op(grpc_call_element *elem, gpr_mu_unlock(&chand->mu); } } else { - /* check to see if we should initiate a connection (if we're not already), - but don't do so until outside the lock to avoid re-entrancy problems if + /* check to see if we should initiate a connection (if we're not + already), + but don't do so until outside the lock to avoid re-entrancy + problems if the callback is immediate */ int initiate_transport_setup = 0; if (!chand->transport_setup_initiated) { @@ -212,9 +215,9 @@ static void cc_start_transport_op(grpc_call_element *elem, if (chand->waiting_child_count == chand->waiting_child_capacity) { chand->waiting_child_capacity = GPR_MAX(chand->waiting_child_capacity * 2, 8); - chand->waiting_children = - gpr_realloc(chand->waiting_children, - chand->waiting_child_capacity * sizeof(call_data *)); + chand->waiting_children = gpr_realloc( + chand->waiting_children, + chand->waiting_child_capacity * sizeof(call_data *)); } calld->s.waiting_op = *op; chand->waiting_children[chand->waiting_child_count++] = calld; @@ -236,8 +239,10 @@ static void cc_start_transport_op(grpc_call_element *elem, handle_op_after_cancellation(elem, &waiting_op); handle_op_after_cancellation(elem, op); } else { - GPR_ASSERT((calld->s.waiting_op.send_ops == NULL) != (op->send_ops == NULL)); - GPR_ASSERT((calld->s.waiting_op.recv_ops == NULL) != (op->recv_ops == NULL)); + GPR_ASSERT((calld->s.waiting_op.send_ops == NULL) != + (op->send_ops == NULL)); + GPR_ASSERT((calld->s.waiting_op.recv_ops == NULL) != + (op->recv_ops == NULL)); if (op->send_ops) { calld->s.waiting_op.send_ops = op->send_ops; calld->s.waiting_op.is_last_send = op->is_last_send; diff --git a/src/core/security/auth.c b/src/core/security/auth.c index b6a002d43c..2322c12aa5 100644 --- a/src/core/security/auth.c +++ b/src/core/security/auth.c @@ -76,7 +76,8 @@ static void on_credentials_metadata(void *user_data, grpc_mdelem **md_elems, grpc_metadata_batch *mdb; size_t i; GPR_ASSERT(num_md <= MAX_CREDENTIALS_METADATA_COUNT); - GPR_ASSERT(op->send_ops && op->send_ops->nops > calld->op_md_idx && op->send_ops->ops[calld->op_md_idx].type == GRPC_OP_METADATA); + GPR_ASSERT(op->send_ops && op->send_ops->nops > calld->op_md_idx && + op->send_ops->ops[calld->op_md_idx].type == GRPC_OP_METADATA); mdb = &op->send_ops->ops[calld->op_md_idx].data.metadata; for (i = 0; i < num_md; i++) { grpc_metadata_batch_add_tail(mdb, &calld->md_links[i], @@ -105,7 +106,8 @@ static char *build_service_url(const char *url_scheme, call_data *calld) { return service_url; } -static void send_security_metadata(grpc_call_element *elem, grpc_transport_op *op) { +static void send_security_metadata(grpc_call_element *elem, + grpc_transport_op *op) { /* grab pointers to our data from the call element */ call_data *calld = elem->call_data; channel_data *channeld = elem->channel_data; @@ -144,7 +146,9 @@ static void on_host_checked(void *user_data, grpc_security_status status) { char *error_msg; gpr_asprintf(&error_msg, "Invalid host %s set in :authority metadata.", grpc_mdstr_as_c_string(calld->host)); - grpc_transport_op_add_cancellation(&calld->op, GRPC_STATUS_UNAUTHENTICATED, grpc_mdstr_from_string(chand->md_ctx, error_msg)); + grpc_transport_op_add_cancellation( + &calld->op, GRPC_STATUS_UNAUTHENTICATED, + grpc_mdstr_from_string(chand->md_ctx, error_msg)); gpr_free(error_msg); grpc_call_next_op(elem, &calld->op); } @@ -155,8 +159,8 @@ static void on_host_checked(void *user_data, grpc_security_status status) { - a network event (or similar) from below, to receive something op contains type and call direction information, in addition to the data that is being sent or received. */ -static void auth_start_transport_op(grpc_call_element *elem, - grpc_transport_op *op) { +static void auth_start_transport_op(grpc_call_element *elem, + grpc_transport_op *op) { /* grab pointers to our data from the call element */ call_data *calld = elem->call_data; channel_data *channeld = elem->channel_data; @@ -195,7 +199,9 @@ static void auth_start_transport_op(grpc_call_element *elem, gpr_asprintf(&error_msg, "Invalid host %s set in :authority metadata.", call_host); - grpc_transport_op_add_cancellation(&calld->op, GRPC_STATUS_UNAUTHENTICATED, grpc_mdstr_from_string(channeld->md_ctx, error_msg)); + grpc_transport_op_add_cancellation( + &calld->op, GRPC_STATUS_UNAUTHENTICATED, + grpc_mdstr_from_string(channeld->md_ctx, error_msg)); gpr_free(error_msg); grpc_call_next_op(elem, &calld->op); } @@ -220,7 +226,8 @@ static void channel_op(grpc_channel_element *elem, /* Constructor for call_data */ static void init_call_elem(grpc_call_element *elem, - const void *server_transport_data, grpc_transport_op *initial_op) { + const void *server_transport_data, + grpc_transport_op *initial_op) { /* TODO(jboeuf): Find a way to pass-in the credentials from the caller here. */ call_data *calld = elem->call_data; @@ -297,5 +304,6 @@ static void destroy_channel_elem(grpc_channel_element *elem) { } const grpc_channel_filter grpc_client_auth_filter = { - auth_start_transport_op, channel_op, sizeof(call_data), init_call_elem, destroy_call_elem, - sizeof(channel_data), init_channel_elem, destroy_channel_elem, "auth"}; + auth_start_transport_op, channel_op, sizeof(call_data), init_call_elem, + destroy_call_elem, sizeof(channel_data), init_channel_elem, + destroy_channel_elem, "auth"}; diff --git a/src/core/security/server_secure_chttp2.c b/src/core/security/server_secure_chttp2.c index 0698161b6d..db9d545c0e 100644 --- a/src/core/security/server_secure_chttp2.c +++ b/src/core/security/server_secure_chttp2.c @@ -72,7 +72,8 @@ static void state_unref(grpc_server_secure_state *state) { static grpc_transport_setup_result setup_transport(void *server, grpc_transport *transport, grpc_mdctx *mdctx) { - static grpc_channel_filter const *extra_filters[] = {&grpc_http_server_filter}; + static grpc_channel_filter const *extra_filters[] = { + &grpc_http_server_filter}; return grpc_server_setup_transport(server, transport, extra_filters, GPR_ARRAY_SIZE(extra_filters), mdctx); } diff --git a/src/core/surface/call.c b/src/core/surface/call.c index 2f5bd94cff..7c91ca917c 100644 --- a/src/core/surface/call.c +++ b/src/core/surface/call.c @@ -423,8 +423,7 @@ static void unlock(grpc_call *call) { memset(&op, 0, sizeof(op)); - if (!call->receiving && - need_more_data(call)) { + if (!call->receiving && need_more_data(call)) { op.recv_ops = &call->recv_ops; op.recv_state = &call->recv_state; op.on_done_recv = call_on_done_recv; diff --git a/src/core/surface/secure_channel_create.c b/src/core/surface/secure_channel_create.c index 0bcbe38131..3e331293b5 100644 --- a/src/core/surface/secure_channel_create.c +++ b/src/core/surface/secure_channel_create.c @@ -210,7 +210,7 @@ grpc_channel *grpc_secure_channel_create(grpc_credentials *creds, grpc_arg connector_arg; grpc_channel_args *args_copy; grpc_channel_args *new_args_from_connector; - grpc_channel_security_connector* connector; + grpc_channel_security_connector *connector; grpc_mdctx *mdctx; #define MAX_FILTERS 3 const grpc_channel_filter *filters[MAX_FILTERS]; diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index 7c3f40e3b9..52de8c218f 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -361,7 +361,7 @@ static void cancel_stream_id(transport *t, gpr_uint32 id, grpc_chttp2_error_code error_code, int send_rst); static void cancel_stream(transport *t, stream *s, grpc_status_code local_status, - grpc_chttp2_error_code error_code, + grpc_chttp2_error_code error_code, grpc_mdstr *optional_message, int send_rst); static void finalize_cancellations(transport *t); static stream *lookup_stream(transport *t, gpr_uint32 id); @@ -731,8 +731,8 @@ static void stream_list_join(transport *t, stream *s, stream_list_id id) { static void remove_from_stream_map(transport *t, stream *s) { if (s->id == 0) return; - IF_TRACING(gpr_log(GPR_DEBUG, "HTTP:%s: Removing stream %d", t->is_client? "CLI" : "SVR", - s->id)); + IF_TRACING(gpr_log(GPR_DEBUG, "HTTP:%s: Removing stream %d", + t->is_client ? "CLI" : "SVR", s->id)); if (grpc_chttp2_stream_map_delete(&t->stream_map, s->id)) { maybe_start_some_streams(t); } @@ -1001,7 +1001,8 @@ static void maybe_start_some_streams(transport *t) { stream *s = stream_list_remove_head(t, WAITING_FOR_CONCURRENCY); if (!s) break; - IF_TRACING(gpr_log(GPR_DEBUG, "HTTP:%s: Allocating new stream %p to id %d", t->is_client? "CLI" : "SVR", s, t->next_stream_id)); + IF_TRACING(gpr_log(GPR_DEBUG, "HTTP:%s: Allocating new stream %p to id %d", + t->is_client ? "CLI" : "SVR", s, t->next_stream_id)); GPR_ASSERT(s->id == 0); s->id = t->next_stream_id; @@ -1015,7 +1016,8 @@ static void perform_op_locked(transport *t, stream *s, grpc_transport_op *op) { if (op->cancel_with_status != GRPC_STATUS_OK) { cancel_stream( t, s, op->cancel_with_status, - grpc_chttp2_grpc_status_to_http2_error(op->cancel_with_status), op->cancel_message, 1); + grpc_chttp2_grpc_status_to_http2_error(op->cancel_with_status), + op->cancel_message, 1); } if (op->send_ops) { @@ -1028,7 +1030,9 @@ static void perform_op_locked(transport *t, stream *s, grpc_transport_op *op) { s->write_state = WRITE_STATE_QUEUED_CLOSE; } if (s->id == 0) { - IF_TRACING(gpr_log(GPR_DEBUG, "HTTP:%s: New stream %p waiting for concurrency", t->is_client? "CLI" : "SVR", s)); + IF_TRACING(gpr_log(GPR_DEBUG, + "HTTP:%s: New stream %p waiting for concurrency", + t->is_client ? "CLI" : "SVR", s)); stream_list_join(t, s, WAITING_FOR_CONCURRENCY); maybe_start_some_streams(t); } else if (s->outgoing_window > 0) { @@ -1120,8 +1124,7 @@ static void add_incoming_metadata(transport *t, stream *s, grpc_mdelem *elem) { static void cancel_stream_inner(transport *t, stream *s, gpr_uint32 id, grpc_status_code local_status, grpc_chttp2_error_code error_code, - grpc_mdstr *optional_message, - int send_rst) { + grpc_mdstr *optional_message, int send_rst) { int had_outgoing; char buffer[GPR_LTOA_MIN_BUFSIZE]; @@ -1157,7 +1160,12 @@ static void cancel_stream_inner(transport *t, stream *s, gpr_uint32 id, break; } } else { - add_incoming_metadata(t, s, grpc_mdelem_from_metadata_strings(t->metadata_context, grpc_mdstr_from_string(t->metadata_context, "grpc-message"), grpc_mdstr_ref(optional_message))); + add_incoming_metadata( + t, s, + grpc_mdelem_from_metadata_strings( + t->metadata_context, + grpc_mdstr_from_string(t->metadata_context, "grpc-message"), + grpc_mdstr_ref(optional_message))); } add_metadata_batch(t, s); maybe_finish_read(t, s); @@ -1182,8 +1190,10 @@ static void cancel_stream_id(transport *t, gpr_uint32 id, static void cancel_stream(transport *t, stream *s, grpc_status_code local_status, - grpc_chttp2_error_code error_code, grpc_mdstr *optional_message, int send_rst) { - cancel_stream_inner(t, s, s->id, local_status, error_code, optional_message, send_rst); + grpc_chttp2_error_code error_code, + grpc_mdstr *optional_message, int send_rst) { + cancel_stream_inner(t, s, s->id, local_status, error_code, optional_message, + send_rst); } static void cancel_stream_cb(void *user_data, gpr_uint32 id, void *stream) { @@ -1310,10 +1320,9 @@ static void on_header(void *tp, grpc_mdelem *md) { GPR_ASSERT(s); - IF_TRACING(gpr_log(GPR_INFO, "HTTP:%d:%s:HDR: %s: %s", s->id, - t->is_client? "CLI" : "SVR", - grpc_mdstr_as_c_string(md->key), - grpc_mdstr_as_c_string(md->value))); + IF_TRACING(gpr_log( + GPR_INFO, "HTTP:%d:%s:HDR: %s: %s", s->id, t->is_client ? "CLI" : "SVR", + grpc_mdstr_as_c_string(md->key), grpc_mdstr_as_c_string(md->value))); if (md->key == t->str_grpc_timeout) { gpr_timespec *cached_timeout = grpc_mdelem_get_user_data(md, free_timeout); diff --git a/src/core/transport/transport.c b/src/core/transport/transport.c index cc9392177f..d9a1319c42 100644 --- a/src/core/transport/transport.c +++ b/src/core/transport/transport.c @@ -95,7 +95,9 @@ void grpc_transport_op_finish_with_failure(grpc_transport_op *op) { } } -void grpc_transport_op_add_cancellation(grpc_transport_op *op, grpc_status_code status, grpc_mdstr *message) { +void grpc_transport_op_add_cancellation(grpc_transport_op *op, + grpc_status_code status, + grpc_mdstr *message) { if (op->cancel_with_status == GRPC_STATUS_OK) { op->cancel_with_status = status; op->cancel_message = message; diff --git a/src/core/transport/transport.h b/src/core/transport/transport.h index 7c4bed1863..cdea0b9a0b 100644 --- a/src/core/transport/transport.h +++ b/src/core/transport/transport.h @@ -135,7 +135,9 @@ void grpc_transport_destroy_stream(grpc_transport *transport, void grpc_transport_op_finish_with_failure(grpc_transport_op *op); -void grpc_transport_op_add_cancellation(grpc_transport_op *op, grpc_status_code status, grpc_mdstr *message); +void grpc_transport_op_add_cancellation(grpc_transport_op *op, + grpc_status_code status, + grpc_mdstr *message); /* TODO(ctiller): remove this */ void grpc_transport_add_to_pollset(grpc_transport *transport, diff --git a/src/core/transport/transport_op_string.c b/src/core/transport/transport_op_string.c index b9283b7abf..a215710969 100644 --- a/src/core/transport/transport_op_string.c +++ b/src/core/transport/transport_op_string.c @@ -140,7 +140,8 @@ char *grpc_transport_op_string(grpc_transport_op *op) { gpr_asprintf(&tmp, "CANCEL:%d", op->cancel_with_status); gpr_strvec_add(&b, tmp); if (op->cancel_message) { - gpr_asprintf(&tmp, ";msg='%s'", grpc_mdstr_as_c_string(op->cancel_message)); + gpr_asprintf(&tmp, ";msg='%s'", + grpc_mdstr_as_c_string(op->cancel_message)); gpr_strvec_add(&b, tmp); } } diff --git a/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs b/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs index 14add60c72..19d3f57abb 100644 --- a/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs +++ b/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs @@ -1,11 +1,11 @@ #region Copyright notice and license // 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 @@ -15,7 +15,7 @@ // * 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 @@ -34,158 +34,203 @@ using System.Diagnostics; using System.Runtime.InteropServices; using Grpc.Core; -namespace Grpc.Core.Internal -{ - internal delegate void CompletionCallbackDelegate(GRPCOpError error, IntPtr batchContextPtr); - - /// - /// grpc_call from - /// - internal class CallSafeHandle : SafeHandleZeroIsInvalid - { - const uint GRPC_WRITE_BUFFER_HINT = 1; - - [DllImport("grpc_csharp_ext.dll")] - static extern CallSafeHandle grpcsharp_channel_create_call(ChannelSafeHandle channel, CompletionQueueSafeHandle cq, string method, string host, Timespec deadline); - - [DllImport("grpc_csharp_ext.dll")] - static extern GRPCCallError grpcsharp_call_cancel(CallSafeHandle call); - - [DllImport("grpc_csharp_ext.dll")] - static extern GRPCCallError grpcsharp_call_cancel_with_status(CallSafeHandle call, StatusCode status, string description); - - [DllImport("grpc_csharp_ext.dll")] - static extern GRPCCallError grpcsharp_call_start_unary(CallSafeHandle call, - [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback, - byte[] send_buffer, UIntPtr send_buffer_len, MetadataArraySafeHandle metadataArray); - - [DllImport("grpc_csharp_ext.dll")] - static extern void grpcsharp_call_blocking_unary(CallSafeHandle call, CompletionQueueSafeHandle dedicatedCq, - [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback, - byte[] send_buffer, UIntPtr send_buffer_len, MetadataArraySafeHandle metadataArray); - - [DllImport("grpc_csharp_ext.dll")] - static extern GRPCCallError grpcsharp_call_start_client_streaming(CallSafeHandle call, - [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback, - MetadataArraySafeHandle metadataArray); - - [DllImport("grpc_csharp_ext.dll")] - static extern GRPCCallError grpcsharp_call_start_server_streaming(CallSafeHandle call, - [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback, - byte[] send_buffer, UIntPtr send_buffer_len, - MetadataArraySafeHandle metadataArray); - - [DllImport("grpc_csharp_ext.dll")] - static extern GRPCCallError grpcsharp_call_start_duplex_streaming(CallSafeHandle call, - [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback, - MetadataArraySafeHandle metadataArray); - - [DllImport("grpc_csharp_ext.dll")] - static extern GRPCCallError grpcsharp_call_send_message(CallSafeHandle call, - [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback, - byte[] send_buffer, UIntPtr send_buffer_len); - - [DllImport("grpc_csharp_ext.dll")] - static extern GRPCCallError grpcsharp_call_send_close_from_client(CallSafeHandle call, - [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback); - - [DllImport("grpc_csharp_ext.dll")] - static extern GRPCCallError grpcsharp_call_send_status_from_server(CallSafeHandle call, [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback, StatusCode statusCode, string statusMessage); - - [DllImport("grpc_csharp_ext.dll")] - static extern GRPCCallError grpcsharp_call_recv_message(CallSafeHandle call, - [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback); - - [DllImport("grpc_csharp_ext.dll")] - static extern GRPCCallError grpcsharp_call_start_serverside(CallSafeHandle call, - [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback); - - [DllImport("grpc_csharp_ext.dll")] - static extern void grpcsharp_call_destroy(IntPtr call); - - private CallSafeHandle() - { - } - - public static CallSafeHandle Create(ChannelSafeHandle channel, CompletionQueueSafeHandle cq, string method, string host, Timespec deadline) - { - return grpcsharp_channel_create_call(channel, cq, method, host, deadline); - } - - public void StartUnary(byte[] payload, CompletionCallbackDelegate callback, MetadataArraySafeHandle metadataArray) - { - AssertCallOk(grpcsharp_call_start_unary(this, callback, payload, new UIntPtr((ulong)payload.Length), metadataArray)); - } - - public void BlockingUnary(CompletionQueueSafeHandle dedicatedCq, byte[] payload, CompletionCallbackDelegate callback, MetadataArraySafeHandle metadataArray) - { - grpcsharp_call_blocking_unary(this, dedicatedCq, callback, payload, new UIntPtr((ulong)payload.Length), metadataArray); - } - - public void StartClientStreaming(CompletionCallbackDelegate callback, MetadataArraySafeHandle metadataArray) - { - AssertCallOk(grpcsharp_call_start_client_streaming(this, callback, metadataArray)); - } - - public void StartServerStreaming(byte[] payload, CompletionCallbackDelegate callback, MetadataArraySafeHandle metadataArray) - { - AssertCallOk(grpcsharp_call_start_server_streaming(this, callback, payload, new UIntPtr((ulong)payload.Length), metadataArray)); - } - - public void StartDuplexStreaming(CompletionCallbackDelegate callback, MetadataArraySafeHandle metadataArray) - { - AssertCallOk(grpcsharp_call_start_duplex_streaming(this, callback, metadataArray)); - } - - public void StartSendMessage(byte[] payload, CompletionCallbackDelegate callback) - { - AssertCallOk(grpcsharp_call_send_message(this, callback, payload, new UIntPtr((ulong)payload.Length))); - } - - public void StartSendCloseFromClient(CompletionCallbackDelegate callback) - { - AssertCallOk(grpcsharp_call_send_close_from_client(this, callback)); - } - - public void StartSendStatusFromServer(Status status, CompletionCallbackDelegate callback) - { - AssertCallOk(grpcsharp_call_send_status_from_server(this, callback, status.StatusCode, status.Detail)); - } - - public void StartReceiveMessage(CompletionCallbackDelegate callback) - { - AssertCallOk(grpcsharp_call_recv_message(this, callback)); - } - - public void StartServerSide(CompletionCallbackDelegate callback) - { - AssertCallOk(grpcsharp_call_start_serverside(this, callback)); - } - - public void Cancel() - { - AssertCallOk(grpcsharp_call_cancel(this)); - } - - public void CancelWithStatus(Status status) - { - AssertCallOk(grpcsharp_call_cancel_with_status(this, status.StatusCode, status.Detail)); - } - - protected override bool ReleaseHandle() - { - grpcsharp_call_destroy(handle); - return true; - } - - private static void AssertCallOk(GRPCCallError callError) - { - Trace.Assert(callError == GRPCCallError.GRPC_CALL_OK, "Status not GRPC_CALL_OK"); - } - - private static uint GetFlags(bool buffered) - { - return buffered ? 0 : GRPC_WRITE_BUFFER_HINT; - } +namespace Grpc.Core.Internal { + internal delegate void CompletionCallbackDelegate(GRPCOpError error, + IntPtr batchContextPtr); + + /// + /// grpc_call from + /// + internal class CallSafeHandle : SafeHandleZeroIsInvalid { + const uint GRPC_WRITE_BUFFER_HINT = 1; + + [DllImport("grpc_csharp_ext.dll")] static extern CallSafeHandle + grpcsharp_channel_create_call(ChannelSafeHandle channel, + CompletionQueueSafeHandle cq, string method, + string host, Timespec deadline); + + [DllImport("grpc_csharp_ext.dll")] static extern GRPCCallError + grpcsharp_call_cancel(CallSafeHandle call); + + [DllImport("grpc_csharp_ext.dll")] static extern GRPCCallError + grpcsharp_call_cancel_with_status(CallSafeHandle call, StatusCode status, + string description); + + [DllImport("grpc_csharp_ext.dll")] static extern GRPCCallError + grpcsharp_call_start_unary( + CallSafeHandle call, + [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate + callback, + byte[] send_buffer, UIntPtr send_buffer_len, + MetadataArraySafeHandle metadataArray); + + [DllImport("grpc_csharp_ext.dll")] static extern void + grpcsharp_call_blocking_unary( + CallSafeHandle call, CompletionQueueSafeHandle dedicatedCq, + [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate + callback, + byte[] send_buffer, UIntPtr send_buffer_len, + MetadataArraySafeHandle metadataArray); + + [DllImport("grpc_csharp_ext.dll")] static extern GRPCCallError + grpcsharp_call_start_client_streaming( + CallSafeHandle call, + [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate + callback, + MetadataArraySafeHandle metadataArray); + + [DllImport("grpc_csharp_ext.dll")] static extern GRPCCallError + grpcsharp_call_start_server_streaming( + CallSafeHandle call, + [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate + callback, + byte[] send_buffer, UIntPtr send_buffer_len, + MetadataArraySafeHandle metadataArray); + + [DllImport("grpc_csharp_ext.dll")] static extern GRPCCallError + grpcsharp_call_start_duplex_streaming( + CallSafeHandle call, + [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate + callback, + MetadataArraySafeHandle metadataArray); + + [DllImport("grpc_csharp_ext.dll")] static extern GRPCCallError + grpcsharp_call_send_message( + CallSafeHandle call, + [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate + callback, + byte[] send_buffer, UIntPtr send_buffer_len); + + [DllImport("grpc_csharp_ext.dll")] static extern GRPCCallError + grpcsharp_call_send_close_from_client( + CallSafeHandle call, + [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate + callback); + + [DllImport("grpc_csharp_ext.dll")] static extern GRPCCallError + grpcsharp_call_send_status_from_server( + CallSafeHandle call, + [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate + callback, + StatusCode statusCode, string statusMessage); + + [DllImport("grpc_csharp_ext.dll")] static extern GRPCCallError + grpcsharp_call_recv_message( + CallSafeHandle call, + [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate + callback); + + [DllImport("grpc_csharp_ext.dll")] static extern GRPCCallError + grpcsharp_call_start_serverside( + CallSafeHandle call, + [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate + callback); + + [DllImport("grpc_csharp_ext.dll")] static extern void + grpcsharp_call_destroy(IntPtr call); + + private + CallSafeHandle() {} + + public + static CallSafeHandle Create(ChannelSafeHandle channel, + CompletionQueueSafeHandle cq, string method, + string host, Timespec deadline) { + return grpcsharp_channel_create_call(channel, cq, method, host, deadline); } + + public + void StartUnary(byte[] payload, CompletionCallbackDelegate callback, + MetadataArraySafeHandle metadataArray) { + AssertCallOk(grpcsharp_call_start_unary( + this, callback, payload, new UIntPtr((ulong)payload.Length), + metadataArray)); + } + + public + void BlockingUnary(CompletionQueueSafeHandle dedicatedCq, byte[] payload, + CompletionCallbackDelegate callback, + MetadataArraySafeHandle metadataArray) { + grpcsharp_call_blocking_unary(this, dedicatedCq, callback, payload, + new UIntPtr((ulong)payload.Length), + metadataArray); + } + + public + void StartClientStreaming(CompletionCallbackDelegate callback, + MetadataArraySafeHandle metadataArray) { + AssertCallOk( + grpcsharp_call_start_client_streaming(this, callback, metadataArray)); + } + + public + void StartServerStreaming(byte[] payload, + CompletionCallbackDelegate callback, + MetadataArraySafeHandle metadataArray) { + AssertCallOk(grpcsharp_call_start_server_streaming( + this, callback, payload, new UIntPtr((ulong)payload.Length), + metadataArray)); + } + + public + void StartDuplexStreaming(CompletionCallbackDelegate callback, + MetadataArraySafeHandle metadataArray) { + AssertCallOk( + grpcsharp_call_start_duplex_streaming(this, callback, metadataArray)); + } + + public + void StartSendMessage(byte[] payload, CompletionCallbackDelegate callback) { + AssertCallOk(grpcsharp_call_send_message( + this, callback, payload, new UIntPtr((ulong)payload.Length))); + } + + public + void StartSendCloseFromClient(CompletionCallbackDelegate callback) { + AssertCallOk(grpcsharp_call_send_close_from_client(this, callback)); + } + + public + void StartSendStatusFromServer(Status status, + CompletionCallbackDelegate callback) { + AssertCallOk(grpcsharp_call_send_status_from_server( + this, callback, status.StatusCode, status.Detail)); + } + + public + void StartReceiveMessage(CompletionCallbackDelegate callback) { + AssertCallOk(grpcsharp_call_recv_message(this, callback)); + } + + public + void StartServerSide(CompletionCallbackDelegate callback) { + AssertCallOk(grpcsharp_call_start_serverside(this, callback)); + } + + public + void Cancel() { AssertCallOk(grpcsharp_call_cancel(this)); } + + public + void CancelWithStatus(Status status) { + AssertCallOk(grpcsharp_call_cancel_with_status(this, status.StatusCode, + status.Detail)); + } + + protected + override bool ReleaseHandle() { + grpcsharp_call_destroy(handle); + return true; + } + + private + static void AssertCallOk(GRPCCallError callError) { + Trace.Assert(callError == GRPCCallError.GRPC_CALL_OK, + "Status not GRPC_CALL_OK"); + } + + private + static uint GetFlags(bool buffered) { + return buffered ? 0 : GRPC_WRITE_BUFFER_HINT; + } + } } \ No newline at end of file diff --git a/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs b/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs index a59da09822..b56e8d9801 100644 --- a/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs +++ b/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs @@ -36,84 +36,89 @@ using System.Collections.Concurrent; using System.Diagnostics; using System.Runtime.InteropServices; -namespace Grpc.Core.Internal -{ - // TODO: we need to make sure that the delegates are not collected before invoked. - internal delegate void ServerShutdownCallbackDelegate(IntPtr eventPtr); - - /// - /// grpc_server from grpc/grpc.h - /// - internal sealed class ServerSafeHandle : SafeHandleZeroIsInvalid - { - [DllImport("grpc_csharp_ext.dll")] - static extern GRPCCallError grpcsharp_server_request_call(ServerSafeHandle server, CompletionQueueSafeHandle cq, [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate callback); - - [DllImport("grpc_csharp_ext.dll")] - static extern ServerSafeHandle grpcsharp_server_create(CompletionQueueSafeHandle cq, IntPtr args); - - [DllImport("grpc_csharp_ext.dll")] - static extern int grpcsharp_server_add_http2_port(ServerSafeHandle server, string addr); - - [DllImport("grpc_csharp_ext.dll")] - static extern int grpcsharp_server_add_secure_http2_port(ServerSafeHandle server, string addr, ServerCredentialsSafeHandle creds); - - [DllImport("grpc_csharp_ext.dll")] - static extern void grpcsharp_server_start(ServerSafeHandle server); - - [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_destroy(IntPtr server); - - private ServerSafeHandle() - { - } - - public static ServerSafeHandle NewServer(CompletionQueueSafeHandle cq, IntPtr args) - { - return grpcsharp_server_create(cq, args); - } - - public int AddListeningPort(string addr) - { - return grpcsharp_server_add_http2_port(this, addr); - } - - public int AddListeningPort(string addr, ServerCredentialsSafeHandle credentials) - { - return grpcsharp_server_add_secure_http2_port(this, addr, credentials); - } - - public void Start() - { - grpcsharp_server_start(this); - } - - public void Shutdown() - { - grpcsharp_server_shutdown(this); - } - - public void ShutdownAndNotify(ServerShutdownCallbackDelegate callback) - { - grpcsharp_server_shutdown_and_notify_CALLBACK(this, callback); - } - - public GRPCCallError RequestCall(CompletionQueueSafeHandle cq, CompletionCallbackDelegate callback) - { - return grpcsharp_server_request_call(this, cq, callback); - } - - protected override bool ReleaseHandle() - { - grpcsharp_server_destroy(handle); - return true; - } +namespace Grpc.Core.Internal { + // TODO: we need to make sure that the delegates are not collected before + // invoked. + internal delegate void ServerShutdownCallbackDelegate(IntPtr eventPtr); + + /// + /// grpc_server from grpc/grpc.h + /// + internal sealed class ServerSafeHandle : SafeHandleZeroIsInvalid { + [DllImport("grpc_csharp_ext.dll")] static extern GRPCCallError + grpcsharp_server_request_call( + ServerSafeHandle server, CompletionQueueSafeHandle cq, + [MarshalAs(UnmanagedType.FunctionPtr)] CompletionCallbackDelegate + callback); + + [DllImport("grpc_csharp_ext.dll")] static extern ServerSafeHandle + grpcsharp_server_create(CompletionQueueSafeHandle cq, IntPtr args); + + [DllImport("grpc_csharp_ext.dll")] static extern int + grpcsharp_server_add_http2_port(ServerSafeHandle server, string addr); + + [DllImport("grpc_csharp_ext.dll")] static extern int + grpcsharp_server_add_secure_http2_port(ServerSafeHandle server, string addr, + ServerCredentialsSafeHandle creds); + + [DllImport("grpc_csharp_ext.dll")] static extern void + grpcsharp_server_start(ServerSafeHandle server); + + [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_destroy(IntPtr server); + + private + ServerSafeHandle() {} + + public + static ServerSafeHandle NewServer(CompletionQueueSafeHandle cq, + IntPtr args) { + return grpcsharp_server_create(cq, args); } + + public + int AddListeningPort(string addr) { + return grpcsharp_server_add_http2_port(this, addr); + } + + public + int AddListeningPort(string addr, ServerCredentialsSafeHandle credentials) { + return grpcsharp_server_add_secure_http2_port(this, addr, credentials); + } + + public + void Start() { grpcsharp_server_start(this); } + + public + void Shutdown() { grpcsharp_server_shutdown(this); } + + public + void ShutdownAndNotify(ServerShutdownCallbackDelegate callback) { + grpcsharp_server_shutdown_and_notify_CALLBACK(this, callback); + } + + public + GRPCCallError RequestCall(CompletionQueueSafeHandle cq, + CompletionCallbackDelegate callback) { + return grpcsharp_server_request_call(this, cq, callback); + } + + protected + override bool ReleaseHandle() { + grpcsharp_server_destroy(handle); + return true; + } + } } diff --git a/src/csharp/Grpc.Core/Server.cs b/src/csharp/Grpc.Core/Server.cs index f086fa8beb..308fbfb71c 100644 --- a/src/csharp/Grpc.Core/Server.cs +++ b/src/csharp/Grpc.Core/Server.cs @@ -39,205 +39,186 @@ using System.Runtime.InteropServices; using System.Threading.Tasks; using Grpc.Core.Internal; -namespace Grpc.Core -{ - /// - /// Server is implemented only to be able to do - /// in-process testing. - /// - public class Server - { - // TODO: make sure the delegate doesn't get garbage collected while - // native callbacks are in the completion queue. - readonly ServerShutdownCallbackDelegate serverShutdownHandler; - readonly CompletionCallbackDelegate newServerRpcHandler; +namespace Grpc.Core { + /// + /// Server is implemented only to be able to do + /// in-process testing. + /// + public + class Server { + // TODO: make sure the delegate doesn't get garbage collected while + // native callbacks are in the completion queue. + readonly ServerShutdownCallbackDelegate serverShutdownHandler; + readonly CompletionCallbackDelegate newServerRpcHandler; + + readonly BlockingCollection newRpcQueue = + new BlockingCollection(); + readonly ServerSafeHandle handle; + + readonly Dictionary callHandlers = + new Dictionary(); + + readonly TaskCompletionSource shutdownTcs = + new TaskCompletionSource(); + + public + Server() { + this.handle = + ServerSafeHandle.NewServer(GetCompletionQueue(), IntPtr.Zero); + this.newServerRpcHandler = HandleNewServerRpc; + this.serverShutdownHandler = HandleServerShutdown; + } - readonly BlockingCollection newRpcQueue = new BlockingCollection(); - readonly ServerSafeHandle handle; + // only call this before Start() + public + void AddServiceDefinition(ServerServiceDefinition serviceDefinition) { + foreach (var entry in serviceDefinition.CallHandlers) { + callHandlers.Add(entry.Key, entry.Value); + } + } - readonly Dictionary callHandlers = new Dictionary(); + // only call before Start() + public + int AddListeningPort(string addr) { return handle.AddListeningPort(addr); } - readonly TaskCompletionSource shutdownTcs = new TaskCompletionSource(); + // only call before Start() + public + int AddListeningPort(string addr, ServerCredentials credentials) { + using(var nativeCredentials = credentials.ToNativeCredentials()) { + return handle.AddListeningPort(addr, nativeCredentials); + } + } - public Server() - { - this.handle = ServerSafeHandle.NewServer(GetCompletionQueue(), IntPtr.Zero); - this.newServerRpcHandler = HandleNewServerRpc; - this.serverShutdownHandler = HandleServerShutdown; - } + public + void Start() { + handle.Start(); - // only call this before Start() - public void AddServiceDefinition(ServerServiceDefinition serviceDefinition) - { - foreach (var entry in serviceDefinition.CallHandlers) - { - callHandlers.Add(entry.Key, entry.Value); - } - } + // TODO: this basically means the server is single threaded.... + StartHandlingRpcs(); + } - // only call before Start() - public int AddListeningPort(string addr) - { - return handle.AddListeningPort(addr); - } + /// + /// Requests and handles single RPC call. + /// + internal void RunRpc() { + AllowOneRpc(); - // only call before Start() - public int AddListeningPort(string addr, ServerCredentials credentials) - { - using (var nativeCredentials = credentials.ToNativeCredentials()) - { - return handle.AddListeningPort(addr, nativeCredentials); - } - } + try { + var rpcInfo = newRpcQueue.Take(); - public void Start() - { - handle.Start(); + // Console.WriteLine("Server received RPC " + rpcInfo.Method); - // TODO: this basically means the server is single threaded.... - StartHandlingRpcs(); + IServerCallHandler callHandler; + if (!callHandlers.TryGetValue(rpcInfo.Method, out callHandler)) { + callHandler = new NoSuchMethodCallHandler(); } + callHandler.StartCall(rpcInfo.Method, rpcInfo.Call, + GetCompletionQueue()); + } catch (Exception e) { + Console.WriteLine("Exception while handling RPC: " + e); + } + } - /// - /// Requests and handles single RPC call. - /// - internal void RunRpc() - { - AllowOneRpc(); - - try - { - var rpcInfo = newRpcQueue.Take(); - - // Console.WriteLine("Server received RPC " + rpcInfo.Method); - - IServerCallHandler callHandler; - if (!callHandlers.TryGetValue(rpcInfo.Method, out callHandler)) - { - callHandler = new NoSuchMethodCallHandler(); - } - callHandler.StartCall(rpcInfo.Method, rpcInfo.Call, GetCompletionQueue()); - } - catch (Exception e) - { - Console.WriteLine("Exception while handling RPC: " + e); - } - } + /// + /// Requests server shutdown and when there are no more calls being + /// serviced, + /// cleans up used resources. + /// + /// The async. + public + async Task ShutdownAsync() { + handle.ShutdownAndNotify(serverShutdownHandler); + await shutdownTcs.Task; + handle.Dispose(); + } - /// - /// Requests server shutdown and when there are no more calls being serviced, - /// cleans up used resources. - /// - /// The async. - public async Task ShutdownAsync() - { - handle.ShutdownAndNotify(serverShutdownHandler); - await shutdownTcs.Task; - handle.Dispose(); - } + /// + /// To allow awaiting termination of the server. + /// + public + Task ShutdownTask { + get { return shutdownTcs.Task; } + } - /// - /// To allow awaiting termination of the server. - /// - public Task ShutdownTask - { - get - { - return shutdownTcs.Task; - } - } + public + void Kill() { handle.Dispose(); } - public void Kill() - { - handle.Dispose(); - } + private + async Task StartHandlingRpcs() { + while (true) { + await Task.Factory.StartNew(RunRpc); + } + } - private async Task StartHandlingRpcs() - { - while (true) - { - await Task.Factory.StartNew(RunRpc); - } - } + private + void AllowOneRpc() { + AssertCallOk( + handle.RequestCall(GetCompletionQueue(), newServerRpcHandler)); + } - private void AllowOneRpc() - { - AssertCallOk(handle.RequestCall(GetCompletionQueue(), newServerRpcHandler)); - } + private + void HandleNewServerRpc(GRPCOpError error, IntPtr batchContextPtr) { + try { + var ctx = new BatchContextSafeHandleNotOwned(batchContextPtr); - private void HandleNewServerRpc(GRPCOpError error, IntPtr batchContextPtr) - { - try - { - var ctx = new BatchContextSafeHandleNotOwned(batchContextPtr); - - if (error != GRPCOpError.GRPC_OP_OK) - { - // TODO: handle error - } - - var rpcInfo = new NewRpcInfo(ctx.GetServerRpcNewCall(), ctx.GetServerRpcNewMethod()); - - // after server shutdown, the callback returns with null call - if (!rpcInfo.Call.IsInvalid) - { - newRpcQueue.Add(rpcInfo); - } - } - catch (Exception e) - { - Console.WriteLine("Caught exception in a native handler: " + e); - } + if (error != GRPCOpError.GRPC_OP_OK) { + // TODO: handle error } - private void HandleServerShutdown(IntPtr eventPtr) - { - try - { - shutdownTcs.SetResult(null); - } - catch (Exception e) - { - Console.WriteLine("Caught exception in a native handler: " + e); - } - } + var rpcInfo = new NewRpcInfo(ctx.GetServerRpcNewCall(), + ctx.GetServerRpcNewMethod()); - private static void AssertCallOk(GRPCCallError callError) - { - Trace.Assert(callError == GRPCCallError.GRPC_CALL_OK, "Status not GRPC_CALL_OK"); + // after server shutdown, the callback returns with null call + if (!rpcInfo.Call.IsInvalid) { + newRpcQueue.Add(rpcInfo); } + } catch (Exception e) { + Console.WriteLine("Caught exception in a native handler: " + e); + } + } - private static CompletionQueueSafeHandle GetCompletionQueue() - { - return GrpcEnvironment.ThreadPool.CompletionQueue; - } + private + void HandleServerShutdown(IntPtr eventPtr) { + try { + shutdownTcs.SetResult(null); + } catch (Exception e) { + Console.WriteLine("Caught exception in a native handler: " + e); + } + } - private struct NewRpcInfo - { - private CallSafeHandle call; - private string method; - - public NewRpcInfo(CallSafeHandle call, string method) - { - this.call = call; - this.method = method; - } - - public CallSafeHandle Call - { - get - { - return this.call; - } - } - - public string Method - { - get - { - return this.method; - } - } - } + private + static void AssertCallOk(GRPCCallError callError) { + Trace.Assert(callError == GRPCCallError.GRPC_CALL_OK, + "Status not GRPC_CALL_OK"); + } + + private + static CompletionQueueSafeHandle GetCompletionQueue() { + return GrpcEnvironment.ThreadPool.CompletionQueue; + } + + private + struct NewRpcInfo { + private + CallSafeHandle call; + private + string method; + + public + NewRpcInfo(CallSafeHandle call, string method) { + this.call = call; + this.method = method; + } + + public + CallSafeHandle Call { + get { return this.call; } + } + + public + string Method { + get { return this.method; } + } } + } } diff --git a/src/csharp/Grpc.Examples.MathServer/MathServer.cs b/src/csharp/Grpc.Examples.MathServer/MathServer.cs index f7429fb43f..f9a28f4d0d 100644 --- a/src/csharp/Grpc.Examples.MathServer/MathServer.cs +++ b/src/csharp/Grpc.Examples.MathServer/MathServer.cs @@ -34,28 +34,26 @@ using System.Runtime.InteropServices; using System.Threading; using Grpc.Core; -namespace math -{ - class MainClass - { - public static void Main(string[] args) - { - String host = "0.0.0.0"; +namespace math { +class MainClass { + public + static void Main(string[] args) { + String host = "0.0.0.0"; - GrpcEnvironment.Initialize(); + GrpcEnvironment.Initialize(); - Server server = new Server(); - server.AddServiceDefinition(MathGrpc.BindService(new MathServiceImpl())); - int port = server.AddListeningPort(host + ":23456"); - server.Start(); + Server server = new Server(); + server.AddServiceDefinition(MathGrpc.BindService(new MathServiceImpl())); + int port = server.AddListeningPort(host + ":23456"); + server.Start(); - Console.WriteLine("MathServer listening on port " + port); + Console.WriteLine("MathServer listening on port " + port); - Console.WriteLine("Press any key to stop the server..."); - Console.ReadKey(); + Console.WriteLine("Press any key to stop the server..."); + Console.ReadKey(); - server.ShutdownAsync().Wait(); - GrpcEnvironment.Shutdown(); - } - } + server.ShutdownAsync().Wait(); + GrpcEnvironment.Shutdown(); + } +} } diff --git a/src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj b/src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj index 584bf1068d..088c435585 100644 --- a/src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj +++ b/src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj @@ -16,7 +16,8 @@ full false bin\Debug - DEBUG; + DEBUG; + prompt 4 true diff --git a/src/ruby/ext/grpc/rb_call.c b/src/ruby/ext/grpc/rb_call.c index 6da7d3c830..d4e255d2fa 100644 --- a/src/ruby/ext/grpc/rb_call.c +++ b/src/ruby/ext/grpc/rb_call.c @@ -118,35 +118,36 @@ static void grpc_rb_call_destroy(void *p) { } static size_t md_ary_datasize(const void *p) { - const grpc_metadata_array* const ary = (grpc_metadata_array*)p; - size_t i, datasize = sizeof(grpc_metadata_array); - for (i = 0; i < ary->count; ++i) { - const grpc_metadata* const md = &ary->metadata[i]; - datasize += strlen(md->key); - datasize += md->value_length; - } - datasize += ary->capacity * sizeof(grpc_metadata); - return datasize; + const grpc_metadata_array *const ary = (grpc_metadata_array *)p; + size_t i, datasize = sizeof(grpc_metadata_array); + for (i = 0; i < ary->count; ++i) { + const grpc_metadata *const md = &ary->metadata[i]; + datasize += strlen(md->key); + datasize += md->value_length; + } + datasize += ary->capacity * sizeof(grpc_metadata); + return datasize; } static const rb_data_type_t grpc_rb_md_ary_data_type = { "grpc_metadata_array", {GRPC_RB_GC_NOT_MARKED, GRPC_RB_GC_DONT_FREE, md_ary_datasize}, - NULL, NULL, - 0 -}; + NULL, + NULL, + 0}; /* Describes grpc_call struct for RTypedData */ static const rb_data_type_t grpc_call_data_type = { "grpc_call", {GRPC_RB_GC_NOT_MARKED, grpc_rb_call_destroy, GRPC_RB_MEMSIZE_UNAVAILABLE}, - NULL, NULL, - /* it is unsafe to specify RUBY_TYPED_FREE_IMMEDIATELY because grpc_rb_call_destroy + NULL, + NULL, + /* it is unsafe to specify RUBY_TYPED_FREE_IMMEDIATELY because + * grpc_rb_call_destroy * touches a hash object. * TODO(yugui) Directly use st_table and call the free function earlier? */ - 0 -}; + 0}; /* Error code details is a hash containing text strings describing errors */ VALUE rb_error_code_details; @@ -250,7 +251,7 @@ static int grpc_rb_md_ary_fill_hash_cb(VALUE key, VALUE val, VALUE md_ary_obj) { } md_ary->metadata[md_ary->count].value = RSTRING_PTR(rb_ary_entry(val, i)); md_ary->metadata[md_ary->count].value_length = - RSTRING_LEN(rb_ary_entry(val, i)); + RSTRING_LEN(rb_ary_entry(val, i)); md_ary->count += 1; } } else { @@ -290,10 +291,11 @@ static int grpc_rb_md_ary_capacity_hash_cb(VALUE key, VALUE val, /* grpc_rb_md_ary_convert converts a ruby metadata hash into a grpc_metadata_array. */ -static void grpc_rb_md_ary_convert(VALUE md_ary_hash, grpc_metadata_array *md_ary) { +static void grpc_rb_md_ary_convert(VALUE md_ary_hash, + grpc_metadata_array *md_ary) { VALUE md_ary_obj = Qnil; if (md_ary_hash == Qnil) { - return; /* Do nothing if the expected has value is nil */ + return; /* Do nothing if the expected has value is nil */ } if (TYPE(md_ary_hash) != T_HASH) { rb_raise(rb_eTypeError, "md_ary_convert: got <%s>, want ", @@ -303,8 +305,8 @@ static void grpc_rb_md_ary_convert(VALUE md_ary_hash, grpc_metadata_array *md_ar /* Initialize the array, compute it's capacity, then fill it. */ grpc_metadata_array_init(md_ary); - md_ary_obj = TypedData_Wrap_Struct(grpc_rb_cMdAry, &grpc_rb_md_ary_data_type, - md_ary); + md_ary_obj = + TypedData_Wrap_Struct(grpc_rb_cMdAry, &grpc_rb_md_ary_data_type, md_ary); rb_hash_foreach(md_ary_hash, grpc_rb_md_ary_capacity_hash_cb, md_ary_obj); md_ary->metadata = gpr_malloc(md_ary->capacity * sizeof(grpc_metadata)); rb_hash_foreach(md_ary_hash, grpc_rb_md_ary_fill_hash_cb, md_ary_obj); @@ -327,16 +329,14 @@ VALUE grpc_rb_md_ary_to_h(grpc_metadata_array *md_ary) { rb_hash_aset(result, key, value); } else if (TYPE(value) == T_ARRAY) { /* Add the string to the returned array */ - rb_ary_push(value, - rb_str_new(md_ary->metadata[i].value, - md_ary->metadata[i].value_length)); + rb_ary_push(value, rb_str_new(md_ary->metadata[i].value, + md_ary->metadata[i].value_length)); } else { /* Add the current value with this key and the new one to an array */ new_ary = rb_ary_new(); rb_ary_push(new_ary, value); - rb_ary_push(new_ary, - rb_str_new(md_ary->metadata[i].value, - md_ary->metadata[i].value_length)); + rb_ary_push(new_ary, rb_str_new(md_ary->metadata[i].value, + md_ary->metadata[i].value_length)); rb_hash_aset(result, key, new_ary); } } @@ -355,7 +355,7 @@ static int grpc_rb_call_check_op_keys_hash_cb(VALUE key, VALUE val, rb_obj_classname(key)); return ST_STOP; } - switch(NUM2INT(key)) { + switch (NUM2INT(key)) { case GRPC_OP_SEND_INITIAL_METADATA: case GRPC_OP_SEND_MESSAGE: case GRPC_OP_SEND_CLOSE_FROM_CLIENT: @@ -367,8 +367,7 @@ static int grpc_rb_call_check_op_keys_hash_cb(VALUE key, VALUE val, rb_ary_push(ops_ary, key); return ST_CONTINUE; default: - rb_raise(rb_eTypeError, "invalid operation : bad value %d", - NUM2INT(key)); + rb_raise(rb_eTypeError, "invalid operation : bad value %d", NUM2INT(key)); }; return ST_STOP; } @@ -377,8 +376,8 @@ static int grpc_rb_call_check_op_keys_hash_cb(VALUE key, VALUE val, struct to the 'send_status_from_server' portion of an op. */ static void grpc_rb_op_update_status_from_server(grpc_op *op, - grpc_metadata_array* md_ary, - VALUE status) { + grpc_metadata_array *md_ary, + VALUE status) { VALUE code = rb_struct_aref(status, sym_code); VALUE details = rb_struct_aref(status, sym_details); VALUE metadata_hash = rb_struct_aref(status, sym_metadata); @@ -405,8 +404,8 @@ static void grpc_rb_op_update_status_from_server(grpc_op *op, * grpc_rb_call_run_batch function */ typedef struct run_batch_stack { /* The batch ops */ - grpc_op ops[8]; /* 8 is the maximum number of operations */ - size_t op_num; /* tracks the last added operation */ + grpc_op ops[8]; /* 8 is the maximum number of operations */ + size_t op_num; /* tracks the last added operation */ /* Data being sent */ grpc_metadata_array send_metadata; @@ -424,7 +423,7 @@ typedef struct run_batch_stack { /* grpc_run_batch_stack_init ensures the run_batch_stack is properly * initialized */ -static void grpc_run_batch_stack_init(run_batch_stack* st) { +static void grpc_run_batch_stack_init(run_batch_stack *st) { MEMZERO(st, run_batch_stack, 1); grpc_metadata_array_init(&st->send_metadata); grpc_metadata_array_init(&st->send_trailing_metadata); @@ -435,7 +434,7 @@ static void grpc_run_batch_stack_init(run_batch_stack* st) { /* grpc_run_batch_stack_cleanup ensures the run_batch_stack is properly * cleaned up */ -static void grpc_run_batch_stack_cleanup(run_batch_stack* st) { +static void grpc_run_batch_stack_cleanup(run_batch_stack *st) { grpc_metadata_array_destroy(&st->send_metadata); grpc_metadata_array_destroy(&st->send_trailing_metadata); grpc_metadata_array_destroy(&st->recv_metadata); @@ -447,7 +446,7 @@ static void grpc_run_batch_stack_cleanup(run_batch_stack* st) { /* grpc_run_batch_stack_fill_ops fills the run_batch_stack ops array from * ops_hash */ -static void grpc_run_batch_stack_fill_ops(run_batch_stack* st, VALUE ops_hash) { +static void grpc_run_batch_stack_fill_ops(run_batch_stack *st, VALUE ops_hash) { VALUE this_op = Qnil; VALUE this_value = Qnil; VALUE ops_ary = rb_ary_new(); @@ -460,7 +459,7 @@ static void grpc_run_batch_stack_fill_ops(run_batch_stack* st, VALUE ops_hash) { for (i = 0; i < (size_t)RARRAY_LEN(ops_ary); i++) { this_op = rb_ary_entry(ops_ary, i); this_value = rb_hash_aref(ops_hash, this_op); - switch(NUM2INT(this_op)) { + switch (NUM2INT(this_op)) { case GRPC_OP_SEND_INITIAL_METADATA: /* N.B. later there is no need to explicitly delete the metadata keys * and values, they are references to data in ruby objects. */ @@ -471,18 +470,16 @@ static void grpc_run_batch_stack_fill_ops(run_batch_stack* st, VALUE ops_hash) { st->send_metadata.metadata; break; case GRPC_OP_SEND_MESSAGE: - st->ops[st->op_num].data.send_message = - grpc_rb_s_to_byte_buffer(RSTRING_PTR(this_value), - RSTRING_LEN(this_value)); + st->ops[st->op_num].data.send_message = grpc_rb_s_to_byte_buffer( + RSTRING_PTR(this_value), RSTRING_LEN(this_value)); break; case GRPC_OP_SEND_CLOSE_FROM_CLIENT: break; case GRPC_OP_SEND_STATUS_FROM_SERVER: /* N.B. later there is no need to explicitly delete the metadata keys * and values, they are references to data in ruby objects. */ - grpc_rb_op_update_status_from_server(&st->ops[st->op_num], - &st->send_trailing_metadata, - this_value); + grpc_rb_op_update_status_from_server( + &st->ops[st->op_num], &st->send_trailing_metadata, this_value); break; case GRPC_OP_RECV_INITIAL_METADATA: st->ops[st->op_num].data.recv_initial_metadata = &st->recv_metadata; @@ -516,12 +513,12 @@ static void grpc_run_batch_stack_fill_ops(run_batch_stack* st, VALUE ops_hash) { /* grpc_run_batch_stack_build_result fills constructs a ruby BatchResult struct after the results have run */ -static VALUE grpc_run_batch_stack_build_result(run_batch_stack* st) { +static VALUE grpc_run_batch_stack_build_result(run_batch_stack *st) { size_t i = 0; VALUE result = rb_struct_new(grpc_rb_sBatchResult, Qnil, Qnil, Qnil, Qnil, Qnil, Qnil, Qnil, Qnil, NULL); for (i = 0; i < st->op_num; i++) { - switch(st->ops[i].op) { + switch (st->ops[i].op) { case GRPC_OP_SEND_INITIAL_METADATA: rb_struct_aset(result, sym_send_metadata, Qtrue); break; @@ -544,13 +541,11 @@ static VALUE grpc_run_batch_stack_build_result(run_batch_stack* st) { break; case GRPC_OP_RECV_STATUS_ON_CLIENT: rb_struct_aset( - result, - sym_status, - rb_struct_new(grpc_rb_sStatus, - UINT2NUM(st->recv_status), + result, sym_status, + rb_struct_new(grpc_rb_sStatus, UINT2NUM(st->recv_status), (st->recv_status_details == NULL - ? Qnil - : rb_str_new2(st->recv_status_details)), + ? Qnil + : rb_str_new2(st->recv_status_details)), grpc_rb_md_ary_to_h(&st->recv_trailing_metadata), NULL)); break; @@ -682,8 +677,7 @@ static void Init_grpc_error_codes() { static void Init_grpc_op_codes() { /* Constants representing operation type codes in grpc.h */ - VALUE grpc_rb_mCallOps = - rb_define_module_under(grpc_rb_mGrpcCore, "CallOps"); + VALUE grpc_rb_mCallOps = rb_define_module_under(grpc_rb_mGrpcCore, "CallOps"); rb_define_const(grpc_rb_mCallOps, "SEND_INITIAL_METADATA", UINT2NUM(GRPC_OP_SEND_INITIAL_METADATA)); rb_define_const(grpc_rb_mCallOps, "SEND_MESSAGE", @@ -709,14 +703,14 @@ void Init_grpc_call() { grpc_rb_eOutOfTime = rb_define_class_under(grpc_rb_mGrpcCore, "OutOfTime", rb_eException); grpc_rb_cCall = rb_define_class_under(grpc_rb_mGrpcCore, "Call", rb_cObject); - grpc_rb_cMdAry = rb_define_class_under(grpc_rb_mGrpcCore, "MetadataArray", - rb_cObject); + grpc_rb_cMdAry = + rb_define_class_under(grpc_rb_mGrpcCore, "MetadataArray", rb_cObject); /* Prevent allocation or inialization of the Call class */ rb_define_alloc_func(grpc_rb_cCall, grpc_rb_cannot_alloc); rb_define_method(grpc_rb_cCall, "initialize", grpc_rb_cannot_init, 0); - rb_define_method(grpc_rb_cCall, "initialize_copy", - grpc_rb_cannot_init_copy, 1); + rb_define_method(grpc_rb_cCall, "initialize_copy", grpc_rb_cannot_init_copy, + 1); /* Add ruby analogues of the Call methods. */ rb_define_method(grpc_rb_cCall, "run_batch", grpc_rb_call_run_batch, 4); @@ -746,16 +740,8 @@ void Init_grpc_call() { /* The Struct used to return the run_batch result. */ grpc_rb_sBatchResult = rb_struct_define( - "BatchResult", - "send_message", - "send_metadata", - "send_close", - "send_status", - "message", - "metadata", - "status", - "cancelled", - NULL); + "BatchResult", "send_message", "send_metadata", "send_close", + "send_status", "message", "metadata", "status", "cancelled", NULL); /* The hash for reference counting calls, to ensure they can't be destroyed * more than once */ diff --git a/test/core/channel/channel_stack_test.c b/test/core/channel/channel_stack_test.c index 1d1331eb9f..957dee1aa7 100644 --- a/test/core/channel/channel_stack_test.c +++ b/test/core/channel/channel_stack_test.c @@ -55,7 +55,8 @@ static void channel_init_func(grpc_channel_element *elem, } static void call_init_func(grpc_call_element *elem, - const void *server_transport_data, grpc_transport_op *initial_op) { + const void *server_transport_data, + grpc_transport_op *initial_op) { ++*(int *)(elem->channel_data); *(int *)(elem->call_data) = 0; } @@ -66,8 +67,7 @@ static void call_destroy_func(grpc_call_element *elem) { ++*(int *)(elem->channel_data); } -static void call_func(grpc_call_element *elem, - grpc_transport_op *op) { +static void call_func(grpc_call_element *elem, grpc_transport_op *op) { ++*(int *)(elem->call_data); } @@ -78,9 +78,8 @@ static void channel_func(grpc_channel_element *elem, static void test_create_channel_stack(void) { const grpc_channel_filter filter = { - call_func, channel_func, sizeof(int), - call_init_func, call_destroy_func, sizeof(int), - channel_init_func, channel_destroy_func, "some_test_filter"}; + call_func, channel_func, sizeof(int), call_init_func, call_destroy_func, + sizeof(int), channel_init_func, channel_destroy_func, "some_test_filter"}; const grpc_channel_filter *filters = &filter; grpc_channel_stack *channel_stack; grpc_call_stack *call_stack; diff --git a/test/core/end2end/fixtures/chttp2_socket_pair_one_byte_at_a_time.c b/test/core/end2end/fixtures/chttp2_socket_pair_one_byte_at_a_time.c index d861034f8f..ddde585b83 100644 --- a/test/core/end2end/fixtures/chttp2_socket_pair_one_byte_at_a_time.c +++ b/test/core/end2end/fixtures/chttp2_socket_pair_one_byte_at_a_time.c @@ -59,7 +59,8 @@ static grpc_transport_setup_result server_setup_transport( void *ts, grpc_transport *transport, grpc_mdctx *mdctx) { grpc_end2end_test_fixture *f = ts; - static grpc_channel_filter const *extra_filters[] = {&grpc_http_server_filter}; + static grpc_channel_filter const *extra_filters[] = { + &grpc_http_server_filter}; return grpc_server_setup_transport(f->server, transport, extra_filters, GPR_ARRAY_SIZE(extra_filters), mdctx); } @@ -73,9 +74,9 @@ static grpc_transport_setup_result client_setup_transport( void *ts, grpc_transport *transport, grpc_mdctx *mdctx) { sp_client_setup *cs = ts; - const grpc_channel_filter *filters[] = { - &grpc_client_surface_filter, &grpc_http_client_filter, - &grpc_connected_channel_filter}; + const grpc_channel_filter *filters[] = {&grpc_client_surface_filter, + &grpc_http_client_filter, + &grpc_connected_channel_filter}; size_t nfilters = sizeof(filters) / sizeof(*filters); grpc_channel *channel = grpc_channel_create_from_filters( filters, nfilters, cs->client_args, mdctx, 1); -- cgit v1.2.3 From 4df5caea068914474c3a1596dd1d195ebfdfaff1 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Fri, 24 Apr 2015 13:46:12 -0700 Subject: Fix memory leak --- src/core/transport/chttp2_transport.c | 3 +++ 1 file changed, 3 insertions(+) (limited to 'src/core/transport/chttp2_transport.c') diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index 52de8c218f..49cd87f80f 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -420,6 +420,9 @@ static void destruct_transport(transport *t) { } gpr_free(t->pings); + gpr_free(t->pending_callbacks.callbacks); + gpr_free(t->executing_callbacks.callbacks); + for (i = 0; i < t->num_pending_goaways; i++) { gpr_slice_unref(t->pending_goaways[i].debug); } -- cgit v1.2.3 From 48bfcdcfcc42ccefdf711b509cce61d4299655ce Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Fri, 24 Apr 2015 14:24:27 -0700 Subject: Fix memory leak --- src/core/transport/chttp2_transport.c | 60 +++++++++++++++++++++++++---------- src/core/transport/stream_op.c | 24 -------------- 2 files changed, 43 insertions(+), 41 deletions(-) (limited to 'src/core/transport/chttp2_transport.c') diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index 49cd87f80f..0640c848d7 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -661,9 +661,12 @@ static void destroy_stream(grpc_transport *gt, grpc_stream *gs) { gpr_mu_unlock(&t->mu); GPR_ASSERT(s->outgoing_sopb == NULL); + GPR_ASSERT(s->incoming_sopb == NULL); grpc_sopb_destroy(&s->writing_sopb); grpc_sopb_destroy(&s->callback_sopb); grpc_chttp2_data_parser_destroy(&s->parser); + GPR_ASSERT(s->incoming_metadata_count == 0); + gpr_free(s->incoming_metadata); unref_transport(t); } @@ -1522,28 +1525,18 @@ static int is_window_update_legal(gpr_int64 window_update, gpr_int64 window) { static void add_metadata_batch(transport *t, stream *s) { grpc_metadata_batch b; - size_t i; - b.list.head = &s->incoming_metadata[0]; - b.list.tail = &s->incoming_metadata[s->incoming_metadata_count - 1]; + b.list.head = NULL; + /* Store away the last element of the list, so that in patch_metadata_ops + we can reconstitute the list. + We can't do list building here as later incoming metadata may reallocate + the underlying array. */ + b.list.tail = (void*)(gpr_intptr)s->incoming_metadata_count; b.garbage.head = b.garbage.tail = NULL; b.deadline = s->incoming_deadline; - - for (i = 1; i < s->incoming_metadata_count; i++) { - s->incoming_metadata[i].prev = &s->incoming_metadata[i - 1]; - s->incoming_metadata[i - 1].next = &s->incoming_metadata[i]; - } - s->incoming_metadata[0].prev = NULL; - s->incoming_metadata[s->incoming_metadata_count - 1].next = NULL; + s->incoming_deadline = gpr_inf_future; grpc_sopb_add_metadata(&s->parser.incoming_sopb, b); - /* TODO(ctiller): don't leak incoming_metadata */ - - /* reset */ - s->incoming_deadline = gpr_inf_future; - s->incoming_metadata = NULL; - s->incoming_metadata_count = 0; - s->incoming_metadata_capacity = 0; } static int parse_frame_slice(transport *t, gpr_slice slice, int is_last) { @@ -1884,6 +1877,35 @@ static grpc_stream_state compute_state(gpr_uint8 write_closed, return GRPC_STREAM_OPEN; } +static void patch_metadata_ops(stream *s) { + grpc_stream_op *ops = s->incoming_sopb->ops; + size_t nops = s->incoming_sopb->nops; + size_t i; + size_t j; + size_t mdidx = 0; + size_t last_mdidx; + + for (i = 0; i < nops; i++) { + grpc_stream_op *op = &ops[i]; + if (op->type != GRPC_OP_METADATA) continue; + last_mdidx = (size_t)(gpr_intptr)(op->data.metadata.list.tail); + GPR_ASSERT(last_mdidx > mdidx); + GPR_ASSERT(last_mdidx <= s->incoming_metadata_count); + op->data.metadata.list.head = &s->incoming_metadata[mdidx]; + op->data.metadata.list.tail = &s->incoming_metadata[last_mdidx - 1]; + for (j = mdidx + 1; j < last_mdidx; j++) { + s->incoming_metadata[j].prev = &s->incoming_metadata[j-1]; + s->incoming_metadata[j-1].next = &s->incoming_metadata[j]; + } + s->incoming_metadata[mdidx].prev = NULL; + s->incoming_metadata[last_mdidx-1].next = NULL; + mdidx = last_mdidx; + } + GPR_ASSERT(mdidx == s->incoming_metadata_count); + + s->incoming_metadata_count = 0; +} + static void finish_reads(transport *t) { stream *s; @@ -1904,10 +1926,14 @@ static void finish_reads(transport *t) { publish = 1; } if (publish) { + if (s->incoming_metadata_count > 0) { + patch_metadata_ops(s); + } s->incoming_sopb = NULL; schedule_cb(t, s->recv_done_closure, 1); } } + } static void schedule_cb(transport *t, op_closure closure, int success) { diff --git a/src/core/transport/stream_op.c b/src/core/transport/stream_op.c index ea22b0e1c8..e1a75adcb6 100644 --- a/src/core/transport/stream_op.c +++ b/src/core/transport/stream_op.c @@ -88,34 +88,19 @@ void grpc_stream_ops_unref_owned_objects(grpc_stream_op *ops, size_t nops) { } } -static void assert_contained_metadata_ok(grpc_stream_op *ops, size_t nops) { -#ifndef NDEBUG - size_t i; - for (i = 0; i < nops; i++) { - if (ops[i].type == GRPC_OP_METADATA) { - grpc_metadata_batch_assert_ok(&ops[i].data.metadata); - } - } -#endif /* NDEBUG */ -} - static void expandto(grpc_stream_op_buffer *sopb, size_t new_capacity) { sopb->capacity = new_capacity; - assert_contained_metadata_ok(sopb->ops, sopb->nops); if (sopb->ops == sopb->inlined_ops) { sopb->ops = gpr_malloc(sizeof(grpc_stream_op) * new_capacity); memcpy(sopb->ops, sopb->inlined_ops, sopb->nops * sizeof(grpc_stream_op)); } else { sopb->ops = gpr_realloc(sopb->ops, sizeof(grpc_stream_op) * new_capacity); } - assert_contained_metadata_ok(sopb->ops, sopb->nops); } static grpc_stream_op *add(grpc_stream_op_buffer *sopb) { grpc_stream_op *out; - assert_contained_metadata_ok(sopb->ops, sopb->nops); - GPR_ASSERT(sopb->nops <= sopb->capacity); if (sopb->nops == sopb->capacity) { expandto(sopb, GROW(sopb->capacity)); @@ -127,7 +112,6 @@ static grpc_stream_op *add(grpc_stream_op_buffer *sopb) { void grpc_sopb_add_no_op(grpc_stream_op_buffer *sopb) { add(sopb)->type = GRPC_NO_OP; - assert_contained_metadata_ok(sopb->ops, sopb->nops); } void grpc_sopb_add_begin_message(grpc_stream_op_buffer *sopb, gpr_uint32 length, @@ -136,24 +120,19 @@ void grpc_sopb_add_begin_message(grpc_stream_op_buffer *sopb, gpr_uint32 length, op->type = GRPC_OP_BEGIN_MESSAGE; op->data.begin_message.length = length; op->data.begin_message.flags = flags; - assert_contained_metadata_ok(sopb->ops, sopb->nops); } void grpc_sopb_add_metadata(grpc_stream_op_buffer *sopb, grpc_metadata_batch b) { grpc_stream_op *op = add(sopb); - grpc_metadata_batch_assert_ok(&b); op->type = GRPC_OP_METADATA; op->data.metadata = b; - grpc_metadata_batch_assert_ok(&op->data.metadata); - assert_contained_metadata_ok(sopb->ops, sopb->nops); } void grpc_sopb_add_slice(grpc_stream_op_buffer *sopb, gpr_slice slice) { grpc_stream_op *op = add(sopb); op->type = GRPC_OP_SLICE; op->data.slice = slice; - assert_contained_metadata_ok(sopb->ops, sopb->nops); } void grpc_sopb_append(grpc_stream_op_buffer *sopb, grpc_stream_op *ops, @@ -161,15 +140,12 @@ void grpc_sopb_append(grpc_stream_op_buffer *sopb, grpc_stream_op *ops, size_t orig_nops = sopb->nops; size_t new_nops = orig_nops + nops; - assert_contained_metadata_ok(ops, nops); - assert_contained_metadata_ok(sopb->ops, sopb->nops); if (new_nops > sopb->capacity) { expandto(sopb, GPR_MAX(GROW(sopb->capacity), new_nops)); } memcpy(sopb->ops + orig_nops, ops, sizeof(grpc_stream_op) * nops); sopb->nops = new_nops; - assert_contained_metadata_ok(sopb->ops, sopb->nops); } static void assert_valid_list(grpc_mdelem_list *list) { -- cgit v1.2.3 From 9c71b6f5b670098667cd6cba93ad96018b975e44 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Fri, 24 Apr 2015 16:02:00 -0700 Subject: Fix memory leak --- src/core/surface/call.c | 10 ++-------- src/core/transport/chttp2_transport.c | 4 +++- 2 files changed, 5 insertions(+), 9 deletions(-) (limited to 'src/core/transport/chttp2_transport.c') diff --git a/src/core/surface/call.c b/src/core/surface/call.c index e80cf9e293..134759e0c1 100644 --- a/src/core/surface/call.c +++ b/src/core/surface/call.c @@ -288,9 +288,8 @@ grpc_call *grpc_call_create(grpc_channel *channel, grpc_completion_queue *cq, grpc_sopb_init(&call->send_ops); grpc_sopb_init(&call->recv_ops); gpr_slice_buffer_init(&call->incoming_message); - /* one ref is dropped in response to destroy, the other in - stream_closed */ - gpr_ref_init(&call->internal_refcount, 2); + /* dropped in destroy */ + gpr_ref_init(&call->internal_refcount, 1); /* server hack: start reads immediately so we can get initial metadata. TODO(ctiller): figure out a cleaner solution */ if (!call->is_client) { @@ -688,7 +687,6 @@ static int add_slice_to_message(grpc_call *call, gpr_slice slice) { static void call_on_done_recv(void *pc, int success) { grpc_call *call = pc; size_t i; - int unref_due_to_connection_close = 0; gpr_log(GPR_DEBUG, "%s %p", __FUNCTION__, call); lock(call); call->receiving = 0; @@ -716,7 +714,6 @@ static void call_on_done_recv(void *pc, int success) { if (call->recv_state == GRPC_STREAM_CLOSED) { GPR_ASSERT(call->read_state <= READ_STATE_STREAM_CLOSED); call->read_state = READ_STATE_STREAM_CLOSED; - unref_due_to_connection_close = 1; } finish_read_ops(call); } else { @@ -731,9 +728,6 @@ static void call_on_done_recv(void *pc, int success) { unlock(call); grpc_call_internal_unref(call, "receiving", 0); - if (unref_due_to_connection_close) { - grpc_call_internal_unref(call, "live", 0); - } } static grpc_mdelem_list chain_metadata_from_app(grpc_call *call, size_t count, diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index 0640c848d7..238bcacfaa 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -665,7 +665,9 @@ static void destroy_stream(grpc_transport *gt, grpc_stream *gs) { grpc_sopb_destroy(&s->writing_sopb); grpc_sopb_destroy(&s->callback_sopb); grpc_chttp2_data_parser_destroy(&s->parser); - GPR_ASSERT(s->incoming_metadata_count == 0); + for (i = 0; i < s->incoming_metadata_count; i++) { + grpc_mdelem_unref(s->incoming_metadata[i].md); + } gpr_free(s->incoming_metadata); unref_transport(t); -- cgit v1.2.3 From 65f9f81afbb4d78496f3032246e97970720191cd Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Fri, 24 Apr 2015 16:53:20 -0700 Subject: Fix memory leak --- src/core/surface/call.c | 10 ++++++---- src/core/transport/chttp2_transport.c | 3 ++- src/core/transport/transport_op_string.c | 1 + 3 files changed, 9 insertions(+), 5 deletions(-) (limited to 'src/core/transport/chttp2_transport.c') diff --git a/src/core/surface/call.c b/src/core/surface/call.c index 134759e0c1..995e2dab7e 100644 --- a/src/core/surface/call.c +++ b/src/core/surface/call.c @@ -320,7 +320,7 @@ grpc_completion_queue *grpc_call_get_completion_queue(grpc_call *call) { } void grpc_call_internal_ref(grpc_call *c, const char *reason) { -gpr_log(GPR_DEBUG, "grpc_call_internal_ref: %p %s %d -> %d", c, reason, c->internal_refcount.count, c->internal_refcount.count+1); +gpr_log(GPR_DEBUG, "grpc_call_internal_ref: %p %d %s %d -> %d", c, c->is_client, reason, c->internal_refcount.count, c->internal_refcount.count+1); gpr_ref(&c->internal_refcount); } static void destroy_call(void *call, int ignored_success) { @@ -355,7 +355,7 @@ static void destroy_call(void *call, int ignored_success) { } void grpc_call_internal_unref(grpc_call *c, const char *reason, int allow_immediate_deletion) { -gpr_log(GPR_DEBUG, "grpc_call_internal_unref: %p %s %d -> %d", c, reason, c->internal_refcount.count, c->internal_refcount.count-1); +gpr_log(GPR_DEBUG, "grpc_call_internal_unref: %p %d %s %d -> %d", c, c->is_client, reason, c->internal_refcount.count, c->internal_refcount.count-1); if (gpr_unref(&c->internal_refcount)) { if (allow_immediate_deletion) { destroy_call(c, 1); @@ -408,12 +408,13 @@ static int is_op_live(grpc_call *call, grpc_ioreq_op op) { static void lock(grpc_call *call) { gpr_mu_lock(&call->mu); } static int need_more_data(grpc_call *call) { + gpr_log(GPR_DEBUG, "st: %d%d%d%d%d%d%d", is_op_live(call, GRPC_IOREQ_RECV_INITIAL_METADATA), is_op_live(call, GRPC_IOREQ_RECV_MESSAGE), is_op_live(call, GRPC_IOREQ_RECV_TRAILING_METADATA), is_op_live(call, GRPC_IOREQ_RECV_STATUS), is_op_live(call, GRPC_IOREQ_RECV_STATUS_DETAILS), is_op_live(call, GRPC_IOREQ_RECV_CLOSE), (call->write_state == WRITE_STATE_INITIAL && !call->is_client && call->read_state != READ_STATE_STREAM_CLOSED)); return is_op_live(call, GRPC_IOREQ_RECV_INITIAL_METADATA) || is_op_live(call, GRPC_IOREQ_RECV_MESSAGE) || is_op_live(call, GRPC_IOREQ_RECV_TRAILING_METADATA) || is_op_live(call, GRPC_IOREQ_RECV_STATUS) || is_op_live(call, GRPC_IOREQ_RECV_STATUS_DETAILS) || - is_op_live(call, GRPC_IOREQ_RECV_CLOSE) || + (is_op_live(call, GRPC_IOREQ_RECV_CLOSE) && grpc_bbq_empty(&call->incoming_queue)) || (call->write_state == WRITE_STATE_INITIAL && !call->is_client && call->read_state != READ_STATE_STREAM_CLOSED); } @@ -687,7 +688,7 @@ static int add_slice_to_message(grpc_call *call, gpr_slice slice) { static void call_on_done_recv(void *pc, int success) { grpc_call *call = pc; size_t i; - gpr_log(GPR_DEBUG, "%s %p", __FUNCTION__, call); + gpr_log(GPR_DEBUG, "%s %p succ=%d rcvs=%d rds0=%d", __FUNCTION__, call, success, call->recv_state, call->read_state); lock(call); call->receiving = 0; if (success) { @@ -715,6 +716,7 @@ static void call_on_done_recv(void *pc, int success) { GPR_ASSERT(call->read_state <= READ_STATE_STREAM_CLOSED); call->read_state = READ_STATE_STREAM_CLOSED; } + gpr_log(GPR_DEBUG, "%p rds1=%d", call, call->read_state); finish_read_ops(call); } else { finish_ioreq_op(call, GRPC_IOREQ_RECV_MESSAGE, GRPC_OP_ERROR); diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index 238bcacfaa..545d9ef78e 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -957,7 +957,7 @@ static void finish_write_common(transport *t, int success) { } while ((s = stream_list_remove_head(t, WRITTEN_CLOSED))) { s->write_state = WRITE_STATE_SENT_CLOSE; - if (!s->cancelled) { + if (1||!s->cancelled) { maybe_finish_read(t, s); } } @@ -1916,6 +1916,7 @@ static void finish_reads(transport *t) { GPR_ASSERT(s->incoming_sopb); *s->publish_state = compute_state(s->write_state == WRITE_STATE_SENT_CLOSE, s->read_closed); + gpr_log(GPR_DEBUG, "FR: %p pub=%d known=%d ws=%d rc=%d", s, *s->publish_state, s->published_state, s->write_state, s->read_closed); if (*s->publish_state != s->published_state) { s->published_state = *s->publish_state; publish = 1; diff --git a/src/core/transport/transport_op_string.c b/src/core/transport/transport_op_string.c index a215710969..04c1d79022 100644 --- a/src/core/transport/transport_op_string.c +++ b/src/core/transport/transport_op_string.c @@ -87,6 +87,7 @@ char *grpc_sopb_string(grpc_stream_op_buffer *sopb) { break; case GRPC_OP_SLICE: gpr_asprintf(&tmp, "SLICE:%d", GPR_SLICE_LENGTH(op->data.slice)); + gpr_strvec_add(&b, tmp); break; case GRPC_OP_METADATA: gpr_strvec_add(&b, gpr_strdup("METADATA{")); -- cgit v1.2.3 From 64bc3fdaaaae22bab48d1ebda27c9791ea160590 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Fri, 24 Apr 2015 17:07:12 -0700 Subject: remove annotations --- src/core/surface/call.c | 29 ++++++++++++----------------- src/core/surface/call.h | 4 ++-- src/core/surface/completion_queue.c | 4 ++-- src/core/surface/server.c | 2 +- src/core/transport/chttp2_transport.c | 1 - 5 files changed, 17 insertions(+), 23 deletions(-) (limited to 'src/core/transport/chttp2_transport.c') diff --git a/src/core/surface/call.c b/src/core/surface/call.c index 995e2dab7e..a65bfb8903 100644 --- a/src/core/surface/call.c +++ b/src/core/surface/call.c @@ -299,7 +299,7 @@ grpc_call *grpc_call_create(grpc_channel *channel, grpc_completion_queue *cq, initial_op.on_done_recv = call_on_done_recv; initial_op.recv_user_data = call; call->receiving = 1; - grpc_call_internal_ref(call, "receiving-0"); + grpc_call_internal_ref(call); initial_op_ptr = &initial_op; } grpc_call_stack_init(channel_stack, server_transport_data, initial_op_ptr, @@ -319,8 +319,7 @@ grpc_completion_queue *grpc_call_get_completion_queue(grpc_call *call) { return call->cq; } -void grpc_call_internal_ref(grpc_call *c, const char *reason) { -gpr_log(GPR_DEBUG, "grpc_call_internal_ref: %p %d %s %d -> %d", c, c->is_client, reason, c->internal_refcount.count, c->internal_refcount.count+1); +void grpc_call_internal_ref(grpc_call *c) { gpr_ref(&c->internal_refcount); } static void destroy_call(void *call, int ignored_success) { @@ -354,8 +353,7 @@ static void destroy_call(void *call, int ignored_success) { gpr_free(c); } -void grpc_call_internal_unref(grpc_call *c, const char *reason, int allow_immediate_deletion) { -gpr_log(GPR_DEBUG, "grpc_call_internal_unref: %p %d %s %d -> %d", c, c->is_client, reason, c->internal_refcount.count, c->internal_refcount.count-1); +void grpc_call_internal_unref(grpc_call *c, int allow_immediate_deletion) { if (gpr_unref(&c->internal_refcount)) { if (allow_immediate_deletion) { destroy_call(c, 1); @@ -408,7 +406,6 @@ static int is_op_live(grpc_call *call, grpc_ioreq_op op) { static void lock(grpc_call *call) { gpr_mu_lock(&call->mu); } static int need_more_data(grpc_call *call) { - gpr_log(GPR_DEBUG, "st: %d%d%d%d%d%d%d", is_op_live(call, GRPC_IOREQ_RECV_INITIAL_METADATA), is_op_live(call, GRPC_IOREQ_RECV_MESSAGE), is_op_live(call, GRPC_IOREQ_RECV_TRAILING_METADATA), is_op_live(call, GRPC_IOREQ_RECV_STATUS), is_op_live(call, GRPC_IOREQ_RECV_STATUS_DETAILS), is_op_live(call, GRPC_IOREQ_RECV_CLOSE), (call->write_state == WRITE_STATE_INITIAL && !call->is_client && call->read_state != READ_STATE_STREAM_CLOSED)); return is_op_live(call, GRPC_IOREQ_RECV_INITIAL_METADATA) || is_op_live(call, GRPC_IOREQ_RECV_MESSAGE) || is_op_live(call, GRPC_IOREQ_RECV_TRAILING_METADATA) || @@ -433,14 +430,14 @@ static void unlock(grpc_call *call) { op.on_done_recv = call_on_done_recv; op.recv_user_data = call; call->receiving = 1; - grpc_call_internal_ref(call, "receiving"); + grpc_call_internal_ref(call); start_op = 1; } if (!call->sending) { if (fill_send_ops(call, &op)) { call->sending = 1; - grpc_call_internal_ref(call, "sending"); + grpc_call_internal_ref(call); start_op = 1; } } @@ -451,7 +448,7 @@ static void unlock(grpc_call *call) { sizeof(completed_requests)); call->num_completed_requests = 0; call->completing = 1; - grpc_call_internal_ref(call, "completing"); + grpc_call_internal_ref(call); } gpr_mu_unlock(&call->mu); @@ -468,7 +465,7 @@ static void unlock(grpc_call *call) { lock(call); call->completing = 0; unlock(call); - grpc_call_internal_unref(call, "completing", 0); + grpc_call_internal_unref(call, 0); } } @@ -608,7 +605,7 @@ static void call_on_done_send(void *pc, int success) { call->last_send_contains = 0; call->sending = 0; unlock(call); - grpc_call_internal_unref(call, "sending", 0); + grpc_call_internal_unref(call, 0); } static void finish_message(grpc_call *call) { @@ -688,7 +685,6 @@ static int add_slice_to_message(grpc_call *call, gpr_slice slice) { static void call_on_done_recv(void *pc, int success) { grpc_call *call = pc; size_t i; - gpr_log(GPR_DEBUG, "%s %p succ=%d rcvs=%d rds0=%d", __FUNCTION__, call, success, call->recv_state, call->read_state); lock(call); call->receiving = 0; if (success) { @@ -716,7 +712,6 @@ static void call_on_done_recv(void *pc, int success) { GPR_ASSERT(call->read_state <= READ_STATE_STREAM_CLOSED); call->read_state = READ_STATE_STREAM_CLOSED; } - gpr_log(GPR_DEBUG, "%p rds1=%d", call, call->read_state); finish_read_ops(call); } else { finish_ioreq_op(call, GRPC_IOREQ_RECV_MESSAGE, GRPC_OP_ERROR); @@ -729,7 +724,7 @@ static void call_on_done_recv(void *pc, int success) { call->recv_ops.nops = 0; unlock(call); - grpc_call_internal_unref(call, "receiving", 0); + grpc_call_internal_unref(call, 0); } static grpc_mdelem_list chain_metadata_from_app(grpc_call *call, size_t count, @@ -987,7 +982,7 @@ void grpc_call_destroy(grpc_call *c) { cancel = c->read_state != READ_STATE_STREAM_CLOSED; unlock(c); if (cancel) grpc_call_cancel(c); - grpc_call_internal_unref(c, "destroy", 1); + grpc_call_internal_unref(c, 1); } grpc_call_error grpc_call_cancel(grpc_call *call) { @@ -1034,7 +1029,7 @@ static void call_alarm(void *arg, int success) { grpc_call_cancel(call); } } - grpc_call_internal_unref(call, "alarm", 1); + grpc_call_internal_unref(call, 1); } static void set_deadline_alarm(grpc_call *call, gpr_timespec deadline) { @@ -1043,7 +1038,7 @@ static void set_deadline_alarm(grpc_call *call, gpr_timespec deadline) { assert(0); return; } - grpc_call_internal_ref(call, "alarm"); + grpc_call_internal_ref(call); call->have_alarm = 1; grpc_alarm_init(&call->alarm, deadline, call_alarm, call, gpr_now()); } diff --git a/src/core/surface/call.h b/src/core/surface/call.h index f0c31e3a0d..199beb1738 100644 --- a/src/core/surface/call.h +++ b/src/core/surface/call.h @@ -93,8 +93,8 @@ grpc_call *grpc_call_create(grpc_channel *channel, grpc_completion_queue *cq, void grpc_call_set_completion_queue(grpc_call *call, grpc_completion_queue *cq); grpc_completion_queue *grpc_call_get_completion_queue(grpc_call *call); -void grpc_call_internal_ref(grpc_call *call, const char *reason); -void grpc_call_internal_unref(grpc_call *call, const char *reason, int allow_immediate_deletion); +void grpc_call_internal_ref(grpc_call *call); +void grpc_call_internal_unref(grpc_call *call, int allow_immediate_deletion); grpc_call_error grpc_call_start_ioreq_and_call_back( grpc_call *call, const grpc_ioreq *reqs, size_t nreqs, diff --git a/src/core/surface/completion_queue.c b/src/core/surface/completion_queue.c index 4398c43f5e..24f4a05071 100644 --- a/src/core/surface/completion_queue.c +++ b/src/core/surface/completion_queue.c @@ -135,7 +135,7 @@ static event *add_locked(grpc_completion_queue *cc, grpc_completion_type type, void grpc_cq_begin_op(grpc_completion_queue *cc, grpc_call *call, grpc_completion_type type) { gpr_ref(&cc->refs); - if (call) grpc_call_internal_ref(call, "cq"); + if (call) grpc_call_internal_ref(call); #ifndef NDEBUG gpr_atm_no_barrier_fetch_add(&cc->pending_op_count[type], 1); #endif @@ -409,7 +409,7 @@ 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); + grpc_call_internal_unref(ev->base.call, 1); } gpr_free(ev); } diff --git a/src/core/surface/server.c b/src/core/surface/server.c index 254a1a64bd..2f00ad0bc6 100644 --- a/src/core/surface/server.c +++ b/src/core/surface/server.c @@ -1131,7 +1131,7 @@ static void begin_call(grpc_server *server, call_data *calld, break; } - grpc_call_internal_ref(calld->call, "server"); + grpc_call_internal_ref(calld->call); grpc_call_start_ioreq_and_call_back(calld->call, req, r - req, publish, rc->tag); } diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index 545d9ef78e..6dd8f0cde1 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -1916,7 +1916,6 @@ static void finish_reads(transport *t) { GPR_ASSERT(s->incoming_sopb); *s->publish_state = compute_state(s->write_state == WRITE_STATE_SENT_CLOSE, s->read_closed); - gpr_log(GPR_DEBUG, "FR: %p pub=%d known=%d ws=%d rc=%d", s, *s->publish_state, s->published_state, s->write_state, s->read_closed); if (*s->publish_state != s->published_state) { s->published_state = *s->publish_state; publish = 1; -- cgit v1.2.3 From 2b0f7c5b1eb50f2d62e7fad757c1be6f007de28b Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Fri, 24 Apr 2015 17:23:17 -0700 Subject: Thread safety fix --- src/core/transport/chttp2_transport.c | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) (limited to 'src/core/transport/chttp2_transport.c') diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index 6dd8f0cde1..9f9005691b 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -317,6 +317,7 @@ struct stream { grpc_linked_mdelem *incoming_metadata; size_t incoming_metadata_count; size_t incoming_metadata_capacity; + grpc_linked_mdelem *old_incoming_metadata; gpr_timespec incoming_deadline; /* sops from application */ @@ -669,6 +670,7 @@ static void destroy_stream(grpc_transport *gt, grpc_stream *gs) { grpc_mdelem_unref(s->incoming_metadata[i].md); } gpr_free(s->incoming_metadata); + gpr_free(s->old_incoming_metadata); unref_transport(t); } @@ -1059,6 +1061,8 @@ static void perform_op_locked(transport *t, stream *s, grpc_transport_op *op) { s->incoming_sopb = op->recv_ops; s->incoming_sopb->nops = 0; s->publish_state = op->recv_state; + gpr_free(s->old_incoming_metadata); + s->old_incoming_metadata = NULL; maybe_finish_read(t, s); maybe_join_window_updates(t, s); } @@ -1904,8 +1908,10 @@ static void patch_metadata_ops(stream *s) { mdidx = last_mdidx; } GPR_ASSERT(mdidx == s->incoming_metadata_count); - + s->old_incoming_metadata = s->incoming_metadata; + s->incoming_metadata = NULL; s->incoming_metadata_count = 0; + s->incoming_metadata_capacity = 0; } static void finish_reads(transport *t) { -- cgit v1.2.3 From 59abfc234b666c9e5e667cf6dfb7fb786888d40a Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Mon, 27 Apr 2015 09:29:35 -0700 Subject: Properly handle non-full metadata batches --- src/core/transport/chttp2_transport.c | 31 ++++++++++++++++++++++++++----- 1 file changed, 26 insertions(+), 5 deletions(-) (limited to 'src/core/transport/chttp2_transport.c') diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index 9f9005691b..26c550c1f1 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -1890,13 +1890,22 @@ static void patch_metadata_ops(stream *s) { size_t j; size_t mdidx = 0; size_t last_mdidx; + int found_metadata = 0; + /* rework the array of metadata into a linked list, making use + of the breadcrumbs we left in metadata batches during + add_metadata_batch */ for (i = 0; i < nops; i++) { grpc_stream_op *op = &ops[i]; if (op->type != GRPC_OP_METADATA) continue; + found_metadata = 1; + /* we left a breadcrumb indicating where the end of this list is, + and since we add sequentially, we know from the end of the last + segment where this segment begins */ last_mdidx = (size_t)(gpr_intptr)(op->data.metadata.list.tail); GPR_ASSERT(last_mdidx > mdidx); GPR_ASSERT(last_mdidx <= s->incoming_metadata_count); + /* turn the array into a doubly linked list */ op->data.metadata.list.head = &s->incoming_metadata[mdidx]; op->data.metadata.list.tail = &s->incoming_metadata[last_mdidx - 1]; for (j = mdidx + 1; j < last_mdidx; j++) { @@ -1905,13 +1914,25 @@ static void patch_metadata_ops(stream *s) { } s->incoming_metadata[mdidx].prev = NULL; s->incoming_metadata[last_mdidx-1].next = NULL; + /* track where we're up to */ mdidx = last_mdidx; } - GPR_ASSERT(mdidx == s->incoming_metadata_count); - s->old_incoming_metadata = s->incoming_metadata; - s->incoming_metadata = NULL; - s->incoming_metadata_count = 0; - s->incoming_metadata_capacity = 0; + if (found_metadata) { + s->old_incoming_metadata = s->incoming_metadata; + if (mdidx != s->incoming_metadata_count) { + /* we have a partially read metadata batch still in incoming_metadata */ + size_t new_count = s->incoming_metadata_count - mdidx; + size_t copy_bytes = sizeof(*s->incoming_metadata) * new_count; + GPR_ASSERT(mdidx < s->incoming_metadata_count); + s->incoming_metadata = gpr_malloc(copy_bytes); + memcpy(s->old_incoming_metadata + mdidx, s->incoming_metadata, copy_bytes); + s->incoming_metadata_count = s->incoming_metadata_capacity = new_count; + } else { + s->incoming_metadata = NULL; + s->incoming_metadata_count = 0; + s->incoming_metadata_capacity = 0; + } + } } static void finish_reads(transport *t) { -- cgit v1.2.3 From 8a5bce35419abd3d0b51498afcdd091b39683930 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Wed, 29 Apr 2015 07:49:47 -0700 Subject: Incoming stream id validation fixes - correct log message on an invalid stream id - add an additional check that the low bit indicates a client stream id on the server --- src/core/transport/chttp2_transport.c | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) (limited to 'src/core/transport/chttp2_transport.c') diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index e32ee284e0..1bb6e7f960 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -1327,7 +1327,10 @@ static int init_header_frame_parser(transport *t, int is_continuation) { gpr_log(GPR_ERROR, "ignoring out of order new stream request on server; last stream " "id=%d, new stream id=%d", - t->last_incoming_stream_id, t->incoming_stream); + t->last_incoming_stream_id, t->incoming_stream_id); + return init_skip_frame(t, 1); + } else if ((t->incoming_stream_id & 1) == 0) { + gpr_log(GPR_ERROR, "ignoring stream with non-client generated index %d", t->incoming_stream_id); return init_skip_frame(t, 1); } t->incoming_stream = NULL; -- cgit v1.2.3