src/rdb.c

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