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>

3 files changed, 139 insertions, 68 deletions

diff --git a/Makefile b/Makefile
index a293c1a..a2ad84c 100644
--- a/Makefile
+++ b/Makefile
@@ -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)
 	$(CC) -o examples/$@ $(REAL_CFLAGS) -I. $< $(STLIBNAME) $(SSL_STLIBNAME) $(REAL_LDFLAGS) $(SSL_LDFLAGS)
 
-
 ifndef AE_DIR
 hiredis-example-ae:
 	@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)
 endif
 
 ifndef LIBUV_DIR
-hiredis-example-libuv:
-	@echo "Please specify LIBUV_DIR (e.g. ../libuv/)"
-	@false
+# dynamic link libuv.so
+hiredis-example-libuv: examples/example-libuv.c adapters/libuv.h $(STLIBNAME)
+	$(CC) -o examples/$@ $(REAL_CFLAGS) -I. -I$(LIBUV_DIR)/include $< -luv -lpthread -lrt $(STLIBNAME) $(REAL_LDFLAGS)
 else
+# use user provided static lib
 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)
 endif
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
diff --git a/examples/example-libuv.c b/examples/example-libuv.c
index cbde452..53fd04a 100644
--- a/examples/example-libuv.c
+++ b/examples/example-libuv.c
@@ -7,18 +7,33 @@
 #include <async.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) {
     redisReply *reply = r;
-    if (reply == NULL) return;
-    printf("argv[%s]: %s\n", (char*)privdata, reply->str);
+    if (reply == NULL) {
+        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 */
-    redisAsyncDisconnect(c);
+    /* start another request that demonstrate timeout */
+    redisAsyncCommand(c, debugCallback, NULL, "DEBUG SLEEP %f", 1.5);
 }
 
 void connectCallback(const redisAsyncContext *c, int status) {
     if (status != REDIS_OK) {
-        printf("Error: %s\n", c->errstr);
+        printf("connect error: %s\n", c->errstr);
         return;
     }
     printf("Connected...\n");
@@ -26,7 +41,7 @@ void connectCallback(const redisAsyncContext *c, int status) {
 
 void disconnectCallback(const redisAsyncContext *c, int status) {
     if (status != REDIS_OK) {
-        printf("Error: %s\n", c->errstr);
+        printf("disconnect because of error: %s\n", c->errstr);
         return;
     }
     printf("Disconnected...\n");
@@ -49,8 +64,18 @@ int main (int argc, char **argv) {
     redisLibuvAttach(c,loop);
     redisAsyncSetConnectCallback(c,connectCallback);
     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, getCallback, (char*)"end-1", "GET key");
+
     uv_run(loop, UV_RUN_DEFAULT);
     return 0;
 }