redis-benchmark.c

/* Redis benchmark utility.
 *
 * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com>
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 * * Redistributions of source code must retain the above copyright notice,
 * this list of conditions and the following disclaimer.
 * * Redistributions in binary form must reproduce the above copyright
 * notice, this list of conditions and the following disclaimer in the
 * documentation and/or other materials provided with the distribution.
 * * Neither the name of Redis nor the names of its contributors may be used
 * to endorse or promote products derived from this software without
 * specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */

#include"fmacros.h"
#include"version.h"

#include<stdio.h>
#include<string.h>
#include<stdlib.h>
#include<unistd.h>
#include<errno.h>
#include<time.h>
#include<sys/time.h>
#include<signal.h>
#include<assert.h>
#include<math.h>
#include<pthread.h>

#include<sdscompat.h>/* Use hiredis' sds compat header that maps sds calls to their hi_ variants */
#include<sds.h>/* Use hiredis sds. */
#include"ae.h"
#include<hiredis.h>
#ifdefUSE_OPENSSL
#include<openssl/ssl.h>
#include<openssl/err.h>
#include<hiredis_ssl.h>
#endif
#include"adlist.h"
#include"dict.h"
#include"zmalloc.h"
#include"atomicvar.h"
#include"crc16_slottable.h"
#include"hdr_histogram.h"
#include"cli_common.h"
#include"mt19937-64.h"

#defineUNUSED(V) ((void) V)
#defineRANDPTR_INITIAL_SIZE 8
#defineDEFAULT_LATENCY_PRECISION 3
#defineMAX_LATENCY_PRECISION 4
#defineMAX_THREADS 500
#defineCLUSTER_SLOTS 16384
#defineCONFIG_LATENCY_HISTOGRAM_MIN_VALUE 10L /* >= 10 usecs */
#defineCONFIG_LATENCY_HISTOGRAM_MAX_VALUE 3000000L /* <= 30 secs(us precision) */
#defineCONFIG_LATENCY_HISTOGRAM_INSTANT_MAX_VALUE 3000000L /* <= 3 secs(us precision) */

#defineCLIENT_GET_EVENTLOOP(c) \
 (c->thread_id >= 0 ? config.threads[c->thread_id]->el : config.el)

structbenchmarkThread;
structclusterNode;
structredisConfig;

staticstructconfig {
aeEventLoop*el;
constchar*hostip;
inthostport;
constchar*hostsocket;
inttls;
structcliSSLconfigsslconfig;
intnumclients;
redisAtomicintliveclients;
intrequests;
redisAtomicintrequests_issued;
redisAtomicintrequests_finished;
redisAtomicintprevious_requests_finished;
intlast_printed_bytes;
long longprevious_tick;
intkeysize;
intdatasize;
intrandomkeys;
intrandomkeys_keyspacelen;
intkeepalive;
intpipeline;
long longstart;
long longtotlatency;
constchar*title;
list*clients;
intquiet;
intcsv;
intloop;
intidlemode;
intdbnum;
sdsdbnumstr;
char*tests;
char*auth;
constchar*user;
intprecision;
intnum_threads;
structbenchmarkThread**threads;
intcluster_mode;
intcluster_node_count;
structclusterNode**cluster_nodes;
structredisConfig*redis_config;
structhdr_histogram*latency_histogram;
structhdr_histogram*current_sec_latency_histogram;
redisAtomicintis_fetching_slots;
redisAtomicintis_updating_slots;
redisAtomicintslots_last_update;
intenable_tracking;
pthread_mutex_tliveclients_mutex;
pthread_mutex_tis_updating_slots_mutex;
} config;

typedefstruct_client {
redisContext*context;
sdsobuf;
char**randptr; /* Pointers to :rand: strings inside the command buf */
size_trandlen; /* Number of pointers in client->randptr */
size_trandfree; /* Number of unused pointers in client->randptr */
char**stagptr; /* Pointers to slot hashtags (cluster mode only) */
size_tstaglen; /* Number of pointers in client->stagptr */
size_tstagfree; /* Number of unused pointers in client->stagptr */
size_twritten; /* Bytes of 'obuf' already written */
long longstart; /* Start time of a request */
long longlatency; /* Request latency */
intpending; /* Number of pending requests (replies to consume) */
intprefix_pending; /* If non-zero, number of pending prefix commands. Commands
 such as auth and select are prefixed to the pipeline of
 benchmark commands and discarded after the first send. */
intprefixlen; /* Size in bytes of the pending prefix commands */
intthread_id;
structclusterNode*cluster_node;
intslots_last_update;
} *client;

/* Threads. */

typedefstructbenchmarkThread {
intindex;
pthread_tthread;
aeEventLoop*el;
} benchmarkThread;

/* Cluster. */
typedefstructclusterNode {
char*ip;
intport;
sdsname;
intflags;
sdsreplicate; /* Master ID if node is a slave */
int*slots;
intslots_count;
intcurrent_slot_index;
int*updated_slots; /* Used by updateClusterSlotsConfiguration */
intupdated_slots_count; /* Used by updateClusterSlotsConfiguration */
intreplicas_count;
sds*migrating; /* An array of sds where even strings are slots and odd
 * strings are the destination node IDs. */
sds*importing; /* An array of sds where even strings are slots and odd
 * strings are the source node IDs. */
intmigrating_count; /* Length of the migrating array (migrating slots*2) */
intimporting_count; /* Length of the importing array (importing slots*2) */
structredisConfig*redis_config;
} clusterNode;

typedefstructredisConfig {
sdssave;
sdsappendonly;
} redisConfig;

/* Prototypes */
char*redisGitSHA1(void);
char*redisGitDirty(void);
staticvoidwriteHandler(aeEventLoop*el, intfd, void*privdata, intmask);
staticvoidcreateMissingClients(clientc);
staticbenchmarkThread*createBenchmarkThread(intindex);
staticvoidfreeBenchmarkThread(benchmarkThread*thread);
staticvoidfreeBenchmarkThreads();
staticvoid*execBenchmarkThread(void*ptr);
staticclusterNode*createClusterNode(char*ip, intport);
staticredisConfig*getRedisConfig(constchar*ip, intport,
constchar*hostsocket);
staticredisContext*getRedisContext(constchar*ip, intport,
constchar*hostsocket);
staticvoidfreeRedisConfig(redisConfig*cfg);
staticintfetchClusterSlotsConfiguration(clientc);
staticvoidupdateClusterSlotsConfiguration();
intshowThroughput(structaeEventLoop*eventLoop, long longid,
void*clientData);

staticsdsbenchmarkVersion(void) {
sdsversion;
version=sdscatprintf(sdsempty(), "%s", REDIS_VERSION);

/* Add git commit and working tree status when available */
if (strtoll(redisGitSHA1(),NULL,16)) {
version=sdscatprintf(version, " (git:%s", redisGitSHA1());
if (strtoll(redisGitDirty(),NULL,10))
version=sdscatprintf(version, "-dirty");
version=sdscat(version, ")");
 }
returnversion;
}

/* Dict callbacks */
staticuint64_tdictSdsHash(constvoid*key);
staticintdictSdsKeyCompare(void*privdata, constvoid*key1,
constvoid*key2);

/* Implementation */
staticlong longustime(void) {
structtimevaltv;
long longust;

gettimeofday(&tv, NULL);
ust= ((long)tv.tv_sec)*1000000;
ust+=tv.tv_usec;
returnust;
}

staticlong longmstime(void) {
structtimevaltv;
long longmst;

gettimeofday(&tv, NULL);
mst= ((long long)tv.tv_sec)*1000;
mst+=tv.tv_usec/1000;
returnmst;
}

staticuint64_tdictSdsHash(constvoid*key) {
returndictGenHashFunction((unsigned char*)key, sdslen((char*)key));
}

staticintdictSdsKeyCompare(void*privdata, constvoid*key1,
constvoid*key2)
{
intl1,l2;
DICT_NOTUSED(privdata);

l1=sdslen((sds)key1);
l2=sdslen((sds)key2);
if (l1!=l2) return0;
returnmemcmp(key1, key2, l1) ==0;
}

staticredisContext*getRedisContext(constchar*ip, intport,
constchar*hostsocket)
{
redisContext*ctx=NULL;
redisReply*reply=NULL;
if (hostsocket==NULL)
ctx=redisConnect(ip, port);
else
ctx=redisConnectUnix(hostsocket);
if (ctx==NULL||ctx->err) {
fprintf(stderr,"Could not connect to Redis at ");
char*err= (ctx!=NULL ? ctx->errstr : "");
if (hostsocket==NULL)
fprintf(stderr,"%s:%d: %s\n",ip,port,err);
else
fprintf(stderr,"%s: %s\n",hostsocket,err);
 goto cleanup;
 }
if (config.tls==1) {
constchar*err=NULL;
if (cliSecureConnection(ctx, config.sslconfig, &err) ==REDIS_ERR&&err) {
fprintf(stderr, "Could not negotiate a TLS connection: %s\n", err);
 goto cleanup;
 }
 }
if (config.auth==NULL)
returnctx;
if (config.user==NULL)
reply=redisCommand(ctx,"AUTH %s", config.auth);
else
reply=redisCommand(ctx,"AUTH %s %s", config.user, config.auth);
if (reply!=NULL) {
if (reply->type==REDIS_REPLY_ERROR) {
if (hostsocket==NULL)
fprintf(stderr, "Node %s:%d replied with error:\n%s\n", ip, port, reply->str);
else
fprintf(stderr, "Node %s replied with error:\n%s\n", hostsocket, reply->str);
freeReplyObject(reply);
redisFree(ctx);
exit(1);
 }
freeReplyObject(reply);
returnctx;
 }
fprintf(stderr, "ERROR: failed to fetch reply from ");
if (hostsocket==NULL)
fprintf(stderr, "%s:%d\n", ip, port);
else
fprintf(stderr, "%s\n", hostsocket);
cleanup:
freeReplyObject(reply);
redisFree(ctx);
returnNULL;
}



staticredisConfig*getRedisConfig(constchar*ip, intport,
constchar*hostsocket)
{
redisConfig*cfg=zcalloc(sizeof(*cfg));
if (!cfg) returnNULL;
redisContext*c=NULL;
redisReply*reply=NULL, *sub_reply=NULL;
c=getRedisContext(ip, port, hostsocket);
if (c==NULL) {
freeRedisConfig(cfg);
returnNULL;
 }
redisAppendCommand(c, "CONFIG GET %s", "save");
redisAppendCommand(c, "CONFIG GET %s", "appendonly");
inti=0;
void*r=NULL;
for (; i<2; i++) {
intres=redisGetReply(c, &r);
if (reply) freeReplyObject(reply);
reply=res==REDIS_OK ? ((redisReply*) r) : NULL;
if (res!=REDIS_OK|| !r) goto fail;
if (reply->type==REDIS_REPLY_ERROR) {
fprintf(stderr, "ERROR: %s\n", reply->str);
 goto fail;
 }
if (reply->type!=REDIS_REPLY_ARRAY||reply->elements<2) goto fail;
sub_reply=reply->element[1];
char*value=sub_reply->str;
if (!value) value="";
switch (i) {
case0: cfg->save=sdsnew(value); break;
case1: cfg->appendonly=sdsnew(value); break;
 }
 }
freeReplyObject(reply);
redisFree(c);
returncfg;
fail:
fprintf(stderr, "ERROR: failed to fetch CONFIG from ");
if (hostsocket==NULL) fprintf(stderr, "%s:%d\n", ip, port);
elsefprintf(stderr, "%s\n", hostsocket);
intabort_test=0;
if (reply&&reply->type==REDIS_REPLY_ERROR&&
 (!strncmp(reply->str,"NOAUTH",6) ||
 !strncmp(reply->str,"WRONGPASS",9) ||
 !strncmp(reply->str,"NOPERM",6)))
abort_test=1;
freeReplyObject(reply);
redisFree(c);
freeRedisConfig(cfg);
if (abort_test) exit(1);
returnNULL;
}
staticvoidfreeRedisConfig(redisConfig*cfg) {
if (cfg->save) sdsfree(cfg->save);
if (cfg->appendonly) sdsfree(cfg->appendonly);
zfree(cfg);
}

staticvoidfreeClient(clientc) {
aeEventLoop*el=CLIENT_GET_EVENTLOOP(c);
listNode*ln;
aeDeleteFileEvent(el,c->context->fd,AE_WRITABLE);
aeDeleteFileEvent(el,c->context->fd,AE_READABLE);
if (c->thread_id >= 0) {
intrequests_finished=0;
atomicGet(config.requests_finished, requests_finished);
if (requests_finished >= config.requests) {
aeStop(el);
 }
 }
redisFree(c->context);
sdsfree(c->obuf);
zfree(c->randptr);
zfree(c->stagptr);
zfree(c);
if (config.num_threads) pthread_mutex_lock(&(config.liveclients_mutex));
config.liveclients--;
ln=listSearchKey(config.clients,c);
assert(ln!=NULL);
listDelNode(config.clients,ln);
if (config.num_threads) pthread_mutex_unlock(&(config.liveclients_mutex));
}

staticvoidfreeAllClients(void) {
listNode*ln=config.clients->head, *next;

while(ln) {
next=ln->next;
freeClient(ln->value);
ln=next;
 }
}

staticvoidresetClient(clientc) {
aeEventLoop*el=CLIENT_GET_EVENTLOOP(c);
aeDeleteFileEvent(el,c->context->fd,AE_WRITABLE);
aeDeleteFileEvent(el,c->context->fd,AE_READABLE);
aeCreateFileEvent(el,c->context->fd,AE_WRITABLE,writeHandler,c);
c->written=0;
c->pending=config.pipeline;
}

staticvoidrandomizeClientKey(clientc) {
size_ti;

for (i=0; i<c->randlen; i++) {
char*p=c->randptr[i]+11;
size_tr=0;
if (config.randomkeys_keyspacelen!=0)
r=random() % config.randomkeys_keyspacelen;
size_tj;

for (j=0; j<12; j++) {
*p='0'+r%10;
r/=10;
p--;
 }
 }
}

staticvoidsetClusterKeyHashTag(clientc) {
assert(c->thread_id >= 0);
clusterNode*node=c->cluster_node;
assert(node);
assert(node->current_slot_index<node->slots_count);
intis_updating_slots=0;
atomicGet(config.is_updating_slots, is_updating_slots);
/* If updateClusterSlotsConfiguration is updating the slots array,
 * call updateClusterSlotsConfiguration is order to block the thread
 * since the mutex is locked. When the slots will be updated by the
 * thread that's actually performing the update, the execution of
 * updateClusterSlotsConfiguration won't actually do anything, since
 * the updated_slots_count array will be already NULL. */
if (is_updating_slots) updateClusterSlotsConfiguration();
intslot=node->slots[node->current_slot_index];
constchar*tag=crc16_slot_table[slot];
inttaglen=strlen(tag);
size_ti;
for (i=0; i<c->staglen; i++) {
char*p=c->stagptr[i] +1;
p[0] =tag[0];
p[1] = (taglen >= 2 ? tag[1] : '}');
p[2] = (taglen==3 ? tag[2] : '}');
 }
}

staticvoidclientDone(clientc) {
intrequests_finished=0;
atomicGet(config.requests_finished, requests_finished);
if (requests_finished >= config.requests) {
freeClient(c);
if (!config.num_threads&&config.el) aeStop(config.el);
return;
 }
if (config.keepalive) {
resetClient(c);
 } else {
if (config.num_threads) pthread_mutex_lock(&(config.liveclients_mutex));
config.liveclients--;
createMissingClients(c);
config.liveclients++;
if (config.num_threads)
pthread_mutex_unlock(&(config.liveclients_mutex));
freeClient(c);
 }
}

staticvoidreadHandler(aeEventLoop*el, intfd, void*privdata, intmask) {
clientc=privdata;
void*reply=NULL;
UNUSED(el);
UNUSED(fd);
UNUSED(mask);

/* Calculate latency only for the first read event. This means that the
 * server already sent the reply and we need to parse it. Parsing overhead
 * is not part of the latency, so calculate it only once, here. */
if (c->latency<0) c->latency=ustime()-(c->start);

if (redisBufferRead(c->context) !=REDIS_OK) {
fprintf(stderr,"Error: %s\n",c->context->errstr);
exit(1);
 } else {
while(c->pending) {
if (redisGetReply(c->context,&reply) !=REDIS_OK) {
fprintf(stderr,"Error: %s\n",c->context->errstr);
exit(1);
 }
if (reply!=NULL) {
if (reply== (void*)REDIS_REPLY_ERROR) {
fprintf(stderr,"Unexpected error reply, exiting...\n");
exit(1);
 }
redisReply*r=reply;
if (r->type==REDIS_REPLY_ERROR) {
/* Try to update slots configuration if reply error is
 * MOVED/ASK/CLUSTERDOWN and the key(s) used by the command
 * contain(s) the slot hash tag.
 * If the error is not topology-update related then we
 * immediately exit to avoid false results. */
if (c->cluster_node&&c->staglen) {
intfetch_slots=0, do_wait=0;
if (!strncmp(r->str,"MOVED",5) || !strncmp(r->str,"ASK",3))
fetch_slots=1;
elseif (!strncmp(r->str,"CLUSTERDOWN",11)) {
/* Usually the cluster is able to recover itself after
 * a CLUSTERDOWN error, so try to sleep one second
 * before requesting the new configuration. */
fetch_slots=1;
do_wait=1;
fprintf(stderr, "Error from server %s:%d: %s.\n",
c->cluster_node->ip,
c->cluster_node->port,
r->str);
 }
if (do_wait) sleep(1);
if (fetch_slots&& !fetchClusterSlotsConfiguration(c))
exit(1);
 } else {
if (c->cluster_node) {
fprintf(stderr, "Error from server %s:%d: %s\n",
c->cluster_node->ip,
c->cluster_node->port,
r->str);
 } elsefprintf(stderr, "Error from server: %s\n", r->str);
exit(1);
 }
 }

freeReplyObject(reply);
/* This is an OK for prefix commands such as auth and select.*/
if (c->prefix_pending>0) {
c->prefix_pending--;
c->pending--;
/* Discard prefix commands on first response.*/
if (c->prefixlen>0) {
size_tj;
sdsrange(c->obuf, c->prefixlen, -1);
/* We also need to fix the pointers to the strings
 * we need to randomize. */
for (j=0; j<c->randlen; j++)
c->randptr[j] -=c->prefixlen;
/* Fix the pointers to the slot hash tags */
for (j=0; j<c->staglen; j++)
c->stagptr[j] -=c->prefixlen;
c->prefixlen=0;
 }
continue;
 }
intrequests_finished=0;
atomicGetIncr(config.requests_finished, requests_finished, 1);
if (requests_finished<config.requests){
if (config.num_threads==0) {
hdr_record_value(
config.latency_histogram, // Histogram to record to
 (long)c->latency<=CONFIG_LATENCY_HISTOGRAM_MAX_VALUE ? (long)c->latency : CONFIG_LATENCY_HISTOGRAM_MAX_VALUE); // Value to record
hdr_record_value(
config.current_sec_latency_histogram, // Histogram to record to
 (long)c->latency<=CONFIG_LATENCY_HISTOGRAM_INSTANT_MAX_VALUE ? (long)c->latency : CONFIG_LATENCY_HISTOGRAM_INSTANT_MAX_VALUE); // Value to record
 } else {
hdr_record_value_atomic(
config.latency_histogram, // Histogram to record to
 (long)c->latency<=CONFIG_LATENCY_HISTOGRAM_MAX_VALUE ? (long)c->latency : CONFIG_LATENCY_HISTOGRAM_MAX_VALUE); // Value to record
hdr_record_value_atomic(
config.current_sec_latency_histogram, // Histogram to record to
 (long)c->latency<=CONFIG_LATENCY_HISTOGRAM_INSTANT_MAX_VALUE ? (long)c->latency : CONFIG_LATENCY_HISTOGRAM_INSTANT_MAX_VALUE); // Value to record
 }
 }
c->pending--;
if (c->pending==0) {
clientDone(c);
break;
 }
 } else {
break;
 }
 }
 }
}

staticvoidwriteHandler(aeEventLoop*el, intfd, void*privdata, intmask) {
clientc=privdata;
UNUSED(el);
UNUSED(fd);
UNUSED(mask);

/* Initialize request when nothing was written. */
if (c->written==0) {
/* Enforce upper bound to number of requests. */
intrequests_issued=0;
atomicGetIncr(config.requests_issued, requests_issued, config.pipeline);
if (requests_issued >= config.requests) {
return;
 }

/* Really initialize: randomize keys and set start time. */
if (config.randomkeys) randomizeClientKey(c);
if (config.cluster_mode&&c->staglen>0) setClusterKeyHashTag(c);
atomicGet(config.slots_last_update, c->slots_last_update);
c->start=ustime();
c->latency=-1;
 }
constssize_tbuflen=sdslen(c->obuf);
constssize_twriteLen=buflen-c->written;
if (writeLen>0) {
void*ptr=c->obuf+c->written;
while(1) {
/* Optimistically try to write before checking if the file descriptor
 * is actually writable. At worst we get EAGAIN. */
constssize_tnwritten=cliWriteConn(c->context,ptr,writeLen);
if (nwritten!=writeLen) {
if (nwritten==-1&&errno!=EAGAIN) {
if (errno!=EPIPE)
fprintf(stderr, "Error writing to the server: %s\n", strerror(errno));
freeClient(c);
return;
 }
 } else {
aeDeleteFileEvent(el,c->context->fd,AE_WRITABLE);
aeCreateFileEvent(el,c->context->fd,AE_READABLE,readHandler,c);
return;
 }
 }
 }
}

/* Create a benchmark client, configured to send the command passed as 'cmd' of
 * 'len' bytes.
 *
 * The command is copied N times in the client output buffer (that is reused
 * again and again to send the request to the server) accordingly to the configured
 * pipeline size.
 *
 * Also an initial SELECT command is prepended in order to make sure the right
 * database is selected, if needed. The initial SELECT will be discarded as soon
 * as the first reply is received.
 *
 * To create a client from scratch, the 'from' pointer is set to NULL. If instead
 * we want to create a client using another client as reference, the 'from' pointer
 * points to the client to use as reference. In such a case the following
 * information is take from the 'from' client:
 *
 * 1) The command line to use.
 * 2) The offsets of the __rand_int__ elements inside the command line, used
 * for arguments randomization.
 *
 * Even when cloning another client, prefix commands are applied if needed.*/
staticclientcreateClient(char*cmd, size_tlen, clientfrom, intthread_id) {
intj;
intis_cluster_client= (config.cluster_mode&&thread_id >= 0);
clientc=zmalloc(sizeof(struct_client));

constchar*ip=NULL;
intport=0;
c->cluster_node=NULL;
if (config.hostsocket==NULL||is_cluster_client) {
if (!is_cluster_client) {
ip=config.hostip;
port=config.hostport;
 } else {
intnode_idx=0;
if (config.num_threads<config.cluster_node_count)
node_idx=config.liveclients % config.cluster_node_count;
else
node_idx=thread_id % config.cluster_node_count;
clusterNode*node=config.cluster_nodes[node_idx];
assert(node!=NULL);
ip= (constchar*) node->ip;
port=node->port;
c->cluster_node=node;
 }
c->context=redisConnectNonBlock(ip,port);
 } else {
c->context=redisConnectUnixNonBlock(config.hostsocket);
 }
if (c->context->err) {
fprintf(stderr,"Could not connect to Redis at ");
if (config.hostsocket==NULL||is_cluster_client)
fprintf(stderr,"%s:%d: %s\n",ip,port,c->context->errstr);
else
fprintf(stderr,"%s: %s\n",config.hostsocket,c->context->errstr);
exit(1);
 }
if (config.tls==1) {
constchar*err=NULL;
if (cliSecureConnection(c->context, config.sslconfig, &err) ==REDIS_ERR&&err) {
fprintf(stderr, "Could not negotiate a TLS connection: %s\n", err);
exit(1);
 }
 }
c->thread_id=thread_id;
/* Suppress hiredis cleanup of unused buffers for max speed. */
c->context->reader->maxbuf=0;

/* Build the request buffer:
 * Queue N requests accordingly to the pipeline size, or simply clone
 * the example client buffer. */
c->obuf=sdsempty();
/* Prefix the request buffer with AUTH and/or SELECT commands, if applicable.
 * These commands are discarded after the first response, so if the client is
 * reused the commands will not be used again. */
c->prefix_pending=0;
if (config.auth) {
char*buf=NULL;
intlen;
if (config.user==NULL)
len=redisFormatCommand(&buf, "AUTH %s", config.auth);
else
len=redisFormatCommand(&buf, "AUTH %s %s",
config.user, config.auth);
c->obuf=sdscatlen(c->obuf, buf, len);
free(buf);
c->prefix_pending++;
 }

if (config.enable_tracking) {
char*buf=NULL;
intlen=redisFormatCommand(&buf, "CLIENT TRACKING on");
c->obuf=sdscatlen(c->obuf, buf, len);
free(buf);
c->prefix_pending++;
 }

/* If a DB number different than zero is selected, prefix our request
 * buffer with the SELECT command, that will be discarded the first
 * time the replies are received, so if the client is reused the
 * SELECT command will not be used again. */
if (config.dbnum!=0&& !is_cluster_client) {
c->obuf=sdscatprintf(c->obuf,"*2\r\n$6\r\nSELECT\r\n$%d\r\n%s\r\n",
 (int)sdslen(config.dbnumstr),config.dbnumstr);
c->prefix_pending++;
 }
c->prefixlen=sdslen(c->obuf);
/* Append the request itself. */
if (from) {
c->obuf=sdscatlen(c->obuf,
from->obuf+from->prefixlen,
sdslen(from->obuf)-from->prefixlen);
 } else {
for (j=0; j<config.pipeline; j++)
c->obuf=sdscatlen(c->obuf,cmd,len);
 }

c->written=0;
c->pending=config.pipeline+c->prefix_pending;
c->randptr=NULL;
c->randlen=0;
c->stagptr=NULL;
c->staglen=0;

/* Find substrings in the output buffer that need to be randomized. */
if (config.randomkeys) {
if (from) {
c->randlen=from->randlen;
c->randfree=0;
c->randptr=zmalloc(sizeof(char*)*c->randlen);
/* copy the offsets. */
for (j=0; j< (int)c->randlen; j++) {
c->randptr[j] =c->obuf+ (from->randptr[j]-from->obuf);
/* Adjust for the different select prefix length. */
c->randptr[j] +=c->prefixlen-from->prefixlen;
 }
 } else {
char*p=c->obuf;

c->randlen=0;
c->randfree=RANDPTR_INITIAL_SIZE;
c->randptr=zmalloc(sizeof(char*)*c->randfree);
while ((p=strstr(p,"__rand_int__")) !=NULL) {
if (c->randfree==0) {
c->randptr=zrealloc(c->randptr,sizeof(char*)*c->randlen*2);
c->randfree+=c->randlen;
 }
c->randptr[c->randlen++] =p;
c->randfree--;
p+=12; /* 12 is strlen("__rand_int__). */
 }
 }
 }
/* If cluster mode is enabled, set slot hashtags pointers. */
if (config.cluster_mode) {
if (from) {
c->staglen=from->staglen;
c->stagfree=0;
c->stagptr=zmalloc(sizeof(char*)*c->staglen);
/* copy the offsets. */
for (j=0; j< (int)c->staglen; j++) {
c->stagptr[j] =c->obuf+ (from->stagptr[j]-from->obuf);
/* Adjust for the different select prefix length. */
c->stagptr[j] +=c->prefixlen-from->prefixlen;
 }
 } else {
char*p=c->obuf;

c->staglen=0;
c->stagfree=RANDPTR_INITIAL_SIZE;
c->stagptr=zmalloc(sizeof(char*)*c->stagfree);
while ((p=strstr(p,"{tag}")) !=NULL) {
if (c->stagfree==0) {
c->stagptr=zrealloc(c->stagptr,
sizeof(char*) *c->staglen*2);
c->stagfree+=c->staglen;
 }
c->stagptr[c->staglen++] =p;
c->stagfree--;
p+=5; /* 5 is strlen("{tag}"). */
 }
 }
 }
aeEventLoop*el=NULL;
if (thread_id<0) el=config.el;
else {
benchmarkThread*thread=config.threads[thread_id];
el=thread->el;
 }
if (config.idlemode==0)
aeCreateFileEvent(el,c->context->fd,AE_WRITABLE,writeHandler,c);
listAddNodeTail(config.clients,c);
atomicIncr(config.liveclients, 1);
atomicGet(config.slots_last_update, c->slots_last_update);
returnc;
}

staticvoidcreateMissingClients(clientc) {
intn=0;
while(config.liveclients<config.numclients) {
intthread_id=-1;
if (config.num_threads)
thread_id=config.liveclients % config.num_threads;
createClient(NULL,0,c,thread_id);

/* Listen backlog is quite limited on most systems */
if (++n>64) {
usleep(50000);
n=0;
 }
 }
}

staticvoidshowLatencyReport(void) {

constfloatreqpersec= (float)config.requests_finished/((float)config.totlatency/1000.0f);
constfloatp0= ((float) hdr_min(config.latency_histogram))/1000.0f;
constfloatp50=hdr_value_at_percentile(config.latency_histogram, 50.0 )/1000.0f;
constfloatp95=hdr_value_at_percentile(config.latency_histogram, 95.0 )/1000.0f;
constfloatp99=hdr_value_at_percentile(config.latency_histogram, 99.0 )/1000.0f;
constfloatp100= ((float) hdr_max(config.latency_histogram))/1000.0f;
constfloatavg=hdr_mean(config.latency_histogram)/1000.0f;

if (!config.quiet&& !config.csv) {
printf("%*s\r", config.last_printed_bytes, " "); // ensure there is a clean line
printf("====== %s ======\n", config.title);
printf(" %d requests completed in %.2f seconds\n", config.requests_finished,
 (float)config.totlatency/1000);
printf(" %d parallel clients\n", config.numclients);
printf(" %d bytes payload\n", config.datasize);
printf(" keep alive: %d\n", config.keepalive);
if (config.cluster_mode) {
printf(" cluster mode: yes (%d masters)\n",
config.cluster_node_count);
intm ;
for (m=0; m<config.cluster_node_count; m++) {
clusterNode*node=config.cluster_nodes[m];
redisConfig*cfg=node->redis_config;
if (cfg==NULL) continue;
printf(" node [%d] configuration:\n",m );
printf(" save: %s\n",
sdslen(cfg->save) ? cfg->save : "NONE");
printf(" appendonly: %s\n", cfg->appendonly);
 }
 } else {
if (config.redis_config) {
printf(" host configuration \"save\": %s\n",
config.redis_config->save);
printf(" host configuration \"appendonly\": %s\n",
config.redis_config->appendonly);
 }
 }
printf(" multi-thread: %s\n", (config.num_threads ? "yes" : "no"));
if (config.num_threads)
printf(" threads: %d\n", config.num_threads);

printf("\n");
printf("Latency by percentile distribution:\n");
structhdr_iteriter;
long longprevious_cumulative_count=-1;
constlong longtotal_count=config.latency_histogram->total_count;
hdr_iter_percentile_init(&iter, config.latency_histogram, 1);
structhdr_iter_percentiles*percentiles=&iter.specifics.percentiles;
while (hdr_iter_next(&iter))
 {
constdoublevalue=iter.highest_equivalent_value / 1000.0f;
constdoublepercentile=percentiles->percentile;
constlong longcumulative_count=iter.cumulative_count;
if( previous_cumulative_count!=cumulative_count||cumulative_count==total_count ){
printf("%3.3f%% <= %.3f milliseconds (cumulative count %lld)\n", percentile, value, cumulative_count);
 }
previous_cumulative_count=cumulative_count;
 }
printf("\n");
printf("Cumulative distribution of latencies:\n");
previous_cumulative_count=-1;
hdr_iter_linear_init(&iter, config.latency_histogram, 100);
while (hdr_iter_next(&iter))
 {
constdoublevalue=iter.highest_equivalent_value / 1000.0f;
constlong longcumulative_count=iter.cumulative_count;
constdoublepercentile= ((double)cumulative_count/(double)total_count)*100.0;
if( previous_cumulative_count!=cumulative_count||cumulative_count==total_count ){
printf("%3.3f%% <= %.3f milliseconds (cumulative count %lld)\n", percentile, value, cumulative_count);
 }
/* After the 2 milliseconds latency to have percentages split
 * by decimals will just add a lot of noise to the output. */
if(iter.highest_equivalent_value>2000){
hdr_iter_linear_set_value_units_per_bucket(&iter,1000);
 }
previous_cumulative_count=cumulative_count;
 }
printf("\n");
printf("Summary:\n");
printf(" throughput summary: %.2f requests per second\n", reqpersec);
printf(" latency summary (msec):\n");
printf(" %9s %9s %9s %9s %9s %9s\n", "avg", "min", "p50", "p95", "p99", "max");
printf(" %9.3f %9.3f %9.3f %9.3f %9.3f %9.3f\n", avg, p0, p50, p95, p99, p100);
 } elseif (config.csv) {
printf("\"%s\",\"%.2f\",\"%.3f\",\"%.3f\",\"%.3f\",\"%.3f\",\"%.3f\",\"%.3f\"\n", config.title, reqpersec, avg, p0, p50, p95, p99, p100);
 } else {
printf("%*s\r", config.last_printed_bytes, " "); // ensure there is a clean line
printf("%s: %.2f requests per second, p50=%.3f msec\n", config.title, reqpersec, p50);
 }
}

staticvoidinitBenchmarkThreads() {
inti;
if (config.threads) freeBenchmarkThreads();
config.threads=zmalloc(config.num_threads*sizeof(benchmarkThread*));
for (i=0; i<config.num_threads; i++) {
benchmarkThread*thread=createBenchmarkThread(i);
config.threads[i] =thread;
 }
}

staticvoidstartBenchmarkThreads() {
inti;
for (i=0; i<config.num_threads; i++) {
benchmarkThread*t=config.threads[i];
if (pthread_create(&(t->thread), NULL, execBenchmarkThread, t)){
fprintf(stderr, "FATAL: Failed to start thread %d.\n", i);
exit(1);
 }
 }
for (i=0; i<config.num_threads; i++)
pthread_join(config.threads[i]->thread, NULL);
}

staticvoidbenchmark(char*title, char*cmd, intlen) {
clientc;

config.title=title;
config.requests_issued=0;
config.requests_finished=0;
config.previous_requests_finished=0;
config.last_printed_bytes=0;
hdr_init(
CONFIG_LATENCY_HISTOGRAM_MIN_VALUE, // Minimum value
CONFIG_LATENCY_HISTOGRAM_MAX_VALUE, // Maximum value
config.precision, // Number of significant figures
&config.latency_histogram); // Pointer to initialise
hdr_init(
CONFIG_LATENCY_HISTOGRAM_MIN_VALUE, // Minimum value
CONFIG_LATENCY_HISTOGRAM_INSTANT_MAX_VALUE, // Maximum value
config.precision, // Number of significant figures
&config.current_sec_latency_histogram); // Pointer to initialise

if (config.num_threads) initBenchmarkThreads();

intthread_id=config.num_threads>0 ? 0 : -1;
c=createClient(cmd,len,NULL,thread_id);