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