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