async DNS resolution via udns, take 3

.gitignore

@@ -41,6 +41,7 @@ scripts/lt*.m4
 .#*
 \#*#
 *~
+*.swp
 
 # Packages #
 ############

.travis.yml

@@ -4,8 +4,19 @@ matrix:
   include:
   - compiler: clang
     env: COMPILER=clang++ SKIP_CHECK=true SKIP_COVERAGE=true
+    addons:
+      apt:
+        packages:
+        - libudns-dev
   - compiler: clang
     env: COMPILER=clang++ SKIP_COVERAGE=true
+    addons:
+      apt:
+        packages:
+        - libcppunit-dev
+        - libudns-dev
+  - compiler: clang
+    env: COMPILER=clang++ SKIP_COVERAGE=true UDNS=--disable-udns
     addons:
       apt:
         packages:
@@ -20,6 +31,7 @@ matrix:
         packages:
         - clang-3.6
         - libcppunit-dev
+        - libudns-dev
   - compiler: clang
     env: COMPILER=clang++-3.7 SKIP_COVERAGE=true
     addons:
@@ -30,6 +42,7 @@ matrix:
         packages:
         - clang-3.7
         - libcppunit-dev
+        - libudns-dev
   - compiler: clang
     env: COMPILER=clang++-3.8 SKIP_COVERAGE=true
     addons:
@@ -40,15 +53,26 @@ matrix:
         packages:
         - clang-3.8
         - libcppunit-dev
+        - libudns-dev
   - compiler: gcc
     env: COMPILER=g++-4.7 SKIP_CHECK=true SKIP_COVERAGE=true
     addons:
       apt:
         sources: ubuntu-toolchain-r-test
         packages:
         - g++-4.7
+        - libudns-dev
   - compiler: gcc
     env: COMPILER=g++-4.7 SKIP_COVERAGE=true
+    addons:
+      apt:
+        sources: ubuntu-toolchain-r-test
+        packages:
+        - g++-4.7
+        - libcppunit-dev
+        - libudns-dev
+  - compiler: gcc
+    env: COMPILER=g++-4.7 SKIP_COVERAGE=true UDNS=--disable-udns
     addons:
       apt:
         sources: ubuntu-toolchain-r-test
@@ -63,13 +87,14 @@ matrix:
         packages:
         - g++-4.8
         - libcppunit-dev
+        - libudns-dev
 
 before_install:
   - if [ ! $SKIP_COVERAGE ]; then pip install --user cpp-coveralls; fi
 
 script:
 - if [ ! $SKIP_COVERAGE ]; then export CXXFLAGS="-ftest-coverage -fprofile-arcs"; fi
-- ./autogen.sh && CXX="$COMPILER" ./configure && make -j12
+- ./autogen.sh && CXX="$COMPILER" ./configure $UDNS && make -j12
 - if [ ! $SKIP_CHECK ]; then make -j12 check; fi
 
 after_success:

configure.ac

@@ -97,6 +97,8 @@ AC_ARG_ENABLE(cyrus-rc4,
   ]
 )
 
+TORRENT_WITH_UDNS()
+
 AC_CHECK_FUNCS(posix_memalign)
 
 TORRENT_CHECK_MADVISE()

scripts/udns.m4

@@ -0,0 +1,26 @@
+dnl function for enabling/disabling udns support
+AC_DEFUN([TORRENT_WITH_UDNS], [
+  AC_ARG_WITH(
+    [udns],
+    AS_HELP_STRING([--without-udns], [Don't use udns, falling back to synchronous DNS resolution.])
+  )
+dnl neither ubuntu nor fedora ships a pkgconfig file for udns
+  AS_IF(
+    [test "x$with_udns"  != "xno"],
+    [AC_CHECK_HEADERS([udns.h], [have_udns=yes], [have_udns=no])],
+    [have_udns=no]
+  )
+  AS_IF(
+    [test "x$have_udns" = "xyes"],
+    [
+      AC_DEFINE(USE_UDNS, 1, Define to build with udns support.)
+      LIBS="$LIBS -ludns"
+    ],
+    [
+      AS_IF(
+        [test "x$with_udns" = "xyes"],
+        [AC_MSG_ERROR([udns requested but not found])]
+      )
+    ]
+  )
+])

src/torrent/connection_manager.cc

@@ -48,11 +48,69 @@
 #include "exceptions.h"
 #include "manager.h"
 
+#ifdef USE_UDNS
+#include "utils/udnsevent.h"
+#endif
+
 namespace torrent {
 
-// Fix TrackerUdp, etc, if this is made async.
-static ConnectionManager::slot_resolver_result_type*
-resolve_host(const char* host, int family, int socktype, ConnectionManager::slot_resolver_result_type slot) {
+/* Encapsulates whether we do genuine async resolution or fall back to sync. */
+struct AsyncResolver {
+
+  ConnectionManager *m_connection_manager;
+#ifdef USE_UDNS
+  UdnsEvent           m_udnsevent;
+#else
+  struct MockResolve {
+    std::string hostname;
+    int family;
+    resolver_callback *callback;
+  };
+  std::list<MockResolve*> m_mock_resolve_queue;
+#endif
+
+  AsyncResolver(ConnectionManager *cm): m_connection_manager(cm) {}
+
+  void *enqueue_resolve(const char *name, int family, resolver_callback *cbck) {
+#ifdef USE_UDNS
+    return m_udnsevent.enqueue_resolve(name, family, cbck);
+#else
+    MockResolve *mock_resolve = new MockResolve {name, family, cbck};
+    m_mock_resolve_queue.push_back(mock_resolve);
+    return mock_resolve;
+#endif
+  }
+
+  void flush_resolves() {
+#ifdef USE_UDNS
+    m_udnsevent.flush_resolves();
+#else
+    // dequeue all callbacks and resolve them synchronously
+    while (!m_mock_resolve_queue.empty()) {
+      MockResolve *mock_resolve = m_mock_resolve_queue.front();
+      m_mock_resolve_queue.pop_front();
+      m_connection_manager->resolver()(mock_resolve->hostname.c_str(), mock_resolve->family, 0, *(mock_resolve->callback));
+      delete mock_resolve;
+    }
+#endif
+  }
+
+  void cancel_resolve(void *query) {
+#ifdef USE_UDNS
+    m_udnsevent.cancel(static_cast<UdnsQuery*>(query));
+#else
+    MockResolve *mock_resolve = static_cast<MockResolve*>(query);
+    auto it = std::find(std::begin(m_mock_resolve_queue), std::end(m_mock_resolve_queue), mock_resolve);
+    if (it != std::end(m_mock_resolve_queue)) {
+      m_mock_resolve_queue.erase(it);
+      delete mock_resolve;
+    }
+#endif
+  }
+};
+
+static void
+resolve_host(const char* host, int family, int socktype, resolver_callback slot) {
   if (manager->main_thread_main()->is_current())
     thread_base::release_global_lock();
 
@@ -64,7 +122,7 @@ resolve_host(const char* host, int family, int socktype, ConnectionManager::slot
       thread_base::acquire_global_lock();
 
     slot(NULL, err);
-    return NULL;
+    return;
   }
 
   rak::socket_address sa;
@@ -75,7 +133,7 @@ resolve_host(const char* host, int family, int socktype, ConnectionManager::slot
     thread_base::acquire_global_lock();
 
   slot(sa.c_sockaddr(), 0);
-  return NULL;
+  return;
 }
 
 ConnectionManager::ConnectionManager() :
@@ -89,7 +147,8 @@ ConnectionManager::ConnectionManager() :
 
   m_listen(new Listen),
   m_listen_port(0),
-  m_listen_backlog(SOMAXCONN) {
+  m_listen_backlog(SOMAXCONN),
+  m_async_resolver(new AsyncResolver(this)) {
 
   m_bindAddress = (new rak::socket_address())->c_sockaddr();
   m_localAddress = (new rak::socket_address())->c_sockaddr();
@@ -202,4 +261,18 @@ ConnectionManager::set_listen_backlog(int v) {
   m_listen_backlog = v;
 }
 
+void *
+ConnectionManager::enqueue_async_resolve(const char *name, int family, resolver_callback *cbck) {
+  return m_async_resolver->enqueue_resolve(name, family, cbck);
+}
+
+void ConnectionManager::flush_async_resolves() {
+  m_async_resolver->flush_resolves();
+}
+
+void
+ConnectionManager::cancel_async_resolve(void *query) {
+  m_async_resolver->cancel_resolve(query);
+}
+
 }

src/torrent/connection_manager.h

@@ -39,12 +39,13 @@
 #ifndef LIBTORRENT_CONNECTION_MANAGER_H
 #define LIBTORRENT_CONNECTION_MANAGER_H
 
-#include <list>
 #include <arpa/inet.h>
 #include <netinet/in.h>
 #include <netinet/in_systm.h>
 #include <netinet/ip.h>
 #include <sys/socket.h>
+#include <list>
+#include <memory>
 #include lt_tr1_functional
 #include <torrent/common.h>
 
@@ -54,6 +55,12 @@ namespace torrent {
 // First element is upload throttle, second element is download throttle.
 typedef std::pair<Throttle*, Throttle*> ThrottlePair;
 
+// The sockaddr argument in the result call is NULL if the resolve failed,
+// and the int holds the error code.
+typedef std::function<void (const sockaddr*, int)> resolver_callback;
+
+struct AsyncResolver;
+
 class LIBTORRENT_EXPORT ConnectionManager {
 public:
   typedef uint32_t size_type;
@@ -100,9 +107,7 @@ class LIBTORRENT_EXPORT ConnectionManager {
   typedef std::function<uint32_t (const sockaddr*)>     slot_filter_type;
   typedef std::function<ThrottlePair (const sockaddr*)> slot_throttle_type;
 
-  // The sockaddr argument in the result slot call is NULL if the resolve failed, and the int holds the errno.
-  typedef std::function<void (const sockaddr*, int)> slot_resolver_result_type;
-  typedef std::function<slot_resolver_result_type* (const char*, int, int, slot_resolver_result_type)> slot_resolver_type;
+  typedef std::function<void (const char*, int, int, resolver_callback)> slot_resolver_type;
 
   ConnectionManager();
   ~ConnectionManager();
@@ -154,10 +159,23 @@ class LIBTORRENT_EXPORT ConnectionManager {
   void                set_listen_port(port_type p)            { m_listen_port = p; }
   void                set_listen_backlog(int v);
 
-  // The resolver returns a pointer to its copy of the result slot
-  // which the caller may set blocked to prevent the slot from being
-  // called. The pointer must be NULL if the result slot was already
-  // called because the resolve was synchronous.
+  /* Async resolver interface.
+
+     In a build with USE_UDNS, these do genuine asynchronous DNS resolution.
+     In a build without it, they're stubbed out to use a synchronous getaddrinfo(3)
+     call, while exposing the same API.
+  */
+  // this queues a DNS resolve but doesn't send it. it doesn't execute any callbacks
+  // and returns control immediately. the return value is an opaque identifier that
+  // can be used to cancel the query (as long as the callback hasn't been executed yet):
+  void*               enqueue_async_resolve(const char *name, int family, resolver_callback *cbck);
+  // this sends any queued resolves. it can execute arbitrary callbacks
+  // before returning control:
+  void                flush_async_resolves();
+  // this cancels a pending async query (as long as the callback hasn't executed yet):
+  void                cancel_async_resolve(void *query);
+
+  /* Legacy synchronous resolver interface. */
   slot_resolver_type& resolver()          { return m_slot_resolver; }
 
   // The slot returns a ThrottlePair to use for the given address, or
@@ -190,6 +208,8 @@ class LIBTORRENT_EXPORT ConnectionManager {
   slot_filter_type    m_slot_filter;
   slot_resolver_type  m_slot_resolver;
   slot_throttle_type  m_slot_address_throttle;
+
+  std::unique_ptr<AsyncResolver> m_async_resolver;
 };
 
 }