Lines Matching refs:server
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()
313 if ((fd = open(server.cluster_configfile,O_WRONLY|O_CREAT,0644)) 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()
423 if (clusterLockConfig(server.cluster_configfile) == C_ERR) in clusterInit()
427 if (clusterLoadConfig(server.cluster_configfile) == C_ERR) { in clusterInit()
430 myself = server.cluster->myself = in clusterInit()
440 server.cfd_count = 0; in clusterInit()
445 if (server.port > (65535-CLUSTER_PORT_INCR)) { in clusterInit()
454 if (listenToPort(server.port+CLUSTER_PORT_INCR, in clusterInit()
455 server.cfd,&server.cfd_count) == C_ERR) in clusterInit()
461 for (j = 0; j < server.cfd_count; j++) { in clusterInit()
462 if (aeCreateFileEvent(server.el, server.cfd[j], AE_READABLE, in clusterInit()
470 server.cluster->slots_to_keys = zslCreate(); in clusterInit()
474 myself->port = server.port; 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()
562 aeDeleteFileEvent(server.el, link->fd, AE_WRITABLE); in freeClusterLink()
563 aeDeleteFileEvent(server.el, link->fd, AE_READABLE); in freeClusterLink()
585 if (server.masterhost == NULL && server.loading) return; in clusterAcceptHandler()
588 cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip), &cport); in clusterAcceptHandler()
592 "Error accepting cluster node: %s", server.neterr); in clusterAcceptHandler()
607 aeCreateFileEvent(server.el,cfd,AE_READABLE,clusterReadHandler,link); in clusterAcceptHandler()
727 mstime_t maxtime = server.cluster_node_timeout * in clusterNodeCleanupFailureReports()
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()
1091 if (expire < server.unixtime) 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()
1202 (server.cluster_node_timeout * CLUSTER_FAIL_UNDO_TIME_MULT)) in clearNodeFailureIfNeeded()
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()
1911 if (dictSize(server.pubsub_channels) || in clusterProcessPacket()
1912 listLength(server.pubsub_patterns)) 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()
2009 aeDeleteFileEvent(server.el, link->fd, AE_WRITABLE); in clusterWriteHandler()
2085 aeCreateFileEvent(server.el,link->fd,AE_WRITABLE, in clusterSendMessage()
2089 server.cluster->stats_bus_messages_sent++; in clusterSendMessage()
2102 di = dictGetSafeIterator(server.cluster->nodes); in clusterBroadcastMessage()
2141 hdr->port = htons(server.port); in clusterBuildMessageHdr()
2143 hdr->state = server.cluster->state; in clusterBuildMessageHdr()
2146 hdr->currentEpoch = htonu64(server.cluster->currentEpoch); in clusterBuildMessageHdr()
2153 offset = server.master_repl_offset; 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()
2527 if (mstime() - node->slaveof->voted_time < server.cluster_node_timeout * 2) in clusterSendFailoverAuthIfNeeded()
2533 (long long) ((server.cluster_node_timeout*2)- 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()
2621 mstime_t nolog_fail_time = server.cluster_node_timeout + 5000; in clusterLogCantFailover()
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()
2721 auth_timeout = server.cluster_node_timeout*2; in clusterHandleSlaveFailover()
2737 server.cluster->cant_failover_reason = CLUSTER_CANT_FAILOVER_NONE; in clusterHandleSlaveFailover()
2743 if (server.repl_state == REPL_STATE_CONNECTED) { in clusterHandleSlaveFailover()
2744 data_age = (mstime_t)(server.unixtime - server.master->lastinteraction) in clusterHandleSlaveFailover()
2747 data_age = (mstime_t)(server.unixtime - server.repl_down_since) * 1000; in clusterHandleSlaveFailover()
2753 if (data_age > server.cluster_node_timeout) in clusterHandleSlaveFailover()
2754 data_age -= server.cluster_node_timeout; in clusterHandleSlaveFailover()
2760 if (server.cluster_slave_validity_factor && in clusterHandleSlaveFailover()
2762 (((mstime_t)server.repl_ping_slave_period * 1000) + in clusterHandleSlaveFailover()
2763 (server.cluster_node_timeout * server.cluster_slave_validity_factor))) 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()
2913 if (okslaves <= server.cluster_migration_barrier) return; in clusterHandleSlaveMigration()
2926 di = dictGetSafeIterator(server.cluster->nodes); in clusterHandleSlaveMigration()
3017 if (server.cluster->mf_end && clientsArePaused()) { in resetManualFailover()
3018 server.clients_pause_end_time = 0; 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()
3080 handshake_timeout = server.cluster_node_timeout; in clusterCron()
3084 di = dictGetSafeIterator(server.cluster->nodes); in clusterCron()
3102 fd = anetTcpNonBlockBindConnect(server.neterr, node->ip, in clusterCron()
3114 server.neterr); in clusterCron()
3120 aeCreateFileEvent(server.el,link->fd,AE_READABLE, in clusterCron()
3158 de = dictGetRandomKey(server.cluster->nodes); in clusterCron()
3185 di = dictGetSafeIterator(server.cluster->nodes); in clusterCron()
3218 server.cluster_node_timeout && /* was not already reconnected */ in clusterCron()
3222 now - node->ping_sent > server.cluster_node_timeout/2) in clusterCron()
3234 (now - node->pong_received) > server.cluster_node_timeout/2) in clusterCron()
3242 if (server.cluster->mf_end && in clusterCron()
3244 server.cluster->mf_slave == node && in clusterCron()
3259 if (delay > server.cluster_node_timeout) { in clusterCron()
3276 server.masterhost == NULL && 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()
3494 if (server.cluster_require_full_coverage) { 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()
3541 mstime_t rejoin_delay = server.cluster_node_timeout; in clusterUpdateState()
3562 server.cluster->state = new_state; in clusterUpdateState()
3597 for (j = 1; j < server.dbnum; j++) { in verifyClusterConfigWithData()
3598 if (dictSize(server.db[j].dict)) return C_ERR; in verifyClusterConfigWithData()
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()
3881 if (server.cluster_enabled == 0) { in clusterCommand()
3919 if (dictSize(server.db[0].dict) != 0) { in clusterCommand()
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()
4222 (myself->numslots != 0 || dictSize(server.db[0].dict) != 0)) { 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()
4525 server.dirty++; in restoreCommand()
4563 cs = dictFetchValue(server.migrate_cached_sockets,name); in migrateGetSocket()
4566 cs->last_use_time = server.unixtime; in migrateGetSocket()
4571 if (dictSize(server.migrate_cached_sockets) == MIGRATE_SOCKET_CACHE_ITEMS) { in migrateGetSocket()
4573 dictEntry *de = dictGetRandomKey(server.migrate_cached_sockets); in migrateGetSocket()
4577 dictDelete(server.migrate_cached_sockets,dictGetKey(de)); in migrateGetSocket()
4581 fd = anetTcpNonBlockConnect(server.neterr,c->argv[1]->ptr, in migrateGetSocket()
4586 server.neterr); in migrateGetSocket()
4589 anetEnableTcpNoDelay(server.neterr,fd); in migrateGetSocket()
4604 cs->last_use_time = server.unixtime; in migrateGetSocket()
4605 dictAdd(server.migrate_cached_sockets,name,cs); in migrateGetSocket()
4617 cs = dictFetchValue(server.migrate_cached_sockets,name); in migrateCloseSocket()
4625 dictDelete(server.migrate_cached_sockets,name); in migrateCloseSocket()
4630 dictIterator *di = dictGetSafeIterator(server.migrate_cached_sockets); in migrateCloseTimedoutSockets()
4636 if ((server.unixtime - cs->last_use_time) > MIGRATE_SOCKET_CACHE_TTL) { in migrateCloseTimedoutSockets()
4639 dictDelete(server.migrate_cached_sockets,dictGetKey(de)); in migrateCloseTimedoutSockets()
4752 if (server.cluster_enabled) in migrateCommand()
4826 server.dirty++; in migrateCommand()
4916 if (server.cluster_enabled == 0) { in askingCommand()
4928 if (server.cluster_enabled == 0) { in readonlyCommand()
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()
5071 lookupKeyRead(&server.db[0],thiskey) == 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()