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