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