Commit e944ea36 authored by Pieter Noordhuis's avatar Pieter Noordhuis
Browse files

Add reply type for protocol errors, in order to never exit()

parent aec1fbd2
...@@ -48,6 +48,8 @@ typedef struct redisReader { ...@@ -48,6 +48,8 @@ typedef struct redisReader {
static redisReply *redisReadReply(int fd); static redisReply *redisReadReply(int fd);
static redisReply *createReplyObject(int type, sds reply); static redisReply *createReplyObject(int type, sds reply);
static redisReply *createErrorObject(const char *fmt, ...);
static void redisSetReplyReaderError(redisReader *r, redisReply *error);
/* We simply abort on out of memory */ /* We simply abort on out of memory */
static void redisOOM(void) { static void redisOOM(void) {
...@@ -64,7 +66,7 @@ redisReply *redisConnect(int *fd, const char *ip, int port) { ...@@ -64,7 +66,7 @@ redisReply *redisConnect(int *fd, const char *ip, int port) {
*fd = anetTcpConnect(err,ip,port); *fd = anetTcpConnect(err,ip,port);
if (*fd == ANET_ERR) if (*fd == ANET_ERR)
return createReplyObject(REDIS_REPLY_ERROR,sdsnew(err)); return createErrorObject(err);
anetTcpNoDelay(NULL,*fd); anetTcpNoDelay(NULL,*fd);
return NULL; return NULL;
} }
...@@ -99,8 +101,19 @@ void freeReplyObject(redisReply *r) { ...@@ -99,8 +101,19 @@ void freeReplyObject(redisReply *r) {
free(r); free(r);
} }
static redisReply *createErrorObject(const char *fmt, ...) {
va_list ap;
sds err;
redisReply *r;
va_start(ap,fmt);
err = sdscatvprintf(sdsempty(),fmt,ap);
va_end(ap);
r = createReplyObject(REDIS_PROTOCOL_ERROR,err);
return r;
}
static redisReply *redisIOError(void) { static redisReply *redisIOError(void) {
return createReplyObject(REDIS_REPLY_ERROR,sdsnew("I/O error")); return createErrorObject("I/O error");
} }
static char *readBytes(redisReader *r, unsigned int bytes) { static char *readBytes(redisReader *r, unsigned int bytes) {
...@@ -224,6 +237,7 @@ static int processMultiBulkItem(redisReader *r) { ...@@ -224,6 +237,7 @@ static int processMultiBulkItem(redisReader *r) {
static int processItem(redisReader *r) { static int processItem(redisReader *r) {
redisReply *cur = r->rlist[r->rpos]; redisReply *cur = r->rlist[r->rpos];
char *p; char *p;
sds byte;
/* check if we need to read type */ /* check if we need to read type */
if (cur->type < 0) { if (cur->type < 0) {
...@@ -245,8 +259,11 @@ static int processItem(redisReader *r) { ...@@ -245,8 +259,11 @@ static int processItem(redisReader *r) {
cur->type = REDIS_REPLY_ARRAY; cur->type = REDIS_REPLY_ARRAY;
break; break;
default: default:
printf("protocol error, got '%c' as reply type byte\n", p[0]); byte = sdscatrepr(sdsempty(),p,1);
exit(1); redisSetReplyReaderError(r,createErrorObject(
"protocol error, got %s as reply type byte", byte));
sdsfree(byte);
return -1;
} }
} else { } else {
/* could not consume 1 byte */ /* could not consume 1 byte */
...@@ -265,8 +282,9 @@ static int processItem(redisReader *r) { ...@@ -265,8 +282,9 @@ static int processItem(redisReader *r) {
case REDIS_REPLY_ARRAY: case REDIS_REPLY_ARRAY:
return processMultiBulkItem(r); return processMultiBulkItem(r);
default: default:
printf("unknown item type: %d\n", cur->type); redisSetReplyReaderError(r,createErrorObject(
exit(1); "unknown item type '%d'", cur->type));
return -1;
} }
} }
...@@ -308,6 +326,20 @@ void redisFreeReplyReader(void *reader) { ...@@ -308,6 +326,20 @@ void redisFreeReplyReader(void *reader) {
free(r); free(r);
} }
static void redisSetReplyReaderError(redisReader *r, redisReply *error) {
/* Clear remaining buffer when we see a protocol error. */
if (r->buf != NULL) {
sdsfree(r->buf);
r->buf = sdsempty();
r->pos = 0;
}
/* Clear currently allocated objects. */
if (r->rlist[0] != NULL)
freeReplyObject(r->rlist[0]);
r->rlen = r->rpos = 1;
r->rlist[0] = error;
}
void *redisFeedReplyReader(void *reader, char *buf, int len) { void *redisFeedReplyReader(void *reader, char *buf, int len) {
redisReader *r = reader; redisReader *r = reader;
...@@ -357,6 +389,7 @@ void *redisFeedReplyReader(void *reader, char *buf, int len) { ...@@ -357,6 +389,7 @@ void *redisFeedReplyReader(void *reader, char *buf, int len) {
/* Free list of items to process. */ /* Free list of items to process. */
free(r->rlist); free(r->rlist);
r->rlist = NULL;
r->rlen = r->rpos = 0; r->rlen = r->rpos = 0;
return reply; return reply;
} else { } else {
......
...@@ -36,6 +36,7 @@ ...@@ -36,6 +36,7 @@
#define REDIS_REPLY_INTEGER 3 #define REDIS_REPLY_INTEGER 3
#define REDIS_REPLY_NIL 4 #define REDIS_REPLY_NIL 4
#define REDIS_REPLY_STATUS 5 #define REDIS_REPLY_STATUS 5
#define REDIS_PROTOCOL_ERROR 6
#include "sds.h" #include "sds.h"
......
...@@ -29,11 +29,12 @@ int main(void) { ...@@ -29,11 +29,12 @@ int main(void) {
int i, tests = 0, fails = 0; int i, tests = 0, fails = 0;
long long t1, t2; long long t1, t2;
redisReply *reply; redisReply *reply;
void *reader;
__connect(&fd); __connect(&fd);
test("Returns I/O error when the connection is lost: "); test("Returns I/O error when the connection is lost: ");
reply = redisCommand(fd,"QUIT"); reply = redisCommand(fd,"QUIT");
test_cond(reply->type == REDIS_REPLY_ERROR && test_cond(reply->type == REDIS_PROTOCOL_ERROR &&
strcasecmp(reply->reply,"i/o error") == 0); strcasecmp(reply->reply,"i/o error") == 0);
freeReplyObject(reply); freeReplyObject(reply);
__connect(&fd); /* reconnect */ __connect(&fd); /* reconnect */
...@@ -122,6 +123,26 @@ int main(void) { ...@@ -122,6 +123,26 @@ int main(void) {
strcasecmp(reply->element[1]->reply,"pong") == 0); strcasecmp(reply->element[1]->reply,"pong") == 0);
freeReplyObject(reply); freeReplyObject(reply);
test("Error handling in reply parser: ");
reader = redisCreateReplyReader();
reply = redisFeedReplyReader(reader,(char*)"@foo\r\n",6);
test_cond(reply->type == REDIS_PROTOCOL_ERROR &&
strcasecmp(reply->reply,"protocol error, got \"@\" as reply type byte") == 0);
freeReplyObject(reply);
redisFreeReplyReader(reader);
/* when the reply already contains multiple items, they must be free'd
* on an error. valgrind will bark when this doesn't happen. */
test("Memory cleanup in reply parser: ");
reader = redisCreateReplyReader();
redisFeedReplyReader(reader,(char*)"*2\r\n",4);
redisFeedReplyReader(reader,(char*)"$5\r\nhello\r\n",11);
reply = redisFeedReplyReader(reader,(char*)"@foo\r\n",6);
test_cond(reply->type == REDIS_PROTOCOL_ERROR &&
strcasecmp(reply->reply,"protocol error, got \"@\" as reply type byte") == 0);
freeReplyObject(reply);
redisFreeReplyReader(reader);
test("Throughput:\n"); test("Throughput:\n");
for (i = 0; i < 500; i++) for (i = 0; i < 500; i++)
freeReplyObject(redisCommand(fd,"LPUSH mylist foo")); freeReplyObject(redisCommand(fd,"LPUSH mylist foo"));
......
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