src/rdb.c

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