1    /* Create an event handler for accepting new connections in TCP and Unix
 2     * domain sockets. */
 3    if (createSocketAcceptHandler(&server.ipfd, acceptTcpHandler) != C_OK) {
 4        serverPanic("Unrecoverable error creating TCP socket accept handler.");
 5    }
 6    if (createSocketAcceptHandler(&server.tlsfd, acceptTLSHandler) != C_OK) {
 7        serverPanic("Unrecoverable error creating TLS socket accept handler.");
 8    }
 9    if (server.sofd > 0 && aeCreateFileEvent(server.el,server.sofd,AE_READABLE,
10        acceptUnixHandler,NULL) == AE_ERR) serverPanic("Unrecoverable error creating server.sofd file event.");
11

 1/* Create an event handler for accepting new connections in TCP or TLS domain sockets.
 2 * This works atomically for all socket fds */
 3int createSocketAcceptHandler(socketFds *sfd, aeFileProc *accept_handler) {
 4    int j;
 5
 6    for (j = 0; j < sfd->count; j++) {
 7        if (aeCreateFileEvent(server.el, sfd->fd[j], AE_READABLE, accept_handler,NULL) == AE_ERR) {
 8            /* Rollback */
 9            for (j = j-1; j >= 0; j--) aeDeleteFileEvent(server.el, sfd->fd[j], AE_READABLE);
10            return C_ERR;
11        }
12    }
13    return C_OK;
14}
15
16int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
17        aeFileProc *proc, void *clientData)
18{
19    if (fd >= eventLoop->setsize) {
20        errno = ERANGE;
21        return AE_ERR;
22    }
23    aeFileEvent *fe = &eventLoop->events[fd];
24
25    if (aeApiAddEvent(eventLoop, fd, mask) == -1)
26        return AE_ERR;
27    fe->mask |= mask;
28    if (mask & AE_READABLE) fe->rfileProc = proc;
29    if (mask & AE_WRITABLE) fe->wfileProc = proc;
30    fe->clientData = clientData;
31    if (fd > eventLoop->maxfd)
32        eventLoop->maxfd = fd;
33    return AE_OK;
34}

1aeCreateFileEvent(server.el, sfd->fd[j], AE_READABLE, accept_handler,NULL)

 1static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
 2    aeApiState *state = eventLoop->apidata;
 3    struct epoll_event ee = {0}; /* avoid valgrind warning */
 4    /* If the fd was already monitored for some event, we need a MOD
 5     * operation. Otherwise we need an ADD operation. */
 6    int op = eventLoop->events[fd].mask == AE_NONE ?
 7            EPOLL_CTL_ADD : EPOLL_CTL_MOD;
 8
 9    ee.events = 0;
10    mask |= eventLoop->events[fd].mask; /* Merge old events */
11    if (mask & AE_READABLE) ee.events |= EPOLLIN;
12    if (mask & AE_WRITABLE) ee.events |= EPOLLOUT;
13    ee.data.fd = fd;
14    if (epoll_ctl(state->epfd,op,fd,&ee) == -1) return -1;
15    return 0;
16}

 1void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
 2    int cport, cfd, max = MAX_ACCEPTS_PER_CALL;
 3    char cip[NET_IP_STR_LEN];
 4    UNUSED(el);
 5    UNUSED(mask);
 6    UNUSED(privdata);
 7
 8    while(max--) {
 9        cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip), &cport);
10        if (cfd == ANET_ERR) {
11            if (errno != EWOULDBLOCK)
12                serverLog(LL_WARNING,
13                    "Accepting client connection: %s", server.neterr);
14            return;
15        }
16        anetCloexec(cfd);
17        serverLog(LL_VERBOSE,"Accepted %s:%d", cip, cport);
18        acceptCommonHandler(connCreateAcceptedSocket(cfd),0,cip);
19    }
20}

 1static int anetGenericAccept(char *err, int s, struct sockaddr *sa, socklen_t *len) {
 2    int fd;
 3    while(1) {
 4        fd = accept(s,sa,len);
 5        if (fd == -1) {
 6            if (errno == EINTR)
 7                continue;
 8            else {
 9                anetSetError(err, "accept: %s", strerror(errno));
10                return ANET_ERR;
11            }
12        }
13        break;
14    }
15    return fd;
16}
17
18int anetTcpAccept(char *err, int s, char *ip, size_t ip_len, int *port) {
19    int fd;
20    struct sockaddr_storage sa;
21    socklen_t salen = sizeof(sa);
22    if ((fd = anetGenericAccept(err,s,(struct sockaddr*)&sa,&salen)) == -1)
23        return ANET_ERR;
24
25    if (sa.ss_family == AF_INET) {
26        struct sockaddr_in *s = (struct sockaddr_in *)&sa;
27        if (ip) inet_ntop(AF_INET,(void*)&(s->sin_addr),ip,ip_len);
28        if (port) *port = ntohs(s->sin_port);
29    } else {
30        struct sockaddr_in6 *s = (struct sockaddr_in6 *)&sa;
31        if (ip) inet_ntop(AF_INET6,(void*)&(s->sin6_addr),ip,ip_len);
32        if (port) *port = ntohs(s->sin6_port);
33    }
34    return fd;
35}

 1#define MAX_ACCEPTS_PER_CALL 1000
 2static void acceptCommonHandler(connection *conn, int flags, char *ip) {
 3    client *c;
 4    char conninfo[100];
 5    UNUSED(ip);
 6
 7    if (connGetState(conn) != CONN_STATE_ACCEPTING) {
 8        serverLog(LL_VERBOSE,
 9            "Accepted client connection in error state: %s (conn: %s)",
10            connGetLastError(conn),
11            connGetInfo(conn, conninfo, sizeof(conninfo)));
12        connClose(conn);
13        return;
14    }
15
16    /* Limit the number of connections we take at the same time.
17     *
18     * Admission control will happen before a client is created and connAccept()
19     * called, because we don't want to even start transport-level negotiation
20     * if rejected. */
21    if (listLength(server.clients) + getClusterConnectionsCount()
22        >= server.maxclients)
23    {
24        char *err;
25        if (server.cluster_enabled)
26            err = "-ERR max number of clients + cluster "
27                  "connections reached\r\n";
28        else
29            err = "-ERR max number of clients reached\r\n";
30
31        /* That's a best effort error message, don't check write errors.
32         * Note that for TLS connections, no handshake was done yet so nothing
33         * is written and the connection will just drop. */
34        if (connWrite(conn,err,strlen(err)) == -1) {
35            /* Nothing to do, Just to avoid the warning... */
36        }
37        server.stat_rejected_conn++;
38        connClose(conn);
39        return;
40    }
41
42    /* Create connection and client */
43    if ((c = createClient(conn)) == NULL) {
44        serverLog(LL_WARNING,
45            "Error registering fd event for the new client: %s (conn: %s)",
46            connGetLastError(conn),
47            connGetInfo(conn, conninfo, sizeof(conninfo)));
48        connClose(conn); /* May be already closed, just ignore errors */
49        return;
50    }
51
52    /* Last chance to keep flags */
53    c->flags |= flags;
54
55    /* Initiate accept.
56     *
57     * Note that connAccept() is free to do two things here:
58     * 1. Call clientAcceptHandler() immediately;
59     * 2. Schedule a future call to clientAcceptHandler().
60     *
61     * Because of that, we must do nothing else afterwards.
62     */
63    if (connAccept(conn, clientAcceptHandler) == C_ERR) {
64        char conninfo[100];
65        if (connGetState(conn) == CONN_STATE_ERROR)
66            serverLog(LL_WARNING,
67                    "Error accepting a client connection: %s (conn: %s)",
68                    connGetLastError(conn), connGetInfo(conn, conninfo, sizeof(conninfo)));
69        freeClient(connGetPrivateData(conn));
70        return;
71    }
72}

 1struct connection {
 2    /// 连接类型，这是个回调函数的集合，封装了这个connection同步读写、异步读写、等回调。可以认为这个结构描述了connection的行为。
 3    ConnectionType *type;
 4    /// 连接的状态包含 取值我注释在上面的枚举定义中。
 5    ConnectionState state;
 6    /// 
 7    short int flags;
 8    /// 引用的操作数，这里值还在epoll等待事件未返回的操作数量。
 9    short int refs;
10    /// 最后一次错误的错误码，调用connSocketGetLastError时返回
11    int last_errno;
12    /// 自定义的用户数据，这里用来保存client对象指针
13    void *private_data;
14    /// 连接事件处理函数，初始化的时候是置空的。
15    ConnectionCallbackFunc conn_handler;
16    /// 写事件处理函数，初始化的时候是置空的。
17    ConnectionCallbackFunc write_handler;
18    /// 读事件处理函数，初始化的时候是置空的。
19    ConnectionCallbackFunc read_handler;
20    /// Socket 描述符
21    int fd;
22};

 1ConnectionType CT_Socket = {
 2    .ae_handler = connSocketEventHandler,
 3    .close = connSocketClose,
 4    .write = connSocketWrite,
 5    .read = connSocketRead,
 6    .accept = connSocketAccept,
 7    .connect = connSocketConnect,
 8    .set_write_handler = connSocketSetWriteHandler,
 9    .set_read_handler = connSocketSetReadHandler,
10    .get_last_error = connSocketGetLastError,
11    .blocking_connect = connSocketBlockingConnect,
12    .sync_write = connSocketSyncWrite,
13    .sync_read = connSocketSyncRead,
14    .sync_readline = connSocketSyncReadLine,
15    .get_type = connSocketGetType
16};
17
18connection *connCreateSocket() {
19    connection *conn = zcalloc(sizeof(connection));
20    conn->type = &CT_Socket;
21    conn->fd = -1;
22
23    return conn;
24}

1    int connGetState(connection *conn) {
2        return conn->state;
3    }

 1client *createClient(connection *conn) {
 2    client *c = zmalloc(sizeof(client));
 3
 4    /* passing NULL as conn it is possible to create a non connected client.
 5     * This is useful since all the commands needs to be executed
 6     * in the context of a client. When commands are executed in other
 7     * contexts (for instance a Lua script) we need a non connected client. */
 8    if (conn) {
 9        connNonBlock(conn);
10        connEnableTcpNoDelay(conn);
11        if (server.tcpkeepalive)
12            connKeepAlive(conn,server.tcpkeepalive);
13        connSetReadHandler(conn, readQueryFromClient);
14        connSetPrivateData(conn, c);
15    }
16
17    selectDb(c,0);
18    uint64_t client_id;
19    atomicGetIncr(server.next_client_id, client_id, 1);
20    c->id = client_id;
21    c->resp = 2;
22    c->conn = conn;
23    c->name = NULL;
24    c->bufpos = 0;
25    c->qb_pos = 0;
26    c->querybuf = sdsempty();
27    c->pending_querybuf = sdsempty();
28    c->querybuf_peak = 0;
29    c->reqtype = 0;
30    c->argc = 0;
31    c->argv = NULL;
32    c->argv_len_sum = 0;
33    c->original_argc = 0;
34    c->original_argv = NULL;
35    c->cmd = c->lastcmd = NULL;
36    c->multibulklen = 0;
37    c->bulklen = -1;
38    c->sentlen = 0;
39    c->flags = 0;
40    c->ctime = c->lastinteraction = server.unixtime;
41    clientSetDefaultAuth(c);
42    c->replstate = REPL_STATE_NONE;
43    c->repl_put_online_on_ack = 0;
44    c->reploff = 0;
45    c->read_reploff = 0;
46    c->repl_ack_off = 0;
47    c->repl_ack_time = 0;
48    c->repl_last_partial_write = 0;
49    c->slave_listening_port = 0;
50    c->slave_addr = NULL;
51    c->slave_capa = SLAVE_CAPA_NONE;
52    c->reply = listCreate();
53    c->reply_bytes = 0;
54    c->obuf_soft_limit_reached_time = 0;
55    listSetFreeMethod(c->reply,freeClientReplyValue);
56    listSetDupMethod(c->reply,dupClientReplyValue);
57    c->btype = BLOCKED_NONE;
58    c->bpop.timeout = 0;
59    c->bpop.keys = dictCreate(&objectKeyHeapPointerValueDictType,NULL);
60    c->bpop.target = NULL;
61    c->bpop.xread_group = NULL;
62    c->bpop.xread_consumer = NULL;
63    c->bpop.xread_group_noack = 0;
64    c->bpop.numreplicas = 0;
65    c->bpop.reploffset = 0;
66    c->woff = 0;
67    c->watched_keys = listCreate();
68    c->pubsub_channels = dictCreate(&objectKeyPointerValueDictType,NULL);
69    c->pubsub_patterns = listCreate();
70    c->peerid = NULL;
71    c->sockname = NULL;
72    c->client_list_node = NULL;
73    c->paused_list_node = NULL;
74    c->client_tracking_redirection = 0;
75    c->client_tracking_prefixes = NULL;
76    c->client_cron_last_memory_usage = 0;
77    c->client_cron_last_memory_type = CLIENT_TYPE_NORMAL;
78    c->auth_callback = NULL;
79    c->auth_callback_privdata = NULL;
80    c->auth_module = NULL;
81    listSetFreeMethod(c->pubsub_patterns,decrRefCountVoid);
82    listSetMatchMethod(c->pubsub_patterns,listMatchObjects);
83    if (conn) linkClient(c);
84    initClientMultiState(c);
85    return c;
86}

1selectDB(c,0)

1int selectDb(client *c, int id) {
2    if (id < 0 || id >= server.dbnum)
3        return C_ERR;
4    c->db = &server.db[id];
5    return C_OK;
6}

1atomicGetIncr(server.next_client_id, client_id, 1);

1if (conn) linkClient(c);
2initClientMultiState(c);

1void linkClient(client *c) {
2    listAddNodeTail(server.clients,c);
3    /* Note that we remember the linked list node where the client is stored,
4     * this way removing the client in unlinkClient() will not require
5     * a linear scan, but just a constant time operation. */
6    c->client_list_node = listLast(server.clients);
7    uint64_t id = htonu64(c->id);
8    raxInsert(server.clients_index,(unsigned char*)&id,sizeof(id),c,NULL);
9}

1void initClientMultiState(client *c) {
2    c->mstate.commands = NULL;
3    c->mstate.count = 0;
4    c->mstate.cmd_flags = 0;
5    c->mstate.cmd_inv_flags = 0;
6}