# Deps (use make dep to generate this)
adlist.o: adlist.c adlist.h zmalloc.h
-ae.o: ae.c ae.h zmalloc.h
+ae.o: ae.c ae.h zmalloc.h ae_select.c
+ae_select.o: ae_select.c
anet.o: anet.c fmacros.h anet.h
benchmark.o: benchmark.c fmacros.h ae.h anet.h sds.h adlist.h zmalloc.h
dict.o: dict.c fmacros.h dict.h zmalloc.h
lzf_d.o: lzf_d.c lzfP.h
pqsort.o: pqsort.c
redis-cli.o: redis-cli.c fmacros.h anet.h sds.h adlist.h zmalloc.h
-redis.o: redis.c fmacros.h ae.h sds.h anet.h dict.h adlist.h zmalloc.h lzf.h pqsort.h config.h
+redis.o: redis.c fmacros.h config.h redis.h ae.h sds.h anet.h dict.h \
+ adlist.h zmalloc.h lzf.h pqsort.h staticsymbols.h
sds.o: sds.c sds.h zmalloc.h
zmalloc.o: zmalloc.c config.h
#include "ae.h"
#include "zmalloc.h"
+#include "config.h"
+
+/* Include the best multiplexing layer supported by this system.
+ * The following should be ordered by performances, descending. */
+#ifdef HAVE_EPOLL
+#include "ae_epoll.c"
+#else
+#include "ae_select.c"
+#endif
aeEventLoop *aeCreateEventLoop(void) {
aeEventLoop *eventLoop;
+ int i;
eventLoop = zmalloc(sizeof(*eventLoop));
if (!eventLoop) return NULL;
- eventLoop->fileEventHead = NULL;
eventLoop->timeEventHead = NULL;
eventLoop->timeEventNextId = 0;
eventLoop->stop = 0;
+ eventLoop->maxfd = -1;
+ if (aeApiCreate(eventLoop) == -1) {
+ zfree(eventLoop);
+ return NULL;
+ }
+ /* Events with mask == AE_NONE are not set. So let's initialize the
+ * vector with it. */
+ for (i = 0; i < AE_SETSIZE; i++)
+ eventLoop->events[i].mask = AE_NONE;
return eventLoop;
}
void aeDeleteEventLoop(aeEventLoop *eventLoop) {
+ aeApiFree(eventLoop);
zfree(eventLoop);
}
}
int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
- aeFileProc *proc, void *clientData,
- aeEventFinalizerProc *finalizerProc)
+ aeFileProc *proc, void *clientData)
{
- aeFileEvent *fe;
-
- fe = zmalloc(sizeof(*fe));
- if (fe == NULL) return AE_ERR;
- fe->fd = fd;
- fe->mask = mask;
- fe->fileProc = proc;
- fe->finalizerProc = finalizerProc;
+ if (fd >= AE_SETSIZE) return AE_ERR;
+ aeFileEvent *fe = &eventLoop->events[fd];
+
+ if (aeApiAddEvent(eventLoop, fd, mask) == -1)
+ return AE_ERR;
+ fe->mask |= mask;
+ if (mask & AE_READABLE) fe->rfileProc = proc;
+ if (mask & AE_WRITABLE) fe->wfileProc = proc;
+ if (mask & AE_EXCEPTION) fe->efileProc = proc;
fe->clientData = clientData;
- fe->next = eventLoop->fileEventHead;
- eventLoop->fileEventHead = fe;
+ if (fd > eventLoop->maxfd)
+ eventLoop->maxfd = fd;
return AE_OK;
}
void aeDeleteFileEvent(aeEventLoop *eventLoop, int fd, int mask)
{
- aeFileEvent *fe, *prev = NULL;
-
- fe = eventLoop->fileEventHead;
- while(fe) {
- if (fe->fd == fd && fe->mask == mask) {
- if (prev == NULL)
- eventLoop->fileEventHead = fe->next;
- else
- prev->next = fe->next;
- if (fe->finalizerProc)
- fe->finalizerProc(eventLoop, fe->clientData);
- zfree(fe);
- return;
- }
- prev = fe;
- fe = fe->next;
+ if (fd >= AE_SETSIZE) return;
+ aeFileEvent *fe = &eventLoop->events[fd];
+
+ if (fe->mask == AE_NONE) return;
+ fe->mask = fe->mask & (~mask);
+ if (fd == eventLoop->maxfd && fe->mask == AE_NONE) {
+ /* Update the max fd */
+ int j;
+
+ for (j = eventLoop->maxfd-1; j >= 0; j--)
+ if (eventLoop->events[j].mask != AE_NONE) break;
+ eventLoop->maxfd = j;
}
+ aeApiDelEvent(eventLoop, fd, mask);
}
static void aeGetTime(long *seconds, long *milliseconds)
* The function returns the number of events processed. */
int aeProcessEvents(aeEventLoop *eventLoop, int flags)
{
- int maxfd = 0, numfd = 0, processed = 0;
- fd_set rfds, wfds, efds;
- aeFileEvent *fe = eventLoop->fileEventHead;
+ int processed = 0, numevents;
/* Nothing to do? return ASAP */
if (!(flags & AE_TIME_EVENTS) && !(flags & AE_FILE_EVENTS)) return 0;
- FD_ZERO(&rfds);
- FD_ZERO(&wfds);
- FD_ZERO(&efds);
-
- /* Check file events */
- if (flags & AE_FILE_EVENTS) {
- while (fe != NULL) {
- if (fe->mask & AE_READABLE) FD_SET(fe->fd, &rfds);
- if (fe->mask & AE_WRITABLE) FD_SET(fe->fd, &wfds);
- if (fe->mask & AE_EXCEPTION) FD_SET(fe->fd, &efds);
- if (maxfd < fe->fd) maxfd = fe->fd;
- numfd++;
- fe = fe->next;
- }
- }
/* Note that we want call select() even if there are no
* file events to process as long as we want to process time
* events, in order to sleep until the next time event is ready
* to fire. */
- if (numfd || ((flags & AE_TIME_EVENTS) && !(flags & AE_DONT_WAIT))) {
- int retval;
+ if (eventLoop->maxfd != -1 ||
+ ((flags & AE_TIME_EVENTS) && !(flags & AE_DONT_WAIT))) {
+ int j;
aeTimeEvent *shortest = NULL;
struct timeval tv, *tvp;
} else {
tvp->tv_usec = (shortest->when_ms - now_ms)*1000;
}
+ if (tvp->tv_sec < 0) tvp->tv_sec = 0;
+ if (tvp->tv_usec < 0) tvp->tv_usec = 0;
} else {
/* If we have to check for events but need to return
* ASAP because of AE_DONT_WAIT we need to se the timeout
}
}
- retval = select(maxfd+1, &rfds, &wfds, &efds, tvp);
- if (retval > 0) {
- fe = eventLoop->fileEventHead;
- while(fe != NULL) {
- int fd = (int) fe->fd;
-
- if ((fe->mask & AE_READABLE && FD_ISSET(fd, &rfds)) ||
- (fe->mask & AE_WRITABLE && FD_ISSET(fd, &wfds)) ||
- (fe->mask & AE_EXCEPTION && FD_ISSET(fd, &efds)))
- {
- int mask = 0;
-
- if (fe->mask & AE_READABLE && FD_ISSET(fd, &rfds))
- mask |= AE_READABLE;
- if (fe->mask & AE_WRITABLE && FD_ISSET(fd, &wfds))
- mask |= AE_WRITABLE;
- if (fe->mask & AE_EXCEPTION && FD_ISSET(fd, &efds))
- mask |= AE_EXCEPTION;
- fe->fileProc(eventLoop, fe->fd, fe->clientData, mask);
- processed++;
- /* After an event is processed our file event list
- * may no longer be the same, so what we do
- * is to clear the bit for this file descriptor and
- * restart again from the head. */
- fe = eventLoop->fileEventHead;
- FD_CLR(fd, &rfds);
- FD_CLR(fd, &wfds);
- FD_CLR(fd, &efds);
- } else {
- fe = fe->next;
- }
- }
+ numevents = aeApiPoll(eventLoop, tvp);
+ for (j = 0; j < numevents; j++) {
+ aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd];
+ int mask = eventLoop->fired[j].mask;
+ int fd = eventLoop->fired[j].fd;
+
+ /* note the fe->mask & mask & ... code: maybe an already processed
+ * event removed an element that fired and we still didn't
+ * processed, so we check if the event is still valid. */
+ if (fe->mask & mask & AE_READABLE)
+ fe->rfileProc(eventLoop,fd,fe->clientData,mask);
+ if (fe->mask & mask & AE_WRITABLE && fe->wfileProc != fe->rfileProc)
+ fe->wfileProc(eventLoop,fd,fe->clientData,mask);
+ if (fe->mask & mask & AE_EXCEPTION &&
+ fe->efileProc != fe->wfileProc &&
+ fe->efileProc != fe->rfileProc)
+ fe->efileProc(eventLoop,fd,fe->clientData,mask);
+ processed++;
}
}
/* Check time events */
#ifndef __AE_H__
#define __AE_H__
+#define AE_SETSIZE (1024*10) /* Max number of fd supported */
+
+#define AE_OK 0
+#define AE_ERR -1
+
+#define AE_NONE 0
+#define AE_READABLE 1
+#define AE_WRITABLE 2
+#define AE_EXCEPTION 4
+
+#define AE_FILE_EVENTS 1
+#define AE_TIME_EVENTS 2
+#define AE_ALL_EVENTS (AE_FILE_EVENTS|AE_TIME_EVENTS)
+#define AE_DONT_WAIT 4
+
+#define AE_NOMORE -1
+
+/* Macros */
+#define AE_NOTUSED(V) ((void) V)
+
struct aeEventLoop;
/* Types and data structures */
/* File event structure */
typedef struct aeFileEvent {
- int fd;
int mask; /* one of AE_(READABLE|WRITABLE|EXCEPTION) */
- aeFileProc *fileProc;
- aeEventFinalizerProc *finalizerProc;
+ aeFileProc *rfileProc;
+ aeFileProc *wfileProc;
+ aeFileProc *efileProc;
void *clientData;
- struct aeFileEvent *next;
} aeFileEvent;
/* Time event structure */
struct aeTimeEvent *next;
} aeTimeEvent;
+/* A fired event */
+typedef struct aeFiredEvent {
+ int fd;
+ int mask;
+} aeFiredEvent;
+
/* State of an event based program */
typedef struct aeEventLoop {
+ int maxfd;
long long timeEventNextId;
- aeFileEvent *fileEventHead;
+ aeFileEvent events[AE_SETSIZE]; /* Registered events */
+ aeFiredEvent fired[AE_SETSIZE]; /* Fired events */
aeTimeEvent *timeEventHead;
int stop;
+ void *apidata; /* This is used for polling API specific data */
} aeEventLoop;
-/* Defines */
-#define AE_OK 0
-#define AE_ERR -1
-
-#define AE_READABLE 1
-#define AE_WRITABLE 2
-#define AE_EXCEPTION 4
-
-#define AE_FILE_EVENTS 1
-#define AE_TIME_EVENTS 2
-#define AE_ALL_EVENTS (AE_FILE_EVENTS|AE_TIME_EVENTS)
-#define AE_DONT_WAIT 4
-
-#define AE_NOMORE -1
-
-/* Macros */
-#define AE_NOTUSED(V) ((void) V)
-
/* Prototypes */
aeEventLoop *aeCreateEventLoop(void);
void aeDeleteEventLoop(aeEventLoop *eventLoop);
void aeStop(aeEventLoop *eventLoop);
int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
- aeFileProc *proc, void *clientData,
- aeEventFinalizerProc *finalizerProc);
+ aeFileProc *proc, void *clientData);
void aeDeleteFileEvent(aeEventLoop *eventLoop, int fd, int mask);
long long aeCreateTimeEvent(aeEventLoop *eventLoop, long long milliseconds,
aeTimeProc *proc, void *clientData,
list *clients;
int quiet;
int loop;
+ int idlemode;
} config;
typedef struct _client {
static void resetClient(client c) {
aeDeleteFileEvent(config.el,c->fd,AE_WRITABLE);
aeDeleteFileEvent(config.el,c->fd,AE_READABLE);
- aeCreateFileEvent(config.el,c->fd, AE_WRITABLE,writeHandler,c,NULL);
+ aeCreateFileEvent(config.el,c->fd, AE_WRITABLE,writeHandler,c);
sdsfree(c->ibuf);
c->ibuf = sdsempty();
c->readlen = (c->replytype == REPLY_BULK ||
c->written += nwritten;
if (sdslen(c->obuf) == c->written) {
aeDeleteFileEvent(config.el,c->fd,AE_WRITABLE);
- aeCreateFileEvent(config.el,c->fd,AE_READABLE,readHandler,c,NULL);
+ aeCreateFileEvent(config.el,c->fd,AE_READABLE,readHandler,c);
c->state = CLIENT_READREPLY;
}
}
c->written = 0;
c->totreceived = 0;
c->state = CLIENT_CONNECTING;
- aeCreateFileEvent(config.el, c->fd, AE_WRITABLE, writeHandler, c, NULL);
+ aeCreateFileEvent(config.el, c->fd, AE_WRITABLE, writeHandler, c);
config.liveclients++;
listAddNodeTail(config.clients,c);
return c;
config.loop = 1;
} else if (!strcmp(argv[i],"-D")) {
config.debug = 1;
+ } else if (!strcmp(argv[i],"-I")) {
+ config.idlemode = 1;
} else {
printf("Wrong option '%s' or option argument missing\n\n",argv[i]);
printf("Usage: redis-benchmark [-h <host>] [-p <port>] [-c <clients>] [-n <requests]> [-k <boolean>]\n\n");
printf(" range will be allowed.\n");
printf(" -q Quiet. Just show query/sec values\n");
printf(" -l Loop. Run the tests forever\n");
+ printf(" -I Idle mode. Just open N idle connections and wait.\n");
printf(" -D Debug mode. more verbose.\n");
exit(1);
}
config.randomkeys_keyspacelen = 0;
config.quiet = 0;
config.loop = 0;
+ config.idlemode = 0;
config.latency = NULL;
config.clients = listCreate();
config.latency = zmalloc(sizeof(int)*(MAX_LATENCY+1));
printf("WARNING: keepalive disabled, you probably need 'echo 1 > /proc/sys/net/ipv4/tcp_tw_reuse' for Linux and 'sudo sysctl -w net.inet.tcp.msl=1000' for Mac OS X in order to use a lot of clients/requests\n");
}
+ if (config.idlemode) {
+ printf("Creating %d idle connections and waiting forever (Ctrl+C when done)\n", config.numclients);
+ prepareForBenchmark();
+ c = createClient();
+ if (!c) exit(1);
+ c->obuf = sdsempty();
+ prepareClientForReply(c,REPLY_RETCODE); /* will never receive it */
+ createMissingClients(c);
+ aeMain(config.el);
+ /* and will wait for every */
+ }
+
do {
prepareForBenchmark();
c = createClient();
#define HAVE_BACKTRACE 1
#endif
+/* test for polling API */
+#ifdef __linux__
+#define HAVE_EPOLL 1
+#endif
+
#endif
listSetFreeMethod(c->reply,decrRefCount);
listSetDupMethod(c->reply,dupClientReplyValue);
if (aeCreateFileEvent(server.el, c->fd, AE_READABLE,
- readQueryFromClient, c, NULL) == AE_ERR) {
+ readQueryFromClient, c) == AE_ERR) {
freeClient(c);
return NULL;
}
(c->replstate == REDIS_REPL_NONE ||
c->replstate == REDIS_REPL_ONLINE) &&
aeCreateFileEvent(server.el, c->fd, AE_WRITABLE,
- sendReplyToClient, c, NULL) == AE_ERR) return;
+ sendReplyToClient, c) == AE_ERR) return;
if (obj->encoding != REDIS_ENCODING_RAW) {
obj = getDecodedObject(obj);
} else {
aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE);
slave->replstate = REDIS_REPL_ONLINE;
if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE,
- sendReplyToClient, slave, NULL) == AE_ERR) {
+ sendReplyToClient, slave) == AE_ERR) {
freeClient(slave);
return;
}
slave->repldbsize = buf.st_size;
slave->replstate = REDIS_REPL_SEND_BULK;
aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE);
- if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE, sendBulkToSlave, slave, NULL) == AE_ERR) {
+ if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE, sendBulkToSlave, slave) == AE_ERR) {
freeClient(slave);
continue;
}
redisLog(REDIS_NOTICE,"DB loaded from disk");
}
if (aeCreateFileEvent(server.el, server.fd, AE_READABLE,
- acceptHandler, NULL, NULL) == AE_ERR) oom("creating file event");
+ acceptHandler, NULL) == AE_ERR) oom("creating file event");
redisLog(REDIS_NOTICE,"The server is now ready to accept connections on port %d", server.port);
aeMain(server.el);
aeDeleteEventLoop(server.el);
projects / redis.git / commitdiff
