diff options
Diffstat (limited to 'src/core/ext/filters')
24 files changed, 576 insertions, 390 deletions
diff --git a/src/core/ext/filters/client_channel/client_channel.cc b/src/core/ext/filters/client_channel/client_channel.cc index f141aabe70..ea6775a8d8 100644 --- a/src/core/ext/filters/client_channel/client_channel.cc +++ b/src/core/ext/filters/client_channel/client_channel.cc @@ -327,14 +327,16 @@ static void on_resolver_result_changed_locked(void* arg, grpc_error* error) { if (chand->resolver_result != nullptr) { if (chand->resolver != nullptr) { // Find LB policy name. - const char* lb_policy_name = grpc_channel_args_get_string( + const grpc_arg* channel_arg = grpc_channel_args_find( chand->resolver_result, GRPC_ARG_LB_POLICY_NAME); + const char* lb_policy_name = grpc_channel_arg_get_string(channel_arg); // Special case: If at least one balancer address is present, we use // the grpclb policy, regardless of what the resolver actually specified. - grpc_lb_addresses* addresses = - grpc_channel_args_get_pointer<grpc_lb_addresses>( - chand->resolver_result, GRPC_ARG_LB_ADDRESSES); - if (addresses != nullptr) { + channel_arg = + grpc_channel_args_find(chand->resolver_result, GRPC_ARG_LB_ADDRESSES); + if (channel_arg != nullptr && channel_arg->type == GRPC_ARG_POINTER) { + grpc_lb_addresses* addresses = + static_cast<grpc_lb_addresses*>(channel_arg->value.pointer.p); bool found_balancer_address = false; for (size_t i = 0; i < addresses->num_addresses; ++i) { if (addresses->addresses[i].is_balancer) { @@ -398,15 +400,18 @@ static void on_resolver_result_changed_locked(void* arg, grpc_error* error) { // The copy will be saved in chand->lb_policy_name below. lb_policy_name_dup = gpr_strdup(lb_policy_name); // Find service config. - service_config_json = gpr_strdup(grpc_channel_args_get_string( - chand->resolver_result, GRPC_ARG_SERVICE_CONFIG)); + channel_arg = grpc_channel_args_find(chand->resolver_result, + GRPC_ARG_SERVICE_CONFIG); + service_config_json = + gpr_strdup(grpc_channel_arg_get_string(channel_arg)); if (service_config_json != nullptr) { grpc_core::UniquePtr<grpc_core::ServiceConfig> service_config = grpc_core::ServiceConfig::Create(service_config_json); if (service_config != nullptr) { if (chand->enable_retries) { - const char* server_uri = grpc_channel_args_get_string( - chand->resolver_result, GRPC_ARG_SERVER_URI); + channel_arg = grpc_channel_args_find(chand->resolver_result, + GRPC_ARG_SERVER_URI); + const char* server_uri = grpc_channel_arg_get_string(channel_arg); GPR_ASSERT(server_uri != nullptr); grpc_uri* uri = grpc_uri_parse(server_uri, true); GPR_ASSERT(uri->path[0] != '\0'); @@ -643,37 +648,45 @@ static grpc_error* cc_init_channel_elem(grpc_channel_element* elem, "client_channel"); grpc_client_channel_start_backup_polling(chand->interested_parties); // Record max per-RPC retry buffer size. - chand->per_rpc_retry_buffer_size = (size_t)grpc_channel_args_get_integer( - args->channel_args, GRPC_ARG_PER_RPC_RETRY_BUFFER_SIZE, - {DEFAULT_PER_RPC_RETRY_BUFFER_SIZE, 0, INT_MAX}); + const grpc_arg* arg = grpc_channel_args_find( + args->channel_args, GRPC_ARG_PER_RPC_RETRY_BUFFER_SIZE); + chand->per_rpc_retry_buffer_size = (size_t)grpc_channel_arg_get_integer( + arg, {DEFAULT_PER_RPC_RETRY_BUFFER_SIZE, 0, INT_MAX}); // Record enable_retries. - chand->enable_retries = grpc_channel_args_get_bool( - args->channel_args, GRPC_ARG_ENABLE_RETRIES, true); + arg = grpc_channel_args_find(args->channel_args, GRPC_ARG_ENABLE_RETRIES); + chand->enable_retries = grpc_channel_arg_get_bool(arg, true); // Record client channel factory. - grpc_client_channel_factory* client_channel_factory = - grpc_channel_args_get_pointer<grpc_client_channel_factory>( - args->channel_args, GRPC_ARG_CLIENT_CHANNEL_FACTORY); - if (client_channel_factory == nullptr) { + arg = grpc_channel_args_find(args->channel_args, + GRPC_ARG_CLIENT_CHANNEL_FACTORY); + if (arg == nullptr) { return GRPC_ERROR_CREATE_FROM_STATIC_STRING( - "Missing or malformed client channel factory in args for client " - "channel filter"); + "Missing client channel factory in args for client channel filter"); } - grpc_client_channel_factory_ref(client_channel_factory); - chand->client_channel_factory = client_channel_factory; + if (arg->type != GRPC_ARG_POINTER) { + return GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "client channel factory arg must be a pointer"); + } + grpc_client_channel_factory_ref( + static_cast<grpc_client_channel_factory*>(arg->value.pointer.p)); + chand->client_channel_factory = + static_cast<grpc_client_channel_factory*>(arg->value.pointer.p); // Get server name to resolve, using proxy mapper if needed. - char* server_uri = - grpc_channel_args_get_string(args->channel_args, GRPC_ARG_SERVER_URI); - if (server_uri == nullptr) { + arg = grpc_channel_args_find(args->channel_args, GRPC_ARG_SERVER_URI); + if (arg == nullptr) { + return GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "Missing server uri in args for client channel filter"); + } + if (arg->type != GRPC_ARG_STRING) { return GRPC_ERROR_CREATE_FROM_STATIC_STRING( - "Missing or malformed server uri in args for client channel filter"); + "server uri arg must be a string"); } char* proxy_name = nullptr; grpc_channel_args* new_args = nullptr; - grpc_proxy_mappers_map_name(server_uri, args->channel_args, &proxy_name, - &new_args); + grpc_proxy_mappers_map_name(arg->value.string, args->channel_args, + &proxy_name, &new_args); // Instantiate resolver. chand->resolver = grpc_core::ResolverRegistry::CreateResolver( - proxy_name != nullptr ? proxy_name : server_uri, + proxy_name != nullptr ? proxy_name : arg->value.string, new_args != nullptr ? new_args : args->channel_args, chand->interested_parties, chand->combiner); if (proxy_name != nullptr) gpr_free(proxy_name); diff --git a/src/core/ext/filters/client_channel/http_connect_handshaker.cc b/src/core/ext/filters/client_channel/http_connect_handshaker.cc index 58f059829c..4e8b8b71db 100644 --- a/src/core/ext/filters/client_channel/http_connect_handshaker.cc +++ b/src/core/ext/filters/client_channel/http_connect_handshaker.cc @@ -254,8 +254,9 @@ static void http_connect_handshaker_do_handshake( reinterpret_cast<http_connect_handshaker*>(handshaker_in); // Check for HTTP CONNECT channel arg. // If not found, invoke on_handshake_done without doing anything. - char* server_name = - grpc_channel_args_get_string(args->args, GRPC_ARG_HTTP_CONNECT_SERVER); + const grpc_arg* arg = + grpc_channel_args_find(args->args, GRPC_ARG_HTTP_CONNECT_SERVER); + char* server_name = grpc_channel_arg_get_string(arg); if (server_name == nullptr) { // Set shutdown to true so that subsequent calls to // http_connect_handshaker_shutdown() do nothing. @@ -266,8 +267,8 @@ static void http_connect_handshaker_do_handshake( return; } // Get headers from channel args. - char* arg_header_string = - grpc_channel_args_get_string(args->args, GRPC_ARG_HTTP_CONNECT_HEADERS); + arg = grpc_channel_args_find(args->args, GRPC_ARG_HTTP_CONNECT_HEADERS); + char* arg_header_string = grpc_channel_arg_get_string(arg); grpc_http_header* headers = nullptr; size_t num_headers = 0; char** header_strings = nullptr; diff --git a/src/core/ext/filters/client_channel/http_proxy.cc b/src/core/ext/filters/client_channel/http_proxy.cc index e21de35a7d..26d3f479b7 100644 --- a/src/core/ext/filters/client_channel/http_proxy.cc +++ b/src/core/ext/filters/client_channel/http_proxy.cc @@ -37,17 +37,21 @@ #include "src/core/lib/slice/b64.h" /** - * Parses the 'http_proxy' env var and returns the proxy hostname to resolve or - * nullptr on error. Also sets 'user_cred' to user credentials if present in the - * 'http_proxy' env var, otherwise leaves it unchanged. It is caller's - * responsibility to gpr_free user_cred. + * Parses the 'https_proxy' env var (fallback on 'http_proxy') and returns the + * proxy hostname to resolve or nullptr on error. Also sets 'user_cred' to user + * credentials if present in the 'http_proxy' env var, otherwise leaves it + * unchanged. It is caller's responsibility to gpr_free user_cred. */ static char* get_http_proxy_server(char** user_cred) { GPR_ASSERT(user_cred != nullptr); char* proxy_name = nullptr; - char* uri_str = gpr_getenv("http_proxy"); char** authority_strs = nullptr; size_t authority_nstrs; + /* Prefer using 'https_proxy'. Fallback on 'http_proxy' if it is not set. The + * fallback behavior can be removed if there's a demand for it. + */ + char* uri_str = gpr_getenv("https_proxy"); + if (uri_str == nullptr) uri_str = gpr_getenv("http_proxy"); if (uri_str == nullptr) return nullptr; grpc_uri* uri = grpc_uri_parse(uri_str, false /* suppress_errors */); if (uri == nullptr || uri->authority == nullptr) { @@ -88,7 +92,9 @@ done: * should be used. */ bool http_proxy_enabled(const grpc_channel_args* args) { - return grpc_channel_args_get_bool(args, GRPC_ARG_ENABLE_HTTP_PROXY, true); + const grpc_arg* arg = + grpc_channel_args_find(args, GRPC_ARG_ENABLE_HTTP_PROXY); + return grpc_channel_arg_get_bool(arg, true); } static bool proxy_mapper_map_name(grpc_proxy_mapper* mapper, diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc index 0ee4958f3d..263b51ae89 100644 --- a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc +++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.cc @@ -1045,8 +1045,8 @@ GrpcLb::GrpcLb(const grpc_lb_addresses* addresses, grpc_combiner_scheduler(args.combiner)); grpc_connectivity_state_init(&state_tracker_, GRPC_CHANNEL_IDLE, "grpclb"); // Record server name. - const char* server_uri = - grpc_channel_args_get_string(args.args, GRPC_ARG_SERVER_URI); + const grpc_arg* arg = grpc_channel_args_find(args.args, GRPC_ARG_SERVER_URI); + const char* server_uri = grpc_channel_arg_get_string(arg); GPR_ASSERT(server_uri != nullptr); grpc_uri* uri = grpc_uri_parse(server_uri, true); GPR_ASSERT(uri->path[0] != '\0'); @@ -1058,12 +1058,12 @@ GrpcLb::GrpcLb(const grpc_lb_addresses* addresses, } grpc_uri_destroy(uri); // Record LB call timeout. - lb_call_timeout_ms_ = grpc_channel_args_get_integer( - args.args, GRPC_ARG_GRPCLB_CALL_TIMEOUT_MS, {0, 0, INT_MAX}); + arg = grpc_channel_args_find(args.args, GRPC_ARG_GRPCLB_CALL_TIMEOUT_MS); + lb_call_timeout_ms_ = grpc_channel_arg_get_integer(arg, {0, 0, INT_MAX}); // Record fallback timeout. - lb_fallback_timeout_ms_ = grpc_channel_args_get_integer( - args.args, GRPC_ARG_GRPCLB_FALLBACK_TIMEOUT_MS, - {GRPC_GRPCLB_DEFAULT_FALLBACK_TIMEOUT_MS, 0, INT_MAX}); + arg = grpc_channel_args_find(args.args, GRPC_ARG_GRPCLB_FALLBACK_TIMEOUT_MS); + lb_fallback_timeout_ms_ = grpc_channel_arg_get_integer( + arg, {GRPC_GRPCLB_DEFAULT_FALLBACK_TIMEOUT_MS, 0, INT_MAX}); // Process channel args. ProcessChannelArgsLocked(*args.args); } @@ -1284,10 +1284,8 @@ void GrpcLb::NotifyOnStateChangeLocked(grpc_connectivity_state* current, } void GrpcLb::ProcessChannelArgsLocked(const grpc_channel_args& args) { - const grpc_lb_addresses* addresses = - grpc_channel_args_get_pointer<grpc_lb_addresses>(&args, - GRPC_ARG_LB_ADDRESSES); - if (GPR_UNLIKELY(addresses == nullptr)) { + const grpc_arg* arg = grpc_channel_args_find(&args, GRPC_ARG_LB_ADDRESSES); + if (GPR_UNLIKELY(arg == nullptr || arg->type != GRPC_ARG_POINTER)) { // Ignore this update. gpr_log( GPR_ERROR, @@ -1295,6 +1293,8 @@ void GrpcLb::ProcessChannelArgsLocked(const grpc_channel_args& args) { this); return; } + const grpc_lb_addresses* addresses = + static_cast<const grpc_lb_addresses*>(arg->value.pointer.p); // Update fallback address list. if (fallback_backend_addresses_ != nullptr) { grpc_lb_addresses_destroy(fallback_backend_addresses_); @@ -1860,12 +1860,13 @@ class GrpcLbFactory : public LoadBalancingPolicyFactory { OrphanablePtr<LoadBalancingPolicy> CreateLoadBalancingPolicy( const LoadBalancingPolicy::Args& args) const override { /* Count the number of gRPC-LB addresses. There must be at least one. */ - grpc_lb_addresses* addresses = - grpc_channel_args_get_pointer<grpc_lb_addresses>(args.args, - GRPC_ARG_LB_ADDRESSES); - if (addresses == nullptr) { + const grpc_arg* arg = + grpc_channel_args_find(args.args, GRPC_ARG_LB_ADDRESSES); + if (arg == nullptr || arg->type != GRPC_ARG_POINTER) { return nullptr; } + grpc_lb_addresses* addresses = + static_cast<grpc_lb_addresses*>(arg->value.pointer.p); size_t num_grpclb_addrs = 0; for (size_t i = 0; i < addresses->num_addresses; ++i) { if (addresses->addresses[i].is_balancer) ++num_grpclb_addrs; @@ -1892,9 +1893,10 @@ bool maybe_add_client_load_reporting_filter(grpc_channel_stack_builder* builder, void* arg) { const grpc_channel_args* args = grpc_channel_stack_builder_get_channel_arguments(builder); - const char* lb_policy = - grpc_channel_args_get_string(args, GRPC_ARG_LB_POLICY_NAME); - if (lb_policy != nullptr && strcmp(lb_policy, "grpclb") == 0) { + const grpc_arg* channel_arg = + grpc_channel_args_find(args, GRPC_ARG_LB_POLICY_NAME); + if (channel_arg != nullptr && channel_arg->type == GRPC_ARG_STRING && + strcmp(channel_arg->value.string, "grpclb") == 0) { return grpc_channel_stack_builder_append_filter( builder, (const grpc_channel_filter*)arg, nullptr, nullptr); } diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb_channel_secure.cc b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb_channel_secure.cc index 972bdd40d5..441efd5e23 100644 --- a/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb_channel_secure.cc +++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb_channel_secure.cc @@ -72,10 +72,11 @@ grpc_channel_args* grpc_lb_policy_grpclb_modify_lb_channel_args( grpc_arg args_to_add[2]; size_t num_args_to_add = 0; // Add arg for targets info table. + const grpc_arg* arg = grpc_channel_args_find(args, GRPC_ARG_LB_ADDRESSES); + GPR_ASSERT(arg != nullptr); + GPR_ASSERT(arg->type == GRPC_ARG_POINTER); grpc_lb_addresses* addresses = - grpc_channel_args_get_pointer<grpc_lb_addresses>(args, - GRPC_ARG_LB_ADDRESSES); - GPR_ASSERT(addresses != nullptr); + static_cast<grpc_lb_addresses*>(arg->value.pointer.p); grpc_core::RefCountedPtr<grpc_core::TargetAuthorityTable> target_authority_table = grpc_core::CreateTargetAuthorityTable(addresses); args_to_add[num_args_to_add++] = diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.cc b/src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.cc index ed246273c9..c854e7a9cf 100644 --- a/src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.cc +++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.cc @@ -18,9 +18,9 @@ #include <grpc/support/port_platform.h> +#include "pb_decode.h" +#include "pb_encode.h" #include "src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.h" -#include "third_party/nanopb/pb_decode.h" -#include "third_party/nanopb/pb_encode.h" #include <grpc/support/alloc.h> diff --git a/src/core/ext/filters/client_channel/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.h b/src/core/ext/filters/client_channel/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.h index 066c076202..088b6f513a 100644 --- a/src/core/ext/filters/client_channel/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.h +++ b/src/core/ext/filters/client_channel/lb_policy/grpclb/proto/grpc/lb/v1/load_balancer.pb.h @@ -3,7 +3,7 @@ #ifndef PB_GRPC_LB_V1_LOAD_BALANCER_PB_H_INCLUDED #define PB_GRPC_LB_V1_LOAD_BALANCER_PB_H_INCLUDED -#include "third_party/nanopb/pb.h" +#include "pb.h" /* @@protoc_insertion_point(includes) */ #if PB_PROTO_HEADER_VERSION != 30 #error Regenerate this file with the current version of nanopb generator. diff --git a/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc b/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc index 70220e28d3..ff2140e628 100644 --- a/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc +++ b/src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc @@ -281,10 +281,8 @@ void PickFirst::PingOneLocked(grpc_closure* on_initiate, grpc_closure* on_ack) { } void PickFirst::UpdateLocked(const grpc_channel_args& args) { - const grpc_lb_addresses* addresses = - grpc_channel_args_get_pointer<const grpc_lb_addresses>( - &args, GRPC_ARG_LB_ADDRESSES); - if (addresses == nullptr) { + const grpc_arg* arg = grpc_channel_args_find(&args, GRPC_ARG_LB_ADDRESSES); + if (arg == nullptr || arg->type != GRPC_ARG_POINTER) { if (subchannel_list_ == nullptr) { // If we don't have a current subchannel list, go into TRANSIENT FAILURE. grpc_connectivity_state_set( @@ -300,6 +298,8 @@ void PickFirst::UpdateLocked(const grpc_channel_args& args) { } return; } + const grpc_lb_addresses* addresses = + static_cast<const grpc_lb_addresses*>(arg->value.pointer.p); if (grpc_lb_pick_first_trace.enabled()) { gpr_log(GPR_INFO, "Pick First %p received update with %" PRIuPTR " addresses", this, diff --git a/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc b/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc index 066f6ef7a6..b177385065 100644 --- a/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc +++ b/src/core/ext/filters/client_channel/lb_policy/round_robin/round_robin.cc @@ -607,10 +607,8 @@ void RoundRobin::PingOneLocked(grpc_closure* on_initiate, } void RoundRobin::UpdateLocked(const grpc_channel_args& args) { - grpc_lb_addresses* addresses = - grpc_channel_args_get_pointer<grpc_lb_addresses>(&args, - GRPC_ARG_LB_ADDRESSES); - if (GPR_UNLIKELY(addresses == nullptr)) { + const grpc_arg* arg = grpc_channel_args_find(&args, GRPC_ARG_LB_ADDRESSES); + if (GPR_UNLIKELY(arg == nullptr || arg->type != GRPC_ARG_POINTER)) { gpr_log(GPR_ERROR, "[RR %p] update provided no addresses; ignoring", this); // If we don't have a current subchannel list, go into TRANSIENT_FAILURE. // Otherwise, keep using the current subchannel list (ignore this update). @@ -622,6 +620,8 @@ void RoundRobin::UpdateLocked(const grpc_channel_args& args) { } return; } + grpc_lb_addresses* addresses = + static_cast<grpc_lb_addresses*>(arg->value.pointer.p); if (grpc_lb_round_robin_trace.enabled()) { gpr_log(GPR_INFO, "[RR %p] received update with %" PRIuPTR " addresses", this, addresses->num_addresses); diff --git a/src/core/ext/filters/client_channel/lb_policy_factory.cc b/src/core/ext/filters/client_channel/lb_policy_factory.cc index 93dae753bf..7c8cba55b7 100644 --- a/src/core/ext/filters/client_channel/lb_policy_factory.cc +++ b/src/core/ext/filters/client_channel/lb_policy_factory.cc @@ -147,6 +147,9 @@ grpc_arg grpc_lb_addresses_create_channel_arg( grpc_lb_addresses* grpc_lb_addresses_find_channel_arg( const grpc_channel_args* channel_args) { - return grpc_channel_args_get_pointer<grpc_lb_addresses>( - channel_args, GRPC_ARG_LB_ADDRESSES); + const grpc_arg* lb_addresses_arg = + grpc_channel_args_find(channel_args, GRPC_ARG_LB_ADDRESSES); + if (lb_addresses_arg == nullptr || lb_addresses_arg->type != GRPC_ARG_POINTER) + return nullptr; + return static_cast<grpc_lb_addresses*>(lb_addresses_arg->value.pointer.p); } diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.cc b/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.cc index 01156b1e40..f4f6444c5f 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.cc +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.cc @@ -140,11 +140,14 @@ AresDnsResolver::AresDnsResolver(const ResolverArgs& args) dns_server_ = gpr_strdup(args.uri->authority); } channel_args_ = grpc_channel_args_copy(args.args); - request_service_config_ = !grpc_channel_args_get_bool( - channel_args_, GRPC_ARG_SERVICE_CONFIG_DISABLE_RESOLUTION, false); - min_time_between_resolutions_ = grpc_channel_args_get_integer( - channel_args_, GRPC_ARG_DNS_MIN_TIME_BETWEEN_RESOLUTIONS_MS, - {1000, 0, INT_MAX}); + const grpc_arg* arg = grpc_channel_args_find( + channel_args_, GRPC_ARG_SERVICE_CONFIG_DISABLE_RESOLUTION); + request_service_config_ = !grpc_channel_arg_get_integer( + arg, (grpc_integer_options){false, false, true}); + arg = grpc_channel_args_find(channel_args_, + GRPC_ARG_DNS_MIN_TIME_BETWEEN_RESOLUTIONS_MS); + min_time_between_resolutions_ = + grpc_channel_arg_get_integer(arg, {1000, 0, INT_MAX}); interested_parties_ = grpc_pollset_set_create(); if (args.pollset_set != nullptr) { grpc_pollset_set_add_pollset_set(interested_parties_, args.pollset_set); diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.cc b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.cc new file mode 100644 index 0000000000..06a6e853f5 --- /dev/null +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.cc @@ -0,0 +1,311 @@ +/* + * + * Copyright 2016 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ +#include <grpc/support/port_platform.h> + +#include "src/core/lib/iomgr/port.h" +#if GRPC_ARES == 1 && defined(GRPC_POSIX_SOCKET_ARES_EV_DRIVER) + +#include <ares.h> +#include <string.h> +#include <sys/ioctl.h> + +#include "src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.h" + +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> +#include <grpc/support/string_util.h> +#include <grpc/support/time.h> +#include "src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h" +#include "src/core/lib/gpr/string.h" +#include "src/core/lib/iomgr/ev_posix.h" +#include "src/core/lib/iomgr/iomgr_internal.h" +#include "src/core/lib/iomgr/sockaddr_utils.h" + +typedef struct fd_node { + /** the owner of this fd node */ + grpc_ares_ev_driver* ev_driver; + /** a closure wrapping on_readable_locked, which should be + invoked when the grpc_fd in this node becomes readable. */ + grpc_closure read_closure; + /** a closure wrapping on_writable_locked, which should be + invoked when the grpc_fd in this node becomes writable. */ + grpc_closure write_closure; + /** next fd node in the list */ + struct fd_node* next; + + /** wrapped fd that's polled by grpc's poller for the current platform */ + grpc_core::GrpcPolledFd* grpc_polled_fd; + /** if the readable closure has been registered */ + bool readable_registered; + /** if the writable closure has been registered */ + bool writable_registered; + /** if the fd has been shutdown yet from grpc iomgr perspective */ + bool already_shutdown; +} fd_node; + +struct grpc_ares_ev_driver { + /** the ares_channel owned by this event driver */ + ares_channel channel; + /** pollset set for driving the IO events of the channel */ + grpc_pollset_set* pollset_set; + /** refcount of the event driver */ + gpr_refcount refs; + + /** combiner to synchronize c-ares and I/O callbacks on */ + grpc_combiner* combiner; + /** a list of grpc_fd that this event driver is currently using. */ + fd_node* fds; + /** is this event driver currently working? */ + bool working; + /** is this event driver being shut down */ + bool shutting_down; +}; + +static void grpc_ares_notify_on_event_locked(grpc_ares_ev_driver* ev_driver); + +static grpc_ares_ev_driver* grpc_ares_ev_driver_ref( + grpc_ares_ev_driver* ev_driver) { + gpr_log(GPR_DEBUG, "Ref ev_driver %" PRIuPTR, (uintptr_t)ev_driver); + gpr_ref(&ev_driver->refs); + return ev_driver; +} + +static void grpc_ares_ev_driver_unref(grpc_ares_ev_driver* ev_driver) { + gpr_log(GPR_DEBUG, "Unref ev_driver %" PRIuPTR, (uintptr_t)ev_driver); + if (gpr_unref(&ev_driver->refs)) { + gpr_log(GPR_DEBUG, "destroy ev_driver %" PRIuPTR, (uintptr_t)ev_driver); + GPR_ASSERT(ev_driver->fds == nullptr); + GRPC_COMBINER_UNREF(ev_driver->combiner, "free ares event driver"); + ares_destroy(ev_driver->channel); + gpr_free(ev_driver); + } +} + +static void fd_node_destroy_locked(fd_node* fdn) { + gpr_log(GPR_DEBUG, "delete fd: %s", fdn->grpc_polled_fd->GetName()); + GPR_ASSERT(!fdn->readable_registered); + GPR_ASSERT(!fdn->writable_registered); + GPR_ASSERT(fdn->already_shutdown); + grpc_core::Delete(fdn->grpc_polled_fd); + gpr_free(fdn); +} + +static void fd_node_shutdown_locked(fd_node* fdn, const char* reason) { + if (!fdn->already_shutdown) { + fdn->already_shutdown = true; + fdn->grpc_polled_fd->ShutdownLocked( + GRPC_ERROR_CREATE_FROM_STATIC_STRING(reason)); + } +} + +grpc_error* grpc_ares_ev_driver_create_locked(grpc_ares_ev_driver** ev_driver, + grpc_pollset_set* pollset_set, + grpc_combiner* combiner) { + *ev_driver = static_cast<grpc_ares_ev_driver*>( + gpr_malloc(sizeof(grpc_ares_ev_driver))); + ares_options opts; + memset(&opts, 0, sizeof(opts)); + opts.flags |= ARES_FLAG_STAYOPEN; + int status = ares_init_options(&(*ev_driver)->channel, &opts, ARES_OPT_FLAGS); + grpc_core::ConfigureAresChannelLocked(&(*ev_driver)->channel); + gpr_log(GPR_DEBUG, "grpc_ares_ev_driver_create_locked"); + if (status != ARES_SUCCESS) { + char* err_msg; + gpr_asprintf(&err_msg, "Failed to init ares channel. C-ares error: %s", + ares_strerror(status)); + grpc_error* err = GRPC_ERROR_CREATE_FROM_COPIED_STRING(err_msg); + gpr_free(err_msg); + gpr_free(*ev_driver); + return err; + } + (*ev_driver)->combiner = GRPC_COMBINER_REF(combiner, "ares event driver"); + gpr_ref_init(&(*ev_driver)->refs, 1); + (*ev_driver)->pollset_set = pollset_set; + (*ev_driver)->fds = nullptr; + (*ev_driver)->working = false; + (*ev_driver)->shutting_down = false; + return GRPC_ERROR_NONE; +} + +void grpc_ares_ev_driver_destroy_locked(grpc_ares_ev_driver* ev_driver) { + // We mark the event driver as being shut down. If the event driver + // is working, grpc_ares_notify_on_event_locked will shut down the + // fds; if it's not working, there are no fds to shut down. + ev_driver->shutting_down = true; + grpc_ares_ev_driver_unref(ev_driver); +} + +void grpc_ares_ev_driver_shutdown_locked(grpc_ares_ev_driver* ev_driver) { + ev_driver->shutting_down = true; + fd_node* fn = ev_driver->fds; + while (fn != nullptr) { + fd_node_shutdown_locked(fn, "grpc_ares_ev_driver_shutdown"); + fn = fn->next; + } +} + +// Search fd in the fd_node list head. This is an O(n) search, the max possible +// value of n is ARES_GETSOCK_MAXNUM (16). n is typically 1 - 2 in our tests. +static fd_node* pop_fd_node_locked(fd_node** head, ares_socket_t as) { + fd_node dummy_head; + dummy_head.next = *head; + fd_node* node = &dummy_head; + while (node->next != nullptr) { + if (node->next->grpc_polled_fd->GetWrappedAresSocketLocked() == as) { + fd_node* ret = node->next; + node->next = node->next->next; + *head = dummy_head.next; + return ret; + } + node = node->next; + } + return nullptr; +} + +static void on_readable_locked(void* arg, grpc_error* error) { + fd_node* fdn = static_cast<fd_node*>(arg); + grpc_ares_ev_driver* ev_driver = fdn->ev_driver; + const ares_socket_t as = fdn->grpc_polled_fd->GetWrappedAresSocketLocked(); + fdn->readable_registered = false; + gpr_log(GPR_DEBUG, "readable on %s", fdn->grpc_polled_fd->GetName()); + if (error == GRPC_ERROR_NONE) { + do { + ares_process_fd(ev_driver->channel, as, ARES_SOCKET_BAD); + } while (fdn->grpc_polled_fd->IsFdStillReadableLocked()); + } else { + // If error is not GRPC_ERROR_NONE, it means the fd has been shutdown or + // timed out. The pending lookups made on this ev_driver will be cancelled + // by the following ares_cancel() and the on_done callbacks will be invoked + // with a status of ARES_ECANCELLED. The remaining file descriptors in this + // ev_driver will be cleaned up in the follwing + // grpc_ares_notify_on_event_locked(). + ares_cancel(ev_driver->channel); + } + grpc_ares_notify_on_event_locked(ev_driver); + grpc_ares_ev_driver_unref(ev_driver); +} + +static void on_writable_locked(void* arg, grpc_error* error) { + fd_node* fdn = static_cast<fd_node*>(arg); + grpc_ares_ev_driver* ev_driver = fdn->ev_driver; + const ares_socket_t as = fdn->grpc_polled_fd->GetWrappedAresSocketLocked(); + fdn->writable_registered = false; + gpr_log(GPR_DEBUG, "writable on %s", fdn->grpc_polled_fd->GetName()); + if (error == GRPC_ERROR_NONE) { + ares_process_fd(ev_driver->channel, ARES_SOCKET_BAD, as); + } else { + // If error is not GRPC_ERROR_NONE, it means the fd has been shutdown or + // timed out. The pending lookups made on this ev_driver will be cancelled + // by the following ares_cancel() and the on_done callbacks will be invoked + // with a status of ARES_ECANCELLED. The remaining file descriptors in this + // ev_driver will be cleaned up in the follwing + // grpc_ares_notify_on_event_locked(). + ares_cancel(ev_driver->channel); + } + grpc_ares_notify_on_event_locked(ev_driver); + grpc_ares_ev_driver_unref(ev_driver); +} + +ares_channel* grpc_ares_ev_driver_get_channel_locked( + grpc_ares_ev_driver* ev_driver) { + return &ev_driver->channel; +} + +// Get the file descriptors used by the ev_driver's ares channel, register +// driver_closure with these filedescriptors. +static void grpc_ares_notify_on_event_locked(grpc_ares_ev_driver* ev_driver) { + fd_node* new_list = nullptr; + if (!ev_driver->shutting_down) { + ares_socket_t socks[ARES_GETSOCK_MAXNUM]; + int socks_bitmask = + ares_getsock(ev_driver->channel, socks, ARES_GETSOCK_MAXNUM); + for (size_t i = 0; i < ARES_GETSOCK_MAXNUM; i++) { + if (ARES_GETSOCK_READABLE(socks_bitmask, i) || + ARES_GETSOCK_WRITABLE(socks_bitmask, i)) { + fd_node* fdn = pop_fd_node_locked(&ev_driver->fds, socks[i]); + // Create a new fd_node if sock[i] is not in the fd_node list. + if (fdn == nullptr) { + fdn = static_cast<fd_node*>(gpr_malloc(sizeof(fd_node))); + fdn->grpc_polled_fd = grpc_core::NewGrpcPolledFdLocked( + socks[i], ev_driver->pollset_set); + gpr_log(GPR_DEBUG, "new fd: %s", fdn->grpc_polled_fd->GetName()); + fdn->ev_driver = ev_driver; + fdn->readable_registered = false; + fdn->writable_registered = false; + fdn->already_shutdown = false; + GRPC_CLOSURE_INIT(&fdn->read_closure, on_readable_locked, fdn, + grpc_combiner_scheduler(ev_driver->combiner)); + GRPC_CLOSURE_INIT(&fdn->write_closure, on_writable_locked, fdn, + grpc_combiner_scheduler(ev_driver->combiner)); + } + fdn->next = new_list; + new_list = fdn; + // Register read_closure if the socket is readable and read_closure has + // not been registered with this socket. + if (ARES_GETSOCK_READABLE(socks_bitmask, i) && + !fdn->readable_registered) { + grpc_ares_ev_driver_ref(ev_driver); + gpr_log(GPR_DEBUG, "notify read on: %s", + fdn->grpc_polled_fd->GetName()); + fdn->grpc_polled_fd->RegisterForOnReadableLocked(&fdn->read_closure); + fdn->readable_registered = true; + } + // Register write_closure if the socket is writable and write_closure + // has not been registered with this socket. + if (ARES_GETSOCK_WRITABLE(socks_bitmask, i) && + !fdn->writable_registered) { + gpr_log(GPR_DEBUG, "notify write on: %s", + fdn->grpc_polled_fd->GetName()); + grpc_ares_ev_driver_ref(ev_driver); + fdn->grpc_polled_fd->RegisterForOnWriteableLocked( + &fdn->write_closure); + fdn->writable_registered = true; + } + } + } + } + // Any remaining fds in ev_driver->fds were not returned by ares_getsock() and + // are therefore no longer in use, so they can be shut down and removed from + // the list. + while (ev_driver->fds != nullptr) { + fd_node* cur = ev_driver->fds; + ev_driver->fds = ev_driver->fds->next; + fd_node_shutdown_locked(cur, "c-ares fd shutdown"); + if (!cur->readable_registered && !cur->writable_registered) { + fd_node_destroy_locked(cur); + } else { + cur->next = new_list; + new_list = cur; + } + } + ev_driver->fds = new_list; + // If the ev driver has no working fd, all the tasks are done. + if (new_list == nullptr) { + ev_driver->working = false; + gpr_log(GPR_DEBUG, "ev driver stop working"); + } +} + +void grpc_ares_ev_driver_start_locked(grpc_ares_ev_driver* ev_driver) { + if (!ev_driver->working) { + ev_driver->working = true; + grpc_ares_notify_on_event_locked(ev_driver); + } +} + +#endif /* GRPC_ARES == 1 && defined(GRPC_POSIX_SOCKET_ARES_EV_DRIVER) */ diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.h b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.h index 27d1511d94..7002c8f95f 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.h +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.h @@ -22,6 +22,7 @@ #include <grpc/support/port_platform.h> #include <ares.h> +#include "src/core/lib/gprpp/abstract.h" #include "src/core/lib/iomgr/pollset_set.h" typedef struct grpc_ares_ev_driver grpc_ares_ev_driver; @@ -51,5 +52,40 @@ void grpc_ares_ev_driver_destroy_locked(grpc_ares_ev_driver* ev_driver); /* Shutdown all the grpc_fds used by \a ev_driver */ void grpc_ares_ev_driver_shutdown_locked(grpc_ares_ev_driver* ev_driver); +namespace grpc_core { + +/* A wrapped fd that integrates with the grpc iomgr of the current platform. + * A GrpcPolledFd knows how to create grpc platform-specific iomgr endpoints + * from "ares_socket_t" sockets, and then sign up for readability/writeability + * with that poller, and do shutdown and destruction. */ +class GrpcPolledFd { + public: + virtual ~GrpcPolledFd() {} + /* Called when c-ares library is interested and there's no pending callback */ + virtual void RegisterForOnReadableLocked(grpc_closure* read_closure) + GRPC_ABSTRACT; + /* Called when c-ares library is interested and there's no pending callback */ + virtual void RegisterForOnWriteableLocked(grpc_closure* write_closure) + GRPC_ABSTRACT; + /* Indicates if there is data left even after just being read from */ + virtual bool IsFdStillReadableLocked() GRPC_ABSTRACT; + /* Called once and only once. Must cause cancellation of any pending + * read/write callbacks. */ + virtual void ShutdownLocked(grpc_error* error) GRPC_ABSTRACT; + /* Get the underlying ares_socket_t that this was created from */ + virtual ares_socket_t GetWrappedAresSocketLocked() GRPC_ABSTRACT; + /* A unique name, for logging */ + virtual const char* GetName() GRPC_ABSTRACT; + + GRPC_ABSTRACT_BASE_CLASS +}; + +/* Creates a new wrapped fd for the current platform */ +GrpcPolledFd* NewGrpcPolledFdLocked(ares_socket_t as, + grpc_pollset_set* driver_pollset_set); +void ConfigureAresChannelLocked(ares_channel* channel); + +} // namespace grpc_core + #endif /* GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_RESOLVER_DNS_C_ARES_GRPC_ARES_EV_DRIVER_H \ */ diff --git a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver_posix.cc b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver_posix.cc index b73e979e9f..5db832baf8 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver_posix.cc +++ b/src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver_posix.cc @@ -36,286 +36,60 @@ #include "src/core/lib/iomgr/iomgr_internal.h" #include "src/core/lib/iomgr/sockaddr_utils.h" -typedef struct fd_node { - /** the owner of this fd node */ - grpc_ares_ev_driver* ev_driver; - /** a closure wrapping on_readable_locked, which should be - invoked when the grpc_fd in this node becomes readable. */ - grpc_closure read_closure; - /** a closure wrapping on_writable_locked, which should be - invoked when the grpc_fd in this node becomes writable. */ - grpc_closure write_closure; - /** next fd node in the list */ - struct fd_node* next; - - /** the grpc_fd owned by this fd node */ - grpc_fd* fd; - /** if the readable closure has been registered */ - bool readable_registered; - /** if the writable closure has been registered */ - bool writable_registered; - /** if the fd has been shutdown yet from grpc iomgr perspective */ - bool already_shutdown; -} fd_node; - -struct grpc_ares_ev_driver { - /** the ares_channel owned by this event driver */ - ares_channel channel; - /** pollset set for driving the IO events of the channel */ - grpc_pollset_set* pollset_set; - /** refcount of the event driver */ - gpr_refcount refs; - - /** combiner to synchronize c-ares and I/O callbacks on */ - grpc_combiner* combiner; - /** a list of grpc_fd that this event driver is currently using. */ - fd_node* fds; - /** is this event driver currently working? */ - bool working; - /** is this event driver being shut down */ - bool shutting_down; -}; - -static void grpc_ares_notify_on_event_locked(grpc_ares_ev_driver* ev_driver); - -static grpc_ares_ev_driver* grpc_ares_ev_driver_ref( - grpc_ares_ev_driver* ev_driver) { - gpr_log(GPR_DEBUG, "Ref ev_driver %" PRIuPTR, (uintptr_t)ev_driver); - gpr_ref(&ev_driver->refs); - return ev_driver; -} - -static void grpc_ares_ev_driver_unref(grpc_ares_ev_driver* ev_driver) { - gpr_log(GPR_DEBUG, "Unref ev_driver %" PRIuPTR, (uintptr_t)ev_driver); - if (gpr_unref(&ev_driver->refs)) { - gpr_log(GPR_DEBUG, "destroy ev_driver %" PRIuPTR, (uintptr_t)ev_driver); - GPR_ASSERT(ev_driver->fds == nullptr); - GRPC_COMBINER_UNREF(ev_driver->combiner, "free ares event driver"); - ares_destroy(ev_driver->channel); - gpr_free(ev_driver); +namespace grpc_core { + +class GrpcPolledFdPosix : public GrpcPolledFd { + public: + GrpcPolledFdPosix(ares_socket_t as, grpc_pollset_set* driver_pollset_set) + : as_(as) { + gpr_asprintf(&name_, "c-ares fd: %d", (int)as); + fd_ = grpc_fd_create((int)as, name_, false); + grpc_pollset_set_add_fd(driver_pollset_set, fd_); } -} - -static void fd_node_destroy_locked(fd_node* fdn) { - gpr_log(GPR_DEBUG, "delete fd: %d", grpc_fd_wrapped_fd(fdn->fd)); - GPR_ASSERT(!fdn->readable_registered); - GPR_ASSERT(!fdn->writable_registered); - GPR_ASSERT(fdn->already_shutdown); - /* c-ares library will close the fd inside grpc_fd. This fd may be picked up - immediately by another thread, and should not be closed by the following - grpc_fd_orphan. */ - int dummy_release_fd; - grpc_fd_orphan(fdn->fd, nullptr, &dummy_release_fd, "c-ares query finished"); - gpr_free(fdn); -} -static void fd_node_shutdown_locked(fd_node* fdn, const char* reason) { - if (!fdn->already_shutdown) { - fdn->already_shutdown = true; - grpc_fd_shutdown(fdn->fd, GRPC_ERROR_CREATE_FROM_STATIC_STRING(reason)); + ~GrpcPolledFdPosix() { + gpr_free(name_); + /* c-ares library will close the fd inside grpc_fd. This fd may be picked up + immediately by another thread, and should not be closed by the following + grpc_fd_orphan. */ + int dummy_release_fd; + grpc_fd_orphan(fd_, nullptr, &dummy_release_fd, "c-ares query finished"); } -} -grpc_error* grpc_ares_ev_driver_create_locked(grpc_ares_ev_driver** ev_driver, - grpc_pollset_set* pollset_set, - grpc_combiner* combiner) { - *ev_driver = static_cast<grpc_ares_ev_driver*>( - gpr_malloc(sizeof(grpc_ares_ev_driver))); - ares_options opts; - memset(&opts, 0, sizeof(opts)); - opts.flags |= ARES_FLAG_STAYOPEN; - int status = ares_init_options(&(*ev_driver)->channel, &opts, ARES_OPT_FLAGS); - gpr_log(GPR_DEBUG, "grpc_ares_ev_driver_create_locked"); - if (status != ARES_SUCCESS) { - char* err_msg; - gpr_asprintf(&err_msg, "Failed to init ares channel. C-ares error: %s", - ares_strerror(status)); - grpc_error* err = GRPC_ERROR_CREATE_FROM_COPIED_STRING(err_msg); - gpr_free(err_msg); - gpr_free(*ev_driver); - return err; + void RegisterForOnReadableLocked(grpc_closure* read_closure) override { + grpc_fd_notify_on_read(fd_, read_closure); } - (*ev_driver)->combiner = GRPC_COMBINER_REF(combiner, "ares event driver"); - gpr_ref_init(&(*ev_driver)->refs, 1); - (*ev_driver)->pollset_set = pollset_set; - (*ev_driver)->fds = nullptr; - (*ev_driver)->working = false; - (*ev_driver)->shutting_down = false; - return GRPC_ERROR_NONE; -} -void grpc_ares_ev_driver_destroy_locked(grpc_ares_ev_driver* ev_driver) { - // We mark the event driver as being shut down. If the event driver - // is working, grpc_ares_notify_on_event_locked will shut down the - // fds; if it's not working, there are no fds to shut down. - ev_driver->shutting_down = true; - grpc_ares_ev_driver_unref(ev_driver); -} + void RegisterForOnWriteableLocked(grpc_closure* write_closure) override { + grpc_fd_notify_on_write(fd_, write_closure); + } -void grpc_ares_ev_driver_shutdown_locked(grpc_ares_ev_driver* ev_driver) { - ev_driver->shutting_down = true; - fd_node* fn = ev_driver->fds; - while (fn != nullptr) { - fd_node_shutdown_locked(fn, "grpc_ares_ev_driver_shutdown"); - fn = fn->next; + bool IsFdStillReadableLocked() override { + size_t bytes_available = 0; + return ioctl(grpc_fd_wrapped_fd(fd_), FIONREAD, &bytes_available) == 0 && + bytes_available > 0; } -} -// Search fd in the fd_node list head. This is an O(n) search, the max possible -// value of n is ARES_GETSOCK_MAXNUM (16). n is typically 1 - 2 in our tests. -static fd_node* pop_fd_node_locked(fd_node** head, int fd) { - fd_node dummy_head; - dummy_head.next = *head; - fd_node* node = &dummy_head; - while (node->next != nullptr) { - if (grpc_fd_wrapped_fd(node->next->fd) == fd) { - fd_node* ret = node->next; - node->next = node->next->next; - *head = dummy_head.next; - return ret; - } - node = node->next; + void ShutdownLocked(grpc_error* error) override { + grpc_fd_shutdown(fd_, error); } - return nullptr; -} -/* Check if \a fd is still readable */ -static bool grpc_ares_is_fd_still_readable_locked( - grpc_ares_ev_driver* ev_driver, int fd) { - size_t bytes_available = 0; - return ioctl(fd, FIONREAD, &bytes_available) == 0 && bytes_available > 0; -} + ares_socket_t GetWrappedAresSocketLocked() override { return as_; } -static void on_readable_locked(void* arg, grpc_error* error) { - fd_node* fdn = static_cast<fd_node*>(arg); - grpc_ares_ev_driver* ev_driver = fdn->ev_driver; - const int fd = grpc_fd_wrapped_fd(fdn->fd); - fdn->readable_registered = false; - gpr_log(GPR_DEBUG, "readable on %d", fd); - if (error == GRPC_ERROR_NONE) { - do { - ares_process_fd(ev_driver->channel, fd, ARES_SOCKET_BAD); - } while (grpc_ares_is_fd_still_readable_locked(ev_driver, fd)); - } else { - // If error is not GRPC_ERROR_NONE, it means the fd has been shutdown or - // timed out. The pending lookups made on this ev_driver will be cancelled - // by the following ares_cancel() and the on_done callbacks will be invoked - // with a status of ARES_ECANCELLED. The remaining file descriptors in this - // ev_driver will be cleaned up in the follwing - // grpc_ares_notify_on_event_locked(). - ares_cancel(ev_driver->channel); - } - grpc_ares_notify_on_event_locked(ev_driver); - grpc_ares_ev_driver_unref(ev_driver); -} + const char* GetName() override { return name_; } -static void on_writable_locked(void* arg, grpc_error* error) { - fd_node* fdn = static_cast<fd_node*>(arg); - grpc_ares_ev_driver* ev_driver = fdn->ev_driver; - const int fd = grpc_fd_wrapped_fd(fdn->fd); - fdn->writable_registered = false; - gpr_log(GPR_DEBUG, "writable on %d", fd); - if (error == GRPC_ERROR_NONE) { - ares_process_fd(ev_driver->channel, ARES_SOCKET_BAD, fd); - } else { - // If error is not GRPC_ERROR_NONE, it means the fd has been shutdown or - // timed out. The pending lookups made on this ev_driver will be cancelled - // by the following ares_cancel() and the on_done callbacks will be invoked - // with a status of ARES_ECANCELLED. The remaining file descriptors in this - // ev_driver will be cleaned up in the follwing - // grpc_ares_notify_on_event_locked(). - ares_cancel(ev_driver->channel); - } - grpc_ares_notify_on_event_locked(ev_driver); - grpc_ares_ev_driver_unref(ev_driver); -} + char* name_; + ares_socket_t as_; + grpc_fd* fd_; +}; -ares_channel* grpc_ares_ev_driver_get_channel_locked( - grpc_ares_ev_driver* ev_driver) { - return &ev_driver->channel; +GrpcPolledFd* NewGrpcPolledFdLocked(ares_socket_t as, + grpc_pollset_set* driver_pollset_set) { + return grpc_core::New<GrpcPolledFdPosix>(as, driver_pollset_set); } -// Get the file descriptors used by the ev_driver's ares channel, register -// driver_closure with these filedescriptors. -static void grpc_ares_notify_on_event_locked(grpc_ares_ev_driver* ev_driver) { - fd_node* new_list = nullptr; - if (!ev_driver->shutting_down) { - ares_socket_t socks[ARES_GETSOCK_MAXNUM]; - int socks_bitmask = - ares_getsock(ev_driver->channel, socks, ARES_GETSOCK_MAXNUM); - for (size_t i = 0; i < ARES_GETSOCK_MAXNUM; i++) { - if (ARES_GETSOCK_READABLE(socks_bitmask, i) || - ARES_GETSOCK_WRITABLE(socks_bitmask, i)) { - fd_node* fdn = pop_fd_node_locked(&ev_driver->fds, socks[i]); - // Create a new fd_node if sock[i] is not in the fd_node list. - if (fdn == nullptr) { - char* fd_name; - gpr_asprintf(&fd_name, "ares_ev_driver-%" PRIuPTR, i); - fdn = static_cast<fd_node*>(gpr_malloc(sizeof(fd_node))); - gpr_log(GPR_DEBUG, "new fd: %d", socks[i]); - fdn->fd = grpc_fd_create(socks[i], fd_name, false); - fdn->ev_driver = ev_driver; - fdn->readable_registered = false; - fdn->writable_registered = false; - fdn->already_shutdown = false; - GRPC_CLOSURE_INIT(&fdn->read_closure, on_readable_locked, fdn, - grpc_combiner_scheduler(ev_driver->combiner)); - GRPC_CLOSURE_INIT(&fdn->write_closure, on_writable_locked, fdn, - grpc_combiner_scheduler(ev_driver->combiner)); - grpc_pollset_set_add_fd(ev_driver->pollset_set, fdn->fd); - gpr_free(fd_name); - } - fdn->next = new_list; - new_list = fdn; - // Register read_closure if the socket is readable and read_closure has - // not been registered with this socket. - if (ARES_GETSOCK_READABLE(socks_bitmask, i) && - !fdn->readable_registered) { - grpc_ares_ev_driver_ref(ev_driver); - gpr_log(GPR_DEBUG, "notify read on: %d", grpc_fd_wrapped_fd(fdn->fd)); - grpc_fd_notify_on_read(fdn->fd, &fdn->read_closure); - fdn->readable_registered = true; - } - // Register write_closure if the socket is writable and write_closure - // has not been registered with this socket. - if (ARES_GETSOCK_WRITABLE(socks_bitmask, i) && - !fdn->writable_registered) { - gpr_log(GPR_DEBUG, "notify write on: %d", - grpc_fd_wrapped_fd(fdn->fd)); - grpc_ares_ev_driver_ref(ev_driver); - grpc_fd_notify_on_write(fdn->fd, &fdn->write_closure); - fdn->writable_registered = true; - } - } - } - } - // Any remaining fds in ev_driver->fds were not returned by ares_getsock() and - // are therefore no longer in use, so they can be shut down and removed from - // the list. - while (ev_driver->fds != nullptr) { - fd_node* cur = ev_driver->fds; - ev_driver->fds = ev_driver->fds->next; - fd_node_shutdown_locked(cur, "c-ares fd shutdown"); - if (!cur->readable_registered && !cur->writable_registered) { - fd_node_destroy_locked(cur); - } else { - cur->next = new_list; - new_list = cur; - } - } - ev_driver->fds = new_list; - // If the ev driver has no working fd, all the tasks are done. - if (new_list == nullptr) { - ev_driver->working = false; - gpr_log(GPR_DEBUG, "ev driver stop working"); - } -} +void ConfigureAresChannelLocked(ares_channel* channel) {} -void grpc_ares_ev_driver_start_locked(grpc_ares_ev_driver* ev_driver) { - if (!ev_driver->working) { - ev_driver->working = true; - grpc_ares_notify_on_event_locked(ev_driver); - } -} +} // namespace grpc_core #endif /* GRPC_ARES == 1 && defined(GRPC_POSIX_SOCKET_ARES_EV_DRIVER) */ diff --git a/src/core/ext/filters/client_channel/resolver/dns/native/dns_resolver.cc b/src/core/ext/filters/client_channel/resolver/dns/native/dns_resolver.cc index 25e92a078d..fae4c33a17 100644 --- a/src/core/ext/filters/client_channel/resolver/dns/native/dns_resolver.cc +++ b/src/core/ext/filters/client_channel/resolver/dns/native/dns_resolver.cc @@ -116,9 +116,10 @@ NativeDnsResolver::NativeDnsResolver(const ResolverArgs& args) if (path[0] == '/') ++path; name_to_resolve_ = gpr_strdup(path); channel_args_ = grpc_channel_args_copy(args.args); - min_time_between_resolutions_ = grpc_channel_args_get_integer( - args.args, GRPC_ARG_DNS_MIN_TIME_BETWEEN_RESOLUTIONS_MS, - {1000, 0, INT_MAX}); + const grpc_arg* arg = grpc_channel_args_find( + args.args, GRPC_ARG_DNS_MIN_TIME_BETWEEN_RESOLUTIONS_MS); + min_time_between_resolutions_ = + grpc_channel_arg_get_integer(arg, {1000, 0, INT_MAX}); interested_parties_ = grpc_pollset_set_create(); if (args.pollset_set != nullptr) { grpc_pollset_set_add_pollset_set(interested_parties_, args.pollset_set); diff --git a/src/core/ext/filters/client_channel/resolver/fake/fake_resolver.cc b/src/core/ext/filters/client_channel/resolver/fake/fake_resolver.cc index 9250dbd6b3..99a33f2277 100644 --- a/src/core/ext/filters/client_channel/resolver/fake/fake_resolver.cc +++ b/src/core/ext/filters/client_channel/resolver/fake/fake_resolver.cc @@ -252,15 +252,20 @@ static const grpc_arg_pointer_vtable response_generator_arg_vtable = { grpc_arg FakeResolverResponseGenerator::MakeChannelArg( FakeResolverResponseGenerator* generator) { - return grpc_channel_arg_pointer_create( - const_cast<char*>(GRPC_ARG_FAKE_RESOLVER_RESPONSE_GENERATOR), generator, - &response_generator_arg_vtable); + grpc_arg arg; + arg.type = GRPC_ARG_POINTER; + arg.key = (char*)GRPC_ARG_FAKE_RESOLVER_RESPONSE_GENERATOR; + arg.value.pointer.p = generator; + arg.value.pointer.vtable = &response_generator_arg_vtable; + return arg; } FakeResolverResponseGenerator* FakeResolverResponseGenerator::GetFromArgs( const grpc_channel_args* args) { - return grpc_channel_args_get_pointer<FakeResolverResponseGenerator>( - args, GRPC_ARG_FAKE_RESOLVER_RESPONSE_GENERATOR); + const grpc_arg* arg = + grpc_channel_args_find(args, GRPC_ARG_FAKE_RESOLVER_RESPONSE_GENERATOR); + if (arg == nullptr || arg->type != GRPC_ARG_POINTER) return nullptr; + return static_cast<FakeResolverResponseGenerator*>(arg->value.pointer.p); } // diff --git a/src/core/ext/filters/client_channel/subchannel.cc b/src/core/ext/filters/client_channel/subchannel.cc index 5db5d1f6c0..f010002ab9 100644 --- a/src/core/ext/filters/client_channel/subchannel.cc +++ b/src/core/ext/filters/client_channel/subchannel.cc @@ -736,8 +736,9 @@ void grpc_get_subchannel_address_arg(const grpc_channel_args* args, } const char* grpc_get_subchannel_address_uri_arg(const grpc_channel_args* args) { - const char* addr_str = - grpc_channel_args_get_string(args, GRPC_ARG_SUBCHANNEL_ADDRESS); + const grpc_arg* addr_arg = + grpc_channel_args_find(args, GRPC_ARG_SUBCHANNEL_ADDRESS); + const char* addr_str = grpc_channel_arg_get_string(addr_arg); GPR_ASSERT(addr_str != nullptr); // Should have been set by LB policy. return addr_str; } diff --git a/src/core/ext/filters/deadline/deadline_filter.cc b/src/core/ext/filters/deadline/deadline_filter.cc index caa547204b..e0a41a3637 100644 --- a/src/core/ext/filters/deadline/deadline_filter.cc +++ b/src/core/ext/filters/deadline/deadline_filter.cc @@ -358,8 +358,8 @@ const grpc_channel_filter grpc_server_deadline_filter = { }; bool grpc_deadline_checking_enabled(const grpc_channel_args* channel_args) { - return grpc_channel_args_get_bool( - channel_args, GRPC_ARG_ENABLE_DEADLINE_CHECKS, + return grpc_channel_arg_get_bool( + grpc_channel_args_find(channel_args, GRPC_ARG_ENABLE_DEADLINE_CHECKS), !grpc_channel_args_want_minimal_stack(channel_args)); } diff --git a/src/core/ext/filters/http/client/http_client_filter.cc b/src/core/ext/filters/http/client/http_client_filter.cc index fe7f2ba705..ae94ce47b9 100644 --- a/src/core/ext/filters/http/client/http_client_filter.cc +++ b/src/core/ext/filters/http/client/http_client_filter.cc @@ -23,7 +23,6 @@ #include <stdint.h> #include <string.h> #include "src/core/ext/filters/http/client/http_client_filter.h" -#include "src/core/lib/channel/channel_args.h" #include "src/core/lib/gpr/string.h" #include "src/core/lib/gprpp/manual_constructor.h" #include "src/core/lib/profiling/timers.h" @@ -436,43 +435,64 @@ static void destroy_call_elem(grpc_call_element* elem, const grpc_call_final_info* final_info, grpc_closure* ignored) {} -static grpc_mdelem scheme_from_args(const grpc_channel_args* channel_args) { +static grpc_mdelem scheme_from_args(const grpc_channel_args* args) { + unsigned i; + size_t j; grpc_mdelem valid_schemes[] = {GRPC_MDELEM_SCHEME_HTTP, GRPC_MDELEM_SCHEME_HTTPS}; - char* scheme = - grpc_channel_args_get_string(channel_args, GRPC_ARG_HTTP2_SCHEME); - if (scheme != nullptr) { - for (size_t i = 0; i < GPR_ARRAY_SIZE(valid_schemes); i++) { - if (0 == grpc_slice_str_cmp(GRPC_MDVALUE(valid_schemes[i]), scheme)) { - return valid_schemes[i]; + if (args != nullptr) { + for (i = 0; i < args->num_args; ++i) { + if (args->args[i].type == GRPC_ARG_STRING && + strcmp(args->args[i].key, GRPC_ARG_HTTP2_SCHEME) == 0) { + for (j = 0; j < GPR_ARRAY_SIZE(valid_schemes); j++) { + if (0 == grpc_slice_str_cmp(GRPC_MDVALUE(valid_schemes[j]), + args->args[i].value.string)) { + return valid_schemes[j]; + } + } } } } return GRPC_MDELEM_SCHEME_HTTP; } -static size_t max_payload_size_from_args( - const grpc_channel_args* channel_args) { - return grpc_channel_args_get_integer( - channel_args, GRPC_ARG_MAX_PAYLOAD_SIZE_FOR_GET, - {kMaxPayloadSizeForGet, 0, kMaxPayloadSizeForGet}); +static size_t max_payload_size_from_args(const grpc_channel_args* args) { + if (args != nullptr) { + for (size_t i = 0; i < args->num_args; ++i) { + if (0 == strcmp(args->args[i].key, GRPC_ARG_MAX_PAYLOAD_SIZE_FOR_GET)) { + if (args->args[i].type != GRPC_ARG_INTEGER) { + gpr_log(GPR_ERROR, "%s: must be an integer", + GRPC_ARG_MAX_PAYLOAD_SIZE_FOR_GET); + } else { + return static_cast<size_t>(args->args[i].value.integer); + } + } + } + } + return kMaxPayloadSizeForGet; } static grpc_slice user_agent_from_args(const grpc_channel_args* args, const char* transport_name) { gpr_strvec v; + size_t i; int is_first = 1; char* tmp; grpc_slice result; gpr_strvec_init(&v); - char* user_agent_str = - grpc_channel_args_get_string(args, GRPC_ARG_PRIMARY_USER_AGENT_STRING); - if (user_agent_str != nullptr) { - if (!is_first) gpr_strvec_add(&v, gpr_strdup(" ")); - is_first = 0; - gpr_strvec_add(&v, gpr_strdup(user_agent_str)); + 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; %s; %s)", is_first ? "" : " ", @@ -481,11 +501,17 @@ static grpc_slice user_agent_from_args(const grpc_channel_args* args, is_first = 0; gpr_strvec_add(&v, tmp); - user_agent_str = - grpc_channel_args_get_string(args, GRPC_ARG_SECONDARY_USER_AGENT_STRING); - if (user_agent_str != nullptr) { - gpr_strvec_add(&v, gpr_strdup(" ")); - gpr_strvec_add(&v, gpr_strdup(user_agent_str)); + 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, nullptr); diff --git a/src/core/ext/filters/http/http_filters_plugin.cc b/src/core/ext/filters/http/http_filters_plugin.cc index d3cbe50016..f03fa0141d 100644 --- a/src/core/ext/filters/http/http_filters_plugin.cc +++ b/src/core/ext/filters/http/http_filters_plugin.cc @@ -48,8 +48,8 @@ static bool maybe_add_optional_filter(grpc_channel_stack_builder* builder, optional_filter* filtarg = static_cast<optional_filter*>(arg); const grpc_channel_args* channel_args = grpc_channel_stack_builder_get_channel_arguments(builder); - bool enable = grpc_channel_args_get_bool( - channel_args, filtarg->control_channel_arg, + bool enable = grpc_channel_arg_get_bool( + grpc_channel_args_find(channel_args, filtarg->control_channel_arg), !grpc_channel_args_want_minimal_stack(channel_args)); return enable ? grpc_channel_stack_builder_prepend_filter( builder, filtarg->filter, nullptr, nullptr) diff --git a/src/core/ext/filters/http/server/http_server_filter.h b/src/core/ext/filters/http/server/http_server_filter.h index 4eb130b1fd..e4e0546f6d 100644 --- a/src/core/ext/filters/http/server/http_server_filter.h +++ b/src/core/ext/filters/http/server/http_server_filter.h @@ -23,7 +23,7 @@ #include "src/core/lib/channel/channel_stack.h" -/* Processes metadata on the client side for HTTP2 transports */ +/* Processes metadata on the server side for HTTP2 transports */ extern const grpc_channel_filter grpc_http_server_filter; #endif /* GRPC_CORE_EXT_FILTERS_HTTP_SERVER_HTTP_SERVER_FILTER_H */ diff --git a/src/core/ext/filters/load_reporting/server_load_reporting_plugin.cc b/src/core/ext/filters/load_reporting/server_load_reporting_plugin.cc index 73bcf242d8..667c0c56ef 100644 --- a/src/core/ext/filters/load_reporting/server_load_reporting_plugin.cc +++ b/src/core/ext/filters/load_reporting/server_load_reporting_plugin.cc @@ -33,7 +33,8 @@ #include "src/core/lib/surface/channel_init.h" static bool is_load_reporting_enabled(const grpc_channel_args* a) { - return grpc_channel_args_get_bool(a, GRPC_ARG_ENABLE_LOAD_REPORTING, false); + return grpc_channel_arg_get_bool( + grpc_channel_args_find(a, GRPC_ARG_ENABLE_LOAD_REPORTING), false); } static bool maybe_add_server_load_reporting_filter( diff --git a/src/core/ext/filters/max_age/max_age_filter.cc b/src/core/ext/filters/max_age/max_age_filter.cc index bbe4998f00..1fe8288bd0 100644 --- a/src/core/ext/filters/max_age/max_age_filter.cc +++ b/src/core/ext/filters/max_age/max_age_filter.cc @@ -519,12 +519,13 @@ static bool maybe_add_max_age_filter(grpc_channel_stack_builder* builder, void* arg) { const grpc_channel_args* channel_args = grpc_channel_stack_builder_get_channel_arguments(builder); - bool enable = grpc_channel_args_get_integer( - channel_args, GRPC_ARG_MAX_CONNECTION_AGE_MS, - MAX_CONNECTION_AGE_INTEGER_OPTIONS) != INT_MAX || - grpc_channel_args_get_integer( - channel_args, GRPC_ARG_MAX_CONNECTION_IDLE_MS, - MAX_CONNECTION_IDLE_INTEGER_OPTIONS) != INT_MAX; + bool enable = + grpc_channel_arg_get_integer( + grpc_channel_args_find(channel_args, GRPC_ARG_MAX_CONNECTION_AGE_MS), + MAX_CONNECTION_AGE_INTEGER_OPTIONS) != INT_MAX || + grpc_channel_arg_get_integer( + grpc_channel_args_find(channel_args, GRPC_ARG_MAX_CONNECTION_IDLE_MS), + MAX_CONNECTION_IDLE_INTEGER_OPTIONS) != INT_MAX; if (enable) { return grpc_channel_stack_builder_prepend_filter( builder, &grpc_max_age_filter, nullptr, nullptr); diff --git a/src/core/ext/filters/message_size/message_size_filter.cc b/src/core/ext/filters/message_size/message_size_filter.cc index 9815c93145..c7fc3f2e62 100644 --- a/src/core/ext/filters/message_size/message_size_filter.cc +++ b/src/core/ext/filters/message_size/message_size_filter.cc @@ -254,8 +254,9 @@ static grpc_error* init_channel_elem(grpc_channel_element* elem, channel_data* chand = static_cast<channel_data*>(elem->channel_data); chand->limits = get_message_size_limits(args->channel_args); // Get method config table from channel args. - const char* service_config_str = - grpc_channel_args_get_string(args->channel_args, GRPC_ARG_SERVICE_CONFIG); + const grpc_arg* channel_arg = + grpc_channel_args_find(args->channel_args, GRPC_ARG_SERVICE_CONFIG); + const char* service_config_str = grpc_channel_arg_get_string(channel_arg); if (service_config_str != nullptr) { grpc_core::UniquePtr<grpc_core::ServiceConfig> service_config = grpc_core::ServiceConfig::Create(service_config_str); |