Commit e6043981 authored by antirez's avatar antirez
Browse files

ae.c: introduce the concept of read->write barrier.

AOF fsync=always, and certain Redis Cluster bus operations, require to
fsync data on disk before replying with an acknowledge.
In such case, in order to implement Group Commits, we want to be sure
that queries that are read in a given cycle of the event loop, are never
served to clients in the same event loop iteration. This way, by using
the event loop "before sleep" callback, we can fsync the information
just one time before returning into the event loop for the next cycle.
This is much more efficient compared to calling fsync() multiple times.

Unfortunately because of a bug, this was not always guaranteed: the
actual way the events are installed was the sole thing that could
control. Normally this problem is hard to trigger when AOF is enabled
with fsync=always, because we try to flush the output buffers to the
socekt directly in the beforeSleep() function of Redis. However if the
output buffers are full, we actually install a write event, and in suc...
parent fd5f229f
...@@ -158,6 +158,10 @@ void aeDeleteFileEvent(aeEventLoop *eventLoop, int fd, int mask) ...@@ -158,6 +158,10 @@ void aeDeleteFileEvent(aeEventLoop *eventLoop, int fd, int mask)
aeFileEvent *fe = &eventLoop->events[fd]; aeFileEvent *fe = &eventLoop->events[fd];
if (fe->mask == AE_NONE) return; if (fe->mask == AE_NONE) return;
/* We want to always remove AE_BARRIER if set when AE_WRITABLE
* is removed. */
if (mask & AE_WRITABLE) mask |= AE_BARRIER;
aeApiDelEvent(eventLoop, fd, mask); aeApiDelEvent(eventLoop, fd, mask);
fe->mask = fe->mask & (~mask); fe->mask = fe->mask & (~mask);
if (fd == eventLoop->maxfd && fe->mask == AE_NONE) { if (fd == eventLoop->maxfd && fe->mask == AE_NONE) {
...@@ -412,8 +416,22 @@ int aeProcessEvents(aeEventLoop *eventLoop, int flags) ...@@ -412,8 +416,22 @@ int aeProcessEvents(aeEventLoop *eventLoop, int flags)
fe->rfileProc(eventLoop,fd,fe->clientData,mask); fe->rfileProc(eventLoop,fd,fe->clientData,mask);
} }
if (fe->mask & mask & AE_WRITABLE) { if (fe->mask & mask & AE_WRITABLE) {
if (!rfired || fe->wfileProc != fe->rfileProc) int can_fire = 1;
fe->wfileProc(eventLoop,fd,fe->clientData,mask); if (rfired) {
/* The previous event fired? We do not want this to
* fire again if:
*
* 1. The handler is the same as the READABLE event.
* 2. If there AE_BARRIER is set, to signal that we
* are never allowed to fire WRITABLE after READABLE
* in the same iteration. */
if (fe->wfileProc == fe->rfileProc ||
fe->mask & AE_BARRIER)
{
can_fire = 0;
}
}
if (can_fire) fe->wfileProc(eventLoop,fd,fe->clientData,mask);
} }
processed++; processed++;
} }
......
...@@ -38,9 +38,14 @@ ...@@ -38,9 +38,14 @@
#define AE_OK 0 #define AE_OK 0
#define AE_ERR -1 #define AE_ERR -1
#define AE_NONE 0 #define AE_NONE 0 /* No events registered. */
#define AE_READABLE 1 #define AE_READABLE 1 /* Fire when descriptor is readable. */
#define AE_WRITABLE 2 #define AE_WRITABLE 2 /* Fire when descriptor is writable. */
#define AE_BARRIER 4 /* With WRITABLE, never fire the event if the
READABLE event already fired in the same event
loop iteration. Useful when you want to persist
things to disk before sending replies, and want
to do that in a group fashion. */
#define AE_FILE_EVENTS 1 #define AE_FILE_EVENTS 1
#define AE_TIME_EVENTS 2 #define AE_TIME_EVENTS 2
...@@ -63,7 +68,7 @@ typedef void aeBeforeSleepProc(struct aeEventLoop *eventLoop); ...@@ -63,7 +68,7 @@ typedef void aeBeforeSleepProc(struct aeEventLoop *eventLoop);
/* File event structure */ /* File event structure */
typedef struct aeFileEvent { typedef struct aeFileEvent {
int mask; /* one of AE_(READABLE|WRITABLE) */ int mask; /* one of AE_(READABLE|WRITABLE|BARRIER) */
aeFileProc *rfileProc; aeFileProc *rfileProc;
aeFileProc *wfileProc; aeFileProc *wfileProc;
void *clientData; void *clientData;
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment