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 dictGetEntryVal(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 < 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 }
 497
 498 /* When this function is called, there is a packet to process starting
 499  * at node->rcvbuf. Releasing the buffer is up to the caller, so this
 500  * function should just handle the higher level stuff of processing the
 501  * packet, modifying the cluster state if needed.
 502  *
 503  * The function returns 1 if the link is still valid after the packet
 504  * was processed, otherwise 0 if the link was freed since the packet
 505  * processing lead to some inconsistency error (for instance a PONG
 506  * received from the wrong sender ID). */
 507 int clusterProcessPacket(clusterLink *link) {
 508     clusterMsg *hdr = (clusterMsg*) link->rcvbuf;
 509     uint32_t totlen = ntohl(hdr->totlen);
 510     uint16_t type = ntohs(hdr->type);
 511     clusterNode *sender;
 512
 513     redisLog(REDIS_DEBUG,"--- packet to process %lu bytes (%lu) ---",
 514         (unsigned long) totlen, sdslen(link->rcvbuf));
 515     if (totlen < 8) return 1;
 516     if (totlen > sdslen(link->rcvbuf)) return 1;
 517     if (type == CLUSTERMSG_TYPE_PING || type == CLUSTERMSG_TYPE_PONG ||
 518         type == CLUSTERMSG_TYPE_MEET)
 519     {
 520         uint16_t count = ntohs(hdr->count);
 521         uint32_t explen; /* expected length of this packet */
 522
 523         explen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
 524         explen += (sizeof(clusterMsgDataGossip)*count);
 525         if (totlen != explen) return 1;
 526     }
 527     if (type == CLUSTERMSG_TYPE_FAIL) {
 528         uint32_t explen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
 529
 530         explen += sizeof(clusterMsgDataFail);
 531         if (totlen != explen) return 1;
 532     }
 533
 534     sender = clusterLookupNode(hdr->sender);
 535     if (type == CLUSTERMSG_TYPE_PING || type == CLUSTERMSG_TYPE_MEET) {
 536         int update_config = 0;
 537         redisLog(REDIS_DEBUG,"Ping packet received: %p", link->node);
 538
 539         /* Add this node if it is new for us and the msg type is MEET.
 540          * In this stage we don't try to add the node with the right
 541          * flags, slaveof pointer, and so forth, as this details will be
 542          * resolved when we'll receive PONGs from the server. */
 543         if (!sender && type == CLUSTERMSG_TYPE_MEET) {
 544             clusterNode *node;
 545
 546             node = createClusterNode(NULL,REDIS_NODE_HANDSHAKE);
 547             nodeIp2String(node->ip,link);
 548             node->port = ntohs(hdr->port);
 549             clusterAddNode(node);
 550             update_config = 1;
 551         }
 552
 553         /* Get info from the gossip section */
 554         clusterProcessGossipSection(hdr,link);
 555
 556         /* Anyway reply with a PONG */
 557         clusterSendPing(link,CLUSTERMSG_TYPE_PONG);
 558
 559         /* Update config if needed */
 560         if (update_config) clusterSaveConfigOrDie();
 561     } else if (type == CLUSTERMSG_TYPE_PONG) {
 562         int update_state = 0;
 563         int update_config = 0;
 564
 565         redisLog(REDIS_DEBUG,"Pong packet received: %p", link->node);
 566         if (link->node) {
 567             if (link->node->flags & REDIS_NODE_HANDSHAKE) {
 568                 /* If we already have this node, try to change the
 569                  * IP/port of the node with the new one. */
 570                 if (sender) {
 571                     redisLog(REDIS_WARNING,
 572                         "Handshake error: we already know node %.40s, updating the address if needed.", sender->name);
 573                     nodeUpdateAddress(sender,link,ntohs(hdr->port));
 574                     freeClusterNode(link->node); /* will free the link too */
 575                     return 0;
 576                 }
 577
 578                 /* First thing to do is replacing the random name with the
 579                  * right node name if this was an handshake stage. */
 580                 clusterRenameNode(link->node, hdr->sender);
 581                 redisLog(REDIS_DEBUG,"Handshake with node %.40s completed.",
 582                     link->node->name);
 583                 link->node->flags &= ~REDIS_NODE_HANDSHAKE;
 584                 update_config = 1;
 585             } else if (memcmp(link->node->name,hdr->sender,
 586                         REDIS_CLUSTER_NAMELEN) != 0)
 587             {
 588                 /* If the reply has a non matching node ID we
 589                  * disconnect this node and set it as not having an associated
 590                  * address. */
 591                 redisLog(REDIS_DEBUG,"PONG contains mismatching sender ID");
 592                 link->node->flags |= REDIS_NODE_NOADDR;
 593                 freeClusterLink(link);
 594                 update_config = 1;
 595                 /* FIXME: remove this node if we already have it.
 596                  *
 597                  * If we already have it but the IP is different, use
 598                  * the new one if the old node is in FAIL, PFAIL, or NOADDR
 599                  * status... */
 600                 return 0;
 601             }
 602         }
 603         /* Update our info about the node */
 604         link->node->pong_received = time(NULL);
 605
 606         /* Update master/slave info */
 607         if (sender) {
 608             if (!memcmp(hdr->slaveof,REDIS_NODE_NULL_NAME,
 609                 sizeof(hdr->slaveof)))
 610             {
 611                 sender->flags &= ~REDIS_NODE_SLAVE;
 612                 sender->flags |= REDIS_NODE_MASTER;
 613                 sender->slaveof = NULL;
 614             } else {
 615                 clusterNode *master = clusterLookupNode(hdr->slaveof);
 616
 617                 sender->flags &= ~REDIS_NODE_MASTER;
 618                 sender->flags |= REDIS_NODE_SLAVE;
 619                 if (sender->numslaves) clusterNodeResetSlaves(sender);
 620                 if (master) clusterNodeAddSlave(master,sender);
 621             }
 622         }
 623
 624         /* Update our info about served slots if this new node is serving
 625          * slots that are not served from our point of view. */
 626         if (sender && sender->flags & REDIS_NODE_MASTER) {
 627             int newslots, j;
 628
 629             newslots =
 630                 memcmp(sender->slots,hdr->myslots,sizeof(hdr->myslots)) != 0;
 631             memcpy(sender->slots,hdr->myslots,sizeof(hdr->myslots));
 632             if (newslots) {
 633                 for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) {
 634                     if (clusterNodeGetSlotBit(sender,j)) {
 635                         if (server.cluster.slots[j] == sender) continue;
 636                         if (server.cluster.slots[j] == NULL ||
 637                             server.cluster.slots[j]->flags & REDIS_NODE_FAIL)
 638                         {
 639                             server.cluster.slots[j] = sender;
 640                             update_state = update_config = 1;
 641                         }
 642                     }
 643                 }
 644             }
 645         }
 646
 647         /* Get info from the gossip section */
 648         clusterProcessGossipSection(hdr,link);
 649
 650         /* Update the cluster state if needed */
 651         if (update_state) clusterUpdateState();
 652         if (update_config) clusterSaveConfigOrDie();
 653     } else if (type == CLUSTERMSG_TYPE_FAIL && sender) {
 654         clusterNode *failing;
 655
 656         failing = clusterLookupNode(hdr->data.fail.about.nodename);
 657         if (failing && !(failing->flags & (REDIS_NODE_FAIL|REDIS_NODE_MYSELF)))
 658         {
 659             redisLog(REDIS_NOTICE,
 660                 "FAIL message received from %.40s about %.40s",
 661                 hdr->sender, hdr->data.fail.about.nodename);
 662             failing->flags |= REDIS_NODE_FAIL;
 663             failing->flags &= ~REDIS_NODE_PFAIL;
 664             clusterUpdateState();
 665             clusterSaveConfigOrDie();
 666         }
 667     } else {
 668         redisLog(REDIS_NOTICE,"Received unknown packet type: %d", type);
 669     }
 670     return 1;
 671 }
 672
 673 /* This function is called when we detect the link with this node is lost.
 674    We set the node as no longer connected. The Cluster Cron will detect
 675    this connection and will try to get it connected again.
 676
 677    Instead if the node is a temporary node used to accept a query, we
 678    completely free the node on error. */
 679 void handleLinkIOError(clusterLink *link) {
 680     freeClusterLink(link);
 681 }
 682
 683 /* Send data. This is handled using a trivial send buffer that gets
 684  * consumed by write(). We don't try to optimize this for speed too much
 685  * as this is a very low traffic channel. */
 686 void clusterWriteHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
 687     clusterLink *link = (clusterLink*) privdata;
 688     ssize_t nwritten;
 689     REDIS_NOTUSED(el);
 690     REDIS_NOTUSED(mask);
 691
 692     nwritten = write(fd, link->sndbuf, sdslen(link->sndbuf));
 693     if (nwritten <= 0) {
 694         redisLog(REDIS_NOTICE,"I/O error writing to node link: %s",
 695             strerror(errno));
 696         handleLinkIOError(link);
 697         return;
 698     }
 699     link->sndbuf = sdsrange(link->sndbuf,nwritten,-1);
 700     if (sdslen(link->sndbuf) == 0)
 701         aeDeleteFileEvent(server.el, link->fd, AE_WRITABLE);
 702 }
 703
 704 /* Read data. Try to read the first field of the header first to check the
 705  * full length of the packet. When a whole packet is in memory this function
 706  * will call the function to process the packet. And so forth. */
 707 void clusterReadHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
 708     char buf[1024];
 709     ssize_t nread;
 710     clusterMsg *hdr;
 711     clusterLink *link = (clusterLink*) privdata;
 712     int readlen;
 713     REDIS_NOTUSED(el);
 714     REDIS_NOTUSED(mask);
 715
 716 again:
 717     if (sdslen(link->rcvbuf) >= 4) {
 718         hdr = (clusterMsg*) link->rcvbuf;
 719         readlen = ntohl(hdr->totlen) - sdslen(link->rcvbuf);
 720     } else {
 721         readlen = 4 - sdslen(link->rcvbuf);
 722     }
 723
 724     nread = read(fd,buf,readlen);
 725     if (nread == -1 && errno == EAGAIN) return; /* Just no data */
 726
 727     if (nread <= 0) {
 728         /* I/O error... */
 729         redisLog(REDIS_NOTICE,"I/O error reading from node link: %s",
 730             (nread == 0) ? "connection closed" : strerror(errno));
 731         handleLinkIOError(link);
 732         return;
 733     } else {
 734         /* Read data and recast the pointer to the new buffer. */
 735         link->rcvbuf = sdscatlen(link->rcvbuf,buf,nread);
 736         hdr = (clusterMsg*) link->rcvbuf;
 737     }
 738
 739     /* Total length obtained? read the payload now instead of burning
 740      * cycles waiting for a new event to fire. */
 741     if (sdslen(link->rcvbuf) == 4) goto again;
 742
 743     /* Whole packet in memory? We can process it. */
 744     if (sdslen(link->rcvbuf) == ntohl(hdr->totlen)) {
 745         if (clusterProcessPacket(link)) {
 746             sdsfree(link->rcvbuf);
 747             link->rcvbuf = sdsempty();
 748         }
 749     }
 750 }
 751
 752 /* Put stuff into the send buffer. */
 753 void clusterSendMessage(clusterLink *link, unsigned char *msg, size_t msglen) {
 754     if (sdslen(link->sndbuf) == 0 && msglen != 0)
 755         aeCreateFileEvent(server.el,link->fd,AE_WRITABLE,
 756                     clusterWriteHandler,link);
 757
 758     link->sndbuf = sdscatlen(link->sndbuf, msg, msglen);
 759 }
 760
 761 /* Build the message header */
 762 void clusterBuildMessageHdr(clusterMsg *hdr, int type) {
 763     int totlen;
 764
 765     memset(hdr,0,sizeof(*hdr));
 766     hdr->type = htons(type);
 767     memcpy(hdr->sender,server.cluster.myself->name,REDIS_CLUSTER_NAMELEN);
 768     memcpy(hdr->myslots,server.cluster.myself->slots,
 769         sizeof(hdr->myslots));
 770     memset(hdr->slaveof,0,REDIS_CLUSTER_NAMELEN);
 771     if (server.cluster.myself->slaveof != NULL) {
 772         memcpy(hdr->slaveof,server.cluster.myself->slaveof->name,
 773                                     REDIS_CLUSTER_NAMELEN);
 774     }
 775     hdr->port = htons(server.port);
 776     hdr->state = server.cluster.state;
 777     memset(hdr->configdigest,0,32); /* FIXME: set config digest */
 778
 779     if (type == CLUSTERMSG_TYPE_FAIL) {
 780         totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
 781         totlen += sizeof(clusterMsgDataFail);
 782     }
 783     hdr->totlen = htonl(totlen);
 784     /* For PING, PONG, and MEET, fixing the totlen field is up to the caller */
 785 }
 786
 787 /* Send a PING or PONG packet to the specified node, making sure to add enough
 788  * gossip informations. */
 789 void clusterSendPing(clusterLink *link, int type) {
 790     unsigned char buf[1024];
 791     clusterMsg *hdr = (clusterMsg*) buf;
 792     int gossipcount = 0, totlen;
 793     /* freshnodes is the number of nodes we can still use to populate the
 794      * gossip section of the ping packet. Basically we start with the nodes
 795      * we have in memory minus two (ourself and the node we are sending the
 796      * message to). Every time we add a node we decrement the counter, so when
 797      * it will drop to <= zero we know there is no more gossip info we can
 798      * send. */
 799     int freshnodes = dictSize(server.cluster.nodes)-2;
 800
 801     if (link->node && type == CLUSTERMSG_TYPE_PING)
 802         link->node->ping_sent = time(NULL);
 803     clusterBuildMessageHdr(hdr,type);
 804
 805     /* Populate the gossip fields */
 806     while(freshnodes > 0 && gossipcount < 3) {
 807         struct dictEntry *de = dictGetRandomKey(server.cluster.nodes);
 808         clusterNode *this = dictGetEntryVal(de);
 809         clusterMsgDataGossip *gossip;
 810         int j;
 811
 812         /* Not interesting to gossip about ourself.
 813          * Nor to send gossip info about HANDSHAKE state nodes (zero info). */
 814         if (this == server.cluster.myself ||
 815             this->flags & REDIS_NODE_HANDSHAKE) {
 816                 freshnodes--; /* otherwise we may loop forever. */
 817                 continue;
 818         }
 819
 820         /* Check if we already added this node */
 821         for (j = 0; j < gossipcount; j++) {
 822             if (memcmp(hdr->data.ping.gossip[j].nodename,this->name,
 823                     REDIS_CLUSTER_NAMELEN) == 0) break;
 824         }
 825         if (j != gossipcount) continue;
 826
 827         /* Add it */
 828         freshnodes--;
 829         gossip = &(hdr->data.ping.gossip[gossipcount]);
 830         memcpy(gossip->nodename,this->name,REDIS_CLUSTER_NAMELEN);
 831         gossip->ping_sent = htonl(this->ping_sent);
 832         gossip->pong_received = htonl(this->pong_received);
 833         memcpy(gossip->ip,this->ip,sizeof(this->ip));
 834         gossip->port = htons(this->port);
 835         gossip->flags = htons(this->flags);
 836         gossipcount++;
 837     }
 838     totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
 839     totlen += (sizeof(clusterMsgDataGossip)*gossipcount);
 840     hdr->count = htons(gossipcount);
 841     hdr->totlen = htonl(totlen);
 842     clusterSendMessage(link,buf,totlen);
 843 }
 844
 845 /* Send a message to all the nodes with a reliable link */
 846 void clusterBroadcastMessage(void *buf, size_t len) {
 847     dictIterator *di;
 848     dictEntry *de;
 849
 850     di = dictGetIterator(server.cluster.nodes);
 851     while((de = dictNext(di)) != NULL) {
 852         clusterNode *node = dictGetEntryVal(de);
 853
 854         if (!node->link) continue;
 855         if (node->flags & (REDIS_NODE_MYSELF|REDIS_NODE_NOADDR)) continue;
 856         clusterSendMessage(node->link,buf,len);
 857     }
 858     dictReleaseIterator(di);
 859 }
 860
 861 /* Send a FAIL message to all the nodes we are able to contact.
 862  * The FAIL message is sent when we detect that a node is failing
 863  * (REDIS_NODE_PFAIL) and we also receive a gossip confirmation of this:
 864  * we switch the node state to REDIS_NODE_FAIL and ask all the other
 865  * nodes to do the same ASAP. */
 866 void clusterSendFail(char *nodename) {
 867     unsigned char buf[1024];
 868     clusterMsg *hdr = (clusterMsg*) buf;
 869
 870     clusterBuildMessageHdr(hdr,CLUSTERMSG_TYPE_FAIL);
 871     memcpy(hdr->data.fail.about.nodename,nodename,REDIS_CLUSTER_NAMELEN);
 872     clusterBroadcastMessage(buf,ntohl(hdr->totlen));
 873 }
 874
 875 /* -----------------------------------------------------------------------------
 876  * CLUSTER cron job
 877  * -------------------------------------------------------------------------- */
 878
 879 /* This is executed 1 time every second */
 880 void clusterCron(void) {
 881     dictIterator *di;
 882     dictEntry *de;
 883     int j;
 884     time_t min_ping_sent = 0;
 885     clusterNode *min_ping_node = NULL;
 886
 887     /* Check if we have disconnected nodes and reestablish the connection. */
 888     di = dictGetIterator(server.cluster.nodes);
 889     while((de = dictNext(di)) != NULL) {
 890         clusterNode *node = dictGetEntryVal(de);
 891
 892         if (node->flags & (REDIS_NODE_MYSELF|REDIS_NODE_NOADDR)) continue;
 893         if (node->link == NULL) {
 894             int fd;
 895             clusterLink *link;
 896
 897             fd = anetTcpNonBlockConnect(server.neterr, node->ip,
 898                 node->port+REDIS_CLUSTER_PORT_INCR);
 899             if (fd == -1) continue;
 900             link = createClusterLink(node);
 901             link->fd = fd;
 902             node->link = link;
 903             aeCreateFileEvent(server.el,link->fd,AE_READABLE,clusterReadHandler,link);
 904             /* If the node is flagged as MEET, we send a MEET message instead
 905              * of a PING one, to force the receiver to add us in its node
 906              * table. */
 907             clusterSendPing(link, node->flags & REDIS_NODE_MEET ?
 908                     CLUSTERMSG_TYPE_MEET : CLUSTERMSG_TYPE_PING);
 909             /* We can clear the flag after the first packet is sent.
 910              * If we'll never receive a PONG, we'll never send new packets
 911              * to this node. Instead after the PONG is received and we
 912              * are no longer in meet/handshake status, we want to send
 913              * normal PING packets. */
 914             node->flags &= ~REDIS_NODE_MEET;
 915
 916             redisLog(REDIS_NOTICE,"Connecting with Node %.40s at %s:%d", node->name, node->ip, node->port+REDIS_CLUSTER_PORT_INCR);
 917         }
 918     }
 919     dictReleaseIterator(di);
 920
 921     /* Ping some random node. Check a few random nodes and ping the one with
 922      * the oldest ping_sent time */
 923     for (j = 0; j < 5; j++) {
 924         de = dictGetRandomKey(server.cluster.nodes);
 925         clusterNode *this = dictGetEntryVal(de);
 926
 927         if (this->link == NULL) continue;
 928         if (this->flags & (REDIS_NODE_MYSELF|REDIS_NODE_HANDSHAKE)) continue;
 929         if (min_ping_node == NULL || min_ping_sent > this->ping_sent) {
 930             min_ping_node = this;
 931             min_ping_sent = this->ping_sent;
 932         }
 933     }
 934     if (min_ping_node) {
 935         redisLog(REDIS_DEBUG,"Pinging node %40s", min_ping_node->name);
 936         clusterSendPing(min_ping_node->link, CLUSTERMSG_TYPE_PING);
 937     }
 938
 939     /* Iterate nodes to check if we need to flag something as failing */
 940     di = dictGetIterator(server.cluster.nodes);
 941     while((de = dictNext(di)) != NULL) {
 942         clusterNode *node = dictGetEntryVal(de);
 943         int delay;
 944
 945         if (node->flags &
 946             (REDIS_NODE_MYSELF|REDIS_NODE_NOADDR|REDIS_NODE_HANDSHAKE))
 947                 continue;
 948         /* Check only if we already sent a ping and did not received
 949          * a reply yet. */
 950         if (node->ping_sent == 0 ||
 951             node->ping_sent <= node->pong_received) continue;
 952
 953         delay = time(NULL) - node->pong_received;
 954         if (delay < server.cluster.node_timeout) {
 955             /* The PFAIL condition can be reversed without external
 956              * help if it is not transitive (that is, if it does not
 957              * turn into a FAIL state).
 958              *
 959              * The FAIL condition is also reversible if there are no slaves
 960              * for this host, so no slave election should be in progress.
 961              *
 962              * TODO: consider all the implications of resurrecting a
 963              * FAIL node. */
 964             if (node->flags & REDIS_NODE_PFAIL) {
 965                 node->flags &= ~REDIS_NODE_PFAIL;
 966             } else if (node->flags & REDIS_NODE_FAIL && !node->numslaves) {
 967                 node->flags &= ~REDIS_NODE_FAIL;
 968                 clusterUpdateState();
 969             }
 970         } else {
 971             /* Timeout reached. Set the noad se possibly failing if it is
 972              * not already in this state. */
 973             if (!(node->flags & (REDIS_NODE_PFAIL|REDIS_NODE_FAIL))) {
 974                 redisLog(REDIS_DEBUG,"*** NODE %.40s possibly failing",
 975                     node->name);
 976                 node->flags |= REDIS_NODE_PFAIL;
 977             }
 978         }
 979     }
 980     dictReleaseIterator(di);
 981 }
 982
 983 /* -----------------------------------------------------------------------------
 984  * Slots management
 985  * -------------------------------------------------------------------------- */
 986
 987 /* Set the slot bit and return the old value. */
 988 int clusterNodeSetSlotBit(clusterNode *n, int slot) {
 989     off_t byte = slot/8;
 990     int bit = slot&7;
 991     int old = (n->slots[byte] & (1<<bit)) != 0;
 992     n->slots[byte] |= 1<<bit;
 993     return old;
 994 }
 995
 996 /* Clear the slot bit and return the old value. */
 997 int clusterNodeClearSlotBit(clusterNode *n, int slot) {
 998     off_t byte = slot/8;
 999     int bit = slot&7;
1000     int old = (n->slots[byte] & (1<<bit)) != 0;
1001     n->slots[byte] &= ~(1<<bit);
1002     return old;
1003 }
1004
1005 /* Return the slot bit from the cluster node structure. */
1006 int clusterNodeGetSlotBit(clusterNode *n, int slot) {
1007     off_t byte = slot/8;
1008     int bit = slot&7;
1009     return (n->slots[byte] & (1<<bit)) != 0;
1010 }
1011
1012 /* Add the specified slot to the list of slots that node 'n' will
1013  * serve. Return REDIS_OK if the operation ended with success.
1014  * If the slot is already assigned to another instance this is considered
1015  * an error and REDIS_ERR is returned. */
1016 int clusterAddSlot(clusterNode *n, int slot) {
1017     if (clusterNodeSetSlotBit(n,slot) != 0)
1018         return REDIS_ERR;
1019     server.cluster.slots[slot] = n;
1020     return REDIS_OK;
1021 }
1022
1023 /* Delete the specified slot marking it as unassigned.
1024  * Returns REDIS_OK if the slot was assigned, otherwise if the slot was
1025  * already unassigned REDIS_ERR is returned. */
1026 int clusterDelSlot(int slot) {
1027     clusterNode *n = server.cluster.slots[slot];
1028
1029     if (!n) return REDIS_ERR;
1030     redisAssert(clusterNodeClearSlotBit(n,slot) == 1);
1031     server.cluster.slots[slot] = NULL;
1032     return REDIS_OK;
1033 }
1034
1035 /* -----------------------------------------------------------------------------
1036  * Cluster state evaluation function
1037  * -------------------------------------------------------------------------- */
1038 void clusterUpdateState(void) {
1039     int ok = 1;
1040     int j;
1041
1042     for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) {
1043         if (server.cluster.slots[j] == NULL ||
1044             server.cluster.slots[j]->flags & (REDIS_NODE_FAIL))
1045         {
1046             ok = 0;
1047             break;
1048         }
1049     }
1050     if (ok) {
1051         if (server.cluster.state == REDIS_CLUSTER_NEEDHELP) {
1052             server.cluster.state = REDIS_CLUSTER_NEEDHELP;
1053         } else {
1054             server.cluster.state = REDIS_CLUSTER_OK;
1055         }
1056     } else {
1057         server.cluster.state = REDIS_CLUSTER_FAIL;
1058     }
1059 }
1060
1061 /* -----------------------------------------------------------------------------
1062  * CLUSTER command
1063  * -------------------------------------------------------------------------- */
1064
1065 sds clusterGenNodesDescription(void) {
1066     sds ci = sdsempty();
1067     dictIterator *di;
1068     dictEntry *de;
1069     int j, start;
1070
1071     di = dictGetIterator(server.cluster.nodes);
1072     while((de = dictNext(di)) != NULL) {
1073         clusterNode *node = dictGetEntryVal(de);
1074
1075         /* Node coordinates */
1076         ci = sdscatprintf(ci,"%.40s %s:%d ",
1077             node->name,
1078             node->ip,
1079             node->port);
1080
1081         /* Flags */
1082         if (node->flags == 0) ci = sdscat(ci,"noflags,");
1083         if (node->flags & REDIS_NODE_MYSELF) ci = sdscat(ci,"myself,");
1084         if (node->flags & REDIS_NODE_MASTER) ci = sdscat(ci,"master,");
1085         if (node->flags & REDIS_NODE_SLAVE) ci = sdscat(ci,"slave,");
1086         if (node->flags & REDIS_NODE_PFAIL) ci = sdscat(ci,"fail?,");
1087         if (node->flags & REDIS_NODE_FAIL) ci = sdscat(ci,"fail,");
1088         if (node->flags & REDIS_NODE_HANDSHAKE) ci =sdscat(ci,"handshake,");
1089         if (node->flags & REDIS_NODE_NOADDR) ci = sdscat(ci,"noaddr,");
1090         if (ci[sdslen(ci)-1] == ',') ci[sdslen(ci)-1] = ' ';
1091
1092         /* Slave of... or just "-" */
1093         if (node->slaveof)
1094             ci = sdscatprintf(ci,"%.40s ",node->slaveof->name);
1095         else
1096             ci = sdscatprintf(ci,"- ");
1097
1098         /* Latency from the POV of this node, link status */
1099         ci = sdscatprintf(ci,"%ld %ld %s",
1100             (long) node->ping_sent,
1101             (long) node->pong_received,
1102             node->link ? "connected" : "disconnected");
1103
1104         /* Slots served by this instance */
1105         start = -1;
1106         for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) {
1107             int bit;
1108
1109             if ((bit = clusterNodeGetSlotBit(node,j)) != 0) {
1110                 if (start == -1) start = j;
1111             }
1112             if (start != -1 && (!bit || j == REDIS_CLUSTER_SLOTS-1)) {
1113                 if (j == REDIS_CLUSTER_SLOTS-1) j++;
1114
1115                 if (start == j-1) {
1116                     ci = sdscatprintf(ci," %d",start);
1117                 } else {
1118                     ci = sdscatprintf(ci," %d-%d",start,j-1);
1119                 }
1120                 start = -1;
1121             }
1122         }
1123
1124         /* Just for MYSELF node we also dump info about slots that
1125          * we are migrating to other instances or importing from other
1126          * instances. */
1127         if (node->flags & REDIS_NODE_MYSELF) {
1128             for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) {
1129                 if (server.cluster.migrating_slots_to[j]) {
1130                     ci = sdscatprintf(ci," [%d->-%.40s]",j,
1131                         server.cluster.migrating_slots_to[j]->name);
1132                 } else if (server.cluster.importing_slots_from[j]) {
1133                     ci = sdscatprintf(ci," [%d-<-%.40s]",j,
1134                         server.cluster.importing_slots_from[j]->name);
1135                 }
1136             }
1137         }
1138         ci = sdscatlen(ci,"\n",1);
1139     }
1140     dictReleaseIterator(di);
1141     return ci;
1142 }
1143
1144 int getSlotOrReply(redisClient *c, robj *o) {
1145     long long slot;
1146
1147     if (getLongLongFromObject(o,&slot) != REDIS_OK ||
1148         slot < 0 || slot > REDIS_CLUSTER_SLOTS)
1149     {
1150         addReplyError(c,"Invalid or out of range slot");
1151         return -1;
1152     }
1153     return (int) slot;
1154 }
1155
1156 void clusterCommand(redisClient *c) {
1157     if (server.cluster_enabled == 0) {
1158         addReplyError(c,"This instance has cluster support disabled");
1159         return;
1160     }
1161
1162     if (!strcasecmp(c->argv[1]->ptr,"meet") && c->argc == 4) {
1163         clusterNode *n;
1164         struct sockaddr_in sa;
1165         long port;
1166
1167         /* Perform sanity checks on IP/port */
1168         if (inet_aton(c->argv[2]->ptr,&sa.sin_addr) == 0) {
1169             addReplyError(c,"Invalid IP address in MEET");
1170             return;
1171         }
1172         if (getLongFromObjectOrReply(c, c->argv[3], &port, NULL) != REDIS_OK ||
1173                     port < 0 || port > (65535-REDIS_CLUSTER_PORT_INCR))
1174         {
1175             addReplyError(c,"Invalid TCP port specified");
1176             return;
1177         }
1178
1179         /* Finally add the node to the cluster with a random name, this
1180          * will get fixed in the first handshake (ping/pong). */
1181         n = createClusterNode(NULL,REDIS_NODE_HANDSHAKE|REDIS_NODE_MEET);
1182         strncpy(n->ip,inet_ntoa(sa.sin_addr),sizeof(n->ip));
1183         n->port = port;
1184         clusterAddNode(n);
1185         addReply(c,shared.ok);
1186     } else if (!strcasecmp(c->argv[1]->ptr,"nodes") && c->argc == 2) {
1187         robj *o;
1188         sds ci = clusterGenNodesDescription();
1189
1190         o = createObject(REDIS_STRING,ci);
1191         addReplyBulk(c,o);
1192         decrRefCount(o);
1193     } else if ((!strcasecmp(c->argv[1]->ptr,"addslots") ||
1194                !strcasecmp(c->argv[1]->ptr,"delslots")) && c->argc >= 3) {
1195         int j, slot;
1196         unsigned char *slots = zmalloc(REDIS_CLUSTER_SLOTS);
1197         int del = !strcasecmp(c->argv[1]->ptr,"delslots");
1198
1199         memset(slots,0,REDIS_CLUSTER_SLOTS);
1200         /* Check that all the arguments are parsable and that all the
1201          * slots are not already busy. */
1202         for (j = 2; j < c->argc; j++) {
1203             if ((slot = getSlotOrReply(c,c->argv[j])) == -1) {
1204                 zfree(slots);
1205                 return;
1206             }
1207             if (del && server.cluster.slots[slot] == NULL) {
1208                 addReplyErrorFormat(c,"Slot %d is already unassigned", slot);
1209                 zfree(slots);
1210                 return;
1211             } else if (!del && server.cluster.slots[slot]) {
1212                 addReplyErrorFormat(c,"Slot %d is already busy", slot);
1213                 zfree(slots);
1214                 return;
1215             }
1216             if (slots[slot]++ == 1) {
1217                 addReplyErrorFormat(c,"Slot %d specified multiple times",
1218                     (int)slot);
1219                 zfree(slots);
1220                 return;
1221             }
1222         }
1223         for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) {
1224             if (slots[j]) {
1225                 int retval;
1226
1227                 /* If this slot was set as importing we can clear this
1228                  * state as now we are the real owner of the slot. */
1229                 if (server.cluster.importing_slots_from[j])
1230                     server.cluster.importing_slots_from[j] = NULL;
1231
1232                 retval = del ? clusterDelSlot(j) :
1233                                clusterAddSlot(server.cluster.myself,j);
1234                 redisAssert(retval == REDIS_OK);
1235             }
1236         }
1237         zfree(slots);
1238         clusterUpdateState();
1239         clusterSaveConfigOrDie();
1240         addReply(c,shared.ok);
1241     } else if (!strcasecmp(c->argv[1]->ptr,"setslot") && c->argc >= 4) {
1242         /* SETSLOT 10 MIGRATING <instance ID> */
1243         /* SETSLOT 10 IMPORTING <instance ID> */
1244         /* SETSLOT 10 STABLE */
1245         int slot;
1246         clusterNode *n;
1247
1248         if ((slot = getSlotOrReply(c,c->argv[2])) == -1) return;
1249
1250         if (!strcasecmp(c->argv[3]->ptr,"migrating") && c->argc == 5) {
1251             if (server.cluster.slots[slot] != server.cluster.myself) {
1252                 addReplyErrorFormat(c,"I'm not the owner of hash slot %u",slot);
1253                 return;
1254             }
1255             if ((n = clusterLookupNode(c->argv[4]->ptr)) == NULL) {
1256                 addReplyErrorFormat(c,"I don't know about node %s",
1257                     (char*)c->argv[4]->ptr);
1258                 return;
1259             }
1260             server.cluster.migrating_slots_to[slot] = n;
1261         } else if (!strcasecmp(c->argv[3]->ptr,"importing") && c->argc == 5) {
1262             if (server.cluster.slots[slot] == server.cluster.myself) {
1263                 addReplyErrorFormat(c,
1264                     "I'm already the owner of hash slot %u",slot);
1265                 return;
1266             }
1267             if ((n = clusterLookupNode(c->argv[4]->ptr)) == NULL) {
1268                 addReplyErrorFormat(c,"I don't know about node %s",
1269                     (char*)c->argv[3]->ptr);
1270                 return;
1271             }
1272             server.cluster.importing_slots_from[slot] = n;
1273         } else if (!strcasecmp(c->argv[3]->ptr,"stable") && c->argc == 4) {
1274             /* CLUSTER SETSLOT <SLOT> STABLE */
1275             server.cluster.importing_slots_from[slot] = NULL;
1276             server.cluster.migrating_slots_to[slot] = NULL;
1277         } else if (!strcasecmp(c->argv[3]->ptr,"node") && c->argc == 4) {
1278             /* CLUSTER SETSLOT <SLOT> NODE <NODE ID> */
1279             clusterNode *n = clusterLookupNode(c->argv[4]->ptr);
1280
1281             if (!n) addReplyErrorFormat(c,"Unknown node %s",
1282                 (char*)c->argv[4]->ptr);
1283             /* If this hash slot was served by 'myself' before to switch
1284              * make sure there are no longer local keys for this hash slot. */
1285             if (server.cluster.slots[slot] == server.cluster.myself &&
1286                 n != server.cluster.myself)
1287             {
1288                 int numkeys;
1289                 robj **keys;
1290
1291                 keys = zmalloc(sizeof(robj*)*1);
1292                 numkeys = GetKeysInSlot(slot, keys, 1);
1293                 zfree(keys);
1294                 if (numkeys == 0) {
1295                     addReplyErrorFormat(c, "Can't assign hashslot %d to a different node while I still hold keys for this hash slot.", slot);
1296                     return;
1297                 }
1298             }
1299             /* If this node was the slot owner and the slot was marked as
1300              * migrating, assigning the slot to another node will clear
1301              * the migratig status. */
1302             if (server.cluster.slots[slot] == server.cluster.myself &&
1303                 server.cluster.migrating_slots_to[slot])
1304                 server.cluster.migrating_slots_to[slot] = NULL;
1305
1306             clusterDelSlot(slot);
1307             clusterAddSlot(n,slot);
1308         } else {
1309             addReplyError(c,"Invalid CLUSTER SETSLOT action or number of arguments");
1310             return;
1311         }
1312         clusterSaveConfigOrDie();
1313         addReply(c,shared.ok);
1314     } else if (!strcasecmp(c->argv[1]->ptr,"info") && c->argc == 2) {
1315         char *statestr[] = {"ok","fail","needhelp"};
1316         int slots_assigned = 0, slots_ok = 0, slots_pfail = 0, slots_fail = 0;
1317         int j;
1318
1319         for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) {
1320             clusterNode *n = server.cluster.slots[j];
1321
1322             if (n == NULL) continue;
1323             slots_assigned++;
1324             if (n->flags & REDIS_NODE_FAIL) {
1325                 slots_fail++;
1326             } else if (n->flags & REDIS_NODE_PFAIL) {
1327                 slots_pfail++;
1328             } else {
1329                 slots_ok++;
1330             }
1331         }
1332
1333         sds info = sdscatprintf(sdsempty(),
1334             "cluster_state:%s\r\n"
1335             "cluster_slots_assigned:%d\r\n"
1336             "cluster_slots_ok:%d\r\n"
1337             "cluster_slots_pfail:%d\r\n"
1338             "cluster_slots_fail:%d\r\n"
1339             "cluster_known_nodes:%lu\r\n"
1340             , statestr[server.cluster.state],
1341             slots_assigned,
1342             slots_ok,
1343             slots_pfail,
1344             slots_fail,
1345             dictSize(server.cluster.nodes)
1346         );
1347         addReplySds(c,sdscatprintf(sdsempty(),"$%lu\r\n",
1348             (unsigned long)sdslen(info)));
1349         addReplySds(c,info);
1350         addReply(c,shared.crlf);
1351     } else if (!strcasecmp(c->argv[1]->ptr,"keyslot") && c->argc == 3) {
1352         sds key = c->argv[2]->ptr;
1353
1354         addReplyLongLong(c,keyHashSlot(key,sdslen(key)));
1355     } else if (!strcasecmp(c->argv[1]->ptr,"getkeysinslot") && c->argc == 4) {
1356         long long maxkeys, slot;
1357         unsigned int numkeys, j;
1358         robj **keys;
1359
1360         if (getLongLongFromObjectOrReply(c,c->argv[2],&slot,NULL) != REDIS_OK)
1361             return;
1362         if (getLongLongFromObjectOrReply(c,c->argv[3],&maxkeys,NULL) != REDIS_OK)
1363             return;
1364         if (slot < 0 || slot >= REDIS_CLUSTER_SLOTS || maxkeys < 0 ||
1365             maxkeys > 1024*1024) {
1366             addReplyError(c,"Invalid slot or number of keys");
1367             return;
1368         }
1369
1370         keys = zmalloc(sizeof(robj*)*maxkeys);
1371         numkeys = GetKeysInSlot(slot, keys, maxkeys);
1372         addReplyMultiBulkLen(c,numkeys);
1373         for (j = 0; j < numkeys; j++) addReplyBulk(c,keys[j]);
1374         zfree(keys);
1375     } else {
1376         addReplyError(c,"Wrong CLUSTER subcommand or number of arguments");
1377     }
1378 }
1379
1380 /* -----------------------------------------------------------------------------
1381  * RESTORE and MIGRATE commands
1382  * -------------------------------------------------------------------------- */
1383
1384 /* RESTORE key ttl serialized-value */
1385 void restoreCommand(redisClient *c) {
1386     FILE *fp;
1387     char buf[64];
1388     robj *o;
1389     unsigned char *data;
1390     long ttl;
1391
1392     /* Make sure this key does not already exist here... */
1393     if (lookupKeyWrite(c->db,c->argv[1]) != NULL) {
1394         addReplyError(c,"Target key name is busy.");
1395         return;
1396     }
1397
1398     /* Check if the TTL value makes sense */
1399     if (getLongFromObjectOrReply(c,c->argv[2],&ttl,NULL) != REDIS_OK) {
1400         return;
1401     } else if (ttl < 0) {
1402         addReplyError(c,"Invalid TTL value, must be >= 0");
1403         return;
1404     }
1405
1406     /* rdbLoadObject() only works against file descriptors so we need to
1407      * dump the serialized object into a file and reload. */
1408     snprintf(buf,sizeof(buf),"redis-restore-%d.tmp",getpid());
1409     fp = fopen(buf,"w+");
1410     if (!fp) {
1411         redisLog(REDIS_WARNING,"Can't open tmp file for RESTORE: %s",
1412             strerror(errno));
1413         addReplyErrorFormat(c,"RESTORE failed, tmp file creation error: %s",
1414             strerror(errno));
1415         return;
1416     }
1417     unlink(buf);
1418
1419     /* Write the actual data and rewind the file */
1420     data = (unsigned char*) c->argv[3]->ptr;
1421     if (fwrite(data+1,sdslen((sds)data)-1,1,fp) != 1) {
1422         redisLog(REDIS_WARNING,"Can't write against tmp file for RESTORE: %s",
1423             strerror(errno));
1424         addReplyError(c,"RESTORE failed, tmp file I/O error.");
1425         fclose(fp);
1426         return;
1427     }
1428     rewind(fp);
1429
1430     /* Finally create the object from the serialized dump and
1431      * store it at the specified key. */
1432     if ((data[0] > 4 && data[0] < 9) ||
1433          data[0] > 11 ||
1434         (o = rdbLoadObject(data[0],fp)) == NULL)
1435     {
1436         addReplyError(c,"Bad data format.");
1437         fclose(fp);
1438         return;
1439     }
1440     fclose(fp);
1441
1442     /* Create the key and set the TTL if any */
1443     dbAdd(c->db,c->argv[1],o);
1444     if (ttl) setExpire(c->db,c->argv[1],time(NULL)+ttl);
1445     addReply(c,shared.ok);
1446 }
1447
1448 /* MIGRATE host port key dbid timeout */
1449 void migrateCommand(redisClient *c) {
1450     int fd;
1451     long timeout;
1452     long dbid;
1453     char buf[64];
1454     FILE *fp;
1455     time_t ttl;
1456     robj *o;
1457     unsigned char type;
1458     off_t payload_len;
1459
1460     /* Sanity check */
1461     if (getLongFromObjectOrReply(c,c->argv[5],&timeout,NULL) != REDIS_OK)
1462         return;
1463     if (getLongFromObjectOrReply(c,c->argv[4],&dbid,NULL) != REDIS_OK)
1464         return;
1465     if (timeout <= 0) timeout = 1;
1466
1467     /* Check if the key is here. If not we reply with success as there is
1468      * nothing to migrate (for instance the key expired in the meantime), but
1469      * we include such information in the reply string. */
1470     if ((o = lookupKeyRead(c->db,c->argv[3])) == NULL) {
1471         addReplySds(c,sdsnew("+NOKEY"));
1472         return;
1473     }
1474
1475     /* Connect */
1476     fd = anetTcpNonBlockConnect(server.neterr,c->argv[1]->ptr,
1477                 atoi(c->argv[2]->ptr));
1478     if (fd == -1) {
1479         addReplyErrorFormat(c,"Can't connect to target node: %s",
1480             server.neterr);
1481         return;
1482     }
1483     if ((aeWait(fd,AE_WRITABLE,timeout*1000) & AE_WRITABLE) == 0) {
1484         addReplyError(c,"Timeout connecting to the client");
1485         return;
1486     }
1487
1488     /* Create temp file */
1489     snprintf(buf,sizeof(buf),"redis-migrate-%d.tmp",getpid());
1490     fp = fopen(buf,"w+");
1491     if (!fp) {
1492         redisLog(REDIS_WARNING,"Can't open tmp file for MIGRATE: %s",
1493             strerror(errno));
1494         addReplyErrorFormat(c,"MIGRATE failed, tmp file creation error: %s.",
1495             strerror(errno));
1496         return;
1497     }
1498     unlink(buf);
1499
1500     /* Build the SELECT + RESTORE query writing it in our temp file. */
1501     if (fwriteBulkCount(fp,'*',2) == 0) goto file_wr_err;
1502     if (fwriteBulkString(fp,"SELECT",6) == 0) goto file_wr_err;
1503     if (fwriteBulkLongLong(fp,dbid) == 0) goto file_wr_err;
1504
1505     ttl = getExpire(c->db,c->argv[3]);
1506     type = o->type;
1507     if (fwriteBulkCount(fp,'*',4) == 0) goto file_wr_err;
1508     if (fwriteBulkString(fp,"RESTORE",7) == 0) goto file_wr_err;
1509     if (fwriteBulkObject(fp,c->argv[3]) == 0) goto file_wr_err;
1510     if (fwriteBulkLongLong(fp, (ttl == -1) ? 0 : ttl) == 0) goto file_wr_err;
1511
1512     /* Finally the last argument that is the serailized object payload
1513      * in the form: <type><rdb-serailized-object>. */
1514     payload_len = rdbSavedObjectLen(o);
1515     if (fwriteBulkCount(fp,'$',payload_len+1) == 0) goto file_wr_err;
1516     if (fwrite(&type,1,1,fp) == 0) goto file_wr_err;
1517     if (rdbSaveObject(fp,o) == -1) goto file_wr_err;
1518     if (fwrite("\r\n",2,1,fp) == 0) goto file_wr_err;
1519
1520     /* Tranfer the query to the other node */
1521     rewind(fp);
1522     {
1523         char buf[4096];
1524         size_t nread;
1525
1526         while ((nread = fread(buf,1,sizeof(buf),fp)) != 0) {
1527             int nwritten;
1528
1529             nwritten = syncWrite(fd,buf,nread,timeout);
1530             if (nwritten != (signed)nread) goto socket_wr_err;
1531         }
1532         if (ferror(fp)) goto file_rd_err;
1533     }
1534
1535     /* Read back the reply */
1536     {
1537         char buf1[1024];
1538         char buf2[1024];
1539
1540         /* Read the two replies */
1541         if (syncReadLine(fd, buf1, sizeof(buf1), timeout) <= 0)
1542             goto socket_rd_err;
1543         if (syncReadLine(fd, buf2, sizeof(buf2), timeout) <= 0)
1544                 goto socket_rd_err;
1545         if (buf1[0] == '-' || buf2[0] == '-') {
1546             addReplyErrorFormat(c,"Target instance replied with error: %s",
1547                 (buf1[0] == '-') ? buf1+1 : buf2+1);
1548         } else {
1549             dbDelete(c->db,c->argv[3]);
1550             addReply(c,shared.ok);
1551         }
1552     }
1553     fclose(fp);
1554     close(fd);
1555     return;
1556
1557 file_wr_err:
1558     redisLog(REDIS_WARNING,"Can't write on tmp file for MIGRATE: %s",
1559         strerror(errno));
1560     addReplyErrorFormat(c,"MIGRATE failed, tmp file write error: %s.",
1561         strerror(errno));
1562     fclose(fp);
1563     close(fd);
1564     return;
1565
1566 file_rd_err:
1567     redisLog(REDIS_WARNING,"Can't read from tmp file for MIGRATE: %s",
1568         strerror(errno));
1569     addReplyErrorFormat(c,"MIGRATE failed, tmp file read error: %s.",
1570         strerror(errno));
1571     fclose(fp);
1572     close(fd);
1573     return;
1574
1575 socket_wr_err:
1576     redisLog(REDIS_NOTICE,"Can't write to target node for MIGRATE: %s",
1577         strerror(errno));
1578     addReplyErrorFormat(c,"MIGRATE failed, writing to target node: %s.",
1579         strerror(errno));
1580     fclose(fp);
1581     close(fd);
1582     return;
1583
1584 socket_rd_err:
1585     redisLog(REDIS_NOTICE,"Can't read from target node for MIGRATE: %s",
1586         strerror(errno));
1587     addReplyErrorFormat(c,"MIGRATE failed, reading from target node: %s.",
1588         strerror(errno));
1589     fclose(fp);
1590     close(fd);
1591     return;
1592 }
1593
1594 /* DUMP keyname
1595  * DUMP is actually not used by Redis Cluster but it is the obvious
1596  * complement of RESTORE and can be useful for different applications. */
1597 void dumpCommand(redisClient *c) {
1598     char buf[64];
1599     FILE *fp;
1600     robj *o, *dumpobj;
1601     sds dump = NULL;
1602     off_t payload_len;
1603     unsigned int type;
1604
1605     /* Check if the key is here. */
1606     if ((o = lookupKeyRead(c->db,c->argv[1])) == NULL) {
1607         addReply(c,shared.nullbulk);
1608         return;
1609     }
1610
1611     /* Create temp file */
1612     snprintf(buf,sizeof(buf),"redis-dump-%d.tmp",getpid());
1613     fp = fopen(buf,"w+");
1614     if (!fp) {
1615         redisLog(REDIS_WARNING,"Can't open tmp file for MIGRATE: %s",
1616             strerror(errno));
1617         addReplyErrorFormat(c,"DUMP failed, tmp file creation error: %s.",
1618             strerror(errno));
1619         return;
1620     }
1621     unlink(buf);
1622
1623     /* Dump the serailized object and read it back in memory.
1624      * We prefix it with a one byte containing the type ID.
1625      * This is the serialization format understood by RESTORE. */
1626     if (rdbSaveObject(fp,o) == -1) goto file_wr_err;
1627     payload_len = ftello(fp);
1628     if (fseeko(fp,0,SEEK_SET) == -1) goto file_rd_err;
1629     dump = sdsnewlen(NULL,payload_len+1);
1630     if (payload_len && fread(dump+1,payload_len,1,fp) != 1) goto file_rd_err;
1631     fclose(fp);
1632     type = o->type;
1633     if (type == REDIS_LIST && o->encoding == REDIS_ENCODING_ZIPLIST)
1634         type = REDIS_LIST_ZIPLIST;
1635     else if (type == REDIS_HASH && o->encoding == REDIS_ENCODING_ZIPMAP)
1636         type = REDIS_HASH_ZIPMAP;
1637     else if (type == REDIS_SET && o->encoding == REDIS_ENCODING_INTSET)
1638         type = REDIS_SET_INTSET;
1639     else
1640         type = o->type;
1641     dump[0] = type;
1642
1643     /* Transfer to the client */
1644     dumpobj = createObject(REDIS_STRING,dump);
1645     addReplyBulk(c,dumpobj);
1646     decrRefCount(dumpobj);
1647     return;
1648
1649 file_wr_err:
1650     redisLog(REDIS_WARNING,"Can't write on tmp file for DUMP: %s",
1651         strerror(errno));
1652     addReplyErrorFormat(c,"DUMP failed, tmp file write error: %s.",
1653         strerror(errno));
1654     sdsfree(dump);
1655     fclose(fp);
1656     return;
1657
1658 file_rd_err:
1659     redisLog(REDIS_WARNING,"Can't read from tmp file for DUMP: %s",
1660         strerror(errno));
1661     addReplyErrorFormat(c,"DUMP failed, tmp file read error: %s.",
1662         strerror(errno));
1663     sdsfree(dump);
1664     fclose(fp);
1665     return;
1666 }
1667
1668 /* -----------------------------------------------------------------------------
1669  * Cluster functions related to serving / redirecting clients
1670  * -------------------------------------------------------------------------- */
1671
1672 /* Return the pointer to the cluster node that is able to serve the query
1673  * as all the keys belong to hash slots for which the node is in charge.
1674  *
1675  * If the returned node should be used only for this request, the *ask
1676  * integer is set to '1', otherwise to '0'. This is used in order to
1677  * let the caller know if we should reply with -MOVED or with -ASK.
1678  *
1679  * If the request contains more than a single key NULL is returned,
1680  * however a request with more then a key argument where the key is always
1681  * the same is valid, like in: RPOPLPUSH mylist mylist.*/
1682 clusterNode *getNodeByQuery(redisClient *c, struct redisCommand *cmd, robj **argv, int argc, int *hashslot, int *ask) {
1683     clusterNode *n = NULL;
1684     robj *firstkey = NULL;
1685     multiState *ms, _ms;
1686     multiCmd mc;
1687     int i, slot = 0;
1688
1689     /* We handle all the cases as if they were EXEC commands, so we have
1690      * a common code path for everything */
1691     if (cmd->proc == execCommand) {
1692         /* If REDIS_MULTI flag is not set EXEC is just going to return an
1693          * error. */
1694         if (!(c->flags & REDIS_MULTI)) return server.cluster.myself;
1695         ms = &c->mstate;
1696     } else {
1697         /* In order to have a single codepath create a fake Multi State
1698          * structure if the client is not in MULTI/EXEC state, this way
1699          * we have a single codepath below. */
1700         ms = &_ms;
1701         _ms.commands = &mc;
1702         _ms.count = 1;
1703         mc.argv = argv;
1704         mc.argc = argc;
1705         mc.cmd = cmd;
1706     }
1707
1708     /* Check that all the keys are the same key, and get the slot and
1709      * node for this key. */
1710     for (i = 0; i < ms->count; i++) {
1711         struct redisCommand *mcmd;
1712         robj **margv;
1713         int margc, *keyindex, numkeys, j;
1714
1715         mcmd = ms->commands[i].cmd;
1716         margc = ms->commands[i].argc;
1717         margv = ms->commands[i].argv;
1718
1719         keyindex = getKeysFromCommand(mcmd,margv,margc,&numkeys,
1720                                       REDIS_GETKEYS_ALL);
1721         for (j = 0; j < numkeys; j++) {
1722             if (firstkey == NULL) {
1723                 /* This is the first key we see. Check what is the slot
1724                  * and node. */
1725                 firstkey = margv[keyindex[j]];
1726
1727                 slot = keyHashSlot((char*)firstkey->ptr, sdslen(firstkey->ptr));
1728                 n = server.cluster.slots[slot];
1729                 redisAssert(n != NULL);
1730             } else {
1731                 /* If it is not the first key, make sure it is exactly
1732                  * the same key as the first we saw. */
1733                 if (!equalStringObjects(firstkey,margv[keyindex[j]])) {
1734                     decrRefCount(firstkey);
1735                     getKeysFreeResult(keyindex);
1736                     return NULL;
1737                 }
1738             }
1739         }
1740         getKeysFreeResult(keyindex);
1741     }
1742     if (ask) *ask = 0; /* This is the default. Set to 1 if needed later. */
1743     /* No key at all in command? then we can serve the request
1744      * without redirections. */
1745     if (n == NULL) return server.cluster.myself;
1746     if (hashslot) *hashslot = slot;
1747     /* This request is about a slot we are migrating into another instance?
1748      * Then we need to check if we have the key. If we have it we can reply.
1749      * If instead is a new key, we pass the request to the node that is
1750      * receiving the slot. */
1751     if (n == server.cluster.myself &&
1752         server.cluster.migrating_slots_to[slot] != NULL)
1753     {
1754         if (lookupKeyRead(&server.db[0],firstkey) == NULL) {
1755             if (ask) *ask = 1;
1756             return server.cluster.migrating_slots_to[slot];
1757         }
1758     }
1759     /* Handle the case in which we are receiving this hash slot from
1760      * another instance, so we'll accept the query even if in the table
1761      * it is assigned to a different node. */
1762     if (server.cluster.importing_slots_from[slot] != NULL)
1763         return server.cluster.myself;
1764     /* It's not a -ASK case. Base case: just return the right node. */
1765     return n;
1766 }