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 the object type of object "o". */
 370 int rdbSaveObjectType(rio *rdb, robj *o) {
 371     switch (o->type) {
 372     case REDIS_STRING:
 373         return rdbSaveType(rdb,REDIS_RDB_TYPE_STRING);
 374     case REDIS_LIST:
 375         if (o->encoding == REDIS_ENCODING_ZIPLIST)
 376             return rdbSaveType(rdb,REDIS_RDB_TYPE_LIST_ZIPLIST);
 377         else if (o->encoding == REDIS_ENCODING_LINKEDLIST)
 378             return rdbSaveType(rdb,REDIS_RDB_TYPE_LIST);
 379         else
 380             redisPanic("Unknown list encoding");
 381     case REDIS_SET:
 382         if (o->encoding == REDIS_ENCODING_INTSET)
 383             return rdbSaveType(rdb,REDIS_RDB_TYPE_SET_INTSET);
 384         else if (o->encoding == REDIS_ENCODING_HT)
 385             return rdbSaveType(rdb,REDIS_RDB_TYPE_SET);
 386         else
 387             redisPanic("Unknown set encoding");
 388     case REDIS_ZSET:
 389         if (o->encoding == REDIS_ENCODING_ZIPLIST)
 390             return rdbSaveType(rdb,REDIS_RDB_TYPE_ZSET_ZIPLIST);
 391         else if (o->encoding == REDIS_ENCODING_SKIPLIST)
 392             return rdbSaveType(rdb,REDIS_RDB_TYPE_ZSET);
 393         else
 394             redisPanic("Unknown sorted set encoding");
 395     case REDIS_HASH:
 396         if (o->encoding == REDIS_ENCODING_ZIPMAP)
 397             return rdbSaveType(rdb,REDIS_RDB_TYPE_HASH_ZIPMAP);
 398         else if (o->encoding == REDIS_ENCODING_HT)
 399             return rdbSaveType(rdb,REDIS_RDB_TYPE_HASH);
 400         else
 401             redisPanic("Unknown hash encoding");
 402     default:
 403         redisPanic("Unknown object type");
 404     }
 405     return -1; /* avoid warning */
 406 }
 407
 408 /* Load object type. Return -1 when the byte doesn't contain an object type. */
 409 int rdbLoadObjectType(rio *rdb) {
 410     int type;
 411     if ((type = rdbLoadType(rdb)) == -1) return -1;
 412     if (!rdbIsObjectType(type)) return -1;
 413     return type;
 414 }
 415
 416 /* Save a key-value pair, with expire time, type, key, value.
 417  * On error -1 is returned.
 418  * On success if the key was actaully saved 1 is returned, otherwise 0
 419  * is returned (the key was already expired). */
 420 int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val,
 421                         time_t expiretime, time_t now)
 422 {
 423     /* Save the expire time */
 424     if (expiretime != -1) {
 425         /* If this key is already expired skip it */
 426         if (expiretime < now) return 0;
 427         if (rdbSaveType(rdb,REDIS_RDB_OPCODE_EXPIRETIME) == -1) return -1;
 428         if (rdbSaveTime(rdb,expiretime) == -1) return -1;
 429     }
 430
 431     /* Save type, key, value */
 432     if (rdbSaveObjectType(rdb,val) == -1) return -1;
 433     if (rdbSaveStringObject(rdb,key) == -1) return -1;
 434     if (rdbSaveObject(rdb,val) == -1) return -1;
 435     return 1;
 436 }
 437
 438 /* Save the DB on disk. Return REDIS_ERR on error, REDIS_OK on success */
 439 int rdbSave(char *filename) {
 440     dictIterator *di = NULL;
 441     dictEntry *de;
 442     char tmpfile[256];
 443     int j;
 444     time_t now = time(NULL);
 445     FILE *fp;
 446     rio rdb;
 447
 448     if (server.ds_enabled) {
 449         cacheForcePointInTime();
 450         return dsRdbSave(filename);
 451     }
 452
 453     snprintf(tmpfile,256,"temp-%d.rdb", (int) getpid());
 454     fp = fopen(tmpfile,"w");
 455     if (!fp) {
 456         redisLog(REDIS_WARNING, "Failed opening .rdb for saving: %s",
 457             strerror(errno));
 458         return REDIS_ERR;
 459     }
 460
 461     rdb = rioInitWithFile(fp);
 462     if (rdbWriteRaw(&rdb,"REDIS0002",9) == -1) goto werr;
 463
 464     for (j = 0; j < server.dbnum; j++) {
 465         redisDb *db = server.db+j;
 466         dict *d = db->dict;
 467         if (dictSize(d) == 0) continue;
 468         di = dictGetIterator(d);
 469         if (!di) {
 470             fclose(fp);
 471             return REDIS_ERR;
 472         }
 473
 474         /* Write the SELECT DB opcode */
 475         if (rdbSaveType(&rdb,REDIS_RDB_OPCODE_SELECTDB) == -1) goto werr;
 476         if (rdbSaveLen(&rdb,j) == -1) goto werr;
 477
 478         /* Iterate this DB writing every entry */
 479         while((de = dictNext(di)) != NULL) {
 480             sds keystr = dictGetEntryKey(de);
 481             robj key, *o = dictGetEntryVal(de);
 482             time_t expire;
 483
 484             initStaticStringObject(key,keystr);
 485             expire = getExpire(db,&key);
 486             if (rdbSaveKeyValuePair(&rdb,&key,o,expire,now) == -1) goto werr;
 487         }
 488         dictReleaseIterator(di);
 489     }
 490     /* EOF opcode */
 491     if (rdbSaveType(&rdb,REDIS_RDB_OPCODE_EOF) == -1) goto werr;
 492
 493     /* Make sure data will not remain on the OS's output buffers */
 494     fflush(fp);
 495     fsync(fileno(fp));
 496     fclose(fp);
 497
 498     /* Use RENAME to make sure the DB file is changed atomically only
 499      * if the generate DB file is ok. */
 500     if (rename(tmpfile,filename) == -1) {
 501         redisLog(REDIS_WARNING,"Error moving temp DB file on the final destination: %s", strerror(errno));
 502         unlink(tmpfile);
 503         return REDIS_ERR;
 504     }
 505     redisLog(REDIS_NOTICE,"DB saved on disk");
 506     server.dirty = 0;
 507     server.lastsave = time(NULL);
 508     return REDIS_OK;
 509
 510 werr:
 511     fclose(fp);
 512     unlink(tmpfile);
 513     redisLog(REDIS_WARNING,"Write error saving DB on disk: %s", strerror(errno));
 514     if (di) dictReleaseIterator(di);
 515     return REDIS_ERR;
 516 }
 517
 518 int rdbSaveBackground(char *filename) {
 519     pid_t childpid;
 520
 521     if (server.bgsavechildpid != -1 ||
 522         server.bgsavethread != (pthread_t) -1) return REDIS_ERR;
 523
 524     server.dirty_before_bgsave = server.dirty;
 525
 526     if (server.ds_enabled) {
 527         cacheForcePointInTime();
 528         return dsRdbSaveBackground(filename);
 529     }
 530
 531     if ((childpid = fork()) == 0) {
 532         int retval;
 533
 534         /* Child */
 535         if (server.ipfd > 0) close(server.ipfd);
 536         if (server.sofd > 0) close(server.sofd);
 537         retval = rdbSave(filename);
 538         _exit((retval == REDIS_OK) ? 0 : 1);
 539     } else {
 540         /* Parent */
 541         if (childpid == -1) {
 542             redisLog(REDIS_WARNING,"Can't save in background: fork: %s",
 543                 strerror(errno));
 544             return REDIS_ERR;
 545         }
 546         redisLog(REDIS_NOTICE,"Background saving started by pid %d",childpid);
 547         server.bgsavechildpid = childpid;
 548         updateDictResizePolicy();
 549         return REDIS_OK;
 550     }
 551     return REDIS_OK; /* unreached */
 552 }
 553
 554 void rdbRemoveTempFile(pid_t childpid) {
 555     char tmpfile[256];
 556
 557     snprintf(tmpfile,256,"temp-%d.rdb", (int) childpid);
 558     unlink(tmpfile);
 559 }
 560
 561 int rdbLoadType(rio *rdb) {
 562     unsigned char type;
 563     if (rioRead(rdb,&type,1) == 0) return -1;
 564     return type;
 565 }
 566
 567 time_t rdbLoadTime(rio *rdb) {
 568     int32_t t32;
 569     if (rioRead(rdb,&t32,4) == 0) return -1;
 570     return (time_t) t32;
 571 }
 572
 573 /* Load an encoded length from the DB, see the REDIS_RDB_* defines on the top
 574  * of this file for a description of how this are stored on disk.
 575  *
 576  * isencoded is set to 1 if the readed length is not actually a length but
 577  * an "encoding type", check the above comments for more info */
 578 uint32_t rdbLoadLen(rio *rdb, int *isencoded) {
 579     unsigned char buf[2];
 580     uint32_t len;
 581     int type;
 582
 583     if (isencoded) *isencoded = 0;
 584     if (rioRead(rdb,buf,1) == 0) return REDIS_RDB_LENERR;
 585     type = (buf[0]&0xC0)>>6;
 586     if (type == REDIS_RDB_6BITLEN) {
 587         /* Read a 6 bit len */
 588         return buf[0]&0x3F;
 589     } else if (type == REDIS_RDB_ENCVAL) {
 590         /* Read a 6 bit len encoding type */
 591         if (isencoded) *isencoded = 1;
 592         return buf[0]&0x3F;
 593     } else if (type == REDIS_RDB_14BITLEN) {
 594         /* Read a 14 bit len */
 595         if (rioRead(rdb,buf+1,1) == 0) return REDIS_RDB_LENERR;
 596         return ((buf[0]&0x3F)<<8)|buf[1];
 597     } else {
 598         /* Read a 32 bit len */
 599         if (rioRead(rdb,&len,4) == 0) return REDIS_RDB_LENERR;
 600         return ntohl(len);
 601     }
 602 }
 603
 604 /* Load an integer-encoded object from file 'fp', with the specified
 605  * encoding type 'enctype'. If encode is true the function may return
 606  * an integer-encoded object as reply, otherwise the returned object
 607  * will always be encoded as a raw string. */
 608 robj *rdbLoadIntegerObject(rio *rdb, int enctype, int encode) {
 609     unsigned char enc[4];
 610     long long val;
 611
 612     if (enctype == REDIS_RDB_ENC_INT8) {
 613         if (rioRead(rdb,enc,1) == 0) return NULL;
 614         val = (signed char)enc[0];
 615     } else if (enctype == REDIS_RDB_ENC_INT16) {
 616         uint16_t v;
 617         if (rioRead(rdb,enc,2) == 0) return NULL;
 618         v = enc[0]|(enc[1]<<8);
 619         val = (int16_t)v;
 620     } else if (enctype == REDIS_RDB_ENC_INT32) {
 621         uint32_t v;
 622         if (rioRead(rdb,enc,4) == 0) return NULL;
 623         v = enc[0]|(enc[1]<<8)|(enc[2]<<16)|(enc[3]<<24);
 624         val = (int32_t)v;
 625     } else {
 626         val = 0; /* anti-warning */
 627         redisPanic("Unknown RDB integer encoding type");
 628     }
 629     if (encode)
 630         return createStringObjectFromLongLong(val);
 631     else
 632         return createObject(REDIS_STRING,sdsfromlonglong(val));
 633 }
 634
 635 robj *rdbLoadLzfStringObject(rio *rdb) {
 636     unsigned int len, clen;
 637     unsigned char *c = NULL;
 638     sds val = NULL;
 639
 640     if ((clen = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL;
 641     if ((len = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL;
 642     if ((c = zmalloc(clen)) == NULL) goto err;
 643     if ((val = sdsnewlen(NULL,len)) == NULL) goto err;
 644     if (rioRead(rdb,c,clen) == 0) goto err;
 645     if (lzf_decompress(c,clen,val,len) == 0) goto err;
 646     zfree(c);
 647     return createObject(REDIS_STRING,val);
 648 err:
 649     zfree(c);
 650     sdsfree(val);
 651     return NULL;
 652 }
 653
 654 robj *rdbGenericLoadStringObject(rio *rdb, int encode) {
 655     int isencoded;
 656     uint32_t len;
 657     sds val;
 658
 659     len = rdbLoadLen(rdb,&isencoded);
 660     if (isencoded) {
 661         switch(len) {
 662         case REDIS_RDB_ENC_INT8:
 663         case REDIS_RDB_ENC_INT16:
 664         case REDIS_RDB_ENC_INT32:
 665             return rdbLoadIntegerObject(rdb,len,encode);
 666         case REDIS_RDB_ENC_LZF:
 667             return rdbLoadLzfStringObject(rdb);
 668         default:
 669             redisPanic("Unknown RDB encoding type");
 670         }
 671     }
 672
 673     if (len == REDIS_RDB_LENERR) return NULL;
 674     val = sdsnewlen(NULL,len);
 675     if (len && rioRead(rdb,val,len) == 0) {
 676         sdsfree(val);
 677         return NULL;
 678     }
 679     return createObject(REDIS_STRING,val);
 680 }
 681
 682 robj *rdbLoadStringObject(rio *rdb) {
 683     return rdbGenericLoadStringObject(rdb,0);
 684 }
 685
 686 robj *rdbLoadEncodedStringObject(rio *rdb) {
 687     return rdbGenericLoadStringObject(rdb,1);
 688 }
 689
 690 /* For information about double serialization check rdbSaveDoubleValue() */
 691 int rdbLoadDoubleValue(rio *rdb, double *val) {
 692     char buf[128];
 693     unsigned char len;
 694
 695     if (rioRead(rdb,&len,1) == 0) return -1;
 696     switch(len) {
 697     case 255: *val = R_NegInf; return 0;
 698     case 254: *val = R_PosInf; return 0;
 699     case 253: *val = R_Nan; return 0;
 700     default:
 701         if (rioRead(rdb,buf,len) == 0) return -1;
 702         buf[len] = '\0';
 703         sscanf(buf, "%lg", val);
 704         return 0;
 705     }
 706 }
 707
 708 /* Load a Redis object of the specified type from the specified file.
 709  * On success a newly allocated object is returned, otherwise NULL. */
 710 robj *rdbLoadObject(int rdbtype, rio *rdb) {
 711     robj *o, *ele, *dec;
 712     size_t len;
 713     unsigned int i;
 714
 715     redisLog(REDIS_DEBUG,"LOADING OBJECT %d (at %d)\n",rdbtype,rdb->tell(rdb));
 716     if (rdbtype == REDIS_RDB_TYPE_STRING) {
 717         /* Read string value */
 718         if ((o = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL;
 719         o = tryObjectEncoding(o);
 720     } else if (rdbtype == REDIS_RDB_TYPE_LIST) {
 721         /* Read list value */
 722         if ((len = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL;
 723
 724         /* Use a real list when there are too many entries */
 725         if (len > server.list_max_ziplist_entries) {
 726             o = createListObject();
 727         } else {
 728             o = createZiplistObject();
 729         }
 730
 731         /* Load every single element of the list */
 732         while(len--) {
 733             if ((ele = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL;
 734
 735             /* If we are using a ziplist and the value is too big, convert
 736              * the object to a real list. */
 737             if (o->encoding == REDIS_ENCODING_ZIPLIST &&
 738                 ele->encoding == REDIS_ENCODING_RAW &&
 739                 sdslen(ele->ptr) > server.list_max_ziplist_value)
 740                     listTypeConvert(o,REDIS_ENCODING_LINKEDLIST);
 741
 742             if (o->encoding == REDIS_ENCODING_ZIPLIST) {
 743                 dec = getDecodedObject(ele);
 744                 o->ptr = ziplistPush(o->ptr,dec->ptr,sdslen(dec->ptr),REDIS_TAIL);
 745                 decrRefCount(dec);
 746                 decrRefCount(ele);
 747             } else {
 748                 ele = tryObjectEncoding(ele);
 749                 listAddNodeTail(o->ptr,ele);
 750             }
 751         }
 752     } else if (rdbtype == REDIS_RDB_TYPE_SET) {
 753         /* Read list/set value */
 754         if ((len = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL;
 755
 756         /* Use a regular set when there are too many entries. */
 757         if (len > server.set_max_intset_entries) {
 758             o = createSetObject();
 759             /* It's faster to expand the dict to the right size asap in order
 760              * to avoid rehashing */
 761             if (len > DICT_HT_INITIAL_SIZE)
 762                 dictExpand(o->ptr,len);
 763         } else {
 764             o = createIntsetObject();
 765         }
 766
 767         /* Load every single element of the list/set */
 768         for (i = 0; i < len; i++) {
 769             long long llval;
 770             if ((ele = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL;
 771             ele = tryObjectEncoding(ele);
 772
 773             if (o->encoding == REDIS_ENCODING_INTSET) {
 774                 /* Fetch integer value from element */
 775                 if (isObjectRepresentableAsLongLong(ele,&llval) == REDIS_OK) {
 776                     o->ptr = intsetAdd(o->ptr,llval,NULL);
 777                 } else {
 778                     setTypeConvert(o,REDIS_ENCODING_HT);
 779                     dictExpand(o->ptr,len);
 780                 }
 781             }
 782
 783             /* This will also be called when the set was just converted
 784              * to regular hashtable encoded set */
 785             if (o->encoding == REDIS_ENCODING_HT) {
 786                 dictAdd((dict*)o->ptr,ele,NULL);
 787             } else {
 788                 decrRefCount(ele);
 789             }
 790         }
 791     } else if (rdbtype == REDIS_RDB_TYPE_ZSET) {
 792         /* Read list/set value */
 793         size_t zsetlen;
 794         size_t maxelelen = 0;
 795         zset *zs;
 796
 797         if ((zsetlen = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL;
 798         o = createZsetObject();
 799         zs = o->ptr;
 800
 801         /* Load every single element of the list/set */
 802         while(zsetlen--) {
 803             robj *ele;
 804             double score;
 805             zskiplistNode *znode;
 806
 807             if ((ele = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL;
 808             ele = tryObjectEncoding(ele);
 809             if (rdbLoadDoubleValue(rdb,&score) == -1) return NULL;
 810
 811             /* Don't care about integer-encoded strings. */
 812             if (ele->encoding == REDIS_ENCODING_RAW &&
 813                 sdslen(ele->ptr) > maxelelen)
 814                     maxelelen = sdslen(ele->ptr);
 815
 816             znode = zslInsert(zs->zsl,score,ele);
 817             dictAdd(zs->dict,ele,&znode->score);
 818             incrRefCount(ele); /* added to skiplist */
 819         }
 820
 821         /* Convert *after* loading, since sorted sets are not stored ordered. */
 822         if (zsetLength(o) <= server.zset_max_ziplist_entries &&
 823             maxelelen <= server.zset_max_ziplist_value)
 824                 zsetConvert(o,REDIS_ENCODING_ZIPLIST);
 825     } else if (rdbtype == REDIS_RDB_TYPE_HASH) {
 826         size_t hashlen;
 827
 828         if ((hashlen = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL;
 829         o = createHashObject();
 830         /* Too many entries? Use an hash table. */
 831         if (hashlen > server.hash_max_zipmap_entries)
 832             convertToRealHash(o);
 833         /* Load every key/value, then set it into the zipmap or hash
 834          * table, as needed. */
 835         while(hashlen--) {
 836             robj *key, *val;
 837
 838             if ((key = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL;
 839             if ((val = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL;
 840             /* If we are using a zipmap and there are too big values
 841              * the object is converted to real hash table encoding. */
 842             if (o->encoding != REDIS_ENCODING_HT &&
 843                ((key->encoding == REDIS_ENCODING_RAW &&
 844                 sdslen(key->ptr) > server.hash_max_zipmap_value) ||
 845                 (val->encoding == REDIS_ENCODING_RAW &&
 846                 sdslen(val->ptr) > server.hash_max_zipmap_value)))
 847             {
 848                     convertToRealHash(o);
 849             }
 850
 851             if (o->encoding == REDIS_ENCODING_ZIPMAP) {
 852                 unsigned char *zm = o->ptr;
 853                 robj *deckey, *decval;
 854
 855                 /* We need raw string objects to add them to the zipmap */
 856                 deckey = getDecodedObject(key);
 857                 decval = getDecodedObject(val);
 858                 zm = zipmapSet(zm,deckey->ptr,sdslen(deckey->ptr),
 859                                   decval->ptr,sdslen(decval->ptr),NULL);
 860                 o->ptr = zm;
 861                 decrRefCount(deckey);
 862                 decrRefCount(decval);
 863                 decrRefCount(key);
 864                 decrRefCount(val);
 865             } else {
 866                 key = tryObjectEncoding(key);
 867                 val = tryObjectEncoding(val);
 868                 dictAdd((dict*)o->ptr,key,val);
 869             }
 870         }
 871     } else if (rdbtype == REDIS_RDB_TYPE_HASH_ZIPMAP  ||
 872                rdbtype == REDIS_RDB_TYPE_LIST_ZIPLIST ||
 873                rdbtype == REDIS_RDB_TYPE_SET_INTSET   ||
 874                rdbtype == REDIS_RDB_TYPE_ZSET_ZIPLIST)
 875     {
 876         robj *aux = rdbLoadStringObject(rdb);
 877
 878         if (aux == NULL) return NULL;
 879         o = createObject(REDIS_STRING,NULL); /* string is just placeholder */
 880         o->ptr = zmalloc(sdslen(aux->ptr));
 881         memcpy(o->ptr,aux->ptr,sdslen(aux->ptr));
 882         decrRefCount(aux);
 883
 884         /* Fix the object encoding, and make sure to convert the encoded
 885          * data type into the base type if accordingly to the current
 886          * configuration there are too many elements in the encoded data
 887          * type. Note that we only check the length and not max element
 888          * size as this is an O(N) scan. Eventually everything will get
 889          * converted. */
 890         switch(rdbtype) {
 891             case REDIS_RDB_TYPE_HASH_ZIPMAP:
 892                 o->type = REDIS_HASH;
 893                 o->encoding = REDIS_ENCODING_ZIPMAP;
 894                 if (zipmapLen(o->ptr) > server.hash_max_zipmap_entries)
 895                     convertToRealHash(o);
 896                 break;
 897             case REDIS_RDB_TYPE_LIST_ZIPLIST:
 898                 o->type = REDIS_LIST;
 899                 o->encoding = REDIS_ENCODING_ZIPLIST;
 900                 if (ziplistLen(o->ptr) > server.list_max_ziplist_entries)
 901                     listTypeConvert(o,REDIS_ENCODING_LINKEDLIST);
 902                 break;
 903             case REDIS_RDB_TYPE_SET_INTSET:
 904                 o->type = REDIS_SET;
 905                 o->encoding = REDIS_ENCODING_INTSET;
 906                 if (intsetLen(o->ptr) > server.set_max_intset_entries)
 907                     setTypeConvert(o,REDIS_ENCODING_HT);
 908                 break;
 909             case REDIS_RDB_TYPE_ZSET_ZIPLIST:
 910                 o->type = REDIS_ZSET;
 911                 o->encoding = REDIS_ENCODING_ZIPLIST;
 912                 if (zsetLength(o) > server.zset_max_ziplist_entries)
 913                     zsetConvert(o,REDIS_ENCODING_SKIPLIST);
 914                 break;
 915             default:
 916                 redisPanic("Unknown encoding");
 917                 break;
 918         }
 919     } else {
 920         redisPanic("Unknown object type");
 921     }
 922     return o;
 923 }
 924
 925 /* Mark that we are loading in the global state and setup the fields
 926  * needed to provide loading stats. */
 927 void startLoading(FILE *fp) {
 928     struct stat sb;
 929
 930     /* Load the DB */
 931     server.loading = 1;
 932     server.loading_start_time = time(NULL);
 933     if (fstat(fileno(fp), &sb) == -1) {
 934         server.loading_total_bytes = 1; /* just to avoid division by zero */
 935     } else {
 936         server.loading_total_bytes = sb.st_size;
 937     }
 938 }
 939
 940 /* Refresh the loading progress info */
 941 void loadingProgress(off_t pos) {
 942     server.loading_loaded_bytes = pos;
 943 }
 944
 945 /* Loading finished */
 946 void stopLoading(void) {
 947     server.loading = 0;
 948 }
 949
 950 int rdbLoad(char *filename) {
 951     uint32_t dbid;
 952     int type, retval, rdbver;
 953     redisDb *db = server.db+0;
 954     char buf[1024];
 955     time_t expiretime, now = time(NULL);
 956     long loops = 0;
 957     FILE *fp;
 958     rio rdb;
 959
 960     fp = fopen(filename,"r");
 961     if (!fp) return REDIS_ERR;
 962     if (fread(buf,9,1,fp) == 0) goto eoferr;
 963     buf[9] = '\0';
 964     if (memcmp(buf,"REDIS",5) != 0) {
 965         fclose(fp);
 966         redisLog(REDIS_WARNING,"Wrong signature trying to load DB from file");
 967         return REDIS_ERR;
 968     }
 969     rdbver = atoi(buf+5);
 970     if (rdbver < 1 || rdbver > 2) {
 971         fclose(fp);
 972         redisLog(REDIS_WARNING,"Can't handle RDB format version %d",rdbver);
 973         return REDIS_ERR;
 974     }
 975
 976     startLoading(fp);
 977     rdb = rioInitWithFile(fp);
 978     while(1) {
 979         robj *key, *val;
 980         expiretime = -1;
 981
 982         /* Serve the clients from time to time */
 983         if (!(loops++ % 1000)) {
 984             loadingProgress(rdb.tell(&rdb));
 985             aeProcessEvents(server.el, AE_FILE_EVENTS|AE_DONT_WAIT);
 986         }
 987
 988         /* Read type. */
 989         if ((type = rdbLoadType(&rdb)) == -1) goto eoferr;
 990         if (type == REDIS_RDB_OPCODE_EXPIRETIME) {
 991             if ((expiretime = rdbLoadTime(&rdb)) == -1) goto eoferr;
 992             /* We read the time so we need to read the object type again. */
 993             if ((type = rdbLoadType(&rdb)) == -1) goto eoferr;
 994         }
 995
 996         if (type == REDIS_RDB_OPCODE_EOF)
 997             break;
 998
 999         /* Handle SELECT DB opcode as a special case */
1000         if (type == REDIS_RDB_OPCODE_SELECTDB) {
1001             if ((dbid = rdbLoadLen(&rdb,NULL)) == REDIS_RDB_LENERR)
1002                 goto eoferr;
1003             if (dbid >= (unsigned)server.dbnum) {
1004                 redisLog(REDIS_WARNING,"FATAL: Data file was created with a Redis server configured to handle more than %d databases. Exiting\n", server.dbnum);
1005                 exit(1);
1006             }
1007             db = server.db+dbid;
1008             continue;
1009         }
1010         /* Read key */
1011         if ((key = rdbLoadStringObject(&rdb)) == NULL) goto eoferr;
1012         /* Read value */
1013         if ((val = rdbLoadObject(type,&rdb)) == NULL) goto eoferr;
1014         /* Check if the key already expired */
1015         if (expiretime != -1 && expiretime < now) {
1016             decrRefCount(key);
1017             decrRefCount(val);
1018             continue;
1019         }
1020         /* Add the new object in the hash table */
1021         retval = dbAdd(db,key,val);
1022         if (retval == REDIS_ERR) {
1023             redisLog(REDIS_WARNING,"Loading DB, duplicated key (%s) found! Unrecoverable error, exiting now.", key->ptr);
1024             exit(1);
1025         }
1026         /* Set the expire time if needed */
1027         if (expiretime != -1) setExpire(db,key,expiretime);
1028
1029         decrRefCount(key);
1030     }
1031     fclose(fp);
1032     stopLoading();
1033     return REDIS_OK;
1034
1035 eoferr: /* unexpected end of file is handled here with a fatal exit */
1036     redisLog(REDIS_WARNING,"Short read or OOM loading DB. Unrecoverable error, aborting now.");
1037     exit(1);
1038     return REDIS_ERR; /* Just to avoid warning */
1039 }
1040
1041 /* A background saving child (BGSAVE) terminated its work. Handle this. */
1042 void backgroundSaveDoneHandler(int exitcode, int bysignal) {
1043     if (!bysignal && exitcode == 0) {
1044         redisLog(REDIS_NOTICE,
1045             "Background saving terminated with success");
1046         server.dirty = server.dirty - server.dirty_before_bgsave;
1047         server.lastsave = time(NULL);
1048     } else if (!bysignal && exitcode != 0) {
1049         redisLog(REDIS_WARNING, "Background saving error");
1050     } else {
1051         redisLog(REDIS_WARNING,
1052             "Background saving terminated by signal %d", bysignal);
1053         rdbRemoveTempFile(server.bgsavechildpid);
1054     }
1055     server.bgsavechildpid = -1;
1056     server.bgsavethread = (pthread_t) -1;
1057     server.bgsavethread_state = REDIS_BGSAVE_THREAD_UNACTIVE;
1058     /* Possibly there are slaves waiting for a BGSAVE in order to be served
1059      * (the first stage of SYNC is a bulk transfer of dump.rdb) */
1060     updateSlavesWaitingBgsave(exitcode == 0 ? REDIS_OK : REDIS_ERR);
1061 }
1062
1063 void saveCommand(redisClient *c) {
1064     if (server.bgsavechildpid != -1 || server.bgsavethread != (pthread_t)-1) {
1065         addReplyError(c,"Background save already in progress");
1066         return;
1067     }
1068     if (rdbSave(server.dbfilename) == REDIS_OK) {
1069         addReply(c,shared.ok);
1070     } else {
1071         addReply(c,shared.err);
1072     }
1073 }
1074
1075 void bgsaveCommand(redisClient *c) {
1076     if (server.bgsavechildpid != -1 || server.bgsavethread != (pthread_t)-1) {
1077         addReplyError(c,"Background save already in progress");
1078         return;
1079     }
1080     if (rdbSaveBackground(server.dbfilename) == REDIS_OK) {
1081         addReplyStatus(c,"Background saving started");
1082     } else {
1083         addReply(c,shared.err);
1084     }
1085 }