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

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 \

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;
 }
 
 /* -----------------------------------------------------------------------------

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);

src/rdb.h

+#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

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);

src/rio.c

+#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);
+}

src/rio.h

+#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