Unverified Commit e73ab2f2 authored by MichaelSuen's avatar MichaelSuen Committed by GitHub
Browse files

Add timeout support for libuv adapter (#1016)



Add timeout to libuv adapter
Co-authored-by: default avatarsunmingqi <sunmingqi@corp.netease.com>
Co-authored-by: default avatarsunmingqi <smq222@126.com>
Co-authored-by: default avatarMichael Grunder <michael.grunder@gmail.com>
Co-authored-by: default avatarViktor Söderqvist <viktor@zuiderkwast.se>
parent f2ce5980
...@@ -192,7 +192,6 @@ hiredis-example-macosx: examples/example-macosx.c adapters/macosx.h $(STLIBNAME) ...@@ -192,7 +192,6 @@ hiredis-example-macosx: examples/example-macosx.c adapters/macosx.h $(STLIBNAME)
hiredis-example-ssl: examples/example-ssl.c $(STLIBNAME) $(SSL_STLIBNAME) hiredis-example-ssl: examples/example-ssl.c $(STLIBNAME) $(SSL_STLIBNAME)
$(CC) -o examples/$@ $(REAL_CFLAGS) -I. $< $(STLIBNAME) $(SSL_STLIBNAME) $(REAL_LDFLAGS) $(SSL_LDFLAGS) $(CC) -o examples/$@ $(REAL_CFLAGS) -I. $< $(STLIBNAME) $(SSL_STLIBNAME) $(REAL_LDFLAGS) $(SSL_LDFLAGS)
ifndef AE_DIR ifndef AE_DIR
hiredis-example-ae: hiredis-example-ae:
@echo "Please specify AE_DIR (e.g. <redis repository>/src)" @echo "Please specify AE_DIR (e.g. <redis repository>/src)"
...@@ -203,10 +202,11 @@ hiredis-example-ae: examples/example-ae.c adapters/ae.h $(STLIBNAME) ...@@ -203,10 +202,11 @@ hiredis-example-ae: examples/example-ae.c adapters/ae.h $(STLIBNAME)
endif endif
ifndef LIBUV_DIR ifndef LIBUV_DIR
hiredis-example-libuv: # dynamic link libuv.so
@echo "Please specify LIBUV_DIR (e.g. ../libuv/)" hiredis-example-libuv: examples/example-libuv.c adapters/libuv.h $(STLIBNAME)
@false $(CC) -o examples/$@ $(REAL_CFLAGS) -I. -I$(LIBUV_DIR)/include $< -luv -lpthread -lrt $(STLIBNAME) $(REAL_LDFLAGS)
else else
# use user provided static lib
hiredis-example-libuv: examples/example-libuv.c adapters/libuv.h $(STLIBNAME) hiredis-example-libuv: examples/example-libuv.c adapters/libuv.h $(STLIBNAME)
$(CC) -o examples/$@ $(REAL_CFLAGS) -I. -I$(LIBUV_DIR)/include $< $(LIBUV_DIR)/.libs/libuv.a -lpthread -lrt $(STLIBNAME) $(REAL_LDFLAGS) $(CC) -o examples/$@ $(REAL_CFLAGS) -I. -I$(LIBUV_DIR)/include $< $(LIBUV_DIR)/.libs/libuv.a -lpthread -lrt $(STLIBNAME) $(REAL_LDFLAGS)
endif endif
......
...@@ -7,111 +7,157 @@ ...@@ -7,111 +7,157 @@
#include <string.h> #include <string.h>
typedef struct redisLibuvEvents { typedef struct redisLibuvEvents {
redisAsyncContext* context; redisAsyncContext* context;
uv_poll_t handle; uv_poll_t handle;
int events; uv_timer_t timer;
int events;
} redisLibuvEvents; } redisLibuvEvents;
static void redisLibuvPoll(uv_poll_t* handle, int status, int events) { static void redisLibuvPoll(uv_poll_t* handle, int status, int events) {
redisLibuvEvents* p = (redisLibuvEvents*)handle->data; redisLibuvEvents* p = (redisLibuvEvents*)handle->data;
int ev = (status ? p->events : events); int ev = (status ? p->events : events);
if (p->context != NULL && (ev & UV_READABLE)) { if (p->context != NULL && (ev & UV_READABLE)) {
redisAsyncHandleRead(p->context); redisAsyncHandleRead(p->context);
} }
if (p->context != NULL && (ev & UV_WRITABLE)) { if (p->context != NULL && (ev & UV_WRITABLE)) {
redisAsyncHandleWrite(p->context); redisAsyncHandleWrite(p->context);
} }
} }
static void redisLibuvAddRead(void *privdata) { 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) { static void redisLibuvDelRead(void *privdata) {
redisLibuvEvents* p = (redisLibuvEvents*)privdata; redisLibuvEvents* p = (redisLibuvEvents*)privdata;
p->events &= ~UV_READABLE; p->events &= ~UV_READABLE;
if (p->events) { if (p->events) {
uv_poll_start(&p->handle, p->events, redisLibuvPoll); uv_poll_start(&p->handle, p->events, redisLibuvPoll);
} else { } else {
uv_poll_stop(&p->handle); uv_poll_stop(&p->handle);
} }
} }
static void redisLibuvAddWrite(void *privdata) { 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) { static void redisLibuvDelWrite(void *privdata) {
redisLibuvEvents* p = (redisLibuvEvents*)privdata; redisLibuvEvents* p = (redisLibuvEvents*)privdata;
p->events &= ~UV_WRITABLE; p->events &= ~UV_WRITABLE;
if (p->events) { if (p->events) {
uv_poll_start(&p->handle, p->events, redisLibuvPoll); uv_poll_start(&p->handle, p->events, redisLibuvPoll);
} else { } else {
uv_poll_stop(&p->handle); 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) { static void on_handle_close(uv_handle_t *handle) {
redisLibuvEvents* p = (redisLibuvEvents*)handle->data; 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) { static void redisLibuvCleanup(void *privdata) {
redisLibuvEvents* p = (redisLibuvEvents*)privdata; redisLibuvEvents* p = (redisLibuvEvents*)privdata;
p->context = NULL; // indicate that context might no longer exist p->context = NULL; // indicate that context might no longer exist
uv_close((uv_handle_t*)&p->handle, on_close); 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) { static int redisLibuvAttach(redisAsyncContext* ac, uv_loop_t* loop) {
redisContext *c = &(ac->c); redisContext *c = &(ac->c);
if (ac->ev.data != NULL) { if (ac->ev.data != NULL) {
return REDIS_ERR; return REDIS_ERR;
} }
ac->ev.addRead = redisLibuvAddRead; ac->ev.addRead = redisLibuvAddRead;
ac->ev.delRead = redisLibuvDelRead; ac->ev.delRead = redisLibuvDelRead;
ac->ev.addWrite = redisLibuvAddWrite; ac->ev.addWrite = redisLibuvAddWrite;
ac->ev.delWrite = redisLibuvDelWrite; ac->ev.delWrite = redisLibuvDelWrite;
ac->ev.cleanup = redisLibuvCleanup; ac->ev.cleanup = redisLibuvCleanup;
ac->ev.scheduleTimer = redisLibuvSetTimeout;
redisLibuvEvents* p = (redisLibuvEvents*)hi_malloc(sizeof(*p)); redisLibuvEvents* p = (redisLibuvEvents*)hi_malloc(sizeof(*p));
if (p == NULL) if (p == NULL)
return REDIS_ERR; return REDIS_ERR;
memset(p, 0, sizeof(*p)); memset(p, 0, sizeof(*p));
if (uv_poll_init_socket(loop, &p->handle, c->fd) != 0) { if (uv_poll_init_socket(loop, &p->handle, c->fd) != 0) {
return REDIS_ERR; return REDIS_ERR;
} }
ac->ev.data = p; ac->ev.data = p;
p->handle.data = p; p->handle.data = p;
p->context = ac; p->context = ac;
return REDIS_OK; return REDIS_OK;
} }
#endif #endif
...@@ -7,18 +7,33 @@ ...@@ -7,18 +7,33 @@
#include <async.h> #include <async.h>
#include <adapters/libuv.h> #include <adapters/libuv.h>
void debugCallback(redisAsyncContext *c, void *r, void *privdata) {
(void)privdata; //unused
redisReply *reply = r;
if (reply == NULL) {
/* The DEBUG SLEEP command will almost always fail, because we have set a 1 second timeout */
printf("`DEBUG SLEEP` error: %s\n", c->errstr ? c->errstr : "unknown error");
return;
}
/* Disconnect after receiving the reply of DEBUG SLEEP (which will not)*/
redisAsyncDisconnect(c);
}
void getCallback(redisAsyncContext *c, void *r, void *privdata) { void getCallback(redisAsyncContext *c, void *r, void *privdata) {
redisReply *reply = r; redisReply *reply = r;
if (reply == NULL) return; if (reply == NULL) {
printf("argv[%s]: %s\n", (char*)privdata, reply->str); printf("`GET key` error: %s\n", c->errstr ? c->errstr : "unknown error");
return;
}
printf("`GET key` result: argv[%s]: %s\n", (char*)privdata, reply->str);
/* Disconnect after receiving the reply to GET */ /* start another request that demonstrate timeout */
redisAsyncDisconnect(c); redisAsyncCommand(c, debugCallback, NULL, "DEBUG SLEEP %f", 1.5);
} }
void connectCallback(const redisAsyncContext *c, int status) { void connectCallback(const redisAsyncContext *c, int status) {
if (status != REDIS_OK) { if (status != REDIS_OK) {
printf("Error: %s\n", c->errstr); printf("connect error: %s\n", c->errstr);
return; return;
} }
printf("Connected...\n"); printf("Connected...\n");
...@@ -26,7 +41,7 @@ void connectCallback(const redisAsyncContext *c, int status) { ...@@ -26,7 +41,7 @@ void connectCallback(const redisAsyncContext *c, int status) {
void disconnectCallback(const redisAsyncContext *c, int status) { void disconnectCallback(const redisAsyncContext *c, int status) {
if (status != REDIS_OK) { if (status != REDIS_OK) {
printf("Error: %s\n", c->errstr); printf("disconnect because of error: %s\n", c->errstr);
return; return;
} }
printf("Disconnected...\n"); printf("Disconnected...\n");
...@@ -49,8 +64,18 @@ int main (int argc, char **argv) { ...@@ -49,8 +64,18 @@ int main (int argc, char **argv) {
redisLibuvAttach(c,loop); redisLibuvAttach(c,loop);
redisAsyncSetConnectCallback(c,connectCallback); redisAsyncSetConnectCallback(c,connectCallback);
redisAsyncSetDisconnectCallback(c,disconnectCallback); redisAsyncSetDisconnectCallback(c,disconnectCallback);
redisAsyncSetTimeout(c, (struct timeval){ .tv_sec = 1, .tv_usec = 0});
/*
In this demo, we first `set key`, then `get key` to demonstrate the basic usage of libuv adapter.
Then in `getCallback`, we start a `debug sleep` command to create 1.5 second long request.
Because we have set a 1 second timeout to the connection, the command will always fail with a
timeout error, which is shown in the `debugCallback`.
*/
redisAsyncCommand(c, NULL, NULL, "SET key %b", argv[argc-1], strlen(argv[argc-1])); redisAsyncCommand(c, NULL, NULL, "SET key %b", argv[argc-1], strlen(argv[argc-1]));
redisAsyncCommand(c, getCallback, (char*)"end-1", "GET key"); redisAsyncCommand(c, getCallback, (char*)"end-1", "GET key");
uv_run(loop, UV_RUN_DEFAULT); uv_run(loop, UV_RUN_DEFAULT);
return 0; return 0;
} }
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment