src/rdb.c

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