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