src/cluster.c

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