PROTON-2812: Implement async name lookup with c-ares

* TODO: Would be useful to implement a specific async name lookup call
  with a specific proactor event signalling the completed lookup for use
  in code that just provides us with fds.

Author: astitcher

.github/workflows/build.yml

@@ -45,7 +45,7 @@ jobs:
     - name: Install Linux dependencies
       if: runner.os == 'Linux'
       run: |
-        sudo apt install -y swig libpython3-dev libsasl2-dev libjsoncpp-dev softhsm2 opensc
+        sudo apt install -y swig libpython3-dev libsasl2-dev libjsoncpp-dev libc-ares-dev softhsm2 opensc
     - name: Install Windows dependencies
       if: runner.os == 'Windows'
       run: |

azure-pipelines/azure-pipelines.yml

@@ -36,7 +36,7 @@ jobs:
   steps:
   - script: |
       sudo apt-get update
-      sudo apt-get install -y swig libpython3-dev libsasl2-dev libjsoncpp-dev
+      sudo apt-get install -y swig libpython3-dev libsasl2-dev libjsoncpp-dev libc-ares-dev
     name: InstallExtraStuff
   - template: steps.yml
 - job: MacOS

c/CMakeLists.txt

@@ -368,6 +368,14 @@ if (PROACTOR STREQUAL "epoll" OR (NOT PROACTOR AND NOT BUILD_PROACTOR))
     set (PROACTOR_OK epoll)
     set (qpid-proton-proactor src/proactor/epoll.c src/proactor/epoll_raw_connection.c src/proactor/epoll_timer.c ${qpid-proton-proactor-common})
     set (PROACTOR_LIBS Threads::Threads ${TIME_LIB})
+    find_package(c-ares 1.16 CONFIG)
+    option(ENABLE_ASYNC_DNS "Enable async DNS lookups (Using c-ares)" ${c-ares_FOUND})
+    if (ENABLE_ASYNC_DNS)
+      list(APPEND qpid-proton-proactor src/proactor/epoll_name_lookup_async.c)
+      list(APPEND PROACTOR_LIBS c-ares::cares)
+    else()
+      list(APPEND qpid-proton-proactor src/proactor/epoll_name_lookup_sync.c)
+    endif()
   endif()
 endif()
 

c/src/proactor/epoll-internal.h

@@ -61,7 +61,8 @@ typedef enum {
   LISTENER_IO,
   PCONNECTION_IO,
   RAW_CONNECTION_IO,
-  TIMER
+  TIMER,
+  NAME_LOOKUP_EPOLL  /* Inner epoll for async name lookup (e.g. c-ares) */
 } epoll_type_t;
 
 // Data to use with epoll.
@@ -78,7 +79,8 @@ typedef enum {
   PCONNECTION,
   LISTENER,
   RAW_CONNECTION,
-  TIMER_MANAGER
+  TIMER_MANAGER,
+  NAME_LOOKUP
 } task_type_t;
 
 typedef struct task_t {
@@ -142,9 +144,16 @@ typedef struct pni_timer_manager_t {
   bool sched_timeout;
 } pni_timer_manager_t;
 
+typedef struct pname_lookup_t {
+  task_t task;
+  epoll_extended_t epoll_name_lookup;
+  void *impl;   /* NULL for sync; for async: implementation context */
+} pname_lookup_t;
+
 struct pn_proactor_t {
   task_t task;
   pni_timer_manager_t timer_manager;
+  pname_lookup_t name_lookup;
   epoll_extended_t epoll_schedule;     /* ready list */
   epoll_extended_t epoll_interrupt;
   pn_event_batch_t batch;
@@ -364,7 +373,7 @@ bool proactor_remove(task_t *tsk);
 bool unassign_thread(pn_proactor_t *p, tslot_t *ts, tslot_state new_state, tslot_t **resume_thread);
 
 void task_init(task_t *tsk, task_type_t t, pn_proactor_t *p);
-static void task_finalize(task_t* tsk) {
+static inline void task_finalize(task_t* tsk) {
   pmutex_finalize(&tsk->mutex);
 }
 
@@ -377,7 +386,6 @@ bool start_polling(epoll_extended_t *ee, int epollfd);
 void stop_polling(epoll_extended_t *ee, int epollfd);
 void rearm_polling(epoll_extended_t *ee, int epollfd);
 
-int pgetaddrinfo(const char *host, const char *port, int flags, struct addrinfo **res);
 void configure_socket(int sock);
 
 accepted_t *listener_accepted_next(pn_listener_t *listener);

c/src/proactor/epoll.c

@@ -63,6 +63,7 @@
 #undef _GNU_SOURCE
 
 #include "epoll-internal.h"
+#include "epoll_name_lookup.h"
 #include "proactor-internal.h"
 #include "core/engine-internal.h"
 #include "core/logger_private.h"
@@ -632,6 +633,10 @@ static inline pn_listener_t *task_listener(task_t *t) {
   return t->type == LISTENER ? containerof(t, pn_listener_t, task) : NULL;
 }
 
+static inline pname_lookup_t *task_name_lookup(task_t *t) {
+  return t->type == NAME_LOOKUP ? containerof(t, pname_lookup_t, task) : NULL;
+}
+
 static pn_event_t *listener_batch_next(pn_event_batch_t *batch);
 static pn_event_t *proactor_batch_next(pn_event_batch_t *batch);
 static pn_event_t *pconnection_batch_next(pn_event_batch_t *batch);
@@ -1408,15 +1413,6 @@ static void pconnection_maybe_connect_lh(pconnection_t *pc) {
   pc->disconnected = true;
 }
 
-int pgetaddrinfo(const char *host, const char *port, int flags, struct addrinfo **res)
-{
-  // NOTE: getaddrinfo can block on DNS lookup (PROTON-2812).
-  struct addrinfo hints = { 0 };
-  hints.ai_family = AF_UNSPEC;
-  hints.ai_socktype = SOCK_STREAM;
-  hints.ai_flags = AI_V4MAPPED | AI_ADDRCONFIG | flags;
-  return getaddrinfo(host, port, &hints, res);
-}
 
 static inline bool is_inactive(pn_proactor_t *p) {
   return (!p->tasks && !p->disconnects_pending && !p->timeout_set && !p->shutting_down);
@@ -1431,23 +1427,39 @@ bool schedule_if_inactive(pn_proactor_t *p) {
   return false;
 }
 
+/* Called when connection name lookup completes (from name_lookup done_cb). Call with task lock held. */
+static void connection_lookup_done_lh(pconnection_t *pc, struct addrinfo *ai, int gai_error) {
+  pn_proactor_t *p = pc->task.proactor;
+  bool notify = false;
+  if (gai_error) {
+    psocket_gai_error(&pc->psocket, gai_error, "connect to ");
+  } else if (ai) {
+    pc->addrinfo = ai;
+    pc->ai = ai;
+    pconnection_maybe_connect_lh(pc);
+    if (pc->psocket.epoll_io.fd != -1 && !pc->queued_disconnect && !pni_task_wake_pending(&pc->task)) {
+      return;
+    }
+  }
+  notify = schedule(&pc->task);
+  if (notify) notify_poller(p);
+}
+
+static void connection_done_cb(void *user_data, struct addrinfo *ai, int gai_error) {
+  pconnection_t *pc = (pconnection_t *)user_data;
+  lock(&pc->task.mutex);
+  connection_lookup_done_lh(pc, ai, gai_error);
+  unlock(&pc->task.mutex);
+}
+
 // Call from pconnection_process with task lock held.
 // Return true if the socket is connecting and there are no Proton events to deliver.
 static bool pconnection_first_connect_lh(pconnection_t *pc) {
+  pn_proactor_t *p = pc->task.proactor;
   unlock(&pc->task.mutex);
-  // TODO: move this step to a separate worker thread that scales in response to multiple blocking DNS lookups.
-  int gai_error = pgetaddrinfo(pc->host, pc->port, 0, &pc->addrinfo);
+  bool rc = pni_name_lookup_start(&p->name_lookup, pc->host, pc->port, pc, connection_done_cb);
   lock(&pc->task.mutex);
-
-  if (!gai_error) {
-    pc->ai = pc->addrinfo;
-    pconnection_maybe_connect_lh(pc); /* Start connection attempts */
-    if (pc->psocket.epoll_io.fd != -1 && !pc->queued_disconnect && !pni_task_wake_pending(&pc->task))
-      return true;
-  } else {
-    psocket_gai_error(&pc->psocket, gai_error, "connect to ");
-  }
-  return false;
+  return rc;
 }
 
 void pn_proactor_connect2(pn_proactor_t *p, pn_connection_t *c, pn_transport_t *t, const char *addr) {
@@ -1579,7 +1591,7 @@ void pn_proactor_listen(pn_proactor_t *p, pn_listener_t *l, const char *addr, in
   pni_parse_addr(addr, l->addr_buf, sizeof(l->addr_buf), &l->host, &l->port);
 
   struct addrinfo *addrinfo = NULL;
-  int gai_err = pgetaddrinfo(l->host, l->port, AI_PASSIVE | AI_ALL, &addrinfo);
+  int gai_err = pni_name_lookup_blocking(l->host, l->port, AI_PASSIVE | AI_ALL, &addrinfo);
   if (!gai_err) {
     /* Count addresses, allocate enough space for sockets */
     size_t len = 0;
@@ -2021,23 +2033,27 @@ pn_proactor_t *pn_proactor(void) {
   if ((p->epollfd = epoll_create(1)) >= 0) {
     if ((p->eventfd = eventfd(0, EFD_NONBLOCK)) >= 0) {
       if ((p->interruptfd = eventfd(0, EFD_NONBLOCK)) >= 0) {
-        if (pni_timer_manager_init(&p->timer_manager))
-          if ((p->collector = pn_collector()) != NULL) {
-            p->batch.next_event = &proactor_batch_next;
-            start_polling(&p->timer_manager.epoll_timer, p->epollfd);  // TODO: check for error
-            epoll_eventfd_init(&p->epoll_schedule, p->eventfd, p->epollfd, true);
-            epoll_eventfd_init(&p->epoll_interrupt, p->interruptfd, p->epollfd, false);
-            p->tslot_map = pn_hash(PN_VOID, 0, 0.75);
-            grow_poller_bufs(p);
-            p->ready_list_generation = 1;
-            return p;
+        if (pni_timer_manager_init(&p->timer_manager)) {
+          if (pni_name_lookup_init(&p->name_lookup, p)) {
+            if ((p->collector = pn_collector()) != NULL) {
+              p->batch.next_event = &proactor_batch_next;
+              start_polling(&p->timer_manager.epoll_timer, p->epollfd);  // TODO: check for error
+              epoll_eventfd_init(&p->epoll_schedule, p->eventfd, p->epollfd, true);
+              epoll_eventfd_init(&p->epoll_interrupt, p->interruptfd, p->epollfd, false);
+              p->tslot_map = pn_hash(PN_VOID, 0, 0.75);
+              grow_poller_bufs(p);
+              p->ready_list_generation = 1;
+              return p;
+            }
           }
+        }
       }
     }
   }
   if (p->epollfd >= 0) close(p->epollfd);
   if (p->eventfd >= 0) close(p->eventfd);
   if (p->interruptfd >= 0) close(p->interruptfd);
+  pni_name_lookup_cleanup(&p->name_lookup, p);
   pni_timer_manager_finalize(&p->timer_manager);
   pmutex_finalize(&p->timeout_mutex);
   pmutex_finalize(&p->tslot_mutex);
@@ -2071,11 +2087,15 @@ void pn_proactor_free(pn_proactor_t *p) {
      case RAW_CONNECTION:
       pni_raw_connection_forced_shutdown(pni_task_raw_connection(tsk));
       break;
+     case NAME_LOOKUP:
+      pni_name_lookup_forced_shutdown(task_name_lookup(tsk));
+      break;
      default:
       break;
     }
   }
 
+  pni_name_lookup_cleanup(&p->name_lookup, p);
   pni_timer_manager_finalize(&p->timer_manager);
   pn_collector_free(p->collector);
   pmutex_finalize(&p->timeout_mutex);
@@ -2309,6 +2329,11 @@ static pn_event_batch_t *process(task_t *tsk) {
     batch = pni_timer_manager_process(tm, timeout, tsk_ready);
     break;
   }
+  case NAME_LOOKUP:
+    unlock(&p->sched_mutex);
+    pni_name_lookup_process_events(&p->name_lookup);
+    batch = NULL;
+    break;
   default:
     assert(NULL);
   }
@@ -2350,6 +2375,11 @@ static task_t *post_event(pn_proactor_t *p, struct epoll_event *evp) {
     }
     // else if (ee->fd == p->eventfd)... schedule_ready_list already performed by poller task.
     break;
+  case NAME_LOOKUP_EPOLL: {
+    tsk = &p->name_lookup.task;
+    tsk->sched_pending = true;
+    break;
+  }
   case PCONNECTION_IO: {
     psocket_t *ps = containerof(ee, psocket_t, epoll_io);
     pconnection_t *pc = psocket_pconnection(ps);
@@ -2581,11 +2611,10 @@ static bool poller_do_epoll(struct pn_proactor_t* p, tslot_t *ts, bool can_block
       unlock(&p->eventfd_mutex);
     }
 
-    int timeout = (epoll_immediate) ? 0 : -1;
-    p->poller_suspended = (timeout == -1);
+    p->poller_suspended = !epoll_immediate;
     unlock(&p->sched_mutex);
 
-    n_events = epoll_wait(p->epollfd, p->kevents, p->kevents_capacity, timeout);
+    n_events = epoll_wait(p->epollfd, p->kevents, p->kevents_capacity, epoll_immediate ? 0 : -1);
 
     lock(&p->sched_mutex);
     p->poller_suspended = false;
@@ -2612,8 +2641,9 @@ static bool poller_do_epoll(struct pn_proactor_t* p, tslot_t *ts, bool can_block
     unlock(&p->eventfd_mutex);
 
     if (n_events < 0) {
-      if (errno != EINTR)
+      if (errno != EINTR) {
         perror("epoll_wait"); // TODO: proper log
+      }
       if (!can_block && !unpolled_work)
         return true;
       else
@@ -2622,8 +2652,9 @@ static bool poller_do_epoll(struct pn_proactor_t* p, tslot_t *ts, bool can_block
       if (!can_block && !unpolled_work)
         return true;
       else {
-        if (!epoll_immediate)
+        if (!epoll_immediate) {
           perror("epoll_wait unexpected timeout"); // TODO: proper log
+        }
         if (!unpolled_work)
           continue;
       }

c/src/proactor/epoll_name_lookup.h

@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ *
+ */
+
+#ifndef PROACTOR_NAME_LOOKUP_H
+#define PROACTOR_NAME_LOOKUP_H
+
+#include "epoll-internal.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/* Blocking name lookup - used by listeners and as fallback for connections.
+ * Same signature as getaddrinfo.
+ */
+int pni_name_lookup_blocking(const char *host, const char *port, int flags, struct addrinfo **res);
+
+/* Initialize name lookup subsystem. Returns false if error. */
+bool pni_name_lookup_init(pname_lookup_t *nl, pn_proactor_t *p);
+
+/* Cleanup name lookup subsystem. */
+void pni_name_lookup_cleanup(pname_lookup_t *nl, pn_proactor_t *p);
+
+/* Callback when lookup completes. Called with (user_data, addrinfo or NULL, gai_error). */
+typedef void (*pni_nl_done_cb)(void *user_data, struct addrinfo *ai, int gai_error);
+
+/* Start a name lookup. Returns true if started successfully. */
+bool pni_name_lookup_start(pname_lookup_t *nl, const char *host, const char *port, void *user_data, pni_nl_done_cb done_cb);
+
+/* Clear lookup state when forcing shutdown. */
+void pni_name_lookup_forced_shutdown(pname_lookup_t *nl);
+
+/* Process events. */
+void pni_name_lookup_process_events(pname_lookup_t *nl);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* PROACTOR_NAME_LOOKUP_H */