redict/src/rdb.c

1684 lines
61 KiB
C
Raw Normal View History

/*
* Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com>
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are met:
*
* * Redistributions of source code must retain the above copyright notice,
* this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
* * Neither the name of Redis nor the names of its contributors may be used
* to endorse or promote products derived from this software without
* specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
* POSSIBILITY OF SUCH DAMAGE.
*/
#include "redis.h"
#include "lzf.h" /* LZF compression library */
2012-01-03 01:14:10 -05:00
#include "zipmap.h"
#include "endianconv.h"
#include <math.h>
#include <sys/types.h>
#include <sys/time.h>
#include <sys/resource.h>
#include <sys/wait.h>
#include <arpa/inet.h>
#include <sys/stat.h>
#define RDB_LOAD_NONE 0
#define RDB_LOAD_ENC (1<<0)
#define RDB_LOAD_PLAIN (1<<1)
static int rdbWriteRaw(rio *rdb, void *p, size_t len) {
if (rdb && rioWrite(rdb,p,len) == 0)
return -1;
return len;
}
int rdbSaveType(rio *rdb, unsigned char type) {
return rdbWriteRaw(rdb,&type,1);
}
/* Load a "type" in RDB format, that is a one byte unsigned integer.
* This function is not only used to load object types, but also special
* "types" like the end-of-file type, the EXPIRE type, and so forth. */
2011-05-13 17:24:19 -04:00
int rdbLoadType(rio *rdb) {
unsigned char type;
if (rioRead(rdb,&type,1) == 0) return -1;
return type;
}
2011-05-13 17:24:19 -04:00
time_t rdbLoadTime(rio *rdb) {
int32_t t32;
if (rioRead(rdb,&t32,4) == 0) return -1;
return (time_t)t32;
}
int rdbSaveMillisecondTime(rio *rdb, long long t) {
int64_t t64 = (int64_t) t;
return rdbWriteRaw(rdb,&t64,8);
}
long long rdbLoadMillisecondTime(rio *rdb) {
int64_t t64;
if (rioRead(rdb,&t64,8) == 0) return -1;
return (long long)t64;
}
2011-05-13 17:24:19 -04:00
/* Saves an encoded length. The first two bits in the first byte are used to
* hold the encoding type. See the REDIS_RDB_* definitions for more information
* on the types of encoding. */
int rdbSaveLen(rio *rdb, uint32_t len) {
unsigned char buf[2];
size_t nwritten;
if (len < (1<<6)) {
/* Save a 6 bit len */
buf[0] = (len&0xFF)|(REDIS_RDB_6BITLEN<<6);
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(rdb,buf,2) == -1) return -1;
nwritten = 2;
} else {
/* Save a 32 bit len */
buf[0] = (REDIS_RDB_32BITLEN<<6);
if (rdbWriteRaw(rdb,buf,1) == -1) return -1;
len = htonl(len);
if (rdbWriteRaw(rdb,&len,4) == -1) return -1;
nwritten = 1+4;
}
return nwritten;
}
2011-05-13 17:24:19 -04:00
/* Load an encoded length. The "isencoded" argument is set to 1 if the length
* is not actually a length but an "encoding type". See the REDIS_RDB_ENC_*
* definitions in rdb.h for more information. */
uint32_t rdbLoadLen(rio *rdb, int *isencoded) {
unsigned char buf[2];
uint32_t len;
int type;
if (isencoded) *isencoded = 0;
if (rioRead(rdb,buf,1) == 0) return REDIS_RDB_LENERR;
type = (buf[0]&0xC0)>>6;
if (type == REDIS_RDB_ENCVAL) {
/* Read a 6 bit encoding type. */
if (isencoded) *isencoded = 1;
return buf[0]&0x3F;
} else if (type == REDIS_RDB_6BITLEN) {
/* Read a 6 bit len. */
return buf[0]&0x3F;
} else if (type == REDIS_RDB_14BITLEN) {
/* Read a 14 bit len. */
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 (rioRead(rdb,&len,4) == 0) return REDIS_RDB_LENERR;
return ntohl(len);
}
}
/* Encodes the "value" argument as integer when it fits in the supported ranges
* for encoded types. If the function successfully encodes the integer, the
* representation is stored in the buffer pointer to by "enc" and the string
* length is returned. Otherwise 0 is returned. */
int rdbEncodeInteger(long long value, unsigned char *enc) {
if (value >= -(1<<7) && value <= (1<<7)-1) {
enc[0] = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_INT8;
enc[1] = value&0xFF;
return 2;
} else if (value >= -(1<<15) && value <= (1<<15)-1) {
enc[0] = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_INT16;
enc[1] = value&0xFF;
enc[2] = (value>>8)&0xFF;
return 3;
} else if (value >= -((long long)1<<31) && value <= ((long long)1<<31)-1) {
enc[0] = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_INT32;
enc[1] = value&0xFF;
enc[2] = (value>>8)&0xFF;
enc[3] = (value>>16)&0xFF;
enc[4] = (value>>24)&0xFF;
return 5;
} else {
return 0;
}
}
2011-05-13 17:24:19 -04:00
/* Loads an integer-encoded object with the specified encoding type "enctype".
* The returned value changes according to the flags, see
* rdbGenerincLoadStringObject() for more info. */
void *rdbLoadIntegerObject(rio *rdb, int enctype, int flags) {
int plain = flags & RDB_LOAD_PLAIN;
int encode = flags & RDB_LOAD_ENC;
2011-05-13 17:24:19 -04:00
unsigned char enc[4];
long long val;
if (enctype == REDIS_RDB_ENC_INT8) {
if (rioRead(rdb,enc,1) == 0) return NULL;
val = (signed char)enc[0];
} else if (enctype == REDIS_RDB_ENC_INT16) {
uint16_t v;
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 (rioRead(rdb,enc,4) == 0) return NULL;
v = enc[0]|(enc[1]<<8)|(enc[2]<<16)|(enc[3]<<24);
val = (int32_t)v;
} else {
val = 0; /* anti-warning */
redisPanic("Unknown RDB integer encoding type");
}
if (plain) {
char buf[REDIS_LONGSTR_SIZE], *p;
int len = ll2string(buf,sizeof(buf),val);
p = zmalloc(len);
memcpy(p,buf,len);
return p;
} else if (encode) {
2011-05-13 17:24:19 -04:00
return createStringObjectFromLongLong(val);
} else {
2011-05-13 17:24:19 -04:00
return createObject(REDIS_STRING,sdsfromlonglong(val));
}
2011-05-13 17:24:19 -04:00
}
/* String objects in the form "2391" "-100" without any space and with a
* range of values that can fit in an 8, 16 or 32 bit signed value can be
* encoded as integers to save space */
int rdbTryIntegerEncoding(char *s, size_t len, unsigned char *enc) {
long long value;
char *endptr, buf[32];
/* Check if it's possible to encode this value as a number */
value = strtoll(s, &endptr, 10);
if (endptr[0] != '\0') return 0;
ll2string(buf,32,value);
/* If the number converted back into a string is not identical
* then it's not possible to encode the string as integer */
if (strlen(buf) != len || memcmp(buf,s,len)) return 0;
return rdbEncodeInteger(value,enc);
}
int rdbSaveLzfBlob(rio *rdb, void *data, size_t compress_len,
size_t original_len) {
unsigned char byte;
int n, nwritten = 0;
/* Data compressed! Let's save it on disk */
byte = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_LZF;
if ((n = rdbWriteRaw(rdb,&byte,1)) == -1) goto writeerr;
nwritten += n;
if ((n = rdbSaveLen(rdb,compress_len)) == -1) goto writeerr;
nwritten += n;
if ((n = rdbSaveLen(rdb,original_len)) == -1) goto writeerr;
nwritten += n;
if ((n = rdbWriteRaw(rdb,data,compress_len)) == -1) goto writeerr;
nwritten += n;
return nwritten;
writeerr:
return -1;
}
int rdbSaveLzfStringObject(rio *rdb, unsigned char *s, size_t len) {
size_t comprlen, outlen;
void *out;
/* We require at least four bytes compression for this to be worth it */
if (len <= 4) return 0;
outlen = len-4;
if ((out = zmalloc(outlen+1)) == NULL) return 0;
comprlen = lzf_compress(s, len, out, outlen);
if (comprlen == 0) {
zfree(out);
return 0;
}
size_t nwritten = rdbSaveLzfBlob(rdb, out, comprlen, len);
zfree(out);
return nwritten;
}
/* Load an LZF compressed string in RDB format. The returned value
* changes according to 'flags'. For more info check the
* rdbGenericLoadStringObject() function. */
void *rdbLoadLzfStringObject(rio *rdb, int flags) {
int plain = flags & RDB_LOAD_PLAIN;
2011-05-13 17:24:19 -04:00
unsigned int len, clen;
unsigned char *c = NULL;
sds val = 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;
/* Allocate our target according to the uncompressed size. */
if (plain) {
val = zmalloc(len);
} else {
if ((val = sdsnewlen(NULL,len)) == NULL) goto err;
}
/* Load the compressed representation and uncompress it to target. */
2011-05-13 17:24:19 -04:00
if (rioRead(rdb,c,clen) == 0) goto err;
if (lzf_decompress(c,clen,val,len) == 0) goto err;
zfree(c);
if (plain)
return val;
else
return createObject(REDIS_STRING,val);
2011-05-13 17:24:19 -04:00
err:
zfree(c);
if (plain)
zfree(val);
else
sdsfree(val);
2011-05-13 17:24:19 -04:00
return NULL;
}
2013-01-16 12:00:20 -05:00
/* Save a string object 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(rio *rdb, unsigned char *s, size_t len) {
int enclen;
int n, nwritten = 0;
/* Try integer encoding */
if (len <= 11) {
unsigned char buf[5];
if ((enclen = rdbTryIntegerEncoding((char*)s,len,buf)) > 0) {
if (rdbWriteRaw(rdb,buf,enclen) == -1) return -1;
return enclen;
}
}
/* Try LZF compression - under 20 bytes it's unable to compress even
* aaaaaaaaaaaaaaaaaa so skip it */
2011-12-21 06:22:13 -05:00
if (server.rdb_compression && len > 20) {
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(rdb,len)) == -1) return -1;
nwritten += n;
if (len > 0) {
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(rio *rdb, long long value) {
unsigned char buf[32];
int n, nwritten = 0;
int enclen = rdbEncodeInteger(value,buf);
if (enclen > 0) {
return rdbWriteRaw(rdb,buf,enclen);
} else {
/* Encode as string */
enclen = ll2string((char*)buf,32,value);
redisAssert(enclen < 32);
if ((n = rdbSaveLen(rdb,enclen)) == -1) return -1;
nwritten += n;
if ((n = rdbWriteRaw(rdb,buf,enclen)) == -1) return -1;
nwritten += n;
}
return nwritten;
}
/* Like rdbSaveStringObjectRaw() but handle encoded objects */
int rdbSaveStringObject(rio *rdb, robj *obj) {
/* Avoid to decode the object, then encode it again, if the
2013-01-16 12:00:20 -05:00
* object is already integer encoded. */
if (obj->encoding == REDIS_ENCODING_INT) {
return rdbSaveLongLongAsStringObject(rdb,(long)obj->ptr);
} else {
redisAssertWithInfo(NULL,obj,sdsEncodedObject(obj));
return rdbSaveRawString(rdb,obj->ptr,sdslen(obj->ptr));
}
}
/* Load a string object from an RDB file according to flags:
*
* RDB_LOAD_NONE (no flags): load an RDB object, unencoded.
* RDB_LOAD_ENC: If the returned type is a Redis object, try to
* encode it in a special way to be more memory
* efficient. When this flag is passed the function
* no longer guarantees that obj->ptr is an SDS string.
* RDB_LOAD_PLAIN: Return a plain string allocated with zmalloc()
* instead of a Redis object.
*/
void *rdbGenericLoadStringObject(rio *rdb, int flags) {
int encode = flags & RDB_LOAD_ENC;
int plain = flags & RDB_LOAD_PLAIN;
2011-05-13 17:24:19 -04:00
int isencoded;
uint32_t len;
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(rdb,len,flags);
2011-05-13 17:24:19 -04:00
case REDIS_RDB_ENC_LZF:
return rdbLoadLzfStringObject(rdb,flags);
2011-05-13 17:24:19 -04:00
default:
redisPanic("Unknown RDB encoding type");
}
}
if (len == REDIS_RDB_LENERR) return NULL;
if (!plain) {
robj *o = encode ? createStringObject(NULL,len) :
createRawStringObject(NULL,len);
if (len && rioRead(rdb,o->ptr,len) == 0) {
decrRefCount(o);
return NULL;
}
return o;
} else {
void *buf = zmalloc(len);
if (len && rioRead(rdb,buf,len) == 0) {
zfree(buf);
return NULL;
}
return buf;
2011-05-13 17:24:19 -04:00
}
}
robj *rdbLoadStringObject(rio *rdb) {
return rdbGenericLoadStringObject(rdb,RDB_LOAD_NONE);
2011-05-13 17:24:19 -04:00
}
robj *rdbLoadEncodedStringObject(rio *rdb) {
return rdbGenericLoadStringObject(rdb,RDB_LOAD_ENC);
2011-05-13 17:24:19 -04:00
}
/* Save a double value. Doubles are saved as strings prefixed by an unsigned
2013-01-16 12:00:20 -05:00
* 8 bit integer specifying the length of the representation.
* This 8 bit integer has special values in order to specify the following
* conditions:
* 253: not a number
* 254: + inf
* 255: - inf
*/
int rdbSaveDoubleValue(rio *rdb, double val) {
unsigned char buf[128];
int len;
if (isnan(val)) {
buf[0] = 253;
len = 1;
} else if (!isfinite(val)) {
len = 1;
buf[0] = (val < 0) ? 255 : 254;
} else {
#if (DBL_MANT_DIG >= 52) && (LLONG_MAX == 0x7fffffffffffffffLL)
/* Check if the float is in a safe range to be casted into a
* long long. We are assuming that long long is 64 bit here.
* Also we are assuming that there are no implementations around where
* double has precision < 52 bit.
*
* Under this assumptions we test if a double is inside an interval
* where casting to long long is safe. Then using two castings we
* make sure the decimal part is zero. If all this is true we use
* integer printing function that is much faster. */
double min = -4503599627370495; /* (2^52)-1 */
double max = 4503599627370496; /* -(2^52) */
if (val > min && val < max && val == ((double)((long long)val)))
2014-05-12 05:05:18 -04:00
ll2string((char*)buf+1,sizeof(buf)-1,(long long)val);
else
#endif
snprintf((char*)buf+1,sizeof(buf)-1,"%.17g",val);
buf[0] = strlen((char*)buf+1);
len = buf[0]+1;
}
return rdbWriteRaw(rdb,buf,len);
}
2011-05-13 17:24:19 -04:00
/* For information about double serialization check rdbSaveDoubleValue() */
int rdbLoadDoubleValue(rio *rdb, double *val) {
char buf[256];
2011-05-13 17:24:19 -04:00
unsigned char len;
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 (rioRead(rdb,buf,len) == 0) return -1;
buf[len] = '\0';
sscanf(buf, "%lg", val);
return 0;
}
}
/* Save the object type of object "o". */
int rdbSaveObjectType(rio *rdb, robj *o) {
switch (o->type) {
case REDIS_STRING:
return rdbSaveType(rdb,REDIS_RDB_TYPE_STRING);
case REDIS_LIST:
if (o->encoding == REDIS_ENCODING_QUICKLIST)
return rdbSaveType(rdb,REDIS_RDB_TYPE_LIST_QUICKLIST);
2011-05-13 17:24:19 -04:00
else
redisPanic("Unknown list encoding");
case REDIS_SET:
if (o->encoding == REDIS_ENCODING_INTSET)
return rdbSaveType(rdb,REDIS_RDB_TYPE_SET_INTSET);
else if (o->encoding == REDIS_ENCODING_HT)
return rdbSaveType(rdb,REDIS_RDB_TYPE_SET);
else
redisPanic("Unknown set encoding");
case REDIS_ZSET:
if (o->encoding == REDIS_ENCODING_ZIPLIST)
return rdbSaveType(rdb,REDIS_RDB_TYPE_ZSET_ZIPLIST);
else if (o->encoding == REDIS_ENCODING_SKIPLIST)
return rdbSaveType(rdb,REDIS_RDB_TYPE_ZSET);
else
redisPanic("Unknown sorted set encoding");
case REDIS_HASH:
2012-01-03 01:14:10 -05:00
if (o->encoding == REDIS_ENCODING_ZIPLIST)
return rdbSaveType(rdb,REDIS_RDB_TYPE_HASH_ZIPLIST);
2011-05-13 17:24:19 -04:00
else if (o->encoding == REDIS_ENCODING_HT)
return rdbSaveType(rdb,REDIS_RDB_TYPE_HASH);
else
redisPanic("Unknown hash encoding");
default:
redisPanic("Unknown object type");
}
return -1; /* avoid warning */
}
/* Use rdbLoadType() to load a TYPE in RDB format, but returns -1 if the
* type is not specifically a valid Object Type. */
2011-05-13 17:24:19 -04:00
int rdbLoadObjectType(rio *rdb) {
int type;
if ((type = rdbLoadType(rdb)) == -1) return -1;
if (!rdbIsObjectType(type)) return -1;
return type;
}
/* Save a Redis object. Returns -1 on error, number of bytes written on success. */
int rdbSaveObject(rio *rdb, robj *o) {
int n = 0, nwritten = 0;
if (o->type == REDIS_STRING) {
/* Save a string value */
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_QUICKLIST) {
quicklist *ql = o->ptr;
quicklistNode *node = ql->head;
if ((n = rdbSaveLen(rdb,ql->len)) == -1) return -1;
nwritten += n;
do {
if (quicklistNodeIsCompressed(node)) {
void *data;
size_t compress_len = quicklistGetLzf(node, &data);
if ((n = rdbSaveLzfBlob(rdb,data,compress_len,node->sz)) == -1) return -1;
nwritten += n;
} else {
if ((n = rdbSaveRawString(rdb,node->zl,node->sz)) == -1) return -1;
nwritten += n;
}
} while ((node = node->next));
} else {
redisPanic("Unknown list encoding");
}
} else if (o->type == REDIS_SET) {
/* Save a set value */
if (o->encoding == REDIS_ENCODING_HT) {
dict *set = o->ptr;
dictIterator *di = dictGetIterator(set);
dictEntry *de;
if ((n = rdbSaveLen(rdb,dictSize(set))) == -1) return -1;
nwritten += n;
while((de = dictNext(di)) != NULL) {
robj *eleobj = dictGetKey(de);
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(rdb,o->ptr,l)) == -1) return -1;
nwritten += n;
} else {
redisPanic("Unknown set encoding");
}
} else if (o->type == REDIS_ZSET) {
/* Save a sorted set value */
if (o->encoding == REDIS_ENCODING_ZIPLIST) {
size_t l = ziplistBlobLen((unsigned char*)o->ptr);
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(rdb,dictSize(zs->dict))) == -1) return -1;
nwritten += n;
while((de = dictNext(di)) != NULL) {
robj *eleobj = dictGetKey(de);
double *score = dictGetVal(de);
if ((n = rdbSaveStringObject(rdb,eleobj)) == -1) return -1;
nwritten += n;
if ((n = rdbSaveDoubleValue(rdb,*score)) == -1) return -1;
nwritten += n;
}
dictReleaseIterator(di);
} else {
2011-04-06 10:15:15 -04:00
redisPanic("Unknown sorted set encoding");
}
} else if (o->type == REDIS_HASH) {
/* Save a hash value */
2012-01-03 01:14:10 -05:00
if (o->encoding == REDIS_ENCODING_ZIPLIST) {
size_t l = ziplistBlobLen((unsigned char*)o->ptr);
if ((n = rdbSaveRawString(rdb,o->ptr,l)) == -1) return -1;
nwritten += n;
2012-01-03 01:14:10 -05:00
} else if (o->encoding == REDIS_ENCODING_HT) {
dictIterator *di = dictGetIterator(o->ptr);
dictEntry *de;
if ((n = rdbSaveLen(rdb,dictSize((dict*)o->ptr))) == -1) return -1;
nwritten += n;
while((de = dictNext(di)) != NULL) {
robj *key = dictGetKey(de);
robj *val = dictGetVal(de);
if ((n = rdbSaveStringObject(rdb,key)) == -1) return -1;
nwritten += n;
if ((n = rdbSaveStringObject(rdb,val)) == -1) return -1;
nwritten += n;
}
dictReleaseIterator(di);
2012-01-03 01:14:10 -05:00
} else {
redisPanic("Unknown hash encoding");
}
2012-01-03 01:14:10 -05:00
} else {
redisPanic("Unknown object type");
}
return nwritten;
}
/* Return the length the object will have on disk if saved with
* the rdbSaveObject() function. Currently we use a trick to get
* this length with very little changes to the code. In the future
* we could switch to a faster solution. */
off_t rdbSavedObjectLen(robj *o) {
int len = rdbSaveObject(NULL,o);
redisAssertWithInfo(NULL,o,len != -1);
return len;
}
/* Save a key-value pair, with expire time, type, key, value.
* On error -1 is returned.
2013-01-16 12:00:20 -05:00
* On success if the key was actually saved 1 is returned, otherwise 0
* is returned (the key was already expired). */
int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val,
long long expiretime, long long now)
{
/* Save the expire time */
if (expiretime != -1) {
/* If this key is already expired skip it */
if (expiretime < now) return 0;
if (rdbSaveType(rdb,REDIS_RDB_OPCODE_EXPIRETIME_MS) == -1) return -1;
if (rdbSaveMillisecondTime(rdb,expiretime) == -1) return -1;
}
/* Save type, key, value */
if (rdbSaveObjectType(rdb,val) == -1) return -1;
if (rdbSaveStringObject(rdb,key) == -1) return -1;
if (rdbSaveObject(rdb,val) == -1) return -1;
return 1;
}
/* Save an AUX field. */
int rdbSaveAuxField(rio *rdb, void *key, size_t keylen, void *val, size_t vallen) {
if (rdbSaveType(rdb,REDIS_RDB_OPCODE_AUX) == -1) return -1;
if (rdbSaveRawString(rdb,key,keylen) == -1) return -1;
if (rdbSaveRawString(rdb,val,vallen) == -1) return -1;
return 1;
}
/* Wrapper for rdbSaveAuxField() used when key/val length can be obtained
* with strlen(). */
int rdbSaveAuxFieldStrStr(rio *rdb, char *key, char *val) {
return rdbSaveAuxField(rdb,key,strlen(key),val,strlen(val));
}
/* Wrapper for strlen(key) + integer type (up to long long range). */
int rdbSaveAuxFieldStrInt(rio *rdb, char *key, long long val) {
char buf[REDIS_LONGSTR_SIZE];
int vlen = ll2string(buf,sizeof(buf),val);
return rdbSaveAuxField(rdb,key,strlen(key),buf,vlen);
}
/* Save a few default AUX fields with information about the RDB generated. */
int rdbSaveInfoAuxFields(rio *rdb) {
int redis_bits = (sizeof(void*) == 8) ? 64 : 32;
/* Add a few fields about the state when the RDB was created. */
if (rdbSaveAuxFieldStrStr(rdb,"redis-ver",REDIS_VERSION) == -1) return -1;
if (rdbSaveAuxFieldStrInt(rdb,"redis-bits",redis_bits) == -1) return -1;
if (rdbSaveAuxFieldStrInt(rdb,"ctime",time(NULL)) == -1) return -1;
if (rdbSaveAuxFieldStrInt(rdb,"used-mem",zmalloc_used_memory()) == -1) return -1;
return 1;
}
/* Produces a dump of the database in RDB format sending it to the specified
* Redis I/O channel. On success REDIS_OK is returned, otherwise REDIS_ERR
* is returned and part of the output, or all the output, can be
* missing because of I/O errors.
*
* When the function returns REDIS_ERR and if 'error' is not NULL, the
* integer pointed by 'error' is set to the value of errno just after the I/O
* error. */
int rdbSaveRio(rio *rdb, int *error) {
dictIterator *di = NULL;
dictEntry *de;
char magic[10];
int j;
long long now = mstime();
uint64_t cksum;
if (server.rdb_checksum)
rdb->update_cksum = rioGenericUpdateChecksum;
snprintf(magic,sizeof(magic),"REDIS%04d",REDIS_RDB_VERSION);
if (rdbWriteRaw(rdb,magic,9) == -1) goto werr;
if (rdbSaveInfoAuxFields(rdb) == -1) goto werr;
for (j = 0; j < server.dbnum; j++) {
redisDb *db = server.db+j;
dict *d = db->dict;
if (dictSize(d) == 0) continue;
di = dictGetSafeIterator(d);
if (!di) return REDIS_ERR;
/* Write the SELECT DB opcode */
if (rdbSaveType(rdb,REDIS_RDB_OPCODE_SELECTDB) == -1) goto werr;
if (rdbSaveLen(rdb,j) == -1) goto werr;
/* Write the RESIZE DB opcode. We trim the size to UINT32_MAX, which
* is currently the largest type we are able to represent in RDB sizes.
* However this does not limit the actual size of the DB to load since
* these sizes are just hints to resize the hash tables. */
uint32_t db_size, expires_size;
db_size = (dictSize(db->dict) <= UINT32_MAX) ?
dictSize(db->dict) :
UINT32_MAX;
expires_size = (dictSize(db->dict) <= UINT32_MAX) ?
dictSize(db->expires) :
UINT32_MAX;
if (rdbSaveType(rdb,REDIS_RDB_OPCODE_RESIZEDB) == -1) goto werr;
if (rdbSaveLen(rdb,db_size) == -1) goto werr;
if (rdbSaveLen(rdb,expires_size) == -1) goto werr;
/* Iterate this DB writing every entry */
while((de = dictNext(di)) != NULL) {
sds keystr = dictGetKey(de);
robj key, *o = dictGetVal(de);
long long expire;
initStaticStringObject(key,keystr);
expire = getExpire(db,&key);
if (rdbSaveKeyValuePair(rdb,&key,o,expire,now) == -1) goto werr;
}
dictReleaseIterator(di);
}
di = NULL; /* So that we don't release it again on error. */
/* EOF opcode */
if (rdbSaveType(rdb,REDIS_RDB_OPCODE_EOF) == -1) goto werr;
/* CRC64 checksum. It will be zero if checksum computation is disabled, the
* loading code skips the check in this case. */
cksum = rdb->cksum;
memrev64ifbe(&cksum);
if (rioWrite(rdb,&cksum,8) == 0) goto werr;
return REDIS_OK;
werr:
if (error) *error = errno;
if (di) dictReleaseIterator(di);
return REDIS_ERR;
}
/* This is just a wrapper to rdbSaveRio() that additionally adds a prefix
* and a suffix to the generated RDB dump. The prefix is:
*
* $EOF:<40 bytes unguessable hex string>\r\n
*
* While the suffix is the 40 bytes hex string we announced in the prefix.
* This way processes receiving the payload can understand when it ends
* without doing any processing of the content. */
int rdbSaveRioWithEOFMark(rio *rdb, int *error) {
char eofmark[REDIS_EOF_MARK_SIZE];
getRandomHexChars(eofmark,REDIS_EOF_MARK_SIZE);
if (error) *error = 0;
if (rioWrite(rdb,"$EOF:",5) == 0) goto werr;
if (rioWrite(rdb,eofmark,REDIS_EOF_MARK_SIZE) == 0) goto werr;
if (rioWrite(rdb,"\r\n",2) == 0) goto werr;
if (rdbSaveRio(rdb,error) == REDIS_ERR) goto werr;
if (rioWrite(rdb,eofmark,REDIS_EOF_MARK_SIZE) == 0) goto werr;
return REDIS_OK;
werr: /* Write error. */
/* Set 'error' only if not already set by rdbSaveRio() call. */
if (error && *error == 0) *error = errno;
return REDIS_ERR;
}
/* Save the DB on disk. Return REDIS_ERR on error, REDIS_OK on success. */
int rdbSave(char *filename) {
char tmpfile[256];
FILE *fp;
rio rdb;
int error = 0;
snprintf(tmpfile,256,"temp-%d.rdb", (int) getpid());
fp = fopen(tmpfile,"w");
if (!fp) {
redisLog(REDIS_WARNING, "Failed opening .rdb for saving: %s",
strerror(errno));
return REDIS_ERR;
}
rioInitWithFile(&rdb,fp);
if (rdbSaveRio(&rdb,&error) == REDIS_ERR) {
errno = error;
goto werr;
}
/* Make sure data will not remain on the OS's output buffers */
if (fflush(fp) == EOF) goto werr;
if (fsync(fileno(fp)) == -1) goto werr;
if (fclose(fp) == EOF) goto werr;
/* Use RENAME to make sure the DB file is changed atomically only
* if the generate DB file is ok. */
if (rename(tmpfile,filename) == -1) {
redisLog(REDIS_WARNING,"Error moving temp DB file on the final destination: %s", strerror(errno));
unlink(tmpfile);
return REDIS_ERR;
}
redisLog(REDIS_NOTICE,"DB saved on disk");
server.dirty = 0;
server.lastsave = time(NULL);
server.lastbgsave_status = REDIS_OK;
return REDIS_OK;
werr:
fclose(fp);
unlink(tmpfile);
redisLog(REDIS_WARNING,"Write error saving DB on disk: %s", strerror(errno));
return REDIS_ERR;
}
int rdbSaveBackground(char *filename) {
pid_t childpid;
long long start;
2011-12-21 06:22:13 -05:00
if (server.rdb_child_pid != -1) return REDIS_ERR;
2011-01-05 12:38:31 -05:00
server.dirty_before_bgsave = server.dirty;
server.lastbgsave_try = time(NULL);
2011-01-05 12:38:31 -05:00
start = ustime();
if ((childpid = fork()) == 0) {
2011-01-05 12:38:31 -05:00
int retval;
/* Child */
closeListeningSockets(0);
redisSetProcTitle("redis-rdb-bgsave");
retval = rdbSave(filename);
if (retval == REDIS_OK) {
size_t private_dirty = zmalloc_get_private_dirty();
if (private_dirty) {
redisLog(REDIS_NOTICE,
"RDB: %zu MB of memory used by copy-on-write",
private_dirty/(1024*1024));
}
}
exitFromChild((retval == REDIS_OK) ? 0 : 1);
} else {
/* Parent */
server.stat_fork_time = ustime()-start;
server.stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / server.stat_fork_time / (1024*1024*1024); /* GB per second. */
latencyAddSampleIfNeeded("fork",server.stat_fork_time/1000);
if (childpid == -1) {
server.lastbgsave_status = REDIS_ERR;
redisLog(REDIS_WARNING,"Can't save in background: fork: %s",
strerror(errno));
return REDIS_ERR;
}
redisLog(REDIS_NOTICE,"Background saving started by pid %d",childpid);
server.rdb_save_time_start = time(NULL);
2011-12-21 06:22:13 -05:00
server.rdb_child_pid = childpid;
server.rdb_child_type = REDIS_RDB_CHILD_TYPE_DISK;
updateDictResizePolicy();
return REDIS_OK;
}
return REDIS_OK; /* unreached */
}
void rdbRemoveTempFile(pid_t childpid) {
char tmpfile[256];
snprintf(tmpfile,sizeof(tmpfile),"temp-%d.rdb", (int) childpid);
unlink(tmpfile);
}
/* 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 rdbtype, rio *rdb) {
robj *o, *ele, *dec;
size_t len;
unsigned int i;
if (rdbtype == REDIS_RDB_TYPE_STRING) {
/* Read string value */
if ((o = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL;
o = tryObjectEncoding(o);
} else if (rdbtype == REDIS_RDB_TYPE_LIST) {
/* Read list value */
if ((len = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL;
o = createQuicklistObject();
quicklistSetOptions(o->ptr, server.list_max_ziplist_size,
server.list_compress_depth);
/* Load every single element of the list */
while(len--) {
if ((ele = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL;
dec = getDecodedObject(ele);
size_t len = sdslen(dec->ptr);
quicklistPushTail(o->ptr, dec->ptr, len);
decrRefCount(dec);
decrRefCount(ele);
}
} else if (rdbtype == REDIS_RDB_TYPE_SET) {
/* Read list/set value */
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) {
o = createSetObject();
/* It's faster to expand the dict to the right size asap in order
* to avoid rehashing */
if (len > DICT_HT_INITIAL_SIZE)
dictExpand(o->ptr,len);
} else {
o = createIntsetObject();
}
/* Load every single element of the list/set */
for (i = 0; i < len; i++) {
long long llval;
if ((ele = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL;
ele = tryObjectEncoding(ele);
if (o->encoding == REDIS_ENCODING_INTSET) {
/* Fetch integer value from element */
2010-08-26 13:10:40 -04:00
if (isObjectRepresentableAsLongLong(ele,&llval) == REDIS_OK) {
o->ptr = intsetAdd(o->ptr,llval,NULL);
} else {
setTypeConvert(o,REDIS_ENCODING_HT);
dictExpand(o->ptr,len);
}
}
/* This will also be called when the set was just converted
2012-04-07 08:48:30 -04:00
* to a regular hash table encoded set */
if (o->encoding == REDIS_ENCODING_HT) {
dictAdd((dict*)o->ptr,ele,NULL);
} else {
decrRefCount(ele);
}
}
} else if (rdbtype == REDIS_RDB_TYPE_ZSET) {
/* Read list/set value */
size_t zsetlen;
size_t maxelelen = 0;
zset *zs;
if ((zsetlen = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL;
o = createZsetObject();
zs = o->ptr;
/* Load every single element of the list/set */
while(zsetlen--) {
robj *ele;
double score;
zskiplistNode *znode;
if ((ele = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL;
ele = tryObjectEncoding(ele);
if (rdbLoadDoubleValue(rdb,&score) == -1) return NULL;
/* Don't care about integer-encoded strings. */
if (sdsEncodedObject(ele) && sdslen(ele->ptr) > maxelelen)
maxelelen = sdslen(ele->ptr);
znode = zslInsert(zs->zsl,score,ele);
dictAdd(zs->dict,ele,&znode->score);
incrRefCount(ele); /* added to skiplist */
}
/* Convert *after* loading, since sorted sets are not stored ordered. */
if (zsetLength(o) <= server.zset_max_ziplist_entries &&
maxelelen <= server.zset_max_ziplist_value)
zsetConvert(o,REDIS_ENCODING_ZIPLIST);
} else if (rdbtype == REDIS_RDB_TYPE_HASH) {
2012-01-03 01:14:10 -05:00
size_t len;
int ret;
len = rdbLoadLen(rdb, NULL);
if (len == REDIS_RDB_LENERR) return NULL;
o = createHashObject();
2012-01-03 01:14:10 -05:00
/* Too many entries? Use a hash table. */
2012-01-03 01:14:10 -05:00
if (len > server.hash_max_ziplist_entries)
hashTypeConvert(o, REDIS_ENCODING_HT);
/* Load every field and value into the ziplist */
while (o->encoding == REDIS_ENCODING_ZIPLIST && len > 0) {
2012-01-03 01:14:10 -05:00
robj *field, *value;
len--;
2012-01-03 01:14:10 -05:00
/* Load raw strings */
field = rdbLoadStringObject(rdb);
if (field == NULL) return NULL;
redisAssert(sdsEncodedObject(field));
2012-01-03 01:14:10 -05:00
value = rdbLoadStringObject(rdb);
if (value == NULL) return NULL;
redisAssert(sdsEncodedObject(value));
2012-01-03 01:14:10 -05:00
/* Add pair to ziplist */
o->ptr = ziplistPush(o->ptr, field->ptr, sdslen(field->ptr), ZIPLIST_TAIL);
o->ptr = ziplistPush(o->ptr, value->ptr, sdslen(value->ptr), ZIPLIST_TAIL);
2012-01-03 01:14:10 -05:00
/* Convert to hash table if size threshold is exceeded */
if (sdslen(field->ptr) > server.hash_max_ziplist_value ||
sdslen(value->ptr) > server.hash_max_ziplist_value)
{
2012-03-23 15:21:19 -04:00
decrRefCount(field);
decrRefCount(value);
2012-01-03 01:14:10 -05:00
hashTypeConvert(o, REDIS_ENCODING_HT);
break;
}
2012-03-23 15:21:19 -04:00
decrRefCount(field);
decrRefCount(value);
}
2012-01-03 01:14:10 -05:00
/* Load remaining fields and values into the hash table */
while (o->encoding == REDIS_ENCODING_HT && len > 0) {
2012-01-03 01:14:10 -05:00
robj *field, *value;
len--;
2012-01-03 01:14:10 -05:00
/* Load encoded strings */
field = rdbLoadEncodedStringObject(rdb);
if (field == NULL) return NULL;
value = rdbLoadEncodedStringObject(rdb);
if (value == NULL) return NULL;
field = tryObjectEncoding(field);
value = tryObjectEncoding(value);
/* Add pair to hash table */
ret = dictAdd((dict*)o->ptr, field, value);
redisAssert(ret == DICT_OK);
2012-01-03 01:14:10 -05:00
}
/* All pairs should be read by now */
redisAssert(len == 0);
} else if (rdbtype == REDIS_RDB_TYPE_LIST_QUICKLIST) {
if ((len = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL;
o = createQuicklistObject();
quicklistSetOptions(o->ptr, server.list_max_ziplist_size,
server.list_compress_depth);
2012-01-03 01:14:10 -05:00
while (len--) {
unsigned char *zl = rdbGenericLoadStringObject(rdb,RDB_LOAD_PLAIN);
if (zl == NULL) return NULL;
quicklistAppendZiplist(o->ptr, zl);
}
} else if (rdbtype == REDIS_RDB_TYPE_HASH_ZIPMAP ||
rdbtype == REDIS_RDB_TYPE_LIST_ZIPLIST ||
rdbtype == REDIS_RDB_TYPE_SET_INTSET ||
2012-01-03 01:14:10 -05:00
rdbtype == REDIS_RDB_TYPE_ZSET_ZIPLIST ||
rdbtype == REDIS_RDB_TYPE_HASH_ZIPLIST)
{
unsigned char *encoded = rdbGenericLoadStringObject(rdb,RDB_LOAD_PLAIN);
if (encoded == NULL) return NULL;
o = createObject(REDIS_STRING,encoded); /* Obj type fixed below. */
/* Fix the object encoding, and make sure to convert the encoded
* data type into the base type if accordingly to the current
* configuration there are too many elements in the encoded data
* type. Note that we only check the length and not max element
* size as this is an O(N) scan. Eventually everything will get
* converted. */
switch(rdbtype) {
case REDIS_RDB_TYPE_HASH_ZIPMAP:
2012-01-03 01:14:10 -05:00
/* Convert to ziplist encoded hash. This must be deprecated
* when loading dumps created by Redis 2.4 gets deprecated. */
{
unsigned char *zl = ziplistNew();
unsigned char *zi = zipmapRewind(o->ptr);
unsigned char *fstr, *vstr;
unsigned int flen, vlen;
unsigned int maxlen = 0;
2012-01-03 01:14:10 -05:00
while ((zi = zipmapNext(zi, &fstr, &flen, &vstr, &vlen)) != NULL) {
if (flen > maxlen) maxlen = flen;
if (vlen > maxlen) maxlen = vlen;
2012-01-03 01:14:10 -05:00
zl = ziplistPush(zl, fstr, flen, ZIPLIST_TAIL);
zl = ziplistPush(zl, vstr, vlen, ZIPLIST_TAIL);
}
zfree(o->ptr);
o->ptr = zl;
o->type = REDIS_HASH;
o->encoding = REDIS_ENCODING_ZIPLIST;
if (hashTypeLength(o) > server.hash_max_ziplist_entries ||
maxlen > server.hash_max_ziplist_value)
{
2012-01-03 01:14:10 -05:00
hashTypeConvert(o, REDIS_ENCODING_HT);
}
2012-01-03 01:14:10 -05:00
}
break;
case REDIS_RDB_TYPE_LIST_ZIPLIST:
o->type = REDIS_LIST;
o->encoding = REDIS_ENCODING_ZIPLIST;
listTypeConvert(o,REDIS_ENCODING_QUICKLIST);
break;
case REDIS_RDB_TYPE_SET_INTSET:
o->type = REDIS_SET;
o->encoding = REDIS_ENCODING_INTSET;
if (intsetLen(o->ptr) > server.set_max_intset_entries)
setTypeConvert(o,REDIS_ENCODING_HT);
break;
case REDIS_RDB_TYPE_ZSET_ZIPLIST:
o->type = REDIS_ZSET;
o->encoding = REDIS_ENCODING_ZIPLIST;
if (zsetLength(o) > server.zset_max_ziplist_entries)
zsetConvert(o,REDIS_ENCODING_SKIPLIST);
break;
2012-01-03 01:14:10 -05:00
case REDIS_RDB_TYPE_HASH_ZIPLIST:
o->type = REDIS_HASH;
o->encoding = REDIS_ENCODING_ZIPLIST;
if (hashTypeLength(o) > server.hash_max_ziplist_entries)
hashTypeConvert(o, REDIS_ENCODING_HT);
break;
default:
redisPanic("Unknown encoding");
break;
}
} else {
redisPanic("Unknown object type");
}
return o;
}
/* Mark that we are loading in the global state and setup the fields
* needed to provide loading stats. */
void startLoading(FILE *fp) {
struct stat sb;
/* Load the DB */
server.loading = 1;
server.loading_start_time = time(NULL);
server.loading_loaded_bytes = 0;
if (fstat(fileno(fp), &sb) == -1) {
server.loading_total_bytes = 0;
} else {
server.loading_total_bytes = sb.st_size;
}
}
/* Refresh the loading progress info */
void loadingProgress(off_t pos) {
server.loading_loaded_bytes = pos;
if (server.stat_peak_memory < zmalloc_used_memory())
server.stat_peak_memory = zmalloc_used_memory();
}
/* Loading finished */
void stopLoading(void) {
server.loading = 0;
}
/* Track loading progress in order to serve client's from time to time
and if needed calculate rdb checksum */
void rdbLoadProgressCallback(rio *r, const void *buf, size_t len) {
if (server.rdb_checksum)
rioGenericUpdateChecksum(r, buf, len);
if (server.loading_process_events_interval_bytes &&
Slaves heartbeat while loading RDB files. Starting with Redis 2.8 masters are able to detect timed out slaves, while before 2.8 only slaves were able to detect a timed out master. Now that timeout detection is bi-directional the following problem happens as described "in the field" by issue #1449: 1) Master and slave setup with big dataset. 2) Slave performs the first synchronization, or a full sync after a failed partial resync. 3) Master sends the RDB payload to the slave. 4) Slave loads this payload. 5) Master detects the slave as timed out since does not receive back the REPLCONF ACK acknowledges. Here the problem is that the master has no way to know how much the slave will take to load the RDB file in memory. The obvious solution is to use a greater replication timeout setting, but this is a shame since for the 0.1% of operation time we are forced to use a timeout that is not what is suited for 99.9% of operation time. This commit tries to fix this problem with a solution that is a bit of an hack, but that modifies little of the replication internals, in order to be back ported to 2.8 safely. During the RDB loading time, we send the master newlines to avoid being sensed as timed out. This is the same that the master already does while saving the RDB file to still signal its presence to the slave. The single newline is used because: 1) It can't desync the protocol, as it is only transmitted all or nothing. 2) It can be safely sent while we don't have a client structure for the master or in similar situations just with write(2).
2013-12-09 07:32:44 -05:00
(r->processed_bytes + len)/server.loading_process_events_interval_bytes > r->processed_bytes/server.loading_process_events_interval_bytes)
{
/* The DB can take some non trivial amount of time to load. Update
* our cached time since it is used to create and update the last
* interaction time with clients and for other important things. */
updateCachedTime();
if (server.masterhost && server.repl_state == REDIS_REPL_TRANSFER)
replicationSendNewlineToMaster();
loadingProgress(r->processed_bytes);
processEventsWhileBlocked();
}
}
int rdbLoad(char *filename) {
uint32_t dbid;
int type, rdbver;
redisDb *db = server.db+0;
char buf[1024];
long long expiretime, now = mstime();
FILE *fp;
rio rdb;
if ((fp = fopen(filename,"r")) == NULL) return REDIS_ERR;
rioInitWithFile(&rdb,fp);
rdb.update_cksum = rdbLoadProgressCallback;
rdb.max_processing_chunk = server.loading_process_events_interval_bytes;
2011-05-14 06:15:08 -04:00
if (rioRead(&rdb,buf,9) == 0) goto eoferr;
buf[9] = '\0';
if (memcmp(buf,"REDIS",5) != 0) {
fclose(fp);
redisLog(REDIS_WARNING,"Wrong signature trying to load DB from file");
errno = EINVAL;
return REDIS_ERR;
}
rdbver = atoi(buf+5);
if (rdbver < 1 || rdbver > REDIS_RDB_VERSION) {
fclose(fp);
redisLog(REDIS_WARNING,"Can't handle RDB format version %d",rdbver);
errno = EINVAL;
return REDIS_ERR;
}
startLoading(fp);
while(1) {
robj *key, *val;
expiretime = -1;
/* Read type. */
if ((type = rdbLoadType(&rdb)) == -1) goto eoferr;
/* Handle special types. */
if (type == REDIS_RDB_OPCODE_EXPIRETIME) {
/* EXPIRETIME: load an expire associated with the next key
* to load. Note that after loading an expire we need to
* load the actual type, and continue. */
if ((expiretime = rdbLoadTime(&rdb)) == -1) goto eoferr;
/* We read the time so we need to read the object type again. */
if ((type = rdbLoadType(&rdb)) == -1) goto eoferr;
2011-11-09 15:59:27 -05:00
/* the EXPIRETIME opcode specifies time in seconds, so convert
2013-01-16 12:00:20 -05:00
* into milliseconds. */
expiretime *= 1000;
} else if (type == REDIS_RDB_OPCODE_EXPIRETIME_MS) {
/* EXPIRETIME_MS: milliseconds precision expire times introduced
* with RDB v3. Like EXPIRETIME but no with more precision. */
if ((expiretime = rdbLoadMillisecondTime(&rdb)) == -1) goto eoferr;
/* We read the time so we need to read the object type again. */
if ((type = rdbLoadType(&rdb)) == -1) goto eoferr;
} else if (type == REDIS_RDB_OPCODE_EOF) {
/* EOF: End of file, exit the main loop. */
break;
} else if (type == REDIS_RDB_OPCODE_SELECTDB) {
/* SELECTDB: Select the specified database. */
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);
exit(1);
}
db = server.db+dbid;
continue; /* Read type again. */
} else if (type == REDIS_RDB_OPCODE_RESIZEDB) {
/* RESIZEDB: Hint about the size of the keys in the currently
* selected data base, in order to avoid useless rehashing. */
uint32_t db_size, expires_size;
if ((db_size = rdbLoadLen(&rdb,NULL)) == REDIS_RDB_LENERR)
goto eoferr;
if ((expires_size = rdbLoadLen(&rdb,NULL)) == REDIS_RDB_LENERR)
goto eoferr;
dictExpand(db->dict,db_size);
dictExpand(db->expires,expires_size);
continue; /* Read type again. */
} else if (type == REDIS_RDB_OPCODE_AUX) {
/* AUX: generic string-string fields. Use to add state to RDB
* which is backward compatible. Implementations of RDB loading
* are requierd to skip AUX fields they don't understand.
*
* An AUX field is composed of two strings: key and value. */
robj *auxkey, *auxval;
if ((auxkey = rdbLoadStringObject(&rdb)) == NULL) goto eoferr;
if ((auxval = rdbLoadStringObject(&rdb)) == NULL) goto eoferr;
if (((char*)auxkey->ptr)[0] == '%') {
/* All the fields with a name staring with '%' are considered
* information fields and are logged at startup with a log
* level of NOTICE. */
redisLog(REDIS_NOTICE,"RDB '%s': %s", auxkey->ptr, auxval->ptr);
} else {
/* We ignore fields we don't understand, as by AUX field
* contract. */
redisLog(REDIS_DEBUG,"Unrecognized RDB AUX field: '%s'",
auxkey->ptr);
}
2015-01-08 16:23:48 -05:00
decrRefCount(auxkey);
decrRefCount(auxval);
continue; /* Read type again. */
}
/* Read key */
if ((key = rdbLoadStringObject(&rdb)) == NULL) goto eoferr;
/* Read value */
if ((val = rdbLoadObject(type,&rdb)) == NULL) goto eoferr;
/* Check if the key already expired. This function is used when loading
* an RDB file from disk, either at startup, or when an RDB was
* received from the master. In the latter case, the master is
* responsible for key expiry. If we would expire keys here, the
* snapshot taken by the master may not be reflected on the slave. */
if (server.masterhost == NULL && expiretime != -1 && expiretime < now) {
decrRefCount(key);
decrRefCount(val);
continue;
}
/* Add the new object in the hash table */
dbAdd(db,key,val);
/* Set the expire time if needed */
if (expiretime != -1) setExpire(db,key,expiretime);
decrRefCount(key);
}
/* Verify the checksum if RDB version is >= 5 */
if (rdbver >= 5 && server.rdb_checksum) {
uint64_t cksum, expected = rdb.cksum;
if (rioRead(&rdb,&cksum,8) == 0) goto eoferr;
memrev64ifbe(&cksum);
if (cksum == 0) {
redisLog(REDIS_WARNING,"RDB file was saved with checksum disabled: no check performed.");
} else if (cksum != expected) {
redisLog(REDIS_WARNING,"Wrong RDB checksum. Aborting now.");
exit(1);
}
}
fclose(fp);
stopLoading();
return REDIS_OK;
eoferr: /* unexpected end of file is handled here with a fatal exit */
redisLog(REDIS_WARNING,"Short read or OOM loading DB. Unrecoverable error, aborting now.");
exit(1);
return REDIS_ERR; /* Just to avoid warning */
}
/* A background saving child (BGSAVE) terminated its work. Handle this.
* This function covers the case of actual BGSAVEs. */
void backgroundSaveDoneHandlerDisk(int exitcode, int bysignal) {
if (!bysignal && exitcode == 0) {
redisLog(REDIS_NOTICE,
"Background saving terminated with success");
server.dirty = server.dirty - server.dirty_before_bgsave;
server.lastsave = time(NULL);
server.lastbgsave_status = REDIS_OK;
} else if (!bysignal && exitcode != 0) {
redisLog(REDIS_WARNING, "Background saving error");
server.lastbgsave_status = REDIS_ERR;
} else {
mstime_t latency;
redisLog(REDIS_WARNING,
"Background saving terminated by signal %d", bysignal);
latencyStartMonitor(latency);
2011-12-21 06:22:13 -05:00
rdbRemoveTempFile(server.rdb_child_pid);
latencyEndMonitor(latency);
latencyAddSampleIfNeeded("rdb-unlink-temp-file",latency);
/* SIGUSR1 is whitelisted, so we have a way to kill a child without
* tirggering an error conditon. */
if (bysignal != SIGUSR1)
server.lastbgsave_status = REDIS_ERR;
}
2011-12-21 06:22:13 -05:00
server.rdb_child_pid = -1;
server.rdb_child_type = REDIS_RDB_CHILD_TYPE_NONE;
server.rdb_save_time_last = time(NULL)-server.rdb_save_time_start;
server.rdb_save_time_start = -1;
/* Possibly there are slaves waiting for a BGSAVE in order to be served
* (the first stage of SYNC is a bulk transfer of dump.rdb) */
updateSlavesWaitingBgsave((!bysignal && exitcode == 0) ? REDIS_OK : REDIS_ERR, REDIS_RDB_CHILD_TYPE_DISK);
}
/* A background saving child (BGSAVE) terminated its work. Handle this.
* This function covers the case of RDB -> Salves socket transfers for
* diskless replication. */
void backgroundSaveDoneHandlerSocket(int exitcode, int bysignal) {
uint64_t *ok_slaves;
if (!bysignal && exitcode == 0) {
redisLog(REDIS_NOTICE,
"Background RDB transfer terminated with success");
} else if (!bysignal && exitcode != 0) {
redisLog(REDIS_WARNING, "Background transfer error");
} else {
redisLog(REDIS_WARNING,
"Background transfer terminated by signal %d", bysignal);
}
server.rdb_child_pid = -1;
server.rdb_child_type = REDIS_RDB_CHILD_TYPE_NONE;
server.rdb_save_time_start = -1;
/* If the child returns an OK exit code, read the set of slave client
* IDs and the associated status code. We'll terminate all the slaves
* in error state.
*
* If the process returned an error, consider the list of slaves that
* can continue to be emtpy, so that it's just a special case of the
* normal code path. */
ok_slaves = zmalloc(sizeof(uint64_t)); /* Make space for the count. */
ok_slaves[0] = 0;
if (!bysignal && exitcode == 0) {
int readlen = sizeof(uint64_t);
if (read(server.rdb_pipe_read_result_from_child, ok_slaves, readlen) ==
readlen)
{
readlen = ok_slaves[0]*sizeof(uint64_t)*2;
/* Make space for enough elements as specified by the first
* uint64_t element in the array. */
ok_slaves = zrealloc(ok_slaves,sizeof(uint64_t)+readlen);
if (readlen &&
read(server.rdb_pipe_read_result_from_child, ok_slaves+1,
readlen) != readlen)
{
ok_slaves[0] = 0;
}
}
}
close(server.rdb_pipe_read_result_from_child);
close(server.rdb_pipe_write_result_to_parent);
/* We can continue the replication process with all the slaves that
* correctly received the full payload. Others are terminated. */
listNode *ln;
listIter li;
listRewind(server.slaves,&li);
while((ln = listNext(&li))) {
redisClient *slave = ln->value;
if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) {
uint64_t j;
int errorcode = 0;
/* Search for the slave ID in the reply. In order for a slave to
* continue the replication process, we need to find it in the list,
* and it must have an error code set to 0 (which means success). */
for (j = 0; j < ok_slaves[0]; j++) {
if (slave->id == ok_slaves[2*j+1]) {
errorcode = ok_slaves[2*j+2];
break; /* Found in slaves list. */
}
}
if (j == ok_slaves[0] || errorcode != 0) {
redisLog(REDIS_WARNING,
"Closing slave %s: child->slave RDB transfer failed: %s",
replicationGetSlaveName(slave),
(errorcode == 0) ? "RDB transfer child aborted"
: strerror(errorcode));
freeClient(slave);
} else {
redisLog(REDIS_WARNING,
"Slave %s correctly received the streamed RDB file.",
replicationGetSlaveName(slave));
/* Restore the socket as non-blocking. */
anetNonBlock(NULL,slave->fd);
anetSendTimeout(NULL,slave->fd,0);
}
}
}
zfree(ok_slaves);
updateSlavesWaitingBgsave((!bysignal && exitcode == 0) ? REDIS_OK : REDIS_ERR, REDIS_RDB_CHILD_TYPE_SOCKET);
}
/* When a background RDB saving/transfer terminates, call the right handler. */
void backgroundSaveDoneHandler(int exitcode, int bysignal) {
switch(server.rdb_child_type) {
case REDIS_RDB_CHILD_TYPE_DISK:
backgroundSaveDoneHandlerDisk(exitcode,bysignal);
break;
case REDIS_RDB_CHILD_TYPE_SOCKET:
backgroundSaveDoneHandlerSocket(exitcode,bysignal);
break;
default:
redisPanic("Unknown RDB child type.");
break;
}
}
/* Spawn an RDB child that writes the RDB to the sockets of the slaves
* that are currently in REDIS_REPL_WAIT_BGSAVE_START state. */
int rdbSaveToSlavesSockets(void) {
int *fds;
uint64_t *clientids;
int numfds;
listNode *ln;
listIter li;
pid_t childpid;
long long start;
int pipefds[2];
if (server.rdb_child_pid != -1) return REDIS_ERR;
/* Before to fork, create a pipe that will be used in order to
* send back to the parent the IDs of the slaves that successfully
* received all the writes. */
if (pipe(pipefds) == -1) return REDIS_ERR;
server.rdb_pipe_read_result_from_child = pipefds[0];
server.rdb_pipe_write_result_to_parent = pipefds[1];
/* Collect the file descriptors of the slaves we want to transfer
* the RDB to, which are i WAIT_BGSAVE_START state. */
fds = zmalloc(sizeof(int)*listLength(server.slaves));
/* We also allocate an array of corresponding client IDs. This will
* be useful for the child process in order to build the report
* (sent via unix pipe) that will be sent to the parent. */
clientids = zmalloc(sizeof(uint64_t)*listLength(server.slaves));
numfds = 0;
listRewind(server.slaves,&li);
while((ln = listNext(&li))) {
redisClient *slave = ln->value;
if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) {
clientids[numfds] = slave->id;
fds[numfds++] = slave->fd;
slave->replstate = REDIS_REPL_WAIT_BGSAVE_END;
/* Put the socket in non-blocking mode to simplify RDB transfer.
* We'll restore it when the children returns (since duped socket
* will share the O_NONBLOCK attribute with the parent). */
anetBlock(NULL,slave->fd);
anetSendTimeout(NULL,slave->fd,server.repl_timeout*1000);
}
}
/* Create the child process. */
start = ustime();
if ((childpid = fork()) == 0) {
/* Child */
int retval;
rio slave_sockets;
rioInitWithFdset(&slave_sockets,fds,numfds);
zfree(fds);
closeListeningSockets(0);
redisSetProcTitle("redis-rdb-to-slaves");
retval = rdbSaveRioWithEOFMark(&slave_sockets,NULL);
if (retval == REDIS_OK && rioFlush(&slave_sockets) == 0)
retval = REDIS_ERR;
if (retval == REDIS_OK) {
size_t private_dirty = zmalloc_get_private_dirty();
if (private_dirty) {
redisLog(REDIS_NOTICE,
"RDB: %zu MB of memory used by copy-on-write",
private_dirty/(1024*1024));
}
/* If we are returning OK, at least one slave was served
* with the RDB file as expected, so we need to send a report
* to the parent via the pipe. The format of the message is:
*
* <len> <slave[0].id> <slave[0].error> ...
*
* len, slave IDs, and slave errors, are all uint64_t integers,
* so basically the reply is composed of 64 bits for the len field
* plus 2 additional 64 bit integers for each entry, for a total
* of 'len' entries.
*
* The 'id' represents the slave's client ID, so that the master
* can match the report with a specific slave, and 'error' is
* set to 0 if the replication process terminated with a success
* or the error code if an error occurred. */
void *msg = zmalloc(sizeof(uint64_t)*(1+2*numfds));
uint64_t *len = msg;
uint64_t *ids = len+1;
int j, msglen;
*len = numfds;
for (j = 0; j < numfds; j++) {
*ids++ = clientids[j];
*ids++ = slave_sockets.io.fdset.state[j];
}
/* Write the message to the parent. If we have no good slaves or
* we are unable to transfer the message to the parent, we exit
* with an error so that the parent will abort the replication
* process with all the childre that were waiting. */
msglen = sizeof(uint64_t)*(1+2*numfds);
if (*len == 0 ||
write(server.rdb_pipe_write_result_to_parent,msg,msglen)
!= msglen)
{
retval = REDIS_ERR;
}
zfree(msg);
}
zfree(clientids);
exitFromChild((retval == REDIS_OK) ? 0 : 1);
} else {
/* Parent */
zfree(clientids); /* Not used by parent. Free ASAP. */
server.stat_fork_time = ustime()-start;
server.stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / server.stat_fork_time / (1024*1024*1024); /* GB per second. */
latencyAddSampleIfNeeded("fork",server.stat_fork_time/1000);
if (childpid == -1) {
redisLog(REDIS_WARNING,"Can't save in background: fork: %s",
strerror(errno));
zfree(fds);
close(pipefds[0]);
close(pipefds[1]);
return REDIS_ERR;
}
redisLog(REDIS_NOTICE,"Background RDB transfer started by pid %d",childpid);
server.rdb_save_time_start = time(NULL);
server.rdb_child_pid = childpid;
server.rdb_child_type = REDIS_RDB_CHILD_TYPE_SOCKET;
updateDictResizePolicy();
zfree(fds);
return REDIS_OK;
}
return REDIS_OK; /* unreached */
}
void saveCommand(redisClient *c) {
2011-12-21 06:22:13 -05:00
if (server.rdb_child_pid != -1) {
addReplyError(c,"Background save already in progress");
return;
}
2011-12-21 06:22:13 -05:00
if (rdbSave(server.rdb_filename) == REDIS_OK) {
addReply(c,shared.ok);
} else {
addReply(c,shared.err);
}
}
void bgsaveCommand(redisClient *c) {
2011-12-21 06:22:13 -05:00
if (server.rdb_child_pid != -1) {
addReplyError(c,"Background save already in progress");
2011-12-21 06:17:02 -05:00
} else if (server.aof_child_pid != -1) {
addReplyError(c,"Can't BGSAVE while AOF log rewriting is in progress");
2011-12-21 06:22:13 -05:00
} else if (rdbSaveBackground(server.rdb_filename) == REDIS_OK) {
addReplyStatus(c,"Background saving started");
} else {
addReply(c,shared.err);
}
}