WAIT AOF: deny if AOF is off. Trigger sync in proper place.

src/aof.c

@@ -197,10 +197,40 @@ ssize_t aofRewriteBufferWrite(int fd) {
  * AOF file implementation
  * ------------------------------------------------------------------------- */
 
+/* Return true if an AOf fsync is currently already in progress in a
+ * BIO thread. */
+int aofFsyncInProgress(void) {
+    return bioPendingJobsOfType(BIO_AOF_FSYNC) != 0;
+}
+
 /* Starts a background task that performs fsync() against the specified
  * file descriptor (the one of the AOF file) in another thread. */
-void aof_background_fsync(int fd) {
-    bioCreateBackgroundJob(BIO_AOF_FSYNC,(void*)(long)fd,NULL,NULL);
+void aofStartBackgroundFsync(void) {
+    if (aofFsyncInProgress()) {
+        /* We never want to start another fsync if one is in progress. */
+        return;
+    }
+
+    /* Before starting a new fsync, we need to flush the AOF buffer to
+     * disk if there are clients blocked in WAIT AOF, otherwise we are
+     * not going to sync their data. */
+    if (server.blocked_clients_by_type[BLOCKED_AOF])
+        flushAppendOnlyFile(0);
+
+    /* No fsync is in progress. If there was one, the new epoch is stored
+     * in server.aof_fsync_in_progress_epoch. So update the current fsync
+     * epoch with the one of the fsync in progress. */
+    server.aof_fsync_epoch = server.aof_fsync_in_progress_epoch;
+    bioCreateBackgroundJob(BIO_AOF_FSYNC,(void*)(long)server.aof_fd,NULL,NULL);
+    server.aof_fsync_in_progress_epoch++;
+}
+
+/* Returns an AOF epoch so that, when such epoch is reached by
+ * server.aof_fsync_epoch, it means that everything that was written in the
+ * AOF up to the moment this function returned the epoch, is now flushed
+ * on disk. */
+uint64_t aofNextEpoch(void) {
+    return server.aof_fsync_in_progress_epoch + 1;
 }
 
 /* Kills an AOFRW child process if exists */
@@ -336,7 +366,7 @@ void flushAppendOnlyFile(int force) {
     if (sdslen(server.aof_buf) == 0) return;
 
     if (server.aof_fsync == AOF_FSYNC_EVERYSEC)
-        sync_in_progress = bioPendingJobsOfType(BIO_AOF_FSYNC) != 0;
+        sync_in_progress = aofFsyncInProgress();
 
     if (server.aof_fsync == AOF_FSYNC_EVERYSEC && !force) {
         /* With this append fsync policy we do background fsyncing.
@@ -485,7 +515,11 @@ void flushAppendOnlyFile(int force) {
         server.aof_last_fsync = server.unixtime;
     } else if ((server.aof_fsync == AOF_FSYNC_EVERYSEC &&
                 server.unixtime > server.aof_last_fsync)) {
-        if (!sync_in_progress) aof_background_fsync(server.aof_fd);
+        aofStartBackgroundFsync();
+        /* Note that we update this time regardless of the fact fsync
+         * was actually started or nmot. server.aof_last_fsync is not
+         * used to really know when the latest fsync succeeded, but just
+         * as a timer to *try* to fsync once every second. */
         server.aof_last_fsync = server.unixtime;
     }
 }
@@ -1608,10 +1642,11 @@ void backgroundRewriteDoneHandler(int exitcode, int bysignal) {
             /* AOF enabled, replace the old fd with the new one. */
             oldfd = server.aof_fd;
             server.aof_fd = newfd;
-            if (server.aof_fsync == AOF_FSYNC_ALWAYS)
+            if (server.aof_fsync == AOF_FSYNC_ALWAYS) {
                 aof_fsync(newfd);
-            else if (server.aof_fsync == AOF_FSYNC_EVERYSEC)
-                aof_background_fsync(newfd);
+            } else if (server.aof_fsync == AOF_FSYNC_EVERYSEC) {
+                aofStartBackgroundFsync();
+            }
             server.aof_selected_db = -1; /* Make sure SELECT is re-issued */
             aofUpdateCurrentSize();
             server.aof_rewrite_base_size = server.aof_current_size;

src/blocked.c

@@ -139,6 +139,8 @@ void unblockClient(client *c) {
         unblockClientWaitingData(c);
     } else if (c->btype == BLOCKED_WAIT) {
         unblockClientWaitingReplicas(c);
+    } else if (c->btype == BLOCKED_AOF) {
+        unblockClientWaitingReplicas(c);
     } else if (c->btype == BLOCKED_MODULE) {
         unblockClientFromModule(c);
     } else {
@@ -166,6 +168,8 @@ void replyToBlockedClientTimedOut(client *c) {
         addReply(c,shared.nullmultibulk);
     } else if (c->btype == BLOCKED_WAIT) {
         addReplyLongLong(c,replicationCountAcksByOffset(c->bpop.reploffset));
+    } else if (c->btype == BLOCKED_AOF) {
+        addReply(c,shared.czero);
     } else if (c->btype == BLOCKED_MODULE) {
         moduleBlockedClientTimedOut(c);
     } else {

src/networking.c

@@ -138,6 +138,7 @@ client *createClient(int fd) {
     c->bpop.target = NULL;
     c->bpop.xread_group = NULL;
     c->bpop.numreplicas = 0;
+    c->bpop.aofepoch = 0;
     c->bpop.reploffset = 0;
     c->woff = 0;
     c->watched_keys = listCreate();

src/replication.c

@@ -2394,15 +2394,26 @@ void waitCommand(client *c) {
     mstime_t timeout;
     long numreplicas, ackreplicas;
     long long offset = c->woff;
+    int waitaof = 0; /* True if the user requested to wait for AOF sync. */
 
     if (server.masterhost) {
         addReplyError(c,"WAIT cannot be used with slave instances. Please also note that since Redis 4.0 if a slave is configured to be writable (which is not the default) writes to slaves are just local and are not propagated.");
         return;
     }
 
-    /* Argument parsing. */
-    if (getLongFromObjectOrReply(c,c->argv[1],&numreplicas,NULL) != C_OK)
-        return;
+    /* AOF or number of replicas argument parsing. */
+    if (!strcasecmp(c->argv[1]->ptr,"AOF")) {
+        waitaof = 1;
+        if (server.aof_state != AOF_ON) {
+            addReplyError(c,"WAIT AOF is only allowed when AOF is enabled");
+            return;
+        }
+    } else {
+        if (getLongFromObjectOrReply(c,c->argv[1],&numreplicas,NULL) != C_OK)
+            return;
+    }
+
+    /* Timeout parsing. */
     if (getTimeoutFromObjectOrReply(c,c->argv[2],&timeout,UNIT_MILLISECONDS)
         != C_OK) return;
 
@@ -2415,11 +2426,17 @@ void waitCommand(client *c) {
 
     /* Otherwise block the client and put it into our list of clients
      * waiting for ack from slaves. */
-    c->bpop.timeout = timeout;
-    c->bpop.reploffset = offset;
-    c->bpop.numreplicas = numreplicas;
-    listAddNodeTail(server.clients_waiting_acks,c);
-    blockClient(c,BLOCKED_WAIT);
+    if (waitaof) {
+        c->bpop.aofepoch = aofNextEpoch();
+        listAddNodeTail(server.clients_waiting_acks,c);
+        blockClient(c,BLOCKED_AOF);
+    } else {
+        c->bpop.timeout = timeout;
+        c->bpop.reploffset = offset;
+        c->bpop.numreplicas = numreplicas;
+        listAddNodeTail(server.clients_waiting_acks,c);
+        blockClient(c,BLOCKED_WAIT);
+    }
 
     /* Make sure that the server will send an ACK request to all the slaves
      * before returning to the event loop. */
@@ -2438,7 +2455,7 @@ void unblockClientWaitingReplicas(client *c) {
 
 /* Check if there are clients blocked in WAIT that can be unblocked since
  * we received enough ACKs from slaves. */
-void processClientsWaitingReplicas(void) {
+void processClientsBlockedInWait(void) {
     long long last_offset = 0;
     int last_numreplicas = 0;
 
@@ -2449,23 +2466,36 @@ void processClientsWaitingReplicas(void) {
     while((ln = listNext(&li))) {
         client *c = ln->value;
 
-        /* Every time we find a client that is satisfied for a given
-         * offset and number of replicas, we remember it so the next client
-         * may be unblocked without calling replicationCountAcksByOffset()
-         * if the requested offset / replicas were equal or less. */
-        if (last_offset && last_offset > c->bpop.reploffset &&
-                           last_numreplicas > c->bpop.numreplicas)
-        {
-            unblockClient(c);
-            addReplyLongLong(c,last_numreplicas);
-        } else {
-            int numreplicas = replicationCountAcksByOffset(c->bpop.reploffset);
-
-            if (numreplicas >= c->bpop.numreplicas) {
-                last_offset = c->bpop.reploffset;
-                last_numreplicas = numreplicas;
+        if (c->btype == BLOCKED_AOF) {
+            /* Handle WAIT AOF. */
+            if (server.aof_fsync == AOF_FSYNC_ALWAYS ||
+                c->bpop.aofepoch <= server.aof_fsync_epoch)
+            {
                 unblockClient(c);
-                addReplyLongLong(c,numreplicas);
+                addReply(c,shared.cone);
+            }
+        } else {
+            /* Handle WAIT for slaves to ACK.
+             *
+             * Every time we find a client that is satisfied for a given
+             * offset and number of replicas, we remember it so the next client
+             * may be unblocked without calling replicationCountAcksByOffset()
+             * if the requested offset / replicas were equal or less. */
+            if (last_offset && last_offset > c->bpop.reploffset &&
+                               last_numreplicas > c->bpop.numreplicas)
+            {
+                unblockClient(c);
+                addReplyLongLong(c,last_numreplicas);
+            } else {
+                int numreplicas =
+                    replicationCountAcksByOffset(c->bpop.reploffset);
+
+                if (numreplicas >= c->bpop.numreplicas) {
+                    last_offset = c->bpop.reploffset;
+                    last_numreplicas = numreplicas;
+                    unblockClient(c);
+                    addReplyLongLong(c,numreplicas);
+                }
             }
         }
     }

src/server.c

@@ -1229,9 +1229,19 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
     }
 
     /* Unblock all the clients blocked for synchronous replication
-     * in WAIT. */
-    if (listLength(server.clients_waiting_acks))
-        processClientsWaitingReplicas();
+     * or AOF sync in WAIT. */
+    if (listLength(server.clients_waiting_acks)) {
+        processClientsBlockedInWait();
+
+        /* If after this cycle we have still clients blocked waiting for
+         * AOF fsync, try to start a new sync cycle. Note that if one is
+         * already in progress, the call does nothing. */
+        if (server.blocked_clients_by_type[BLOCKED_AOF] &&
+            server.aof_state == AOF_ON)
+        {
+            aofStartBackgroundFsync();
+        }
+    }
 
     /* Check if there are clients unblocked by modules that implement
      * blocking commands. */
@@ -1418,6 +1428,8 @@ void initServerConfig(void) {
     server.aof_rewrite_incremental_fsync = CONFIG_DEFAULT_AOF_REWRITE_INCREMENTAL_FSYNC;
     server.aof_load_truncated = CONFIG_DEFAULT_AOF_LOAD_TRUNCATED;
     server.aof_use_rdb_preamble = CONFIG_DEFAULT_AOF_USE_RDB_PREAMBLE;
+    server.aof_fsync_in_progress_epoch = 0;
+    server.aof_fsync_epoch = 0;
     server.pidfile = NULL;
     server.rdb_filename = zstrdup(CONFIG_DEFAULT_RDB_FILENAME);
     server.aof_filename = zstrdup(CONFIG_DEFAULT_AOF_FILENAME);

src/server.h

@@ -258,7 +258,8 @@ typedef long long mstime_t; /* millisecond time type. */
 #define BLOCKED_WAIT 2    /* WAIT for synchronous replication. */
 #define BLOCKED_MODULE 3  /* Blocked by a loadable module. */
 #define BLOCKED_STREAM 4  /* XREAD. */
-#define BLOCKED_NUM 5     /* Number of blocked states. */
+#define BLOCKED_AOF 5     /* WAIT AOF waiting for AOF sync. */
+#define BLOCKED_NUM 6     /* Number of blocked states. */
 
 /* Client request types */
 #define PROTO_REQ_INLINE 1
@@ -660,6 +661,9 @@ typedef struct blockingState {
     int numreplicas;        /* Number of replicas we are waiting for ACK. */
     long long reploffset;   /* Replication offset to reach. */
 
+    /* BLOCKED_AOF */
+    uint64_t aofepoch;      /* WAIT AOF: the sync epoch we are waiting for. */
+
     /* BLOCKED_MODULE */
     void *module_blocked_handle; /* RedisModuleBlockedClient structure.
                                     which is opaque for the Redis core, only
@@ -1016,7 +1020,7 @@ struct redisServer {
     int aof_fd;       /* File descriptor of currently selected AOF file */
     int aof_selected_db; /* Currently selected DB in AOF */
     time_t aof_flush_postponed_start; /* UNIX time of postponed AOF flush */
-    time_t aof_last_fsync;            /* UNIX time of last fsync() */
+    time_t aof_last_fsync;            /* UNIX time of last fsync() *attempt* */
     time_t aof_rewrite_time_last;   /* Time used by last AOF rewrite run. */
     time_t aof_rewrite_time_start;  /* Current AOF rewrite start time. */
     int aof_lastbgrewrite_status;   /* C_OK or C_ERR */
@@ -1026,6 +1030,8 @@ struct redisServer {
     int aof_last_write_errno;       /* Valid if aof_last_write_status is ERR */
     int aof_load_truncated;         /* Don't stop on unexpected AOF EOF. */
     int aof_use_rdb_preamble;       /* Use RDB preamble on AOF rewrites. */
+    uint64_t aof_fsync_epoch;       /* AOF sync epoch used for WAIT AOF. */
+    uint64_t aof_fsync_in_progress_epoch; /* Current ongoing AOF sync epoch. */
     /* AOF pipes used to communicate between parent and child during rewrite. */
     int aof_pipe_write_data_to_child;
     int aof_pipe_read_data_from_parent;
@@ -1513,7 +1519,7 @@ void replicationScriptCacheInit(void);
 void replicationScriptCacheFlush(void);
 void replicationScriptCacheAdd(sds sha1);
 int replicationScriptCacheExists(sds sha1);
-void processClientsWaitingReplicas(void);
+void processClientsBlockedInWait(void);
 void unblockClientWaitingReplicas(client *c);
 int replicationCountAcksByOffset(long long offset);
 void replicationSendNewlineToMaster(void);
@@ -1548,6 +1554,8 @@ void backgroundRewriteDoneHandler(int exitcode, int bysignal);
 void aofRewriteBufferReset(void);
 unsigned long aofRewriteBufferSize(void);
 ssize_t aofReadDiffFromParent(void);
+void aofStartBackgroundFsync(void);
+uint64_t aofNextEpoch(void);
 
 /* Child info */
 void openChildInfoPipe(void);