src/rdb.c

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