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_save_time_start = time(NULL);
 712         server.rdb_child_pid = childpid;
 713         updateDictResizePolicy();
 714         return REDIS_OK;
 715     }
 716     return REDIS_OK; /* unreached */
 717 }
 718
 719 void rdbRemoveTempFile(pid_t childpid) {
 720     char tmpfile[256];
 721
 722     snprintf(tmpfile,256,"temp-%d.rdb", (int) childpid);
 723     unlink(tmpfile);
 724 }
 725
 726 /* Load a Redis object of the specified type from the specified file.
 727  * On success a newly allocated object is returned, otherwise NULL. */
 728 robj *rdbLoadObject(int rdbtype, rio *rdb) {
 729     robj *o, *ele, *dec;
 730     size_t len;
 731     unsigned int i;
 732
 733     redisLog(REDIS_DEBUG,"LOADING OBJECT %d (at %d)\n",rdbtype,rioTell(rdb));
 734     if (rdbtype == REDIS_RDB_TYPE_STRING) {
 735         /* Read string value */
 736         if ((o = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL;
 737         o = tryObjectEncoding(o);
 738     } else if (rdbtype == REDIS_RDB_TYPE_LIST) {
 739         /* Read list value */
 740         if ((len = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL;
 741
 742         /* Use a real list when there are too many entries */
 743         if (len > server.list_max_ziplist_entries) {
 744             o = createListObject();
 745         } else {
 746             o = createZiplistObject();
 747         }
 748
 749         /* Load every single element of the list */
 750         while(len--) {
 751             if ((ele = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL;
 752
 753             /* If we are using a ziplist and the value is too big, convert
 754              * the object to a real list. */
 755             if (o->encoding == REDIS_ENCODING_ZIPLIST &&
 756                 ele->encoding == REDIS_ENCODING_RAW &&
 757                 sdslen(ele->ptr) > server.list_max_ziplist_value)
 758                     listTypeConvert(o,REDIS_ENCODING_LINKEDLIST);
 759
 760             if (o->encoding == REDIS_ENCODING_ZIPLIST) {
 761                 dec = getDecodedObject(ele);
 762                 o->ptr = ziplistPush(o->ptr,dec->ptr,sdslen(dec->ptr),REDIS_TAIL);
 763                 decrRefCount(dec);
 764                 decrRefCount(ele);
 765             } else {
 766                 ele = tryObjectEncoding(ele);
 767                 listAddNodeTail(o->ptr,ele);
 768             }
 769         }
 770     } else if (rdbtype == REDIS_RDB_TYPE_SET) {
 771         /* Read list/set value */
 772         if ((len = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL;
 773
 774         /* Use a regular set when there are too many entries. */
 775         if (len > server.set_max_intset_entries) {
 776             o = createSetObject();
 777             /* It's faster to expand the dict to the right size asap in order
 778              * to avoid rehashing */
 779             if (len > DICT_HT_INITIAL_SIZE)
 780                 dictExpand(o->ptr,len);
 781         } else {
 782             o = createIntsetObject();
 783         }
 784
 785         /* Load every single element of the list/set */
 786         for (i = 0; i < len; i++) {
 787             long long llval;
 788             if ((ele = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL;
 789             ele = tryObjectEncoding(ele);
 790
 791             if (o->encoding == REDIS_ENCODING_INTSET) {
 792                 /* Fetch integer value from element */
 793                 if (isObjectRepresentableAsLongLong(ele,&llval) == REDIS_OK) {
 794                     o->ptr = intsetAdd(o->ptr,llval,NULL);
 795                 } else {
 796                     setTypeConvert(o,REDIS_ENCODING_HT);
 797                     dictExpand(o->ptr,len);
 798                 }
 799             }
 800
 801             /* This will also be called when the set was just converted
 802              * to regular hashtable encoded set */
 803             if (o->encoding == REDIS_ENCODING_HT) {
 804                 dictAdd((dict*)o->ptr,ele,NULL);
 805             } else {
 806                 decrRefCount(ele);
 807             }
 808         }
 809     } else if (rdbtype == REDIS_RDB_TYPE_ZSET) {
 810         /* Read list/set value */
 811         size_t zsetlen;
 812         size_t maxelelen = 0;
 813         zset *zs;
 814
 815         if ((zsetlen = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL;
 816         o = createZsetObject();
 817         zs = o->ptr;
 818
 819         /* Load every single element of the list/set */
 820         while(zsetlen--) {
 821             robj *ele;
 822             double score;
 823             zskiplistNode *znode;
 824
 825             if ((ele = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL;
 826             ele = tryObjectEncoding(ele);
 827             if (rdbLoadDoubleValue(rdb,&score) == -1) return NULL;
 828
 829             /* Don't care about integer-encoded strings. */
 830             if (ele->encoding == REDIS_ENCODING_RAW &&
 831                 sdslen(ele->ptr) > maxelelen)
 832                     maxelelen = sdslen(ele->ptr);
 833
 834             znode = zslInsert(zs->zsl,score,ele);
 835             dictAdd(zs->dict,ele,&znode->score);
 836             incrRefCount(ele); /* added to skiplist */
 837         }
 838
 839         /* Convert *after* loading, since sorted sets are not stored ordered. */
 840         if (zsetLength(o) <= server.zset_max_ziplist_entries &&
 841             maxelelen <= server.zset_max_ziplist_value)
 842                 zsetConvert(o,REDIS_ENCODING_ZIPLIST);
 843     } else if (rdbtype == REDIS_RDB_TYPE_HASH) {
 844         size_t len;
 845         int ret;
 846
 847         len = rdbLoadLen(rdb, NULL);
 848         if (len == REDIS_RDB_LENERR) return NULL;
 849
 850         o = createHashObject();
 851
 852         /* Too many entries? Use an hash table. */
 853         if (len > server.hash_max_ziplist_entries)
 854             hashTypeConvert(o, REDIS_ENCODING_HT);
 855
 856         /* Load every field and value into the ziplist */
 857         while (o->encoding == REDIS_ENCODING_ZIPLIST && len > 0) {
 858             robj *field, *value;
 859
 860             len--;
 861             /* Load raw strings */
 862             field = rdbLoadStringObject(rdb);
 863             if (field == NULL) return NULL;
 864             redisAssert(field->encoding == REDIS_ENCODING_RAW);
 865             value = rdbLoadStringObject(rdb);
 866             if (value == NULL) return NULL;
 867             redisAssert(field->encoding == REDIS_ENCODING_RAW);
 868
 869             /* Add pair to ziplist */
 870             o->ptr = ziplistPush(o->ptr, field->ptr, sdslen(field->ptr), ZIPLIST_TAIL);
 871             o->ptr = ziplistPush(o->ptr, value->ptr, sdslen(value->ptr), ZIPLIST_TAIL);
 872             /* Convert to hash table if size threshold is exceeded */
 873             if (sdslen(field->ptr) > server.hash_max_ziplist_value ||
 874                 sdslen(value->ptr) > server.hash_max_ziplist_value)
 875             {
 876                 decrRefCount(field);
 877                 decrRefCount(value);
 878                 hashTypeConvert(o, REDIS_ENCODING_HT);
 879                 break;
 880             }
 881             decrRefCount(field);
 882             decrRefCount(value);
 883         }
 884
 885         /* Load remaining fields and values into the hash table */
 886         while (o->encoding == REDIS_ENCODING_HT && len > 0) {
 887             robj *field, *value;
 888
 889             len--;
 890             /* Load encoded strings */
 891             field = rdbLoadEncodedStringObject(rdb);
 892             if (field == NULL) return NULL;
 893             value = rdbLoadEncodedStringObject(rdb);
 894             if (value == NULL) return NULL;
 895
 896             field = tryObjectEncoding(field);
 897             value = tryObjectEncoding(value);
 898
 899             /* Add pair to hash table */
 900             ret = dictAdd((dict*)o->ptr, field, value);
 901             redisAssert(ret == REDIS_OK);
 902         }
 903
 904         /* All pairs should be read by now */
 905         redisAssert(len == 0);
 906
 907     } else if (rdbtype == REDIS_RDB_TYPE_HASH_ZIPMAP  ||
 908                rdbtype == REDIS_RDB_TYPE_LIST_ZIPLIST ||
 909                rdbtype == REDIS_RDB_TYPE_SET_INTSET   ||
 910                rdbtype == REDIS_RDB_TYPE_ZSET_ZIPLIST ||
 911                rdbtype == REDIS_RDB_TYPE_HASH_ZIPLIST)
 912     {
 913         robj *aux = rdbLoadStringObject(rdb);
 914
 915         if (aux == NULL) return NULL;
 916         o = createObject(REDIS_STRING,NULL); /* string is just placeholder */
 917         o->ptr = zmalloc(sdslen(aux->ptr));
 918         memcpy(o->ptr,aux->ptr,sdslen(aux->ptr));
 919         decrRefCount(aux);
 920
 921         /* Fix the object encoding, and make sure to convert the encoded
 922          * data type into the base type if accordingly to the current
 923          * configuration there are too many elements in the encoded data
 924          * type. Note that we only check the length and not max element
 925          * size as this is an O(N) scan. Eventually everything will get
 926          * converted. */
 927         switch(rdbtype) {
 928             case REDIS_RDB_TYPE_HASH_ZIPMAP:
 929                 /* Convert to ziplist encoded hash. This must be deprecated
 930                  * when loading dumps created by Redis 2.4 gets deprecated. */
 931                 {
 932                     unsigned char *zl = ziplistNew();
 933                     unsigned char *zi = zipmapRewind(o->ptr);
 934                     unsigned char *fstr, *vstr;
 935                     unsigned int flen, vlen;
 936                     unsigned int maxlen = 0;
 937
 938                     while ((zi = zipmapNext(zi, &fstr, &flen, &vstr, &vlen)) != NULL) {
 939                         if (flen > maxlen) maxlen = flen;
 940                         if (vlen > maxlen) maxlen = vlen;
 941                         zl = ziplistPush(zl, fstr, flen, ZIPLIST_TAIL);
 942                         zl = ziplistPush(zl, vstr, vlen, ZIPLIST_TAIL);
 943                     }
 944
 945                     zfree(o->ptr);
 946                     o->ptr = zl;
 947                     o->type = REDIS_HASH;
 948                     o->encoding = REDIS_ENCODING_ZIPLIST;
 949
 950                     if (hashTypeLength(o) > server.hash_max_ziplist_entries ||
 951                         maxlen > server.hash_max_ziplist_value)
 952                     {
 953                         hashTypeConvert(o, REDIS_ENCODING_HT);
 954                     }
 955                 }
 956                 break;
 957             case REDIS_RDB_TYPE_LIST_ZIPLIST:
 958                 o->type = REDIS_LIST;
 959                 o->encoding = REDIS_ENCODING_ZIPLIST;
 960                 if (ziplistLen(o->ptr) > server.list_max_ziplist_entries)
 961                     listTypeConvert(o,REDIS_ENCODING_LINKEDLIST);
 962                 break;
 963             case REDIS_RDB_TYPE_SET_INTSET:
 964                 o->type = REDIS_SET;
 965                 o->encoding = REDIS_ENCODING_INTSET;
 966                 if (intsetLen(o->ptr) > server.set_max_intset_entries)
 967                     setTypeConvert(o,REDIS_ENCODING_HT);
 968                 break;
 969             case REDIS_RDB_TYPE_ZSET_ZIPLIST:
 970                 o->type = REDIS_ZSET;
 971                 o->encoding = REDIS_ENCODING_ZIPLIST;
 972                 if (zsetLength(o) > server.zset_max_ziplist_entries)
 973                     zsetConvert(o,REDIS_ENCODING_SKIPLIST);
 974                 break;
 975             case REDIS_RDB_TYPE_HASH_ZIPLIST:
 976                 o->type = REDIS_HASH;
 977                 o->encoding = REDIS_ENCODING_ZIPLIST;
 978                 if (hashTypeLength(o) > server.hash_max_ziplist_entries)
 979                     hashTypeConvert(o, REDIS_ENCODING_HT);
 980                 break;
 981             default:
 982                 redisPanic("Unknown encoding");
 983                 break;
 984         }
 985     } else {
 986         redisPanic("Unknown object type");
 987     }
 988     return o;
 989 }
 990
 991 /* Mark that we are loading in the global state and setup the fields
 992  * needed to provide loading stats. */
 993 void startLoading(FILE *fp) {
 994     struct stat sb;
 995
 996     /* Load the DB */
 997     server.loading = 1;
 998     server.loading_start_time = time(NULL);
 999     if (fstat(fileno(fp), &sb) == -1) {
1000         server.loading_total_bytes = 1; /* just to avoid division by zero */
1001     } else {
1002         server.loading_total_bytes = sb.st_size;
1003     }
1004 }
1005
1006 /* Refresh the loading progress info */
1007 void loadingProgress(off_t pos) {
1008     server.loading_loaded_bytes = pos;
1009 }
1010
1011 /* Loading finished */
1012 void stopLoading(void) {
1013     server.loading = 0;
1014 }
1015
1016 int rdbLoad(char *filename) {
1017     uint32_t dbid;
1018     int type, rdbver;
1019     redisDb *db = server.db+0;
1020     char buf[1024];
1021     long long expiretime, now = mstime();
1022     long loops = 0;
1023     FILE *fp;
1024     rio rdb;
1025
1026     fp = fopen(filename,"r");
1027     if (!fp) {
1028         errno = ENOENT;
1029         return REDIS_ERR;
1030     }
1031     rioInitWithFile(&rdb,fp);
1032     if (server.rdb_checksum)
1033         rdb.update_cksum = rioGenericUpdateChecksum;
1034     if (rioRead(&rdb,buf,9) == 0) goto eoferr;
1035     buf[9] = '\0';
1036     if (memcmp(buf,"REDIS",5) != 0) {
1037         fclose(fp);
1038         redisLog(REDIS_WARNING,"Wrong signature trying to load DB from file");
1039         errno = EINVAL;
1040         return REDIS_ERR;
1041     }
1042     rdbver = atoi(buf+5);
1043     if (rdbver < 1 || rdbver > REDIS_RDB_VERSION) {
1044         fclose(fp);
1045         redisLog(REDIS_WARNING,"Can't handle RDB format version %d",rdbver);
1046         errno = EINVAL;
1047         return REDIS_ERR;
1048     }
1049
1050     startLoading(fp);
1051     while(1) {
1052         robj *key, *val;
1053         expiretime = -1;
1054
1055         /* Serve the clients from time to time */
1056         if (!(loops++ % 1000)) {
1057             loadingProgress(rioTell(&rdb));
1058             aeProcessEvents(server.el, AE_FILE_EVENTS|AE_DONT_WAIT);
1059         }
1060
1061         /* Read type. */
1062         if ((type = rdbLoadType(&rdb)) == -1) goto eoferr;
1063         if (type == REDIS_RDB_OPCODE_EXPIRETIME) {
1064             if ((expiretime = rdbLoadTime(&rdb)) == -1) goto eoferr;
1065             /* We read the time so we need to read the object type again. */
1066             if ((type = rdbLoadType(&rdb)) == -1) goto eoferr;
1067             /* the EXPIRETIME opcode specifies time in seconds, so convert
1068              * into milliesconds. */
1069             expiretime *= 1000;
1070         } else if (type == REDIS_RDB_OPCODE_EXPIRETIME_MS) {
1071             /* Milliseconds precision expire times introduced with RDB
1072              * version 3. */
1073             if ((expiretime = rdbLoadMillisecondTime(&rdb)) == -1) goto eoferr;
1074             /* We read the time so we need to read the object type again. */
1075             if ((type = rdbLoadType(&rdb)) == -1) goto eoferr;
1076         }
1077
1078         if (type == REDIS_RDB_OPCODE_EOF)
1079             break;
1080
1081         /* Handle SELECT DB opcode as a special case */
1082         if (type == REDIS_RDB_OPCODE_SELECTDB) {
1083             if ((dbid = rdbLoadLen(&rdb,NULL)) == REDIS_RDB_LENERR)
1084                 goto eoferr;
1085             if (dbid >= (unsigned)server.dbnum) {
1086                 redisLog(REDIS_WARNING,"FATAL: Data file was created with a Redis server configured to handle more than %d databases. Exiting\n", server.dbnum);
1087                 exit(1);
1088             }
1089             db = server.db+dbid;
1090             continue;
1091         }
1092         /* Read key */
1093         if ((key = rdbLoadStringObject(&rdb)) == NULL) goto eoferr;
1094         /* Read value */
1095         if ((val = rdbLoadObject(type,&rdb)) == NULL) goto eoferr;
1096         /* Check if the key already expired. This function is used when loading
1097          * an RDB file from disk, either at startup, or when an RDB was
1098          * received from the master. In the latter case, the master is
1099          * responsible for key expiry. If we would expire keys here, the
1100          * snapshot taken by the master may not be reflected on the slave. */
1101         if (server.masterhost == NULL && expiretime != -1 && expiretime < now) {
1102             decrRefCount(key);
1103             decrRefCount(val);
1104             continue;
1105         }
1106         /* Add the new object in the hash table */
1107         dbAdd(db,key,val);
1108
1109         /* Set the expire time if needed */
1110         if (expiretime != -1) setExpire(db,key,expiretime);
1111
1112         decrRefCount(key);
1113     }
1114     /* Verify the checksum if RDB version is >= 5 */
1115     if (rdbver >= 5 && server.rdb_checksum) {
1116         uint64_t cksum, expected = rdb.cksum;
1117
1118         if (rioRead(&rdb,&cksum,8) == 0) goto eoferr;
1119         memrev64ifbe(&cksum);
1120         if (cksum == 0) {
1121             redisLog(REDIS_WARNING,"RDB file was saved with checksum disabled: no check performed.");
1122         } else if (cksum != expected) {
1123             redisLog(REDIS_WARNING,"Wrong RDB checksum. Aborting now.");
1124             exit(1);
1125         }
1126     }
1127
1128     fclose(fp);
1129     stopLoading();
1130     return REDIS_OK;
1131
1132 eoferr: /* unexpected end of file is handled here with a fatal exit */
1133     redisLog(REDIS_WARNING,"Short read or OOM loading DB. Unrecoverable error, aborting now.");
1134     exit(1);
1135     return REDIS_ERR; /* Just to avoid warning */
1136 }
1137
1138 /* A background saving child (BGSAVE) terminated its work. Handle this. */
1139 void backgroundSaveDoneHandler(int exitcode, int bysignal) {
1140     if (!bysignal && exitcode == 0) {
1141         redisLog(REDIS_NOTICE,
1142             "Background saving terminated with success");
1143         server.dirty = server.dirty - server.dirty_before_bgsave;
1144         server.lastsave = time(NULL);
1145         server.lastbgsave_status = REDIS_OK;
1146     } else if (!bysignal && exitcode != 0) {
1147         redisLog(REDIS_WARNING, "Background saving error");
1148         server.lastbgsave_status = REDIS_ERR;
1149     } else {
1150         redisLog(REDIS_WARNING,
1151             "Background saving terminated by signal %d", bysignal);
1152         rdbRemoveTempFile(server.rdb_child_pid);
1153         server.lastbgsave_status = REDIS_ERR;
1154     }
1155     server.rdb_child_pid = -1;
1156     server.rdb_save_time_last = time(NULL)-server.rdb_save_time_start;
1157     server.rdb_save_time_start = -1;
1158     /* Possibly there are slaves waiting for a BGSAVE in order to be served
1159      * (the first stage of SYNC is a bulk transfer of dump.rdb) */
1160     updateSlavesWaitingBgsave(exitcode == 0 ? REDIS_OK : REDIS_ERR);
1161 }
1162
1163 void saveCommand(redisClient *c) {
1164     if (server.rdb_child_pid != -1) {
1165         addReplyError(c,"Background save already in progress");
1166         return;
1167     }
1168     if (rdbSave(server.rdb_filename) == REDIS_OK) {
1169         addReply(c,shared.ok);
1170     } else {
1171         addReply(c,shared.err);
1172     }
1173 }
1174
1175 void bgsaveCommand(redisClient *c) {
1176     if (server.rdb_child_pid != -1) {
1177         addReplyError(c,"Background save already in progress");
1178     } else if (server.aof_child_pid != -1) {
1179         addReplyError(c,"Can't BGSAVE while AOF log rewriting is in progress");
1180     } else if (rdbSaveBackground(server.rdb_filename) == REDIS_OK) {
1181         addReplyStatus(c,"Background saving started");
1182     } else {
1183         addReply(c,shared.err);
1184     }
1185 }