Commit 2fc0d875 authored by Pieter Noordhuis's avatar Pieter Noordhuis
Browse files

Use select(2) for enforce a timeout on blocking connect(2)

parent 663d6d12
...@@ -9,7 +9,8 @@ int main(void) { ...@@ -9,7 +9,8 @@ int main(void) {
redisContext *c; redisContext *c;
redisReply *reply; redisReply *reply;
c = redisConnect((char*)"127.0.0.1", 6379); struct timeval timeout = { 1, 500000 }; // 1.5 seconds
c = redisConnectWithTimeout((char*)"127.0.0.2", 6379, timeout);
if (c->err) { if (c->err) {
printf("Connection error: %s\n", c->errstr); printf("Connection error: %s\n", c->errstr);
exit(1); exit(1);
......
...@@ -821,28 +821,42 @@ void redisFree(redisContext *c) { ...@@ -821,28 +821,42 @@ void redisFree(redisContext *c) {
redisContext *redisConnect(const char *ip, int port) { redisContext *redisConnect(const char *ip, int port) {
redisContext *c = redisContextInit(); redisContext *c = redisContextInit();
c->flags |= REDIS_BLOCK; c->flags |= REDIS_BLOCK;
redisContextConnectTcp(c,ip,port); redisContextConnectTcp(c,ip,port,NULL);
return c;
}
redisContext *redisConnectWithTimeout(const char *ip, int port, struct timeval tv) {
redisContext *c = redisContextInit();
c->flags |= REDIS_BLOCK;
redisContextConnectTcp(c,ip,port,&tv);
return c; return c;
} }
redisContext *redisConnectNonBlock(const char *ip, int port) { redisContext *redisConnectNonBlock(const char *ip, int port) {
redisContext *c = redisContextInit(); redisContext *c = redisContextInit();
c->flags &= ~REDIS_BLOCK; c->flags &= ~REDIS_BLOCK;
redisContextConnectTcp(c,ip,port); redisContextConnectTcp(c,ip,port,NULL);
return c; return c;
} }
redisContext *redisConnectUnix(const char *path) { redisContext *redisConnectUnix(const char *path) {
redisContext *c = redisContextInit(); redisContext *c = redisContextInit();
c->flags |= REDIS_BLOCK; c->flags |= REDIS_BLOCK;
redisContextConnectUnix(c,path); redisContextConnectUnix(c,path,NULL);
return c;
}
redisContext *redisConnectUnixWithTimeout(const char *path, struct timeval tv) {
redisContext *c = redisContextInit();
c->flags |= REDIS_BLOCK;
redisContextConnectUnix(c,path,&tv);
return c; return c;
} }
redisContext *redisConnectUnixNonBlock(const char *path) { redisContext *redisConnectUnixNonBlock(const char *path) {
redisContext *c = redisContextInit(); redisContext *c = redisContextInit();
c->flags &= ~REDIS_BLOCK; c->flags &= ~REDIS_BLOCK;
redisContextConnectUnix(c,path); redisContextConnectUnix(c,path,NULL);
return c; return c;
} }
......
...@@ -144,8 +144,10 @@ int redisFormatCommand(char **target, const char *format, ...); ...@@ -144,8 +144,10 @@ int redisFormatCommand(char **target, const char *format, ...);
int redisFormatCommandArgv(char **target, int argc, const char **argv, const size_t *argvlen); int redisFormatCommandArgv(char **target, int argc, const char **argv, const size_t *argvlen);
redisContext *redisConnect(const char *ip, int port); redisContext *redisConnect(const char *ip, int port);
redisContext *redisConnectWithTimeout(const char *ip, int port, struct timeval tv);
redisContext *redisConnectNonBlock(const char *ip, int port); redisContext *redisConnectNonBlock(const char *ip, int port);
redisContext *redisConnectUnix(const char *path); redisContext *redisConnectUnix(const char *path);
redisContext *redisConnectUnixWithTimeout(const char *path, struct timeval tv);
redisContext *redisConnectUnixNonBlock(const char *path); redisContext *redisConnectUnixNonBlock(const char *path);
int redisSetTimeout(redisContext *c, struct timeval tv); int redisSetTimeout(redisContext *c, struct timeval tv);
int redisSetReplyObjectFunctions(redisContext *c, redisReplyObjectFunctions *fn); int redisSetReplyObjectFunctions(redisContext *c, redisReplyObjectFunctions *fn);
......
...@@ -33,6 +33,7 @@ ...@@ -33,6 +33,7 @@
#include "fmacros.h" #include "fmacros.h"
#include <sys/types.h> #include <sys/types.h>
#include <sys/socket.h> #include <sys/socket.h>
#include <sys/select.h>
#include <sys/un.h> #include <sys/un.h>
#include <netinet/in.h> #include <netinet/in.h>
#include <netinet/tcp.h> #include <netinet/tcp.h>
...@@ -67,7 +68,7 @@ static int redisCreateSocket(redisContext *c, int type) { ...@@ -67,7 +68,7 @@ static int redisCreateSocket(redisContext *c, int type) {
return s; return s;
} }
static int redisSetNonBlock(redisContext *c, int fd) { static int redisSetBlocking(redisContext *c, int fd, int blocking) {
int flags; int flags;
/* Set the socket nonblocking. /* Set the socket nonblocking.
...@@ -79,9 +80,15 @@ static int redisSetNonBlock(redisContext *c, int fd) { ...@@ -79,9 +80,15 @@ static int redisSetNonBlock(redisContext *c, int fd) {
close(fd); close(fd);
return REDIS_ERR; return REDIS_ERR;
} }
if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) == -1) {
if (blocking)
flags &= ~O_NONBLOCK;
else
flags |= O_NONBLOCK;
if (fcntl(fd, F_SETFL, flags) == -1) {
__redisSetError(c,REDIS_ERR_IO, __redisSetError(c,REDIS_ERR_IO,
sdscatprintf(sdsempty(), "fcntl(F_SETFL,O_NONBLOCK): %s", strerror(errno))); sdscatprintf(sdsempty(), "fcntl(F_SETFL): %s", strerror(errno)));
close(fd); close(fd);
return REDIS_ERR; return REDIS_ERR;
} }
...@@ -93,9 +100,65 @@ static int redisSetTcpNoDelay(redisContext *c, int fd) { ...@@ -93,9 +100,65 @@ static int redisSetTcpNoDelay(redisContext *c, int fd) {
if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &yes, sizeof(yes)) == -1) { if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &yes, sizeof(yes)) == -1) {
__redisSetError(c,REDIS_ERR_IO, __redisSetError(c,REDIS_ERR_IO,
sdscatprintf(sdsempty(), "setsockopt(TCP_NODELAY): %s", strerror(errno))); sdscatprintf(sdsempty(), "setsockopt(TCP_NODELAY): %s", strerror(errno)));
close(fd);
return REDIS_ERR;
}
return REDIS_OK;
}
static int redisContextWaitReady(redisContext *c, int fd, const struct timeval *timeout) {
struct timeval to;
struct timeval *toptr = NULL;
fd_set wfd;
int err;
socklen_t errlen;
/* Only use timeout when not NULL. */
if (timeout != NULL) {
memcpy(&to,timeout,sizeof(timeout));
toptr = &to;
}
if (errno == EINPROGRESS) {
FD_ZERO(&wfd);
FD_SET(fd, &wfd);
if (select(FD_SETSIZE, NULL, &wfd, NULL, toptr) == -1) {
__redisSetError(c,REDIS_ERR_IO,
sdscatprintf(sdsempty(), "select(2): %s", strerror(errno)));
close(fd);
return REDIS_ERR;
}
if (!FD_ISSET(fd, &wfd)) {
errno = ETIMEDOUT;
__redisSetError(c,REDIS_ERR_IO,NULL);
close(fd);
return REDIS_ERR;
}
err = 0;
errlen = sizeof(err);
if (getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &errlen) == -1) {
__redisSetError(c,REDIS_ERR_IO,
sdscatprintf(sdsempty(), "getsockopt(SO_ERROR): %s", strerror(errno)));
close(fd);
return REDIS_ERR;
}
if (err) {
errno = err;
__redisSetError(c,REDIS_ERR_IO,NULL);
close(fd);
return REDIS_ERR; return REDIS_ERR;
} }
return REDIS_OK; return REDIS_OK;
}
__redisSetError(c,REDIS_ERR_IO,NULL);
close(fd);
return REDIS_ERR;
} }
int redisContextSetTimeout(redisContext *c, struct timeval tv) { int redisContextSetTimeout(redisContext *c, struct timeval tv) {
...@@ -112,14 +175,14 @@ int redisContextSetTimeout(redisContext *c, struct timeval tv) { ...@@ -112,14 +175,14 @@ int redisContextSetTimeout(redisContext *c, struct timeval tv) {
return REDIS_OK; return REDIS_OK;
} }
int redisContextConnectTcp(redisContext *c, const char *addr, int port) { int redisContextConnectTcp(redisContext *c, const char *addr, int port, struct timeval *timeout) {
int s; int s;
int blocking = (c->flags & REDIS_BLOCK); int blocking = (c->flags & REDIS_BLOCK);
struct sockaddr_in sa; struct sockaddr_in sa;
if ((s = redisCreateSocket(c,AF_INET)) == REDIS_ERR) if ((s = redisCreateSocket(c,AF_INET)) < 0)
return REDIS_ERR; return REDIS_ERR;
if (!blocking && redisSetNonBlock(c,s) == REDIS_ERR) if (redisSetBlocking(c,s,0) != REDIS_OK)
return REDIS_ERR; return REDIS_ERR;
sa.sin_family = AF_INET; sa.sin_family = AF_INET;
...@@ -141,30 +204,31 @@ int redisContextConnectTcp(redisContext *c, const char *addr, int port) { ...@@ -141,30 +204,31 @@ int redisContextConnectTcp(redisContext *c, const char *addr, int port) {
if (errno == EINPROGRESS && !blocking) { if (errno == EINPROGRESS && !blocking) {
/* This is ok. */ /* This is ok. */
} else { } else {
__redisSetError(c,REDIS_ERR_IO,NULL); if (redisContextWaitReady(c,s,timeout) != REDIS_OK)
close(s);
return REDIS_ERR; return REDIS_ERR;
} }
} }
if (redisSetTcpNoDelay(c,s) != REDIS_OK) { /* Reset socket to be blocking after connect(2). */
close(s); if (blocking && redisSetBlocking(c,s,1) != REDIS_OK)
return REDIS_ERR;
if (redisSetTcpNoDelay(c,s) != REDIS_OK)
return REDIS_ERR; return REDIS_ERR;
}
c->fd = s; c->fd = s;
c->flags |= REDIS_CONNECTED; c->flags |= REDIS_CONNECTED;
return REDIS_OK; return REDIS_OK;
} }
int redisContextConnectUnix(redisContext *c, const char *path) { int redisContextConnectUnix(redisContext *c, const char *path, struct timeval *timeout) {
int s; int s;
int blocking = (c->flags & REDIS_BLOCK); int blocking = (c->flags & REDIS_BLOCK);
struct sockaddr_un sa; struct sockaddr_un sa;
if ((s = redisCreateSocket(c,AF_LOCAL)) == REDIS_ERR) if ((s = redisCreateSocket(c,AF_LOCAL)) < 0)
return REDIS_ERR; return REDIS_ERR;
if (!blocking && redisSetNonBlock(c,s) != REDIS_OK) if (redisSetBlocking(c,s,0) != REDIS_OK)
return REDIS_ERR; return REDIS_ERR;
sa.sun_family = AF_LOCAL; sa.sun_family = AF_LOCAL;
...@@ -173,12 +237,15 @@ int redisContextConnectUnix(redisContext *c, const char *path) { ...@@ -173,12 +237,15 @@ int redisContextConnectUnix(redisContext *c, const char *path) {
if (errno == EINPROGRESS && !blocking) { if (errno == EINPROGRESS && !blocking) {
/* This is ok. */ /* This is ok. */
} else { } else {
__redisSetError(c,REDIS_ERR_IO,NULL); if (redisContextWaitReady(c,s,timeout) != REDIS_OK)
close(s);
return REDIS_ERR; return REDIS_ERR;
} }
} }
/* Reset socket to be blocking after connect(2). */
if (blocking && redisSetBlocking(c,s,1) != REDIS_OK)
return REDIS_ERR;
c->fd = s; c->fd = s;
c->flags |= REDIS_CONNECTED; c->flags |= REDIS_CONNECTED;
return REDIS_OK; return REDIS_OK;
......
...@@ -40,7 +40,7 @@ ...@@ -40,7 +40,7 @@
#endif #endif
int redisContextSetTimeout(redisContext *c, struct timeval tv); int redisContextSetTimeout(redisContext *c, struct timeval tv);
int redisContextConnectTcp(redisContext *c, const char *addr, int port); int redisContextConnectTcp(redisContext *c, const char *addr, int port, struct timeval *timeout);
int redisContextConnectUnix(redisContext *c, const char *path); int redisContextConnectUnix(redisContext *c, const char *path, struct timeval *timeout);
#endif #endif
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