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,"info") && c->argc == 2) {
1172         char *statestr[] = {"ok","fail","needhelp"};
1173         int slots_assigned = 0, slots_ok = 0, slots_pfail = 0, slots_fail = 0;
1174         int j;
1175
1176         for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) {
1177             clusterNode *n = server.cluster.slots[j];
1178
1179             if (n == NULL) continue;
1180             slots_assigned++;
1181             if (n->flags & REDIS_NODE_FAIL) {
1182                 slots_fail++;
1183             } else if (n->flags & REDIS_NODE_PFAIL) {
1184                 slots_pfail++;
1185             } else {
1186                 slots_ok++;
1187             }
1188         }
1189
1190         sds info = sdscatprintf(sdsempty(),
1191             "cluster_state:%s\r\n"
1192             "cluster_slots_assigned:%d\r\n"
1193             "cluster_slots_ok:%d\r\n"
1194             "cluster_slots_pfail:%d\r\n"
1195             "cluster_slots_fail:%d\r\n"
1196             "cluster_known_nodes:%lu\r\n"
1197             , statestr[server.cluster.state],
1198             slots_assigned,
1199             slots_ok,
1200             slots_pfail,
1201             slots_fail,
1202             dictSize(server.cluster.nodes)
1203         );
1204         addReplySds(c,sdscatprintf(sdsempty(),"$%lu\r\n",
1205             (unsigned long)sdslen(info)));
1206         addReplySds(c,info);
1207         addReply(c,shared.crlf);
1208     } else if (!strcasecmp(c->argv[1]->ptr,"keyslot") && c->argc == 3) {
1209         sds key = c->argv[2]->ptr;
1210
1211         addReplyLongLong(c,keyHashSlot(key,sdslen(key)));
1212     } else if (!strcasecmp(c->argv[1]->ptr,"getkeysinslot") && c->argc == 4) {
1213         long long maxkeys, slot;
1214         unsigned int numkeys;
1215         int j;
1216         robj **keys;
1217
1218         if (getLongLongFromObjectOrReply(c,c->argv[2],&slot,NULL) != REDIS_OK)
1219             return;
1220         if (getLongLongFromObjectOrReply(c,c->argv[3],&maxkeys,NULL) != REDIS_OK)
1221             return;
1222         if (slot < 0 || slot >= REDIS_CLUSTER_SLOTS || maxkeys < 0 ||
1223             maxkeys > 1024*1024) {
1224             addReplyError(c,"Invalid slot or number of keys");
1225             return;
1226         }
1227
1228         keys = zmalloc(sizeof(robj*)*maxkeys);
1229         numkeys = GetKeysInSlot(slot, keys, maxkeys);
1230         addReplyMultiBulkLen(c,numkeys);
1231         for (j = 0; j < numkeys; j++) addReplyBulk(c,keys[j]);
1232         zfree(keys);
1233     } else {
1234         addReplyError(c,"Wrong CLUSTER subcommand or number of arguments");
1235     }
1236 }
1237
1238 /* -----------------------------------------------------------------------------
1239  * RESTORE and MIGRATE commands
1240  * -------------------------------------------------------------------------- */
1241
1242 /* RESTORE key ttl serialized-value */
1243 void restoreCommand(redisClient *c) {
1244     FILE *fp;
1245     char buf[64];
1246     robj *o;
1247     unsigned char *data;
1248     long ttl;
1249
1250     /* Make sure this key does not already exist here... */
1251     if (dbExists(c->db,c->argv[1])) {
1252         addReplyError(c,"Target key name is busy.");
1253         return;
1254     }
1255
1256     /* Check if the TTL value makes sense */
1257     if (getLongFromObjectOrReply(c,c->argv[2],&ttl,NULL) != REDIS_OK) {
1258         return;
1259     } else if (ttl < 0) {
1260         addReplyError(c,"Invalid TTL value, must be >= 0");
1261         return;
1262     }
1263
1264     /* rdbLoadObject() only works against file descriptors so we need to
1265      * dump the serialized object into a file and reload. */
1266     snprintf(buf,sizeof(buf),"redis-restore-%d.tmp",getpid());
1267     fp = fopen(buf,"w+");
1268     if (!fp) {
1269         redisLog(REDIS_WARNING,"Can't open tmp file for RESTORE: %s",
1270             strerror(errno));
1271         addReplyErrorFormat(c,"RESTORE failed, tmp file creation error: %s",
1272             strerror(errno));
1273         return;
1274     }
1275     unlink(buf);
1276
1277     /* Write the actual data and rewind the file */
1278     data = (unsigned char*) c->argv[3]->ptr;
1279     if (fwrite(data+1,sdslen((sds)data)-1,1,fp) != 1) {
1280         redisLog(REDIS_WARNING,"Can't write against tmp file for RESTORE: %s",
1281             strerror(errno));
1282         addReplyError(c,"RESTORE failed, tmp file I/O error.");
1283         fclose(fp);
1284         return;
1285     }
1286     rewind(fp);
1287
1288     /* Finally create the object from the serialized dump and
1289      * store it at the specified key. */
1290     if ((data[0] > 4 && data[0] < 9) ||
1291          data[0] > 11 ||
1292         (o = rdbLoadObject(data[0],fp)) == NULL)
1293     {
1294         addReplyError(c,"Bad data format.");
1295         fclose(fp);
1296         return;
1297     }
1298     fclose(fp);
1299
1300     /* Create the key and set the TTL if any */
1301     dbAdd(c->db,c->argv[1],o);
1302     if (ttl) setExpire(c->db,c->argv[1],time(NULL)+ttl);
1303     addReply(c,shared.ok);
1304 }
1305
1306 /* MIGRATE host port key dbid timeout */
1307 void migrateCommand(redisClient *c) {
1308     int fd;
1309     long timeout;
1310     long dbid;
1311     char buf[64];
1312     FILE *fp;
1313     time_t ttl;
1314     robj *o;
1315     unsigned char type;
1316     off_t payload_len;
1317
1318     /* Sanity check */
1319     if (getLongFromObjectOrReply(c,c->argv[5],&timeout,NULL) != REDIS_OK)
1320         return;
1321     if (getLongFromObjectOrReply(c,c->argv[4],&dbid,NULL) != REDIS_OK)
1322         return;
1323     if (timeout <= 0) timeout = 1;
1324
1325     /* Check if the key is here. If not we reply with success as there is
1326      * nothing to migrate (for instance the key expired in the meantime), but
1327      * we include such information in the reply string. */
1328     if ((o = lookupKeyRead(c->db,c->argv[3])) == NULL) {
1329         addReplySds(c,sdsnew("+NOKEY"));
1330         return;
1331     }
1332
1333     /* Connect */
1334     fd = anetTcpNonBlockConnect(server.neterr,c->argv[1]->ptr,
1335                 atoi(c->argv[2]->ptr));
1336     if (fd == -1) {
1337         addReplyErrorFormat(c,"Can't connect to target node: %s",
1338             server.neterr);
1339         return;
1340     }
1341     if ((aeWait(fd,AE_WRITABLE,timeout*1000) & AE_WRITABLE) == 0) {
1342         addReplyError(c,"Timeout connecting to the client");
1343         return;
1344     }
1345
1346     /* Create temp file */
1347     snprintf(buf,sizeof(buf),"redis-migrate-%d.tmp",getpid());
1348     fp = fopen(buf,"w+");
1349     if (!fp) {
1350         redisLog(REDIS_WARNING,"Can't open tmp file for MIGRATE: %s",
1351             strerror(errno));
1352         addReplyErrorFormat(c,"MIGRATE failed, tmp file creation error: %s.",
1353             strerror(errno));
1354         return;
1355     }
1356     unlink(buf);
1357
1358     /* Build the SELECT + RESTORE query writing it in our temp file. */
1359     if (fwriteBulkCount(fp,'*',2) == 0) goto file_wr_err;
1360     if (fwriteBulkString(fp,"SELECT",6) == 0) goto file_wr_err;
1361     if (fwriteBulkLongLong(fp,dbid) == 0) goto file_wr_err;
1362
1363     ttl = getExpire(c->db,c->argv[3]);
1364     type = o->type;
1365     if (fwriteBulkCount(fp,'*',4) == 0) goto file_wr_err;
1366     if (fwriteBulkString(fp,"RESTORE",7) == 0) goto file_wr_err;
1367     if (fwriteBulkObject(fp,c->argv[3]) == 0) goto file_wr_err;
1368     if (fwriteBulkLongLong(fp, (ttl == -1) ? 0 : ttl) == 0) goto file_wr_err;
1369
1370     /* Finally the last argument that is the serailized object payload
1371      * in the form: <type><rdb-serailized-object>. */
1372     payload_len = rdbSavedObjectLen(o);
1373     if (fwriteBulkCount(fp,'$',payload_len+1) == 0) goto file_wr_err;
1374     if (fwrite(&type,1,1,fp) == 0) goto file_wr_err;
1375     if (rdbSaveObject(fp,o) == -1) goto file_wr_err;
1376     if (fwrite("\r\n",2,1,fp) == 0) goto file_wr_err;
1377
1378     /* Tranfer the query to the other node */
1379     rewind(fp);
1380     {
1381         char buf[4096];
1382         size_t nread;
1383
1384         while ((nread = fread(buf,1,sizeof(buf),fp)) != 0) {
1385             int nwritten;
1386
1387             nwritten = syncWrite(fd,buf,nread,timeout);
1388             if (nwritten != (signed)nread) goto socket_wr_err;
1389         }
1390         if (ferror(fp)) goto file_rd_err;
1391     }
1392
1393     /* Read back the reply */
1394     {
1395         char buf1[1024];
1396         char buf2[1024];
1397
1398         /* Read the two replies */
1399         if (syncReadLine(fd, buf1, sizeof(buf1), timeout) <= 0)
1400             goto socket_rd_err;
1401         if (syncReadLine(fd, buf2, sizeof(buf2), timeout) <= 0)
1402                 goto socket_rd_err;
1403         if (buf1[0] == '-' || buf2[0] == '-') {
1404             addReplyErrorFormat(c,"Target instance replied with error: %s",
1405                 (buf1[0] == '-') ? buf1+1 : buf2+1);
1406         } else {
1407             dbDelete(c->db,c->argv[3]);
1408             addReply(c,shared.ok);
1409         }
1410     }
1411     fclose(fp);
1412     close(fd);
1413     return;
1414
1415 file_wr_err:
1416     redisLog(REDIS_WARNING,"Can't write on tmp file for MIGRATE: %s",
1417         strerror(errno));
1418     addReplyErrorFormat(c,"MIGRATE failed, tmp file write error: %s.",
1419         strerror(errno));
1420     fclose(fp);
1421     close(fd);
1422     return;
1423
1424 file_rd_err:
1425     redisLog(REDIS_WARNING,"Can't read from tmp file for MIGRATE: %s",
1426         strerror(errno));
1427     addReplyErrorFormat(c,"MIGRATE failed, tmp file read error: %s.",
1428         strerror(errno));
1429     fclose(fp);
1430     close(fd);
1431     return;
1432
1433 socket_wr_err:
1434     redisLog(REDIS_NOTICE,"Can't write to target node for MIGRATE: %s",
1435         strerror(errno));
1436     addReplyErrorFormat(c,"MIGRATE failed, writing to target node: %s.",
1437         strerror(errno));
1438     fclose(fp);
1439     close(fd);
1440     return;
1441
1442 socket_rd_err:
1443     redisLog(REDIS_NOTICE,"Can't read from target node for MIGRATE: %s",
1444         strerror(errno));
1445     addReplyErrorFormat(c,"MIGRATE failed, reading from target node: %s.",
1446         strerror(errno));
1447     fclose(fp);
1448     close(fd);
1449     return;
1450 }
1451
1452 /* DUMP keyname
1453  * DUMP is actually not used by Redis Cluster but it is the obvious
1454  * complement of RESTORE and can be useful for different applications. */
1455 void dumpCommand(redisClient *c) {
1456     char buf[64];
1457     FILE *fp;
1458     robj *o, *dumpobj;
1459     sds dump = NULL;
1460     off_t payload_len;
1461     unsigned int type;
1462
1463     /* Check if the key is here. */
1464     if ((o = lookupKeyRead(c->db,c->argv[1])) == NULL) {
1465         addReply(c,shared.nullbulk);
1466         return;
1467     }
1468
1469     /* Create temp file */
1470     snprintf(buf,sizeof(buf),"redis-dump-%d.tmp",getpid());
1471     fp = fopen(buf,"w+");
1472     if (!fp) {
1473         redisLog(REDIS_WARNING,"Can't open tmp file for MIGRATE: %s",
1474             strerror(errno));
1475         addReplyErrorFormat(c,"DUMP failed, tmp file creation error: %s.",
1476             strerror(errno));
1477         return;
1478     }
1479     unlink(buf);
1480
1481     /* Dump the serailized object and read it back in memory.
1482      * We prefix it with a one byte containing the type ID.
1483      * This is the serialization format understood by RESTORE. */
1484     if (rdbSaveObject(fp,o) == -1) goto file_wr_err;
1485     payload_len = ftello(fp);
1486     if (fseeko(fp,0,SEEK_SET) == -1) goto file_rd_err;
1487     dump = sdsnewlen(NULL,payload_len+1);
1488     if (payload_len && fread(dump+1,payload_len,1,fp) != 1) goto file_rd_err;
1489     fclose(fp);
1490     type = o->type;
1491     if (type == REDIS_LIST && o->encoding == REDIS_ENCODING_ZIPLIST)
1492         type = REDIS_LIST_ZIPLIST;
1493     else if (type == REDIS_HASH && o->encoding == REDIS_ENCODING_ZIPMAP)
1494         type = REDIS_HASH_ZIPMAP;
1495     else if (type == REDIS_SET && o->encoding == REDIS_ENCODING_INTSET)
1496         type = REDIS_SET_INTSET;
1497     else
1498         type = o->type;
1499     dump[0] = type;
1500
1501     /* Transfer to the client */
1502     dumpobj = createObject(REDIS_STRING,dump);
1503     addReplyBulk(c,dumpobj);
1504     decrRefCount(dumpobj);
1505     return;
1506
1507 file_wr_err:
1508     redisLog(REDIS_WARNING,"Can't write on tmp file for DUMP: %s",
1509         strerror(errno));
1510     addReplyErrorFormat(c,"DUMP failed, tmp file write error: %s.",
1511         strerror(errno));
1512     sdsfree(dump);
1513     fclose(fp);
1514     return;
1515
1516 file_rd_err:
1517     redisLog(REDIS_WARNING,"Can't read from tmp file for DUMP: %s",
1518         strerror(errno));
1519     addReplyErrorFormat(c,"DUMP failed, tmp file read error: %s.",
1520         strerror(errno));
1521     sdsfree(dump);
1522     fclose(fp);
1523     return;
1524 }
1525
1526 /* -----------------------------------------------------------------------------
1527  * Cluster functions related to serving / redirecting clients
1528  * -------------------------------------------------------------------------- */
1529
1530 /* Return the pointer to the cluster node that is able to serve the query
1531  * as all the keys belong to hash slots for which the node is in charge.
1532  *
1533  * If keys in query spawn multiple nodes NULL is returned. */
1534 clusterNode *getNodeByQuery(redisClient *c, struct redisCommand *cmd, robj **argv, int argc, int *hashslot) {
1535     clusterNode *n = NULL;
1536     multiState *ms, _ms;
1537     multiCmd mc;
1538     int i;
1539
1540     /* We handle all the cases as if they were EXEC commands, so we have
1541      * a common code path for everything */
1542     if (cmd->proc == execCommand) {
1543         /* If REDIS_MULTI flag is not set EXEC is just going to return an
1544          * error. */
1545         if (!(c->flags & REDIS_MULTI)) return server.cluster.myself;
1546         ms = &c->mstate;
1547     } else {
1548         /* Create a fake Multi State structure, with just one command */
1549         ms = &_ms;
1550         _ms.commands = &mc;
1551         _ms.count = 1;
1552         mc.argv = argv;
1553         mc.argc = argc;
1554         mc.cmd = cmd;
1555     }
1556
1557     for (i = 0; i < ms->count; i++) {
1558         struct redisCommand *mcmd;
1559         robj **margv;
1560         int margc, *keyindex, numkeys, j;
1561
1562         mcmd = ms->commands[i].cmd;
1563         margc = ms->commands[i].argc;
1564         margv = ms->commands[i].argv;
1565
1566         keyindex = getKeysFromCommand(mcmd,margv,margc,&numkeys,
1567                                       REDIS_GETKEYS_PRELOAD);
1568         for (j = 0; j < numkeys; j++) {
1569             int slot = keyHashSlot((char*)margv[keyindex[j]]->ptr,
1570                 sdslen(margv[keyindex[j]]->ptr));
1571             struct clusterNode *slotnode;
1572
1573             slotnode = server.cluster.slots[slot];
1574             if (hashslot) *hashslot = slot;
1575             /* Node not assigned? (Should never happen actually
1576              * if we reached this function).
1577              * Different node than the previous one?
1578              * Return NULL, the cluster can't serve multi-node requests */
1579             if (slotnode == NULL || (n && slotnode != n)) {
1580                 getKeysFreeResult(keyindex);
1581                 return NULL;
1582             } else {
1583                 n = slotnode;
1584             }
1585         }
1586         getKeysFreeResult(keyindex);
1587     }
1588     return (n == NULL) ? server.cluster.myself : n;
1589 }