src/rdb.c

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