]>
Commit | Line | Data |
---|---|---|
e2641e09 | 1 | #include "redis.h" |
2 | ||
3 | #include <signal.h> | |
4 | #include <fcntl.h> | |
5 | #include <sys/stat.h> | |
3688d7f3 | 6 | #include <sys/types.h> |
7 | #include <sys/time.h> | |
8 | #include <sys/resource.h> | |
9 | #include <sys/wait.h> | |
e2641e09 | 10 | |
11 | /* Called when the user switches from "appendonly yes" to "appendonly no" | |
12 | * at runtime using the CONFIG command. */ | |
13 | void stopAppendOnly(void) { | |
14 | flushAppendOnlyFile(); | |
15 | aof_fsync(server.appendfd); | |
16 | close(server.appendfd); | |
17 | ||
18 | server.appendfd = -1; | |
19 | server.appendseldb = -1; | |
20 | server.appendonly = 0; | |
21 | /* rewrite operation in progress? kill it, wait child exit */ | |
22 | if (server.bgsavechildpid != -1) { | |
23 | int statloc; | |
24 | ||
25 | if (kill(server.bgsavechildpid,SIGKILL) != -1) | |
26 | wait3(&statloc,0,NULL); | |
27 | /* reset the buffer accumulating changes while the child saves */ | |
28 | sdsfree(server.bgrewritebuf); | |
29 | server.bgrewritebuf = sdsempty(); | |
30 | server.bgsavechildpid = -1; | |
31 | } | |
32 | } | |
33 | ||
34 | /* Called when the user switches from "appendonly no" to "appendonly yes" | |
35 | * at runtime using the CONFIG command. */ | |
36 | int startAppendOnly(void) { | |
37 | server.appendonly = 1; | |
38 | server.lastfsync = time(NULL); | |
39 | server.appendfd = open(server.appendfilename,O_WRONLY|O_APPEND|O_CREAT,0644); | |
40 | if (server.appendfd == -1) { | |
41 | redisLog(REDIS_WARNING,"Used tried to switch on AOF via CONFIG, but I can't open the AOF file: %s",strerror(errno)); | |
42 | return REDIS_ERR; | |
43 | } | |
44 | if (rewriteAppendOnlyFileBackground() == REDIS_ERR) { | |
45 | server.appendonly = 0; | |
46 | close(server.appendfd); | |
47 | redisLog(REDIS_WARNING,"Used tried to switch on AOF via CONFIG, I can't trigger a background AOF rewrite operation. Check the above logs for more info about the error.",strerror(errno)); | |
48 | return REDIS_ERR; | |
49 | } | |
50 | return REDIS_OK; | |
51 | } | |
52 | ||
53 | /* Write the append only file buffer on disk. | |
54 | * | |
55 | * Since we are required to write the AOF before replying to the client, | |
56 | * and the only way the client socket can get a write is entering when the | |
57 | * the event loop, we accumulate all the AOF writes in a memory | |
58 | * buffer and write it on disk using this function just before entering | |
59 | * the event loop again. */ | |
60 | void flushAppendOnlyFile(void) { | |
61 | time_t now; | |
62 | ssize_t nwritten; | |
63 | ||
64 | if (sdslen(server.aofbuf) == 0) return; | |
65 | ||
66 | /* We want to perform a single write. This should be guaranteed atomic | |
67 | * at least if the filesystem we are writing is a real physical one. | |
68 | * While this will save us against the server being killed I don't think | |
69 | * there is much to do about the whole server stopping for power problems | |
70 | * or alike */ | |
71 | nwritten = write(server.appendfd,server.aofbuf,sdslen(server.aofbuf)); | |
72 | if (nwritten != (signed)sdslen(server.aofbuf)) { | |
73 | /* Ooops, we are in troubles. The best thing to do for now is | |
74 | * aborting instead of giving the illusion that everything is | |
75 | * working as expected. */ | |
76 | if (nwritten == -1) { | |
77 | redisLog(REDIS_WARNING,"Exiting on error writing to the append-only file: %s",strerror(errno)); | |
78 | } else { | |
79 | redisLog(REDIS_WARNING,"Exiting on short write while writing to the append-only file: %s",strerror(errno)); | |
80 | } | |
81 | exit(1); | |
82 | } | |
83 | sdsfree(server.aofbuf); | |
84 | server.aofbuf = sdsempty(); | |
85 | ||
86 | /* Don't Fsync if no-appendfsync-on-rewrite is set to yes and we have | |
87 | * childs performing heavy I/O on disk. */ | |
88 | if (server.no_appendfsync_on_rewrite && | |
89 | (server.bgrewritechildpid != -1 || server.bgsavechildpid != -1)) | |
90 | return; | |
91 | /* Fsync if needed */ | |
92 | now = time(NULL); | |
93 | if (server.appendfsync == APPENDFSYNC_ALWAYS || | |
94 | (server.appendfsync == APPENDFSYNC_EVERYSEC && | |
95 | now-server.lastfsync > 1)) | |
96 | { | |
97 | /* aof_fsync is defined as fdatasync() for Linux in order to avoid | |
98 | * flushing metadata. */ | |
99 | aof_fsync(server.appendfd); /* Let's try to get this data on the disk */ | |
100 | server.lastfsync = now; | |
101 | } | |
102 | } | |
103 | ||
104 | sds catAppendOnlyGenericCommand(sds buf, int argc, robj **argv) { | |
105 | int j; | |
106 | buf = sdscatprintf(buf,"*%d\r\n",argc); | |
107 | for (j = 0; j < argc; j++) { | |
108 | robj *o = getDecodedObject(argv[j]); | |
109 | buf = sdscatprintf(buf,"$%lu\r\n",(unsigned long)sdslen(o->ptr)); | |
110 | buf = sdscatlen(buf,o->ptr,sdslen(o->ptr)); | |
111 | buf = sdscatlen(buf,"\r\n",2); | |
112 | decrRefCount(o); | |
113 | } | |
114 | return buf; | |
115 | } | |
116 | ||
117 | sds catAppendOnlyExpireAtCommand(sds buf, robj *key, robj *seconds) { | |
118 | int argc = 3; | |
119 | long when; | |
120 | robj *argv[3]; | |
121 | ||
122 | /* Make sure we can use strtol */ | |
123 | seconds = getDecodedObject(seconds); | |
124 | when = time(NULL)+strtol(seconds->ptr,NULL,10); | |
125 | decrRefCount(seconds); | |
126 | ||
127 | argv[0] = createStringObject("EXPIREAT",8); | |
128 | argv[1] = key; | |
129 | argv[2] = createObject(REDIS_STRING, | |
130 | sdscatprintf(sdsempty(),"%ld",when)); | |
131 | buf = catAppendOnlyGenericCommand(buf, argc, argv); | |
132 | decrRefCount(argv[0]); | |
133 | decrRefCount(argv[2]); | |
134 | return buf; | |
135 | } | |
136 | ||
137 | void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int argc) { | |
138 | sds buf = sdsempty(); | |
139 | robj *tmpargv[3]; | |
140 | ||
141 | /* The DB this command was targetting is not the same as the last command | |
142 | * we appendend. To issue a SELECT command is needed. */ | |
143 | if (dictid != server.appendseldb) { | |
144 | char seldb[64]; | |
145 | ||
146 | snprintf(seldb,sizeof(seldb),"%d",dictid); | |
147 | buf = sdscatprintf(buf,"*2\r\n$6\r\nSELECT\r\n$%lu\r\n%s\r\n", | |
148 | (unsigned long)strlen(seldb),seldb); | |
149 | server.appendseldb = dictid; | |
150 | } | |
151 | ||
152 | if (cmd->proc == expireCommand) { | |
153 | /* Translate EXPIRE into EXPIREAT */ | |
154 | buf = catAppendOnlyExpireAtCommand(buf,argv[1],argv[2]); | |
155 | } else if (cmd->proc == setexCommand) { | |
156 | /* Translate SETEX to SET and EXPIREAT */ | |
157 | tmpargv[0] = createStringObject("SET",3); | |
158 | tmpargv[1] = argv[1]; | |
159 | tmpargv[2] = argv[3]; | |
160 | buf = catAppendOnlyGenericCommand(buf,3,tmpargv); | |
161 | decrRefCount(tmpargv[0]); | |
162 | buf = catAppendOnlyExpireAtCommand(buf,argv[1],argv[2]); | |
163 | } else { | |
164 | buf = catAppendOnlyGenericCommand(buf,argc,argv); | |
165 | } | |
166 | ||
167 | /* Append to the AOF buffer. This will be flushed on disk just before | |
168 | * of re-entering the event loop, so before the client will get a | |
169 | * positive reply about the operation performed. */ | |
170 | server.aofbuf = sdscatlen(server.aofbuf,buf,sdslen(buf)); | |
171 | ||
172 | /* If a background append only file rewriting is in progress we want to | |
173 | * accumulate the differences between the child DB and the current one | |
174 | * in a buffer, so that when the child process will do its work we | |
175 | * can append the differences to the new append only file. */ | |
176 | if (server.bgrewritechildpid != -1) | |
177 | server.bgrewritebuf = sdscatlen(server.bgrewritebuf,buf,sdslen(buf)); | |
178 | ||
179 | sdsfree(buf); | |
180 | } | |
181 | ||
182 | /* In Redis commands are always executed in the context of a client, so in | |
183 | * order to load the append only file we need to create a fake client. */ | |
184 | struct redisClient *createFakeClient(void) { | |
185 | struct redisClient *c = zmalloc(sizeof(*c)); | |
186 | ||
187 | selectDb(c,0); | |
188 | c->fd = -1; | |
189 | c->querybuf = sdsempty(); | |
190 | c->argc = 0; | |
191 | c->argv = NULL; | |
2403fc9f | 192 | c->bufpos = 0; |
e2641e09 | 193 | c->flags = 0; |
194 | /* We set the fake client as a slave waiting for the synchronization | |
195 | * so that Redis will not try to send replies to this client. */ | |
196 | c->replstate = REDIS_REPL_WAIT_BGSAVE_START; | |
197 | c->reply = listCreate(); | |
b67d2345 | 198 | c->watched_keys = listCreate(); |
e2641e09 | 199 | listSetFreeMethod(c->reply,decrRefCount); |
200 | listSetDupMethod(c->reply,dupClientReplyValue); | |
201 | initClientMultiState(c); | |
202 | return c; | |
203 | } | |
204 | ||
205 | void freeFakeClient(struct redisClient *c) { | |
206 | sdsfree(c->querybuf); | |
207 | listRelease(c->reply); | |
b67d2345 | 208 | listRelease(c->watched_keys); |
e2641e09 | 209 | freeClientMultiState(c); |
210 | zfree(c); | |
211 | } | |
212 | ||
213 | /* Replay the append log file. On error REDIS_OK is returned. On non fatal | |
214 | * error (the append only file is zero-length) REDIS_ERR is returned. On | |
215 | * fatal error an error message is logged and the program exists. */ | |
216 | int loadAppendOnlyFile(char *filename) { | |
217 | struct redisClient *fakeClient; | |
218 | FILE *fp = fopen(filename,"r"); | |
219 | struct redis_stat sb; | |
220 | int appendonly = server.appendonly; | |
221 | ||
222 | if (redis_fstat(fileno(fp),&sb) != -1 && sb.st_size == 0) | |
223 | return REDIS_ERR; | |
224 | ||
225 | if (fp == NULL) { | |
226 | redisLog(REDIS_WARNING,"Fatal error: can't open the append log file for reading: %s",strerror(errno)); | |
227 | exit(1); | |
228 | } | |
229 | ||
230 | /* Temporarily disable AOF, to prevent EXEC from feeding a MULTI | |
231 | * to the same file we're about to read. */ | |
232 | server.appendonly = 0; | |
233 | ||
234 | fakeClient = createFakeClient(); | |
235 | while(1) { | |
236 | int argc, j; | |
237 | unsigned long len; | |
238 | robj **argv; | |
239 | char buf[128]; | |
240 | sds argsds; | |
241 | struct redisCommand *cmd; | |
242 | int force_swapout; | |
243 | ||
244 | if (fgets(buf,sizeof(buf),fp) == NULL) { | |
245 | if (feof(fp)) | |
246 | break; | |
247 | else | |
248 | goto readerr; | |
249 | } | |
250 | if (buf[0] != '*') goto fmterr; | |
251 | argc = atoi(buf+1); | |
252 | argv = zmalloc(sizeof(robj*)*argc); | |
253 | for (j = 0; j < argc; j++) { | |
254 | if (fgets(buf,sizeof(buf),fp) == NULL) goto readerr; | |
255 | if (buf[0] != '$') goto fmterr; | |
256 | len = strtol(buf+1,NULL,10); | |
257 | argsds = sdsnewlen(NULL,len); | |
258 | if (len && fread(argsds,len,1,fp) == 0) goto fmterr; | |
259 | argv[j] = createObject(REDIS_STRING,argsds); | |
260 | if (fread(buf,2,1,fp) == 0) goto fmterr; /* discard CRLF */ | |
261 | } | |
262 | ||
263 | /* Command lookup */ | |
264 | cmd = lookupCommand(argv[0]->ptr); | |
265 | if (!cmd) { | |
266 | redisLog(REDIS_WARNING,"Unknown command '%s' reading the append only file", argv[0]->ptr); | |
267 | exit(1); | |
268 | } | |
269 | /* Try object encoding */ | |
270 | if (cmd->flags & REDIS_CMD_BULK) | |
271 | argv[argc-1] = tryObjectEncoding(argv[argc-1]); | |
272 | /* Run the command in the context of a fake client */ | |
273 | fakeClient->argc = argc; | |
274 | fakeClient->argv = argv; | |
275 | cmd->proc(fakeClient); | |
57b07380 PN |
276 | |
277 | /* The fake client should not have a reply */ | |
278 | redisAssert(fakeClient->bufpos == 0 && listLength(fakeClient->reply) == 0); | |
279 | ||
e2641e09 | 280 | /* Clean up, ready for the next command */ |
281 | for (j = 0; j < argc; j++) decrRefCount(argv[j]); | |
282 | zfree(argv); | |
57b07380 | 283 | |
e2641e09 | 284 | /* Handle swapping while loading big datasets when VM is on */ |
285 | force_swapout = 0; | |
286 | if ((zmalloc_used_memory() - server.vm_max_memory) > 1024*1024*32) | |
287 | force_swapout = 1; | |
288 | ||
289 | if (server.vm_enabled && force_swapout) { | |
290 | while (zmalloc_used_memory() > server.vm_max_memory) { | |
291 | if (vmSwapOneObjectBlocking() == REDIS_ERR) break; | |
292 | } | |
293 | } | |
294 | } | |
295 | ||
296 | /* This point can only be reached when EOF is reached without errors. | |
297 | * If the client is in the middle of a MULTI/EXEC, log error and quit. */ | |
298 | if (fakeClient->flags & REDIS_MULTI) goto readerr; | |
299 | ||
300 | fclose(fp); | |
301 | freeFakeClient(fakeClient); | |
302 | server.appendonly = appendonly; | |
303 | return REDIS_OK; | |
304 | ||
305 | readerr: | |
306 | if (feof(fp)) { | |
307 | redisLog(REDIS_WARNING,"Unexpected end of file reading the append only file"); | |
308 | } else { | |
309 | redisLog(REDIS_WARNING,"Unrecoverable error reading the append only file: %s", strerror(errno)); | |
310 | } | |
311 | exit(1); | |
312 | fmterr: | |
412e457c | 313 | 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>"); |
e2641e09 | 314 | exit(1); |
315 | } | |
316 | ||
317 | /* Write binary-safe string into a file in the bulkformat | |
318 | * $<count>\r\n<payload>\r\n */ | |
319 | int fwriteBulkString(FILE *fp, char *s, unsigned long len) { | |
320 | char cbuf[128]; | |
321 | int clen; | |
322 | cbuf[0] = '$'; | |
323 | clen = 1+ll2string(cbuf+1,sizeof(cbuf)-1,len); | |
324 | cbuf[clen++] = '\r'; | |
325 | cbuf[clen++] = '\n'; | |
326 | if (fwrite(cbuf,clen,1,fp) == 0) return 0; | |
327 | if (len > 0 && fwrite(s,len,1,fp) == 0) return 0; | |
328 | if (fwrite("\r\n",2,1,fp) == 0) return 0; | |
329 | return 1; | |
330 | } | |
331 | ||
332 | /* Write a double value in bulk format $<count>\r\n<payload>\r\n */ | |
333 | int fwriteBulkDouble(FILE *fp, double d) { | |
334 | char buf[128], dbuf[128]; | |
335 | ||
336 | snprintf(dbuf,sizeof(dbuf),"%.17g\r\n",d); | |
337 | snprintf(buf,sizeof(buf),"$%lu\r\n",(unsigned long)strlen(dbuf)-2); | |
338 | if (fwrite(buf,strlen(buf),1,fp) == 0) return 0; | |
339 | if (fwrite(dbuf,strlen(dbuf),1,fp) == 0) return 0; | |
340 | return 1; | |
341 | } | |
342 | ||
343 | /* Write a long value in bulk format $<count>\r\n<payload>\r\n */ | |
344 | int fwriteBulkLongLong(FILE *fp, long long l) { | |
345 | char bbuf[128], lbuf[128]; | |
346 | unsigned int blen, llen; | |
347 | llen = ll2string(lbuf,32,l); | |
348 | blen = snprintf(bbuf,sizeof(bbuf),"$%u\r\n%s\r\n",llen,lbuf); | |
349 | if (fwrite(bbuf,blen,1,fp) == 0) return 0; | |
350 | return 1; | |
351 | } | |
352 | ||
353 | /* Delegate writing an object to writing a bulk string or bulk long long. */ | |
354 | int fwriteBulkObject(FILE *fp, robj *obj) { | |
355 | /* Avoid using getDecodedObject to help copy-on-write (we are often | |
356 | * in a child process when this function is called). */ | |
357 | if (obj->encoding == REDIS_ENCODING_INT) { | |
358 | return fwriteBulkLongLong(fp,(long)obj->ptr); | |
359 | } else if (obj->encoding == REDIS_ENCODING_RAW) { | |
360 | return fwriteBulkString(fp,obj->ptr,sdslen(obj->ptr)); | |
361 | } else { | |
362 | redisPanic("Unknown string encoding"); | |
363 | } | |
364 | } | |
365 | ||
366 | /* Write a sequence of commands able to fully rebuild the dataset into | |
367 | * "filename". Used both by REWRITEAOF and BGREWRITEAOF. */ | |
368 | int rewriteAppendOnlyFile(char *filename) { | |
369 | dictIterator *di = NULL; | |
370 | dictEntry *de; | |
371 | FILE *fp; | |
372 | char tmpfile[256]; | |
373 | int j; | |
374 | time_t now = time(NULL); | |
375 | ||
376 | /* Note that we have to use a different temp name here compared to the | |
377 | * one used by rewriteAppendOnlyFileBackground() function. */ | |
378 | snprintf(tmpfile,256,"temp-rewriteaof-%d.aof", (int) getpid()); | |
379 | fp = fopen(tmpfile,"w"); | |
380 | if (!fp) { | |
381 | redisLog(REDIS_WARNING, "Failed rewriting the append only file: %s", strerror(errno)); | |
382 | return REDIS_ERR; | |
383 | } | |
384 | for (j = 0; j < server.dbnum; j++) { | |
385 | char selectcmd[] = "*2\r\n$6\r\nSELECT\r\n"; | |
386 | redisDb *db = server.db+j; | |
387 | dict *d = db->dict; | |
388 | if (dictSize(d) == 0) continue; | |
389 | di = dictGetIterator(d); | |
390 | if (!di) { | |
391 | fclose(fp); | |
392 | return REDIS_ERR; | |
393 | } | |
394 | ||
395 | /* SELECT the new DB */ | |
396 | if (fwrite(selectcmd,sizeof(selectcmd)-1,1,fp) == 0) goto werr; | |
397 | if (fwriteBulkLongLong(fp,j) == 0) goto werr; | |
398 | ||
399 | /* Iterate this DB writing every entry */ | |
400 | while((de = dictNext(di)) != NULL) { | |
401 | sds keystr = dictGetEntryKey(de); | |
402 | robj key, *o; | |
403 | time_t expiretime; | |
404 | int swapped; | |
405 | ||
406 | keystr = dictGetEntryKey(de); | |
407 | o = dictGetEntryVal(de); | |
408 | initStaticStringObject(key,keystr); | |
409 | /* If the value for this key is swapped, load a preview in memory. | |
410 | * We use a "swapped" flag to remember if we need to free the | |
411 | * value object instead to just increment the ref count anyway | |
412 | * in order to avoid copy-on-write of pages if we are forked() */ | |
413 | if (!server.vm_enabled || o->storage == REDIS_VM_MEMORY || | |
414 | o->storage == REDIS_VM_SWAPPING) { | |
415 | swapped = 0; | |
416 | } else { | |
417 | o = vmPreviewObject(o); | |
418 | swapped = 1; | |
419 | } | |
420 | expiretime = getExpire(db,&key); | |
421 | ||
422 | /* Save the key and associated value */ | |
423 | if (o->type == REDIS_STRING) { | |
424 | /* Emit a SET command */ | |
425 | char cmd[]="*3\r\n$3\r\nSET\r\n"; | |
426 | if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr; | |
427 | /* Key and value */ | |
428 | if (fwriteBulkObject(fp,&key) == 0) goto werr; | |
429 | if (fwriteBulkObject(fp,o) == 0) goto werr; | |
430 | } else if (o->type == REDIS_LIST) { | |
431 | /* Emit the RPUSHes needed to rebuild the list */ | |
432 | char cmd[]="*3\r\n$5\r\nRPUSH\r\n"; | |
433 | if (o->encoding == REDIS_ENCODING_ZIPLIST) { | |
434 | unsigned char *zl = o->ptr; | |
435 | unsigned char *p = ziplistIndex(zl,0); | |
436 | unsigned char *vstr; | |
437 | unsigned int vlen; | |
438 | long long vlong; | |
439 | ||
440 | while(ziplistGet(p,&vstr,&vlen,&vlong)) { | |
441 | if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr; | |
442 | if (fwriteBulkObject(fp,&key) == 0) goto werr; | |
443 | if (vstr) { | |
444 | if (fwriteBulkString(fp,(char*)vstr,vlen) == 0) | |
445 | goto werr; | |
446 | } else { | |
447 | if (fwriteBulkLongLong(fp,vlong) == 0) | |
448 | goto werr; | |
449 | } | |
450 | p = ziplistNext(zl,p); | |
451 | } | |
452 | } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) { | |
453 | list *list = o->ptr; | |
454 | listNode *ln; | |
455 | listIter li; | |
456 | ||
457 | listRewind(list,&li); | |
458 | while((ln = listNext(&li))) { | |
459 | robj *eleobj = listNodeValue(ln); | |
460 | ||
461 | if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr; | |
462 | if (fwriteBulkObject(fp,&key) == 0) goto werr; | |
463 | if (fwriteBulkObject(fp,eleobj) == 0) goto werr; | |
464 | } | |
465 | } else { | |
466 | redisPanic("Unknown list encoding"); | |
467 | } | |
468 | } else if (o->type == REDIS_SET) { | |
2767f1c0 | 469 | char cmd[]="*3\r\n$4\r\nSADD\r\n"; |
e2641e09 | 470 | |
2767f1c0 PN |
471 | /* Emit the SADDs needed to rebuild the set */ |
472 | if (o->encoding == REDIS_ENCODING_INTSET) { | |
473 | int ii = 0; | |
23c64fe5 | 474 | int64_t llval; |
2767f1c0 PN |
475 | while(intsetGet(o->ptr,ii++,&llval)) { |
476 | if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr; | |
477 | if (fwriteBulkObject(fp,&key) == 0) goto werr; | |
478 | if (fwriteBulkLongLong(fp,llval) == 0) goto werr; | |
479 | } | |
480 | } else if (o->encoding == REDIS_ENCODING_HT) { | |
481 | dictIterator *di = dictGetIterator(o->ptr); | |
482 | dictEntry *de; | |
483 | while((de = dictNext(di)) != NULL) { | |
484 | robj *eleobj = dictGetEntryKey(de); | |
485 | if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr; | |
486 | if (fwriteBulkObject(fp,&key) == 0) goto werr; | |
487 | if (fwriteBulkObject(fp,eleobj) == 0) goto werr; | |
488 | } | |
489 | dictReleaseIterator(di); | |
490 | } else { | |
491 | redisPanic("Unknown set encoding"); | |
e2641e09 | 492 | } |
e2641e09 | 493 | } else if (o->type == REDIS_ZSET) { |
494 | /* Emit the ZADDs needed to rebuild the sorted set */ | |
495 | zset *zs = o->ptr; | |
496 | dictIterator *di = dictGetIterator(zs->dict); | |
497 | dictEntry *de; | |
498 | ||
499 | while((de = dictNext(di)) != NULL) { | |
500 | char cmd[]="*4\r\n$4\r\nZADD\r\n"; | |
501 | robj *eleobj = dictGetEntryKey(de); | |
502 | double *score = dictGetEntryVal(de); | |
503 | ||
504 | if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr; | |
505 | if (fwriteBulkObject(fp,&key) == 0) goto werr; | |
506 | if (fwriteBulkDouble(fp,*score) == 0) goto werr; | |
507 | if (fwriteBulkObject(fp,eleobj) == 0) goto werr; | |
508 | } | |
509 | dictReleaseIterator(di); | |
510 | } else if (o->type == REDIS_HASH) { | |
511 | char cmd[]="*4\r\n$4\r\nHSET\r\n"; | |
512 | ||
513 | /* Emit the HSETs needed to rebuild the hash */ | |
514 | if (o->encoding == REDIS_ENCODING_ZIPMAP) { | |
515 | unsigned char *p = zipmapRewind(o->ptr); | |
516 | unsigned char *field, *val; | |
517 | unsigned int flen, vlen; | |
518 | ||
519 | while((p = zipmapNext(p,&field,&flen,&val,&vlen)) != NULL) { | |
520 | if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr; | |
521 | if (fwriteBulkObject(fp,&key) == 0) goto werr; | |
daf2049d | 522 | if (fwriteBulkString(fp,(char*)field,flen) == 0) |
5bd09cd4 | 523 | goto werr; |
daf2049d | 524 | if (fwriteBulkString(fp,(char*)val,vlen) == 0) |
5bd09cd4 | 525 | goto werr; |
e2641e09 | 526 | } |
527 | } else { | |
528 | dictIterator *di = dictGetIterator(o->ptr); | |
529 | dictEntry *de; | |
530 | ||
531 | while((de = dictNext(di)) != NULL) { | |
532 | robj *field = dictGetEntryKey(de); | |
533 | robj *val = dictGetEntryVal(de); | |
534 | ||
535 | if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr; | |
536 | if (fwriteBulkObject(fp,&key) == 0) goto werr; | |
5bd09cd4 | 537 | if (fwriteBulkObject(fp,field) == 0) goto werr; |
538 | if (fwriteBulkObject(fp,val) == 0) goto werr; | |
e2641e09 | 539 | } |
540 | dictReleaseIterator(di); | |
541 | } | |
542 | } else { | |
543 | redisPanic("Unknown object type"); | |
544 | } | |
545 | /* Save the expire time */ | |
546 | if (expiretime != -1) { | |
547 | char cmd[]="*3\r\n$8\r\nEXPIREAT\r\n"; | |
548 | /* If this key is already expired skip it */ | |
549 | if (expiretime < now) continue; | |
550 | if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr; | |
551 | if (fwriteBulkObject(fp,&key) == 0) goto werr; | |
552 | if (fwriteBulkLongLong(fp,expiretime) == 0) goto werr; | |
553 | } | |
554 | if (swapped) decrRefCount(o); | |
555 | } | |
556 | dictReleaseIterator(di); | |
557 | } | |
558 | ||
559 | /* Make sure data will not remain on the OS's output buffers */ | |
560 | fflush(fp); | |
561 | aof_fsync(fileno(fp)); | |
562 | fclose(fp); | |
563 | ||
564 | /* Use RENAME to make sure the DB file is changed atomically only | |
565 | * if the generate DB file is ok. */ | |
566 | if (rename(tmpfile,filename) == -1) { | |
567 | redisLog(REDIS_WARNING,"Error moving temp append only file on the final destination: %s", strerror(errno)); | |
568 | unlink(tmpfile); | |
569 | return REDIS_ERR; | |
570 | } | |
571 | redisLog(REDIS_NOTICE,"SYNC append only file rewrite performed"); | |
572 | return REDIS_OK; | |
573 | ||
574 | werr: | |
575 | fclose(fp); | |
576 | unlink(tmpfile); | |
577 | redisLog(REDIS_WARNING,"Write error writing append only file on disk: %s", strerror(errno)); | |
578 | if (di) dictReleaseIterator(di); | |
579 | return REDIS_ERR; | |
580 | } | |
581 | ||
582 | /* This is how rewriting of the append only file in background works: | |
583 | * | |
584 | * 1) The user calls BGREWRITEAOF | |
585 | * 2) Redis calls this function, that forks(): | |
586 | * 2a) the child rewrite the append only file in a temp file. | |
587 | * 2b) the parent accumulates differences in server.bgrewritebuf. | |
588 | * 3) When the child finished '2a' exists. | |
589 | * 4) The parent will trap the exit code, if it's OK, will append the | |
590 | * data accumulated into server.bgrewritebuf into the temp file, and | |
591 | * finally will rename(2) the temp file in the actual file name. | |
592 | * The the new file is reopened as the new append only file. Profit! | |
593 | */ | |
594 | int rewriteAppendOnlyFileBackground(void) { | |
595 | pid_t childpid; | |
596 | ||
597 | if (server.bgrewritechildpid != -1) return REDIS_ERR; | |
598 | if (server.vm_enabled) waitEmptyIOJobsQueue(); | |
599 | if ((childpid = fork()) == 0) { | |
600 | /* Child */ | |
601 | char tmpfile[256]; | |
602 | ||
603 | if (server.vm_enabled) vmReopenSwapFile(); | |
604 | close(server.fd); | |
605 | snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid()); | |
606 | if (rewriteAppendOnlyFile(tmpfile) == REDIS_OK) { | |
607 | _exit(0); | |
608 | } else { | |
609 | _exit(1); | |
610 | } | |
611 | } else { | |
612 | /* Parent */ | |
613 | if (childpid == -1) { | |
614 | redisLog(REDIS_WARNING, | |
615 | "Can't rewrite append only file in background: fork: %s", | |
616 | strerror(errno)); | |
617 | return REDIS_ERR; | |
618 | } | |
619 | redisLog(REDIS_NOTICE, | |
620 | "Background append only file rewriting started by pid %d",childpid); | |
621 | server.bgrewritechildpid = childpid; | |
622 | updateDictResizePolicy(); | |
623 | /* We set appendseldb to -1 in order to force the next call to the | |
624 | * feedAppendOnlyFile() to issue a SELECT command, so the differences | |
625 | * accumulated by the parent into server.bgrewritebuf will start | |
626 | * with a SELECT statement and it will be safe to merge. */ | |
627 | server.appendseldb = -1; | |
628 | return REDIS_OK; | |
629 | } | |
630 | return REDIS_OK; /* unreached */ | |
631 | } | |
632 | ||
633 | void bgrewriteaofCommand(redisClient *c) { | |
634 | if (server.bgrewritechildpid != -1) { | |
3ab20376 | 635 | addReplyError(c,"Background append only file rewriting already in progress"); |
e2641e09 | 636 | return; |
637 | } | |
638 | if (rewriteAppendOnlyFileBackground() == REDIS_OK) { | |
3ab20376 | 639 | addReplyStatus(c,"Background append only file rewriting started"); |
e2641e09 | 640 | } else { |
641 | addReply(c,shared.err); | |
642 | } | |
643 | } | |
644 | ||
645 | void aofRemoveTempFile(pid_t childpid) { | |
646 | char tmpfile[256]; | |
647 | ||
648 | snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) childpid); | |
649 | unlink(tmpfile); | |
650 | } | |
651 | ||
652 | /* A background append only file rewriting (BGREWRITEAOF) terminated its work. | |
653 | * Handle this. */ | |
654 | void backgroundRewriteDoneHandler(int statloc) { | |
655 | int exitcode = WEXITSTATUS(statloc); | |
656 | int bysignal = WIFSIGNALED(statloc); | |
657 | ||
658 | if (!bysignal && exitcode == 0) { | |
659 | int fd; | |
660 | char tmpfile[256]; | |
661 | ||
662 | redisLog(REDIS_NOTICE, | |
663 | "Background append only file rewriting terminated with success"); | |
664 | /* Now it's time to flush the differences accumulated by the parent */ | |
665 | snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) server.bgrewritechildpid); | |
666 | fd = open(tmpfile,O_WRONLY|O_APPEND); | |
667 | if (fd == -1) { | |
668 | redisLog(REDIS_WARNING, "Not able to open the temp append only file produced by the child: %s", strerror(errno)); | |
669 | goto cleanup; | |
670 | } | |
671 | /* Flush our data... */ | |
672 | if (write(fd,server.bgrewritebuf,sdslen(server.bgrewritebuf)) != | |
673 | (signed) sdslen(server.bgrewritebuf)) { | |
674 | redisLog(REDIS_WARNING, "Error or short write trying to flush the parent diff of the append log file in the child temp file: %s", strerror(errno)); | |
675 | close(fd); | |
676 | goto cleanup; | |
677 | } | |
678 | redisLog(REDIS_NOTICE,"Parent diff flushed into the new append log file with success (%lu bytes)",sdslen(server.bgrewritebuf)); | |
679 | /* Now our work is to rename the temp file into the stable file. And | |
680 | * switch the file descriptor used by the server for append only. */ | |
681 | if (rename(tmpfile,server.appendfilename) == -1) { | |
682 | redisLog(REDIS_WARNING,"Can't rename the temp append only file into the stable one: %s", strerror(errno)); | |
683 | close(fd); | |
684 | goto cleanup; | |
685 | } | |
686 | /* Mission completed... almost */ | |
687 | redisLog(REDIS_NOTICE,"Append only file successfully rewritten."); | |
688 | if (server.appendfd != -1) { | |
689 | /* If append only is actually enabled... */ | |
690 | close(server.appendfd); | |
691 | server.appendfd = fd; | |
692 | if (server.appendfsync != APPENDFSYNC_NO) aof_fsync(fd); | |
693 | server.appendseldb = -1; /* Make sure it will issue SELECT */ | |
694 | redisLog(REDIS_NOTICE,"The new append only file was selected for future appends."); | |
695 | } else { | |
696 | /* If append only is disabled we just generate a dump in this | |
697 | * format. Why not? */ | |
698 | close(fd); | |
699 | } | |
700 | } else if (!bysignal && exitcode != 0) { | |
701 | redisLog(REDIS_WARNING, "Background append only file rewriting error"); | |
702 | } else { | |
703 | redisLog(REDIS_WARNING, | |
704 | "Background append only file rewriting terminated by signal %d", | |
705 | WTERMSIG(statloc)); | |
706 | } | |
707 | cleanup: | |
708 | sdsfree(server.bgrewritebuf); | |
709 | server.bgrewritebuf = sdsempty(); | |
710 | aofRemoveTempFile(server.bgrewritechildpid); | |
711 | server.bgrewritechildpid = -1; | |
712 | } |