From: Pieter Noordhuis <pcnoordhuis@gmail.com>
Date: Fri, 13 May 2011 15:31:00 +0000 (+0200)
Subject: Abstract file/buffer I/O to support in-memory serialization
X-Git-Url: https://git.saurik.com/redis.git/commitdiff_plain/2e4b0e7727743cf03d25da0f535ecc02aad82d1f

Abstract file/buffer I/O to support in-memory serialization
---

diff --git a/src/Makefile b/src/Makefile
index e08c6987..bce8b6e5 100644
--- a/src/Makefile
+++ b/src/Makefile
@@ -47,7 +47,7 @@ BINCOLOR="\033[37;1m"
 MAKECOLOR="\033[32;1m"
 ENDCOLOR="\033[0m"
 
-OBJ = adlist.o ae.o anet.o dict.o redis.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o dscache.o pubsub.o multi.o debug.o sort.o intset.o syncio.o diskstore.o cluster.o crc16.o endian.o
+OBJ = adlist.o ae.o anet.o dict.o redis.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o dscache.o pubsub.o multi.o debug.o sort.o intset.o syncio.o diskstore.o cluster.o crc16.o endian.o rio.o
 BENCHOBJ = ae.o anet.o redis-benchmark.o sds.o adlist.o zmalloc.o
 CLIOBJ = anet.o sds.o adlist.o redis-cli.o zmalloc.o release.o
 CHECKDUMPOBJ = redis-check-dump.o lzf_c.o lzf_d.o
@@ -109,6 +109,7 @@ redis.o: redis.c redis.h fmacros.h config.h ae.h sds.h dict.h adlist.h \
 release.o: release.c release.h
 replication.o: replication.c redis.h fmacros.h config.h ae.h sds.h dict.h \
   adlist.h zmalloc.h anet.h zipmap.h ziplist.h intset.h version.h
+rio.o: rio.c sds.h
 sds.o: sds.c sds.h zmalloc.h
 sha1.o: sha1.c sha1.h
 sort.o: sort.c redis.h fmacros.h config.h ae.h sds.h dict.h adlist.h \
diff --git a/src/cluster.c b/src/cluster.c
index 3e65af78..1ae84e1e 100644
--- a/src/cluster.c
+++ b/src/cluster.c
@@ -1383,11 +1383,10 @@ void clusterCommand(redisClient *c) {
 
 /* RESTORE key ttl serialized-value */
 void restoreCommand(redisClient *c) {
-    FILE *fp;
-    char buf[64];
     robj *o;
-    unsigned char *data;
     long ttl;
+    rio payload;
+    unsigned char *data;
 
     /* Make sure this key does not already exist here... */
     if (dbExists(c->db,c->argv[1])) {
@@ -1403,41 +1402,20 @@ void restoreCommand(redisClient *c) {
         return;
     }
 
-    /* rdbLoadObject() only works against file descriptors so we need to
-     * dump the serialized object into a file and reload. */
-    snprintf(buf,sizeof(buf),"redis-restore-%d.tmp",getpid());
-    fp = fopen(buf,"w+");
-    if (!fp) {
-        redisLog(REDIS_WARNING,"Can't open tmp file for RESTORE: %s",
-            strerror(errno));
-        addReplyErrorFormat(c,"RESTORE failed, tmp file creation error: %s",
-            strerror(errno));
-        return;
-    }
-    unlink(buf);
+    /* Temporary hack to get RDB-aligned payload. */
+    payload = rioInitWithBuffer(sdsnewlen(c->argv[3]->ptr+1, sdslen(c->argv[3]->ptr)-1));
+    data = c->argv[3]->ptr;
 
-    /* Write the actual data and rewind the file */
-    data = (unsigned char*) c->argv[3]->ptr;
-    if (fwrite(data+1,sdslen((sds)data)-1,1,fp) != 1) {
-        redisLog(REDIS_WARNING,"Can't write against tmp file for RESTORE: %s",
-            strerror(errno));
-        addReplyError(c,"RESTORE failed, tmp file I/O error.");
-        fclose(fp);
-        return;
-    }
-    rewind(fp);
-
-    /* Finally create the object from the serialized dump and
-     * store it at the specified key. */
+    /* Create the object from the serialized dump. */
     if ((data[0] > 4 && data[0] < 9) ||
          data[0] > 11 ||
-        (o = rdbLoadObject(data[0],fp)) == NULL)
+        (o = rdbLoadObject(data[0],&payload)) == NULL)
     {
         addReplyError(c,"Bad data format.");
-        fclose(fp);
+        sdsfree(payload.io.buffer.ptr);
         return;
     }
-    fclose(fp);
+    sdsfree(payload.io.buffer.ptr);
 
     /* Create the key and set the TTL if any */
     dbAdd(c->db,c->argv[1],o);
@@ -1450,12 +1428,10 @@ void migrateCommand(redisClient *c) {
     int fd;
     long timeout;
     long dbid;
-    char buf[64];
-    FILE *fp;
     time_t ttl;
     robj *o;
     unsigned char type;
-    off_t payload_len;
+    rio cmd, payload;
 
     /* Sanity check */
     if (getLongFromObjectOrReply(c,c->argv[5],&timeout,NULL) != REDIS_OK)
@@ -1485,54 +1461,51 @@ void migrateCommand(redisClient *c) {
         return;
     }
 
-    /* Create temp file */
-    snprintf(buf,sizeof(buf),"redis-migrate-%d.tmp",getpid());
-    fp = fopen(buf,"w+");
-    if (!fp) {
-        redisLog(REDIS_WARNING,"Can't open tmp file for MIGRATE: %s",
-            strerror(errno));
-        addReplyErrorFormat(c,"MIGRATE failed, tmp file creation error: %s.",
-            strerror(errno));
-        return;
-    }
-    unlink(buf);
-
-    /* Build the SELECT + RESTORE query writing it in our temp file. */
-    if (fwriteBulkCount(fp,'*',2) == 0) goto file_wr_err;
-    if (fwriteBulkString(fp,"SELECT",6) == 0) goto file_wr_err;
-    if (fwriteBulkLongLong(fp,dbid) == 0) goto file_wr_err;
+    cmd = rioInitWithBuffer(sdsempty());
+    redisAssert(rioWriteBulkCount(&cmd,'*',2));
+    redisAssert(rioWriteBulkString(&cmd,"SELECT",6));
+    redisAssert(rioWriteBulkLongLong(&cmd,dbid));
 
     ttl = getExpire(c->db,c->argv[3]);
     type = o->type;
-    if (fwriteBulkCount(fp,'*',4) == 0) goto file_wr_err;
-    if (fwriteBulkString(fp,"RESTORE",7) == 0) goto file_wr_err;
-    if (fwriteBulkObject(fp,c->argv[3]) == 0) goto file_wr_err;
-    if (fwriteBulkLongLong(fp, (ttl == -1) ? 0 : ttl) == 0) goto file_wr_err;
+    if (type == REDIS_LIST && o->encoding == REDIS_ENCODING_ZIPLIST)
+        type = REDIS_LIST_ZIPLIST;
+    else if (type == REDIS_HASH && o->encoding == REDIS_ENCODING_ZIPMAP)
+        type = REDIS_HASH_ZIPMAP;
+    else if (type == REDIS_SET && o->encoding == REDIS_ENCODING_INTSET)
+        type = REDIS_SET_INTSET;
+    else
+        type = o->type;
+
+    redisAssert(rioWriteBulkCount(&cmd,'*',4));
+    redisAssert(rioWriteBulkString(&cmd,"RESTORE",7));
+    redisAssert(c->argv[3]->encoding == REDIS_ENCODING_RAW);
+    redisAssert(rioWriteBulkString(&cmd,c->argv[3]->ptr,sdslen(c->argv[3]->ptr)));
+    redisAssert(rioWriteBulkLongLong(&cmd,(ttl == -1) ? 0 : ttl));
 
     /* Finally the last argument that is the serailized object payload
-     * in the form: <type><rdb-serailized-object>. */
-    payload_len = rdbSavedObjectLen(o);
-    if (fwriteBulkCount(fp,'$',payload_len+1) == 0) goto file_wr_err;
-    if (fwrite(&type,1,1,fp) == 0) goto file_wr_err;
-    if (rdbSaveObject(fp,o) == -1) goto file_wr_err;
-    if (fwrite("\r\n",2,1,fp) == 0) goto file_wr_err;
-
-    /* Tranfer the query to the other node */
-    rewind(fp);
+     * in the form: <type><rdb-serialized-object>. */
+    payload = rioInitWithBuffer(sdsempty());
+    redisAssert(rioWrite(&payload,&type,1));
+    redisAssert(rdbSaveObject(&payload,o) != -1);
+    redisAssert(rioWriteBulkString(&cmd,payload.io.buffer.ptr,sdslen(payload.io.buffer.ptr)));
+    sdsfree(payload.io.buffer.ptr);
+
+    /* Tranfer the query to the other node in 64K chunks. */
     {
-        char buf[4096];
-        size_t nread;
-
-        while ((nread = fread(buf,1,sizeof(buf),fp)) != 0) {
-            int nwritten;
-
-            nwritten = syncWrite(fd,buf,nread,timeout);
-            if (nwritten != (signed)nread) goto socket_wr_err;
+        sds buf = cmd.io.buffer.ptr;
+        size_t pos = 0, towrite;
+        int nwritten = 0;
+
+        while ((towrite = sdslen(buf)-pos) > 0) {
+            towrite = (towrite > (64*1024) ? (64*1024) : towrite);
+            nwritten = syncWrite(fd,buf+nwritten,towrite,timeout);
+            if (nwritten != (signed)towrite) goto socket_wr_err;
+            pos += nwritten;
         }
-        if (ferror(fp)) goto file_rd_err;
     }
 
-    /* Read back the reply */
+    /* Read back the reply. */
     {
         char buf1[1024];
         char buf2[1024];
@@ -1541,7 +1514,7 @@ void migrateCommand(redisClient *c) {
         if (syncReadLine(fd, buf1, sizeof(buf1), timeout) <= 0)
             goto socket_rd_err;
         if (syncReadLine(fd, buf2, sizeof(buf2), timeout) <= 0)
-                goto socket_rd_err;
+            goto socket_rd_err;
         if (buf1[0] == '-' || buf2[0] == '-') {
             addReplyErrorFormat(c,"Target instance replied with error: %s",
                 (buf1[0] == '-') ? buf1+1 : buf2+1);
@@ -1550,25 +1523,8 @@ void migrateCommand(redisClient *c) {
             addReply(c,shared.ok);
         }
     }
-    fclose(fp);
-    close(fd);
-    return;
 
-file_wr_err:
-    redisLog(REDIS_WARNING,"Can't write on tmp file for MIGRATE: %s",
-        strerror(errno));
-    addReplyErrorFormat(c,"MIGRATE failed, tmp file write error: %s.",
-        strerror(errno));
-    fclose(fp);
-    close(fd);
-    return;
-
-file_rd_err:
-    redisLog(REDIS_WARNING,"Can't read from tmp file for MIGRATE: %s",
-        strerror(errno));
-    addReplyErrorFormat(c,"MIGRATE failed, tmp file read error: %s.",
-        strerror(errno));
-    fclose(fp);
+    sdsfree(cmd.io.buffer.ptr);
     close(fd);
     return;
 
@@ -1577,7 +1533,7 @@ socket_wr_err:
         strerror(errno));
     addReplyErrorFormat(c,"MIGRATE failed, writing to target node: %s.",
         strerror(errno));
-    fclose(fp);
+    sdsfree(cmd.io.buffer.ptr);
     close(fd);
     return;
 
@@ -1586,7 +1542,7 @@ socket_rd_err:
         strerror(errno));
     addReplyErrorFormat(c,"MIGRATE failed, reading from target node: %s.",
         strerror(errno));
-    fclose(fp);
+    sdsfree(cmd.io.buffer.ptr);
     close(fd);
     return;
 }
@@ -1595,11 +1551,9 @@ socket_rd_err:
  * DUMP is actually not used by Redis Cluster but it is the obvious
  * complement of RESTORE and can be useful for different applications. */
 void dumpCommand(redisClient *c) {
-    char buf[64];
-    FILE *fp;
     robj *o, *dumpobj;
     sds dump = NULL;
-    off_t payload_len;
+    rio payload;
     unsigned int type;
 
     /* Check if the key is here. */
@@ -1608,27 +1562,15 @@ void dumpCommand(redisClient *c) {
         return;
     }
     
-    /* Create temp file */
-    snprintf(buf,sizeof(buf),"redis-dump-%d.tmp",getpid());
-    fp = fopen(buf,"w+");
-    if (!fp) {
-        redisLog(REDIS_WARNING,"Can't open tmp file for MIGRATE: %s",
-            strerror(errno));
-        addReplyErrorFormat(c,"DUMP failed, tmp file creation error: %s.",
-            strerror(errno));
-        return;
-    }
-    unlink(buf);
-
-    /* Dump the serailized object and read it back in memory.
-     * We prefix it with a one byte containing the type ID.
-     * This is the serialization format understood by RESTORE. */
-    if (rdbSaveObject(fp,o) == -1) goto file_wr_err;
-    payload_len = ftello(fp);
-    if (fseeko(fp,0,SEEK_SET) == -1) goto file_rd_err;
-    dump = sdsnewlen(NULL,payload_len+1);
-    if (payload_len && fread(dump+1,payload_len,1,fp) != 1) goto file_rd_err;
-    fclose(fp);
+    /* Dump the serailized object and read it back in memory. We prefix it
+     * with a one byte containing the type ID.  This is the serialization
+     * format understood by RESTORE. */
+    payload = rioInitWithBuffer(sdsempty());
+    redisAssert(rdbSaveObject(&payload,o)); /* always write >= 1 bytes. */
+    dump = sdsnewlen(NULL,sdslen(payload.io.buffer.ptr)+1);
+    memcpy(dump+1,payload.io.buffer.ptr,sdslen(payload.io.buffer.ptr));
+    sdsfree(payload.io.buffer.ptr);
+
     type = o->type;
     if (type == REDIS_LIST && o->encoding == REDIS_ENCODING_ZIPLIST)
         type = REDIS_LIST_ZIPLIST;
@@ -1645,24 +1587,6 @@ void dumpCommand(redisClient *c) {
     addReplyBulk(c,dumpobj);
     decrRefCount(dumpobj);
     return;
-
-file_wr_err:
-    redisLog(REDIS_WARNING,"Can't write on tmp file for DUMP: %s",
-        strerror(errno));
-    addReplyErrorFormat(c,"DUMP failed, tmp file write error: %s.",
-        strerror(errno));
-    sdsfree(dump);
-    fclose(fp);
-    return;
-
-file_rd_err:
-    redisLog(REDIS_WARNING,"Can't read from tmp file for DUMP: %s",
-        strerror(errno));
-    addReplyErrorFormat(c,"DUMP failed, tmp file read error: %s.",
-        strerror(errno));
-    sdsfree(dump);
-    fclose(fp);
-    return;
 }
 
 /* -----------------------------------------------------------------------------
diff --git a/src/diskstore.c b/src/diskstore.c
index 9e86364e..9e45305f 100644
--- a/src/diskstore.c
+++ b/src/diskstore.c
@@ -185,8 +185,9 @@ int dsKeyToPath(redisDb *db, char *buf, robj *key) {
 
 int dsSet(redisDb *db, robj *key, robj *val, time_t expire) {
     char buf[1024], buf2[1024];
-    FILE *fp;
     int retval, len;
+    FILE *fp;
+    rio rdb;
 
     len = dsKeyToPath(db,buf,key);
     memcpy(buf2,buf,len);
@@ -201,7 +202,9 @@ int dsSet(redisDb *db, robj *key, robj *val, time_t expire) {
             redisPanic("Unrecoverable diskstore error. Exiting.");
         }
     }
-    if ((retval = rdbSaveKeyValuePair(fp,key,val,expire,time(NULL))) == -1)
+
+    rdb = rioInitWithFile(fp);
+    if ((retval = rdbSaveKeyValuePair(&rdb,key,val,expire,time(NULL))) == -1)
         return REDIS_ERR;
     fclose(fp);
     if (retval == 0) {
@@ -226,6 +229,7 @@ robj *dsGet(redisDb *db, robj *key, time_t *expire) {
     robj *dskey; /* Key as loaded from disk. */
     robj *val;
     FILE *fp;
+    rio rdb;
 
     dsKeyToPath(db,buf,key);
     fp = fopen(buf,"r");
@@ -236,16 +240,17 @@ robj *dsGet(redisDb *db, robj *key, time_t *expire) {
         goto readerr;
     }
 
-    if ((type = rdbLoadType(fp)) == -1) goto readerr;
+    rdb = rioInitWithFile(fp);
+    if ((type = rdbLoadType(&rdb)) == -1) goto readerr;
     if (type == REDIS_EXPIRETIME) {
-        if ((expiretime = rdbLoadTime(fp)) == -1) goto readerr;
+        if ((expiretime = rdbLoadTime(&rdb)) == -1) goto readerr;
         /* We read the time so we need to read the object type again */
-        if ((type = rdbLoadType(fp)) == -1) goto readerr;
+        if ((type = rdbLoadType(&rdb)) == -1) goto readerr;
     }
     /* Read key */
-    if ((dskey = rdbLoadStringObject(fp)) == NULL) goto readerr;
+    if ((dskey = rdbLoadStringObject(&rdb)) == NULL) goto readerr;
     /* Read value */
-    if ((val = rdbLoadObject(type,fp)) == NULL) goto readerr;
+    if ((val = rdbLoadObject(type,&rdb)) == NULL) goto readerr;
     fclose(fp);
 
     /* The key we asked, and the key returned, must be the same */
@@ -362,6 +367,7 @@ void *dsRdbSave_thread(void *arg) {
     struct dirent *dp, de;
     int j, i, last_dbid = -1;
     FILE *fp;
+    rio rdb;
 
     /* Change state to ACTIVE, to signal there is a saving thead working. */
     redisLog(REDIS_NOTICE,"Diskstore BGSAVE thread started");
@@ -375,7 +381,9 @@ void *dsRdbSave_thread(void *arg) {
         dsRdbSaveSetState(REDIS_BGSAVE_THREAD_DONE_ERR);
         return NULL;
     }
-    if (fwrite("REDIS0001",9,1,fp) == 0) goto werr;
+
+    rdb = rioInitWithFile(fp);
+    if (rioWrite(&rdb,"REDIS0001",9) == 0) goto werr;
 
     sleep(5);
 
@@ -408,8 +416,8 @@ void *dsRdbSave_thread(void *arg) {
                 dbid = dsGetDbidFromFilename(dp->d_name);
                 if (dbid != last_dbid) {
                     last_dbid = dbid;
-                    if (rdbSaveType(fp,REDIS_SELECTDB) == -1) goto werr;
-                    if (rdbSaveLen(fp,dbid) == -1) goto werr;
+                    if (rdbSaveType(&rdb,REDIS_SELECTDB) == -1) goto werr;
+                    if (rdbSaveLen(&rdb,dbid) == -1) goto werr;
                 }
 
                 /* Let's copy this file into the target .rdb */
@@ -422,7 +430,7 @@ void *dsRdbSave_thread(void *arg) {
                     goto werr;
                 }
                 while(1) {
-                    int nread = fread(buf,1,sizeof(buf),entryfp);
+                    size_t nread = fread(buf,1,sizeof(buf),entryfp);
 
                     if (nread == 0) {
                         if (ferror(entryfp)) {
@@ -433,7 +441,7 @@ void *dsRdbSave_thread(void *arg) {
                             break;
                         }
                     }
-                    if (fwrite(buf,1,nread,fp) != (unsigned)nread) {
+                    if (rioWrite(&rdb,buf,nread) == 0) {
                         closedir(dir);
                         goto werr;
                     }
@@ -445,7 +453,7 @@ void *dsRdbSave_thread(void *arg) {
     }
     
     /* Output the end of file opcode */
-    if (rdbSaveType(fp,REDIS_EOF) == -1) goto werr;
+    if (rdbSaveType(&rdb,REDIS_EOF) == -1) goto werr;
 
     /* Make sure data will not remain on the OS's output buffers */
     fflush(fp);
diff --git a/src/rdb.c b/src/rdb.c
index eeafc053..1e1df2ad 100644
--- a/src/rdb.c
+++ b/src/rdb.c
@@ -1,6 +1,3 @@
-#include "redis.h"
-#include "lzf.h"    /* LZF compression library */
-
 #include <math.h>
 #include <sys/types.h>
 #include <sys/time.h>
@@ -8,47 +5,46 @@
 #include <sys/wait.h>
 #include <arpa/inet.h>
 #include <sys/stat.h>
+#include "rdb.h"
+#include "lzf.h" /* LZF compression library */
 
-/* Convenience wrapper around fwrite, that returns the number of bytes written
- * to the file instead of the number of objects (see fwrite(3)) and -1 in the
- * case of an error. It also supports a NULL *fp to skip writing altogether
- * instead of writing to /dev/null. */
-static int rdbWriteRaw(FILE *fp, void *p, size_t len) {
-    if (fp != NULL && fwrite(p,len,1,fp) == 0) return -1;
-    return len;
+static int rdbWriteRaw(rio *rdb, void *p, size_t len) {
+    if (rioWrite(rdb,p,len) == 0)
+        return -1;
+    return 1;
 }
 
-int rdbSaveType(FILE *fp, unsigned char type) {
-    return rdbWriteRaw(fp,&type,1);
+int rdbSaveType(rio *rdb, unsigned char type) {
+    return rdbWriteRaw(rdb,&type,1);
 }
 
-int rdbSaveTime(FILE *fp, time_t t) {
+int rdbSaveTime(rio *rdb, time_t t) {
     int32_t t32 = (int32_t) t;
-    return rdbWriteRaw(fp,&t32,4);
+    return rdbWriteRaw(rdb,&t32,4);
 }
 
 /* check rdbLoadLen() comments for more info */
-int rdbSaveLen(FILE *fp, uint32_t len) {
+int rdbSaveLen(rio *rdb, uint32_t len) {
     unsigned char buf[2];
-    int nwritten;
+    size_t nwritten;
 
     if (len < (1<<6)) {
         /* Save a 6 bit len */
         buf[0] = (len&0xFF)|(REDIS_RDB_6BITLEN<<6);
-        if (rdbWriteRaw(fp,buf,1) == -1) return -1;
+        if (rdbWriteRaw(rdb,buf,1) == -1) return -1;
         nwritten = 1;
     } else if (len < (1<<14)) {
         /* Save a 14 bit len */
         buf[0] = ((len>>8)&0xFF)|(REDIS_RDB_14BITLEN<<6);
         buf[1] = len&0xFF;
-        if (rdbWriteRaw(fp,buf,2) == -1) return -1;
+        if (rdbWriteRaw(rdb,buf,2) == -1) return -1;
         nwritten = 2;
     } else {
         /* Save a 32 bit len */
         buf[0] = (REDIS_RDB_32BITLEN<<6);
-        if (rdbWriteRaw(fp,buf,1) == -1) return -1;
+        if (rdbWriteRaw(rdb,buf,1) == -1) return -1;
         len = htonl(len);
-        if (rdbWriteRaw(fp,&len,4) == -1) return -1;
+        if (rdbWriteRaw(rdb,&len,4) == -4) return -1;
         nwritten = 1+4;
     }
     return nwritten;
@@ -101,7 +97,7 @@ int rdbTryIntegerEncoding(char *s, size_t len, unsigned char *enc) {
     return rdbEncodeInteger(value,enc);
 }
 
-int rdbSaveLzfStringObject(FILE *fp, unsigned char *s, size_t len) {
+int rdbSaveLzfStringObject(rio *rdb, unsigned char *s, size_t len) {
     size_t comprlen, outlen;
     unsigned char byte;
     int n, nwritten = 0;
@@ -118,16 +114,16 @@ int rdbSaveLzfStringObject(FILE *fp, unsigned char *s, size_t len) {
     }
     /* Data compressed! Let's save it on disk */
     byte = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_LZF;
-    if ((n = rdbWriteRaw(fp,&byte,1)) == -1) goto writeerr;
+    if ((n = rdbWriteRaw(rdb,&byte,1)) == -1) goto writeerr;
     nwritten += n;
 
-    if ((n = rdbSaveLen(fp,comprlen)) == -1) goto writeerr;
+    if ((n = rdbSaveLen(rdb,comprlen)) == -1) goto writeerr;
     nwritten += n;
 
-    if ((n = rdbSaveLen(fp,len)) == -1) goto writeerr;
+    if ((n = rdbSaveLen(rdb,len)) == -1) goto writeerr;
     nwritten += n;
 
-    if ((n = rdbWriteRaw(fp,out,comprlen)) == -1) goto writeerr;
+    if ((n = rdbWriteRaw(rdb,out,comprlen)) == -1) goto writeerr;
     nwritten += n;
 
     zfree(out);
@@ -140,7 +136,7 @@ writeerr:
 
 /* Save a string objet as [len][data] on disk. If the object is a string
  * representation of an integer value we try to save it in a special form */
-int rdbSaveRawString(FILE *fp, unsigned char *s, size_t len) {
+int rdbSaveRawString(rio *rdb, unsigned char *s, size_t len) {
     int enclen;
     int n, nwritten = 0;
 
@@ -148,7 +144,7 @@ int rdbSaveRawString(FILE *fp, unsigned char *s, size_t len) {
     if (len <= 11) {
         unsigned char buf[5];
         if ((enclen = rdbTryIntegerEncoding((char*)s,len,buf)) > 0) {
-            if (rdbWriteRaw(fp,buf,enclen) == -1) return -1;
+            if (rdbWriteRaw(rdb,buf,enclen) == -1) return -1;
             return enclen;
         }
     }
@@ -156,50 +152,50 @@ int rdbSaveRawString(FILE *fp, unsigned char *s, size_t len) {
     /* Try LZF compression - under 20 bytes it's unable to compress even
      * aaaaaaaaaaaaaaaaaa so skip it */
     if (server.rdbcompression && len > 20) {
-        n = rdbSaveLzfStringObject(fp,s,len);
+        n = rdbSaveLzfStringObject(rdb,s,len);
         if (n == -1) return -1;
         if (n > 0) return n;
         /* Return value of 0 means data can't be compressed, save the old way */
     }
 
     /* Store verbatim */
-    if ((n = rdbSaveLen(fp,len)) == -1) return -1;
+    if ((n = rdbSaveLen(rdb,len)) == -1) return -1;
     nwritten += n;
     if (len > 0) {
-        if (rdbWriteRaw(fp,s,len) == -1) return -1;
+        if (rdbWriteRaw(rdb,s,len) == -1) return -1;
         nwritten += len;
     }
     return nwritten;
 }
 
 /* Save a long long value as either an encoded string or a string. */
-int rdbSaveLongLongAsStringObject(FILE *fp, long long value) {
+int rdbSaveLongLongAsStringObject(rio *rdb, long long value) {
     unsigned char buf[32];
     int n, nwritten = 0;
     int enclen = rdbEncodeInteger(value,buf);
     if (enclen > 0) {
-        return rdbWriteRaw(fp,buf,enclen);
+        return rdbWriteRaw(rdb,buf,enclen);
     } else {
         /* Encode as string */
         enclen = ll2string((char*)buf,32,value);
         redisAssert(enclen < 32);
-        if ((n = rdbSaveLen(fp,enclen)) == -1) return -1;
+        if ((n = rdbSaveLen(rdb,enclen)) == -1) return -1;
         nwritten += n;
-        if ((n = rdbWriteRaw(fp,buf,enclen)) == -1) return -1;
+        if ((n = rdbWriteRaw(rdb,buf,enclen)) == -1) return -1;
         nwritten += n;
     }
     return nwritten;
 }
 
 /* Like rdbSaveStringObjectRaw() but handle encoded objects */
-int rdbSaveStringObject(FILE *fp, robj *obj) {
+int rdbSaveStringObject(rio *rdb, robj *obj) {
     /* Avoid to decode the object, then encode it again, if the
      * object is alrady integer encoded. */
     if (obj->encoding == REDIS_ENCODING_INT) {
-        return rdbSaveLongLongAsStringObject(fp,(long)obj->ptr);
+        return rdbSaveLongLongAsStringObject(rdb,(long)obj->ptr);
     } else {
         redisAssert(obj->encoding == REDIS_ENCODING_RAW);
-        return rdbSaveRawString(fp,obj->ptr,sdslen(obj->ptr));
+        return rdbSaveRawString(rdb,obj->ptr,sdslen(obj->ptr));
     }
 }
 
@@ -211,7 +207,7 @@ int rdbSaveStringObject(FILE *fp, robj *obj) {
  * 254: + inf
  * 255: - inf
  */
-int rdbSaveDoubleValue(FILE *fp, double val) {
+int rdbSaveDoubleValue(rio *rdb, double val) {
     unsigned char buf[128];
     int len;
 
@@ -242,36 +238,36 @@ int rdbSaveDoubleValue(FILE *fp, double val) {
         buf[0] = strlen((char*)buf+1);
         len = buf[0]+1;
     }
-    return rdbWriteRaw(fp,buf,len);
+    return rdbWriteRaw(rdb,buf,len);
 }
 
 /* Save a Redis object. Returns -1 on error, 0 on success. */
-int rdbSaveObject(FILE *fp, robj *o) {
+int rdbSaveObject(rio *rdb, robj *o) {
     int n, nwritten = 0;
 
     if (o->type == REDIS_STRING) {
         /* Save a string value */
-        if ((n = rdbSaveStringObject(fp,o)) == -1) return -1;
+        if ((n = rdbSaveStringObject(rdb,o)) == -1) return -1;
         nwritten += n;
     } else if (o->type == REDIS_LIST) {
         /* Save a list value */
         if (o->encoding == REDIS_ENCODING_ZIPLIST) {
             size_t l = ziplistBlobLen((unsigned char*)o->ptr);
 
-            if ((n = rdbSaveRawString(fp,o->ptr,l)) == -1) return -1;
+            if ((n = rdbSaveRawString(rdb,o->ptr,l)) == -1) return -1;
             nwritten += n;
         } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) {
             list *list = o->ptr;
             listIter li;
             listNode *ln;
 
-            if ((n = rdbSaveLen(fp,listLength(list))) == -1) return -1;
+            if ((n = rdbSaveLen(rdb,listLength(list))) == -1) return -1;
             nwritten += n;
 
             listRewind(list,&li);
             while((ln = listNext(&li))) {
                 robj *eleobj = listNodeValue(ln);
-                if ((n = rdbSaveStringObject(fp,eleobj)) == -1) return -1;
+                if ((n = rdbSaveStringObject(rdb,eleobj)) == -1) return -1;
                 nwritten += n;
             }
         } else {
@@ -284,19 +280,19 @@ int rdbSaveObject(FILE *fp, robj *o) {
             dictIterator *di = dictGetIterator(set);
             dictEntry *de;
 
-            if ((n = rdbSaveLen(fp,dictSize(set))) == -1) return -1;
+            if ((n = rdbSaveLen(rdb,dictSize(set))) == -1) return -1;
             nwritten += n;
 
             while((de = dictNext(di)) != NULL) {
                 robj *eleobj = dictGetEntryKey(de);
-                if ((n = rdbSaveStringObject(fp,eleobj)) == -1) return -1;
+                if ((n = rdbSaveStringObject(rdb,eleobj)) == -1) return -1;
                 nwritten += n;
             }
             dictReleaseIterator(di);
         } else if (o->encoding == REDIS_ENCODING_INTSET) {
             size_t l = intsetBlobLen((intset*)o->ptr);
 
-            if ((n = rdbSaveRawString(fp,o->ptr,l)) == -1) return -1;
+            if ((n = rdbSaveRawString(rdb,o->ptr,l)) == -1) return -1;
             nwritten += n;
         } else {
             redisPanic("Unknown set encoding");
@@ -306,23 +302,23 @@ int rdbSaveObject(FILE *fp, robj *o) {
         if (o->encoding == REDIS_ENCODING_ZIPLIST) {
             size_t l = ziplistBlobLen((unsigned char*)o->ptr);
 
-            if ((n = rdbSaveRawString(fp,o->ptr,l)) == -1) return -1;
+            if ((n = rdbSaveRawString(rdb,o->ptr,l)) == -1) return -1;
             nwritten += n;
         } else if (o->encoding == REDIS_ENCODING_SKIPLIST) {
             zset *zs = o->ptr;
             dictIterator *di = dictGetIterator(zs->dict);
             dictEntry *de;
 
-            if ((n = rdbSaveLen(fp,dictSize(zs->dict))) == -1) return -1;
+            if ((n = rdbSaveLen(rdb,dictSize(zs->dict))) == -1) return -1;
             nwritten += n;
 
             while((de = dictNext(di)) != NULL) {
                 robj *eleobj = dictGetEntryKey(de);
                 double *score = dictGetEntryVal(de);
 
-                if ((n = rdbSaveStringObject(fp,eleobj)) == -1) return -1;
+                if ((n = rdbSaveStringObject(rdb,eleobj)) == -1) return -1;
                 nwritten += n;
-                if ((n = rdbSaveDoubleValue(fp,*score)) == -1) return -1;
+                if ((n = rdbSaveDoubleValue(rdb,*score)) == -1) return -1;
                 nwritten += n;
             }
             dictReleaseIterator(di);
@@ -334,22 +330,22 @@ int rdbSaveObject(FILE *fp, robj *o) {
         if (o->encoding == REDIS_ENCODING_ZIPMAP) {
             size_t l = zipmapBlobLen((unsigned char*)o->ptr);
 
-            if ((n = rdbSaveRawString(fp,o->ptr,l)) == -1) return -1;
+            if ((n = rdbSaveRawString(rdb,o->ptr,l)) == -1) return -1;
             nwritten += n;
         } else {
             dictIterator *di = dictGetIterator(o->ptr);
             dictEntry *de;
 
-            if ((n = rdbSaveLen(fp,dictSize((dict*)o->ptr))) == -1) return -1;
+            if ((n = rdbSaveLen(rdb,dictSize((dict*)o->ptr))) == -1) return -1;
             nwritten += n;
 
             while((de = dictNext(di)) != NULL) {
                 robj *key = dictGetEntryKey(de);
                 robj *val = dictGetEntryVal(de);
 
-                if ((n = rdbSaveStringObject(fp,key)) == -1) return -1;
+                if ((n = rdbSaveStringObject(rdb,key)) == -1) return -1;
                 nwritten += n;
-                if ((n = rdbSaveStringObject(fp,val)) == -1) return -1;
+                if ((n = rdbSaveStringObject(rdb,val)) == -1) return -1;
                 nwritten += n;
             }
             dictReleaseIterator(di);
@@ -374,7 +370,7 @@ off_t rdbSavedObjectLen(robj *o) {
  * On error -1 is returned.
  * On success if the key was actaully saved 1 is returned, otherwise 0
  * is returned (the key was already expired). */
-int rdbSaveKeyValuePair(FILE *fp, robj *key, robj *val,
+int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val,
                         time_t expiretime, time_t now)
 {
     int vtype;
@@ -383,8 +379,8 @@ int rdbSaveKeyValuePair(FILE *fp, robj *key, robj *val,
     if (expiretime != -1) {
         /* If this key is already expired skip it */
         if (expiretime < now) return 0;
-        if (rdbSaveType(fp,REDIS_EXPIRETIME) == -1) return -1;
-        if (rdbSaveTime(fp,expiretime) == -1) return -1;
+        if (rdbSaveType(rdb,REDIS_EXPIRETIME) == -1) return -1;
+        if (rdbSaveTime(rdb,expiretime) == -1) return -1;
     }
     /* Fix the object type if needed, to support saving zipmaps, ziplists,
      * and intsets, directly as blobs of bytes: they are already serialized. */
@@ -398,9 +394,9 @@ int rdbSaveKeyValuePair(FILE *fp, robj *key, robj *val,
     else if (vtype == REDIS_ZSET && val->encoding == REDIS_ENCODING_ZIPLIST)
         vtype = REDIS_ZSET_ZIPLIST;
     /* Save type, key, value */
-    if (rdbSaveType(fp,vtype) == -1) return -1;
-    if (rdbSaveStringObject(fp,key) == -1) return -1;
-    if (rdbSaveObject(fp,val) == -1) return -1;
+    if (rdbSaveType(rdb,vtype) == -1) return -1;
+    if (rdbSaveStringObject(rdb,key) == -1) return -1;
+    if (rdbSaveObject(rdb,val) == -1) return -1;
     return 1;
 }
 
@@ -408,10 +404,11 @@ int rdbSaveKeyValuePair(FILE *fp, robj *key, robj *val,
 int rdbSave(char *filename) {
     dictIterator *di = NULL;
     dictEntry *de;
-    FILE *fp;
     char tmpfile[256];
     int j;
     time_t now = time(NULL);
+    FILE *fp;
+    rio rdb;
 
     if (server.ds_enabled) {
         cacheForcePointInTime();
@@ -425,7 +422,10 @@ int rdbSave(char *filename) {
             strerror(errno));
         return REDIS_ERR;
     }
-    if (fwrite("REDIS0002",9,1,fp) == 0) goto werr;
+
+    rdb = rioInitWithFile(fp);
+    if (rdbWriteRaw(&rdb,"REDIS0002",9) == -1) goto werr;
+
     for (j = 0; j < server.dbnum; j++) {
         redisDb *db = server.db+j;
         dict *d = db->dict;
@@ -437,8 +437,8 @@ int rdbSave(char *filename) {
         }
 
         /* Write the SELECT DB opcode */
-        if (rdbSaveType(fp,REDIS_SELECTDB) == -1) goto werr;
-        if (rdbSaveLen(fp,j) == -1) goto werr;
+        if (rdbSaveType(&rdb,REDIS_SELECTDB) == -1) goto werr;
+        if (rdbSaveLen(&rdb,j) == -1) goto werr;
 
         /* Iterate this DB writing every entry */
         while((de = dictNext(di)) != NULL) {
@@ -448,12 +448,12 @@ int rdbSave(char *filename) {
             
             initStaticStringObject(key,keystr);
             expire = getExpire(db,&key);
-            if (rdbSaveKeyValuePair(fp,&key,o,expire,now) == -1) goto werr;
+            if (rdbSaveKeyValuePair(&rdb,&key,o,expire,now) == -1) goto werr;
         }
         dictReleaseIterator(di);
     }
     /* EOF opcode */
-    if (rdbSaveType(fp,REDIS_EOF) == -1) goto werr;
+    if (rdbSaveType(&rdb,REDIS_EOF) == -1) goto werr;
 
     /* Make sure data will not remain on the OS's output buffers */
     fflush(fp);
@@ -523,15 +523,15 @@ void rdbRemoveTempFile(pid_t childpid) {
     unlink(tmpfile);
 }
 
-int rdbLoadType(FILE *fp) {
+int rdbLoadType(rio *rdb) {
     unsigned char type;
-    if (fread(&type,1,1,fp) == 0) return -1;
+    if (rioRead(rdb,&type,1) == 0) return -1;
     return type;
 }
 
-time_t rdbLoadTime(FILE *fp) {
+time_t rdbLoadTime(rio *rdb) {
     int32_t t32;
-    if (fread(&t32,4,1,fp) == 0) return -1;
+    if (rioRead(rdb,&t32,4) == 0) return -1;
     return (time_t) t32;
 }
 
@@ -540,13 +540,13 @@ time_t rdbLoadTime(FILE *fp) {
  *
  * isencoded is set to 1 if the readed length is not actually a length but
  * an "encoding type", check the above comments for more info */
-uint32_t rdbLoadLen(FILE *fp, int *isencoded) {
+uint32_t rdbLoadLen(rio *rdb, int *isencoded) {
     unsigned char buf[2];
     uint32_t len;
     int type;
 
     if (isencoded) *isencoded = 0;
-    if (fread(buf,1,1,fp) == 0) return REDIS_RDB_LENERR;
+    if (rioRead(rdb,buf,1) == 0) return REDIS_RDB_LENERR;
     type = (buf[0]&0xC0)>>6;
     if (type == REDIS_RDB_6BITLEN) {
         /* Read a 6 bit len */
@@ -557,11 +557,11 @@ uint32_t rdbLoadLen(FILE *fp, int *isencoded) {
         return buf[0]&0x3F;
     } else if (type == REDIS_RDB_14BITLEN) {
         /* Read a 14 bit len */
-        if (fread(buf+1,1,1,fp) == 0) return REDIS_RDB_LENERR;
+        if (rioRead(rdb,buf+1,1) == 0) return REDIS_RDB_LENERR;
         return ((buf[0]&0x3F)<<8)|buf[1];
     } else {
         /* Read a 32 bit len */
-        if (fread(&len,4,1,fp) == 0) return REDIS_RDB_LENERR;
+        if (rioRead(rdb,&len,4) == 0) return REDIS_RDB_LENERR;
         return ntohl(len);
     }
 }
@@ -570,21 +570,21 @@ uint32_t rdbLoadLen(FILE *fp, int *isencoded) {
  * encoding type 'enctype'. If encode is true the function may return
  * an integer-encoded object as reply, otherwise the returned object
  * will always be encoded as a raw string. */
-robj *rdbLoadIntegerObject(FILE *fp, int enctype, int encode) {
+robj *rdbLoadIntegerObject(rio *rdb, int enctype, int encode) {
     unsigned char enc[4];
     long long val;
 
     if (enctype == REDIS_RDB_ENC_INT8) {
-        if (fread(enc,1,1,fp) == 0) return NULL;
+        if (rioRead(rdb,enc,1) == 0) return NULL;
         val = (signed char)enc[0];
     } else if (enctype == REDIS_RDB_ENC_INT16) {
         uint16_t v;
-        if (fread(enc,2,1,fp) == 0) return NULL;
+        if (rioRead(rdb,enc,2) == 0) return NULL;
         v = enc[0]|(enc[1]<<8);
         val = (int16_t)v;
     } else if (enctype == REDIS_RDB_ENC_INT32) {
         uint32_t v;
-        if (fread(enc,4,1,fp) == 0) return NULL;
+        if (rioRead(rdb,enc,4) == 0) return NULL;
         v = enc[0]|(enc[1]<<8)|(enc[2]<<16)|(enc[3]<<24);
         val = (int32_t)v;
     } else {
@@ -597,16 +597,16 @@ robj *rdbLoadIntegerObject(FILE *fp, int enctype, int encode) {
         return createObject(REDIS_STRING,sdsfromlonglong(val));
 }
 
-robj *rdbLoadLzfStringObject(FILE*fp) {
+robj *rdbLoadLzfStringObject(rio *rdb) {
     unsigned int len, clen;
     unsigned char *c = NULL;
     sds val = NULL;
 
-    if ((clen = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR) return NULL;
-    if ((len = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR) return NULL;
+    if ((clen = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL;
+    if ((len = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL;
     if ((c = zmalloc(clen)) == NULL) goto err;
     if ((val = sdsnewlen(NULL,len)) == NULL) goto err;
-    if (fread(c,clen,1,fp) == 0) goto err;
+    if (rioRead(rdb,c,clen) == 0) goto err;
     if (lzf_decompress(c,clen,val,len) == 0) goto err;
     zfree(c);
     return createObject(REDIS_STRING,val);
@@ -616,20 +616,20 @@ err:
     return NULL;
 }
 
-robj *rdbGenericLoadStringObject(FILE*fp, int encode) {
+robj *rdbGenericLoadStringObject(rio *rdb, int encode) {
     int isencoded;
     uint32_t len;
     sds val;
 
-    len = rdbLoadLen(fp,&isencoded);
+    len = rdbLoadLen(rdb,&isencoded);
     if (isencoded) {
         switch(len) {
         case REDIS_RDB_ENC_INT8:
         case REDIS_RDB_ENC_INT16:
         case REDIS_RDB_ENC_INT32:
-            return rdbLoadIntegerObject(fp,len,encode);
+            return rdbLoadIntegerObject(rdb,len,encode);
         case REDIS_RDB_ENC_LZF:
-            return rdbLoadLzfStringObject(fp);
+            return rdbLoadLzfStringObject(rdb);
         default:
             redisPanic("Unknown RDB encoding type");
         }
@@ -637,33 +637,33 @@ robj *rdbGenericLoadStringObject(FILE*fp, int encode) {
 
     if (len == REDIS_RDB_LENERR) return NULL;
     val = sdsnewlen(NULL,len);
-    if (len && fread(val,len,1,fp) == 0) {
+    if (len && rioRead(rdb,val,len) == 0) {
         sdsfree(val);
         return NULL;
     }
     return createObject(REDIS_STRING,val);
 }
 
-robj *rdbLoadStringObject(FILE *fp) {
-    return rdbGenericLoadStringObject(fp,0);
+robj *rdbLoadStringObject(rio *rdb) {
+    return rdbGenericLoadStringObject(rdb,0);
 }
 
-robj *rdbLoadEncodedStringObject(FILE *fp) {
-    return rdbGenericLoadStringObject(fp,1);
+robj *rdbLoadEncodedStringObject(rio *rdb) {
+    return rdbGenericLoadStringObject(rdb,1);
 }
 
 /* For information about double serialization check rdbSaveDoubleValue() */
-int rdbLoadDoubleValue(FILE *fp, double *val) {
+int rdbLoadDoubleValue(rio *rdb, double *val) {
     char buf[128];
     unsigned char len;
 
-    if (fread(&len,1,1,fp) == 0) return -1;
+    if (rioRead(rdb,&len,1) == 0) return -1;
     switch(len) {
     case 255: *val = R_NegInf; return 0;
     case 254: *val = R_PosInf; return 0;
     case 253: *val = R_Nan; return 0;
     default:
-        if (fread(buf,len,1,fp) == 0) return -1;
+        if (rioRead(rdb,buf,len) == 0) return -1;
         buf[len] = '\0';
         sscanf(buf, "%lg", val);
         return 0;
@@ -672,19 +672,19 @@ int rdbLoadDoubleValue(FILE *fp, double *val) {
 
 /* Load a Redis object of the specified type from the specified file.
  * On success a newly allocated object is returned, otherwise NULL. */
-robj *rdbLoadObject(int type, FILE *fp) {
+robj *rdbLoadObject(int type, rio *rdb) {
     robj *o, *ele, *dec;
     size_t len;
     unsigned int i;
 
-    redisLog(REDIS_DEBUG,"LOADING OBJECT %d (at %d)\n",type,ftell(fp));
+    redisLog(REDIS_DEBUG,"LOADING OBJECT %d (at %d)\n",type,rdb->tell(rdb));
     if (type == REDIS_STRING) {
         /* Read string value */
-        if ((o = rdbLoadEncodedStringObject(fp)) == NULL) return NULL;
+        if ((o = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL;
         o = tryObjectEncoding(o);
     } else if (type == REDIS_LIST) {
         /* Read list value */
-        if ((len = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR) return NULL;
+        if ((len = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL;
 
         /* Use a real list when there are too many entries */
         if (len > server.list_max_ziplist_entries) {
@@ -695,7 +695,7 @@ robj *rdbLoadObject(int type, FILE *fp) {
 
         /* Load every single element of the list */
         while(len--) {
-            if ((ele = rdbLoadEncodedStringObject(fp)) == NULL) return NULL;
+            if ((ele = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL;
 
             /* If we are using a ziplist and the value is too big, convert
              * the object to a real list. */
@@ -716,7 +716,7 @@ robj *rdbLoadObject(int type, FILE *fp) {
         }
     } else if (type == REDIS_SET) {
         /* Read list/set value */
-        if ((len = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR) return NULL;
+        if ((len = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL;
 
         /* Use a regular set when there are too many entries. */
         if (len > server.set_max_intset_entries) {
@@ -732,7 +732,7 @@ robj *rdbLoadObject(int type, FILE *fp) {
         /* Load every single element of the list/set */
         for (i = 0; i < len; i++) {
             long long llval;
-            if ((ele = rdbLoadEncodedStringObject(fp)) == NULL) return NULL;
+            if ((ele = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL;
             ele = tryObjectEncoding(ele);
 
             if (o->encoding == REDIS_ENCODING_INTSET) {
@@ -759,7 +759,7 @@ robj *rdbLoadObject(int type, FILE *fp) {
         size_t maxelelen = 0;
         zset *zs;
 
-        if ((zsetlen = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR) return NULL;
+        if ((zsetlen = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL;
         o = createZsetObject();
         zs = o->ptr;
 
@@ -769,9 +769,9 @@ robj *rdbLoadObject(int type, FILE *fp) {
             double score;
             zskiplistNode *znode;
 
-            if ((ele = rdbLoadEncodedStringObject(fp)) == NULL) return NULL;
+            if ((ele = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL;
             ele = tryObjectEncoding(ele);
-            if (rdbLoadDoubleValue(fp,&score) == -1) return NULL;
+            if (rdbLoadDoubleValue(rdb,&score) == -1) return NULL;
 
             /* Don't care about integer-encoded strings. */
             if (ele->encoding == REDIS_ENCODING_RAW &&
@@ -790,7 +790,7 @@ robj *rdbLoadObject(int type, FILE *fp) {
     } else if (type == REDIS_HASH) {
         size_t hashlen;
 
-        if ((hashlen = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR) return NULL;
+        if ((hashlen = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL;
         o = createHashObject();
         /* Too many entries? Use an hash table. */
         if (hashlen > server.hash_max_zipmap_entries)
@@ -800,8 +800,8 @@ robj *rdbLoadObject(int type, FILE *fp) {
         while(hashlen--) {
             robj *key, *val;
 
-            if ((key = rdbLoadEncodedStringObject(fp)) == NULL) return NULL;
-            if ((val = rdbLoadEncodedStringObject(fp)) == NULL) return NULL;
+            if ((key = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL;
+            if ((val = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL;
             /* If we are using a zipmap and there are too big values
              * the object is converted to real hash table encoding. */
             if (o->encoding != REDIS_ENCODING_HT &&
@@ -838,7 +838,7 @@ robj *rdbLoadObject(int type, FILE *fp) {
                type == REDIS_SET_INTSET ||
                type == REDIS_ZSET_ZIPLIST)
     {
-        robj *aux = rdbLoadStringObject(fp);
+        robj *aux = rdbLoadStringObject(rdb);
 
         if (aux == NULL) return NULL;
         o = createObject(REDIS_STRING,NULL); /* string is just placeholder */
@@ -913,13 +913,14 @@ void stopLoading(void) {
 }
 
 int rdbLoad(char *filename) {
-    FILE *fp;
     uint32_t dbid;
     int type, retval, rdbver;
     redisDb *db = server.db+0;
     char buf[1024];
     time_t expiretime, now = time(NULL);
     long loops = 0;
+    FILE *fp;
+    rio rdb;
 
     fp = fopen(filename,"r");
     if (!fp) return REDIS_ERR;
@@ -938,27 +939,28 @@ int rdbLoad(char *filename) {
     }
 
     startLoading(fp);
+    rdb = rioInitWithFile(fp);
     while(1) {
         robj *key, *val;
         expiretime = -1;
 
         /* Serve the clients from time to time */
         if (!(loops++ % 1000)) {
-            loadingProgress(ftello(fp));
+            loadingProgress(rdb.tell(&rdb));
             aeProcessEvents(server.el, AE_FILE_EVENTS|AE_DONT_WAIT);
         }
 
         /* Read type. */
-        if ((type = rdbLoadType(fp)) == -1) goto eoferr;
+        if ((type = rdbLoadType(&rdb)) == -1) goto eoferr;
         if (type == REDIS_EXPIRETIME) {
-            if ((expiretime = rdbLoadTime(fp)) == -1) goto eoferr;
+            if ((expiretime = rdbLoadTime(&rdb)) == -1) goto eoferr;
             /* We read the time so we need to read the object type again */
-            if ((type = rdbLoadType(fp)) == -1) goto eoferr;
+            if ((type = rdbLoadType(&rdb)) == -1) goto eoferr;
         }
         if (type == REDIS_EOF) break;
         /* Handle SELECT DB opcode as a special case */
         if (type == REDIS_SELECTDB) {
-            if ((dbid = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR)
+            if ((dbid = rdbLoadLen(&rdb,NULL)) == REDIS_RDB_LENERR)
                 goto eoferr;
             if (dbid >= (unsigned)server.dbnum) {
                 redisLog(REDIS_WARNING,"FATAL: Data file was created with a Redis server configured to handle more than %d databases. Exiting\n", server.dbnum);
@@ -968,9 +970,9 @@ int rdbLoad(char *filename) {
             continue;
         }
         /* Read key */
-        if ((key = rdbLoadStringObject(fp)) == NULL) goto eoferr;
+        if ((key = rdbLoadStringObject(&rdb)) == NULL) goto eoferr;
         /* Read value */
-        if ((val = rdbLoadObject(type,fp)) == NULL) goto eoferr;
+        if ((val = rdbLoadObject(type,&rdb)) == NULL) goto eoferr;
         /* Check if the key already expired */
         if (expiretime != -1 && expiretime < now) {
             decrRefCount(key);
diff --git a/src/rdb.h b/src/rdb.h
new file mode 100644
index 00000000..a4a8c096
--- /dev/null
+++ b/src/rdb.h
@@ -0,0 +1,26 @@
+#ifndef __REDIS_RDB_H
+#define __REDIS_RDB_H
+
+#include <stdio.h>
+#include "rio.h"
+
+/* TBD: include only necessary headers. */
+#include "redis.h"
+
+int rdbLoad(char *filename);
+int rdbSaveBackground(char *filename);
+void rdbRemoveTempFile(pid_t childpid);
+int rdbSave(char *filename);
+int rdbSaveObject(rio *rdb, robj *o);
+off_t rdbSavedObjectLen(robj *o);
+off_t rdbSavedObjectPages(robj *o);
+robj *rdbLoadObject(int type, rio *rdb);
+void backgroundSaveDoneHandler(int exitcode, int bysignal);
+int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val, time_t expireitme, time_t now);
+int rdbLoadType(rio *rdb);
+time_t rdbLoadTime(rio *rdb);
+robj *rdbLoadStringObject(rio *rdb);
+int rdbSaveType(rio *rdb, unsigned char type);
+int rdbSaveLen(rio *rdb, uint32_t len);
+
+#endif
diff --git a/src/redis.h b/src/redis.h
index 5934b6a6..a172bca4 100644
--- a/src/redis.h
+++ b/src/redis.h
@@ -898,21 +898,7 @@ void loadingProgress(off_t pos);
 void stopLoading(void);
 
 /* RDB persistence */
-int rdbLoad(char *filename);
-int rdbSaveBackground(char *filename);
-void rdbRemoveTempFile(pid_t childpid);
-int rdbSave(char *filename);
-int rdbSaveObject(FILE *fp, robj *o);
-off_t rdbSavedObjectLen(robj *o);
-off_t rdbSavedObjectPages(robj *o);
-robj *rdbLoadObject(int type, FILE *fp);
-void backgroundSaveDoneHandler(int exitcode, int bysignal);
-int rdbSaveKeyValuePair(FILE *fp, robj *key, robj *val, time_t expireitme, time_t now);
-int rdbLoadType(FILE *fp);
-time_t rdbLoadTime(FILE *fp);
-robj *rdbLoadStringObject(FILE *fp);
-int rdbSaveType(FILE *fp, unsigned char type);
-int rdbSaveLen(FILE *fp, uint32_t len);
+#include "rdb.h"
 
 /* AOF persistence */
 void flushAppendOnlyFile(void);
diff --git a/src/rio.c b/src/rio.c
new file mode 100644
index 00000000..e69d939f
--- /dev/null
+++ b/src/rio.c
@@ -0,0 +1,106 @@
+#include <string.h>
+#include "rio.h"
+#include "util.h"
+
+/* Returns 1 or 0 for success/failure. */
+static size_t rioBufferWrite(rio *r, const void *buf, size_t len) {
+    r->io.buffer.ptr = sdscatlen(r->io.buffer.ptr,(char*)buf,len);
+    r->io.buffer.pos += len;
+    return len;
+}
+
+/* Returns 1 or 0 for success/failure. */
+static size_t rioBufferRead(rio *r, void *buf, size_t len) {
+    if (sdslen(r->io.buffer.ptr)-r->io.buffer.pos < len)
+        return 0;
+    memcpy(buf,r->io.buffer.ptr+r->io.buffer.pos,len);
+    r->io.buffer.pos += len;
+    return 1;
+}
+
+/* Returns read/write position in buffer. */
+static off_t rioBufferTell(rio *r) {
+    return r->io.buffer.pos;
+}
+
+/* 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);
+}
+
+/* Returns 1 or 0 for success/failure. */
+static size_t rioFileRead(rio *r, void *buf, size_t len) {
+    return fread(buf,len,1,r->io.file.fp);
+}
+
+/* Returns read/write position in file. */
+static off_t rioFileTell(rio *r) {
+    return ftello(r->io.file.fp);
+}
+
+static const rio rioBufferIO = {
+    rioBufferRead,
+    rioBufferWrite,
+    rioBufferTell,
+    { { NULL, 0 } } /* union for io-specific vars */
+};
+
+static const rio rioFileIO = {
+    rioFileRead,
+    rioFileWrite,
+    rioFileTell,
+    { { NULL, 0 } } /* union for io-specific vars */
+};
+
+rio rioInitWithFile(FILE *fp) {
+    rio r = rioFileIO;
+    r.io.file.fp = fp;
+    return r;
+}
+rio rioInitWithBuffer(sds s) {
+    rio r = rioBufferIO;
+    r.io.buffer.ptr = s;
+    r.io.buffer.pos = 0;
+    return r;
+}
+
+/* Write multi bulk count in the format: "*<count>\r\n". */
+size_t rioWriteBulkCount(rio *r, char prefix, int count) {
+    char cbuf[128];
+    int clen;
+
+    cbuf[0] = prefix;
+    clen = 1+ll2string(cbuf+1,sizeof(cbuf)-1,count);
+    cbuf[clen++] = '\r';
+    cbuf[clen++] = '\n';
+    if (rioWrite(r,cbuf,clen) == 0) return 0;
+    return clen;
+}
+
+/* Write binary-safe string in the format: "$<count>\r\n<payload>\r\n". */
+size_t rioWriteBulkString(rio *r, const char *buf, size_t len) {
+    size_t nwritten;
+
+    if ((nwritten = rioWriteBulkCount(r,'$',len)) == 0) return 0;
+    if (len > 0 && rioWrite(r,buf,len) == 0) return 0;
+    if (rioWrite(r,"\r\n",2) == 0) return 0;
+    return nwritten+len+2;
+}
+
+/* Write a long long value in format: "$<count>\r\n<payload>\r\n". */
+size_t rioWriteBulkLongLong(rio *r, long long l) {
+    char lbuf[32];
+    unsigned int llen;
+
+    llen = ll2string(lbuf,sizeof(lbuf),l);
+    return rioWriteBulkString(r,lbuf,llen);
+}
+
+/* Write a double value in the format: "$<count>\r\n<payload>\r\n" */
+size_t rioWriteBulkDouble(rio *r, double d) {
+    char dbuf[128];
+    unsigned int dlen;
+
+    dlen = snprintf(dbuf,sizeof(dbuf),"%.17g",d);
+    return rioWriteBulkString(r,dbuf,dlen);
+}
diff --git a/src/rio.h b/src/rio.h
new file mode 100644
index 00000000..fe863cb5
--- /dev/null
+++ b/src/rio.h
@@ -0,0 +1,39 @@
+#ifndef __REDIS_RIO_H
+#define __REDIS_RIO_H
+
+#include <stdio.h>
+#include "sds.h"
+
+struct _rio {
+    /* Backend functions. Both read and write should return 0 for short reads
+     * or writes, identical to the return values of fread/fwrite. */
+    size_t (*read)(struct _rio *, void *buf, size_t len);
+    size_t (*write)(struct _rio *, const void *buf, size_t len);
+    off_t (*tell)(struct _rio *);
+
+    /* Backend-specific vars. */
+    union {
+        struct {
+            sds ptr;
+            off_t pos;
+        } buffer;
+        struct {
+            FILE *fp;
+        } file;
+    } io;
+};
+
+typedef struct _rio rio;
+
+#define rioWrite(rio,buf,len) ((rio)->write((rio),(buf),(len)))
+#define rioRead(rio,buf,len) ((rio)->read((rio),(buf),(len)))
+
+rio rioInitWithFile(FILE *fp);
+rio rioInitWithBuffer(sds s);
+
+size_t rioWriteBulkCount(rio *r, char prefix, int count);
+size_t rioWriteBulkString(rio *r, const char *buf, size_t len);
+size_t rioWriteBulkLongLong(rio *r, long long l);
+size_t rioWriteBulkDouble(rio *r, double d);
+
+#endif