gvsafronov/futriix
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Use hiredis from redis-benchmark

Browse Source
This commit is contained in:
Pieter Noordhuis 2010-11-04 13:37:05 +01:00
parent afc156c2d8
commit ec8f06675a
2 changed files with 66 additions and 236 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/Makefile
View File

@ -109,7 +109,11 @@ redis-server: $(OBJ)
@echo "" @echo ""
redis-benchmark: $(BENCHOBJ) redis-benchmark: $(BENCHOBJ)
$(CC) -o $(BENCHPRGNAME) $(CCOPT) $(DEBUG) $(BENCHOBJ) cd ../deps/hiredis && make static
$(CC) -o $(BENCHPRGNAME) $(CCOPT) $(DEBUG) $(BENCHOBJ) ../deps/hiredis/libhiredis.a
redis-benchmark.o:
$(CC) -c $(CFLAGS) -I../deps/hiredis $(DEBUG) $(COMPILE_TIME) $<
redis-cli: $(CLIOBJ) redis-cli: $(CLIOBJ)
cd ../deps/hiredis && make static cd ../deps/hiredis && make static

296
src/redis-benchmark.c
View File

@ -40,7 +40,7 @@
#include <assert.h> #include <assert.h>
#include "ae.h" #include "ae.h"
#include "anet.h" #include "hiredis.h"
#include "sds.h" #include "sds.h"
#include "adlist.h" #include "adlist.h"
#include "zmalloc.h" #include "zmalloc.h"
@ -84,13 +84,9 @@ static struct config {
} config; } config;
typedef struct _client { typedef struct _client {
redisContext *context;
int state; int state;
int fd;
sds obuf; sds obuf;
sds ibuf;
int mbulk; /* Number of elements in an mbulk reply */
int readlen; /* readlen == -1 means read a single line */
int totreceived;
unsigned int written; /* bytes of 'obuf' already written */ unsigned int written; /* bytes of 'obuf' already written */
int replytype; int replytype;
long long start; /* start time in milliseconds */ long long start; /* start time in milliseconds */
@ -113,12 +109,10 @@ static long long mstime(void) {
static void freeClient(client c) { static void freeClient(client c) {
listNode *ln; listNode *ln;
aeDeleteFileEvent(config.el,c->context->fd,AE_WRITABLE);
aeDeleteFileEvent(config.el,c->fd,AE_WRITABLE); aeDeleteFileEvent(config.el,c->context->fd,AE_READABLE);
aeDeleteFileEvent(config.el,c->fd,AE_READABLE); redisFree(c->context);
sdsfree(c->ibuf);
sdsfree(c->obuf); sdsfree(c->obuf);
close(c->fd);
zfree(c); zfree(c);
config.liveclients--; config.liveclients--;
ln = listSearchKey(config.clients,c); ln = listSearchKey(config.clients,c);
@ -137,19 +131,12 @@ static void freeAllClients(void) {
} }
static void resetClient(client c) { static void resetClient(client c) {
aeDeleteFileEvent(config.el,c->fd,AE_WRITABLE); aeDeleteFileEvent(config.el,c->context->fd,AE_WRITABLE);
aeDeleteFileEvent(config.el,c->fd,AE_READABLE); aeDeleteFileEvent(config.el,c->context->fd,AE_READABLE);
aeCreateFileEvent(config.el,c->fd, AE_WRITABLE,writeHandler,c); aeCreateFileEvent(config.el,c->context->fd,AE_WRITABLE,writeHandler,c);
sdsfree(c->ibuf);
c->ibuf = sdsempty();
c->readlen = (c->replytype == REPLY_BULK ||
c->replytype == REPLY_MBULK) ? -1 : 0;
c->mbulk = -1;
c->written = 0; c->written = 0;
c->totreceived = 0;
c->state = CLIENT_SENDQUERY; c->state = CLIENT_SENDQUERY;
c->start = mstime(); c->start = mstime();
createMissingClients(c);
} }
static void randomizeClientKey(client c) { static void randomizeClientKey(client c) {
@ -165,33 +152,13 @@ static void randomizeClientKey(client c) {
memcpy(p,buf,strlen(buf)); memcpy(p,buf,strlen(buf));
} }
static void prepareClientForReply(client c, int type) {
if (type == REPLY_BULK) {
c->replytype = REPLY_BULK;
c->readlen = -1;
} else if (type == REPLY_MBULK) {
c->replytype = REPLY_MBULK;
c->readlen = -1;
c->mbulk = -1;
} else {
c->replytype = type;
c->readlen = 0;
}
}
static void clientDone(client c) { static void clientDone(client c) {
static int last_tot_received = 1;
long long latency; long long latency;
config.donerequests ++; config.donerequests ++;
latency = mstime() - c->start; latency = mstime() - c->start;
if (latency > MAX_LATENCY) latency = MAX_LATENCY; if (latency > MAX_LATENCY) latency = MAX_LATENCY;
config.latency[latency]++; config.latency[latency]++;
if (config.debug && last_tot_received != c->totreceived) {
printf("Tot bytes received: %d\n", c->totreceived);
last_tot_received = c->totreceived;
}
if (config.donerequests == config.requests) { if (config.donerequests == config.requests) {
freeClient(c); freeClient(c);
aeStop(config.el); aeStop(config.el);
@ -208,126 +175,27 @@ static void clientDone(client c) {
} }
} }
/* Read a length from the buffer pointed to by *p, store the length in *len, static void readHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
* and return the number of bytes that the cursor advanced. */
static int readLen(char *p, int *len) {
char *tail = strstr(p,"\r\n");
if (tail == NULL)
return 0;
*tail = '\0';
*len = atoi(p+1);
return tail+2-p;
}
static void readHandler(aeEventLoop *el, int fd, void *privdata, int mask)
{
char buf[1024], *p;
int nread, pos=0, len=0;
client c = privdata; client c = privdata;
void *reply = NULL;
REDIS_NOTUSED(el); REDIS_NOTUSED(el);
REDIS_NOTUSED(fd); REDIS_NOTUSED(fd);
REDIS_NOTUSED(mask); REDIS_NOTUSED(mask);
nread = read(c->fd,buf,sizeof(buf)); if (redisBufferRead(c->context) != REDIS_OK) {
if (nread == -1) { fprintf(stderr,"Error: %s\n",c->context->errstr);
fprintf(stderr, "Reading from socket: %s\n", strerror(errno)); exit(1);
freeClient(c); } else {
return; if (redisGetReply(c->context,&reply) != REDIS_OK) {
fprintf(stderr,"Error: %s\n",c->context->errstr);
exit(1);
}
if (reply != NULL)
clientDone(c);
} }
if (nread == 0) {
fprintf(stderr, "EOF from client\n");
freeClient(c);
return;
}
c->totreceived += nread;
c->ibuf = sdscatlen(c->ibuf,buf,nread);
len = sdslen(c->ibuf);
if (c->replytype == REPLY_INT ||
c->replytype == REPLY_RETCODE)
{
/* Check if the first line is complete. This is everything we need
* when waiting for an integer or status code reply.*/
if ((p = strstr(c->ibuf,"\r\n")) != NULL)
goto done;
} else if (c->replytype == REPLY_BULK) {
int advance = 0;
if (c->readlen < 0) {
advance = readLen(c->ibuf+pos,&c->readlen);
if (advance) {
pos += advance;
if (c->readlen == -1) {
goto done;
} else {
/* include the trailing \r\n */
c->readlen += 2;
}
} else {
goto skip;
}
}
int canconsume;
if (c->readlen > 0) {
canconsume = c->readlen > (len-pos) ? (len-pos) : c->readlen;
c->readlen -= canconsume;
pos += canconsume;
}
if (c->readlen == 0)
goto done;
} else if (c->replytype == REPLY_MBULK) {
int advance = 0;
if (c->mbulk == -1) {
advance = readLen(c->ibuf+pos,&c->mbulk);
if (advance) {
pos += advance;
if (c->mbulk == -1)
goto done;
} else {
goto skip;
}
}
int canconsume;
while(c->mbulk > 0 && pos < len) {
if (c->readlen > 0) {
canconsume = c->readlen > (len-pos) ? (len-pos) : c->readlen;
c->readlen -= canconsume;
pos += canconsume;
if (c->readlen == 0)
c->mbulk--;
} else {
advance = readLen(c->ibuf+pos,&c->readlen);
if (advance) {
pos += advance;
if (c->readlen == -1) {
c->mbulk--;
continue;
} else {
/* include the trailing \r\n */
c->readlen += 2;
}
} else {
goto skip;
}
}
}
if (c->mbulk == 0)
goto done;
}
skip:
c->ibuf = sdsrange(c->ibuf,pos,-1);
return;
done:
clientDone(c);
return;
} }
static void writeHandler(aeEventLoop *el, int fd, void *privdata, int mask) static void writeHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
{
client c = privdata; client c = privdata;
REDIS_NOTUSED(el); REDIS_NOTUSED(el);
REDIS_NOTUSED(fd); REDIS_NOTUSED(fd);
@ -339,8 +207,7 @@ static void writeHandler(aeEventLoop *el, int fd, void *privdata, int mask)
} }
if (sdslen(c->obuf) > c->written) { if (sdslen(c->obuf) > c->written) {
void *ptr = c->obuf+c->written; void *ptr = c->obuf+c->written;
int len = sdslen(c->obuf) - c->written; int nwritten = write(c->context->fd,ptr,sdslen(c->obuf)-c->written);
int nwritten = write(c->fd, ptr, len);
if (nwritten == -1) { if (nwritten == -1) {
if (errno != EPIPE) if (errno != EPIPE)
fprintf(stderr, "Writing to socket: %s\n", strerror(errno)); fprintf(stderr, "Writing to socket: %s\n", strerror(errno));
@ -349,49 +216,45 @@ static void writeHandler(aeEventLoop *el, int fd, void *privdata, int mask)
} }
c->written += nwritten; c->written += nwritten;
if (sdslen(c->obuf) == c->written) { if (sdslen(c->obuf) == c->written) {
aeDeleteFileEvent(config.el,c->fd,AE_WRITABLE); aeDeleteFileEvent(config.el,c->context->fd,AE_WRITABLE);
aeCreateFileEvent(config.el,c->fd,AE_READABLE,readHandler,c); aeCreateFileEvent(config.el,c->context->fd,AE_READABLE,readHandler,c);
c->state = CLIENT_READREPLY; c->state = CLIENT_READREPLY;
} }
} }
} }
static client createClient(void) { static client createClient(int replytype) {
client c = zmalloc(sizeof(struct _client)); client c = zmalloc(sizeof(struct _client));
char err[ANET_ERR_LEN]; if (config.hostsocket == NULL) {
c->context = redisConnectNonBlock(config.hostip,config.hostport);
if (config.hostsocket == NULL) } else {
c->fd = anetTcpNonBlockConnect(err,config.hostip,config.hostport); c->context = redisConnectUnixNonBlock(config.hostsocket);
else
c->fd = anetUnixNonBlockConnect(err,config.hostsocket);
if (c->fd == ANET_ERR) {
zfree(c);
fprintf(stderr,"Connect: %s\n",err);
return NULL;
} }
anetTcpNoDelay(NULL,c->fd); if (c->context->err) {
c->obuf = sdsempty(); fprintf(stderr,"Could not connect to Redis at ");
c->ibuf = sdsempty(); if (config.hostsocket == NULL)
c->mbulk = -1; fprintf(stderr,"%s:%d: %s\n",config.hostip,config.hostport,c->context->errstr);
c->readlen = 0; else
c->written = 0; fprintf(stderr,"%s: %s\n",config.hostsocket,c->context->errstr);
c->totreceived = 0; exit(1);
}
c->replytype = replytype;
c->state = CLIENT_CONNECTING; c->state = CLIENT_CONNECTING;
aeCreateFileEvent(config.el, c->fd, AE_WRITABLE, writeHandler, c); c->obuf = sdsempty();
config.liveclients++; c->written = 0;
redisSetReplyObjectFunctions(c->context,NULL);
aeCreateFileEvent(config.el,c->context->fd,AE_WRITABLE,writeHandler,c);
listAddNodeTail(config.clients,c); listAddNodeTail(config.clients,c);
config.liveclients++;
return c; return c;
} }
static void createMissingClients(client c) { static void createMissingClients(client c) {
while(config.liveclients < config.numclients) { while(config.liveclients < config.numclients) {
client new = createClient(); client new = createClient(c->replytype);
if (!new) continue;
sdsfree(new->obuf); sdsfree(new->obuf);
new->obuf = sdsdup(c->obuf); new->obuf = sdsdup(c->obuf);
if (config.randomkeys) randomizeClientKey(c); if (config.randomkeys) randomizeClientKey(c);
prepareClientForReply(new,c->replytype);
} }
} }
@ -450,12 +313,7 @@ void parseOptions(int argc, char **argv) {
config.keepalive = atoi(argv[i+1]); config.keepalive = atoi(argv[i+1]);
i++; i++;
} else if (!strcmp(argv[i],"-h") && !lastarg) { } else if (!strcmp(argv[i],"-h") && !lastarg) {
char *ip = zmalloc(32); config.hostip = argv[i+1];
if (anetResolve(NULL,argv[i+1],ip) == ANET_ERR) {
printf("Can't resolve %s\n", argv[i]);
exit(1);
}
config.hostip = ip;
i++; i++;
} else if (!strcmp(argv[i],"-p") && !lastarg) { } else if (!strcmp(argv[i],"-p") && !lastarg) {
config.hostport = atoi(argv[i+1]); config.hostport = atoi(argv[i+1]);
@ -557,10 +415,8 @@ int main(int argc, char **argv) {
if (config.idlemode) { if (config.idlemode) {
printf("Creating %d idle connections and waiting forever (Ctrl+C when done)\n", config.numclients); printf("Creating %d idle connections and waiting forever (Ctrl+C when done)\n", config.numclients);
prepareForBenchmark("IDLE"); prepareForBenchmark("IDLE");
c = createClient(); c = createClient(0); /* will never receive a reply */
if (!c) exit(1);
c->obuf = sdsempty(); c->obuf = sdsempty();
prepareClientForReply(c,REPLY_RETCODE); /* will never receive it */
createMissingClients(c); createMissingClients(c);
aeMain(config.el); aeMain(config.el);
/* and will wait for every */ /* and will wait for every */
@ -568,26 +424,21 @@ int main(int argc, char **argv) {
do { do {
prepareForBenchmark("PING"); prepareForBenchmark("PING");
c = createClient(); c = createClient(REDIS_REPLY_STATUS);
if (!c) exit(1);
c->obuf = sdscat(c->obuf,"PING\r\n"); c->obuf = sdscat(c->obuf,"PING\r\n");
prepareClientForReply(c,REPLY_RETCODE);
createMissingClients(c); createMissingClients(c);
aeMain(config.el); aeMain(config.el);
endBenchmark(); endBenchmark();
prepareForBenchmark("PING (multi bulk)"); prepareForBenchmark("PING (multi bulk)");
c = createClient(); c = createClient(REDIS_REPLY_STATUS);
if (!c) exit(1);
c->obuf = sdscat(c->obuf,"*1\r\n$4\r\nPING\r\n"); c->obuf = sdscat(c->obuf,"*1\r\n$4\r\nPING\r\n");
prepareClientForReply(c,REPLY_RETCODE);
createMissingClients(c); createMissingClients(c);
aeMain(config.el); aeMain(config.el);
endBenchmark(); endBenchmark();
prepareForBenchmark("MSET (10 keys, multi bulk)"); prepareForBenchmark("MSET (10 keys, multi bulk)");
c = createClient(); c = createClient(REDIS_REPLY_ARRAY);
if (!c) exit(1);
c->obuf = sdscatprintf(c->obuf,"*%d\r\n$4\r\nMSET\r\n", 11); c->obuf = sdscatprintf(c->obuf,"*%d\r\n$4\r\nMSET\r\n", 11);
{ {
int i; int i;
@ -598,14 +449,12 @@ int main(int argc, char **argv) {
} }
zfree(data); zfree(data);
} }
prepareClientForReply(c,REPLY_RETCODE);
createMissingClients(c); createMissingClients(c);
aeMain(config.el); aeMain(config.el);
endBenchmark(); endBenchmark();
prepareForBenchmark("SET"); prepareForBenchmark("SET");
c = createClient(); c = createClient(REDIS_REPLY_STATUS);
if (!c) exit(1);
c->obuf = sdscat(c->obuf,"SET foo_rand000000000000 "); c->obuf = sdscat(c->obuf,"SET foo_rand000000000000 ");
{ {
char *data = zmalloc(config.datasize+2); char *data = zmalloc(config.datasize+2);
@ -614,106 +463,83 @@ int main(int argc, char **argv) {
data[config.datasize+1] = '\n'; data[config.datasize+1] = '\n';
c->obuf = sdscatlen(c->obuf,data,config.datasize+2); c->obuf = sdscatlen(c->obuf,data,config.datasize+2);
} }
prepareClientForReply(c,REPLY_RETCODE);
createMissingClients(c); createMissingClients(c);
aeMain(config.el); aeMain(config.el);
endBenchmark(); endBenchmark();
prepareForBenchmark("GET"); prepareForBenchmark("GET");
c = createClient(); c = createClient(REDIS_REPLY_STRING);
if (!c) exit(1);
c->obuf = sdscat(c->obuf,"GET foo_rand000000000000\r\n"); c->obuf = sdscat(c->obuf,"GET foo_rand000000000000\r\n");
prepareClientForReply(c,REPLY_BULK);
createMissingClients(c); createMissingClients(c);
aeMain(config.el); aeMain(config.el);
endBenchmark(); endBenchmark();
prepareForBenchmark("INCR"); prepareForBenchmark("INCR");
c = createClient(); c = createClient(REDIS_REPLY_INTEGER);
if (!c) exit(1);
c->obuf = sdscat(c->obuf,"INCR counter_rand000000000000\r\n"); c->obuf = sdscat(c->obuf,"INCR counter_rand000000000000\r\n");
prepareClientForReply(c,REPLY_INT);
createMissingClients(c); createMissingClients(c);
aeMain(config.el); aeMain(config.el);
endBenchmark(); endBenchmark();
prepareForBenchmark("LPUSH"); prepareForBenchmark("LPUSH");
c = createClient(); c = createClient(REDIS_REPLY_INTEGER);
if (!c) exit(1);
c->obuf = sdscat(c->obuf,"LPUSH mylist bar\r\n"); c->obuf = sdscat(c->obuf,"LPUSH mylist bar\r\n");
prepareClientForReply(c,REPLY_INT);
createMissingClients(c); createMissingClients(c);
aeMain(config.el); aeMain(config.el);
endBenchmark(); endBenchmark();
prepareForBenchmark("LPOP"); prepareForBenchmark("LPOP");
c = createClient(); c = createClient(REDIS_REPLY_STRING);
if (!c) exit(1);
c->obuf = sdscat(c->obuf,"LPOP mylist\r\n"); c->obuf = sdscat(c->obuf,"LPOP mylist\r\n");
prepareClientForReply(c,REPLY_BULK);
createMissingClients(c); createMissingClients(c);
aeMain(config.el); aeMain(config.el);
endBenchmark(); endBenchmark();
prepareForBenchmark("SADD"); prepareForBenchmark("SADD");
c = createClient(); c = createClient(REDIS_REPLY_STATUS);
if (!c) exit(1);
c->obuf = sdscat(c->obuf,"SADD myset counter_rand000000000000\r\n"); c->obuf = sdscat(c->obuf,"SADD myset counter_rand000000000000\r\n");
prepareClientForReply(c,REPLY_RETCODE);
createMissingClients(c); createMissingClients(c);
aeMain(config.el); aeMain(config.el);
endBenchmark(); endBenchmark();
prepareForBenchmark("SPOP"); prepareForBenchmark("SPOP");
c = createClient(); c = createClient(REDIS_REPLY_STRING);
if (!c) exit(1);
c->obuf = sdscat(c->obuf,"SPOP myset\r\n"); c->obuf = sdscat(c->obuf,"SPOP myset\r\n");
prepareClientForReply(c,REPLY_BULK);
createMissingClients(c); createMissingClients(c);
aeMain(config.el); aeMain(config.el);
endBenchmark(); endBenchmark();
prepareForBenchmark("LPUSH (again, in order to bench LRANGE)"); prepareForBenchmark("LPUSH (again, in order to bench LRANGE)");
c = createClient(); c = createClient(REDIS_REPLY_STATUS);
if (!c) exit(1);
c->obuf = sdscat(c->obuf,"LPUSH mylist bar\r\n"); c->obuf = sdscat(c->obuf,"LPUSH mylist bar\r\n");
prepareClientForReply(c,REPLY_RETCODE);
createMissingClients(c); createMissingClients(c);
aeMain(config.el); aeMain(config.el);
endBenchmark(); endBenchmark();
prepareForBenchmark("LRANGE (first 100 elements)"); prepareForBenchmark("LRANGE (first 100 elements)");
c = createClient(); c = createClient(REDIS_REPLY_ARRAY);
if (!c) exit(1);
c->obuf = sdscat(c->obuf,"LRANGE mylist 0 99\r\n"); c->obuf = sdscat(c->obuf,"LRANGE mylist 0 99\r\n");
prepareClientForReply(c,REPLY_MBULK);
createMissingClients(c); createMissingClients(c);
aeMain(config.el); aeMain(config.el);
endBenchmark(); endBenchmark();
prepareForBenchmark("LRANGE (first 300 elements)"); prepareForBenchmark("LRANGE (first 300 elements)");
c = createClient(); c = createClient(REDIS_REPLY_ARRAY);
if (!c) exit(1);
c->obuf = sdscat(c->obuf,"LRANGE mylist 0 299\r\n"); c->obuf = sdscat(c->obuf,"LRANGE mylist 0 299\r\n");
prepareClientForReply(c,REPLY_MBULK);
createMissingClients(c); createMissingClients(c);
aeMain(config.el); aeMain(config.el);
endBenchmark(); endBenchmark();
prepareForBenchmark("LRANGE (first 450 elements)"); prepareForBenchmark("LRANGE (first 450 elements)");
c = createClient(); c = createClient(REDIS_REPLY_ARRAY);
if (!c) exit(1);
c->obuf = sdscat(c->obuf,"LRANGE mylist 0 449\r\n"); c->obuf = sdscat(c->obuf,"LRANGE mylist 0 449\r\n");
prepareClientForReply(c,REPLY_MBULK);
createMissingClients(c); createMissingClients(c);
aeMain(config.el); aeMain(config.el);
endBenchmark(); endBenchmark();
prepareForBenchmark("LRANGE (first 600 elements)"); prepareForBenchmark("LRANGE (first 600 elements)");
c = createClient(); c = createClient(REDIS_REPLY_ARRAY);
if (!c) exit(1);
c->obuf = sdscat(c->obuf,"LRANGE mylist 0 599\r\n"); c->obuf = sdscat(c->obuf,"LRANGE mylist 0 599\r\n");
prepareClientForReply(c,REPLY_MBULK);
createMissingClients(c); createMissingClients(c);
aeMain(config.el); aeMain(config.el);
endBenchmark(); endBenchmark();