| 1 | #include "redis.h" |
| 2 | #include "bio.h" |
| 3 | #include "rio.h" |
| 4 | |
| 5 | #include <signal.h> |
| 6 | #include <fcntl.h> |
| 7 | #include <sys/stat.h> |
| 8 | #include <sys/types.h> |
| 9 | #include <sys/time.h> |
| 10 | #include <sys/resource.h> |
| 11 | #include <sys/wait.h> |
| 12 | |
| 13 | void aofUpdateCurrentSize(void); |
| 14 | |
| 15 | void aof_background_fsync(int fd) { |
| 16 | bioCreateBackgroundJob(REDIS_BIO_AOF_FSYNC,(void*)(long)fd,NULL,NULL); |
| 17 | } |
| 18 | |
| 19 | /* Called when the user switches from "appendonly yes" to "appendonly no" |
| 20 | * at runtime using the CONFIG command. */ |
| 21 | void stopAppendOnly(void) { |
| 22 | flushAppendOnlyFile(1); |
| 23 | aof_fsync(server.appendfd); |
| 24 | close(server.appendfd); |
| 25 | |
| 26 | server.appendfd = -1; |
| 27 | server.appendseldb = -1; |
| 28 | server.appendonly = 0; |
| 29 | /* rewrite operation in progress? kill it, wait child exit */ |
| 30 | if (server.bgrewritechildpid != -1) { |
| 31 | int statloc; |
| 32 | |
| 33 | if (kill(server.bgrewritechildpid,SIGKILL) != -1) |
| 34 | wait3(&statloc,0,NULL); |
| 35 | /* reset the buffer accumulating changes while the child saves */ |
| 36 | sdsfree(server.bgrewritebuf); |
| 37 | server.bgrewritebuf = sdsempty(); |
| 38 | server.bgrewritechildpid = -1; |
| 39 | } |
| 40 | } |
| 41 | |
| 42 | /* Called when the user switches from "appendonly no" to "appendonly yes" |
| 43 | * at runtime using the CONFIG command. */ |
| 44 | int startAppendOnly(void) { |
| 45 | server.appendonly = 1; |
| 46 | server.lastfsync = time(NULL); |
| 47 | server.appendfd = open(server.appendfilename,O_WRONLY|O_APPEND|O_CREAT,0644); |
| 48 | if (server.appendfd == -1) { |
| 49 | redisLog(REDIS_WARNING,"Used tried to switch on AOF via CONFIG, but I can't open the AOF file: %s",strerror(errno)); |
| 50 | return REDIS_ERR; |
| 51 | } |
| 52 | if (rewriteAppendOnlyFileBackground() == REDIS_ERR) { |
| 53 | server.appendonly = 0; |
| 54 | close(server.appendfd); |
| 55 | redisLog(REDIS_WARNING,"User tried turning on AOF with CONFIG SET but I can't trigger a background AOF rewrite operation. Check the above logs for more info about the error."); |
| 56 | return REDIS_ERR; |
| 57 | } |
| 58 | return REDIS_OK; |
| 59 | } |
| 60 | |
| 61 | /* Write the append only file buffer on disk. |
| 62 | * |
| 63 | * Since we are required to write the AOF before replying to the client, |
| 64 | * and the only way the client socket can get a write is entering when the |
| 65 | * the event loop, we accumulate all the AOF writes in a memory |
| 66 | * buffer and write it on disk using this function just before entering |
| 67 | * the event loop again. |
| 68 | * |
| 69 | * About the 'force' argument: |
| 70 | * |
| 71 | * When the fsync policy is set to 'everysec' we may delay the flush if there |
| 72 | * is still an fsync() going on in the background thread, since for instance |
| 73 | * on Linux write(2) will be blocked by the background fsync anyway. |
| 74 | * When this happens we remember that there is some aof buffer to be |
| 75 | * flushed ASAP, and will try to do that in the serverCron() function. |
| 76 | * |
| 77 | * However if force is set to 1 we'll write regardless of the background |
| 78 | * fsync. */ |
| 79 | void flushAppendOnlyFile(int force) { |
| 80 | ssize_t nwritten; |
| 81 | int sync_in_progress = 0; |
| 82 | |
| 83 | if (sdslen(server.aofbuf) == 0) return; |
| 84 | |
| 85 | if (server.appendfsync == APPENDFSYNC_EVERYSEC) |
| 86 | sync_in_progress = bioPendingJobsOfType(REDIS_BIO_AOF_FSYNC) != 0; |
| 87 | |
| 88 | if (server.appendfsync == APPENDFSYNC_EVERYSEC && !force) { |
| 89 | /* With this append fsync policy we do background fsyncing. |
| 90 | * If the fsync is still in progress we can try to delay |
| 91 | * the write for a couple of seconds. */ |
| 92 | if (sync_in_progress) { |
| 93 | if (server.aof_flush_postponed_start == 0) { |
| 94 | /* No previous write postponinig, remember that we are |
| 95 | * postponing the flush and return. */ |
| 96 | server.aof_flush_postponed_start = server.unixtime; |
| 97 | return; |
| 98 | } else if (server.unixtime - server.aof_flush_postponed_start < 2) { |
| 99 | /* We were already waiting for fsync to finish, but for less |
| 100 | * than two seconds this is still ok. Postpone again. */ |
| 101 | return; |
| 102 | } |
| 103 | /* Otherwise fall trough, and go write since we can't wait |
| 104 | * over two seconds. */ |
| 105 | redisLog(REDIS_NOTICE,"Asynchronous AOF fsync is taking too long (disk is busy?). Writing the AOF buffer without waiting for fsync to complete, this may slow down Redis."); |
| 106 | } |
| 107 | } |
| 108 | /* If you are following this code path, then we are going to write so |
| 109 | * set reset the postponed flush sentinel to zero. */ |
| 110 | server.aof_flush_postponed_start = 0; |
| 111 | |
| 112 | /* We want to perform a single write. This should be guaranteed atomic |
| 113 | * at least if the filesystem we are writing is a real physical one. |
| 114 | * While this will save us against the server being killed I don't think |
| 115 | * there is much to do about the whole server stopping for power problems |
| 116 | * or alike */ |
| 117 | nwritten = write(server.appendfd,server.aofbuf,sdslen(server.aofbuf)); |
| 118 | if (nwritten != (signed)sdslen(server.aofbuf)) { |
| 119 | /* Ooops, we are in troubles. The best thing to do for now is |
| 120 | * aborting instead of giving the illusion that everything is |
| 121 | * working as expected. */ |
| 122 | if (nwritten == -1) { |
| 123 | redisLog(REDIS_WARNING,"Exiting on error writing to the append-only file: %s",strerror(errno)); |
| 124 | } else { |
| 125 | redisLog(REDIS_WARNING,"Exiting on short write while writing to the append-only file: %s",strerror(errno)); |
| 126 | } |
| 127 | exit(1); |
| 128 | } |
| 129 | server.appendonly_current_size += nwritten; |
| 130 | |
| 131 | /* Re-use AOF buffer when it is small enough. The maximum comes from the |
| 132 | * arena size of 4k minus some overhead (but is otherwise arbitrary). */ |
| 133 | if ((sdslen(server.aofbuf)+sdsavail(server.aofbuf)) < 4000) { |
| 134 | sdsclear(server.aofbuf); |
| 135 | } else { |
| 136 | sdsfree(server.aofbuf); |
| 137 | server.aofbuf = sdsempty(); |
| 138 | } |
| 139 | |
| 140 | /* Don't fsync if no-appendfsync-on-rewrite is set to yes and there are |
| 141 | * children doing I/O in the background. */ |
| 142 | if (server.no_appendfsync_on_rewrite && |
| 143 | (server.bgrewritechildpid != -1 || server.bgsavechildpid != -1)) |
| 144 | return; |
| 145 | |
| 146 | /* Perform the fsync if needed. */ |
| 147 | if (server.appendfsync == APPENDFSYNC_ALWAYS) { |
| 148 | /* aof_fsync is defined as fdatasync() for Linux in order to avoid |
| 149 | * flushing metadata. */ |
| 150 | aof_fsync(server.appendfd); /* Let's try to get this data on the disk */ |
| 151 | server.lastfsync = server.unixtime; |
| 152 | } else if ((server.appendfsync == APPENDFSYNC_EVERYSEC && |
| 153 | server.unixtime > server.lastfsync)) { |
| 154 | if (!sync_in_progress) aof_background_fsync(server.appendfd); |
| 155 | server.lastfsync = server.unixtime; |
| 156 | } |
| 157 | } |
| 158 | |
| 159 | sds catAppendOnlyGenericCommand(sds dst, int argc, robj **argv) { |
| 160 | char buf[32]; |
| 161 | int len, j; |
| 162 | robj *o; |
| 163 | |
| 164 | buf[0] = '*'; |
| 165 | len = 1+ll2string(buf+1,sizeof(buf)-1,argc); |
| 166 | buf[len++] = '\r'; |
| 167 | buf[len++] = '\n'; |
| 168 | dst = sdscatlen(dst,buf,len); |
| 169 | |
| 170 | for (j = 0; j < argc; j++) { |
| 171 | o = getDecodedObject(argv[j]); |
| 172 | buf[0] = '$'; |
| 173 | len = 1+ll2string(buf+1,sizeof(buf)-1,sdslen(o->ptr)); |
| 174 | buf[len++] = '\r'; |
| 175 | buf[len++] = '\n'; |
| 176 | dst = sdscatlen(dst,buf,len); |
| 177 | dst = sdscatlen(dst,o->ptr,sdslen(o->ptr)); |
| 178 | dst = sdscatlen(dst,"\r\n",2); |
| 179 | decrRefCount(o); |
| 180 | } |
| 181 | return dst; |
| 182 | } |
| 183 | |
| 184 | /* Create the sds representation of an PEXPIREAT command, using |
| 185 | * 'seconds' as time to live and 'cmd' to understand what command |
| 186 | * we are translating into a PEXPIREAT. |
| 187 | * |
| 188 | * This command is used in order to translate EXPIRE and PEXPIRE commands |
| 189 | * into PEXPIREAT command so that we retain precision in the append only |
| 190 | * file, and the time is always absolute and not relative. */ |
| 191 | sds catAppendOnlyExpireAtCommand(sds buf, struct redisCommand *cmd, robj *key, robj *seconds) { |
| 192 | long long when; |
| 193 | robj *argv[3]; |
| 194 | |
| 195 | /* Make sure we can use strtol */ |
| 196 | seconds = getDecodedObject(seconds); |
| 197 | when = strtoll(seconds->ptr,NULL,10); |
| 198 | /* Convert argument into milliseconds for EXPIRE, SETEX, EXPIREAT */ |
| 199 | if (cmd->proc == expireCommand || cmd->proc == setexCommand || |
| 200 | cmd->proc == expireatCommand) |
| 201 | { |
| 202 | when *= 1000; |
| 203 | } |
| 204 | /* Convert into absolute time for EXPIRE, PEXPIRE, SETEX, PSETEX */ |
| 205 | if (cmd->proc == expireCommand || cmd->proc == pexpireCommand || |
| 206 | cmd->proc == setexCommand || cmd->proc == psetexCommand) |
| 207 | { |
| 208 | when += mstime(); |
| 209 | } |
| 210 | decrRefCount(seconds); |
| 211 | |
| 212 | argv[0] = createStringObject("PEXPIREAT",9); |
| 213 | argv[1] = key; |
| 214 | argv[2] = createStringObjectFromLongLong(when); |
| 215 | buf = catAppendOnlyGenericCommand(buf, 3, argv); |
| 216 | decrRefCount(argv[0]); |
| 217 | decrRefCount(argv[2]); |
| 218 | return buf; |
| 219 | } |
| 220 | |
| 221 | void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int argc) { |
| 222 | sds buf = sdsempty(); |
| 223 | robj *tmpargv[3]; |
| 224 | |
| 225 | /* The DB this command was targetting is not the same as the last command |
| 226 | * we appendend. To issue a SELECT command is needed. */ |
| 227 | if (dictid != server.appendseldb) { |
| 228 | char seldb[64]; |
| 229 | |
| 230 | snprintf(seldb,sizeof(seldb),"%d",dictid); |
| 231 | buf = sdscatprintf(buf,"*2\r\n$6\r\nSELECT\r\n$%lu\r\n%s\r\n", |
| 232 | (unsigned long)strlen(seldb),seldb); |
| 233 | server.appendseldb = dictid; |
| 234 | } |
| 235 | |
| 236 | if (cmd->proc == expireCommand || cmd->proc == pexpireCommand || |
| 237 | cmd->proc == expireatCommand) { |
| 238 | /* Translate EXPIRE/PEXPIRE/EXPIREAT into PEXPIREAT */ |
| 239 | buf = catAppendOnlyExpireAtCommand(buf,cmd,argv[1],argv[2]); |
| 240 | } else if (cmd->proc == setexCommand || cmd->proc == psetexCommand) { |
| 241 | /* Translate SETEX/PSETEX to SET and PEXPIREAT */ |
| 242 | tmpargv[0] = createStringObject("SET",3); |
| 243 | tmpargv[1] = argv[1]; |
| 244 | tmpargv[2] = argv[3]; |
| 245 | buf = catAppendOnlyGenericCommand(buf,3,tmpargv); |
| 246 | decrRefCount(tmpargv[0]); |
| 247 | buf = catAppendOnlyExpireAtCommand(buf,cmd,argv[1],argv[2]); |
| 248 | } else { |
| 249 | /* All the other commands don't need translation or need the |
| 250 | * same translation already operated in the command vector |
| 251 | * for the replication itself. */ |
| 252 | buf = catAppendOnlyGenericCommand(buf,argc,argv); |
| 253 | } |
| 254 | |
| 255 | /* Append to the AOF buffer. This will be flushed on disk just before |
| 256 | * of re-entering the event loop, so before the client will get a |
| 257 | * positive reply about the operation performed. */ |
| 258 | server.aofbuf = sdscatlen(server.aofbuf,buf,sdslen(buf)); |
| 259 | |
| 260 | /* If a background append only file rewriting is in progress we want to |
| 261 | * accumulate the differences between the child DB and the current one |
| 262 | * in a buffer, so that when the child process will do its work we |
| 263 | * can append the differences to the new append only file. */ |
| 264 | if (server.bgrewritechildpid != -1) |
| 265 | server.bgrewritebuf = sdscatlen(server.bgrewritebuf,buf,sdslen(buf)); |
| 266 | |
| 267 | sdsfree(buf); |
| 268 | } |
| 269 | |
| 270 | /* In Redis commands are always executed in the context of a client, so in |
| 271 | * order to load the append only file we need to create a fake client. */ |
| 272 | struct redisClient *createFakeClient(void) { |
| 273 | struct redisClient *c = zmalloc(sizeof(*c)); |
| 274 | |
| 275 | selectDb(c,0); |
| 276 | c->fd = -1; |
| 277 | c->querybuf = sdsempty(); |
| 278 | c->argc = 0; |
| 279 | c->argv = NULL; |
| 280 | c->bufpos = 0; |
| 281 | c->flags = 0; |
| 282 | /* We set the fake client as a slave waiting for the synchronization |
| 283 | * so that Redis will not try to send replies to this client. */ |
| 284 | c->replstate = REDIS_REPL_WAIT_BGSAVE_START; |
| 285 | c->reply = listCreate(); |
| 286 | c->watched_keys = listCreate(); |
| 287 | listSetFreeMethod(c->reply,decrRefCount); |
| 288 | listSetDupMethod(c->reply,dupClientReplyValue); |
| 289 | initClientMultiState(c); |
| 290 | return c; |
| 291 | } |
| 292 | |
| 293 | void freeFakeClient(struct redisClient *c) { |
| 294 | sdsfree(c->querybuf); |
| 295 | listRelease(c->reply); |
| 296 | listRelease(c->watched_keys); |
| 297 | freeClientMultiState(c); |
| 298 | zfree(c); |
| 299 | } |
| 300 | |
| 301 | /* Replay the append log file. On error REDIS_OK is returned. On non fatal |
| 302 | * error (the append only file is zero-length) REDIS_ERR is returned. On |
| 303 | * fatal error an error message is logged and the program exists. */ |
| 304 | int loadAppendOnlyFile(char *filename) { |
| 305 | struct redisClient *fakeClient; |
| 306 | FILE *fp = fopen(filename,"r"); |
| 307 | struct redis_stat sb; |
| 308 | int appendonly = server.appendonly; |
| 309 | long loops = 0; |
| 310 | |
| 311 | if (fp && redis_fstat(fileno(fp),&sb) != -1 && sb.st_size == 0) { |
| 312 | server.appendonly_current_size = 0; |
| 313 | fclose(fp); |
| 314 | return REDIS_ERR; |
| 315 | } |
| 316 | |
| 317 | if (fp == NULL) { |
| 318 | redisLog(REDIS_WARNING,"Fatal error: can't open the append log file for reading: %s",strerror(errno)); |
| 319 | exit(1); |
| 320 | } |
| 321 | |
| 322 | /* Temporarily disable AOF, to prevent EXEC from feeding a MULTI |
| 323 | * to the same file we're about to read. */ |
| 324 | server.appendonly = 0; |
| 325 | |
| 326 | fakeClient = createFakeClient(); |
| 327 | startLoading(fp); |
| 328 | |
| 329 | while(1) { |
| 330 | int argc, j; |
| 331 | unsigned long len; |
| 332 | robj **argv; |
| 333 | char buf[128]; |
| 334 | sds argsds; |
| 335 | struct redisCommand *cmd; |
| 336 | |
| 337 | /* Serve the clients from time to time */ |
| 338 | if (!(loops++ % 1000)) { |
| 339 | loadingProgress(ftello(fp)); |
| 340 | aeProcessEvents(server.el, AE_FILE_EVENTS|AE_DONT_WAIT); |
| 341 | } |
| 342 | |
| 343 | if (fgets(buf,sizeof(buf),fp) == NULL) { |
| 344 | if (feof(fp)) |
| 345 | break; |
| 346 | else |
| 347 | goto readerr; |
| 348 | } |
| 349 | if (buf[0] != '*') goto fmterr; |
| 350 | argc = atoi(buf+1); |
| 351 | if (argc < 1) goto fmterr; |
| 352 | |
| 353 | argv = zmalloc(sizeof(robj*)*argc); |
| 354 | for (j = 0; j < argc; j++) { |
| 355 | if (fgets(buf,sizeof(buf),fp) == NULL) goto readerr; |
| 356 | if (buf[0] != '$') goto fmterr; |
| 357 | len = strtol(buf+1,NULL,10); |
| 358 | argsds = sdsnewlen(NULL,len); |
| 359 | if (len && fread(argsds,len,1,fp) == 0) goto fmterr; |
| 360 | argv[j] = createObject(REDIS_STRING,argsds); |
| 361 | if (fread(buf,2,1,fp) == 0) goto fmterr; /* discard CRLF */ |
| 362 | } |
| 363 | |
| 364 | /* Command lookup */ |
| 365 | cmd = lookupCommand(argv[0]->ptr); |
| 366 | if (!cmd) { |
| 367 | redisLog(REDIS_WARNING,"Unknown command '%s' reading the append only file", argv[0]->ptr); |
| 368 | exit(1); |
| 369 | } |
| 370 | /* Run the command in the context of a fake client */ |
| 371 | fakeClient->argc = argc; |
| 372 | fakeClient->argv = argv; |
| 373 | cmd->proc(fakeClient); |
| 374 | |
| 375 | /* The fake client should not have a reply */ |
| 376 | redisAssert(fakeClient->bufpos == 0 && listLength(fakeClient->reply) == 0); |
| 377 | /* The fake client should never get blocked */ |
| 378 | redisAssert((fakeClient->flags & REDIS_BLOCKED) == 0); |
| 379 | |
| 380 | /* Clean up. Command code may have changed argv/argc so we use the |
| 381 | * argv/argc of the client instead of the local variables. */ |
| 382 | for (j = 0; j < fakeClient->argc; j++) |
| 383 | decrRefCount(fakeClient->argv[j]); |
| 384 | zfree(fakeClient->argv); |
| 385 | } |
| 386 | |
| 387 | /* This point can only be reached when EOF is reached without errors. |
| 388 | * If the client is in the middle of a MULTI/EXEC, log error and quit. */ |
| 389 | if (fakeClient->flags & REDIS_MULTI) goto readerr; |
| 390 | |
| 391 | fclose(fp); |
| 392 | freeFakeClient(fakeClient); |
| 393 | server.appendonly = appendonly; |
| 394 | stopLoading(); |
| 395 | aofUpdateCurrentSize(); |
| 396 | server.auto_aofrewrite_base_size = server.appendonly_current_size; |
| 397 | return REDIS_OK; |
| 398 | |
| 399 | readerr: |
| 400 | if (feof(fp)) { |
| 401 | redisLog(REDIS_WARNING,"Unexpected end of file reading the append only file"); |
| 402 | } else { |
| 403 | redisLog(REDIS_WARNING,"Unrecoverable error reading the append only file: %s", strerror(errno)); |
| 404 | } |
| 405 | exit(1); |
| 406 | fmterr: |
| 407 | redisLog(REDIS_WARNING,"Bad file format reading the append only file: make a backup of your AOF file, then use ./redis-check-aof --fix <filename>"); |
| 408 | exit(1); |
| 409 | } |
| 410 | |
| 411 | /* Delegate writing an object to writing a bulk string or bulk long long. |
| 412 | * This is not placed in rio.c since that adds the redis.h dependency. */ |
| 413 | int rioWriteBulkObject(rio *r, robj *obj) { |
| 414 | /* Avoid using getDecodedObject to help copy-on-write (we are often |
| 415 | * in a child process when this function is called). */ |
| 416 | if (obj->encoding == REDIS_ENCODING_INT) { |
| 417 | return rioWriteBulkLongLong(r,(long)obj->ptr); |
| 418 | } else if (obj->encoding == REDIS_ENCODING_RAW) { |
| 419 | return rioWriteBulkString(r,obj->ptr,sdslen(obj->ptr)); |
| 420 | } else { |
| 421 | redisPanic("Unknown string encoding"); |
| 422 | } |
| 423 | } |
| 424 | |
| 425 | /* Write a sequence of commands able to fully rebuild the dataset into |
| 426 | * "filename". Used both by REWRITEAOF and BGREWRITEAOF. */ |
| 427 | int rewriteAppendOnlyFile(char *filename) { |
| 428 | dictIterator *di = NULL; |
| 429 | dictEntry *de; |
| 430 | rio aof; |
| 431 | FILE *fp; |
| 432 | char tmpfile[256]; |
| 433 | int j; |
| 434 | long long now = mstime(); |
| 435 | |
| 436 | /* Note that we have to use a different temp name here compared to the |
| 437 | * one used by rewriteAppendOnlyFileBackground() function. */ |
| 438 | snprintf(tmpfile,256,"temp-rewriteaof-%d.aof", (int) getpid()); |
| 439 | fp = fopen(tmpfile,"w"); |
| 440 | if (!fp) { |
| 441 | redisLog(REDIS_WARNING, "Failed rewriting the append only file: %s", strerror(errno)); |
| 442 | return REDIS_ERR; |
| 443 | } |
| 444 | |
| 445 | rioInitWithFile(&aof,fp); |
| 446 | for (j = 0; j < server.dbnum; j++) { |
| 447 | char selectcmd[] = "*2\r\n$6\r\nSELECT\r\n"; |
| 448 | redisDb *db = server.db+j; |
| 449 | dict *d = db->dict; |
| 450 | if (dictSize(d) == 0) continue; |
| 451 | di = dictGetSafeIterator(d); |
| 452 | if (!di) { |
| 453 | fclose(fp); |
| 454 | return REDIS_ERR; |
| 455 | } |
| 456 | |
| 457 | /* SELECT the new DB */ |
| 458 | if (rioWrite(&aof,selectcmd,sizeof(selectcmd)-1) == 0) goto werr; |
| 459 | if (rioWriteBulkLongLong(&aof,j) == 0) goto werr; |
| 460 | |
| 461 | /* Iterate this DB writing every entry */ |
| 462 | while((de = dictNext(di)) != NULL) { |
| 463 | sds keystr; |
| 464 | robj key, *o; |
| 465 | long long expiretime; |
| 466 | |
| 467 | keystr = dictGetKey(de); |
| 468 | o = dictGetVal(de); |
| 469 | initStaticStringObject(key,keystr); |
| 470 | |
| 471 | expiretime = getExpire(db,&key); |
| 472 | |
| 473 | /* Save the key and associated value */ |
| 474 | if (o->type == REDIS_STRING) { |
| 475 | /* Emit a SET command */ |
| 476 | char cmd[]="*3\r\n$3\r\nSET\r\n"; |
| 477 | if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr; |
| 478 | /* Key and value */ |
| 479 | if (rioWriteBulkObject(&aof,&key) == 0) goto werr; |
| 480 | if (rioWriteBulkObject(&aof,o) == 0) goto werr; |
| 481 | } else if (o->type == REDIS_LIST) { |
| 482 | /* Emit the RPUSHes needed to rebuild the list */ |
| 483 | char cmd[]="*3\r\n$5\r\nRPUSH\r\n"; |
| 484 | if (o->encoding == REDIS_ENCODING_ZIPLIST) { |
| 485 | unsigned char *zl = o->ptr; |
| 486 | unsigned char *p = ziplistIndex(zl,0); |
| 487 | unsigned char *vstr; |
| 488 | unsigned int vlen; |
| 489 | long long vlong; |
| 490 | |
| 491 | while(ziplistGet(p,&vstr,&vlen,&vlong)) { |
| 492 | if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr; |
| 493 | if (rioWriteBulkObject(&aof,&key) == 0) goto werr; |
| 494 | if (vstr) { |
| 495 | if (rioWriteBulkString(&aof,(char*)vstr,vlen) == 0) |
| 496 | goto werr; |
| 497 | } else { |
| 498 | if (rioWriteBulkLongLong(&aof,vlong) == 0) |
| 499 | goto werr; |
| 500 | } |
| 501 | p = ziplistNext(zl,p); |
| 502 | } |
| 503 | } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) { |
| 504 | list *list = o->ptr; |
| 505 | listNode *ln; |
| 506 | listIter li; |
| 507 | |
| 508 | listRewind(list,&li); |
| 509 | while((ln = listNext(&li))) { |
| 510 | robj *eleobj = listNodeValue(ln); |
| 511 | |
| 512 | if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr; |
| 513 | if (rioWriteBulkObject(&aof,&key) == 0) goto werr; |
| 514 | if (rioWriteBulkObject(&aof,eleobj) == 0) goto werr; |
| 515 | } |
| 516 | } else { |
| 517 | redisPanic("Unknown list encoding"); |
| 518 | } |
| 519 | } else if (o->type == REDIS_SET) { |
| 520 | char cmd[]="*3\r\n$4\r\nSADD\r\n"; |
| 521 | |
| 522 | /* Emit the SADDs needed to rebuild the set */ |
| 523 | if (o->encoding == REDIS_ENCODING_INTSET) { |
| 524 | int ii = 0; |
| 525 | int64_t llval; |
| 526 | while(intsetGet(o->ptr,ii++,&llval)) { |
| 527 | if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr; |
| 528 | if (rioWriteBulkObject(&aof,&key) == 0) goto werr; |
| 529 | if (rioWriteBulkLongLong(&aof,llval) == 0) goto werr; |
| 530 | } |
| 531 | } else if (o->encoding == REDIS_ENCODING_HT) { |
| 532 | dictIterator *di = dictGetIterator(o->ptr); |
| 533 | dictEntry *de; |
| 534 | while((de = dictNext(di)) != NULL) { |
| 535 | robj *eleobj = dictGetKey(de); |
| 536 | if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr; |
| 537 | if (rioWriteBulkObject(&aof,&key) == 0) goto werr; |
| 538 | if (rioWriteBulkObject(&aof,eleobj) == 0) goto werr; |
| 539 | } |
| 540 | dictReleaseIterator(di); |
| 541 | } else { |
| 542 | redisPanic("Unknown set encoding"); |
| 543 | } |
| 544 | } else if (o->type == REDIS_ZSET) { |
| 545 | /* Emit the ZADDs needed to rebuild the sorted set */ |
| 546 | char cmd[]="*4\r\n$4\r\nZADD\r\n"; |
| 547 | |
| 548 | if (o->encoding == REDIS_ENCODING_ZIPLIST) { |
| 549 | unsigned char *zl = o->ptr; |
| 550 | unsigned char *eptr, *sptr; |
| 551 | unsigned char *vstr; |
| 552 | unsigned int vlen; |
| 553 | long long vll; |
| 554 | double score; |
| 555 | |
| 556 | eptr = ziplistIndex(zl,0); |
| 557 | redisAssert(eptr != NULL); |
| 558 | sptr = ziplistNext(zl,eptr); |
| 559 | redisAssert(sptr != NULL); |
| 560 | |
| 561 | while (eptr != NULL) { |
| 562 | redisAssert(ziplistGet(eptr,&vstr,&vlen,&vll)); |
| 563 | score = zzlGetScore(sptr); |
| 564 | |
| 565 | if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr; |
| 566 | if (rioWriteBulkObject(&aof,&key) == 0) goto werr; |
| 567 | if (rioWriteBulkDouble(&aof,score) == 0) goto werr; |
| 568 | if (vstr != NULL) { |
| 569 | if (rioWriteBulkString(&aof,(char*)vstr,vlen) == 0) |
| 570 | goto werr; |
| 571 | } else { |
| 572 | if (rioWriteBulkLongLong(&aof,vll) == 0) |
| 573 | goto werr; |
| 574 | } |
| 575 | zzlNext(zl,&eptr,&sptr); |
| 576 | } |
| 577 | } else if (o->encoding == REDIS_ENCODING_SKIPLIST) { |
| 578 | zset *zs = o->ptr; |
| 579 | dictIterator *di = dictGetIterator(zs->dict); |
| 580 | dictEntry *de; |
| 581 | |
| 582 | while((de = dictNext(di)) != NULL) { |
| 583 | robj *eleobj = dictGetKey(de); |
| 584 | double *score = dictGetVal(de); |
| 585 | |
| 586 | if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr; |
| 587 | if (rioWriteBulkObject(&aof,&key) == 0) goto werr; |
| 588 | if (rioWriteBulkDouble(&aof,*score) == 0) goto werr; |
| 589 | if (rioWriteBulkObject(&aof,eleobj) == 0) goto werr; |
| 590 | } |
| 591 | dictReleaseIterator(di); |
| 592 | } else { |
| 593 | redisPanic("Unknown sorted set encoding"); |
| 594 | } |
| 595 | } else if (o->type == REDIS_HASH) { |
| 596 | char cmd[]="*4\r\n$4\r\nHSET\r\n"; |
| 597 | |
| 598 | /* Emit the HSETs needed to rebuild the hash */ |
| 599 | if (o->encoding == REDIS_ENCODING_ZIPMAP) { |
| 600 | unsigned char *p = zipmapRewind(o->ptr); |
| 601 | unsigned char *field, *val; |
| 602 | unsigned int flen, vlen; |
| 603 | |
| 604 | while((p = zipmapNext(p,&field,&flen,&val,&vlen)) != NULL) { |
| 605 | if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr; |
| 606 | if (rioWriteBulkObject(&aof,&key) == 0) goto werr; |
| 607 | if (rioWriteBulkString(&aof,(char*)field,flen) == 0) |
| 608 | goto werr; |
| 609 | if (rioWriteBulkString(&aof,(char*)val,vlen) == 0) |
| 610 | goto werr; |
| 611 | } |
| 612 | } else { |
| 613 | dictIterator *di = dictGetIterator(o->ptr); |
| 614 | dictEntry *de; |
| 615 | |
| 616 | while((de = dictNext(di)) != NULL) { |
| 617 | robj *field = dictGetKey(de); |
| 618 | robj *val = dictGetVal(de); |
| 619 | |
| 620 | if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr; |
| 621 | if (rioWriteBulkObject(&aof,&key) == 0) goto werr; |
| 622 | if (rioWriteBulkObject(&aof,field) == 0) goto werr; |
| 623 | if (rioWriteBulkObject(&aof,val) == 0) goto werr; |
| 624 | } |
| 625 | dictReleaseIterator(di); |
| 626 | } |
| 627 | } else { |
| 628 | redisPanic("Unknown object type"); |
| 629 | } |
| 630 | /* Save the expire time */ |
| 631 | if (expiretime != -1) { |
| 632 | char cmd[]="*3\r\n$9\r\nPEXPIREAT\r\n"; |
| 633 | /* If this key is already expired skip it */ |
| 634 | if (expiretime < now) continue; |
| 635 | if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr; |
| 636 | if (rioWriteBulkObject(&aof,&key) == 0) goto werr; |
| 637 | if (rioWriteBulkLongLong(&aof,expiretime) == 0) goto werr; |
| 638 | } |
| 639 | } |
| 640 | dictReleaseIterator(di); |
| 641 | } |
| 642 | |
| 643 | /* Make sure data will not remain on the OS's output buffers */ |
| 644 | fflush(fp); |
| 645 | aof_fsync(fileno(fp)); |
| 646 | fclose(fp); |
| 647 | |
| 648 | /* Use RENAME to make sure the DB file is changed atomically only |
| 649 | * if the generate DB file is ok. */ |
| 650 | if (rename(tmpfile,filename) == -1) { |
| 651 | redisLog(REDIS_WARNING,"Error moving temp append only file on the final destination: %s", strerror(errno)); |
| 652 | unlink(tmpfile); |
| 653 | return REDIS_ERR; |
| 654 | } |
| 655 | redisLog(REDIS_NOTICE,"SYNC append only file rewrite performed"); |
| 656 | return REDIS_OK; |
| 657 | |
| 658 | werr: |
| 659 | fclose(fp); |
| 660 | unlink(tmpfile); |
| 661 | redisLog(REDIS_WARNING,"Write error writing append only file on disk: %s", strerror(errno)); |
| 662 | if (di) dictReleaseIterator(di); |
| 663 | return REDIS_ERR; |
| 664 | } |
| 665 | |
| 666 | /* This is how rewriting of the append only file in background works: |
| 667 | * |
| 668 | * 1) The user calls BGREWRITEAOF |
| 669 | * 2) Redis calls this function, that forks(): |
| 670 | * 2a) the child rewrite the append only file in a temp file. |
| 671 | * 2b) the parent accumulates differences in server.bgrewritebuf. |
| 672 | * 3) When the child finished '2a' exists. |
| 673 | * 4) The parent will trap the exit code, if it's OK, will append the |
| 674 | * data accumulated into server.bgrewritebuf into the temp file, and |
| 675 | * finally will rename(2) the temp file in the actual file name. |
| 676 | * The the new file is reopened as the new append only file. Profit! |
| 677 | */ |
| 678 | int rewriteAppendOnlyFileBackground(void) { |
| 679 | pid_t childpid; |
| 680 | long long start; |
| 681 | |
| 682 | if (server.bgrewritechildpid != -1) return REDIS_ERR; |
| 683 | start = ustime(); |
| 684 | if ((childpid = fork()) == 0) { |
| 685 | char tmpfile[256]; |
| 686 | |
| 687 | /* Child */ |
| 688 | if (server.ipfd > 0) close(server.ipfd); |
| 689 | if (server.sofd > 0) close(server.sofd); |
| 690 | snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid()); |
| 691 | if (rewriteAppendOnlyFile(tmpfile) == REDIS_OK) { |
| 692 | _exit(0); |
| 693 | } else { |
| 694 | _exit(1); |
| 695 | } |
| 696 | } else { |
| 697 | /* Parent */ |
| 698 | server.stat_fork_time = ustime()-start; |
| 699 | if (childpid == -1) { |
| 700 | redisLog(REDIS_WARNING, |
| 701 | "Can't rewrite append only file in background: fork: %s", |
| 702 | strerror(errno)); |
| 703 | return REDIS_ERR; |
| 704 | } |
| 705 | redisLog(REDIS_NOTICE, |
| 706 | "Background append only file rewriting started by pid %d",childpid); |
| 707 | server.aofrewrite_scheduled = 0; |
| 708 | server.bgrewritechildpid = childpid; |
| 709 | updateDictResizePolicy(); |
| 710 | /* We set appendseldb to -1 in order to force the next call to the |
| 711 | * feedAppendOnlyFile() to issue a SELECT command, so the differences |
| 712 | * accumulated by the parent into server.bgrewritebuf will start |
| 713 | * with a SELECT statement and it will be safe to merge. */ |
| 714 | server.appendseldb = -1; |
| 715 | return REDIS_OK; |
| 716 | } |
| 717 | return REDIS_OK; /* unreached */ |
| 718 | } |
| 719 | |
| 720 | void bgrewriteaofCommand(redisClient *c) { |
| 721 | if (server.bgrewritechildpid != -1) { |
| 722 | addReplyError(c,"Background append only file rewriting already in progress"); |
| 723 | } else if (server.bgsavechildpid != -1) { |
| 724 | server.aofrewrite_scheduled = 1; |
| 725 | addReplyStatus(c,"Background append only file rewriting scheduled"); |
| 726 | } else if (rewriteAppendOnlyFileBackground() == REDIS_OK) { |
| 727 | addReplyStatus(c,"Background append only file rewriting started"); |
| 728 | } else { |
| 729 | addReply(c,shared.err); |
| 730 | } |
| 731 | } |
| 732 | |
| 733 | void aofRemoveTempFile(pid_t childpid) { |
| 734 | char tmpfile[256]; |
| 735 | |
| 736 | snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) childpid); |
| 737 | unlink(tmpfile); |
| 738 | } |
| 739 | |
| 740 | /* Update the server.appendonly_current_size filed explicitly using stat(2) |
| 741 | * to check the size of the file. This is useful after a rewrite or after |
| 742 | * a restart, normally the size is updated just adding the write length |
| 743 | * to the current lenght, that is much faster. */ |
| 744 | void aofUpdateCurrentSize(void) { |
| 745 | struct redis_stat sb; |
| 746 | |
| 747 | if (redis_fstat(server.appendfd,&sb) == -1) { |
| 748 | redisLog(REDIS_WARNING,"Unable to check the AOF length: %s", |
| 749 | strerror(errno)); |
| 750 | } else { |
| 751 | server.appendonly_current_size = sb.st_size; |
| 752 | } |
| 753 | } |
| 754 | |
| 755 | /* A background append only file rewriting (BGREWRITEAOF) terminated its work. |
| 756 | * Handle this. */ |
| 757 | void backgroundRewriteDoneHandler(int exitcode, int bysignal) { |
| 758 | if (!bysignal && exitcode == 0) { |
| 759 | int newfd, oldfd; |
| 760 | int nwritten; |
| 761 | char tmpfile[256]; |
| 762 | long long now = ustime(); |
| 763 | |
| 764 | redisLog(REDIS_NOTICE, |
| 765 | "Background AOF rewrite terminated with success"); |
| 766 | |
| 767 | /* Flush the differences accumulated by the parent to the |
| 768 | * rewritten AOF. */ |
| 769 | snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", |
| 770 | (int)server.bgrewritechildpid); |
| 771 | newfd = open(tmpfile,O_WRONLY|O_APPEND); |
| 772 | if (newfd == -1) { |
| 773 | redisLog(REDIS_WARNING, |
| 774 | "Unable to open the temporary AOF produced by the child: %s", strerror(errno)); |
| 775 | goto cleanup; |
| 776 | } |
| 777 | |
| 778 | nwritten = write(newfd,server.bgrewritebuf,sdslen(server.bgrewritebuf)); |
| 779 | if (nwritten != (signed)sdslen(server.bgrewritebuf)) { |
| 780 | if (nwritten == -1) { |
| 781 | redisLog(REDIS_WARNING, |
| 782 | "Error trying to flush the parent diff to the rewritten AOF: %s", strerror(errno)); |
| 783 | } else { |
| 784 | redisLog(REDIS_WARNING, |
| 785 | "Short write trying to flush the parent diff to the rewritten AOF: %s", strerror(errno)); |
| 786 | } |
| 787 | close(newfd); |
| 788 | goto cleanup; |
| 789 | } |
| 790 | |
| 791 | redisLog(REDIS_NOTICE, |
| 792 | "Parent diff successfully flushed to the rewritten AOF (%lu bytes)", nwritten); |
| 793 | |
| 794 | /* The only remaining thing to do is to rename the temporary file to |
| 795 | * the configured file and switch the file descriptor used to do AOF |
| 796 | * writes. We don't want close(2) or rename(2) calls to block the |
| 797 | * server on old file deletion. |
| 798 | * |
| 799 | * There are two possible scenarios: |
| 800 | * |
| 801 | * 1) AOF is DISABLED and this was a one time rewrite. The temporary |
| 802 | * file will be renamed to the configured file. When this file already |
| 803 | * exists, it will be unlinked, which may block the server. |
| 804 | * |
| 805 | * 2) AOF is ENABLED and the rewritten AOF will immediately start |
| 806 | * receiving writes. After the temporary file is renamed to the |
| 807 | * configured file, the original AOF file descriptor will be closed. |
| 808 | * Since this will be the last reference to that file, closing it |
| 809 | * causes the underlying file to be unlinked, which may block the |
| 810 | * server. |
| 811 | * |
| 812 | * To mitigate the blocking effect of the unlink operation (either |
| 813 | * caused by rename(2) in scenario 1, or by close(2) in scenario 2), we |
| 814 | * use a background thread to take care of this. First, we |
| 815 | * make scenario 1 identical to scenario 2 by opening the target file |
| 816 | * when it exists. The unlink operation after the rename(2) will then |
| 817 | * be executed upon calling close(2) for its descriptor. Everything to |
| 818 | * guarantee atomicity for this switch has already happened by then, so |
| 819 | * we don't care what the outcome or duration of that close operation |
| 820 | * is, as long as the file descriptor is released again. */ |
| 821 | if (server.appendfd == -1) { |
| 822 | /* AOF disabled */ |
| 823 | |
| 824 | /* Don't care if this fails: oldfd will be -1 and we handle that. |
| 825 | * One notable case of -1 return is if the old file does |
| 826 | * not exist. */ |
| 827 | oldfd = open(server.appendfilename,O_RDONLY|O_NONBLOCK); |
| 828 | } else { |
| 829 | /* AOF enabled */ |
| 830 | oldfd = -1; /* We'll set this to the current AOF filedes later. */ |
| 831 | } |
| 832 | |
| 833 | /* Rename the temporary file. This will not unlink the target file if |
| 834 | * it exists, because we reference it with "oldfd". */ |
| 835 | if (rename(tmpfile,server.appendfilename) == -1) { |
| 836 | redisLog(REDIS_WARNING, |
| 837 | "Error trying to rename the temporary AOF: %s", strerror(errno)); |
| 838 | close(newfd); |
| 839 | if (oldfd != -1) close(oldfd); |
| 840 | goto cleanup; |
| 841 | } |
| 842 | |
| 843 | if (server.appendfd == -1) { |
| 844 | /* AOF disabled, we don't need to set the AOF file descriptor |
| 845 | * to this new file, so we can close it. */ |
| 846 | close(newfd); |
| 847 | } else { |
| 848 | /* AOF enabled, replace the old fd with the new one. */ |
| 849 | oldfd = server.appendfd; |
| 850 | server.appendfd = newfd; |
| 851 | if (server.appendfsync == APPENDFSYNC_ALWAYS) |
| 852 | aof_fsync(newfd); |
| 853 | else if (server.appendfsync == APPENDFSYNC_EVERYSEC) |
| 854 | aof_background_fsync(newfd); |
| 855 | server.appendseldb = -1; /* Make sure SELECT is re-issued */ |
| 856 | aofUpdateCurrentSize(); |
| 857 | server.auto_aofrewrite_base_size = server.appendonly_current_size; |
| 858 | |
| 859 | /* Clear regular AOF buffer since its contents was just written to |
| 860 | * the new AOF from the background rewrite buffer. */ |
| 861 | sdsfree(server.aofbuf); |
| 862 | server.aofbuf = sdsempty(); |
| 863 | } |
| 864 | |
| 865 | redisLog(REDIS_NOTICE, "Background AOF rewrite successful"); |
| 866 | |
| 867 | /* Asynchronously close the overwritten AOF. */ |
| 868 | if (oldfd != -1) bioCreateBackgroundJob(REDIS_BIO_CLOSE_FILE,(void*)(long)oldfd,NULL,NULL); |
| 869 | |
| 870 | redisLog(REDIS_VERBOSE, |
| 871 | "Background AOF rewrite signal handler took %lldus", ustime()-now); |
| 872 | } else if (!bysignal && exitcode != 0) { |
| 873 | redisLog(REDIS_WARNING, |
| 874 | "Background AOF rewrite terminated with error"); |
| 875 | } else { |
| 876 | redisLog(REDIS_WARNING, |
| 877 | "Background AOF rewrite terminated by signal %d", bysignal); |
| 878 | } |
| 879 | |
| 880 | cleanup: |
| 881 | sdsfree(server.bgrewritebuf); |
| 882 | server.bgrewritebuf = sdsempty(); |
| 883 | aofRemoveTempFile(server.bgrewritechildpid); |
| 884 | server.bgrewritechildpid = -1; |
| 885 | } |