Redis Cluster 的實現 - cluster 消息的接收和分包

由於針對 READ 事件的 clusterReadHandler 處理器主要工做就是解析 cluster bus 上接收的數據並進行消息分包，而後對消息進行處理，而對於消息的分包首先須要瞭解一下消息結構，Redis Cluster 節點之間通訊的消息結構定義以下:
 

typedef struct {
    char sig[4];        /* Siganture "RCmb" (Redis Cluster message bus). */
    uint32_t totlen;    /* Total length of this message */
    uint16_t ver;       /* Protocol version, currently set to 0. */
    uint16_t notused0;  /* 2 bytes not used. */
    
    /* Message type，如：PING, PONG，定義參考宏定義 CLUSTERMSG_TYPE_* */
    uint16_t type;
    
    uint16_t count;     /* Only used for some kind of messages. */
    uint64_t currentEpoch;  /* The epoch accordingly to the sending node. */
    uint64_t configEpoch;   /* The config epoch if it's a master, or the last
                               epoch advertised by its master if it is a
                               slave. */
    uint64_t offset;    /* Master replication offset if node is a master or
                           processed replication offset if node is a slave. */
                           
    // 節點發送方，爲 NodeID 表示，如: 123ed65d59ff22370f2f09546f410d31207789f6
    char sender[REDIS_CLUSTER_NAMELEN]; /* Name of the sender node */
    
    // 本節點維護的 slots bits
    unsigned char myslots[REDIS_CLUSTER_SLOTS/8];
    
    // 若是本節點爲 slave 節點，則 slaveof 記錄對應的 master 節點ID
    char slaveof[REDIS_CLUSTER_NAMELEN];
    
    char notused1[32];  /* 32 bytes reserved for future usage. */
    uint16_t port;      /* Sender TCP base port */
    uint16_t flags;     /* Sender node flags */
    
    // cluster 狀態， 如：REDIS_CLUSTER_OK, REDIS_CLUSTER_FAIL ...
    unsigned char state; /* Cluster state from the POV of the sender */
    unsigned char mflags[3]; /* Message flags: CLUSTERMSG_FLAG[012]_... */
    
    // 指向不一樣 消息類型 的消息體
    // 參考 clusterMsgData 結構的說明
    union clusterMsgData data;
} clusterMsg;

 

從上面結構能夠看到消息分包，主要解析前 8 個字節，分別爲:
 
  - char sig[4];      // 消息簽名，對於 cluster 消息，固定爲字符序列 RCmb
  - uint32_t totlen;  // 消息總長度
 
其餘結構成員都是在處理消息時使用的，後續講解消息處理流程時進行分析。

 

/* Read data. Try to read the first field of the header first to check the
 * full length of the packet. When a whole packet is in memory this function
 * will call the function to process the packet. And so forth. */
void clusterReadHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
    char buf[sizeof(clusterMsg)];
    ssize_t nread;
    clusterMsg *hdr;
    clusterLink *link = (clusterLink*) privdata;
    int readlen, rcvbuflen;
    REDIS_NOTUSED(el);
    REDIS_NOTUSED(mask);
    while(1) { /* Read as long as there is data to read. */
        rcvbuflen = sdslen(link->rcvbuf);
        if (rcvbuflen < 8) {
            /* First, obtain the first 8 bytes to get the full message
             * length. */
            readlen = 8 - rcvbuflen;
        } else {
            // 已經知道了本條消息的長度
            // 本塊代碼主要計算剩餘還需讀入的字節數(readlen)纔是完整的消息
            /* Finally read the full message. */
            hdr = (clusterMsg*) link->rcvbuf;
            if (rcvbuflen == 8) {
                /* Perform some sanity check on the message signature
                 * and length. */
                if (memcmp(hdr->sig,"RCmb",4) != 0 ||
                    ntohl(hdr->totlen) < CLUSTERMSG_MIN_LEN)
                {
                    redisLog(REDIS_WARNING,
                        "Bad message length or signature received "
                        "from Cluster bus.");
                    handleLinkIOError(link);
                    return;
                }
            }
            readlen = ntohl(hdr->totlen) - rcvbuflen;
            if (readlen > sizeof(buf)) readlen = sizeof(buf);
        }
        
        // 讀入本條消息記錄的剩餘 readlen 個字節的數據
        // 由於這裏的 fd 是非阻塞的，因此須要判斷 EAGAIN
        nread = read(fd,buf,readlen);
        if (nread == -1 && errno == EAGAIN) return; /* No more data ready. */
        if (nread <= 0) {
            /* I/O error... */
            redisLog(REDIS_DEBUG,"I/O error reading from node link: %s",
                (nread == 0) ? "connection closed" : strerror(errno));
            handleLinkIOError(link);
            return;
        } else {
            /* Read data and recast the pointer to the new buffer. */
            link->rcvbuf = sdscatlen(link->rcvbuf,buf,nread);
            hdr = (clusterMsg*) link->rcvbuf;
            rcvbuflen += nread;
        }
        /* Total length obtained? Process this packet. */
        if (rcvbuflen >= 8 && rcvbuflen == ntohl(hdr->totlen)) {
            // 代表 link 上的 rcvbuf 已是一個完整的 cluster 消息
            // 下面開始處理此消息
            if (clusterProcessPacket(link)) {
                sdsfree(link->rcvbuf);
                link->rcvbuf = sdsempty();
            } else {
                return; /* Link no longer valid. */
            }
        }
    }
}