Periodic fsync on aof rewrite and rdb snapshots to avoid high IO stalls when large page cache is available.

src/aof.c

@@ -863,7 +863,7 @@ int rewriteAppendOnlyFile(char *filename) {
         return REDIS_ERR;
     }
 
-    rioInitWithFile(&aof,fp);
+    rioInitWithFileAndFsyncInterval(&aof,fp, 1024*1024*16);
     for (j = 0; j < server.dbnum; j++) {
         char selectcmd[] = "*2\r\n$6\r\nSELECT\r\n";
         redisDb *db = server.db+j;

src/rdb.c

@@ -648,7 +648,7 @@ int rdbSave(char *filename, int dbnum) {
         return REDIS_ERR;
     }
 
-    rioInitWithFile(&rdb,fp);
+    rioInitWithFileAndFsyncInterval(&rdb,fp, 1024*1024*16);
     if (server.rdb_checksum)
         rdb.update_cksum = rioGenericUpdateChecksum;
     snprintf(magic,sizeof(magic),"REDIS%04d",REDIS_RDB_VERSION);

src/rio.c

@@ -48,6 +48,7 @@
 #include "fmacros.h"
 #include <string.h>
 #include <stdio.h>
+#include <unistd.h>
 #include "rio.h"
 #include "util.h"
 
@@ -76,7 +77,17 @@ static off_t rioBufferTell(rio *r) {
 
 /* Returns 1 or 0 for success/failure. */
 static size_t rioFileWrite(rio *r, const void *buf, size_t len) {
-    return fwrite(buf,len,1,r->io.file.fp);
+    size_t bytes_written = 0;
+    while (len) {
+        size_t bytes_to_write = (r->io.file.fsync_interval && r->io.file.fsync_interval < len) ? r->io.file.fsync_interval : len;
+        if (fwrite((char*)buf + bytes_written,bytes_to_write,1,r->io.file.fp) != 1)
+            return 0;
+        bytes_written += bytes_to_write;
+        len -= bytes_to_write;
+        if (r->io.file.fsync_interval && r->processed_bytes/r->io.file.fsync_interval < (r->processed_bytes + bytes_written)/r->io.file.fsync_interval)
+            fsync(fileno(r->io.file.fp));
+    }
+    return 1;
 }
 
 /* Returns 1 or 0 for success/failure. */
@@ -116,6 +127,12 @@ void rioInitWithFile(rio *r, FILE *fp) {
     r->io.file.fp = fp;
 }
 
+void rioInitWithFileAndFsyncInterval(rio *r, FILE *fp, size_t fsyncInterval) {
+    *r = rioFileIO;
+    r->io.file.fp = fp;
+    r->io.file.fsync_interval = fsyncInterval;
+}
+
 void rioInitWithBuffer(rio *r, sds s) {
     *r = rioBufferIO;
     r->io.buffer.ptr = s;

src/rio.h

@@ -66,6 +66,7 @@ struct _rio {
         } buffer;
         struct {
             FILE *fp;
+            size_t fsync_interval;
         } file;
     } io;
 };
@@ -107,6 +108,7 @@ static inline off_t rioTell(rio *r) {
 }
 
 void rioInitWithFile(rio *r, FILE *fp);
+void rioInitWithFileAndFsyncInterval(rio *r, FILE *fp, size_t fsyncInterval);
 void rioInitWithBuffer(rio *r, sds s);
 
 size_t rioWriteBulkCount(rio *r, char prefix, int count);