X-Git-Url: https://git.saurik.com/redis.git/blobdiff_plain/24f753a8b996a364273019ea791fe89f87cff678..2ebd2720b37dcca3b6e0c18377bd69e9eaf541fc:/deps/hiredis/hiredis.c diff --git a/deps/hiredis/hiredis.c b/deps/hiredis/hiredis.c index 3c8736dc..976e94f9 100644 --- a/deps/hiredis/hiredis.c +++ b/deps/hiredis/hiredis.c @@ -1,5 +1,7 @@ /* * Copyright (c) 2009-2010, Salvatore Sanfilippo + * Copyright (c) 2010, Pieter Noordhuis + * * All rights reserved. * * Redistribution and use in source and binary forms, with or without @@ -32,6 +34,7 @@ #include #include #include +#include #include "hiredis.h" #include "net.h" @@ -44,10 +47,12 @@ typedef struct redisReader { void *reply; /* holds temporary reply */ sds buf; /* read buffer */ - unsigned int pos; /* buffer cursor */ + size_t pos; /* buffer cursor */ + size_t len; /* buffer length */ - redisReadTask rstack[3]; /* stack of read tasks */ + redisReadTask rstack[9]; /* stack of read tasks */ int ridx; /* index of stack */ + void *privdata; /* user-settable arbitrary field */ } redisReader; static redisReply *createReplyObject(int type); @@ -68,7 +73,7 @@ static redisReplyObjectFunctions defaultFunctions = { /* Create a reply object */ static redisReply *createReplyObject(int type) { - redisReply *r = calloc(sizeof(*r),1); + redisReply *r = malloc(sizeof(*r)); if (!r) redisOOM(); r->type = type; @@ -88,9 +93,10 @@ void freeReplyObject(void *reply) { if (r->element[j]) freeReplyObject(r->element[j]); free(r->element); break; - default: - if (r->str != NULL) - free(r->str); + case REDIS_REPLY_ERROR: + case REDIS_REPLY_STATUS: + case REDIS_REPLY_STRING: + free(r->str); break; } free(r); @@ -111,7 +117,7 @@ static void *createStringObject(const redisReadTask *task, char *str, size_t len r->len = len; if (task->parent) { - redisReply *parent = task->parent; + redisReply *parent = task->parent->obj; assert(parent->type == REDIS_REPLY_ARRAY); parent->element[task->idx] = r; } @@ -124,7 +130,7 @@ static void *createArrayObject(const redisReadTask *task, int elements) { if ((r->element = calloc(sizeof(redisReply*),elements)) == NULL) redisOOM(); if (task->parent) { - redisReply *parent = task->parent; + redisReply *parent = task->parent->obj; assert(parent->type == REDIS_REPLY_ARRAY); parent->element[task->idx] = r; } @@ -135,7 +141,7 @@ static void *createIntegerObject(const redisReadTask *task, long long value) { redisReply *r = createReplyObject(REDIS_REPLY_INTEGER); r->integer = value; if (task->parent) { - redisReply *parent = task->parent; + redisReply *parent = task->parent->obj; assert(parent->type == REDIS_REPLY_ARRAY); parent->element[task->idx] = r; } @@ -145,7 +151,7 @@ static void *createIntegerObject(const redisReadTask *task, long long value) { static void *createNilObject(const redisReadTask *task) { redisReply *r = createReplyObject(REDIS_REPLY_NIL); if (task->parent) { - redisReply *parent = task->parent; + redisReply *parent = task->parent->obj; assert(parent->type == REDIS_REPLY_ARRAY); parent->element[task->idx] = r; } @@ -154,7 +160,7 @@ static void *createNilObject(const redisReadTask *task) { static char *readBytes(redisReader *r, unsigned int bytes) { char *p; - if (sdslen(r->buf)-r->pos >= bytes) { + if (r->len-r->pos >= bytes) { p = r->buf+r->pos; r->pos += bytes; return p; @@ -162,11 +168,69 @@ static char *readBytes(redisReader *r, unsigned int bytes) { return NULL; } +/* Find pointer to \r\n. */ +static char *seekNewline(char *s, size_t len) { + int pos = 0; + int _len = len-1; + + /* Position should be < len-1 because the character at "pos" should be + * followed by a \n. Note that strchr cannot be used because it doesn't + * allow to search a limited length and the buffer that is being searched + * might not have a trailing NULL character. */ + while (pos < _len) { + while(pos < _len && s[pos] != '\r') pos++; + if (s[pos] != '\r') { + /* Not found. */ + return NULL; + } else { + if (s[pos+1] == '\n') { + /* Found. */ + return s+pos; + } else { + /* Continue searching. */ + pos++; + } + } + } + return NULL; +} + +/* Read a long long value starting at *s, under the assumption that it will be + * terminated by \r\n. Ambiguously returns -1 for unexpected input. */ +static long long readLongLong(char *s) { + long long v = 0; + int dec, mult = 1; + char c; + + if (*s == '-') { + mult = -1; + s++; + } else if (*s == '+') { + mult = 1; + s++; + } + + while ((c = *(s++)) != '\r') { + dec = c - '0'; + if (dec >= 0 && dec < 10) { + v *= 10; + v += dec; + } else { + /* Should not happen... */ + return -1; + } + } + + return mult*v; +} + static char *readLine(redisReader *r, int *_len) { - char *p, *s = strstr(r->buf+r->pos,"\r\n"); + char *p, *s; int len; + + p = r->buf+r->pos; + s = seekNewline(p,(r->len-r->pos)); if (s != NULL) { - p = r->buf+r->pos; len = s-(r->buf+r->pos); r->pos += len+2; /* skip \r\n */ if (_len) *_len = len; @@ -177,26 +241,26 @@ static char *readLine(redisReader *r, int *_len) { static void moveToNextTask(redisReader *r) { redisReadTask *cur, *prv; - assert(r->ridx >= 0); - - /* Return a.s.a.p. when the stack is now empty. */ - if (r->ridx == 0) { - r->ridx--; - return; - } + while (r->ridx >= 0) { + /* Return a.s.a.p. when the stack is now empty. */ + if (r->ridx == 0) { + r->ridx--; + return; + } - cur = &(r->rstack[r->ridx]); - prv = &(r->rstack[r->ridx-1]); - assert(prv->type == REDIS_REPLY_ARRAY); - if (cur->idx == prv->elements-1) { - r->ridx--; - moveToNextTask(r); - } else { - /* Reset the type because the next item can be anything */ - assert(cur->idx < prv->elements); - cur->type = -1; - cur->elements = -1; - cur->idx++; + cur = &(r->rstack[r->ridx]); + prv = &(r->rstack[r->ridx-1]); + assert(prv->type == REDIS_REPLY_ARRAY); + if (cur->idx == prv->elements-1) { + r->ridx--; + } else { + /* Reset the type because the next item can be anything */ + assert(cur->idx < prv->elements); + cur->type = -1; + cur->elements = -1; + cur->idx++; + return; + } } } @@ -208,14 +272,20 @@ static int processLineItem(redisReader *r) { if ((p = readLine(r,&len)) != NULL) { if (cur->type == REDIS_REPLY_INTEGER) { - obj = r->fn->createInteger(cur,strtoll(p,NULL,10)); + if (r->fn && r->fn->createInteger) + obj = r->fn->createInteger(cur,readLongLong(p)); + else + obj = (void*)REDIS_REPLY_INTEGER; } else { - obj = r->fn->createString(cur,p,len); + /* Type will be error or status. */ + if (r->fn && r->fn->createString) + obj = r->fn->createString(cur,p,len); + else + obj = (void*)(size_t)(cur->type); } - /* If there is no root yet, register this object as root. */ - if (r->reply == NULL) - r->reply = obj; + /* Set reply if this is the root object. */ + if (r->ridx == 0) r->reply = obj; moveToNextTask(r); return 0; } @@ -228,30 +298,40 @@ static int processBulkItem(redisReader *r) { char *p, *s; long len; unsigned long bytelen; + int success = 0; p = r->buf+r->pos; - s = strstr(p,"\r\n"); + s = seekNewline(p,r->len-r->pos); if (s != NULL) { p = r->buf+r->pos; bytelen = s-(r->buf+r->pos)+2; /* include \r\n */ - len = strtol(p,NULL,10); + len = readLongLong(p); if (len < 0) { /* The nil object can always be created. */ - obj = r->fn->createNil(cur); + if (r->fn && r->fn->createNil) + obj = r->fn->createNil(cur); + else + obj = (void*)REDIS_REPLY_NIL; + success = 1; } else { /* Only continue when the buffer contains the entire bulk item. */ bytelen += len+2; /* include \r\n */ - if (r->pos+bytelen <= sdslen(r->buf)) { - obj = r->fn->createString(cur,s+2,len); + if (r->pos+bytelen <= r->len) { + if (r->fn && r->fn->createString) + obj = r->fn->createString(cur,s+2,len); + else + obj = (void*)REDIS_REPLY_STRING; + success = 1; } } /* Proceed when obj was created. */ - if (obj != NULL) { + if (success) { r->pos += bytelen; - if (r->reply == NULL) - r->reply = obj; + + /* Set reply if this is the root object. */ + if (r->ridx == 0) r->reply = obj; moveToNextTask(r); return 0; } @@ -264,31 +344,49 @@ static int processMultiBulkItem(redisReader *r) { void *obj; char *p; long elements; + int root = 0; + + /* Set error for nested multi bulks with depth > 1 */ + if (r->ridx == 8) { + redisSetReplyReaderError(r,sdscatprintf(sdsempty(), + "No support for nested multi bulk replies with depth > 7")); + return -1; + } if ((p = readLine(r,NULL)) != NULL) { - elements = strtol(p,NULL,10); + elements = readLongLong(p); + root = (r->ridx == 0); + if (elements == -1) { - obj = r->fn->createNil(cur); + if (r->fn && r->fn->createNil) + obj = r->fn->createNil(cur); + else + obj = (void*)REDIS_REPLY_NIL; moveToNextTask(r); } else { - obj = r->fn->createArray(cur,elements); + if (r->fn && r->fn->createArray) + obj = r->fn->createArray(cur,elements); + else + obj = (void*)REDIS_REPLY_ARRAY; /* Modify task stack when there are more than 0 elements. */ if (elements > 0) { cur->elements = elements; + cur->obj = obj; r->ridx++; r->rstack[r->ridx].type = -1; r->rstack[r->ridx].elements = -1; - r->rstack[r->ridx].parent = obj; r->rstack[r->ridx].idx = 0; + r->rstack[r->ridx].obj = NULL; + r->rstack[r->ridx].parent = cur; + r->rstack[r->ridx].privdata = r->privdata; } else { moveToNextTask(r); } } - /* Object was created, so we can always continue. */ - if (r->reply == NULL) - r->reply = obj; + /* Set reply if this is the root object. */ + if (root) r->reply = obj; return 0; } return -1; @@ -321,7 +419,7 @@ static int processItem(redisReader *r) { default: byte = sdscatrepr(sdsempty(),p,1); redisSetReplyReaderError(r,sdscatprintf(sdsempty(), - "protocol error, got %s as reply type byte", byte)); + "Protocol error, got %s as reply type byte", byte)); sdsfree(byte); return -1; } @@ -342,21 +440,42 @@ static int processItem(redisReader *r) { case REDIS_REPLY_ARRAY: return processMultiBulkItem(r); default: - redisSetReplyReaderError(r,sdscatprintf(sdsempty(), - "unknown item type '%d'", cur->type)); + assert(NULL); return -1; } } -void *redisReplyReaderCreate(redisReplyObjectFunctions *fn) { +void *redisReplyReaderCreate(void) { redisReader *r = calloc(sizeof(redisReader),1); r->error = NULL; - r->fn = fn == NULL ? &defaultFunctions : fn; + r->fn = &defaultFunctions; r->buf = sdsempty(); r->ridx = -1; return r; } +/* Set the function set to build the reply. Returns REDIS_OK when there + * is no temporary object and it can be set, REDIS_ERR otherwise. */ +int redisReplyReaderSetReplyObjectFunctions(void *reader, redisReplyObjectFunctions *fn) { + redisReader *r = reader; + if (r->reply == NULL) { + r->fn = fn; + return REDIS_OK; + } + return REDIS_ERR; +} + +/* Set the private data field that is used in the read tasks. This argument can + * be used to curry arbitrary data to the custom reply object functions. */ +int redisReplyReaderSetPrivdata(void *reader, void *privdata) { + redisReader *r = reader; + if (r->reply == NULL) { + r->privdata = privdata; + return REDIS_OK; + } + return REDIS_ERR; +} + /* External libraries wrapping hiredis might need access to the temporary * variable while the reply is built up. When the reader contains an * object in between receiving some bytes to parse, this object might @@ -370,7 +489,7 @@ void redisReplyReaderFree(void *reader) { redisReader *r = reader; if (r->error != NULL) sdsfree(r->error); - if (r->reply != NULL) + if (r->reply != NULL && r->fn) r->fn->freeObject(r->reply); if (r->buf != NULL) sdsfree(r->buf); @@ -385,7 +504,7 @@ static void redisSetReplyReaderError(redisReader *r, sds err) { if (r->buf != NULL) { sdsfree(r->buf); r->buf = sdsempty(); - r->pos = 0; + r->pos = r->len = 0; } r->ridx = -1; r->error = err; @@ -396,12 +515,22 @@ char *redisReplyReaderGetError(void *reader) { return r->error; } -void redisReplyReaderFeed(void *reader, char *buf, int len) { +void redisReplyReaderFeed(void *reader, const char *buf, size_t len) { redisReader *r = reader; /* Copy the provided buffer. */ - if (buf != NULL && len >= 1) + if (buf != NULL && len >= 1) { +#if 0 + /* Destroy internal buffer when it is empty and is quite large. */ + if (r->len == 0 && sdsavail(r->buf) > 16*1024) { + sdsfree(r->buf); + r->buf = sdsempty(); + r->pos = 0; + } +#endif r->buf = sdscatlen(r->buf,buf,len); + r->len = sdslen(r->buf); + } } int redisReplyReaderGetReply(void *reader, void **reply) { @@ -409,15 +538,17 @@ int redisReplyReaderGetReply(void *reader, void **reply) { if (reply != NULL) *reply = NULL; /* When the buffer is empty, there will never be a reply. */ - if (sdslen(r->buf) == 0) + if (r->len == 0) return REDIS_OK; /* Set first item to process when the stack is empty. */ if (r->ridx == -1) { r->rstack[0].type = -1; r->rstack[0].elements = -1; - r->rstack[0].parent = NULL; r->rstack[0].idx = -1; + r->rstack[0].obj = NULL; + r->rstack[0].parent = NULL; + r->rstack[0].privdata = r->privdata; r->ridx = 0; } @@ -426,16 +557,12 @@ int redisReplyReaderGetReply(void *reader, void **reply) { if (processItem(r) < 0) break; - /* Discard the consumed part of the buffer. */ - if (r->pos > 0) { - if (r->pos == sdslen(r->buf)) { - /* sdsrange has a quirck on this edge case. */ - sdsfree(r->buf); - r->buf = sdsempty(); - } else { - r->buf = sdsrange(r->buf,r->pos,sdslen(r->buf)); - } + /* Discard part of the buffer when we've consumed at least 1k, to avoid + * doing unnecessary calls to memmove() in sds.c. */ + if (r->pos >= 1024) { + r->buf = sdsrange(r->buf,r->pos,-1); r->pos = 0; + r->len = sdslen(r->buf); } /* Emit a reply when there is one. */ @@ -443,13 +570,6 @@ int redisReplyReaderGetReply(void *reader, void **reply) { void *aux = r->reply; r->reply = NULL; - /* Destroy the buffer when it is empty and is quite large. */ - if (sdslen(r->buf) == 0 && sdsavail(r->buf) > 16*1024) { - sdsfree(r->buf); - r->buf = sdsempty(); - r->pos = 0; - } - /* Check if there actually *is* a reply. */ if (r->error != NULL) { return REDIS_ERR; @@ -488,6 +608,7 @@ int redisvFormatCommand(char **target, const char *format, va_list ap) { char *cmd = NULL; /* final command */ int pos; /* position in final command */ sds current; /* current argument */ + int touched = 0; /* was the current argument touched? */ char **argv = NULL; int argc = 0, j; int totlen = 0; @@ -501,35 +622,93 @@ int redisvFormatCommand(char **target, const char *format, va_list ap) { while(*c != '\0') { if (*c != '%' || c[1] == '\0') { if (*c == ' ') { - if (sdslen(current) != 0) { + if (touched) { addArgument(current, &argv, &argc, &totlen); current = sdsempty(); + touched = 0; } } else { current = sdscatlen(current,c,1); + touched = 1; } } else { switch(c[1]) { case 's': arg = va_arg(ap,char*); - current = sdscat(current,arg); + size = strlen(arg); + if (size > 0) + current = sdscatlen(current,arg,size); break; case 'b': arg = va_arg(ap,char*); size = va_arg(ap,size_t); - current = sdscatlen(current,arg,size); + if (size > 0) + current = sdscatlen(current,arg,size); break; case '%': - cmd = sdscat(cmd,"%"); + current = sdscat(current,"%"); break; + default: + /* Try to detect printf format */ + { + char _format[16]; + const char *_p = c+1; + size_t _l = 0; + va_list _cpy; + + /* Flags */ + if (*_p != '\0' && *_p == '#') _p++; + if (*_p != '\0' && *_p == '0') _p++; + if (*_p != '\0' && *_p == '-') _p++; + if (*_p != '\0' && *_p == ' ') _p++; + if (*_p != '\0' && *_p == '+') _p++; + + /* Field width */ + while (*_p != '\0' && isdigit(*_p)) _p++; + + /* Precision */ + if (*_p == '.') { + _p++; + while (*_p != '\0' && isdigit(*_p)) _p++; + } + + /* Modifiers */ + if (*_p != '\0') { + if (*_p == 'h' || *_p == 'l') { + /* Allow a single repetition for these modifiers */ + if (_p[0] == _p[1]) _p++; + _p++; + } + } + + /* Conversion specifier */ + if (*_p != '\0' && strchr("diouxXeEfFgGaA",*_p) != NULL) { + _l = (_p+1)-c; + if (_l < sizeof(_format)-2) { + memcpy(_format,c,_l); + _format[_l] = '\0'; + va_copy(_cpy,ap); + current = sdscatvprintf(current,_format,_cpy); + va_end(_cpy); + + /* Update current position (note: outer blocks + * increment c twice so compensate here) */ + c = _p-1; + } + } + + /* Consume and discard vararg */ + va_arg(ap,void); + } } + touched = 1; c++; } c++; } /* Add the last argument if needed */ - if (sdslen(current) != 0) { + if (touched) { addArgument(current, &argv, &argc, &totlen); } else { sdsfree(current); @@ -625,7 +804,7 @@ void __redisSetError(redisContext *c, int type, const sds errstr) { } } -static redisContext *redisContextInit() { +static redisContext *redisContextInit(void) { redisContext *c = calloc(sizeof(redisContext),1); c->err = 0; c->errstr = NULL; @@ -636,8 +815,7 @@ static redisContext *redisContextInit() { } void redisFree(redisContext *c) { - /* Disconnect before free'ing if not yet disconnected. */ - if (c->flags & REDIS_CONNECTED) + if (c->fd > 0) close(c->fd); if (c->errstr != NULL) sdsfree(c->errstr); @@ -654,35 +832,52 @@ void redisFree(redisContext *c) { redisContext *redisConnect(const char *ip, int port) { redisContext *c = redisContextInit(); c->flags |= REDIS_BLOCK; - c->flags |= REDIS_CONNECTED; - 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; } redisContext *redisConnectNonBlock(const char *ip, int port) { redisContext *c = redisContextInit(); c->flags &= ~REDIS_BLOCK; - c->flags |= REDIS_CONNECTED; - redisContextConnectTcp(c,ip,port); + redisContextConnectTcp(c,ip,port,NULL); return c; } redisContext *redisConnectUnix(const char *path) { redisContext *c = redisContextInit(); c->flags |= REDIS_BLOCK; - c->flags |= REDIS_CONNECTED; - 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; } redisContext *redisConnectUnixNonBlock(const char *path) { redisContext *c = redisContextInit(); c->flags &= ~REDIS_BLOCK; - c->flags |= REDIS_CONNECTED; - redisContextConnectUnix(c,path); + redisContextConnectUnix(c,path,NULL); return c; } +/* Set read/write timeout on a blocking socket. */ +int redisSetTimeout(redisContext *c, struct timeval tv) { + if (c->flags & REDIS_BLOCK) + return redisContextSetTimeout(c,tv); + return REDIS_ERR; +} + /* Set the replyObjectFunctions to use. Returns REDIS_ERR when the reader * was already initialized and the function set could not be re-set. * Return REDIS_OK when they could be set. */ @@ -695,8 +890,10 @@ int redisSetReplyObjectFunctions(redisContext *c, redisReplyObjectFunctions *fn) /* Helper function to lazily create a reply reader. */ static void __redisCreateReplyReader(redisContext *c) { - if (c->reader == NULL) - c->reader = redisReplyReaderCreate(c->fn); + if (c->reader == NULL) { + c->reader = redisReplyReaderCreate(); + assert(redisReplyReaderSetReplyObjectFunctions(c->reader,c->fn) == REDIS_OK); + } } /* Use this function to handle a read event on the descriptor. It will try @@ -705,10 +902,10 @@ static void __redisCreateReplyReader(redisContext *c) { * After this function is called, you may use redisContextReadReply to * see if there is a reply available. */ int redisBufferRead(redisContext *c) { - char buf[2048]; + char buf[1024*16]; int nread = read(c->fd,buf,sizeof(buf)); if (nread == -1) { - if (errno == EAGAIN) { + if (errno == EAGAIN && !(c->flags & REDIS_BLOCK)) { /* Try again later */ } else { __redisSetError(c,REDIS_ERR_IO,NULL); @@ -739,7 +936,7 @@ int redisBufferWrite(redisContext *c, int *done) { if (sdslen(c->obuf) > 0) { nwritten = write(c->fd,c->obuf,sdslen(c->obuf)); if (nwritten == -1) { - if (errno == EAGAIN) { + if (errno == EAGAIN && !(c->flags & REDIS_BLOCK)) { /* Try again later */ } else { __redisSetError(c,REDIS_ERR_IO,NULL);