Cancel still-active c-ares queries after 10 seconds to avoid chance of deadlock

include/grpc/impl/codegen/grpc_types.h

@@ -350,6 +350,11 @@ typedef struct {
 /** If set, inhibits health checking (which may be enabled via the
  *  service config.) */
 #define GRPC_ARG_INHIBIT_HEALTH_CHECKING "grpc.inhibit_health_checking"
+/** If set, determines the number of milliseconds that the c-ares based
+ * DNS resolver will wait on queries before cancelling them. The default value
+ * is 10000. Setting this to "0" will disable c-ares query timeouts
+ * entirely. */
+#define GRPC_ARG_DNS_ARES_QUERY_TIMEOUT_MS "grpc.dns_ares_query_timeout"
 /** \} */
 
 /** Result of a grpc call. If the caller satisfies the prerequisites of a

src/core/ext/filters/client_channel/resolver/dns/c_ares/dns_resolver_ares.cc

@@ -122,6 +122,8 @@ class AresDnsResolver : public Resolver {
   char* service_config_json_ = nullptr;
   // has shutdown been initiated
   bool shutdown_initiated_ = false;
+  // timeout in milliseconds for active DNS queries
+  int query_timeout_ms_;
 };
 
 AresDnsResolver::AresDnsResolver(const ResolverArgs& args)
@@ -159,6 +161,11 @@ AresDnsResolver::AresDnsResolver(const ResolverArgs& args)
                     grpc_combiner_scheduler(combiner()));
   GRPC_CLOSURE_INIT(&on_resolved_, OnResolvedLocked, this,
                     grpc_combiner_scheduler(combiner()));
+  const grpc_arg* query_timeout_ms_arg =
+      grpc_channel_args_find(channel_args_, GRPC_ARG_DNS_ARES_QUERY_TIMEOUT_MS);
+  query_timeout_ms_ = grpc_channel_arg_get_integer(
+      query_timeout_ms_arg,
+      {GRPC_DNS_ARES_DEFAULT_QUERY_TIMEOUT_MS, 0, INT_MAX});
 }
 
 AresDnsResolver::~AresDnsResolver() {
@@ -410,7 +417,8 @@ void AresDnsResolver::StartResolvingLocked() {
   pending_request_ = grpc_dns_lookup_ares_locked(
       dns_server_, name_to_resolve_, kDefaultPort, interested_parties_,
       &on_resolved_, &lb_addresses_, true /* check_grpclb */,
-      request_service_config_ ? &service_config_json_ : nullptr, combiner());
+      request_service_config_ ? &service_config_json_ : nullptr,
+      query_timeout_ms_, combiner());
   last_resolution_timestamp_ = grpc_core::ExecCtx::Get()->Now();
 }
 

src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.cc

@@ -33,6 +33,7 @@
 #include "src/core/lib/gpr/string.h"
 #include "src/core/lib/iomgr/iomgr_internal.h"
 #include "src/core/lib/iomgr/sockaddr_utils.h"
+#include "src/core/lib/iomgr/timer.h"
 
 typedef struct fd_node {
   /** the owner of this fd node */
@@ -76,6 +77,12 @@ struct grpc_ares_ev_driver {
   grpc_ares_request* request;
   /** Owned by the ev_driver. Creates new GrpcPolledFd's */
   grpc_core::UniquePtr<grpc_core::GrpcPolledFdFactory> polled_fd_factory;
+  /** query timeout in milliseconds */
+  int query_timeout_ms;
+  /** alarm to cancel active queries */
+  grpc_timer query_timeout;
+  /** cancels queries on a timeout */
+  grpc_closure on_timeout_locked;
 };
 
 static void grpc_ares_notify_on_event_locked(grpc_ares_ev_driver* ev_driver);
@@ -116,8 +123,11 @@ static void fd_node_shutdown_locked(fd_node* fdn, const char* reason) {
   }
 }
 
+static void on_timeout_locked(void* arg, grpc_error* error);
+
 grpc_error* grpc_ares_ev_driver_create_locked(grpc_ares_ev_driver** ev_driver,
                                               grpc_pollset_set* pollset_set,
+                                              int query_timeout_ms,
                                               grpc_combiner* combiner,
                                               grpc_ares_request* request) {
   *ev_driver = grpc_core::New<grpc_ares_ev_driver>();
@@ -146,6 +156,9 @@ grpc_error* grpc_ares_ev_driver_create_locked(grpc_ares_ev_driver** ev_driver,
       grpc_core::NewGrpcPolledFdFactory((*ev_driver)->combiner);
   (*ev_driver)
       ->polled_fd_factory->ConfigureAresChannelLocked((*ev_driver)->channel);
+  GRPC_CLOSURE_INIT(&(*ev_driver)->on_timeout_locked, on_timeout_locked,
+                    *ev_driver, grpc_combiner_scheduler(combiner));
+  (*ev_driver)->query_timeout_ms = query_timeout_ms;
   return GRPC_ERROR_NONE;
 }
 
@@ -155,6 +168,7 @@ void grpc_ares_ev_driver_on_queries_complete_locked(
   // 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_timer_cancel(&ev_driver->query_timeout);
   grpc_ares_ev_driver_unref(ev_driver);
 }
 
@@ -185,6 +199,17 @@ static fd_node* pop_fd_node_locked(fd_node** head, ares_socket_t as) {
   return nullptr;
 }
 
+static void on_timeout_locked(void* arg, grpc_error* error) {
+  grpc_ares_ev_driver* driver = static_cast<grpc_ares_ev_driver*>(arg);
+  GRPC_CARES_TRACE_LOG(
+      "ev_driver=%p on_timeout_locked. driver->shutting_down=%d. err=%s",
+      driver, driver->shutting_down, grpc_error_string(error));
+  if (!driver->shutting_down && error == GRPC_ERROR_NONE) {
+    grpc_ares_ev_driver_shutdown_locked(driver);
+  }
+  grpc_ares_ev_driver_unref(driver);
+}
+
 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;
@@ -314,6 +339,17 @@ 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);
+    grpc_millis timeout =
+        ev_driver->query_timeout_ms == 0
+            ? GRPC_MILLIS_INF_FUTURE
+            : ev_driver->query_timeout_ms + grpc_core::ExecCtx::Get()->Now();
+    GRPC_CARES_TRACE_LOG(
+        "ev_driver=%p grpc_ares_ev_driver_start_locked. timeout in %" PRId64
+        " ms",
+        ev_driver, timeout);
+    grpc_ares_ev_driver_ref(ev_driver);
+    grpc_timer_init(&ev_driver->query_timeout, timeout,
+                    &ev_driver->on_timeout_locked);
   }
 }
 

src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_ev_driver.h

@@ -43,6 +43,7 @@ ares_channel* grpc_ares_ev_driver_get_channel_locked(
    created successfully. */
 grpc_error* grpc_ares_ev_driver_create_locked(grpc_ares_ev_driver** ev_driver,
                                               grpc_pollset_set* pollset_set,
+                                              int query_timeout_ms,
                                               grpc_combiner* combiner,
                                               grpc_ares_request* request);
 

src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.cc

@@ -359,7 +359,7 @@ done:
 void grpc_dns_lookup_ares_continue_after_check_localhost_and_ip_literals_locked(
     grpc_ares_request* r, const char* dns_server, const char* name,
     const char* default_port, grpc_pollset_set* interested_parties,
-    bool check_grpclb, grpc_combiner* combiner) {
+    bool check_grpclb, int query_timeout_ms, grpc_combiner* combiner) {
   grpc_error* error = GRPC_ERROR_NONE;
   grpc_ares_hostbyname_request* hr = nullptr;
   ares_channel* channel = nullptr;
@@ -388,7 +388,7 @@ void grpc_dns_lookup_ares_continue_after_check_localhost_and_ip_literals_locked(
     port = gpr_strdup(default_port);
   }
   error = grpc_ares_ev_driver_create_locked(&r->ev_driver, interested_parties,
-                                            combiner, r);
+                                            query_timeout_ms, combiner, r);
   if (error != GRPC_ERROR_NONE) goto error_cleanup;
   channel = grpc_ares_ev_driver_get_channel_locked(r->ev_driver);
   // If dns_server is specified, use it.
@@ -522,7 +522,7 @@ static grpc_ares_request* grpc_dns_lookup_ares_locked_impl(
     const char* dns_server, const char* name, const char* default_port,
     grpc_pollset_set* interested_parties, grpc_closure* on_done,
     grpc_lb_addresses** addrs, bool check_grpclb, char** service_config_json,
-    grpc_combiner* combiner) {
+    int query_timeout_ms, grpc_combiner* combiner) {
   grpc_ares_request* r =
       static_cast<grpc_ares_request*>(gpr_zalloc(sizeof(grpc_ares_request)));
   r->ev_driver = nullptr;
@@ -546,7 +546,7 @@ static grpc_ares_request* grpc_dns_lookup_ares_locked_impl(
   // Look up name using c-ares lib.
   grpc_dns_lookup_ares_continue_after_check_localhost_and_ip_literals_locked(
       r, dns_server, name, default_port, interested_parties, check_grpclb,
-      combiner);
+      query_timeout_ms, combiner);
   return r;
 }
 
@@ -554,6 +554,7 @@ grpc_ares_request* (*grpc_dns_lookup_ares_locked)(
     const char* dns_server, const char* name, const char* default_port,
     grpc_pollset_set* interested_parties, grpc_closure* on_done,
     grpc_lb_addresses** addrs, bool check_grpclb, char** service_config_json,
+    int query_timeout_ms,
     grpc_combiner* combiner) = grpc_dns_lookup_ares_locked_impl;
 
 static void grpc_cancel_ares_request_locked_impl(grpc_ares_request* r) {
@@ -648,7 +649,8 @@ static void grpc_resolve_address_invoke_dns_lookup_ares_locked(
   r->ares_request = grpc_dns_lookup_ares_locked(
       nullptr /* dns_server */, r->name, r->default_port, r->interested_parties,
       &r->on_dns_lookup_done_locked, &r->lb_addrs, false /* check_grpclb */,
-      nullptr /* service_config_json */, r->combiner);
+      nullptr /* service_config_json */, GRPC_DNS_ARES_DEFAULT_QUERY_TIMEOUT_MS,
+      r->combiner);
 }
 
 static void grpc_resolve_address_ares_impl(const char* name,

src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper.h

@@ -26,6 +26,8 @@
 #include "src/core/lib/iomgr/polling_entity.h"
 #include "src/core/lib/iomgr/resolve_address.h"
 
+#define GRPC_DNS_ARES_DEFAULT_QUERY_TIMEOUT_MS 10000
+
 extern grpc_core::TraceFlag grpc_trace_cares_address_sorting;
 
 extern grpc_core::TraceFlag grpc_trace_cares_resolver;
@@ -60,7 +62,7 @@ extern grpc_ares_request* (*grpc_dns_lookup_ares_locked)(
     const char* dns_server, const char* name, const char* default_port,
     grpc_pollset_set* interested_parties, grpc_closure* on_done,
     grpc_lb_addresses** addresses, bool check_grpclb,
-    char** service_config_json, grpc_combiner* combiner);
+    char** service_config_json, int query_timeout_ms, grpc_combiner* combiner);
 
 /* Cancel the pending grpc_ares_request \a request */
 extern void (*grpc_cancel_ares_request_locked)(grpc_ares_request* request);

src/core/ext/filters/client_channel/resolver/dns/c_ares/grpc_ares_wrapper_fallback.cc

@@ -30,7 +30,7 @@ static grpc_ares_request* grpc_dns_lookup_ares_locked_impl(
     const char* dns_server, const char* name, const char* default_port,
     grpc_pollset_set* interested_parties, grpc_closure* on_done,
     grpc_lb_addresses** addrs, bool check_grpclb, char** service_config_json,
-    grpc_combiner* combiner) {
+    int query_timeout_ms, grpc_combiner* combiner) {
   return NULL;
 }
 
@@ -38,6 +38,7 @@ grpc_ares_request* (*grpc_dns_lookup_ares_locked)(
     const char* dns_server, const char* name, const char* default_port,
     grpc_pollset_set* interested_parties, grpc_closure* on_done,
     grpc_lb_addresses** addrs, bool check_grpclb, char** service_config_json,
+    int query_timeout_ms,
     grpc_combiner* combiner) = grpc_dns_lookup_ares_locked_impl;
 
 static void grpc_cancel_ares_request_locked_impl(grpc_ares_request* r) {}

src/core/lib/iomgr/resolve_address.h

@@ -65,7 +65,7 @@ void grpc_set_resolver_impl(grpc_address_resolver_vtable* vtable);
 
 /* Asynchronously resolve addr. Use default_port if a port isn't designated
    in addr, otherwise use the port in addr. */
-/* TODO(ctiller): add a timeout here */
+/* TODO(apolcyn): add a timeout here */
 void grpc_resolve_address(const char* addr, const char* default_port,
                           grpc_pollset_set* interested_parties,
                           grpc_closure* on_done,

test/core/client_channel/resolvers/dns_resolver_connectivity_test.cc

@@ -64,7 +64,7 @@ static grpc_ares_request* my_dns_lookup_ares_locked(
     const char* dns_server, const char* addr, const char* default_port,
     grpc_pollset_set* interested_parties, grpc_closure* on_done,
     grpc_lb_addresses** lb_addrs, bool check_grpclb, char** service_config_json,
-    grpc_combiner* combiner) {
+    int query_timeout_ms, grpc_combiner* combiner) {
   gpr_mu_lock(&g_mu);
   GPR_ASSERT(0 == strcmp("test", addr));
   grpc_error* error = GRPC_ERROR_NONE;

test/core/client_channel/resolvers/dns_resolver_cooldown_test.cc

@@ -41,7 +41,7 @@ static grpc_ares_request* (*g_default_dns_lookup_ares_locked)(
     const char* dns_server, const char* name, const char* default_port,
     grpc_pollset_set* interested_parties, grpc_closure* on_done,
     grpc_lb_addresses** addrs, bool check_grpclb, char** service_config_json,
-    grpc_combiner* combiner);
+    int query_timeout_ms, grpc_combiner* combiner);
 
 // Counter incremented by test_resolve_address_impl indicating the number of
 // times a system-level resolution has happened.
@@ -91,10 +91,10 @@ static grpc_ares_request* test_dns_lookup_ares_locked(
     const char* dns_server, const char* name, const char* default_port,
     grpc_pollset_set* interested_parties, grpc_closure* on_done,
     grpc_lb_addresses** addrs, bool check_grpclb, char** service_config_json,
-    grpc_combiner* combiner) {
+    int query_timeout_ms, grpc_combiner* combiner) {
   grpc_ares_request* result = g_default_dns_lookup_ares_locked(
       dns_server, name, default_port, g_iomgr_args.pollset_set, on_done, addrs,
-      check_grpclb, service_config_json, combiner);
+      check_grpclb, service_config_json, query_timeout_ms, combiner);
   ++g_resolution_count;
   static grpc_millis last_resolution_time = 0;
   if (last_resolution_time == 0) {

test/core/end2end/fuzzers/api_fuzzer.cc

@@ -378,7 +378,7 @@ grpc_ares_request* my_dns_lookup_ares_locked(
     const char* dns_server, const char* addr, const char* default_port,
     grpc_pollset_set* interested_parties, grpc_closure* on_done,
     grpc_lb_addresses** lb_addrs, bool check_grpclb, char** service_config_json,
-    grpc_combiner* combiner) {
+    int query_timeout, grpc_combiner* combiner) {
   addr_req* r = static_cast<addr_req*>(gpr_malloc(sizeof(*r)));
   r->addr = gpr_strdup(addr);
   r->on_done = on_done;

test/core/end2end/goaway_server_test.cc

@@ -48,7 +48,7 @@ static grpc_ares_request* (*iomgr_dns_lookup_ares_locked)(
     const char* dns_server, const char* addr, const char* default_port,
     grpc_pollset_set* interested_parties, grpc_closure* on_done,
     grpc_lb_addresses** addresses, bool check_grpclb,
-    char** service_config_json, grpc_combiner* combiner);
+    char** service_config_json, int query_timeout_ms, grpc_combiner* combiner);
 
 static void (*iomgr_cancel_ares_request_locked)(grpc_ares_request* request);
 
@@ -104,11 +104,11 @@ static grpc_ares_request* my_dns_lookup_ares_locked(
     const char* dns_server, const char* addr, const char* default_port,
     grpc_pollset_set* interested_parties, grpc_closure* on_done,
     grpc_lb_addresses** lb_addrs, bool check_grpclb, char** service_config_json,
-    grpc_combiner* combiner) {
+    int query_timeout_ms, grpc_combiner* combiner) {
   if (0 != strcmp(addr, "test")) {
     return iomgr_dns_lookup_ares_locked(
         dns_server, addr, default_port, interested_parties, on_done, lb_addrs,
-        check_grpclb, service_config_json, combiner);
+        check_grpclb, service_config_json, query_timeout_ms, combiner);
   }
 
   grpc_error* error = GRPC_ERROR_NONE;

test/cpp/naming/cancel_ares_query_test.cc

@@ -260,8 +260,15 @@ TEST(CancelDuringAresQuery, TestFdsAreDeletedFromPollsetSet) {
   grpc_pollset_set_destroy(fake_other_pollset_set);
 }
 
-TEST(CancelDuringAresQuery,
+// Settings for TestCancelDuringActiveQuery test
-     TestHitDeadlineAndDestroyChannelDuringAresResolutionIsGraceful) {
+typedef enum {
+  NONE,
+  SHORT,
+  ZERO,
+} cancellation_test_query_timeout_setting;
+
+void TestCancelDuringActiveQuery(
+    cancellation_test_query_timeout_setting query_timeout_setting) {
   // Start up fake non responsive DNS server
   int fake_dns_port = grpc_pick_unused_port_or_die();
   FakeNonResponsiveDNSServer fake_dns_server(fake_dns_port);
@@ -271,9 +278,33 @@ TEST(CancelDuringAresQuery,
       &client_target,
       "dns://[::1]:%d/dont-care-since-wont-be-resolved.test.com:1234",
       fake_dns_port));
+  gpr_log(GPR_DEBUG, "TestCancelActiveDNSQuery. query timeout setting: %d",
+          query_timeout_setting);
+  grpc_channel_args* client_args = nullptr;
+  grpc_status_code expected_status_code = GRPC_STATUS_OK;
+  if (query_timeout_setting == NONE) {
+    expected_status_code = GRPC_STATUS_DEADLINE_EXCEEDED;
+    client_args = nullptr;
+  } else if (query_timeout_setting == SHORT) {
+    expected_status_code = GRPC_STATUS_UNAVAILABLE;
+    grpc_arg arg;
+    arg.type = GRPC_ARG_INTEGER;
+    arg.key = const_cast<char*>(GRPC_ARG_DNS_ARES_QUERY_TIMEOUT_MS);
+    arg.value.integer =
+        1;  // Set this shorter than the call deadline so that it goes off.
+    client_args = grpc_channel_args_copy_and_add(nullptr, &arg, 1);
+  } else if (query_timeout_setting == ZERO) {
+    expected_status_code = GRPC_STATUS_DEADLINE_EXCEEDED;
+    grpc_arg arg;
+    arg.type = GRPC_ARG_INTEGER;
+    arg.key = const_cast<char*>(GRPC_ARG_DNS_ARES_QUERY_TIMEOUT_MS);
+    arg.value.integer = 0;  // Set this to zero to disable query timeouts.
+    client_args = grpc_channel_args_copy_and_add(nullptr, &arg, 1);
+  } else {
+    abort();
+  }
   grpc_channel* client =
-      grpc_insecure_channel_create(client_target,
+      grpc_insecure_channel_create(client_target, client_args, nullptr);
-                                   /* client_args */ nullptr, nullptr);
   gpr_free(client_target);
   grpc_completion_queue* cq = grpc_completion_queue_create_for_next(nullptr);
   cq_verifier* cqv = cq_verifier_create(cq);
@@ -325,8 +356,9 @@ TEST(CancelDuringAresQuery,
   EXPECT_EQ(GRPC_CALL_OK, error);
   CQ_EXPECT_COMPLETION(cqv, Tag(1), 1);
   cq_verify(cqv);
-  EXPECT_EQ(status, GRPC_STATUS_DEADLINE_EXCEEDED);
+  EXPECT_EQ(status, expected_status_code);
   // Teardown
+  grpc_channel_args_destroy(client_args);
   grpc_slice_unref(details);
   gpr_free((void*)error_string);
   grpc_metadata_array_destroy(&initial_metadata_recv);
@@ -338,6 +370,23 @@ TEST(CancelDuringAresQuery,
   EndTest(client, cq);
 }
 
+TEST(CancelDuringAresQuery,
+     TestHitDeadlineAndDestroyChannelDuringAresResolutionIsGraceful) {
+  TestCancelDuringActiveQuery(NONE /* don't set query timeouts */);
+}
+
+TEST(
+    CancelDuringAresQuery,
+    TestHitDeadlineAndDestroyChannelDuringAresResolutionWithQueryTimeoutIsGraceful) {
+  TestCancelDuringActiveQuery(SHORT /* set short query timeout */);
+}
+
+TEST(
+    CancelDuringAresQuery,
+    TestHitDeadlineAndDestroyChannelDuringAresResolutionWithZeroQueryTimeoutIsGraceful) {
+  TestCancelDuringActiveQuery(ZERO /* disable query timeouts */);
+}
+
 }  // namespace
 
 int main(int argc, char** argv) {