src/cluster.c

   1 #include "redis.h"
   2
   3 #include <arpa/inet.h>
   4 #include <fcntl.h>
   5 #include <unistd.h>
   6
   7 void clusterAcceptHandler(aeEventLoop *el, int fd, void *privdata, int mask);
   8 void clusterReadHandler(aeEventLoop *el, int fd, void *privdata, int mask);
   9 void clusterSendPing(clusterLink *link, int type);
  10 void clusterSendFail(char *nodename);
  11 void clusterUpdateState(void);
  12 int clusterNodeGetSlotBit(clusterNode *n, int slot);
  13 sds clusterGenNodesDescription(void);
  14 clusterNode *clusterLookupNode(char *name);
  15 int clusterNodeAddSlave(clusterNode *master, clusterNode *slave);
  16 int clusterAddSlot(clusterNode *n, int slot);
  17
  18 /* -----------------------------------------------------------------------------
  19  * Initialization
  20  * -------------------------------------------------------------------------- */
  21
  22 void clusterGetRandomName(char *p) {
  23     FILE *fp = fopen("/dev/urandom","r");
  24     char *charset = "0123456789abcdef";
  25     int j;
  26
  27     if (fp == NULL || fread(p,REDIS_CLUSTER_NAMELEN,1,fp) == 0) {
  28         for (j = 0; j < REDIS_CLUSTER_NAMELEN; j++)
  29             p[j] = rand();
  30     }
  31     for (j = 0; j < REDIS_CLUSTER_NAMELEN; j++)
  32         p[j] = charset[p[j] & 0x0F];
  33     fclose(fp);
  34 }
  35
  36 int clusterLoadConfig(char *filename) {
  37     FILE *fp = fopen(filename,"r");
  38     char *line;
  39     int maxline, j;
  40
  41     if (fp == NULL) return REDIS_ERR;
  42
  43     /* Parse the file. Note that single liens of the cluster config file can
  44      * be really long as they include all the hash slots of the node.
  45      * This means in the worst possible case REDIS_CLUSTER_SLOTS/2 integers.
  46      * To simplify we allocate 1024+REDIS_CLUSTER_SLOTS*16 bytes per line. */
  47     maxline = 1024+REDIS_CLUSTER_SLOTS*16;
  48     line = zmalloc(maxline);
  49     while(fgets(line,maxline,fp) != NULL) {
  50         int argc;
  51         sds *argv = sdssplitargs(line,&argc);
  52         clusterNode *n, *master;
  53         char *p, *s;
  54
  55         /* Create this node if it does not exist */
  56         n = clusterLookupNode(argv[0]);
  57         if (!n) {
  58             n = createClusterNode(argv[0],0);
  59             clusterAddNode(n);
  60         }
  61         /* Address and port */
  62         if ((p = strchr(argv[1],':')) == NULL) goto fmterr;
  63         *p = '\0';
  64         memcpy(n->ip,argv[1],strlen(argv[1])+1);
  65         n->port = atoi(p+1);
  66
  67         /* Parse flags */
  68         p = s = argv[2];
  69         while(p) {
  70             p = strchr(s,',');
  71             if (p) *p = '\0';
  72             if (!strcasecmp(s,"myself")) {
  73                 redisAssert(server.cluster.myself == NULL);
  74                 server.cluster.myself = n;
  75                 n->flags |= REDIS_NODE_MYSELF;
  76             } else if (!strcasecmp(s,"master")) {
  77                 n->flags |= REDIS_NODE_MASTER;
  78             } else if (!strcasecmp(s,"slave")) {
  79                 n->flags |= REDIS_NODE_SLAVE;
  80             } else if (!strcasecmp(s,"fail?")) {
  81                 n->flags |= REDIS_NODE_PFAIL;
  82             } else if (!strcasecmp(s,"fail")) {
  83                 n->flags |= REDIS_NODE_FAIL;
  84             } else if (!strcasecmp(s,"handshake")) {
  85                 n->flags |= REDIS_NODE_HANDSHAKE;
  86             } else if (!strcasecmp(s,"noaddr")) {
  87                 n->flags |= REDIS_NODE_NOADDR;
  88             } else if (!strcasecmp(s,"noflags")) {
  89                 /* nothing to do */
  90             } else {
  91                 redisPanic("Unknown flag in redis cluster config file");
  92             }
  93             if (p) s = p+1;
  94         }
  95
  96         /* Get master if any. Set the master and populate master's
  97          * slave list. */
  98         if (argv[3][0] != '-') {
  99             master = clusterLookupNode(argv[3]);
 100             if (!master) {
 101                 master = createClusterNode(argv[3],0);
 102                 clusterAddNode(master);
 103             }
 104             n->slaveof = master;
 105             clusterNodeAddSlave(master,n);
 106         }
 107
 108         /* Set ping sent / pong received timestamps */
 109         if (atoi(argv[4])) n->ping_sent = time(NULL);
 110         if (atoi(argv[5])) n->pong_received = time(NULL);
 111
 112         /* Populate hash slots served by this instance. */
 113         for (j = 7; j < argc; j++) {
 114             int start, stop;
 115
 116             if (argv[j][0] == '[') {
 117                 /* Here we handle migrating / importing slots */
 118                 int slot;
 119                 char direction;
 120                 clusterNode *cn;
 121
 122                 p = strchr(argv[j],'-');
 123                 redisAssert(p != NULL);
 124                 *p = '\0';
 125                 direction = p[1]; /* Either '>' or '<' */
 126                 slot = atoi(argv[j]+1);
 127                 p += 3;
 128                 cn = clusterLookupNode(p);
 129                 if (!cn) {
 130                     cn = createClusterNode(p,0);
 131                     clusterAddNode(cn);
 132                 }
 133                 if (direction == '>') {
 134                     server.cluster.migrating_slots_to[slot] = cn;
 135                 } else {
 136                     server.cluster.importing_slots_from[slot] = cn;
 137                 }
 138                 continue;
 139             } else if ((p = strchr(argv[j],'-')) != NULL) {
 140                 *p = '\0';
 141                 start = atoi(argv[j]);
 142                 stop = atoi(p+1);
 143             } else {
 144                 start = stop = atoi(argv[j]);
 145             }
 146             while(start <= stop) clusterAddSlot(n, start++);
 147         }
 148
 149         sdssplitargs_free(argv,argc);
 150     }
 151     zfree(line);
 152     fclose(fp);
 153
 154     /* Config sanity check */
 155     redisAssert(server.cluster.myself != NULL);
 156     redisLog(REDIS_NOTICE,"Node configuration loaded, I'm %.40s",
 157         server.cluster.myself->name);
 158     clusterUpdateState();
 159     return REDIS_OK;
 160
 161 fmterr:
 162     redisLog(REDIS_WARNING,"Unrecovarable error: corrupted cluster config file.");
 163     fclose(fp);
 164     exit(1);
 165 }
 166
 167 /* Cluster node configuration is exactly the same as CLUSTER NODES output.
 168  *
 169  * This function writes the node config and returns 0, on error -1
 170  * is returned. */
 171 int clusterSaveConfig(void) {
 172     sds ci = clusterGenNodesDescription();
 173     int fd;
 174
 175     if ((fd = open(server.cluster.configfile,O_WRONLY|O_CREAT|O_TRUNC,0644))
 176         == -1) goto err;
 177     if (write(fd,ci,sdslen(ci)) != (ssize_t)sdslen(ci)) goto err;
 178     close(fd);
 179     sdsfree(ci);
 180     return 0;
 181
 182 err:
 183     sdsfree(ci);
 184     return -1;
 185 }
 186
 187 void clusterSaveConfigOrDie(void) {
 188     if (clusterSaveConfig() == -1) {
 189         redisLog(REDIS_WARNING,"Fatal: can't update cluster config file.");
 190         exit(1);
 191     }
 192 }
 193
 194 void clusterInit(void) {
 195     int saveconf = 0;
 196
 197     server.cluster.myself = NULL;
 198     server.cluster.state = REDIS_CLUSTER_FAIL;
 199     server.cluster.nodes = dictCreate(&clusterNodesDictType,NULL);
 200     server.cluster.node_timeout = 15;
 201     memset(server.cluster.migrating_slots_to,0,
 202         sizeof(server.cluster.migrating_slots_to));
 203     memset(server.cluster.importing_slots_from,0,
 204         sizeof(server.cluster.importing_slots_from));
 205     memset(server.cluster.slots,0,
 206         sizeof(server.cluster.slots));
 207     if (clusterLoadConfig(server.cluster.configfile) == REDIS_ERR) {
 208         /* No configuration found. We will just use the random name provided
 209          * by the createClusterNode() function. */
 210         server.cluster.myself = createClusterNode(NULL,REDIS_NODE_MYSELF);
 211         redisLog(REDIS_NOTICE,"No cluster configuration found, I'm %.40s",
 212             server.cluster.myself->name);
 213         clusterAddNode(server.cluster.myself);
 214         saveconf = 1;
 215     }
 216     if (saveconf) clusterSaveConfigOrDie();
 217     /* We need a listening TCP port for our cluster messaging needs */
 218     server.cfd = anetTcpServer(server.neterr,
 219             server.port+REDIS_CLUSTER_PORT_INCR, server.bindaddr);
 220     if (server.cfd == -1) {
 221         redisLog(REDIS_WARNING, "Opening cluster TCP port: %s", server.neterr);
 222         exit(1);
 223     }
 224     if (aeCreateFileEvent(server.el, server.cfd, AE_READABLE,
 225         clusterAcceptHandler, NULL) == AE_ERR) oom("creating file event");
 226     server.cluster.slots_to_keys = zslCreate();
 227 }
 228
 229 /* -----------------------------------------------------------------------------
 230  * CLUSTER communication link
 231  * -------------------------------------------------------------------------- */
 232
 233 clusterLink *createClusterLink(clusterNode *node) {
 234     clusterLink *link = zmalloc(sizeof(*link));
 235     link->sndbuf = sdsempty();
 236     link->rcvbuf = sdsempty();
 237     link->node = node;
 238     link->fd = -1;
 239     return link;
 240 }
 241
 242 /* Free a cluster link, but does not free the associated node of course.
 243  * Just this function will make sure that the original node associated
 244  * with this link will have the 'link' field set to NULL. */
 245 void freeClusterLink(clusterLink *link) {
 246     if (link->fd != -1) {
 247         aeDeleteFileEvent(server.el, link->fd, AE_WRITABLE);
 248         aeDeleteFileEvent(server.el, link->fd, AE_READABLE);
 249     }
 250     sdsfree(link->sndbuf);
 251     sdsfree(link->rcvbuf);
 252     if (link->node)
 253         link->node->link = NULL;
 254     close(link->fd);
 255     zfree(link);
 256 }
 257
 258 void clusterAcceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
 259     int cport, cfd;
 260     char cip[128];
 261     clusterLink *link;
 262     REDIS_NOTUSED(el);
 263     REDIS_NOTUSED(mask);
 264     REDIS_NOTUSED(privdata);
 265
 266     cfd = anetTcpAccept(server.neterr, fd, cip, &cport);
 267     if (cfd == AE_ERR) {
 268         redisLog(REDIS_VERBOSE,"Accepting cluster node: %s", server.neterr);
 269         return;
 270     }
 271     redisLog(REDIS_VERBOSE,"Accepted cluster node %s:%d", cip, cport);
 272     /* We need to create a temporary node in order to read the incoming
 273      * packet in a valid contest. This node will be released once we
 274      * read the packet and reply. */
 275     link = createClusterLink(NULL);
 276     link->fd = cfd;
 277     aeCreateFileEvent(server.el,cfd,AE_READABLE,clusterReadHandler,link);
 278 }
 279
 280 /* -----------------------------------------------------------------------------
 281  * Key space handling
 282  * -------------------------------------------------------------------------- */
 283
 284 /* We have 4096 hash slots. The hash slot of a given key is obtained
 285  * as the least significant 12 bits of the crc16 of the key. */
 286 unsigned int keyHashSlot(char *key, int keylen) {
 287     return crc16(key,keylen) & 0x0FFF;
 288 }
 289
 290 /* -----------------------------------------------------------------------------
 291  * CLUSTER node API
 292  * -------------------------------------------------------------------------- */
 293
 294 /* Create a new cluster node, with the specified flags.
 295  * If "nodename" is NULL this is considered a first handshake and a random
 296  * node name is assigned to this node (it will be fixed later when we'll
 297  * receive the first pong).
 298  *
 299  * The node is created and returned to the user, but it is not automatically
 300  * added to the nodes hash table. */
 301 clusterNode *createClusterNode(char *nodename, int flags) {
 302     clusterNode *node = zmalloc(sizeof(*node));
 303
 304     if (nodename)
 305         memcpy(node->name, nodename, REDIS_CLUSTER_NAMELEN);
 306     else
 307         clusterGetRandomName(node->name);
 308     node->flags = flags;
 309     memset(node->slots,0,sizeof(node->slots));
 310     node->numslaves = 0;
 311     node->slaves = NULL;
 312     node->slaveof = NULL;
 313     node->ping_sent = node->pong_received = 0;
 314     node->configdigest = NULL;
 315     node->configdigest_ts = 0;
 316     node->link = NULL;
 317     return node;
 318 }
 319
 320 int clusterNodeRemoveSlave(clusterNode *master, clusterNode *slave) {
 321     int j;
 322
 323     for (j = 0; j < master->numslaves; j++) {
 324         if (master->slaves[j] == slave) {
 325             memmove(master->slaves+j,master->slaves+(j+1),
 326                 (master->numslaves-1)-j);
 327             master->numslaves--;
 328             return REDIS_OK;
 329         }
 330     }
 331     return REDIS_ERR;
 332 }
 333
 334 int clusterNodeAddSlave(clusterNode *master, clusterNode *slave) {
 335     int j;
 336
 337     /* If it's already a slave, don't add it again. */
 338     for (j = 0; j < master->numslaves; j++)
 339         if (master->slaves[j] == slave) return REDIS_ERR;
 340     master->slaves = zrealloc(master->slaves,
 341         sizeof(clusterNode*)*(master->numslaves+1));
 342     master->slaves[master->numslaves] = slave;
 343     master->numslaves++;
 344     return REDIS_OK;
 345 }
 346
 347 void clusterNodeResetSlaves(clusterNode *n) {
 348     zfree(n->slaves);
 349     n->numslaves = 0;
 350 }
 351
 352 void freeClusterNode(clusterNode *n) {
 353     sds nodename;
 354
 355     nodename = sdsnewlen(n->name, REDIS_CLUSTER_NAMELEN);
 356     redisAssert(dictDelete(server.cluster.nodes,nodename) == DICT_OK);
 357     sdsfree(nodename);
 358     if (n->slaveof) clusterNodeRemoveSlave(n->slaveof, n);
 359     if (n->link) freeClusterLink(n->link);
 360     zfree(n);
 361 }
 362
 363 /* Add a node to the nodes hash table */
 364 int clusterAddNode(clusterNode *node) {
 365     int retval;
 366
 367     retval = dictAdd(server.cluster.nodes,
 368             sdsnewlen(node->name,REDIS_CLUSTER_NAMELEN), node);
 369     return (retval == DICT_OK) ? REDIS_OK : REDIS_ERR;
 370 }
 371
 372 /* Node lookup by name */
 373 clusterNode *clusterLookupNode(char *name) {
 374     sds s = sdsnewlen(name, REDIS_CLUSTER_NAMELEN);
 375     struct dictEntry *de;
 376
 377     de = dictFind(server.cluster.nodes,s);
 378     sdsfree(s);
 379     if (de == NULL) return NULL;
 380     return dictGetVal(de);
 381 }
 382
 383 /* This is only used after the handshake. When we connect a given IP/PORT
 384  * as a result of CLUSTER MEET we don't have the node name yet, so we
 385  * pick a random one, and will fix it when we receive the PONG request using
 386  * this function. */
 387 void clusterRenameNode(clusterNode *node, char *newname) {
 388     int retval;
 389     sds s = sdsnewlen(node->name, REDIS_CLUSTER_NAMELEN);
 390
 391     redisLog(REDIS_DEBUG,"Renaming node %.40s into %.40s",
 392         node->name, newname);
 393     retval = dictDelete(server.cluster.nodes, s);
 394     sdsfree(s);
 395     redisAssert(retval == DICT_OK);
 396     memcpy(node->name, newname, REDIS_CLUSTER_NAMELEN);
 397     clusterAddNode(node);
 398 }
 399
 400 /* -----------------------------------------------------------------------------
 401  * CLUSTER messages exchange - PING/PONG and gossip
 402  * -------------------------------------------------------------------------- */
 403
 404 /* Process the gossip section of PING or PONG packets.
 405  * Note that this function assumes that the packet is already sanity-checked
 406  * by the caller, not in the content of the gossip section, but in the
 407  * length. */
 408 void clusterProcessGossipSection(clusterMsg *hdr, clusterLink *link) {
 409     uint16_t count = ntohs(hdr->count);
 410     clusterMsgDataGossip *g = (clusterMsgDataGossip*) hdr->data.ping.gossip;
 411     clusterNode *sender = link->node ? link->node : clusterLookupNode(hdr->sender);
 412
 413     while(count--) {
 414         sds ci = sdsempty();
 415         uint16_t flags = ntohs(g->flags);
 416         clusterNode *node;
 417
 418         if (flags == 0) ci = sdscat(ci,"noflags,");
 419         if (flags & REDIS_NODE_MYSELF) ci = sdscat(ci,"myself,");
 420         if (flags & REDIS_NODE_MASTER) ci = sdscat(ci,"master,");
 421         if (flags & REDIS_NODE_SLAVE) ci = sdscat(ci,"slave,");
 422         if (flags & REDIS_NODE_PFAIL) ci = sdscat(ci,"fail?,");
 423         if (flags & REDIS_NODE_FAIL) ci = sdscat(ci,"fail,");
 424         if (flags & REDIS_NODE_HANDSHAKE) ci = sdscat(ci,"handshake,");
 425         if (flags & REDIS_NODE_NOADDR) ci = sdscat(ci,"noaddr,");
 426         if (ci[sdslen(ci)-1] == ',') ci[sdslen(ci)-1] = ' ';
 427
 428         redisLog(REDIS_DEBUG,"GOSSIP %.40s %s:%d %s",
 429             g->nodename,
 430             g->ip,
 431             ntohs(g->port),
 432             ci);
 433         sdsfree(ci);
 434
 435         /* Update our state accordingly to the gossip sections */
 436         node = clusterLookupNode(g->nodename);
 437         if (node != NULL) {
 438             /* We already know this node. Let's start updating the last
 439              * time PONG figure if it is newer than our figure.
 440              * Note that it's not a problem if we have a PING already
 441              * in progress against this node. */
 442             if (node->pong_received < (signed) ntohl(g->pong_received)) {
 443                  redisLog(REDIS_DEBUG,"Node pong_received updated by gossip");
 444                 node->pong_received = ntohl(g->pong_received);
 445             }
 446             /* Mark this node as FAILED if we think it is possibly failing
 447              * and another node also thinks it's failing. */
 448             if (node->flags & REDIS_NODE_PFAIL &&
 449                 (flags & (REDIS_NODE_FAIL|REDIS_NODE_PFAIL)))
 450             {
 451                 redisLog(REDIS_NOTICE,"Received a PFAIL acknowledge from node %.40s, marking node %.40s as FAIL!", hdr->sender, node->name);
 452                 node->flags &= ~REDIS_NODE_PFAIL;
 453                 node->flags |= REDIS_NODE_FAIL;
 454                 /* Broadcast the failing node name to everybody */
 455                 clusterSendFail(node->name);
 456                 clusterUpdateState();
 457                 clusterSaveConfigOrDie();
 458             }
 459         } else {
 460             /* If it's not in NOADDR state and we don't have it, we
 461              * start an handshake process against this IP/PORT pairs.
 462              *
 463              * Note that we require that the sender of this gossip message
 464              * is a well known node in our cluster, otherwise we risk
 465              * joining another cluster. */
 466             if (sender && !(flags & REDIS_NODE_NOADDR)) {
 467                 clusterNode *newnode;
 468
 469                 redisLog(REDIS_DEBUG,"Adding the new node");
 470                 newnode = createClusterNode(NULL,REDIS_NODE_HANDSHAKE);
 471                 memcpy(newnode->ip,g->ip,sizeof(g->ip));
 472                 newnode->port = ntohs(g->port);
 473                 clusterAddNode(newnode);
 474             }
 475         }
 476
 477         /* Next node */
 478         g++;
 479     }
 480 }
 481
 482 /* IP -> string conversion. 'buf' is supposed to at least be 16 bytes. */
 483 void nodeIp2String(char *buf, clusterLink *link) {
 484     struct sockaddr_in sa;
 485     socklen_t salen = sizeof(sa);
 486
 487     if (getpeername(link->fd, (struct sockaddr*) &sa, &salen) == -1)
 488         redisPanic("getpeername() failed.");
 489     strncpy(buf,inet_ntoa(sa.sin_addr),sizeof(link->node->ip));
 490 }
 491
 492
 493 /* Update the node address to the IP address that can be extracted
 494  * from link->fd, and at the specified port. */
 495 void nodeUpdateAddress(clusterNode *node, clusterLink *link, int port) {
 496     /* TODO */
 497 }
 498
 499 /* When this function is called, there is a packet to process starting
 500  * at node->rcvbuf. Releasing the buffer is up to the caller, so this
 501  * function should just handle the higher level stuff of processing the
 502  * packet, modifying the cluster state if needed.
 503  *
 504  * The function returns 1 if the link is still valid after the packet
 505  * was processed, otherwise 0 if the link was freed since the packet
 506  * processing lead to some inconsistency error (for instance a PONG
 507  * received from the wrong sender ID). */
 508 int clusterProcessPacket(clusterLink *link) {
 509     clusterMsg *hdr = (clusterMsg*) link->rcvbuf;
 510     uint32_t totlen = ntohl(hdr->totlen);
 511     uint16_t type = ntohs(hdr->type);
 512     clusterNode *sender;
 513
 514     redisLog(REDIS_DEBUG,"--- Processing packet of type %d, %lu bytes",
 515         type, (unsigned long) totlen);
 516
 517     /* Perform sanity checks */
 518     if (totlen < 8) return 1;
 519     if (totlen > sdslen(link->rcvbuf)) return 1;
 520     if (type == CLUSTERMSG_TYPE_PING || type == CLUSTERMSG_TYPE_PONG ||
 521         type == CLUSTERMSG_TYPE_MEET)
 522     {
 523         uint16_t count = ntohs(hdr->count);
 524         uint32_t explen; /* expected length of this packet */
 525
 526         explen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
 527         explen += (sizeof(clusterMsgDataGossip)*count);
 528         if (totlen != explen) return 1;
 529     }
 530     if (type == CLUSTERMSG_TYPE_FAIL) {
 531         uint32_t explen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
 532
 533         explen += sizeof(clusterMsgDataFail);
 534         if (totlen != explen) return 1;
 535     }
 536     if (type == CLUSTERMSG_TYPE_PUBLISH) {
 537         uint32_t explen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
 538
 539         explen += sizeof(clusterMsgDataPublish) +
 540                 ntohl(hdr->data.publish.msg.channel_len) +
 541                 ntohl(hdr->data.publish.msg.message_len);
 542         if (totlen != explen) return 1;
 543     }
 544
 545     /* Ready to process the packet. Dispatch by type. */
 546     sender = clusterLookupNode(hdr->sender);
 547     if (type == CLUSTERMSG_TYPE_PING || type == CLUSTERMSG_TYPE_MEET) {
 548         int update_config = 0;
 549         redisLog(REDIS_DEBUG,"Ping packet received: %p", link->node);
 550
 551         /* Add this node if it is new for us and the msg type is MEET.
 552          * In this stage we don't try to add the node with the right
 553          * flags, slaveof pointer, and so forth, as this details will be
 554          * resolved when we'll receive PONGs from the server. */
 555         if (!sender && type == CLUSTERMSG_TYPE_MEET) {
 556             clusterNode *node;
 557
 558             node = createClusterNode(NULL,REDIS_NODE_HANDSHAKE);
 559             nodeIp2String(node->ip,link);
 560             node->port = ntohs(hdr->port);
 561             clusterAddNode(node);
 562             update_config = 1;
 563         }
 564
 565         /* Get info from the gossip section */
 566         clusterProcessGossipSection(hdr,link);
 567
 568         /* Anyway reply with a PONG */
 569         clusterSendPing(link,CLUSTERMSG_TYPE_PONG);
 570
 571         /* Update config if needed */
 572         if (update_config) clusterSaveConfigOrDie();
 573     } else if (type == CLUSTERMSG_TYPE_PONG) {
 574         int update_state = 0;
 575         int update_config = 0;
 576
 577         redisLog(REDIS_DEBUG,"Pong packet received: %p", link->node);
 578         if (link->node) {
 579             if (link->node->flags & REDIS_NODE_HANDSHAKE) {
 580                 /* If we already have this node, try to change the
 581                  * IP/port of the node with the new one. */
 582                 if (sender) {
 583                     redisLog(REDIS_WARNING,
 584                         "Handshake error: we already know node %.40s, updating the address if needed.", sender->name);
 585                     nodeUpdateAddress(sender,link,ntohs(hdr->port));
 586                     freeClusterNode(link->node); /* will free the link too */
 587                     return 0;
 588                 }
 589
 590                 /* First thing to do is replacing the random name with the
 591                  * right node name if this was an handshake stage. */
 592                 clusterRenameNode(link->node, hdr->sender);
 593                 redisLog(REDIS_DEBUG,"Handshake with node %.40s completed.",
 594                     link->node->name);
 595                 link->node->flags &= ~REDIS_NODE_HANDSHAKE;
 596                 update_config = 1;
 597             } else if (memcmp(link->node->name,hdr->sender,
 598                         REDIS_CLUSTER_NAMELEN) != 0)
 599             {
 600                 /* If the reply has a non matching node ID we
 601                  * disconnect this node and set it as not having an associated
 602                  * address. */
 603                 redisLog(REDIS_DEBUG,"PONG contains mismatching sender ID");
 604                 link->node->flags |= REDIS_NODE_NOADDR;
 605                 freeClusterLink(link);
 606                 update_config = 1;
 607                 /* FIXME: remove this node if we already have it.
 608                  *
 609                  * If we already have it but the IP is different, use
 610                  * the new one if the old node is in FAIL, PFAIL, or NOADDR
 611                  * status... */
 612                 return 0;
 613             }
 614         }
 615         /* Update our info about the node */
 616         if (link->node) link->node->pong_received = time(NULL);
 617
 618         /* Update master/slave info */
 619         if (sender) {
 620             if (!memcmp(hdr->slaveof,REDIS_NODE_NULL_NAME,
 621                 sizeof(hdr->slaveof)))
 622             {
 623                 sender->flags &= ~REDIS_NODE_SLAVE;
 624                 sender->flags |= REDIS_NODE_MASTER;
 625                 sender->slaveof = NULL;
 626             } else {
 627                 clusterNode *master = clusterLookupNode(hdr->slaveof);
 628
 629                 sender->flags &= ~REDIS_NODE_MASTER;
 630                 sender->flags |= REDIS_NODE_SLAVE;
 631                 if (sender->numslaves) clusterNodeResetSlaves(sender);
 632                 if (master) clusterNodeAddSlave(master,sender);
 633             }
 634         }
 635
 636         /* Update our info about served slots if this new node is serving
 637          * slots that are not served from our point of view. */
 638         if (sender && sender->flags & REDIS_NODE_MASTER) {
 639             int newslots, j;
 640
 641             newslots =
 642                 memcmp(sender->slots,hdr->myslots,sizeof(hdr->myslots)) != 0;
 643             memcpy(sender->slots,hdr->myslots,sizeof(hdr->myslots));
 644             if (newslots) {
 645                 for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) {
 646                     if (clusterNodeGetSlotBit(sender,j)) {
 647                         if (server.cluster.slots[j] == sender) continue;
 648                         if (server.cluster.slots[j] == NULL ||
 649                             server.cluster.slots[j]->flags & REDIS_NODE_FAIL)
 650                         {
 651                             server.cluster.slots[j] = sender;
 652                             update_state = update_config = 1;
 653                         }
 654                     }
 655                 }
 656             }
 657         }
 658
 659         /* Get info from the gossip section */
 660         clusterProcessGossipSection(hdr,link);
 661
 662         /* Update the cluster state if needed */
 663         if (update_state) clusterUpdateState();
 664         if (update_config) clusterSaveConfigOrDie();
 665     } else if (type == CLUSTERMSG_TYPE_FAIL && sender) {
 666         clusterNode *failing;
 667
 668         failing = clusterLookupNode(hdr->data.fail.about.nodename);
 669         if (failing && !(failing->flags & (REDIS_NODE_FAIL|REDIS_NODE_MYSELF)))
 670         {
 671             redisLog(REDIS_NOTICE,
 672                 "FAIL message received from %.40s about %.40s",
 673                 hdr->sender, hdr->data.fail.about.nodename);
 674             failing->flags |= REDIS_NODE_FAIL;
 675             failing->flags &= ~REDIS_NODE_PFAIL;
 676             clusterUpdateState();
 677             clusterSaveConfigOrDie();
 678         }
 679     } else if (type == CLUSTERMSG_TYPE_PUBLISH) {
 680         robj *channel, *message;
 681         uint32_t channel_len, message_len;
 682
 683         /* Don't bother creating useless objects if there are no Pub/Sub subscribers. */
 684         if (dictSize(server.pubsub_channels) || listLength(server.pubsub_patterns)) {
 685             channel_len = ntohl(hdr->data.publish.msg.channel_len);
 686             message_len = ntohl(hdr->data.publish.msg.message_len);
 687             channel = createStringObject(
 688                         (char*)hdr->data.publish.msg.bulk_data,channel_len);
 689             message = createStringObject(
 690                         (char*)hdr->data.publish.msg.bulk_data+channel_len, message_len);
 691             pubsubPublishMessage(channel,message);
 692             decrRefCount(channel);
 693             decrRefCount(message);
 694         }
 695     } else {
 696         redisLog(REDIS_WARNING,"Received unknown packet type: %d", type);
 697     }
 698     return 1;
 699 }
 700
 701 /* This function is called when we detect the link with this node is lost.
 702    We set the node as no longer connected. The Cluster Cron will detect
 703    this connection and will try to get it connected again.
 704
 705    Instead if the node is a temporary node used to accept a query, we
 706    completely free the node on error. */
 707 void handleLinkIOError(clusterLink *link) {
 708     freeClusterLink(link);
 709 }
 710
 711 /* Send data. This is handled using a trivial send buffer that gets
 712  * consumed by write(). We don't try to optimize this for speed too much
 713  * as this is a very low traffic channel. */
 714 void clusterWriteHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
 715     clusterLink *link = (clusterLink*) privdata;
 716     ssize_t nwritten;
 717     REDIS_NOTUSED(el);
 718     REDIS_NOTUSED(mask);
 719
 720     nwritten = write(fd, link->sndbuf, sdslen(link->sndbuf));
 721     if (nwritten <= 0) {
 722         redisLog(REDIS_NOTICE,"I/O error writing to node link: %s",
 723             strerror(errno));
 724         handleLinkIOError(link);
 725         return;
 726     }
 727     link->sndbuf = sdsrange(link->sndbuf,nwritten,-1);
 728     if (sdslen(link->sndbuf) == 0)
 729         aeDeleteFileEvent(server.el, link->fd, AE_WRITABLE);
 730 }
 731
 732 /* Read data. Try to read the first field of the header first to check the
 733  * full length of the packet. When a whole packet is in memory this function
 734  * will call the function to process the packet. And so forth. */
 735 void clusterReadHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
 736     char buf[1024];
 737     ssize_t nread;
 738     clusterMsg *hdr;
 739     clusterLink *link = (clusterLink*) privdata;
 740     int readlen;
 741     REDIS_NOTUSED(el);
 742     REDIS_NOTUSED(mask);
 743
 744 again:
 745     if (sdslen(link->rcvbuf) >= 4) {
 746         hdr = (clusterMsg*) link->rcvbuf;
 747         readlen = ntohl(hdr->totlen) - sdslen(link->rcvbuf);
 748     } else {
 749         readlen = 4 - sdslen(link->rcvbuf);
 750     }
 751
 752     nread = read(fd,buf,readlen);
 753     if (nread == -1 && errno == EAGAIN) return; /* Just no data */
 754
 755     if (nread <= 0) {
 756         /* I/O error... */
 757         redisLog(REDIS_NOTICE,"I/O error reading from node link: %s",
 758             (nread == 0) ? "connection closed" : strerror(errno));
 759         handleLinkIOError(link);
 760         return;
 761     } else {
 762         /* Read data and recast the pointer to the new buffer. */
 763         link->rcvbuf = sdscatlen(link->rcvbuf,buf,nread);
 764         hdr = (clusterMsg*) link->rcvbuf;
 765     }
 766
 767     /* Total length obtained? read the payload now instead of burning
 768      * cycles waiting for a new event to fire. */
 769     if (sdslen(link->rcvbuf) == 4) goto again;
 770
 771     /* Whole packet in memory? We can process it. */
 772     if (sdslen(link->rcvbuf) == ntohl(hdr->totlen)) {
 773         if (clusterProcessPacket(link)) {
 774             sdsfree(link->rcvbuf);
 775             link->rcvbuf = sdsempty();
 776         }
 777     }
 778 }
 779
 780 /* Put stuff into the send buffer. */
 781 void clusterSendMessage(clusterLink *link, unsigned char *msg, size_t msglen) {
 782     if (sdslen(link->sndbuf) == 0 && msglen != 0)
 783         aeCreateFileEvent(server.el,link->fd,AE_WRITABLE,
 784                     clusterWriteHandler,link);
 785
 786     link->sndbuf = sdscatlen(link->sndbuf, msg, msglen);
 787 }
 788
 789 /* Send a message to all the nodes with a reliable link */
 790 void clusterBroadcastMessage(void *buf, size_t len) {
 791     dictIterator *di;
 792     dictEntry *de;
 793
 794     di = dictGetIterator(server.cluster.nodes);
 795     while((de = dictNext(di)) != NULL) {
 796         clusterNode *node = dictGetVal(de);
 797
 798         if (!node->link) continue;
 799         if (node->flags & (REDIS_NODE_MYSELF|REDIS_NODE_NOADDR)) continue;
 800         clusterSendMessage(node->link,buf,len);
 801     }
 802     dictReleaseIterator(di);
 803 }
 804
 805 /* Build the message header */
 806 void clusterBuildMessageHdr(clusterMsg *hdr, int type) {
 807     int totlen = 0;
 808
 809     memset(hdr,0,sizeof(*hdr));
 810     hdr->type = htons(type);
 811     memcpy(hdr->sender,server.cluster.myself->name,REDIS_CLUSTER_NAMELEN);
 812     memcpy(hdr->myslots,server.cluster.myself->slots,
 813         sizeof(hdr->myslots));
 814     memset(hdr->slaveof,0,REDIS_CLUSTER_NAMELEN);
 815     if (server.cluster.myself->slaveof != NULL) {
 816         memcpy(hdr->slaveof,server.cluster.myself->slaveof->name,
 817                                     REDIS_CLUSTER_NAMELEN);
 818     }
 819     hdr->port = htons(server.port);
 820     hdr->state = server.cluster.state;
 821     memset(hdr->configdigest,0,32); /* FIXME: set config digest */
 822
 823     if (type == CLUSTERMSG_TYPE_FAIL) {
 824         totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
 825         totlen += sizeof(clusterMsgDataFail);
 826     }
 827     hdr->totlen = htonl(totlen);
 828     /* For PING, PONG, and MEET, fixing the totlen field is up to the caller */
 829 }
 830
 831 /* Send a PING or PONG packet to the specified node, making sure to add enough
 832  * gossip informations. */
 833 void clusterSendPing(clusterLink *link, int type) {
 834     unsigned char buf[1024];
 835     clusterMsg *hdr = (clusterMsg*) buf;
 836     int gossipcount = 0, totlen;
 837     /* freshnodes is the number of nodes we can still use to populate the
 838      * gossip section of the ping packet. Basically we start with the nodes
 839      * we have in memory minus two (ourself and the node we are sending the
 840      * message to). Every time we add a node we decrement the counter, so when
 841      * it will drop to <= zero we know there is no more gossip info we can
 842      * send. */
 843     int freshnodes = dictSize(server.cluster.nodes)-2;
 844
 845     if (link->node && type == CLUSTERMSG_TYPE_PING)
 846         link->node->ping_sent = time(NULL);
 847     clusterBuildMessageHdr(hdr,type);
 848
 849     /* Populate the gossip fields */
 850     while(freshnodes > 0 && gossipcount < 3) {
 851         struct dictEntry *de = dictGetRandomKey(server.cluster.nodes);
 852         clusterNode *this = dictGetVal(de);
 853         clusterMsgDataGossip *gossip;
 854         int j;
 855
 856         /* Not interesting to gossip about ourself.
 857          * Nor to send gossip info about HANDSHAKE state nodes (zero info). */
 858         if (this == server.cluster.myself ||
 859             this->flags & REDIS_NODE_HANDSHAKE) {
 860                 freshnodes--; /* otherwise we may loop forever. */
 861                 continue;
 862         }
 863
 864         /* Check if we already added this node */
 865         for (j = 0; j < gossipcount; j++) {
 866             if (memcmp(hdr->data.ping.gossip[j].nodename,this->name,
 867                     REDIS_CLUSTER_NAMELEN) == 0) break;
 868         }
 869         if (j != gossipcount) continue;
 870
 871         /* Add it */
 872         freshnodes--;
 873         gossip = &(hdr->data.ping.gossip[gossipcount]);
 874         memcpy(gossip->nodename,this->name,REDIS_CLUSTER_NAMELEN);
 875         gossip->ping_sent = htonl(this->ping_sent);
 876         gossip->pong_received = htonl(this->pong_received);
 877         memcpy(gossip->ip,this->ip,sizeof(this->ip));
 878         gossip->port = htons(this->port);
 879         gossip->flags = htons(this->flags);
 880         gossipcount++;
 881     }
 882     totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
 883     totlen += (sizeof(clusterMsgDataGossip)*gossipcount);
 884     hdr->count = htons(gossipcount);
 885     hdr->totlen = htonl(totlen);
 886     clusterSendMessage(link,buf,totlen);
 887 }
 888
 889 /* Send a PUBLISH message.
 890  *
 891  * If link is NULL, then the message is broadcasted to the whole cluster. */
 892 void clusterSendPublish(clusterLink *link, robj *channel, robj *message) {
 893     unsigned char buf[4096], *payload;
 894     clusterMsg *hdr = (clusterMsg*) buf;
 895     uint32_t totlen;
 896     uint32_t channel_len, message_len;
 897
 898     channel = getDecodedObject(channel);
 899     message = getDecodedObject(message);
 900     channel_len = sdslen(channel->ptr);
 901     message_len = sdslen(message->ptr);
 902
 903     clusterBuildMessageHdr(hdr,CLUSTERMSG_TYPE_PUBLISH);
 904     totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
 905     totlen += sizeof(clusterMsgDataPublish) + channel_len + message_len;
 906
 907     hdr->data.publish.msg.channel_len = htonl(channel_len);
 908     hdr->data.publish.msg.message_len = htonl(message_len);
 909     hdr->totlen = htonl(totlen);
 910
 911     /* Try to use the local buffer if possible */
 912     if (totlen < sizeof(buf)) {
 913         payload = buf;
 914     } else {
 915         payload = zmalloc(totlen);
 916         hdr = (clusterMsg*) payload;
 917         memcpy(payload,hdr,sizeof(hdr));
 918     }
 919     memcpy(hdr->data.publish.msg.bulk_data,channel->ptr,sdslen(channel->ptr));
 920     memcpy(hdr->data.publish.msg.bulk_data+sdslen(channel->ptr),
 921         message->ptr,sdslen(message->ptr));
 922
 923     if (link)
 924         clusterSendMessage(link,payload,totlen);
 925     else
 926         clusterBroadcastMessage(payload,totlen);
 927
 928     decrRefCount(channel);
 929     decrRefCount(message);
 930     if (payload != buf) zfree(payload);
 931 }
 932
 933 /* Send a FAIL message to all the nodes we are able to contact.
 934  * The FAIL message is sent when we detect that a node is failing
 935  * (REDIS_NODE_PFAIL) and we also receive a gossip confirmation of this:
 936  * we switch the node state to REDIS_NODE_FAIL and ask all the other
 937  * nodes to do the same ASAP. */
 938 void clusterSendFail(char *nodename) {
 939     unsigned char buf[1024];
 940     clusterMsg *hdr = (clusterMsg*) buf;
 941
 942     clusterBuildMessageHdr(hdr,CLUSTERMSG_TYPE_FAIL);
 943     memcpy(hdr->data.fail.about.nodename,nodename,REDIS_CLUSTER_NAMELEN);
 944     clusterBroadcastMessage(buf,ntohl(hdr->totlen));
 945 }
 946
 947 /* -----------------------------------------------------------------------------
 948  * CLUSTER Pub/Sub support
 949  *
 950  * For now we do very little, just propagating PUBLISH messages across the whole
 951  * cluster. In the future we'll try to get smarter and avoiding propagating those
 952  * messages to hosts without receives for a given channel.
 953  * -------------------------------------------------------------------------- */
 954 void clusterPropagatePublish(robj *channel, robj *message) {
 955     clusterSendPublish(NULL, channel, message);
 956 }
 957
 958 /* -----------------------------------------------------------------------------
 959  * CLUSTER cron job
 960  * -------------------------------------------------------------------------- */
 961
 962 /* This is executed 1 time every second */
 963 void clusterCron(void) {
 964     dictIterator *di;
 965     dictEntry *de;
 966     int j;
 967     time_t min_ping_sent = 0;
 968     clusterNode *min_ping_node = NULL;
 969
 970     /* Check if we have disconnected nodes and reestablish the connection. */
 971     di = dictGetIterator(server.cluster.nodes);
 972     while((de = dictNext(di)) != NULL) {
 973         clusterNode *node = dictGetVal(de);
 974
 975         if (node->flags & (REDIS_NODE_MYSELF|REDIS_NODE_NOADDR)) continue;
 976         if (node->link == NULL) {
 977             int fd;
 978             clusterLink *link;
 979
 980             fd = anetTcpNonBlockConnect(server.neterr, node->ip,
 981                 node->port+REDIS_CLUSTER_PORT_INCR);
 982             if (fd == -1) continue;
 983             link = createClusterLink(node);
 984             link->fd = fd;
 985             node->link = link;
 986             aeCreateFileEvent(server.el,link->fd,AE_READABLE,clusterReadHandler,link);
 987             /* If the node is flagged as MEET, we send a MEET message instead
 988              * of a PING one, to force the receiver to add us in its node
 989              * table. */
 990             clusterSendPing(link, node->flags & REDIS_NODE_MEET ?
 991                     CLUSTERMSG_TYPE_MEET : CLUSTERMSG_TYPE_PING);
 992             /* We can clear the flag after the first packet is sent.
 993              * If we'll never receive a PONG, we'll never send new packets
 994              * to this node. Instead after the PONG is received and we
 995              * are no longer in meet/handshake status, we want to send
 996              * normal PING packets. */
 997             node->flags &= ~REDIS_NODE_MEET;
 998
 999             redisLog(REDIS_NOTICE,"Connecting with Node %.40s at %s:%d", node->name, node->ip, node->port+REDIS_CLUSTER_PORT_INCR);
1000         }
1001     }
1002     dictReleaseIterator(di);
1003
1004     /* Ping some random node. Check a few random nodes and ping the one with
1005      * the oldest ping_sent time */
1006     for (j = 0; j < 5; j++) {
1007         de = dictGetRandomKey(server.cluster.nodes);
1008         clusterNode *this = dictGetVal(de);
1009
1010         if (this->link == NULL) continue;
1011         if (this->flags & (REDIS_NODE_MYSELF|REDIS_NODE_HANDSHAKE)) continue;
1012         if (min_ping_node == NULL || min_ping_sent > this->ping_sent) {
1013             min_ping_node = this;
1014             min_ping_sent = this->ping_sent;
1015         }
1016     }
1017     if (min_ping_node) {
1018         redisLog(REDIS_DEBUG,"Pinging node %40s", min_ping_node->name);
1019         clusterSendPing(min_ping_node->link, CLUSTERMSG_TYPE_PING);
1020     }
1021
1022     /* Iterate nodes to check if we need to flag something as failing */
1023     di = dictGetIterator(server.cluster.nodes);
1024     while((de = dictNext(di)) != NULL) {
1025         clusterNode *node = dictGetVal(de);
1026         int delay;
1027
1028         if (node->flags &
1029             (REDIS_NODE_MYSELF|REDIS_NODE_NOADDR|REDIS_NODE_HANDSHAKE))
1030                 continue;
1031         /* Check only if we already sent a ping and did not received
1032          * a reply yet. */
1033         if (node->ping_sent == 0 ||
1034             node->ping_sent <= node->pong_received) continue;
1035
1036         delay = time(NULL) - node->pong_received;
1037         if (delay < server.cluster.node_timeout) {
1038             /* The PFAIL condition can be reversed without external
1039              * help if it is not transitive (that is, if it does not
1040              * turn into a FAIL state).
1041              *
1042              * The FAIL condition is also reversible if there are no slaves
1043              * for this host, so no slave election should be in progress.
1044              *
1045              * TODO: consider all the implications of resurrecting a
1046              * FAIL node. */
1047             if (node->flags & REDIS_NODE_PFAIL) {
1048                 node->flags &= ~REDIS_NODE_PFAIL;
1049             } else if (node->flags & REDIS_NODE_FAIL && !node->numslaves) {
1050                 node->flags &= ~REDIS_NODE_FAIL;
1051                 clusterUpdateState();
1052             }
1053         } else {
1054             /* Timeout reached. Set the noad se possibly failing if it is
1055              * not already in this state. */
1056             if (!(node->flags & (REDIS_NODE_PFAIL|REDIS_NODE_FAIL))) {
1057                 redisLog(REDIS_DEBUG,"*** NODE %.40s possibly failing",
1058                     node->name);
1059                 node->flags |= REDIS_NODE_PFAIL;
1060             }
1061         }
1062     }
1063     dictReleaseIterator(di);
1064 }
1065
1066 /* -----------------------------------------------------------------------------
1067  * Slots management
1068  * -------------------------------------------------------------------------- */
1069
1070 /* Set the slot bit and return the old value. */
1071 int clusterNodeSetSlotBit(clusterNode *n, int slot) {
1072     off_t byte = slot/8;
1073     int bit = slot&7;
1074     int old = (n->slots[byte] & (1<<bit)) != 0;
1075     n->slots[byte] |= 1<<bit;
1076     return old;
1077 }
1078
1079 /* Clear the slot bit and return the old value. */
1080 int clusterNodeClearSlotBit(clusterNode *n, int slot) {
1081     off_t byte = slot/8;
1082     int bit = slot&7;
1083     int old = (n->slots[byte] & (1<<bit)) != 0;
1084     n->slots[byte] &= ~(1<<bit);
1085     return old;
1086 }
1087
1088 /* Return the slot bit from the cluster node structure. */
1089 int clusterNodeGetSlotBit(clusterNode *n, int slot) {
1090     off_t byte = slot/8;
1091     int bit = slot&7;
1092     return (n->slots[byte] & (1<<bit)) != 0;
1093 }
1094
1095 /* Add the specified slot to the list of slots that node 'n' will
1096  * serve. Return REDIS_OK if the operation ended with success.
1097  * If the slot is already assigned to another instance this is considered
1098  * an error and REDIS_ERR is returned. */
1099 int clusterAddSlot(clusterNode *n, int slot) {
1100     if (clusterNodeSetSlotBit(n,slot) != 0)
1101         return REDIS_ERR;
1102     server.cluster.slots[slot] = n;
1103     return REDIS_OK;
1104 }
1105
1106 /* Delete the specified slot marking it as unassigned.
1107  * Returns REDIS_OK if the slot was assigned, otherwise if the slot was
1108  * already unassigned REDIS_ERR is returned. */
1109 int clusterDelSlot(int slot) {
1110     clusterNode *n = server.cluster.slots[slot];
1111
1112     if (!n) return REDIS_ERR;
1113     redisAssert(clusterNodeClearSlotBit(n,slot) == 1);
1114     server.cluster.slots[slot] = NULL;
1115     return REDIS_OK;
1116 }
1117
1118 /* -----------------------------------------------------------------------------
1119  * Cluster state evaluation function
1120  * -------------------------------------------------------------------------- */
1121 void clusterUpdateState(void) {
1122     int ok = 1;
1123     int j;
1124
1125     for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) {
1126         if (server.cluster.slots[j] == NULL ||
1127             server.cluster.slots[j]->flags & (REDIS_NODE_FAIL))
1128         {
1129             ok = 0;
1130             break;
1131         }
1132     }
1133     if (ok) {
1134         if (server.cluster.state == REDIS_CLUSTER_NEEDHELP) {
1135             server.cluster.state = REDIS_CLUSTER_NEEDHELP;
1136         } else {
1137             server.cluster.state = REDIS_CLUSTER_OK;
1138         }
1139     } else {
1140         server.cluster.state = REDIS_CLUSTER_FAIL;
1141     }
1142 }
1143
1144 /* -----------------------------------------------------------------------------
1145  * CLUSTER command
1146  * -------------------------------------------------------------------------- */
1147
1148 sds clusterGenNodesDescription(void) {
1149     sds ci = sdsempty();
1150     dictIterator *di;
1151     dictEntry *de;
1152     int j, start;
1153
1154     di = dictGetIterator(server.cluster.nodes);
1155     while((de = dictNext(di)) != NULL) {
1156         clusterNode *node = dictGetVal(de);
1157
1158         /* Node coordinates */
1159         ci = sdscatprintf(ci,"%.40s %s:%d ",
1160             node->name,
1161             node->ip,
1162             node->port);
1163
1164         /* Flags */
1165         if (node->flags == 0) ci = sdscat(ci,"noflags,");
1166         if (node->flags & REDIS_NODE_MYSELF) ci = sdscat(ci,"myself,");
1167         if (node->flags & REDIS_NODE_MASTER) ci = sdscat(ci,"master,");
1168         if (node->flags & REDIS_NODE_SLAVE) ci = sdscat(ci,"slave,");
1169         if (node->flags & REDIS_NODE_PFAIL) ci = sdscat(ci,"fail?,");
1170         if (node->flags & REDIS_NODE_FAIL) ci = sdscat(ci,"fail,");
1171         if (node->flags & REDIS_NODE_HANDSHAKE) ci =sdscat(ci,"handshake,");
1172         if (node->flags & REDIS_NODE_NOADDR) ci = sdscat(ci,"noaddr,");
1173         if (ci[sdslen(ci)-1] == ',') ci[sdslen(ci)-1] = ' ';
1174
1175         /* Slave of... or just "-" */
1176         if (node->slaveof)
1177             ci = sdscatprintf(ci,"%.40s ",node->slaveof->name);
1178         else
1179             ci = sdscatprintf(ci,"- ");
1180
1181         /* Latency from the POV of this node, link status */
1182         ci = sdscatprintf(ci,"%ld %ld %s",
1183             (long) node->ping_sent,
1184             (long) node->pong_received,
1185             (node->link || node->flags & REDIS_NODE_MYSELF) ?
1186                         "connected" : "disconnected");
1187
1188         /* Slots served by this instance */
1189         start = -1;
1190         for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) {
1191             int bit;
1192
1193             if ((bit = clusterNodeGetSlotBit(node,j)) != 0) {
1194                 if (start == -1) start = j;
1195             }
1196             if (start != -1 && (!bit || j == REDIS_CLUSTER_SLOTS-1)) {
1197                 if (j == REDIS_CLUSTER_SLOTS-1) j++;
1198
1199                 if (start == j-1) {
1200                     ci = sdscatprintf(ci," %d",start);
1201                 } else {
1202                     ci = sdscatprintf(ci," %d-%d",start,j-1);
1203                 }
1204                 start = -1;
1205             }
1206         }
1207
1208         /* Just for MYSELF node we also dump info about slots that
1209          * we are migrating to other instances or importing from other
1210          * instances. */
1211         if (node->flags & REDIS_NODE_MYSELF) {
1212             for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) {
1213                 if (server.cluster.migrating_slots_to[j]) {
1214                     ci = sdscatprintf(ci," [%d->-%.40s]",j,
1215                         server.cluster.migrating_slots_to[j]->name);
1216                 } else if (server.cluster.importing_slots_from[j]) {
1217                     ci = sdscatprintf(ci," [%d-<-%.40s]",j,
1218                         server.cluster.importing_slots_from[j]->name);
1219                 }
1220             }
1221         }
1222         ci = sdscatlen(ci,"\n",1);
1223     }
1224     dictReleaseIterator(di);
1225     return ci;
1226 }
1227
1228 int getSlotOrReply(redisClient *c, robj *o) {
1229     long long slot;
1230
1231     if (getLongLongFromObject(o,&slot) != REDIS_OK ||
1232         slot < 0 || slot > REDIS_CLUSTER_SLOTS)
1233     {
1234         addReplyError(c,"Invalid or out of range slot");
1235         return -1;
1236     }
1237     return (int) slot;
1238 }
1239
1240 void clusterCommand(redisClient *c) {
1241     if (server.cluster_enabled == 0) {
1242         addReplyError(c,"This instance has cluster support disabled");
1243         return;
1244     }
1245
1246     if (!strcasecmp(c->argv[1]->ptr,"meet") && c->argc == 4) {
1247         clusterNode *n;
1248         struct sockaddr_in sa;
1249         long port;
1250
1251         /* Perform sanity checks on IP/port */
1252         if (inet_aton(c->argv[2]->ptr,&sa.sin_addr) == 0) {
1253             addReplyError(c,"Invalid IP address in MEET");
1254             return;
1255         }
1256         if (getLongFromObjectOrReply(c, c->argv[3], &port, NULL) != REDIS_OK ||
1257                     port < 0 || port > (65535-REDIS_CLUSTER_PORT_INCR))
1258         {
1259             addReplyError(c,"Invalid TCP port specified");
1260             return;
1261         }
1262
1263         /* Finally add the node to the cluster with a random name, this
1264          * will get fixed in the first handshake (ping/pong). */
1265         n = createClusterNode(NULL,REDIS_NODE_HANDSHAKE|REDIS_NODE_MEET);
1266         strncpy(n->ip,inet_ntoa(sa.sin_addr),sizeof(n->ip));
1267         n->port = port;
1268         clusterAddNode(n);
1269         addReply(c,shared.ok);
1270     } else if (!strcasecmp(c->argv[1]->ptr,"nodes") && c->argc == 2) {
1271         robj *o;
1272         sds ci = clusterGenNodesDescription();
1273
1274         o = createObject(REDIS_STRING,ci);
1275         addReplyBulk(c,o);
1276         decrRefCount(o);
1277     } else if ((!strcasecmp(c->argv[1]->ptr,"addslots") ||
1278                !strcasecmp(c->argv[1]->ptr,"delslots")) && c->argc >= 3)
1279     {
1280         /* CLUSTER ADDSLOTS <slot> [slot] ... */
1281         /* CLUSTER DELSLOTS <slot> [slot] ... */
1282         int j, slot;
1283         unsigned char *slots = zmalloc(REDIS_CLUSTER_SLOTS);
1284         int del = !strcasecmp(c->argv[1]->ptr,"delslots");
1285
1286         memset(slots,0,REDIS_CLUSTER_SLOTS);
1287         /* Check that all the arguments are parsable and that all the
1288          * slots are not already busy. */
1289         for (j = 2; j < c->argc; j++) {
1290             if ((slot = getSlotOrReply(c,c->argv[j])) == -1) {
1291                 zfree(slots);
1292                 return;
1293             }
1294             if (del && server.cluster.slots[slot] == NULL) {
1295                 addReplyErrorFormat(c,"Slot %d is already unassigned", slot);
1296                 zfree(slots);
1297                 return;
1298             } else if (!del && server.cluster.slots[slot]) {
1299                 addReplyErrorFormat(c,"Slot %d is already busy", slot);
1300                 zfree(slots);
1301                 return;
1302             }
1303             if (slots[slot]++ == 1) {
1304                 addReplyErrorFormat(c,"Slot %d specified multiple times",
1305                     (int)slot);
1306                 zfree(slots);
1307                 return;
1308             }
1309         }
1310         for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) {
1311             if (slots[j]) {
1312                 int retval;
1313
1314                 /* If this slot was set as importing we can clear this
1315                  * state as now we are the real owner of the slot. */
1316                 if (server.cluster.importing_slots_from[j])
1317                     server.cluster.importing_slots_from[j] = NULL;
1318
1319                 retval = del ? clusterDelSlot(j) :
1320                                clusterAddSlot(server.cluster.myself,j);
1321                 redisAssertWithInfo(c,NULL,retval == REDIS_OK);
1322             }
1323         }
1324         zfree(slots);
1325         clusterUpdateState();
1326         clusterSaveConfigOrDie();
1327         addReply(c,shared.ok);
1328     } else if (!strcasecmp(c->argv[1]->ptr,"setslot") && c->argc >= 4) {
1329         /* SETSLOT 10 MIGRATING <node ID> */
1330         /* SETSLOT 10 IMPORTING <node ID> */
1331         /* SETSLOT 10 STABLE */
1332         /* SETSLOT 10 NODE <node ID> */
1333         int slot;
1334         clusterNode *n;
1335
1336         if ((slot = getSlotOrReply(c,c->argv[2])) == -1) return;
1337
1338         if (!strcasecmp(c->argv[3]->ptr,"migrating") && c->argc == 5) {
1339             if (server.cluster.slots[slot] != server.cluster.myself) {
1340                 addReplyErrorFormat(c,"I'm not the owner of hash slot %u",slot);
1341                 return;
1342             }
1343             if ((n = clusterLookupNode(c->argv[4]->ptr)) == NULL) {
1344                 addReplyErrorFormat(c,"I don't know about node %s",
1345                     (char*)c->argv[4]->ptr);
1346                 return;
1347             }
1348             server.cluster.migrating_slots_to[slot] = n;
1349         } else if (!strcasecmp(c->argv[3]->ptr,"importing") && c->argc == 5) {
1350             if (server.cluster.slots[slot] == server.cluster.myself) {
1351                 addReplyErrorFormat(c,
1352                     "I'm already the owner of hash slot %u",slot);
1353                 return;
1354             }
1355             if ((n = clusterLookupNode(c->argv[4]->ptr)) == NULL) {
1356                 addReplyErrorFormat(c,"I don't know about node %s",
1357                     (char*)c->argv[3]->ptr);
1358                 return;
1359             }
1360             server.cluster.importing_slots_from[slot] = n;
1361         } else if (!strcasecmp(c->argv[3]->ptr,"stable") && c->argc == 4) {
1362             /* CLUSTER SETSLOT <SLOT> STABLE */
1363             server.cluster.importing_slots_from[slot] = NULL;
1364             server.cluster.migrating_slots_to[slot] = NULL;
1365         } else if (!strcasecmp(c->argv[3]->ptr,"node") && c->argc == 5) {
1366             /* CLUSTER SETSLOT <SLOT> NODE <NODE ID> */
1367             clusterNode *n = clusterLookupNode(c->argv[4]->ptr);
1368
1369             if (!n) addReplyErrorFormat(c,"Unknown node %s",
1370                 (char*)c->argv[4]->ptr);
1371             /* If this hash slot was served by 'myself' before to switch
1372              * make sure there are no longer local keys for this hash slot. */
1373             if (server.cluster.slots[slot] == server.cluster.myself &&
1374                 n != server.cluster.myself)
1375             {
1376                 int numkeys;
1377                 robj **keys;
1378
1379                 keys = zmalloc(sizeof(robj*)*1);
1380                 numkeys = GetKeysInSlot(slot, keys, 1);
1381                 zfree(keys);
1382                 if (numkeys != 0) {
1383                     addReplyErrorFormat(c, "Can't assign hashslot %d to a different node while I still hold keys for this hash slot.", slot);
1384                     return;
1385                 }
1386             }
1387             /* If this node was the slot owner and the slot was marked as
1388              * migrating, assigning the slot to another node will clear
1389              * the migratig status. */
1390             if (server.cluster.slots[slot] == server.cluster.myself &&
1391                 server.cluster.migrating_slots_to[slot])
1392                 server.cluster.migrating_slots_to[slot] = NULL;
1393
1394             /* If this node was importing this slot, assigning the slot to
1395              * itself also clears the importing status. */
1396             if (n == server.cluster.myself && server.cluster.importing_slots_from[slot])
1397                 server.cluster.importing_slots_from[slot] = NULL;
1398
1399             clusterDelSlot(slot);
1400             clusterAddSlot(n,slot);
1401         } else {
1402             addReplyError(c,"Invalid CLUSTER SETSLOT action or number of arguments");
1403             return;
1404         }
1405         clusterSaveConfigOrDie();
1406         addReply(c,shared.ok);
1407     } else if (!strcasecmp(c->argv[1]->ptr,"info") && c->argc == 2) {
1408         char *statestr[] = {"ok","fail","needhelp"};
1409         int slots_assigned = 0, slots_ok = 0, slots_pfail = 0, slots_fail = 0;
1410         int j;
1411
1412         for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) {
1413             clusterNode *n = server.cluster.slots[j];
1414
1415             if (n == NULL) continue;
1416             slots_assigned++;
1417             if (n->flags & REDIS_NODE_FAIL) {
1418                 slots_fail++;
1419             } else if (n->flags & REDIS_NODE_PFAIL) {
1420                 slots_pfail++;
1421             } else {
1422                 slots_ok++;
1423             }
1424         }
1425
1426         sds info = sdscatprintf(sdsempty(),
1427             "cluster_state:%s\r\n"
1428             "cluster_slots_assigned:%d\r\n"
1429             "cluster_slots_ok:%d\r\n"
1430             "cluster_slots_pfail:%d\r\n"
1431             "cluster_slots_fail:%d\r\n"
1432             "cluster_known_nodes:%lu\r\n"
1433             , statestr[server.cluster.state],
1434             slots_assigned,
1435             slots_ok,
1436             slots_pfail,
1437             slots_fail,
1438             dictSize(server.cluster.nodes)
1439         );
1440         addReplySds(c,sdscatprintf(sdsempty(),"$%lu\r\n",
1441             (unsigned long)sdslen(info)));
1442         addReplySds(c,info);
1443         addReply(c,shared.crlf);
1444     } else if (!strcasecmp(c->argv[1]->ptr,"keyslot") && c->argc == 3) {
1445         sds key = c->argv[2]->ptr;
1446
1447         addReplyLongLong(c,keyHashSlot(key,sdslen(key)));
1448     } else if (!strcasecmp(c->argv[1]->ptr,"getkeysinslot") && c->argc == 4) {
1449         long long maxkeys, slot;
1450         unsigned int numkeys, j;
1451         robj **keys;
1452
1453         if (getLongLongFromObjectOrReply(c,c->argv[2],&slot,NULL) != REDIS_OK)
1454             return;
1455         if (getLongLongFromObjectOrReply(c,c->argv[3],&maxkeys,NULL) != REDIS_OK)
1456             return;
1457         if (slot < 0 || slot >= REDIS_CLUSTER_SLOTS || maxkeys < 0 ||
1458             maxkeys > 1024*1024) {
1459             addReplyError(c,"Invalid slot or number of keys");
1460             return;
1461         }
1462
1463         keys = zmalloc(sizeof(robj*)*maxkeys);
1464         numkeys = GetKeysInSlot(slot, keys, maxkeys);
1465         addReplyMultiBulkLen(c,numkeys);
1466         for (j = 0; j < numkeys; j++) addReplyBulk(c,keys[j]);
1467         zfree(keys);
1468     } else {
1469         addReplyError(c,"Wrong CLUSTER subcommand or number of arguments");
1470     }
1471 }
1472
1473 /* -----------------------------------------------------------------------------
1474  * RESTORE and MIGRATE commands
1475  * -------------------------------------------------------------------------- */
1476
1477 /* RESTORE key ttl serialized-value */
1478 void restoreCommand(redisClient *c) {
1479     long ttl;
1480     rio payload;
1481     int type;
1482     robj *obj;
1483
1484     /* Make sure this key does not already exist here... */
1485     if (lookupKeyWrite(c->db,c->argv[1]) != NULL) {
1486         addReplyError(c,"Target key name is busy.");
1487         return;
1488     }
1489
1490     /* Check if the TTL value makes sense */
1491     if (getLongFromObjectOrReply(c,c->argv[2],&ttl,NULL) != REDIS_OK) {
1492         return;
1493     } else if (ttl < 0) {
1494         addReplyError(c,"Invalid TTL value, must be >= 0");
1495         return;
1496     }
1497
1498     rioInitWithBuffer(&payload,c->argv[3]->ptr);
1499     if (((type = rdbLoadObjectType(&payload)) == -1) ||
1500         ((obj = rdbLoadObject(type,&payload)) == NULL))
1501     {
1502         addReplyError(c,"Bad data format");
1503         return;
1504     }
1505
1506     /* Create the key and set the TTL if any */
1507     dbAdd(c->db,c->argv[1],obj);
1508     if (ttl) setExpire(c->db,c->argv[1],time(NULL)+ttl);
1509     signalModifiedKey(c->db,c->argv[1]);
1510     addReply(c,shared.ok);
1511     server.dirty++;
1512 }
1513
1514 /* MIGRATE host port key dbid timeout */
1515 void migrateCommand(redisClient *c) {
1516     int fd;
1517     long timeout;
1518     long dbid;
1519     time_t ttl;
1520     robj *o;
1521     rio cmd, payload;
1522
1523     /* Sanity check */
1524     if (getLongFromObjectOrReply(c,c->argv[5],&timeout,NULL) != REDIS_OK)
1525         return;
1526     if (getLongFromObjectOrReply(c,c->argv[4],&dbid,NULL) != REDIS_OK)
1527         return;
1528     if (timeout <= 0) timeout = 1;
1529
1530     /* Check if the key is here. If not we reply with success as there is
1531      * nothing to migrate (for instance the key expired in the meantime), but
1532      * we include such information in the reply string. */
1533     if ((o = lookupKeyRead(c->db,c->argv[3])) == NULL) {
1534         addReplySds(c,sdsnew("+NOKEY\r\n"));
1535         return;
1536     }
1537
1538     /* Connect */
1539     fd = anetTcpNonBlockConnect(server.neterr,c->argv[1]->ptr,
1540                 atoi(c->argv[2]->ptr));
1541     if (fd == -1) {
1542         addReplyErrorFormat(c,"Can't connect to target node: %s",
1543             server.neterr);
1544         return;
1545     }
1546     if ((aeWait(fd,AE_WRITABLE,timeout*1000) & AE_WRITABLE) == 0) {
1547         addReplyError(c,"Timeout connecting to the client");
1548         return;
1549     }
1550
1551     rioInitWithBuffer(&cmd,sdsempty());
1552     redisAssertWithInfo(c,NULL,rioWriteBulkCount(&cmd,'*',2));
1553     redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,"SELECT",6));
1554     redisAssertWithInfo(c,NULL,rioWriteBulkLongLong(&cmd,dbid));
1555
1556     ttl = getExpire(c->db,c->argv[3]);
1557     redisAssertWithInfo(c,NULL,rioWriteBulkCount(&cmd,'*',4));
1558     redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,"RESTORE",7));
1559     redisAssertWithInfo(c,NULL,c->argv[3]->encoding == REDIS_ENCODING_RAW);
1560     redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,c->argv[3]->ptr,sdslen(c->argv[3]->ptr)));
1561     redisAssertWithInfo(c,NULL,rioWriteBulkLongLong(&cmd,(ttl == -1) ? 0 : ttl));
1562
1563     /* Finally the last argument that is the serailized object payload
1564      * in the form: <type><rdb-serialized-object>. */
1565     rioInitWithBuffer(&payload,sdsempty());
1566     redisAssertWithInfo(c,NULL,rdbSaveObjectType(&payload,o));
1567     redisAssertWithInfo(c,NULL,rdbSaveObject(&payload,o) != -1);
1568     redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,payload.io.buffer.ptr,sdslen(payload.io.buffer.ptr)));
1569     sdsfree(payload.io.buffer.ptr);
1570
1571     /* Tranfer the query to the other node in 64K chunks. */
1572     {
1573         sds buf = cmd.io.buffer.ptr;
1574         size_t pos = 0, towrite;
1575         int nwritten = 0;
1576
1577         while ((towrite = sdslen(buf)-pos) > 0) {
1578             towrite = (towrite > (64*1024) ? (64*1024) : towrite);
1579             nwritten = syncWrite(fd,buf+nwritten,towrite,timeout);
1580             if (nwritten != (signed)towrite) goto socket_wr_err;
1581             pos += nwritten;
1582         }
1583     }
1584
1585     /* Read back the reply. */
1586     {
1587         char buf1[1024];
1588         char buf2[1024];
1589
1590         /* Read the two replies */
1591         if (syncReadLine(fd, buf1, sizeof(buf1), timeout) <= 0)
1592             goto socket_rd_err;
1593         if (syncReadLine(fd, buf2, sizeof(buf2), timeout) <= 0)
1594             goto socket_rd_err;
1595         if (buf1[0] == '-' || buf2[0] == '-') {
1596             addReplyErrorFormat(c,"Target instance replied with error: %s",
1597                 (buf1[0] == '-') ? buf1+1 : buf2+1);
1598         } else {
1599             robj *aux;
1600
1601             dbDelete(c->db,c->argv[3]);
1602             signalModifiedKey(c->db,c->argv[3]);
1603             addReply(c,shared.ok);
1604             server.dirty++;
1605
1606             /* Translate MIGRATE as DEL for replication/AOF. */
1607             aux = createStringObject("DEL",3);
1608             rewriteClientCommandVector(c,2,aux,c->argv[3]);
1609             decrRefCount(aux);
1610         }
1611     }
1612
1613     sdsfree(cmd.io.buffer.ptr);
1614     close(fd);
1615     return;
1616
1617 socket_wr_err:
1618     redisLog(REDIS_NOTICE,"Can't write to target node for MIGRATE: %s",
1619         strerror(errno));
1620     addReplyErrorFormat(c,"MIGRATE failed, writing to target node: %s.",
1621         strerror(errno));
1622     sdsfree(cmd.io.buffer.ptr);
1623     close(fd);
1624     return;
1625
1626 socket_rd_err:
1627     redisLog(REDIS_NOTICE,"Can't read from target node for MIGRATE: %s",
1628         strerror(errno));
1629     addReplyErrorFormat(c,"MIGRATE failed, reading from target node: %s.",
1630         strerror(errno));
1631     sdsfree(cmd.io.buffer.ptr);
1632     close(fd);
1633     return;
1634 }
1635
1636 /* DUMP keyname
1637  * DUMP is actually not used by Redis Cluster but it is the obvious
1638  * complement of RESTORE and can be useful for different applications. */
1639 void dumpCommand(redisClient *c) {
1640     robj *o, *dumpobj;
1641     rio payload;
1642
1643     /* Check if the key is here. */
1644     if ((o = lookupKeyRead(c->db,c->argv[1])) == NULL) {
1645         addReply(c,shared.nullbulk);
1646         return;
1647     }
1648
1649     /* Serialize the object in a RDB-like format. It consist of an object type
1650      * byte followed by the serialized object. This is understood by RESTORE. */
1651     rioInitWithBuffer(&payload,sdsempty());
1652     redisAssertWithInfo(c,NULL,rdbSaveObjectType(&payload,o));
1653     redisAssertWithInfo(c,NULL,rdbSaveObject(&payload,o));
1654
1655     /* Transfer to the client */
1656     dumpobj = createObject(REDIS_STRING,payload.io.buffer.ptr);
1657     addReplyBulk(c,dumpobj);
1658     decrRefCount(dumpobj);
1659     return;
1660 }
1661
1662 /* The ASKING command is required after a -ASK redirection.
1663  * The client should issue ASKING before to actualy send the command to
1664  * the target instance. See the Redis Cluster specification for more
1665  * information. */
1666 void askingCommand(redisClient *c) {
1667     if (server.cluster_enabled == 0) {
1668         addReplyError(c,"This instance has cluster support disabled");
1669         return;
1670     }
1671     c->flags |= REDIS_ASKING;
1672     addReply(c,shared.ok);
1673 }
1674
1675 /* -----------------------------------------------------------------------------
1676  * Cluster functions related to serving / redirecting clients
1677  * -------------------------------------------------------------------------- */
1678
1679 /* Return the pointer to the cluster node that is able to serve the query
1680  * as all the keys belong to hash slots for which the node is in charge.
1681  *
1682  * If the returned node should be used only for this request, the *ask
1683  * integer is set to '1', otherwise to '0'. This is used in order to
1684  * let the caller know if we should reply with -MOVED or with -ASK.
1685  *
1686  * If the request contains more than a single key NULL is returned,
1687  * however a request with more then a key argument where the key is always
1688  * the same is valid, like in: RPOPLPUSH mylist mylist.*/
1689 clusterNode *getNodeByQuery(redisClient *c, struct redisCommand *cmd, robj **argv, int argc, int *hashslot, int *ask) {
1690     clusterNode *n = NULL;
1691     robj *firstkey = NULL;
1692     multiState *ms, _ms;
1693     multiCmd mc;
1694     int i, slot = 0;
1695
1696     /* We handle all the cases as if they were EXEC commands, so we have
1697      * a common code path for everything */
1698     if (cmd->proc == execCommand) {
1699         /* If REDIS_MULTI flag is not set EXEC is just going to return an
1700          * error. */
1701         if (!(c->flags & REDIS_MULTI)) return server.cluster.myself;
1702         ms = &c->mstate;
1703     } else {
1704         /* In order to have a single codepath create a fake Multi State
1705          * structure if the client is not in MULTI/EXEC state, this way
1706          * we have a single codepath below. */
1707         ms = &_ms;
1708         _ms.commands = &mc;
1709         _ms.count = 1;
1710         mc.argv = argv;
1711         mc.argc = argc;
1712         mc.cmd = cmd;
1713     }
1714
1715     /* Check that all the keys are the same key, and get the slot and
1716      * node for this key. */
1717     for (i = 0; i < ms->count; i++) {
1718         struct redisCommand *mcmd;
1719         robj **margv;
1720         int margc, *keyindex, numkeys, j;
1721
1722         mcmd = ms->commands[i].cmd;
1723         margc = ms->commands[i].argc;
1724         margv = ms->commands[i].argv;
1725
1726         keyindex = getKeysFromCommand(mcmd,margv,margc,&numkeys,
1727                                       REDIS_GETKEYS_ALL);
1728         for (j = 0; j < numkeys; j++) {
1729             if (firstkey == NULL) {
1730                 /* This is the first key we see. Check what is the slot
1731                  * and node. */
1732                 firstkey = margv[keyindex[j]];
1733
1734                 slot = keyHashSlot((char*)firstkey->ptr, sdslen(firstkey->ptr));
1735                 n = server.cluster.slots[slot];
1736                 redisAssertWithInfo(c,firstkey,n != NULL);
1737             } else {
1738                 /* If it is not the first key, make sure it is exactly
1739                  * the same key as the first we saw. */
1740                 if (!equalStringObjects(firstkey,margv[keyindex[j]])) {
1741                     decrRefCount(firstkey);
1742                     getKeysFreeResult(keyindex);
1743                     return NULL;
1744                 }
1745             }
1746         }
1747         getKeysFreeResult(keyindex);
1748     }
1749     if (ask) *ask = 0; /* This is the default. Set to 1 if needed later. */
1750     /* No key at all in command? then we can serve the request
1751      * without redirections. */
1752     if (n == NULL) return server.cluster.myself;
1753     if (hashslot) *hashslot = slot;
1754     /* This request is about a slot we are migrating into another instance?
1755      * Then we need to check if we have the key. If we have it we can reply.
1756      * If instead is a new key, we pass the request to the node that is
1757      * receiving the slot. */
1758     if (n == server.cluster.myself &&
1759         server.cluster.migrating_slots_to[slot] != NULL)
1760     {
1761         if (lookupKeyRead(&server.db[0],firstkey) == NULL) {
1762             if (ask) *ask = 1;
1763             return server.cluster.migrating_slots_to[slot];
1764         }
1765     }
1766     /* Handle the case in which we are receiving this hash slot from
1767      * another instance, so we'll accept the query even if in the table
1768      * it is assigned to a different node, but only if the client
1769      * issued an ASKING command before. */
1770     if (server.cluster.importing_slots_from[slot] != NULL &&
1771         c->flags & REDIS_ASKING) {
1772         return server.cluster.myself;
1773     }
1774     /* It's not a -ASK case. Base case: just return the right node. */
1775     return n;
1776 }