src/rdb.c

   1 #include "redis.h"
   2 #include "lzf.h"    /* LZF compression library */
   3 #include "zipmap.h"
   4
   5 #include <math.h>
   6 #include <sys/types.h>
   7 #include <sys/time.h>
   8 #include <sys/resource.h>
   9 #include <sys/wait.h>
  10 #include <arpa/inet.h>
  11 #include <sys/stat.h>
  12
  13 static int rdbWriteRaw(rio *rdb, void *p, size_t len) {
  14     if (rdb && rioWrite(rdb,p,len) == 0)
  15         return -1;
  16     return len;
  17 }
  18
  19 int rdbSaveType(rio *rdb, unsigned char type) {
  20     return rdbWriteRaw(rdb,&type,1);
  21 }
  22
  23 int rdbLoadType(rio *rdb) {
  24     unsigned char type;
  25     if (rioRead(rdb,&type,1) == 0) return -1;
  26     return type;
  27 }
  28
  29 int rdbSaveTime(rio *rdb, time_t t) {
  30     int32_t t32 = (int32_t) t;
  31     return rdbWriteRaw(rdb,&t32,4);
  32 }
  33
  34 time_t rdbLoadTime(rio *rdb) {
  35     int32_t t32;
  36     if (rioRead(rdb,&t32,4) == 0) return -1;
  37     return (time_t)t32;
  38 }
  39
  40 int rdbSaveMillisecondTime(rio *rdb, long long t) {
  41     int64_t t64 = (int64_t) t;
  42     return rdbWriteRaw(rdb,&t64,8);
  43 }
  44
  45 long long rdbLoadMillisecondTime(rio *rdb) {
  46     int64_t t64;
  47     if (rioRead(rdb,&t64,8) == 0) return -1;
  48     return (long long)t64;
  49 }
  50
  51 /* Saves an encoded length. The first two bits in the first byte are used to
  52  * hold the encoding type. See the REDIS_RDB_* definitions for more information
  53  * on the types of encoding. */
  54 int rdbSaveLen(rio *rdb, uint32_t len) {
  55     unsigned char buf[2];
  56     size_t nwritten;
  57
  58     if (len < (1<<6)) {
  59         /* Save a 6 bit len */
  60         buf[0] = (len&0xFF)|(REDIS_RDB_6BITLEN<<6);
  61         if (rdbWriteRaw(rdb,buf,1) == -1) return -1;
  62         nwritten = 1;
  63     } else if (len < (1<<14)) {
  64         /* Save a 14 bit len */
  65         buf[0] = ((len>>8)&0xFF)|(REDIS_RDB_14BITLEN<<6);
  66         buf[1] = len&0xFF;
  67         if (rdbWriteRaw(rdb,buf,2) == -1) return -1;
  68         nwritten = 2;
  69     } else {
  70         /* Save a 32 bit len */
  71         buf[0] = (REDIS_RDB_32BITLEN<<6);
  72         if (rdbWriteRaw(rdb,buf,1) == -1) return -1;
  73         len = htonl(len);
  74         if (rdbWriteRaw(rdb,&len,4) == -4) return -1;
  75         nwritten = 1+4;
  76     }
  77     return nwritten;
  78 }
  79
  80 /* Load an encoded length. The "isencoded" argument is set to 1 if the length
  81  * is not actually a length but an "encoding type". See the REDIS_RDB_ENC_*
  82  * definitions in rdb.h for more information. */
  83 uint32_t rdbLoadLen(rio *rdb, int *isencoded) {
  84     unsigned char buf[2];
  85     uint32_t len;
  86     int type;
  87
  88     if (isencoded) *isencoded = 0;
  89     if (rioRead(rdb,buf,1) == 0) return REDIS_RDB_LENERR;
  90     type = (buf[0]&0xC0)>>6;
  91     if (type == REDIS_RDB_ENCVAL) {
  92         /* Read a 6 bit encoding type. */
  93         if (isencoded) *isencoded = 1;
  94         return buf[0]&0x3F;
  95     } else if (type == REDIS_RDB_6BITLEN) {
  96         /* Read a 6 bit len. */
  97         return buf[0]&0x3F;
  98     } else if (type == REDIS_RDB_14BITLEN) {
  99         /* Read a 14 bit len. */
 100         if (rioRead(rdb,buf+1,1) == 0) return REDIS_RDB_LENERR;
 101         return ((buf[0]&0x3F)<<8)|buf[1];
 102     } else {
 103         /* Read a 32 bit len. */
 104         if (rioRead(rdb,&len,4) == 0) return REDIS_RDB_LENERR;
 105         return ntohl(len);
 106     }
 107 }
 108
 109 /* Encodes the "value" argument as integer when it fits in the supported ranges
 110  * for encoded types. If the function successfully encodes the integer, the
 111  * representation is stored in the buffer pointer to by "enc" and the string
 112  * length is returned. Otherwise 0 is returned. */
 113 int rdbEncodeInteger(long long value, unsigned char *enc) {
 114     if (value >= -(1<<7) && value <= (1<<7)-1) {
 115         enc[0] = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_INT8;
 116         enc[1] = value&0xFF;
 117         return 2;
 118     } else if (value >= -(1<<15) && value <= (1<<15)-1) {
 119         enc[0] = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_INT16;
 120         enc[1] = value&0xFF;
 121         enc[2] = (value>>8)&0xFF;
 122         return 3;
 123     } else if (value >= -((long long)1<<31) && value <= ((long long)1<<31)-1) {
 124         enc[0] = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_INT32;
 125         enc[1] = value&0xFF;
 126         enc[2] = (value>>8)&0xFF;
 127         enc[3] = (value>>16)&0xFF;
 128         enc[4] = (value>>24)&0xFF;
 129         return 5;
 130     } else {
 131         return 0;
 132     }
 133 }
 134
 135 /* Loads an integer-encoded object with the specified encoding type "enctype".
 136  * If the "encode" argument is set the function may return an integer-encoded
 137  * string object, otherwise it always returns a raw string object. */
 138 robj *rdbLoadIntegerObject(rio *rdb, int enctype, int encode) {
 139     unsigned char enc[4];
 140     long long val;
 141
 142     if (enctype == REDIS_RDB_ENC_INT8) {
 143         if (rioRead(rdb,enc,1) == 0) return NULL;
 144         val = (signed char)enc[0];
 145     } else if (enctype == REDIS_RDB_ENC_INT16) {
 146         uint16_t v;
 147         if (rioRead(rdb,enc,2) == 0) return NULL;
 148         v = enc[0]|(enc[1]<<8);
 149         val = (int16_t)v;
 150     } else if (enctype == REDIS_RDB_ENC_INT32) {
 151         uint32_t v;
 152         if (rioRead(rdb,enc,4) == 0) return NULL;
 153         v = enc[0]|(enc[1]<<8)|(enc[2]<<16)|(enc[3]<<24);
 154         val = (int32_t)v;
 155     } else {
 156         val = 0; /* anti-warning */
 157         redisPanic("Unknown RDB integer encoding type");
 158     }
 159     if (encode)
 160         return createStringObjectFromLongLong(val);
 161     else
 162         return createObject(REDIS_STRING,sdsfromlonglong(val));
 163 }
 164
 165 /* String objects in the form "2391" "-100" without any space and with a
 166  * range of values that can fit in an 8, 16 or 32 bit signed value can be
 167  * encoded as integers to save space */
 168 int rdbTryIntegerEncoding(char *s, size_t len, unsigned char *enc) {
 169     long long value;
 170     char *endptr, buf[32];
 171
 172     /* Check if it's possible to encode this value as a number */
 173     value = strtoll(s, &endptr, 10);
 174     if (endptr[0] != '\0') return 0;
 175     ll2string(buf,32,value);
 176
 177     /* If the number converted back into a string is not identical
 178      * then it's not possible to encode the string as integer */
 179     if (strlen(buf) != len || memcmp(buf,s,len)) return 0;
 180
 181     return rdbEncodeInteger(value,enc);
 182 }
 183
 184 int rdbSaveLzfStringObject(rio *rdb, unsigned char *s, size_t len) {
 185     size_t comprlen, outlen;
 186     unsigned char byte;
 187     int n, nwritten = 0;
 188     void *out;
 189
 190     /* We require at least four bytes compression for this to be worth it */
 191     if (len <= 4) return 0;
 192     outlen = len-4;
 193     if ((out = zmalloc(outlen+1)) == NULL) return 0;
 194     comprlen = lzf_compress(s, len, out, outlen);
 195     if (comprlen == 0) {
 196         zfree(out);
 197         return 0;
 198     }
 199     /* Data compressed! Let's save it on disk */
 200     byte = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_LZF;
 201     if ((n = rdbWriteRaw(rdb,&byte,1)) == -1) goto writeerr;
 202     nwritten += n;
 203
 204     if ((n = rdbSaveLen(rdb,comprlen)) == -1) goto writeerr;
 205     nwritten += n;
 206
 207     if ((n = rdbSaveLen(rdb,len)) == -1) goto writeerr;
 208     nwritten += n;
 209
 210     if ((n = rdbWriteRaw(rdb,out,comprlen)) == -1) goto writeerr;
 211     nwritten += n;
 212
 213     zfree(out);
 214     return nwritten;
 215
 216 writeerr:
 217     zfree(out);
 218     return -1;
 219 }
 220
 221 robj *rdbLoadLzfStringObject(rio *rdb) {
 222     unsigned int len, clen;
 223     unsigned char *c = NULL;
 224     sds val = NULL;
 225
 226     if ((clen = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL;
 227     if ((len = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL;
 228     if ((c = zmalloc(clen)) == NULL) goto err;
 229     if ((val = sdsnewlen(NULL,len)) == NULL) goto err;
 230     if (rioRead(rdb,c,clen) == 0) goto err;
 231     if (lzf_decompress(c,clen,val,len) == 0) goto err;
 232     zfree(c);
 233     return createObject(REDIS_STRING,val);
 234 err:
 235     zfree(c);
 236     sdsfree(val);
 237     return NULL;
 238 }
 239
 240 /* Save a string objet as [len][data] on disk. If the object is a string
 241  * representation of an integer value we try to save it in a special form */
 242 int rdbSaveRawString(rio *rdb, unsigned char *s, size_t len) {
 243     int enclen;
 244     int n, nwritten = 0;
 245
 246     /* Try integer encoding */
 247     if (len <= 11) {
 248         unsigned char buf[5];
 249         if ((enclen = rdbTryIntegerEncoding((char*)s,len,buf)) > 0) {
 250             if (rdbWriteRaw(rdb,buf,enclen) == -1) return -1;
 251             return enclen;
 252         }
 253     }
 254
 255     /* Try LZF compression - under 20 bytes it's unable to compress even
 256      * aaaaaaaaaaaaaaaaaa so skip it */
 257     if (server.rdb_compression && len > 20) {
 258         n = rdbSaveLzfStringObject(rdb,s,len);
 259         if (n == -1) return -1;
 260         if (n > 0) return n;
 261         /* Return value of 0 means data can't be compressed, save the old way */
 262     }
 263
 264     /* Store verbatim */
 265     if ((n = rdbSaveLen(rdb,len)) == -1) return -1;
 266     nwritten += n;
 267     if (len > 0) {
 268         if (rdbWriteRaw(rdb,s,len) == -1) return -1;
 269         nwritten += len;
 270     }
 271     return nwritten;
 272 }
 273
 274 /* Save a long long value as either an encoded string or a string. */
 275 int rdbSaveLongLongAsStringObject(rio *rdb, long long value) {
 276     unsigned char buf[32];
 277     int n, nwritten = 0;
 278     int enclen = rdbEncodeInteger(value,buf);
 279     if (enclen > 0) {
 280         return rdbWriteRaw(rdb,buf,enclen);
 281     } else {
 282         /* Encode as string */
 283         enclen = ll2string((char*)buf,32,value);
 284         redisAssert(enclen < 32);
 285         if ((n = rdbSaveLen(rdb,enclen)) == -1) return -1;
 286         nwritten += n;
 287         if ((n = rdbWriteRaw(rdb,buf,enclen)) == -1) return -1;
 288         nwritten += n;
 289     }
 290     return nwritten;
 291 }
 292
 293 /* Like rdbSaveStringObjectRaw() but handle encoded objects */
 294 int rdbSaveStringObject(rio *rdb, robj *obj) {
 295     /* Avoid to decode the object, then encode it again, if the
 296      * object is alrady integer encoded. */
 297     if (obj->encoding == REDIS_ENCODING_INT) {
 298         return rdbSaveLongLongAsStringObject(rdb,(long)obj->ptr);
 299     } else {
 300         redisAssertWithInfo(NULL,obj,obj->encoding == REDIS_ENCODING_RAW);
 301         return rdbSaveRawString(rdb,obj->ptr,sdslen(obj->ptr));
 302     }
 303 }
 304
 305 robj *rdbGenericLoadStringObject(rio *rdb, int encode) {
 306     int isencoded;
 307     uint32_t len;
 308     sds val;
 309
 310     len = rdbLoadLen(rdb,&isencoded);
 311     if (isencoded) {
 312         switch(len) {
 313         case REDIS_RDB_ENC_INT8:
 314         case REDIS_RDB_ENC_INT16:
 315         case REDIS_RDB_ENC_INT32:
 316             return rdbLoadIntegerObject(rdb,len,encode);
 317         case REDIS_RDB_ENC_LZF:
 318             return rdbLoadLzfStringObject(rdb);
 319         default:
 320             redisPanic("Unknown RDB encoding type");
 321         }
 322     }
 323
 324     if (len == REDIS_RDB_LENERR) return NULL;
 325     val = sdsnewlen(NULL,len);
 326     if (len && rioRead(rdb,val,len) == 0) {
 327         sdsfree(val);
 328         return NULL;
 329     }
 330     return createObject(REDIS_STRING,val);
 331 }
 332
 333 robj *rdbLoadStringObject(rio *rdb) {
 334     return rdbGenericLoadStringObject(rdb,0);
 335 }
 336
 337 robj *rdbLoadEncodedStringObject(rio *rdb) {
 338     return rdbGenericLoadStringObject(rdb,1);
 339 }
 340
 341 /* Save a double value. Doubles are saved as strings prefixed by an unsigned
 342  * 8 bit integer specifing the length of the representation.
 343  * This 8 bit integer has special values in order to specify the following
 344  * conditions:
 345  * 253: not a number
 346  * 254: + inf
 347  * 255: - inf
 348  */
 349 int rdbSaveDoubleValue(rio *rdb, double val) {
 350     unsigned char buf[128];
 351     int len;
 352
 353     if (isnan(val)) {
 354         buf[0] = 253;
 355         len = 1;
 356     } else if (!isfinite(val)) {
 357         len = 1;
 358         buf[0] = (val < 0) ? 255 : 254;
 359     } else {
 360 #if (DBL_MANT_DIG >= 52) && (LLONG_MAX == 0x7fffffffffffffffLL)
 361         /* Check if the float is in a safe range to be casted into a
 362          * long long. We are assuming that long long is 64 bit here.
 363          * Also we are assuming that there are no implementations around where
 364          * double has precision < 52 bit.
 365          *
 366          * Under this assumptions we test if a double is inside an interval
 367          * where casting to long long is safe. Then using two castings we
 368          * make sure the decimal part is zero. If all this is true we use
 369          * integer printing function that is much faster. */
 370         double min = -4503599627370495; /* (2^52)-1 */
 371         double max = 4503599627370496; /* -(2^52) */
 372         if (val > min && val < max && val == ((double)((long long)val)))
 373             ll2string((char*)buf+1,sizeof(buf),(long long)val);
 374         else
 375 #endif
 376             snprintf((char*)buf+1,sizeof(buf)-1,"%.17g",val);
 377         buf[0] = strlen((char*)buf+1);
 378         len = buf[0]+1;
 379     }
 380     return rdbWriteRaw(rdb,buf,len);
 381 }
 382
 383 /* For information about double serialization check rdbSaveDoubleValue() */
 384 int rdbLoadDoubleValue(rio *rdb, double *val) {
 385     char buf[128];
 386     unsigned char len;
 387
 388     if (rioRead(rdb,&len,1) == 0) return -1;
 389     switch(len) {
 390     case 255: *val = R_NegInf; return 0;
 391     case 254: *val = R_PosInf; return 0;
 392     case 253: *val = R_Nan; return 0;
 393     default:
 394         if (rioRead(rdb,buf,len) == 0) return -1;
 395         buf[len] = '\0';
 396         sscanf(buf, "%lg", val);
 397         return 0;
 398     }
 399 }
 400
 401 /* Save the object type of object "o". */
 402 int rdbSaveObjectType(rio *rdb, robj *o) {
 403     switch (o->type) {
 404     case REDIS_STRING:
 405         return rdbSaveType(rdb,REDIS_RDB_TYPE_STRING);
 406     case REDIS_LIST:
 407         if (o->encoding == REDIS_ENCODING_ZIPLIST)
 408             return rdbSaveType(rdb,REDIS_RDB_TYPE_LIST_ZIPLIST);
 409         else if (o->encoding == REDIS_ENCODING_LINKEDLIST)
 410             return rdbSaveType(rdb,REDIS_RDB_TYPE_LIST);
 411         else
 412             redisPanic("Unknown list encoding");
 413     case REDIS_SET:
 414         if (o->encoding == REDIS_ENCODING_INTSET)
 415             return rdbSaveType(rdb,REDIS_RDB_TYPE_SET_INTSET);
 416         else if (o->encoding == REDIS_ENCODING_HT)
 417             return rdbSaveType(rdb,REDIS_RDB_TYPE_SET);
 418         else
 419             redisPanic("Unknown set encoding");
 420     case REDIS_ZSET:
 421         if (o->encoding == REDIS_ENCODING_ZIPLIST)
 422             return rdbSaveType(rdb,REDIS_RDB_TYPE_ZSET_ZIPLIST);
 423         else if (o->encoding == REDIS_ENCODING_SKIPLIST)
 424             return rdbSaveType(rdb,REDIS_RDB_TYPE_ZSET);
 425         else
 426             redisPanic("Unknown sorted set encoding");
 427     case REDIS_HASH:
 428         if (o->encoding == REDIS_ENCODING_ZIPLIST)
 429             return rdbSaveType(rdb,REDIS_RDB_TYPE_HASH_ZIPLIST);
 430         else if (o->encoding == REDIS_ENCODING_HT)
 431             return rdbSaveType(rdb,REDIS_RDB_TYPE_HASH);
 432         else
 433             redisPanic("Unknown hash encoding");
 434     default:
 435         redisPanic("Unknown object type");
 436     }
 437     return -1; /* avoid warning */
 438 }
 439
 440 /* Load object type. Return -1 when the byte doesn't contain an object type. */
 441 int rdbLoadObjectType(rio *rdb) {
 442     int type;
 443     if ((type = rdbLoadType(rdb)) == -1) return -1;
 444     if (!rdbIsObjectType(type)) return -1;
 445     return type;
 446 }
 447
 448 /* Save a Redis object. Returns -1 on error, 0 on success. */
 449 int rdbSaveObject(rio *rdb, robj *o) {
 450     int n, nwritten = 0;
 451
 452     if (o->type == REDIS_STRING) {
 453         /* Save a string value */
 454         if ((n = rdbSaveStringObject(rdb,o)) == -1) return -1;
 455         nwritten += n;
 456     } else if (o->type == REDIS_LIST) {
 457         /* Save a list value */
 458         if (o->encoding == REDIS_ENCODING_ZIPLIST) {
 459             size_t l = ziplistBlobLen((unsigned char*)o->ptr);
 460
 461             if ((n = rdbSaveRawString(rdb,o->ptr,l)) == -1) return -1;
 462             nwritten += n;
 463         } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) {
 464             list *list = o->ptr;
 465             listIter li;
 466             listNode *ln;
 467
 468             if ((n = rdbSaveLen(rdb,listLength(list))) == -1) return -1;
 469             nwritten += n;
 470
 471             listRewind(list,&li);
 472             while((ln = listNext(&li))) {
 473                 robj *eleobj = listNodeValue(ln);
 474                 if ((n = rdbSaveStringObject(rdb,eleobj)) == -1) return -1;
 475                 nwritten += n;
 476             }
 477         } else {
 478             redisPanic("Unknown list encoding");
 479         }
 480     } else if (o->type == REDIS_SET) {
 481         /* Save a set value */
 482         if (o->encoding == REDIS_ENCODING_HT) {
 483             dict *set = o->ptr;
 484             dictIterator *di = dictGetIterator(set);
 485             dictEntry *de;
 486
 487             if ((n = rdbSaveLen(rdb,dictSize(set))) == -1) return -1;
 488             nwritten += n;
 489
 490             while((de = dictNext(di)) != NULL) {
 491                 robj *eleobj = dictGetKey(de);
 492                 if ((n = rdbSaveStringObject(rdb,eleobj)) == -1) return -1;
 493                 nwritten += n;
 494             }
 495             dictReleaseIterator(di);
 496         } else if (o->encoding == REDIS_ENCODING_INTSET) {
 497             size_t l = intsetBlobLen((intset*)o->ptr);
 498
 499             if ((n = rdbSaveRawString(rdb,o->ptr,l)) == -1) return -1;
 500             nwritten += n;
 501         } else {
 502             redisPanic("Unknown set encoding");
 503         }
 504     } else if (o->type == REDIS_ZSET) {
 505         /* Save a sorted set value */
 506         if (o->encoding == REDIS_ENCODING_ZIPLIST) {
 507             size_t l = ziplistBlobLen((unsigned char*)o->ptr);
 508
 509             if ((n = rdbSaveRawString(rdb,o->ptr,l)) == -1) return -1;
 510             nwritten += n;
 511         } else if (o->encoding == REDIS_ENCODING_SKIPLIST) {
 512             zset *zs = o->ptr;
 513             dictIterator *di = dictGetIterator(zs->dict);
 514             dictEntry *de;
 515
 516             if ((n = rdbSaveLen(rdb,dictSize(zs->dict))) == -1) return -1;
 517             nwritten += n;
 518
 519             while((de = dictNext(di)) != NULL) {
 520                 robj *eleobj = dictGetKey(de);
 521                 double *score = dictGetVal(de);
 522
 523                 if ((n = rdbSaveStringObject(rdb,eleobj)) == -1) return -1;
 524                 nwritten += n;
 525                 if ((n = rdbSaveDoubleValue(rdb,*score)) == -1) return -1;
 526                 nwritten += n;
 527             }
 528             dictReleaseIterator(di);
 529         } else {
 530             redisPanic("Unknown sorted set encoding");
 531         }
 532     } else if (o->type == REDIS_HASH) {
 533         /* Save a hash value */
 534         if (o->encoding == REDIS_ENCODING_ZIPLIST) {
 535             size_t l = ziplistBlobLen((unsigned char*)o->ptr);
 536
 537             if ((n = rdbSaveRawString(rdb,o->ptr,l)) == -1) return -1;
 538             nwritten += n;
 539
 540         } else if (o->encoding == REDIS_ENCODING_HT) {
 541             dictIterator *di = dictGetIterator(o->ptr);
 542             dictEntry *de;
 543
 544             if ((n = rdbSaveLen(rdb,dictSize((dict*)o->ptr))) == -1) return -1;
 545             nwritten += n;
 546
 547             while((de = dictNext(di)) != NULL) {
 548                 robj *key = dictGetKey(de);
 549                 robj *val = dictGetVal(de);
 550
 551                 if ((n = rdbSaveStringObject(rdb,key)) == -1) return -1;
 552                 nwritten += n;
 553                 if ((n = rdbSaveStringObject(rdb,val)) == -1) return -1;
 554                 nwritten += n;
 555             }
 556             dictReleaseIterator(di);
 557
 558         } else {
 559             redisPanic("Unknown hash encoding");
 560         }
 561
 562     } else {
 563         redisPanic("Unknown object type");
 564     }
 565     return nwritten;
 566 }
 567
 568 /* Return the length the object will have on disk if saved with
 569  * the rdbSaveObject() function. Currently we use a trick to get
 570  * this length with very little changes to the code. In the future
 571  * we could switch to a faster solution. */
 572 off_t rdbSavedObjectLen(robj *o) {
 573     int len = rdbSaveObject(NULL,o);
 574     redisAssertWithInfo(NULL,o,len != -1);
 575     return len;
 576 }
 577
 578 /* Save a key-value pair, with expire time, type, key, value.
 579  * On error -1 is returned.
 580  * On success if the key was actaully saved 1 is returned, otherwise 0
 581  * is returned (the key was already expired). */
 582 int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val,
 583                         long long expiretime, long long now)
 584 {
 585     /* Save the expire time */
 586     if (expiretime != -1) {
 587         /* If this key is already expired skip it */
 588         if (expiretime < now) return 0;
 589         if (rdbSaveType(rdb,REDIS_RDB_OPCODE_EXPIRETIME_MS) == -1) return -1;
 590         if (rdbSaveMillisecondTime(rdb,expiretime) == -1) return -1;
 591     }
 592
 593     /* Save type, key, value */
 594     if (rdbSaveObjectType(rdb,val) == -1) return -1;
 595     if (rdbSaveStringObject(rdb,key) == -1) return -1;
 596     if (rdbSaveObject(rdb,val) == -1) return -1;
 597     return 1;
 598 }
 599
 600 /* Save the DB on disk. Return REDIS_ERR on error, REDIS_OK on success */
 601 int rdbSave(char *filename) {
 602     dictIterator *di = NULL;
 603     dictEntry *de;
 604     char tmpfile[256];
 605     char magic[10];
 606     int j;
 607     long long now = mstime();
 608     FILE *fp;
 609     rio rdb;
 610
 611     snprintf(tmpfile,256,"temp-%d.rdb", (int) getpid());
 612     fp = fopen(tmpfile,"w");
 613     if (!fp) {
 614         redisLog(REDIS_WARNING, "Failed opening .rdb for saving: %s",
 615             strerror(errno));
 616         return REDIS_ERR;
 617     }
 618
 619     rioInitWithFile(&rdb,fp);
 620     snprintf(magic,sizeof(magic),"REDIS%04d",REDIS_RDB_VERSION);
 621     if (rdbWriteRaw(&rdb,magic,9) == -1) goto werr;
 622
 623     for (j = 0; j < server.dbnum; j++) {
 624         redisDb *db = server.db+j;
 625         dict *d = db->dict;
 626         if (dictSize(d) == 0) continue;
 627         di = dictGetSafeIterator(d);
 628         if (!di) {
 629             fclose(fp);
 630             return REDIS_ERR;
 631         }
 632
 633         /* Write the SELECT DB opcode */
 634         if (rdbSaveType(&rdb,REDIS_RDB_OPCODE_SELECTDB) == -1) goto werr;
 635         if (rdbSaveLen(&rdb,j) == -1) goto werr;
 636
 637         /* Iterate this DB writing every entry */
 638         while((de = dictNext(di)) != NULL) {
 639             sds keystr = dictGetKey(de);
 640             robj key, *o = dictGetVal(de);
 641             long long expire;
 642
 643             initStaticStringObject(key,keystr);
 644             expire = getExpire(db,&key);
 645             if (rdbSaveKeyValuePair(&rdb,&key,o,expire,now) == -1) goto werr;
 646         }
 647         dictReleaseIterator(di);
 648     }
 649     /* EOF opcode */
 650     if (rdbSaveType(&rdb,REDIS_RDB_OPCODE_EOF) == -1) goto werr;
 651
 652     /* Make sure data will not remain on the OS's output buffers */
 653     fflush(fp);
 654     fsync(fileno(fp));
 655     fclose(fp);
 656
 657     /* Use RENAME to make sure the DB file is changed atomically only
 658      * if the generate DB file is ok. */
 659     if (rename(tmpfile,filename) == -1) {
 660         redisLog(REDIS_WARNING,"Error moving temp DB file on the final destination: %s", strerror(errno));
 661         unlink(tmpfile);
 662         return REDIS_ERR;
 663     }
 664     redisLog(REDIS_NOTICE,"DB saved on disk");
 665     server.dirty = 0;
 666     server.lastsave = time(NULL);
 667     server.lastbgsave_status = REDIS_OK;
 668     return REDIS_OK;
 669
 670 werr:
 671     fclose(fp);
 672     unlink(tmpfile);
 673     redisLog(REDIS_WARNING,"Write error saving DB on disk: %s", strerror(errno));
 674     if (di) dictReleaseIterator(di);
 675     return REDIS_ERR;
 676 }
 677
 678 int rdbSaveBackground(char *filename) {
 679     pid_t childpid;
 680     long long start;
 681
 682     if (server.rdb_child_pid != -1) return REDIS_ERR;
 683
 684     server.dirty_before_bgsave = server.dirty;
 685
 686     start = ustime();
 687     if ((childpid = fork()) == 0) {
 688         int retval;
 689
 690         /* Child */
 691         if (server.ipfd > 0) close(server.ipfd);
 692         if (server.sofd > 0) close(server.sofd);
 693         retval = rdbSave(filename);
 694         _exit((retval == REDIS_OK) ? 0 : 1);
 695     } else {
 696         /* Parent */
 697         server.stat_fork_time = ustime()-start;
 698         if (childpid == -1) {
 699             redisLog(REDIS_WARNING,"Can't save in background: fork: %s",
 700                 strerror(errno));
 701             return REDIS_ERR;
 702         }
 703         redisLog(REDIS_NOTICE,"Background saving started by pid %d",childpid);
 704         server.rdb_child_pid = childpid;
 705         updateDictResizePolicy();
 706         return REDIS_OK;
 707     }
 708     return REDIS_OK; /* unreached */
 709 }
 710
 711 void rdbRemoveTempFile(pid_t childpid) {
 712     char tmpfile[256];
 713
 714     snprintf(tmpfile,256,"temp-%d.rdb", (int) childpid);
 715     unlink(tmpfile);
 716 }
 717
 718 /* Load a Redis object of the specified type from the specified file.
 719  * On success a newly allocated object is returned, otherwise NULL. */
 720 robj *rdbLoadObject(int rdbtype, rio *rdb) {
 721     robj *o, *ele, *dec;
 722     size_t len;
 723     unsigned int i;
 724
 725     redisLog(REDIS_DEBUG,"LOADING OBJECT %d (at %d)\n",rdbtype,rdb->tell(rdb));
 726     if (rdbtype == REDIS_RDB_TYPE_STRING) {
 727         /* Read string value */
 728         if ((o = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL;
 729         o = tryObjectEncoding(o);
 730     } else if (rdbtype == REDIS_RDB_TYPE_LIST) {
 731         /* Read list value */
 732         if ((len = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL;
 733
 734         /* Use a real list when there are too many entries */
 735         if (len > server.list_max_ziplist_entries) {
 736             o = createListObject();
 737         } else {
 738             o = createZiplistObject();
 739         }
 740
 741         /* Load every single element of the list */
 742         while(len--) {
 743             if ((ele = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL;
 744
 745             /* If we are using a ziplist and the value is too big, convert
 746              * the object to a real list. */
 747             if (o->encoding == REDIS_ENCODING_ZIPLIST &&
 748                 ele->encoding == REDIS_ENCODING_RAW &&
 749                 sdslen(ele->ptr) > server.list_max_ziplist_value)
 750                     listTypeConvert(o,REDIS_ENCODING_LINKEDLIST);
 751
 752             if (o->encoding == REDIS_ENCODING_ZIPLIST) {
 753                 dec = getDecodedObject(ele);
 754                 o->ptr = ziplistPush(o->ptr,dec->ptr,sdslen(dec->ptr),REDIS_TAIL);
 755                 decrRefCount(dec);
 756                 decrRefCount(ele);
 757             } else {
 758                 ele = tryObjectEncoding(ele);
 759                 listAddNodeTail(o->ptr,ele);
 760             }
 761         }
 762     } else if (rdbtype == REDIS_RDB_TYPE_SET) {
 763         /* Read list/set value */
 764         if ((len = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL;
 765
 766         /* Use a regular set when there are too many entries. */
 767         if (len > server.set_max_intset_entries) {
 768             o = createSetObject();
 769             /* It's faster to expand the dict to the right size asap in order
 770              * to avoid rehashing */
 771             if (len > DICT_HT_INITIAL_SIZE)
 772                 dictExpand(o->ptr,len);
 773         } else {
 774             o = createIntsetObject();
 775         }
 776
 777         /* Load every single element of the list/set */
 778         for (i = 0; i < len; i++) {
 779             long long llval;
 780             if ((ele = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL;
 781             ele = tryObjectEncoding(ele);
 782
 783             if (o->encoding == REDIS_ENCODING_INTSET) {
 784                 /* Fetch integer value from element */
 785                 if (isObjectRepresentableAsLongLong(ele,&llval) == REDIS_OK) {
 786                     o->ptr = intsetAdd(o->ptr,llval,NULL);
 787                 } else {
 788                     setTypeConvert(o,REDIS_ENCODING_HT);
 789                     dictExpand(o->ptr,len);
 790                 }
 791             }
 792
 793             /* This will also be called when the set was just converted
 794              * to regular hashtable encoded set */
 795             if (o->encoding == REDIS_ENCODING_HT) {
 796                 dictAdd((dict*)o->ptr,ele,NULL);
 797             } else {
 798                 decrRefCount(ele);
 799             }
 800         }
 801     } else if (rdbtype == REDIS_RDB_TYPE_ZSET) {
 802         /* Read list/set value */
 803         size_t zsetlen;
 804         size_t maxelelen = 0;
 805         zset *zs;
 806
 807         if ((zsetlen = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL;
 808         o = createZsetObject();
 809         zs = o->ptr;
 810
 811         /* Load every single element of the list/set */
 812         while(zsetlen--) {
 813             robj *ele;
 814             double score;
 815             zskiplistNode *znode;
 816
 817             if ((ele = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL;
 818             ele = tryObjectEncoding(ele);
 819             if (rdbLoadDoubleValue(rdb,&score) == -1) return NULL;
 820
 821             /* Don't care about integer-encoded strings. */
 822             if (ele->encoding == REDIS_ENCODING_RAW &&
 823                 sdslen(ele->ptr) > maxelelen)
 824                     maxelelen = sdslen(ele->ptr);
 825
 826             znode = zslInsert(zs->zsl,score,ele);
 827             dictAdd(zs->dict,ele,&znode->score);
 828             incrRefCount(ele); /* added to skiplist */
 829         }
 830
 831         /* Convert *after* loading, since sorted sets are not stored ordered. */
 832         if (zsetLength(o) <= server.zset_max_ziplist_entries &&
 833             maxelelen <= server.zset_max_ziplist_value)
 834                 zsetConvert(o,REDIS_ENCODING_ZIPLIST);
 835     } else if (rdbtype == REDIS_RDB_TYPE_HASH) {
 836         size_t len;
 837         int ret;
 838
 839         len = rdbLoadLen(rdb, NULL);
 840         if (len == REDIS_RDB_LENERR) return NULL;
 841
 842         o = createHashObject();
 843
 844         /* Too many entries? Use an hash table. */
 845         if (len > server.hash_max_ziplist_entries)
 846             hashTypeConvert(o, REDIS_ENCODING_HT);
 847
 848         /* Load every field and value into the ziplist */
 849         while (o->encoding == REDIS_ENCODING_ZIPLIST && len > 0) {
 850             robj *field, *value;
 851
 852             len--;
 853             /* Load raw strings */
 854             field = rdbLoadStringObject(rdb);
 855             if (field == NULL) return NULL;
 856             redisAssert(field->encoding == REDIS_ENCODING_RAW);
 857             value = rdbLoadStringObject(rdb);
 858             if (value == NULL) return NULL;
 859             redisAssert(field->encoding == REDIS_ENCODING_RAW);
 860
 861             /* Add pair to ziplist */
 862             o->ptr = ziplistPush(o->ptr, field->ptr, sdslen(field->ptr), ZIPLIST_TAIL);
 863             o->ptr = ziplistPush(o->ptr, value->ptr, sdslen(value->ptr), ZIPLIST_TAIL);
 864             /* Convert to hash table if size threshold is exceeded */
 865             if (sdslen(field->ptr) > server.hash_max_ziplist_value ||
 866                 sdslen(value->ptr) > server.hash_max_ziplist_value)
 867             {
 868                 decrRefCount(field);
 869                 decrRefCount(value);
 870                 hashTypeConvert(o, REDIS_ENCODING_HT);
 871                 break;
 872             }
 873             decrRefCount(field);
 874             decrRefCount(value);
 875         }
 876
 877         /* Load remaining fields and values into the hash table */
 878         while (o->encoding == REDIS_ENCODING_HT && len > 0) {
 879             robj *field, *value;
 880
 881             len--;
 882             /* Load encoded strings */
 883             field = rdbLoadEncodedStringObject(rdb);
 884             if (field == NULL) return NULL;
 885             value = rdbLoadEncodedStringObject(rdb);
 886             if (value == NULL) return NULL;
 887
 888             field = tryObjectEncoding(field);
 889             value = tryObjectEncoding(value);
 890
 891             /* Add pair to hash table */
 892             ret = dictAdd((dict*)o->ptr, field, value);
 893             redisAssert(ret == REDIS_OK);
 894         }
 895
 896         /* All pairs should be read by now */
 897         redisAssert(len == 0);
 898
 899     } else if (rdbtype == REDIS_RDB_TYPE_HASH_ZIPMAP  ||
 900                rdbtype == REDIS_RDB_TYPE_LIST_ZIPLIST ||
 901                rdbtype == REDIS_RDB_TYPE_SET_INTSET   ||
 902                rdbtype == REDIS_RDB_TYPE_ZSET_ZIPLIST ||
 903                rdbtype == REDIS_RDB_TYPE_HASH_ZIPLIST)
 904     {
 905         robj *aux = rdbLoadStringObject(rdb);
 906
 907         if (aux == NULL) return NULL;
 908         o = createObject(REDIS_STRING,NULL); /* string is just placeholder */
 909         o->ptr = zmalloc(sdslen(aux->ptr));
 910         memcpy(o->ptr,aux->ptr,sdslen(aux->ptr));
 911         decrRefCount(aux);
 912
 913         /* Fix the object encoding, and make sure to convert the encoded
 914          * data type into the base type if accordingly to the current
 915          * configuration there are too many elements in the encoded data
 916          * type. Note that we only check the length and not max element
 917          * size as this is an O(N) scan. Eventually everything will get
 918          * converted. */
 919         switch(rdbtype) {
 920             case REDIS_RDB_TYPE_HASH_ZIPMAP:
 921                 /* Convert to ziplist encoded hash. This must be deprecated
 922                  * when loading dumps created by Redis 2.4 gets deprecated. */
 923                 {
 924                     unsigned char *zl = ziplistNew();
 925                     unsigned char *zi = zipmapRewind(o->ptr);
 926                     unsigned char *fstr, *vstr;
 927                     unsigned int flen, vlen;
 928                     unsigned int maxlen = 0;
 929
 930                     while ((zi = zipmapNext(zi, &fstr, &flen, &vstr, &vlen)) != NULL) {
 931                         if (flen > maxlen) maxlen = flen;
 932                         if (vlen > maxlen) maxlen = vlen;
 933                         zl = ziplistPush(zl, fstr, flen, ZIPLIST_TAIL);
 934                         zl = ziplistPush(zl, vstr, vlen, ZIPLIST_TAIL);
 935                     }
 936
 937                     zfree(o->ptr);
 938                     o->ptr = zl;
 939                     o->type = REDIS_HASH;
 940                     o->encoding = REDIS_ENCODING_ZIPLIST;
 941
 942                     if (hashTypeLength(o) > server.hash_max_ziplist_entries ||
 943                         maxlen > server.hash_max_ziplist_value)
 944                     {
 945                         hashTypeConvert(o, REDIS_ENCODING_HT);
 946                     }
 947                 }
 948                 break;
 949             case REDIS_RDB_TYPE_LIST_ZIPLIST:
 950                 o->type = REDIS_LIST;
 951                 o->encoding = REDIS_ENCODING_ZIPLIST;
 952                 if (ziplistLen(o->ptr) > server.list_max_ziplist_entries)
 953                     listTypeConvert(o,REDIS_ENCODING_LINKEDLIST);
 954                 break;
 955             case REDIS_RDB_TYPE_SET_INTSET:
 956                 o->type = REDIS_SET;
 957                 o->encoding = REDIS_ENCODING_INTSET;
 958                 if (intsetLen(o->ptr) > server.set_max_intset_entries)
 959                     setTypeConvert(o,REDIS_ENCODING_HT);
 960                 break;
 961             case REDIS_RDB_TYPE_ZSET_ZIPLIST:
 962                 o->type = REDIS_ZSET;
 963                 o->encoding = REDIS_ENCODING_ZIPLIST;
 964                 if (zsetLength(o) > server.zset_max_ziplist_entries)
 965                     zsetConvert(o,REDIS_ENCODING_SKIPLIST);
 966                 break;
 967             case REDIS_RDB_TYPE_HASH_ZIPLIST:
 968                 o->type = REDIS_HASH;
 969                 o->encoding = REDIS_ENCODING_ZIPLIST;
 970                 if (hashTypeLength(o) > server.hash_max_ziplist_entries)
 971                     hashTypeConvert(o, REDIS_ENCODING_HT);
 972                 break;
 973             default:
 974                 redisPanic("Unknown encoding");
 975                 break;
 976         }
 977     } else {
 978         redisPanic("Unknown object type");
 979     }
 980     return o;
 981 }
 982
 983 /* Mark that we are loading in the global state and setup the fields
 984  * needed to provide loading stats. */
 985 void startLoading(FILE *fp) {
 986     struct stat sb;
 987
 988     /* Load the DB */
 989     server.loading = 1;
 990     server.loading_start_time = time(NULL);
 991     if (fstat(fileno(fp), &sb) == -1) {
 992         server.loading_total_bytes = 1; /* just to avoid division by zero */
 993     } else {
 994         server.loading_total_bytes = sb.st_size;
 995     }
 996 }
 997
 998 /* Refresh the loading progress info */
 999 void loadingProgress(off_t pos) {
1000     server.loading_loaded_bytes = pos;
1001 }
1002
1003 /* Loading finished */
1004 void stopLoading(void) {
1005     server.loading = 0;
1006 }
1007
1008 int rdbLoad(char *filename) {
1009     uint32_t dbid;
1010     int type, rdbver;
1011     redisDb *db = server.db+0;
1012     char buf[1024];
1013     long long expiretime, now = mstime();
1014     long loops = 0;
1015     FILE *fp;
1016     rio rdb;
1017
1018     fp = fopen(filename,"r");
1019     if (!fp) {
1020         errno = ENOENT;
1021         return REDIS_ERR;
1022     }
1023     rioInitWithFile(&rdb,fp);
1024     if (rioRead(&rdb,buf,9) == 0) goto eoferr;
1025     buf[9] = '\0';
1026     if (memcmp(buf,"REDIS",5) != 0) {
1027         fclose(fp);
1028         redisLog(REDIS_WARNING,"Wrong signature trying to load DB from file");
1029         errno = EINVAL;
1030         return REDIS_ERR;
1031     }
1032     rdbver = atoi(buf+5);
1033     if (rdbver < 1 || rdbver > 4) {
1034         fclose(fp);
1035         redisLog(REDIS_WARNING,"Can't handle RDB format version %d",rdbver);
1036         errno = EINVAL;
1037         return REDIS_ERR;
1038     }
1039
1040     startLoading(fp);
1041     while(1) {
1042         robj *key, *val;
1043         expiretime = -1;
1044
1045         /* Serve the clients from time to time */
1046         if (!(loops++ % 1000)) {
1047             loadingProgress(rdb.tell(&rdb));
1048             aeProcessEvents(server.el, AE_FILE_EVENTS|AE_DONT_WAIT);
1049         }
1050
1051         /* Read type. */
1052         if ((type = rdbLoadType(&rdb)) == -1) goto eoferr;
1053         if (type == REDIS_RDB_OPCODE_EXPIRETIME) {
1054             if ((expiretime = rdbLoadTime(&rdb)) == -1) goto eoferr;
1055             /* We read the time so we need to read the object type again. */
1056             if ((type = rdbLoadType(&rdb)) == -1) goto eoferr;
1057             /* the EXPIRETIME opcode specifies time in seconds, so convert
1058              * into milliesconds. */
1059             expiretime *= 1000;
1060         } else if (type == REDIS_RDB_OPCODE_EXPIRETIME_MS) {
1061             /* Milliseconds precision expire times introduced with RDB
1062              * version 3. */
1063             if ((expiretime = rdbLoadMillisecondTime(&rdb)) == -1) goto eoferr;
1064             /* We read the time so we need to read the object type again. */
1065             if ((type = rdbLoadType(&rdb)) == -1) goto eoferr;
1066         }
1067
1068         if (type == REDIS_RDB_OPCODE_EOF)
1069             break;
1070
1071         /* Handle SELECT DB opcode as a special case */
1072         if (type == REDIS_RDB_OPCODE_SELECTDB) {
1073             if ((dbid = rdbLoadLen(&rdb,NULL)) == REDIS_RDB_LENERR)
1074                 goto eoferr;
1075             if (dbid >= (unsigned)server.dbnum) {
1076                 redisLog(REDIS_WARNING,"FATAL: Data file was created with a Redis server configured to handle more than %d databases. Exiting\n", server.dbnum);
1077                 exit(1);
1078             }
1079             db = server.db+dbid;
1080             continue;
1081         }
1082         /* Read key */
1083         if ((key = rdbLoadStringObject(&rdb)) == NULL) goto eoferr;
1084         /* Read value */
1085         if ((val = rdbLoadObject(type,&rdb)) == NULL) goto eoferr;
1086         /* Check if the key already expired. This function is used when loading
1087          * an RDB file from disk, either at startup, or when an RDB was
1088          * received from the master. In the latter case, the master is
1089          * responsible for key expiry. If we would expire keys here, the
1090          * snapshot taken by the master may not be reflected on the slave. */
1091         if (server.masterhost == NULL && expiretime != -1 && expiretime < now) {
1092             decrRefCount(key);
1093             decrRefCount(val);
1094             continue;
1095         }
1096         /* Add the new object in the hash table */
1097         dbAdd(db,key,val);
1098
1099         /* Set the expire time if needed */
1100         if (expiretime != -1) setExpire(db,key,expiretime);
1101
1102         decrRefCount(key);
1103     }
1104     fclose(fp);
1105     stopLoading();
1106     return REDIS_OK;
1107
1108 eoferr: /* unexpected end of file is handled here with a fatal exit */
1109     redisLog(REDIS_WARNING,"Short read or OOM loading DB. Unrecoverable error, aborting now.");
1110     exit(1);
1111     return REDIS_ERR; /* Just to avoid warning */
1112 }
1113
1114 /* A background saving child (BGSAVE) terminated its work. Handle this. */
1115 void backgroundSaveDoneHandler(int exitcode, int bysignal) {
1116     if (!bysignal && exitcode == 0) {
1117         redisLog(REDIS_NOTICE,
1118             "Background saving terminated with success");
1119         server.dirty = server.dirty - server.dirty_before_bgsave;
1120         server.lastsave = time(NULL);
1121         server.lastbgsave_status = REDIS_OK;
1122     } else if (!bysignal && exitcode != 0) {
1123         redisLog(REDIS_WARNING, "Background saving error");
1124         server.lastbgsave_status = REDIS_ERR;
1125     } else {
1126         redisLog(REDIS_WARNING,
1127             "Background saving terminated by signal %d", bysignal);
1128         rdbRemoveTempFile(server.rdb_child_pid);
1129         server.lastbgsave_status = REDIS_ERR;
1130     }
1131     server.rdb_child_pid = -1;
1132     /* Possibly there are slaves waiting for a BGSAVE in order to be served
1133      * (the first stage of SYNC is a bulk transfer of dump.rdb) */
1134     updateSlavesWaitingBgsave(exitcode == 0 ? REDIS_OK : REDIS_ERR);
1135 }
1136
1137 void saveCommand(redisClient *c) {
1138     if (server.rdb_child_pid != -1) {
1139         addReplyError(c,"Background save already in progress");
1140         return;
1141     }
1142     if (rdbSave(server.rdb_filename) == REDIS_OK) {
1143         addReply(c,shared.ok);
1144     } else {
1145         addReply(c,shared.err);
1146     }
1147 }
1148
1149 void bgsaveCommand(redisClient *c) {
1150     if (server.rdb_child_pid != -1) {
1151         addReplyError(c,"Background save already in progress");
1152     } else if (server.aof_child_pid != -1) {
1153         addReplyError(c,"Can't BGSAVE while AOF log rewriting is in progress");
1154     } else if (rdbSaveBackground(server.rdb_filename) == REDIS_OK) {
1155         addReplyStatus(c,"Background saving started");
1156     } else {
1157         addReply(c,shared.err);
1158     }
1159 }