redis源码分析

转自： http://www.hoterran.info/

 

REDIS源代码分析 – HASH TABLE

By hoterran on 2011/06/21, 15:58

hashtable的实现有很多，redis的dict.c 是其中之一。

dict 包含了2个dictht hashtable ht[0], ht[1]。
client版本的dict是没有dictht的概念。加入dictht的概念存在2个ht的目的是为了在rehash的时候可以平滑的迁移bucket里的数据，而不像client的dict要把老的hash table里的一次性的全部数据迁移到新的hash table，这在造成一个密集型的操作，在业务高峰期不可取。

ht是hashtable的简称，实际上是一个指针数组，数组的个数由dictht->size决定，是DICT_HT_INITIAL_SIZE的整数倍。每个元素（bucket）指向一个dictEntry的单链表来解决hash的conflict。查询某个key，需要先hash，定位到bucket,再通过链表遍历。

key经过hash函数后，与dictht->sizemask求与均分到ht的每个bucket上。dictht->used表示这个ht里包含的key的个数，也就是dictEntry的个数，每次dictAdd成功+1。链表的加入为头指针的方法加入，这样dictAdd更加的方便。

随着key不断的添加，bucket下的单链表越来越长，查找、删除效率越来越低，需要对ht进行expand，增加bucket个数，让链表的长度减少。bucket数量的增多，原有bucket的key需要迁移到新的bucket上，于是有了rehash的这个过程。

ht[1]就是为了rehash而产生，新的ht size是ht[0]的两倍2， 随着dictAdd，dictFind函数的调用，ht[0]的每个bucket会rehash加入到ht[1]里。dict->rehashidx 是ht[0] 需要rehash就是迁移到ht[1]的bucket的索引，从0开始直到ht->used==0。
rehash除了每次伴随dictAdd，dcitFind而迁移一个bucket的所有dictEntry，还有一种一次hash100个bucket，直到消耗了某个时间点为止的做法。

rehash的步骤：
拿到一个bucket， 遍历这个链表的每个kv，对key进行hash然后于sizemask求与，定位ht[1]的新bucket位置，
加入到链表，迁移后ht[0].used–，ht[1].used++。
直到ht[0].used=0，释放ht[0]的table，再赋值ht[0]= ht[1]，再把则ht[1]reset。

rehash的期间：
由于ht[1]是ht[0]size的2倍，每次dictAdd的时候都会rehash一次，不会出现后ht[1] 满了，而ht[0]还有数据的事情。
查询会先查ht[0]再查询ht[1]，在rehash的过程中，不会出现再次expand。
新的key加到ht[1]。

expand的条件：
table的位置已经满了，糟糕的hash函数造成的skrew导致永远不会expand。
key的个数除以table的大小，超过了dict_force_resize_ratio。

 

REDIS源代码分析 – EVENT LIBRARY

By hoterran on 2011/06/22, 21:51

每个cs程序尤其是高并发的网络服务端程序都有自己的网络异步事件处理库，redis不例外。

事件库仅仅包括ae.c、ae.h，还有3个不同的多路复用（本文仅描述epoll）的wrapper文件，事件库封装了框架调用的主循环函数，暴露了时间、文件事件注册和销毁函数，典型的依赖反转模式。
网络操作都在networking.c里，封装了常见的socket操作。

我们从redis启动的main函数开始，从用户发出连接键入命令开始遍历网络事件库所涉及的函数，unix套接口相关函数不表。

首先对几个最常用对象进行解释。

//redis启动的时候（init_server()）会创建的一个全局使用的事件循环结构
typedef struct aeEventLoop {
int maxfd;		//仅仅是select 使用
long long timeEventNextId;
aeFileEvent events[AE_SETSIZE]; //用于保存epoll需要关注的文件事件的fd、触发条件、注册函数。
aeFiredEvent fired[AE_SETSIZE]; //epoll_wait之后获得可读或者可写的fd数组，通过aeFiredEvent->fd再定位到events。
aeTimeEvent *timeEventHead;	//以链表形式保存多个时间事件，每隔一段时间机会触发注册的函数。
int stop;
void *apidata;			//每种多路复用方法的使用的私有变量，例如epoll就是epfd和一个事件数组;而select是保存rset、wset。
aeBeforeSleepProc *beforesleep;	// sleep之前调用的函数，有些事情是每次循环必须做的，并非文件、时间事件。
} aeEventLoop;

//文件可读写事件
typedef struct aeFileEvent {
int mask; 			//触发条件：读、写
aeFileProc *rfileProc;		//当fd可读时执行的事件（accept，read）
aeFileProc *wfileProc;		//当fd可写时执行的事件（write）
void *clientData;               //caller 传入的数据指针
} aeFileEvent;

//超时时间事件
typedef struct aeTimeEvent {
long long id; /* time event identifier. */
long when_sec; /* seconds */
long when_ms; /* milliseconds */
aeTimeProc *timeProc;		//当出现超时的时候所执行的事件
aeEventFinalizerProc *finalizerProc;
void *clientData;                //caller传入的数据指针
struct aeTimeEvent *next;	//单链表指向下一个时间事件
} aeTimeEvent;

//从epoll_wait或者select返回的，已经触发的文件事件
typedef struct aeFiredEvent {
int fd;
int mask;
} aeFiredEvent;

我们来模拟redis server 启动和用户键入命令的过程，先上图。

好吧，从main函数开始吧。

// src/redis.c 853
server.el = aeCreateEventLoop();

创建一个aeEventLoop结构体，成员apidata指向一个aeApiState对象，如果使用epoll，fd是由epoll_create创建的全局epoll fd ，events[] 用于保存epoll_wait返回的事件组。

// src/redis.c 866
server.ipfd = anetTcpServer(server.neterr,server.port,server.bindaddr);

创建监听。

//   src/redis.c 903
aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL);

注册一个时间事件serverCron，作用以后再讲。

 //   src/redis.c 906
aeCreateFileEvent(server.el,server.ipfd,AE_READABLE,  acceptTcpHandler,NULL)

为监听fd的注册一个文件事件，首先把listenfd和触发条件epoll_ctl加入到全局的epoll fd进行监控。再以fd作为文件事件event数组index定位，mask填入只读，rfileProc填入acceptTcpHandler函数。

// src/redis.c 1571
aeSetBeforeSleepProc(server.el,beforeSleep);

aeEventLoop注册一个beforeSleep函数，这个函数在主循环里每次会被调用，作用以后再讲。

// src/ redis.c 1572
aeMain()

网络事件库的核心循环函数。

// src/ae.c  379
aeCreateLoop->beforesleep()

就是上面注册的beforeSleep函数。

// src/ae.c 275
aeProcessEvents(eventLoop, AE_ALL_EVENTS);

先调度aeApiPoll，用epoll_wait处理文件事件，返回的fd和触发条件先存储在eventLoop->fired[]里，然后根据fired[]每个事件的的fd，定位到events，根据触发条件调用已经注册的事件。

文件事件处理完毕后，接下来就是处理超时时间事件，这里不表。

假如有个用户连接上redis，则从redis servere就从上面的aeApiPoll的poll_wait返回，产生的只读事件会调度上面注册了acceptTcpHandler函数。

// src/networking.c 390
acceptTcpHandler(eventLoop, fd, NULL, AE_READABLE)

accept返回产生了一个新连接的fd（这里省略了很多步骤），需要从新的连接读取数据，于是为这个fd注册可读事件。

// src/networking.c 20
aeCreateFileEvent(server.el,fd,AE_READABLE,  readQueryFromClient, c)

于是调用了aeCreateFileEvent 为只读事件注册, 这里的步骤和上面的listen fd 一样。用epoll_ctl将fd加入到epoll fd里等待下次epoll_wait，再注册触发条件和readQueyFromClient函数，接着aeMain()继续执行等待用户的数据。

假如用户在这个连接键入redis命令例如：set foo bar，redis server端将会从aeApiPoll的epoll_wait返回，和accept一样的处理方式。返回的fd填入到fired[]数组，通过fired[fd]->fd找到是哪个文件事件，找到events[fd]->rfielProc这个函数，就是上面注册的readQueyFromClient 函数。

//  src/networking.c 816
readQueyFromClient(server.el, fd, NULL, AE_READABLE)

这个函数首先会执行read从tcp buffer读取用户键入的命令（非阻塞io），然后处理buffer，找到对应的command（lookupCommand），接下来执行这个command（call(c,cmd)），命令执行完毕需要返回结果的时候，会再次注册一个文件事件

// src/networking.c  71
aeCreateFileEvent(server.el, c->fd, AE_WRITABLE, sendReplyToClient, c)

这样下次循环epoll_wait的时候就发现这个fd可写，于是就会执行sendReplyToClient，讲结果发送给client。

redis的这个网络事件库是比较标准的网络框架的模式，实现的功能不算多但够用。

 

REDIS源代码分析- REPLICATION

By hoterran on 2011/06/30, 23:00

redis的复制方法和机制都比较简单。

slaveof masterip port

在slave端键入命令之后，就开启了从master到slave的复制。一个master可以有多个slave，master有变化的时候会主动的把命令传播给每个slave。slave同时可以作为其他的slave的master，前提条件是这个slave已经处于稳定状态（REDIS_REPL_CONNECTED）。slave在复制的开始阶段处于阻塞状态（sync_readline）无法对外提供服务。

数据的有向图会让redis的运维很有搞头。

slaveof no one

从slave状态的转换回master状态，切断与原master的数据同步。

下面根据图形来描述一个复制全过程。

复制的顺序是从左到右。绿色的过程表示replstate复制状态的变迁，复制相关的函数主要在src/replication.c里，红色为master的复制函数，蓝色的为slave使用的复制函数。

slave端接收到客户端的slaveof masterip ort命令之后，根据readonlyCommandTable找到对应的函数为slaveofCommand。slaveofCommand会保存masterip，masterport，并把server.replstate改成REDIS_REPL_CONNECT，然后返回给客户端OK。

slave端主线程在已经注册时间事件serverConn（redis.c 518）里执行replicationCron（redis.c 646）来开始与master的连接。调用syncWithMaster()开始与master的通信。经过校验之后，会发送一个SYNC command给master端，然后打开一个临时文件用于存储接下来master发过来的rdb文件数据。再添加一个文件事件注册readSyncBulkPayload函数，这个就是接收rdb文件的数据的函数，然后修改状态为REDIS_REPL_TRANSFER。

master接收到SYNC command后，跳转到syncCommand函数（replication.c 556）。syncCommand会调度rdbSaveBackground函数，启动一个子进程做一个全库的持久化rdb文件，并把状态改为REDIS_REPL_WAIT_BGSAVE_END。

master的主线程的serverCron会等待做持久化的子进程的退出，并判断退出的状态是否是正常。

if ((pid = wait3(&statloc,WNOHANG,NULL)) != 0) {
if (pid == server.bgsavechildpid) {
backgroundSaveDoneHandler(statloc);
} else {
....

如果子进程正常退出，会转到backgroundSaveDoneHandler函数。backgroundSaveDoneHandler 处理每次rdbSaveBackground成功后的收尾工作，并打开刚刚产生的rdb文件。然后注册一个sendBulkToSlave函数用于发送rdb文件，状态切换至REDIS_REPL_SEND_BULK。

sendBulkToSlave作用就是根据上面打开的rdb文件，读取并发送到slave端，当文件全部发送完毕之后修改状态为REDIS_REPL_ONLINE。

我们回到slave，上面讲到slave通过readSyncBulkPayload接收rdb数据，接收完整个rdb文件后，会清空整个数据库emptyDb()(replication.c 374)。然后就通过rdbLoad函数载入接收到的rdb文件，于是slave和master数据就一致了，再把状态修改为REDIS_REPL_CONNECTED。

接下来就是master和slave之间增量的传递的增量数据，另外slave和master在应用层有心跳检测（replication.c 543），和超时退出（replication.c 511）。

最后再给出一个replstate状态图

slave端复制状态

REDIS_REPL_NONE：未复制的状态

REDIS_REPL_CONNECT：已经接收到slaveof命令，但未发出sync命令给master

REDIS_REPL_TRANSFER：已经发出sync，但还没接收完rdb文件

REDIS_REPL_CONNECTED：已经接收完rdb文件，开始增量接收复制内容

maste端redis_client的复制状态

REDIS_REPL_WAIT_BGSAVE_START：bgsave被抢占，等待bgsave为其持久化进行工作。

REDIS_REPL_WAIT_BGSAVE_END：等待bgsave持久化rdb文件。

REDIS_REPL_SEND_BULK：rdb文件已经生成。开始拷贝给slave。

REDIS_REPL_ONLINE：rdb拷贝完毕，开始增量复制。

另外在syncCommand函数里有几种可能性

bgsave进程未启动，则启动。

bsave已启动，且已经存在另外的一个slave连接，已经由这个连接开启了bgsave，则等待bgsave完毕，共享这次的持久化。

bsave已启动，但不是由slave连接所启动的，则等待下次bgsave的退出。在updateSlavesWaitingBgsave函数里再次启动bgsave进程。

update:

 

 

 

REDIS源代码分析 – PERSISTENCE

By hoterran on 2011/07/03, 02:15

redis有全量（save/bgsave）和增量（aof）的持久化命令。

全量的原理就是遍历里所有的DB，在每个bucket，读取链表的key和value并写入dump.rdb文件（rdb.c 405）。

save命令直接调度rdbSave函数，这会阻塞主线程的工作，通常我们使用bgsave。

bgsave命令调度rdbSaveBackground函数启动了一个子进程然后调度了rdbSave函数，子进程的退出状态由 serverCron的backgroundSaveDoneHandler来判断，这个在复制这篇文章里讲到过，这里就不罗嗦了。

除了直接的save、bgsave命令之外，还有几个地方还调用到rdbSaveBackground和rdbSave函数。

shutdown：redis关闭时需要对数据持久化，保证重启后数据一致性，会调用rdbSave()。

flushallCommand：清空redis数据后，如果不做立即执行一个rdbSave()，出现crash后，可能会载入含有老数据的dump.rdb。

void flushallCommand(redisClient *c) {
  touchWatchedKeysOnFlush(-1);
  server.dirty += emptyDb();      // 清空数据
  addReply(c,shared.ok);
  if (server.bgsavechildpid != -1) {
    kill(server.bgsavechildpid,SIGKILL);
    rdbRemoveTempFile(server.bgsavechildpid);
  }
  rdbSave(server.dbfilename);    //没有数据的dump.db
  server.dirty++;
}

sync：当master接收到slave发来的该命令的时候，会执行rdbSaveBackground，这个以前也有提过。

数据发生变化：在多少秒内出现了多少次变化则触发一次bgsave，这个可以在conf里配置

for (j = 0; j < server.saveparamslen; j++) {
  struct saveparam *sp = server.saveparams+j;
   if (server.dirty >= sp->changes && now-server.lastsave > sp->seconds) {
     rdbSaveBackground(server.dbfilename);
     break;
  }
}

增量备份就是aof，原理有点类似redo log。每次执行命令后如出现数据变化，会调用feedAppendOnlyFile，把数据写到server.aofbuf里。

void call(redisClient *c, struct redisCommand *cmd) {
  long long dirty;
  dirty = server.dirty;
  cmd->proc(c);        //执行命令
  dirty = server.dirty-dirty;
  if (server.appendonly && dirty)
    feedAppendOnlyFile(cmd,c->db->id,c->argv,c->argc);

待到下次循环的before_sleep函数会通过flushAppendOnlyFile函数把server.aofbuf里的数据write到append file里。
可以在redis.conf里配置每次write到append file后从page cache刷新到disk的规律。

# appendfsync always
appendfsync everysec
# appendfsync no

参数的原理MySQL的innodb_flush_log_at_trx_commit一样，是个比较影响io的一个参数，需要在高性能和不丢数据之间做tradeoff。软件的优化就是tradeoff的过程，没有银弹。

一个疑问先写到server.aofbuf，然后再写到数据文件，过程中如果crash会不会丢数据呢？

答案是不会，为何？我们来看函数执行的步骤：

call()
feedAppendOnlyFile()
下一次循环
beforeSleep()-->flushAppendOnlyFile()
aeMain()--->sendReplyToClient()

只有执行完了flush之后才会通知客户端数据写成功了，所以如果在feed和flush之间crash，客户接会因为进程退出接受到一个fin包，也就是一个错误的返回，所以数据并没有丢，只是执行出了错。

redis crash后，重启除了利用rdb重新载入数据外，还会读取append file(redis.c 1561)加载镜像之后的数据。

激活aof，可以在redis.conf配置文件里设置

appendonly yes

也可以通过config命令在运行态启动aof

cofig set appendonly yes

aof最大的问题就是随着时间append file会变的很大，所以我们需要bgrewriteaof命令重新整理文件，只保留最新的kv数据。

下面这个根据图形来描述一下aof的全过程，绿色的为aof.c里的函数，顺序从左到右。

启动appendly，或者config set appendonly yes 都会触发函数rewriteAppendOnlyFileBackground，bgrewriteaof也调用该函数。

实际最后调用的函数是rewriteAppendOnlyFile,这个函数与rdbSave和类似。保存全库的kv数据。

在子进程做快照的过程中，kv的变化是先写到aofbuf里。如果存在bgrewritechildpid进程，变化数据还要写到server.bgrewritebuf里（aof.c 177）。
等子进程完成快照退出之时，由backgroundRewriteDoneHandler函数再把bgrwritebuf和全镜像两部分数据进行合并（aof.c 673）。

合并后的aof文件才是最新的全库的镜像数据。

REDIS源代码分析 – PROTOCOL

By hoterran on 2011/08/01, 00:18

我们跟踪一个普通的get命令来遍历几个关键函数，熟悉协议处理的过程。

你可以通过telnet或者redis_cli、利用lib库发送请求给redis server。前者的是一种裸协议的请求发送到服务端，而后两者会对键入的请求进行协议组装帮助更好的解析（常见的是长度放到前头，还有添加阿协议类型）。

Requests格式

*参数的个数 CRLF
$第一个参数的长度CRLF
第一个参数CRLF
...
$第N个参数的长度CRLF
第N个参数CRLF

例如在redis_cli里键入get a，经过协议组装后的请求为

2\r\n$3\r\nget\r\n$1\r\na\r\n

下面这个图涵盖了接收request，处理请求，调用函数，发送reply的过程。

redis的网络事件库，我们在前面的文章已经讲过，readQueryFromClient先从fd中读取数据，先存储在c->querybuf里(networking.c 823)。接下来函数processInputBuffer来解析querybuf，上面说过如果是telnet发送的裸协议数据是没有*打头的表示参数个数的辅助信息，针对telnet的数据跳到processInlineBuffer函数，而其他则通过函数processMultibulkBuffer。
这两个函数的作用一样，解析c->querybuf的字符串，分解成多参数到c->argc和c->argv里面，argc表示参数的个数，argv是个redis_object的指针数组，每个指针指向一个redis_object, object的ptr里存储具体的内容，对于”get a“的请求转化后，argc就是2，argv就是

(gdb) p (char*)(*c->argv[0])->ptr
$28 = 0x80ea5ec "get"
(gdb) p (char*)(*c->argv[1])->ptr
$26 = 0x80e9fc4 "a"

协议解析后就执行命令。processCommand首先调用lookupCommand找到get对应的函数。在redis server 启动的时候会调用populateCommandTable函数（redis.c 830）把readonlyCommandTable数组转化成一个hash table（server.commands），lookupCommand就是一个简单的hash取值过程，通过key（get）找到相应的命令函数指针getCommand（ t_string.c 437）。
getCommand比较简单，通过另一个全局的server.db这个hash table来查找key，并返回redis object，然后通过addReplyBulk函数返回结果。

下面介绍一下reply协议。
bulk replies是以$打头消息体，格式$值长度\r\n值\r\n，一般的get命令返回的结果就是这种个格式。

redis>get aaa
$3\r\nbbb\r\n

对应的的处理函数addReplyBulk

addReplyBulkLen(c,obj);
addReply(c,obj);
addReply(c,shared.crlf);

error messag是以-ERR 打头的消息体，后面跟着出错的信息，以\r\n结尾，针对命令出错。

redis>d
-ERR unknown command 'd'\r\n

处理的函数是addReplyError

addReplyString(c,"-ERR ",5);
addReplyString(c,s,len);
addReplyString(c,"\r\n",2);

integer reply 是以:打头，后面跟着数字和\r\n。

redis>incr a
:2\r\n

处理函数是

addReply(c,shared.colon);
addReply(c,o);
addReply(c,shared.crlf);

status reply以+打头，后面直接跟状态内容和\r\n

redis>ping
+PONG\r\n

这里要注意reply经过协议加工后，都会先保存在c->buf里，c->bufpos表示buf的长度。待到事件分离器转到写出操作（sendReplyToClient）的时候，就把c->buf的内容写入到fd里，c->sentlen表示写出长度。当c->sentlen=c->bufpos才算写完。

还有一种Multi-bulk replies，lrange、hgetall这类函数通常需要返回多个值，消息结构与请求的格式一模一样。相关的函数是setDeferredMultiBulkLength。临时数据存储在链表c->reply里，处理方式同其他的协议格式。