Mine/add timer system

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)
 

examples/Makefile

@@ -16,6 +16,7 @@ endif
 # Define examples
 EXAMPLES=example-blocking example-blocking-push example-async-libevent \
 	 example-async-libev example-async-glib example-async-poll \
+	 example-async-disable-timeout \
 	 example-cluster-async example-cluster-clientside-caching-async \
 	 example-cluster-simple
 
@@ -71,6 +72,9 @@ example-async-macosx: async-macosx.c $(STLIBNAME)
 example-async-poll: async-poll.c $(STLIBNAME)
 	$(CC) -o $@ $(CFLAGS) $< $(STLIBNAME)
 
+example-async-disable-timeout: async-disable-timeout.c $(STLIBNAME)
+	$(CC) -o $@ $(CFLAGS) $< $(STLIBNAME)
+
 ifndef AE_DIR
 example-async-ae:
 	@echo "Please specify AE_DIR (e.g. <valkey repository>/src)"

examples/async-disable-timeout.c

@@ -0,0 +1,110 @@
+#include <valkey/async.h>
+
+#include <valkey/adapters/poll.h>
+
+#include <signal.h>
+#include <stdio.h>
+#include <stdlib.h>
+
+static int exit_loop = 0;
+static int final_status = 1;
+static double debug_sleep_seconds = 2.0;
+
+static void debugSleepCallback(valkeyAsyncContext *ac, void *reply, void *privdata) {
+    (void)privdata;
+
+    if (reply == NULL) {
+        printf("DEBUG SLEEP callback received NULL reply: %s\n",
+               ac->errstr ? ac->errstr : "unknown error");
+        return;
+    }
+
+    valkeyReply *r = reply;
+    printf("DEBUG SLEEP %.3g completed with reply type %d\n",
+           debug_sleep_seconds, r->type);
+    printf("The disabled command timeout did not tear down the connection.\n");
+    final_status = 0;
+    valkeyAsyncDisconnect(ac);
+}
+
+static void connectCallback(valkeyAsyncContext *ac, int status) {
+    if (status != VALKEY_OK) {
+        printf("Connect failed: %s\n", ac->errstr);
+        exit_loop = 1;
+        return;
+    }
+
+    printf("Connected. Arming a 500ms command timeout.\n");
+    if (valkeyAsyncSetTimeout(ac, (struct timeval){.tv_sec = 0, .tv_usec = 500000}) != VALKEY_OK) {
+        printf("valkeyAsyncSetTimeout failed: %s\n", ac->errstr);
+        exit_loop = 1;
+        return;
+    }
+
+    printf("Queueing DEBUG SLEEP %.3g; this arms the internal command timer.\n",
+           debug_sleep_seconds);
+    if (valkeyAsyncCommand(ac, debugSleepCallback, NULL, "DEBUG SLEEP %f",
+                           debug_sleep_seconds) != VALKEY_OK) {
+        printf("valkeyAsyncCommand failed: %s\n", ac->errstr);
+        exit_loop = 1;
+        return;
+    }
+
+    printf("Disabling the timeout with valkeyAsyncSetTimeout({0,0}).\n");
+    if (valkeyAsyncSetTimeout(ac, (struct timeval){0, 0}) != VALKEY_OK) {
+        printf("valkeyAsyncSetTimeout disable failed: %s\n", ac->errstr);
+        exit_loop = 1;
+    }
+}
+
+static void disconnectCallback(const valkeyAsyncContext *ac, int status) {
+    exit_loop = 1;
+    if (status != VALKEY_OK) {
+        printf("Disconnected with error: %s\n", ac->errstr);
+        return;
+    }
+
+    printf("Disconnected cleanly.\n");
+}
+
+int main(int argc, char **argv) {
+    const char *host = "127.0.0.1";
+    int port = 9999;
+
+    if (argc > 1)
+        port = atoi(argv[1]);
+    if (argc > 2)
+        debug_sleep_seconds = atof(argv[2]);
+
+#ifndef _WIN32
+    signal(SIGPIPE, SIG_IGN);
+#endif
+
+    printf("Using %s:%d. Expected old behavior: timeout after ~500ms.\n", host, port);
+    printf("Expected fixed behavior: DEBUG SLEEP reply after %.3g seconds.\n",
+           debug_sleep_seconds);
+
+    valkeyAsyncContext *ac = valkeyAsyncConnect(host, port);
+    if (ac == NULL) {
+        printf("valkeyAsyncConnect returned NULL\n");
+        return 1;
+    }
+    if (ac->err) {
+        printf("Connect setup failed: %s\n", ac->errstr);
+        return 1;
+    }
+
+    if (valkeyPollAttach(ac) != VALKEY_OK) {
+        printf("valkeyPollAttach failed\n");
+        valkeyAsyncFree(ac);
+        return 1;
+    }
+
+    valkeyAsyncSetConnectCallback(ac, connectCallback);
+    valkeyAsyncSetDisconnectCallback(ac, disconnectCallback);
+
+    while (!exit_loop)
+        valkeyPollTick(ac, 0.05);
+
+    return final_status;
+}

include/valkey/adapters/libhv.h

@@ -6,6 +6,7 @@
 #include "../valkey.h"
 
 #include <hv/hloop.h>
+#include <stdint.h>
 
 typedef struct valkeyLibhvEvents {
     hio_t *io;
@@ -61,13 +62,28 @@ static void valkeyLibhvTimeout(htimer_t *timer) {
     valkeyAsyncHandleTimeout((valkeyAsyncContext *)hevent_userdata(io));
 }
 
+static uint32_t valkeyLibhvTimevalToMillis(struct timeval tv) {
+    uint64_t millis = 0;
+
+    if (tv.tv_sec > 0) {
+        if ((uint64_t)tv.tv_sec > UINT32_MAX / 1000)
+            return UINT32_MAX;
+        millis = (uint64_t)tv.tv_sec * 1000;
+    }
+
+    if (tv.tv_usec)
+        millis += ((uint64_t)tv.tv_usec + 999) / 1000;
+
+    return millis > UINT32_MAX ? UINT32_MAX : (uint32_t)millis;
+}
+
 static void valkeyLibhvSetTimeout(void *privdata, struct timeval tv) {
     valkeyLibhvEvents *events;
     uint32_t millis;
     hloop_t *loop;
 
     events = (valkeyLibhvEvents *)privdata;
-    millis = tv.tv_sec * 1000 + tv.tv_usec / 1000;
+    millis = valkeyLibhvTimevalToMillis(tv);
 
     if (millis == 0) {
         /* Libhv disallows zero'd timers so treat this as a delete or NO OP */

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. */
@@ -124,8 +122,12 @@ 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;
 } valkeyAsyncContext;