Skip to content

Commit

Permalink
Add function to retrieve formatted reply
Browse files Browse the repository at this point in the history 
This is done by only truncating the read buffer once a full reply has
been read. The buffer is no longer truncated halfway through reading a
reply. In addition: pass offset/length of protocol and content via the
read tasks.
  • Loading branch information
@pietern
pietern committed May 5, 2011
1 parent df203bc commit 77540aa
Show file tree
Hide file tree
Showing 3 changed files with 140 additions and 12 deletions.
50 changes: 39 additions & 11 deletions hiredis.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -356,7 +356,12 @@ static int processLineItem(redisReader *r) {
char *p;
int len;

cur->poff = (r->pos-r->roff)-1;
cur->coff = cur->poff+1;
if ((p = readLine(r,&len)) != NULL) {
cur->plen = 1+len+2; /* include \r\n */
cur->clen = len;

if (cur->type == REDIS_REPLY_INTEGER) {
if (r->fn && r->fn->createInteger)
obj = r->fn->createInteger(cur,readLongLong(p));
Expand Down Expand Up @@ -395,10 +400,13 @@ static int processBulkItem(redisReader *r) {
p = r->buf+r->pos;
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 = readLongLong(p);
cur->poff = (r->pos-r->roff)-1;
cur->plen = bytelen+1;
cur->coff = cur->poff+1+bytelen;
cur->clen = 0;

len = readLongLong(p);
if (len < 0) {
/* The nil object can always be created. */
if (r->fn && r->fn->createNil)
Expand All @@ -410,6 +418,8 @@ static int processBulkItem(redisReader *r) {
/* Only continue when the buffer contains the entire bulk item. */
bytelen += len+2; /* include \r\n */
if (r->pos+bytelen <= r->len) {
cur->plen += len+2;
cur->clen = len;
if (r->fn && r->fn->createString)
obj = r->fn->createString(cur,s+2,len);
else
Expand Down Expand Up @@ -451,7 +461,12 @@ static int processMultiBulkItem(redisReader *r) {
return REDIS_ERR;
}

cur->poff = (r->pos-r->roff)-1;
cur->coff = 0;
if ((p = readLine(r,NULL)) != NULL) {
cur->plen = (r->pos-r->roff)-cur->poff; /* includes \r\n */
cur->clen = 0;

elements = readLongLong(p);
root = (r->ridx == 0);

Expand Down Expand Up @@ -588,7 +603,7 @@ int redisReaderFeed(redisReader *r, const char *buf, size_t len) {

/* Copy the provided buffer. */
if (buf != NULL && len >= 1) {
/* Destroy internal buffer when it is empty and is quite large. */
/* Destroy buffer when it is empty and is quite large. */
if (r->len == 0 && sdsavail(r->buf) > 16*1024) {
sdsfree(r->buf);
r->buf = sdsempty();
Expand All @@ -598,6 +613,15 @@ int redisReaderFeed(redisReader *r, const char *buf, size_t len) {
assert(r->buf != NULL);
}

/* Discard consumed part of the buffer when the offset for the reply
* that is currently being read is high enough. */
if (r->roff >= 1024) {
r->buf = sdsrange(r->buf,r->roff,-1);
r->pos -= r->roff;
r->roff = 0;
r->len = sdslen(r->buf);
}

newbuf = sdscatlen(r->buf,buf,len);
if (newbuf == NULL) {
__redisReaderSetErrorOOM(r);
Expand Down Expand Up @@ -633,6 +657,7 @@ int redisReaderGetReply(redisReader *r, void **reply) {
r->rstack[0].parent = NULL;
r->rstack[0].privdata = r->privdata;
r->ridx = 0;
r->roff = r->pos; /* Start offset in buffer. */
}

/* Process items in reply. */
Expand All @@ -644,14 +669,6 @@ int redisReaderGetReply(redisReader *r, void **reply) {
if (r->err)
return REDIS_ERR;

/* 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. */
if (r->ridx == -1) {
if (reply != NULL)
Expand All @@ -661,6 +678,17 @@ int redisReaderGetReply(redisReader *r, void **reply) {
return REDIS_OK;
}

const char *redisReaderGetRaw(redisReader *r, size_t *len) {
/* ridx == -1: No or a full reply has been read. */
/* pos > roff: Buffer position is larger than start offset, meaning
* the buffer has not yet been truncated. */
if (r->ridx == -1 && r->pos > r->roff) {
if (len) *len = (r->pos-r->roff);
return r->buf+r->roff;
}
return NULL;
}

/* Calculate the number of bytes needed to represent an integer as string. */
static int intlen(int i) {
int len = 0;
Expand Down
7 changes: 7 additions & 0 deletions hiredis.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,11 @@ typedef struct redisReply {
} redisReply;

typedef struct redisReadTask {
size_t poff; /* Protocol offset */
size_t plen; /* Protocol length */
size_t coff; /* Content offset */
size_t clen; /* Content length */

int type;
int elements; /* number of elements in multibulk container */
int idx; /* index in parent (array) object */
Expand All @@ -122,6 +127,7 @@ typedef struct redisReader {
char *buf; /* Read buffer */
size_t pos; /* Buffer cursor */
size_t len; /* Buffer length */
size_t roff; /* Reply offset */

redisReadTask rstack[3];
int ridx; /* Index of current read task */
Expand All @@ -136,6 +142,7 @@ redisReader *redisReaderCreate(void);
void redisReaderFree(redisReader *r);
int redisReaderFeed(redisReader *r, const char *buf, size_t len);
int redisReaderGetReply(redisReader *r, void **reply);
const char *redisReaderGetRaw(redisReader *r, size_t *len);

/* Backwards compatibility, can be removed on big version bump. */
#define redisReplyReaderCreate redisReaderCreate
Expand Down
95 changes: 94 additions & 1 deletion test.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -331,6 +331,98 @@ static void test_reply_reader(void) {
test_cond(ret == REDIS_ERR && reply == NULL);
}

static void *test_create_string(const redisReadTask *task, char *str, size_t len) {
redisReader *r = (redisReader*)task->privdata;
const char *roff = r->buf+r->roff;
((void)str); ((void)len);

assert(task->plen > 0);
assert(task->clen > 0);
switch(task->type) {
case REDIS_REPLY_STATUS:
assert(strncmp("+status\r\n", roff+task->poff, task->plen) == 0);
assert(strncmp("status", roff+task->coff, task->clen) == 0);
break;
case REDIS_REPLY_ERROR:
assert(strncmp("-error\r\n", roff+task->poff, task->plen) == 0);
assert(strncmp("error", roff+task->coff, task->clen) == 0);
break;
case REDIS_REPLY_STRING: /* bulk */
assert(strncmp("$4\r\nbulk\r\n", roff+task->poff, task->plen) == 0);
assert(strncmp("bulk", roff+task->coff, task->clen) == 0);
break;
default:
assert(NULL);
}
return (void*)1;
}

static void *test_create_array(const redisReadTask *task, int len) {
redisReader *r = (redisReader*)task->privdata;
const char *roff = r->buf+r->roff;
((void)len);

assert(task->plen > 0);
assert(task->clen == 0);
assert(strncmp("*5\r\n", roff+task->poff, task->plen) == 0);
return (void*)1;
}

static void *test_create_integer(const redisReadTask *task, long long value) {
redisReader *r = (redisReader*)task->privdata;
const char *roff = r->buf+r->roff;
((void)value);

assert(task->plen > 0);
assert(task->clen > 0);
assert(strncmp(":1234\r\n", roff+task->poff, task->plen) == 0);
assert(strncmp("1234", roff+task->coff, task->clen) == 0);
return (void*)1;
}

static void *test_create_nil(const redisReadTask *task) {
redisReader *r = (redisReader*)task->privdata;
const char *roff = r->buf+r->roff;

assert(task->plen > 0);
assert(task->clen == 0);
assert(strncmp("$-1\r\n", roff+task->poff, task->plen) == 0);
return (void*)1;
}

static redisReplyObjectFunctions test_reader_fn = {
test_create_string,
test_create_array,
test_create_integer,
test_create_nil,
NULL
};

static void test_reader_functions(void) {
redisReader *reader;
const char *input;
int ret;
void *obj;

input =
"*5\r\n"
"$-1\r\n"
":1234\r\n"
"+status\r\n"
"-error\r\n"
"$4\r\nbulk\r\n";

test("Custom object functions in reply reader: ");
reader = redisReaderCreate();
reader->fn = &test_reader_fn;
reader->privdata = reader;

redisReaderFeed(reader,input,strlen(input));
ret = redisReaderGetReply(reader,&obj);
test_cond(ret == REDIS_OK && obj == (void*)1);
redisReaderFree(reader);
}

static void test_throughput(void) {
int i, num;
long long t1, t2;
Expand Down Expand Up @@ -513,8 +605,9 @@ int main(int argc, char **argv) {

signal(SIGPIPE, SIG_IGN);
test_format_commands();
test_blocking_connection();
test_reply_reader();
test_reader_functions();
test_blocking_connection();
// test_nonblocking_connection();
test_throughput();
cleanup();
Expand Down

0 comments on commit 77540aa

Please sign in to comment.