redis接收客戶端連接通過設(shè)置的acceptTcpHandler 進(jìn)行處理

void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
    // MAX_ACCEPTS_PER_CALL = 1000
    int cport, cfd, max = MAX_ACCEPTS_PER_CALL;
    char cip[NET_IP_STR_LEN];
    UNUSED(el);
    UNUSED(mask);
    UNUSED(privdata);
    
    // 接收客戶端連接
    while(max--) {
        // accpet
        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;
        }
        serverLog(LL_VERBOSE,"Accepted %s:%d", cip, cport);
        // handler accept
        acceptCommonHandler(cfd,0,cip);
    }
}

accpet過程

MAX_ACCEPTS_PER_CALL 標(biāo)識(shí)每次最多accept 1000個(gè)客戶端妙啃，如果超過1000牍戚，等到下次epoll通知在進(jìn)行處理。

redis_max_accepts_per_call.png

需要注意的是，這種處理方式只能在LT模式下進(jìn)行，ET下是不可以的，原因在于：

• ET模式下，epollo告訴系統(tǒng)fd發(fā)生READABLE事件時(shí)，必須一次讀完呕臂，否則會(huì)丟事件。
• LT模式下肪跋，epollo告訴系統(tǒng)fd發(fā)生READABLE事件時(shí)歧蒋，不需要一次讀完，沒有讀的事件州既，下次epoll_wait時(shí)依然會(huì)告知谜洽。

redis對(duì)這部分的處理模式是LT的，也就是epoll默認(rèn)的模式吴叶。

int anetTcpAccept(char *err, int s, char *ip, size_t ip_len, int *port) {
    int fd;
    struct sockaddr_storage sa;
    socklen_t salen = sizeof(sa);
    if ((fd = anetGenericAccept(err,s,(struct sockaddr*)&sa,&salen)) == -1)
        return ANET_ERR;

    if (sa.ss_family == AF_INET) {
        struct sockaddr_in *s = (struct sockaddr_in *)&sa;
        if (ip) inet_ntop(AF_INET,(void*)&(s->sin_addr),ip,ip_len);
        if (port) *port = ntohs(s->sin_port);
    } else {
        struct sockaddr_in6 *s = (struct sockaddr_in6 *)&sa;
        if (ip) inet_ntop(AF_INET6,(void*)&(s->sin6_addr),ip,ip_len);
        if (port) *port = ntohs(s->sin6_port);
    }
    return fd;
}

anetTcpAccpet 會(huì)調(diào)用anetGenericAccpet接收客戶端連接阐虚，然后解析IP地址和端口

static int anetGenericAccept(char *err, int s, struct sockaddr *sa, socklen_t *len) {
    int fd;
    while(1) {
        fd = accept(s,sa,len);
        if (fd == -1) {
            // EINTR 錯(cuò)誤
            if (errno == EINTR)
                continue;
            else {
                anetSetError(err, "accept: %s", strerror(errno));
                return ANET_ERR;
            }
        }
        break;
    }
    return fd;
}

anetGenericAccept 調(diào)用socket的accpet 接收連接，當(dāng)發(fā)生EINTR 系統(tǒng)中斷時(shí), 重啟accept

handler過程

accept成功會(huì)通過acceptCommonHandler處理客戶端的請(qǐng)求

static void acceptCommonHandler(int fd, int flags, char *ip) {
    // 創(chuàng)建client
    client *c;
    if ((c = createClient(fd)) == NULL) {
        serverLog(LL_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;
    }
    
    // client過多
    // 這里沒有直接close(fd)而是接收了fd并創(chuàng)建了客戶端,原因在于redis希望把錯(cuò)誤原因通過
    // IO告訴client, 之后再關(guān)閉client 
    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;
    }

    // 服務(wù)器處于protected_mode(保護(hù)模式)且沒有設(shè)置密碼, 并且綁定的端口也非特定端口, 
    // 同時(shí)ip是本地lo網(wǎng)卡, 出于安全性考慮, redis會(huì)關(guān)閉client, 并告知client錯(cuò)誤原因
    if (server.protected_mode &&
        server.bindaddr_count == 0 &&
        server.requirepass == NULL &&
        !(flags & CLIENT_UNIX_SOCKET) &&
        ip != NULL)
    {
        if (strcmp(ip,"127.0.0.1") && strcmp(ip,"::1")) {
            char *err =
                "-DENIED Redis is running in protected mode because protected "
                "mode is enabled, no bind address was specified, no "
                "authentication password is requested to clients. In this mode "
                "connections are only accepted from the loopback interface. "
                "If you want to connect from external computers to Redis you "
                "may adopt one of the following solutions: "
                "1) Just disable protected mode sending the command "
                "'CONFIG SET protected-mode no' from the loopback interface "
                "by connecting to Redis from the same host the server is "
                "running, however MAKE SURE Redis is not publicly accessible "
                "from internet if you do so. Use CONFIG REWRITE to make this "
                "change permanent. "
                "2) Alternatively you can just disable the protected mode by "
                "editing the Redis configuration file, and setting the protected "
                "mode option to 'no', and then restarting the server. "
                "3) If you started the server manually just for testing, restart "
                "it with the '--protected-mode no' option. "
                "4) Setup a bind address or an authentication password. "
                "NOTE: You only need to do one of the above things in order for "
                "the server to start accepting connections from the outside.\r\n";
            if (write(c->fd,err,strlen(err)) == -1) {
                /* Nothing to do, Just to avoid the warning... */
            }
            server.stat_rejected_conn++;
            freeClient(c);
            return;
        }
    }

    server.stat_numconnections++;
    c->flags |= flags;
}

最關(guān)鍵的部分在createClient

client *createClient(int fd) {
    client *c = zmalloc(sizeof(client));
    
    // fd != -1時(shí), 設(shè)置fd對(duì)應(yīng)的TCP狀態(tài)
    if (fd != -1) {
        // 設(shè)置fd非阻塞IO
        anetNonBlock(NULL,fd);
        // 設(shè)置TCPNoDeplay（非常重要蚌卤，必須設(shè)置）
        anetEnableTcpNoDelay(NULL,fd);
        // 根據(jù)配置設(shè)置keeapalive (tcp層面)
        if (server.tcpkeepalive)
            anetKeepAlive(NULL,fd,server.tcpkeepalive);
        // 設(shè)置事件處理回調(diào)
        if (aeCreateFileEvent(server.el,fd,AE_READABLE,
            readQueryFromClient, c) == AE_ERR)
        {
            close(fd);
            zfree(c);
            return NULL;
        }
    }

    // 選擇db
    selectDb(c,0);
    uint64_t client_id;
    // 根據(jù)當(dāng)前client_id設(shè)置下一個(gè)client_id
    atomicGetIncr(server.next_client_id,client_id,1);
    // 省略....設(shè)置client各種狀態(tài)
    return c;
}

createClient 在fd == -1 的情況下還是會(huì)創(chuàng)建一個(gè)client实束，只不過這個(gè)client是 “空的client” （沒有連接）奥秆，原因在于client是一個(gè)抽象概念，client并不等于實(shí)際的TCP連接咸灿，除了通過網(wǎng)絡(luò)連接創(chuàng)建的client外构订，redis內(nèi)部一些執(zhí)行需要client對(duì)象。

對(duì)真正的連接析显，createClient會(huì)設(shè)置對(duì)應(yīng)的TCP狀態(tài)鲫咽，其實(shí)就是常用的網(wǎng)絡(luò)編程三板斧

• 事件驅(qū)動(dòng)下必須設(shè)置的非阻塞IO
• 必須設(shè)置的Tcp No Deplay
• tcp 的 keepalive

之后會(huì)創(chuàng)建FileEvent，監(jiān)聽client的可讀事件谷异，并設(shè)置readQueryFromClient 作為其回調(diào)函數(shù)。

激發(fā)maxclients錯(cuò)誤

為了測(cè)試該錯(cuò)誤锦聊，首先將 redis.conf中的 maxclients設(shè)置為100歹嘹，然后使用go client去連接

package main

import (
    "context"
    "flag"
    "fmt"
    "sync"

    "github.com/go-redis/redis/v8"
)

var (
    host            = flag.String("host", "localhost:6379", "redis host")
    redisMaxClients = flag.Int("maxclients", 100, "redis.conf maxclients options")
)

func main() {
    flag.Parse()
    wg := sync.WaitGroup{}
    wg.Add(*redisMaxClients)
    for i := 0; i < *redisMaxClients; i++ {
        go func(client int) {
            defer wg.Done()
            ctx := context.Background()
            rdb := redis.NewClient(&redis.Options{
                Addr:     *host,
                Password: "", // no password set
                DB:       0,  // use default DB
            })

            _, err := rdb.Ping(ctx).Result()
            if err != nil {
                fmt.Printf("%d client error: %s\n", client, err.Error())
            }
        }(i)
    }
    wg.Wait()
}

執(zhí)行結(jié)果

?  go git:(dev) ? ./maxclient -maxclients=101
32 client error: ERR max number of clients reached

?  go git:(dev) ? ./maxclient -maxclients=100