Merge pull request #8795 from markdroth/client_channel_init_cleanup

Client channel init cleanup
pull/9067/head^2
Mark D. Roth 8 years ago committed by GitHub
commit 5904a95274
  1. 10
      include/grpc/impl/codegen/grpc_types.h
  2. 7
      src/core/ext/census/grpc_filter.c
  3. 56
      src/core/ext/client_channel/client_channel.c
  4. 10
      src/core/ext/client_channel/client_channel.h
  5. 32
      src/core/ext/client_channel/client_channel_factory.c
  6. 6
      src/core/ext/client_channel/client_channel_factory.h
  7. 31
      src/core/ext/client_channel/http_connect_handshaker.c
  8. 5
      src/core/ext/client_channel/http_connect_handshaker.h
  9. 3
      src/core/ext/client_channel/lb_policy_factory.h
  10. 32
      src/core/ext/client_channel/resolver_registry.c
  11. 4
      src/core/ext/client_channel/resolver_registry.h
  12. 16
      src/core/ext/client_channel/subchannel.c
  13. 2
      src/core/ext/client_channel/subchannel.h
  14. 4
      src/core/ext/client_channel/subchannel_index.c
  15. 31
      src/core/ext/lb_policy/grpclb/grpclb.c
  16. 12
      src/core/ext/lb_policy/pick_first/pick_first.c
  17. 12
      src/core/ext/lb_policy/round_robin/round_robin.c
  18. 8
      src/core/ext/load_reporting/load_reporting_filter.c
  19. 7
      src/core/ext/resolver/dns/native/dns_resolver.c
  20. 7
      src/core/ext/resolver/sockaddr/sockaddr_resolver.c
  21. 11
      src/core/ext/transport/chttp2/client/chttp2_connector.c
  22. 3
      src/core/ext/transport/chttp2/client/chttp2_connector.h
  23. 33
      src/core/ext/transport/chttp2/client/insecure/channel_create.c
  24. 47
      src/core/ext/transport/chttp2/client/secure/secure_channel_create.c
  25. 24
      src/core/lib/channel/channel_stack.c
  26. 16
      src/core/lib/channel/channel_stack.h
  27. 46
      src/core/lib/channel/channel_stack_builder.c
  28. 11
      src/core/lib/channel/channel_stack_builder.h
  29. 7
      src/core/lib/channel/compress_filter.c
  30. 7
      src/core/lib/channel/connected_channel.c
  31. 7
      src/core/lib/channel/deadline_filter.c
  32. 7
      src/core/lib/channel/http_client_filter.c
  33. 7
      src/core/lib/channel/http_server_filter.c
  34. 7
      src/core/lib/channel/message_size_filter.c
  35. 7
      src/core/lib/security/transport/client_auth_filter.c
  36. 7
      src/core/lib/security/transport/server_auth_filter.c
  37. 119
      src/core/lib/surface/channel.c
  38. 7
      src/core/lib/surface/lame_client.c
  39. 7
      src/core/lib/surface/server.c
  40. 13
      src/cpp/common/channel_filter.h
  41. 7
      test/core/channel/channel_stack_test.c
  42. 5
      test/core/client_channel/resolvers/sockaddr_resolver_test.c
  43. 7
      test/core/end2end/fake_resolver.c
  44. 8
      test/core/end2end/tests/filter_call_init_fails.c
  45. 8
      test/core/end2end/tests/filter_causes_close.c
  46. 8
      test/core/end2end/tests/filter_latency.c
  47. 12
      test/core/surface/channel_create_test.c

@ -206,18 +206,12 @@ typedef struct {
/** If non-zero, allow the use of SO_REUSEPORT if it's available (default 1) */
#define GRPC_ARG_ALLOW_REUSEPORT "grpc.so_reuseport"
/** If non-zero, a pointer to a buffer pool (use grpc_resource_quota_arg_vtable
to fetch an appropriate pointer arg vtable */
to fetch an appropriate pointer arg vtable) */
#define GRPC_ARG_RESOURCE_QUOTA "grpc.resource_quota"
/** Service config data, to be passed to subchannels.
Not intended for external use. */
/** Service config data in JSON form. Not intended for use outside of tests. */
#define GRPC_ARG_SERVICE_CONFIG "grpc.service_config"
/** LB policy name. */
#define GRPC_ARG_LB_POLICY_NAME "grpc.lb_policy_name"
/** Server name. Not intended for external use. */
#define GRPC_ARG_SERVER_NAME "grpc.server_name"
/** Resolved addresses in a form used by the LB policy.
Not intended for external use. */
#define GRPC_ARG_LB_ADDRESSES "grpc.lb_addresses"
/** The grpc_socket_mutator instance that set the socket options. A pointer. */
#define GRPC_ARG_SOCKET_MUTATOR "grpc.socket_mutator"
/** \} */

@ -167,11 +167,12 @@ static void server_destroy_call_elem(grpc_exec_ctx *exec_ctx,
/* TODO(hongyu): record rpc server stats and census_tracing_end_op here */
}
static void init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
channel_data *chand = elem->channel_data;
GPR_ASSERT(chand != NULL);
return GRPC_ERROR_NONE;
}
static void destroy_channel_elem(grpc_exec_ctx *exec_ctx,

@ -44,6 +44,7 @@
#include <grpc/support/useful.h>
#include "src/core/ext/client_channel/lb_policy_registry.h"
#include "src/core/ext/client_channel/resolver_registry.h"
#include "src/core/ext/client_channel/subchannel.h"
#include "src/core/lib/channel/channel_args.h"
#include "src/core/lib/channel/connected_channel.h"
@ -499,24 +500,37 @@ static void cc_get_channel_info(grpc_exec_ctx *exec_ctx,
}
/* Constructor for channel_data */
static void cc_init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
static grpc_error *cc_init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
channel_data *chand = elem->channel_data;
memset(chand, 0, sizeof(*chand));
GPR_ASSERT(args->is_last);
GPR_ASSERT(elem->filter == &grpc_client_channel_filter);
// Initialize data members.
gpr_mu_init(&chand->mu);
chand->owning_stack = args->channel_stack;
grpc_closure_init(&chand->on_resolver_result_changed,
on_resolver_result_changed, chand);
chand->owning_stack = args->channel_stack;
chand->interested_parties = grpc_pollset_set_create();
grpc_connectivity_state_init(&chand->state_tracker, GRPC_CHANNEL_IDLE,
"client_channel");
chand->interested_parties = grpc_pollset_set_create();
// Record client channel factory.
const grpc_arg *arg = grpc_channel_args_find(args->channel_args,
GRPC_ARG_CLIENT_CHANNEL_FACTORY);
GPR_ASSERT(arg != NULL);
GPR_ASSERT(arg->type == GRPC_ARG_POINTER);
grpc_client_channel_factory_ref(arg->value.pointer.p);
chand->client_channel_factory = arg->value.pointer.p;
// Instantiate resolver.
arg = grpc_channel_args_find(args->channel_args, GRPC_ARG_SERVER_URI);
GPR_ASSERT(arg != NULL);
GPR_ASSERT(arg->type == GRPC_ARG_STRING);
chand->resolver = grpc_resolver_create(arg->value.string, args->channel_args);
if (chand->resolver == NULL) {
return GRPC_ERROR_CREATE("resolver creation failed");
}
return GRPC_ERROR_NONE;
}
/* Destructor for channel_data */
@ -1135,30 +1149,6 @@ const grpc_channel_filter grpc_client_channel_filter = {
"client-channel",
};
void grpc_client_channel_finish_initialization(
grpc_exec_ctx *exec_ctx, grpc_channel_stack *channel_stack,
grpc_resolver *resolver,
grpc_client_channel_factory *client_channel_factory) {
/* post construction initialization: set the transport setup pointer */
GPR_ASSERT(client_channel_factory != NULL);
grpc_channel_element *elem = grpc_channel_stack_last_element(channel_stack);
channel_data *chand = elem->channel_data;
gpr_mu_lock(&chand->mu);
GPR_ASSERT(!chand->resolver);
chand->resolver = resolver;
GRPC_RESOLVER_REF(resolver, "channel");
if (!grpc_closure_list_empty(chand->waiting_for_config_closures) ||
chand->exit_idle_when_lb_policy_arrives) {
chand->started_resolving = true;
GRPC_CHANNEL_STACK_REF(chand->owning_stack, "resolver");
grpc_resolver_next(exec_ctx, resolver, &chand->resolver_result,
&chand->on_resolver_result_changed);
}
chand->client_channel_factory = client_channel_factory;
grpc_client_channel_factory_ref(client_channel_factory);
gpr_mu_unlock(&chand->mu);
}
grpc_connectivity_state grpc_client_channel_check_connectivity_state(
grpc_exec_ctx *exec_ctx, grpc_channel_element *elem, int try_to_connect) {
channel_data *chand = elem->channel_data;

@ -38,6 +38,9 @@
#include "src/core/ext/client_channel/resolver.h"
#include "src/core/lib/channel/channel_stack.h"
// Channel arg key for server URI string.
#define GRPC_ARG_SERVER_URI "grpc.server_uri"
/* A client channel is a channel that begins disconnected, and can connect
to some endpoint on demand. If that endpoint disconnects, it will be
connected to again later.
@ -47,13 +50,6 @@
extern const grpc_channel_filter grpc_client_channel_filter;
/* Post-construction initializer to give the client channel its resolver
and factory. */
void grpc_client_channel_finish_initialization(
grpc_exec_ctx *exec_ctx, grpc_channel_stack *channel_stack,
grpc_resolver *resolver,
grpc_client_channel_factory *client_channel_factory);
grpc_connectivity_state grpc_client_channel_check_connectivity_state(
grpc_exec_ctx *exec_ctx, grpc_channel_element *elem, int try_to_connect);

@ -55,3 +55,35 @@ grpc_channel* grpc_client_channel_factory_create_channel(
return factory->vtable->create_client_channel(exec_ctx, factory, target, type,
args);
}
static void* factory_arg_copy(void* factory) {
grpc_client_channel_factory_ref(factory);
return factory;
}
static void factory_arg_destroy(void* factory) {
// TODO(roth): Remove local exec_ctx when
// https://github.com/grpc/grpc/pull/8705 is merged.
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
grpc_client_channel_factory_unref(&exec_ctx, factory);
grpc_exec_ctx_finish(&exec_ctx);
}
static int factory_arg_cmp(void* factory1, void* factory2) {
if (factory1 < factory2) return -1;
if (factory1 > factory2) return 1;
return 0;
}
static const grpc_arg_pointer_vtable factory_arg_vtable = {
factory_arg_copy, factory_arg_destroy, factory_arg_cmp};
grpc_arg grpc_client_channel_factory_create_channel_arg(
grpc_client_channel_factory* factory) {
grpc_arg arg;
arg.type = GRPC_ARG_POINTER;
arg.key = GRPC_ARG_CLIENT_CHANNEL_FACTORY;
arg.value.pointer.p = factory;
arg.value.pointer.vtable = &factory_arg_vtable;
return arg;
}

@ -39,6 +39,9 @@
#include "src/core/ext/client_channel/subchannel.h"
#include "src/core/lib/channel/channel_stack.h"
// Channel arg key for client channel factory.
#define GRPC_ARG_CLIENT_CHANNEL_FACTORY "grpc.client_channel_factory"
typedef struct grpc_client_channel_factory grpc_client_channel_factory;
typedef struct grpc_client_channel_factory_vtable
grpc_client_channel_factory_vtable;
@ -83,4 +86,7 @@ grpc_channel *grpc_client_channel_factory_create_channel(
const char *target, grpc_client_channel_type type,
const grpc_channel_args *args);
grpc_arg grpc_client_channel_factory_create_channel_arg(
grpc_client_channel_factory *factory);
#endif /* GRPC_CORE_EXT_CLIENT_CHANNEL_CLIENT_CHANNEL_FACTORY_H */

@ -40,6 +40,8 @@
#include <grpc/support/log.h>
#include <grpc/support/string_util.h>
#include "src/core/ext/client_channel/client_channel.h"
#include "src/core/ext/client_channel/resolver_registry.h"
#include "src/core/ext/client_channel/uri_parser.h"
#include "src/core/lib/channel/channel_args.h"
#include "src/core/lib/http/format_request.h"
@ -51,7 +53,6 @@ typedef struct http_connect_handshaker {
grpc_handshaker base;
char* proxy_server;
char* server_name;
gpr_refcount refcount;
gpr_mu mu;
@ -86,7 +87,6 @@ static void http_connect_handshaker_unref(grpc_exec_ctx* exec_ctx,
gpr_free(handshaker->read_buffer_to_destroy);
}
gpr_free(handshaker->proxy_server);
gpr_free(handshaker->server_name);
grpc_slice_buffer_destroy(&handshaker->write_buffer);
grpc_http_parser_destroy(&handshaker->http_parser);
grpc_http_response_destroy(&handshaker->http_response);
@ -265,18 +265,27 @@ static void http_connect_handshaker_do_handshake(
grpc_tcp_server_acceptor* acceptor, grpc_closure* on_handshake_done,
grpc_handshaker_args* args) {
http_connect_handshaker* handshaker = (http_connect_handshaker*)handshaker_in;
gpr_mu_lock(&handshaker->mu);
// Get server name from channel args.
const grpc_arg* arg = grpc_channel_args_find(args->args, GRPC_ARG_SERVER_URI);
GPR_ASSERT(arg != NULL);
GPR_ASSERT(arg->type == GRPC_ARG_STRING);
char* canonical_uri =
grpc_resolver_factory_add_default_prefix_if_needed(arg->value.string);
grpc_uri* uri = grpc_uri_parse(canonical_uri, 1);
char* server_name = uri->path;
if (server_name[0] == '/') ++server_name;
// Save state in the handshaker object.
gpr_mu_lock(&handshaker->mu);
handshaker->args = args;
handshaker->on_handshake_done = on_handshake_done;
// Send HTTP CONNECT request.
gpr_log(GPR_INFO, "Connecting to server %s via HTTP proxy %s",
handshaker->server_name, handshaker->proxy_server);
gpr_log(GPR_INFO, "Connecting to server %s via HTTP proxy %s", server_name,
handshaker->proxy_server);
grpc_httpcli_request request;
memset(&request, 0, sizeof(request));
request.host = handshaker->proxy_server;
request.host = server_name;
request.http.method = "CONNECT";
request.http.path = handshaker->server_name;
request.http.path = server_name;
request.handshaker = &grpc_httpcli_plaintext;
grpc_slice request_slice = grpc_httpcli_format_connect_request(&request);
grpc_slice_buffer_add(&handshaker->write_buffer, request_slice);
@ -285,23 +294,23 @@ static void http_connect_handshaker_do_handshake(
grpc_endpoint_write(exec_ctx, args->endpoint, &handshaker->write_buffer,
&handshaker->request_done_closure);
gpr_mu_unlock(&handshaker->mu);
// Clean up.
gpr_free(canonical_uri);
grpc_uri_destroy(uri);
}
static const grpc_handshaker_vtable http_connect_handshaker_vtable = {
http_connect_handshaker_destroy, http_connect_handshaker_shutdown,
http_connect_handshaker_do_handshake};
grpc_handshaker* grpc_http_connect_handshaker_create(const char* proxy_server,
const char* server_name) {
grpc_handshaker* grpc_http_connect_handshaker_create(const char* proxy_server) {
GPR_ASSERT(proxy_server != NULL);
GPR_ASSERT(server_name != NULL);
http_connect_handshaker* handshaker = gpr_malloc(sizeof(*handshaker));
memset(handshaker, 0, sizeof(*handshaker));
grpc_handshaker_init(&http_connect_handshaker_vtable, &handshaker->base);
gpr_mu_init(&handshaker->mu);
gpr_ref_init(&handshaker->refcount, 1);
handshaker->proxy_server = gpr_strdup(proxy_server);
handshaker->server_name = gpr_strdup(server_name);
grpc_slice_buffer_init(&handshaker->write_buffer);
grpc_closure_init(&handshaker->request_done_closure, on_write_done,
handshaker);

@ -36,9 +36,8 @@
#include "src/core/lib/channel/handshaker.h"
/// Does NOT take ownership of \a proxy_server or \a server_name.
grpc_handshaker* grpc_http_connect_handshaker_create(const char* proxy_server,
const char* server_name);
/// Does NOT take ownership of \a proxy_server.
grpc_handshaker* grpc_http_connect_handshaker_create(const char* proxy_server);
/// Returns the name of the proxy to use, or NULL if no proxy is configured.
/// Caller takes ownership of result.

@ -40,6 +40,9 @@
#include "src/core/lib/iomgr/exec_ctx.h"
#include "src/core/lib/iomgr/resolve_address.h"
// Channel arg key for grpc_lb_addresses.
#define GRPC_ARG_LB_ADDRESSES "grpc.lb_addresses"
typedef struct grpc_lb_policy_factory grpc_lb_policy_factory;
typedef struct grpc_lb_policy_factory_vtable grpc_lb_policy_factory_vtable;

@ -109,8 +109,8 @@ static grpc_resolver_factory *lookup_factory_by_uri(grpc_uri *uri) {
}
static grpc_resolver_factory *resolve_factory(const char *target,
grpc_uri **uri) {
char *tmp;
grpc_uri **uri,
char **canonical_target) {
grpc_resolver_factory *factory = NULL;
GPR_ASSERT(uri != NULL);
@ -118,15 +118,15 @@ static grpc_resolver_factory *resolve_factory(const char *target,
factory = lookup_factory_by_uri(*uri);
if (factory == NULL) {
grpc_uri_destroy(*uri);
gpr_asprintf(&tmp, "%s%s", g_default_resolver_prefix, target);
*uri = grpc_uri_parse(tmp, 1);
gpr_asprintf(canonical_target, "%s%s", g_default_resolver_prefix, target);
*uri = grpc_uri_parse(*canonical_target, 1);
factory = lookup_factory_by_uri(*uri);
if (factory == NULL) {
grpc_uri_destroy(grpc_uri_parse(target, 0));
grpc_uri_destroy(grpc_uri_parse(tmp, 0));
gpr_log(GPR_ERROR, "don't know how to resolve '%s' or '%s'", target, tmp);
grpc_uri_destroy(grpc_uri_parse(*canonical_target, 0));
gpr_log(GPR_ERROR, "don't know how to resolve '%s' or '%s'", target,
*canonical_target);
}
gpr_free(tmp);
}
return factory;
}
@ -134,7 +134,9 @@ static grpc_resolver_factory *resolve_factory(const char *target,
grpc_resolver *grpc_resolver_create(const char *target,
const grpc_channel_args *args) {
grpc_uri *uri = NULL;
grpc_resolver_factory *factory = resolve_factory(target, &uri);
char *canonical_target = NULL;
grpc_resolver_factory *factory =
resolve_factory(target, &uri, &canonical_target);
grpc_resolver *resolver;
grpc_resolver_args resolver_args;
memset(&resolver_args, 0, sizeof(resolver_args));
@ -142,13 +144,25 @@ grpc_resolver *grpc_resolver_create(const char *target,
resolver_args.args = args;
resolver = grpc_resolver_factory_create_resolver(factory, &resolver_args);
grpc_uri_destroy(uri);
gpr_free(canonical_target);
return resolver;
}
char *grpc_get_default_authority(const char *target) {
grpc_uri *uri = NULL;
grpc_resolver_factory *factory = resolve_factory(target, &uri);
char *canonical_target = NULL;
grpc_resolver_factory *factory =
resolve_factory(target, &uri, &canonical_target);
char *authority = grpc_resolver_factory_get_default_authority(factory, uri);
grpc_uri_destroy(uri);
gpr_free(canonical_target);
return authority;
}
char *grpc_resolver_factory_add_default_prefix_if_needed(const char *target) {
grpc_uri *uri = NULL;
char *canonical_target = NULL;
resolve_factory(target, &uri, &canonical_target);
grpc_uri_destroy(uri);
return canonical_target == NULL ? gpr_strdup(target) : canonical_target;
}

@ -71,4 +71,8 @@ grpc_resolver_factory *grpc_resolver_factory_lookup(const char *name);
representing the default authority to pass from a client. */
char *grpc_get_default_authority(const char *target);
/** Returns a newly allocated string containing \a target, adding the
default prefix if needed. */
char *grpc_resolver_factory_add_default_prefix_if_needed(const char *target);
#endif /* GRPC_CORE_EXT_CLIENT_CHANNEL_RESOLVER_REGISTRY_H */

@ -604,14 +604,20 @@ static void publish_transport_locked(grpc_exec_ctx *exec_ctx,
grpc_channel_stack_builder_set_transport(builder,
c->connecting_result.transport);
if (grpc_channel_init_create_stack(exec_ctx, builder,
GRPC_CLIENT_SUBCHANNEL)) {
con = grpc_channel_stack_builder_finish(exec_ctx, builder, 0, 1,
connection_destroy, NULL);
} else {
if (!grpc_channel_init_create_stack(exec_ctx, builder,
GRPC_CLIENT_SUBCHANNEL)) {
grpc_channel_stack_builder_destroy(builder);
abort(); /* TODO(ctiller): what to do here (previously we just crashed) */
}
grpc_error *error = grpc_channel_stack_builder_finish(
exec_ctx, builder, 0, 1, connection_destroy, NULL, (void **)&con);
if (error != GRPC_ERROR_NONE) {
const char *msg = grpc_error_string(error);
gpr_log(GPR_ERROR, "error initializing subchannel stack: %s", msg);
grpc_error_free_string(msg);
GRPC_ERROR_UNREF(error);
abort(); /* TODO(ctiller): what to do here? */
}
stk = CHANNEL_STACK_FROM_CONNECTION(con);
memset(&c->connecting_result, 0, sizeof(c->connecting_result));

@ -164,8 +164,6 @@ struct grpc_subchannel_args {
size_t filter_count;
/** Channel arguments to be supplied to the newly created channel */
const grpc_channel_args *args;
/** Server name */
const char *server_name;
/** Address to connect to */
grpc_resolved_address *addr;
};

@ -86,7 +86,6 @@ static grpc_subchannel_key *create_key(
} else {
k->args.filters = NULL;
}
k->args.server_name = gpr_strdup(args->server_name);
k->args.addr = gpr_malloc(sizeof(grpc_resolved_address));
k->args.addr->len = args->addr->len;
if (k->args.addr->len > 0) {
@ -113,8 +112,6 @@ static int subchannel_key_compare(grpc_subchannel_key *a,
if (c != 0) return c;
c = GPR_ICMP(a->args.filter_count, b->args.filter_count);
if (c != 0) return c;
c = strcmp(a->args.server_name, b->args.server_name);
if (c != 0) return c;
if (a->args.addr->len) {
c = memcmp(a->args.addr->addr, b->args.addr->addr, a->args.addr->len);
if (c != 0) return c;
@ -132,7 +129,6 @@ void grpc_subchannel_key_destroy(grpc_exec_ctx *exec_ctx,
grpc_connector_unref(exec_ctx, k->connector);
gpr_free((grpc_channel_args *)k->args.filters);
grpc_channel_args_destroy((grpc_channel_args *)k->args.args);
gpr_free((void *)k->args.server_name);
gpr_free(k->args.addr);
gpr_free(k);
}

@ -106,6 +106,7 @@
#include <grpc/support/string_util.h>
#include <grpc/support/time.h>
#include "src/core/ext/client_channel/client_channel.h"
#include "src/core/ext/client_channel/client_channel_factory.h"
#include "src/core/ext/client_channel/lb_policy_factory.h"
#include "src/core/ext/client_channel/lb_policy_registry.h"
@ -743,12 +744,6 @@ static void glb_rr_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
static grpc_lb_policy *glb_create(grpc_exec_ctx *exec_ctx,
grpc_lb_policy_factory *factory,
grpc_lb_policy_args *args) {
/* Get server name. */
const grpc_arg *arg =
grpc_channel_args_find(args->args, GRPC_ARG_SERVER_NAME);
const char *server_name =
arg != NULL && arg->type == GRPC_ARG_STRING ? arg->value.string : NULL;
/* Count the number of gRPC-LB addresses. There must be at least one.
* TODO(roth): For now, we ignore non-balancer addresses, but in the
* future, we may change the behavior such that we fall back to using
@ -756,7 +751,8 @@ static grpc_lb_policy *glb_create(grpc_exec_ctx *exec_ctx,
* time, this should be changed to allow a list with no balancer addresses,
* since the resolver might fail to return a balancer address even when
* this is the right LB policy to use. */
arg = grpc_channel_args_find(args->args, GRPC_ARG_LB_ADDRESSES);
const grpc_arg *arg =
grpc_channel_args_find(args->args, GRPC_ARG_LB_ADDRESSES);
GPR_ASSERT(arg != NULL && arg->type == GRPC_ARG_POINTER);
grpc_lb_addresses *addresses = arg->value.pointer.p;
size_t num_grpclb_addrs = 0;
@ -768,13 +764,19 @@ static grpc_lb_policy *glb_create(grpc_exec_ctx *exec_ctx,
glb_lb_policy *glb_policy = gpr_malloc(sizeof(*glb_policy));
memset(glb_policy, 0, sizeof(*glb_policy));
/* Get server name. */
arg = grpc_channel_args_find(args->args, GRPC_ARG_SERVER_URI);
GPR_ASSERT(arg != NULL);
GPR_ASSERT(arg->type == GRPC_ARG_STRING);
grpc_uri *uri = grpc_uri_parse(arg->value.string, 1);
glb_policy->server_name = gpr_strdup(uri->path);
grpc_uri_destroy(uri);
/* All input addresses in addresses come from a resolver that claims
* they are LB services. It's the resolver's responsibility to make sure
* this
* policy is only instantiated and used in that case.
* this policy is only instantiated and used in that case.
*
* Create a client channel over them to communicate with a LB service */
glb_policy->server_name = gpr_strdup(server_name);
glb_policy->cc_factory = args->client_channel_factory;
glb_policy->args = grpc_channel_args_copy(args->args);
GPR_ASSERT(glb_policy->cc_factory != NULL);
@ -818,9 +820,14 @@ static grpc_lb_policy *glb_create(grpc_exec_ctx *exec_ctx,
* We need the LB channel to return addresses with is_balancer=false
* so that it does not wind up recursively using the grpclb LB policy,
* as per the special case logic in client_channel.c.
*
* Finally, we also strip out the channel arg for the server URI,
* since that will be different for the LB channel than for the parent
* channel. (The client channel factory will re-add this arg with
* the right value.)
*/
static const char *keys_to_remove[] = {GRPC_ARG_LB_POLICY_NAME,
GRPC_ARG_LB_ADDRESSES};
static const char *keys_to_remove[] = {
GRPC_ARG_LB_POLICY_NAME, GRPC_ARG_LB_ADDRESSES, GRPC_ARG_SERVER_URI};
grpc_channel_args *new_args = grpc_channel_args_copy_and_remove(
args->args, keys_to_remove, GPR_ARRAY_SIZE(keys_to_remove));
glb_policy->lb_channel = grpc_client_channel_factory_create_channel(

@ -438,15 +438,10 @@ static grpc_lb_policy *create_pick_first(grpc_exec_ctx *exec_ctx,
grpc_lb_policy_args *args) {
GPR_ASSERT(args->client_channel_factory != NULL);
/* Get server name. */
const grpc_arg *arg =
grpc_channel_args_find(args->args, GRPC_ARG_SERVER_NAME);
const char *server_name =
arg != NULL && arg->type == GRPC_ARG_STRING ? arg->value.string : NULL;
/* Find the number of backend addresses. We ignore balancer
* addresses, since we don't know how to handle them. */
arg = grpc_channel_args_find(args->args, GRPC_ARG_LB_ADDRESSES);
const grpc_arg *arg =
grpc_channel_args_find(args->args, GRPC_ARG_LB_ADDRESSES);
GPR_ASSERT(arg != NULL && arg->type == GRPC_ARG_POINTER);
grpc_lb_addresses *addresses = arg->value.pointer.p;
size_t num_addrs = 0;
@ -472,9 +467,6 @@ static grpc_lb_policy *create_pick_first(grpc_exec_ctx *exec_ctx,
}
memset(&sc_args, 0, sizeof(grpc_subchannel_args));
/* server_name will be copied as part of the subchannel creation. This makes
* the copying of server_name (a borrowed pointer) OK. */
sc_args.server_name = server_name;
sc_args.addr = &addresses->addresses[i].address;
sc_args.args = args->args;

@ -703,15 +703,10 @@ static grpc_lb_policy *round_robin_create(grpc_exec_ctx *exec_ctx,
grpc_lb_policy_args *args) {
GPR_ASSERT(args->client_channel_factory != NULL);
/* Get server name. */
const grpc_arg *arg =
grpc_channel_args_find(args->args, GRPC_ARG_SERVER_NAME);
const char *server_name =
arg != NULL && arg->type == GRPC_ARG_STRING ? arg->value.string : NULL;
/* Find the number of backend addresses. We ignore balancer
* addresses, since we don't know how to handle them. */
arg = grpc_channel_args_find(args->args, GRPC_ARG_LB_ADDRESSES);
const grpc_arg *arg =
grpc_channel_args_find(args->args, GRPC_ARG_LB_ADDRESSES);
GPR_ASSERT(arg != NULL && arg->type == GRPC_ARG_POINTER);
grpc_lb_addresses *addresses = arg->value.pointer.p;
size_t num_addrs = 0;
@ -734,9 +729,6 @@ static grpc_lb_policy *round_robin_create(grpc_exec_ctx *exec_ctx,
if (addresses->addresses[i].is_balancer) continue;
memset(&sc_args, 0, sizeof(grpc_subchannel_args));
/* server_name will be copied as part of the subchannel creation. This makes
* the copying of server_name (a borrowed pointer) OK. */
sc_args.server_name = server_name;
sc_args.addr = &addresses->addresses[i].address;
sc_args.args = args->args;

@ -152,9 +152,9 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
}
/* Constructor for channel_data */
static void init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
GPR_ASSERT(!args->is_last);
channel_data *chand = elem->channel_data;
@ -171,6 +171,8 @@ static void init_channel_elem(grpc_exec_ctx *exec_ctx,
NULL,
NULL};
*/
return GRPC_ERROR_NONE;
}
/* Destructor for channel data */

@ -264,12 +264,7 @@ static grpc_resolver *dns_create(grpc_resolver_args *args,
grpc_resolver_init(&r->base, &dns_resolver_vtable);
r->name_to_resolve = proxy_name == NULL ? gpr_strdup(path) : proxy_name;
r->default_port = gpr_strdup(default_port);
grpc_arg server_name_arg;
server_name_arg.type = GRPC_ARG_STRING;
server_name_arg.key = GRPC_ARG_SERVER_NAME;
server_name_arg.value.string = (char *)path;
r->channel_args =
grpc_channel_args_copy_and_add(args->args, &server_name_arg, 1);
r->channel_args = grpc_channel_args_copy(args->args);
gpr_backoff_init(&r->backoff_state, GRPC_DNS_INITIAL_CONNECT_BACKOFF_SECONDS,
GRPC_DNS_RECONNECT_BACKOFF_MULTIPLIER,
GRPC_DNS_RECONNECT_JITTER,

@ -198,12 +198,7 @@ static grpc_resolver *sockaddr_create(grpc_resolver_args *args,
sockaddr_resolver *r = gpr_malloc(sizeof(sockaddr_resolver));
memset(r, 0, sizeof(*r));
r->addresses = addresses;
grpc_arg server_name_arg;
server_name_arg.type = GRPC_ARG_STRING;
server_name_arg.key = GRPC_ARG_SERVER_NAME;
server_name_arg.value.string = args->uri->path;
r->channel_args =
grpc_channel_args_copy_and_add(args->args, &server_name_arg, 1);
r->channel_args = grpc_channel_args_copy(args->args);
gpr_mu_init(&r->mu);
grpc_resolver_init(&r->base, &sockaddr_resolver_vtable);
return &r->base;

@ -58,7 +58,6 @@ typedef struct {
bool shutdown;
bool connecting;
char *server_name;
grpc_chttp2_add_handshakers_func add_handshakers;
void *add_handshakers_user_data;
@ -89,7 +88,6 @@ static void chttp2_connector_unref(grpc_exec_ctx *exec_ctx,
// If handshaking is not yet in progress, destroy the endpoint.
// Otherwise, the handshaker will do this for us.
if (c->endpoint != NULL) grpc_endpoint_destroy(exec_ctx, c->endpoint);
gpr_free(c->server_name);
gpr_free(c);
}
}
@ -155,9 +153,8 @@ static void start_handshake_locked(grpc_exec_ctx *exec_ctx,
c->handshake_mgr = grpc_handshake_manager_create();
char *proxy_name = grpc_get_http_proxy_server();
if (proxy_name != NULL) {
grpc_handshake_manager_add(
c->handshake_mgr,
grpc_http_connect_handshaker_create(proxy_name, c->server_name));
grpc_handshake_manager_add(c->handshake_mgr,
grpc_http_connect_handshaker_create(proxy_name));
gpr_free(proxy_name);
}
if (c->add_handshakers != NULL) {
@ -254,15 +251,13 @@ static const grpc_connector_vtable chttp2_connector_vtable = {
chttp2_connector_connect};
grpc_connector *grpc_chttp2_connector_create(
grpc_exec_ctx *exec_ctx, const char *server_name,
grpc_chttp2_add_handshakers_func add_handshakers,
grpc_exec_ctx *exec_ctx, grpc_chttp2_add_handshakers_func add_handshakers,
void *add_handshakers_user_data) {
chttp2_connector *c = gpr_malloc(sizeof(*c));
memset(c, 0, sizeof(*c));
c->base.vtable = &chttp2_connector_vtable;
gpr_mu_init(&c->mu);
gpr_ref_init(&c->refs, 1);
c->server_name = gpr_strdup(server_name);
c->add_handshakers = add_handshakers;
c->add_handshakers_user_data = add_handshakers_user_data;
return &c->base;

@ -45,8 +45,7 @@ typedef void (*grpc_chttp2_add_handshakers_func)(
/// If \a add_handshakers is non-NULL, it will be called with
/// \a add_handshakers_user_data to add handshakers.
grpc_connector* grpc_chttp2_connector_create(
grpc_exec_ctx* exec_ctx, const char* server_name,
grpc_chttp2_add_handshakers_func add_handshakers,
grpc_exec_ctx* exec_ctx, grpc_chttp2_add_handshakers_func add_handshakers,
void* add_handshakers_user_data);
#endif /* GRPC_CORE_EXT_TRANSPORT_CHTTP2_CLIENT_CHTTP2_CONNECTOR_H */

@ -39,8 +39,8 @@
#include <grpc/support/string_util.h>
#include "src/core/ext/client_channel/client_channel.h"
#include "src/core/ext/client_channel/resolver_registry.h"
#include "src/core/ext/transport/chttp2/client/chttp2_connector.h"
#include "src/core/lib/channel/channel_args.h"
#include "src/core/lib/surface/api_trace.h"
#include "src/core/lib/surface/channel.h"
@ -54,8 +54,7 @@ static grpc_subchannel *client_channel_factory_create_subchannel(
grpc_exec_ctx *exec_ctx, grpc_client_channel_factory *cc_factory,
const grpc_subchannel_args *args) {
grpc_connector *connector = grpc_chttp2_connector_create(
exec_ctx, args->server_name, NULL /* add_handshakers */,
NULL /* user_data */);
exec_ctx, NULL /* add_handshakers */, NULL /* user_data */);
grpc_subchannel *s = grpc_subchannel_create(exec_ctx, connector, args);
grpc_connector_unref(exec_ctx, connector);
return s;
@ -65,17 +64,15 @@ 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,
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 == NULL) {
GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, channel,
"client_channel_factory_create_channel");
return NULL;
}
grpc_client_channel_finish_initialization(
exec_ctx, grpc_channel_get_channel_stack(channel), resolver, cc_factory);
GRPC_RESOLVER_UNREF(exec_ctx, resolver, "create_channel");
// Add channel arg containing the server URI.
grpc_arg arg;
arg.type = GRPC_ARG_STRING;
arg.key = GRPC_ARG_SERVER_URI;
arg.value.string = (char *)target;
grpc_channel_args *new_args = grpc_channel_args_copy_and_add(args, &arg, 1);
grpc_channel *channel = grpc_channel_create(exec_ctx, target, new_args,
GRPC_CLIENT_CHANNEL, NULL);
grpc_channel_args_destroy(new_args);
return channel;
}
@ -101,8 +98,14 @@ grpc_channel *grpc_insecure_channel_create(const char *target,
GPR_ASSERT(reserved == NULL);
grpc_client_channel_factory *factory =
(grpc_client_channel_factory *)&client_channel_factory;
// Add channel arg containing the client channel factory.
grpc_arg arg = grpc_client_channel_factory_create_channel_arg(factory);
grpc_channel_args *new_args = grpc_channel_args_copy_and_add(args, &arg, 1);
// Create channel.
grpc_channel *channel = client_channel_factory_create_channel(
&exec_ctx, factory, target, GRPC_CLIENT_CHANNEL_TYPE_REGULAR, args);
&exec_ctx, factory, target, GRPC_CLIENT_CHANNEL_TYPE_REGULAR, new_args);
// Clean up.
grpc_channel_args_destroy(new_args);
grpc_client_channel_factory_unref(&exec_ctx, factory);
grpc_exec_ctx_finish(&exec_ctx);
return channel != NULL ? channel : grpc_lame_client_channel_create(

@ -39,7 +39,6 @@
#include <grpc/support/string_util.h>
#include "src/core/ext/client_channel/client_channel.h"
#include "src/core/ext/client_channel/resolver_registry.h"
#include "src/core/ext/transport/chttp2/client/chttp2_connector.h"
#include "src/core/lib/channel/channel_args.h"
#include "src/core/lib/security/credentials/credentials.h"
@ -80,7 +79,7 @@ static grpc_subchannel *client_channel_factory_create_subchannel(
const grpc_subchannel_args *args) {
client_channel_factory *f = (client_channel_factory *)cc_factory;
grpc_connector *connector = grpc_chttp2_connector_create(
exec_ctx, args->server_name, add_handshakers, f->security_connector);
exec_ctx, add_handshakers, f->security_connector);
grpc_subchannel *s = grpc_subchannel_create(exec_ctx, connector, args);
grpc_connector_unref(exec_ctx, connector);
return s;
@ -90,18 +89,15 @@ 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,
const grpc_channel_args *args) {
client_channel_factory *f = (client_channel_factory *)cc_factory;
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_CHANNEL_INTERNAL_UNREF(exec_ctx, channel,
"client_channel_factory_create_channel");
return NULL;
}
grpc_client_channel_finish_initialization(
exec_ctx, grpc_channel_get_channel_stack(channel), resolver, &f->base);
GRPC_RESOLVER_UNREF(exec_ctx, resolver, "create_channel");
// Add channel arg containing the server URI.
grpc_arg arg;
arg.type = GRPC_ARG_STRING;
arg.key = GRPC_ARG_SERVER_URI;
arg.value.string = (char *)target;
grpc_channel_args *new_args = grpc_channel_args_copy_and_add(args, &arg, 1);
grpc_channel *channel = grpc_channel_create(exec_ctx, target, new_args,
GRPC_CLIENT_CHANNEL, NULL);
grpc_channel_args_destroy(new_args);
return channel;
}
@ -142,14 +138,6 @@ 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.");
}
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);
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));
@ -158,13 +146,24 @@ grpc_channel *grpc_secure_channel_create(grpc_channel_credentials *creds,
GRPC_SECURITY_CONNECTOR_REF(&security_connector->base,
"grpc_secure_channel_create");
f->security_connector = security_connector;
// Add channel args containing the client channel factory and security
// connector.
grpc_arg new_args[2];
new_args[0] = grpc_client_channel_factory_create_channel_arg(&f->base);
new_args[1] = grpc_security_connector_to_arg(&security_connector->base);
grpc_channel_args *args_copy = grpc_channel_args_copy_and_add(
new_args_from_connector != NULL ? new_args_from_connector : args,
new_args, GPR_ARRAY_SIZE(new_args));
if (new_args_from_connector != NULL) {
grpc_channel_args_destroy(new_args_from_connector);
}
// Create channel.
grpc_channel *channel = client_channel_factory_create_channel(
&exec_ctx, &f->base, target, GRPC_CLIENT_CHANNEL_TYPE_REGULAR, new_args);
&exec_ctx, &f->base, target, GRPC_CLIENT_CHANNEL_TYPE_REGULAR, args_copy);
// Clean up.
GRPC_SECURITY_CONNECTOR_UNREF(&f->security_connector->base,
"secure_client_channel_factory_create_channel");
grpc_channel_args_destroy(new_args);
grpc_channel_args_destroy(args_copy);
grpc_client_channel_factory_unref(&exec_ctx, &f->base);
grpc_exec_ctx_finish(&exec_ctx);
return channel; /* may be NULL */

@ -102,13 +102,11 @@ grpc_call_element *grpc_call_stack_element(grpc_call_stack *call_stack,
return CALL_ELEMS_FROM_STACK(call_stack) + index;
}
void grpc_channel_stack_init(grpc_exec_ctx *exec_ctx, int initial_refs,
grpc_iomgr_cb_func destroy, void *destroy_arg,
const grpc_channel_filter **filters,
size_t filter_count,
const grpc_channel_args *channel_args,
grpc_transport *optional_transport,
const char *name, grpc_channel_stack *stack) {
grpc_error *grpc_channel_stack_init(
grpc_exec_ctx *exec_ctx, int initial_refs, grpc_iomgr_cb_func destroy,
void *destroy_arg, const grpc_channel_filter **filters, size_t filter_count,
const grpc_channel_args *channel_args, grpc_transport *optional_transport,
const char *name, grpc_channel_stack *stack) {
size_t call_size =
ROUND_UP_TO_ALIGNMENT_SIZE(sizeof(grpc_call_stack)) +
ROUND_UP_TO_ALIGNMENT_SIZE(filter_count * sizeof(grpc_call_element));
@ -126,6 +124,7 @@ void grpc_channel_stack_init(grpc_exec_ctx *exec_ctx, int initial_refs,
ROUND_UP_TO_ALIGNMENT_SIZE(filter_count * sizeof(grpc_channel_element));
/* init per-filter data */
grpc_error *first_error = GRPC_ERROR_NONE;
for (i = 0; i < filter_count; i++) {
args.channel_stack = stack;
args.channel_args = channel_args;
@ -134,7 +133,15 @@ void grpc_channel_stack_init(grpc_exec_ctx *exec_ctx, int initial_refs,
args.is_last = i == (filter_count - 1);
elems[i].filter = filters[i];
elems[i].channel_data = user_data;
elems[i].filter->init_channel_elem(exec_ctx, &elems[i], &args);
grpc_error *error =
elems[i].filter->init_channel_elem(exec_ctx, &elems[i], &args);
if (error != GRPC_ERROR_NONE) {
if (first_error == GRPC_ERROR_NONE) {
first_error = error;
} else {
GRPC_ERROR_UNREF(error);
}
}
user_data += ROUND_UP_TO_ALIGNMENT_SIZE(filters[i]->sizeof_channel_data);
call_size += ROUND_UP_TO_ALIGNMENT_SIZE(filters[i]->sizeof_call_data);
}
@ -144,6 +151,7 @@ void grpc_channel_stack_init(grpc_exec_ctx *exec_ctx, int initial_refs,
grpc_channel_stack_size(filters, filter_count));
stack->call_stack_size = call_size;
return first_error;
}
void grpc_channel_stack_destroy(grpc_exec_ctx *exec_ctx,

@ -146,8 +146,9 @@ typedef struct {
is_first, is_last designate this elements position in the stack, and are
useful for asserting correct configuration by upper layer code.
The filter does not need to do any chaining */
void (*init_channel_elem)(grpc_exec_ctx *exec_ctx, grpc_channel_element *elem,
grpc_channel_element_args *args);
grpc_error *(*init_channel_elem)(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args);
/* Destroy per channel data.
The filter does not need to do any chaining */
void (*destroy_channel_elem)(grpc_exec_ctx *exec_ctx,
@ -214,12 +215,11 @@ grpc_call_element *grpc_call_stack_element(grpc_call_stack *stack, size_t i);
size_t grpc_channel_stack_size(const grpc_channel_filter **filters,
size_t filter_count);
/* Initialize a channel stack given some filters */
void grpc_channel_stack_init(grpc_exec_ctx *exec_ctx, int initial_refs,
grpc_iomgr_cb_func destroy, void *destroy_arg,
const grpc_channel_filter **filters,
size_t filter_count, const grpc_channel_args *args,
grpc_transport *optional_transport,
const char *name, grpc_channel_stack *stack);
grpc_error *grpc_channel_stack_init(
grpc_exec_ctx *exec_ctx, int initial_refs, grpc_iomgr_cb_func destroy,
void *destroy_arg, const grpc_channel_filter **filters, size_t filter_count,
const grpc_channel_args *args, grpc_transport *optional_transport,
const char *name, grpc_channel_stack *stack);
/* Destroy a channel stack */
void grpc_channel_stack_destroy(grpc_exec_ctx *exec_ctx,
grpc_channel_stack *stack);

@ -227,11 +227,10 @@ void grpc_channel_stack_builder_destroy(grpc_channel_stack_builder *builder) {
gpr_free(builder);
}
void *grpc_channel_stack_builder_finish(grpc_exec_ctx *exec_ctx,
grpc_channel_stack_builder *builder,
size_t prefix_bytes, int initial_refs,
grpc_iomgr_cb_func destroy,
void *destroy_arg) {
grpc_error *grpc_channel_stack_builder_finish(
grpc_exec_ctx *exec_ctx, grpc_channel_stack_builder *builder,
size_t prefix_bytes, int initial_refs, grpc_iomgr_cb_func destroy,
void *destroy_arg, void **result) {
// count the number of filters
size_t num_filters = 0;
for (filter_node *p = builder->begin.next; p != &builder->end; p = p->next) {
@ -250,28 +249,35 @@ void *grpc_channel_stack_builder_finish(grpc_exec_ctx *exec_ctx,
size_t channel_stack_size = grpc_channel_stack_size(filters, num_filters);
// allocate memory, with prefix_bytes followed by channel_stack_size
char *result = gpr_malloc(prefix_bytes + channel_stack_size);
*result = gpr_malloc(prefix_bytes + channel_stack_size);
// fetch a pointer to the channel stack
grpc_channel_stack *channel_stack =
(grpc_channel_stack *)(result + prefix_bytes);
(grpc_channel_stack *)((char *)(*result) + prefix_bytes);
// and initialize it
grpc_channel_stack_init(exec_ctx, initial_refs, destroy,
destroy_arg == NULL ? result : destroy_arg, filters,
num_filters, builder->args, builder->transport,
builder->name, channel_stack);
// run post-initialization functions
i = 0;
for (filter_node *p = builder->begin.next; p != &builder->end; p = p->next) {
if (p->init != NULL) {
p->init(channel_stack, grpc_channel_stack_element(channel_stack, i),
p->init_arg);
grpc_error *error = grpc_channel_stack_init(
exec_ctx, initial_refs, destroy,
destroy_arg == NULL ? *result : destroy_arg, filters, num_filters,
builder->args, builder->transport, builder->name, channel_stack);
if (error != GRPC_ERROR_NONE) {
grpc_channel_stack_destroy(exec_ctx, channel_stack);
gpr_free(*result);
*result = NULL;
} else {
// run post-initialization functions
i = 0;
for (filter_node *p = builder->begin.next; p != &builder->end;
p = p->next) {
if (p->init != NULL) {
p->init(channel_stack, grpc_channel_stack_element(channel_stack, i),
p->init_arg);
}
i++;
}
i++;
}
grpc_channel_stack_builder_destroy(builder);
gpr_free((grpc_channel_filter **)filters);
return result;
return error;
}

@ -146,16 +146,15 @@ bool grpc_channel_stack_builder_append_filter(
void grpc_channel_stack_builder_iterator_destroy(
grpc_channel_stack_builder_iterator *iterator);
/// Destroy the builder, return the freshly minted channel stack
/// Destroy the builder, return the freshly minted channel stack in \a result.
/// Allocates \a prefix_bytes bytes before the channel stack
/// Returns the base pointer of the allocated block
/// \a initial_refs, \a destroy, \a destroy_arg are as per
/// grpc_channel_stack_init
void *grpc_channel_stack_builder_finish(grpc_exec_ctx *exec_ctx,
grpc_channel_stack_builder *builder,
size_t prefix_bytes, int initial_refs,
grpc_iomgr_cb_func destroy,
void *destroy_arg);
grpc_error *grpc_channel_stack_builder_finish(
grpc_exec_ctx *exec_ctx, grpc_channel_stack_builder *builder,
size_t prefix_bytes, int initial_refs, grpc_iomgr_cb_func destroy,
void *destroy_arg, void **result);
/// Destroy the builder without creating a channel stack
void grpc_channel_stack_builder_destroy(grpc_channel_stack_builder *builder);

@ -285,9 +285,9 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
}
/* Constructor for channel_data */
static void init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
channel_data *channeld = elem->channel_data;
channeld->enabled_algorithms_bitset =
@ -315,6 +315,7 @@ static void init_channel_elem(grpc_exec_ctx *exec_ctx,
}
GPR_ASSERT(!args->is_last);
return GRPC_ERROR_NONE;
}
/* Destructor for channel data */

@ -114,12 +114,13 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
}
/* Constructor for channel_data */
static void init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
channel_data *cd = (channel_data *)elem->channel_data;
GPR_ASSERT(args->is_last);
cd->transport = NULL;
return GRPC_ERROR_NONE;
}
/* Destructor for channel_data */

@ -207,10 +207,11 @@ void grpc_deadline_state_client_start_transport_stream_op(
//
// Constructor for channel_data. Used for both client and server filters.
static void init_channel_elem(grpc_exec_ctx* exec_ctx,
grpc_channel_element* elem,
grpc_channel_element_args* args) {
static grpc_error* init_channel_elem(grpc_exec_ctx* exec_ctx,
grpc_channel_element* elem,
grpc_channel_element_args* args) {
GPR_ASSERT(!args->is_last);
return GRPC_ERROR_NONE;
}
// Destructor for channel_data. Used for both client and server filters.

@ -457,9 +457,9 @@ static grpc_mdstr *user_agent_from_args(const grpc_channel_args *args,
}
/* Constructor for channel_data */
static void init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
channel_data *chand = elem->channel_data;
GPR_ASSERT(!args->is_last);
GPR_ASSERT(args->optional_transport != NULL);
@ -470,6 +470,7 @@ static void init_channel_elem(grpc_exec_ctx *exec_ctx,
GRPC_MDSTR_USER_AGENT,
user_agent_from_args(args->channel_args,
args->optional_transport->vtable->name));
return GRPC_ERROR_NONE;
}
/* Destructor for channel data */

@ -350,10 +350,11 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
}
/* Constructor for channel_data */
static void init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
GPR_ASSERT(!args->is_last);
return GRPC_ERROR_NONE;
}
/* Destructor for channel data */

@ -192,9 +192,9 @@ static void destroy_call_elem(grpc_exec_ctx* exec_ctx, grpc_call_element* elem,
void* ignored) {}
// Constructor for channel_data.
static void init_channel_elem(grpc_exec_ctx* exec_ctx,
grpc_channel_element* elem,
grpc_channel_element_args* args) {
static grpc_error* init_channel_elem(grpc_exec_ctx* exec_ctx,
grpc_channel_element* elem,
grpc_channel_element_args* args) {
GPR_ASSERT(!args->is_last);
channel_data* chand = elem->channel_data;
memset(chand, 0, sizeof(*chand));
@ -231,6 +231,7 @@ static void init_channel_elem(grpc_exec_ctx* exec_ctx,
grpc_service_config_destroy(service_config);
}
}
return GRPC_ERROR_NONE;
}
// Destructor for channel_data.

@ -303,9 +303,9 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
}
/* Constructor for channel_data */
static void init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
grpc_security_connector *sc =
grpc_find_security_connector_in_args(args->channel_args);
grpc_auth_context *auth_context =
@ -327,6 +327,7 @@ static void init_channel_elem(grpc_exec_ctx *exec_ctx,
sc, "client_auth_filter");
chand->auth_context =
GRPC_AUTH_CONTEXT_REF(auth_context, "client_auth_filter");
return GRPC_ERROR_NONE;
}
/* Destructor for channel data */

@ -238,9 +238,9 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
void *ignored) {}
/* Constructor for channel_data */
static void init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
grpc_auth_context *auth_context =
grpc_find_auth_context_in_args(args->channel_args);
grpc_server_credentials *creds =
@ -256,6 +256,7 @@ static void init_channel_elem(grpc_exec_ctx *exec_ctx,
chand->auth_context =
GRPC_AUTH_CONTEXT_REF(auth_context, "server_auth_filter");
chand->creds = grpc_server_credentials_ref(creds);
return GRPC_ERROR_NONE;
}
/* Destructor for channel data */

@ -86,87 +86,90 @@ grpc_channel *grpc_channel_create(grpc_exec_ctx *exec_ctx, const char *target,
const grpc_channel_args *input_args,
grpc_channel_stack_type channel_stack_type,
grpc_transport *optional_transport) {
bool is_client = grpc_channel_stack_type_is_client(channel_stack_type);
grpc_channel_stack_builder *builder = grpc_channel_stack_builder_create();
grpc_channel_stack_builder_set_channel_arguments(builder, input_args);
grpc_channel_stack_builder_set_target(builder, target);
grpc_channel_stack_builder_set_transport(builder, optional_transport);
grpc_channel *channel;
grpc_channel_args *args;
if (!grpc_channel_init_create_stack(exec_ctx, builder, channel_stack_type)) {
grpc_channel_stack_builder_destroy(builder);
return NULL;
} else {
args = grpc_channel_args_copy(
grpc_channel_stack_builder_get_channel_arguments(builder));
channel = grpc_channel_stack_builder_finish(
exec_ctx, builder, sizeof(grpc_channel), 1, destroy_channel, NULL);
}
grpc_channel_args *args = grpc_channel_args_copy(
grpc_channel_stack_builder_get_channel_arguments(builder));
grpc_channel *channel;
grpc_error *error = grpc_channel_stack_builder_finish(
exec_ctx, builder, sizeof(grpc_channel), 1, destroy_channel, NULL,
(void **)&channel);
if (error != GRPC_ERROR_NONE) {
const char *msg = grpc_error_string(error);
gpr_log(GPR_ERROR, "channel stack builder failed: %s", msg);
grpc_error_free_string(msg);
GRPC_ERROR_UNREF(error);
goto done;
}
memset(channel, 0, sizeof(*channel));
channel->target = gpr_strdup(target);
channel->is_client = is_client;
channel->is_client = grpc_channel_stack_type_is_client(channel_stack_type);
gpr_mu_init(&channel->registered_call_mu);
channel->registered_calls = NULL;
grpc_compression_options_init(&channel->compression_options);
if (args) {
for (size_t i = 0; i < args->num_args; i++) {
if (0 == strcmp(args->args[i].key, GRPC_ARG_DEFAULT_AUTHORITY)) {
if (args->args[i].type != GRPC_ARG_STRING) {
gpr_log(GPR_ERROR, "%s ignored: it must be a string",
GRPC_ARG_DEFAULT_AUTHORITY);
} else {
if (channel->default_authority) {
/* setting this takes precedence over anything else */
GRPC_MDELEM_UNREF(channel->default_authority);
}
channel->default_authority = grpc_mdelem_from_strings(
":authority", args->args[i].value.string);
for (size_t i = 0; i < args->num_args; i++) {
if (0 == strcmp(args->args[i].key, GRPC_ARG_DEFAULT_AUTHORITY)) {
if (args->args[i].type != GRPC_ARG_STRING) {
gpr_log(GPR_ERROR, "%s ignored: it must be a string",
GRPC_ARG_DEFAULT_AUTHORITY);
} else {
if (channel->default_authority) {
/* setting this takes precedence over anything else */
GRPC_MDELEM_UNREF(channel->default_authority);
}
} else if (0 ==
strcmp(args->args[i].key, GRPC_SSL_TARGET_NAME_OVERRIDE_ARG)) {
if (args->args[i].type != GRPC_ARG_STRING) {
gpr_log(GPR_ERROR, "%s ignored: it must be a string",
channel->default_authority =
grpc_mdelem_from_strings(":authority", args->args[i].value.string);
}
} else if (0 ==
strcmp(args->args[i].key, GRPC_SSL_TARGET_NAME_OVERRIDE_ARG)) {
if (args->args[i].type != GRPC_ARG_STRING) {
gpr_log(GPR_ERROR, "%s ignored: it must be a string",
GRPC_SSL_TARGET_NAME_OVERRIDE_ARG);
} else {
if (channel->default_authority) {
/* other ways of setting this (notably ssl) take precedence */
gpr_log(GPR_ERROR,
"%s ignored: default host already set some other way",
GRPC_SSL_TARGET_NAME_OVERRIDE_ARG);
} else {
if (channel->default_authority) {
/* other ways of setting this (notably ssl) take precedence */
gpr_log(GPR_ERROR,
"%s ignored: default host already set some other way",
GRPC_SSL_TARGET_NAME_OVERRIDE_ARG);
} else {
channel->default_authority = grpc_mdelem_from_strings(
":authority", args->args[i].value.string);
}
channel->default_authority = grpc_mdelem_from_strings(
":authority", args->args[i].value.string);
}
} else if (0 == strcmp(args->args[i].key,
GRPC_COMPRESSION_CHANNEL_DEFAULT_LEVEL)) {
channel->compression_options.default_level.is_set = true;
GPR_ASSERT(args->args[i].value.integer >= 0 &&
args->args[i].value.integer < GRPC_COMPRESS_LEVEL_COUNT);
channel->compression_options.default_level.level =
(grpc_compression_level)args->args[i].value.integer;
} else if (0 == strcmp(args->args[i].key,
GRPC_COMPRESSION_CHANNEL_DEFAULT_ALGORITHM)) {
channel->compression_options.default_algorithm.is_set = true;
GPR_ASSERT(args->args[i].value.integer >= 0 &&
args->args[i].value.integer <
GRPC_COMPRESS_ALGORITHMS_COUNT);
channel->compression_options.default_algorithm.algorithm =
(grpc_compression_algorithm)args->args[i].value.integer;
} else if (0 ==
strcmp(args->args[i].key,
GRPC_COMPRESSION_CHANNEL_ENABLED_ALGORITHMS_BITSET)) {
channel->compression_options.enabled_algorithms_bitset =
(uint32_t)args->args[i].value.integer |
0x1; /* always support no compression */
}
} else if (0 == strcmp(args->args[i].key,
GRPC_COMPRESSION_CHANNEL_DEFAULT_LEVEL)) {
channel->compression_options.default_level.is_set = true;
GPR_ASSERT(args->args[i].value.integer >= 0 &&
args->args[i].value.integer < GRPC_COMPRESS_LEVEL_COUNT);
channel->compression_options.default_level.level =
(grpc_compression_level)args->args[i].value.integer;
} else if (0 == strcmp(args->args[i].key,
GRPC_COMPRESSION_CHANNEL_DEFAULT_ALGORITHM)) {
channel->compression_options.default_algorithm.is_set = true;
GPR_ASSERT(args->args[i].value.integer >= 0 &&
args->args[i].value.integer < GRPC_COMPRESS_ALGORITHMS_COUNT);
channel->compression_options.default_algorithm.algorithm =
(grpc_compression_algorithm)args->args[i].value.integer;
} else if (0 ==
strcmp(args->args[i].key,
GRPC_COMPRESSION_CHANNEL_ENABLED_ALGORITHMS_BITSET)) {
channel->compression_options.enabled_algorithms_bitset =
(uint32_t)args->args[i].value.integer |
0x1; /* always support no compression */
}
grpc_channel_args_destroy(args);
}
done:
grpc_channel_args_destroy(args);
return channel;
}

@ -123,11 +123,12 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
gpr_free(and_free_memory);
}
static void init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
GPR_ASSERT(args->is_first);
GPR_ASSERT(args->is_last);
return GRPC_ERROR_NONE;
}
static void destroy_channel_elem(grpc_exec_ctx *exec_ctx,

@ -914,9 +914,9 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
server_unref(exec_ctx, chand->server);
}
static void init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
channel_data *chand = elem->channel_data;
GPR_ASSERT(args->is_first);
GPR_ASSERT(!args->is_last);
@ -927,6 +927,7 @@ static void init_channel_elem(grpc_exec_ctx *exec_ctx,
chand->connectivity_state = GRPC_CHANNEL_IDLE;
grpc_closure_init(&chand->channel_connectivity_changed,
channel_connectivity_changed, chand);
return GRPC_ERROR_NONE;
}
static void destroy_channel_elem(grpc_exec_ctx *exec_ctx,

@ -220,6 +220,9 @@ class ChannelData {
if (peer_) gpr_free((void *)peer_);
}
/// Initializes the call data.
virtual grpc_error *Init() { return GRPC_ERROR_NONE; }
/// Caller does NOT take ownership of result.
const char *peer() const { return peer_; }
@ -276,15 +279,17 @@ class ChannelFilter final {
public:
static const size_t channel_data_size = sizeof(ChannelDataType);
static void InitChannelElement(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
static grpc_error *InitChannelElement(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
const char *peer =
args->optional_transport
? grpc_transport_get_peer(exec_ctx, args->optional_transport)
: nullptr;
// Construct the object in the already-allocated memory.
new (elem->channel_data) ChannelDataType(*args->channel_args, peer);
ChannelDataType *channel_data =
new (elem->channel_data) ChannelDataType(*args->channel_args, peer);
return channel_data->Init();
}
static void DestroyChannelElement(grpc_exec_ctx *exec_ctx,

@ -41,9 +41,9 @@
#include "test/core/util/test_config.h"
static void channel_init_func(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
static grpc_error *channel_init_func(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
GPR_ASSERT(args->channel_args->num_args == 1);
GPR_ASSERT(args->channel_args->args[0].type == GRPC_ARG_INTEGER);
GPR_ASSERT(0 == strcmp(args->channel_args->args[0].key, "test_key"));
@ -51,6 +51,7 @@ static void channel_init_func(grpc_exec_ctx *exec_ctx,
GPR_ASSERT(args->is_first);
GPR_ASSERT(args->is_last);
*(int *)(elem->channel_data) = 0;
return GRPC_ERROR_NONE;
}
static grpc_error *call_init_func(grpc_exec_ctx *exec_ctx,

@ -49,11 +49,6 @@ typedef struct on_resolution_arg {
void on_resolution_cb(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
on_resolution_arg *res = arg;
const grpc_arg *channel_arg =
grpc_channel_args_find(res->resolver_result, GRPC_ARG_SERVER_NAME);
GPR_ASSERT(channel_arg != NULL);
GPR_ASSERT(channel_arg->type == GRPC_ARG_STRING);
GPR_ASSERT(strcmp(res->expected_server_name, channel_arg->value.string) == 0);
grpc_channel_args_destroy(res->resolver_result);
}

@ -181,12 +181,7 @@ static grpc_resolver* fake_resolver_create(grpc_resolver_factory* factory,
// Instantiate resolver.
fake_resolver* r = gpr_malloc(sizeof(fake_resolver));
memset(r, 0, sizeof(*r));
grpc_arg server_name_arg;
server_name_arg.type = GRPC_ARG_STRING;
server_name_arg.key = GRPC_ARG_SERVER_NAME;
server_name_arg.value.string = args->uri->path;
r->channel_args =
grpc_channel_args_copy_and_add(args->args, &server_name_arg, 1);
r->channel_args = grpc_channel_args_copy(args->args);
r->addresses = addresses;
gpr_mu_init(&r->mu);
grpc_resolver_init(&r->base, &fake_resolver_vtable);

@ -216,9 +216,11 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
const grpc_call_final_info *final_info,
void *and_free_memory) {}
static void init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {}
static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
return GRPC_ERROR_NONE;
}
static void destroy_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem) {}

@ -243,9 +243,11 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
const grpc_call_final_info *final_info,
void *and_free_memory) {}
static void init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {}
static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
return GRPC_ERROR_NONE;
}
static void destroy_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem) {}

@ -281,9 +281,11 @@ static void server_destroy_call_elem(grpc_exec_ctx *exec_ctx,
gpr_mu_unlock(&g_mu);
}
static void init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {}
static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem,
grpc_channel_element_args *args) {
return GRPC_ERROR_NONE;
}
static void destroy_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem) {}

@ -31,9 +31,14 @@
*
*/
#include <string.h>
#include <grpc/grpc.h>
#include <grpc/support/log.h>
#include "src/core/ext/client_channel/resolver_registry.h"
#include "src/core/lib/channel/channel_stack.h"
#include "src/core/lib/surface/channel.h"
#include "test/core/util/test_config.h"
void test_unknown_scheme_target(void) {
@ -44,6 +49,13 @@ void test_unknown_scheme_target(void) {
chan = grpc_insecure_channel_create("blah://blah", NULL, NULL);
GPR_ASSERT(chan != NULL);
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
grpc_channel_element *elem =
grpc_channel_stack_element(grpc_channel_get_channel_stack(chan), 0);
GPR_ASSERT(0 == strcmp(elem->filter->name, "lame-client"));
grpc_exec_ctx_finish(&exec_ctx);
grpc_channel_destroy(chan);
}

Loading…
Cancel
Save