1 /* 2 * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com> 3 * All rights reserved. 4 * 5 * Redistribution and use in source and binary forms, with or without 6 * modification, are permitted provided that the following conditions are met: 7 * 8 * * Redistributions of source code must retain the above copyright notice, 9 * this list of conditions and the following disclaimer. 10 * * Redistributions in binary form must reproduce the above copyright 11 * notice, this list of conditions and the following disclaimer in the 12 * documentation and/or other materials provided with the distribution. 13 * * Neither the name of Redis nor the names of its contributors may be used 14 * to endorse or promote products derived from this software without 15 * specific prior written permission. 16 * 17 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 18 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 19 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 20 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE 21 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 22 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 23 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 24 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 25 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) 26 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 27 * POSSIBILITY OF SUCH DAMAGE. 28 */ 29 30 #include "redis.h" 31 #include "lzf.h" /* LZF compression library */ 32 #include "zipmap.h" 33 #include "endianconv.h" 34 35 #include <math.h> 36 #include <sys/types.h> 37 #include <sys/time.h> 38 #include <sys/resource.h> 39 #include <sys/wait.h> 40 #include <arpa/inet.h> 41 #include <sys/stat.h> 42 43 #define RDB_LOAD_NONE 0 44 #define RDB_LOAD_ENC (1<<0) 45 #define RDB_LOAD_PLAIN (1<<1) 46 47 static int rdbWriteRaw(rio *rdb, void *p, size_t len) { 48 if (rdb && rioWrite(rdb,p,len) == 0) 49 return -1; 50 return len; 51 } 52 53 int rdbSaveType(rio *rdb, unsigned char type) { 54 return rdbWriteRaw(rdb,&type,1); 55 } 56 57 /* Load a "type" in RDB format, that is a one byte unsigned integer. 58 * This function is not only used to load object types, but also special 59 * "types" like the end-of-file type, the EXPIRE type, and so forth. */ 60 int rdbLoadType(rio *rdb) { 61 unsigned char type; 62 if (rioRead(rdb,&type,1) == 0) return -1; 63 return type; 64 } 65 66 time_t rdbLoadTime(rio *rdb) { 67 int32_t t32; 68 if (rioRead(rdb,&t32,4) == 0) return -1; 69 return (time_t)t32; 70 } 71 72 int rdbSaveMillisecondTime(rio *rdb, long long t) { 73 int64_t t64 = (int64_t) t; 74 return rdbWriteRaw(rdb,&t64,8); 75 } 76 77 long long rdbLoadMillisecondTime(rio *rdb) { 78 int64_t t64; 79 if (rioRead(rdb,&t64,8) == 0) return -1; 80 return (long long)t64; 81 } 82 83 /* Saves an encoded length. The first two bits in the first byte are used to 84 * hold the encoding type. See the REDIS_RDB_* definitions for more information 85 * on the types of encoding. */ 86 int rdbSaveLen(rio *rdb, uint32_t len) { 87 unsigned char buf[2]; 88 size_t nwritten; 89 90 if (len < (1<<6)) { 91 /* Save a 6 bit len */ 92 buf[0] = (len&0xFF)|(REDIS_RDB_6BITLEN<<6); 93 if (rdbWriteRaw(rdb,buf,1) == -1) return -1; 94 nwritten = 1; 95 } else if (len < (1<<14)) { 96 /* Save a 14 bit len */ 97 buf[0] = ((len>>8)&0xFF)|(REDIS_RDB_14BITLEN<<6); 98 buf[1] = len&0xFF; 99 if (rdbWriteRaw(rdb,buf,2) == -1) return -1; 100 nwritten = 2; 101 } else { 102 /* Save a 32 bit len */ 103 buf[0] = (REDIS_RDB_32BITLEN<<6); 104 if (rdbWriteRaw(rdb,buf,1) == -1) return -1; 105 len = htonl(len); 106 if (rdbWriteRaw(rdb,&len,4) == -1) return -1; 107 nwritten = 1+4; 108 } 109 return nwritten; 110 } 111 112 /* Load an encoded length. The "isencoded" argument is set to 1 if the length 113 * is not actually a length but an "encoding type". See the REDIS_RDB_ENC_* 114 * definitions in rdb.h for more information. */ 115 uint32_t rdbLoadLen(rio *rdb, int *isencoded) { 116 unsigned char buf[2]; 117 uint32_t len; 118 int type; 119 120 if (isencoded) *isencoded = 0; 121 if (rioRead(rdb,buf,1) == 0) return REDIS_RDB_LENERR; 122 type = (buf[0]&0xC0)>>6; 123 if (type == REDIS_RDB_ENCVAL) { 124 /* Read a 6 bit encoding type. */ 125 if (isencoded) *isencoded = 1; 126 return buf[0]&0x3F; 127 } else if (type == REDIS_RDB_6BITLEN) { 128 /* Read a 6 bit len. */ 129 return buf[0]&0x3F; 130 } else if (type == REDIS_RDB_14BITLEN) { 131 /* Read a 14 bit len. */ 132 if (rioRead(rdb,buf+1,1) == 0) return REDIS_RDB_LENERR; 133 return ((buf[0]&0x3F)<<8)|buf[1]; 134 } else { 135 /* Read a 32 bit len. */ 136 if (rioRead(rdb,&len,4) == 0) return REDIS_RDB_LENERR; 137 return ntohl(len); 138 } 139 } 140 141 /* Encodes the "value" argument as integer when it fits in the supported ranges 142 * for encoded types. If the function successfully encodes the integer, the 143 * representation is stored in the buffer pointer to by "enc" and the string 144 * length is returned. Otherwise 0 is returned. */ 145 int rdbEncodeInteger(long long value, unsigned char *enc) { 146 if (value >= -(1<<7) && value <= (1<<7)-1) { 147 enc[0] = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_INT8; 148 enc[1] = value&0xFF; 149 return 2; 150 } else if (value >= -(1<<15) && value <= (1<<15)-1) { 151 enc[0] = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_INT16; 152 enc[1] = value&0xFF; 153 enc[2] = (value>>8)&0xFF; 154 return 3; 155 } else if (value >= -((long long)1<<31) && value <= ((long long)1<<31)-1) { 156 enc[0] = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_INT32; 157 enc[1] = value&0xFF; 158 enc[2] = (value>>8)&0xFF; 159 enc[3] = (value>>16)&0xFF; 160 enc[4] = (value>>24)&0xFF; 161 return 5; 162 } else { 163 return 0; 164 } 165 } 166 167 /* Loads an integer-encoded object with the specified encoding type "enctype". 168 * The returned value changes according to the flags, see 169 * rdbGenerincLoadStringObject() for more info. */ 170 void *rdbLoadIntegerObject(rio *rdb, int enctype, int flags) { 171 int plain = flags & RDB_LOAD_PLAIN; 172 int encode = flags & RDB_LOAD_ENC; 173 unsigned char enc[4]; 174 long long val; 175 176 if (enctype == REDIS_RDB_ENC_INT8) { 177 if (rioRead(rdb,enc,1) == 0) return NULL; 178 val = (signed char)enc[0]; 179 } else if (enctype == REDIS_RDB_ENC_INT16) { 180 uint16_t v; 181 if (rioRead(rdb,enc,2) == 0) return NULL; 182 v = enc[0]|(enc[1]<<8); 183 val = (int16_t)v; 184 } else if (enctype == REDIS_RDB_ENC_INT32) { 185 uint32_t v; 186 if (rioRead(rdb,enc,4) == 0) return NULL; 187 v = enc[0]|(enc[1]<<8)|(enc[2]<<16)|(enc[3]<<24); 188 val = (int32_t)v; 189 } else { 190 val = 0; /* anti-warning */ 191 redisPanic("Unknown RDB integer encoding type"); 192 } 193 if (plain) { 194 char buf[REDIS_LONGSTR_SIZE], *p; 195 int len = ll2string(buf,sizeof(buf),val); 196 p = zmalloc(len); 197 memcpy(p,buf,len); 198 return p; 199 } else if (encode) { 200 return createStringObjectFromLongLong(val); 201 } else { 202 return createObject(REDIS_STRING,sdsfromlonglong(val)); 203 } 204 } 205 206 /* String objects in the form "2391" "-100" without any space and with a 207 * range of values that can fit in an 8, 16 or 32 bit signed value can be 208 * encoded as integers to save space */ 209 int rdbTryIntegerEncoding(char *s, size_t len, unsigned char *enc) { 210 long long value; 211 char *endptr, buf[32]; 212 213 /* Check if it's possible to encode this value as a number */ 214 value = strtoll(s, &endptr, 10); 215 if (endptr[0] != '\0') return 0; 216 ll2string(buf,32,value); 217 218 /* If the number converted back into a string is not identical 219 * then it's not possible to encode the string as integer */ 220 if (strlen(buf) != len || memcmp(buf,s,len)) return 0; 221 222 return rdbEncodeInteger(value,enc); 223 } 224 225 int rdbSaveLzfBlob(rio *rdb, void *data, size_t compress_len, 226 size_t original_len) { 227 unsigned char byte; 228 int n, nwritten = 0; 229 230 /* Data compressed! Let's save it on disk */ 231 byte = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_LZF; 232 if ((n = rdbWriteRaw(rdb,&byte,1)) == -1) goto writeerr; 233 nwritten += n; 234 235 if ((n = rdbSaveLen(rdb,compress_len)) == -1) goto writeerr; 236 nwritten += n; 237 238 if ((n = rdbSaveLen(rdb,original_len)) == -1) goto writeerr; 239 nwritten += n; 240 241 if ((n = rdbWriteRaw(rdb,data,compress_len)) == -1) goto writeerr; 242 nwritten += n; 243 244 return nwritten; 245 246 writeerr: 247 return -1; 248 } 249 250 int rdbSaveLzfStringObject(rio *rdb, unsigned char *s, size_t len) { 251 size_t comprlen, outlen; 252 void *out; 253 254 /* We require at least four bytes compression for this to be worth it */ 255 if (len <= 4) return 0; 256 outlen = len-4; 257 if ((out = zmalloc(outlen+1)) == NULL) return 0; 258 comprlen = lzf_compress(s, len, out, outlen); 259 if (comprlen == 0) { 260 zfree(out); 261 return 0; 262 } 263 size_t nwritten = rdbSaveLzfBlob(rdb, out, comprlen, len); 264 zfree(out); 265 return nwritten; 266 } 267 268 /* Load an LZF compressed string in RDB format. The returned value 269 * changes according to 'flags'. For more info check the 270 * rdbGenericLoadStringObject() function. */ 271 void *rdbLoadLzfStringObject(rio *rdb, int flags) { 272 int plain = flags & RDB_LOAD_PLAIN; 273 unsigned int len, clen; 274 unsigned char *c = NULL; 275 sds val = NULL; 276 277 if ((clen = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL; 278 if ((len = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL; 279 if ((c = zmalloc(clen)) == NULL) goto err; 280 281 /* Allocate our target according to the uncompressed size. */ 282 if (plain) { 283 val = zmalloc(len); 284 } else { 285 if ((val = sdsnewlen(NULL,len)) == NULL) goto err; 286 } 287 288 /* Load the compressed representation and uncompress it to target. */ 289 if (rioRead(rdb,c,clen) == 0) goto err; 290 if (lzf_decompress(c,clen,val,len) == 0) goto err; 291 zfree(c); 292 293 if (plain) 294 return val; 295 else 296 return createObject(REDIS_STRING,val); 297 err: 298 zfree(c); 299 if (plain) 300 zfree(val); 301 else 302 sdsfree(val); 303 return NULL; 304 } 305 306 /* Save a string object as [len][data] on disk. If the object is a string 307 * representation of an integer value we try to save it in a special form */ 308 int rdbSaveRawString(rio *rdb, unsigned char *s, size_t len) { 309 int enclen; 310 int n, nwritten = 0; 311 312 /* Try integer encoding */ 313 if (len <= 11) { 314 unsigned char buf[5]; 315 if ((enclen = rdbTryIntegerEncoding((char*)s,len,buf)) > 0) { 316 if (rdbWriteRaw(rdb,buf,enclen) == -1) return -1; 317 return enclen; 318 } 319 } 320 321 /* Try LZF compression - under 20 bytes it's unable to compress even 322 * aaaaaaaaaaaaaaaaaa so skip it */ 323 if (server.rdb_compression && len > 20) { 324 n = rdbSaveLzfStringObject(rdb,s,len); 325 if (n == -1) return -1; 326 if (n > 0) return n; 327 /* Return value of 0 means data can't be compressed, save the old way */ 328 } 329 330 /* Store verbatim */ 331 if ((n = rdbSaveLen(rdb,len)) == -1) return -1; 332 nwritten += n; 333 if (len > 0) { 334 if (rdbWriteRaw(rdb,s,len) == -1) return -1; 335 nwritten += len; 336 } 337 return nwritten; 338 } 339 340 /* Save a long long value as either an encoded string or a string. */ 341 int rdbSaveLongLongAsStringObject(rio *rdb, long long value) { 342 unsigned char buf[32]; 343 int n, nwritten = 0; 344 int enclen = rdbEncodeInteger(value,buf); 345 if (enclen > 0) { 346 return rdbWriteRaw(rdb,buf,enclen); 347 } else { 348 /* Encode as string */ 349 enclen = ll2string((char*)buf,32,value); 350 redisAssert(enclen < 32); 351 if ((n = rdbSaveLen(rdb,enclen)) == -1) return -1; 352 nwritten += n; 353 if ((n = rdbWriteRaw(rdb,buf,enclen)) == -1) return -1; 354 nwritten += n; 355 } 356 return nwritten; 357 } 358 359 /* Like rdbSaveStringObjectRaw() but handle encoded objects */ 360 int rdbSaveStringObject(rio *rdb, robj *obj) { 361 /* Avoid to decode the object, then encode it again, if the 362 * object is already integer encoded. */ 363 if (obj->encoding == REDIS_ENCODING_INT) { 364 return rdbSaveLongLongAsStringObject(rdb,(long)obj->ptr); 365 } else { 366 redisAssertWithInfo(NULL,obj,sdsEncodedObject(obj)); 367 return rdbSaveRawString(rdb,obj->ptr,sdslen(obj->ptr)); 368 } 369 } 370 371 /* Load a string object from an RDB file according to flags: 372 * 373 * RDB_LOAD_NONE (no flags): load an RDB object, unencoded. 374 * RDB_LOAD_ENC: If the returned type is a Redis object, try to 375 * encode it in a special way to be more memory 376 * efficient. When this flag is passed the function 377 * no longer guarantees that obj->ptr is an SDS string. 378 * RDB_LOAD_PLAIN: Return a plain string allocated with zmalloc() 379 * instead of a Redis object. 380 */ 381 void *rdbGenericLoadStringObject(rio *rdb, int flags) { 382 int encode = flags & RDB_LOAD_ENC; 383 int plain = flags & RDB_LOAD_PLAIN; 384 int isencoded; 385 uint32_t len; 386 387 len = rdbLoadLen(rdb,&isencoded); 388 if (isencoded) { 389 switch(len) { 390 case REDIS_RDB_ENC_INT8: 391 case REDIS_RDB_ENC_INT16: 392 case REDIS_RDB_ENC_INT32: 393 return rdbLoadIntegerObject(rdb,len,flags); 394 case REDIS_RDB_ENC_LZF: 395 return rdbLoadLzfStringObject(rdb,flags); 396 default: 397 redisPanic("Unknown RDB encoding type"); 398 } 399 } 400 401 if (len == REDIS_RDB_LENERR) return NULL; 402 if (!plain) { 403 robj *o = encode ? createStringObject(NULL,len) : 404 createRawStringObject(NULL,len); 405 if (len && rioRead(rdb,o->ptr,len) == 0) { 406 decrRefCount(o); 407 return NULL; 408 } 409 return o; 410 } else { 411 void *buf = zmalloc(len); 412 if (len && rioRead(rdb,buf,len) == 0) { 413 zfree(buf); 414 return NULL; 415 } 416 return buf; 417 } 418 } 419 420 robj *rdbLoadStringObject(rio *rdb) { 421 return rdbGenericLoadStringObject(rdb,RDB_LOAD_NONE); 422 } 423 424 robj *rdbLoadEncodedStringObject(rio *rdb) { 425 return rdbGenericLoadStringObject(rdb,RDB_LOAD_ENC); 426 } 427 428 /* Save a double value. Doubles are saved as strings prefixed by an unsigned 429 * 8 bit integer specifying the length of the representation. 430 * This 8 bit integer has special values in order to specify the following 431 * conditions: 432 * 253: not a number 433 * 254: + inf 434 * 255: - inf 435 */ 436 int rdbSaveDoubleValue(rio *rdb, double val) { 437 unsigned char buf[128]; 438 int len; 439 440 if (isnan(val)) { 441 buf[0] = 253; 442 len = 1; 443 } else if (!isfinite(val)) { 444 len = 1; 445 buf[0] = (val < 0) ? 255 : 254; 446 } else { 447 #if (DBL_MANT_DIG >= 52) && (LLONG_MAX == 0x7fffffffffffffffLL) 448 /* Check if the float is in a safe range to be casted into a 449 * long long. We are assuming that long long is 64 bit here. 450 * Also we are assuming that there are no implementations around where 451 * double has precision < 52 bit. 452 * 453 * Under this assumptions we test if a double is inside an interval 454 * where casting to long long is safe. Then using two castings we 455 * make sure the decimal part is zero. If all this is true we use 456 * integer printing function that is much faster. */ 457 double min = -4503599627370495; /* (2^52)-1 */ 458 double max = 4503599627370496; /* -(2^52) */ 459 if (val > min && val < max && val == ((double)((long long)val))) 460 ll2string((char*)buf+1,sizeof(buf)-1,(long long)val); 461 else 462 #endif 463 snprintf((char*)buf+1,sizeof(buf)-1,"%.17g",val); 464 buf[0] = strlen((char*)buf+1); 465 len = buf[0]+1; 466 } 467 return rdbWriteRaw(rdb,buf,len); 468 } 469 470 /* For information about double serialization check rdbSaveDoubleValue() */ 471 int rdbLoadDoubleValue(rio *rdb, double *val) { 472 char buf[256]; 473 unsigned char len; 474 475 if (rioRead(rdb,&len,1) == 0) return -1; 476 switch(len) { 477 case 255: *val = R_NegInf; return 0; 478 case 254: *val = R_PosInf; return 0; 479 case 253: *val = R_Nan; return 0; 480 default: 481 if (rioRead(rdb,buf,len) == 0) return -1; 482 buf[len] = '\0'; 483 sscanf(buf, "%lg", val); 484 return 0; 485 } 486 } 487 488 /* Save the object type of object "o". */ 489 int rdbSaveObjectType(rio *rdb, robj *o) { 490 switch (o->type) { 491 case REDIS_STRING: 492 return rdbSaveType(rdb,REDIS_RDB_TYPE_STRING); 493 case REDIS_LIST: 494 if (o->encoding == REDIS_ENCODING_QUICKLIST) 495 return rdbSaveType(rdb,REDIS_RDB_TYPE_LIST_QUICKLIST); 496 else 497 redisPanic("Unknown list encoding"); 498 case REDIS_SET: 499 if (o->encoding == REDIS_ENCODING_INTSET) 500 return rdbSaveType(rdb,REDIS_RDB_TYPE_SET_INTSET); 501 else if (o->encoding == REDIS_ENCODING_HT) 502 return rdbSaveType(rdb,REDIS_RDB_TYPE_SET); 503 else 504 redisPanic("Unknown set encoding"); 505 case REDIS_ZSET: 506 if (o->encoding == REDIS_ENCODING_ZIPLIST) 507 return rdbSaveType(rdb,REDIS_RDB_TYPE_ZSET_ZIPLIST); 508 else if (o->encoding == REDIS_ENCODING_SKIPLIST) 509 return rdbSaveType(rdb,REDIS_RDB_TYPE_ZSET); 510 else 511 redisPanic("Unknown sorted set encoding"); 512 case REDIS_HASH: 513 if (o->encoding == REDIS_ENCODING_ZIPLIST) 514 return rdbSaveType(rdb,REDIS_RDB_TYPE_HASH_ZIPLIST); 515 else if (o->encoding == REDIS_ENCODING_HT) 516 return rdbSaveType(rdb,REDIS_RDB_TYPE_HASH); 517 else 518 redisPanic("Unknown hash encoding"); 519 default: 520 redisPanic("Unknown object type"); 521 } 522 return -1; /* avoid warning */ 523 } 524 525 /* Use rdbLoadType() to load a TYPE in RDB format, but returns -1 if the 526 * type is not specifically a valid Object Type. */ 527 int rdbLoadObjectType(rio *rdb) { 528 int type; 529 if ((type = rdbLoadType(rdb)) == -1) return -1; 530 if (!rdbIsObjectType(type)) return -1; 531 return type; 532 } 533 534 /* Save a Redis object. Returns -1 on error, number of bytes written on success. */ 535 int rdbSaveObject(rio *rdb, robj *o) { 536 int n = 0, nwritten = 0; 537 538 if (o->type == REDIS_STRING) { 539 /* Save a string value */ 540 if ((n = rdbSaveStringObject(rdb,o)) == -1) return -1; 541 nwritten += n; 542 } else if (o->type == REDIS_LIST) { 543 /* Save a list value */ 544 if (o->encoding == REDIS_ENCODING_QUICKLIST) { 545 quicklist *ql = o->ptr; 546 quicklistNode *node = ql->head; 547 548 if ((n = rdbSaveLen(rdb,ql->len)) == -1) return -1; 549 nwritten += n; 550 551 do { 552 if (quicklistNodeIsCompressed(node)) { 553 void *data; 554 size_t compress_len = quicklistGetLzf(node, &data); 555 if ((n = rdbSaveLzfBlob(rdb,data,compress_len,node->sz)) == -1) return -1; 556 nwritten += n; 557 } else { 558 if ((n = rdbSaveRawString(rdb,node->zl,node->sz)) == -1) return -1; 559 nwritten += n; 560 } 561 } while ((node = node->next)); 562 } else { 563 redisPanic("Unknown list encoding"); 564 } 565 } else if (o->type == REDIS_SET) { 566 /* Save a set value */ 567 if (o->encoding == REDIS_ENCODING_HT) { 568 dict *set = o->ptr; 569 dictIterator *di = dictGetIterator(set); 570 dictEntry *de; 571 572 if ((n = rdbSaveLen(rdb,dictSize(set))) == -1) return -1; 573 nwritten += n; 574 575 while((de = dictNext(di)) != NULL) { 576 robj *eleobj = dictGetKey(de); 577 if ((n = rdbSaveStringObject(rdb,eleobj)) == -1) return -1; 578 nwritten += n; 579 } 580 dictReleaseIterator(di); 581 } else if (o->encoding == REDIS_ENCODING_INTSET) { 582 size_t l = intsetBlobLen((intset*)o->ptr); 583 584 if ((n = rdbSaveRawString(rdb,o->ptr,l)) == -1) return -1; 585 nwritten += n; 586 } else { 587 redisPanic("Unknown set encoding"); 588 } 589 } else if (o->type == REDIS_ZSET) { 590 /* Save a sorted set value */ 591 if (o->encoding == REDIS_ENCODING_ZIPLIST) { 592 size_t l = ziplistBlobLen((unsigned char*)o->ptr); 593 594 if ((n = rdbSaveRawString(rdb,o->ptr,l)) == -1) return -1; 595 nwritten += n; 596 } else if (o->encoding == REDIS_ENCODING_SKIPLIST) { 597 zset *zs = o->ptr; 598 dictIterator *di = dictGetIterator(zs->dict); 599 dictEntry *de; 600 601 if ((n = rdbSaveLen(rdb,dictSize(zs->dict))) == -1) return -1; 602 nwritten += n; 603 604 while((de = dictNext(di)) != NULL) { 605 robj *eleobj = dictGetKey(de); 606 double *score = dictGetVal(de); 607 608 if ((n = rdbSaveStringObject(rdb,eleobj)) == -1) return -1; 609 nwritten += n; 610 if ((n = rdbSaveDoubleValue(rdb,*score)) == -1) return -1; 611 nwritten += n; 612 } 613 dictReleaseIterator(di); 614 } else { 615 redisPanic("Unknown sorted set encoding"); 616 } 617 } else if (o->type == REDIS_HASH) { 618 /* Save a hash value */ 619 if (o->encoding == REDIS_ENCODING_ZIPLIST) { 620 size_t l = ziplistBlobLen((unsigned char*)o->ptr); 621 622 if ((n = rdbSaveRawString(rdb,o->ptr,l)) == -1) return -1; 623 nwritten += n; 624 625 } else if (o->encoding == REDIS_ENCODING_HT) { 626 dictIterator *di = dictGetIterator(o->ptr); 627 dictEntry *de; 628 629 if ((n = rdbSaveLen(rdb,dictSize((dict*)o->ptr))) == -1) return -1; 630 nwritten += n; 631 632 while((de = dictNext(di)) != NULL) { 633 robj *key = dictGetKey(de); 634 robj *val = dictGetVal(de); 635 636 if ((n = rdbSaveStringObject(rdb,key)) == -1) return -1; 637 nwritten += n; 638 if ((n = rdbSaveStringObject(rdb,val)) == -1) return -1; 639 nwritten += n; 640 } 641 dictReleaseIterator(di); 642 643 } else { 644 redisPanic("Unknown hash encoding"); 645 } 646 647 } else { 648 redisPanic("Unknown object type"); 649 } 650 return nwritten; 651 } 652 653 /* Return the length the object will have on disk if saved with 654 * the rdbSaveObject() function. Currently we use a trick to get 655 * this length with very little changes to the code. In the future 656 * we could switch to a faster solution. */ 657 off_t rdbSavedObjectLen(robj *o) { 658 int len = rdbSaveObject(NULL,o); 659 redisAssertWithInfo(NULL,o,len != -1); 660 return len; 661 } 662 663 /* Save a key-value pair, with expire time, type, key, value. 664 * On error -1 is returned. 665 * On success if the key was actually saved 1 is returned, otherwise 0 666 * is returned (the key was already expired). */ 667 int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val, 668 long long expiretime, long long now) 669 { 670 /* Save the expire time */ 671 if (expiretime != -1) { 672 /* If this key is already expired skip it */ 673 if (expiretime < now) return 0; 674 if (rdbSaveType(rdb,REDIS_RDB_OPCODE_EXPIRETIME_MS) == -1) return -1; 675 if (rdbSaveMillisecondTime(rdb,expiretime) == -1) return -1; 676 } 677 678 /* Save type, key, value */ 679 if (rdbSaveObjectType(rdb,val) == -1) return -1; 680 if (rdbSaveStringObject(rdb,key) == -1) return -1; 681 if (rdbSaveObject(rdb,val) == -1) return -1; 682 return 1; 683 } 684 685 /* Save an AUX field. */ 686 int rdbSaveAuxField(rio *rdb, void *key, size_t keylen, void *val, size_t vallen) { 687 if (rdbSaveType(rdb,REDIS_RDB_OPCODE_AUX) == -1) return -1; 688 if (rdbSaveRawString(rdb,key,keylen) == -1) return -1; 689 if (rdbSaveRawString(rdb,val,vallen) == -1) return -1; 690 return 1; 691 } 692 693 /* Wrapper for rdbSaveAuxField() used when key/val length can be obtained 694 * with strlen(). */ 695 int rdbSaveAuxFieldStrStr(rio *rdb, char *key, char *val) { 696 return rdbSaveAuxField(rdb,key,strlen(key),val,strlen(val)); 697 } 698 699 /* Wrapper for strlen(key) + integer type (up to long long range). */ 700 int rdbSaveAuxFieldStrInt(rio *rdb, char *key, long long val) { 701 char buf[REDIS_LONGSTR_SIZE]; 702 int vlen = ll2string(buf,sizeof(buf),val); 703 return rdbSaveAuxField(rdb,key,strlen(key),buf,vlen); 704 } 705 706 /* Save a few default AUX fields with information about the RDB generated. */ 707 int rdbSaveInfoAuxFields(rio *rdb) { 708 int redis_bits = (sizeof(void*) == 8) ? 64 : 32; 709 710 /* Add a few fields about the state when the RDB was created. */ 711 if (rdbSaveAuxFieldStrStr(rdb,"redis-ver",REDIS_VERSION) == -1) return -1; 712 if (rdbSaveAuxFieldStrInt(rdb,"redis-bits",redis_bits) == -1) return -1; 713 if (rdbSaveAuxFieldStrInt(rdb,"ctime",time(NULL)) == -1) return -1; 714 if (rdbSaveAuxFieldStrInt(rdb,"used-mem",zmalloc_used_memory()) == -1) return -1; 715 return 1; 716 } 717 718 /* Produces a dump of the database in RDB format sending it to the specified 719 * Redis I/O channel. On success REDIS_OK is returned, otherwise REDIS_ERR 720 * is returned and part of the output, or all the output, can be 721 * missing because of I/O errors. 722 * 723 * When the function returns REDIS_ERR and if 'error' is not NULL, the 724 * integer pointed by 'error' is set to the value of errno just after the I/O 725 * error. */ 726 int rdbSaveRio(rio *rdb, int *error) { 727 dictIterator *di = NULL; 728 dictEntry *de; 729 char magic[10]; 730 int j; 731 long long now = mstime(); 732 uint64_t cksum; 733 734 if (server.rdb_checksum) 735 rdb->update_cksum = rioGenericUpdateChecksum; 736 snprintf(magic,sizeof(magic),"REDIS%04d",REDIS_RDB_VERSION); 737 if (rdbWriteRaw(rdb,magic,9) == -1) goto werr; 738 if (rdbSaveInfoAuxFields(rdb) == -1) goto werr; 739 740 for (j = 0; j < server.dbnum; j++) { 741 redisDb *db = server.db+j; 742 dict *d = db->dict; 743 if (dictSize(d) == 0) continue; 744 di = dictGetSafeIterator(d); 745 if (!di) return REDIS_ERR; 746 747 /* Write the SELECT DB opcode */ 748 if (rdbSaveType(rdb,REDIS_RDB_OPCODE_SELECTDB) == -1) goto werr; 749 if (rdbSaveLen(rdb,j) == -1) goto werr; 750 751 /* Write the RESIZE DB opcode. We trim the size to UINT32_MAX, which 752 * is currently the largest type we are able to represent in RDB sizes. 753 * However this does not limit the actual size of the DB to load since 754 * these sizes are just hints to resize the hash tables. */ 755 uint32_t db_size, expires_size; 756 db_size = (dictSize(db->dict) <= UINT32_MAX) ? 757 dictSize(db->dict) : 758 UINT32_MAX; 759 expires_size = (dictSize(db->dict) <= UINT32_MAX) ? 760 dictSize(db->expires) : 761 UINT32_MAX; 762 if (rdbSaveType(rdb,REDIS_RDB_OPCODE_RESIZEDB) == -1) goto werr; 763 if (rdbSaveLen(rdb,db_size) == -1) goto werr; 764 if (rdbSaveLen(rdb,expires_size) == -1) goto werr; 765 766 /* Iterate this DB writing every entry */ 767 while((de = dictNext(di)) != NULL) { 768 sds keystr = dictGetKey(de); 769 robj key, *o = dictGetVal(de); 770 long long expire; 771 772 initStaticStringObject(key,keystr); 773 expire = getExpire(db,&key); 774 if (rdbSaveKeyValuePair(rdb,&key,o,expire,now) == -1) goto werr; 775 } 776 dictReleaseIterator(di); 777 } 778 di = NULL; /* So that we don't release it again on error. */ 779 780 /* EOF opcode */ 781 if (rdbSaveType(rdb,REDIS_RDB_OPCODE_EOF) == -1) goto werr; 782 783 /* CRC64 checksum. It will be zero if checksum computation is disabled, the 784 * loading code skips the check in this case. */ 785 cksum = rdb->cksum; 786 memrev64ifbe(&cksum); 787 if (rioWrite(rdb,&cksum,8) == 0) goto werr; 788 return REDIS_OK; 789 790 werr: 791 if (error) *error = errno; 792 if (di) dictReleaseIterator(di); 793 return REDIS_ERR; 794 } 795 796 /* This is just a wrapper to rdbSaveRio() that additionally adds a prefix 797 * and a suffix to the generated RDB dump. The prefix is: 798 * 799 * $EOF:<40 bytes unguessable hex string>\r\n 800 * 801 * While the suffix is the 40 bytes hex string we announced in the prefix. 802 * This way processes receiving the payload can understand when it ends 803 * without doing any processing of the content. */ 804 int rdbSaveRioWithEOFMark(rio *rdb, int *error) { 805 char eofmark[REDIS_EOF_MARK_SIZE]; 806 807 getRandomHexChars(eofmark,REDIS_EOF_MARK_SIZE); 808 if (error) *error = 0; 809 if (rioWrite(rdb,"$EOF:",5) == 0) goto werr; 810 if (rioWrite(rdb,eofmark,REDIS_EOF_MARK_SIZE) == 0) goto werr; 811 if (rioWrite(rdb,"\r\n",2) == 0) goto werr; 812 if (rdbSaveRio(rdb,error) == REDIS_ERR) goto werr; 813 if (rioWrite(rdb,eofmark,REDIS_EOF_MARK_SIZE) == 0) goto werr; 814 return REDIS_OK; 815 816 werr: /* Write error. */ 817 /* Set 'error' only if not already set by rdbSaveRio() call. */ 818 if (error && *error == 0) *error = errno; 819 return REDIS_ERR; 820 } 821 822 /* Save the DB on disk. Return REDIS_ERR on error, REDIS_OK on success. */ 823 int rdbSave(char *filename) { 824 char tmpfile[256]; 825 FILE *fp; 826 rio rdb; 827 int error = 0; 828 829 snprintf(tmpfile,256,"temp-%d.rdb", (int) getpid()); 830 fp = fopen(tmpfile,"w"); 831 if (!fp) { 832 redisLog(REDIS_WARNING, "Failed opening .rdb for saving: %s", 833 strerror(errno)); 834 return REDIS_ERR; 835 } 836 837 rioInitWithFile(&rdb,fp); 838 if (rdbSaveRio(&rdb,&error) == REDIS_ERR) { 839 errno = error; 840 goto werr; 841 } 842 843 /* Make sure data will not remain on the OS's output buffers */ 844 if (fflush(fp) == EOF) goto werr; 845 if (fsync(fileno(fp)) == -1) goto werr; 846 if (fclose(fp) == EOF) goto werr; 847 848 /* Use RENAME to make sure the DB file is changed atomically only 849 * if the generate DB file is ok. */ 850 if (rename(tmpfile,filename) == -1) { 851 redisLog(REDIS_WARNING,"Error moving temp DB file on the final destination: %s", strerror(errno)); 852 unlink(tmpfile); 853 return REDIS_ERR; 854 } 855 redisLog(REDIS_NOTICE,"DB saved on disk"); 856 server.dirty = 0; 857 server.lastsave = time(NULL); 858 server.lastbgsave_status = REDIS_OK; 859 return REDIS_OK; 860 861 werr: 862 fclose(fp); 863 unlink(tmpfile); 864 redisLog(REDIS_WARNING,"Write error saving DB on disk: %s", strerror(errno)); 865 return REDIS_ERR; 866 } 867 868 int rdbSaveBackground(char *filename) { 869 pid_t childpid; 870 long long start; 871 872 if (server.rdb_child_pid != -1) return REDIS_ERR; 873 874 server.dirty_before_bgsave = server.dirty; 875 server.lastbgsave_try = time(NULL); 876 877 start = ustime(); 878 if ((childpid = fork()) == 0) { 879 int retval; 880 881 /* Child */ 882 closeListeningSockets(0); 883 redisSetProcTitle("redis-rdb-bgsave"); 884 retval = rdbSave(filename); 885 if (retval == REDIS_OK) { 886 size_t private_dirty = zmalloc_get_private_dirty(); 887 888 if (private_dirty) { 889 redisLog(REDIS_NOTICE, 890 "RDB: %zu MB of memory used by copy-on-write", 891 private_dirty/(1024*1024)); 892 } 893 } 894 exitFromChild((retval == REDIS_OK) ? 0 : 1); 895 } else { 896 /* Parent */ 897 server.stat_fork_time = ustime()-start; 898 server.stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / server.stat_fork_time / (1024*1024*1024); /* GB per second. */ 899 latencyAddSampleIfNeeded("fork",server.stat_fork_time/1000); 900 if (childpid == -1) { 901 server.lastbgsave_status = REDIS_ERR; 902 redisLog(REDIS_WARNING,"Can't save in background: fork: %s", 903 strerror(errno)); 904 return REDIS_ERR; 905 } 906 redisLog(REDIS_NOTICE,"Background saving started by pid %d",childpid); 907 server.rdb_save_time_start = time(NULL); 908 server.rdb_child_pid = childpid; 909 server.rdb_child_type = REDIS_RDB_CHILD_TYPE_DISK; 910 updateDictResizePolicy(); 911 return REDIS_OK; 912 } 913 return REDIS_OK; /* unreached */ 914 } 915 916 void rdbRemoveTempFile(pid_t childpid) { 917 char tmpfile[256]; 918 919 snprintf(tmpfile,sizeof(tmpfile),"temp-%d.rdb", (int) childpid); 920 unlink(tmpfile); 921 } 922 923 /* Load a Redis object of the specified type from the specified file. 924 * On success a newly allocated object is returned, otherwise NULL. */ 925 robj *rdbLoadObject(int rdbtype, rio *rdb) { 926 robj *o, *ele, *dec; 927 size_t len; 928 unsigned int i; 929 930 if (rdbtype == REDIS_RDB_TYPE_STRING) { 931 /* Read string value */ 932 if ((o = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL; 933 o = tryObjectEncoding(o); 934 } else if (rdbtype == REDIS_RDB_TYPE_LIST) { 935 /* Read list value */ 936 if ((len = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL; 937 938 o = createQuicklistObject(); 939 quicklistSetOptions(o->ptr, server.list_max_ziplist_size, 940 server.list_compress_depth); 941 942 /* Load every single element of the list */ 943 while(len--) { 944 if ((ele = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL; 945 dec = getDecodedObject(ele); 946 size_t len = sdslen(dec->ptr); 947 quicklistPushTail(o->ptr, dec->ptr, len); 948 decrRefCount(dec); 949 decrRefCount(ele); 950 } 951 } else if (rdbtype == REDIS_RDB_TYPE_SET) { 952 /* Read list/set value */ 953 if ((len = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL; 954 955 /* Use a regular set when there are too many entries. */ 956 if (len > server.set_max_intset_entries) { 957 o = createSetObject(); 958 /* It's faster to expand the dict to the right size asap in order 959 * to avoid rehashing */ 960 if (len > DICT_HT_INITIAL_SIZE) 961 dictExpand(o->ptr,len); 962 } else { 963 o = createIntsetObject(); 964 } 965 966 /* Load every single element of the list/set */ 967 for (i = 0; i < len; i++) { 968 long long llval; 969 if ((ele = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL; 970 ele = tryObjectEncoding(ele); 971 972 if (o->encoding == REDIS_ENCODING_INTSET) { 973 /* Fetch integer value from element */ 974 if (isObjectRepresentableAsLongLong(ele,&llval) == REDIS_OK) { 975 o->ptr = intsetAdd(o->ptr,llval,NULL); 976 } else { 977 setTypeConvert(o,REDIS_ENCODING_HT); 978 dictExpand(o->ptr,len); 979 } 980 } 981 982 /* This will also be called when the set was just converted 983 * to a regular hash table encoded set */ 984 if (o->encoding == REDIS_ENCODING_HT) { 985 dictAdd((dict*)o->ptr,ele,NULL); 986 } else { 987 decrRefCount(ele); 988 } 989 } 990 } else if (rdbtype == REDIS_RDB_TYPE_ZSET) { 991 /* Read list/set value */ 992 size_t zsetlen; 993 size_t maxelelen = 0; 994 zset *zs; 995 996 if ((zsetlen = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL; 997 o = createZsetObject(); 998 zs = o->ptr; 999 1000 /* Load every single element of the list/set */ 1001 while(zsetlen--) { 1002 robj *ele; 1003 double score; 1004 zskiplistNode *znode; 1005 1006 if ((ele = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL; 1007 ele = tryObjectEncoding(ele); 1008 if (rdbLoadDoubleValue(rdb,&score) == -1) return NULL; 1009 1010 /* Don't care about integer-encoded strings. */ 1011 if (sdsEncodedObject(ele) && sdslen(ele->ptr) > maxelelen) 1012 maxelelen = sdslen(ele->ptr); 1013 1014 znode = zslInsert(zs->zsl,score,ele); 1015 dictAdd(zs->dict,ele,&znode->score); 1016 incrRefCount(ele); /* added to skiplist */ 1017 } 1018 1019 /* Convert *after* loading, since sorted sets are not stored ordered. */ 1020 if (zsetLength(o) <= server.zset_max_ziplist_entries && 1021 maxelelen <= server.zset_max_ziplist_value) 1022 zsetConvert(o,REDIS_ENCODING_ZIPLIST); 1023 } else if (rdbtype == REDIS_RDB_TYPE_HASH) { 1024 size_t len; 1025 int ret; 1026 1027 len = rdbLoadLen(rdb, NULL); 1028 if (len == REDIS_RDB_LENERR) return NULL; 1029 1030 o = createHashObject(); 1031 1032 /* Too many entries? Use a hash table. */ 1033 if (len > server.hash_max_ziplist_entries) 1034 hashTypeConvert(o, REDIS_ENCODING_HT); 1035 1036 /* Load every field and value into the ziplist */ 1037 while (o->encoding == REDIS_ENCODING_ZIPLIST && len > 0) { 1038 robj *field, *value; 1039 1040 len--; 1041 /* Load raw strings */ 1042 field = rdbLoadStringObject(rdb); 1043 if (field == NULL) return NULL; 1044 redisAssert(sdsEncodedObject(field)); 1045 value = rdbLoadStringObject(rdb); 1046 if (value == NULL) return NULL; 1047 redisAssert(sdsEncodedObject(value)); 1048 1049 /* Add pair to ziplist */ 1050 o->ptr = ziplistPush(o->ptr, field->ptr, sdslen(field->ptr), ZIPLIST_TAIL); 1051 o->ptr = ziplistPush(o->ptr, value->ptr, sdslen(value->ptr), ZIPLIST_TAIL); 1052 /* Convert to hash table if size threshold is exceeded */ 1053 if (sdslen(field->ptr) > server.hash_max_ziplist_value || 1054 sdslen(value->ptr) > server.hash_max_ziplist_value) 1055 { 1056 decrRefCount(field); 1057 decrRefCount(value); 1058 hashTypeConvert(o, REDIS_ENCODING_HT); 1059 break; 1060 } 1061 decrRefCount(field); 1062 decrRefCount(value); 1063 } 1064 1065 /* Load remaining fields and values into the hash table */ 1066 while (o->encoding == REDIS_ENCODING_HT && len > 0) { 1067 robj *field, *value; 1068 1069 len--; 1070 /* Load encoded strings */ 1071 field = rdbLoadEncodedStringObject(rdb); 1072 if (field == NULL) return NULL; 1073 value = rdbLoadEncodedStringObject(rdb); 1074 if (value == NULL) return NULL; 1075 1076 field = tryObjectEncoding(field); 1077 value = tryObjectEncoding(value); 1078 1079 /* Add pair to hash table */ 1080 ret = dictAdd((dict*)o->ptr, field, value); 1081 redisAssert(ret == DICT_OK); 1082 } 1083 1084 /* All pairs should be read by now */ 1085 redisAssert(len == 0); 1086 } else if (rdbtype == REDIS_RDB_TYPE_LIST_QUICKLIST) { 1087 if ((len = rdbLoadLen(rdb,NULL)) == REDIS_RDB_LENERR) return NULL; 1088 o = createQuicklistObject(); 1089 quicklistSetOptions(o->ptr, server.list_max_ziplist_size, 1090 server.list_compress_depth); 1091 1092 while (len--) { 1093 unsigned char *zl = rdbGenericLoadStringObject(rdb,RDB_LOAD_PLAIN); 1094 if (zl == NULL) return NULL; 1095 quicklistAppendZiplist(o->ptr, zl); 1096 } 1097 } else if (rdbtype == REDIS_RDB_TYPE_HASH_ZIPMAP || 1098 rdbtype == REDIS_RDB_TYPE_LIST_ZIPLIST || 1099 rdbtype == REDIS_RDB_TYPE_SET_INTSET || 1100 rdbtype == REDIS_RDB_TYPE_ZSET_ZIPLIST || 1101 rdbtype == REDIS_RDB_TYPE_HASH_ZIPLIST) 1102 { 1103 unsigned char *encoded = rdbGenericLoadStringObject(rdb,RDB_LOAD_PLAIN); 1104 if (encoded == NULL) return NULL; 1105 o = createObject(REDIS_STRING,encoded); /* Obj type fixed below. */ 1106 1107 /* Fix the object encoding, and make sure to convert the encoded 1108 * data type into the base type if accordingly to the current 1109 * configuration there are too many elements in the encoded data 1110 * type. Note that we only check the length and not max element 1111 * size as this is an O(N) scan. Eventually everything will get 1112 * converted. */ 1113 switch(rdbtype) { 1114 case REDIS_RDB_TYPE_HASH_ZIPMAP: 1115 /* Convert to ziplist encoded hash. This must be deprecated 1116 * when loading dumps created by Redis 2.4 gets deprecated. */ 1117 { 1118 unsigned char *zl = ziplistNew(); 1119 unsigned char *zi = zipmapRewind(o->ptr); 1120 unsigned char *fstr, *vstr; 1121 unsigned int flen, vlen; 1122 unsigned int maxlen = 0; 1123 1124 while ((zi = zipmapNext(zi, &fstr, &flen, &vstr, &vlen)) != NULL) { 1125 if (flen > maxlen) maxlen = flen; 1126 if (vlen > maxlen) maxlen = vlen; 1127 zl = ziplistPush(zl, fstr, flen, ZIPLIST_TAIL); 1128 zl = ziplistPush(zl, vstr, vlen, ZIPLIST_TAIL); 1129 } 1130 1131 zfree(o->ptr); 1132 o->ptr = zl; 1133 o->type = REDIS_HASH; 1134 o->encoding = REDIS_ENCODING_ZIPLIST; 1135 1136 if (hashTypeLength(o) > server.hash_max_ziplist_entries || 1137 maxlen > server.hash_max_ziplist_value) 1138 { 1139 hashTypeConvert(o, REDIS_ENCODING_HT); 1140 } 1141 } 1142 break; 1143 case REDIS_RDB_TYPE_LIST_ZIPLIST: 1144 o->type = REDIS_LIST; 1145 o->encoding = REDIS_ENCODING_ZIPLIST; 1146 listTypeConvert(o,REDIS_ENCODING_QUICKLIST); 1147 break; 1148 case REDIS_RDB_TYPE_SET_INTSET: 1149 o->type = REDIS_SET; 1150 o->encoding = REDIS_ENCODING_INTSET; 1151 if (intsetLen(o->ptr) > server.set_max_intset_entries) 1152 setTypeConvert(o,REDIS_ENCODING_HT); 1153 break; 1154 case REDIS_RDB_TYPE_ZSET_ZIPLIST: 1155 o->type = REDIS_ZSET; 1156 o->encoding = REDIS_ENCODING_ZIPLIST; 1157 if (zsetLength(o) > server.zset_max_ziplist_entries) 1158 zsetConvert(o,REDIS_ENCODING_SKIPLIST); 1159 break; 1160 case REDIS_RDB_TYPE_HASH_ZIPLIST: 1161 o->type = REDIS_HASH; 1162 o->encoding = REDIS_ENCODING_ZIPLIST; 1163 if (hashTypeLength(o) > server.hash_max_ziplist_entries) 1164 hashTypeConvert(o, REDIS_ENCODING_HT); 1165 break; 1166 default: 1167 redisPanic("Unknown encoding"); 1168 break; 1169 } 1170 } else { 1171 redisPanic("Unknown object type"); 1172 } 1173 return o; 1174 } 1175 1176 /* Mark that we are loading in the global state and setup the fields 1177 * needed to provide loading stats. */ 1178 void startLoading(FILE *fp) { 1179 struct stat sb; 1180 1181 /* Load the DB */ 1182 server.loading = 1; 1183 server.loading_start_time = time(NULL); 1184 server.loading_loaded_bytes = 0; 1185 if (fstat(fileno(fp), &sb) == -1) { 1186 server.loading_total_bytes = 0; 1187 } else { 1188 server.loading_total_bytes = sb.st_size; 1189 } 1190 } 1191 1192 /* Refresh the loading progress info */ 1193 void loadingProgress(off_t pos) { 1194 server.loading_loaded_bytes = pos; 1195 if (server.stat_peak_memory < zmalloc_used_memory()) 1196 server.stat_peak_memory = zmalloc_used_memory(); 1197 } 1198 1199 /* Loading finished */ 1200 void stopLoading(void) { 1201 server.loading = 0; 1202 } 1203 1204 /* Track loading progress in order to serve client's from time to time 1205 and if needed calculate rdb checksum */ 1206 void rdbLoadProgressCallback(rio *r, const void *buf, size_t len) { 1207 if (server.rdb_checksum) 1208 rioGenericUpdateChecksum(r, buf, len); 1209 if (server.loading_process_events_interval_bytes && 1210 (r->processed_bytes + len)/server.loading_process_events_interval_bytes > r->processed_bytes/server.loading_process_events_interval_bytes) 1211 { 1212 /* The DB can take some non trivial amount of time to load. Update 1213 * our cached time since it is used to create and update the last 1214 * interaction time with clients and for other important things. */ 1215 updateCachedTime(); 1216 if (server.masterhost && server.repl_state == REDIS_REPL_TRANSFER) 1217 replicationSendNewlineToMaster(); 1218 loadingProgress(r->processed_bytes); 1219 processEventsWhileBlocked(); 1220 } 1221 } 1222 1223 int rdbLoad(char *filename) { 1224 uint32_t dbid; 1225 int type, rdbver; 1226 redisDb *db = server.db+0; 1227 char buf[1024]; 1228 long long expiretime, now = mstime(); 1229 FILE *fp; 1230 rio rdb; 1231 1232 if ((fp = fopen(filename,"r")) == NULL) return REDIS_ERR; 1233 1234 rioInitWithFile(&rdb,fp); 1235 rdb.update_cksum = rdbLoadProgressCallback; 1236 rdb.max_processing_chunk = server.loading_process_events_interval_bytes; 1237 if (rioRead(&rdb,buf,9) == 0) goto eoferr; 1238 buf[9] = '\0'; 1239 if (memcmp(buf,"REDIS",5) != 0) { 1240 fclose(fp); 1241 redisLog(REDIS_WARNING,"Wrong signature trying to load DB from file"); 1242 errno = EINVAL; 1243 return REDIS_ERR; 1244 } 1245 rdbver = atoi(buf+5); 1246 if (rdbver < 1 || rdbver > REDIS_RDB_VERSION) { 1247 fclose(fp); 1248 redisLog(REDIS_WARNING,"Can't handle RDB format version %d",rdbver); 1249 errno = EINVAL; 1250 return REDIS_ERR; 1251 } 1252 1253 startLoading(fp); 1254 while(1) { 1255 robj *key, *val; 1256 expiretime = -1; 1257 1258 /* Read type. */ 1259 if ((type = rdbLoadType(&rdb)) == -1) goto eoferr; 1260 1261 /* Handle special types. */ 1262 if (type == REDIS_RDB_OPCODE_EXPIRETIME) { 1263 /* EXPIRETIME: load an expire associated with the next key 1264 * to load. Note that after loading an expire we need to 1265 * load the actual type, and continue. */ 1266 if ((expiretime = rdbLoadTime(&rdb)) == -1) goto eoferr; 1267 /* We read the time so we need to read the object type again. */ 1268 if ((type = rdbLoadType(&rdb)) == -1) goto eoferr; 1269 /* the EXPIRETIME opcode specifies time in seconds, so convert 1270 * into milliseconds. */ 1271 expiretime *= 1000; 1272 } else if (type == REDIS_RDB_OPCODE_EXPIRETIME_MS) { 1273 /* EXPIRETIME_MS: milliseconds precision expire times introduced 1274 * with RDB v3. Like EXPIRETIME but no with more precision. */ 1275 if ((expiretime = rdbLoadMillisecondTime(&rdb)) == -1) goto eoferr; 1276 /* We read the time so we need to read the object type again. */ 1277 if ((type = rdbLoadType(&rdb)) == -1) goto eoferr; 1278 } else if (type == REDIS_RDB_OPCODE_EOF) { 1279 /* EOF: End of file, exit the main loop. */ 1280 break; 1281 } else if (type == REDIS_RDB_OPCODE_SELECTDB) { 1282 /* SELECTDB: Select the specified database. */ 1283 if ((dbid = rdbLoadLen(&rdb,NULL)) == REDIS_RDB_LENERR) 1284 goto eoferr; 1285 if (dbid >= (unsigned)server.dbnum) { 1286 redisLog(REDIS_WARNING, 1287 "FATAL: Data file was created with a Redis " 1288 "server configured to handle more than %d " 1289 "databases. Exiting\n", server.dbnum); 1290 exit(1); 1291 } 1292 db = server.db+dbid; 1293 continue; /* Read type again. */ 1294 } else if (type == REDIS_RDB_OPCODE_RESIZEDB) { 1295 /* RESIZEDB: Hint about the size of the keys in the currently 1296 * selected data base, in order to avoid useless rehashing. */ 1297 uint32_t db_size, expires_size; 1298 if ((db_size = rdbLoadLen(&rdb,NULL)) == REDIS_RDB_LENERR) 1299 goto eoferr; 1300 if ((expires_size = rdbLoadLen(&rdb,NULL)) == REDIS_RDB_LENERR) 1301 goto eoferr; 1302 dictExpand(db->dict,db_size); 1303 dictExpand(db->expires,expires_size); 1304 continue; /* Read type again. */ 1305 } else if (type == REDIS_RDB_OPCODE_AUX) { 1306 /* AUX: generic string-string fields. Use to add state to RDB 1307 * which is backward compatible. Implementations of RDB loading 1308 * are requierd to skip AUX fields they don't understand. 1309 * 1310 * An AUX field is composed of two strings: key and value. */ 1311 robj *auxkey, *auxval; 1312 if ((auxkey = rdbLoadStringObject(&rdb)) == NULL) goto eoferr; 1313 if ((auxval = rdbLoadStringObject(&rdb)) == NULL) goto eoferr; 1314 1315 if (((char*)auxkey->ptr)[0] == '%') { 1316 /* All the fields with a name staring with '%' are considered 1317 * information fields and are logged at startup with a log 1318 * level of NOTICE. */ 1319 redisLog(REDIS_NOTICE,"RDB '%s': %s", 1320 (char*)auxkey->ptr, 1321 (char*)auxval->ptr); 1322 } else { 1323 /* We ignore fields we don't understand, as by AUX field 1324 * contract. */ 1325 redisLog(REDIS_DEBUG,"Unrecognized RDB AUX field: '%s'", 1326 (char*)auxkey->ptr); 1327 } 1328 1329 decrRefCount(auxkey); 1330 decrRefCount(auxval); 1331 continue; /* Read type again. */ 1332 } 1333 1334 /* Read key */ 1335 if ((key = rdbLoadStringObject(&rdb)) == NULL) goto eoferr; 1336 /* Read value */ 1337 if ((val = rdbLoadObject(type,&rdb)) == NULL) goto eoferr; 1338 /* Check if the key already expired. This function is used when loading 1339 * an RDB file from disk, either at startup, or when an RDB was 1340 * received from the master. In the latter case, the master is 1341 * responsible for key expiry. If we would expire keys here, the 1342 * snapshot taken by the master may not be reflected on the slave. */ 1343 if (server.masterhost == NULL && expiretime != -1 && expiretime < now) { 1344 decrRefCount(key); 1345 decrRefCount(val); 1346 continue; 1347 } 1348 /* Add the new object in the hash table */ 1349 dbAdd(db,key,val); 1350 1351 /* Set the expire time if needed */ 1352 if (expiretime != -1) setExpire(db,key,expiretime); 1353 1354 decrRefCount(key); 1355 } 1356 /* Verify the checksum if RDB version is >= 5 */ 1357 if (rdbver >= 5 && server.rdb_checksum) { 1358 uint64_t cksum, expected = rdb.cksum; 1359 1360 if (rioRead(&rdb,&cksum,8) == 0) goto eoferr; 1361 memrev64ifbe(&cksum); 1362 if (cksum == 0) { 1363 redisLog(REDIS_WARNING,"RDB file was saved with checksum disabled: no check performed."); 1364 } else if (cksum != expected) { 1365 redisLog(REDIS_WARNING,"Wrong RDB checksum. Aborting now."); 1366 exit(1); 1367 } 1368 } 1369 1370 fclose(fp); 1371 stopLoading(); 1372 return REDIS_OK; 1373 1374 eoferr: /* unexpected end of file is handled here with a fatal exit */ 1375 redisLog(REDIS_WARNING,"Short read or OOM loading DB. Unrecoverable error, aborting now."); 1376 exit(1); 1377 return REDIS_ERR; /* Just to avoid warning */ 1378 } 1379 1380 /* A background saving child (BGSAVE) terminated its work. Handle this. 1381 * This function covers the case of actual BGSAVEs. */ 1382 void backgroundSaveDoneHandlerDisk(int exitcode, int bysignal) { 1383 if (!bysignal && exitcode == 0) { 1384 redisLog(REDIS_NOTICE, 1385 "Background saving terminated with success"); 1386 server.dirty = server.dirty - server.dirty_before_bgsave; 1387 server.lastsave = time(NULL); 1388 server.lastbgsave_status = REDIS_OK; 1389 } else if (!bysignal && exitcode != 0) { 1390 redisLog(REDIS_WARNING, "Background saving error"); 1391 server.lastbgsave_status = REDIS_ERR; 1392 } else { 1393 mstime_t latency; 1394 1395 redisLog(REDIS_WARNING, 1396 "Background saving terminated by signal %d", bysignal); 1397 latencyStartMonitor(latency); 1398 rdbRemoveTempFile(server.rdb_child_pid); 1399 latencyEndMonitor(latency); 1400 latencyAddSampleIfNeeded("rdb-unlink-temp-file",latency); 1401 /* SIGUSR1 is whitelisted, so we have a way to kill a child without 1402 * tirggering an error conditon. */ 1403 if (bysignal != SIGUSR1) 1404 server.lastbgsave_status = REDIS_ERR; 1405 } 1406 server.rdb_child_pid = -1; 1407 server.rdb_child_type = REDIS_RDB_CHILD_TYPE_NONE; 1408 server.rdb_save_time_last = time(NULL)-server.rdb_save_time_start; 1409 server.rdb_save_time_start = -1; 1410 /* Possibly there are slaves waiting for a BGSAVE in order to be served 1411 * (the first stage of SYNC is a bulk transfer of dump.rdb) */ 1412 updateSlavesWaitingBgsave((!bysignal && exitcode == 0) ? REDIS_OK : REDIS_ERR, REDIS_RDB_CHILD_TYPE_DISK); 1413 } 1414 1415 /* A background saving child (BGSAVE) terminated its work. Handle this. 1416 * This function covers the case of RDB -> Salves socket transfers for 1417 * diskless replication. */ 1418 void backgroundSaveDoneHandlerSocket(int exitcode, int bysignal) { 1419 uint64_t *ok_slaves; 1420 1421 if (!bysignal && exitcode == 0) { 1422 redisLog(REDIS_NOTICE, 1423 "Background RDB transfer terminated with success"); 1424 } else if (!bysignal && exitcode != 0) { 1425 redisLog(REDIS_WARNING, "Background transfer error"); 1426 } else { 1427 redisLog(REDIS_WARNING, 1428 "Background transfer terminated by signal %d", bysignal); 1429 } 1430 server.rdb_child_pid = -1; 1431 server.rdb_child_type = REDIS_RDB_CHILD_TYPE_NONE; 1432 server.rdb_save_time_start = -1; 1433 1434 /* If the child returns an OK exit code, read the set of slave client 1435 * IDs and the associated status code. We'll terminate all the slaves 1436 * in error state. 1437 * 1438 * If the process returned an error, consider the list of slaves that 1439 * can continue to be emtpy, so that it's just a special case of the 1440 * normal code path. */ 1441 ok_slaves = zmalloc(sizeof(uint64_t)); /* Make space for the count. */ 1442 ok_slaves[0] = 0; 1443 if (!bysignal && exitcode == 0) { 1444 int readlen = sizeof(uint64_t); 1445 1446 if (read(server.rdb_pipe_read_result_from_child, ok_slaves, readlen) == 1447 readlen) 1448 { 1449 readlen = ok_slaves[0]*sizeof(uint64_t)*2; 1450 1451 /* Make space for enough elements as specified by the first 1452 * uint64_t element in the array. */ 1453 ok_slaves = zrealloc(ok_slaves,sizeof(uint64_t)+readlen); 1454 if (readlen && 1455 read(server.rdb_pipe_read_result_from_child, ok_slaves+1, 1456 readlen) != readlen) 1457 { 1458 ok_slaves[0] = 0; 1459 } 1460 } 1461 } 1462 1463 close(server.rdb_pipe_read_result_from_child); 1464 close(server.rdb_pipe_write_result_to_parent); 1465 1466 /* We can continue the replication process with all the slaves that 1467 * correctly received the full payload. Others are terminated. */ 1468 listNode *ln; 1469 listIter li; 1470 1471 listRewind(server.slaves,&li); 1472 while((ln = listNext(&li))) { 1473 redisClient *slave = ln->value; 1474 1475 if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) { 1476 uint64_t j; 1477 int errorcode = 0; 1478 1479 /* Search for the slave ID in the reply. In order for a slave to 1480 * continue the replication process, we need to find it in the list, 1481 * and it must have an error code set to 0 (which means success). */ 1482 for (j = 0; j < ok_slaves[0]; j++) { 1483 if (slave->id == ok_slaves[2*j+1]) { 1484 errorcode = ok_slaves[2*j+2]; 1485 break; /* Found in slaves list. */ 1486 } 1487 } 1488 if (j == ok_slaves[0] || errorcode != 0) { 1489 redisLog(REDIS_WARNING, 1490 "Closing slave %s: child->slave RDB transfer failed: %s", 1491 replicationGetSlaveName(slave), 1492 (errorcode == 0) ? "RDB transfer child aborted" 1493 : strerror(errorcode)); 1494 freeClient(slave); 1495 } else { 1496 redisLog(REDIS_WARNING, 1497 "Slave %s correctly received the streamed RDB file.", 1498 replicationGetSlaveName(slave)); 1499 /* Restore the socket as non-blocking. */ 1500 anetNonBlock(NULL,slave->fd); 1501 anetSendTimeout(NULL,slave->fd,0); 1502 } 1503 } 1504 } 1505 zfree(ok_slaves); 1506 1507 updateSlavesWaitingBgsave((!bysignal && exitcode == 0) ? REDIS_OK : REDIS_ERR, REDIS_RDB_CHILD_TYPE_SOCKET); 1508 } 1509 1510 /* When a background RDB saving/transfer terminates, call the right handler. */ 1511 void backgroundSaveDoneHandler(int exitcode, int bysignal) { 1512 switch(server.rdb_child_type) { 1513 case REDIS_RDB_CHILD_TYPE_DISK: 1514 backgroundSaveDoneHandlerDisk(exitcode,bysignal); 1515 break; 1516 case REDIS_RDB_CHILD_TYPE_SOCKET: 1517 backgroundSaveDoneHandlerSocket(exitcode,bysignal); 1518 break; 1519 default: 1520 redisPanic("Unknown RDB child type."); 1521 break; 1522 } 1523 } 1524 1525 /* Spawn an RDB child that writes the RDB to the sockets of the slaves 1526 * that are currently in REDIS_REPL_WAIT_BGSAVE_START state. */ 1527 int rdbSaveToSlavesSockets(void) { 1528 int *fds; 1529 uint64_t *clientids; 1530 int numfds; 1531 listNode *ln; 1532 listIter li; 1533 pid_t childpid; 1534 long long start; 1535 int pipefds[2]; 1536 1537 if (server.rdb_child_pid != -1) return REDIS_ERR; 1538 1539 /* Before to fork, create a pipe that will be used in order to 1540 * send back to the parent the IDs of the slaves that successfully 1541 * received all the writes. */ 1542 if (pipe(pipefds) == -1) return REDIS_ERR; 1543 server.rdb_pipe_read_result_from_child = pipefds[0]; 1544 server.rdb_pipe_write_result_to_parent = pipefds[1]; 1545 1546 /* Collect the file descriptors of the slaves we want to transfer 1547 * the RDB to, which are i WAIT_BGSAVE_START state. */ 1548 fds = zmalloc(sizeof(int)*listLength(server.slaves)); 1549 /* We also allocate an array of corresponding client IDs. This will 1550 * be useful for the child process in order to build the report 1551 * (sent via unix pipe) that will be sent to the parent. */ 1552 clientids = zmalloc(sizeof(uint64_t)*listLength(server.slaves)); 1553 numfds = 0; 1554 1555 listRewind(server.slaves,&li); 1556 while((ln = listNext(&li))) { 1557 redisClient *slave = ln->value; 1558 1559 if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) { 1560 clientids[numfds] = slave->id; 1561 fds[numfds++] = slave->fd; 1562 slave->replstate = REDIS_REPL_WAIT_BGSAVE_END; 1563 /* Put the socket in non-blocking mode to simplify RDB transfer. 1564 * We'll restore it when the children returns (since duped socket 1565 * will share the O_NONBLOCK attribute with the parent). */ 1566 anetBlock(NULL,slave->fd); 1567 anetSendTimeout(NULL,slave->fd,server.repl_timeout*1000); 1568 } 1569 } 1570 1571 /* Create the child process. */ 1572 start = ustime(); 1573 if ((childpid = fork()) == 0) { 1574 /* Child */ 1575 int retval; 1576 rio slave_sockets; 1577 1578 rioInitWithFdset(&slave_sockets,fds,numfds); 1579 zfree(fds); 1580 1581 closeListeningSockets(0); 1582 redisSetProcTitle("redis-rdb-to-slaves"); 1583 1584 retval = rdbSaveRioWithEOFMark(&slave_sockets,NULL); 1585 if (retval == REDIS_OK && rioFlush(&slave_sockets) == 0) 1586 retval = REDIS_ERR; 1587 1588 if (retval == REDIS_OK) { 1589 size_t private_dirty = zmalloc_get_private_dirty(); 1590 1591 if (private_dirty) { 1592 redisLog(REDIS_NOTICE, 1593 "RDB: %zu MB of memory used by copy-on-write", 1594 private_dirty/(1024*1024)); 1595 } 1596 1597 /* If we are returning OK, at least one slave was served 1598 * with the RDB file as expected, so we need to send a report 1599 * to the parent via the pipe. The format of the message is: 1600 * 1601 * <len> <slave[0].id> <slave[0].error> ... 1602 * 1603 * len, slave IDs, and slave errors, are all uint64_t integers, 1604 * so basically the reply is composed of 64 bits for the len field 1605 * plus 2 additional 64 bit integers for each entry, for a total 1606 * of 'len' entries. 1607 * 1608 * The 'id' represents the slave's client ID, so that the master 1609 * can match the report with a specific slave, and 'error' is 1610 * set to 0 if the replication process terminated with a success 1611 * or the error code if an error occurred. */ 1612 void *msg = zmalloc(sizeof(uint64_t)*(1+2*numfds)); 1613 uint64_t *len = msg; 1614 uint64_t *ids = len+1; 1615 int j, msglen; 1616 1617 *len = numfds; 1618 for (j = 0; j < numfds; j++) { 1619 *ids++ = clientids[j]; 1620 *ids++ = slave_sockets.io.fdset.state[j]; 1621 } 1622 1623 /* Write the message to the parent. If we have no good slaves or 1624 * we are unable to transfer the message to the parent, we exit 1625 * with an error so that the parent will abort the replication 1626 * process with all the childre that were waiting. */ 1627 msglen = sizeof(uint64_t)*(1+2*numfds); 1628 if (*len == 0 || 1629 write(server.rdb_pipe_write_result_to_parent,msg,msglen) 1630 != msglen) 1631 { 1632 retval = REDIS_ERR; 1633 } 1634 zfree(msg); 1635 } 1636 zfree(clientids); 1637 exitFromChild((retval == REDIS_OK) ? 0 : 1); 1638 } else { 1639 /* Parent */ 1640 zfree(clientids); /* Not used by parent. Free ASAP. */ 1641 server.stat_fork_time = ustime()-start; 1642 server.stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / server.stat_fork_time / (1024*1024*1024); /* GB per second. */ 1643 latencyAddSampleIfNeeded("fork",server.stat_fork_time/1000); 1644 if (childpid == -1) { 1645 redisLog(REDIS_WARNING,"Can't save in background: fork: %s", 1646 strerror(errno)); 1647 zfree(fds); 1648 close(pipefds[0]); 1649 close(pipefds[1]); 1650 return REDIS_ERR; 1651 } 1652 redisLog(REDIS_NOTICE,"Background RDB transfer started by pid %d",childpid); 1653 server.rdb_save_time_start = time(NULL); 1654 server.rdb_child_pid = childpid; 1655 server.rdb_child_type = REDIS_RDB_CHILD_TYPE_SOCKET; 1656 updateDictResizePolicy(); 1657 zfree(fds); 1658 return REDIS_OK; 1659 } 1660 return REDIS_OK; /* unreached */ 1661 } 1662 1663 void saveCommand(redisClient *c) { 1664 if (server.rdb_child_pid != -1) { 1665 addReplyError(c,"Background save already in progress"); 1666 return; 1667 } 1668 if (rdbSave(server.rdb_filename) == REDIS_OK) { 1669 addReply(c,shared.ok); 1670 } else { 1671 addReply(c,shared.err); 1672 } 1673 } 1674 1675 void bgsaveCommand(redisClient *c) { 1676 if (server.rdb_child_pid != -1) { 1677 addReplyError(c,"Background save already in progress"); 1678 } else if (server.aof_child_pid != -1) { 1679 addReplyError(c,"Can't BGSAVE while AOF log rewriting is in progress"); 1680 } else if (rdbSaveBackground(server.rdb_filename) == REDIS_OK) { 1681 addReplyStatus(c,"Background saving started"); 1682 } else { 1683 addReply(c,shared.err); 1684 } 1685 } 1686