src/cluster.c

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