diff options
Diffstat (limited to 'src/core')
65 files changed, 840 insertions, 320 deletions
diff --git a/src/core/census/record_stat.c b/src/core/census/record_stat.c new file mode 100644 index 0000000000..3dd918618b --- /dev/null +++ b/src/core/census/record_stat.c @@ -0,0 +1,38 @@ +/* + * + * 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/census.h> +#include "src/core/census/rpc_stat_id.h" + +void census_record_stat(census_context *context, census_stat *stats, + size_t nstats) {} diff --git a/src/core/census/rpc_stat_id.h b/src/core/census/rpc_stat_id.h new file mode 100644 index 0000000000..fc0aa6f43f --- /dev/null +++ b/src/core/census/rpc_stat_id.h @@ -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. + * + */ + +#ifndef CENSUS_RPC_STAT_ID_H +#define CENSUS_RPC_STAT_ID_H + +/* Stats ID's used for RPC measurements. */ +#define CENSUS_INVALID_STAT_ID 0 /* ID 0 is always invalid */ +#define CENSUS_RPC_CLIENT_REQUESTS 1 /* Count of client requests sent. */ +#define CENSUS_RPC_SERVER_REQUESTS 2 /* Count of server requests sent. */ +#define CENSUS_RPC_CLIENT_ERRORS 3 /* Client error counts. */ +#define CENSUS_RPC_SERVER_ERRORS 4 /* Server error counts. */ +#define CENSUS_RPC_CLIENT_LATENCY 5 /* Client side request latency. */ +#define CENSUS_RPC_SERVER_LATENCY 6 /* Server side request latency. */ + +#endif /* CENSUS_RPC_STAT_ID_H */ diff --git a/src/core/channel/channel_stack.c b/src/core/channel/channel_stack.c index e38dcb58b7..cd7c182ef2 100644 --- a/src/core/channel/channel_stack.c +++ b/src/core/channel/channel_stack.c @@ -191,6 +191,11 @@ void grpc_call_next_op(grpc_call_element *elem, grpc_transport_stream_op *op) { next_elem->filter->start_transport_stream_op(next_elem, op); } +char *grpc_call_next_get_peer(grpc_call_element *elem) { + grpc_call_element *next_elem = elem + 1; + return next_elem->filter->get_peer(next_elem); +} + void grpc_channel_next_op(grpc_channel_element *elem, grpc_transport_op *op) { grpc_channel_element *next_elem = elem + 1; next_elem->filter->start_transport_op(next_elem, op); diff --git a/src/core/channel/channel_stack.h b/src/core/channel/channel_stack.h index 785be8925b..4a608b956e 100644 --- a/src/core/channel/channel_stack.h +++ b/src/core/channel/channel_stack.h @@ -104,6 +104,9 @@ typedef struct { The filter does not need to do any chaining */ void (*destroy_channel_elem)(grpc_channel_element *elem); + /* Implement grpc_call_get_peer() */ + char *(*get_peer)(grpc_call_element *elem); + /* The name of this filter */ const char *name; } grpc_channel_filter; @@ -173,6 +176,8 @@ void grpc_call_next_op(grpc_call_element *elem, grpc_transport_stream_op *op); /* Call the next operation (depending on call directionality) in a channel stack */ void grpc_channel_next_op(grpc_channel_element *elem, grpc_transport_op *op); +/* Pass through a request to get_peer to the next child element */ +char *grpc_call_next_get_peer(grpc_call_element *elem); /* Given the top element of a channel stack, get the channel stack itself */ grpc_channel_stack *grpc_channel_stack_from_top_element( diff --git a/src/core/channel/client_channel.c b/src/core/channel/client_channel.c index 10e01ebbb4..108a6dfdf1 100644 --- a/src/core/channel/client_channel.c +++ b/src/core/channel/client_channel.c @@ -236,21 +236,6 @@ static void picked_target(void *arg, int iomgr_success) { } } -static void pick_target(grpc_lb_policy *lb_policy, call_data *calld) { - grpc_metadata_batch *initial_metadata; - grpc_transport_stream_op *op = &calld->waiting_op; - - GPR_ASSERT(op->bind_pollset); - GPR_ASSERT(op->send_ops); - GPR_ASSERT(op->send_ops->nops >= 1); - GPR_ASSERT(op->send_ops->ops[0].type == GRPC_OP_METADATA); - initial_metadata = &op->send_ops->ops[0].data.metadata; - - grpc_iomgr_closure_init(&calld->async_setup_task, picked_target, calld); - grpc_lb_policy_pick(lb_policy, op->bind_pollset, initial_metadata, - &calld->picked_channel, &calld->async_setup_task); -} - static grpc_iomgr_closure *merge_into_waiting_op( grpc_call_element *elem, grpc_transport_stream_op *new_op) { call_data *calld = elem->call_data; @@ -280,6 +265,26 @@ static grpc_iomgr_closure *merge_into_waiting_op( return consumed_op; } +static char *cc_get_peer(grpc_call_element *elem) { + call_data *calld = elem->call_data; + channel_data *chand = elem->channel_data; + grpc_subchannel_call *subchannel_call; + char *result; + + gpr_mu_lock(&calld->mu_state); + if (calld->state == CALL_ACTIVE) { + subchannel_call = calld->subchannel_call; + GRPC_SUBCHANNEL_CALL_REF(subchannel_call, "get_peer"); + gpr_mu_unlock(&calld->mu_state); + result = grpc_subchannel_call_get_peer(subchannel_call); + GRPC_SUBCHANNEL_CALL_UNREF(subchannel_call, "get_peer"); + return result; + } else { + gpr_mu_unlock(&calld->mu_state); + return grpc_channel_get_target(chand->master); + } +} + static void perform_transport_stream_op(grpc_call_element *elem, grpc_transport_stream_op *op, int continuation) { @@ -358,12 +363,23 @@ static void perform_transport_stream_op(grpc_call_element *elem, gpr_mu_lock(&chand->mu_config); lb_policy = chand->lb_policy; if (lb_policy) { + grpc_transport_stream_op *op = &calld->waiting_op; + grpc_pollset *bind_pollset = op->bind_pollset; + grpc_metadata_batch *initial_metadata = &op->send_ops->ops[0].data.metadata; GRPC_LB_POLICY_REF(lb_policy, "pick"); gpr_mu_unlock(&chand->mu_config); calld->state = CALL_WAITING_FOR_PICK; + + GPR_ASSERT(op->bind_pollset); + GPR_ASSERT(op->send_ops); + GPR_ASSERT(op->send_ops->nops >= 1); + GPR_ASSERT( + op->send_ops->ops[0].type == GRPC_OP_METADATA); gpr_mu_unlock(&calld->mu_state); - pick_target(lb_policy, calld); + grpc_iomgr_closure_init(&calld->async_setup_task, picked_target, calld); + grpc_lb_policy_pick(lb_policy, bind_pollset, initial_metadata, + &calld->picked_channel, &calld->async_setup_task); GRPC_LB_POLICY_UNREF(lb_policy, "pick"); } else if (chand->resolver != NULL) { @@ -594,6 +610,7 @@ const grpc_channel_filter grpc_client_channel_filter = { sizeof(channel_data), init_channel_elem, destroy_channel_elem, + cc_get_peer, "client-channel", }; diff --git a/src/core/channel/compress_filter.c b/src/core/channel/compress_filter.c index 9b9e82caba..4bf24e7db3 100644 --- a/src/core/channel/compress_filter.c +++ b/src/core/channel/compress_filter.c @@ -202,7 +202,7 @@ static void process_send_ops(grpc_call_element *elem, channeld->default_compression_algorithm; calld->has_compression_algorithm = 1; /* GPR_TRUE */ } - grpc_metadata_batch_add_head( + grpc_metadata_batch_add_tail( &(sop->data.metadata), &calld->compression_algorithm_storage, grpc_mdelem_ref(channeld->mdelem_compression_algorithms [calld->compression_algorithm])); @@ -324,4 +324,5 @@ const grpc_channel_filter grpc_compress_filter = { sizeof(channel_data), init_channel_elem, destroy_channel_elem, + grpc_call_next_get_peer, "compress"}; diff --git a/src/core/channel/connected_channel.c b/src/core/channel/connected_channel.c index 34d07de519..b95ed06f2b 100644 --- a/src/core/channel/connected_channel.c +++ b/src/core/channel/connected_channel.c @@ -119,6 +119,11 @@ static void destroy_channel_elem(grpc_channel_element *elem) { grpc_transport_destroy(cd->transport); } +static char *con_get_peer(grpc_call_element *elem) { + channel_data *chand = elem->channel_data; + return grpc_transport_get_peer(chand->transport); +} + const grpc_channel_filter grpc_connected_channel_filter = { con_start_transport_stream_op, con_start_transport_op, @@ -128,6 +133,7 @@ const grpc_channel_filter grpc_connected_channel_filter = { sizeof(channel_data), init_channel_elem, destroy_channel_elem, + con_get_peer, "connected", }; diff --git a/src/core/channel/http_client_filter.c b/src/core/channel/http_client_filter.c index 63e4912397..6ae8488070 100644 --- a/src/core/channel/http_client_filter.c +++ b/src/core/channel/http_client_filter.c @@ -32,13 +32,17 @@ #include "src/core/channel/http_client_filter.h" #include <string.h> +#include <grpc/support/alloc.h> #include <grpc/support/log.h> +#include <grpc/support/string_util.h> +#include "src/core/support/string.h" typedef struct call_data { grpc_linked_mdelem method; grpc_linked_mdelem scheme; grpc_linked_mdelem te_trailers; grpc_linked_mdelem content_type; + grpc_linked_mdelem user_agent; int sent_initial_metadata; int got_initial_metadata; @@ -58,6 +62,8 @@ typedef struct channel_data { grpc_mdelem *scheme; grpc_mdelem *content_type; grpc_mdelem *status; + /** complete user agent mdelem */ + grpc_mdelem *user_agent; } channel_data; /* used to silence 'variable not used' warnings */ @@ -92,6 +98,18 @@ static void hc_on_recv(void *user_data, int success) { calld->on_done_recv->cb(calld->on_done_recv->cb_arg, success); } +static grpc_mdelem *client_strip_filter(void *user_data, grpc_mdelem *md) { + grpc_call_element *elem = user_data; + channel_data *channeld = elem->channel_data; + /* eat the things we'd like to set ourselves */ + if (md->key == channeld->method->key) return NULL; + if (md->key == channeld->scheme->key) return NULL; + if (md->key == channeld->te_trailers->key) return NULL; + if (md->key == channeld->content_type->key) return NULL; + if (md->key == channeld->user_agent->key) return NULL; + return md; +} + static void hc_mutate_op(grpc_call_element *elem, grpc_transport_stream_op *op) { /* grab pointers to our data from the call element */ @@ -105,6 +123,7 @@ static void hc_mutate_op(grpc_call_element *elem, grpc_stream_op *op = &ops[i]; if (op->type != GRPC_OP_METADATA) continue; calld->sent_initial_metadata = 1; + grpc_metadata_batch_filter(&op->data.metadata, client_strip_filter, elem); /* Send : prefixed headers, which have to be before any application layer headers. */ grpc_metadata_batch_add_head(&op->data.metadata, &calld->method, @@ -115,6 +134,8 @@ static void hc_mutate_op(grpc_call_element *elem, GRPC_MDELEM_REF(channeld->te_trailers)); grpc_metadata_batch_add_tail(&op->data.metadata, &calld->content_type, GRPC_MDELEM_REF(channeld->content_type)); + grpc_metadata_batch_add_tail(&op->data.metadata, &calld->user_agent, + GRPC_MDELEM_REF(channeld->user_agent)); break; } } @@ -169,6 +190,55 @@ static const char *scheme_from_args(const grpc_channel_args *args) { return "http"; } +static grpc_mdstr *user_agent_from_args(grpc_mdctx *mdctx, + const grpc_channel_args *args) { + gpr_strvec v; + size_t i; + int is_first = 1; + char *tmp; + grpc_mdstr *result; + + gpr_strvec_init(&v); + + for (i = 0; args && i < args->num_args; i++) { + if (0 == strcmp(args->args[i].key, GRPC_ARG_PRIMARY_USER_AGENT_STRING)) { + if (args->args[i].type != GRPC_ARG_STRING) { + gpr_log(GPR_ERROR, "Channel argument '%s' should be a string", + GRPC_ARG_PRIMARY_USER_AGENT_STRING); + } else { + if (!is_first) gpr_strvec_add(&v, gpr_strdup(" ")); + is_first = 0; + gpr_strvec_add(&v, gpr_strdup(args->args[i].value.string)); + } + } + } + + gpr_asprintf(&tmp, "%sgrpc-c/%s (%s)", is_first ? "" : " ", + grpc_version_string(), GPR_PLATFORM_STRING); + is_first = 0; + gpr_strvec_add(&v, tmp); + + for (i = 0; args && i < args->num_args; i++) { + if (0 == strcmp(args->args[i].key, GRPC_ARG_SECONDARY_USER_AGENT_STRING)) { + if (args->args[i].type != GRPC_ARG_STRING) { + gpr_log(GPR_ERROR, "Channel argument '%s' should be a string", + GRPC_ARG_SECONDARY_USER_AGENT_STRING); + } else { + if (!is_first) gpr_strvec_add(&v, gpr_strdup(" ")); + is_first = 0; + gpr_strvec_add(&v, gpr_strdup(args->args[i].value.string)); + } + } + } + + tmp = gpr_strvec_flatten(&v, NULL); + gpr_strvec_destroy(&v); + result = grpc_mdstr_from_string(mdctx, tmp); + gpr_free(tmp); + + return result; +} + /* Constructor for channel_data */ static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master, const grpc_channel_args *args, grpc_mdctx *mdctx, @@ -189,6 +259,9 @@ static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master, channeld->content_type = grpc_mdelem_from_strings(mdctx, "content-type", "application/grpc"); channeld->status = grpc_mdelem_from_strings(mdctx, ":status", "200"); + channeld->user_agent = grpc_mdelem_from_metadata_strings( + mdctx, grpc_mdstr_from_string(mdctx, "user-agent"), + user_agent_from_args(mdctx, args)); } /* Destructor for channel data */ @@ -201,9 +274,11 @@ static void destroy_channel_elem(grpc_channel_element *elem) { GRPC_MDELEM_UNREF(channeld->scheme); GRPC_MDELEM_UNREF(channeld->content_type); GRPC_MDELEM_UNREF(channeld->status); + GRPC_MDELEM_UNREF(channeld->user_agent); } const grpc_channel_filter grpc_http_client_filter = { hc_start_transport_op, grpc_channel_next_op, sizeof(call_data), init_call_elem, destroy_call_elem, sizeof(channel_data), - init_channel_elem, destroy_channel_elem, "http-client"}; + init_channel_elem, destroy_channel_elem, grpc_call_next_get_peer, + "http-client"}; diff --git a/src/core/channel/http_server_filter.c b/src/core/channel/http_server_filter.c index a6cbb5a7f4..7c798d2fb4 100644 --- a/src/core/channel/http_server_filter.c +++ b/src/core/channel/http_server_filter.c @@ -280,4 +280,5 @@ static void destroy_channel_elem(grpc_channel_element *elem) { const grpc_channel_filter grpc_http_server_filter = { hs_start_transport_op, grpc_channel_next_op, sizeof(call_data), init_call_elem, destroy_call_elem, sizeof(channel_data), - init_channel_elem, destroy_channel_elem, "http-server"}; + init_channel_elem, destroy_channel_elem, grpc_call_next_get_peer, + "http-server"}; diff --git a/src/core/channel/noop_filter.c b/src/core/channel/noop_filter.c index 5117723617..d631885aaf 100644 --- a/src/core/channel/noop_filter.c +++ b/src/core/channel/noop_filter.c @@ -127,4 +127,5 @@ const grpc_channel_filter grpc_no_op_filter = {noop_start_transport_stream_op, sizeof(channel_data), init_channel_elem, destroy_channel_elem, + grpc_call_next_get_peer, "no-op"}; diff --git a/src/core/client_config/README.md b/src/core/client_config/README.md index d0700cfb13..fff7a5af5b 100644 --- a/src/core/client_config/README.md +++ b/src/core/client_config/README.md @@ -60,3 +60,7 @@ unix:path - the unix scheme is used to create and connect to unix domain sockets - the authority must be empty, and the path represents the absolute or relative path to the desired socket + +ipv4:host:port - a pre-resolved ipv4 dotted decimal address/port combination + +ipv6:[host]:port - a pre-resolved ipv6 address/port combination diff --git a/src/core/client_config/resolvers/sockaddr_resolver.c b/src/core/client_config/resolvers/sockaddr_resolver.c new file mode 100644 index 0000000000..74584e7e2c --- /dev/null +++ b/src/core/client_config/resolvers/sockaddr_resolver.c @@ -0,0 +1,299 @@ +/* + * + * 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> + +#include "src/core/client_config/resolvers/sockaddr_resolver.h" + +#include <stdio.h> +#include <string.h> +#ifdef GPR_POSIX_SOCKET +#include <sys/un.h> +#endif + +#include <grpc/support/alloc.h> +#include <grpc/support/host_port.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_storage 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; +} sockaddr_resolver; + +static void sockaddr_destroy(grpc_resolver *r); + +static void sockaddr_maybe_finish_next_locked(sockaddr_resolver *r); + +static void sockaddr_shutdown(grpc_resolver *r); +static void sockaddr_channel_saw_error(grpc_resolver *r, + struct sockaddr *failing_address, + int failing_address_len); +static void sockaddr_next(grpc_resolver *r, grpc_client_config **target_config, + grpc_iomgr_closure *on_complete); + +static const grpc_resolver_vtable sockaddr_resolver_vtable = { + sockaddr_destroy, sockaddr_shutdown, sockaddr_channel_saw_error, + sockaddr_next}; + +static void sockaddr_shutdown(grpc_resolver *resolver) { + sockaddr_resolver *r = (sockaddr_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 sockaddr_channel_saw_error(grpc_resolver *resolver, + struct sockaddr *sa, int len) {} + +static void sockaddr_next(grpc_resolver *resolver, + grpc_client_config **target_config, + grpc_iomgr_closure *on_complete) { + sockaddr_resolver *r = (sockaddr_resolver *)resolver; + gpr_mu_lock(&r->mu); + GPR_ASSERT(!r->next_completion); + r->next_completion = on_complete; + r->target_config = target_config; + sockaddr_maybe_finish_next_locked(r); + gpr_mu_unlock(&r->mu); +} + +static void sockaddr_maybe_finish_next_locked(sockaddr_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 sockaddr_destroy(grpc_resolver *gr) { + sockaddr_resolver *r = (sockaddr_resolver *)gr; + gpr_mu_destroy(&r->mu); + grpc_subchannel_factory_unref(r->subchannel_factory); + gpr_free(r); +} + +#ifdef GPR_POSIX_SOCKET +static int parse_unix(grpc_uri *uri, struct sockaddr_storage *addr, int *len) { + struct sockaddr_un *un = (struct sockaddr_un *)addr; + + un->sun_family = AF_UNIX; + strcpy(un->sun_path, uri->path); + *len = strlen(un->sun_path) + sizeof(un->sun_family) + 1; + + return 1; +} +#endif + +static int parse_ipv4(grpc_uri *uri, struct sockaddr_storage *addr, int *len) { + const char *host_port = uri->path; + char *host; + char *port; + int port_num; + int result = 0; + struct sockaddr_in *in = (struct sockaddr_in *)addr; + + if (*host_port == '/') ++host_port; + if (!gpr_split_host_port(host_port, &host, &port)) { + return 0; + } + + memset(in, 0, sizeof(*in)); + *len = sizeof(*in); + in->sin_family = AF_INET; + if (inet_pton(AF_INET, host, &in->sin_addr) == 0) { + gpr_log(GPR_ERROR, "invalid ipv4 address: '%s'", host); + goto done; + } + + if (port != NULL) { + if (sscanf(port, "%d", &port_num) != 1 || port_num < 0 || + port_num > 65535) { + gpr_log(GPR_ERROR, "invalid ipv4 port: '%s'", port); + goto done; + } + in->sin_port = htons(port_num); + } else { + gpr_log(GPR_ERROR, "no port given for ipv4 scheme"); + goto done; + } + + result = 1; +done: + gpr_free(host); + gpr_free(port); + return result; +} + +static int parse_ipv6(grpc_uri *uri, struct sockaddr_storage *addr, int *len) { + const char *host_port = uri->path; + char *host; + char *port; + int port_num; + int result = 0; + struct sockaddr_in6 *in6 = (struct sockaddr_in6 *)addr; + + if (*host_port == '/') ++host_port; + if (!gpr_split_host_port(host_port, &host, &port)) { + return 0; + } + + memset(in6, 0, sizeof(*in6)); + *len = sizeof(*in6); + in6->sin6_family = AF_INET6; + if (inet_pton(AF_INET6, host, &in6->sin6_addr) == 0) { + gpr_log(GPR_ERROR, "invalid ipv6 address: '%s'", host); + goto done; + } + + if (port != NULL) { + if (sscanf(port, "%d", &port_num) != 1 || port_num < 0 || + port_num > 65535) { + gpr_log(GPR_ERROR, "invalid ipv6 port: '%s'", port); + goto done; + } + in6->sin6_port = htons(port_num); + } else { + gpr_log(GPR_ERROR, "no port given for ipv6 scheme"); + goto done; + } + + result = 1; +done: + gpr_free(host); + gpr_free(port); + return result; +} + +static grpc_resolver *sockaddr_create( + grpc_uri *uri, + grpc_lb_policy *(*lb_policy_factory)(grpc_subchannel **subchannels, + size_t num_subchannels), + grpc_subchannel_factory *subchannel_factory, + int parse(grpc_uri *uri, struct sockaddr_storage *dst, int *len)) { + sockaddr_resolver *r; + + if (0 != strcmp(uri->authority, "")) { + gpr_log(GPR_ERROR, "authority based uri's not supported"); + return NULL; + } + + r = gpr_malloc(sizeof(sockaddr_resolver)); + memset(r, 0, sizeof(*r)); + if (!parse(uri, &r->addr, &r->addr_len)) { + gpr_free(r); + return NULL; + } + + gpr_ref_init(&r->refs, 1); + gpr_mu_init(&r->mu); + grpc_resolver_init(&r->base, &sockaddr_resolver_vtable); + r->subchannel_factory = subchannel_factory; + r->lb_policy_factory = lb_policy_factory; + + grpc_subchannel_factory_ref(subchannel_factory); + return &r->base; +} + +/* + * FACTORY + */ + +static void sockaddr_factory_ref(grpc_resolver_factory *factory) {} + +static void sockaddr_factory_unref(grpc_resolver_factory *factory) {} + +#define DECL_FACTORY(name) \ + static grpc_resolver *name##_factory_create_resolver( \ + grpc_resolver_factory *factory, grpc_uri *uri, \ + grpc_subchannel_factory *subchannel_factory) { \ + return sockaddr_create(uri, grpc_create_pick_first_lb_policy, \ + subchannel_factory, parse_##name); \ + } \ + static const grpc_resolver_factory_vtable name##_factory_vtable = { \ + sockaddr_factory_ref, sockaddr_factory_unref, \ + name##_factory_create_resolver}; \ + static grpc_resolver_factory name##_resolver_factory = { \ + &name##_factory_vtable}; \ + grpc_resolver_factory *grpc_##name##_resolver_factory_create() { \ + return &name##_resolver_factory; \ + } + +#ifdef GPR_POSIX_SOCKET +DECL_FACTORY(unix) +#endif +DECL_FACTORY(ipv4) +DECL_FACTORY(ipv6) diff --git a/src/core/client_config/resolvers/unix_resolver_posix.h b/src/core/client_config/resolvers/sockaddr_resolver.h index 57ace59e21..1b7a18f9c2 100644 --- a/src/core/client_config/resolvers/unix_resolver_posix.h +++ b/src/core/client_config/resolvers/sockaddr_resolver.h @@ -38,7 +38,13 @@ #include "src/core/client_config/resolver_factory.h" +grpc_resolver_factory *grpc_ipv4_resolver_factory_create(void); + +grpc_resolver_factory *grpc_ipv6_resolver_factory_create(void); + +#ifdef GPR_POSIX_SOCKET /** Create a unix resolver factory */ grpc_resolver_factory *grpc_unix_resolver_factory_create(void); +#endif #endif /* GRPC_INTERNAL_CORE_CLIENT_CONFIG_RESOLVERS_UNIX_RESOLVER_H */ diff --git a/src/core/client_config/resolvers/unix_resolver_posix.c b/src/core/client_config/resolvers/unix_resolver_posix.c deleted file mode 100644 index be515d2689..0000000000 --- a/src/core/client_config/resolvers/unix_resolver_posix.c +++ /dev/null @@ -1,195 +0,0 @@ -/* - * - * Copyright 2015, Google Inc. - * All rights reserved. - * - * Redistribution and use in source and binary forms, with or without - * modification, are permitted provided that the following conditions are - * met: - * - * * Redistributions of source code must retain the above copyright - * notice, this list of conditions and the following disclaimer. - * * Redistributions in binary form must reproduce the above - * copyright notice, this list of conditions and the following disclaimer - * in the documentation and/or other materials provided with the - * distribution. - * * Neither the name of Google Inc. nor the names of its - * contributors may be used to endorse or promote products derived from - * this software without specific prior written permission. - * - * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS - * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT - * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR - * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT - * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, - * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT - * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, - * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY - * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT - * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE - * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. - * - */ - -#include <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/subchannel.c b/src/core/client_config/subchannel.c index 35f172683a..358b907e59 100644 --- a/src/core/client_config/subchannel.c +++ b/src/core/client_config/subchannel.c @@ -300,7 +300,7 @@ static void continue_connect(grpc_subchannel *c) { } static void start_connect(grpc_subchannel *c) { - gpr_timespec now = gpr_now(GPR_CLOCK_REALTIME); + gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC); c->next_attempt = now; c->backoff_delta = gpr_time_from_seconds(1, GPR_TIMESPAN); @@ -585,7 +585,7 @@ static void subchannel_connected(void *arg, int iomgr_success) { 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_CLOCK_REALTIME)); + grpc_alarm_init(&c->alarm, c->next_attempt, on_alarm, c, gpr_now(GPR_CLOCK_MONOTONIC)); gpr_mu_unlock(&c->mu); } } @@ -640,6 +640,12 @@ void grpc_subchannel_call_unref( } } +char *grpc_subchannel_call_get_peer(grpc_subchannel_call *call) { + grpc_call_stack *call_stack = SUBCHANNEL_CALL_TO_CALL_STACK(call); + grpc_call_element *top_elem = grpc_call_stack_element(call_stack, 0); + return top_elem->filter->get_peer(top_elem); +} + 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); diff --git a/src/core/client_config/subchannel.h b/src/core/client_config/subchannel.h index a23a623277..d1cd33b2af 100644 --- a/src/core/client_config/subchannel.h +++ b/src/core/client_config/subchannel.h @@ -100,6 +100,9 @@ void grpc_subchannel_del_interested_party(grpc_subchannel *channel, void grpc_subchannel_call_process_op(grpc_subchannel_call *subchannel_call, grpc_transport_stream_op *op); +/** continue querying for peer */ +char *grpc_subchannel_call_get_peer(grpc_subchannel_call *subchannel_call); + struct grpc_subchannel_args { /** Channel filters for this channel - wrapped factories will likely want to mutate this */ diff --git a/src/core/iomgr/alarm.c b/src/core/iomgr/alarm.c index 5b9a37e5dd..931f746f75 100644 --- a/src/core/iomgr/alarm.c +++ b/src/core/iomgr/alarm.c @@ -36,6 +36,7 @@ #include "src/core/iomgr/alarm_heap.h" #include "src/core/iomgr/alarm_internal.h" #include "src/core/iomgr/time_averaged_stats.h" +#include <grpc/support/log.h> #include <grpc/support/sync.h> #include <grpc/support/useful.h> @@ -67,6 +68,7 @@ typedef struct { static gpr_mu g_mu; /* Allow only one run_some_expired_alarms at once */ static gpr_mu g_checker_mu; +static gpr_clock_type g_clock_type; static shard_type g_shards[NUM_SHARDS]; /* Protected by g_mu */ static shard_type *g_shard_queue[NUM_SHARDS]; @@ -85,6 +87,7 @@ void grpc_alarm_list_init(gpr_timespec now) { gpr_mu_init(&g_mu); gpr_mu_init(&g_checker_mu); + g_clock_type = now.clock_type; for (i = 0; i < NUM_SHARDS; i++) { shard_type *shard = &g_shards[i]; @@ -102,7 +105,7 @@ void grpc_alarm_list_init(gpr_timespec now) { void grpc_alarm_list_shutdown(void) { int i; - while (run_some_expired_alarms(NULL, gpr_inf_future(GPR_CLOCK_REALTIME), NULL, + while (run_some_expired_alarms(NULL, gpr_inf_future(g_clock_type), NULL, 0)) ; for (i = 0; i < NUM_SHARDS; i++) { @@ -175,6 +178,8 @@ void grpc_alarm_init(grpc_alarm *alarm, gpr_timespec deadline, gpr_timespec now) { int is_first_alarm = 0; shard_type *shard = &g_shards[shard_idx(alarm)]; + GPR_ASSERT(deadline.clock_type == g_clock_type); + GPR_ASSERT(now.clock_type == g_clock_type); alarm->cb = alarm_cb; alarm->cb_arg = alarm_cb_arg; alarm->deadline = deadline; @@ -355,6 +360,7 @@ static int run_some_expired_alarms(gpr_mu *drop_mu, gpr_timespec now, } int grpc_alarm_check(gpr_mu *drop_mu, gpr_timespec now, gpr_timespec *next) { + GPR_ASSERT(now.clock_type == g_clock_type); return run_some_expired_alarms(drop_mu, now, next, 1); } diff --git a/src/core/iomgr/endpoint.c b/src/core/iomgr/endpoint.c index 96487958a7..f2b44aad03 100644 --- a/src/core/iomgr/endpoint.c +++ b/src/core/iomgr/endpoint.c @@ -53,3 +53,7 @@ void grpc_endpoint_add_to_pollset(grpc_endpoint *ep, grpc_pollset *pollset) { void grpc_endpoint_shutdown(grpc_endpoint *ep) { ep->vtable->shutdown(ep); } void grpc_endpoint_destroy(grpc_endpoint *ep) { ep->vtable->destroy(ep); } + +char *grpc_endpoint_get_peer(grpc_endpoint *ep) { + return ep->vtable->get_peer(ep); +} diff --git a/src/core/iomgr/endpoint.h b/src/core/iomgr/endpoint.h index 881e851800..ee0becf3d6 100644 --- a/src/core/iomgr/endpoint.h +++ b/src/core/iomgr/endpoint.h @@ -72,12 +72,15 @@ struct grpc_endpoint_vtable { void (*add_to_pollset)(grpc_endpoint *ep, grpc_pollset *pollset); void (*shutdown)(grpc_endpoint *ep); void (*destroy)(grpc_endpoint *ep); + char *(*get_peer)(grpc_endpoint *ep); }; /* When data is available on the connection, calls the callback with slices. */ void grpc_endpoint_notify_on_read(grpc_endpoint *ep, grpc_endpoint_read_cb cb, void *user_data); +char *grpc_endpoint_get_peer(grpc_endpoint *ep); + /* Write slices out to the socket. If the connection is ready for more data after the end of the call, it diff --git a/src/core/iomgr/endpoint_pair_posix.c b/src/core/iomgr/endpoint_pair_posix.c index fa2d2555d6..deae9c6875 100644 --- a/src/core/iomgr/endpoint_pair_posix.c +++ b/src/core/iomgr/endpoint_pair_posix.c @@ -66,12 +66,12 @@ grpc_endpoint_pair grpc_iomgr_create_endpoint_pair(const char *name, create_sockets(sv); gpr_asprintf(&final_name, "%s:client", name); - p.client = - grpc_tcp_create(grpc_fd_create(sv[1], final_name), read_slice_size); + p.client = grpc_tcp_create(grpc_fd_create(sv[1], final_name), read_slice_size, + "socketpair-server"); gpr_free(final_name); gpr_asprintf(&final_name, "%s:server", name); - p.server = - grpc_tcp_create(grpc_fd_create(sv[0], final_name), read_slice_size); + p.server = grpc_tcp_create(grpc_fd_create(sv[0], final_name), read_slice_size, + "socketpair-client"); gpr_free(final_name); return p; } diff --git a/src/core/iomgr/endpoint_pair_windows.c b/src/core/iomgr/endpoint_pair_windows.c index c6790b2937..7d81470a78 100644 --- a/src/core/iomgr/endpoint_pair_windows.c +++ b/src/core/iomgr/endpoint_pair_windows.c @@ -81,8 +81,8 @@ grpc_endpoint_pair grpc_iomgr_create_endpoint_pair(const char *name, size_t read SOCKET sv[2]; grpc_endpoint_pair p; create_sockets(sv); - p.client = grpc_tcp_create(grpc_winsocket_create(sv[1], "endpoint:client")); - p.server = grpc_tcp_create(grpc_winsocket_create(sv[0], "endpoint:server")); + p.client = grpc_tcp_create(grpc_winsocket_create(sv[1], "endpoint:client"), "endpoint:server"); + p.server = grpc_tcp_create(grpc_winsocket_create(sv[0], "endpoint:server"), "endpoint:client"); return p; } diff --git a/src/core/iomgr/iomgr.c b/src/core/iomgr/iomgr.c index 0244f689b1..a18c176b30 100644 --- a/src/core/iomgr/iomgr.c +++ b/src/core/iomgr/iomgr.c @@ -57,9 +57,9 @@ static grpc_iomgr_object g_root_object; static void background_callback_executor(void *ignored) { gpr_mu_lock(&g_mu); while (!g_shutdown) { - gpr_timespec deadline = gpr_inf_future(GPR_CLOCK_REALTIME); + gpr_timespec deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC); gpr_timespec short_deadline = gpr_time_add( - gpr_now(GPR_CLOCK_REALTIME), gpr_time_from_millis(100, GPR_TIMESPAN)); + gpr_now(GPR_CLOCK_MONOTONIC), gpr_time_from_millis(100, GPR_TIMESPAN)); if (g_cbs_head) { grpc_iomgr_closure *closure = g_cbs_head; g_cbs_head = closure->next; @@ -67,7 +67,7 @@ static void background_callback_executor(void *ignored) { gpr_mu_unlock(&g_mu); closure->cb(closure->cb_arg, closure->success); gpr_mu_lock(&g_mu); - } else if (grpc_alarm_check(&g_mu, gpr_now(GPR_CLOCK_REALTIME), + } else if (grpc_alarm_check(&g_mu, gpr_now(GPR_CLOCK_MONOTONIC), &deadline)) { } else { gpr_mu_unlock(&g_mu); @@ -90,7 +90,7 @@ void grpc_iomgr_init(void) { gpr_thd_id id; gpr_mu_init(&g_mu); gpr_cv_init(&g_rcv); - grpc_alarm_list_init(gpr_now(GPR_CLOCK_REALTIME)); + grpc_alarm_list_init(gpr_now(GPR_CLOCK_MONOTONIC)); g_root_object.next = g_root_object.prev = &g_root_object; g_root_object.name = "root"; grpc_iomgr_platform_init(); @@ -145,7 +145,7 @@ void grpc_iomgr_shutdown(void) { } while (g_cbs_head); continue; } - if (grpc_alarm_check(&g_mu, gpr_inf_future(GPR_CLOCK_REALTIME), NULL)) { + if (grpc_alarm_check(&g_mu, gpr_inf_future(GPR_CLOCK_MONOTONIC), NULL)) { gpr_log(GPR_DEBUG, "got late alarm"); continue; } diff --git a/src/core/iomgr/pollset_posix.c b/src/core/iomgr/pollset_posix.c index efb301d81c..c8646af615 100644 --- a/src/core/iomgr/pollset_posix.c +++ b/src/core/iomgr/pollset_posix.c @@ -136,7 +136,7 @@ static void finish_shutdown(grpc_pollset *pollset) { int grpc_pollset_work(grpc_pollset *pollset, gpr_timespec deadline) { /* pollset->mu already held */ - gpr_timespec now = gpr_now(GPR_CLOCK_REALTIME); + gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC); if (gpr_time_cmp(now, deadline) > 0) { return 0; } @@ -205,7 +205,7 @@ int grpc_poll_deadline_to_millis_timeout(gpr_timespec deadline, gpr_timespec now) { gpr_timespec timeout; static const int max_spin_polling_us = 10; - if (gpr_time_cmp(deadline, gpr_inf_future(GPR_CLOCK_REALTIME)) == 0) { + if (gpr_time_cmp(deadline, gpr_inf_future(deadline.clock_type)) == 0) { return -1; } if (gpr_time_cmp(deadline, gpr_time_add(now, gpr_time_from_micros( diff --git a/src/core/iomgr/pollset_windows.c b/src/core/iomgr/pollset_windows.c index 24226cc980..a9c4739c7c 100644 --- a/src/core/iomgr/pollset_windows.c +++ b/src/core/iomgr/pollset_windows.c @@ -70,7 +70,7 @@ void grpc_pollset_destroy(grpc_pollset *pollset) { int grpc_pollset_work(grpc_pollset *pollset, gpr_timespec deadline) { gpr_timespec now; - now = gpr_now(GPR_CLOCK_REALTIME); + now = gpr_now(GPR_CLOCK_MONOTONIC); if (gpr_time_cmp(now, deadline) > 0) { return 0 /* GPR_FALSE */; } diff --git a/src/core/iomgr/sockaddr_utils.c b/src/core/iomgr/sockaddr_utils.c index e91b94f8c8..71ac12e87b 100644 --- a/src/core/iomgr/sockaddr_utils.c +++ b/src/core/iomgr/sockaddr_utils.c @@ -36,12 +36,18 @@ #include <errno.h> #include <string.h> -#include "src/core/support/string.h" +#ifdef GPR_POSIX_SOCKET +#include <sys/un.h> +#endif + +#include <grpc/support/alloc.h> #include <grpc/support/host_port.h> #include <grpc/support/log.h> #include <grpc/support/port_platform.h> #include <grpc/support/string_util.h> +#include "src/core/support/string.h" + static const gpr_uint8 kV4MappedPrefix[] = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff}; @@ -161,6 +167,31 @@ int grpc_sockaddr_to_string(char **out, const struct sockaddr *addr, return ret; } +char *grpc_sockaddr_to_uri(const struct sockaddr *addr) { + char *temp; + char *result; + + switch (addr->sa_family) { + case AF_INET: + grpc_sockaddr_to_string(&temp, addr, 0); + gpr_asprintf(&result, "ipv4:%s", temp); + gpr_free(temp); + return result; + case AF_INET6: + grpc_sockaddr_to_string(&temp, addr, 0); + gpr_asprintf(&result, "ipv6:%s", temp); + gpr_free(temp); + return result; +#ifdef GPR_POSIX_SOCKET + case AF_UNIX: + gpr_asprintf(&result, "unix:%s", ((struct sockaddr_un *)addr)->sun_path); + return result; +#endif + } + + return NULL; +} + int grpc_sockaddr_get_port(const struct sockaddr *addr) { switch (addr->sa_family) { case AF_INET: diff --git a/src/core/iomgr/sockaddr_utils.h b/src/core/iomgr/sockaddr_utils.h index bdfb83479b..99f1ed54da 100644 --- a/src/core/iomgr/sockaddr_utils.h +++ b/src/core/iomgr/sockaddr_utils.h @@ -84,4 +84,6 @@ int grpc_sockaddr_set_port(const struct sockaddr *addr, int port); int grpc_sockaddr_to_string(char **out, const struct sockaddr *addr, int normalize); +char *grpc_sockaddr_to_uri(const struct sockaddr *addr); + #endif /* GRPC_INTERNAL_CORE_IOMGR_SOCKADDR_UTILS_H */ diff --git a/src/core/iomgr/tcp_client_posix.c b/src/core/iomgr/tcp_client_posix.c index dc0489e64f..392eda999e 100644 --- a/src/core/iomgr/tcp_client_posix.c +++ b/src/core/iomgr/tcp_client_posix.c @@ -64,6 +64,7 @@ typedef struct { int refs; grpc_iomgr_closure write_closure; grpc_pollset_set *interested_parties; + char *addr_str; } async_connect; static int prepare_socket(const struct sockaddr *addr, int fd) { @@ -99,6 +100,7 @@ static void on_alarm(void *acp, int success) { gpr_mu_unlock(&ac->mu); if (done) { gpr_mu_destroy(&ac->mu); + gpr_free(ac->addr_str); gpr_free(ac); } } @@ -114,6 +116,8 @@ static void on_writable(void *acp, int success) { void (*cb)(void *arg, grpc_endpoint *tcp) = ac->cb; void *cb_arg = ac->cb_arg; + grpc_alarm_cancel(&ac->alarm); + gpr_mu_lock(&ac->mu); if (success) { do { @@ -156,7 +160,8 @@ static void on_writable(void *acp, int success) { } } else { grpc_pollset_set_del_fd(ac->interested_parties, ac->fd); - ep = grpc_tcp_create(ac->fd, GRPC_TCP_DEFAULT_READ_SLICE_SIZE); + ep = grpc_tcp_create(ac->fd, GRPC_TCP_DEFAULT_READ_SLICE_SIZE, + ac->addr_str); goto finish; } } else { @@ -177,9 +182,8 @@ finish: gpr_mu_unlock(&ac->mu); if (done) { gpr_mu_destroy(&ac->mu); + gpr_free(ac->addr_str); gpr_free(ac); - } else { - grpc_alarm_cancel(&ac->alarm); } cb(cb_arg, ep); } @@ -223,13 +227,13 @@ void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *ep), err = connect(fd, addr, addr_len); } while (err < 0 && errno == EINTR); - grpc_sockaddr_to_string(&addr_str, addr, 1); + addr_str = grpc_sockaddr_to_uri(addr); gpr_asprintf(&name, "tcp-client:%s", addr_str); fdobj = grpc_fd_create(fd, name); if (err >= 0) { - cb(arg, grpc_tcp_create(fdobj, GRPC_TCP_DEFAULT_READ_SLICE_SIZE)); + cb(arg, grpc_tcp_create(fdobj, GRPC_TCP_DEFAULT_READ_SLICE_SIZE, addr_str)); goto done; } @@ -247,14 +251,16 @@ void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *ep), ac->cb_arg = arg; ac->fd = fdobj; ac->interested_parties = interested_parties; + ac->addr_str = addr_str; + addr_str = NULL; gpr_mu_init(&ac->mu); ac->refs = 2; ac->write_closure.cb = on_writable; ac->write_closure.cb_arg = ac; gpr_mu_lock(&ac->mu); - grpc_alarm_init(&ac->alarm, deadline, on_alarm, ac, - gpr_now(GPR_CLOCK_REALTIME)); + grpc_alarm_init(&ac->alarm, gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC), + on_alarm, ac, gpr_now(GPR_CLOCK_MONOTONIC)); grpc_fd_notify_on_write(ac->fd, &ac->write_closure); gpr_mu_unlock(&ac->mu); diff --git a/src/core/iomgr/tcp_client_windows.c b/src/core/iomgr/tcp_client_windows.c index 16741452b9..79a58fe2af 100644 --- a/src/core/iomgr/tcp_client_windows.c +++ b/src/core/iomgr/tcp_client_windows.c @@ -58,6 +58,7 @@ typedef struct { grpc_winsocket *socket; gpr_timespec deadline; grpc_alarm alarm; + char *addr_name; int refs; int aborted; } async_connect; @@ -67,6 +68,7 @@ static void async_connect_cleanup(async_connect *ac) { gpr_mu_unlock(&ac->mu); if (done) { gpr_mu_destroy(&ac->mu); + gpr_free(ac->addr_name); gpr_free(ac); } } @@ -107,7 +109,7 @@ static void on_connect(void *acp, int from_iocp) { gpr_log(GPR_ERROR, "on_connect error: %s", utf8_message); gpr_free(utf8_message); } else if (!aborted) { - ep = grpc_tcp_create(ac->socket); + ep = grpc_tcp_create(ac->socket, ac->addr_name); } } else { gpr_log(GPR_ERROR, "on_connect is shutting down"); @@ -213,10 +215,11 @@ void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *tcp), ac->socket = socket; gpr_mu_init(&ac->mu); ac->refs = 2; + ac->addr_name = grpc_sockaddr_to_uri(addr); ac->aborted = 0; grpc_alarm_init(&ac->alarm, deadline, on_alarm, ac, - gpr_now(GPR_CLOCK_REALTIME)); + gpr_now(GPR_CLOCK_MONOTONIC)); socket->write_info.outstanding = 1; grpc_socket_notify_on_write(socket, on_connect, ac); return; diff --git a/src/core/iomgr/tcp_posix.c b/src/core/iomgr/tcp_posix.c index b6d6efc9fb..1e8432d463 100644 --- a/src/core/iomgr/tcp_posix.c +++ b/src/core/iomgr/tcp_posix.c @@ -44,15 +44,17 @@ #include <sys/socket.h> #include <unistd.h> -#include "src/core/support/string.h" -#include "src/core/debug/trace.h" -#include "src/core/profiling/timers.h" #include <grpc/support/alloc.h> #include <grpc/support/log.h> #include <grpc/support/slice.h> +#include <grpc/support/string_util.h> #include <grpc/support/sync.h> #include <grpc/support/time.h> +#include "src/core/support/string.h" +#include "src/core/debug/trace.h" +#include "src/core/profiling/timers.h" + #ifdef GPR_HAVE_MSG_NOSIGNAL #define SENDMSG_FLAGS MSG_NOSIGNAL #else @@ -282,6 +284,8 @@ typedef struct { grpc_iomgr_closure write_closure; grpc_iomgr_closure handle_read_closure; + + char *peer_string; } grpc_tcp; static void grpc_tcp_handle_read(void *arg /* grpc_tcp */, int success); @@ -296,6 +300,7 @@ static void grpc_tcp_unref(grpc_tcp *tcp) { int refcount_zero = gpr_unref(&tcp->refcount); if (refcount_zero) { grpc_fd_orphan(tcp->em_fd, NULL, "tcp_unref_orphan"); + gpr_free(tcp->peer_string); gpr_free(tcp); } } @@ -567,13 +572,20 @@ static void grpc_tcp_add_to_pollset(grpc_endpoint *ep, grpc_pollset *pollset) { grpc_pollset_add_fd(pollset, tcp->em_fd); } +static char *grpc_tcp_get_peer(grpc_endpoint *ep) { + grpc_tcp *tcp = (grpc_tcp *)ep; + return gpr_strdup(tcp->peer_string); +} + static const grpc_endpoint_vtable vtable = { - grpc_tcp_notify_on_read, grpc_tcp_write, grpc_tcp_add_to_pollset, - grpc_tcp_shutdown, grpc_tcp_destroy}; + grpc_tcp_notify_on_read, grpc_tcp_write, grpc_tcp_add_to_pollset, + grpc_tcp_shutdown, grpc_tcp_destroy, grpc_tcp_get_peer}; -grpc_endpoint *grpc_tcp_create(grpc_fd *em_fd, size_t slice_size) { +grpc_endpoint *grpc_tcp_create(grpc_fd *em_fd, size_t slice_size, + const char *peer_string) { grpc_tcp *tcp = (grpc_tcp *)gpr_malloc(sizeof(grpc_tcp)); tcp->base.vtable = &vtable; + tcp->peer_string = gpr_strdup(peer_string); tcp->fd = em_fd->fd; tcp->read_cb = NULL; tcp->write_cb = NULL; diff --git a/src/core/iomgr/tcp_posix.h b/src/core/iomgr/tcp_posix.h index 44279d5a26..d752feaeea 100644 --- a/src/core/iomgr/tcp_posix.h +++ b/src/core/iomgr/tcp_posix.h @@ -53,6 +53,7 @@ extern int grpc_tcp_trace; /* Create a tcp endpoint given a file desciptor and a read slice size. Takes ownership of fd. */ -grpc_endpoint *grpc_tcp_create(grpc_fd *fd, size_t read_slice_size); +grpc_endpoint *grpc_tcp_create(grpc_fd *fd, size_t read_slice_size, + const char *peer_string); #endif /* GRPC_INTERNAL_CORE_IOMGR_TCP_POSIX_H */ diff --git a/src/core/iomgr/tcp_server_posix.c b/src/core/iomgr/tcp_server_posix.c index 5854031c9b..8538600112 100644 --- a/src/core/iomgr/tcp_server_posix.c +++ b/src/core/iomgr/tcp_server_posix.c @@ -332,7 +332,7 @@ static void on_read(void *arg, int success) { grpc_set_socket_no_sigpipe_if_possible(fd); - grpc_sockaddr_to_string(&addr_str, (struct sockaddr *)&addr, 1); + addr_str = grpc_sockaddr_to_uri((struct sockaddr *)&addr); gpr_asprintf(&name, "tcp-server-connection:%s", addr_str); fdobj = grpc_fd_create(fd, name); @@ -342,8 +342,9 @@ static void on_read(void *arg, int success) { for (i = 0; i < sp->server->pollset_count; i++) { grpc_pollset_add_fd(sp->server->pollsets[i], fdobj); } - sp->server->cb(sp->server->cb_arg, - grpc_tcp_create(fdobj, GRPC_TCP_DEFAULT_READ_SLICE_SIZE)); + sp->server->cb( + sp->server->cb_arg, + grpc_tcp_create(fdobj, GRPC_TCP_DEFAULT_READ_SLICE_SIZE, addr_str)); gpr_free(name); gpr_free(addr_str); diff --git a/src/core/iomgr/tcp_server_windows.c b/src/core/iomgr/tcp_server_windows.c index 187009b2c8..8f634fcd7a 100644 --- a/src/core/iomgr/tcp_server_windows.c +++ b/src/core/iomgr/tcp_server_windows.c @@ -243,6 +243,10 @@ static void on_accept(void *arg, int from_iocp) { SOCKET sock = sp->new_socket; grpc_winsocket_callback_info *info = &sp->socket->read_info; grpc_endpoint *ep = NULL; + struct sockaddr_storage peer_name; + char *peer_name_string; + char *fd_name; + int peer_name_len = sizeof(peer_name); DWORD transfered_bytes; DWORD flags; BOOL wsa_success; @@ -277,8 +281,12 @@ static void on_accept(void *arg, int from_iocp) { } } else { if (!sp->shutting_down) { - /* TODO(ctiller): add sockaddr address to label */ - ep = grpc_tcp_create(grpc_winsocket_create(sock, "server")); + getpeername(sock, (struct sockaddr*)&peer_name, &peer_name_len); + peer_name_string = grpc_sockaddr_to_uri((struct sockaddr*)&peer_name); + gpr_asprintf(&fd_name, "tcp_server:%s", peer_name_string); + ep = grpc_tcp_create(grpc_winsocket_create(sock, fd_name), peer_name_string); + gpr_free(fd_name); + gpr_free(peer_name_string); } } diff --git a/src/core/iomgr/tcp_windows.c b/src/core/iomgr/tcp_windows.c index 1bf81a73e0..d68e6aee79 100644 --- a/src/core/iomgr/tcp_windows.c +++ b/src/core/iomgr/tcp_windows.c @@ -96,6 +96,8 @@ typedef struct grpc_tcp { to protect ourselves when requesting a shutdown. */ gpr_mu mu; int shutting_down; + + char *peer_string; } grpc_tcp; static void tcp_ref(grpc_tcp *tcp) { @@ -107,6 +109,7 @@ static void tcp_unref(grpc_tcp *tcp) { gpr_slice_buffer_destroy(&tcp->write_slices); grpc_winsocket_orphan(tcp->socket); gpr_mu_destroy(&tcp->mu); + gpr_free(tcp->peer_string); gpr_free(tcp); } } @@ -393,11 +396,16 @@ static void win_destroy(grpc_endpoint *ep) { tcp_unref(tcp); } +static char *win_get_peer(grpc_endpoint *ep) { + grpc_tcp *tcp = (grpc_tcp *)ep; + return gpr_strdup(tcp->peer_string); +} + static grpc_endpoint_vtable vtable = { - win_notify_on_read, win_write, win_add_to_pollset, win_shutdown, win_destroy + win_notify_on_read, win_write, win_add_to_pollset, win_shutdown, win_destroy, win_get_peer }; -grpc_endpoint *grpc_tcp_create(grpc_winsocket *socket) { +grpc_endpoint *grpc_tcp_create(grpc_winsocket *socket, char *peer_string) { grpc_tcp *tcp = (grpc_tcp *) gpr_malloc(sizeof(grpc_tcp)); memset(tcp, 0, sizeof(grpc_tcp)); tcp->base.vtable = &vtable; @@ -405,6 +413,7 @@ grpc_endpoint *grpc_tcp_create(grpc_winsocket *socket) { gpr_mu_init(&tcp->mu); gpr_slice_buffer_init(&tcp->write_slices); gpr_ref_init(&tcp->refcount, 1); + tcp->peer_string = gpr_strdup(peer_string); return &tcp->base; } diff --git a/src/core/iomgr/tcp_windows.h b/src/core/iomgr/tcp_windows.h index 4cbc12c53a..7e301db250 100644 --- a/src/core/iomgr/tcp_windows.h +++ b/src/core/iomgr/tcp_windows.h @@ -50,7 +50,7 @@ /* Create a tcp endpoint given a winsock handle. * Takes ownership of the handle. */ -grpc_endpoint *grpc_tcp_create(grpc_winsocket *socket); +grpc_endpoint *grpc_tcp_create(grpc_winsocket *socket, char *peer_string); int grpc_tcp_prepare_socket(SOCKET sock); diff --git a/src/core/security/client_auth_filter.c b/src/core/security/client_auth_filter.c index 9e49a807f1..9a69f53a5a 100644 --- a/src/core/security/client_auth_filter.c +++ b/src/core/security/client_auth_filter.c @@ -344,6 +344,8 @@ static void destroy_channel_elem(grpc_channel_element *elem) { } const grpc_channel_filter grpc_client_auth_filter = { - auth_start_transport_op, grpc_channel_next_op, sizeof(call_data), - init_call_elem, destroy_call_elem, sizeof(channel_data), - init_channel_elem, destroy_channel_elem, "client-auth"}; + auth_start_transport_op, grpc_channel_next_op, + sizeof(call_data), init_call_elem, + destroy_call_elem, sizeof(channel_data), + init_channel_elem, destroy_channel_elem, + grpc_call_next_get_peer, "client-auth"}; diff --git a/src/core/security/credentials.c b/src/core/security/credentials.c index fb59fa4b0e..38612cf308 100644 --- a/src/core/security/credentials.c +++ b/src/core/security/credentials.c @@ -315,7 +315,7 @@ grpc_server_credentials *grpc_ssl_server_credentials_create( /* -- Jwt credentials -- */ -static void jwt_reset_cache(grpc_jwt_credentials *c) { +static void jwt_reset_cache(grpc_service_account_jwt_access_credentials *c) { if (c->cached.jwt_md != NULL) { grpc_credentials_md_store_unref(c->cached.jwt_md); c->cached.jwt_md = NULL; @@ -328,7 +328,8 @@ static void jwt_reset_cache(grpc_jwt_credentials *c) { } static void jwt_destroy(grpc_credentials *creds) { - grpc_jwt_credentials *c = (grpc_jwt_credentials *)creds; + grpc_service_account_jwt_access_credentials *c = + (grpc_service_account_jwt_access_credentials *)creds; grpc_auth_json_key_destruct(&c->key); jwt_reset_cache(c); gpr_mu_destroy(&c->cache_mu); @@ -346,7 +347,8 @@ static void jwt_get_request_metadata(grpc_credentials *creds, const char *service_url, grpc_credentials_metadata_cb cb, void *user_data) { - grpc_jwt_credentials *c = (grpc_jwt_credentials *)creds; + grpc_service_account_jwt_access_credentials *c = + (grpc_service_account_jwt_access_credentials *)creds; gpr_timespec refresh_threshold = gpr_time_from_seconds( GRPC_SECURE_TOKEN_REFRESH_THRESHOLD_SECS, GPR_TIMESPAN); @@ -399,15 +401,16 @@ static grpc_credentials_vtable jwt_vtable = { jwt_destroy, jwt_has_request_metadata, jwt_has_request_metadata_only, jwt_get_request_metadata, NULL}; -grpc_credentials *grpc_jwt_credentials_create_from_auth_json_key( +grpc_credentials * +grpc_service_account_jwt_access_credentials_create_from_auth_json_key( grpc_auth_json_key key, gpr_timespec token_lifetime) { - grpc_jwt_credentials *c; + grpc_service_account_jwt_access_credentials *c; if (!grpc_auth_json_key_is_valid(&key)) { gpr_log(GPR_ERROR, "Invalid input for jwt credentials creation"); return NULL; } - c = gpr_malloc(sizeof(grpc_jwt_credentials)); - memset(c, 0, sizeof(grpc_jwt_credentials)); + c = gpr_malloc(sizeof(grpc_service_account_jwt_access_credentials)); + memset(c, 0, sizeof(grpc_service_account_jwt_access_credentials)); c->base.type = GRPC_CREDENTIALS_TYPE_JWT; gpr_ref_init(&c->base.refcount, 1); c->base.vtable = &jwt_vtable; @@ -418,9 +421,9 @@ grpc_credentials *grpc_jwt_credentials_create_from_auth_json_key( return &c->base; } -grpc_credentials *grpc_jwt_credentials_create(const char *json_key, - gpr_timespec token_lifetime) { - return grpc_jwt_credentials_create_from_auth_json_key( +grpc_credentials *grpc_service_account_jwt_access_credentials_create( + const char *json_key, gpr_timespec token_lifetime) { + return grpc_service_account_jwt_access_credentials_create_from_auth_json_key( grpc_auth_json_key_create_from_string(json_key), token_lifetime); } diff --git a/src/core/security/credentials.h b/src/core/security/credentials.h index d988901cf7..7f4141967d 100644 --- a/src/core/security/credentials.h +++ b/src/core/security/credentials.h @@ -188,7 +188,8 @@ grpc_credentials *grpc_fake_oauth2_credentials_create( /* Private constructor for jwt credentials from an already parsed json key. Takes ownership of the key. */ -grpc_credentials *grpc_jwt_credentials_create_from_auth_json_key( +grpc_credentials * +grpc_service_account_jwt_access_credentials_create_from_auth_json_key( grpc_auth_json_key key, gpr_timespec token_lifetime); /* Private constructor for refresh token credentials from an already parsed @@ -240,7 +241,7 @@ typedef struct { grpc_auth_json_key key; gpr_timespec jwt_lifetime; -} grpc_jwt_credentials; +} grpc_service_account_jwt_access_credentials; /* -- Oauth2TokenFetcher credentials -- diff --git a/src/core/security/google_default_credentials.c b/src/core/security/google_default_credentials.c index 833484310f..de1929fe76 100644 --- a/src/core/security/google_default_credentials.c +++ b/src/core/security/google_default_credentials.c @@ -140,8 +140,9 @@ static grpc_credentials *create_default_creds_from_path(char *creds_path) { /* First, try an auth json key. */ key = grpc_auth_json_key_create_from_json(json); if (grpc_auth_json_key_is_valid(&key)) { - result = grpc_jwt_credentials_create_from_auth_json_key( - key, grpc_max_auth_token_lifetime); + result = + grpc_service_account_jwt_access_credentials_create_from_auth_json_key( + key, grpc_max_auth_token_lifetime); goto end; } diff --git a/src/core/security/secure_endpoint.c b/src/core/security/secure_endpoint.c index 3548198046..e189380ec5 100644 --- a/src/core/security/secure_endpoint.c +++ b/src/core/security/secure_endpoint.c @@ -331,9 +331,14 @@ static void endpoint_add_to_pollset(grpc_endpoint *secure_ep, grpc_endpoint_add_to_pollset(ep->wrapped_ep, pollset); } +static char *endpoint_get_peer(grpc_endpoint *secure_ep) { + secure_endpoint *ep = (secure_endpoint *)secure_ep; + return grpc_endpoint_get_peer(ep->wrapped_ep); +} + static const grpc_endpoint_vtable vtable = { endpoint_notify_on_read, endpoint_write, endpoint_add_to_pollset, - endpoint_shutdown, endpoint_unref}; + endpoint_shutdown, endpoint_unref, endpoint_get_peer}; grpc_endpoint *grpc_secure_endpoint_create( struct tsi_frame_protector *protector, grpc_endpoint *transport, diff --git a/src/core/security/server_auth_filter.c b/src/core/security/server_auth_filter.c index 10eef6d237..69789c2f0d 100644 --- a/src/core/security/server_auth_filter.c +++ b/src/core/security/server_auth_filter.c @@ -120,6 +120,8 @@ static void destroy_channel_elem(grpc_channel_element *elem) { } const grpc_channel_filter grpc_server_auth_filter = { - auth_start_transport_op, grpc_channel_next_op, sizeof(call_data), - init_call_elem, destroy_call_elem, sizeof(channel_data), - init_channel_elem, destroy_channel_elem, "server-auth"}; + auth_start_transport_op, grpc_channel_next_op, + sizeof(call_data), init_call_elem, + destroy_call_elem, sizeof(channel_data), + init_channel_elem, destroy_channel_elem, + grpc_call_next_get_peer, "server-auth"}; diff --git a/src/core/support/host_port.c b/src/core/support/host_port.c index 0906ebc2a3..a28f04df9c 100644 --- a/src/core/support/host_port.c +++ b/src/core/support/host_port.c @@ -50,7 +50,7 @@ int gpr_join_host_port(char **out, const char *host, int port) { } } -void gpr_split_host_port(const char *name, char **host, char **port) { +int gpr_split_host_port(const char *name, char **host, char **port) { const char *host_start; size_t host_len; const char *port_start; @@ -63,7 +63,7 @@ void gpr_split_host_port(const char *name, char **host, char **port) { const char *rbracket = strchr(name, ']'); if (rbracket == NULL) { /* Unmatched [ */ - return; + return 0; } if (rbracket[1] == '\0') { /* ]<end> */ @@ -73,14 +73,14 @@ void gpr_split_host_port(const char *name, char **host, char **port) { port_start = rbracket + 2; } else { /* ]<invalid> */ - return; + return 0; } host_start = name + 1; host_len = (size_t)(rbracket - host_start); if (memchr(host_start, ':', host_len) == NULL) { /* Require all bracketed hosts to contain a colon, because a hostname or IPv4 address should never use brackets. */ - return; + return 0; } } else { const char *colon = strchr(name, ':'); @@ -105,4 +105,6 @@ void gpr_split_host_port(const char *name, char **host, char **port) { if (port_start != NULL) { *port = gpr_strdup(port_start); } + + return 1; } diff --git a/src/core/support/sync_posix.c b/src/core/support/sync_posix.c index 41af8ceb0a..61572b9a8e 100644 --- a/src/core/support/sync_posix.c +++ b/src/core/support/sync_posix.c @@ -63,10 +63,11 @@ void gpr_cv_destroy(gpr_cv *cv) { GPR_ASSERT(pthread_cond_destroy(cv) == 0); } int gpr_cv_wait(gpr_cv *cv, gpr_mu *mu, gpr_timespec abs_deadline) { int err = 0; - if (gpr_time_cmp(abs_deadline, gpr_inf_future(GPR_CLOCK_REALTIME)) == 0) { + if (gpr_time_cmp(abs_deadline, gpr_inf_future(abs_deadline.clock_type)) == 0) { err = pthread_cond_wait(cv, mu); } else { struct timespec abs_deadline_ts; + abs_deadline = gpr_convert_clock_type(abs_deadline, GPR_CLOCK_REALTIME); abs_deadline_ts.tv_sec = abs_deadline.tv_sec; abs_deadline_ts.tv_nsec = abs_deadline.tv_nsec; err = pthread_cond_timedwait(cv, mu, &abs_deadline_ts); diff --git a/src/core/support/sync_win32.c b/src/core/support/sync_win32.c index 63196d10d3..54f84a46ac 100644 --- a/src/core/support/sync_win32.c +++ b/src/core/support/sync_win32.c @@ -83,10 +83,10 @@ int gpr_cv_wait(gpr_cv *cv, gpr_mu *mu, gpr_timespec abs_deadline) { int timeout = 0; DWORD timeout_max_ms; mu->locked = 0; - if (gpr_time_cmp(abs_deadline, gpr_inf_future(GPR_CLOCK_REALTIME)) == 0) { + if (gpr_time_cmp(abs_deadline, gpr_inf_future(abs_deadline.clock_type)) == 0) { SleepConditionVariableCS(cv, &mu->cs, INFINITE); } else { - gpr_timespec now = gpr_now(GPR_CLOCK_REALTIME); + gpr_timespec now = gpr_now(abs_deadline.clock_type); gpr_int64 now_ms = now.tv_sec * 1000 + now.tv_nsec / 1000000; gpr_int64 deadline_ms = abs_deadline.tv_sec * 1000 + abs_deadline.tv_nsec / 1000000; diff --git a/src/core/support/time.c b/src/core/support/time.c index 570f195bd1..b523ae01cc 100644 --- a/src/core/support/time.c +++ b/src/core/support/time.c @@ -290,3 +290,30 @@ gpr_int32 gpr_time_to_millis(gpr_timespec t) { double gpr_timespec_to_micros(gpr_timespec t) { return (double)t.tv_sec * GPR_US_PER_SEC + t.tv_nsec * 1e-3; } + +gpr_timespec gpr_convert_clock_type(gpr_timespec t, gpr_clock_type clock_type) { + if (t.clock_type == clock_type) { + return t; + } + + if (t.tv_nsec == 0) { + if (t.tv_sec == TYPE_MAX(time_t)) { + t.clock_type = clock_type; + return t; + } + if (t.tv_sec == TYPE_MIN(time_t)) { + t.clock_type = clock_type; + return t; + } + } + + if (clock_type == GPR_TIMESPAN) { + return gpr_time_sub(t, gpr_now(t.clock_type)); + } + + if (t.clock_type == GPR_TIMESPAN) { + return gpr_time_add(gpr_now(clock_type), t); + } + + return gpr_time_add(gpr_now(clock_type), gpr_time_sub(t, gpr_now(t.clock_type))); +} diff --git a/src/core/support/time_posix.c b/src/core/support/time_posix.c index 258b2e640e..841485c4b4 100644 --- a/src/core/support/time_posix.c +++ b/src/core/support/time_posix.c @@ -120,7 +120,7 @@ void gpr_sleep_until(gpr_timespec until) { for (;;) { /* We could simplify by using clock_nanosleep instead, but it might be * slightly less portable. */ - now = gpr_now(GPR_CLOCK_REALTIME); + now = gpr_now(until.clock_type); if (gpr_time_cmp(until, now) <= 0) { return; } diff --git a/src/core/support/time_win32.c b/src/core/support/time_win32.c index 238cd07ebc..7f64c80e27 100644 --- a/src/core/support/time_win32.c +++ b/src/core/support/time_win32.c @@ -80,7 +80,7 @@ void gpr_sleep_until(gpr_timespec until) { for (;;) { /* We could simplify by using clock_nanosleep instead, but it might be * slightly less portable. */ - now = gpr_now(GPR_CLOCK_REALTIME); + now = gpr_now(until.clock_type); if (gpr_time_cmp(until, now) <= 0) { return; } diff --git a/src/core/surface/call.c b/src/core/surface/call.c index 6e643b591c..a1da822113 100644 --- a/src/core/surface/call.c +++ b/src/core/surface/call.c @@ -348,7 +348,7 @@ grpc_call *grpc_call_create(grpc_channel *channel, grpc_completion_queue *cq, } grpc_call_stack_init(channel_stack, server_transport_data, initial_op_ptr, CALL_STACK_FROM_CALL(call)); - if (gpr_time_cmp(send_deadline, gpr_inf_future(GPR_CLOCK_REALTIME)) != 0) { + if (gpr_time_cmp(send_deadline, gpr_inf_future(send_deadline.clock_type)) != 0) { set_deadline_alarm(call, send_deadline); } return call; @@ -1253,6 +1253,11 @@ static void execute_op(grpc_call *call, grpc_transport_stream_op *op) { elem->filter->start_transport_stream_op(elem, op); } +char *grpc_call_get_peer(grpc_call *call) { + grpc_call_element *elem = CALL_ELEM_FROM_CALL(call, 0); + return elem->filter->get_peer(elem); +} + grpc_call *grpc_call_from_top_element(grpc_call_element *elem) { return CALL_FROM_TOP_ELEM(elem); } @@ -1278,8 +1283,8 @@ static void set_deadline_alarm(grpc_call *call, gpr_timespec deadline) { } GRPC_CALL_INTERNAL_REF(call, "alarm"); call->have_alarm = 1; - grpc_alarm_init(&call->alarm, deadline, call_alarm, call, - gpr_now(GPR_CLOCK_REALTIME)); + grpc_alarm_init(&call->alarm, gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC), call_alarm, call, + gpr_now(GPR_CLOCK_MONOTONIC)); } /* we offset status by a small amount when storing it into transport metadata @@ -1368,7 +1373,8 @@ static void recv_metadata(grpc_call *call, grpc_metadata_batch *md) { l->md = 0; } } - if (gpr_time_cmp(md->deadline, gpr_inf_future(GPR_CLOCK_REALTIME)) != 0) { + if (gpr_time_cmp(md->deadline, gpr_inf_future(md->deadline.clock_type)) != + 0) { set_deadline_alarm(call, md->deadline); } if (!is_trailing) { diff --git a/src/core/surface/call.h b/src/core/surface/call.h index 3b6f9c942e..265638d519 100644 --- a/src/core/surface/call.h +++ b/src/core/surface/call.h @@ -134,6 +134,10 @@ void grpc_server_log_request_call(char *file, int line, grpc_completion_queue *cq_for_notification, void *tag); +void grpc_server_log_shutdown(char *file, int line, gpr_log_severity severity, + grpc_server *server, grpc_completion_queue *cq, + void *tag); + /* Set a context pointer. No thread safety guarantees are made wrt this value. */ void grpc_call_context_set(grpc_call *call, grpc_context_index elem, @@ -151,6 +155,9 @@ void *grpc_call_context_get(grpc_call *call, grpc_context_index elem); grpc_server_log_request_call(sev, server, call, details, initial_metadata, \ cq_bound_to_call, cq_for_notifications, tag) +#define GRPC_SERVER_LOG_SHUTDOWN(sev, server, cq, tag) \ + if (grpc_trace_batch) grpc_server_log_shutdown(sev, server, cq, tag) + gpr_uint8 grpc_call_is_client(grpc_call *call); #endif /* GRPC_INTERNAL_CORE_SURFACE_CALL_H */ diff --git a/src/core/surface/call_log_batch.c b/src/core/surface/call_log_batch.c index 997046d954..7bf8cafc24 100644 --- a/src/core/surface/call_log_batch.c +++ b/src/core/surface/call_log_batch.c @@ -136,3 +136,11 @@ void grpc_server_log_request_call(char *file, int line, "tag=%p)", server, call, details, initial_metadata, cq_bound_to_call, cq_for_notification, tag); } + +void grpc_server_log_shutdown(char *file, int line, gpr_log_severity severity, + grpc_server *server, grpc_completion_queue *cq, + void *tag) { + gpr_log(file, line, severity, + "grpc_server_shutdown_and_notify(server=%p, cq=%p, tag=%p)", server, + cq, tag); +} diff --git a/src/core/surface/channel.c b/src/core/surface/channel.c index a6438ff512..4052c65cc6 100644 --- a/src/core/surface/channel.c +++ b/src/core/surface/channel.c @@ -36,12 +36,14 @@ #include <stdlib.h> #include <string.h> +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> +#include <grpc/support/string_util.h> + #include "src/core/iomgr/iomgr.h" #include "src/core/support/string.h" #include "src/core/surface/call.h" #include "src/core/surface/init.h" -#include <grpc/support/alloc.h> -#include <grpc/support/log.h> /** Cache grpc-status: X mdelems for X = 0..NUM_CACHED_STATUS_ELEMS. * Avoids needing to take a metadata context lock for sending status @@ -73,6 +75,7 @@ struct grpc_channel { gpr_mu registered_call_mu; registered_call *registered_calls; grpc_iomgr_closure destroy_closure; + char *target; }; #define CHANNEL_STACK_FROM_CHANNEL(c) ((grpc_channel_stack *)((c) + 1)) @@ -85,13 +88,14 @@ struct grpc_channel { #define DEFAULT_MAX_MESSAGE_LENGTH (100 * 1024 * 1024) grpc_channel *grpc_channel_create_from_filters( - const grpc_channel_filter **filters, size_t num_filters, + const char *target, const grpc_channel_filter **filters, size_t num_filters, const grpc_channel_args *args, grpc_mdctx *mdctx, int is_client) { size_t i; size_t size = sizeof(grpc_channel) + grpc_channel_stack_size(filters, num_filters); grpc_channel *channel = gpr_malloc(size); memset(channel, 0, sizeof(*channel)); + channel->target = gpr_strdup(target); GPR_ASSERT(grpc_is_initialized() && "call grpc_init()"); channel->is_client = is_client; /* decremented by grpc_channel_destroy */ @@ -137,6 +141,10 @@ grpc_channel *grpc_channel_create_from_filters( return channel; } +char *grpc_channel_get_target(grpc_channel *channel) { + return gpr_strdup(channel->target); +} + static grpc_call *grpc_channel_create_call_internal( grpc_channel *channel, grpc_completion_queue *cq, grpc_mdelem *path_mdelem, grpc_mdelem *authority_mdelem, gpr_timespec deadline) { @@ -222,6 +230,7 @@ static void destroy_channel(void *p, int ok) { } grpc_mdctx_unref(channel->metadata_context); gpr_mu_destroy(&channel->registered_call_mu); + gpr_free(channel->target); gpr_free(channel); } diff --git a/src/core/surface/channel.h b/src/core/surface/channel.h index 4e03eb4411..9e0646efaa 100644 --- a/src/core/surface/channel.h +++ b/src/core/surface/channel.h @@ -38,7 +38,7 @@ #include "src/core/client_config/subchannel_factory.h" grpc_channel *grpc_channel_create_from_filters( - const grpc_channel_filter **filters, size_t count, + const char *target, const grpc_channel_filter **filters, size_t count, const grpc_channel_args *args, grpc_mdctx *mdctx, int is_client); /** Get a (borrowed) pointer to this channels underlying channel stack */ diff --git a/src/core/surface/channel_create.c b/src/core/surface/channel_create.c index 91c7b35550..778f7108fd 100644 --- a/src/core/surface/channel_create.c +++ b/src/core/surface/channel_create.c @@ -179,7 +179,8 @@ grpc_channel *grpc_channel_create(const char *target, return NULL; } - channel = grpc_channel_create_from_filters(filters, n, args, mdctx, 1); + channel = + grpc_channel_create_from_filters(target, filters, n, args, mdctx, 1); grpc_client_channel_set_resolver(grpc_channel_get_channel_stack(channel), resolver); GRPC_RESOLVER_UNREF(resolver, "create"); diff --git a/src/core/surface/completion_queue.c b/src/core/surface/completion_queue.c index 8484418247..3f60b0b0ba 100644 --- a/src/core/surface/completion_queue.c +++ b/src/core/surface/completion_queue.c @@ -148,6 +148,8 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc, gpr_timespec deadline) { grpc_event ret; + deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC); + GRPC_CQ_INTERNAL_REF(cc, "next"); gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset)); for (;;) { @@ -188,6 +190,8 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag, grpc_cq_completion *c; grpc_cq_completion *prev; + deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC); + GRPC_CQ_INTERNAL_REF(cc, "pluck"); gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset)); for (;;) { diff --git a/src/core/surface/init.c b/src/core/surface/init.c index 04e27d30ac..5cba479317 100644 --- a/src/core/surface/init.c +++ b/src/core/surface/init.c @@ -39,6 +39,7 @@ #include "src/core/channel/channel_stack.h" #include "src/core/client_config/resolver_registry.h" #include "src/core/client_config/resolvers/dns_resolver.h" +#include "src/core/client_config/resolvers/sockaddr_resolver.h" #include "src/core/debug/trace.h" #include "src/core/iomgr/iomgr.h" #include "src/core/profiling/timers.h" @@ -47,10 +48,6 @@ #include "src/core/surface/surface_trace.h" #include "src/core/transport/chttp2_transport.h" -#ifdef GPR_POSIX_SOCKET -#include "src/core/client_config/resolvers/unix_resolver_posix.h" -#endif - static gpr_once g_basic_init = GPR_ONCE_INIT; static gpr_mu g_init_mu; static int g_initializations; @@ -68,6 +65,8 @@ void grpc_init(void) { gpr_time_init(); grpc_resolver_registry_init("dns:///"); grpc_register_resolver_type("dns", grpc_dns_resolver_factory_create()); + grpc_register_resolver_type("ipv4", grpc_ipv4_resolver_factory_create()); + grpc_register_resolver_type("ipv6", grpc_ipv6_resolver_factory_create()); #ifdef GPR_POSIX_SOCKET grpc_register_resolver_type("unix", grpc_unix_resolver_factory_create()); #endif diff --git a/src/core/surface/lame_client.c b/src/core/surface/lame_client.c index 3f2bb5c8a9..c4215a2cfb 100644 --- a/src/core/surface/lame_client.c +++ b/src/core/surface/lame_client.c @@ -47,7 +47,10 @@ typedef struct { grpc_linked_mdelem details; } call_data; -typedef struct { grpc_mdctx *mdctx; } channel_data; +typedef struct { + grpc_mdctx *mdctx; + grpc_channel *master; +} channel_data; static void lame_start_transport_stream_op(grpc_call_element *elem, grpc_transport_stream_op *op) { @@ -82,6 +85,11 @@ static void lame_start_transport_stream_op(grpc_call_element *elem, } } +static char *lame_get_peer(grpc_call_element *elem) { + channel_data *chand = elem->channel_data; + return grpc_channel_get_target(chand->master); +} + static void lame_start_transport_op(grpc_channel_element *elem, grpc_transport_op *op) { if (op->on_connectivity_state_change) { @@ -112,6 +120,7 @@ static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master, GPR_ASSERT(is_first); GPR_ASSERT(is_last); chand->mdctx = mdctx; + chand->master = master; } static void destroy_channel_elem(grpc_channel_element *elem) {} @@ -125,11 +134,12 @@ static const grpc_channel_filter lame_filter = { sizeof(channel_data), init_channel_elem, destroy_channel_elem, + lame_get_peer, "lame-client", }; -grpc_channel *grpc_lame_client_channel_create(void) { +grpc_channel *grpc_lame_client_channel_create(const char *target) { static const grpc_channel_filter *filters[] = {&lame_filter}; - return grpc_channel_create_from_filters(filters, 1, NULL, grpc_mdctx_create(), - 1); + return grpc_channel_create_from_filters(target, filters, 1, NULL, + grpc_mdctx_create(), 1); } diff --git a/src/core/surface/secure_channel_create.c b/src/core/surface/secure_channel_create.c index d87ec97b53..a280311ba0 100644 --- a/src/core/surface/secure_channel_create.c +++ b/src/core/surface/secure_channel_create.c @@ -196,13 +196,13 @@ grpc_channel *grpc_secure_channel_create(grpc_credentials *creds, if (grpc_find_security_connector_in_args(args) != NULL) { gpr_log(GPR_ERROR, "Cannot set security context in channel args."); - return grpc_lame_client_channel_create(); + return grpc_lame_client_channel_create(target); } if (grpc_credentials_create_security_connector( creds, target, args, NULL, &connector, &new_args_from_connector) != GRPC_SECURITY_OK) { - return grpc_lame_client_channel_create(); + return grpc_lame_client_channel_create(target); } mdctx = grpc_mdctx_create(); @@ -231,7 +231,8 @@ grpc_channel *grpc_secure_channel_create(grpc_credentials *creds, return NULL; } - channel = grpc_channel_create_from_filters(filters, n, args_copy, mdctx, 1); + channel = + grpc_channel_create_from_filters(target, filters, n, args_copy, mdctx, 1); grpc_client_channel_set_resolver(grpc_channel_get_channel_stack(channel), resolver); GRPC_RESOLVER_UNREF(resolver, "create"); diff --git a/src/core/surface/server.c b/src/core/surface/server.c index f2d6b11bc7..d3233cb3c8 100644 --- a/src/core/surface/server.c +++ b/src/core/surface/server.c @@ -530,6 +530,7 @@ static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) { static void server_on_recv(void *ptr, int success) { grpc_call_element *elem = ptr; call_data *calld = elem->call_data; + gpr_timespec op_deadline; if (success && !calld->got_initial_metadata) { size_t i; @@ -539,8 +540,9 @@ static void server_on_recv(void *ptr, int success) { grpc_stream_op *op = &ops[i]; if (op->type != GRPC_OP_METADATA) continue; grpc_metadata_batch_filter(&op->data.metadata, server_filter, elem); - if (0 != gpr_time_cmp(op->data.metadata.deadline, - gpr_inf_future(GPR_CLOCK_REALTIME))) { + op_deadline = op->data.metadata.deadline; + if (0 != + gpr_time_cmp(op_deadline, gpr_inf_future(op_deadline.clock_type))) { calld->deadline = op->data.metadata.deadline; } calld->got_initial_metadata = 1; @@ -722,6 +724,7 @@ static const grpc_channel_filter server_surface_filter = { sizeof(channel_data), init_channel_elem, destroy_channel_elem, + grpc_call_next_get_peer, "server", }; @@ -878,8 +881,8 @@ void grpc_server_setup_transport(grpc_server *s, grpc_transport *transport, grpc_transport_perform_op(transport, &op); } - channel = - grpc_channel_create_from_filters(filters, num_filters, args, mdctx, 0); + channel = grpc_channel_create_from_filters(NULL, filters, num_filters, args, + mdctx, 0); chand = (channel_data *)grpc_channel_stack_element( grpc_channel_get_channel_stack(channel), 0) ->channel_data; @@ -980,6 +983,8 @@ void grpc_server_shutdown_and_notify(grpc_server *server, channel_broadcaster broadcaster; request_killer reqkill; + GRPC_SERVER_LOG_SHUTDOWN(GPR_INFO, server, cq, tag); + /* lock, and gather up some stuff to do */ gpr_mu_lock(&server->mu_global); grpc_cq_begin_op(cq); diff --git a/src/core/transport/chttp2/internal.h b/src/core/transport/chttp2/internal.h index e5e6f445b7..e7901da510 100644 --- a/src/core/transport/chttp2/internal.h +++ b/src/core/transport/chttp2/internal.h @@ -286,6 +286,7 @@ struct grpc_chttp2_transport { grpc_endpoint *ep; grpc_mdctx *metadata_context; gpr_refcount refs; + char *peer_string; gpr_mu mu; diff --git a/src/core/transport/chttp2/parsing.c b/src/core/transport/chttp2/parsing.c index aa32f2e44a..50a2f752f6 100644 --- a/src/core/transport/chttp2/parsing.c +++ b/src/core/transport/chttp2/parsing.c @@ -588,7 +588,7 @@ static void on_header(void *tp, grpc_mdelem *md) { GPR_ASSERT(stream_parsing); GRPC_CHTTP2_IF_TRACING(gpr_log( - GPR_INFO, "HTTP:%d:HDR: %s: %s", stream_parsing->id, + GPR_INFO, "HTTP:%d:HDR:%s: %s: %s", stream_parsing->id, transport_parsing->is_client ? "CLI" : "SVR", grpc_mdstr_as_c_string(md->key), grpc_mdstr_as_c_string(md->value))); @@ -607,7 +607,7 @@ static void on_header(void *tp, grpc_mdelem *md) { } grpc_chttp2_incoming_metadata_buffer_set_deadline( &stream_parsing->incoming_metadata, - gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), *cached_timeout)); + gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC), *cached_timeout)); GRPC_MDELEM_UNREF(md); } else { grpc_chttp2_incoming_metadata_buffer_add(&stream_parsing->incoming_metadata, diff --git a/src/core/transport/chttp2/stream_encoder.c b/src/core/transport/chttp2/stream_encoder.c index d30059abf8..65b31a5afd 100644 --- a/src/core/transport/chttp2/stream_encoder.c +++ b/src/core/transport/chttp2/stream_encoder.c @@ -438,7 +438,7 @@ static void deadline_enc(grpc_chttp2_hpack_compressor *c, gpr_timespec deadline, char timeout_str[GRPC_CHTTP2_TIMEOUT_ENCODE_MIN_BUFSIZE]; grpc_mdelem *mdelem; grpc_chttp2_encode_timeout( - gpr_time_sub(deadline, gpr_now(GPR_CLOCK_REALTIME)), timeout_str); + gpr_time_sub(deadline, gpr_now(deadline.clock_type)), timeout_str); mdelem = grpc_mdelem_from_metadata_strings( c->mdctx, GRPC_MDSTR_REF(c->timeout_key_str), grpc_mdstr_from_string(c->mdctx, timeout_str)); @@ -560,6 +560,7 @@ void grpc_chttp2_encode(grpc_stream_op *ops, size_t ops_count, int eof, grpc_mdctx *mdctx = compressor->mdctx; grpc_linked_mdelem *l; int need_unref = 0; + gpr_timespec deadline; GPR_ASSERT(stream_id != 0); @@ -589,9 +590,9 @@ void grpc_chttp2_encode(grpc_stream_op *ops, size_t ops_count, int eof, l->md = hpack_enc(compressor, l->md, &st); need_unref |= l->md != NULL; } - if (gpr_time_cmp(op->data.metadata.deadline, - gpr_inf_future(GPR_CLOCK_REALTIME)) != 0) { - deadline_enc(compressor, op->data.metadata.deadline, &st); + deadline = op->data.metadata.deadline; + if (gpr_time_cmp(deadline, gpr_inf_future(deadline.clock_type)) != 0) { + deadline_enc(compressor, deadline, &st); } curop++; break; diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index c923d5e42f..eb435a2ee8 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -168,6 +168,7 @@ static void destruct_transport(grpc_chttp2_transport *t) { grpc_mdctx_unref(t->metadata_context); + gpr_free(t->peer_string); gpr_free(t); } @@ -217,6 +218,7 @@ static void init_transport(grpc_chttp2_transport *t, gpr_ref_init(&t->refs, 2); gpr_mu_init(&t->mu); grpc_mdctx_ref(mdctx); + t->peer_string = grpc_endpoint_get_peer(ep); t->metadata_context = mdctx; t->endpoint_reading = 1; t->global.next_stream_id = is_client ? 1 : 2; @@ -1069,9 +1071,17 @@ void grpc_chttp2_flowctl_trace(const char *file, int line, const char *reason, * INTEGRATION GLUE */ -static const grpc_transport_vtable vtable = { - sizeof(grpc_chttp2_stream), init_stream, perform_stream_op, - perform_transport_op, destroy_stream, destroy_transport}; +static char *chttp2_get_peer(grpc_transport *t) { + return gpr_strdup(((grpc_chttp2_transport *)t)->peer_string); +} + +static const grpc_transport_vtable vtable = {sizeof(grpc_chttp2_stream), + init_stream, + perform_stream_op, + perform_transport_op, + destroy_stream, + destroy_transport, + chttp2_get_peer}; grpc_transport *grpc_create_chttp2_transport( const grpc_channel_args *channel_args, grpc_endpoint *ep, grpc_mdctx *mdctx, diff --git a/src/core/transport/transport.c b/src/core/transport/transport.c index 2689e3028a..69c00b6a4f 100644 --- a/src/core/transport/transport.c +++ b/src/core/transport/transport.c @@ -65,6 +65,10 @@ void grpc_transport_destroy_stream(grpc_transport *transport, transport->vtable->destroy_stream(transport, stream); } +char *grpc_transport_get_peer(grpc_transport *transport) { + return transport->vtable->get_peer(transport); +} + void grpc_transport_stream_op_finish_with_failure( grpc_transport_stream_op *op) { if (op->send_ops) { diff --git a/src/core/transport/transport.h b/src/core/transport/transport.h index 64503604ee..553779602a 100644 --- a/src/core/transport/transport.h +++ b/src/core/transport/transport.h @@ -182,4 +182,7 @@ void grpc_transport_close(grpc_transport *transport); /* Destroy the transport */ void grpc_transport_destroy(grpc_transport *transport); +/* Get the transports peer */ +char *grpc_transport_get_peer(grpc_transport *transport); + #endif /* GRPC_INTERNAL_CORE_TRANSPORT_TRANSPORT_H */ diff --git a/src/core/transport/transport_impl.h b/src/core/transport/transport_impl.h index 515721dfb6..d3bbdf6c27 100644 --- a/src/core/transport/transport_impl.h +++ b/src/core/transport/transport_impl.h @@ -58,6 +58,9 @@ typedef struct grpc_transport_vtable { /* implementation of grpc_transport_destroy */ void (*destroy)(grpc_transport *self); + + /* implementation of grpc_transport_get_peer */ + char *(*get_peer)(grpc_transport *self); } grpc_transport_vtable; /* an instance of a grpc transport */ diff --git a/src/core/transport/transport_op_string.c b/src/core/transport/transport_op_string.c index 9d127c5472..10d796fc15 100644 --- a/src/core/transport/transport_op_string.c +++ b/src/core/transport/transport_op_string.c @@ -61,7 +61,7 @@ static void put_metadata_list(gpr_strvec *b, grpc_metadata_batch md) { if (m != md.list.head) gpr_strvec_add(b, gpr_strdup(", ")); put_metadata(b, m->md); } - if (gpr_time_cmp(md.deadline, gpr_inf_future(GPR_CLOCK_REALTIME)) != 0) { + if (gpr_time_cmp(md.deadline, gpr_inf_future(md.deadline.clock_type)) != 0) { char *tmp; gpr_asprintf(&tmp, " deadline=%d.%09d", md.deadline.tv_sec, md.deadline.tv_nsec); |