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