Redis源码分析：snapshot

winston

源码版本：redis 2.4.4

redis的snapshot通过将内存中的所有数据写入文件，以达到持久化的目的。
需要注意的是：
1）snapshot方式不是追加，而是将内存所有数据写入文件，snapshot间隔短的话，会造成磁盘IO频繁
2）在上一次做snapshot到当前，如果机器crash，期间修改过的数据会丢失
redis支持两种方式做snapshot
1）客户端发送bgsave命令（save命令也可以，其会阻塞主线程执行，不推荐）
2）根据配置的周期，定期执行

snapshot配置（redis.conf）：
save A  B
在A秒后，如果至少B个数据发生变化则做snapshot。eg： save 3000100即在3000s后，如果至少100个key发生变化则做snapshot

rdbcompression yes
是否启用压缩（压缩消耗cpu）

dbfilename dump.rdb
snapshot dump的文件名

dir ./
指定工作目录，snapshot文件将保存在该目录，同时，aof文件也将保存在该目录

源码
执行bgsave时执行：

2. void bgsaveCommand(redisClient *c) {
3.     if (server.bgsavechildpid != -1) { //无snapshot子进程运行
4.         addReplyError(c,"Background save already in progress");
5.     } else if (server.bgrewritechildpid != -1) { //无重写aof进程运行
6.         addReplyError(c,"Can't BGSAVE while AOF log rewriting is in progress");
7.     } else if (rdbSaveBackground(server.dbfilename) == REDIS_OK) { //执行snapshot
8.         addReplyStatus(c,"Background saving started");
9.     } else {
10.         addReply(c,shared.err);
11.     }
12. }

复制代码

检查是否满足snapshot条件：

2. int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData)
3. {
4.         .....
5.         /* If there is not a background saving in progress check if
6.          * we have to save now */
7.          for (j = 0; j < server.saveparamslen; j++) {
8.             struct saveparam *sp = server.saveparams+j;
10.             if (server.dirty >= sp->changes &&
11.                 now-server.lastsave > sp->seconds) {
12.                 redisLog(REDIS_NOTICE,"%d changes in %d seconds. Saving...",
13.                     sp->changes, sp->seconds);
14.                 rdbSaveBackground(server.dbfilename);
15.                 break;
16.             }
17.          }
18.         ......
19. }

复制代码

执行snapshot：

3. int rdbSaveBackground(char *filename) {
4.     pid_t childpid;
5.     long long start;
7.     if (server.bgsavechildpid != -1) return REDIS_ERR;
8.     if (server.vm_enabled) waitEmptyIOJobsQueue();
9.     server.dirty_before_bgsave = server.dirty;
10.     start = ustime();
11.     if ((childpid = fork()) == 0) {  //fork子进程做snapshot
12.         /* Child */
13.         if (server.vm_enabled) vmReopenSwapFile();
14.         if (server.ipfd > 0) close(server.ipfd);
15.         if (server.sofd > 0) close(server.sofd);
16.         if (rdbSave(filename) == REDIS_OK) {
17.             _exit(0);
18.         } else {
19.             _exit(1);
20.         }
21.     } else {
22.         /* Parent */
23.         server.stat_fork_time = ustime()-start;
24.         if (childpid == -1) {
25.             redisLog(REDIS_WARNING,"Can't save in background: fork: %s",
26.                 strerror(errno));
27.             return REDIS_ERR;
28.         }
29.         redisLog(REDIS_NOTICE,"Background saving started by pid %d",childpid);
30.         server.bgsavechildpid = childpid;
31.         updateDictResizePolicy();
32.         return REDIS_OK;
33.     }
34.     return REDIS_OK; /* unreached */
35. }

复制代码

利用copy on write调用rdbsave将当前数据状态写入文件。

2. int rdbSave(char *filename) {
3.     dictIterator *di = NULL;
4.     dictEntry *de;
5.     FILE *fp;
6.     char tmpfile[256];
7.     int j;
8.     time_t now = time(NULL);
10.     /* Wait for I/O therads to terminate, just in case this is a
11.      * foreground-saving, to avoid seeking the swap file descriptor at the
12.      * same time. */
13.     if (server.vm_enabled)
14.         waitEmptyIOJobsQueue();
16.     snprintf(tmpfile,256,"temp-%d.rdb", (int) getpid());
17.     fp = fopen(tmpfile,"w");
18.     if (!fp) {
19.         redisLog(REDIS_WARNING, "Failed saving the DB: %s", strerror(errno));
20.         return REDIS_ERR;
21.     }
22.     if (fwrite("REDIS0002",9,1,fp) == 0) goto werr;
23.     for (j = 0; j < server.dbnum; j++) {
24.         redisDb *db = server.db+j;
25.         dict *d = db->dict;
26.         if (dictSize(d) == 0) continue;
27.         di = dictGetSafeIterator(d);
28.         if (!di) {
29.             fclose(fp);
30.             return REDIS_ERR;
31.         }
33.         /* Write the SELECT DB opcode */
34.         if (rdbSaveType(fp,REDIS_SELECTDB) == -1) goto werr;
35.         if (rdbSaveLen(fp,j) == -1) goto werr;
37.         /* Iterate this DB writing every entry */
38.         while((de = dictNext(di)) != NULL) {
39.             sds keystr = dictGetEntryKey(de);
40.             robj key, *o = dictGetEntryVal(de);
41.             time_t expiretime;
42.             
43.             initStaticStringObject(key,keystr);
44.             expiretime = getExpire(db,&key);
46.             /* Save the expire time */
47.             if (expiretime != -1) {
48.                 /* If this key is already expired skip it */
49.                 if (expiretime < now) continue;
50.                 if (rdbSaveType(fp,REDIS_EXPIRETIME) == -1) goto werr;
51.                 if (rdbSaveTime(fp,expiretime) == -1) goto werr;
52.             }
53.             /* Save the key and associated value. This requires special
54.              * handling if the value is swapped out. */
55.             if (!server.vm_enabled || o->storage == REDIS_VM_MEMORY ||
56.                                       o->storage == REDIS_VM_SWAPPING) {
57.                 int otype = getObjectSaveType(o);
59.                 /* Save type, key, value */
60.                 if (rdbSaveType(fp,otype) == -1) goto werr;
61.                 if (rdbSaveStringObject(fp,&key) == -1) goto werr;
62.                 if (rdbSaveObject(fp,o) == -1) goto werr;
63.             } else {
64.                 /* REDIS_VM_SWAPPED or REDIS_VM_LOADING */
65.                 robj *po;
66.                 /* Get a preview of the object in memory */
67.                 po = vmPreviewObject(o);
68.                 /* Save type, key, value */
69.                 if (rdbSaveType(fp,getObjectSaveType(po)) == -1)
70.                     goto werr;
71.                 if (rdbSaveStringObject(fp,&key) == -1) goto werr;
72.                 if (rdbSaveObject(fp,po) == -1) goto werr;
73.                 /* Remove the loaded object from memory */
74.                 decrRefCount(po);
75.             }
76.         }
77.         dictReleaseIterator(di);
78.     }
79.     /* EOF opcode */
80.     if (rdbSaveType(fp,REDIS_EOF) == -1) goto werr;
82.     /* Make sure data will not remain on the OS's output buffers */
83.     fflush(fp);
84.     fsync(fileno(fp));
85.     fclose(fp);
87.     /* Use RENAME to make sure the DB file is changed atomically only
88.      * if the generate DB file is ok. */
89.     if (rename(tmpfile,filename) == -1) {
90.         redisLog(REDIS_WARNING,"Error moving temp DB file on the final destination: %s", strerror(errno));
91.         unlink(tmpfile);
92.         return REDIS_ERR;
93.     }
94.     redisLog(REDIS_NOTICE,"DB saved on disk");
95.     server.dirty = 0;
96.     server.lastsave = time(NULL);
97.     return REDIS_OK;
99. werr:
100.     fclose(fp);
101.     unlink(tmpfile);
102.     redisLog(REDIS_WARNING,"Write error saving DB on disk: %s", strerror(errno));
103.     if (di) dictReleaseIterator(di);
104.     return REDIS_ERR;
105. }

复制代码

在serverCron  中检查做bgsave的子进程是否结束   

1. /* Check if a background saving or AOF rewrite in progress terminated */
2.     if (server.bgsavechildpid != -1 || server.bgrewritechildpid != -1) { //是否有做snapshot或者aof rewrite的子进程存在
3.         int statloc;
4.         pid_t pid;
6.         if ((pid = wait3(&statloc,WNOHANG,NULL)) != 0) {
7.             if (pid == server.bgsavechildpid) {  //结束了对其做后处理
8.                 backgroundSaveDoneHandler(statloc);
9.             } else {
10.                 backgroundRewriteDoneHandler(statloc);
11.             }
12.             updateDictResizePolicy();
13.         }
14.     } else {
15.          time_t now = time(NULL);
18. void backgroundSaveDoneHandler(int statloc) {
19.     int exitcode = WEXITSTATUS(statloc);
20.     int bysignal = WIFSIGNALED(statloc);
21.    
22.     //判断做snapshot是否成功
23.     if (!bysignal && exitcode == 0) {
24.         redisLog(REDIS_NOTICE,
25.             "Background saving terminated with success");
26.         server.dirty = server.dirty - server.dirty_before_bgsave;
27.         server.lastsave = time(NULL);
28.     } else if (!bysignal && exitcode != 0) {
29.         redisLog(REDIS_WARNING, "Background saving error");
30.     } else {
31.         redisLog(REDIS_WARNING,
32.             "Background saving terminated by signal %d", WTERMSIG(statloc));
33.         rdbRemoveTempFile(server.bgsavechildpid);
34.     }
35.     server.bgsavechildpid = -1;
36.     /* Possibly there are slaves waiting for a BGSAVE in order to be served
37.      * (the first stage of SYNC is a bulk transfer of dump.rdb) */
38.     updateSlavesWaitingBgsave(exitcode == 0 ? REDIS_OK : REDIS_ERR);
39. }

复制代码

               作者：yfkiss 发表于2012-1-13 18:16:53 原文链接