Add non-blocking DNS for the async API via c-ares event-loop integration

CMakeLists.txt

@@ -53,6 +53,7 @@ set(valkey_sources
     src/net.c
     src/read.c
     src/sockcompat.c
+    src/timer.c
     src/valkey.c
     src/vkutil.c)
 

README.md

@@ -21,7 +21,7 @@ Libvalkey is the official C client for the [Valkey](https://valkey.io) database.
 - Supports both `RESP2` and `RESP3` protocol versions.
 - Supports both synchronous and asynchronous operation.
 - Optional support for `MPTCP`, `TLS` and `RDMA` connections.
-- Optional timeout-bounded DNS resolution via [c-ares](https://github.com/c-ares/c-ares).
+- Optional non-blocking DNS resolution via [c-ares](https://github.com/c-ares/c-ares) with timeout support.
 - Asynchronous API with several event libraries to choose from.
 - Supports both standalone and cluster mode operation.
 - Can be compiled with either `make` or `CMake`.

docs/standalone.md

@@ -21,6 +21,7 @@ This document describes using `libvalkey` in standalone (non-cluster) mode, incl
     - [Allocator injection](#allocator-injection)
 - [Asynchronous API](#asynchronous-api)
   - [Connecting](#connecting-1)
+    - [Attaching an adapter via connection options](#attaching-an-adapter-via-connection-options)
   - [Executing commands](#executing-commands-1)
   - [Disconnecting/cleanup](#disconnecting-cleanup-1)
 - [TLS support](#tls-support)
@@ -49,6 +50,10 @@ When a hostname resolves to multiple addresses, libvalkey will try each address
 
 When built with c-ares support (`USE_CARES=1` / `-DENABLE_CARES=1`), DNS resolution uses c-ares instead of `getaddrinfo()`. This provides timeout-bounded DNS resolution using `connect_timeout` (defaulting to 5 seconds if unset), preventing indefinite hangs when DNS servers are slow or unreachable.
 
+For the synchronous API, DNS resolution blocks with a `poll()` loop bounded by the timeout.
+For the asynchronous API, DNS is fully non-blocking when the connection is set up via the options path (`valkeyAsyncConnectWithOptions` with `attach_fn`) using an adapter that implements the c-ares hooks; the libevent adapter does this and drives the c-ares file descriptors from the event loop.
+All other cases, the legacy connect-then-attach path and adapters without the hooks, use the blocking, timeout-bounded resolution.
+
 c-ares support is available on Linux, macOS, and FreeBSD (not Windows).
 
 ```c
@@ -330,6 +335,72 @@ The asynchronous context _should_ hold a connect callback function that is calle
 It _can_ also hold a disconnect callback function that is called when the connection is disconnected (either because of an error or per user request).
 The context object is always freed after the disconnect callback fired.
 
+#### Attaching an adapter via connection options
+
+The example above uses the two-step pattern: create the context, then attach an event-loop adapter separately.
+As an alternative, you can specify the adapter as part of the `valkeyOptions` using the `attach_fn` and `attach_data` fields.
+When set, `valkeyAsyncConnectWithOptions()` attaches the adapter and registers the file descriptor with the event loop for you, in a single call.
+You can also supply the connect and disconnect callbacks via `async_connect_callback` and `async_disconnect_callback`.
+These are registered before the connect is initiated, so no connect event is missed even if the event loop is already running.
+
+```c
+valkeyOptions options = {0};
+VALKEY_OPTIONS_SET_TCP(&options, "localhost", 6379);
+
+// Specify the event-loop adapter as part of the options.
+options.attach_fn = valkeyLibevAttachAdapter;
+options.attach_data = EV_DEFAULT; // the default libev loop; or a `struct ev_loop *`
+
+// Optionally supply the connect/disconnect callbacks too.
+options.async_connect_callback = my_connect_callback;
+options.async_disconnect_callback = my_disconnect_callback;
+
+valkeyAsyncContext *ac = valkeyAsyncConnectWithOptions(&options);
+if (ac == NULL || ac->err) {
+    fprintf(stderr, "Error: %s\n", ac ? ac->errstr : "OOM");
+    // ... handle error / cleanup ...
+}
+
+ev_run(EV_DEFAULT_ 0);
+```
+
+For TCP connections, the actual connect is deferred until after the context is fully initialized, so the adapter always receives a valid file descriptor and works without modification.
+Unix socket connections connect immediately.
+
+The legacy pattern of attaching the adapter after connecting (as shown in the example above) remains fully supported; all of these option fields are optional.
+The callbacks can still be set afterwards with `valkeyAsyncSetConnectCallback` / `valkeyAsyncSetDisconnectCallback`, but doing so via the options is race-free when the loop is already running.
+
+#### Custom adapters and c-ares
+
+When built with c-ares, the asynchronous connect defers DNS resolution until the adapter is attached (this only happens on the options path, when `attach_fn` is set).
+Custom adapters need no special handling for this: if the adapter does not implement the c-ares hooks, `valkeyAsyncConnectWithOptions` resolves DNS synchronously (timeout-bounded) and connects before returning, so the adapter always sees a valid `ac->c.fd`.
+
+To support fully non-blocking DNS, an adapter implements the c-ares socket hooks and initiates the asynchronous resolution from its attach function:
+
+```c
+static int myAdapterAttach(valkeyAsyncContext *ac, void *loopdata) {
+    if (ac->ev.data != NULL)
+        return VALKEY_ERR;
+
+    /* ... register addRead/addWrite/etc. hooks ... */
+#ifdef VALKEY_USE_CARES
+    ac->ev.addCaresSocket = myAdapterAddCaresSocket; /* register a c-ares fd */
+    ac->ev.delCaresSocket = myAdapterDelCaresSocket; /* deregister a c-ares fd */
+
+    if (ac->c.flags & VALKEY_CONNECT_DEFERRED) {
+        /* No fd yet, drive DNS via the event loop. On completion the
+         * connect proceeds and the fd is registered automatically. */
+        return valkeyResolveAsyncStart(ac, ac->c.tcp.host, ac->c.tcp.port);
+    }
+#endif
+    /* ... otherwise register ac->c.fd normally ... */
+    return VALKEY_OK;
+}
+```
+
+The adapter's c-ares `fd` callbacks forward activity to `valkeyResolveAsyncHandleEvent(ac, fd, readable, writable)`.
+See the libevent adapter for a complete reference implementation.
+
 ### Executing commands
 
 Executing commands in an asynchronous context work similarly to the synchronous context, except that you can pass a callback that will be invoked when the reply is received.

examples/async-libev.c

@@ -39,16 +39,20 @@ int main(int argc, char **argv) {
     signal(SIGPIPE, SIG_IGN);
 #endif
 
-    valkeyAsyncContext *c = valkeyAsyncConnect("127.0.0.1", 6379);
+    valkeyOptions options = {0};
+    VALKEY_OPTIONS_SET_TCP(&options, "127.0.0.1", 6379);
+    options.attach_fn = valkeyLibevAttachAdapter;
+    options.attach_data = EV_DEFAULT;
+    options.async_connect_callback = connectCallback;
+    options.async_disconnect_callback = disconnectCallback;
+
+    valkeyAsyncContext *c = valkeyAsyncConnectWithOptions(&options);
     if (c->err) {
         /* Let *c leak for now... */
         printf("Error: %s\n", c->errstr);
         return 1;
     }
 
-    valkeyLibevAttach(EV_DEFAULT_ c);
-    valkeyAsyncSetConnectCallback(c, connectCallback);
-    valkeyAsyncSetDisconnectCallback(c, disconnectCallback);
     valkeyAsyncCommand(
         c, NULL, NULL, "SET key %b", argv[argc - 1], strlen(argv[argc - 1]));
     valkeyAsyncCommand(c, getCallback, (char *)"end-1", "GET key");

examples/async-libevent.c

@@ -49,6 +49,10 @@ int main(int argc, char **argv) {
     struct timeval tv = {0};
     tv.tv_sec = 1;
     options.connect_timeout = &tv;
+    options.attach_fn = valkeyLibeventAttachAdapter;
+    options.attach_data = base;
+    options.async_connect_callback = connectCallback;
+    options.async_disconnect_callback = disconnectCallback;
 
     valkeyAsyncContext *c = valkeyAsyncConnectWithOptions(&options);
     if (c->err) {
@@ -57,9 +61,6 @@ int main(int argc, char **argv) {
         return 1;
     }
 
-    valkeyLibeventAttach(c, base);
-    valkeyAsyncSetConnectCallback(c, connectCallback);
-    valkeyAsyncSetDisconnectCallback(c, disconnectCallback);
     valkeyAsyncCommand(
         c, NULL, NULL, "SET key %b", argv[argc - 1], strlen(argv[argc - 1]));
     valkeyAsyncCommand(c, getCallback, (char *)"end-1", "GET key");

include/valkey/adapters/libevent.h

@@ -39,6 +39,16 @@
 #define VALKEY_LIBEVENT_DELETED 0x01
 #define VALKEY_LIBEVENT_ENTERED 0x02
 
+#ifdef VALKEY_USE_CARES
+
+#define VALKEY_LIBEVENT_MAX_CARES_FDS 16
+
+typedef struct valkeyLibeventCaresEvent {
+    int fd;
+    struct event *ev;
+} valkeyLibeventCaresEvent;
+#endif
+
 typedef struct valkeyLibeventEvents {
     valkeyAsyncContext *context;
     struct event *ev;
@@ -47,6 +57,10 @@ typedef struct valkeyLibeventEvents {
     struct timeval tv;
     short flags;
     short state;
+#ifdef VALKEY_USE_CARES
+    valkeyLibeventCaresEvent cares_fds[VALKEY_LIBEVENT_MAX_CARES_FDS];
+    int cares_nfds;
+#endif
 } valkeyLibeventEvents;
 
 static void valkeyLibeventDestroy(valkeyLibeventEvents *e) {
@@ -102,7 +116,11 @@ static void valkeyLibeventUpdate(void *privdata, short flag, int isRemove) {
         }
     }
 
-    event_del(e->ev);
+    if (e->ev) {
+        event_del(e->ev);
+    } else {
+        e->ev = event_new(e->base, e->context->c.fd, 0, valkeyLibeventHandler, privdata);
+    }
     event_assign(e->ev, e->base, e->context->c.fd, e->flags | EV_PERSIST,
                  valkeyLibeventHandler, privdata);
     event_add(e->ev, NULL);
@@ -129,13 +147,22 @@ static void valkeyLibeventCleanup(void *privdata) {
     if (!e) {
         return;
     }
-    event_del(e->ev);
-    event_free(e->ev);
-    e->ev = NULL;
+    if (e->ev) {
+        event_del(e->ev);
+        event_free(e->ev);
+        e->ev = NULL;
+    }
     if (e->timer) {
         event_free(e->timer);
         e->timer = NULL;
     }
+#ifdef VALKEY_USE_CARES
+    for (int i = 0; i < e->cares_nfds; i++) {
+        if (e->cares_fds[i].ev)
+            event_free(e->cares_fds[i].ev);
+    }
+    e->cares_nfds = 0;
+#endif
 
     if (e->state & VALKEY_LIBEVENT_ENTERED) {
         e->state |= VALKEY_LIBEVENT_DELETED;
@@ -154,6 +181,60 @@ static void valkeyLibeventSetTimeout(void *privdata, struct timeval tv) {
         evtimer_add(e->timer, &tv);
 }
 
+#ifdef VALKEY_USE_CARES
+static void valkeyLibeventCaresHandler(evutil_socket_t fd, short event, void *arg) {
+    valkeyLibeventEvents *e = (valkeyLibeventEvents *)arg;
+    int readable = (event & EV_READ) != 0;
+    int writable = (event & EV_WRITE) != 0;
+    valkeyResolveAsyncHandleEvent(e->context, (int)fd, readable, writable);
+}
+
+static void valkeyLibeventAddCaresSocket(void *privdata, int fd, int readable, int writable) {
+    valkeyLibeventEvents *e = (valkeyLibeventEvents *)privdata;
+    int i;
+
+    /* Find existing or add new. */
+    for (i = 0; i < e->cares_nfds; i++) {
+        if (e->cares_fds[i].fd == fd)
+            break;
+    }
+    if (i == e->cares_nfds) {
+        if (e->cares_nfds >= VALKEY_LIBEVENT_MAX_CARES_FDS)
+            return;
+        e->cares_fds[i].fd = fd;
+        e->cares_fds[i].ev = NULL;
+        e->cares_nfds++;
+    }
+
+    short flags = EV_PERSIST;
+    if (readable)
+        flags |= EV_READ;
+    if (writable)
+        flags |= EV_WRITE;
+
+    if (e->cares_fds[i].ev)
+        event_free(e->cares_fds[i].ev);
+    e->cares_fds[i].ev = event_new(e->base, fd, flags, valkeyLibeventCaresHandler, e);
+    event_add(e->cares_fds[i].ev, NULL);
+}
+
+static void valkeyLibeventDelCaresSocket(void *privdata, int fd) {
+    valkeyLibeventEvents *e = (valkeyLibeventEvents *)privdata;
+
+    for (int i = 0; i < e->cares_nfds; i++) {
+        if (e->cares_fds[i].fd == fd) {
+            if (e->cares_fds[i].ev) {
+                event_free(e->cares_fds[i].ev);
+                e->cares_fds[i].ev = NULL;
+            }
+            e->cares_fds[i] = e->cares_fds[e->cares_nfds - 1];
+            e->cares_nfds--;
+            break;
+        }
+    }
+}
+#endif /* VALKEY_USE_CARES */
+
 static int valkeyLibeventAttach(valkeyAsyncContext *ac, struct event_base *base) {
     valkeyContext *c = &(ac->c);
     valkeyLibeventEvents *e;
@@ -176,11 +257,30 @@ static int valkeyLibeventAttach(valkeyAsyncContext *ac, struct event_base *base)
     ac->ev.delWrite = valkeyLibeventDelWrite;
     ac->ev.cleanup = valkeyLibeventCleanup;
     ac->ev.scheduleTimer = valkeyLibeventSetTimeout;
+#ifdef VALKEY_USE_CARES
+    ac->ev.addCaresSocket = valkeyLibeventAddCaresSocket;
+    ac->ev.delCaresSocket = valkeyLibeventDelCaresSocket;
+#endif
     ac->ev.data = e;
 
-    /* Initialize and install read/write events */
-    e->ev = event_new(base, c->fd, EV_READ | EV_WRITE, valkeyLibeventHandler, e);
     e->base = base;
+#ifdef VALKEY_USE_CARES
+    if (c->flags & VALKEY_CONNECT_DEFERRED) {
+        /* DNS is deferred, no fd yet. Initiate async DNS now that the
+         * adapter is attached and can drive c-ares fds. The connect flow
+         * continues from the DNS completion callback. */
+        e->ev = NULL;
+        if (valkeyResolveAsyncStart(ac, c->tcp.host, c->tcp.port) != VALKEY_OK) {
+            vk_free(e);
+            ac->ev.data = NULL;
+            return VALKEY_ERR;
+        }
+    } else
+#endif
+    {
+        /* Initialize and install read/write events */
+        e->ev = event_new(base, c->fd, EV_READ | EV_WRITE, valkeyLibeventHandler, e);
+    }
     return VALKEY_OK;
 }
 

include/valkey/async.h

@@ -69,8 +69,6 @@ typedef void(valkeyDisconnectCallback)(const struct valkeyAsyncContext *, int st
 typedef void(valkeyConnectCallback)(struct valkeyAsyncContext *, int status);
 typedef void(valkeyTimerCallback)(void *timer, void *privdata);
 
-#define VALKEY_TIMEOUT_INACTIVE -1
-
 /* Context for an async connection to Valkey */
 typedef struct valkeyAsyncContext {
     /* Hold the regular context, so it can be realloc'ed. */
@@ -96,6 +94,10 @@ typedef struct valkeyAsyncContext {
         void (*delWrite)(void *privdata);
         void (*cleanup)(void *privdata);
         void (*scheduleTimer)(void *privdata, struct timeval tv);
+        /* c-ares async DNS hooks. If addCaresSocket is NULL, the blocking
+         * fallback (valkeyResolveSync with timeout) is used instead. */
+        void (*addCaresSocket)(void *privdata, int fd, int readable, int writable);
+        void (*delCaresSocket)(void *privdata, int fd);
     } ev;
 
     /* Called when either the connection is terminated due to an error or per
@@ -124,9 +126,16 @@ typedef struct valkeyAsyncContext {
     /* Any configured RESP3 PUSH handler */
     valkeyAsyncPushFn *push_cb;
 
-    /* Replies received since command timeout timer was started, or
-     * VALKEY_TIMEOUT_INACTIVE when no timer is scheduled. */
+    /* Internal timer state */
+    struct valkeyTimerList *timer_list;
+    struct valkeyTimer *connect_timer;
+    struct valkeyTimer *command_timer;
+
+    /* Replies received since command timeout timer was started. */
     int timeout_reply_count;
+
+    /* Async DNS resolution state (used when c-ares is enabled). */
+    void *dns_state;
 } valkeyAsyncContext;
 
 LIBVALKEY_API valkeyAsyncContext *valkeyAsyncConnectWithOptions(const valkeyOptions *options);
@@ -157,6 +166,13 @@ LIBVALKEY_API int valkeyAsyncCommand(valkeyAsyncContext *ac, valkeyCallbackFn *f
 LIBVALKEY_API int valkeyAsyncCommandArgv(valkeyAsyncContext *ac, valkeyCallbackFn *fn, void *privdata, int argc, const char **argv, const size_t *argvlen);
 LIBVALKEY_API int valkeyAsyncFormattedCommand(valkeyAsyncContext *ac, valkeyCallbackFn *fn, void *privdata, const char *cmd, size_t len);
 
+#ifdef VALKEY_USE_CARES
+/* Async DNS resolution (c-ares integration). Used by event-loop adapters. */
+LIBVALKEY_API int valkeyResolveAsyncStart(valkeyAsyncContext *ac, const char *host, int port);
+LIBVALKEY_API void valkeyResolveAsyncHandleEvent(valkeyAsyncContext *ac, int fd, int readable, int writable);
+LIBVALKEY_API void valkeyResolveAsyncFree(valkeyAsyncContext *ac);
+#endif
+
 #ifdef __cplusplus
 }
 #endif

include/valkey/net.h

@@ -38,12 +38,18 @@
 #include "valkey.h"
 #include "visibility.h"
 
+#include <sys/types.h>
+struct addrinfo;
+
 LIBVALKEY_API void valkeyNetClose(valkeyContext *c);
 
 LIBVALKEY_API int valkeyHasMptcp(void);
 LIBVALKEY_API int valkeyCheckSocketError(valkeyContext *c);
 LIBVALKEY_API int valkeyTcpSetTimeout(valkeyContext *c, const struct timeval tv);
 LIBVALKEY_API int valkeyContextConnectTcp(valkeyContext *c, const valkeyOptions *options);
+#ifdef VALKEY_USE_CARES
+LIBVALKEY_API int valkeyTcpConnectNonBlock(valkeyContext *c, struct addrinfo *servinfo);
+#endif
 LIBVALKEY_API int valkeyKeepAlive(valkeyContext *c, int interval);
 LIBVALKEY_API int valkeyCheckConnectDone(valkeyContext *c, int *completed);