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