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