diff --git a/CMakeLists.txt b/CMakeLists.txt index f6e6b6a8..c8905610 100644 --- a/CMakeLists.txt +++ b/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) diff --git a/README.md b/README.md index 2705d4b5..a4aacc9c 100644 --- a/README.md +++ b/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`. diff --git a/docs/standalone.md b/docs/standalone.md index 82b666e0..15913171 100644 --- a/docs/standalone.md +++ b/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. diff --git a/examples/async-libev.c b/examples/async-libev.c index 37c85604..dbcdbc6e 100644 --- a/examples/async-libev.c +++ b/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"); diff --git a/examples/async-libevent.c b/examples/async-libevent.c index 943c05a0..a9885439 100644 --- a/examples/async-libevent.c +++ b/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"); diff --git a/include/valkey/adapters/libevent.h b/include/valkey/adapters/libevent.h index ecdf48d0..e33096ff 100644 --- a/include/valkey/adapters/libevent.h +++ b/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; } diff --git a/include/valkey/async.h b/include/valkey/async.h index 365b6435..1ddf2f40 100644 --- a/include/valkey/async.h +++ b/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 diff --git a/include/valkey/net.h b/include/valkey/net.h index 7e844cb8..11bd9571 100644 --- a/include/valkey/net.h +++ b/include/valkey/net.h @@ -38,12 +38,18 @@ #include "valkey.h" #include "visibility.h" +#include +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); diff --git a/include/valkey/valkey.h b/include/valkey/valkey.h index 586cdea8..fdd50d9a 100644 --- a/include/valkey/valkey.h +++ b/include/valkey/valkey.h @@ -103,6 +103,9 @@ typedef SSIZE_T ssize_t; /* Flag specific to use Multipath TCP (MPTCP) */ #define VALKEY_MPTCP 0x2000 +/* Flag indicating connect is deferred (endpoint info stored, connect later). */ +#define VALKEY_CONNECT_DEFERRED 0x4000 + #define VALKEY_KEEPALIVE_INTERVAL 15 /* seconds */ /* number of times we retry to connect in the case of EADDRNOTAVAIL and @@ -226,6 +229,19 @@ typedef struct { /* A user defined PUSH message callback */ valkeyPushFn *push_cb; valkeyAsyncPushFn *async_push_cb; + + /* Optional event-loop adapter. When set, valkeyAsyncConnectWithOptions + * attaches the adapter automatically. */ + int (*attach_fn)(struct valkeyAsyncContext *ac, void *attach_data); + void *attach_data; + + /* Optional async connect/disconnect callbacks. When set, + * valkeyAsyncConnectWithOptions registers them on the context before the + * connect is initiated, so no connect event can be missed even when the + * event loop is already running. Equivalent to calling + * valkeyAsyncSetConnectCallback / valkeyAsyncSetDisconnectCallback. */ + void (*async_connect_callback)(struct valkeyAsyncContext *ac, int status); + void (*async_disconnect_callback)(const struct valkeyAsyncContext *ac, int status); } valkeyOptions; /** diff --git a/src/async.c b/src/async.c index 123c1c05..c0210cc3 100644 --- a/src/async.c +++ b/src/async.c @@ -46,6 +46,7 @@ #include "async_private.h" #include "dict.h" #include "net.h" +#include "timer.h" #include "valkey_private.h" #include "vkutil.h" @@ -139,6 +140,8 @@ static valkeyAsyncContext *valkeyAsyncInitialize(valkeyContext *c) { ac->ev.delWrite = NULL; ac->ev.cleanup = NULL; ac->ev.scheduleTimer = NULL; + ac->ev.addCaresSocket = NULL; + ac->ev.delCaresSocket = NULL; ac->onConnect = NULL; ac->onDisconnect = NULL; @@ -152,7 +155,11 @@ static valkeyAsyncContext *valkeyAsyncInitialize(valkeyContext *c) { ac->sub.schannels = schannels; ac->sub.pending_unsubs = 0; - ac->timeout_reply_count = VALKEY_TIMEOUT_INACTIVE; + ac->timer_list = NULL; + ac->connect_timer = NULL; + ac->command_timer = NULL; + ac->timeout_reply_count = 0; + ac->dns_state = NULL; return ac; oom: @@ -164,7 +171,7 @@ static valkeyAsyncContext *valkeyAsyncInitialize(valkeyContext *c) { /* We want the error field to be accessible directly instead of requiring * an indirection to the valkeyContext struct. */ -static void valkeyAsyncCopyError(valkeyAsyncContext *ac) { +void valkeyAsyncCopyError(valkeyAsyncContext *ac) { if (!ac) return; @@ -194,10 +201,56 @@ valkeyAsyncContext *valkeyAsyncConnectWithOptions(const valkeyOptions *options) valkeyFree(c); return NULL; } + c = &ac->c; /* c was reallocated by valkeyAsyncInitialize */ /* Set any configured async push handler */ valkeyAsyncSetPushCallback(ac, myOptions.async_push_cb); + /* Register connect/disconnect callbacks before connect is initiated. */ + if (myOptions.async_connect_callback) + valkeyAsyncSetConnectCallback(ac, myOptions.async_connect_callback); + if (myOptions.async_disconnect_callback) + valkeyAsyncSetDisconnectCallback(ac, myOptions.async_disconnect_callback); + + /* Attach adapter and initiate connect if adapter was provided. */ + if (myOptions.attach_fn) { + /* Attach the adapter first so its hooks (including any c-ares async + * DNS hooks) are registered before the connect is initiated. While + * VALKEY_CONNECT_DEFERRED is set, the adapter installs its hooks but + * does not register the (not-yet-valid) fd; a c-ares aware adapter + * initiates async DNS from within its attach function. */ + if (myOptions.attach_fn(ac, myOptions.attach_data) != VALKEY_OK) { + valkeySetError(c, VALKEY_ERR_OTHER, "Failed to attach event adapter"); + valkeyAsyncCopyError(ac); + return ac; + } + + if (c->flags & VALKEY_CONNECT_DEFERRED) { + /* When the adapter started async DNS it leaves dns_state set and + * keeps the deferred flag; the connect flow then continues from + * the DNS callback, which adds the write watch once the fd is + * valid. There is no valid fd yet, so return without registering + * read/write interest. */ + if (ac->dns_state != NULL) { + valkeyAsyncCopyError(ac); + return ac; + } + /* Otherwise the adapter has no async DNS support. Perform the + * deferred DNS + connect synchronously now. */ + c->flags &= ~VALKEY_CONNECT_DEFERRED; + c->funcs->connect(c, &myOptions); + if (c->err) { + valkeyAsyncCopyError(ac); + return ac; + } + /* Non-blocking connect in progress, let the event loop + * complete it via valkeyAsyncHandleConnect. */ + c->flags &= ~VALKEY_CONNECTED; + } + + _EL_ADD_WRITE(ac); + } + valkeyAsyncCopyError(ac); return ac; } @@ -240,8 +293,13 @@ int valkeyAsyncSetConnectCallback(valkeyAsyncContext *ac, valkeyConnectCallback /* The common way to detect an established connection is to wait for * the first write event to be fired. This assumes the related event - * library functions are already set. */ - _EL_ADD_WRITE(ac); + * library functions are already set. + * + * While the connect is deferred (async DNS in progress via the options + * path) there is no valid fd yet, so skip registering write interest. + * The write watch is added once the connection is initiated. */ + if (!(ac->c.flags & VALKEY_CONNECT_DEFERRED)) + _EL_ADD_WRITE(ac); return VALKEY_OK; } @@ -377,6 +435,18 @@ static void valkeyAsyncFreeInternal(valkeyAsyncContext *ac) { dictRelease(ac->sub.schannels); } + /* Free internal timers. */ + if (ac->timer_list) { + valkeyTimerListFree(ac->timer_list); + vk_free(ac->timer_list); + ac->timer_list = NULL; + } + + /* Free any in-flight async DNS state before tearing down the event loop. */ +#ifdef VALKEY_USE_CARES + valkeyResolveAsyncFree(ac); +#endif + /* Signal event lib to clean up */ _EL_CLEANUP(ac); @@ -586,7 +656,7 @@ void valkeyProcessCallbacks(valkeyAsyncContext *ac) { c->flags |= VALKEY_SUPPORTS_PUSH; /* Any data from the server means it's alive. */ - if (ac->timeout_reply_count != VALKEY_TIMEOUT_INACTIVE) + if (ac->command_timer != NULL) ac->timeout_reply_count++; /* Send any non-subscribe related PUSH messages to our PUSH handler @@ -665,7 +735,7 @@ void valkeyProcessCallbacks(valkeyAsyncContext *ac) { valkeyAsyncDisconnectInternal(ac); } -static void valkeyAsyncHandleConnectFailure(valkeyAsyncContext *ac) { +void valkeyAsyncHandleConnectFailure(valkeyAsyncContext *ac) { valkeyRunConnectCallback(ac, VALKEY_ERR); valkeyAsyncDisconnectInternal(ac); } @@ -695,6 +765,10 @@ static int valkeyAsyncHandleConnect(valkeyAsyncContext *ac) { * to disconnect. For that reason, permit the function * to delete the context here after callback return. */ + if (ac->connect_timer) { + valkeyTimerDel(ac->timer_list, ac->connect_timer); + ac->connect_timer = NULL; + } c->flags |= VALKEY_CONNECTED; valkeyRunConnectCallback(ac, VALKEY_OK); if ((ac->c.flags & VALKEY_DISCONNECTING)) { @@ -777,32 +851,69 @@ void valkeyAsyncHandleWrite(valkeyAsyncContext *ac) { c->funcs->async_write(ac); } -void valkeyAsyncHandleTimeout(valkeyAsyncContext *ac) { +/* Add a timer and notify the adapter if rescheduling is needed. */ +valkeyTimer *valkeyAsyncAddTimer(valkeyAsyncContext *ac, struct timeval timeout, + valkeyTimerProc proc, void *privdata) { + if (ac->timer_list == NULL) { + ac->timer_list = vk_malloc(sizeof(valkeyTimerList)); + if (ac->timer_list == NULL) + return NULL; + valkeyTimerListInit(ac->timer_list); + } + valkeyTimerList *list = ac->timer_list; + valkeyTimer *old_head = list->head; + valkeyTimer *t = valkeyTimerAdd(list, timeout, proc, privdata); + if (t == NULL) + return NULL; + + /* New timer has the earliest deadline, tell the adapter to wake sooner. */ + if (list->head != old_head && ac->ev.scheduleTimer) + ac->ev.scheduleTimer(ac->ev.data, timeout); + + return t; +} + +#define VALKEY_TIMER_ISSET(tvp) \ + (tvp && ((tvp)->tv_sec || (tvp)->tv_usec)) + +/* Timer callback for connect timeout. */ +static void valkeyAsyncConnectTimeoutCallback(void *privdata) { + valkeyAsyncContext *ac = (valkeyAsyncContext *)privdata; + valkeyContext *c = &(ac->c); + + ac->connect_timer = NULL; + + if (c->flags & VALKEY_CONNECTED) + return; /* Connect completed before timer fired, ignore. */ + + if (!c->err) { + valkeySetError(c, VALKEY_ERR_TIMEOUT, "Timeout"); + valkeyAsyncCopyError(ac); + } + + valkeyRunConnectCallback(ac, VALKEY_ERR); + valkeyAsyncDisconnectInternal(ac); +} + +/* Timer callback for command timeout. */ +static void valkeyAsyncCommandTimeoutCallback(void *privdata) { + valkeyAsyncContext *ac = (valkeyAsyncContext *)privdata; valkeyContext *c = &(ac->c); valkeyCallback cb; - /* must not be called from a callback */ - assert(!(c->flags & VALKEY_IN_CALLBACK)); - if ((c->flags & VALKEY_CONNECTED)) { - if (ac->replies.head == NULL && ac->sub.replies.head == NULL) { - /* Nothing to do - just an idle timeout */ - ac->timeout_reply_count = VALKEY_TIMEOUT_INACTIVE; - return; - } + ac->command_timer = NULL; - if (!ac->c.command_timeout || - (!ac->c.command_timeout->tv_sec && !ac->c.command_timeout->tv_usec)) { - /* A belated connect timeout arriving, ignore */ - return; - } + if (ac->replies.head == NULL && ac->sub.replies.head == NULL) { + /* Nothing to do - just an idle timeout */ + return; + } - /* If replies were received since the timer started, the server is - * alive. Restart the timer rather than timing out. */ - if (ac->timeout_reply_count > 0) { - ac->timeout_reply_count = VALKEY_TIMEOUT_INACTIVE; - refreshTimeout(ac); - return; - } + /* If replies were received since the timer started, the server is + * alive. Restart the timer rather than timing out. */ + if (ac->timeout_reply_count > 0) { + ac->timeout_reply_count = 0; + refreshTimeout(ac); + return; } if (!c->err) { @@ -810,21 +921,56 @@ void valkeyAsyncHandleTimeout(valkeyAsyncContext *ac) { valkeyAsyncCopyError(ac); } - if (!(c->flags & VALKEY_CONNECTED)) { - valkeyRunConnectCallback(ac, VALKEY_ERR); - } - while (valkeyShiftCallback(&ac->replies, &cb) == VALKEY_OK) { valkeyRunCallback(ac, &cb, NULL); } - /** - * TODO: Don't automatically sever the connection, - * rather, allow to ignore responses before the queue is clear - */ valkeyAsyncDisconnectInternal(ac); } +void refreshTimeout(valkeyAsyncContext *ac) { + if (ac->c.flags & VALKEY_CONNECTED) { + struct timeval *tvp = ac->c.command_timeout; + if (!VALKEY_TIMER_ISSET(tvp)) + return; + + /* Don't reset the timer if already active, prevents the timeout from + * never firing when commands are written continuously. */ + if (ac->command_timer != NULL) + return; + + ac->command_timer = valkeyAsyncAddTimer(ac, *tvp, + valkeyAsyncCommandTimeoutCallback, ac); + ac->timeout_reply_count = 0; + } else { + struct timeval *tvp = ac->c.connect_timeout; + if (!VALKEY_TIMER_ISSET(tvp)) + return; + + if (ac->connect_timer != NULL) + return; + + ac->connect_timer = valkeyAsyncAddTimer(ac, *tvp, + valkeyAsyncConnectTimeoutCallback, ac); + } +} + +/* Called by adapters when the scheduled timer expires. Dispatches internal + * timers and reschedules the adapter if more timers are pending. */ +void valkeyAsyncHandleTimeout(valkeyAsyncContext *ac) { + valkeyContext *c = &(ac->c); + struct timeval remaining; + /* must not be called from a callback */ + assert(!(c->flags & VALKEY_IN_CALLBACK)); + + /* Process internal timers. */ + if (ac->timer_list == NULL) + return; + struct timeval *tv = valkeyProcessTimers(ac->timer_list, &remaining); + if (tv && ac->ev.scheduleTimer) + ac->ev.scheduleTimer(ac->ev.data, *tv); +} + static inline int vk_isdigit_ascii(char c) { return (unsigned)(c - '0') < 10; } diff --git a/src/async_private.h b/src/async_private.h index aeea13fd..b9eeb915 100644 --- a/src/async_private.h +++ b/src/async_private.h @@ -31,6 +31,7 @@ #ifndef VALKEY_ASYNC_PRIVATE_H #define VALKEY_ASYNC_PRIVATE_H +#include "timer.h" #include "visibility.h" #define _EL_ADD_READ(ctx) \ @@ -62,28 +63,15 @@ ctx->ev.cleanup = NULL; \ } while (0) -static inline void refreshTimeout(valkeyAsyncContext *ctx) { -#define VALKEY_TIMER_ISSET(tvp) \ - (tvp && ((tvp)->tv_sec || (tvp)->tv_usec)) - - if (ctx->c.flags & VALKEY_CONNECTED) { - /* Don't reset the timer if already active, prevents the timeout from - * never firing when commands are written continuously. */ - if (ctx->timeout_reply_count != VALKEY_TIMEOUT_INACTIVE) - return; - if (ctx->ev.scheduleTimer && VALKEY_TIMER_ISSET(ctx->c.command_timeout)) { - ctx->ev.scheduleTimer(ctx->ev.data, *ctx->c.command_timeout); - ctx->timeout_reply_count = 0; - } - } else { - if (ctx->ev.scheduleTimer && VALKEY_TIMER_ISSET(ctx->c.connect_timeout)) { - ctx->ev.scheduleTimer(ctx->ev.data, *ctx->c.connect_timeout); - } - } -} - /* Visible although private since required by libvalkey_tls.so */ +LIBVALKEY_API void refreshTimeout(valkeyAsyncContext *ac); LIBVALKEY_API void valkeyAsyncDisconnectInternal(valkeyAsyncContext *ac); LIBVALKEY_API void valkeyProcessCallbacks(valkeyAsyncContext *ac); +/* Visible although private since required by dns.c (c-ares async connect). */ +void valkeyAsyncCopyError(valkeyAsyncContext *ac); +void valkeyAsyncHandleConnectFailure(valkeyAsyncContext *ac); +valkeyTimer *valkeyAsyncAddTimer(valkeyAsyncContext *ac, struct timeval timeout, + valkeyTimerProc proc, void *privdata); + #endif /* VALKEY_ASYNC_PRIVATE_H */ diff --git a/src/dns.c b/src/dns.c index 72b5aad6..d27a27a6 100644 --- a/src/dns.c +++ b/src/dns.c @@ -37,6 +37,10 @@ #ifdef VALKEY_USE_CARES #include "alloc.h" +#include "async.h" +#include "async_private.h" +#include "net.h" +#include "valkey_private.h" #include @@ -227,6 +231,17 @@ static void caresPollLoop(ares_channel_t *channel, struct caresSockState *st, } } +/* Determine address family from context flags and hostname. */ +static int caresHintsFamily(const char *host, int flags) { + if ((flags & VALKEY_PREFER_IPV6) && (flags & VALKEY_PREFER_IPV4)) + return AF_UNSPEC; + if (flags & VALKEY_PREFER_IPV6) + return AF_INET6; + if (strchr(host, ':') != NULL) + return AF_INET6; /* IPv6 literal */ + return AF_INET; +} + /* Resolve hostname using c-ares with a poll loop bounded by timeout_ms. * Returns 0 on success (result set), or a getaddrinfo-compatible error code. */ static int valkeyResolveCares(const char *host, int port, int flags, @@ -257,15 +272,7 @@ static int valkeyResolveCares(const char *host, int port, int flags, memset(&hints, 0, sizeof(hints)); hints.ai_socktype = SOCK_STREAM; - - if ((flags & VALKEY_PREFER_IPV6) && (flags & VALKEY_PREFER_IPV4)) - hints.ai_family = AF_UNSPEC; - else if (flags & VALKEY_PREFER_IPV6) - hints.ai_family = AF_INET6; - else if (strchr(host, ':') != NULL) - hints.ai_family = AF_INET6; /* IPv6 literal */ - else - hints.ai_family = AF_INET; + hints.ai_family = caresHintsFamily(host, flags); char portstr[6]; snprintf(portstr, sizeof(portstr), "%d", port); @@ -317,6 +324,242 @@ static int valkeyResolveCares(const char *host, int port, int flags, ares_destroy(channel); return rv; } + +/* --- Async DNS resolution --- */ + +typedef struct valkeyAsyncDns { + ares_channel_t *channel; + struct valkeyAsyncContext *ac; + struct valkeyTimer *timer; + char *host; + int port; + int flags; + int ai_family; + int done; + int failed; +} valkeyAsyncDns; + +static void caresAsyncSockStateCb(void *data, ares_socket_t fd, int readable, int writable) { + valkeyAsyncDns *dns = (valkeyAsyncDns *)data; + valkeyAsyncContext *ac = dns->ac; + + if (!readable && !writable) { + ac->ev.delCaresSocket(ac->ev.data, (int)fd); + } else { + ac->ev.addCaresSocket(ac->ev.data, (int)fd, readable, writable); + } +} + +static void caresAsyncTimerCb(void *privdata); + +static void caresAsyncScheduleTimer(valkeyAsyncDns *dns) { + valkeyAsyncContext *ac = dns->ac; + struct timeval tv, maxtv; + + maxtv.tv_sec = 1; + maxtv.tv_usec = 0; + struct timeval *tvp = ares_timeout(dns->channel, &maxtv, &tv); + + /* Cancel previous timer if active. */ + if (dns->timer) { + valkeyTimerDel(ac->timer_list, dns->timer); + dns->timer = NULL; + } + dns->timer = valkeyAsyncAddTimer(ac, *tvp, caresAsyncTimerCb, dns); +} + +/* Forward declaration. */ +static void caresAsyncCallback(void *arg, int status, int timeouts, struct ares_addrinfo *res); + +/* Mark DNS resolution as complete and record failure. */ +static void caresAsyncFail(valkeyAsyncDns *dns) { + valkeyAsyncContext *ac = dns->ac; + valkeyAsyncCopyError(ac); + dns->done = 1; + dns->failed = 1; +} + +static void caresAsyncConnectWithResult(valkeyAsyncDns *dns, struct ares_addrinfo *ai) { + valkeyAsyncContext *ac = dns->ac; + valkeyContext *c = &ac->c; + struct addrinfo *servinfo = NULL; + + if (caresAddrInfoToAddrInfo(ai, &servinfo) != 0) { + valkeySetError(c, VALKEY_ERR_OOM, "Out of memory"); + ares_freeaddrinfo(ai); + caresAsyncFail(dns); + return; + } + ares_freeaddrinfo(ai); + + if (valkeyTcpConnectNonBlock(c, servinfo) != VALKEY_OK) { + valkeyFreeAddrInfo(servinfo); + caresAsyncFail(dns); + return; + } + + c->flags &= ~VALKEY_CONNECT_DEFERRED; + valkeyFreeAddrInfo(servinfo); + dns->done = 1; + _EL_ADD_WRITE(ac); +} + +static void caresAsyncCallback(void *arg, int status, int timeouts, struct ares_addrinfo *res) { + (void)timeouts; + valkeyAsyncDns *dns = (valkeyAsyncDns *)arg; + valkeyAsyncContext *ac = dns->ac; + valkeyContext *c = &ac->c; + + /* Channel is being destroyed (e.g. during valkeyAsyncFree). */ + if (status == ARES_EDESTRUCTION) { + if (res) + ares_freeaddrinfo(res); + return; + } + + if (status == ARES_SUCCESS && res) { + caresAsyncConnectWithResult(dns, res); + return; + } + + /* Retry with other family if applicable. */ + if ((status == ARES_ENOTFOUND || status == ARES_ENODATA) && + dns->ai_family != AF_UNSPEC) { + if (res) + ares_freeaddrinfo(res); + + dns->ai_family = (dns->ai_family == AF_INET) ? AF_INET6 : AF_INET; + struct ares_addrinfo_hints hints = {0}; + hints.ai_socktype = SOCK_STREAM; + hints.ai_family = dns->ai_family; + + char portstr[6]; + snprintf(portstr, sizeof(portstr), "%d", dns->port); + ares_getaddrinfo(dns->channel, dns->host, portstr, &hints, caresAsyncCallback, dns); + caresAsyncScheduleTimer(dns); + return; + } + + /* DNS failed. */ + if (res) + ares_freeaddrinfo(res); + + int eai = caresStatusToEai(status); + if (eai == EAI_MEMORY) + valkeySetError(c, VALKEY_ERR_OOM, "Out of memory"); + else + valkeySetError(c, VALKEY_ERR_OTHER, gai_strerror(eai)); + caresAsyncFail(dns); +} + +int valkeyResolveAsyncStart(struct valkeyAsyncContext *ac, const char *host, int port) { + valkeyContext *c = &ac->c; + valkeyAsyncDns *dns; + + pthread_once(&cares_init_once, valkeyCaresLibraryInit); + + dns = vk_calloc(1, sizeof(*dns)); + if (dns == NULL) + return VALKEY_ERR; + + dns->ac = ac; + dns->host = vk_strdup(host); + if (dns->host == NULL) { + vk_free(dns); + return VALKEY_ERR; + } + dns->port = port; + dns->flags = c->flags; + dns->ai_family = caresHintsFamily(host, c->flags); + + struct ares_options opts = {0}; + opts.sock_state_cb = caresAsyncSockStateCb; + opts.sock_state_cb_data = dns; + int optmask = ARES_OPT_SOCK_STATE_CB; + + int rv = ares_init_options(&dns->channel, &opts, optmask); + if (rv != ARES_SUCCESS) { + vk_free(dns->host); + vk_free(dns); + return VALKEY_ERR; + } + + /* Store dns state in the context for later access. */ + ac->dns_state = dns; + + struct ares_addrinfo_hints hints = {0}; + hints.ai_socktype = SOCK_STREAM; + hints.ai_family = dns->ai_family; + + char portstr[6]; + snprintf(portstr, sizeof(portstr), "%d", port); + + ares_getaddrinfo(dns->channel, host, portstr, &hints, caresAsyncCallback, dns); + + /* If c-ares resolved synchronously (e.g. IP literals), the callback + * has already fired and set dns->done. Clean up now. */ + if (dns->done) { + int failed = dns->failed; + valkeyResolveAsyncFree(ac); + if (failed) + return VALKEY_ERR; + } else { + caresAsyncScheduleTimer(dns); + } + + return VALKEY_OK; +} + +void valkeyResolveAsyncHandleEvent(struct valkeyAsyncContext *ac, int fd, int readable, int writable) { + valkeyAsyncDns *dns = (valkeyAsyncDns *)ac->dns_state; + if (dns == NULL || dns->channel == NULL) + return; + ares_socket_t rfd = readable ? (ares_socket_t)fd : ARES_SOCKET_BAD; + ares_socket_t wfd = writable ? (ares_socket_t)fd : ARES_SOCKET_BAD; + ares_process_fd(dns->channel, rfd, wfd); + if (dns->done) { + int failed = dns->failed; + valkeyResolveAsyncFree(ac); + if (failed) + valkeyAsyncHandleConnectFailure(ac); + } else { + caresAsyncScheduleTimer(dns); + } +} + +static void caresAsyncTimerCb(void *privdata) { + valkeyAsyncDns *dns = (valkeyAsyncDns *)privdata; + valkeyAsyncContext *ac = dns->ac; + dns->timer = NULL; /* one-shot, already removed */ + if (dns->channel == NULL) + return; + ares_process_fd(dns->channel, ARES_SOCKET_BAD, ARES_SOCKET_BAD); + if (dns->done) { + int failed = dns->failed; + valkeyResolveAsyncFree(ac); + if (failed) + valkeyAsyncHandleConnectFailure(ac); + } else + caresAsyncScheduleTimer(dns); +} + +void valkeyResolveAsyncFree(struct valkeyAsyncContext *ac) { + valkeyAsyncDns *dns = (valkeyAsyncDns *)ac->dns_state; + if (dns == NULL) + return; + if (dns->timer && ac->timer_list) { + valkeyTimerDel(ac->timer_list, dns->timer); + dns->timer = NULL; + } + if (dns->channel) { + ares_destroy(dns->channel); + dns->channel = NULL; + } + vk_free(dns->host); + vk_free(dns); + ac->dns_state = NULL; +} + #endif /* VALKEY_USE_CARES */ int valkeyResolveSync(const char *host, int port, int flags, diff --git a/src/dns.h b/src/dns.h index 8cb7b933..ae359b97 100644 --- a/src/dns.h +++ b/src/dns.h @@ -46,6 +46,20 @@ int valkeyResolveSync(const char *host, int port, int flags, * freeaddrinfo-compatible or c-ares-allocated results. */ #ifdef VALKEY_USE_CARES void valkeyFreeAddrInfo(struct addrinfo *ai); + +struct valkeyAsyncContext; + +/* Initiate async DNS resolution. Returns VALKEY_OK if the query was started + * (completion is via the event loop), or VALKEY_ERR on immediate failure. + * When DNS completes, the connect flow continues on the event loop. */ +int valkeyResolveAsyncStart(struct valkeyAsyncContext *ac, const char *host, int port); + +/* Called by the adapter when a c-ares fd is readable/writable. */ +void valkeyResolveAsyncHandleEvent(struct valkeyAsyncContext *ac, int fd, int readable, int writable); + +/* Free async DNS state. Called during context cleanup. */ +void valkeyResolveAsyncFree(struct valkeyAsyncContext *ac); + #else #define valkeyFreeAddrInfo(ai) freeaddrinfo(ai) #endif diff --git a/src/net.c b/src/net.c index e99f2a3d..46df17b5 100644 --- a/src/net.c +++ b/src/net.c @@ -392,6 +392,82 @@ static int valkeyTcpGetProtocol(int is_mptcp_enabled) { } #endif /* IPPROTO_MPTCP */ +#ifdef VALKEY_USE_CARES +/* Try each address in servinfo, create a non-blocking socket, bind source_addr + * if set, and initiate connect(). Returns VALKEY_OK on first successful + * connect (or EINPROGRESS), VALKEY_ERR if all addresses fail. Sets c->fd. */ +int valkeyTcpConnectNonBlock(valkeyContext *c, struct addrinfo *servinfo) { + struct addrinfo *p, *bservinfo, *b; + valkeyFD s; + int rv, n; + int reuseaddr = (c->flags & VALKEY_REUSEADDR); + + for (p = servinfo; p != NULL; p = p->ai_next) { + s = socket(p->ai_family, p->ai_socktype, valkeyTcpGetProtocol(c->flags & VALKEY_MPTCP)); + if (s == VALKEY_INVALID_FD) + continue; + + c->fd = s; + if (valkeySetBlocking(c, 0) != VALKEY_OK) { + valkeyNetClose(c); + continue; + } + + if (c->tcp.source_addr) { + int bound = 0; + struct addrinfo hints = {0}; + hints.ai_family = p->ai_family; + hints.ai_socktype = p->ai_socktype; + if ((rv = getaddrinfo(c->tcp.source_addr, NULL, &hints, &bservinfo)) != 0) { + char buf[128]; + snprintf(buf, sizeof(buf), "Can't get addr: %s", gai_strerror(rv)); + valkeySetError(c, VALKEY_ERR_OTHER, buf); + valkeyNetClose(c); + return VALKEY_ERR; + } + if (reuseaddr) { + n = 1; + if (setsockopt(s, SOL_SOCKET, SO_REUSEADDR, (char *)&n, sizeof(n)) < 0) { + freeaddrinfo(bservinfo); + valkeyNetClose(c); + continue; + } + } + for (b = bservinfo; b != NULL; b = b->ai_next) { + if (bind(s, b->ai_addr, b->ai_addrlen) != -1) { + bound = 1; + break; + } + } + freeaddrinfo(bservinfo); + if (!bound) { + valkeyNetClose(c); + continue; + } + } + + vk_free(c->saddr); + c->saddr = vk_malloc(p->ai_addrlen); + if (c->saddr == NULL) { + valkeyNetClose(c); + valkeySetError(c, VALKEY_ERR_OOM, "Out of memory"); + return VALKEY_ERR; + } + memcpy(c->saddr, p->ai_addr, p->ai_addrlen); + c->addrlen = p->ai_addrlen; + + if (connect(s, p->ai_addr, p->ai_addrlen) == 0 || errno == EINPROGRESS) { + return VALKEY_OK; + } + + valkeyNetClose(c); + } + + valkeySetErrorFromErrno(c, VALKEY_ERR_IO, NULL); + return VALKEY_ERR; +} +#endif /* VALKEY_USE_CARES */ + int valkeyContextConnectTcp(valkeyContext *c, const valkeyOptions *options) { const struct timeval *timeout = options->connect_timeout; const char *addr = options->endpoint.tcp.ip; @@ -444,6 +520,10 @@ int valkeyContextConnectTcp(valkeyContext *c, const valkeyOptions *options) { c->tcp.source_addr = vk_strdup(source_addr); } + /* Endpoint stored. Return early if connect is deferred. */ + if (c->flags & VALKEY_CONNECT_DEFERRED) + return VALKEY_OK; + /* DNS lookup */ /* TODO: Decide if DNS + TCP connect should share a single connect_timeout * budget rather than each getting the full timeout independently. */ diff --git a/src/timer.c b/src/timer.c new file mode 100644 index 00000000..0d51392c --- /dev/null +++ b/src/timer.c @@ -0,0 +1,162 @@ +/* + * Copyright (c) 2026, the libvalkey contributors + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are met: + * + * * Redistributions of source code must retain the above copyright notice, + * this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above copyright + * notice, this list of conditions and the following disclaimer in the + * documentation and/or other materials provided with the distribution. + * * Neither the name of the copyright holder nor the names of its + * contributors may be used to endorse or promote products derived + * from this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" + * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE + * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE + * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE + * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR + * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF + * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS + * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN + * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) + * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE + * POSSIBILITY OF SUCH DAMAGE. + */ + +#include "fmacros.h" + +#include "timer.h" + +#include +#ifndef _MSC_VER +#include +#else +#include +#include +#endif + +static void valkeyTimerGetMonotonic(struct timeval *tv) { +#ifndef _MSC_VER + struct timespec ts; + clock_gettime(CLOCK_MONOTONIC, &ts); + tv->tv_sec = ts.tv_sec; + tv->tv_usec = (int)(ts.tv_nsec / 1000); +#else + LARGE_INTEGER counter, frequency; + QueryPerformanceCounter(&counter); + QueryPerformanceFrequency(&frequency); + int64_t usec = counter.QuadPart * 1000000 / frequency.QuadPart; + tv->tv_sec = (long)(usec / 1000000); + tv->tv_usec = (long)(usec % 1000000); +#endif +} + +static long long tvdiff_us(const struct timeval *a, const struct timeval *b) { + return (long long)(a->tv_sec - b->tv_sec) * 1000000 + + (a->tv_usec - b->tv_usec); +} + +static void tvadd(struct timeval *result, const struct timeval *a, const struct timeval *b) { + result->tv_sec = a->tv_sec + b->tv_sec; + result->tv_usec = a->tv_usec + b->tv_usec; + if (result->tv_usec >= 1000000) { + result->tv_sec++; + result->tv_usec -= 1000000; + } +} + +/* Insert timer into sorted active list (earliest deadline first). */ +static void timerInsert(valkeyTimerList *list, valkeyTimer *timer) { + valkeyTimer **pp = &list->head; + while (*pp && tvdiff_us(&(*pp)->deadline, &timer->deadline) <= 0) + pp = &(*pp)->next; + timer->next = *pp; + *pp = timer; +} + +void valkeyTimerListInit(valkeyTimerList *list) { + memset(list, 0, sizeof(*list)); +} + +valkeyTimer *valkeyTimerAdd(valkeyTimerList *list, struct timeval timeout, + valkeyTimerProc proc, void *privdata) { + /* Find a free slot. */ + valkeyTimer *t = NULL; + for (int i = 0; i < VALKEY_MAX_TIMERS; i++) { + if (list->timers[i].proc == NULL) { + t = &list->timers[i]; + break; + } + } + if (t == NULL || proc == NULL) + return NULL; + + struct timeval now; + valkeyTimerGetMonotonic(&now); + tvadd(&t->deadline, &now, &timeout); + t->proc = proc; + t->privdata = privdata; + t->next = NULL; + + timerInsert(list, t); + return t; +} + +void valkeyTimerDel(valkeyTimerList *list, valkeyTimer *timer) { + if (timer == NULL || timer->proc == NULL) + return; + + /* Remove from active list. */ + valkeyTimer **pp = &list->head; + while (*pp) { + if (*pp == timer) { + *pp = timer->next; + break; + } + pp = &(*pp)->next; + } + + /* Mark slot as free. */ + timer->proc = NULL; + timer->next = NULL; +} + +struct timeval *valkeyProcessTimers(valkeyTimerList *list, struct timeval *remaining) { + struct timeval now; + valkeyTimerGetMonotonic(&now); + + /* Process at most one expired timer per call. The callback may free + * the context (and this list), so we must not access list after. */ + if (list->head && tvdiff_us(&now, &list->head->deadline) >= 0) { + valkeyTimer *t = list->head; + list->head = t->next; + t->next = NULL; + + valkeyTimerProc proc = t->proc; + void *privdata = t->privdata; + + t->proc = NULL; + + proc(privdata); + /* Context may be freed here, caller must not access list. */ + return NULL; + } + + if (list->head == NULL) + return NULL; + + long long diff_us = tvdiff_us(&list->head->deadline, &now); + remaining->tv_sec = (long)(diff_us / 1000000); + remaining->tv_usec = (long)(diff_us % 1000000); + return remaining; +} + +void valkeyTimerListFree(valkeyTimerList *list) { + for (int i = 0; i < VALKEY_MAX_TIMERS; i++) + list->timers[i].proc = NULL; + list->head = NULL; +} diff --git a/src/timer.h b/src/timer.h new file mode 100644 index 00000000..9c6653f4 --- /dev/null +++ b/src/timer.h @@ -0,0 +1,72 @@ +/* + * Copyright (c) 2026, the libvalkey contributors + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are met: + * + * * Redistributions of source code must retain the above copyright notice, + * this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above copyright + * notice, this list of conditions and the following disclaimer in the + * documentation and/or other materials provided with the distribution. + * * Neither the name of the copyright holder nor the names of its + * contributors may be used to endorse or promote products derived + * from this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" + * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE + * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE + * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE + * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR + * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF + * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS + * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN + * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) + * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE + * POSSIBILITY OF SUCH DAMAGE. + */ + +#ifndef VALKEY_TIMER_H +#define VALKEY_TIMER_H + +#ifndef _MSC_VER +#include +#else +#include +#include +#endif + +#define VALKEY_MAX_TIMERS 4 + +typedef void (*valkeyTimerProc)(void *privdata); + +typedef struct valkeyTimer { + struct timeval deadline; + valkeyTimerProc proc; /* NULL = slot is free */ + void *privdata; + struct valkeyTimer *next; /* sorted active list link */ +} valkeyTimer; + +typedef struct valkeyTimerList { + valkeyTimer timers[VALKEY_MAX_TIMERS]; + valkeyTimer *head; +} valkeyTimerList; + +/* Initialize a timer list (all slots free). */ +void valkeyTimerListInit(valkeyTimerList *list); + +/* Activate a timer. Returns handle or NULL if pool exhausted. */ +valkeyTimer *valkeyTimerAdd(valkeyTimerList *list, struct timeval timeout, + valkeyTimerProc proc, void *privdata); + +/* Deactivate a timer. */ +void valkeyTimerDel(valkeyTimerList *list, valkeyTimer *timer); + +/* Process expired timers. Returns time until next deadline, or NULL if none. */ +struct timeval *valkeyProcessTimers(valkeyTimerList *list, struct timeval *remaining); + +/* Deactivate all timers. */ +void valkeyTimerListFree(valkeyTimerList *list); + +#endif /* VALKEY_TIMER_H */ diff --git a/src/valkey.c b/src/valkey.c index e4667130..33c39075 100644 --- a/src/valkey.c +++ b/src/valkey.c @@ -894,6 +894,9 @@ valkeyContext *valkeyConnectWithOptions(const valkeyOptions *options) { return c; } + if (options->attach_fn && options->type == VALKEY_CONN_TCP) + c->flags |= VALKEY_CONNECT_DEFERRED; + c->funcs->connect(c, options); if (c->err == 0 && c->fd != VALKEY_INVALID_FD && options->command_timeout != NULL && (c->flags & VALKEY_BLOCK)) { diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index ad7c2e20..cdd13aaf 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -97,6 +97,11 @@ target_include_directories(ut_slotmap_update PRIVATE "${PROJECT_SOURCE_DIR}/src" target_link_libraries(ut_slotmap_update valkey_unittest) add_test(NAME ut_slotmap_update COMMAND "$") +add_executable(ut_timer ut_timer.c) +target_include_directories(ut_timer PRIVATE "${PROJECT_SOURCE_DIR}/src") +target_link_libraries(ut_timer valkey_unittest) +add_test(NAME ut_timer COMMAND "$") + if(NOT WIN32 AND NOT CYGWIN AND NOT ENABLE_CARES) add_executable(ut_connect_fallback ut_connect_fallback.c) target_compile_options(ut_connect_fallback PRIVATE -Wno-pedantic) diff --git a/tests/client_test.c b/tests/client_test.c index 7d58b425..07660881 100644 --- a/tests/client_test.c +++ b/tests/client_test.c @@ -1952,6 +1952,100 @@ void test_async_command_parsing(struct config config) { valkeyAsyncFree(ac); } +static void async_ping_cb(valkeyAsyncContext *ac, void *reply, void *privdata) { + int *count = (int *)privdata; + valkeyReply *r = reply; + assert(r != NULL && r->type == VALKEY_REPLY_STATUS); + assert(strcmp(r->str, "PONG") == 0); + (*count)++; + if (*count == 2) { + valkeyAsyncDisconnect(ac); + event_base_loopbreak(base); + } +} + +static void test_async_connect_with_attach_in_options(struct config config) { + test("Async connect with adapter in options (deferred connect): "); + base = event_base_new(); + struct event *timeout = evtimer_new(base, timeout_cb, NULL); + struct timeval timeout_tv = {.tv_sec = 3}; + evtimer_add(timeout, &timeout_tv); + + valkeyOptions options = get_server_tcp_options(config); + options.attach_fn = valkeyLibeventAttachAdapter; + options.attach_data = base; + valkeyAsyncContext *ac = valkeyAsyncConnectWithOptions(&options); + assert(ac != NULL && ac->err == 0); + + int count = 0; + valkeyAsyncCommand(ac, async_ping_cb, &count, "PING"); + valkeyAsyncCommand(ac, async_ping_cb, &count, "PING"); + event_base_dispatch(base); + + event_free(timeout); + event_base_free(base); + test_cond(count == 2); +} + +static void test_async_connect_with_attach_in_options_dns_fail(void) { + test("Async connect with adapter in options (DNS failure): "); + struct event_base *b = event_base_new(); + valkeyOptions options = {0}; + VALKEY_OPTIONS_SET_TCP(&options, "this.host.does.not.exist.invalid", 6379); + options.attach_fn = valkeyLibeventAttachAdapter; + options.attach_data = b; + valkeyAsyncContext *ac = valkeyAsyncConnectWithOptions(&options); + assert(ac != NULL); + test_cond(ac->err != 0); + valkeyAsyncFree(ac); + event_base_free(b); +} + +static void test_async_connect_unix(struct config config) { + test("Async Unix connect: "); + base = event_base_new(); + struct event *timeout = evtimer_new(base, timeout_cb, NULL); + struct timeval timeout_tv = {.tv_sec = 3}; + evtimer_add(timeout, &timeout_tv); + + valkeyAsyncContext *ac = valkeyAsyncConnectUnix(config.unix_sock.path); + assert(ac != NULL && ac->err == 0); + valkeyLibeventAttach(ac, base); + + int count = 0; + valkeyAsyncCommand(ac, async_ping_cb, &count, "PING"); + valkeyAsyncCommand(ac, async_ping_cb, &count, "PING"); + event_base_dispatch(base); + + event_free(timeout); + event_base_free(base); + test_cond(count == 2); +} + +static void test_async_connect_unix_with_attach_in_options(struct config config) { + test("Async Unix connect with adapter in options: "); + base = event_base_new(); + struct event *timeout = evtimer_new(base, timeout_cb, NULL); + struct timeval timeout_tv = {.tv_sec = 3}; + evtimer_add(timeout, &timeout_tv); + + valkeyOptions options = {0}; + VALKEY_OPTIONS_SET_UNIX(&options, config.unix_sock.path); + options.attach_fn = valkeyLibeventAttachAdapter; + options.attach_data = base; + valkeyAsyncContext *ac = valkeyAsyncConnectWithOptions(&options); + assert(ac != NULL && ac->err == 0); + + int count = 0; + valkeyAsyncCommand(ac, async_ping_cb, &count, "PING"); + valkeyAsyncCommand(ac, async_ping_cb, &count, "PING"); + event_base_dispatch(base); + + event_free(timeout); + event_base_free(base); + test_cond(count == 2); +} + static void test_pubsub_handling(struct config config) { test("Subscribe, handle published message and unsubscribe: "); /* Setup event dispatcher with a testcase timeout */ @@ -2997,6 +3091,8 @@ int main(int argc, char **argv) { disconnect(c, 0); test_async_command_parsing(cfg); + test_async_connect_with_attach_in_options(cfg); + test_async_connect_with_attach_in_options_dns_fail(); test_pubsub_handling(cfg); test_pubsub_multiple_channels(cfg); test_monitor(cfg); @@ -3028,6 +3124,13 @@ int main(int argc, char **argv) { } #endif /* IPPROTO_MPTCP */ + if (test_unix_socket) { + printf("\nTesting asynchronous API against Unix socket (%s):\n", cfg.unix_sock.path); + cfg.type = CONN_UNIX; + test_async_connect_unix(cfg); + test_async_connect_unix_with_attach_in_options(cfg); + } + #endif /* VALKEY_TEST_ASYNC */ cfg.type = CONN_TCP; diff --git a/tests/ut_timer.c b/tests/ut_timer.c new file mode 100644 index 00000000..4093e6bf --- /dev/null +++ b/tests/ut_timer.c @@ -0,0 +1,129 @@ +/* + * Unit tests for src/timer.c + */ + +#define _DEFAULT_SOURCE /* for usleep() */ + +#include "timer.h" + +#include +#include +#ifdef _MSC_VER +#include +#define usleep(us) Sleep((us) / 1000) +#else +#include +#endif + +static int fired_count; +static void *fired_data; + +static void test_cb(void *privdata) { + fired_count++; + fired_data = privdata; +} + +static void test_add_and_fire(void) { + printf(" test_add_and_fire: "); + valkeyTimerList list; + valkeyTimerListInit(&list); + + fired_count = 0; + fired_data = NULL; + int data = 42; + struct timeval iv = {.tv_sec = 0, .tv_usec = 10000}; /* 10ms */ + valkeyTimer *t = valkeyTimerAdd(&list, iv, test_cb, &data); + assert(t != NULL); + assert(list.head == t); + + /* Not yet expired. */ + struct timeval next; + valkeyProcessTimers(&list, &next); + assert(fired_count == 0); + + /* Wait for timer to expire. */ + usleep(15000); /* 15ms */ + valkeyProcessTimers(&list, &next); + assert(fired_count == 1); + assert(fired_data == &data); + + /* Timer is one-shot, list should be empty. */ + assert(list.head == NULL); + + valkeyTimerListFree(&list); + printf("PASSED\n"); +} + +static void test_ordering(void) { + printf(" test_ordering: "); + valkeyTimerList list; + valkeyTimerListInit(&list); + + struct timeval iv1 = {.tv_sec = 0, .tv_usec = 50000}; /* 50ms */ + struct timeval iv2 = {.tv_sec = 0, .tv_usec = 10000}; /* 10ms */ + struct timeval iv3 = {.tv_sec = 0, .tv_usec = 30000}; /* 30ms */ + + valkeyTimer *t1 = valkeyTimerAdd(&list, iv1, test_cb, NULL); + valkeyTimer *t2 = valkeyTimerAdd(&list, iv2, test_cb, NULL); + valkeyTimer *t3 = valkeyTimerAdd(&list, iv3, test_cb, NULL); + + /* Should be ordered: t2 (10ms) -> t3 (30ms) -> t1 (50ms) */ + assert(list.head == t2); + assert(t2->next == t3); + assert(t3->next == t1); + assert(t1->next == NULL); + + valkeyTimerListFree(&list); + printf("PASSED\n"); +} + +static void test_cancel(void) { + printf(" test_cancel: "); + valkeyTimerList list; + valkeyTimerListInit(&list); + + fired_count = 0; + struct timeval iv = {.tv_sec = 0, .tv_usec = 10000}; + valkeyTimer *t = valkeyTimerAdd(&list, iv, test_cb, NULL); + assert(list.head == t); + + valkeyTimerDel(&list, t); + assert(list.head == NULL); + + usleep(15000); + struct timeval next; + struct timeval *ret = valkeyProcessTimers(&list, &next); + assert(ret == NULL); /* No timers */ + assert(fired_count == 0); + + printf("PASSED\n"); +} + +static void test_next_deadline(void) { + printf(" test_next_deadline: "); + valkeyTimerList list; + valkeyTimerListInit(&list); + + struct timeval iv = {.tv_sec = 1, .tv_usec = 0}; /* 1s */ + valkeyTimerAdd(&list, iv, test_cb, NULL); + + struct timeval next; + struct timeval *ret = valkeyProcessTimers(&list, &next); + /* Should be close to 1s remaining. */ + assert(ret != NULL); + long remaining_us = next.tv_sec * 1000000 + next.tv_usec; + assert(remaining_us > 900000 && remaining_us <= 1000000); + + valkeyTimerListFree(&list); + printf("PASSED\n"); +} + +int main(void) { + printf("Testing timer module:\n"); + test_add_and_fire(); + test_ordering(); + test_cancel(); + test_next_deadline(); + printf("All timer tests passed.\n"); + return 0; +}