命令处理过程

void readQueryFromClient(connection *conn) {
    // [1] 从 connection 中获取 client 数据结构
    client *c = connGetPrivateData(conn);
    int nread, readlen;
    size_t qblen;
    // [2] 如果开启了 I/0 线程，则交给 I/0 线程去处理，这里直接返回
    if (postponeClientRead(c)) return;
    /* 原子更增加服务器总读取操作的计数器 */
    atomicIncr(server.stat_total_reads_processed, 1);
    // [3] 设置最大读取字节数
    readlen = PROTO_IOBUF_LEN;
    // [4] 如果是多批量请求，且发生了拆包，这里做特殊处理
    if (c->reqtype == PROTO_REQ_MULTIBULK && c->multibulklen && c->bulklen != -1
        && c->bulklen >= PROTO_MBULK_BIG_ARG)
    {
        // bulklen 为指令总长度，remaining 为剩余未读长度 + 2(\r\n)
        ssize_t remaining = (size_t)(c->bulklen+2)-sdslen(c->querybuf);
        // 如果剩余未读长度小于 readlen，则更新 readlen 为 remaining
        if (remaining > 0 && remaining < readlen) readlen = remaining;
    }
    // [5] 计算当前查询缓冲区已被占用的长度
    qblen = sdslen(c->querybuf);
    // 并更新 querybuf_peak 变量以跟踪查询缓冲区的最大长度
    if (c->querybuf_peak < qblen) c->querybuf_peak = qblen;
    // 确保查询缓冲区有足够的空间来存放数据，如果不够则扩容
    c->querybuf = sdsMakeRoomFor(c->querybuf, readlen);
    // [6] 读取来自 socket 中的数据流
    nread = connRead(c->conn, c->querybuf+qblen, readlen);
    if (nread == -1) {
        // 异常处理...
    } else if (nread == 0) {
        // 断开连接 ...
        return;
    } else if (c->flags & CLIENT_MASTER) { // 如果 client 是集群主节点
        // 将查询缓冲区的数据追加到主服务器的挂起查询缓冲区
        c->pending_querybuf = sdscatlen(c->pending_querybuf,
                                        c->querybuf+qblen,nread);
    }

    sdsIncrLen(c->querybuf,nread); //更新 sds 数据长度
    c->lastinteraction = server.unixtime;
    // 增加这个 master 客户端的读取复制偏移量
    if (c->flags & CLIENT_MASTER) c->read_reploff += nread;
    atomicIncr(server.stat_net_input_bytes, nread);
    //[7]查询缓冲长度超了，则断开连接
    if (sdslen(c->querybuf) > server.client_max_querybuf_len) {
        // free...
        return;
    }

    // [8] 处理已经读取到缓冲区中的数据
    processInputBuffer(c);
}

void processInputBuffer(client *c) {
    // [1] 循环，直到处理完读取缓冲区中的数据
    while(c->qb_pos < sdslen(c->querybuf)) {
        // [2] 判断是否要终止处理:
        // 如果客户端正处于某种中止状态，跳出
        if (c->flags & CLIENT_BLOCKED) break;
        // 如果当前客户端的请求数据已经读取完毕，则跳出循环
        if (c->flags & CLIENT_PENDING_COMMAND) break;
        // 如果收到请求的服务器处于 Lua 脚本执行状态，
        //  且发送请求的客户端是 Master 节点，也中止处理，用于累积复制流
        if (server.lua_timedout && c->flags & CLIENT_MASTER) break;
        // 如果客户端包含以下两个状态，表示需要关闭连接，也立刻终止处理
        if (c->flags & (CLIENT_CLOSE_AFTER_REPLY|CLIENT_CLOSE_ASAP)) break;

        // [3] 如果是第一次处理这个请求，则设定请求类型
        if (!c->reqtype) {
            if (c->querybuf[c->qb_pos] == '*') {
                // 如果请求以 * 开头，则表示 PROTO REQ MULTIBULK
                c->reqtype = PROTO_REQ_MULTIBULK;
            } else {
                c->reqtype = PROTO_REQ_INLINE;
            }
        }
        // [4] 根据请求类型，调用不同的函数处理
        if (c->reqtype == PROTO_REQ_INLINE) {
            if (processInlineBuffer(c) != C_OK) break;
            if (server.gopher_enabled && !server.io_threads_do_reads &&
                ((c->argc == 1 && ((char*)(c->argv[0]->ptr))[0] == '/') ||
                  c->argc == 0))
            {
                processGopherRequest(c);
                resetClient(c);
                c->flags |= CLIENT_CLOSE_AFTER_REPLY;
                break;
            }
        } else if (c->reqtype == PROTO_REQ_MULTIBULK) {
            if (processMultibulkBuffer(c) != C_OK) break;
        } else {
            serverPanic("Unknown request type");
        }

        // 到这里，请求就读取完毕了
        if (c->argc == 0) {
            // [5] 如果命令未携带参数，则重置客户端
            resetClient(c);
        } else { // 如果有参数
            // [6] 如果启用了 I/0 线程，则跳出循环先不处理
            if (c->flags & CLIENT_PENDING_READ) {
                c->flags |= CLIENT_PENDING_COMMAND;
                break;
            }
            // 否则先执行命令，再重置客户端 (即返回响应)
            if (processCommandAndResetClient(c) == C_ERR) {
                return;
            }
        }
    }

    // [8] 命令执行成功，
        // 删除查询缓冲中已经处理过的请求，并初始化 qb_pos
    if (c->qb_pos) {
        sdsrange(c->querybuf,c->qb_pos,-1);
        c->qb_pos = 0;
    }
}

int processMultibulkBuffer(client *c) {
    char *newline = NULL;
    int ok;
    long long ll;
    // [1] 读取 Multibulk 请求头部的指令行数
    if (c->multibulklen == 0) {
        ...
        newline = strchr(c->querybuf+c->qb_pos,'\r');
        ...
        ok = string2ll(c->querybuf+1+c->qb_pos,newline-(c->querybuf+1+c->qb_pos),&ll);
        ...
        /* 更新 qb_pos：
         *        newline
         *           |
         *           v
         *   *123123\r\n...\r\n...\r\n
         *   ^          ^
         *   |          |
         * querybuf   qb_pos
         */
        c->qb_pos = (newline-c->querybuf)+2;
        ...
        c->multibulklen = ll; // 存储指令行数
        ...
    }
    ...
    // [2] 每次处理一行数据，直到处理完
    while(c->multibulklen) {

int processCommandAndResetClient(client *c) {
    int deadclient = 0;
    client *old_client = server.current_client;
    // 先将当前客户端设为 current_client
    server.current_client = c;
    // 执行命令
    if (processCommand(c) == C_OK) {
        // 执行成功后，做一些收尾工作
        commandProcessed(c);
    }
    if (server.current_client == NULL) deadclient = 1;
    // current_client 恢复至原客户端
    server.current_client = old_client;
    return deadclient ? C_ERR : C_OK;
}

int processCommand(client *c) {
    // [1] Mudule Filter
    moduleCallCommandFilters(c);
    // [2] 优先处理退出命令
    if (!strcasecmp(c->argv[0]->ptr,"quit")) {
        addReply(c,shared.ok);
        c->flags |= CLIENT_CLOSE_AFTER_REPLY;
        return C_ERR;
    }
    // [3] 根据指令名称，查找指令格式
    c->cmd = c->lastcmd = lookupCommand(c->argv[0]->ptr);
    ...

    // [4] 执行以下逻辑： 
    // 获取请求类型
    int is_read_command = (c->cmd->flags & CMD_READONLY) ||
                           (c->cmd->proc == execCommand && (c->mstate.cmd_flags & CMD_READONLY));
    ...
    // 如果启用了用户验证，则在这里完成验证
    if (authRequired(c)) {...}
    // 权鉴
    int acl_errpos;
    int acl_retval = ACLCheckAllPerm(c,&acl_errpos);
    // cluster 模式特别处理
    if (server.cluster_enabled &&
        !(c->flags & CLIENT_MASTER) &&
        !(c->flags & CLIENT_LUA &&
          server.lua_caller->flags & CLIENT_MASTER) &&
        !(!cmdHasMovableKeys(c->cmd) && c->cmd->firstkey == 0 &&
          c->cmd->proc != execCommand)) {...}
    // 如果服务器配置了内存最大限制，则按需淘汰数据
    if (server.maxmemory && !server.lua_timedout) {...}
    // tracking 机制要求记录查询过的 key
    if (server.tracking_clients) trackingLimitUsedSlots();
    // 如果当前服务是 master，当持久化异常了或者正常的从节点数量小于阈值，拒绝执行请求
    int deny_write_type = writeCommandsDeniedByDiskError();
    ...
    // 如果当前服务是 slave，且客户端并非 master，则拒绝写命令
    if (server.masterhost && server.repl_slave_ro &&
        !(c->flags & CLIENT_MASTER) &&
        is_write_command) {...}
    // 如果客户端处于 pub/sub 模式，而且使用的是 RESP2 协议，
    //   则只允许执行 ping、sub、unsub、psub、punsub、reset 这几种命令
    //   其他类型命令都拒绝执行
    if ((c->flags & CLIENT_PUBSUB && c->resp == 2) &&
        c->cmd->proc != pingCommand && ... {...}
    // 如果当前服务是 slave，且与 master 断连，则拒绝执行 is_denystale_command 类型命令
    if (server.masterhost && server.repl_state != REPL_STATE_CONNECTED &&
        server.repl_serve_stale_data == 0 &&
        is_denystale_command) {...}
    // 服务器正在加载数据期间，只允许执行执行特定命令
    if (server.loading && is_denyloading_command) {...}
    // lua 脚本执行超时期间，只有特定命令能执行
    if (server.lua_timedout &&
          c->cmd->proc != authCommand && ... {...}
    // 阻止 slave 发送访问密钥空间的命令
    if ((c->flags & CLIENT_SLAVE) && (is_may_replicate_command || is_write_command || is_read_command)) {...}
    // 如果服务器已暂停，则将非 slave 客户端阻塞
    if (!(c->flags & CLIENT_SLAVE) && 
        ((server.client_pause_type == CLIENT_PAUSE_ALL) ||
        (server.client_pause_type == CLIENT_PAUSE_WRITE && is_may_replicate_command)))
    {...}

    /* [5] Exec the command */
    if (c->flags & CLIENT_MULTI &&
        c->cmd->proc != execCommand && c->cmd->proc != discardCommand &&
        c->cmd->proc != multiCommand && c->cmd->proc != watchCommand &&
        c->cmd->proc != resetCommand)
    {
        // 如果当前 client 处于事务上下文，将非 exec、discard、watch、reset 命令放入事务队列
        queueMultiCommand(c);
        addReply(c,shared.queued);
    } else {
        // 非事务上下文，或者是 exec、discard、watch、reset 命令，则直接执行
        call(c,CMD_CALL_FULL);
        c->woff = server.master_repl_offset;
        if (listLength(server.ready_keys))
            handleClientsBlockedOnKeys();
    }
    
    return C_OK;
}

void call(client *c, int flags) {
    ...
    // [1] 清空传播控制标志
    c->flags &= ~(CLIENT_FORCE_AOF|CLIENT_FORCE_REPL|CLIENT_PREVENT_PROP);
    ...
    // [2] 调用命令处理函数 proc，执行命令处理逻辑
    c->cmd->proc(c);
    ...
    // [3] 根据状态设定 propagate_flags 标志，再根据 propagate_flags 将命令记录到 AOF 文件或发送到从服务器
    if (flags & CMD_CALL_PROPAGATE &&
        (c->flags & CLIENT_PREVENT_PROP) != CLIENT_PREVENT_PROP)
    {
        int propagate_flags = PROPAGATE_NONE;
        // 产生脏数据，即修改了数据库
        if (dirty) propagate_flags |= (PROPAGATE_AOF|PROPAGATE_REPL);
        ...
        if (propagate_flags != PROPAGATE_NONE && !(c->cmd->flags & CMD_MODULE))
            propagate(c->cmd,c->db->id,c->argv,c->argc,propagate_flags);
    }

    // [4] 补偿先前被去清空的传播控制标志
    c->flags &= ~(CLIENT_FORCE_AOF|CLIENT_FORCE_REPL|CLIENT_PREVENT_PROP);
    c->flags |= client_old_flags &
        (CLIENT_FORCE_AOF|CLIENT_FORCE_REPL|CLIENT_PREVENT_PROP);
    // [5] 记录 also_propagate 中存放的额外传播控制命令
    if (server.also_propagate.numops) {...}
    server.also_propagate = prev_also_propagate;
    ...
    // [6] 如果是查询类型的命令，那么记忆该命令，后续当 key 变化时通知客户端（Tracking 机制）
     if (c->cmd->flags & CMD_READONLY) {
        client *caller = (c->flags 
            & CLIENT_LUA && server.lua_caller) 
            ? server.lua_caller : c;
        if (caller->flags & CLIENT_TRACKING &&
            !(caller->flags & CLIENT_TRACKING_BCAST))
        {
            // 存储当前请求的所有 keys
            trackingRememberKeys(caller);
        }
     }
     // clean
     ...
}

void commandProcessed(client *c) {
    if (c->flags & CLIENT_BLOCKED) return;
    // [1] 重置客户端
    resetClient(c);
    long long prev_offset = c->reploff;
    // [2] 更新已同步命令的偏移量
    if (c->flags & CLIENT_MASTER && !(c->flags & CLIENT_MULTI)) {
        c->reploff = c->read_reploff - sdslen(c->querybuf) + c->qb_pos;
    }
    // [3] 如果客户端是 master，则将这个命令复制到当前节点的从节点
    if (c->flags & CLIENT_MASTER) {
        long long applied = c->reploff - prev_offset;
        if (applied) {
            replicationFeedSlavesFromMasterStream(server.slaves,
                    c->pending_querybuf, applied);
            sdsrange(c->pending_querybuf,applied,-1);
        }
    }
}

void addReplySds(client *c, sds s) {
    // [1] 如果客户端允许响应回复，则将其加到阻塞队列
    if (prepareClientToWrite(c) != C_OK) {
        /* The caller expects the sds to be free'd. */
        sdsfree(s);
        return;
    }
    // [2] 先尝试将其写入 client.buf
    if (_addReplyToBuffer(c,s,sdslen(s)) != C_OK)
        // buf 存不下，则写入 list
        _addReplyProtoToList(c,s,sdslen(s));
    sdsfree(s);
}

int handleClientsWithPendingWrites(void) {
    listIter li;
    listNode *ln;
    // [1] 遍历 clients_pending_write
    int processed = listLength(server.clients_pending_write);
    listRewind(server.clients_pending_write,&li);
    while((ln = listNext(&li))) {
        client *c = listNodeValue(ln);
        c->flags &= ~CLIENT_PENDING_WRITE;
        listDelNode(server.clients_pending_write,ln);
        ...
        // [2] 写入 tcp 发送缓冲
        if (writeToClient(c,0) == C_ERR) continue;

        // [3] 如果回复缓冲中数据太多，则注册回调慢慢发送
        if (clientHasPendingReplies(c)) {
            int ae_barrier = 0;
            if (server.aof_state == AOF_ON &&
                server.aof_fsync == AOF_FSYNC_ALWAYS)
            {
                ae_barrier = 1;
            }
            if (connSetWriteHandlerWithBarrier(c->conn, sendReplyToClient, ae_barrier) == C_ERR) {
                freeClientAsync(c);
            }
        }
    }
    return processed;
}