Redis源码解析（十五）---aof-appendonlyfile解析

/* AOF的下面的一些代码都用到了一个简单buffer缓存块来进行存储，存储了数据的一些改变操作记录，等到
	缓冲中的达到一定的数据规模时，在持久化地写入到一个文件中，redis采用的方式是append追加的形式，这意味
	每次追加都要调整存储的块的大小，但是不可能会有无限大小的块空间，所以redis在这里引入了块列表的概念，
	设定死一个块的大小，超过单位块大小，存入另一个块中,这里定义每个块的大小为10M. */
#define AOF_RW_BUF_BLOCK_SIZE (1024*1024*10) /* 10 MB per block */

/* 标准的aof文件读写块 */
typedef struct aofrwblock {
	//当前文件块被使用了多少，空闲的大小
 unsigned long used, free;
 //具体存储内容，大小10M
 char buf[AOF_RW_BUF_BLOCK_SIZE];
} aofrwblock;

/* Append data to the AOF rewrite buffer, allocating new blocks if needed. */
/* 在缓冲区中追加数据，如果超出空间，会新申请一个缓冲块 */
void aofRewriteBufferAppend(unsigned char *s, unsigned long len) {
 listNode *ln = listLast(server.aof_rewrite_buf_blocks);
 //定位到缓冲区的最后一块，在最后一块的位置上进行追加写操作
 aofrwblock *block = ln ? ln->value : NULL;

 while(len) {
 /* If we already got at least an allocated block, try appending
 * at least some piece into it. */
 if (block) {
 	//如果当前的缓冲块的剩余空闲能支持len长度的内容时，直接写入
 unsigned long thislen = (block->free < len) ? block->free : len;
 if (thislen) { /* The current block is not already full. */
 memcpy(block->buf+block->used, s, thislen);
 block->used += thislen;
 block->free -= thislen;
 s += thislen;
 len -= thislen;
 }
 }

 if (len) { /* First block to allocate, or need another block. */
 int numblocks;
	//如果不够的话，需要新创建，进行写操作
 block = zmalloc(sizeof(*block));
 block->free = AOF_RW_BUF_BLOCK_SIZE;
 block->used = 0;
 //还要把缓冲块追加到服务端的buffer列表中
 listAddNodeTail(server.aof_rewrite_buf_blocks,block);

 /* Log every time we cross more 10 or 100 blocks, respectively
 * as a notice or warning. */
 numblocks = listLength(server.aof_rewrite_buf_blocks);
 if (((numblocks+1) % 10) == 0) {
 int level = ((numblocks+1) % 100) == 0 ? REDIS_WARNING :
 REDIS_NOTICE;
 redisLog(level,"Background AOF buffer size: %lu MB",
 aofRewriteBufferSize()/(1024*1024));
 }
 }
 }
}

/* Write the append only file buffer on disk.
 *
 * Since we are required to write the AOF before replying to the client,
 * and the only way the client socket can get a write is entering when the
 * the event loop, we accumulate all the AOF writes in a memory
 * buffer and write it on disk using this function just before entering
 * the event loop again.
 *
 * About the 'force' argument:
 *
 * When the fsync policy is set to 'everysec' we may delay the flush if there
 * is still an fsync() going on in the background thread, since for instance
 * on Linux write(2) will be blocked by the background fsync anyway.
 * When this happens we remember that there is some aof buffer to be
 * flushed ASAP, and will try to do that in the serverCron() function.
 *
 * However if force is set to 1 we'll write regardless of the background
 * fsync. */
#define AOF_WRITE_LOG_ERROR_RATE 30 /* Seconds between errors logging. */
/* 刷新缓存区的内容到磁盘中 */
void flushAppendOnlyFile(int force) {
 ssize_t nwritten;
 int sync_in_progress = 0;
 mstime_t latency;

 if (sdslen(server.aof_buf) == 0) return;

 if (server.aof_fsync == AOF_FSYNC_EVERYSEC)
 sync_in_progress = bioPendingJobsOfType(REDIS_BIO_AOF_FSYNC) != 0;

 if (server.aof_fsync == AOF_FSYNC_EVERYSEC && !force) {
 /* With this append fsync policy we do background fsyncing.
 * If the fsync is still in progress we can try to delay
 * the write for a couple of seconds. */
 if (sync_in_progress) {
 if (server.aof_flush_postponed_start == 0) {
 /* No previous write postponinig, remember that we are
 * postponing the flush and return. */
 server.aof_flush_postponed_start = server.unixtime;
 return;
 } else if (server.unixtime - server.aof_flush_postponed_start < 2) {
 /* We were already waiting for fsync to finish, but for less
 * than two seconds this is still ok. Postpone again. */
 return;
 }
 /* Otherwise fall trough, and go write since we can't wait
 * over two seconds. */
 server.aof_delayed_fsync++;
 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.");
 }
 }
 /* We want to perform a single write. This should be guaranteed atomic
 * at least if the filesystem we are writing is a real physical one.
 * While this will save us against the server being killed I don't think
 * there is much to do about the whole server stopping for power problems
 * or alike */

	//在进行写入操作的时候，还监听了延迟
 latencyStartMonitor(latency);
 nwritten = write(server.aof_fd,server.aof_buf,sdslen(server.aof_buf));
 latencyEndMonitor(latency);
 /* We want to capture different events for delayed writes:
 * when the delay happens with a pending fsync, or with a saving child
 * active, and when the above two conditions are missing.
 * We also use an additional event name to save all samples which is
 * useful for graphing / monitoring purposes. */
 if (sync_in_progress) {
 latencyAddSampleIfNeeded("aof-write-pending-fsync",latency);
 } else if (server.aof_child_pid != -1 || server.rdb_child_pid != -1) {
 latencyAddSampleIfNeeded("aof-write-active-child",latency);
 } else {
 latencyAddSampleIfNeeded("aof-write-alone",latency);
 }
 latencyAddSampleIfNeeded("aof-write",latency);

 /* We performed the write so reset the postponed flush sentinel to zero. */
 server.aof_flush_postponed_start = 0;

 if (nwritten != (signed)sdslen(server.aof_buf)) {
 static time_t last_write_error_log = 0;
 int can_log = 0;

 /* Limit logging rate to 1 line per AOF_WRITE_LOG_ERROR_RATE seconds. */
 if ((server.unixtime - last_write_error_log) > AOF_WRITE_LOG_ERROR_RATE) {
 can_log = 1;
 last_write_error_log = server.unixtime;
 }

 /* Lof the AOF write error and record the error code. */
 if (nwritten == -1) {
 if (can_log) {
 redisLog(REDIS_WARNING,"Error writing to the AOF file: %s",
 strerror(errno));
 server.aof_last_write_errno = errno;
 }
 } else {
 if (can_log) {
 redisLog(REDIS_WARNING,"Short write while writing to "
 "the AOF file: (nwritten=%lld, "
 "expected=%lld)",
 (long long)nwritten,
 (long long)sdslen(server.aof_buf));
 }

 if (ftruncate(server.aof_fd, server.aof_current_size) == -1) {
 if (can_log) {
 redisLog(REDIS_WARNING, "Could not remove short write "
 "from the append-only file. Redis may refuse "
 "to load the AOF the next time it starts. "
 "ftruncate: %s", strerror(errno));
 }
 } else {
 /* If the ftrunacate() succeeded we can set nwritten to
 * -1 since there is no longer partial data into the AOF. */
 nwritten = -1;
 }
 server.aof_last_write_errno = ENOSPC;
 }

 /* Handle the AOF write error. */
 if (server.aof_fsync == AOF_FSYNC_ALWAYS) {
 /* We can't recover when the fsync policy is ALWAYS since the
 * reply for the client is already in the output buffers, and we
 * have the contract with the user that on acknowledged write data
 * is synched on disk. */
 redisLog(REDIS_WARNING,"Can't recover from AOF write error when the AOF fsync policy is 'always'. Exiting...");
 exit(1);
 } else {
 /* Recover from failed write leaving data into the buffer. However
 * set an error to stop accepting writes as long as the error
 * condition is not cleared. */
 server.aof_last_write_status = REDIS_ERR;

 /* Trim the sds buffer if there was a partial write, and there
 * was no way to undo it with ftruncate(2). */
 if (nwritten > 0) {
 server.aof_current_size += nwritten;
 sdsrange(server.aof_buf,nwritten,-1);
 }
 return; /* We'll try again on the next call... */
 }
 } else {
 /* Successful write(2). If AOF was in error state, restore the
 * OK state and log the event. */
 if (server.aof_last_write_status == REDIS_ERR) {
 redisLog(REDIS_WARNING,
 "AOF write error looks solved, Redis can write again.");
 server.aof_last_write_status = REDIS_OK;
 }
 }
 server.aof_current_size += nwritten;

 /* Re-use AOF buffer when it is small enough. The maximum comes from the
 * arena size of 4k minus some overhead (but is otherwise arbitrary). */
 if ((sdslen(server.aof_buf)+sdsavail(server.aof_buf)) < 4000) {
 sdsclear(server.aof_buf);
 } else {
 sdsfree(server.aof_buf);
 server.aof_buf = sdsempty();
 }

 /* Don't fsync if no-appendfsync-on-rewrite is set to yes and there are
 * children doing I/O in the background. */
 if (server.aof_no_fsync_on_rewrite &&
 (server.aof_child_pid != -1 || server.rdb_child_pid != -1))
 return;

 /* Perform the fsync if needed. */
 if (server.aof_fsync == AOF_FSYNC_ALWAYS) {
 /* aof_fsync is defined as fdatasync() for Linux in order to avoid
 * flushing metadata. */
 latencyStartMonitor(latency);
 aof_fsync(server.aof_fd); /* Let's try to get this data on the disk */
 latencyEndMonitor(latency);
 latencyAddSampleIfNeeded("aof-fsync-always",latency);
 server.aof_last_fsync = server.unixtime;
 } else if ((server.aof_fsync == AOF_FSYNC_EVERYSEC &&
 server.unixtime > server.aof_last_fsync)) {
 if (!sync_in_progress) aof_background_fsync(server.aof_fd);
 server.aof_last_fsync = server.unixtime;
 }
}

/* Write a sequence of commands able to fully rebuild the dataset into
 * "filename". Used both by REWRITEAOF and BGREWRITEAOF.
 *
 * In order to minimize the number of commands needed in the rewritten
 * log Redis uses variadic commands when possible, such as RPUSH, SADD
 * and ZADD. However at max REDIS_AOF_REWRITE_ITEMS_PER_CMD items per time
 * are inserted using a single command. */
/* 将数据库的内容按照键值，再次完全重写入文件中 */
int rewriteAppendOnlyFile(char *filename) {
 dictIterator *di = NULL;
 dictEntry *de;
 rio aof;
 FILE *fp;
 char tmpfile[256];
 int j;
 long long now = mstime();

 /* Note that we have to use a different temp name here compared to the
 * one used by rewriteAppendOnlyFileBackground() function. */
 snprintf(tmpfile,256,"temp-rewriteaof-%d.aof", (int) getpid());
 fp = fopen(tmpfile,"w");
 if (!fp) {
 redisLog(REDIS_WARNING, "Opening the temp file for AOF rewrite in rewriteAppendOnlyFile(): %s", strerror(errno));
 return REDIS_ERR;
 }

 rioInitWithFile(&aof,fp);
 if (server.aof_rewrite_incremental_fsync)
 rioSetAutoSync(&aof,REDIS_AOF_AUTOSYNC_BYTES);
 for (j = 0; j < server.dbnum; j++) {
 char selectcmd[] = "*2\r\n$6\r\nSELECT\r\n";
 redisDb *db = server.db+j;
 dict *d = db->dict;
 if (dictSize(d) == 0) continue;
 di = dictGetSafeIterator(d);
 if (!di) {
 fclose(fp);
 return REDIS_ERR;
 }

 /* SELECT the new DB */
 if (rioWrite(&aof,selectcmd,sizeof(selectcmd)-1) == 0) goto werr;
 if (rioWriteBulkLongLong(&aof,j) == 0) goto werr;

 /* Iterate this DB writing every entry */
 //遍历数据库中的每条记录，进行日志记录
 while((de = dictNext(di)) != NULL) {
 sds keystr;
 robj key, *o;
 long long expiretime;

 keystr = dictGetKey(de);
 o = dictGetVal(de);
 initStaticStringObject(key,keystr);

 expiretime = getExpire(db,&key);

 /* If this key is already expired skip it */
 if (expiretime != -1 && expiretime < now) continue;

 /* Save the key and associated value */
 if (o->type == REDIS_STRING) {
 /* Emit a SET command */
 char cmd[]="*3\r\n$3\r\nSET\r\n";
 if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
 /* Key and value */
 if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
 if (rioWriteBulkObject(&aof,o) == 0) goto werr;
 } else if (o->type == REDIS_LIST) {
 if (rewriteListObject(&aof,&key,o) == 0) goto werr;
 } else if (o->type == REDIS_SET) {
 if (rewriteSetObject(&aof,&key,o) == 0) goto werr;
 } else if (o->type == REDIS_ZSET) {
 if (rewriteSortedSetObject(&aof,&key,o) == 0) goto werr;
 } else if (o->type == REDIS_HASH) {
 if (rewriteHashObject(&aof,&key,o) == 0) goto werr;
 } else {
 redisPanic("Unknown object type");
 }
 /* Save the expire time */
 if (expiretime != -1) {
 char cmd[]="*3\r\n$9\r\nPEXPIREAT\r\n";
 if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
 if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
 if (rioWriteBulkLongLong(&aof,expiretime) == 0) goto werr;
 }
 }
 dictReleaseIterator(di);
 }

 /* Make sure data will not remain on the OS's output buffers */
 if (fflush(fp) == EOF) goto werr;
 if (fsync(fileno(fp)) == -1) goto werr;
 if (fclose(fp) == EOF) goto werr;

 /* Use RENAME to make sure the DB file is changed atomically only
 * if the generate DB file is ok. */
 if (rename(tmpfile,filename) == -1) {
 redisLog(REDIS_WARNING,"Error moving temp append only file on the final destination: %s", strerror(errno));
 unlink(tmpfile);
 return REDIS_ERR;
 }
 redisLog(REDIS_NOTICE,"SYNC append only file rewrite performed");
 return REDIS_OK;

werr:
 fclose(fp);
 unlink(tmpfile);
 redisLog(REDIS_WARNING,"Write error writing append only file on disk: %s", strerror(errno));
 if (di) dictReleaseIterator(di);
 return REDIS_ERR;
}

/* This is how rewriting of the append only file in background works:
 *
 * 1) The user calls BGREWRITEAOF
 * 2) Redis calls this function, that forks():
 * 2a) the child rewrite the append only file in a temp file.
 * 2b) the parent accumulates differences in server.aof_rewrite_buf.
 * 3) When the child finished '2a' exists.
 * 4) The parent will trap the exit code, if it's OK, will append the
 * data accumulated into server.aof_rewrite_buf into the temp file, and
 * finally will rename(2) the temp file in the actual file name.
 * The the new file is reopened as the new append only file. Profit!
 */
/* 后台进行AOF数据文件写入操作 */
int rewriteAppendOnlyFileBackground(void)

/* aof.c 中的API */
void aofRewriteBufferReset(void) /* 释放server中旧的buffer，并创建一份新的buffer */
unsigned long aofRewriteBufferSize(void) /* 返回当前AOF的buffer的总大小 */
void aofRewriteBufferAppend(unsigned char *s, unsigned long len) /* 在缓冲区中追加数据，如果超出空间，会新申请一个缓冲块 */
ssize_t aofRewriteBufferWrite(int fd) /* 将保存内存中的buffer内容写入到文件中，也是分块分块的写入 */
void aof_background_fsync(int fd) /* 开启后台线程进行文件同步操作 */
void stopAppendOnly(void) /* 停止追加数据操作,这里用的是一个命令模式 */
int startAppendOnly(void) /* 开启追加模式 */
void flushAppendOnlyFile(int force) /* 刷新缓存区的内容到磁盘中 */
sds catAppendOnlyGenericCommand(sds dst, int argc, robj **argv) /* 根据输入的字符串，进行参数包装，再次