src/rdb.c

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