#include "server.h"

#include "cluster.h"

#include "endianconv.h"

#include <sys/types.h>

#include <sys/socket.h>

#include <arpa/inet.h>

#include <fcntl.h>

#include <unistd.h>

#include <sys/stat.h>

#include <sys/file.h>

#include <math.h>

#include <ctype.h>

clusterNode *myself = NULL;

clusterNode *createClusterNode(char *nodename, int flags);

void clusterAddNode(clusterNode *node);

void clusterAcceptHandler(aeEventLoop *el, int fd, void *privdata, int mask);

void clusterReadHandler(connection *conn);

void clusterSendPing(clusterLink *link, int type);

void clusterSendFail(char *nodename);

void clusterSendFailoverAuthIfNeeded(clusterNode *node, clusterMsg *request);

void clusterUpdateState(void);

int clusterNodeGetSlotBit(clusterNode *n, int slot);

sds clusterGenNodesDescription(int filter, int use_pport);

list *clusterGetNodesInMyShard(clusterNode *node);

int clusterNodeAddSlave(clusterNode *master, clusterNode *slave);

int clusterAddSlot(clusterNode *n, int slot);

int clusterDelSlot(int slot);

int clusterDelNodeSlots(clusterNode *node);

int clusterNodeSetSlotBit(clusterNode *n, int slot);

void clusterSetMaster(clusterNode *n);

void clusterHandleSlaveFailover(void);

void clusterHandleSlaveMigration(int max_slaves);

int bitmapTestBit(unsigned char *bitmap, int pos);

void clusterDoBeforeSleep(int flags);

void clusterSendUpdate(clusterLink *link, clusterNode *node);

void resetManualFailover(void);

void clusterCloseAllSlots(void);

void clusterSetNodeAsMaster(clusterNode *n);

void clusterDelNode(clusterNode *delnode);

sds representClusterNodeFlags(sds ci, uint16_t flags);

sds representSlotInfo(sds ci, uint16_t *slot_info_pairs, int slot_info_pairs_count);

void clusterFreeNodesSlotsInfo(clusterNode *n);

uint64_t clusterGetMaxEpoch(void);

int clusterBumpConfigEpochWithoutConsensus(void);

void moduleCallClusterReceivers(const char *sender_id, uint64_t module_id, uint8_t type, const unsigned char *payload, uint32_t len);

const char *clusterGetMessageTypeString(int type);

void removeChannelsInSlot(unsigned int slot);

unsigned int countKeysInSlot(unsigned int hashslot);

unsigned int countChannelsInSlot(unsigned int hashslot);

unsigned int delKeysInSlot(unsigned int hashslot);

void clusterAddNodeToShard(const char *shard_id, clusterNode *node);

list *clusterLookupNodeListByShardId(const char *shard_id);

void clusterRemoveNodeFromShard(clusterNode *node);

int auxShardIdSetter(clusterNode *n, void *value, int length);

sds auxShardIdGetter(clusterNode *n, sds s);

static void clusterBuildMessageHdr(clusterMsg *hdr, int type, size_t msglen);

#define dictEntryNextInSlot(de) \

(((clusterDictEntryMetadata *)dictMetadata(de))->next)

#define dictEntryPrevInSlot(de) \

(((clusterDictEntryMetadata *)dictMetadata(de))->prev)

#define RCVBUF_INIT_LEN 1024

#define RCVBUF_MAX_PREALLOC (1<<20) /* 1MB */

dictType clusterNodesDictType = {

dictSdsHash, /* hash function */

NULL, /* key dup */

NULL, /* val dup */

dictSdsKeyCompare, /* key compare */

dictSdsDestructor, /* key destructor */

NULL, /* val destructor */

NULL /* allow to expand */

};

dictType clusterNodesBlackListDictType = {

dictSdsCaseHash, /* hash function */

NULL, /* key dup */

NULL, /* val dup */

dictSdsKeyCaseCompare, /* key compare */

dictSdsDestructor, /* key destructor */

NULL, /* val destructor */

NULL /* allow to expand */

};

static ConnectionType *connTypeOfCluster() {

if (server.tls_cluster) {

return connectionTypeTls();

}

return connectionTypeTcp();

}

dictType clusterSdsToListType = {

dictSdsHash, /* hash function */

NULL, /* key dup */

NULL, /* val dup */

dictSdsKeyCompare, /* key compare */

dictSdsDestructor, /* key destructor */

dictListDestructor, /* val destructor */

NULL /* allow to expand */

};

typedef int (aux_value_setter) (clusterNode* n, void *value, int length);

typedef sds (aux_value_getter) (clusterNode* n, sds s);

typedef struct {

char *field;

aux_value_setter *setter;

aux_value_getter *getter;

int present;

} auxFieldHandler;

typedef enum {

af_start,

af_shard_id = af_start,

af_count,

} auxFieldIndex;

auxFieldHandler auxFieldHandlers[] = {

{"shard-id", auxShardIdSetter, auxShardIdGetter, 0},

};

int isValidAuxChar(int c) {

return isalnum(c) || (strchr("!#$%&()*+-.:;<>?@[]^_{|}~", c) != NULL);

}

int isValidAuxString(sds s) {

for (unsigned i = 0; i < sdslen(s); i++) {

if (!isValidAuxChar(s[i])) return 0;

}

return 1;

}

int auxShardIdSetter(clusterNode *n, void *value, int length) {

if (verifyClusterNodeId(value, length) == C_ERR) {

return C_ERR;

}

memcpy(n->shard_id, value, CLUSTER_NAMELEN);

/* if n already has replicas, make sure they all agree

for (int i = 0; i < n->numslaves; i++) {

if (memcmp(n->slaves[i]->shard_id, n->shard_id, CLUSTER_NAMELEN) != 0) {

return C_ERR;

}

}

clusterAddNodeToShard(value, n);

return C_OK;

}

sds auxShardIdGetter(clusterNode *n, sds s) {

return sdscatprintf(s, "%.40s", n->shard_id);

}

typedef struct {

size_t totlen; /* Total length of this block including the message */

int refcount; /* Number of cluster link send msg queues containing the message */

clusterMsg msg;

} clusterMsgSendBlock;

int clusterLoadConfig(char *filename) {

FILE *fp = fopen(filename,"r");

struct stat sb;

char *line;

int maxline, j;

if (fp == NULL) {

if (errno == ENOENT) {

return C_ERR;

} else {

serverLog(LL_WARNING,

"Loading the cluster node config from %s: %s",

filename, strerror(errno));

exit(1);

}

}

if (redis_fstat(fileno(fp),&sb) == -1) {

serverLog(LL_WARNING,

"Unable to obtain the cluster node config file stat %s: %s",

filename, strerror(errno));

exit(1);

}

/* Check if the file is zero-length: if so return C_ERR to signal

if (sb.st_size == 0) {

fclose(fp);

return C_ERR;

}

/* Parse the file. Note that single lines of the cluster config file can

maxline = 1024+CLUSTER_SLOTS*128;

line = zmalloc(maxline);

while(fgets(line,maxline,fp) != NULL) {

int argc, aux_argc;

sds *argv, *aux_argv;

clusterNode *n, *master;

char *p, *s;

/* Skip blank lines, they can be created either by users manually

if (line[0] == '\n' || line[0] == '\0') continue;

/* Split the line into arguments for processing. */

argv = sdssplitargs(line,&argc);

if (argv == NULL) goto fmterr;

/* Handle the special "vars" line. Don't pretend it is the last

if (strcasecmp(argv[0],"vars") == 0) {

if (!(argc % 2)) goto fmterr;

for (j = 1; j < argc; j += 2) {

if (strcasecmp(argv[j],"currentEpoch") == 0) {

server.cluster->currentEpoch =

strtoull(argv[j+1],NULL,10);

} else if (strcasecmp(argv[j],"lastVoteEpoch") == 0) {

server.cluster->lastVoteEpoch =

strtoull(argv[j+1],NULL,10);

} else {

serverLog(LL_WARNING,

"Skipping unknown cluster config variable '%s'",

argv[j]);

}

}

sdsfreesplitres(argv,argc);

continue;

}

/* Regular config lines have at least eight fields */

if (argc < 8) {

sdsfreesplitres(argv,argc);

goto fmterr;

}

/* Create this node if it does not exist */

if (verifyClusterNodeId(argv[0], sdslen(argv[0])) == C_ERR) {

sdsfreesplitres(argv, argc);

goto fmterr;

}

n = clusterLookupNode(argv[0], sdslen(argv[0]));

if (!n) {

n = createClusterNode(argv[0],0);

clusterAddNode(n);

}

/* Format for the node address and auxiliary argument information:

aux_argv = sdssplitlen(argv[1], sdslen(argv[1]), ",", 1, &aux_argc);

if (aux_argv == NULL) {

sdsfreesplitres(argv,argc);

goto fmterr;

}

/* Hostname is an optional argument that defines the endpoint

if (aux_argc > 1 && sdslen(aux_argv[1]) > 0) {

n->hostname = sdscpy(n->hostname, aux_argv[1]);

} else if (sdslen(n->hostname) != 0) {

sdsclear(n->hostname);

}

/* All fields after hostname are auxiliary and they take on

for (int i = 2; i < aux_argc; i++) {

int field_argc;

sds *field_argv;

field_argv = sdssplitlen(aux_argv[i], sdslen(aux_argv[i]), "=", 1, &field_argc);

if (field_argv == NULL || field_argc != 2) {

/* Invalid aux field format */

if (field_argv != NULL) sdsfreesplitres(field_argv, field_argc);

sdsfreesplitres(argv,argc);

goto fmterr;

}

/* Validate that both aux and value contain valid characters only */

for (unsigned j = 0; j < 2; j++) {

if (!isValidAuxString(field_argv[j])) {

/* Invalid aux field format */

sdsfreesplitres(field_argv, field_argc);

sdsfreesplitres(argv,argc);

goto fmterr;

}

}

/* Note that we don't expect lots of aux fields in the foreseeable

int field_found = 0;

for (unsigned j = 0; j < numElements(auxFieldHandlers); j++) {

if (sdslen(field_argv[0]) != strlen(auxFieldHandlers[j].field) ||

memcmp(field_argv[0], auxFieldHandlers[j].field, sdslen(field_argv[0])) != 0) {

continue;

}

field_found = 1;

if (auxFieldHandlers[j].setter(n, field_argv[1], sdslen(field_argv[1])) != C_OK) {

/* Invalid aux field format */

sdsfreesplitres(field_argv, field_argc);

sdsfreesplitres(argv,argc);

goto fmterr;

}

auxFieldHandlers[j].present = 1;

}

if (field_found == 0) {

/* Invalid aux field format */

sdsfreesplitres(field_argv, field_argc);

sdsfreesplitres(argv,argc);

goto fmterr;

}

sdsfreesplitres(field_argv, field_argc);

}

/* Address and port */

if ((p = strrchr(aux_argv[0],':')) == NULL) {

sdsfreesplitres(aux_argv, aux_argc);

sdsfreesplitres(argv,argc);

goto fmterr;

}

*p = '\0';

memcpy(n->ip,aux_argv[0],strlen(aux_argv[0])+1);

char *port = p+1;

char *busp = strchr(port,'@');

if (busp) {

*busp = '\0';

busp++;

}

n->port = atoi(port);

/* In older versions of nodes.conf the "@busport" part is missing.

n->cport = busp ? atoi(busp) : n->port + CLUSTER_PORT_INCR;

/* The plaintext port for client in a TLS cluster (n->pport) is not

sdsfreesplitres(aux_argv, aux_argc);

/* Parse flags */

p = s = argv[2];

while(p) {

p = strchr(s,',');

if (p) *p = '\0';

if (!strcasecmp(s,"myself")) {

serverAssert(server.cluster->myself == NULL);

myself = server.cluster->myself = n;

n->flags |= CLUSTER_NODE_MYSELF;

} else if (!strcasecmp(s,"master")) {

n->flags |= CLUSTER_NODE_MASTER;

} else if (!strcasecmp(s,"slave")) {

n->flags |= CLUSTER_NODE_SLAVE;

} else if (!strcasecmp(s,"fail?")) {

n->flags |= CLUSTER_NODE_PFAIL;

} else if (!strcasecmp(s,"fail")) {

n->flags |= CLUSTER_NODE_FAIL;

n->fail_time = mstime();

} else if (!strcasecmp(s,"handshake")) {

n->flags |= CLUSTER_NODE_HANDSHAKE;

} else if (!strcasecmp(s,"noaddr")) {

n->flags |= CLUSTER_NODE_NOADDR;

} else if (!strcasecmp(s,"nofailover")) {

n->flags |= CLUSTER_NODE_NOFAILOVER;

} else if (!strcasecmp(s,"noflags")) {

/* nothing to do */

} else {

serverPanic("Unknown flag in redis cluster config file");

}

if (p) s = p+1;

}

/* Get master if any. Set the master and populate master's

if (argv[3][0] != '-') {

if (verifyClusterNodeId(argv[3], sdslen(argv[3])) == C_ERR) {

sdsfreesplitres(argv, argc);

goto fmterr;

}

master = clusterLookupNode(argv[3], sdslen(argv[3]));

if (!master) {

master = createClusterNode(argv[3],0);

clusterAddNode(master);

}

/* shard_id can be absent if we are loading a nodes.conf generated

if (auxFieldHandlers[af_shard_id].present == 0) {

memcpy(n->shard_id, master->shard_id, CLUSTER_NAMELEN);

clusterAddNodeToShard(master->shard_id, n);

} else if (clusterGetNodesInMyShard(master) != NULL &&

memcmp(master->shard_id, n->shard_id, CLUSTER_NAMELEN) != 0)

{

/* If the primary has been added to a shard, make sure this

goto fmterr;

}

n->slaveof = master;

clusterNodeAddSlave(master,n);

} else if (auxFieldHandlers[af_shard_id].present == 0) {

/* n is a primary but it does not have a persisted shard_id.

clusterAddNodeToShard(n->shard_id, n);

}

/* Set ping sent / pong received timestamps */

if (atoi(argv[4])) n->ping_sent = mstime();

if (atoi(argv[5])) n->pong_received = mstime();

/* Set configEpoch for this node.

n->configEpoch = (nodeIsSlave(n) && n->slaveof) ? 0 : strtoull(argv[6],NULL,10);

/* Populate hash slots served by this instance. */

for (j = 8; j < argc; j++) {

int start, stop;

if (argv[j][0] == '[') {

/* Here we handle migrating / importing slots */

int slot;

char direction;

clusterNode *cn;

p = strchr(argv[j],'-');

serverAssert(p != NULL);

*p = '\0';

direction = p[1]; /* Either '>' or '<' */

slot = atoi(argv[j]+1);

if (slot < 0 || slot >= CLUSTER_SLOTS) {

sdsfreesplitres(argv,argc);

goto fmterr;

}

p += 3;

char *pr = strchr(p, ']');

size_t node_len = pr - p;

if (pr == NULL || verifyClusterNodeId(p, node_len) == C_ERR) {

sdsfreesplitres(argv, argc);

goto fmterr;

}

cn = clusterLookupNode(p, CLUSTER_NAMELEN);

if (!cn) {

cn = createClusterNode(p,0);

clusterAddNode(cn);

}

if (direction == '>') {

server.cluster->migrating_slots_to[slot] = cn;

} else {

server.cluster->importing_slots_from[slot] = cn;

}

continue;

} else if ((p = strchr(argv[j],'-')) != NULL) {

*p = '\0';

start = atoi(argv[j]);

stop = atoi(p+1);

} else {

start = stop = atoi(argv[j]);

}

if (start < 0 || start >= CLUSTER_SLOTS ||

stop < 0 || stop >= CLUSTER_SLOTS)

{

sdsfreesplitres(argv,argc);

goto fmterr;

}

while(start <= stop) clusterAddSlot(n, start++);

}

sdsfreesplitres(argv,argc);

}

/* Config sanity check */

if (server.cluster->myself == NULL) goto fmterr;

zfree(line);

fclose(fp);

serverLog(LL_NOTICE,"Node configuration loaded, I'm %.40s", myself->name);

/* Something that should never happen: currentEpoch smaller than

if (clusterGetMaxEpoch() > server.cluster->currentEpoch) {

server.cluster->currentEpoch = clusterGetMaxEpoch();

}

return C_OK;

fmterr:

serverLog(LL_WARNING,

"Unrecoverable error: corrupted cluster config file \"%s\".", line);

zfree(line);

if (fp) fclose(fp);

exit(1);

}

int clusterSaveConfig(int do_fsync) {

sds ci,tmpfilename;

size_t content_size,offset = 0;

ssize_t written_bytes;

int fd = -1;

int retval = C_ERR;

server.cluster->todo_before_sleep &= ~CLUSTER_TODO_SAVE_CONFIG;

/* Get the nodes description and concatenate our "vars" directive to

ci = clusterGenNodesDescription(CLUSTER_NODE_HANDSHAKE, 0);

ci = sdscatprintf(ci,"vars currentEpoch %llu lastVoteEpoch %llu\n",

(unsigned long long) server.cluster->currentEpoch,

(unsigned long long) server.cluster->lastVoteEpoch);

content_size = sdslen(ci);

/* Create a temp file with the new content. */

tmpfilename = sdscatfmt(sdsempty(),"%s.tmp-%i-%I",

server.cluster_configfile,(int) getpid(),mstime());

if ((fd = open(tmpfilename,O_WRONLY|O_CREAT,0644)) == -1) {

serverLog(LL_WARNING,"Could not open temp cluster config file: %s",strerror(errno));

goto cleanup;

}

while (offset < content_size) {

written_bytes = write(fd,ci + offset,content_size - offset);

if (written_bytes <= 0) {

if (errno == EINTR) continue;

serverLog(LL_WARNING,"Failed after writing (%zd) bytes to tmp cluster config file: %s",

offset,strerror(errno));

goto cleanup;

}

offset += written_bytes;

}

if (do_fsync) {

server.cluster->todo_before_sleep &= ~CLUSTER_TODO_FSYNC_CONFIG;

if (redis_fsync(fd) == -1) {

serverLog(LL_WARNING,"Could not sync tmp cluster config file: %s",strerror(errno));

goto cleanup;

}

}

if (rename(tmpfilename, server.cluster_configfile) == -1) {

serverLog(LL_WARNING,"Could not rename tmp cluster config file: %s",strerror(errno));

goto cleanup;

}

if (do_fsync) {

if (fsyncFileDir(server.cluster_configfile) == -1) {

serverLog(LL_WARNING,"Could not sync cluster config file dir: %s",strerror(errno));

goto cleanup;

}

}

retval = C_OK; /* If we reached this point, everything is fine. */

cleanup:

if (fd != -1) close(fd);

if (retval) unlink(tmpfilename);

sdsfree(tmpfilename);

sdsfree(ci);

return retval;

}

void clusterSaveConfigOrDie(int do_fsync) {

if (clusterSaveConfig(do_fsync) == -1) {

serverLog(LL_WARNING,"Fatal: can't update cluster config file.");

exit(1);

}

}

int clusterLockConfig(char *filename) {

#if !defined(__sun)

/* To lock it, we need to open the file in a way it is created if

int fd = open(filename,O_WRONLY|O_CREAT|O_CLOEXEC,0644);

if (fd == -1) {

serverLog(LL_WARNING,

"Can't open %s in order to acquire a lock: %s",

filename, strerror(errno));

return C_ERR;

}

if (flock(fd,LOCK_EX|LOCK_NB) == -1) {

if (errno == EWOULDBLOCK) {

serverLog(LL_WARNING,

"Sorry, the cluster configuration file %s is already used "

"by a different Redis Cluster node. Please make sure that "

"different nodes use different cluster configuration "

"files.", filename);

} else {

serverLog(LL_WARNING,

"Impossible to lock %s: %s", filename, strerror(errno));

}

close(fd);

return C_ERR;

}

/* Lock acquired: leak the 'fd' by not closing it until shutdown time, so that

server.cluster_config_file_lock_fd = fd;

UNUSED(filename);

#endif /* __sun */

return C_OK;

}

void deriveAnnouncedPorts(int *announced_port, int *announced_pport,

int *announced_cport) {

int port = server.tls_cluster ? server.tls_port : server.port;

/* Default announced ports. */

*announced_port = port;

*announced_pport = server.tls_cluster ? server.port : 0;

*announced_cport = server.cluster_port ? server.cluster_port : port + CLUSTER_PORT_INCR;

/* Config overriding announced ports. */

if (server.tls_cluster && server.cluster_announce_tls_port) {

*announced_port = server.cluster_announce_tls_port;

*announced_pport = server.cluster_announce_port;

} else if (server.cluster_announce_port) {

*announced_port = server.cluster_announce_port;

}

if (server.cluster_announce_bus_port) {

*announced_cport = server.cluster_announce_bus_port;

}

}

void clusterUpdateMyselfFlags(void) {

if (!myself) return;

int oldflags = myself->flags;

int nofailover = server.cluster_slave_no_failover ?

CLUSTER_NODE_NOFAILOVER : 0;

myself->flags &= ~CLUSTER_NODE_NOFAILOVER;

myself->flags |= nofailover;

if (myself->flags != oldflags) {

clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG|

CLUSTER_TODO_UPDATE_STATE);

}

}

void clusterUpdateMyselfAnnouncedPorts(void) {

if (!myself) return;

deriveAnnouncedPorts(&myself->port,&myself->pport,&myself->cport);

}

void clusterUpdateMyselfIp(void) {

if (!myself) return;

static char *prev_ip = NULL;

char *curr_ip = server.cluster_announce_ip;

int changed = 0;

if (prev_ip == NULL && curr_ip != NULL) changed = 1;

else if (prev_ip != NULL && curr_ip == NULL) changed = 1;

else if (prev_ip && curr_ip && strcmp(prev_ip,curr_ip)) changed = 1;

if (changed) {

if (prev_ip) zfree(prev_ip);

prev_ip = curr_ip;

if (curr_ip) {

/* We always take a copy of the previous IP address, by

prev_ip = zstrdup(prev_ip);

redis_strlcpy(myself->ip,server.cluster_announce_ip,NET_IP_STR_LEN);

} else {

myself->ip[0] = '\0'; /* Force autodetection. */

}

}

}

static void updateAnnouncedHostname(clusterNode *node, char *new) {

/* Previous and new hostname are the same, no need to update. */

if (new && !strcmp(new, node->hostname)) {

return;

} else if (!new && (sdslen(node->hostname) == 0)) {

return;

}

if (new) {

node->hostname = sdscpy(node->hostname, new);

} else if (sdslen(node->hostname) != 0) {

sdsclear(node->hostname);

}

clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG);

}

static void updateShardId(clusterNode *node, const char *shard_id) {

if (memcmp(node->shard_id, shard_id, CLUSTER_NAMELEN) != 0) {

clusterRemoveNodeFromShard(node);

memcpy(node->shard_id, shard_id, CLUSTER_NAMELEN);

clusterAddNodeToShard(shard_id, node);

clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG);

}

if (myself != node && myself->slaveof == node) {

if (memcmp(myself->shard_id, shard_id, CLUSTER_NAMELEN) != 0) {

/* shard-id can diverge right after a rolling upgrade

clusterRemoveNodeFromShard(myself);

memcpy(myself->shard_id, shard_id, CLUSTER_NAMELEN);

clusterAddNodeToShard(shard_id, myself);

clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG|CLUSTER_TODO_FSYNC_CONFIG);

}

}

}

void clusterUpdateMyselfHostname(void) {

if (!myself) return;

updateAnnouncedHostname(myself, server.cluster_announce_hostname);

}

void clusterInit(void) {

int saveconf = 0;

server.cluster = zmalloc(sizeof(clusterState));

server.cluster->myself = NULL;

server.cluster->currentEpoch = 0;

server.cluster->state = CLUSTER_FAIL;

server.cluster->size = 1;

server.cluster->todo_before_sleep = 0;

server.cluster->nodes = dictCreate(&clusterNodesDictType);

server.cluster->shards = dictCreate(&clusterSdsToListType);

server.cluster->nodes_black_list =

dictCreate(&clusterNodesBlackListDictType);

server.cluster->failover_auth_time = 0;

server.cluster->failover_auth_count = 0;

server.cluster->failover_auth_rank = 0;

server.cluster->failover_auth_epoch = 0;

server.cluster->cant_failover_reason = CLUSTER_CANT_FAILOVER_NONE;

server.cluster->lastVoteEpoch = 0;

/* Initialize stats */

for (int i = 0; i < CLUSTERMSG_TYPE_COUNT; i++) {

server.cluster->stats_bus_messages_sent[i] = 0;

server.cluster->stats_bus_messages_received[i] = 0;

}

server.cluster->stats_pfail_nodes = 0;

server.cluster->stat_cluster_links_buffer_limit_exceeded = 0;

memset(server.cluster->slots,0, sizeof(server.cluster->slots));

clusterCloseAllSlots();

/* Lock the cluster config file to make sure every node uses

server.cluster_config_file_lock_fd = -1;

if (clusterLockConfig(server.cluster_configfile) == C_ERR)

exit(1);

/* Load or create a new nodes configuration. */

if (clusterLoadConfig(server.cluster_configfile) == C_ERR) {

/* No configuration found. We will just use the random name provided

myself = server.cluster->myself =

createClusterNode(NULL,CLUSTER_NODE_MYSELF|CLUSTER_NODE_MASTER);

serverLog(LL_NOTICE,"No cluster configuration found, I'm %.40s",

myself->name);

clusterAddNode(myself);

clusterAddNodeToShard(myself->shard_id, myself);

saveconf = 1;

}

if (saveconf) clusterSaveConfigOrDie(1);

/* Port sanity check II

int port = server.tls_cluster ? server.tls_port : server.port;

if (!server.cluster_port && port > (65535-CLUSTER_PORT_INCR)) {

serverLog(LL_WARNING, "Redis port number too high. "

"Cluster communication port is 10,000 port "

"numbers higher than your Redis port. "

"Your Redis port number must be 55535 or less.");

exit(1);

}

if (!server.bindaddr_count) {

serverLog(LL_WARNING, "No bind address is configured, but it is required for the Cluster bus.");

exit(1);

}

/* Initialize data for the Slot to key API. */

slotToKeyInit(server.db);

/* The slots -> channels map is a radix tree. Initialize it here. */

server.cluster->slots_to_channels = raxNew();

/* Set myself->port/cport/pport to my listening ports, we'll just need to

deriveAnnouncedPorts(&myself->port, &myself->pport, &myself->cport);

server.cluster->mf_end = 0;

server.cluster->mf_slave = NULL;

resetManualFailover();

clusterUpdateMyselfFlags();

clusterUpdateMyselfIp();

clusterUpdateMyselfHostname();

}

void clusterInitListeners(void) {

if (connectionIndexByType(connTypeOfCluster()->get_type(NULL)) < 0) {

serverLog(LL_WARNING, "Missing connection type %s, but it is required for the Cluster bus.", connTypeOfCluster()->get_type(NULL));

exit(1);

}

int port = server.tls_cluster ? server.tls_port : server.port;

connListener *listener = &server.clistener;

listener->count = 0;

listener->bindaddr = server.bindaddr;

listener->bindaddr_count = server.bindaddr_count;

listener->port = server.cluster_port ? server.cluster_port : port + CLUSTER_PORT_INCR;

listener->ct = connTypeOfCluster();

if (connListen(listener) == C_ERR ) {

/* Note: the following log text is matched by the test suite. */

serverLog(LL_WARNING, "Failed listening on port %u (cluster), aborting.", listener->port);

exit(1);

}

if (createSocketAcceptHandler(&server.clistener, clusterAcceptHandler) != C_OK) {

serverPanic("Unrecoverable error creating Redis Cluster socket accept handler.");

}

}

void clusterReset(int hard) {

dictIterator *di;

dictEntry *de;

int j;

/* Turn into master. */

if (nodeIsSlave(myself)) {

clusterSetNodeAsMaster(myself);

replicationUnsetMaster();

emptyData(-1,EMPTYDB_NO_FLAGS,NULL);

}

/* Close slots, reset manual failover state. */

clusterCloseAllSlots();

resetManualFailover();

/* Unassign all the slots. */

for (j = 0; j < CLUSTER_SLOTS; j++) clusterDelSlot(j);

/* Recreate shards dict */

dictEmpty(server.cluster->shards, NULL);

/* Forget all the nodes, but myself. */

di = dictGetSafeIterator(server.cluster->nodes);

while((de = dictNext(di)) != NULL) {

clusterNode *node = dictGetVal(de);

if (node == myself) continue;

clusterDelNode(node);

}

dictReleaseIterator(di);

/* Hard reset only: set epochs to 0, change node ID. */

if (hard) {

sds oldname;

server.cluster->currentEpoch = 0;

server.cluster->lastVoteEpoch = 0;

myself->configEpoch = 0;

serverLog(LL_WARNING, "configEpoch set to 0 via CLUSTER RESET HARD");

/* To change the Node ID we need to remove the old name from the

oldname = sdsnewlen(myself->name, CLUSTER_NAMELEN);

dictDelete(server.cluster->nodes,oldname);

sdsfree(oldname);

getRandomHexChars(myself->name, CLUSTER_NAMELEN);

getRandomHexChars(myself->shard_id, CLUSTER_NAMELEN);

clusterAddNode(myself);