1 /* rio.c is a simple stream-oriented I/O abstraction that provides an interface 2 * to write code that can consume/produce data using different concrete input 3 * and output devices. For instance the same rdb.c code using the rio 4 * abstraction can be used to read and write the RDB format using in-memory 5 * buffers or files. 6 * 7 * A rio object provides the following methods: 8 * read: read from stream. 9 * write: write to stream. 10 * tell: get the current offset. 11 * 12 * It is also possible to set a 'checksum' method that is used by rio.c in order 13 * to compute a checksum of the data written or read, or to query the rio object 14 * for the current checksum. 15 * 16 * ---------------------------------------------------------------------------- 17 * 18 * Copyright (c) 2009-2012, Pieter Noordhuis <pcnoordhuis at gmail dot com> 19 * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com> 20 * All rights reserved. 21 * 22 * Redistribution and use in source and binary forms, with or without 23 * modification, are permitted provided that the following conditions are met: 24 * 25 * * Redistributions of source code must retain the above copyright notice, 26 * this list of conditions and the following disclaimer. 27 * * Redistributions in binary form must reproduce the above copyright 28 * notice, this list of conditions and the following disclaimer in the 29 * documentation and/or other materials provided with the distribution. 30 * * Neither the name of Redis nor the names of its contributors may be used 31 * to endorse or promote products derived from this software without 32 * specific prior written permission. 33 * 34 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 35 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 36 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 37 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE 38 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 39 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 40 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 41 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 42 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) 43 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 44 * POSSIBILITY OF SUCH DAMAGE. 45 */ 46 47 48 #include "fmacros.h" 49 #include <string.h> 50 #include <stdio.h> 51 #include <unistd.h> 52 #include "rio.h" 53 #include "util.h" 54 #include "crc64.h" 55 #include "config.h" 56 #include "redis.h" 57 58 /* Returns 1 or 0 for success/failure. */ 59 static size_t rioBufferWrite(rio *r, const void *buf, size_t len) { 60 r->io.buffer.ptr = sdscatlen(r->io.buffer.ptr,(char*)buf,len); 61 r->io.buffer.pos += len; 62 return 1; 63 } 64 65 /* Returns 1 or 0 for success/failure. */ 66 static size_t rioBufferRead(rio *r, void *buf, size_t len) { 67 if (sdslen(r->io.buffer.ptr)-r->io.buffer.pos < len) 68 return 0; /* not enough buffer to return len bytes. */ 69 memcpy(buf,r->io.buffer.ptr+r->io.buffer.pos,len); 70 r->io.buffer.pos += len; 71 return 1; 72 } 73 74 /* Returns read/write position in buffer. */ 75 static off_t rioBufferTell(rio *r) { 76 return r->io.buffer.pos; 77 } 78 79 /* Returns 1 or 0 for success/failure. */ 80 static size_t rioFileWrite(rio *r, const void *buf, size_t len) { 81 size_t retval; 82 83 retval = fwrite(buf,len,1,r->io.file.fp); 84 r->io.file.buffered += len; 85 86 if (r->io.file.autosync && 87 r->io.file.buffered >= r->io.file.autosync) 88 { 89 fflush(r->io.file.fp); 90 aof_fsync(fileno(r->io.file.fp)); 91 r->io.file.buffered = 0; 92 } 93 return retval; 94 } 95 96 /* Returns 1 or 0 for success/failure. */ 97 static size_t rioFileRead(rio *r, void *buf, size_t len) { 98 return fread(buf,len,1,r->io.file.fp); 99 } 100 101 /* Returns read/write position in file. */ 102 static off_t rioFileTell(rio *r) { 103 return ftello(r->io.file.fp); 104 } 105 106 static const rio rioBufferIO = { 107 rioBufferRead, 108 rioBufferWrite, 109 rioBufferTell, 110 NULL, /* update_checksum */ 111 0, /* current checksum */ 112 0, /* bytes read or written */ 113 0, /* read/write chunk size */ 114 { { NULL, 0 } } /* union for io-specific vars */ 115 }; 116 117 static const rio rioFileIO = { 118 rioFileRead, 119 rioFileWrite, 120 rioFileTell, 121 NULL, /* update_checksum */ 122 0, /* current checksum */ 123 0, /* bytes read or written */ 124 0, /* read/write chunk size */ 125 { { NULL, 0 } } /* union for io-specific vars */ 126 }; 127 128 void rioInitWithFile(rio *r, FILE *fp) { 129 *r = rioFileIO; 130 r->io.file.fp = fp; 131 r->io.file.buffered = 0; 132 r->io.file.autosync = 0; 133 } 134 135 void rioInitWithBuffer(rio *r, sds s) { 136 *r = rioBufferIO; 137 r->io.buffer.ptr = s; 138 r->io.buffer.pos = 0; 139 } 140 141 /* This function can be installed both in memory and file streams when checksum 142 * computation is needed. */ 143 void rioGenericUpdateChecksum(rio *r, const void *buf, size_t len) { 144 r->cksum = crc64(r->cksum,buf,len); 145 } 146 147 /* Set the file-based rio object to auto-fsync every 'bytes' file written. 148 * By default this is set to zero that means no automatic file sync is 149 * performed. 150 * 151 * This feature is useful in a few contexts since when we rely on OS write 152 * buffers sometimes the OS buffers way too much, resulting in too many 153 * disk I/O concentrated in very little time. When we fsync in an explicit 154 * way instead the I/O pressure is more distributed across time. */ 155 void rioSetAutoSync(rio *r, off_t bytes) { 156 redisAssert(r->read == rioFileIO.read); 157 r->io.file.autosync = bytes; 158 } 159 160 /* ------------------------------ Higher level interface --------------------------- 161 * The following higher level functions use lower level rio.c functions to help 162 * generating the Redis protocol for the Append Only File. */ 163 164 /* Write multi bulk count in the format: "*<count>\r\n". */ 165 size_t rioWriteBulkCount(rio *r, char prefix, int count) { 166 char cbuf[128]; 167 int clen; 168 169 cbuf[0] = prefix; 170 clen = 1+ll2string(cbuf+1,sizeof(cbuf)-1,count); 171 cbuf[clen++] = '\r'; 172 cbuf[clen++] = '\n'; 173 if (rioWrite(r,cbuf,clen) == 0) return 0; 174 return clen; 175 } 176 177 /* Write binary-safe string in the format: "$<count>\r\n<payload>\r\n". */ 178 size_t rioWriteBulkString(rio *r, const char *buf, size_t len) { 179 size_t nwritten; 180 181 if ((nwritten = rioWriteBulkCount(r,'$',len)) == 0) return 0; 182 if (len > 0 && rioWrite(r,buf,len) == 0) return 0; 183 if (rioWrite(r,"\r\n",2) == 0) return 0; 184 return nwritten+len+2; 185 } 186 187 /* Write a long long value in format: "$<count>\r\n<payload>\r\n". */ 188 size_t rioWriteBulkLongLong(rio *r, long long l) { 189 char lbuf[32]; 190 unsigned int llen; 191 192 llen = ll2string(lbuf,sizeof(lbuf),l); 193 return rioWriteBulkString(r,lbuf,llen); 194 } 195 196 /* Write a double value in the format: "$<count>\r\n<payload>\r\n" */ 197 size_t rioWriteBulkDouble(rio *r, double d) { 198 char dbuf[128]; 199 unsigned int dlen; 200 201 dlen = snprintf(dbuf,sizeof(dbuf),"%.17g",d); 202 return rioWriteBulkString(r,dbuf,dlen); 203 } 204