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