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)) {
 636             redisLog(REDIS_NOTICE,
 637                 "FAIL message received from %.40s about %.40s",
 638                 hdr->sender, hdr->data.fail.about.nodename);
 639             failing->flags |= REDIS_NODE_FAIL;
 640             failing->flags &= ~REDIS_NODE_PFAIL;
 641             clusterUpdateState();
 642             clusterSaveConfigOrDie();
 643         }
 644     } else {
 645         redisLog(REDIS_NOTICE,"Received unknown packet type: %d", type);
 646     }
 647     return 1;
 648 }
 649
 650 /* This function is called when we detect the link with this node is lost.
 651    We set the node as no longer connected. The Cluster Cron will detect
 652    this connection and will try to get it connected again.
 653
 654    Instead if the node is a temporary node used to accept a query, we
 655    completely free the node on error. */
 656 void handleLinkIOError(clusterLink *link) {
 657     freeClusterLink(link);
 658 }
 659
 660 /* Send data. This is handled using a trivial send buffer that gets
 661  * consumed by write(). We don't try to optimize this for speed too much
 662  * as this is a very low traffic channel. */
 663 void clusterWriteHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
 664     clusterLink *link = (clusterLink*) privdata;
 665     ssize_t nwritten;
 666     REDIS_NOTUSED(el);
 667     REDIS_NOTUSED(mask);
 668
 669     nwritten = write(fd, link->sndbuf, sdslen(link->sndbuf));
 670     if (nwritten <= 0) {
 671         redisLog(REDIS_NOTICE,"I/O error writing to node link: %s",
 672             strerror(errno));
 673         handleLinkIOError(link);
 674         return;
 675     }
 676     link->sndbuf = sdsrange(link->sndbuf,nwritten,-1);
 677     if (sdslen(link->sndbuf) == 0)
 678         aeDeleteFileEvent(server.el, link->fd, AE_WRITABLE);
 679 }
 680
 681 /* Read data. Try to read the first field of the header first to check the
 682  * full length of the packet. When a whole packet is in memory this function
 683  * will call the function to process the packet. And so forth. */
 684 void clusterReadHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
 685     char buf[1024];
 686     ssize_t nread;
 687     clusterMsg *hdr;
 688     clusterLink *link = (clusterLink*) privdata;
 689     int readlen;
 690     REDIS_NOTUSED(el);
 691     REDIS_NOTUSED(mask);
 692
 693 again:
 694     if (sdslen(link->rcvbuf) >= 4) {
 695         hdr = (clusterMsg*) link->rcvbuf;
 696         readlen = ntohl(hdr->totlen) - sdslen(link->rcvbuf);
 697     } else {
 698         readlen = 4 - sdslen(link->rcvbuf);
 699     }
 700
 701     nread = read(fd,buf,readlen);
 702     if (nread == -1 && errno == EAGAIN) return; /* Just no data */
 703
 704     if (nread <= 0) {
 705         /* I/O error... */
 706         redisLog(REDIS_NOTICE,"I/O error reading from node link: %s",
 707             (nread == 0) ? "connection closed" : strerror(errno));
 708         handleLinkIOError(link);
 709         return;
 710     } else {
 711         /* Read data and recast the pointer to the new buffer. */
 712         link->rcvbuf = sdscatlen(link->rcvbuf,buf,nread);
 713         hdr = (clusterMsg*) link->rcvbuf;
 714     }
 715
 716     /* Total length obtained? read the payload now instead of burning
 717      * cycles waiting for a new event to fire. */
 718     if (sdslen(link->rcvbuf) == 4) goto again;
 719
 720     /* Whole packet in memory? We can process it. */
 721     if (sdslen(link->rcvbuf) == ntohl(hdr->totlen)) {
 722         if (clusterProcessPacket(link)) {
 723             sdsfree(link->rcvbuf);
 724             link->rcvbuf = sdsempty();
 725         }
 726     }
 727 }
 728
 729 /* Put stuff into the send buffer. */
 730 void clusterSendMessage(clusterLink *link, unsigned char *msg, size_t msglen) {
 731     if (sdslen(link->sndbuf) == 0 && msglen != 0)
 732         aeCreateFileEvent(server.el,link->fd,AE_WRITABLE,
 733                     clusterWriteHandler,link);
 734
 735     link->sndbuf = sdscatlen(link->sndbuf, msg, msglen);
 736 }
 737
 738 /* Build the message header */
 739 void clusterBuildMessageHdr(clusterMsg *hdr, int type) {
 740     int totlen;
 741
 742     memset(hdr,0,sizeof(*hdr));
 743     hdr->type = htons(type);
 744     memcpy(hdr->sender,server.cluster.myself->name,REDIS_CLUSTER_NAMELEN);
 745     memcpy(hdr->myslots,server.cluster.myself->slots,
 746         sizeof(hdr->myslots));
 747     memset(hdr->slaveof,0,REDIS_CLUSTER_NAMELEN);
 748     if (server.cluster.myself->slaveof != NULL) {
 749         memcpy(hdr->slaveof,server.cluster.myself->slaveof->name,
 750                                     REDIS_CLUSTER_NAMELEN);
 751     }
 752     hdr->port = htons(server.port);
 753     hdr->state = server.cluster.state;
 754     memset(hdr->configdigest,0,32); /* FIXME: set config digest */
 755
 756     if (type == CLUSTERMSG_TYPE_FAIL) {
 757         totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
 758         totlen += sizeof(clusterMsgDataFail);
 759     }
 760     hdr->totlen = htonl(totlen);
 761     /* For PING, PONG, and MEET, fixing the totlen field is up to the caller */
 762 }
 763
 764 /* Send a PING or PONG packet to the specified node, making sure to add enough
 765  * gossip informations. */
 766 void clusterSendPing(clusterLink *link, int type) {
 767     unsigned char buf[1024];
 768     clusterMsg *hdr = (clusterMsg*) buf;
 769     int gossipcount = 0, totlen;
 770     /* freshnodes is the number of nodes we can still use to populate the
 771      * gossip section of the ping packet. Basically we start with the nodes
 772      * we have in memory minus two (ourself and the node we are sending the
 773      * message to). Every time we add a node we decrement the counter, so when
 774      * it will drop to <= zero we know there is no more gossip info we can
 775      * send. */
 776     int freshnodes = dictSize(server.cluster.nodes)-2;
 777
 778     if (link->node && type == CLUSTERMSG_TYPE_PING)
 779         link->node->ping_sent = time(NULL);
 780     clusterBuildMessageHdr(hdr,type);
 781
 782     /* Populate the gossip fields */
 783     while(freshnodes > 0 && gossipcount < 3) {
 784         struct dictEntry *de = dictGetRandomKey(server.cluster.nodes);
 785         clusterNode *this = dictGetEntryVal(de);
 786         clusterMsgDataGossip *gossip;
 787         int j;
 788
 789         /* Not interesting to gossip about ourself.
 790          * Nor to send gossip info about HANDSHAKE state nodes (zero info). */
 791         if (this == server.cluster.myself ||
 792             this->flags & REDIS_NODE_HANDSHAKE) {
 793                 freshnodes--; /* otherwise we may loop forever. */
 794                 continue;
 795         }
 796
 797         /* Check if we already added this node */
 798         for (j = 0; j < gossipcount; j++) {
 799             if (memcmp(hdr->data.ping.gossip[j].nodename,this->name,
 800                     REDIS_CLUSTER_NAMELEN) == 0) break;
 801         }
 802         if (j != gossipcount) continue;
 803
 804         /* Add it */
 805         freshnodes--;
 806         gossip = &(hdr->data.ping.gossip[gossipcount]);
 807         memcpy(gossip->nodename,this->name,REDIS_CLUSTER_NAMELEN);
 808         gossip->ping_sent = htonl(this->ping_sent);
 809         gossip->pong_received = htonl(this->pong_received);
 810         memcpy(gossip->ip,this->ip,sizeof(this->ip));
 811         gossip->port = htons(this->port);
 812         gossip->flags = htons(this->flags);
 813         gossipcount++;
 814     }
 815     totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData);
 816     totlen += (sizeof(clusterMsgDataGossip)*gossipcount);
 817     hdr->count = htons(gossipcount);
 818     hdr->totlen = htonl(totlen);
 819     clusterSendMessage(link,buf,totlen);
 820 }
 821
 822 /* Send a message to all the nodes with a reliable link */
 823 void clusterBroadcastMessage(void *buf, size_t len) {
 824     dictIterator *di;
 825     dictEntry *de;
 826
 827     di = dictGetIterator(server.cluster.nodes);
 828     while((de = dictNext(di)) != NULL) {
 829         clusterNode *node = dictGetEntryVal(de);
 830
 831         if (!node->link) continue;
 832         if (node->flags & (REDIS_NODE_MYSELF|REDIS_NODE_NOADDR)) continue;
 833         clusterSendMessage(node->link,buf,len);
 834     }
 835     dictReleaseIterator(di);
 836 }
 837
 838 /* Send a FAIL message to all the nodes we are able to contact.
 839  * The FAIL message is sent when we detect that a node is failing
 840  * (REDIS_NODE_PFAIL) and we also receive a gossip confirmation of this:
 841  * we switch the node state to REDIS_NODE_FAIL and ask all the other
 842  * nodes to do the same ASAP. */
 843 void clusterSendFail(char *nodename) {
 844     unsigned char buf[1024];
 845     clusterMsg *hdr = (clusterMsg*) buf;
 846
 847     clusterBuildMessageHdr(hdr,CLUSTERMSG_TYPE_FAIL);
 848     memcpy(hdr->data.fail.about.nodename,nodename,REDIS_CLUSTER_NAMELEN);
 849     clusterBroadcastMessage(buf,ntohl(hdr->totlen));
 850 }
 851
 852 /* -----------------------------------------------------------------------------
 853  * CLUSTER cron job
 854  * -------------------------------------------------------------------------- */
 855
 856 /* This is executed 1 time every second */
 857 void clusterCron(void) {
 858     dictIterator *di;
 859     dictEntry *de;
 860     int j;
 861     time_t min_ping_sent = 0;
 862     clusterNode *min_ping_node = NULL;
 863
 864     /* Check if we have disconnected nodes and reestablish the connection. */
 865     di = dictGetIterator(server.cluster.nodes);
 866     while((de = dictNext(di)) != NULL) {
 867         clusterNode *node = dictGetEntryVal(de);
 868
 869         if (node->flags & (REDIS_NODE_MYSELF|REDIS_NODE_NOADDR)) continue;
 870         if (node->link == NULL) {
 871             int fd;
 872             clusterLink *link;
 873
 874             fd = anetTcpNonBlockConnect(server.neterr, node->ip,
 875                 node->port+REDIS_CLUSTER_PORT_INCR);
 876             if (fd == -1) continue;
 877             link = createClusterLink(node);
 878             link->fd = fd;
 879             node->link = link;
 880             aeCreateFileEvent(server.el,link->fd,AE_READABLE,clusterReadHandler,link);
 881             /* If the node is flagged as MEET, we send a MEET message instead
 882              * of a PING one, to force the receiver to add us in its node
 883              * table. */
 884             clusterSendPing(link, node->flags & REDIS_NODE_MEET ?
 885                     CLUSTERMSG_TYPE_MEET : CLUSTERMSG_TYPE_PING);
 886             /* We can clear the flag after the first packet is sent.
 887              * If we'll never receive a PONG, we'll never send new packets
 888              * to this node. Instead after the PONG is received and we
 889              * are no longer in meet/handshake status, we want to send
 890              * normal PING packets. */
 891             node->flags &= ~REDIS_NODE_MEET;
 892
 893             redisLog(REDIS_NOTICE,"Connecting with Node %.40s at %s:%d", node->name, node->ip, node->port+REDIS_CLUSTER_PORT_INCR);
 894         }
 895     }
 896     dictReleaseIterator(di);
 897
 898     /* Ping some random node. Check a few random nodes and ping the one with
 899      * the oldest ping_sent time */
 900     for (j = 0; j < 5; j++) {
 901         de = dictGetRandomKey(server.cluster.nodes);
 902         clusterNode *this = dictGetEntryVal(de);
 903
 904         if (this->link == NULL) continue;
 905         if (this->flags & (REDIS_NODE_MYSELF|REDIS_NODE_HANDSHAKE)) continue;
 906         if (min_ping_node == NULL || min_ping_sent > this->ping_sent) {
 907             min_ping_node = this;
 908             min_ping_sent = this->ping_sent;
 909         }
 910     }
 911     if (min_ping_node) {
 912         redisLog(REDIS_DEBUG,"Pinging node %40s", min_ping_node->name);
 913         clusterSendPing(min_ping_node->link, CLUSTERMSG_TYPE_PING);
 914     }
 915
 916     /* Iterate nodes to check if we need to flag something as failing */
 917     di = dictGetIterator(server.cluster.nodes);
 918     while((de = dictNext(di)) != NULL) {
 919         clusterNode *node = dictGetEntryVal(de);
 920         int delay;
 921
 922         if (node->flags &
 923             (REDIS_NODE_MYSELF|REDIS_NODE_NOADDR|REDIS_NODE_HANDSHAKE))
 924                 continue;
 925         /* Check only if we already sent a ping and did not received
 926          * a reply yet. */
 927         if (node->ping_sent == 0 ||
 928             node->ping_sent <= node->pong_received) continue;
 929
 930         delay = time(NULL) - node->pong_received;
 931         if (delay < server.cluster.node_timeout) {
 932             /* The PFAIL condition can be reversed without external
 933              * help if it is not transitive (that is, if it does not
 934              * turn into a FAIL state).
 935              *
 936              * The FAIL condition is also reversible if there are no slaves
 937              * for this host, so no slave election should be in progress.
 938              *
 939              * TODO: consider all the implications of resurrecting a
 940              * FAIL node. */
 941             if (node->flags & REDIS_NODE_PFAIL) {
 942                 node->flags &= ~REDIS_NODE_PFAIL;
 943             } else if (node->flags & REDIS_NODE_FAIL && !node->numslaves) {
 944                 node->flags &= ~REDIS_NODE_FAIL;
 945             }
 946         } else {
 947             /* Timeout reached. Set the noad se possibly failing if it is
 948              * not already in this state. */
 949             if (!(node->flags & (REDIS_NODE_PFAIL|REDIS_NODE_FAIL))) {
 950                 redisLog(REDIS_DEBUG,"*** NODE %.40s possibly failing",
 951                     node->name);
 952                 node->flags |= REDIS_NODE_PFAIL;
 953             }
 954         }
 955     }
 956     dictReleaseIterator(di);
 957 }
 958
 959 /* -----------------------------------------------------------------------------
 960  * Slots management
 961  * -------------------------------------------------------------------------- */
 962
 963 /* Set the slot bit and return the old value. */
 964 int clusterNodeSetSlotBit(clusterNode *n, int slot) {
 965     off_t byte = slot/8;
 966     int bit = slot&7;
 967     int old = (n->slots[byte] & (1<<bit)) != 0;
 968     n->slots[byte] |= 1<<bit;
 969     return old;
 970 }
 971
 972 /* Clear the slot bit and return the old value. */
 973 int clusterNodeClearSlotBit(clusterNode *n, int slot) {
 974     off_t byte = slot/8;
 975     int bit = slot&7;
 976     int old = (n->slots[byte] & (1<<bit)) != 0;
 977     n->slots[byte] &= ~(1<<bit);
 978     return old;
 979 }
 980
 981 /* Return the slot bit from the cluster node structure. */
 982 int clusterNodeGetSlotBit(clusterNode *n, int slot) {
 983     off_t byte = slot/8;
 984     int bit = slot&7;
 985     return (n->slots[byte] & (1<<bit)) != 0;
 986 }
 987
 988 /* Add the specified slot to the list of slots that node 'n' will
 989  * serve. Return REDIS_OK if the operation ended with success.
 990  * If the slot is already assigned to another instance this is considered
 991  * an error and REDIS_ERR is returned. */
 992 int clusterAddSlot(clusterNode *n, int slot) {
 993     redisAssert(clusterNodeSetSlotBit(n,slot) == 0);
 994     server.cluster.slots[slot] = server.cluster.myself;
 995     printf("SLOT %d added to %.40s\n", slot, n->name);
 996     return REDIS_OK;
 997 }
 998
 999 /* -----------------------------------------------------------------------------
1000  * Cluster state evaluation function
1001  * -------------------------------------------------------------------------- */
1002 void clusterUpdateState(void) {
1003     int ok = 1;
1004     int j;
1005
1006     for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) {
1007         if (server.cluster.slots[j] == NULL ||
1008             server.cluster.slots[j]->flags & (REDIS_NODE_FAIL))
1009         {
1010             ok = 0;
1011             break;
1012         }
1013     }
1014     if (ok) {
1015         if (server.cluster.state == REDIS_CLUSTER_NEEDHELP) {
1016             server.cluster.state = REDIS_CLUSTER_NEEDHELP;
1017         } else {
1018             server.cluster.state = REDIS_CLUSTER_OK;
1019         }
1020     } else {
1021         server.cluster.state = REDIS_CLUSTER_FAIL;
1022     }
1023 }
1024
1025 /* -----------------------------------------------------------------------------
1026  * CLUSTER command
1027  * -------------------------------------------------------------------------- */
1028
1029 sds clusterGenNodesDescription(void) {
1030     sds ci = sdsempty();
1031     dictIterator *di;
1032     dictEntry *de;
1033     int j, start;
1034
1035     di = dictGetIterator(server.cluster.nodes);
1036     while((de = dictNext(di)) != NULL) {
1037         clusterNode *node = dictGetEntryVal(de);
1038
1039         /* Node coordinates */
1040         ci = sdscatprintf(ci,"%.40s %s:%d ",
1041             node->name,
1042             node->ip,
1043             node->port);
1044
1045         /* Flags */
1046         if (node->flags == 0) ci = sdscat(ci,"noflags,");
1047         if (node->flags & REDIS_NODE_MYSELF) ci = sdscat(ci,"myself,");
1048         if (node->flags & REDIS_NODE_MASTER) ci = sdscat(ci,"master,");
1049         if (node->flags & REDIS_NODE_SLAVE) ci = sdscat(ci,"slave,");
1050         if (node->flags & REDIS_NODE_PFAIL) ci = sdscat(ci,"fail?,");
1051         if (node->flags & REDIS_NODE_FAIL) ci = sdscat(ci,"fail,");
1052         if (node->flags & REDIS_NODE_HANDSHAKE) ci =sdscat(ci,"handshake,");
1053         if (node->flags & REDIS_NODE_NOADDR) ci = sdscat(ci,"noaddr,");
1054         if (ci[sdslen(ci)-1] == ',') ci[sdslen(ci)-1] = ' ';
1055
1056         /* Slave of... or just "-" */
1057         if (node->slaveof)
1058             ci = sdscatprintf(ci,"%.40s ",node->slaveof->name);
1059         else
1060             ci = sdscatprintf(ci,"- ");
1061
1062         /* Latency from the POV of this node, link status */
1063         ci = sdscatprintf(ci,"%ld %ld %s",
1064             (long) node->ping_sent,
1065             (long) node->pong_received,
1066             node->link ? "connected" : "disconnected");
1067
1068         /* Slots served by this instance */
1069         start = -1;
1070         for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) {
1071             int bit;
1072
1073             if ((bit = clusterNodeGetSlotBit(node,j)) != 0) {
1074                 if (start == -1) start = j;
1075             }
1076             if (start != -1 && (!bit || j == REDIS_CLUSTER_SLOTS-1)) {
1077                 if (j == REDIS_CLUSTER_SLOTS-1) j++;
1078
1079                 if (start == j-1) {
1080                     ci = sdscatprintf(ci," %d",start);
1081                 } else {
1082                     ci = sdscatprintf(ci," %d-%d",start,j-1);
1083                 }
1084                 start = -1;
1085             }
1086         }
1087         ci = sdscatlen(ci,"\n",1);
1088     }
1089     dictReleaseIterator(di);
1090     return ci;
1091 }
1092
1093 void clusterCommand(redisClient *c) {
1094     if (server.cluster_enabled == 0) {
1095         addReplyError(c,"This instance has cluster support disabled");
1096         return;
1097     }
1098
1099     if (!strcasecmp(c->argv[1]->ptr,"meet") && c->argc == 4) {
1100         clusterNode *n;
1101         struct sockaddr_in sa;
1102         long port;
1103
1104         /* Perform sanity checks on IP/port */
1105         if (inet_aton(c->argv[2]->ptr,&sa.sin_addr) == 0) {
1106             addReplyError(c,"Invalid IP address in MEET");
1107             return;
1108         }
1109         if (getLongFromObjectOrReply(c, c->argv[3], &port, NULL) != REDIS_OK ||
1110                     port < 0 || port > (65535-REDIS_CLUSTER_PORT_INCR))
1111         {
1112             addReplyError(c,"Invalid TCP port specified");
1113             return;
1114         }
1115
1116         /* Finally add the node to the cluster with a random name, this
1117          * will get fixed in the first handshake (ping/pong). */
1118         n = createClusterNode(NULL,REDIS_NODE_HANDSHAKE|REDIS_NODE_MEET);
1119         strncpy(n->ip,inet_ntoa(sa.sin_addr),sizeof(n->ip));
1120         n->port = port;
1121         clusterAddNode(n);
1122         addReply(c,shared.ok);
1123     } else if (!strcasecmp(c->argv[1]->ptr,"nodes") && c->argc == 2) {
1124         robj *o;
1125         sds ci = clusterGenNodesDescription();
1126
1127         o = createObject(REDIS_STRING,ci);
1128         addReplyBulk(c,o);
1129         decrRefCount(o);
1130     } else if (!strcasecmp(c->argv[1]->ptr,"addslots") && c->argc >= 3) {
1131         int j;
1132         long long slot;
1133         unsigned char *slots = zmalloc(REDIS_CLUSTER_SLOTS);
1134
1135         memset(slots,0,REDIS_CLUSTER_SLOTS);
1136         /* Check that all the arguments are parsable and that all the
1137          * slots are not already busy. */
1138         for (j = 2; j < c->argc; j++) {
1139             if (getLongLongFromObject(c->argv[j],&slot) != REDIS_OK ||
1140                 slot < 0 || slot > REDIS_CLUSTER_SLOTS)
1141             {
1142                 addReplyError(c,"Invalid or out of range slot index");
1143                 zfree(slots);
1144                 return;
1145             }
1146             if (server.cluster.slots[slot]) {
1147                 addReplyErrorFormat(c,"Slot %lld is already busy", slot);
1148                 zfree(slots);
1149                 return;
1150             }
1151             if (slots[slot]++ == 1) {
1152                 addReplyErrorFormat(c,"Slot %d specified multiple times",
1153                     (int)slot);
1154                 zfree(slots);
1155                 return;
1156             }
1157         }
1158         for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) {
1159             if (slots[j]) {
1160                 int retval = clusterAddSlot(server.cluster.myself,j);
1161
1162                 redisAssert(retval == REDIS_OK);
1163             }
1164         }
1165         zfree(slots);
1166         clusterUpdateState();
1167         clusterSaveConfigOrDie();
1168         addReply(c,shared.ok);
1169     } else if (!strcasecmp(c->argv[1]->ptr,"info") && c->argc == 2) {
1170         char *statestr[] = {"ok","fail","needhelp"};
1171         int slots_assigned = 0, slots_ok = 0, slots_pfail = 0, slots_fail = 0;
1172         int j;
1173
1174         for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) {
1175             clusterNode *n = server.cluster.slots[j];
1176
1177             if (n == NULL) continue;
1178             slots_assigned++;
1179             if (n->flags & REDIS_NODE_FAIL) {
1180                 slots_fail++;
1181             } else if (n->flags & REDIS_NODE_PFAIL) {
1182                 slots_pfail++;
1183             } else {
1184                 slots_ok++;
1185             }
1186         }
1187
1188         sds info = sdscatprintf(sdsempty(),
1189             "cluster_state:%s\r\n"
1190             "cluster_slots_assigned:%d\r\n"
1191             "cluster_slots_ok:%d\r\n"
1192             "cluster_slots_pfail:%d\r\n"
1193             "cluster_slots_fail:%d\r\n"
1194             , statestr[server.cluster.state],
1195             slots_assigned,
1196             slots_ok,
1197             slots_pfail,
1198             slots_fail
1199         );
1200         addReplySds(c,sdscatprintf(sdsempty(),"$%lu\r\n",
1201             (unsigned long)sdslen(info)));
1202         addReplySds(c,info);
1203         addReply(c,shared.crlf);
1204     } else {
1205         addReplyError(c,"Wrong CLUSTER subcommand or number of arguments");
1206     }
1207 }
1208
1209 /* -----------------------------------------------------------------------------
1210  * RESTORE and MIGRATE commands
1211  * -------------------------------------------------------------------------- */
1212
1213 /* RESTORE key ttl serialized-value */
1214 void restoreCommand(redisClient *c) {
1215     FILE *fp;
1216     char buf[64];
1217     robj *o;
1218     unsigned char *data;
1219     long ttl;
1220
1221     /* Make sure this key does not already exist here... */
1222     if (dbExists(c->db,c->argv[1])) {
1223         addReplyError(c,"Target key name is busy.");
1224         return;
1225     }
1226
1227     /* Check if the TTL value makes sense */
1228     if (getLongFromObjectOrReply(c,c->argv[2],&ttl,NULL) != REDIS_OK) {
1229         return;
1230     } else if (ttl < 0) {
1231         addReplyError(c,"Invalid TTL value, must be >= 0");
1232         return;
1233     }
1234
1235     /* rdbLoadObject() only works against file descriptors so we need to
1236      * dump the serialized object into a file and reload. */
1237     snprintf(buf,sizeof(buf),"redis-restore-%d.tmp",getpid());
1238     fp = fopen(buf,"w+");
1239     if (!fp) {
1240         redisLog(REDIS_WARNING,"Can't open tmp file for RESTORE: %s",
1241             strerror(errno));
1242         addReplyErrorFormat(c,"RESTORE failed, tmp file creation error: %s",
1243             strerror(errno));
1244         return;
1245     }
1246     unlink(buf);
1247
1248     /* Write the actual data and rewind the file */
1249     data = (unsigned char*) c->argv[3]->ptr;
1250     if (fwrite(data+1,sdslen((sds)data)-1,1,fp) != 1) {
1251         redisLog(REDIS_WARNING,"Can't write against tmp file for RESTORE: %s",
1252             strerror(errno));
1253         addReplyError(c,"RESTORE failed, tmp file I/O error.");
1254         fclose(fp);
1255         return;
1256     }
1257     rewind(fp);
1258
1259     /* Finally create the object from the serialized dump and
1260      * store it at the specified key. */
1261     if ((data[0] > 4 && data[0] < 9) ||
1262          data[0] > 11 ||
1263         (o = rdbLoadObject(data[0],fp)) == NULL)
1264     {
1265         addReplyError(c,"Bad data format.");
1266         fclose(fp);
1267         return;
1268     }
1269     fclose(fp);
1270
1271     /* Create the key and set the TTL if any */
1272     dbAdd(c->db,c->argv[1],o);
1273     if (ttl) setExpire(c->db,c->argv[1],time(NULL)+ttl);
1274     addReply(c,shared.ok);
1275 }
1276
1277 /* MIGRATE host port key dbid timeout */
1278 void migrateCommand(redisClient *c) {
1279     int fd;
1280     long timeout;
1281     long dbid;
1282     char buf[64];
1283     FILE *fp;
1284     time_t ttl;
1285     robj *o;
1286     unsigned char type;
1287     off_t payload_len;
1288
1289     /* Sanity check */
1290     if (getLongFromObjectOrReply(c,c->argv[5],&timeout,NULL) != REDIS_OK)
1291         return;
1292     if (getLongFromObjectOrReply(c,c->argv[4],&dbid,NULL) != REDIS_OK)
1293         return;
1294     if (timeout <= 0) timeout = 1;
1295
1296     /* Check if the key is here. If not we reply with success as there is
1297      * nothing to migrate (for instance the key expired in the meantime), but
1298      * we include such information in the reply string. */
1299     if ((o = lookupKeyRead(c->db,c->argv[3])) == NULL) {
1300         addReplySds(c,sdsnew("+NOKEY"));
1301         return;
1302     }
1303
1304     /* Connect */
1305     fd = anetTcpNonBlockConnect(server.neterr,c->argv[1]->ptr,
1306                 atoi(c->argv[2]->ptr));
1307     if (fd == -1) {
1308         addReplyErrorFormat(c,"Can't connect to target node: %s",
1309             server.neterr);
1310         return;
1311     }
1312     if ((aeWait(fd,AE_WRITABLE,timeout*1000) & AE_WRITABLE) == 0) {
1313         addReplyError(c,"Timeout connecting to the client");
1314         return;
1315     }
1316
1317     /* Create temp file */
1318     snprintf(buf,sizeof(buf),"redis-migrate-%d.tmp",getpid());
1319     fp = fopen(buf,"w+");
1320     if (!fp) {
1321         redisLog(REDIS_WARNING,"Can't open tmp file for MIGRATE: %s",
1322             strerror(errno));
1323         addReplyErrorFormat(c,"MIGRATE failed, tmp file creation error: %s.",
1324             strerror(errno));
1325         return;
1326     }
1327     unlink(buf);
1328
1329     /* Build the SELECT + RESTORE query writing it in our temp file. */
1330     if (fwriteBulkCount(fp,'*',2) == 0) goto file_wr_err;
1331     if (fwriteBulkString(fp,"SELECT",6) == 0) goto file_wr_err;
1332     if (fwriteBulkLongLong(fp,dbid) == 0) goto file_wr_err;
1333
1334     ttl = getExpire(c->db,c->argv[3]);
1335     type = o->type;
1336     if (fwriteBulkCount(fp,'*',4) == 0) goto file_wr_err;
1337     if (fwriteBulkString(fp,"RESTORE",7) == 0) goto file_wr_err;
1338     if (fwriteBulkObject(fp,c->argv[3]) == 0) goto file_wr_err;
1339     if (fwriteBulkLongLong(fp, (ttl == -1) ? 0 : ttl) == 0) goto file_wr_err;
1340
1341     /* Finally the last argument that is the serailized object payload
1342      * in the form: <type><rdb-serailized-object>. */
1343     payload_len = rdbSavedObjectLen(o);
1344     if (fwriteBulkCount(fp,'$',payload_len+1) == 0) goto file_wr_err;
1345     if (fwrite(&type,1,1,fp) == 0) goto file_wr_err;
1346     if (rdbSaveObject(fp,o) == -1) goto file_wr_err;
1347     if (fwrite("\r\n",2,1,fp) == 0) goto file_wr_err;
1348
1349     /* Tranfer the query to the other node */
1350     rewind(fp);
1351     {
1352         char buf[4096];
1353         size_t nread;
1354
1355         while ((nread = fread(buf,1,sizeof(buf),fp)) != 0) {
1356             int nwritten;
1357
1358             nwritten = syncWrite(fd,buf,nread,timeout);
1359             if (nwritten != (signed)nread) goto socket_wr_err;
1360         }
1361         if (ferror(fp)) goto file_rd_err;
1362     }
1363
1364     /* Read back the reply */
1365     {
1366         char buf1[1024];
1367         char buf2[1024];
1368
1369         /* Read the two replies */
1370         if (syncReadLine(fd, buf1, sizeof(buf1), timeout) <= 0)
1371             goto socket_rd_err;
1372         if (syncReadLine(fd, buf2, sizeof(buf2), timeout) <= 0)
1373                 goto socket_rd_err;
1374         if (buf1[0] == '-' || buf2[0] == '-') {
1375             addReplyErrorFormat(c,"Target instance replied with error: %s",
1376                 (buf1[0] == '-') ? buf1+1 : buf2+1);
1377         } else {
1378             dbDelete(c->db,c->argv[3]);
1379             addReply(c,shared.ok);
1380         }
1381     }
1382     fclose(fp);
1383     close(fd);
1384     return;
1385
1386 file_wr_err:
1387     redisLog(REDIS_WARNING,"Can't write on tmp file for MIGRATE: %s",
1388         strerror(errno));
1389     addReplyErrorFormat(c,"MIGRATE failed, tmp file write error: %s.",
1390         strerror(errno));
1391     fclose(fp);
1392     close(fd);
1393     return;
1394
1395 file_rd_err:
1396     redisLog(REDIS_WARNING,"Can't read from tmp file for MIGRATE: %s",
1397         strerror(errno));
1398     addReplyErrorFormat(c,"MIGRATE failed, tmp file read error: %s.",
1399         strerror(errno));
1400     fclose(fp);
1401     close(fd);
1402     return;
1403
1404 socket_wr_err:
1405     redisLog(REDIS_NOTICE,"Can't write to target node for MIGRATE: %s",
1406         strerror(errno));
1407     addReplyErrorFormat(c,"MIGRATE failed, writing to target node: %s.",
1408         strerror(errno));
1409     fclose(fp);
1410     close(fd);
1411     return;
1412
1413 socket_rd_err:
1414     redisLog(REDIS_NOTICE,"Can't read from target node for MIGRATE: %s",
1415         strerror(errno));
1416     addReplyErrorFormat(c,"MIGRATE failed, reading from target node: %s.",
1417         strerror(errno));
1418     fclose(fp);
1419     close(fd);
1420     return;
1421 }
1422
1423 /* DUMP keyname
1424  * DUMP is actually not used by Redis Cluster but it is the obvious
1425  * complement of RESTORE and can be useful for different applications. */
1426 void dumpCommand(redisClient *c) {
1427     char buf[64];
1428     FILE *fp;
1429     robj *o, *dumpobj;
1430     sds dump = NULL;
1431     off_t payload_len;
1432     unsigned int type;
1433
1434     /* Check if the key is here. */
1435     if ((o = lookupKeyRead(c->db,c->argv[1])) == NULL) {
1436         addReply(c,shared.nullbulk);
1437         return;
1438     }
1439
1440     /* Create temp file */
1441     snprintf(buf,sizeof(buf),"redis-dump-%d.tmp",getpid());
1442     fp = fopen(buf,"w+");
1443     if (!fp) {
1444         redisLog(REDIS_WARNING,"Can't open tmp file for MIGRATE: %s",
1445             strerror(errno));
1446         addReplyErrorFormat(c,"DUMP failed, tmp file creation error: %s.",
1447             strerror(errno));
1448         return;
1449     }
1450     unlink(buf);
1451
1452     /* Dump the serailized object and read it back in memory.
1453      * We prefix it with a one byte containing the type ID.
1454      * This is the serialization format understood by RESTORE. */
1455     if (rdbSaveObject(fp,o) == -1) goto file_wr_err;
1456     payload_len = ftello(fp);
1457     if (fseeko(fp,0,SEEK_SET) == -1) goto file_rd_err;
1458     dump = sdsnewlen(NULL,payload_len+1);
1459     if (payload_len && fread(dump+1,payload_len,1,fp) != 1) goto file_rd_err;
1460     fclose(fp);
1461     type = o->type;
1462     if (type == REDIS_LIST && o->encoding == REDIS_ENCODING_ZIPLIST)
1463         type = REDIS_LIST_ZIPLIST;
1464     else if (type == REDIS_HASH && o->encoding == REDIS_ENCODING_ZIPMAP)
1465         type = REDIS_HASH_ZIPMAP;
1466     else if (type == REDIS_SET && o->encoding == REDIS_ENCODING_INTSET)
1467         type = REDIS_SET_INTSET;
1468     else
1469         type = o->type;
1470     dump[0] = type;
1471
1472     /* Transfer to the client */
1473     dumpobj = createObject(REDIS_STRING,dump);
1474     addReplyBulk(c,dumpobj);
1475     decrRefCount(dumpobj);
1476     return;
1477
1478 file_wr_err:
1479     redisLog(REDIS_WARNING,"Can't write on tmp file for DUMP: %s",
1480         strerror(errno));
1481     addReplyErrorFormat(c,"DUMP failed, tmp file write error: %s.",
1482         strerror(errno));
1483     sdsfree(dump);
1484     fclose(fp);
1485     return;
1486
1487 file_rd_err:
1488     redisLog(REDIS_WARNING,"Can't read from tmp file for DUMP: %s",
1489         strerror(errno));
1490     addReplyErrorFormat(c,"DUMP failed, tmp file read error: %s.",
1491         strerror(errno));
1492     sdsfree(dump);
1493     fclose(fp);
1494     return;
1495 }
1496
1497 /* -----------------------------------------------------------------------------
1498  * Cluster functions related to serving / redirecting clients
1499  * -------------------------------------------------------------------------- */
1500
1501 /* Return the pointer to the cluster node that is able to serve the query
1502  * as all the keys belong to hash slots for which the node is in charge.
1503  *
1504  * If keys in query spawn multiple nodes NULL is returned. */
1505 clusterNode *getNodeByQuery(redisClient *c, struct redisCommand *cmd, robj **argv, int argc, int *hashslot) {
1506     clusterNode *n = NULL;
1507     multiState *ms, _ms;
1508     multiCmd mc;
1509     int i;
1510
1511     /* We handle all the cases as if they were EXEC commands, so we have
1512      * a common code path for everything */
1513     if (cmd->proc == execCommand) {
1514         /* If REDIS_MULTI flag is not set EXEC is just going to return an
1515          * error. */
1516         if (!(c->flags & REDIS_MULTI)) return server.cluster.myself;
1517         ms = &c->mstate;
1518     } else {
1519         /* Create a fake Multi State structure, with just one command */
1520         ms = &_ms;
1521         _ms.commands = &mc;
1522         _ms.count = 1;
1523         mc.argv = argv;
1524         mc.argc = argc;
1525         mc.cmd = cmd;
1526     }
1527
1528     for (i = 0; i < ms->count; i++) {
1529         struct redisCommand *mcmd;
1530         robj **margv;
1531         int margc, *keyindex, numkeys, j;
1532
1533         mcmd = ms->commands[i].cmd;
1534         margc = ms->commands[i].argc;
1535         margv = ms->commands[i].argv;
1536
1537         keyindex = getKeysFromCommand(mcmd,margv,margc,&numkeys,
1538                                       REDIS_GETKEYS_PRELOAD);
1539         for (j = 0; j < numkeys; j++) {
1540             int slot = keyHashSlot((char*)margv[keyindex[j]]->ptr,
1541                 sdslen(margv[keyindex[j]]->ptr));
1542             struct clusterNode *slotnode;
1543
1544             slotnode = server.cluster.slots[slot];
1545             if (hashslot) *hashslot = slot;
1546             /* Node not assigned? (Should never happen actually
1547              * if we reached this function).
1548              * Different node than the previous one?
1549              * Return NULL, the cluster can't serve multi-node requests */
1550             if (slotnode == NULL || (n && slotnode != n)) {
1551                 getKeysFreeResult(keyindex);
1552                 return NULL;
1553             } else {
1554                 n = slotnode;
1555             }
1556         }
1557         getKeysFreeResult(keyindex);
1558     }
1559     return (n == NULL) ? server.cluster.myself : n;
1560 }