54 void clusterSendPing(clusterLink *link, int type);
71 void clusterSendUpdate(clusterLink *link, clusterNode *node);
548     clusterLink *link = zmalloc(sizeof(*link));  in createClusterLink()  local
549     link->ctime = mstime();  in createClusterLink()
550     link->sndbuf = sdsempty();  in createClusterLink()
551     link->rcvbuf = sdsempty();  in createClusterLink()
552     link->node = node;  in createClusterLink()
553     link->fd = -1;  in createClusterLink()
554     return link;  in createClusterLink()
560 void freeClusterLink(clusterLink *link) {  in freeClusterLink()  argument
561     if (link->fd != -1) {  in freeClusterLink()
562         aeDeleteFileEvent(server.el, link->fd, AE_WRITABLE);  in freeClusterLink()
563         aeDeleteFileEvent(server.el, link->fd, AE_READABLE);  in freeClusterLink()
565     sdsfree(link->sndbuf);  in freeClusterLink()
566     sdsfree(link->rcvbuf);  in freeClusterLink()
567     if (link->node)  in freeClusterLink()
568         link->node->link = NULL;  in freeClusterLink()
569     close(link->fd);  in freeClusterLink()
570     zfree(link);  in freeClusterLink()
578     clusterLink *link;  in clusterAcceptHandler()  local
605         link = createClusterLink(NULL);  in clusterAcceptHandler()
606         link->fd = cfd;  in clusterAcceptHandler()
607         aeCreateFileEvent(server.el,cfd,AE_READABLE,clusterReadHandler,link);  in clusterAcceptHandler()
670     node->link = NULL;  in createClusterNode()
837     if (n->link) freeClusterLink(n->link);  in freeClusterNode()
1293 void clusterProcessGossipSection(clusterMsg *hdr, clusterLink *link) {  in clusterProcessGossipSection()  argument
1296     clusterNode *sender = link->node ? link->node : clusterLookupNode(hdr->sender);  in clusterProcessGossipSection()
1343                 if (node->link) freeClusterLink(node->link);  in clusterProcessGossipSection()
1369 void nodeIp2String(char *buf, clusterLink *link) {  in nodeIp2String()  argument
1370     anetPeerToString(link->fd, buf, NET_IP_STR_LEN, NULL);  in nodeIp2String()
1383 int nodeUpdateAddressIfNeeded(clusterNode *node, clusterLink *link, int port) {  in nodeUpdateAddressIfNeeded()  argument
1392     if (link == node->link) return 0;  in nodeUpdateAddressIfNeeded()
1394     nodeIp2String(ip,link);  in nodeUpdateAddressIfNeeded()
1400     if (node->link) freeClusterLink(node->link);  in nodeUpdateAddressIfNeeded()
1541 int clusterProcessPacket(clusterLink *link) {  in clusterProcessPacket()  argument
1542     clusterMsg *hdr = (clusterMsg*) link->rcvbuf;  in clusterProcessPacket()
1552     if (totlen > sdslen(link->rcvbuf)) return 1;  in clusterProcessPacket()
1634         serverLog(LL_DEBUG,"Ping packet received: %p", (void*)link->node);  in clusterProcessPacket()
1650             if (anetSockName(link->fd,ip,sizeof(ip),NULL) != -1 &&  in clusterProcessPacket()
1668             nodeIp2String(node->ip,link);  in clusterProcessPacket()
1678             clusterProcessGossipSection(hdr,link);  in clusterProcessPacket()
1681         clusterSendPing(link,CLUSTERMSG_TYPE_PONG);  in clusterProcessPacket()
1690             (void*)link->node);  in clusterProcessPacket()
1691         if (link->node) {  in clusterProcessPacket()
1692             if (nodeInHandshake(link->node)) {  in clusterProcessPacket()
1699                     if (nodeUpdateAddressIfNeeded(sender,link,ntohs(hdr->port)))  in clusterProcessPacket()
1706                     clusterDelNode(link->node);  in clusterProcessPacket()
1712                 clusterRenameNode(link->node, hdr->sender);  in clusterProcessPacket()
1714                     link->node->name);  in clusterProcessPacket()
1715                 link->node->flags &= ~CLUSTER_NODE_HANDSHAKE;  in clusterProcessPacket()
1716                 link->node->flags |= flags&(CLUSTER_NODE_MASTER|CLUSTER_NODE_SLAVE);  in clusterProcessPacket()
1718             } else if (memcmp(link->node->name,hdr->sender,  in clusterProcessPacket()
1725                     link->node->name,  in clusterProcessPacket()
1726                     (int)(mstime()-(link->node->ctime)),  in clusterProcessPacket()
1727                     link->node->flags);  in clusterProcessPacket()
1728                 link->node->flags |= CLUSTER_NODE_NOADDR;  in clusterProcessPacket()
1729                 link->node->ip[0] = '\0';  in clusterProcessPacket()
1730                 link->node->port = 0;  in clusterProcessPacket()
1731                 freeClusterLink(link);  in clusterProcessPacket()
1740             nodeUpdateAddressIfNeeded(sender,link,ntohs(hdr->port)))  in clusterProcessPacket()
1747         if (link->node && type == CLUSTERMSG_TYPE_PONG) {  in clusterProcessPacket()
1748             link->node->pong_received = mstime();  in clusterProcessPacket()
1749             link->node->ping_sent = 0;  in clusterProcessPacket()
1757             if (nodeTimedOut(link->node)) {  in clusterProcessPacket()
1758                 link->node->flags &= ~CLUSTER_NODE_PFAIL;  in clusterProcessPacket()
1761             } else if (nodeFailed(link->node)) {  in clusterProcessPacket()
1762                 clearNodeFailureIfNeeded(link->node);  in clusterProcessPacket()
1860                         clusterSendUpdate(sender->link,  in clusterProcessPacket()
1882         if (sender) clusterProcessGossipSection(hdr,link);  in clusterProcessPacket()
1987 void handleLinkIOError(clusterLink *link) {  in handleLinkIOError()  argument
1988     freeClusterLink(link);  in handleLinkIOError()
1995     clusterLink *link = (clusterLink*) privdata;  in clusterWriteHandler()  local
2000     nwritten = write(fd, link->sndbuf, sdslen(link->sndbuf));  in clusterWriteHandler()
2004         handleLinkIOError(link);  in clusterWriteHandler()
2007     sdsrange(link->sndbuf,nwritten,-1);  in clusterWriteHandler()
2008     if (sdslen(link->sndbuf) == 0)  in clusterWriteHandler()
2009         aeDeleteFileEvent(server.el, link->fd, AE_WRITABLE);  in clusterWriteHandler()
2019     clusterLink *link = (clusterLink*) privdata;  in clusterReadHandler()  local
2025         rcvbuflen = sdslen(link->rcvbuf);  in clusterReadHandler()
2032             hdr = (clusterMsg*) link->rcvbuf;  in clusterReadHandler()
2042                     handleLinkIOError(link);  in clusterReadHandler()
2057             handleLinkIOError(link);  in clusterReadHandler()
2061             link->rcvbuf = sdscatlen(link->rcvbuf,buf,nread);  in clusterReadHandler()
2062             hdr = (clusterMsg*) link->rcvbuf;  in clusterReadHandler()
2068             if (clusterProcessPacket(link)) {  in clusterReadHandler()
2069                 sdsfree(link->rcvbuf);  in clusterReadHandler()
2070                 link->rcvbuf = sdsempty();  in clusterReadHandler()
2083 void clusterSendMessage(clusterLink *link, unsigned char *msg, size_t msglen) {  in clusterSendMessage()  argument
2084     if (sdslen(link->sndbuf) == 0 && msglen != 0)  in clusterSendMessage()
2085         aeCreateFileEvent(server.el,link->fd,AE_WRITABLE,  in clusterSendMessage()
2086                     clusterWriteHandler,link);  in clusterSendMessage()
2088     link->sndbuf = sdscatlen(link->sndbuf, msg, msglen);  in clusterSendMessage()
2106         if (!node->link) continue;  in clusterBroadcastMessage()
2109         clusterSendMessage(node->link,buf,len);  in clusterBroadcastMessage()
2175 void clusterSendPing(clusterLink *link, int type) {  in clusterSendPing()  argument
2229     if (link->node && type == CLUSTERMSG_TYPE_PING)  in clusterSendPing()
2230         link->node->ping_sent = mstime();  in clusterSendPing()
2256             (this->link == NULL && this->numslots == 0))  in clusterSendPing()
2289     clusterSendMessage(link,buf,totlen);  in clusterSendPing()
2317         if (!node->link) continue;  in clusterBroadcastPong()
2325         clusterSendPing(node->link,CLUSTERMSG_TYPE_PONG);  in clusterBroadcastPong()
2333 void clusterSendPublish(clusterLink *link, robj *channel, robj *message) {  in clusterSendPublish()  argument
2364     if (link)  in clusterSendPublish()
2365         clusterSendMessage(link,payload,totlen);  in clusterSendPublish()
2391 void clusterSendUpdate(clusterLink *link, clusterNode *node) {  in clusterSendUpdate()  argument
2395     if (link == NULL) return;  in clusterSendUpdate()
2400     clusterSendMessage(link,buf,ntohl(hdr->totlen));  in clusterSendUpdate()
2445     if (!node->link) return;  in clusterSendFailoverAuth()
2449     clusterSendMessage(node->link,buf,totlen);  in clusterSendFailoverAuth()
2458     if (!node->link) return;  in clusterSendMFStart()
2462     clusterSendMessage(node->link,buf,totlen);  in clusterSendMFStart()
3097         if (node->link == NULL) {  in clusterCron()
3100             clusterLink *link;  in clusterCron()  local
3117             link = createClusterLink(node);  in clusterCron()
3118             link->fd = fd;  in clusterCron()
3119             node->link = link;  in clusterCron()
3120             aeCreateFileEvent(server.el,link->fd,AE_READABLE,  in clusterCron()
3121                     clusterReadHandler,link);  in clusterCron()
3129             clusterSendPing(link, node->flags & CLUSTER_NODE_MEET ?  in clusterCron()
3162             if (this->link == NULL || this->ping_sent != 0) continue;  in clusterCron()
3172             clusterSendPing(min_pong_node->link, CLUSTERMSG_TYPE_PING);  in clusterCron()
3216         if (node->link && /* is connected */  in clusterCron()
3217             now - node->link->ctime >  in clusterCron()
3225             freeClusterLink(node->link);  in clusterCron()
3232         if (node->link &&  in clusterCron()
3236             clusterSendPing(node->link, CLUSTERMSG_TYPE_PING);  in clusterCron()
3245             node->link)  in clusterCron()
3247             clusterSendPing(node->link, CLUSTERMSG_TYPE_PING);  in clusterCron()
3719         (node->link || node->flags & CLUSTER_NODE_MYSELF) ?  in clusterGenNodeDescription()
4294                     myself->slaveof->link == NULL))  in clusterCommand()