Lines Matching refs:cluster
143 server.cluster->currentEpoch = in clusterLoadConfig()
146 server.cluster->lastVoteEpoch = in clusterLoadConfig()
179 serverAssert(server.cluster->myself == NULL); in clusterLoadConfig()
180 myself = server.cluster->myself = n; in clusterLoadConfig()
244 server.cluster->migrating_slots_to[slot] = cn; in clusterLoadConfig()
246 server.cluster->importing_slots_from[slot] = cn; in clusterLoadConfig()
262 if (server.cluster->myself == NULL) goto fmterr; in clusterLoadConfig()
272 if (clusterGetMaxEpoch() > server.cluster->currentEpoch) { in clusterLoadConfig()
273 server.cluster->currentEpoch = clusterGetMaxEpoch(); in clusterLoadConfig()
303 server.cluster->todo_before_sleep &= ~CLUSTER_TODO_SAVE_CONFIG; in clusterSaveConfig()
309 (unsigned long long) server.cluster->currentEpoch, in clusterSaveConfig()
310 (unsigned long long) server.cluster->lastVoteEpoch); in clusterSaveConfig()
325 server.cluster->todo_before_sleep &= ~CLUSTER_TODO_FSYNC_CONFIG; in clusterSaveConfig()
401 server.cluster = zmalloc(sizeof(clusterState)); in clusterInit()
402 server.cluster->myself = NULL; in clusterInit()
403 server.cluster->currentEpoch = 0; in clusterInit()
404 server.cluster->state = CLUSTER_FAIL; in clusterInit()
405 server.cluster->size = 1; in clusterInit()
406 server.cluster->todo_before_sleep = 0; in clusterInit()
407 server.cluster->nodes = dictCreate(&clusterNodesDictType,NULL); in clusterInit()
408 server.cluster->nodes_black_list = in clusterInit()
410 server.cluster->failover_auth_time = 0; in clusterInit()
411 server.cluster->failover_auth_count = 0; in clusterInit()
412 server.cluster->failover_auth_rank = 0; in clusterInit()
413 server.cluster->failover_auth_epoch = 0; in clusterInit()
414 server.cluster->cant_failover_reason = CLUSTER_CANT_FAILOVER_NONE; in clusterInit()
415 server.cluster->lastVoteEpoch = 0; in clusterInit()
416 server.cluster->stats_bus_messages_sent = 0; in clusterInit()
417 server.cluster->stats_bus_messages_received = 0; in clusterInit()
418 memset(server.cluster->slots,0, sizeof(server.cluster->slots)); in clusterInit()
430 myself = server.cluster->myself = in clusterInit()
470 server.cluster->slots_to_keys = zslCreate(); in clusterInit()
476 server.cluster->mf_end = 0; in clusterInit()
509 di = dictGetSafeIterator(server.cluster->nodes); in clusterReset()
522 server.cluster->currentEpoch = 0; in clusterReset()
523 server.cluster->lastVoteEpoch = 0; in clusterReset()
530 dictDelete(server.cluster->nodes,oldname); in clusterReset()
833 serverAssert(dictDelete(server.cluster->nodes,nodename) == DICT_OK); in freeClusterNode()
847 retval = dictAdd(server.cluster->nodes, in clusterAddNode()
870 if (server.cluster->importing_slots_from[j] == delnode) in clusterDelNode()
871 server.cluster->importing_slots_from[j] = NULL; in clusterDelNode()
872 if (server.cluster->migrating_slots_to[j] == delnode) in clusterDelNode()
873 server.cluster->migrating_slots_to[j] = NULL; in clusterDelNode()
874 if (server.cluster->slots[j] == delnode) in clusterDelNode()
879 di = dictGetSafeIterator(server.cluster->nodes); in clusterDelNode()
897 de = dictFind(server.cluster->nodes,s); in clusterLookupNode()
913 retval = dictDelete(server.cluster->nodes, s); in clusterRenameNode()
931 di = dictGetSafeIterator(server.cluster->nodes); in clusterGetMaxEpoch()
937 if (max < server.cluster->currentEpoch) max = server.cluster->currentEpoch; in clusterGetMaxEpoch()
976 server.cluster->currentEpoch++; in clusterBumpConfigEpochWithoutConsensus()
977 myself->configEpoch = server.cluster->currentEpoch; in clusterBumpConfigEpochWithoutConsensus()
1042 server.cluster->currentEpoch++; in clusterHandleConfigEpochCollision()
1043 myself->configEpoch = server.cluster->currentEpoch; in clusterHandleConfigEpochCollision()
1087 di = dictGetSafeIterator(server.cluster->nodes_black_list); in clusterBlacklistCleanup()
1092 dictDelete(server.cluster->nodes_black_list,dictGetKey(de)); in clusterBlacklistCleanup()
1103 if (dictAdd(server.cluster->nodes_black_list,id,NULL) == DICT_OK) { in clusterBlacklistAddNode()
1108 de = dictFind(server.cluster->nodes_black_list,id); in clusterBlacklistAddNode()
1121 retval = dictFind(server.cluster->nodes_black_list,id) != NULL; in clusterBlacklistExists()
1153 int needed_quorum = (server.cluster->size / 2) + 1; in markNodeAsFailingIfNeeded()
1219 di = dictGetSafeIterator(server.cluster->nodes); in clusterHandshakeInProgress()
1468 if (server.cluster->slots[j] == sender) continue; in clusterUpdateSlotsConfigWith()
1474 if (server.cluster->importing_slots_from[j]) continue; in clusterUpdateSlotsConfigWith()
1480 if (server.cluster->slots[j] == NULL || in clusterUpdateSlotsConfigWith()
1481 server.cluster->slots[j]->configEpoch < senderConfigEpoch) in clusterUpdateSlotsConfigWith()
1485 if (server.cluster->slots[j] == myself && in clusterUpdateSlotsConfigWith()
1493 if (server.cluster->slots[j] == curmaster) in clusterUpdateSlotsConfigWith()
1546 server.cluster->stats_bus_messages_received++; in clusterProcessPacket()
1605 if (senderCurrentEpoch > server.cluster->currentEpoch) in clusterProcessPacket()
1606 server.cluster->currentEpoch = senderCurrentEpoch; in clusterProcessPacket()
1618 if (server.cluster->mf_end && in clusterProcessPacket()
1622 server.cluster->mf_master_offset == 0) in clusterProcessPacket()
1624 server.cluster->mf_master_offset = sender->repl_offset; in clusterProcessPacket()
1628 server.cluster->mf_master_offset); in clusterProcessPacket()
1851 if (server.cluster->slots[j] == sender || in clusterProcessPacket()
1852 server.cluster->slots[j] == NULL) continue; in clusterProcessPacket()
1853 if (server.cluster->slots[j]->configEpoch > in clusterProcessPacket()
1859 sender->name, server.cluster->slots[j]->name); in clusterProcessPacket()
1861 server.cluster->slots[j]); in clusterProcessPacket()
1934 senderCurrentEpoch >= server.cluster->failover_auth_epoch) in clusterProcessPacket()
1936 server.cluster->failover_auth_count++; in clusterProcessPacket()
1948 server.cluster->mf_end = mstime() + CLUSTER_MF_TIMEOUT; in clusterProcessPacket()
1949 server.cluster->mf_slave = sender; in clusterProcessPacket()
2089 server.cluster->stats_bus_messages_sent++; in clusterSendMessage()
2102 di = dictGetSafeIterator(server.cluster->nodes); in clusterBroadcastMessage()
2143 hdr->state = server.cluster->state; in clusterBuildMessageHdr()
2146 hdr->currentEpoch = htonu64(server.cluster->currentEpoch); in clusterBuildMessageHdr()
2157 if (nodeIsMaster(myself) && server.cluster->mf_end) in clusterBuildMessageHdr()
2185 int freshnodes = dictSize(server.cluster->nodes)-2; in clusterSendPing()
2213 wanted = floor(dictSize(server.cluster->nodes)/10); in clusterSendPing()
2236 dictEntry *de = dictGetRandomKey(server.cluster->nodes); in clusterSendPing()
2313 di = dictGetSafeIterator(server.cluster->nodes); in clusterBroadcastPong()
2433 if (server.cluster->mf_end) hdr->mflags[0] |= CLUSTERMSG_FLAG0_FORCEACK; in clusterRequestFailoverAuth()
2484 if (requestCurrentEpoch < server.cluster->currentEpoch) { in clusterSendFailoverAuthIfNeeded()
2489 (unsigned long long) server.cluster->currentEpoch); in clusterSendFailoverAuthIfNeeded()
2494 if (server.cluster->lastVoteEpoch == server.cluster->currentEpoch) { in clusterSendFailoverAuthIfNeeded()
2498 (unsigned long long) server.cluster->currentEpoch); in clusterSendFailoverAuthIfNeeded()
2543 if (server.cluster->slots[j] == NULL || in clusterSendFailoverAuthIfNeeded()
2544 server.cluster->slots[j]->configEpoch <= requestConfigEpoch) in clusterSendFailoverAuthIfNeeded()
2555 (unsigned long long) server.cluster->slots[j]->configEpoch, in clusterSendFailoverAuthIfNeeded()
2562 server.cluster->lastVoteEpoch = server.cluster->currentEpoch; in clusterSendFailoverAuthIfNeeded()
2565 node->name, (unsigned long long) server.cluster->currentEpoch); in clusterSendFailoverAuthIfNeeded()
2624 if (reason == server.cluster->cant_failover_reason && in clusterLogCantFailover()
2628 server.cluster->cant_failover_reason = reason; in clusterLogCantFailover()
2706 mstime_t auth_age = mstime() - server.cluster->failover_auth_time; in clusterHandleSlaveFailover()
2707 int needed_quorum = (server.cluster->size / 2) + 1; in clusterHandleSlaveFailover()
2708 int manual_failover = server.cluster->mf_end != 0 && in clusterHandleSlaveFailover()
2709 server.cluster->mf_can_start; in clusterHandleSlaveFailover()
2712 server.cluster->todo_before_sleep &= ~CLUSTER_TODO_HANDLE_FAILOVER; in clusterHandleSlaveFailover()
2737 server.cluster->cant_failover_reason = CLUSTER_CANT_FAILOVER_NONE; in clusterHandleSlaveFailover()
2774 server.cluster->failover_auth_time = mstime() + in clusterHandleSlaveFailover()
2777 server.cluster->failover_auth_count = 0; in clusterHandleSlaveFailover()
2778 server.cluster->failover_auth_sent = 0; in clusterHandleSlaveFailover()
2779 server.cluster->failover_auth_rank = clusterGetSlaveRank(); in clusterHandleSlaveFailover()
2783 server.cluster->failover_auth_time += in clusterHandleSlaveFailover()
2784 server.cluster->failover_auth_rank * 1000; in clusterHandleSlaveFailover()
2786 if (server.cluster->mf_end) { in clusterHandleSlaveFailover()
2787 server.cluster->failover_auth_time = mstime(); in clusterHandleSlaveFailover()
2788 server.cluster->failover_auth_rank = 0; in clusterHandleSlaveFailover()
2793 server.cluster->failover_auth_time - mstime(), in clusterHandleSlaveFailover()
2794 server.cluster->failover_auth_rank, in clusterHandleSlaveFailover()
2808 if (server.cluster->failover_auth_sent == 0 && in clusterHandleSlaveFailover()
2809 server.cluster->mf_end == 0) in clusterHandleSlaveFailover()
2812 if (newrank > server.cluster->failover_auth_rank) { in clusterHandleSlaveFailover()
2814 (newrank - server.cluster->failover_auth_rank) * 1000; in clusterHandleSlaveFailover()
2815 server.cluster->failover_auth_time += added_delay; in clusterHandleSlaveFailover()
2816 server.cluster->failover_auth_rank = newrank; in clusterHandleSlaveFailover()
2824 if (mstime() < server.cluster->failover_auth_time) { in clusterHandleSlaveFailover()
2836 if (server.cluster->failover_auth_sent == 0) { in clusterHandleSlaveFailover()
2837 server.cluster->currentEpoch++; in clusterHandleSlaveFailover()
2838 server.cluster->failover_auth_epoch = server.cluster->currentEpoch; in clusterHandleSlaveFailover()
2840 (unsigned long long) server.cluster->currentEpoch); in clusterHandleSlaveFailover()
2842 server.cluster->failover_auth_sent = 1; in clusterHandleSlaveFailover()
2850 if (server.cluster->failover_auth_count >= needed_quorum) { in clusterHandleSlaveFailover()
2857 if (myself->configEpoch < server.cluster->failover_auth_epoch) { in clusterHandleSlaveFailover()
2858 myself->configEpoch = server.cluster->failover_auth_epoch; in clusterHandleSlaveFailover()
2905 if (server.cluster->state != CLUSTER_OK) return; in clusterHandleSlaveMigration()
2926 di = dictGetSafeIterator(server.cluster->nodes); in clusterHandleSlaveMigration()
3017 if (server.cluster->mf_end && clientsArePaused()) { in resetManualFailover()
3021 server.cluster->mf_end = 0; /* No manual failover in progress. */ in resetManualFailover()
3022 server.cluster->mf_can_start = 0; in resetManualFailover()
3023 server.cluster->mf_slave = NULL; in resetManualFailover()
3024 server.cluster->mf_master_offset = 0; in resetManualFailover()
3029 if (server.cluster->mf_end && server.cluster->mf_end < mstime()) { in manualFailoverCheckTimeout()
3039 if (server.cluster->mf_end == 0) return; in clusterHandleManualFailover()
3043 if (server.cluster->mf_can_start) return; in clusterHandleManualFailover()
3045 if (server.cluster->mf_master_offset == 0) return; /* Wait for offset... */ in clusterHandleManualFailover()
3047 if (server.cluster->mf_master_offset == replicationGetSlaveOffset()) { in clusterHandleManualFailover()
3050 server.cluster->mf_can_start = 1; in clusterHandleManualFailover()
3084 di = dictGetSafeIterator(server.cluster->nodes); in clusterCron()
3158 de = dictGetRandomKey(server.cluster->nodes); in clusterCron()
3185 di = dictGetSafeIterator(server.cluster->nodes); in clusterCron()
3242 if (server.cluster->mf_end && in clusterCron()
3244 server.cluster->mf_slave == node && in clusterCron()
3298 if (update_state || server.cluster->state == CLUSTER_FAIL) in clusterCron()
3310 if (server.cluster->todo_before_sleep & CLUSTER_TODO_HANDLE_FAILOVER) in clusterBeforeSleep()
3314 if (server.cluster->todo_before_sleep & CLUSTER_TODO_UPDATE_STATE) in clusterBeforeSleep()
3318 if (server.cluster->todo_before_sleep & CLUSTER_TODO_SAVE_CONFIG) { in clusterBeforeSleep()
3319 int fsync = server.cluster->todo_before_sleep & in clusterBeforeSleep()
3326 server.cluster->todo_before_sleep = 0; in clusterBeforeSleep()
3330 server.cluster->todo_before_sleep |= flags; in clusterDoBeforeSleep()
3363 dictIterator *di = dictGetSafeIterator(server.cluster->nodes); in clusterMastersHaveSlaves()
3419 if (server.cluster->slots[slot]) return C_ERR; in clusterAddSlot()
3421 server.cluster->slots[slot] = n; in clusterAddSlot()
3429 clusterNode *n = server.cluster->slots[slot]; in clusterDelSlot()
3433 server.cluster->slots[slot] = NULL; in clusterDelSlot()
3452 memset(server.cluster->migrating_slots_to,0, in clusterCloseAllSlots()
3453 sizeof(server.cluster->migrating_slots_to)); in clusterCloseAllSlots()
3454 memset(server.cluster->importing_slots_from,0, in clusterCloseAllSlots()
3455 sizeof(server.cluster->importing_slots_from)); in clusterCloseAllSlots()
3476 server.cluster->todo_before_sleep &= ~CLUSTER_TODO_UPDATE_STATE; in clusterUpdateState()
3486 server.cluster->state == CLUSTER_FAIL && in clusterUpdateState()
3496 if (server.cluster->slots[j] == NULL || in clusterUpdateState()
3497 server.cluster->slots[j]->flags & (CLUSTER_NODE_FAIL)) in clusterUpdateState()
3514 server.cluster->size = 0; in clusterUpdateState()
3515 di = dictGetSafeIterator(server.cluster->nodes); in clusterUpdateState()
3520 server.cluster->size++; in clusterUpdateState()
3531 int needed_quorum = (server.cluster->size / 2) + 1; in clusterUpdateState()
3540 if (new_state != server.cluster->state) { in clusterUpdateState()
3562 server.cluster->state = new_state; in clusterUpdateState()
3608 if (server.cluster->slots[j] == myself || in verifyClusterConfigWithData()
3609 server.cluster->importing_slots_from[j] != NULL) continue; in verifyClusterConfigWithData()
3617 if (server.cluster->slots[j] == NULL) { in verifyClusterConfigWithData()
3625 server.cluster->importing_slots_from[j] = server.cluster->slots[j]; in verifyClusterConfigWithData()
3747 if (server.cluster->migrating_slots_to[j]) { in clusterGenNodeDescription()
3749 server.cluster->migrating_slots_to[j]->name); in clusterGenNodeDescription()
3750 } else if (server.cluster->importing_slots_from[j]) { in clusterGenNodeDescription()
3752 server.cluster->importing_slots_from[j]->name); in clusterGenNodeDescription()
3776 di = dictGetSafeIterator(server.cluster->nodes); in clusterGenNodesDescription()
3822 dictIterator *di = dictGetSafeIterator(server.cluster->nodes); in clusterReplyMultiBulkSlots()
3943 if (del && server.cluster->slots[slot] == NULL) { in clusterCommand()
3947 } else if (!del && server.cluster->slots[slot]) { in clusterCommand()
3965 if (server.cluster->importing_slots_from[j]) in clusterCommand()
3966 server.cluster->importing_slots_from[j] = NULL; in clusterCommand()
3992 if (server.cluster->slots[slot] != myself) { in clusterCommand()
4001 server.cluster->migrating_slots_to[slot] = n; in clusterCommand()
4003 if (server.cluster->slots[slot] == myself) { in clusterCommand()
4013 server.cluster->importing_slots_from[slot] = n; in clusterCommand()
4016 server.cluster->importing_slots_from[slot] = NULL; in clusterCommand()
4017 server.cluster->migrating_slots_to[slot] = NULL; in clusterCommand()
4029 if (server.cluster->slots[slot] == myself && n != myself) { in clusterCommand()
4041 server.cluster->migrating_slots_to[slot]) in clusterCommand()
4042 server.cluster->migrating_slots_to[slot] = NULL; in clusterCommand()
4047 server.cluster->importing_slots_from[slot]) in clusterCommand()
4062 server.cluster->importing_slots_from[slot] = NULL; in clusterCommand()
4088 clusterNode *n = server.cluster->slots[j]; in clusterCommand()
4116 , statestr[server.cluster->state], in clusterCommand()
4121 dictSize(server.cluster->nodes), in clusterCommand()
4122 server.cluster->size, in clusterCommand()
4123 (unsigned long long) server.cluster->currentEpoch, in clusterCommand()
4125 server.cluster->stats_bus_messages_sent, in clusterCommand()
4126 server.cluster->stats_bus_messages_received in clusterCommand()
4301 server.cluster->mf_end = mstime() + CLUSTER_MF_TIMEOUT; in clusterCommand()
4316 server.cluster->mf_can_start = 1; in clusterCommand()
4338 } else if (dictSize(server.cluster->nodes) > 1) { in clusterCommand()
4349 if (server.cluster->currentEpoch < (uint64_t)epoch) in clusterCommand()
4350 server.cluster->currentEpoch = epoch; in clusterCommand()
5026 n = server.cluster->slots[slot]; in getNodeByQuery()
5045 server.cluster->migrating_slots_to[slot] != NULL) in getNodeByQuery()
5048 } else if (server.cluster->importing_slots_from[slot] != NULL) { in getNodeByQuery()
5084 if (server.cluster->state != CLUSTER_OK) { in getNodeByQuery()
5102 return server.cluster->migrating_slots_to[slot]; in getNodeByQuery()
5185 if (server.cluster->state == CLUSTER_FAIL) { in clusterRedirectBlockedClientIfNeeded()
5194 clusterNode *node = server.cluster->slots[slot]; in clusterRedirectBlockedClientIfNeeded()
5200 server.cluster->importing_slots_from[slot] == NULL) in clusterRedirectBlockedClientIfNeeded()