Redis源码架构阅读

// 5.0.8 伪代码
int main(int argc,char **argv){
    ...
    initServer();
    ...
    aeCreateFileEvent(fd,acceptxxxHandler,...);
    ...
    aeMain(server.el);
    ...
    return 0;
}

/**
* 1. 创建数据结构，初始化数据结构
* 2. 打开TCP监听端口与Unix本地端口
* 4. 为socket关联应答处理器
* 3. 初始化AOF，集群，脚本系统，BIO等
**/
void initServer() {
...
    /* Open the TCP listening socket for the user commands. */
    // 打开 TCP 监听端口，用于等待客户端的命令请求
    if (server.port != 0 &&
        listenToPort(server.port,server.ipfd,&server.ipfd_count) == REDIS_ERR)
        exit(1);
...

    /* Create the serverCron() time event, that's our main way to process
     * background operations. */
    // 为 serverCron() 创建时间事件,后台线程操作
    if(aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL) == AE_ERR) {
        redisPanic("Can't create the serverCron time event.");
        exit(1);
    }

    /* Create an event handler for accepting new connections in TCP and Unix
     * domain sockets. */
    // 为 TCP 连接关联连接应答（accept）处理器
    // 用于接受并应答客户端的 connect() 调用
    for (j = 0; j < server.ipfd_count; j++) {
        if (aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE,
            acceptTcpHandler,NULL) == AE_ERR)
            {
                redisPanic(
                    "Unrecoverable error creating server.ipfd file event.");
            }
    }

    // 为本地套接字关联应答处理器
    if (server.sofd > 0 && aeCreateFileEvent(server.el,server.sofd,AE_READABLE,
        acceptUnixHandler,NULL) == AE_ERR) redisPanic("Unrecoverable error creating server.sofd file event.");

    /* 32 bit instances are limited to 4GB of address space, so if there is
     * no explicit limit in the user provided configuration we set a limit
     * at 3 GB using maxmemory with 'noeviction' policy'. This avoids
     * useless crashes of the Redis instance for out of memory. */
    // 对于 32 位实例来说，默认将最大可用内存限制在 3 GB
    if (server.arch_bits == 32 && server.maxmemory == 0) {
        redisLog(REDIS_WARNING,"Warning: 32 bit instance detected but no memory limit set. Setting 3 GB maxmemory limit with 'noeviction' policy now.");
        server.maxmemory = 3072LL*(1024*1024); /* 3 GB */
        server.maxmemory_policy = REDIS_MAXMEMORY_NO_EVICTION;
    }
}

/*
 * 根据 mask 参数(比如AE_READABLE)的值，监听 fd 文件的状态，
 * 绑定fd到传入的acceptHandler或writeHandler.etc（即*proc这个函数指针）
 */
int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
        aeFileProc *proc, void *clientData)
{
    if (fd >= eventLoop->setsize) {
        errno = ERANGE;
        return AE_ERR;
    }

    if (fd >= eventLoop->setsize) return AE_ERR;

    // 取出文件事件结构
    aeFileEvent *fe = &eventLoop->events[fd];

    // 监听指定 fd 的指定事件
    // 把fd加入到链表中
    if (aeApiAddEvent(eventLoop, fd, mask) == -1)
        return AE_ERR;

    // 设置文件事件类型，以及事件的处理器
    // 当 fd 可用时，执行 proc 函数
        // 比如绑定fd到传入的acceptHandler
    fe->mask |= mask; // 或运算后再进行可读或可写判断
    if (mask & AE_READABLE) fe->rfileProc = proc;
    if (mask & AE_WRITABLE) fe->wfileProc = proc;

    // 私有数据
    fe->clientData = clientData;

    // 如果有需要，更新事件处理器的最大 fd
    if (fd > eventLoop->maxfd)
        eventLoop->maxfd = fd;

    return AE_OK;
}

/*
 * 事件处理器的主循环
 */
void aeMain(aeEventLoop *eventLoop) {

    eventLoop->stop = 0;
    // 这样写死循环，给后面关闭这个死循环留下了口子，比如测试以及性能评测时
    while (!eventLoop->stop) {

        // 如果有需要在事件处理前执行的函数(beforesleep)，那么运行它
        if (eventLoop->beforesleep != NULL)
            eventLoop->beforesleep(eventLoop);

        // 开始处理事件
        // 多路复用：将所有fd作为select的入参以求内核返回可读或可写的fd再处理
        aeProcessEvents(eventLoop, AE_ALL_EVENTS);
    }
}

/**
* 将fd以aeFileEvent的形式加入aeEventLoop这个大链表当中
**/
static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
    aeApiState *state = eventLoop->apidata;

    if (mask & AE_READABLE) FD_SET(fd,&state->rfds);
    if (mask & AE_WRITABLE) FD_SET(fd,&state->wfds);
    return 0;
}

/* aeCreateFileEvent() -> File event structure */
typedef struct aeFileEvent {
    int mask; /* one of AE_(READABLE|WRITABLE|BARRIER) */
    aeFileProc *rfileProc;
    aeFileProc *wfileProc;
    void *clientData;
} aeFileEvent;

/* State of an event based program */
typedef struct aeEventLoop {
    int maxfd;   /* highest file descriptor currently registered */
    int setsize; /* max number of file descriptors tracked */
    aeFileEvent *events; /* Registered events */
    int stop;
    aeBeforeSleepProc *beforesleep;
    aeBeforeSleepProc *aftersleep;
} aeEventLoop;

/*
 * 创建一个 TCP 连接处理器
 */
void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
...
//     控制最大TCP连接数
    while(max--) {
        // accept 客户端连接--> 由内核自动生成的一个全新的fd，代表与返回client的TCP连接
        cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip), &cport);
        if (cfd == ANET_ERR) {
            if (errno != EWOULDBLOCK)
                serverLog(LL_WARNING,
                    "Accepting client connection: %s", server.neterr);
            return;
        }
        // 为客户端创建客户端状态（redisClient）！
        acceptCommonHandler(cfd,0,cip);
    }
}

/**
* 为每一个链接都创建一个专属的客户端
**/
static void acceptCommonHandler(int fd, int flags) {
    // 创建客户端
    redisClient *c;
    //     createClient比如aeCreateFileEvent(config.el,c->context->fd,AE_WRITABLE,writeHandler,c);
    if ((c = createClient(fd)) == NULL) {
        redisLog(REDIS_WARNING,
            "Error registering fd event for the new client: %s (fd=%d)",
            strerror(errno),fd);
        close(fd); /* May be already closed, just ignore errors */
        return;
    }
    /* If maxclient directive is set and this is one client more... close the
     * connection. Note that we create the client instead to check before
     * for this condition, since now the socket is already set in non-blocking
     * mode and we can send an error for free using the Kernel I/O */
    // 如果新添加的客户端令服务器的最大客户端数量达到了
    // 那么向新客户端写入错误信息，并关闭新客户端
    // 先创建客户端，再进行数量检查是为了方便地进行错误信息写入
    if (listLength(server.clients) > server.maxclients) {
        char *err = "-ERR max number of clients reached\r\n";

        /* That's a best effort error message, don't check write errors */
        if (write(c->fd,err,strlen(err)) == -1) {
            /* Nothing to do, Just to avoid the warning... */
        }
        // 更新拒绝连接数
        server.stat_rejected_conn++;
        freeClient(c);
        return;
    }
}

/**
* 将fd放入链表，同时绑定readQueryFromClient
* （而不是处理新客户端连接的acceptHandler）
**/
client *createClient(int fd) {
    client *c = zmalloc(sizeof(client));

    /* passing -1 as fd it is possible to create a non connected client（like a lua acript）.
     * This is useful since all the commands needs to be executed
     * in the context of a client. When commands are executed in other
     * contexts (for instance a Lua script) we need a non connected client. */
    if (fd != -1) {
        anetNonBlock(NULL,fd);
        anetEnableTcpNoDelay(NULL,fd);
        if (server.tcpkeepalive)
            anetKeepAlive(NULL,fd,server.tcpkeepalive);
            //复用aeCreateFileEvent绑定处理客户端redis命令的函数readQueryFromClient
        if (aeCreateFileEvent(server.el,fd,AE_READABLE,
            readQueryFromClient, c) == AE_ERR)
        {
            close(fd);
            zfree(c);
            return NULL;
        }
    }
    ...
}

// 结构体定义
typedef char *sds;
struct sdshdr {

    // buf 已占用长度
    int len;

    // buf 剩余可用长度
    int free;

    // 实际保存字符串数据的地方
    // 利用c99(C99 specification 6.7.2.1.16)中引入的 flexible array member
    // sizeof的结果不包含结构体空间，作为符号地址存在
    // 通过buf来引用sdshdr后面的地址，
    char buf[];
};


/*
 * 新建字符串对象
 *
 * T = O(N)
 */
sds sdsnewlen(const void *init, size_t initlen) {

    struct sdshdr *sh;

    // 有初始值:sdshdr长度+字符串长度+一个结束符'\0'
    if (init) {
    // zmallloc O(N)
        sh = zmalloc(sizeof(struct sdshdr)+initlen+1);
    } else {
        sh = zcalloc(sizeof(struct sdshdr)+initlen+1);
    }

    // 内存不足，分配失败
    if (sh == NULL) return NULL;

    sh->len = initlen; // 不包括结构体的长度
    sh->free = 0;

    // 如果给定了 init 且 initlen 不为 0 的话
    // 那么将 init 的内容复制至 sds buf
    // memcpy() O(N)
    if (initlen && init)
        memcpy(sh->buf, init, initlen);

    // 加上终结符
    sh->buf[initlen] = '\0';

    // ！！返回 buf 而不是整个 sdshdr
    return (char*)sh->buf;
}

/*
* 得到字符串长度（而不是结构体的长度）
* T=O（1）
*/
static inline size_t sdslen(const sds s) {
    // 如前所说，sizeof不返回柔性数组大小，因此指针运算减去两个int长度后就得到了sds的地址指针sh
    struct sdshdr *sh = (void*)(s-(sizeof(struct sdshdr)));
    return sh->len;
}

/*
 * 字符串复制进入原字符串操作
 * 将一个 char 数组的前 len 个字节复制至 sds
 * 如果 sds 的 buf 不足以容纳要复制的内容，
 * 那么扩展 buf 的长度，让 buf 的长度大于等于 len 。
 *
 * T = O(N)
 */
sds sdscpylen(sds s, const char *t, size_t len) {

    struct sdshdr *sh = (void*) (s-(sizeof(struct sdshdr)));

    // 是否需要扩展 buf ？
    size_t totlen = sh->free+sh->len;
    if (totlen < len) {
        // 扩展 buf 长度，让它的长度大于等于 len
        // 具体的大小请参考 sdsMakeRoomFor 的注释
        // T = O(N)
        s = sdsMakeRoomFor(s,len-sh->len);
        if (s == NULL) return NULL;
        sh = (void*) (s-(sizeof(struct sdshdr)));
        totlen = sh->free+sh->len;
    }

    // O(N)
    memcpy(s, t, len);
    s[len] = '\0';

    sh->len = len;
    sh->free = totlen-len;

    return s;
}

/*
 * 字符串追加操作
 * 按长度 len 扩展 sds ，并将 t 拼接到 sds 的末尾
 *
 * T = O(N)
 */
sds sdscatlen(sds s, const void *t, size_t len) {
    struct sdshdr *sh;
    size_t curlen = sdslen(s);

    // O(N)
    s = sdsMakeRoomFor(s,len);
    if (s == NULL) return NULL;

    // 复制
    // O(N)
    memcpy(s+curlen, t, len);

    // 更新 len 和 free 属性
    // O(1)
    sh = (void*) (s-(sizeof(struct sdshdr)));
    sh->len = curlen+len;
    sh->free = sh->free-len;

    // 终结符
    // O(1)
    s[curlen+len] = '\0';

    return s;
}

/* 
 * 对 sds 的 buf 进行扩展，扩展的长度不少于 addlen 。
 *
 * T = O(N)
 */
sds sdsMakeRoomFor(
    sds s,
    size_t addlen   // 需要增加的空间长度
) 
{
    struct sdshdr *sh, *newsh;
    size_t free = sdsavail(s);
    size_t len, newlen;

    // 剩余空间可以满足需求，无须扩展
    if (free >= addlen) return s;

    sh = (void*) (s-(sizeof(struct sdshdr)));

    // 目前 buf 长度
    len = sdslen(s);
    // 新 buf 长度
    newlen = (len+addlen);
    // 如果新 buf 长度小于 SDS_MAX_PREALLOC 长度
    // 那么将 buf 的长度设为新 buf 长度的两倍
    if (newlen < SDS_MAX_PREALLOC)
        newlen *= 2;
    else
        newlen += SDS_MAX_PREALLOC;

    // 扩展长度
    newsh = zrealloc(sh, sizeof(struct sdshdr)+newlen+1);

    if (newsh == NULL) return NULL;

    newsh->free = newlen - len;

    return newsh->buf;
}

//定义dict字典结构体
typedef struct dict {

    // 对应不同类型，操作键值对的函数不同
    dictType *type;

    // 类型处理函数的私有数据-->提供给dicType的参数不同
    void *privdata;

    // 哈希表（2个）,用于rehash
    dictht ht[2];

    // 记录 rehash 进度的标志，值为-1 表示 rehash 未进行
    int rehashidx;

    // 当前正在运作的安全迭代器数量
    int iterators;      

} dict;


// 定义哈希表
typedef struct dictht {

    // 哈希表节点指针数组（俗称桶，bucket）
    // 数组中的元素都是一个指向dictEntry结构的指针,*table[]
    dictEntry **table;      

    // 指针数组的大小
    unsigned long size;     

    // 指针数组的长度掩码，用于计算索引值
    unsigned long sizemask; 

    // 哈希表现有的节点数量
    unsigned long used;     
}dictht;

// 定义哈希桶
typedef struct dictEntry {

    // 键
    void *key;

    // 值，保存类型包括指针或无符号64位整数或64位整数
    union {
        void *val;
        uint64_t u64;
        int64_t s64;
    } v;

    // 链往后继节点-->开链法解决哈希冲突，头插入
    struct dictEntry *next; 

} dictEntry;

// 定义key与value所指向的对象
typedef struct redisObject {

    // 类型
    unsigned type:4;

    // 编码
    unsigned encoding:4;

    // 对象最后一次被访问的时间
    unsigned lru:REDIS_LRU_BITS; /* lru time (relative to server.lruclock) */

    // 引用计数
    int refcount;

    // 指向底层数据结构的指针：sds/hash/ziplist/quicklist/skiplist/intset
    void *ptr;

} robj;

A fingerprintis a 64 bit number that represents the state of the dictionary
 * at a given time, it's just a few dictproperties xored together.
 * When an unsafe iterator is initialized, weget the dict fingerprint, and check
 * the fingerprint again when the iterator isreleased.
 * If the two fingerprints are different itmeans that the user of the iterator
 * performed forbidden operations against thedictionary while iterating. */
long longdictFingerprint(dict *d) {
    long long integers[6], hash = 0;
    int j;
    //将dict结构体中的几个状态放入到数组中，以便后面应用到64 bit MixFunctions中。
    //dict结构体其实就是一个hash表的实现，而这些状态其实就是第一、第二哈希表的表地址、表大小与
    //已用条目的数量
    integers[0] = (long) d->ht[0].table;
    integers[1] = d->ht[0].size;
    integers[2] = d->ht[0].used;
    integers[3] = (long) d->ht[1].table;
    integers[4] = d->ht[1].size;
    integers[5] = d->ht[1].used;

    /* We hash N integers by summing everysuccessive integer with the integer
     * hashing of the previous sum. Basically:
     *
     * Result =hash(hash(hash(int1)+int2)+int3) ...
     *
     * This way the same set of integers in adifferent order will (likely) hash
     * to a different number. */
    //利用64 bit Mix Functions，将这些状态信息混合到hash中，组成最后的指纹，如果这些状态中有一个
    //出现变化，可以通过一个算法逆推出该状态变化之前的值。例如，d->ht[0].size发生变化，则我们可
    //以通过hash和其他的几个状态，逆推出d->ht[0].size的最初值。
    for (j = 0; j < 6; j++) {
        hash += integers[j];
        /* For the hashing step we use TomasWang's 64 bit integer hash. */
        hash = (~hash) + (hash << 21); //hash = (hash << 21) - hash - 1;
        hash = hash ^ (hash >> 24);
        hash = (hash + (hash << 3)) +(hash << 8); // hash * 265
        hash = hash ^ (hash >> 14);
        hash = (hash + (hash << 2)) +(hash << 4); // hash * 21
        hash = hash ^ (hash >> 28);
        hash = hash + (hash << 31);
    }
    return hash;
}

/* MurmurHash2, byAustin Appleby
 * Note - This code makes a few assumptionsabout how your machine behaves -
 * 1. We can read a 4-byte value from anyaddress without crashing
 * 2. sizeof(int) == 4
 *
 * And it has a few limitations -
 *
 * 1. It will not work incrementally.
 * 2. It will not produce the same results onlittle-endian and big-endian
 *   machines.
 */
unsigned intdictGenHashFunction(const void *key, int len) {
    /* 'm' and 'r' are mixing constantsgenerated offline.
     They're not really 'magic', they justhappen to work well.  */
    uint32_t seed = dict_hash_function_seed;
    const uint32_t m = 0x5bd1e995;
    const int r = 24;

    /* Initialize the hash to a 'random' value*/
    uint32_t h = seed ^ len;

    /* Mix 4 bytes at a time into the hash */
    const unsigned char *data = (const unsignedchar *)key;

    while(len >= 4) {
        uint32_t k = *(uint32_t*)data;

        k *= m;
        k ^= k >> r;
        k *= m;

        h *= m;
        h ^= k;

        data += 4;
        len -= 4;
    }

    /* Handle the last few bytes of the inputarray  */
    switch(len) {
    case 3: h ^= data[2] << 16;
    case 2: h ^= data[1] << 8;
    case 1: h ^= data[0]; h *= m;
    };

    /* Do a few final mixes of the hash toensure the last few
     * bytes are well-incorporated. */
    h ^= h >> 13;
    h *= m;
    h ^= h >> 15;

    return (unsigned int)h;
}