diff options
Diffstat (limited to 'src/core/client_config')
25 files changed, 2904 insertions, 0 deletions
diff --git a/src/core/client_config/README.md b/src/core/client_config/README.md new file mode 100644 index 0000000000..7cb19cd130 --- /dev/null +++ b/src/core/client_config/README.md @@ -0,0 +1,44 @@ +Client Configuration Support for GRPC +===================================== + +This library provides high level configuration machinery to construct client +channels and load balance between them. + +Each grpc_channel is created with a grpc_resolver. It is the resolver's duty +to resolve a name into configuration data for the channel. Such configuration +data might include: + +- a list of (ip, port) addresses to connect to +- a load balancing policy to decide which server to send a request to +- a set of filters to mutate outgoing requests (say, by adding metadata) + +The resolver provides this data as a stream of grpc_client_config objects to +the channel. We represent configuration as a stream so that it can be changed +by the resolver during execution, by reacting to external events (such as a +new configuration file being pushed to some store). + + +Load Balancing +-------------- + +Load balancing configuration is provided by a grpc_lb_policy object, stored as +part of grpc_client_config. + +A load balancing policies primary job is to pick a target server given only the +initial metadata for a request. It does this by providing a grpc_subchannel +object to the owning channel. + + +Sub-Channels +------------ + +A sub-channel provides a connection to a server for a client channel. It has a +connectivity state like a regular channel, and so can be connected or +disconnected. This connectivity state can be used to inform load balancing +decisions (for example, by avoiding disconnected backends). + +Configured sub-channels are fully setup to participate in the grpc data plane. +Their behavior is specified by a set of grpc channel filters defined at their +construction. To customize this behavior, resolvers build grpc_subchannel_factory +objects, which use the decorator pattern to customize construction arguments for +concrete grpc_subchannel instances. diff --git a/src/core/client_config/client_config.c b/src/core/client_config/client_config.c new file mode 100644 index 0000000000..4453824148 --- /dev/null +++ b/src/core/client_config/client_config.c @@ -0,0 +1,74 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include "src/core/client_config/client_config.h" + +#include <string.h> + +#include <grpc/support/alloc.h> + +struct grpc_client_config { + gpr_refcount refs; + grpc_lb_policy *lb_policy; +}; + +grpc_client_config *grpc_client_config_create() { + grpc_client_config *c = gpr_malloc(sizeof(*c)); + memset(c, 0, sizeof(*c)); + gpr_ref_init(&c->refs, 1); + return c; +} + +void grpc_client_config_ref(grpc_client_config *c) { gpr_ref(&c->refs); } + +void grpc_client_config_unref(grpc_client_config *c) { + if (gpr_unref(&c->refs)) { + GRPC_LB_POLICY_UNREF(c->lb_policy, "client_config"); + gpr_free(c); + } +} + +void grpc_client_config_set_lb_policy(grpc_client_config *c, + grpc_lb_policy *lb_policy) { + if (lb_policy) { + GRPC_LB_POLICY_REF(lb_policy, "client_config"); + } + if (c->lb_policy) { + GRPC_LB_POLICY_UNREF(c->lb_policy, "client_config"); + } + c->lb_policy = lb_policy; +} + +grpc_lb_policy *grpc_client_config_get_lb_policy(grpc_client_config *c) { + return c->lb_policy; +} diff --git a/src/core/client_config/client_config.h b/src/core/client_config/client_config.h new file mode 100644 index 0000000000..47612da42c --- /dev/null +++ b/src/core/client_config/client_config.h @@ -0,0 +1,52 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#ifndef GRPC_INTERNAL_CORE_CLIENT_CONFIG_CLIENT_CONFIG_H +#define GRPC_INTERNAL_CORE_CLIENT_CONFIG_CLIENT_CONFIG_H + +#include "src/core/client_config/lb_policy.h" + +/** Total configuration for a client. Provided, and updated, by + grpc_resolver */ +typedef struct grpc_client_config grpc_client_config; + +grpc_client_config *grpc_client_config_create(); +void grpc_client_config_ref(grpc_client_config *client_config); +void grpc_client_config_unref(grpc_client_config *client_config); + +void grpc_client_config_set_lb_policy(grpc_client_config *client_config, + grpc_lb_policy *lb_policy); +grpc_lb_policy *grpc_client_config_get_lb_policy( + grpc_client_config *client_config); + +#endif /* GRPC_INTERNAL_CORE_CLIENT_CONFIG_CLIENT_CONFIG_H */ diff --git a/src/core/client_config/connector.c b/src/core/client_config/connector.c new file mode 100644 index 0000000000..a8cd5fc149 --- /dev/null +++ b/src/core/client_config/connector.c @@ -0,0 +1,49 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include "src/core/client_config/connector.h" + +void grpc_connector_ref(grpc_connector *connector) { + connector->vtable->ref(connector); +} + +void grpc_connector_unref(grpc_connector *connector) { + connector->vtable->unref(connector); +} + +void grpc_connector_connect(grpc_connector *connector, + const grpc_connect_in_args *in_args, + grpc_connect_out_args *out_args, + grpc_iomgr_closure *notify) { + connector->vtable->connect(connector, in_args, out_args, notify); +} diff --git a/src/core/client_config/connector.h b/src/core/client_config/connector.h new file mode 100644 index 0000000000..edcb10a36e --- /dev/null +++ b/src/core/client_config/connector.h @@ -0,0 +1,85 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#ifndef GRPC_INTERNAL_CORE_CLIENT_CONFIG_CONNECTOR_H +#define GRPC_INTERNAL_CORE_CLIENT_CONFIG_CONNECTOR_H + +#include "src/core/channel/channel_stack.h" +#include "src/core/iomgr/sockaddr.h" +#include "src/core/transport/transport.h" + +typedef struct grpc_connector grpc_connector; +typedef struct grpc_connector_vtable grpc_connector_vtable; + +struct grpc_connector { + const grpc_connector_vtable *vtable; +}; + +typedef struct { + /** set of pollsets interested in this connection */ + grpc_pollset_set *interested_parties; + /** address to connect to */ + const struct sockaddr *addr; + int addr_len; + /** deadline for connection */ + gpr_timespec deadline; + /** channel arguments (to be passed to transport) */ + const grpc_channel_args *channel_args; + /** metadata context */ + grpc_mdctx *metadata_context; +} grpc_connect_in_args; + +typedef struct { + /** the connected transport */ + grpc_transport *transport; + /** any additional filters (owned by the caller of connect) */ + const grpc_channel_filter **filters; + size_t num_filters; +} grpc_connect_out_args; + +struct grpc_connector_vtable { + void (*ref)(grpc_connector *connector); + void (*unref)(grpc_connector *connector); + void (*connect)(grpc_connector *connector, + const grpc_connect_in_args *in_args, + grpc_connect_out_args *out_args, grpc_iomgr_closure *notify); +}; + +void grpc_connector_ref(grpc_connector *connector); +void grpc_connector_unref(grpc_connector *connector); +void grpc_connector_connect(grpc_connector *connector, + const grpc_connect_in_args *in_args, + grpc_connect_out_args *out_args, + grpc_iomgr_closure *notify); + +#endif diff --git a/src/core/client_config/lb_policies/pick_first.c b/src/core/client_config/lb_policies/pick_first.c new file mode 100644 index 0000000000..3d57e3136a --- /dev/null +++ b/src/core/client_config/lb_policies/pick_first.c @@ -0,0 +1,270 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include "src/core/client_config/lb_policies/pick_first.h" + +#include <string.h> + +#include <grpc/support/alloc.h> +#include "src/core/transport/connectivity_state.h" + +typedef struct pending_pick { + struct pending_pick *next; + grpc_pollset *pollset; + grpc_subchannel **target; + grpc_iomgr_closure *on_complete; +} pending_pick; + +typedef struct { + /** base policy: must be first */ + grpc_lb_policy base; + /** all our subchannels */ + grpc_subchannel **subchannels; + size_t num_subchannels; + + grpc_iomgr_closure connectivity_changed; + + /** mutex protecting remaining members */ + gpr_mu mu; + /** the selected channel + TODO(ctiller): this should be atomically set so we don't + need to take a mutex in the common case */ + grpc_subchannel *selected; + /** have we started picking? */ + int started_picking; + /** which subchannel are we watching? */ + size_t checking_subchannel; + /** what is the connectivity of that channel? */ + grpc_connectivity_state checking_connectivity; + /** list of picks that are waiting on connectivity */ + pending_pick *pending_picks; + + /** our connectivity state tracker */ + grpc_connectivity_state_tracker state_tracker; +} pick_first_lb_policy; + +void pf_destroy(grpc_lb_policy *pol) { + pick_first_lb_policy *p = (pick_first_lb_policy *)pol; + size_t i; + for (i = 0; i < p->num_subchannels; i++) { + GRPC_SUBCHANNEL_UNREF(p->subchannels[i], "pick_first"); + } + gpr_free(p->subchannels); + gpr_mu_destroy(&p->mu); + gpr_free(p); +} + +void pf_shutdown(grpc_lb_policy *pol) { + pick_first_lb_policy *p = (pick_first_lb_policy *)pol; + pending_pick *pp; + gpr_mu_lock(&p->mu); + while ((pp = p->pending_picks)) { + p->pending_picks = pp->next; + *pp->target = NULL; + grpc_iomgr_add_delayed_callback(pp->on_complete, 0); + gpr_free(pp); + } + gpr_mu_unlock(&p->mu); +} + +void pf_pick(grpc_lb_policy *pol, grpc_pollset *pollset, + grpc_metadata_batch *initial_metadata, grpc_subchannel **target, + grpc_iomgr_closure *on_complete) { + pick_first_lb_policy *p = (pick_first_lb_policy *)pol; + pending_pick *pp; + gpr_mu_lock(&p->mu); + if (p->selected) { + gpr_mu_unlock(&p->mu); + *target = p->selected; + on_complete->cb(on_complete->cb_arg, 1); + } else { + if (!p->started_picking) { + p->started_picking = 1; + p->checking_subchannel = 0; + p->checking_connectivity = GRPC_CHANNEL_IDLE; + GRPC_LB_POLICY_REF(pol, "pick_first_connectivity"); + grpc_subchannel_notify_on_state_change( + p->subchannels[p->checking_subchannel], &p->checking_connectivity, + &p->connectivity_changed); + } + grpc_subchannel_add_interested_party(p->subchannels[p->checking_subchannel], + pollset); + pp = gpr_malloc(sizeof(*pp)); + pp->next = p->pending_picks; + pp->pollset = pollset; + pp->target = target; + pp->on_complete = on_complete; + p->pending_picks = pp; + gpr_mu_unlock(&p->mu); + } +} + +static void del_interested_parties_locked(pick_first_lb_policy *p) { + pending_pick *pp; + for (pp = p->pending_picks; pp; pp = pp->next) { + grpc_subchannel_del_interested_party(p->subchannels[p->checking_subchannel], + pp->pollset); + } +} + +static void add_interested_parties_locked(pick_first_lb_policy *p) { + pending_pick *pp; + for (pp = p->pending_picks; pp; pp = pp->next) { + grpc_subchannel_add_interested_party(p->subchannels[p->checking_subchannel], + pp->pollset); + } +} + +static void pf_connectivity_changed(void *arg, int iomgr_success) { + pick_first_lb_policy *p = arg; + pending_pick *pp; + int unref = 0; + + gpr_mu_lock(&p->mu); +loop: + switch (p->checking_connectivity) { + case GRPC_CHANNEL_READY: + p->selected = p->subchannels[p->checking_subchannel]; + GPR_ASSERT(grpc_subchannel_check_connectivity(p->selected) == + GRPC_CHANNEL_READY); + while ((pp = p->pending_picks)) { + p->pending_picks = pp->next; + *pp->target = p->selected; + grpc_subchannel_del_interested_party(p->selected, pp->pollset); + grpc_iomgr_add_delayed_callback(pp->on_complete, 1); + gpr_free(pp); + } + unref = 1; + break; + case GRPC_CHANNEL_TRANSIENT_FAILURE: + del_interested_parties_locked(p); + p->checking_subchannel = + (p->checking_subchannel + 1) % p->num_subchannels; + p->checking_connectivity = grpc_subchannel_check_connectivity( + p->subchannels[p->checking_subchannel]); + add_interested_parties_locked(p); + goto loop; + case GRPC_CHANNEL_CONNECTING: + case GRPC_CHANNEL_IDLE: + grpc_subchannel_notify_on_state_change( + p->subchannels[p->checking_subchannel], &p->checking_connectivity, + &p->connectivity_changed); + break; + case GRPC_CHANNEL_FATAL_FAILURE: + del_interested_parties_locked(p); + GPR_SWAP(grpc_subchannel *, p->subchannels[p->checking_subchannel], + p->subchannels[p->num_subchannels - 1]); + p->num_subchannels--; + if (p->num_subchannels == 0) { + while ((pp = p->pending_picks)) { + p->pending_picks = pp->next; + *pp->target = NULL; + grpc_iomgr_add_delayed_callback(pp->on_complete, 1); + gpr_free(pp); + } + unref = 1; + } else { + p->checking_subchannel %= p->num_subchannels; + p->checking_connectivity = grpc_subchannel_check_connectivity( + p->subchannels[p->checking_subchannel]); + GRPC_SUBCHANNEL_UNREF(p->subchannels[p->num_subchannels], "pick_first"); + add_interested_parties_locked(p); + goto loop; + } + } + gpr_mu_unlock(&p->mu); + + if (unref) { + GRPC_LB_POLICY_UNREF(&p->base, "pick_first_connectivity"); + } +} + +static void pf_broadcast(grpc_lb_policy *pol, grpc_transport_op *op) { + pick_first_lb_policy *p = (pick_first_lb_policy *)pol; + size_t i; + size_t n; + grpc_subchannel **subchannels; + + gpr_mu_lock(&p->mu); + n = p->num_subchannels; + subchannels = gpr_malloc(n * sizeof(*subchannels)); + for (i = 0; i < n; i++) { + subchannels[i] = p->subchannels[i]; + GRPC_SUBCHANNEL_REF(subchannels[i], "pf_broadcast"); + } + gpr_mu_unlock(&p->mu); + + for (i = 0; i < n; i++) { + grpc_subchannel_process_transport_op(subchannels[i], op); + GRPC_SUBCHANNEL_UNREF(subchannels[i], "pf_broadcast"); + } + gpr_free(subchannels); +} + +static grpc_connectivity_state pf_check_connectivity(grpc_lb_policy *pol) { + pick_first_lb_policy *p = (pick_first_lb_policy *)pol; + grpc_connectivity_state st; + gpr_mu_lock(&p->mu); + st = grpc_connectivity_state_check(&p->state_tracker); + gpr_mu_unlock(&p->mu); + return st; +} + +static void pf_notify_on_state_change(grpc_lb_policy *pol, + grpc_connectivity_state *current, + grpc_iomgr_closure *notify) { + pick_first_lb_policy *p = (pick_first_lb_policy *)pol; + gpr_mu_lock(&p->mu); + grpc_connectivity_state_notify_on_state_change(&p->state_tracker, current, + notify); + gpr_mu_unlock(&p->mu); +} + +static const grpc_lb_policy_vtable pick_first_lb_policy_vtable = { + pf_destroy, pf_shutdown, pf_pick, + pf_broadcast, pf_check_connectivity, pf_notify_on_state_change}; + +grpc_lb_policy *grpc_create_pick_first_lb_policy(grpc_subchannel **subchannels, + size_t num_subchannels) { + pick_first_lb_policy *p = gpr_malloc(sizeof(*p)); + GPR_ASSERT(num_subchannels); + memset(p, 0, sizeof(*p)); + grpc_lb_policy_init(&p->base, &pick_first_lb_policy_vtable); + p->subchannels = gpr_malloc(sizeof(grpc_subchannel *) * num_subchannels); + p->num_subchannels = num_subchannels; + memcpy(p->subchannels, subchannels, + sizeof(grpc_subchannel *) * num_subchannels); + grpc_iomgr_closure_init(&p->connectivity_changed, pf_connectivity_changed, p); + gpr_mu_init(&p->mu); + return &p->base; +} diff --git a/src/core/client_config/lb_policies/pick_first.h b/src/core/client_config/lb_policies/pick_first.h new file mode 100644 index 0000000000..94c2a9f0c7 --- /dev/null +++ b/src/core/client_config/lb_policies/pick_first.h @@ -0,0 +1,42 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#ifndef GRPC_INTERNAL_CORE_CLIENT_CONFIG_PICK_FIRST_H +#define GRPC_INTERNAL_CORE_CLIENT_CONFIG_PICK_FIRST_H + +#include "src/core/client_config/lb_policy.h" + +grpc_lb_policy *grpc_create_pick_first_lb_policy(grpc_subchannel **subchannels, + size_t num_subchannels); + +#endif diff --git a/src/core/client_config/lb_policy.c b/src/core/client_config/lb_policy.c new file mode 100644 index 0000000000..6d1c788742 --- /dev/null +++ b/src/core/client_config/lb_policy.c @@ -0,0 +1,79 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include "src/core/client_config/lb_policy.h" + +void grpc_lb_policy_init(grpc_lb_policy *policy, + const grpc_lb_policy_vtable *vtable) { + policy->vtable = vtable; + gpr_ref_init(&policy->refs, 1); +} + +#ifdef GRPC_LB_POLICY_REFCOUNT_DEBUG +void grpc_lb_policy_ref(grpc_lb_policy *policy, const char *file, int line, + const char *reason) { + gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "LB_POLICY:%p ref %d -> %d %s", + policy, (int)policy->refs.count, (int)policy->refs.count + 1, reason); +#else +void grpc_lb_policy_ref(grpc_lb_policy *policy) { +#endif + gpr_ref(&policy->refs); +} + +#ifdef GRPC_LB_POLICY_REFCOUNT_DEBUG +void grpc_lb_policy_unref(grpc_lb_policy *policy, const char *file, int line, + const char *reason) { + gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "LB_POLICY:%p unref %d -> %d %s", + policy, (int)policy->refs.count, (int)policy->refs.count - 1, reason); +#else +void grpc_lb_policy_unref(grpc_lb_policy *policy) { +#endif + if (gpr_unref(&policy->refs)) { + policy->vtable->destroy(policy); + } +} + +void grpc_lb_policy_shutdown(grpc_lb_policy *policy) { + policy->vtable->shutdown(policy); +} + +void grpc_lb_policy_pick(grpc_lb_policy *policy, grpc_pollset *pollset, + grpc_metadata_batch *initial_metadata, + grpc_subchannel **target, + grpc_iomgr_closure *on_complete) { + policy->vtable->pick(policy, pollset, initial_metadata, target, on_complete); +} + +void grpc_lb_policy_broadcast(grpc_lb_policy *policy, grpc_transport_op *op) { + policy->vtable->broadcast(policy, op); +} diff --git a/src/core/client_config/lb_policy.h b/src/core/client_config/lb_policy.h new file mode 100644 index 0000000000..a468f761cc --- /dev/null +++ b/src/core/client_config/lb_policy.h @@ -0,0 +1,109 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#ifndef GRPC_INTERNAL_CORE_CLIENT_CONFIG_LB_POLICY_H +#define GRPC_INTERNAL_CORE_CLIENT_CONFIG_LB_POLICY_H + +#include "src/core/client_config/subchannel.h" + +/** A load balancing policy: specified by a vtable and a struct (which + is expected to be extended to contain some parameters) */ +typedef struct grpc_lb_policy grpc_lb_policy; +typedef struct grpc_lb_policy_vtable grpc_lb_policy_vtable; + +typedef void (*grpc_lb_completion)(void *cb_arg, grpc_subchannel *subchannel, + grpc_status_code status, const char *errmsg); + +struct grpc_lb_policy { + const grpc_lb_policy_vtable *vtable; + gpr_refcount refs; +}; + +struct grpc_lb_policy_vtable { + void (*destroy)(grpc_lb_policy *policy); + + void (*shutdown)(grpc_lb_policy *policy); + + /** implement grpc_lb_policy_pick */ + void (*pick)(grpc_lb_policy *policy, grpc_pollset *pollset, + grpc_metadata_batch *initial_metadata, grpc_subchannel **target, + grpc_iomgr_closure *on_complete); + + /** broadcast a transport op to all subchannels */ + void (*broadcast)(grpc_lb_policy *policy, grpc_transport_op *op); + + /** check the current connectivity of the lb_policy */ + grpc_connectivity_state (*check_connectivity)(grpc_lb_policy *policy); + + /** call notify when the connectivity state of a channel changes from *state. + Updates *state with the new state of the policy */ + void (*notify_on_state_change)(grpc_lb_policy *policy, + grpc_connectivity_state *state, + grpc_iomgr_closure *closure); +}; + +#ifdef GRPC_LB_POLICY_REFCOUNT_DEBUG +#define GRPC_LB_POLICY_REF(p, r) \ + grpc_lb_policy_ref((p), __FILE__, __LINE__, (r)) +#define GRPC_LB_POLICY_UNREF(p, r) \ + grpc_lb_policy_unref((p), __FILE__, __LINE__, (r)) +void grpc_lb_policy_ref(grpc_lb_policy *policy, const char *file, int line, + const char *reason); +void grpc_lb_policy_unref(grpc_lb_policy *policy, const char *file, int line, + const char *reason); +#else +#define GRPC_LB_POLICY_REF(p, r) grpc_lb_policy_ref((p)) +#define GRPC_LB_POLICY_UNREF(p, r) grpc_lb_policy_unref((p)) +void grpc_lb_policy_ref(grpc_lb_policy *policy); +void grpc_lb_policy_unref(grpc_lb_policy *policy); +#endif + +/** called by concrete implementations to initialize the base struct */ +void grpc_lb_policy_init(grpc_lb_policy *policy, + const grpc_lb_policy_vtable *vtable); + +/** Start shutting down (fail any pending picks) */ +void grpc_lb_policy_shutdown(grpc_lb_policy *policy); + +/** Given initial metadata in \a initial_metadata, find an appropriate + target for this rpc, and 'return' it by calling \a on_complete after setting + \a target. + Picking can be asynchronous. Any IO should be done under \a pollset. */ +void grpc_lb_policy_pick(grpc_lb_policy *policy, grpc_pollset *pollset, + grpc_metadata_batch *initial_metadata, + grpc_subchannel **target, + grpc_iomgr_closure *on_complete); + +void grpc_lb_policy_broadcast(grpc_lb_policy *policy, grpc_transport_op *op); + +#endif /* GRPC_INTERNAL_CORE_CONFIG_LB_POLICY_H */ diff --git a/src/core/client_config/resolver.c b/src/core/client_config/resolver.c new file mode 100644 index 0000000000..91e42bb684 --- /dev/null +++ b/src/core/client_config/resolver.c @@ -0,0 +1,83 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include "src/core/client_config/resolver.h" + +void grpc_resolver_init(grpc_resolver *resolver, + const grpc_resolver_vtable *vtable) { + resolver->vtable = vtable; + gpr_ref_init(&resolver->refs, 1); +} + +#ifdef GRPC_RESOLVER_REFCOUNT_DEBUG +void grpc_resolver_ref(grpc_resolver *resolver, const char *file, int line, + const char *reason) { + gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "RESOLVER:%p ref %d -> %d %s", + resolver, (int)resolver->refs.count, (int)resolver->refs.count + 1, + reason); +#else +void grpc_resolver_ref(grpc_resolver *resolver) { +#endif + gpr_ref(&resolver->refs); +} + +#ifdef GRPC_RESOLVER_REFCOUNT_DEBUG +void grpc_resolver_unref(grpc_resolver *resolver, const char *file, int line, + const char *reason) { + gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "RESOLVER:%p unref %d -> %d %s", + resolver, (int)resolver->refs.count, (int)resolver->refs.count - 1, + reason); +#else +void grpc_resolver_unref(grpc_resolver *resolver) { +#endif + if (gpr_unref(&resolver->refs)) { + resolver->vtable->destroy(resolver); + } +} + +void grpc_resolver_shutdown(grpc_resolver *resolver) { + resolver->vtable->shutdown(resolver); +} + +void grpc_resolver_channel_saw_error(grpc_resolver *resolver, + struct sockaddr *failing_address, + int failing_address_len) { + resolver->vtable->channel_saw_error(resolver, failing_address, + failing_address_len); +} + +void grpc_resolver_next(grpc_resolver *resolver, + grpc_client_config **target_config, + grpc_iomgr_closure *on_complete) { + resolver->vtable->next(resolver, target_config, on_complete); +} diff --git a/src/core/client_config/resolver.h b/src/core/client_config/resolver.h new file mode 100644 index 0000000000..8ad87d789b --- /dev/null +++ b/src/core/client_config/resolver.h @@ -0,0 +1,97 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#ifndef GRPC_INTERNAL_CORE_CLIENT_CONFIG_RESOLVER_H +#define GRPC_INTERNAL_CORE_CLIENT_CONFIG_RESOLVER_H + +#include "src/core/client_config/client_config.h" +#include "src/core/iomgr/iomgr.h" +#include "src/core/iomgr/sockaddr.h" + +typedef struct grpc_resolver grpc_resolver; +typedef struct grpc_resolver_vtable grpc_resolver_vtable; + +/** grpc_resolver provides grpc_client_config objects to grpc_channel + objects */ +struct grpc_resolver { + const grpc_resolver_vtable *vtable; + gpr_refcount refs; +}; + +struct grpc_resolver_vtable { + void (*destroy)(grpc_resolver *resolver); + void (*shutdown)(grpc_resolver *resolver); + void (*channel_saw_error)(grpc_resolver *resolver, + struct sockaddr *failing_address, + int failing_address_len); + void (*next)(grpc_resolver *resolver, grpc_client_config **target_config, + grpc_iomgr_closure *on_complete); +}; + +#ifdef GRPC_RESOLVER_REFCOUNT_DEBUG +#define GRPC_RESOLVER_REF(p, r) grpc_resolver_ref((p), __FILE__, __LINE__, (r)) +#define GRPC_RESOLVER_UNREF(p, r) \ + grpc_resolver_unref((p), __FILE__, __LINE__, (r)) +void grpc_resolver_ref(grpc_resolver *policy, const char *file, int line, + const char *reason); +void grpc_resolver_unref(grpc_resolver *policy, const char *file, int line, + const char *reason); +#else +#define GRPC_RESOLVER_REF(p, r) grpc_resolver_ref((p)) +#define GRPC_RESOLVER_UNREF(p, r) grpc_resolver_unref((p)) +void grpc_resolver_ref(grpc_resolver *policy); +void grpc_resolver_unref(grpc_resolver *policy); +#endif + +void grpc_resolver_init(grpc_resolver *resolver, + const grpc_resolver_vtable *vtable); + +void grpc_resolver_shutdown(grpc_resolver *resolver); + +/** Notification that the channel has seen an error on some address. + Can be used as a hint that re-resolution is desirable soon. */ +void grpc_resolver_channel_saw_error(grpc_resolver *resolver, + struct sockaddr *failing_address, + int failing_address_len); + +/** Get the next client config. Called by the channel to fetch a new + configuration. Expected to set *target_config with a new configuration, + and then schedule on_complete for execution. + + If resolution is fatally broken, set *target_config to NULL and + schedule on_complete. */ +void grpc_resolver_next(grpc_resolver *resolver, + grpc_client_config **target_config, + grpc_iomgr_closure *on_complete); + +#endif /* GRPC_INTERNAL_CORE_CONFIG_RESOLVER_H */ diff --git a/src/core/client_config/resolver_factory.c b/src/core/client_config/resolver_factory.c new file mode 100644 index 0000000000..6721977e21 --- /dev/null +++ b/src/core/client_config/resolver_factory.c @@ -0,0 +1,50 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include "src/core/client_config/resolver_factory.h" + +void grpc_resolver_factory_ref(grpc_resolver_factory *factory) { + factory->vtable->ref(factory); +} + +void grpc_resolver_factory_unref(grpc_resolver_factory *factory) { + factory->vtable->unref(factory); +} + +/** Create a resolver instance for a name */ +grpc_resolver *grpc_resolver_factory_create_resolver( + grpc_resolver_factory *factory, grpc_uri *uri, + grpc_subchannel_factory *subchannel_factory) { + if (!factory) return NULL; + return factory->vtable->create_resolver(factory, uri, subchannel_factory); +} diff --git a/src/core/client_config/resolver_factory.h b/src/core/client_config/resolver_factory.h new file mode 100644 index 0000000000..c5d85499c6 --- /dev/null +++ b/src/core/client_config/resolver_factory.h @@ -0,0 +1,67 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#ifndef GRPC_INTERNAL_CORE_CLIENT_CONFIG_RESOLVER_FACTORY_H +#define GRPC_INTERNAL_CORE_CLIENT_CONFIG_RESOLVER_FACTORY_H + +#include "src/core/client_config/resolver.h" +#include "src/core/client_config/subchannel_factory.h" +#include "src/core/client_config/uri_parser.h" + +typedef struct grpc_resolver_factory grpc_resolver_factory; +typedef struct grpc_resolver_factory_vtable grpc_resolver_factory_vtable; + +/** grpc_resolver provides grpc_client_config objects to grpc_channel + objects */ +struct grpc_resolver_factory { + const grpc_resolver_factory_vtable *vtable; +}; + +struct grpc_resolver_factory_vtable { + void (*ref)(grpc_resolver_factory *factory); + void (*unref)(grpc_resolver_factory *factory); + + grpc_resolver *(*create_resolver)( + grpc_resolver_factory *factory, grpc_uri *uri, + grpc_subchannel_factory *subchannel_factory); +}; + +void grpc_resolver_factory_ref(grpc_resolver_factory *resolver); +void grpc_resolver_factory_unref(grpc_resolver_factory *resolver); + +/** Create a resolver instance for a name */ +grpc_resolver *grpc_resolver_factory_create_resolver( + grpc_resolver_factory *factory, grpc_uri *uri, + grpc_subchannel_factory *subchannel_factory); + +#endif /* GRPC_INTERNAL_CORE_CONFIG_RESOLVER_FACTORY_H */ diff --git a/src/core/client_config/resolver_registry.c b/src/core/client_config/resolver_registry.c new file mode 100644 index 0000000000..16be2da994 --- /dev/null +++ b/src/core/client_config/resolver_registry.c @@ -0,0 +1,124 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include "src/core/client_config/resolver_registry.h" + +#include <string.h> + +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> +#include <grpc/support/string_util.h> + +#define MAX_RESOLVERS 10 + +typedef struct { + char *scheme; + grpc_resolver_factory *factory; +} registered_resolver; + +static registered_resolver g_all_of_the_resolvers[MAX_RESOLVERS]; +static int g_number_of_resolvers = 0; + +static char *g_default_resolver_scheme; + +void grpc_resolver_registry_init(const char *default_resolver_scheme) { + g_number_of_resolvers = 0; + g_default_resolver_scheme = gpr_strdup(default_resolver_scheme); +} + +void grpc_resolver_registry_shutdown(void) { + int i; + for (i = 0; i < g_number_of_resolvers; i++) { + gpr_free(g_all_of_the_resolvers[i].scheme); + grpc_resolver_factory_unref(g_all_of_the_resolvers[i].factory); + } + gpr_free(g_default_resolver_scheme); +} + +void grpc_register_resolver_type(const char *scheme, + grpc_resolver_factory *factory) { + int i; + for (i = 0; i < g_number_of_resolvers; i++) { + GPR_ASSERT(0 != strcmp(scheme, g_all_of_the_resolvers[i].scheme)); + } + GPR_ASSERT(g_number_of_resolvers != MAX_RESOLVERS); + g_all_of_the_resolvers[g_number_of_resolvers].scheme = gpr_strdup(scheme); + grpc_resolver_factory_ref(factory); + g_all_of_the_resolvers[g_number_of_resolvers].factory = factory; + g_number_of_resolvers++; +} + +static grpc_resolver_factory *lookup_factory(grpc_uri *uri) { + int i; + + /* handling NULL uri's here simplifies grpc_resolver_create */ + if (!uri) return NULL; + + for (i = 0; i < g_number_of_resolvers; i++) { + if (0 == strcmp(uri->scheme, g_all_of_the_resolvers[i].scheme)) { + return g_all_of_the_resolvers[i].factory; + } + } + + return NULL; +} + +grpc_resolver *grpc_resolver_create( + const char *name, grpc_subchannel_factory *subchannel_factory) { + grpc_uri *uri; + char *tmp; + grpc_resolver_factory *factory = NULL; + grpc_resolver *resolver; + + uri = grpc_uri_parse(name, 1); + factory = lookup_factory(uri); + if (factory == NULL && g_default_resolver_scheme != NULL) { + grpc_uri_destroy(uri); + gpr_asprintf(&tmp, "%s%s", g_default_resolver_scheme, name); + uri = grpc_uri_parse(tmp, 1); + factory = lookup_factory(uri); + if (factory == NULL) { + grpc_uri_destroy(grpc_uri_parse(name, 0)); + grpc_uri_destroy(grpc_uri_parse(tmp, 0)); + gpr_log(GPR_ERROR, "don't know how to resolve '%s' or '%s'", name, tmp); + } + gpr_free(tmp); + } else if (factory == NULL) { + grpc_uri_destroy(grpc_uri_parse(name, 0)); + gpr_log(GPR_ERROR, "don't know how to resolve '%s'", name); + } + resolver = + grpc_resolver_factory_create_resolver(factory, uri, subchannel_factory); + grpc_uri_destroy(uri); + return resolver; +} diff --git a/src/core/client_config/resolver_registry.h b/src/core/client_config/resolver_registry.h new file mode 100644 index 0000000000..31aa47620a --- /dev/null +++ b/src/core/client_config/resolver_registry.h @@ -0,0 +1,62 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#ifndef GRPC_INTERNAL_CORE_CLIENT_CONFIG_RESOLVER_REGISTRY_H +#define GRPC_INTERNAL_CORE_CLIENT_CONFIG_RESOLVER_REGISTRY_H + +#include "src/core/client_config/resolver_factory.h" + +void grpc_resolver_registry_init(const char *default_prefix); +void grpc_resolver_registry_shutdown(void); + +/** Register a resolver type. + URI's of \a scheme will be resolved with the given resolver. + If \a priority is greater than zero, then the resolver will be eligible + to resolve names that are passed in with no scheme. Higher priority + resolvers will be tried before lower priority schemes. */ +void grpc_register_resolver_type(const char *scheme, + grpc_resolver_factory *factory); + +/** Create a resolver given \a name. + First tries to parse \a name as a URI. If this succeeds, tries + to locate a registered resolver factory based on the URI scheme. + If parsing or location fails, prefixes default_prefix from + grpc_resolver_registry_init to name, and tries again (if default_prefix + was not NULL). + If a resolver factory was found, use it to instantiate a resolver and + return it. + If a resolver factory was not found, return NULL. */ +grpc_resolver *grpc_resolver_create( + const char *name, grpc_subchannel_factory *subchannel_factory); + +#endif /* GRPC_INTERNAL_CORE_CLIENT_CONFIG_RESOLVER_REGISTRY_H */ diff --git a/src/core/client_config/resolvers/dns_resolver.c b/src/core/client_config/resolvers/dns_resolver.c new file mode 100644 index 0000000000..ac401bc4d3 --- /dev/null +++ b/src/core/client_config/resolvers/dns_resolver.c @@ -0,0 +1,246 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include "src/core/client_config/resolvers/dns_resolver.h" + +#include <string.h> + +#include <grpc/support/alloc.h> +#include <grpc/support/string_util.h> + +#include "src/core/client_config/lb_policies/pick_first.h" +#include "src/core/iomgr/resolve_address.h" +#include "src/core/support/string.h" + +typedef struct { + /** base class: must be first */ + grpc_resolver base; + /** refcount */ + gpr_refcount refs; + /** name to resolve */ + char *name; + /** default port to use */ + char *default_port; + /** subchannel factory */ + grpc_subchannel_factory *subchannel_factory; + /** load balancing policy factory */ + grpc_lb_policy *(*lb_policy_factory)(grpc_subchannel **subchannels, + size_t num_subchannels); + + /** mutex guarding the rest of the state */ + gpr_mu mu; + /** are we currently resolving? */ + int resolving; + /** which version of resolved_config have we published? */ + int published_version; + /** which version of resolved_config is current? */ + int resolved_version; + /** pending next completion, or NULL */ + grpc_iomgr_closure *next_completion; + /** target config address for next completion */ + grpc_client_config **target_config; + /** current (fully resolved) config */ + grpc_client_config *resolved_config; +} dns_resolver; + +static void dns_destroy(grpc_resolver *r); + +static void dns_start_resolving_locked(dns_resolver *r); +static void dns_maybe_finish_next_locked(dns_resolver *r); + +static void dns_shutdown(grpc_resolver *r); +static void dns_channel_saw_error(grpc_resolver *r, + struct sockaddr *failing_address, + int failing_address_len); +static void dns_next(grpc_resolver *r, grpc_client_config **target_config, + grpc_iomgr_closure *on_complete); + +static const grpc_resolver_vtable dns_resolver_vtable = { + dns_destroy, dns_shutdown, dns_channel_saw_error, dns_next}; + +static void dns_shutdown(grpc_resolver *resolver) { + dns_resolver *r = (dns_resolver *)resolver; + gpr_mu_lock(&r->mu); + if (r->next_completion != NULL) { + *r->target_config = NULL; + grpc_iomgr_add_callback(r->next_completion); + r->next_completion = NULL; + } + gpr_mu_unlock(&r->mu); +} + +static void dns_channel_saw_error(grpc_resolver *resolver, struct sockaddr *sa, + int len) { + dns_resolver *r = (dns_resolver *)resolver; + gpr_mu_lock(&r->mu); + if (!r->resolving) { + dns_start_resolving_locked(r); + } + gpr_mu_unlock(&r->mu); +} + +static void dns_next(grpc_resolver *resolver, + grpc_client_config **target_config, + grpc_iomgr_closure *on_complete) { + dns_resolver *r = (dns_resolver *)resolver; + gpr_mu_lock(&r->mu); + GPR_ASSERT(!r->next_completion); + r->next_completion = on_complete; + r->target_config = target_config; + if (r->resolved_version == 0 && !r->resolving) { + dns_start_resolving_locked(r); + } else { + dns_maybe_finish_next_locked(r); + } + gpr_mu_unlock(&r->mu); +} + +static void dns_on_resolved(void *arg, grpc_resolved_addresses *addresses) { + dns_resolver *r = arg; + grpc_client_config *config = NULL; + grpc_subchannel **subchannels; + grpc_subchannel_args args; + grpc_lb_policy *lb_policy; + size_t i; + if (addresses) { + config = grpc_client_config_create(); + subchannels = gpr_malloc(sizeof(grpc_subchannel *) * addresses->naddrs); + for (i = 0; i < addresses->naddrs; i++) { + memset(&args, 0, sizeof(args)); + args.addr = (struct sockaddr *)(addresses->addrs[i].addr); + args.addr_len = addresses->addrs[i].len; + subchannels[i] = grpc_subchannel_factory_create_subchannel( + r->subchannel_factory, &args); + } + lb_policy = r->lb_policy_factory(subchannels, addresses->naddrs); + grpc_client_config_set_lb_policy(config, lb_policy); + GRPC_LB_POLICY_UNREF(lb_policy, "construction"); + grpc_resolved_addresses_destroy(addresses); + gpr_free(subchannels); + } + gpr_mu_lock(&r->mu); + GPR_ASSERT(r->resolving); + r->resolving = 0; + if (r->resolved_config) { + grpc_client_config_unref(r->resolved_config); + } + r->resolved_config = config; + r->resolved_version++; + dns_maybe_finish_next_locked(r); + gpr_mu_unlock(&r->mu); + + GRPC_RESOLVER_UNREF(&r->base, "dns-resolving"); +} + +static void dns_start_resolving_locked(dns_resolver *r) { + GRPC_RESOLVER_REF(&r->base, "dns-resolving"); + GPR_ASSERT(!r->resolving); + r->resolving = 1; + grpc_resolve_address(r->name, r->default_port, dns_on_resolved, r); +} + +static void dns_maybe_finish_next_locked(dns_resolver *r) { + if (r->next_completion != NULL && + r->resolved_version != r->published_version) { + *r->target_config = r->resolved_config; + if (r->resolved_config) { + grpc_client_config_ref(r->resolved_config); + } + grpc_iomgr_add_callback(r->next_completion); + r->next_completion = NULL; + r->published_version = r->resolved_version; + } +} + +static void dns_destroy(grpc_resolver *gr) { + dns_resolver *r = (dns_resolver *)gr; + gpr_mu_destroy(&r->mu); + if (r->resolved_config) { + grpc_client_config_unref(r->resolved_config); + } + grpc_subchannel_factory_unref(r->subchannel_factory); + gpr_free(r->name); + gpr_free(r->default_port); + gpr_free(r); +} + +static grpc_resolver *dns_create( + grpc_uri *uri, const char *default_port, + grpc_lb_policy *(*lb_policy_factory)(grpc_subchannel **subchannels, + size_t num_subchannels), + grpc_subchannel_factory *subchannel_factory) { + dns_resolver *r; + const char *path = uri->path; + + if (0 != strcmp(uri->authority, "")) { + gpr_log(GPR_ERROR, "authority based uri's not supported"); + return NULL; + } + + if (path[0] == '/') ++path; + + r = gpr_malloc(sizeof(dns_resolver)); + memset(r, 0, sizeof(*r)); + gpr_ref_init(&r->refs, 1); + gpr_mu_init(&r->mu); + grpc_resolver_init(&r->base, &dns_resolver_vtable); + r->name = gpr_strdup(path); + r->default_port = gpr_strdup(default_port); + r->subchannel_factory = subchannel_factory; + r->lb_policy_factory = lb_policy_factory; + grpc_subchannel_factory_ref(subchannel_factory); + return &r->base; +} + +/* + * FACTORY + */ + +static void dns_factory_ref(grpc_resolver_factory *factory) {} + +static void dns_factory_unref(grpc_resolver_factory *factory) {} + +static grpc_resolver *dns_factory_create_resolver( + grpc_resolver_factory *factory, grpc_uri *uri, + grpc_subchannel_factory *subchannel_factory) { + return dns_create(uri, "https", grpc_create_pick_first_lb_policy, + subchannel_factory); +} + +static const grpc_resolver_factory_vtable dns_factory_vtable = { + dns_factory_ref, dns_factory_unref, dns_factory_create_resolver}; +static grpc_resolver_factory dns_resolver_factory = {&dns_factory_vtable}; + +grpc_resolver_factory *grpc_dns_resolver_factory_create() { + return &dns_resolver_factory; +} diff --git a/src/core/client_config/resolvers/dns_resolver.h b/src/core/client_config/resolvers/dns_resolver.h new file mode 100644 index 0000000000..09322b16cf --- /dev/null +++ b/src/core/client_config/resolvers/dns_resolver.h @@ -0,0 +1,42 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#ifndef GRPC_INTERNAL_CORE_CLIENT_CONFIG_RESOLVERS_DNS_RESOLVER_H +#define GRPC_INTERNAL_CORE_CLIENT_CONFIG_RESOLVERS_DNS_RESOLVER_H + +#include "src/core/client_config/resolver_factory.h" + +/** Create a dns resolver for \a name */ +grpc_resolver_factory *grpc_dns_resolver_factory_create(void); + +#endif /* GRPC_INTERNAL_CORE_CLIENT_CONFIG_RESOLVERS_DNS_RESOLVER_H */ diff --git a/src/core/client_config/resolvers/unix_resolver_posix.c b/src/core/client_config/resolvers/unix_resolver_posix.c new file mode 100644 index 0000000000..be515d2689 --- /dev/null +++ b/src/core/client_config/resolvers/unix_resolver_posix.c @@ -0,0 +1,195 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include <grpc/support/port_platform.h> +#ifdef GPR_POSIX_SOCKET + +#include "src/core/client_config/resolvers/unix_resolver_posix.h" + +#include <string.h> +#include <sys/un.h> + +#include <grpc/support/alloc.h> +#include <grpc/support/string_util.h> + +#include "src/core/client_config/lb_policies/pick_first.h" +#include "src/core/iomgr/resolve_address.h" +#include "src/core/support/string.h" + +typedef struct { + /** base class: must be first */ + grpc_resolver base; + /** refcount */ + gpr_refcount refs; + /** subchannel factory */ + grpc_subchannel_factory *subchannel_factory; + /** load balancing policy factory */ + grpc_lb_policy *(*lb_policy_factory)(grpc_subchannel **subchannels, + size_t num_subchannels); + + /** the address that we've 'resolved' */ + struct sockaddr_un addr; + int addr_len; + + /** mutex guarding the rest of the state */ + gpr_mu mu; + /** have we published? */ + int published; + /** pending next completion, or NULL */ + grpc_iomgr_closure *next_completion; + /** target config address for next completion */ + grpc_client_config **target_config; +} unix_resolver; + +static void unix_destroy(grpc_resolver *r); + +static void unix_maybe_finish_next_locked(unix_resolver *r); + +static void unix_shutdown(grpc_resolver *r); +static void unix_channel_saw_error(grpc_resolver *r, + struct sockaddr *failing_address, + int failing_address_len); +static void unix_next(grpc_resolver *r, grpc_client_config **target_config, + grpc_iomgr_closure *on_complete); + +static const grpc_resolver_vtable unix_resolver_vtable = { + unix_destroy, unix_shutdown, unix_channel_saw_error, unix_next}; + +static void unix_shutdown(grpc_resolver *resolver) { + unix_resolver *r = (unix_resolver *)resolver; + gpr_mu_lock(&r->mu); + if (r->next_completion != NULL) { + *r->target_config = NULL; + /* TODO(ctiller): add delayed callback */ + grpc_iomgr_add_callback(r->next_completion); + r->next_completion = NULL; + } + gpr_mu_unlock(&r->mu); +} + +static void unix_channel_saw_error(grpc_resolver *resolver, struct sockaddr *sa, + int len) {} + +static void unix_next(grpc_resolver *resolver, + grpc_client_config **target_config, + grpc_iomgr_closure *on_complete) { + unix_resolver *r = (unix_resolver *)resolver; + gpr_mu_lock(&r->mu); + GPR_ASSERT(!r->next_completion); + r->next_completion = on_complete; + r->target_config = target_config; + unix_maybe_finish_next_locked(r); + gpr_mu_unlock(&r->mu); +} + +static void unix_maybe_finish_next_locked(unix_resolver *r) { + grpc_client_config *cfg; + grpc_lb_policy *lb_policy; + grpc_subchannel *subchannel; + grpc_subchannel_args args; + + if (r->next_completion != NULL && !r->published) { + cfg = grpc_client_config_create(); + memset(&args, 0, sizeof(args)); + args.addr = (struct sockaddr *)&r->addr; + args.addr_len = r->addr_len; + subchannel = + grpc_subchannel_factory_create_subchannel(r->subchannel_factory, &args); + lb_policy = r->lb_policy_factory(&subchannel, 1); + grpc_client_config_set_lb_policy(cfg, lb_policy); + GRPC_LB_POLICY_UNREF(lb_policy, "unix"); + r->published = 1; + *r->target_config = cfg; + grpc_iomgr_add_callback(r->next_completion); + r->next_completion = NULL; + } +} + +static void unix_destroy(grpc_resolver *gr) { + unix_resolver *r = (unix_resolver *)gr; + gpr_mu_destroy(&r->mu); + grpc_subchannel_factory_unref(r->subchannel_factory); + gpr_free(r); +} + +static grpc_resolver *unix_create( + grpc_uri *uri, + grpc_lb_policy *(*lb_policy_factory)(grpc_subchannel **subchannels, + size_t num_subchannels), + grpc_subchannel_factory *subchannel_factory) { + unix_resolver *r; + + if (0 != strcmp(uri->authority, "")) { + gpr_log(GPR_ERROR, "authority based uri's not supported"); + return NULL; + } + + r = gpr_malloc(sizeof(unix_resolver)); + memset(r, 0, sizeof(*r)); + gpr_ref_init(&r->refs, 1); + gpr_mu_init(&r->mu); + grpc_resolver_init(&r->base, &unix_resolver_vtable); + r->subchannel_factory = subchannel_factory; + r->lb_policy_factory = lb_policy_factory; + + r->addr.sun_family = AF_UNIX; + strcpy(r->addr.sun_path, uri->path); + r->addr_len = strlen(r->addr.sun_path) + sizeof(r->addr.sun_family) + 1; + + grpc_subchannel_factory_ref(subchannel_factory); + return &r->base; +} + +/* + * FACTORY + */ + +static void unix_factory_ref(grpc_resolver_factory *factory) {} + +static void unix_factory_unref(grpc_resolver_factory *factory) {} + +static grpc_resolver *unix_factory_create_resolver( + grpc_resolver_factory *factory, grpc_uri *uri, + grpc_subchannel_factory *subchannel_factory) { + return unix_create(uri, grpc_create_pick_first_lb_policy, subchannel_factory); +} + +static const grpc_resolver_factory_vtable unix_factory_vtable = { + unix_factory_ref, unix_factory_unref, unix_factory_create_resolver}; +static grpc_resolver_factory unix_resolver_factory = {&unix_factory_vtable}; + +grpc_resolver_factory *grpc_unix_resolver_factory_create() { + return &unix_resolver_factory; +} + +#endif diff --git a/src/core/client_config/resolvers/unix_resolver_posix.h b/src/core/client_config/resolvers/unix_resolver_posix.h new file mode 100644 index 0000000000..a84906f4d6 --- /dev/null +++ b/src/core/client_config/resolvers/unix_resolver_posix.h @@ -0,0 +1,44 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#ifndef GRPC_INTERNAL_CORE_CLIENT_CONFIG_RESOLVERS_UNIX_RESOLVER_H +#define GRPC_INTERNAL_CORE_CLIENT_CONFIG_RESOLVERS_UNIX_RESOLVER_H + +#include <grpc/support/port_platform.h> + +#include "src/core/client_config/resolver_factory.h" + +/** Create a dns resolver for \a name */ +grpc_resolver_factory *grpc_unix_resolver_factory_create(void); + +#endif /* GRPC_INTERNAL_CORE_CLIENT_CONFIG_RESOLVERS_UNIX_RESOLVER_H */ diff --git a/src/core/client_config/subchannel.c b/src/core/client_config/subchannel.c new file mode 100644 index 0000000000..6cf9062ab0 --- /dev/null +++ b/src/core/client_config/subchannel.c @@ -0,0 +1,659 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include "src/core/client_config/subchannel.h" + +#include <string.h> + +#include <grpc/support/alloc.h> + +#include "src/core/channel/channel_args.h" +#include "src/core/channel/connected_channel.h" +#include "src/core/iomgr/alarm.h" +#include "src/core/transport/connectivity_state.h" + +typedef struct { + /* all fields protected by subchannel->mu */ + /** refcount */ + int refs; + /** parent subchannel */ + grpc_subchannel *subchannel; +} connection; + +typedef struct { + grpc_iomgr_closure closure; + size_t version; + grpc_subchannel *subchannel; + grpc_connectivity_state connectivity_state; +} state_watcher; + +typedef struct waiting_for_connect { + struct waiting_for_connect *next; + grpc_iomgr_closure *notify; + grpc_pollset *pollset; + grpc_subchannel_call **target; + grpc_subchannel *subchannel; + grpc_iomgr_closure continuation; +} waiting_for_connect; + +struct grpc_subchannel { + grpc_connector *connector; + + /** non-transport related channel filters */ + const grpc_channel_filter **filters; + size_t num_filters; + /** channel arguments */ + grpc_channel_args *args; + /** address to connect to */ + struct sockaddr *addr; + size_t addr_len; + /** metadata context */ + grpc_mdctx *mdctx; + /** master channel - the grpc_channel instance that ultimately owns + this channel_data via its channel stack. + We occasionally use this to bump the refcount on the master channel + to keep ourselves alive through an asynchronous operation. */ + grpc_channel *master; + /** have we seen a disconnection? */ + int disconnected; + + /** set during connection */ + grpc_connect_out_args connecting_result; + + /** callback for connection finishing */ + grpc_iomgr_closure connected; + + /** pollset_set tracking who's interested in a connection + being setup */ + grpc_pollset_set pollset_set; + + /** mutex protecting remaining elements */ + gpr_mu mu; + + /** active connection */ + connection *active; + /** version number for the active connection */ + size_t active_version; + /** refcount */ + int refs; + /** are we connecting */ + int connecting; + /** things waiting for a connection */ + waiting_for_connect *waiting; + /** connectivity state tracking */ + grpc_connectivity_state_tracker state_tracker; + + /** next connect attempt time */ + gpr_timespec next_attempt; + /** amount to backoff each failure */ + gpr_timespec backoff_delta; + /** do we have an active alarm? */ + int have_alarm; + /** our alarm */ + grpc_alarm alarm; +}; + +struct grpc_subchannel_call { + connection *connection; + gpr_refcount refs; +}; + +#define SUBCHANNEL_CALL_TO_CALL_STACK(call) ((grpc_call_stack *)((call) + 1)) +#define CHANNEL_STACK_FROM_CONNECTION(con) ((grpc_channel_stack *)((con) + 1)) + +static grpc_subchannel_call *create_call(connection *con); +static void connectivity_state_changed_locked(grpc_subchannel *c); +static grpc_connectivity_state compute_connectivity_locked(grpc_subchannel *c); +static gpr_timespec compute_connect_deadline(grpc_subchannel *c); +static void subchannel_connected(void *subchannel, int iomgr_success); + +static void subchannel_ref_locked( + grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS); +static int subchannel_unref_locked( + grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) GRPC_MUST_USE_RESULT; +static void connection_ref_locked(connection *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS); +static grpc_subchannel *connection_unref_locked( + connection *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) GRPC_MUST_USE_RESULT; +static void subchannel_destroy(grpc_subchannel *c); + +#ifdef GRPC_SUBCHANNEL_REFCOUNT_DEBUG +#define SUBCHANNEL_REF_LOCKED(p, r) \ + subchannel_ref_locked((p), __FILE__, __LINE__, (r)) +#define SUBCHANNEL_UNREF_LOCKED(p, r) \ + subchannel_unref_locked((p), __FILE__, __LINE__, (r)) +#define CONNECTION_REF_LOCKED(p, r) \ + connection_ref_locked((p), __FILE__, __LINE__, (r)) +#define CONNECTION_UNREF_LOCKED(p, r) \ + connection_unref_locked((p), __FILE__, __LINE__, (r)) +#define REF_PASS_ARGS , file, line, reason +#define REF_LOG(name, p) \ + gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "%s: %p ref %d -> %d %s", \ + (name), (p), (p)->refs, (p)->refs + 1, reason) +#define UNREF_LOG(name, p) \ + gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "%s: %p unref %d -> %d %s", \ + (name), (p), (p)->refs, (p)->refs - 1, reason) +#else +#define SUBCHANNEL_REF_LOCKED(p, r) subchannel_ref_locked((p)) +#define SUBCHANNEL_UNREF_LOCKED(p, r) subchannel_unref_locked((p)) +#define CONNECTION_REF_LOCKED(p, r) connection_ref_locked((p)) +#define CONNECTION_UNREF_LOCKED(p, r) connection_unref_locked((p)) +#define REF_PASS_ARGS +#define REF_LOG(name, p) \ + do { \ + } while (0) +#define UNREF_LOG(name, p) \ + do { \ + } while (0) +#endif + +/* + * connection implementation + */ + +static void connection_destroy(connection *c) { + GPR_ASSERT(c->refs == 0); + grpc_channel_stack_destroy(CHANNEL_STACK_FROM_CONNECTION(c)); + gpr_free(c); +} + +static void connection_ref_locked( + connection *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) { + REF_LOG("CONNECTION", c); + subchannel_ref_locked(c->subchannel REF_PASS_ARGS); + ++c->refs; +} + +static grpc_subchannel *connection_unref_locked( + connection *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) { + grpc_subchannel *destroy = NULL; + UNREF_LOG("CONNECTION", c); + if (subchannel_unref_locked(c->subchannel REF_PASS_ARGS)) { + destroy = c->subchannel; + } + if (--c->refs == 0 && c->subchannel->active != c) { + connection_destroy(c); + } + return destroy; +} + +/* + * grpc_subchannel implementation + */ + +static void subchannel_ref_locked( + grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) { + REF_LOG("SUBCHANNEL", c); + ++c->refs; +} + +static int subchannel_unref_locked( + grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) { + UNREF_LOG("SUBCHANNEL", c); + return --c->refs == 0; +} + +void grpc_subchannel_ref(grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) { + gpr_mu_lock(&c->mu); + subchannel_ref_locked(c REF_PASS_ARGS); + gpr_mu_unlock(&c->mu); +} + +void grpc_subchannel_unref(grpc_subchannel *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) { + int destroy; + gpr_mu_lock(&c->mu); + destroy = subchannel_unref_locked(c REF_PASS_ARGS); + gpr_mu_unlock(&c->mu); + if (destroy) subchannel_destroy(c); +} + +static void subchannel_destroy(grpc_subchannel *c) { + if (c->active != NULL) { + connection_destroy(c->active); + } + gpr_free(c->filters); + grpc_channel_args_destroy(c->args); + gpr_free(c->addr); + grpc_mdctx_unref(c->mdctx); + grpc_pollset_set_destroy(&c->pollset_set); + grpc_connectivity_state_destroy(&c->state_tracker); + grpc_connector_unref(c->connector); + gpr_free(c); +} + +void grpc_subchannel_add_interested_party(grpc_subchannel *c, + grpc_pollset *pollset) { + grpc_pollset_set_add_pollset(&c->pollset_set, pollset); +} + +void grpc_subchannel_del_interested_party(grpc_subchannel *c, + grpc_pollset *pollset) { + grpc_pollset_set_del_pollset(&c->pollset_set, pollset); +} + +grpc_subchannel *grpc_subchannel_create(grpc_connector *connector, + grpc_subchannel_args *args) { + grpc_subchannel *c = gpr_malloc(sizeof(*c)); + memset(c, 0, sizeof(*c)); + c->refs = 1; + c->connector = connector; + grpc_connector_ref(c->connector); + c->num_filters = args->filter_count; + c->filters = gpr_malloc(sizeof(grpc_channel_filter *) * c->num_filters); + memcpy(c->filters, args->filters, + sizeof(grpc_channel_filter *) * c->num_filters); + c->addr = gpr_malloc(args->addr_len); + memcpy(c->addr, args->addr, args->addr_len); + c->addr_len = args->addr_len; + c->args = grpc_channel_args_copy(args->args); + c->mdctx = args->mdctx; + c->master = args->master; + grpc_mdctx_ref(c->mdctx); + grpc_pollset_set_init(&c->pollset_set); + grpc_iomgr_closure_init(&c->connected, subchannel_connected, c); + grpc_connectivity_state_init(&c->state_tracker, GRPC_CHANNEL_IDLE); + gpr_mu_init(&c->mu); + return c; +} + +static void continue_connect(grpc_subchannel *c) { + grpc_connect_in_args args; + + args.interested_parties = &c->pollset_set; + args.addr = c->addr; + args.addr_len = c->addr_len; + args.deadline = compute_connect_deadline(c); + args.channel_args = c->args; + args.metadata_context = c->mdctx; + + grpc_connector_connect(c->connector, &args, &c->connecting_result, + &c->connected); +} + +static void start_connect(grpc_subchannel *c) { + gpr_timespec now = gpr_now(); + c->next_attempt = now; + c->backoff_delta = gpr_time_from_seconds(1); + + continue_connect(c); +} + +static void continue_creating_call(void *arg, int iomgr_success) { + waiting_for_connect *w4c = arg; + grpc_subchannel_create_call(w4c->subchannel, w4c->pollset, w4c->target, + w4c->notify); + GRPC_SUBCHANNEL_UNREF(w4c->subchannel, "waiting_for_connect"); + gpr_free(w4c); +} + +void grpc_subchannel_create_call(grpc_subchannel *c, grpc_pollset *pollset, + grpc_subchannel_call **target, + grpc_iomgr_closure *notify) { + connection *con; + gpr_mu_lock(&c->mu); + if (c->active != NULL) { + con = c->active; + CONNECTION_REF_LOCKED(con, "call"); + gpr_mu_unlock(&c->mu); + + *target = create_call(con); + notify->cb(notify->cb_arg, 1); + } else { + waiting_for_connect *w4c = gpr_malloc(sizeof(*w4c)); + w4c->next = c->waiting; + w4c->notify = notify; + w4c->pollset = pollset; + w4c->target = target; + w4c->subchannel = c; + /* released when clearing w4c */ + SUBCHANNEL_REF_LOCKED(c, "waiting_for_connect"); + grpc_iomgr_closure_init(&w4c->continuation, continue_creating_call, w4c); + c->waiting = w4c; + grpc_subchannel_add_interested_party(c, pollset); + if (!c->connecting) { + c->connecting = 1; + connectivity_state_changed_locked(c); + /* released by connection */ + SUBCHANNEL_REF_LOCKED(c, "connecting"); + gpr_mu_unlock(&c->mu); + + start_connect(c); + } else { + gpr_mu_unlock(&c->mu); + } + } +} + +grpc_connectivity_state grpc_subchannel_check_connectivity(grpc_subchannel *c) { + grpc_connectivity_state state; + gpr_mu_lock(&c->mu); + state = grpc_connectivity_state_check(&c->state_tracker); + gpr_mu_unlock(&c->mu); + return state; +} + +void grpc_subchannel_notify_on_state_change(grpc_subchannel *c, + grpc_connectivity_state *state, + grpc_iomgr_closure *notify) { + int do_connect = 0; + gpr_mu_lock(&c->mu); + if (grpc_connectivity_state_notify_on_state_change(&c->state_tracker, state, + notify)) { + do_connect = 1; + c->connecting = 1; + /* released by connection */ + SUBCHANNEL_REF_LOCKED(c, "connecting"); + connectivity_state_changed_locked(c); + } + gpr_mu_unlock(&c->mu); + if (do_connect) { + start_connect(c); + } +} + +void grpc_subchannel_process_transport_op(grpc_subchannel *c, + grpc_transport_op *op) { + connection *con = NULL; + grpc_subchannel *destroy; + int cancel_alarm = 0; + gpr_mu_lock(&c->mu); + if (op->disconnect) { + c->disconnected = 1; + connectivity_state_changed_locked(c); + if (c->have_alarm) { + cancel_alarm = 1; + } + } + if (c->active != NULL) { + con = c->active; + CONNECTION_REF_LOCKED(con, "transport-op"); + } + gpr_mu_unlock(&c->mu); + + if (con != NULL) { + grpc_channel_stack *channel_stack = CHANNEL_STACK_FROM_CONNECTION(con); + grpc_channel_element *top_elem = + grpc_channel_stack_element(channel_stack, 0); + top_elem->filter->start_transport_op(top_elem, op); + + gpr_mu_lock(&c->mu); + destroy = CONNECTION_UNREF_LOCKED(con, "transport-op"); + gpr_mu_unlock(&c->mu); + if (destroy) { + subchannel_destroy(destroy); + } + } + + if (cancel_alarm) { + grpc_alarm_cancel(&c->alarm); + } +} + +static void on_state_changed(void *p, int iomgr_success) { + state_watcher *sw = p; + grpc_subchannel *c = sw->subchannel; + gpr_mu *mu = &c->mu; + int destroy; + grpc_transport_op op; + grpc_channel_element *elem; + connection *destroy_connection = NULL; + + gpr_mu_lock(mu); + + /* if we failed or there is a version number mismatch, just leave + this closure */ + if (!iomgr_success || sw->subchannel->active_version != sw->version) { + goto done; + } + + switch (sw->connectivity_state) { + case GRPC_CHANNEL_CONNECTING: + case GRPC_CHANNEL_READY: + case GRPC_CHANNEL_IDLE: + /* all is still good: keep watching */ + memset(&op, 0, sizeof(op)); + op.connectivity_state = &sw->connectivity_state; + op.on_connectivity_state_change = &sw->closure; + elem = grpc_channel_stack_element( + CHANNEL_STACK_FROM_CONNECTION(c->active), 0); + elem->filter->start_transport_op(elem, &op); + /* early out */ + gpr_mu_unlock(mu); + return; + case GRPC_CHANNEL_FATAL_FAILURE: + case GRPC_CHANNEL_TRANSIENT_FAILURE: + /* things have gone wrong, deactivate and enter idle */ + if (sw->subchannel->active->refs == 0) { + destroy_connection = sw->subchannel->active; + } + sw->subchannel->active = NULL; + grpc_connectivity_state_set(&c->state_tracker, + GRPC_CHANNEL_TRANSIENT_FAILURE); + break; + } + +done: + connectivity_state_changed_locked(c); + destroy = SUBCHANNEL_UNREF_LOCKED(c, "state_watcher"); + gpr_free(sw); + gpr_mu_unlock(mu); + if (destroy) { + subchannel_destroy(c); + } + if (destroy_connection != NULL) { + connection_destroy(destroy_connection); + } +} + +static void publish_transport(grpc_subchannel *c) { + size_t channel_stack_size; + connection *con; + grpc_channel_stack *stk; + size_t num_filters; + const grpc_channel_filter **filters; + waiting_for_connect *w4c; + grpc_transport_op op; + state_watcher *sw; + connection *destroy_connection = NULL; + grpc_channel_element *elem; + + /* build final filter list */ + num_filters = c->num_filters + c->connecting_result.num_filters + 1; + filters = gpr_malloc(sizeof(*filters) * num_filters); + memcpy(filters, c->filters, sizeof(*filters) * c->num_filters); + memcpy(filters + c->num_filters, c->connecting_result.filters, + sizeof(*filters) * c->connecting_result.num_filters); + filters[num_filters - 1] = &grpc_connected_channel_filter; + + /* construct channel stack */ + channel_stack_size = grpc_channel_stack_size(filters, num_filters); + con = gpr_malloc(sizeof(connection) + channel_stack_size); + stk = (grpc_channel_stack *)(con + 1); + con->refs = 0; + con->subchannel = c; + grpc_channel_stack_init(filters, num_filters, c->master, c->args, c->mdctx, + stk); + grpc_connected_channel_bind_transport(stk, c->connecting_result.transport); + gpr_free(c->connecting_result.filters); + memset(&c->connecting_result, 0, sizeof(c->connecting_result)); + + /* initialize state watcher */ + sw = gpr_malloc(sizeof(*sw)); + grpc_iomgr_closure_init(&sw->closure, on_state_changed, sw); + sw->subchannel = c; + sw->connectivity_state = GRPC_CHANNEL_READY; + + gpr_mu_lock(&c->mu); + + if (c->disconnected) { + gpr_mu_unlock(&c->mu); + gpr_free(sw); + gpr_free(filters); + grpc_channel_stack_destroy(stk); + return; + } + + /* publish */ + if (c->active != NULL && c->active->refs == 0) { + destroy_connection = c->active; + } + c->active = con; + c->active_version++; + sw->version = c->active_version; + c->connecting = 0; + + /* watch for changes; subchannel ref for connecting is donated + to the state watcher */ + memset(&op, 0, sizeof(op)); + op.connectivity_state = &sw->connectivity_state; + op.on_connectivity_state_change = &sw->closure; + SUBCHANNEL_REF_LOCKED(c, "state_watcher"); + GPR_ASSERT(!SUBCHANNEL_UNREF_LOCKED(c, "connecting")); + elem = + grpc_channel_stack_element(CHANNEL_STACK_FROM_CONNECTION(c->active), 0); + elem->filter->start_transport_op(elem, &op); + + /* signal completion */ + connectivity_state_changed_locked(c); + while ((w4c = c->waiting)) { + c->waiting = w4c->next; + grpc_iomgr_add_callback(&w4c->continuation); + } + + gpr_mu_unlock(&c->mu); + + gpr_free(filters); + + if (destroy_connection != NULL) { + connection_destroy(destroy_connection); + } +} + +static void on_alarm(void *arg, int iomgr_success) { + grpc_subchannel *c = arg; + gpr_mu_lock(&c->mu); + c->have_alarm = 0; + if (c->disconnected) { + iomgr_success = 0; + } + connectivity_state_changed_locked(c); + gpr_mu_unlock(&c->mu); + if (iomgr_success) { + continue_connect(c); + } else { + GRPC_SUBCHANNEL_UNREF(c, "connecting"); + } +} + +static void subchannel_connected(void *arg, int iomgr_success) { + grpc_subchannel *c = arg; + if (c->connecting_result.transport != NULL) { + publish_transport(c); + } else { + gpr_mu_lock(&c->mu); + connectivity_state_changed_locked(c); + GPR_ASSERT(!c->have_alarm); + c->have_alarm = 1; + c->next_attempt = gpr_time_add(c->next_attempt, c->backoff_delta); + c->backoff_delta = gpr_time_add(c->backoff_delta, c->backoff_delta); + grpc_alarm_init(&c->alarm, c->next_attempt, on_alarm, c, gpr_now()); + gpr_mu_unlock(&c->mu); + } +} + +static gpr_timespec compute_connect_deadline(grpc_subchannel *c) { + return gpr_time_add(c->next_attempt, c->backoff_delta); +} + +static grpc_connectivity_state compute_connectivity_locked(grpc_subchannel *c) { + if (c->disconnected) { + return GRPC_CHANNEL_FATAL_FAILURE; + } + if (c->connecting) { + if (c->have_alarm) { + return GRPC_CHANNEL_TRANSIENT_FAILURE; + } + return GRPC_CHANNEL_CONNECTING; + } + if (c->active) { + return GRPC_CHANNEL_READY; + } + return GRPC_CHANNEL_IDLE; +} + +static void connectivity_state_changed_locked(grpc_subchannel *c) { + grpc_connectivity_state current = compute_connectivity_locked(c); + grpc_connectivity_state_set(&c->state_tracker, current); +} + +/* + * grpc_subchannel_call implementation + */ + +void grpc_subchannel_call_ref( + grpc_subchannel_call *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) { + gpr_ref(&c->refs); +} + +void grpc_subchannel_call_unref( + grpc_subchannel_call *c GRPC_SUBCHANNEL_REF_EXTRA_ARGS) { + if (gpr_unref(&c->refs)) { + gpr_mu *mu = &c->connection->subchannel->mu; + grpc_subchannel *destroy; + grpc_call_stack_destroy(SUBCHANNEL_CALL_TO_CALL_STACK(c)); + gpr_mu_lock(mu); + destroy = CONNECTION_UNREF_LOCKED(c->connection, "call"); + gpr_mu_unlock(mu); + gpr_free(c); + if (destroy != NULL) { + subchannel_destroy(destroy); + } + } +} + +void grpc_subchannel_call_process_op(grpc_subchannel_call *call, + grpc_transport_stream_op *op) { + grpc_call_stack *call_stack = SUBCHANNEL_CALL_TO_CALL_STACK(call); + grpc_call_element *top_elem = grpc_call_stack_element(call_stack, 0); + top_elem->filter->start_transport_stream_op(top_elem, op); +} + +grpc_subchannel_call *create_call(connection *con) { + grpc_channel_stack *chanstk = CHANNEL_STACK_FROM_CONNECTION(con); + grpc_subchannel_call *call = + gpr_malloc(sizeof(grpc_subchannel_call) + chanstk->call_stack_size); + grpc_call_stack *callstk = SUBCHANNEL_CALL_TO_CALL_STACK(call); + call->connection = con; + gpr_ref_init(&call->refs, 1); + grpc_call_stack_init(chanstk, NULL, NULL, callstk); + return call; +} diff --git a/src/core/client_config/subchannel.h b/src/core/client_config/subchannel.h new file mode 100644 index 0000000000..a23a623277 --- /dev/null +++ b/src/core/client_config/subchannel.h @@ -0,0 +1,124 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#ifndef GRPC_INTERNAL_CORE_CLIENT_CONFIG_SUBCHANNEL_H +#define GRPC_INTERNAL_CORE_CLIENT_CONFIG_SUBCHANNEL_H + +#include "src/core/channel/channel_stack.h" +#include "src/core/client_config/connector.h" + +/** A (sub-)channel that knows how to connect to exactly one target + address. Provides a target for load balancing. */ +typedef struct grpc_subchannel grpc_subchannel; +typedef struct grpc_subchannel_call grpc_subchannel_call; +typedef struct grpc_subchannel_args grpc_subchannel_args; + +#ifdef GRPC_SUBCHANNEL_REFCOUNT_DEBUG +#define GRPC_SUBCHANNEL_REF(p, r) \ + grpc_subchannel_ref((p), __FILE__, __LINE__, (r)) +#define GRPC_SUBCHANNEL_UNREF(p, r) \ + grpc_subchannel_unref((p), __FILE__, __LINE__, (r)) +#define GRPC_SUBCHANNEL_CALL_REF(p, r) \ + grpc_subchannel_call_ref((p), __FILE__, __LINE__, (r)) +#define GRPC_SUBCHANNEL_CALL_UNREF(p, r) \ + grpc_subchannel_call_unref((p), __FILE__, __LINE__, (r)) +#define GRPC_SUBCHANNEL_REF_EXTRA_ARGS \ + , const char *file, int line, const char *reason +#else +#define GRPC_SUBCHANNEL_REF(p, r) grpc_subchannel_ref((p)) +#define GRPC_SUBCHANNEL_UNREF(p, r) grpc_subchannel_unref((p)) +#define GRPC_SUBCHANNEL_CALL_REF(p, r) grpc_subchannel_call_ref((p)) +#define GRPC_SUBCHANNEL_CALL_UNREF(p, r) grpc_subchannel_call_unref((p)) +#define GRPC_SUBCHANNEL_REF_EXTRA_ARGS +#endif + +void grpc_subchannel_ref( + grpc_subchannel *channel GRPC_SUBCHANNEL_REF_EXTRA_ARGS); +void grpc_subchannel_unref( + grpc_subchannel *channel GRPC_SUBCHANNEL_REF_EXTRA_ARGS); +void grpc_subchannel_call_ref( + grpc_subchannel_call *call GRPC_SUBCHANNEL_REF_EXTRA_ARGS); +void grpc_subchannel_call_unref( + grpc_subchannel_call *call GRPC_SUBCHANNEL_REF_EXTRA_ARGS); + +/** construct a call (possibly asynchronously) */ +void grpc_subchannel_create_call(grpc_subchannel *subchannel, + grpc_pollset *pollset, + grpc_subchannel_call **target, + grpc_iomgr_closure *notify); + +/** process a transport level op */ +void grpc_subchannel_process_transport_op(grpc_subchannel *subchannel, + grpc_transport_op *op); + +/** poll the current connectivity state of a channel */ +grpc_connectivity_state grpc_subchannel_check_connectivity( + grpc_subchannel *channel); + +/** call notify when the connectivity state of a channel changes from *state. + Updates *state with the new state of the channel */ +void grpc_subchannel_notify_on_state_change(grpc_subchannel *channel, + grpc_connectivity_state *state, + grpc_iomgr_closure *notify); + +void grpc_subchannel_add_interested_party(grpc_subchannel *channel, + grpc_pollset *pollset); +void grpc_subchannel_del_interested_party(grpc_subchannel *channel, + grpc_pollset *pollset); + +/** continue processing a transport op */ +void grpc_subchannel_call_process_op(grpc_subchannel_call *subchannel_call, + grpc_transport_stream_op *op); + +struct grpc_subchannel_args { + /** Channel filters for this channel - wrapped factories will likely + want to mutate this */ + const grpc_channel_filter **filters; + /** The number of filters in the above array */ + size_t filter_count; + /** Channel arguments to be supplied to the newly created channel */ + const grpc_channel_args *args; + /** Address to connect to */ + struct sockaddr *addr; + size_t addr_len; + /** metadata context to use */ + grpc_mdctx *mdctx; + /** master channel */ + grpc_channel *master; +}; + +/** create a subchannel given a connector */ +grpc_subchannel *grpc_subchannel_create(grpc_connector *connector, + grpc_subchannel_args *args); + +#endif /* GRPC_INTERNAL_CORE_CLIENT_CONFIG_SUBCHANNEL_H */ diff --git a/src/core/client_config/subchannel_factory.c b/src/core/client_config/subchannel_factory.c new file mode 100644 index 0000000000..f71386594c --- /dev/null +++ b/src/core/client_config/subchannel_factory.c @@ -0,0 +1,46 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include "src/core/client_config/subchannel_factory.h" + +void grpc_subchannel_factory_ref(grpc_subchannel_factory *factory) { + factory->vtable->ref(factory); +} +void grpc_subchannel_factory_unref(grpc_subchannel_factory *factory) { + factory->vtable->unref(factory); +} + +grpc_subchannel *grpc_subchannel_factory_create_subchannel( + grpc_subchannel_factory *factory, grpc_subchannel_args *args) { + return factory->vtable->create_subchannel(factory, args); +} diff --git a/src/core/client_config/subchannel_factory.h b/src/core/client_config/subchannel_factory.h new file mode 100644 index 0000000000..d7eae1c964 --- /dev/null +++ b/src/core/client_config/subchannel_factory.h @@ -0,0 +1,63 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#ifndef GRPC_INTERNAL_CORE_CLIENT_CONFIG_SUBCHANNEL_FACTORY_H +#define GRPC_INTERNAL_CORE_CLIENT_CONFIG_SUBCHANNEL_FACTORY_H + +#include "src/core/channel/channel_stack.h" +#include "src/core/client_config/subchannel.h" + +typedef struct grpc_subchannel_factory grpc_subchannel_factory; +typedef struct grpc_subchannel_factory_vtable grpc_subchannel_factory_vtable; + +/** Constructor for new configured channels. + Creating decorators around this type is encouraged to adapt behavior. */ +struct grpc_subchannel_factory { + const grpc_subchannel_factory_vtable *vtable; +}; + +struct grpc_subchannel_factory_vtable { + void (*ref)(grpc_subchannel_factory *factory); + void (*unref)(grpc_subchannel_factory *factory); + grpc_subchannel *(*create_subchannel)(grpc_subchannel_factory *factory, + grpc_subchannel_args *args); +}; + +void grpc_subchannel_factory_ref(grpc_subchannel_factory *factory); +void grpc_subchannel_factory_unref(grpc_subchannel_factory *factory); + +/** Create a new grpc_subchannel */ +grpc_subchannel *grpc_subchannel_factory_create_subchannel( + grpc_subchannel_factory *factory, grpc_subchannel_args *args); + +#endif /* GRPC_INTERNAL_CORE_CLIENT_CONFIG_SUBCHANNEL_FACTORY_H */ diff --git a/src/core/client_config/uri_parser.c b/src/core/client_config/uri_parser.c new file mode 100644 index 0000000000..776a255923 --- /dev/null +++ b/src/core/client_config/uri_parser.c @@ -0,0 +1,149 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include "src/core/client_config/uri_parser.h" + +#include <string.h> + +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> +#include <grpc/support/string_util.h> + +static grpc_uri *bad_uri(const char *uri_text, int pos, const char *section, + int suppress_errors) { + char *line_prefix; + int pfx_len; + + if (!suppress_errors) { + gpr_asprintf(&line_prefix, "bad uri.%s: '", section); + pfx_len = strlen(line_prefix) + pos; + gpr_log(GPR_ERROR, "%s%s'", line_prefix, uri_text); + gpr_free(line_prefix); + + line_prefix = gpr_malloc(pfx_len + 1); + memset(line_prefix, ' ', pfx_len); + line_prefix[pfx_len] = 0; + gpr_log(GPR_ERROR, "%s^ here", line_prefix); + gpr_free(line_prefix); + } + + return NULL; +} + +static char *copy_fragment(const char *src, int begin, int end) { + char *out = gpr_malloc(end - begin + 1); + memcpy(out, src + begin, end - begin); + out[end - begin] = 0; + return out; +} + +grpc_uri *grpc_uri_parse(const char *uri_text, int suppress_errors) { + grpc_uri *uri; + int scheme_begin = 0; + int scheme_end = -1; + int authority_begin = -1; + int authority_end = -1; + int path_begin = -1; + int path_end = -1; + int i; + + for (i = scheme_begin; uri_text[i] != 0; i++) { + if (uri_text[i] == ':') { + scheme_end = i; + break; + } + if (uri_text[i] >= 'a' && uri_text[i] <= 'z') continue; + if (uri_text[i] >= 'A' && uri_text[i] <= 'Z') continue; + if (i != scheme_begin) { + if (uri_text[i] >= '0' && uri_text[i] <= '9') continue; + if (uri_text[i] == '+') continue; + if (uri_text[i] == '-') continue; + if (uri_text[i] == '.') continue; + } + break; + } + if (scheme_end == -1) { + return bad_uri(uri_text, i, "scheme", suppress_errors); + } + + if (uri_text[scheme_end + 1] == '/' && uri_text[scheme_end + 2] == '/') { + authority_begin = scheme_end + 3; + for (i = authority_begin; uri_text[i] != 0; i++) { + if (uri_text[i] == '/') { + authority_end = i; + } + if (uri_text[i] == '?') { + return bad_uri(uri_text, i, "query_not_supported", suppress_errors); + } + if (uri_text[i] == '#') { + return bad_uri(uri_text, i, "fragment_not_supported", suppress_errors); + } + } + if (authority_end == -1 && uri_text[i] == 0) { + authority_end = i; + } + if (authority_end == -1) { + return bad_uri(uri_text, i, "authority", suppress_errors); + } + /* TODO(ctiller): parse the authority correctly */ + path_begin = authority_end; + } else { + path_begin = scheme_end + 1; + } + + for (i = path_begin; uri_text[i] != 0; i++) { + if (uri_text[i] == '?') { + return bad_uri(uri_text, i, "query_not_supported", suppress_errors); + } + if (uri_text[i] == '#') { + return bad_uri(uri_text, i, "fragment_not_supported", suppress_errors); + } + } + path_end = i; + + uri = gpr_malloc(sizeof(*uri)); + memset(uri, 0, sizeof(*uri)); + uri->scheme = copy_fragment(uri_text, scheme_begin, scheme_end); + uri->authority = copy_fragment(uri_text, authority_begin, authority_end); + uri->path = copy_fragment(uri_text, path_begin, path_end); + + return uri; +} + +void grpc_uri_destroy(grpc_uri *uri) { + if (!uri) return; + gpr_free(uri->scheme); + gpr_free(uri->authority); + gpr_free(uri->path); + gpr_free(uri); +} diff --git a/src/core/client_config/uri_parser.h b/src/core/client_config/uri_parser.h new file mode 100644 index 0000000000..ce4e6aecb0 --- /dev/null +++ b/src/core/client_config/uri_parser.h @@ -0,0 +1,49 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#ifndef GRPC_INTERNAL_CORE_CLIENT_CONFIG_URI_PARSER_H +#define GRPC_INTERNAL_CORE_CLIENT_CONFIG_URI_PARSER_H + +typedef struct { + char *scheme; + char *authority; + char *path; +} grpc_uri; + +/** parse a uri, return NULL on failure */ +grpc_uri *grpc_uri_parse(const char *uri_text, int suppress_errors); + +/** destroy a uri */ +void grpc_uri_destroy(grpc_uri *uri); + +#endif |