Lines Matching refs:node
51 int clusterAddNode(clusterNode *node);
56 void clusterSendFailoverAuthIfNeeded(clusterNode *node, clusterMsg *request);
64 int clusterDelNodeSlots(clusterNode *node);
71 void clusterSendUpdate(clusterLink *link, clusterNode *node);
511 clusterNode *node = dictGetVal(de); in clusterReset() local
513 if (node == myself) continue; in clusterReset()
514 clusterDelNode(node); in clusterReset()
547 clusterLink *createClusterLink(clusterNode *node) { in createClusterLink() argument
552 link->node = node; in createClusterLink()
567 if (link->node) in freeClusterLink()
568 link->node->link = NULL; in freeClusterLink()
654 clusterNode *node = zmalloc(sizeof(*node)); in createClusterNode() local
657 memcpy(node->name, nodename, CLUSTER_NAMELEN); in createClusterNode()
659 getRandomHexChars(node->name, CLUSTER_NAMELEN); in createClusterNode()
660 node->ctime = mstime(); in createClusterNode()
661 node->configEpoch = 0; in createClusterNode()
662 node->flags = flags; in createClusterNode()
663 memset(node->slots,0,sizeof(node->slots)); in createClusterNode()
664 node->numslots = 0; in createClusterNode()
665 node->numslaves = 0; in createClusterNode()
666 node->slaves = NULL; in createClusterNode()
667 node->slaveof = NULL; in createClusterNode()
668 node->ping_sent = node->pong_received = 0; in createClusterNode()
669 node->fail_time = 0; in createClusterNode()
670 node->link = NULL; in createClusterNode()
671 memset(node->ip,0,sizeof(node->ip)); in createClusterNode()
672 node->port = 0; in createClusterNode()
673 node->fail_reports = listCreate(); in createClusterNode()
674 node->voted_time = 0; in createClusterNode()
675 node->orphaned_time = 0; in createClusterNode()
676 node->repl_offset_time = 0; in createClusterNode()
677 node->repl_offset = 0; in createClusterNode()
678 listSetFreeMethod(node->fail_reports,zfree); in createClusterNode()
679 return node; in createClusterNode()
703 if (fr->node == sender) { in clusterNodeAddFailureReport()
711 fr->node = sender; in clusterNodeAddFailureReport()
722 void clusterNodeCleanupFailureReports(clusterNode *node) { in clusterNodeCleanupFailureReports() argument
723 list *l = node->fail_reports; in clusterNodeCleanupFailureReports()
749 int clusterNodeDelFailureReport(clusterNode *node, clusterNode *sender) { in clusterNodeDelFailureReport() argument
750 list *l = node->fail_reports; in clusterNodeDelFailureReport()
759 if (fr->node == sender) break; in clusterNodeDelFailureReport()
765 clusterNodeCleanupFailureReports(node); in clusterNodeDelFailureReport()
772 int clusterNodeFailureReportsCount(clusterNode *node) { in clusterNodeFailureReportsCount() argument
773 clusterNodeCleanupFailureReports(node); in clusterNodeFailureReportsCount()
774 return listLength(node->fail_reports); in clusterNodeFailureReportsCount()
844 int clusterAddNode(clusterNode *node) { in clusterAddNode() argument
848 sdsnewlen(node->name,CLUSTER_NAMELEN), node); in clusterAddNode()
881 clusterNode *node = dictGetVal(de); in clusterDelNode() local
883 if (node == delnode) continue; in clusterDelNode()
884 clusterNodeDelFailureReport(node,delnode); in clusterDelNode()
907 void clusterRenameNode(clusterNode *node, char *newname) { in clusterRenameNode() argument
909 sds s = sdsnewlen(node->name, CLUSTER_NAMELEN); in clusterRenameNode()
912 node->name, newname); in clusterRenameNode()
916 memcpy(node->name, newname, CLUSTER_NAMELEN); in clusterRenameNode()
917 clusterAddNode(node); in clusterRenameNode()
933 clusterNode *node = dictGetVal(de); in clusterGetMaxEpoch() local
934 if (node->configEpoch > max) max = node->configEpoch; in clusterGetMaxEpoch()
1098 void clusterBlacklistAddNode(clusterNode *node) { in clusterBlacklistAddNode() argument
1100 sds id = sdsnewlen(node->name,CLUSTER_NAMELEN); in clusterBlacklistAddNode()
1151 void markNodeAsFailingIfNeeded(clusterNode *node) { in markNodeAsFailingIfNeeded() argument
1155 if (!nodeTimedOut(node)) return; /* We can reach it. */ in markNodeAsFailingIfNeeded()
1156 if (nodeFailed(node)) return; /* Already FAILing. */ in markNodeAsFailingIfNeeded()
1158 failures = clusterNodeFailureReportsCount(node); in markNodeAsFailingIfNeeded()
1164 "Marking node %.40s as failing (quorum reached).", node->name); in markNodeAsFailingIfNeeded()
1167 node->flags &= ~CLUSTER_NODE_PFAIL; in markNodeAsFailingIfNeeded()
1168 node->flags |= CLUSTER_NODE_FAIL; in markNodeAsFailingIfNeeded()
1169 node->fail_time = mstime(); in markNodeAsFailingIfNeeded()
1173 if (nodeIsMaster(myself)) clusterSendFail(node->name); in markNodeAsFailingIfNeeded()
1180 void clearNodeFailureIfNeeded(clusterNode *node) { in clearNodeFailureIfNeeded() argument
1183 serverAssert(nodeFailed(node)); in clearNodeFailureIfNeeded()
1187 if (nodeIsSlave(node) || node->numslots == 0) { in clearNodeFailureIfNeeded()
1190 node->name, in clearNodeFailureIfNeeded()
1191 nodeIsSlave(node) ? "slave" : "master without slots"); in clearNodeFailureIfNeeded()
1192 node->flags &= ~CLUSTER_NODE_FAIL; in clearNodeFailureIfNeeded()
1200 if (nodeIsMaster(node) && node->numslots > 0 && in clearNodeFailureIfNeeded()
1201 (now - node->fail_time) > in clearNodeFailureIfNeeded()
1206 node->name); in clearNodeFailureIfNeeded()
1207 node->flags &= ~CLUSTER_NODE_FAIL; in clearNodeFailureIfNeeded()
1221 clusterNode *node = dictGetVal(de); in clusterHandshakeInProgress() local
1223 if (!nodeInHandshake(node)) continue; in clusterHandshakeInProgress()
1224 if (!strcasecmp(node->ip,ip) && node->port == port) break; in clusterHandshakeInProgress()
1296 clusterNode *sender = link->node ? link->node : clusterLookupNode(hdr->sender); in clusterProcessGossipSection()
1300 clusterNode *node; in clusterProcessGossipSection() local
1312 node = clusterLookupNode(g->nodename); in clusterProcessGossipSection()
1313 if (node) { in clusterProcessGossipSection()
1316 if (sender && nodeIsMaster(sender) && node != myself) { in clusterProcessGossipSection()
1318 if (clusterNodeAddFailureReport(node,sender)) { in clusterProcessGossipSection()
1321 sender->name, node->name); in clusterProcessGossipSection()
1323 markNodeAsFailingIfNeeded(node); in clusterProcessGossipSection()
1325 if (clusterNodeDelFailureReport(node,sender)) { in clusterProcessGossipSection()
1328 sender->name, node->name); in clusterProcessGossipSection()
1338 if (node->flags & (CLUSTER_NODE_FAIL|CLUSTER_NODE_PFAIL) && in clusterProcessGossipSection()
1341 (strcasecmp(node->ip,g->ip) || node->port != ntohs(g->port))) in clusterProcessGossipSection()
1343 if (node->link) freeClusterLink(node->link); in clusterProcessGossipSection()
1344 memcpy(node->ip,g->ip,NET_IP_STR_LEN); in clusterProcessGossipSection()
1345 node->port = ntohs(g->port); in clusterProcessGossipSection()
1346 node->flags &= ~CLUSTER_NODE_NOADDR; in clusterProcessGossipSection()
1383 int nodeUpdateAddressIfNeeded(clusterNode *node, clusterLink *link, int port) { in nodeUpdateAddressIfNeeded() argument
1392 if (link == node->link) return 0; in nodeUpdateAddressIfNeeded()
1395 if (node->port == port && strcmp(ip,node->ip) == 0) return 0; in nodeUpdateAddressIfNeeded()
1398 memcpy(node->ip,ip,sizeof(ip)); in nodeUpdateAddressIfNeeded()
1399 node->port = port; in nodeUpdateAddressIfNeeded()
1400 if (node->link) freeClusterLink(node->link); in nodeUpdateAddressIfNeeded()
1401 node->flags &= ~CLUSTER_NODE_NOADDR; in nodeUpdateAddressIfNeeded()
1403 node->name, node->ip, node->port); in nodeUpdateAddressIfNeeded()
1407 if (nodeIsSlave(myself) && myself->slaveof == node) in nodeUpdateAddressIfNeeded()
1408 replicationSetMaster(node->ip, node->port); in nodeUpdateAddressIfNeeded()
1634 serverLog(LL_DEBUG,"Ping packet received: %p", (void*)link->node); in clusterProcessPacket()
1665 clusterNode *node; in clusterProcessPacket() local
1667 node = createClusterNode(NULL,CLUSTER_NODE_HANDSHAKE); in clusterProcessPacket()
1668 nodeIp2String(node->ip,link); in clusterProcessPacket()
1669 node->port = ntohs(hdr->port); in clusterProcessPacket()
1670 clusterAddNode(node); in clusterProcessPacket()
1690 (void*)link->node); in clusterProcessPacket()
1691 if (link->node) { in clusterProcessPacket()
1692 if (nodeInHandshake(link->node)) { 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()
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()
2104 clusterNode *node = dictGetVal(de); in clusterBroadcastMessage() local
2106 if (!node->link) continue; in clusterBroadcastMessage()
2107 if (node->flags & (CLUSTER_NODE_MYSELF|CLUSTER_NODE_HANDSHAKE)) in clusterBroadcastMessage()
2109 clusterSendMessage(node->link,buf,len); in clusterBroadcastMessage()
2229 if (link->node && type == CLUSTERMSG_TYPE_PING) in clusterSendPing()
2230 link->node->ping_sent = mstime(); in clusterSendPing()
2315 clusterNode *node = dictGetVal(de); in clusterBroadcastPong() local
2317 if (!node->link) continue; in clusterBroadcastPong()
2318 if (node == myself || nodeInHandshake(node)) continue; in clusterBroadcastPong()
2321 nodeIsSlave(node) && node->slaveof && in clusterBroadcastPong()
2322 (node->slaveof == myself || node->slaveof == myself->slaveof); in clusterBroadcastPong()
2325 clusterSendPing(node->link,CLUSTERMSG_TYPE_PONG); in clusterBroadcastPong()
2391 void clusterSendUpdate(clusterLink *link, clusterNode *node) { in clusterSendUpdate() argument
2397 memcpy(hdr->data.update.nodecfg.nodename,node->name,CLUSTER_NAMELEN); in clusterSendUpdate()
2398 hdr->data.update.nodecfg.configEpoch = htonu64(node->configEpoch); in clusterSendUpdate()
2399 memcpy(hdr->data.update.nodecfg.slots,node->slots,sizeof(node->slots)); in clusterSendUpdate()
2440 void clusterSendFailoverAuth(clusterNode *node) { in clusterSendFailoverAuth() argument
2445 if (!node->link) return; in clusterSendFailoverAuth()
2449 clusterSendMessage(node->link,buf,totlen); in clusterSendFailoverAuth()
2453 void clusterSendMFStart(clusterNode *node) { in clusterSendMFStart() argument
2458 if (!node->link) return; in clusterSendMFStart()
2462 clusterSendMessage(node->link,buf,totlen); in clusterSendMFStart()
2466 void clusterSendFailoverAuthIfNeeded(clusterNode *node, clusterMsg *request) { in clusterSendFailoverAuthIfNeeded() argument
2467 clusterNode *master = node->slaveof; in clusterSendFailoverAuthIfNeeded()
2487 node->name, in clusterSendFailoverAuthIfNeeded()
2497 node->name, in clusterSendFailoverAuthIfNeeded()
2505 if (nodeIsMaster(node) || master == NULL || in clusterSendFailoverAuthIfNeeded()
2508 if (nodeIsMaster(node)) { in clusterSendFailoverAuthIfNeeded()
2511 node->name); in clusterSendFailoverAuthIfNeeded()
2515 node->name); in clusterSendFailoverAuthIfNeeded()
2519 node->name); in clusterSendFailoverAuthIfNeeded()
2527 if (mstime() - node->slaveof->voted_time < server.cluster_node_timeout * 2) in clusterSendFailoverAuthIfNeeded()
2532 node->name, in clusterSendFailoverAuthIfNeeded()
2534 (mstime() - node->slaveof->voted_time))); in clusterSendFailoverAuthIfNeeded()
2554 node->name, j, in clusterSendFailoverAuthIfNeeded()
2561 clusterSendFailoverAuth(node); in clusterSendFailoverAuthIfNeeded()
2563 node->slaveof->voted_time = mstime(); in clusterSendFailoverAuthIfNeeded()
2565 node->name, (unsigned long long) server.cluster->currentEpoch); in clusterSendFailoverAuthIfNeeded()
2928 clusterNode *node = dictGetVal(de); in clusterHandleSlaveMigration() local
2935 if (nodeIsSlave(node) || nodeFailed(node)) is_orphaned = 0; in clusterHandleSlaveMigration()
2936 if (!(node->flags & CLUSTER_NODE_MIGRATE_TO)) is_orphaned = 0; in clusterHandleSlaveMigration()
2939 if (nodeIsMaster(node)) okslaves = clusterCountNonFailingSlaves(node); in clusterHandleSlaveMigration()
2943 if (!target && node->numslots > 0) target = node; in clusterHandleSlaveMigration()
2947 if (!node->orphaned_time) node->orphaned_time = mstime(); in clusterHandleSlaveMigration()
2949 node->orphaned_time = 0; in clusterHandleSlaveMigration()
2956 for (j = 0; j < node->numslaves; j++) { in clusterHandleSlaveMigration()
2957 if (memcmp(node->slaves[j]->name, in clusterHandleSlaveMigration()
2961 candidate = node->slaves[j]; in clusterHandleSlaveMigration()
3086 clusterNode *node = dictGetVal(de); in clusterCron() local
3088 if (node->flags & (CLUSTER_NODE_MYSELF|CLUSTER_NODE_NOADDR)) continue; in clusterCron()
3092 if (nodeInHandshake(node) && now - node->ctime > handshake_timeout) { in clusterCron()
3093 clusterDelNode(node); in clusterCron()
3097 if (node->link == NULL) { in clusterCron()
3102 fd = anetTcpNonBlockBindConnect(server.neterr, node->ip, in clusterCron()
3103 node->port+CLUSTER_PORT_INCR, NET_FIRST_BIND_ADDR); in clusterCron()
3110 if (node->ping_sent == 0) node->ping_sent = mstime(); in clusterCron()
3112 "Cluster Node [%s]:%d -> %s", node->ip, in clusterCron()
3113 node->port+CLUSTER_PORT_INCR, in clusterCron()
3117 link = createClusterLink(node); in clusterCron()
3119 node->link = link; in clusterCron()
3128 old_ping_sent = node->ping_sent; in clusterCron()
3129 clusterSendPing(link, node->flags & CLUSTER_NODE_MEET ? in clusterCron()
3135 node->ping_sent = old_ping_sent; in clusterCron()
3142 node->flags &= ~CLUSTER_NODE_MEET; in clusterCron()
3145 node->name, node->ip, node->port+CLUSTER_PORT_INCR); in clusterCron()
3187 clusterNode *node = dictGetVal(de); in clusterCron() local
3191 if (node->flags & in clusterCron()
3197 if (nodeIsSlave(myself) && nodeIsMaster(node) && !nodeFailed(node)) { in clusterCron()
3198 int okslaves = clusterCountNonFailingSlaves(node); in clusterCron()
3203 if (okslaves == 0 && node->numslots > 0 && in clusterCron()
3204 node->flags & CLUSTER_NODE_MIGRATE_TO) in clusterCron()
3209 if (nodeIsSlave(myself) && myself->slaveof == node) in clusterCron()
3216 if (node->link && /* is connected */ in clusterCron()
3217 now - node->link->ctime > in clusterCron()
3219 node->ping_sent && /* we already sent a ping */ in clusterCron()
3220 node->pong_received < node->ping_sent && /* still waiting pong */ in clusterCron()
3222 now - node->ping_sent > server.cluster_node_timeout/2) in clusterCron()
3225 freeClusterLink(node->link); in clusterCron()
3232 if (node->link && in clusterCron()
3233 node->ping_sent == 0 && in clusterCron()
3234 (now - node->pong_received) > server.cluster_node_timeout/2) in clusterCron()
3236 clusterSendPing(node->link, CLUSTERMSG_TYPE_PING); in clusterCron()
3244 server.cluster->mf_slave == node && in clusterCron()
3245 node->link) in clusterCron()
3247 clusterSendPing(node->link, CLUSTERMSG_TYPE_PING); in clusterCron()
3252 if (node->ping_sent == 0) continue; in clusterCron()
3257 delay = now - node->ping_sent; in clusterCron()
3262 if (!(node->flags & (CLUSTER_NODE_PFAIL|CLUSTER_NODE_FAIL))) { in clusterCron()
3264 node->name); in clusterCron()
3265 node->flags |= CLUSTER_NODE_PFAIL; in clusterCron()
3367 clusterNode *node = dictGetVal(de); in clusterMastersHaveSlaves() local
3369 if (nodeIsSlave(node)) continue; in clusterMastersHaveSlaves()
3370 slaves += node->numslaves; in clusterMastersHaveSlaves()
3439 int clusterDelNodeSlots(clusterNode *node) { in clusterDelNodeSlots() argument
3443 if (clusterNodeGetSlotBit(node,j)) clusterDelSlot(j); in clusterDelNodeSlots()
3517 clusterNode *node = dictGetVal(de); in clusterUpdateState() local
3519 if (nodeIsMaster(node) && node->numslots) { in clusterUpdateState()
3521 if ((node->flags & (CLUSTER_NODE_FAIL|CLUSTER_NODE_PFAIL)) == 0) in clusterUpdateState()
3695 sds clusterGenNodeDescription(clusterNode *node) { in clusterGenNodeDescription() argument
3701 node->name, in clusterGenNodeDescription()
3702 node->ip, in clusterGenNodeDescription()
3703 node->port); in clusterGenNodeDescription()
3706 ci = representClusterNodeFlags(ci, node->flags); in clusterGenNodeDescription()
3709 if (node->slaveof) in clusterGenNodeDescription()
3710 ci = sdscatprintf(ci," %.40s ",node->slaveof->name); in clusterGenNodeDescription()
3716 (long long) node->ping_sent, in clusterGenNodeDescription()
3717 (long long) node->pong_received, in clusterGenNodeDescription()
3718 (unsigned long long) node->configEpoch, in clusterGenNodeDescription()
3719 (node->link || node->flags & CLUSTER_NODE_MYSELF) ? in clusterGenNodeDescription()
3727 if ((bit = clusterNodeGetSlotBit(node,j)) != 0) { in clusterGenNodeDescription()
3745 if (node->flags & CLUSTER_NODE_MYSELF) { in clusterGenNodeDescription()
3778 clusterNode *node = dictGetVal(de); in clusterGenNodesDescription() local
3780 if (node->flags & filter) continue; in clusterGenNodesDescription()
3781 ni = clusterGenNodeDescription(node); in clusterGenNodesDescription()
3824 clusterNode *node = dictGetVal(de); in clusterReplyMultiBulkSlots() local
3829 if (!nodeIsMaster(node) || node->numslots == 0) continue; in clusterReplyMultiBulkSlots()
3834 if ((bit = clusterNodeGetSlotBit(node,j)) != 0) { in clusterReplyMultiBulkSlots()
3856 addReplyBulkCString(c, node->ip); in clusterReplyMultiBulkSlots()
3857 addReplyLongLong(c, node->port); in clusterReplyMultiBulkSlots()
3858 addReplyBulkCBuffer(c, node->name, CLUSTER_NAMELEN); in clusterReplyMultiBulkSlots()
3861 for (i = 0; i < node->numslaves; i++) { in clusterReplyMultiBulkSlots()
3864 if (nodeFailed(node->slaves[i])) continue; in clusterReplyMultiBulkSlots()
3866 addReplyBulkCString(c, node->slaves[i]->ip); in clusterReplyMultiBulkSlots()
3867 addReplyLongLong(c, node->slaves[i]->port); in clusterReplyMultiBulkSlots()
3868 addReplyBulkCBuffer(c, node->slaves[i]->name, CLUSTER_NAMELEN); in clusterReplyMultiBulkSlots()
5194 clusterNode *node = server.cluster->slots[slot]; in clusterRedirectBlockedClientIfNeeded() local
5199 if (node != myself && in clusterRedirectBlockedClientIfNeeded()
5202 if (node == NULL) { in clusterRedirectBlockedClientIfNeeded()
5206 clusterRedirectClient(c,node,slot, in clusterRedirectBlockedClientIfNeeded()