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