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/examples/Makefile b/examples/Makefile index 3196e62e..99cc942f 100644 --- a/examples/Makefile +++ b/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. /src)" diff --git a/examples/async-disable-timeout.c b/examples/async-disable-timeout.c new file mode 100644 index 00000000..9cb4ac03 --- /dev/null +++ b/examples/async-disable-timeout.c @@ -0,0 +1,110 @@ +#include + +#include + +#include +#include +#include + +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; +} diff --git a/include/valkey/adapters/libhv.h b/include/valkey/adapters/libhv.h index d73ec1a2..5f24bf68 100644 --- a/include/valkey/adapters/libhv.h +++ b/include/valkey/adapters/libhv.h @@ -6,6 +6,7 @@ #include "../valkey.h" #include +#include 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 */ diff --git a/include/valkey/async.h b/include/valkey/async.h index 365b6435..8140c9c6 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. */ @@ -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; diff --git a/src/async.c b/src/async.c index 123c1c05..2158af39 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" @@ -152,7 +153,10 @@ 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; return ac; oom: @@ -377,6 +381,13 @@ 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; + } + /* Signal event lib to clean up */ _EL_CLEANUP(ac); @@ -586,7 +597,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 @@ -695,6 +706,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 +792,72 @@ 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 (!VALKEY_TIMER_ISSET(ac->c.command_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 (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 = 0; + refreshTimeout(ac); + return; } if (!c->err) { @@ -810,21 +865,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; } @@ -1221,5 +1311,11 @@ int valkeyAsyncSetTimeout(valkeyAsyncContext *ac, struct timeval tv) { *ac->c.command_timeout = tv; } + if (tv.tv_sec == 0 && tv.tv_usec == 0 && ac->command_timer != NULL) { + valkeyTimerDel(ac->timer_list, ac->command_timer); + ac->command_timer = NULL; + ac->timeout_reply_count = 0; + } + return VALKEY_OK; } diff --git a/src/async_private.h b/src/async_private.h index aeea13fd..dd30fd05 100644 --- a/src/async_private.h +++ b/src/async_private.h @@ -62,27 +62,8 @@ 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); diff --git a/src/timer.c b/src/timer.c new file mode 100644 index 00000000..b6fb6fb7 --- /dev/null +++ b/src/timer.c @@ -0,0 +1,163 @@ +/* + * 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 sec = counter.QuadPart / frequency.QuadPart; + int64_t rem = counter.QuadPart % frequency.QuadPart; + tv->tv_sec = (long)sec; + tv->tv_usec = (long)(rem * 1000000 / frequency.QuadPart); +#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/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/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; +}