Redis6的IO多线程分析

性能测试

机器配置

C++
Architecture:          x86_64
CPU op-mode(s):        32-bit, 64-bit
Byte Order:            Little Endian
CPU(s):                14
On-line CPU(s) list:   0-13
Mem:            62G
1
2
3
4
5
6
7

性能

配置推荐

官方表示，当使用redis时有性能瓶颈时，才推荐开启该功能，但是消耗更多的cpu time。并且，开启i/o多线程，至少要是4核以上的cpu，并且需要预留一个空闲cpu。比如4核就配io-threads=2，8核至多配io-threads=6。对于4核以上的机器，官方更推荐配置io-threads=4，因为再往上叠加，收益相对较低了，也没有太大必要。
另外，io-threads-do-reads配置默认no。i/o多线程默认是的是写socket多线程，socket连接本身就是epoll多路复用模式，理解下来开启此配置对性能不太有提升空间。

分析

多线程模型

并非是标准的Reactor多线程模型。主线程将就绪的socket交给IO线程去读写，IO线程处理完后主线程开始进行下一步操作。

流程图

IO线程执行函数代码分析

void *IOThreadMain(void *myid) {
    long id = (unsigned long)myid;
    char thdname[16];

    snprintf(thdname, sizeof(thdname), "io_thd_%ld", id);
    redis_set_thread_title(thdname);
    
    //设置线程的CPU亲和性
    redisSetCpuAffinity(server.server_cpulist); 
    
    //设置线程可以在任意时刻被kill
    //pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
    //pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL);
    makeThreadKillable();   

    while(1) {
        /* Wait for start */
        //主线程setIOPendingCount(id, count)时，这里先空转一下。
        for (int j = 0; j < 1000000; j++) {
            if (getIOPendingCount(id) != 0) break;
        }

        /* 如果pending count 小于线程数*2，或者io_threads_num == 1，那么交给主线程自己处理，阻塞子线程。
        *   （入口是：handleClientsWithPendingWritesUsingThreads->stopThreadedIOIfNeeded）
        *  主线程尝试加锁，加锁成功后，子线程就阻塞在这里了。（见：stopThreadedIO）
        */
        if (getIOPendingCount(id) == 0) {
            pthread_mutex_lock(&io_threads_mutex[id]);
            pthread_mutex_unlock(&io_threads_mutex[id]);
            continue;
        }

        serverAssert(getIOPendingCount(id) != 0);

        /* Process: note that the main thread will never touch our list
         * before we drop the pending count to 0. */
        listIter li;
        listNode *ln;
        //每个线程都有自己的list，遍历list执行就序操作。
        listRewind(io_threads_list[id],&li);    
        while((ln = listNext(&li))) {
            client *c = listNodeValue(ln);
            //handleClientsWithPendingWritesUsingThreads()会把就序操作设置为IO_THREADS_OP_WRITE
            if (io_threads_op == IO_THREADS_OP_WRITE) {
                writeToClient(c,0);    //写
            //handleClientsWithPendingReadsUsingThreads()会把就序操作设置为IO_THREADS_OP_READ
            } else if (io_threads_op == IO_THREADS_OP_READ) {
                readQueryFromClient(c->conn);    //读（需开启io-threads-do-reads）
            } else {
                serverPanic("io_threads_op value is unknown");
            }
        }
        listEmpty(io_threads_list[id]);
        // 置0，表示这个线程已经处理完了。
        setIOPendingCount(id, 0);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57