summaryrefslogtreecommitdiff
path: root/adapters
diff options
context:
space:
mode:
authorMichaelSuen <crazy95sun@live.cn>2022-01-19 02:59:18 +0800
committerGitHub <noreply@github.com>2022-01-18 10:59:18 -0800
commite73ab2f2329f33823cc89d1070dda8bfb1356bb0 (patch)
treee145a3c479ed2ce01182cb1ca81a2594ec6e21c5 /adapters
parentf2ce5980e6bc9739b85a5ac882980956c80907ba (diff)
Add timeout support for libuv adapter (#1016)
Add timeout to libuv adapter Co-authored-by: sunmingqi <sunmingqi@corp.netease.com> Co-authored-by: sunmingqi <smq222@126.com> Co-authored-by: Michael Grunder <michael.grunder@gmail.com> Co-authored-by: Viktor Söderqvist <viktor@zuiderkwast.se>
Diffstat (limited to 'adapters')
-rw-r--r--adapters/libuv.h162
1 files changed, 104 insertions, 58 deletions
diff --git a/adapters/libuv.h b/adapters/libuv.h
index c120b1b..df0a845 100644
--- a/adapters/libuv.h
+++ b/adapters/libuv.h
@@ -7,111 +7,157 @@
#include <string.h>
typedef struct redisLibuvEvents {
- redisAsyncContext* context;
- uv_poll_t handle;
- int events;
+ redisAsyncContext* context;
+ uv_poll_t handle;
+ uv_timer_t timer;
+ int events;
} redisLibuvEvents;
static void redisLibuvPoll(uv_poll_t* handle, int status, int events) {
- redisLibuvEvents* p = (redisLibuvEvents*)handle->data;
- int ev = (status ? p->events : events);
-
- if (p->context != NULL && (ev & UV_READABLE)) {
- redisAsyncHandleRead(p->context);
- }
- if (p->context != NULL && (ev & UV_WRITABLE)) {
- redisAsyncHandleWrite(p->context);
- }
+ redisLibuvEvents* p = (redisLibuvEvents*)handle->data;
+ int ev = (status ? p->events : events);
+
+ if (p->context != NULL && (ev & UV_READABLE)) {
+ redisAsyncHandleRead(p->context);
+ }
+ if (p->context != NULL && (ev & UV_WRITABLE)) {
+ redisAsyncHandleWrite(p->context);
+ }
}
static void redisLibuvAddRead(void *privdata) {
- redisLibuvEvents* p = (redisLibuvEvents*)privdata;
+ redisLibuvEvents* p = (redisLibuvEvents*)privdata;
- p->events |= UV_READABLE;
+ p->events |= UV_READABLE;
- uv_poll_start(&p->handle, p->events, redisLibuvPoll);
+ uv_poll_start(&p->handle, p->events, redisLibuvPoll);
}
static void redisLibuvDelRead(void *privdata) {
- redisLibuvEvents* p = (redisLibuvEvents*)privdata;
+ redisLibuvEvents* p = (redisLibuvEvents*)privdata;
- p->events &= ~UV_READABLE;
+ p->events &= ~UV_READABLE;
- if (p->events) {
- uv_poll_start(&p->handle, p->events, redisLibuvPoll);
- } else {
- uv_poll_stop(&p->handle);
- }
+ if (p->events) {
+ uv_poll_start(&p->handle, p->events, redisLibuvPoll);
+ } else {
+ uv_poll_stop(&p->handle);
+ }
}
static void redisLibuvAddWrite(void *privdata) {
- redisLibuvEvents* p = (redisLibuvEvents*)privdata;
+ redisLibuvEvents* p = (redisLibuvEvents*)privdata;
- p->events |= UV_WRITABLE;
+ p->events |= UV_WRITABLE;
- uv_poll_start(&p->handle, p->events, redisLibuvPoll);
+ uv_poll_start(&p->handle, p->events, redisLibuvPoll);
}
static void redisLibuvDelWrite(void *privdata) {
- redisLibuvEvents* p = (redisLibuvEvents*)privdata;
+ redisLibuvEvents* p = (redisLibuvEvents*)privdata;
- p->events &= ~UV_WRITABLE;
+ p->events &= ~UV_WRITABLE;
- if (p->events) {
- uv_poll_start(&p->handle, p->events, redisLibuvPoll);
- } else {
- uv_poll_stop(&p->handle);
- }
+ if (p->events) {
+ uv_poll_start(&p->handle, p->events, redisLibuvPoll);
+ } else {
+ uv_poll_stop(&p->handle);
+ }
}
+static void on_timer_close(uv_handle_t *handle) {
+ redisLibuvEvents* p = (redisLibuvEvents*)handle->data;
+ p->timer.data = NULL;
+ if (!p->handle.data) {
+ // both timer and handle are closed
+ hi_free(p);
+ }
+ // else, wait for `on_handle_close`
+}
-static void on_close(uv_handle_t* handle) {
- redisLibuvEvents* p = (redisLibuvEvents*)handle->data;
+static void on_handle_close(uv_handle_t *handle) {
+ redisLibuvEvents* p = (redisLibuvEvents*)handle->data;
+ p->handle.data = NULL;
+ if (!p->timer.data) {
+ // timer never started, or timer already destroyed
+ hi_free(p);
+ }
+ // else, wait for `on_timer_close`
+}
- hi_free(p);
+// libuv removed `status` parameter since v0.11.23
+// see: https://github.com/libuv/libuv/blob/v0.11.23/include/uv.h
+#if (UV_VERSION_MAJOR == 0 && UV_VERSION_MINOR < 11) || \
+ (UV_VERSION_MAJOR == 0 && UV_VERSION_MINOR == 11 && UV_VERSION_PATCH < 23)
+static void redisLibuvTimeout(uv_timer_t *timer, int status) {
+ (void)status; // unused
+#else
+static void redisLibuvTimeout(uv_timer_t *timer) {
+#endif
+ redisLibuvEvents *e = (redisLibuvEvents*)timer->data;
+ redisAsyncHandleTimeout(e->context);
}
+static void redisLibuvSetTimeout(void *privdata, struct timeval tv) {
+ redisLibuvEvents* p = (redisLibuvEvents*)privdata;
+
+ uint64_t millsec = tv.tv_sec * 1000 + tv.tv_usec / 1000.0;
+ if (!p->timer.data) {
+ // timer is uninitialized
+ if (uv_timer_init(p->handle.loop, &p->timer) != 0) {
+ return;
+ }
+ p->timer.data = p;
+ }
+ // updates the timeout if the timer has already started
+ // or start the timer
+ uv_timer_start(&p->timer, redisLibuvTimeout, millsec, 0);
+}
static void redisLibuvCleanup(void *privdata) {
- redisLibuvEvents* p = (redisLibuvEvents*)privdata;
+ redisLibuvEvents* p = (redisLibuvEvents*)privdata;
- p->context = NULL; // indicate that context might no longer exist
- uv_close((uv_handle_t*)&p->handle, on_close);
+ p->context = NULL; // indicate that context might no longer exist
+ if (p->timer.data) {
+ uv_close((uv_handle_t*)&p->timer, on_timer_close);
+ }
+ uv_close((uv_handle_t*)&p->handle, on_handle_close);
}
static int redisLibuvAttach(redisAsyncContext* ac, uv_loop_t* loop) {
- redisContext *c = &(ac->c);
+ redisContext *c = &(ac->c);
- if (ac->ev.data != NULL) {
- return REDIS_ERR;
- }
+ if (ac->ev.data != NULL) {
+ return REDIS_ERR;
+ }
- ac->ev.addRead = redisLibuvAddRead;
- ac->ev.delRead = redisLibuvDelRead;
- ac->ev.addWrite = redisLibuvAddWrite;
- ac->ev.delWrite = redisLibuvDelWrite;
- ac->ev.cleanup = redisLibuvCleanup;
+ ac->ev.addRead = redisLibuvAddRead;
+ ac->ev.delRead = redisLibuvDelRead;
+ ac->ev.addWrite = redisLibuvAddWrite;
+ ac->ev.delWrite = redisLibuvDelWrite;
+ ac->ev.cleanup = redisLibuvCleanup;
+ ac->ev.scheduleTimer = redisLibuvSetTimeout;
- redisLibuvEvents* p = (redisLibuvEvents*)hi_malloc(sizeof(*p));
- if (p == NULL)
- return REDIS_ERR;
+ redisLibuvEvents* p = (redisLibuvEvents*)hi_malloc(sizeof(*p));
+ if (p == NULL)
+ return REDIS_ERR;
- memset(p, 0, sizeof(*p));
+ memset(p, 0, sizeof(*p));
- if (uv_poll_init_socket(loop, &p->handle, c->fd) != 0) {
- return REDIS_ERR;
- }
+ if (uv_poll_init_socket(loop, &p->handle, c->fd) != 0) {
+ return REDIS_ERR;
+ }
- ac->ev.data = p;
- p->handle.data = p;
- p->context = ac;
+ ac->ev.data = p;
+ p->handle.data = p;
+ p->context = ac;
- return REDIS_OK;
+ return REDIS_OK;
}
#endif