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