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