diff options
author | 2016-10-26 21:00:53 -0700 | |
---|---|---|
committer | 2016-10-26 21:00:53 -0700 | |
commit | 6e080a36515455f2db5b07465abdeacf3e21c3be (patch) | |
tree | 0aa4c9e0297a4408206c93d612efe3852c19fa57 /src/core/ext/transport | |
parent | f6f9cf39c677751fb46d42f303a07687e9bfb26a (diff) | |
parent | 3d357d901c5228cf8989aed8949b20d32baad77b (diff) |
Merge branch 'buffer_pools_for_realsies' into grpc_slice
Diffstat (limited to 'src/core/ext/transport')
6 files changed, 98 insertions, 134 deletions
diff --git a/src/core/ext/transport/chttp2/alpn/alpn.c b/src/core/ext/transport/chttp2/alpn/alpn.c index 48b0217265..55710dc5ae 100644 --- a/src/core/ext/transport/chttp2/alpn/alpn.c +++ b/src/core/ext/transport/chttp2/alpn/alpn.c @@ -36,7 +36,7 @@ #include <grpc/support/useful.h> /* in order of preference */ -static const char *const supported_versions[] = {"h2"}; +static const char *const supported_versions[] = {"grpc-exp", "h2"}; int grpc_chttp2_is_alpn_version_supported(const char *version, size_t size) { size_t i; diff --git a/src/core/ext/transport/chttp2/client/insecure/channel_create.c b/src/core/ext/transport/chttp2/client/insecure/channel_create.c index 0c4dbee8d9..03dc955b27 100644 --- a/src/core/ext/transport/chttp2/client/insecure/channel_create.c +++ b/src/core/ext/transport/chttp2/client/insecure/channel_create.c @@ -52,6 +52,10 @@ #include "src/core/lib/surface/api_trace.h" #include "src/core/lib/surface/channel.h" +// +// connector +// + typedef struct { grpc_connector base; gpr_refcount refs; @@ -151,41 +155,26 @@ static void connector_connect(grpc_exec_ctx *exec_ctx, grpc_connector *con, grpc_closure_init(&c->connected, connected, c); grpc_tcp_client_connect(exec_ctx, &c->connected, &c->tcp, args->interested_parties, args->channel_args, - args->addr, args->addr_len, args->deadline); + args->addr, args->deadline); } static const grpc_connector_vtable connector_vtable = { connector_ref, connector_unref, connector_shutdown, connector_connect}; -typedef struct { - grpc_client_channel_factory base; - gpr_refcount refs; - grpc_channel_args *merge_args; -} client_channel_factory; +// +// client_channel_factory +// static void client_channel_factory_ref( - grpc_client_channel_factory *cc_factory) { - client_channel_factory *f = (client_channel_factory *)cc_factory; - gpr_ref(&f->refs); -} + grpc_client_channel_factory *cc_factory) {} static void client_channel_factory_unref( - grpc_exec_ctx *exec_ctx, grpc_client_channel_factory *cc_factory) { - client_channel_factory *f = (client_channel_factory *)cc_factory; - if (gpr_unref(&f->refs)) { - grpc_channel_args_destroy(f->merge_args); - gpr_free(f); - } -} + grpc_exec_ctx *exec_ctx, grpc_client_channel_factory *cc_factory) {} static grpc_subchannel *client_channel_factory_create_subchannel( grpc_exec_ctx *exec_ctx, grpc_client_channel_factory *cc_factory, - grpc_subchannel_args *args) { - client_channel_factory *f = (client_channel_factory *)cc_factory; + const grpc_subchannel_args *args) { connector *c = gpr_malloc(sizeof(*c)); - grpc_channel_args *final_args = - grpc_channel_args_merge(args->args, f->merge_args); - grpc_subchannel *s; memset(c, 0, sizeof(*c)); c->base.vtable = &connector_vtable; gpr_ref_init(&c->refs, 1); @@ -197,23 +186,18 @@ static grpc_subchannel *client_channel_factory_create_subchannel( grpc_http_connect_handshaker_create(proxy_name, args->server_name)); gpr_free(proxy_name); } - args->args = final_args; - s = grpc_subchannel_create(exec_ctx, &c->base, args); + grpc_subchannel *s = grpc_subchannel_create(exec_ctx, &c->base, args); grpc_connector_unref(exec_ctx, &c->base); - grpc_channel_args_destroy(final_args); return s; } static grpc_channel *client_channel_factory_create_channel( grpc_exec_ctx *exec_ctx, grpc_client_channel_factory *cc_factory, const char *target, grpc_client_channel_type type, - grpc_channel_args *args) { - client_channel_factory *f = (client_channel_factory *)cc_factory; - grpc_channel_args *final_args = grpc_channel_args_merge(args, f->merge_args); - grpc_channel *channel = grpc_channel_create(exec_ctx, target, final_args, - GRPC_CLIENT_CHANNEL, NULL); - grpc_channel_args_destroy(final_args); - grpc_resolver *resolver = grpc_resolver_create(target); + const grpc_channel_args *args) { + grpc_channel *channel = + grpc_channel_create(exec_ctx, target, args, GRPC_CLIENT_CHANNEL, NULL); + grpc_resolver *resolver = grpc_resolver_create(target, args); if (!resolver) { GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, channel, "client_channel_factory_create_channel"); @@ -221,7 +205,7 @@ static grpc_channel *client_channel_factory_create_channel( } grpc_client_channel_finish_initialization( - exec_ctx, grpc_channel_get_channel_stack(channel), resolver, &f->base); + exec_ctx, grpc_channel_get_channel_stack(channel), resolver, cc_factory); GRPC_RESOLVER_UNREF(exec_ctx, resolver, "create_channel"); return channel; @@ -232,6 +216,9 @@ static const grpc_client_channel_factory_vtable client_channel_factory_vtable = client_channel_factory_create_subchannel, client_channel_factory_create_channel}; +static grpc_client_channel_factory client_channel_factory = { + &client_channel_factory_vtable}; + /* Create a client channel: Asynchronously: - resolve target - connect to it (trying alternatives as presented) @@ -245,16 +232,12 @@ grpc_channel *grpc_insecure_channel_create(const char *target, (target, args, reserved)); GPR_ASSERT(!reserved); - client_channel_factory *f = gpr_malloc(sizeof(*f)); - memset(f, 0, sizeof(*f)); - f->base.vtable = &client_channel_factory_vtable; - gpr_ref_init(&f->refs, 1); - f->merge_args = grpc_channel_args_copy(args); - + grpc_client_channel_factory *factory = + (grpc_client_channel_factory *)&client_channel_factory; grpc_channel *channel = client_channel_factory_create_channel( - &exec_ctx, &f->base, target, GRPC_CLIENT_CHANNEL_TYPE_REGULAR, NULL); + &exec_ctx, factory, target, GRPC_CLIENT_CHANNEL_TYPE_REGULAR, args); - grpc_client_channel_factory_unref(&exec_ctx, &f->base); + grpc_client_channel_factory_unref(&exec_ctx, factory); grpc_exec_ctx_finish(&exec_ctx); return channel != NULL ? channel : grpc_lame_client_channel_create( diff --git a/src/core/ext/transport/chttp2/client/secure/secure_channel_create.c b/src/core/ext/transport/chttp2/client/secure/secure_channel_create.c index c135f69820..7f9c365799 100644 --- a/src/core/ext/transport/chttp2/client/secure/secure_channel_create.c +++ b/src/core/ext/transport/chttp2/client/secure/secure_channel_create.c @@ -54,6 +54,10 @@ #include "src/core/lib/surface/channel.h" #include "src/core/lib/tsi/transport_security_interface.h" +// +// connector +// + typedef struct { grpc_connector base; gpr_refcount refs; @@ -208,19 +212,21 @@ static void connector_connect(grpc_exec_ctx *exec_ctx, grpc_connector *con, GPR_ASSERT(c->connecting_endpoint == NULL); gpr_mu_unlock(&c->mu); grpc_closure_init(&c->connected_closure, connected, c); - grpc_tcp_client_connect(exec_ctx, &c->connected_closure, - &c->newly_connecting_endpoint, - args->interested_parties, args->channel_args, - args->addr, args->addr_len, args->deadline); + grpc_tcp_client_connect( + exec_ctx, &c->connected_closure, &c->newly_connecting_endpoint, + args->interested_parties, args->channel_args, args->addr, args->deadline); } static const grpc_connector_vtable connector_vtable = { connector_ref, connector_unref, connector_shutdown, connector_connect}; +// +// client_channel_factory +// + typedef struct { grpc_client_channel_factory base; gpr_refcount refs; - grpc_channel_args *merge_args; grpc_channel_security_connector *security_connector; } client_channel_factory; @@ -236,19 +242,15 @@ static void client_channel_factory_unref( if (gpr_unref(&f->refs)) { GRPC_SECURITY_CONNECTOR_UNREF(&f->security_connector->base, "client_channel_factory"); - grpc_channel_args_destroy(f->merge_args); gpr_free(f); } } static grpc_subchannel *client_channel_factory_create_subchannel( grpc_exec_ctx *exec_ctx, grpc_client_channel_factory *cc_factory, - grpc_subchannel_args *args) { + const grpc_subchannel_args *args) { client_channel_factory *f = (client_channel_factory *)cc_factory; connector *c = gpr_malloc(sizeof(*c)); - grpc_channel_args *final_args = - grpc_channel_args_merge(args->args, f->merge_args); - grpc_subchannel *s; memset(c, 0, sizeof(*c)); c->base.vtable = &connector_vtable; c->security_connector = f->security_connector; @@ -262,25 +264,19 @@ static grpc_subchannel *client_channel_factory_create_subchannel( } gpr_mu_init(&c->mu); gpr_ref_init(&c->refs, 1); - args->args = final_args; - s = grpc_subchannel_create(exec_ctx, &c->base, args); + grpc_subchannel *s = grpc_subchannel_create(exec_ctx, &c->base, args); grpc_connector_unref(exec_ctx, &c->base); - grpc_channel_args_destroy(final_args); return s; } static grpc_channel *client_channel_factory_create_channel( grpc_exec_ctx *exec_ctx, grpc_client_channel_factory *cc_factory, const char *target, grpc_client_channel_type type, - grpc_channel_args *args) { + const grpc_channel_args *args) { client_channel_factory *f = (client_channel_factory *)cc_factory; - - grpc_channel_args *final_args = grpc_channel_args_merge(args, f->merge_args); - grpc_channel *channel = grpc_channel_create(exec_ctx, target, final_args, - GRPC_CLIENT_CHANNEL, NULL); - grpc_channel_args_destroy(final_args); - - grpc_resolver *resolver = grpc_resolver_create(target); + grpc_channel *channel = + grpc_channel_create(exec_ctx, target, args, GRPC_CLIENT_CHANNEL, NULL); + grpc_resolver *resolver = grpc_resolver_create(target, args); if (resolver != NULL) { grpc_client_channel_finish_initialization( exec_ctx, grpc_channel_get_channel_stack(channel), resolver, &f->base); @@ -290,7 +286,6 @@ static grpc_channel *client_channel_factory_create_channel( "client_channel_factory_create_channel"); channel = NULL; } - return channel; } @@ -307,19 +302,13 @@ grpc_channel *grpc_secure_channel_create(grpc_channel_credentials *creds, const char *target, const grpc_channel_args *args, void *reserved) { - grpc_arg connector_arg; - grpc_channel_args *args_copy; - grpc_channel_args *new_args_from_connector; - grpc_channel_security_connector *security_connector; - client_channel_factory *f; grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; - GRPC_API_TRACE( "grpc_secure_channel_create(creds=%p, target=%s, args=%p, " "reserved=%p)", 4, (creds, target, args, reserved)); GPR_ASSERT(reserved == NULL); - + // Make sure security connector does not already exist in args. if (grpc_find_security_connector_in_args(args) != NULL) { gpr_log(GPR_ERROR, "Cannot set security context in channel args."); grpc_exec_ctx_finish(&exec_ctx); @@ -327,7 +316,9 @@ grpc_channel *grpc_secure_channel_create(grpc_channel_credentials *creds, target, GRPC_STATUS_INTERNAL, "Security connector exists in channel args."); } - + // Create security connector and construct new channel args. + grpc_channel_security_connector *security_connector; + grpc_channel_args *new_args_from_connector; if (grpc_channel_credentials_create_security_connector( creds, target, args, &security_connector, &new_args_from_connector) != GRPC_SECURITY_OK) { @@ -335,35 +326,30 @@ grpc_channel *grpc_secure_channel_create(grpc_channel_credentials *creds, return grpc_lame_client_channel_create( target, GRPC_STATUS_INTERNAL, "Failed to create security connector."); } - - connector_arg = grpc_security_connector_to_arg(&security_connector->base); - args_copy = grpc_channel_args_copy_and_add( + grpc_arg connector_arg = + grpc_security_connector_to_arg(&security_connector->base); + grpc_channel_args *new_args = grpc_channel_args_copy_and_add( new_args_from_connector != NULL ? new_args_from_connector : args, &connector_arg, 1); - - f = gpr_malloc(sizeof(*f)); - memset(f, 0, sizeof(*f)); - f->base.vtable = &client_channel_factory_vtable; - gpr_ref_init(&f->refs, 1); - - f->merge_args = grpc_channel_args_copy(args_copy); - grpc_channel_args_destroy(args_copy); if (new_args_from_connector != NULL) { grpc_channel_args_destroy(new_args_from_connector); } - + // Create client channel factory. + client_channel_factory *f = gpr_malloc(sizeof(*f)); + memset(f, 0, sizeof(*f)); + f->base.vtable = &client_channel_factory_vtable; + gpr_ref_init(&f->refs, 1); GRPC_SECURITY_CONNECTOR_REF(&security_connector->base, "grpc_secure_channel_create"); f->security_connector = security_connector; - + // Create channel. grpc_channel *channel = client_channel_factory_create_channel( - &exec_ctx, &f->base, target, GRPC_CLIENT_CHANNEL_TYPE_REGULAR, NULL); - + &exec_ctx, &f->base, target, GRPC_CLIENT_CHANNEL_TYPE_REGULAR, new_args); + // Clean up. GRPC_SECURITY_CONNECTOR_UNREF(&f->security_connector->base, "client_channel_factory_create_channel"); - + grpc_channel_args_destroy(new_args); grpc_client_channel_factory_unref(&exec_ctx, &f->base); grpc_exec_ctx_finish(&exec_ctx); - return channel; /* may be NULL */ } diff --git a/src/core/ext/transport/chttp2/server/insecure/server_chttp2.c b/src/core/ext/transport/chttp2/server/insecure/server_chttp2.c index e5b4025c5f..c18d618f96 100644 --- a/src/core/ext/transport/chttp2/server/insecure/server_chttp2.c +++ b/src/core/ext/transport/chttp2/server/insecure/server_chttp2.c @@ -148,9 +148,7 @@ int grpc_server_add_insecure_http2_port(grpc_server *server, const char *addr) { const size_t naddrs = resolved->naddrs; errors = gpr_malloc(sizeof(*errors) * naddrs); for (i = 0; i < naddrs; i++) { - errors[i] = grpc_tcp_server_add_port( - tcp, (struct sockaddr *)&resolved->addrs[i].addr, - resolved->addrs[i].len, &port_temp); + errors[i] = grpc_tcp_server_add_port(tcp, &resolved->addrs[i], &port_temp); if (errors[i] == GRPC_ERROR_NONE) { if (port_num == -1) { port_num = port_temp; diff --git a/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c b/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c index a998dddd37..942638ad7f 100644 --- a/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c +++ b/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c @@ -287,9 +287,7 @@ int grpc_server_add_secure_http2_port(grpc_server *server, const char *addr, errors = gpr_malloc(sizeof(*errors) * resolved->naddrs); for (i = 0; i < resolved->naddrs; i++) { - errors[i] = grpc_tcp_server_add_port( - tcp, (struct sockaddr *)&resolved->addrs[i].addr, - resolved->addrs[i].len, &port_temp); + errors[i] = grpc_tcp_server_add_port(tcp, &resolved->addrs[i], &port_temp); if (errors[i] == GRPC_ERROR_NONE) { if (port_num == -1) { port_num = port_temp; diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c index 98190f7fb9..07639c1223 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c @@ -883,53 +883,51 @@ static bool contains_non_ok_status(grpc_metadata_batch *batch) { static void add_fetched_slice_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, - grpc_chttp2_stream *s); + grpc_chttp2_stream *s) { + s->fetched_send_message_length += + (uint32_t)GPR_SLICE_LENGTH(s->fetching_slice); + grpc_slice_buffer_add(&s->flow_controlled_buffer, s->fetching_slice); + if (s->id != 0) { + grpc_chttp2_become_writable(exec_ctx, t, s, true, "op.send_message"); + } +} static void continue_fetching_send_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_chttp2_stream *s) { - if (s->fetching_send_message == NULL) { - /* Stream was cancelled before message fetch completed */ - abort(); /* TODO(ctiller): what cleanup here? */ - return; - } - if (s->fetched_send_message_length == s->fetching_send_message->length) { - int64_t notify_offset = s->next_message_end_offset; - if (notify_offset <= s->flow_controlled_bytes_written) { - grpc_chttp2_complete_closure_step( - exec_ctx, t, s, &s->fetching_send_message_finished, GRPC_ERROR_NONE, - "fetching_send_message_finished"); - } else { - grpc_chttp2_write_cb *cb = t->write_cb_pool; - if (cb == NULL) { - cb = gpr_malloc(sizeof(*cb)); + for (;;) { + if (s->fetching_send_message == NULL) { + /* Stream was cancelled before message fetch completed */ + abort(); /* TODO(ctiller): what cleanup here? */ + return; /* early out */ + } + if (s->fetched_send_message_length == s->fetching_send_message->length) { + int64_t notify_offset = s->next_message_end_offset; + if (notify_offset <= s->flow_controlled_bytes_written) { + grpc_chttp2_complete_closure_step( + exec_ctx, t, s, &s->fetching_send_message_finished, GRPC_ERROR_NONE, + "fetching_send_message_finished"); } else { - t->write_cb_pool = cb->next; + grpc_chttp2_write_cb *cb = t->write_cb_pool; + if (cb == NULL) { + cb = gpr_malloc(sizeof(*cb)); + } else { + t->write_cb_pool = cb->next; + } + cb->call_at_byte = notify_offset; + cb->closure = s->fetching_send_message_finished; + s->fetching_send_message_finished = NULL; + cb->next = s->on_write_finished_cbs; + s->on_write_finished_cbs = cb; } - cb->call_at_byte = notify_offset; - cb->closure = s->fetching_send_message_finished; - s->fetching_send_message_finished = NULL; - cb->next = s->on_write_finished_cbs; - s->on_write_finished_cbs = cb; + s->fetching_send_message = NULL; + return; /* early out */ + } else if (grpc_byte_stream_next(exec_ctx, s->fetching_send_message, + &s->fetching_slice, UINT32_MAX, + &s->complete_fetch)) { + add_fetched_slice_locked(exec_ctx, t, s); } - s->fetching_send_message = NULL; - } else if (grpc_byte_stream_next(exec_ctx, s->fetching_send_message, - &s->fetching_slice, UINT32_MAX, - &s->complete_fetch)) { - add_fetched_slice_locked(exec_ctx, t, s); - } -} - -static void add_fetched_slice_locked(grpc_exec_ctx *exec_ctx, - grpc_chttp2_transport *t, - grpc_chttp2_stream *s) { - s->fetched_send_message_length += - (uint32_t)GPR_SLICE_LENGTH(s->fetching_slice); - grpc_slice_buffer_add(&s->flow_controlled_buffer, s->fetching_slice); - if (s->id != 0) { - grpc_chttp2_become_writable(exec_ctx, t, s, true, "op.send_message"); } - continue_fetching_send_locked(exec_ctx, t, s); } static void complete_fetch_locked(grpc_exec_ctx *exec_ctx, void *gs, @@ -938,6 +936,7 @@ static void complete_fetch_locked(grpc_exec_ctx *exec_ctx, void *gs, grpc_chttp2_transport *t = s->t; if (error == GRPC_ERROR_NONE) { add_fetched_slice_locked(exec_ctx, t, s); + continue_fetching_send_locked(exec_ctx, t, s); } else { /* TODO(ctiller): what to do here */ abort(); |