Skip to content

Commit

Permalink
add a command_timeout to redisContextOptions (#839)
Browse files Browse the repository at this point in the history 
Add an additional timeout so the user has a convenient way of controlling distinct connect and command timeouts
  • Loading branch information
@valentinogeron
valentinogeron committed Jul 26, 2020
1 parent 3bb9853 commit 38b5ae5
Show file tree
Hide file tree
Showing 7 changed files with 80 additions and 42 deletions.
12 changes: 6 additions & 6 deletions async.c
View file
Expand Up @@ -868,19 +868,19 @@ redisAsyncPushFn *redisAsyncSetPushCallback(redisAsyncContext *ac, redisAsyncPus
}

int redisAsyncSetTimeout(redisAsyncContext *ac, struct timeval tv) {
if (!ac->c.timeout) {
ac->c.timeout = hi_calloc(1, sizeof(tv));
if (ac->c.timeout == NULL) {
if (!ac->c.command_timeout) {
ac->c.command_timeout = hi_calloc(1, sizeof(tv));
if (ac->c.command_timeout == NULL) {
__redisSetError(&ac->c, REDIS_ERR_OOM, "Out of memory");
__redisAsyncCopyError(ac);
return REDIS_ERR;
}
}

if (tv.tv_sec != ac->c.timeout->tv_sec ||
tv.tv_usec != ac->c.timeout->tv_usec)
if (tv.tv_sec != ac->c.command_timeout->tv_sec ||
tv.tv_usec != ac->c.command_timeout->tv_usec)
{
*ac->c.timeout = tv;
*ac->c.command_timeout = tv;
}

return REDIS_OK;
Expand Down
17 changes: 8 additions & 9 deletions async_private.h
View file
Expand Up @@ -54,15 +54,14 @@
} while(0);

static inline void refreshTimeout(redisAsyncContext *ctx) {
if (ctx->c.timeout && ctx->ev.scheduleTimer &&
(ctx->c.timeout->tv_sec || ctx->c.timeout->tv_usec)) {
ctx->ev.scheduleTimer(ctx->ev.data, *ctx->c.timeout);
// } else {
// printf("Not scheduling timer.. (tmo=%p)\n", ctx->c.timeout);
// if (ctx->c.timeout){
// printf("tv_sec: %u. tv_usec: %u\n", ctx->c.timeout->tv_sec,
// ctx->c.timeout->tv_usec);
// }
if (!(ctx->c.flags & REDIS_CONNECTED)) {
if (ctx->c.connect_timeout && ctx->ev.scheduleTimer &&
(ctx->c.connect_timeout->tv_sec || ctx->c.connect_timeout->tv_usec)) {
ctx->ev.scheduleTimer(ctx->ev.data, *ctx->c.connect_timeout);
}
} else if (ctx->c.command_timeout && ctx->ev.scheduleTimer &&
(ctx->c.command_timeout->tv_sec || ctx->c.command_timeout->tv_usec)) {
ctx->ev.scheduleTimer(ctx->ev.data, *ctx->c.command_timeout);
}
}

Expand Down
2 changes: 1 addition & 1 deletion examples/example-libevent.c
View file
Expand Up @@ -47,7 +47,7 @@ int main (int argc, char **argv) {
REDIS_OPTIONS_SET_TCP(&options, "127.0.0.1", 6379);
struct timeval tv = {0};
tv.tv_sec = 1;
options.timeout = &tv;
options.connect_timeout = &tv;


redisAsyncContext *c = redisAsyncConnectWithOptions(&options);
Expand Down
2 changes: 1 addition & 1 deletion examples/example-ssl.c
View file
Expand Up @@ -33,7 +33,7 @@ int main(int argc, char **argv) {
struct timeval tv = { 1, 500000 }; // 1.5 seconds
redisOptions options = {0};
REDIS_OPTIONS_SET_TCP(&options, hostname, port);
options.timeout = &tv;
options.connect_timeout = &tv;
c = redisConnectWithOptions(&options);

if (c == NULL || c->err) {
Expand Down
38 changes: 29 additions & 9 deletions hiredis.c
View file
Expand Up @@ -44,6 +44,9 @@
#include "async.h"
#include "win32.h"

extern int redisContextUpdateConnectTimeout(redisContext *c, const struct timeval *timeout);
extern int redisContextUpdateCommandTimeout(redisContext *c, const struct timeval *timeout);

static redisContextFuncs redisContextDefaultFuncs = {
.free_privdata = NULL,
.async_read = redisAsyncRead,
Expand Down Expand Up @@ -723,7 +726,8 @@ void redisFree(redisContext *c) {
hi_free(c->tcp.host);
hi_free(c->tcp.source_addr);
hi_free(c->unix_sock.path);
hi_free(c->timeout);
hi_free(c->connect_timeout);
hi_free(c->command_timeout);
hi_free(c->saddr);
if (c->funcs->free_privdata) {
c->funcs->free_privdata(c->privdata);
Expand Down Expand Up @@ -761,18 +765,24 @@ int redisReconnect(redisContext *c) {
return REDIS_ERR;
}

int ret = REDIS_ERR;
if (c->connection_type == REDIS_CONN_TCP) {
return redisContextConnectBindTcp(c, c->tcp.host, c->tcp.port,
c->timeout, c->tcp.source_addr);
ret = redisContextConnectBindTcp(c, c->tcp.host, c->tcp.port,
c->connect_timeout, c->tcp.source_addr);
} else if (c->connection_type == REDIS_CONN_UNIX) {
return redisContextConnectUnix(c, c->unix_sock.path, c->timeout);
ret = redisContextConnectUnix(c, c->unix_sock.path, c->connect_timeout);
} else {
/* Something bad happened here and shouldn't have. There isn't
enough information in the context to reconnect. */
__redisSetError(c,REDIS_ERR_OTHER,"Not enough information to reconnect");
ret = REDIS_ERR;
}

return REDIS_ERR;
if (c->command_timeout != NULL && (c->flags & REDIS_BLOCK) && c->fd != REDIS_INVALID_FD) {
redisContextSetTimeout(c, *c->command_timeout);
}

return ret;
}

redisContext *redisConnectWithOptions(const redisOptions *options) {
Expand All @@ -790,13 +800,19 @@ redisContext *redisConnectWithOptions(const redisOptions *options) {
c->flags |= REDIS_NO_AUTO_FREE;
}

if (redisContextUpdateConnectTimeout(c, options->connect_timeout) != REDIS_OK ||
redisContextUpdateCommandTimeout(c, options->command_timeout) != REDIS_OK) {
__redisSetError(c, REDIS_ERR_OOM, "Out of memory");
return c;
}

if (options->type == REDIS_CONN_TCP) {
redisContextConnectBindTcp(c, options->endpoint.tcp.ip,
options->endpoint.tcp.port, options->timeout,
options->endpoint.tcp.port, options->connect_timeout,
options->endpoint.tcp.source_addr);
} else if (options->type == REDIS_CONN_UNIX) {
redisContextConnectUnix(c, options->endpoint.unix_socket,
options->timeout);
options->connect_timeout);
} else if (options->type == REDIS_CONN_USERFD) {
c->fd = options->endpoint.fd;
c->flags |= REDIS_CONNECTED;
Expand All @@ -805,6 +821,10 @@ redisContext *redisConnectWithOptions(const redisOptions *options) {
return NULL;
}

if (options->command_timeout != NULL && (c->flags & REDIS_BLOCK) && c->fd != REDIS_INVALID_FD) {
redisContextSetTimeout(c, *options->command_timeout);
}

return c;
}

Expand All @@ -820,7 +840,7 @@ redisContext *redisConnect(const char *ip, int port) {
redisContext *redisConnectWithTimeout(const char *ip, int port, const struct timeval tv) {
redisOptions options = {0};
REDIS_OPTIONS_SET_TCP(&options, ip, port);
options.timeout = &tv;
options.connect_timeout = &tv;
return redisConnectWithOptions(&options);
}

Expand Down Expand Up @@ -858,7 +878,7 @@ redisContext *redisConnectUnix(const char *path) {
redisContext *redisConnectUnixWithTimeout(const char *path, const struct timeval tv) {
redisOptions options = {0};
REDIS_OPTIONS_SET_UNIX(&options, path);
options.timeout = &tv;
options.connect_timeout = &tv;
return redisConnectWithOptions(&options);
}

Expand Down
9 changes: 6 additions & 3 deletions hiredis.h
View file
Expand Up @@ -176,8 +176,10 @@ typedef struct {
int type;
/* bit field of REDIS_OPT_xxx */
int options;
/* timeout value. if NULL, no timeout is used */
const struct timeval *timeout;
/* timeout value for connect operation. if NULL, no timeout is used */
const struct timeval *connect_timeout;
/* timeout value for commands. if NULL, no timeout is used. (can be set later on with redisSetTimeout/redisAsyncSetTimeout) */
const struct timeval *command_timeout;
union {
/** use this field for tcp/ip connections */
struct {
Expand Down Expand Up @@ -230,7 +232,8 @@ typedef struct redisContext {
redisReader *reader; /* Protocol reader */

enum redisConnectionType connection_type;
struct timeval *timeout;
struct timeval *connect_timeout;
struct timeval *command_timeout;

struct {
char *host;
Expand Down
42 changes: 29 additions & 13 deletions net.c
View file
Expand Up @@ -217,7 +217,7 @@ int redisSetTcpNoDelay(redisContext *c) {

static int redisContextTimeoutMsec(redisContext *c, long *result)
{
const struct timeval *timeout = c->timeout;
const struct timeval *timeout = c->connect_timeout;
long msec = -1;

/* Only use timeout when not NULL. */
Expand Down Expand Up @@ -328,19 +328,35 @@ int redisContextSetTimeout(redisContext *c, const struct timeval tv) {
return REDIS_OK;
}

static int _redisContextUpdateTimeout(redisContext *c, const struct timeval *timeout) {
int redisContextUpdateConnectTimeout(redisContext *c, const struct timeval *timeout) {
/* Same timeval struct, short circuit */
if (c->timeout == timeout)
if (c->connect_timeout == timeout)
return REDIS_OK;

/* Allocate context timeval if we need to */
if (c->timeout == NULL) {
c->timeout = hi_malloc(sizeof(*c->timeout));
if (c->timeout == NULL)
if (c->connect_timeout == NULL) {
c->connect_timeout = hi_malloc(sizeof(*c->connect_timeout));
if (c->connect_timeout == NULL)
return REDIS_ERR;
}

memcpy(c->timeout, timeout, sizeof(*c->timeout));
memcpy(c->connect_timeout, timeout, sizeof(*c->connect_timeout));
return REDIS_OK;
}

int redisContextUpdateCommandTimeout(redisContext *c, const struct timeval *timeout) {
/* Same timeval struct, short circuit */
if (c->command_timeout == timeout)
return REDIS_OK;

/* Allocate context timeval if we need to */
if (c->command_timeout == NULL) {
c->command_timeout = hi_malloc(sizeof(*c->command_timeout));
if (c->command_timeout == NULL)
return REDIS_ERR;
}

memcpy(c->command_timeout, timeout, sizeof(*c->command_timeout));
return REDIS_OK;
}

Expand Down Expand Up @@ -376,11 +392,11 @@ static int _redisContextConnectTcp(redisContext *c, const char *addr, int port,
}

if (timeout) {
if (_redisContextUpdateTimeout(c, timeout) == REDIS_ERR)
if (redisContextUpdateConnectTimeout(c, timeout) == REDIS_ERR)
goto oom;
} else {
hi_free(c->timeout);
c->timeout = NULL;
hi_free(c->connect_timeout);
c->connect_timeout = NULL;
}

if (redisContextTimeoutMsec(c, &timeout_msec) != REDIS_OK) {
Expand Down Expand Up @@ -549,11 +565,11 @@ int redisContextConnectUnix(redisContext *c, const char *path, const struct time
}

if (timeout) {
if (_redisContextUpdateTimeout(c, timeout) == REDIS_ERR)
if (redisContextUpdateConnectTimeout(c, timeout) == REDIS_ERR)
goto oom;
} else {
hi_free(c->timeout);
c->timeout = NULL;
hi_free(c->connect_timeout);
c->connect_timeout = NULL;
}

if (redisContextTimeoutMsec(c,&timeout_msec) != REDIS_OK)
Expand Down

0 comments on commit 38b5ae5

Please sign in to comment.