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