mirrors/valkey
1
0
Fork 0
mirror of http://github.com/valkey-io/valkey synced 2024-11-22 00:52:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

ae.c now supports multiple polling API modules, even if only ae_select.c is implemented currently. Also adding and removing an event is now O(1).

Browse Source
This commit is contained in:
antirez 2009-11-23 18:50:39 +01:00
parent 5b2a1c292a
commit 266373b283
6 changed files with 147 additions and 126 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
Makefile
View File

@ -25,7 +25,8 @@ all: redis-server redis-benchmark redis-cli
# 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
@ -33,7 +34,8 @@ lzf_c.o: lzf_c.c lzfP.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

144
ae.c
View File

@ -38,20 +38,39 @@
#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);
}
@ -60,42 +79,39 @@ void aeStop(aeEventLoop *eventLoop) {
}
int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
aeFileProc *proc, void *clientData,
aeEventFinalizerProc *finalizerProc)
aeFileProc *proc, void *clientData)
{
aeFileEvent *fe;
if (fd >= AE_SETSIZE) return AE_ERR;
aeFileEvent *fe = &eventLoop->events[fd];
fe = zmalloc(sizeof(*fe));
if (fe == NULL) return AE_ERR;
fe->fd = fd;
fe->mask = mask;
fe->fileProc = proc;
fe->finalizerProc = finalizerProc;
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;
if (fd >= AE_SETSIZE) return;
aeFileEvent *fe = &eventLoop->events[fd];
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 (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)
@ -254,34 +270,18 @@ static int processTimeEvents(aeEventLoop *eventLoop) {
* 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;
@ -301,6 +301,8 @@ int aeProcessEvents(aeEventLoop *eventLoop, int flags)
} 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
@ -314,38 +316,24 @@ int aeProcessEvents(aeEventLoop *eventLoop, int flags)
}
}
retval = select(maxfd+1, &rfds, &wfds, &efds, tvp);
if (retval > 0) {
fe = eventLoop->fileEventHead;
while(fe != NULL) {
int fd = (int) fe->fd;
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;
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;
}
}
/* 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 */

85
ae.h
View File

@ -33,46 +33,12 @@
#ifndef __AE_H__
#define __AE_H__
struct aeEventLoop;
#define AE_SETSIZE (1024*10) /* Max number of fd supported */
/* Types and data structures */
typedef void aeFileProc(struct aeEventLoop *eventLoop, int fd, void *clientData, int mask);
typedef int aeTimeProc(struct aeEventLoop *eventLoop, long long id, void *clientData);
typedef void aeEventFinalizerProc(struct aeEventLoop *eventLoop, void *clientData);
/* File event structure */
typedef struct aeFileEvent {
int fd;
int mask; /* one of AE_(READABLE|WRITABLE|EXCEPTION) */
aeFileProc *fileProc;
aeEventFinalizerProc *finalizerProc;
void *clientData;
struct aeFileEvent *next;
} aeFileEvent;
/* Time event structure */
typedef struct aeTimeEvent {
long long id; /* time event identifier. */
long when_sec; /* seconds */
long when_ms; /* milliseconds */
aeTimeProc *timeProc;
aeEventFinalizerProc *finalizerProc;
void *clientData;
struct aeTimeEvent *next;
} aeTimeEvent;
/* State of an event based program */
typedef struct aeEventLoop {
long long timeEventNextId;
aeFileEvent *fileEventHead;
aeTimeEvent *timeEventHead;
int stop;
} aeEventLoop;
/* Defines */
#define AE_OK 0
#define AE_ERR -1
#define AE_NONE 0
#define AE_READABLE 1
#define AE_WRITABLE 2
#define AE_EXCEPTION 4
@ -87,13 +53,56 @@ typedef struct aeEventLoop {
/* Macros */
#define AE_NOTUSED(V) ((void) V)
struct aeEventLoop;
/* Types and data structures */
typedef void aeFileProc(struct aeEventLoop *eventLoop, int fd, void *clientData, int mask);
typedef int aeTimeProc(struct aeEventLoop *eventLoop, long long id, void *clientData);
typedef void aeEventFinalizerProc(struct aeEventLoop *eventLoop, void *clientData);
/* File event structure */
typedef struct aeFileEvent {
int mask; /* one of AE_(READABLE|WRITABLE|EXCEPTION) */
aeFileProc *rfileProc;
aeFileProc *wfileProc;
aeFileProc *efileProc;
void *clientData;
} aeFileEvent;
/* Time event structure */
typedef struct aeTimeEvent {
long long id; /* time event identifier. */
long when_sec; /* seconds */
long when_ms; /* milliseconds */
aeTimeProc *timeProc;
aeEventFinalizerProc *finalizerProc;
void *clientData;
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 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;
/* 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,

23
benchmark.c
View File

@ -78,6 +78,7 @@ static struct config {
list *clients;
int quiet;
int loop;
int idlemode;
} config;
typedef struct _client {
@ -136,7 +137,7 @@ static void freeAllClients(void) {
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 ||
@ -328,7 +329,7 @@ static void writeHandler(aeEventLoop *el, int fd, void *privdata, int mask)
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;
}
}
@ -352,7 +353,7 @@ static client createClient(void) {
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;
@ -451,6 +452,8 @@ void parseOptions(int argc, char **argv) {
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");
@ -469,6 +472,7 @@ void parseOptions(int argc, char **argv) {
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);
}
@ -493,6 +497,7 @@ int main(int argc, char **argv) {
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));
@ -506,6 +511,18 @@ int main(int argc, char **argv) {
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();

5
config.h
View File

@ -26,4 +26,9 @@
#define HAVE_BACKTRACE 1
#endif
/* test for polling API */
#ifdef __linux__
#define HAVE_EPOLL 1
#endif
#endif

10
redis.c
View File

@ -1912,7 +1912,7 @@ static redisClient *createClient(int fd) {
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;
}
@ -1925,7 +1925,7 @@ static void addReply(redisClient *c, robj *obj) {
(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 {
@ -5304,7 +5304,7 @@ static void sendBulkToSlave(aeEventLoop *el, int fd, void *privdata, int mask) {
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;
}
@ -5348,7 +5348,7 @@ static void updateSlavesWaitingBgsave(int bgsaveerr) {
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;
}
@ -5834,7 +5834,7 @@ int main(int argc, char **argv) {
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);