src/cluster.c

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