1 /* Redis Cluster implementation. 2 * 3 * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com> 4 * All rights reserved. 5 * 6 * Redistribution and use in source and binary forms, with or without 7 * modification, are permitted provided that the following conditions are met: 8 * 9 * * Redistributions of source code must retain the above copyright notice, 10 * this list of conditions and the following disclaimer. 11 * * Redistributions in binary form must reproduce the above copyright 12 * notice, this list of conditions and the following disclaimer in the 13 * documentation and/or other materials provided with the distribution. 14 * * Neither the name of Redis nor the names of its contributors may be used 15 * to endorse or promote products derived from this software without 16 * specific prior written permission. 17 * 18 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 19 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 20 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 21 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE 22 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 23 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 24 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 25 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 26 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) 27 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 28 * POSSIBILITY OF SUCH DAMAGE. 29 */ 30 31 #include "redis.h" 32 #include "cluster.h" 33 #include "endianconv.h" 34 35 #include <sys/types.h> 36 #include <sys/socket.h> 37 #include <arpa/inet.h> 38 #include <fcntl.h> 39 #include <unistd.h> 40 #include <sys/socket.h> 41 #include <sys/stat.h> 42 43 /* A global reference to myself is handy to make code more clear. 44 * Myself always points to server.cluster->myself, that is, the clusterNode 45 * that represents this node. */ 46 clusterNode *myself = NULL; 47 48 clusterNode *createClusterNode(char *nodename, int flags); 49 int clusterAddNode(clusterNode *node); 50 void clusterAcceptHandler(aeEventLoop *el, int fd, void *privdata, int mask); 51 void clusterReadHandler(aeEventLoop *el, int fd, void *privdata, int mask); 52 void clusterSendPing(clusterLink *link, int type); 53 void clusterSendFail(char *nodename); 54 void clusterSendFailoverAuthIfNeeded(clusterNode *node, clusterMsg *request); 55 void clusterUpdateState(void); 56 int clusterNodeGetSlotBit(clusterNode *n, int slot); 57 sds clusterGenNodesDescription(int filter); 58 clusterNode *clusterLookupNode(char *name); 59 int clusterNodeAddSlave(clusterNode *master, clusterNode *slave); 60 int clusterAddSlot(clusterNode *n, int slot); 61 int clusterDelSlot(int slot); 62 int clusterDelNodeSlots(clusterNode *node); 63 int clusterNodeSetSlotBit(clusterNode *n, int slot); 64 void clusterSetMaster(clusterNode *n); 65 void clusterHandleSlaveFailover(void); 66 void clusterHandleSlaveMigration(int max_slaves); 67 int bitmapTestBit(unsigned char *bitmap, int pos); 68 void clusterDoBeforeSleep(int flags); 69 void clusterSendUpdate(clusterLink *link, clusterNode *node); 70 void resetManualFailover(void); 71 void clusterCloseAllSlots(void); 72 73 /* ----------------------------------------------------------------------------- 74 * Initialization 75 * -------------------------------------------------------------------------- */ 76 77 /* Return the greatest configEpoch found in the cluster. */ 78 uint64_t clusterGetMaxEpoch(void) { 79 uint64_t max = 0; 80 dictIterator *di; 81 dictEntry *de; 82 83 di = dictGetSafeIterator(server.cluster->nodes); 84 while((de = dictNext(di)) != NULL) { 85 clusterNode *node = dictGetVal(de); 86 if (node->configEpoch > max) max = node->configEpoch; 87 } 88 dictReleaseIterator(di); 89 return max; 90 } 91 92 int clusterLoadConfig(char *filename) { 93 FILE *fp = fopen(filename,"r"); 94 char *line; 95 int maxline, j; 96 97 if (fp == NULL) return REDIS_ERR; 98 99 /* Parse the file. Note that single liens of the cluster config file can 100 * be really long as they include all the hash slots of the node. 101 * This means in the worst possible case, half of the Redis slots will be 102 * present in a single line, possibly in importing or migrating state, so 103 * together with the node ID of the sender/receiver. 104 * 105 * To simplify we allocate 1024+REDIS_CLUSTER_SLOTS*128 bytes per line. */ 106 maxline = 1024+REDIS_CLUSTER_SLOTS*128; 107 line = zmalloc(maxline); 108 while(fgets(line,maxline,fp) != NULL) { 109 int argc; 110 sds *argv; 111 clusterNode *n, *master; 112 char *p, *s; 113 114 /* Skip blank lines, they can be created either by users manually 115 * editing nodes.conf or by the config writing process if stopped 116 * before the truncate() call. */ 117 if (line[0] == '\n') continue; 118 119 /* Split the line into arguments for processing. */ 120 argv = sdssplitargs(line,&argc); 121 if (argv == NULL) goto fmterr; 122 123 /* Create this node if it does not exist */ 124 n = clusterLookupNode(argv[0]); 125 if (!n) { 126 n = createClusterNode(argv[0],0); 127 clusterAddNode(n); 128 } 129 /* Address and port */ 130 if ((p = strchr(argv[1],':')) == NULL) goto fmterr; 131 *p = '\0'; 132 memcpy(n->ip,argv[1],strlen(argv[1])+1); 133 n->port = atoi(p+1); 134 135 /* Parse flags */ 136 p = s = argv[2]; 137 while(p) { 138 p = strchr(s,','); 139 if (p) *p = '\0'; 140 if (!strcasecmp(s,"myself")) { 141 redisAssert(server.cluster->myself == NULL); 142 myself = server.cluster->myself = n; 143 n->flags |= REDIS_NODE_MYSELF; 144 } else if (!strcasecmp(s,"master")) { 145 n->flags |= REDIS_NODE_MASTER; 146 } else if (!strcasecmp(s,"slave")) { 147 n->flags |= REDIS_NODE_SLAVE; 148 } else if (!strcasecmp(s,"fail?")) { 149 n->flags |= REDIS_NODE_PFAIL; 150 } else if (!strcasecmp(s,"fail")) { 151 n->flags |= REDIS_NODE_FAIL; 152 n->fail_time = mstime(); 153 } else if (!strcasecmp(s,"handshake")) { 154 n->flags |= REDIS_NODE_HANDSHAKE; 155 } else if (!strcasecmp(s,"noaddr")) { 156 n->flags |= REDIS_NODE_NOADDR; 157 } else if (!strcasecmp(s,"noflags")) { 158 /* nothing to do */ 159 } else { 160 redisPanic("Unknown flag in redis cluster config file"); 161 } 162 if (p) s = p+1; 163 } 164 165 /* Get master if any. Set the master and populate master's 166 * slave list. */ 167 if (argv[3][0] != '-') { 168 master = clusterLookupNode(argv[3]); 169 if (!master) { 170 master = createClusterNode(argv[3],0); 171 clusterAddNode(master); 172 } 173 n->slaveof = master; 174 clusterNodeAddSlave(master,n); 175 } 176 177 /* Set ping sent / pong received timestamps */ 178 if (atoi(argv[4])) n->ping_sent = mstime(); 179 if (atoi(argv[5])) n->pong_received = mstime(); 180 181 /* Set configEpoch for this node. */ 182 n->configEpoch = strtoull(argv[6],NULL,10); 183 184 /* Populate hash slots served by this instance. */ 185 for (j = 8; j < argc; j++) { 186 int start, stop; 187 188 if (argv[j][0] == '[') { 189 /* Here we handle migrating / importing slots */ 190 int slot; 191 char direction; 192 clusterNode *cn; 193 194 p = strchr(argv[j],'-'); 195 redisAssert(p != NULL); 196 *p = '\0'; 197 direction = p[1]; /* Either '>' or '<' */ 198 slot = atoi(argv[j]+1); 199 p += 3; 200 cn = clusterLookupNode(p); 201 if (!cn) { 202 cn = createClusterNode(p,0); 203 clusterAddNode(cn); 204 } 205 if (direction == '>') { 206 server.cluster->migrating_slots_to[slot] = cn; 207 } else { 208 server.cluster->importing_slots_from[slot] = cn; 209 } 210 continue; 211 } else if ((p = strchr(argv[j],'-')) != NULL) { 212 *p = '\0'; 213 start = atoi(argv[j]); 214 stop = atoi(p+1); 215 } else { 216 start = stop = atoi(argv[j]); 217 } 218 while(start <= stop) clusterAddSlot(n, start++); 219 } 220 221 sdsfreesplitres(argv,argc); 222 } 223 zfree(line); 224 fclose(fp); 225 226 /* Config sanity check */ 227 redisAssert(server.cluster->myself != NULL); 228 redisLog(REDIS_NOTICE,"Node configuration loaded, I'm %.40s", myself->name); 229 /* Set the currentEpoch to the max epoch found in the master. 230 * FIXME: this should actually be part of the persistent state, as 231 * documented in the Github issue #1479. */ 232 server.cluster->currentEpoch = clusterGetMaxEpoch(); 233 return REDIS_OK; 234 235 fmterr: 236 redisLog(REDIS_WARNING,"Unrecoverable error: corrupted cluster config file."); 237 fclose(fp); 238 exit(1); 239 } 240 241 /* Cluster node configuration is exactly the same as CLUSTER NODES output. 242 * 243 * This function writes the node config and returns 0, on error -1 244 * is returned. 245 * 246 * Note: we need to write the file in an atomic way from the point of view 247 * of the POSIX filesystem semantics, so that if the server is stopped 248 * or crashes during the write, we'll end with either the old file or the 249 * new one. Since we have the full payload to write available we can use 250 * a single write to write the whole file. If the pre-existing file was 251 * bigger we pad our payload with newlines that are anyway ignored and truncate 252 * the file afterward. */ 253 int clusterSaveConfig(int do_fsync) { 254 sds ci = clusterGenNodesDescription(REDIS_NODE_HANDSHAKE); 255 size_t content_size = sdslen(ci); 256 struct stat sb; 257 int fd; 258 259 if ((fd = open(server.cluster_configfile,O_WRONLY|O_CREAT,0644)) 260 == -1) goto err; 261 262 /* Pad the new payload if the existing file length is greater. */ 263 if (fstat(fd,&sb) != -1) { 264 if (sb.st_size > content_size) { 265 ci = sdsgrowzero(ci,sb.st_size); 266 memset(ci+content_size,'\n',sb.st_size-content_size); 267 } 268 } 269 if (write(fd,ci,sdslen(ci)) != (ssize_t)sdslen(ci)) goto err; 270 if (do_fsync) fsync(fd); 271 272 /* Truncate the file if needed to remove the final \n padding that 273 * is just garbage. */ 274 if (content_size != sdslen(ci) && ftruncate(fd,content_size) == -1) { 275 /* ftruncate() failing is not a critical error. */ 276 } 277 close(fd); 278 sdsfree(ci); 279 return 0; 280 281 err: 282 if (fd != -1) close(fd); 283 sdsfree(ci); 284 return -1; 285 } 286 287 void clusterSaveConfigOrDie(int do_fsync) { 288 if (clusterSaveConfig(do_fsync) == -1) { 289 redisLog(REDIS_WARNING,"Fatal: can't update cluster config file."); 290 exit(1); 291 } 292 } 293 294 void clusterInit(void) { 295 int saveconf = 0; 296 297 server.cluster = zmalloc(sizeof(clusterState)); 298 server.cluster->myself = NULL; 299 server.cluster->currentEpoch = 0; 300 server.cluster->state = REDIS_CLUSTER_FAIL; 301 server.cluster->size = 1; 302 server.cluster->todo_before_sleep = 0; 303 server.cluster->nodes = dictCreate(&clusterNodesDictType,NULL); 304 server.cluster->nodes_black_list = 305 dictCreate(&clusterNodesBlackListDictType,NULL); 306 server.cluster->failover_auth_time = 0; 307 server.cluster->failover_auth_count = 0; 308 server.cluster->failover_auth_rank = 0; 309 server.cluster->failover_auth_epoch = 0; 310 server.cluster->last_vote_epoch = 0; 311 server.cluster->stats_bus_messages_sent = 0; 312 server.cluster->stats_bus_messages_received = 0; 313 memset(server.cluster->slots,0, sizeof(server.cluster->slots)); 314 clusterCloseAllSlots(); 315 if (clusterLoadConfig(server.cluster_configfile) == REDIS_ERR) { 316 /* No configuration found. We will just use the random name provided 317 * by the createClusterNode() function. */ 318 myself = server.cluster->myself = 319 createClusterNode(NULL,REDIS_NODE_MYSELF|REDIS_NODE_MASTER); 320 redisLog(REDIS_NOTICE,"No cluster configuration found, I'm %.40s", 321 myself->name); 322 clusterAddNode(myself); 323 saveconf = 1; 324 } 325 if (saveconf) clusterSaveConfigOrDie(1); 326 327 /* We need a listening TCP port for our cluster messaging needs. */ 328 server.cfd_count = 0; 329 330 /* Port sanity check II 331 * The other handshake port check is triggered too late to stop 332 * us from trying to use a too-high cluster port number. */ 333 if (server.port > (65535-REDIS_CLUSTER_PORT_INCR)) { 334 redisLog(REDIS_WARNING, "Redis port number too high. " 335 "Cluster communication port is 10,000 port " 336 "numbers higher than your Redis port. " 337 "Your Redis port number must be " 338 "lower than 55535."); 339 exit(1); 340 } 341 342 if (listenToPort(server.port+REDIS_CLUSTER_PORT_INCR, 343 server.cfd,&server.cfd_count) == REDIS_ERR) 344 { 345 exit(1); 346 } else { 347 int j; 348 349 for (j = 0; j < server.cfd_count; j++) { 350 if (aeCreateFileEvent(server.el, server.cfd[j], AE_READABLE, 351 clusterAcceptHandler, NULL) == AE_ERR) 352 redisPanic("Unrecoverable error creating Redis Cluster " 353 "file event."); 354 } 355 } 356 357 /* The slots -> keys map is a sorted set. Init it. */ 358 server.cluster->slots_to_keys = zslCreate(); 359 resetManualFailover(); 360 } 361 362 /* ----------------------------------------------------------------------------- 363 * CLUSTER communication link 364 * -------------------------------------------------------------------------- */ 365 366 clusterLink *createClusterLink(clusterNode *node) { 367 clusterLink *link = zmalloc(sizeof(*link)); 368 link->ctime = mstime(); 369 link->sndbuf = sdsempty(); 370 link->rcvbuf = sdsempty(); 371 link->node = node; 372 link->fd = -1; 373 return link; 374 } 375 376 /* Free a cluster link, but does not free the associated node of course. 377 * This function will just make sure that the original node associated 378 * with this link will have the 'link' field set to NULL. */ 379 void freeClusterLink(clusterLink *link) { 380 if (link->fd != -1) { 381 aeDeleteFileEvent(server.el, link->fd, AE_WRITABLE); 382 aeDeleteFileEvent(server.el, link->fd, AE_READABLE); 383 } 384 sdsfree(link->sndbuf); 385 sdsfree(link->rcvbuf); 386 if (link->node) 387 link->node->link = NULL; 388 close(link->fd); 389 zfree(link); 390 } 391 392 void clusterAcceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) { 393 int cport, cfd; 394 char cip[REDIS_IP_STR_LEN]; 395 clusterLink *link; 396 REDIS_NOTUSED(el); 397 REDIS_NOTUSED(mask); 398 REDIS_NOTUSED(privdata); 399 400 cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip), &cport); 401 if (cfd == ANET_ERR) { 402 redisLog(REDIS_VERBOSE,"Accepting cluster node: %s", server.neterr); 403 return; 404 } 405 anetNonBlock(NULL,cfd); 406 anetEnableTcpNoDelay(NULL,cfd); 407 408 /* Use non-blocking I/O for cluster messages. */ 409 /* IPV6: might want to wrap a v6 address in [] */ 410 redisLog(REDIS_VERBOSE,"Accepted cluster node %s:%d", cip, cport); 411 /* We need to create a temporary node in order to read the incoming 412 * packet in a valid contest. This node will be released once we 413 * read the packet and reply. */ 414 link = createClusterLink(NULL); 415 link->fd = cfd; 416 aeCreateFileEvent(server.el,cfd,AE_READABLE,clusterReadHandler,link); 417 } 418 419 /* ----------------------------------------------------------------------------- 420 * Key space handling 421 * -------------------------------------------------------------------------- */ 422 423 /* We have 16384 hash slots. The hash slot of a given key is obtained 424 * as the least significant 14 bits of the crc16 of the key. 425 * 426 * However if the key contains the {...} pattern, only the part between 427 * { and } is hashed. This may be useful in the future to force certain 428 * keys to be in the same node (assuming no resharding is in progress). */ 429 unsigned int keyHashSlot(char *key, int keylen) { 430 int s, e; /* start-end indexes of { and } */ 431 432 for (s = 0; s < keylen; s++) 433 if (key[s] == '{') break; 434 435 /* No '{' ? Hash the whole key. This is the base case. */ 436 if (s == keylen) return crc16(key,keylen) & 0x3FFF; 437 438 /* '{' found? Check if we have the corresponding '}'. */ 439 for (e = s+1; e < keylen; e++) 440 if (key[e] == '}') break; 441 442 /* No '}' or nothing betweeen {} ? Hash the whole key. */ 443 if (e == keylen || e == s+1) return crc16(key,keylen) & 0x3FFF; 444 445 /* If we are here there is both a { and a } on its right. Hash 446 * what is in the middle between { and }. */ 447 return crc16(key+s+1,e-s-1) & 0x3FFF; 448 } 449 450 /* ----------------------------------------------------------------------------- 451 * CLUSTER node API 452 * -------------------------------------------------------------------------- */ 453 454 /* Create a new cluster node, with the specified flags. 455 * If "nodename" is NULL this is considered a first handshake and a random 456 * node name is assigned to this node (it will be fixed later when we'll 457 * receive the first pong). 458 * 459 * The node is created and returned to the user, but it is not automatically 460 * added to the nodes hash table. */ 461 clusterNode *createClusterNode(char *nodename, int flags) { 462 clusterNode *node = zmalloc(sizeof(*node)); 463 464 if (nodename) 465 memcpy(node->name, nodename, REDIS_CLUSTER_NAMELEN); 466 else 467 getRandomHexChars(node->name, REDIS_CLUSTER_NAMELEN); 468 node->ctime = mstime(); 469 node->configEpoch = 0; 470 node->flags = flags; 471 memset(node->slots,0,sizeof(node->slots)); 472 node->numslots = 0; 473 node->numslaves = 0; 474 node->slaves = NULL; 475 node->slaveof = NULL; 476 node->ping_sent = node->pong_received = 0; 477 node->fail_time = 0; 478 node->link = NULL; 479 memset(node->ip,0,sizeof(node->ip)); 480 node->port = 0; 481 node->fail_reports = listCreate(); 482 node->voted_time = 0; 483 node->repl_offset_time = 0; 484 node->repl_offset = 0; 485 listSetFreeMethod(node->fail_reports,zfree); 486 return node; 487 } 488 489 /* This function is called every time we get a failure report from a node. 490 * The side effect is to populate the fail_reports list (or to update 491 * the timestamp of an existing report). 492 * 493 * 'failing' is the node that is in failure state according to the 494 * 'sender' node. 495 * 496 * The function returns 0 if it just updates a timestamp of an existing 497 * failure report from the same sender. 1 is returned if a new failure 498 * report is created. */ 499 int clusterNodeAddFailureReport(clusterNode *failing, clusterNode *sender) { 500 list *l = failing->fail_reports; 501 listNode *ln; 502 listIter li; 503 clusterNodeFailReport *fr; 504 505 /* If a failure report from the same sender already exists, just update 506 * the timestamp. */ 507 listRewind(l,&li); 508 while ((ln = listNext(&li)) != NULL) { 509 fr = ln->value; 510 if (fr->node == sender) { 511 fr->time = mstime(); 512 return 0; 513 } 514 } 515 516 /* Otherwise create a new report. */ 517 fr = zmalloc(sizeof(*fr)); 518 fr->node = sender; 519 fr->time = mstime(); 520 listAddNodeTail(l,fr); 521 return 1; 522 } 523 524 /* Remove failure reports that are too old, where too old means reasonably 525 * older than the global node timeout. Note that anyway for a node to be 526 * flagged as FAIL we need to have a local PFAIL state that is at least 527 * older than the global node timeout, so we don't just trust the number 528 * of failure reports from other nodes. */ 529 void clusterNodeCleanupFailureReports(clusterNode *node) { 530 list *l = node->fail_reports; 531 listNode *ln; 532 listIter li; 533 clusterNodeFailReport *fr; 534 mstime_t maxtime = server.cluster_node_timeout * 535 REDIS_CLUSTER_FAIL_REPORT_VALIDITY_MULT; 536 mstime_t now = mstime(); 537 538 listRewind(l,&li); 539 while ((ln = listNext(&li)) != NULL) { 540 fr = ln->value; 541 if (now - fr->time > maxtime) listDelNode(l,ln); 542 } 543 } 544 545 /* Remove the failing report for 'node' if it was previously considered 546 * failing by 'sender'. This function is called when a node informs us via 547 * gossip that a node is OK from its point of view (no FAIL or PFAIL flags). 548 * 549 * Note that this function is called relatively often as it gets called even 550 * when there are no nodes failing, and is O(N), however when the cluster is 551 * fine the failure reports list is empty so the function runs in constant 552 * time. 553 * 554 * The function returns 1 if the failure report was found and removed. 555 * Otherwise 0 is returned. */ 556 int clusterNodeDelFailureReport(clusterNode *node, clusterNode *sender) { 557 list *l = node->fail_reports; 558 listNode *ln; 559 listIter li; 560 clusterNodeFailReport *fr; 561 562 /* Search for a failure report from this sender. */ 563 listRewind(l,&li); 564 while ((ln = listNext(&li)) != NULL) { 565 fr = ln->value; 566 if (fr->node == sender) break; 567 } 568 if (!ln) return 0; /* No failure report from this sender. */ 569 570 /* Remove the failure report. */ 571 listDelNode(l,ln); 572 clusterNodeCleanupFailureReports(node); 573 return 1; 574 } 575 576 /* Return the number of external nodes that believe 'node' is failing, 577 * not including this node, that may have a PFAIL or FAIL state for this 578 * node as well. */ 579 int clusterNodeFailureReportsCount(clusterNode *node) { 580 clusterNodeCleanupFailureReports(node); 581 return listLength(node->fail_reports); 582 } 583 584 int clusterNodeRemoveSlave(clusterNode *master, clusterNode *slave) { 585 int j; 586 587 for (j = 0; j < master->numslaves; j++) { 588 if (master->slaves[j] == slave) { 589 memmove(master->slaves+j,master->slaves+(j+1), 590 (master->numslaves-1)-j); 591 master->numslaves--; 592 return REDIS_OK; 593 } 594 } 595 return REDIS_ERR; 596 } 597 598 int clusterNodeAddSlave(clusterNode *master, clusterNode *slave) { 599 int j; 600 601 /* If it's already a slave, don't add it again. */ 602 for (j = 0; j < master->numslaves; j++) 603 if (master->slaves[j] == slave) return REDIS_ERR; 604 master->slaves = zrealloc(master->slaves, 605 sizeof(clusterNode*)*(master->numslaves+1)); 606 master->slaves[master->numslaves] = slave; 607 master->numslaves++; 608 return REDIS_OK; 609 } 610 611 void clusterNodeResetSlaves(clusterNode *n) { 612 zfree(n->slaves); 613 n->numslaves = 0; 614 n->slaves = NULL; 615 } 616 617 int clusterCountNonFailingSlaves(clusterNode *n) { 618 int j, okslaves = 0; 619 620 for (j = 0; j < n->numslaves; j++) 621 if (!nodeFailed(n->slaves[j])) okslaves++; 622 return okslaves; 623 } 624 625 void freeClusterNode(clusterNode *n) { 626 sds nodename; 627 628 nodename = sdsnewlen(n->name, REDIS_CLUSTER_NAMELEN); 629 redisAssert(dictDelete(server.cluster->nodes,nodename) == DICT_OK); 630 sdsfree(nodename); 631 if (n->slaveof) clusterNodeRemoveSlave(n->slaveof, n); 632 if (n->link) freeClusterLink(n->link); 633 listRelease(n->fail_reports); 634 zfree(n); 635 } 636 637 /* Add a node to the nodes hash table */ 638 int clusterAddNode(clusterNode *node) { 639 int retval; 640 641 retval = dictAdd(server.cluster->nodes, 642 sdsnewlen(node->name,REDIS_CLUSTER_NAMELEN), node); 643 return (retval == DICT_OK) ? REDIS_OK : REDIS_ERR; 644 } 645 646 /* Remove a node from the cluster: 647 * 1) Mark all the nodes handled by it as unassigned. 648 * 2) Remove all the failure reports sent by this node. 649 * 3) Free the node, that will in turn remove it from the hash table 650 * and from the list of slaves of its master, if it is a slave node. 651 */ 652 void clusterDelNode(clusterNode *delnode) { 653 int j; 654 dictIterator *di; 655 dictEntry *de; 656 657 /* 1) Mark slots as unassigned. */ 658 for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) { 659 if (server.cluster->importing_slots_from[j] == delnode) 660 server.cluster->importing_slots_from[j] = NULL; 661 if (server.cluster->migrating_slots_to[j] == delnode) 662 server.cluster->migrating_slots_to[j] = NULL; 663 if (server.cluster->slots[j] == delnode) 664 clusterDelSlot(j); 665 } 666 667 /* 2) Remove failure reports. */ 668 di = dictGetSafeIterator(server.cluster->nodes); 669 while((de = dictNext(di)) != NULL) { 670 clusterNode *node = dictGetVal(de); 671 672 if (node == delnode) continue; 673 clusterNodeDelFailureReport(node,delnode); 674 } 675 dictReleaseIterator(di); 676 677 /* 3) Remove this node from its master's slaves if needed. */ 678 if (nodeIsSlave(delnode) && delnode->slaveof) 679 clusterNodeRemoveSlave(delnode->slaveof,delnode); 680 681 /* 4) Free the node, unlinking it from the cluster. */ 682 freeClusterNode(delnode); 683 } 684 685 /* Node lookup by name */ 686 clusterNode *clusterLookupNode(char *name) { 687 sds s = sdsnewlen(name, REDIS_CLUSTER_NAMELEN); 688 dictEntry *de; 689 690 de = dictFind(server.cluster->nodes,s); 691 sdsfree(s); 692 if (de == NULL) return NULL; 693 return dictGetVal(de); 694 } 695 696 /* This is only used after the handshake. When we connect a given IP/PORT 697 * as a result of CLUSTER MEET we don't have the node name yet, so we 698 * pick a random one, and will fix it when we receive the PONG request using 699 * this function. */ 700 void clusterRenameNode(clusterNode *node, char *newname) { 701 int retval; 702 sds s = sdsnewlen(node->name, REDIS_CLUSTER_NAMELEN); 703 704 redisLog(REDIS_DEBUG,"Renaming node %.40s into %.40s", 705 node->name, newname); 706 retval = dictDelete(server.cluster->nodes, s); 707 sdsfree(s); 708 redisAssert(retval == DICT_OK); 709 memcpy(node->name, newname, REDIS_CLUSTER_NAMELEN); 710 clusterAddNode(node); 711 } 712 713 /* ----------------------------------------------------------------------------- 714 * CLUSTER nodes blacklist 715 * 716 * The nodes blacklist is just a way to ensure that a given node with a given 717 * Node ID is not readded before some time elapsed (this time is specified 718 * in seconds in REDIS_CLUSTER_BLACKLIST_TTL). 719 * 720 * This is useful when we want to remove a node from the cluster completely: 721 * when CLUSTER FORGET is called, it also puts the node into the blacklist so 722 * that even if we receive gossip messages from other nodes that still remember 723 * about the node we want to remove, we don't re-add it before some time. 724 * 725 * Currently the REDIS_CLUSTER_BLACKLIST_TTL is set to 1 minute, this means 726 * that redis-trib has 60 seconds to send CLUSTER FORGET messages to nodes 727 * in the cluster without dealing with the problem of other nodes re-adding 728 * back the node to nodes we already sent the FORGET command to. 729 * 730 * The data structure used is a hash table with an sds string representing 731 * the node ID as key, and the time when it is ok to re-add the node as 732 * value. 733 * -------------------------------------------------------------------------- */ 734 735 #define REDIS_CLUSTER_BLACKLIST_TTL 60 /* 1 minute. */ 736 737 738 /* Before of the addNode() or Exists() operations we always remove expired 739 * entries from the black list. This is an O(N) operation but it is not a 740 * problem since add / exists operations are called very infrequently and 741 * the hash table is supposed to contain very little elements at max. 742 * However without the cleanup during long uptimes and with some automated 743 * node add/removal procedures, entries could accumulate. */ 744 void clusterBlacklistCleanup(void) { 745 dictIterator *di; 746 dictEntry *de; 747 748 di = dictGetSafeIterator(server.cluster->nodes_black_list); 749 while((de = dictNext(di)) != NULL) { 750 int64_t expire = dictGetUnsignedIntegerVal(de); 751 752 if (expire < server.unixtime) 753 dictDelete(server.cluster->nodes_black_list,dictGetKey(de)); 754 } 755 dictReleaseIterator(di); 756 } 757 758 /* Cleanup the blacklist and add a new node ID to the black list. */ 759 void clusterBlacklistAddNode(clusterNode *node) { 760 dictEntry *de; 761 sds id = sdsnewlen(node->name,REDIS_CLUSTER_NAMELEN); 762 763 clusterBlacklistCleanup(); 764 if (dictAdd(server.cluster->nodes_black_list,id,NULL) == DICT_OK) { 765 /* If the key was added, duplicate the sds string representation of 766 * the key for the next lookup. We'll free it at the end. */ 767 id = sdsdup(id); 768 } 769 de = dictFind(server.cluster->nodes_black_list,id); 770 dictSetUnsignedIntegerVal(de,time(NULL)+REDIS_CLUSTER_BLACKLIST_TTL); 771 sdsfree(id); 772 } 773 774 /* Return non-zero if the specified node ID exists in the blacklist. 775 * You don't need to pass an sds string here, any pointer to 40 bytes 776 * will work. */ 777 int clusterBlacklistExists(char *nodeid) { 778 sds id = sdsnewlen(nodeid,REDIS_CLUSTER_NAMELEN); 779 int retval; 780 781 clusterBlacklistCleanup(); 782 retval = dictFind(server.cluster->nodes_black_list,id) != NULL; 783 sdsfree(id); 784 return retval; 785 } 786 787 /* ----------------------------------------------------------------------------- 788 * CLUSTER messages exchange - PING/PONG and gossip 789 * -------------------------------------------------------------------------- */ 790 791 /* This function checks if a given node should be marked as FAIL. 792 * It happens if the following conditions are met: 793 * 794 * 1) We received enough failure reports from other master nodes via gossip. 795 * Enough means that the majority of the masters signaled the node is 796 * down recently. 797 * 2) We believe this node is in PFAIL state. 798 * 799 * If a failure is detected we also inform the whole cluster about this 800 * event trying to force every other node to set the FAIL flag for the node. 801 * 802 * Note that the form of agreement used here is weak, as we collect the majority 803 * of masters state during some time, and even if we force agreement by 804 * propagating the FAIL message, because of partitions we may not reach every 805 * node. However: 806 * 807 * 1) Either we reach the majority and eventually the FAIL state will propagate 808 * to all the cluster. 809 * 2) Or there is no majority so no slave promotion will be authorized and the 810 * FAIL flag will be cleared after some time. 811 */ 812 void markNodeAsFailingIfNeeded(clusterNode *node) { 813 int failures; 814 int needed_quorum = (server.cluster->size / 2) + 1; 815 816 if (!nodeTimedOut(node)) return; /* We can reach it. */ 817 if (nodeFailed(node)) return; /* Already FAILing. */ 818 819 failures = clusterNodeFailureReportsCount(node); 820 /* Also count myself as a voter if I'm a master. */ 821 if (nodeIsMaster(myself)) failures++; 822 if (failures < needed_quorum) return; /* No weak agreement from masters. */ 823 824 redisLog(REDIS_NOTICE, 825 "Marking node %.40s as failing (quorum reached).", node->name); 826 827 /* Mark the node as failing. */ 828 node->flags &= ~REDIS_NODE_PFAIL; 829 node->flags |= REDIS_NODE_FAIL; 830 node->fail_time = mstime(); 831 832 /* Broadcast the failing node name to everybody, forcing all the other 833 * reachable nodes to flag the node as FAIL. */ 834 if (nodeIsMaster(myself)) clusterSendFail(node->name); 835 clusterDoBeforeSleep(CLUSTER_TODO_UPDATE_STATE|CLUSTER_TODO_SAVE_CONFIG); 836 } 837 838 /* This function is called only if a node is marked as FAIL, but we are able 839 * to reach it again. It checks if there are the conditions to undo the FAIL 840 * state. */ 841 void clearNodeFailureIfNeeded(clusterNode *node) { 842 mstime_t now = mstime(); 843 844 redisAssert(nodeFailed(node)); 845 846 /* For slaves we always clear the FAIL flag if we can contact the 847 * node again. */ 848 if (nodeIsSlave(node) || node->numslots == 0) { 849 redisLog(REDIS_NOTICE, 850 "Clear FAIL state for node %.40s: %s is reachable again.", 851 node->name, 852 nodeIsSlave(node) ? "slave" : "master without slots"); 853 node->flags &= ~REDIS_NODE_FAIL; 854 clusterDoBeforeSleep(CLUSTER_TODO_UPDATE_STATE|CLUSTER_TODO_SAVE_CONFIG); 855 } 856 857 /* If it is a master and... 858 * 1) The FAIL state is old enough. 859 * 2) It is yet serving slots from our point of view (not failed over). 860 * Apparently no one is going to fix these slots, clear the FAIL flag. */ 861 if (nodeIsMaster(node) && node->numslots > 0 && 862 (now - node->fail_time) > 863 (server.cluster_node_timeout * REDIS_CLUSTER_FAIL_UNDO_TIME_MULT)) 864 { 865 redisLog(REDIS_NOTICE, 866 "Clear FAIL state for node %.40s: is reachable again and nobody is serving its slots after some time.", 867 node->name); 868 node->flags &= ~REDIS_NODE_FAIL; 869 clusterDoBeforeSleep(CLUSTER_TODO_UPDATE_STATE|CLUSTER_TODO_SAVE_CONFIG); 870 } 871 } 872 873 /* Return true if we already have a node in HANDSHAKE state matching the 874 * specified ip address and port number. This function is used in order to 875 * avoid adding a new handshake node for the same address multiple times. */ 876 int clusterHandshakeInProgress(char *ip, int port) { 877 dictIterator *di; 878 dictEntry *de; 879 880 di = dictGetSafeIterator(server.cluster->nodes); 881 while((de = dictNext(di)) != NULL) { 882 clusterNode *node = dictGetVal(de); 883 884 if (!nodeInHandshake(node)) continue; 885 if (!strcasecmp(node->ip,ip) && node->port == port) break; 886 } 887 dictReleaseIterator(di); 888 return de != NULL; 889 } 890 891 /* Start an handshake with the specified address if there is not one 892 * already in progress. Returns non-zero if the handshake was actually 893 * started. On error zero is returned and errno is set to one of the 894 * following values: 895 * 896 * EAGAIN - There is already an handshake in progress for this address. 897 * EINVAL - IP or port are not valid. */ 898 int clusterStartHandshake(char *ip, int port) { 899 clusterNode *n; 900 char norm_ip[REDIS_IP_STR_LEN]; 901 struct sockaddr_storage sa; 902 903 /* IP sanity check */ 904 if (inet_pton(AF_INET,ip, 905 &(((struct sockaddr_in *)&sa)->sin_addr))) 906 { 907 sa.ss_family = AF_INET; 908 } else if (inet_pton(AF_INET6,ip, 909 &(((struct sockaddr_in6 *)&sa)->sin6_addr))) 910 { 911 sa.ss_family = AF_INET6; 912 } else { 913 errno = EINVAL; 914 return 0; 915 } 916 917 /* Port sanity check */ 918 if (port <= 0 || port > (65535-REDIS_CLUSTER_PORT_INCR)) { 919 errno = EINVAL; 920 return 0; 921 } 922 923 /* Set norm_ip as the normalized string representation of the node 924 * IP address. */ 925 if (sa.ss_family == AF_INET) 926 inet_ntop(AF_INET, 927 (void*)&(((struct sockaddr_in *)&sa)->sin_addr), 928 norm_ip,REDIS_IP_STR_LEN); 929 else 930 inet_ntop(AF_INET6, 931 (void*)&(((struct sockaddr_in6 *)&sa)->sin6_addr), 932 norm_ip,REDIS_IP_STR_LEN); 933 934 if (clusterHandshakeInProgress(norm_ip,port)) { 935 errno = EAGAIN; 936 return 0; 937 } 938 939 /* Add the node with a random address (NULL as first argument to 940 * createClusterNode()). Everything will be fixed during the 941 * handskake. */ 942 n = createClusterNode(NULL,REDIS_NODE_HANDSHAKE|REDIS_NODE_MEET); 943 memcpy(n->ip,norm_ip,sizeof(n->ip)); 944 n->port = port; 945 clusterAddNode(n); 946 return 1; 947 } 948 949 /* Process the gossip section of PING or PONG packets. 950 * Note that this function assumes that the packet is already sanity-checked 951 * by the caller, not in the content of the gossip section, but in the 952 * length. */ 953 void clusterProcessGossipSection(clusterMsg *hdr, clusterLink *link) { 954 uint16_t count = ntohs(hdr->count); 955 clusterMsgDataGossip *g = (clusterMsgDataGossip*) hdr->data.ping.gossip; 956 clusterNode *sender = link->node ? link->node : clusterLookupNode(hdr->sender); 957 958 while(count--) { 959 sds ci = sdsempty(); 960 uint16_t flags = ntohs(g->flags); 961 clusterNode *node; 962 963 if (flags == 0) ci = sdscat(ci,"noflags,"); 964 if (flags & REDIS_NODE_MYSELF) ci = sdscat(ci,"myself,"); 965 if (flags & REDIS_NODE_MASTER) ci = sdscat(ci,"master,"); 966 if (flags & REDIS_NODE_SLAVE) ci = sdscat(ci,"slave,"); 967 if (flags & REDIS_NODE_PFAIL) ci = sdscat(ci,"fail?,"); 968 if (flags & REDIS_NODE_FAIL) ci = sdscat(ci,"fail,"); 969 if (flags & REDIS_NODE_HANDSHAKE) ci = sdscat(ci,"handshake,"); 970 if (flags & REDIS_NODE_NOADDR) ci = sdscat(ci,"noaddr,"); 971 if (ci[sdslen(ci)-1] == ',') ci[sdslen(ci)-1] = ' '; 972 973 redisLog(REDIS_DEBUG,"GOSSIP %.40s %s:%d %s", 974 g->nodename, 975 g->ip, 976 ntohs(g->port), 977 ci); 978 sdsfree(ci); 979 980 /* Update our state accordingly to the gossip sections */ 981 node = clusterLookupNode(g->nodename); 982 if (node) { 983 /* We already know this node. 984 Handle failure reports, only when the sender is a master. */ 985 if (sender && nodeIsMaster(sender) && node != myself) { 986 if (flags & (REDIS_NODE_FAIL|REDIS_NODE_PFAIL)) { 987 if (clusterNodeAddFailureReport(node,sender)) { 988 redisLog(REDIS_VERBOSE, 989 "Node %.40s reported node %.40s as not reachable.", 990 sender->name, node->name); 991 } 992 markNodeAsFailingIfNeeded(node); 993 } else { 994 if (clusterNodeDelFailureReport(node,sender)) { 995 redisLog(REDIS_VERBOSE, 996 "Node %.40s reported node %.40s is back online.", 997 sender->name, node->name); 998 } 999 } 1000 } 1001 1002 /* If we already know this node, but it is not reachable, and 1003 * we see a different address in the gossip section, start an 1004 * handshake with the (possibly) new address: this will result 1005 * into a node address update if the handshake will be 1006 * successful. */ 1007 if (node->flags & (REDIS_NODE_FAIL|REDIS_NODE_PFAIL) && 1008 (strcasecmp(node->ip,g->ip) || node->port != ntohs(g->port))) 1009 { 1010 clusterStartHandshake(g->ip,ntohs(g->port)); 1011 } 1012 } else { 1013 /* If it's not in NOADDR state and we don't have it, we 1014 * start a handshake process against this IP/PORT pairs. 1015 * 1016 * Note that we require that the sender of this gossip message 1017 * is a well known node in our cluster, otherwise we risk 1018 * joining another cluster. */ 1019 if (sender && 1020 !(flags & REDIS_NODE_NOADDR) && 1021 !clusterBlacklistExists(g->nodename)) 1022 { 1023 clusterStartHandshake(g->ip,ntohs(g->port)); 1024 } 1025 } 1026 1027 /* Next node */ 1028 g++; 1029 } 1030 } 1031 1032 /* IP -> string conversion. 'buf' is supposed to at least be 46 bytes. */ 1033 void nodeIp2String(char *buf, clusterLink *link) { 1034 anetPeerToString(link->fd, buf, REDIS_IP_STR_LEN, NULL); 1035 } 1036 1037 /* Update the node address to the IP address that can be extracted 1038 * from link->fd, and at the specified port. 1039 * Also disconnect the node link so that we'll connect again to the new 1040 * address. 1041 * 1042 * If the ip/port pair are already correct no operation is performed at 1043 * all. 1044 * 1045 * The function returns 0 if the node address is still the same, 1046 * otherwise 1 is returned. */ 1047 int nodeUpdateAddressIfNeeded(clusterNode *node, clusterLink *link, int port) { 1048 char ip[REDIS_IP_STR_LEN]; 1049 1050 /* We don't proceed if the link is the same as the sender link, as this 1051 * function is designed to see if the node link is consistent with the 1052 * symmetric link that is used to receive PINGs from the node. 1053 * 1054 * As a side effect this function never frees the passed 'link', so 1055 * it is safe to call during packet processing. */ 1056 if (link == node->link) return 0; 1057 1058 nodeIp2String(ip,link); 1059 if (node->port == port && strcmp(ip,node->ip) == 0) return 0; 1060 1061 /* IP / port is different, update it. */ 1062 memcpy(node->ip,ip,sizeof(ip)); 1063 node->port = port; 1064 if (node->link) freeClusterLink(node->link); 1065 redisLog(REDIS_WARNING,"Address updated for node %.40s, now %s:%d", 1066 node->name, node->ip, node->port); 1067 1068 /* Check if this is our master and we have to change the 1069 * replication target as well. */ 1070 if (nodeIsSlave(myself) && myself->slaveof == node) 1071 replicationSetMaster(node->ip, node->port); 1072 return 1; 1073 } 1074 1075 /* Reconfigure the specified node 'n' as a master. This function is called when 1076 * a node that we believed to be a slave is now acting as master in order to 1077 * update the state of the node. */ 1078 void clusterSetNodeAsMaster(clusterNode *n) { 1079 if (nodeIsMaster(n)) return; 1080 1081 if (n->slaveof) clusterNodeRemoveSlave(n->slaveof,n); 1082 n->flags &= ~REDIS_NODE_SLAVE; 1083 n->flags |= REDIS_NODE_MASTER; 1084 n->slaveof = NULL; 1085 1086 /* Update config and state. */ 1087 clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG| 1088 CLUSTER_TODO_UPDATE_STATE); 1089 } 1090 1091 /* This function is called when we receive a master configuration via a 1092 * PING, PONG or UPDATE packet. What we receive is a node, a configEpoch of the 1093 * node, and the set of slots claimed under this configEpoch. 1094 * 1095 * What we do is to rebind the slots with newer configuration compared to our 1096 * local configuration, and if needed, we turn ourself into a replica of the 1097 * node (see the function comments for more info). 1098 * 1099 * The 'sender' is the node for which we received a configuration update. 1100 * Sometimes it is not actaully the "Sender" of the information, like in the case 1101 * we receive the info via an UPDATE packet. */ 1102 void clusterUpdateSlotsConfigWith(clusterNode *sender, uint64_t senderConfigEpoch, unsigned char *slots) { 1103 int j; 1104 clusterNode *curmaster, *newmaster = NULL; 1105 1106 /* Here we set curmaster to this node or the node this node 1107 * replicates to if it's a slave. In the for loop we are 1108 * interested to check if slots are taken away from curmaster. */ 1109 curmaster = nodeIsMaster(myself) ? myself : myself->slaveof; 1110 1111 for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) { 1112 if (bitmapTestBit(slots,j)) { 1113 /* The slot is already bound to the sender of this message. */ 1114 if (server.cluster->slots[j] == sender) continue; 1115 1116 /* The slot is in importing state, it should be modified only 1117 * manually via redis-trib (example: a resharding is in progress 1118 * and the migrating side slot was already closed and is advertising 1119 * a new config. We still want the slot to be closed manually). */ 1120 if (server.cluster->importing_slots_from[j]) continue; 1121 1122 /* We rebind the slot to the new node claiming it if: 1123 * 1) The slot was unassigned or the new node claims it with a 1124 * greater configEpoch. 1125 * 2) We are not currently importing the slot. */ 1126 if (server.cluster->slots[j] == NULL || 1127 server.cluster->slots[j]->configEpoch < senderConfigEpoch) 1128 { 1129 /* Was this slot mine, and still contains keys? Something 1130 * odd happened, put the slot in importing state so that 1131 * redis-trib fix can detect the condition (and no further 1132 * updates will be processed before the slot gets fixed). */ 1133 if (server.cluster->slots[j] == myself && 1134 countKeysInSlot(j) && 1135 sender != myself) 1136 { 1137 redisLog(REDIS_WARNING,"Slot update for a slot I still have keys received. Putting the slot in IMPORTING state. Please run the 'redis-trib fix' command."); 1138 server.cluster->importing_slots_from[j] = sender; 1139 } 1140 1141 if (server.cluster->slots[j] == curmaster) 1142 newmaster = sender; 1143 clusterDelSlot(j); 1144 clusterAddSlot(sender,j); 1145 clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG| 1146 CLUSTER_TODO_UPDATE_STATE| 1147 CLUSTER_TODO_FSYNC_CONFIG); 1148 } 1149 } 1150 } 1151 1152 /* If at least one slot was reassigned from a node to another node 1153 * with a greater configEpoch, it is possible that: 1154 * 1) We are a master left without slots. This means that we were 1155 * failed over and we should turn into a replica of the new 1156 * master. 1157 * 2) We are a slave and our master is left without slots. We need 1158 * to replicate to the new slots owner. */ 1159 if (newmaster && curmaster->numslots == 0) { 1160 redisLog(REDIS_WARNING,"Configuration change detected. Reconfiguring myself as a replica of %.40s", sender->name); 1161 clusterSetMaster(sender); 1162 clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG| 1163 CLUSTER_TODO_UPDATE_STATE| 1164 CLUSTER_TODO_FSYNC_CONFIG); 1165 } 1166 } 1167 1168 /* When this function is called, there is a packet to process starting 1169 * at node->rcvbuf. Releasing the buffer is up to the caller, so this 1170 * function should just handle the higher level stuff of processing the 1171 * packet, modifying the cluster state if needed. 1172 * 1173 * The function returns 1 if the link is still valid after the packet 1174 * was processed, otherwise 0 if the link was freed since the packet 1175 * processing lead to some inconsistency error (for instance a PONG 1176 * received from the wrong sender ID). */ 1177 int clusterProcessPacket(clusterLink *link) { 1178 clusterMsg *hdr = (clusterMsg*) link->rcvbuf; 1179 uint32_t totlen = ntohl(hdr->totlen); 1180 uint16_t type = ntohs(hdr->type); 1181 uint16_t flags = ntohs(hdr->flags); 1182 uint64_t senderCurrentEpoch = 0, senderConfigEpoch = 0; 1183 clusterNode *sender; 1184 1185 server.cluster->stats_bus_messages_received++; 1186 redisLog(REDIS_DEBUG,"--- Processing packet of type %d, %lu bytes", 1187 type, (unsigned long) totlen); 1188 1189 /* Perform sanity checks */ 1190 if (totlen < 16) return 1; /* At least signature, version, totlen, count. */ 1191 if (ntohs(hdr->ver) != 0) return 1; /* Can't handle versions other than 0. */ 1192 if (totlen > sdslen(link->rcvbuf)) return 1; 1193 if (type == CLUSTERMSG_TYPE_PING || type == CLUSTERMSG_TYPE_PONG || 1194 type == CLUSTERMSG_TYPE_MEET) 1195 { 1196 uint16_t count = ntohs(hdr->count); 1197 uint32_t explen; /* expected length of this packet */ 1198 1199 explen = sizeof(clusterMsg)-sizeof(union clusterMsgData); 1200 explen += (sizeof(clusterMsgDataGossip)*count); 1201 if (totlen != explen) return 1; 1202 } else if (type == CLUSTERMSG_TYPE_FAIL) { 1203 uint32_t explen = sizeof(clusterMsg)-sizeof(union clusterMsgData); 1204 1205 explen += sizeof(clusterMsgDataFail); 1206 if (totlen != explen) return 1; 1207 } else if (type == CLUSTERMSG_TYPE_PUBLISH) { 1208 uint32_t explen = sizeof(clusterMsg)-sizeof(union clusterMsgData); 1209 1210 explen += sizeof(clusterMsgDataPublish) + 1211 ntohl(hdr->data.publish.msg.channel_len) + 1212 ntohl(hdr->data.publish.msg.message_len); 1213 if (totlen != explen) return 1; 1214 } else if (type == CLUSTERMSG_TYPE_FAILOVER_AUTH_REQUEST || 1215 type == CLUSTERMSG_TYPE_FAILOVER_AUTH_ACK || 1216 type == CLUSTERMSG_TYPE_MFSTART) 1217 { 1218 uint32_t explen = sizeof(clusterMsg)-sizeof(union clusterMsgData); 1219 1220 if (totlen != explen) return 1; 1221 } else if (type == CLUSTERMSG_TYPE_UPDATE) { 1222 uint32_t explen = sizeof(clusterMsg)-sizeof(union clusterMsgData); 1223 1224 explen += sizeof(clusterMsgDataUpdate); 1225 if (totlen != explen) return 1; 1226 } 1227 1228 /* Check if the sender is a known node. */ 1229 sender = clusterLookupNode(hdr->sender); 1230 if (sender && !nodeInHandshake(sender)) { 1231 /* Update our curretEpoch if we see a newer epoch in the cluster. */ 1232 senderCurrentEpoch = ntohu64(hdr->currentEpoch); 1233 senderConfigEpoch = ntohu64(hdr->configEpoch); 1234 if (senderCurrentEpoch > server.cluster->currentEpoch) 1235 server.cluster->currentEpoch = senderCurrentEpoch; 1236 /* Update the sender configEpoch if it is publishing a newer one. */ 1237 if (senderConfigEpoch > sender->configEpoch) { 1238 sender->configEpoch = senderConfigEpoch; 1239 clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG|CLUSTER_TODO_FSYNC_CONFIG); 1240 } 1241 /* Update the replication offset info for this node. */ 1242 sender->repl_offset = ntohu64(hdr->offset); 1243 sender->repl_offset_time = mstime(); 1244 /* If we are a slave performing a manual failover and our master 1245 * sent its offset while already paused, populate the MF state. */ 1246 if (server.cluster->mf_end && 1247 nodeIsSlave(myself) && 1248 myself->slaveof == sender && 1249 hdr->mflags[0] & CLUSTERMSG_FLAG0_PAUSED && 1250 server.cluster->mf_master_offset == 0) 1251 { 1252 server.cluster->mf_master_offset = sender->repl_offset; 1253 redisLog(REDIS_WARNING,"Received replication offset for paused master manual failover: %lld", server.cluster->mf_master_offset); 1254 } 1255 } 1256 1257 /* Process packets by type. */ 1258 if (type == CLUSTERMSG_TYPE_PING || type == CLUSTERMSG_TYPE_MEET) { 1259 redisLog(REDIS_DEBUG,"Ping packet received: %p", (void*)link->node); 1260 1261 /* Add this node if it is new for us and the msg type is MEET. 1262 * In this stage we don't try to add the node with the right 1263 * flags, slaveof pointer, and so forth, as this details will be 1264 * resolved when we'll receive PONGs from the node. */ 1265 if (!sender && type == CLUSTERMSG_TYPE_MEET) { 1266 clusterNode *node; 1267 1268 node = createClusterNode(NULL,REDIS_NODE_HANDSHAKE); 1269 nodeIp2String(node->ip,link); 1270 node->port = ntohs(hdr->port); 1271 clusterAddNode(node); 1272 clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG); 1273 } 1274 1275 /* Get info from the gossip section */ 1276 clusterProcessGossipSection(hdr,link); 1277 1278 /* Anyway reply with a PONG */ 1279 clusterSendPing(link,CLUSTERMSG_TYPE_PONG); 1280 } 1281 1282 /* PING or PONG: process config information. */ 1283 if (type == CLUSTERMSG_TYPE_PING || type == CLUSTERMSG_TYPE_PONG || 1284 type == CLUSTERMSG_TYPE_MEET) 1285 { 1286 redisLog(REDIS_DEBUG,"%s packet received: %p", 1287 type == CLUSTERMSG_TYPE_PING ? "ping" : "pong", 1288 (void*)link->node); 1289 if (link->node) { 1290 if (nodeInHandshake(link->node)) { 1291 /* If we already have this node, try to change the 1292 * IP/port of the node with the new one. */ 1293 if (sender) { 1294 redisLog(REDIS_WARNING, 1295 "Handshake error: we already know node %.40s, updating the address if needed.", sender->name); 1296 if (nodeUpdateAddressIfNeeded(sender,link,ntohs(hdr->port))) 1297 { 1298 clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG| 1299 CLUSTER_TODO_UPDATE_STATE); 1300 } 1301 /* Free this node as we alrady have it. This will 1302 * cause the link to be freed as well. */ 1303 freeClusterNode(link->node); 1304 return 0; 1305 } 1306 1307 /* First thing to do is replacing the random name with the 1308 * right node name if this was a handshake stage. */ 1309 clusterRenameNode(link->node, hdr->sender); 1310 redisLog(REDIS_DEBUG,"Handshake with node %.40s completed.", 1311 link->node->name); 1312 link->node->flags &= ~REDIS_NODE_HANDSHAKE; 1313 link->node->flags |= flags&(REDIS_NODE_MASTER|REDIS_NODE_SLAVE); 1314 clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG); 1315 } else if (memcmp(link->node->name,hdr->sender, 1316 REDIS_CLUSTER_NAMELEN) != 0) 1317 { 1318 /* If the reply has a non matching node ID we 1319 * disconnect this node and set it as not having an associated 1320 * address. */ 1321 redisLog(REDIS_DEBUG,"PONG contains mismatching sender ID"); 1322 link->node->flags |= REDIS_NODE_NOADDR; 1323 link->node->ip[0] = '\0'; 1324 link->node->port = 0; 1325 freeClusterLink(link); 1326 clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG); 1327 return 0; 1328 } 1329 } 1330 1331 /* Update the node address if it changed. */ 1332 if (sender && type == CLUSTERMSG_TYPE_PING && 1333 !nodeInHandshake(sender) && 1334 nodeUpdateAddressIfNeeded(sender,link,ntohs(hdr->port))) 1335 { 1336 clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG|CLUSTER_TODO_UPDATE_STATE); 1337 } 1338 1339 /* Update our info about the node */ 1340 if (link->node && type == CLUSTERMSG_TYPE_PONG) { 1341 link->node->pong_received = mstime(); 1342 link->node->ping_sent = 0; 1343 1344 /* The PFAIL condition can be reversed without external 1345 * help if it is momentary (that is, if it does not 1346 * turn into a FAIL state). 1347 * 1348 * The FAIL condition is also reversible under specific 1349 * conditions detected by clearNodeFailureIfNeeded(). */ 1350 if (nodeTimedOut(link->node)) { 1351 link->node->flags &= ~REDIS_NODE_PFAIL; 1352 clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG| 1353 CLUSTER_TODO_UPDATE_STATE); 1354 } else if (nodeFailed(link->node)) { 1355 clearNodeFailureIfNeeded(link->node); 1356 } 1357 } 1358 1359 /* Check for role switch: slave -> master or master -> slave. */ 1360 if (sender) { 1361 if (!memcmp(hdr->slaveof,REDIS_NODE_NULL_NAME, 1362 sizeof(hdr->slaveof))) 1363 { 1364 /* Node is a master. */ 1365 clusterSetNodeAsMaster(sender); 1366 } else { 1367 /* Node is a slave. */ 1368 clusterNode *master = clusterLookupNode(hdr->slaveof); 1369 1370 if (nodeIsMaster(sender)) { 1371 /* Master turned into a slave! Reconfigure the node. */ 1372 clusterDelNodeSlots(sender); 1373 sender->flags &= ~REDIS_NODE_MASTER; 1374 sender->flags |= REDIS_NODE_SLAVE; 1375 1376 /* Remove the list of slaves from the node. */ 1377 if (sender->numslaves) clusterNodeResetSlaves(sender); 1378 1379 /* Update config and state. */ 1380 clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG| 1381 CLUSTER_TODO_UPDATE_STATE); 1382 } 1383 1384 /* Master node changed for this slave? */ 1385 if (master && sender->slaveof != master) { 1386 if (sender->slaveof) 1387 clusterNodeRemoveSlave(sender->slaveof,sender); 1388 clusterNodeAddSlave(master,sender); 1389 sender->slaveof = master; 1390 1391 /* Update config. */ 1392 clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG); 1393 } 1394 } 1395 } 1396 1397 /* Update our info about served slots. 1398 * 1399 * Note: this MUST happen after we update the master/slave state 1400 * so that REDIS_NODE_MASTER flag will be set. */ 1401 1402 /* Many checks are only needed if the set of served slots this 1403 * instance claims is different compared to the set of slots we have 1404 * for it. Check this ASAP to avoid other computational expansive 1405 * checks later. */ 1406 clusterNode *sender_master = NULL; /* Sender or its master if slave. */ 1407 int dirty_slots = 0; /* Sender claimed slots don't match my view? */ 1408 1409 if (sender) { 1410 sender_master = nodeIsMaster(sender) ? sender : sender->slaveof; 1411 if (sender_master) { 1412 dirty_slots = memcmp(sender_master->slots, 1413 hdr->myslots,sizeof(hdr->myslots)) != 0; 1414 } 1415 } 1416 1417 /* 1) If the sender of the message is a master, and we detected that 1418 * the set of slots it claims changed, scan the slots to see if we 1419 * need to update our configuration. */ 1420 if (sender && nodeIsMaster(sender) && dirty_slots) 1421 clusterUpdateSlotsConfigWith(sender,senderConfigEpoch,hdr->myslots); 1422 1423 /* 2) We also check for the reverse condition, that is, the sender 1424 * claims to serve slots we know are served by a master with a 1425 * greater configEpoch. If this happens we inform the sender. 1426 * 1427 * This is useful because sometimes after a partition heals, a 1428 * reappearing master may be the last one to claim a given set of 1429 * hash slots, but with a configuration that other instances know to 1430 * be deprecated. Example: 1431 * 1432 * A and B are master and slave for slots 1,2,3. 1433 * A is partitioned away, B gets promoted. 1434 * B is partitioned away, and A returns available. 1435 * 1436 * Usually B would PING A publishing its set of served slots and its 1437 * configEpoch, but because of the partition B can't inform A of the 1438 * new configuration, so other nodes that have an updated table must 1439 * do it. In this way A will stop to act as a master (or can try to 1440 * failover if there are the conditions to win the election). */ 1441 if (sender && dirty_slots) { 1442 int j; 1443 1444 for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) { 1445 if (bitmapTestBit(hdr->myslots,j)) { 1446 if (server.cluster->slots[j] == sender || 1447 server.cluster->slots[j] == NULL) continue; 1448 if (server.cluster->slots[j]->configEpoch > 1449 senderConfigEpoch) 1450 { 1451 redisLog(REDIS_VERBOSE, 1452 "Node %.40s has old slots configuration, sending " 1453 "an UPDATE message about %.40s", 1454 sender->name, server.cluster->slots[j]->name); 1455 clusterSendUpdate(sender->link,server.cluster->slots[j]); 1456 1457 /* TODO: instead of exiting the loop send every other 1458 * UPDATE packet for other nodes that are the new owner 1459 * of sender's slots. */ 1460 break; 1461 } 1462 } 1463 } 1464 } 1465 1466 /* Get info from the gossip section */ 1467 clusterProcessGossipSection(hdr,link); 1468 } else if (type == CLUSTERMSG_TYPE_FAIL) { 1469 clusterNode *failing; 1470 1471 if (sender) { 1472 failing = clusterLookupNode(hdr->data.fail.about.nodename); 1473 if (failing && 1474 !(failing->flags & (REDIS_NODE_FAIL|REDIS_NODE_MYSELF))) 1475 { 1476 redisLog(REDIS_NOTICE, 1477 "FAIL message received from %.40s about %.40s", 1478 hdr->sender, hdr->data.fail.about.nodename); 1479 failing->flags |= REDIS_NODE_FAIL; 1480 failing->fail_time = mstime(); 1481 failing->flags &= ~REDIS_NODE_PFAIL; 1482 clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG|CLUSTER_TODO_UPDATE_STATE); 1483 } 1484 } else { 1485 redisLog(REDIS_NOTICE, 1486 "Ignoring FAIL message from unknonw node %.40s about %.40s", 1487 hdr->sender, hdr->data.fail.about.nodename); 1488 } 1489 } else if (type == CLUSTERMSG_TYPE_PUBLISH) { 1490 robj *channel, *message; 1491 uint32_t channel_len, message_len; 1492 1493 /* Don't bother creating useless objects if there are no 1494 * Pub/Sub subscribers. */ 1495 if (dictSize(server.pubsub_channels) || 1496 listLength(server.pubsub_patterns)) 1497 { 1498 channel_len = ntohl(hdr->data.publish.msg.channel_len); 1499 message_len = ntohl(hdr->data.publish.msg.message_len); 1500 channel = createStringObject( 1501 (char*)hdr->data.publish.msg.bulk_data,channel_len); 1502 message = createStringObject( 1503 (char*)hdr->data.publish.msg.bulk_data+channel_len, 1504 message_len); 1505 pubsubPublishMessage(channel,message); 1506 decrRefCount(channel); 1507 decrRefCount(message); 1508 } 1509 } else if (type == CLUSTERMSG_TYPE_FAILOVER_AUTH_REQUEST) { 1510 if (!sender) return 1; /* We don't know that node. */ 1511 clusterSendFailoverAuthIfNeeded(sender,hdr); 1512 } else if (type == CLUSTERMSG_TYPE_FAILOVER_AUTH_ACK) { 1513 if (!sender) return 1; /* We don't know that node. */ 1514 /* We consider this vote only if the sender is a master serving 1515 * a non zero number of slots, and its currentEpoch is greater or 1516 * equal to epoch where this node started the election. */ 1517 if (nodeIsMaster(sender) && sender->numslots > 0 && 1518 senderCurrentEpoch >= server.cluster->failover_auth_epoch) 1519 { 1520 server.cluster->failover_auth_count++; 1521 /* Maybe we reached a quorum here, set a flag to make sure 1522 * we check ASAP. */ 1523 clusterDoBeforeSleep(CLUSTER_TODO_HANDLE_FAILOVER); 1524 } 1525 } else if (type == CLUSTERMSG_TYPE_MFSTART) { 1526 /* This message is acceptable only if I'm a master and the sender 1527 * is one of my slaves. */ 1528 if (!sender || sender->slaveof != myself) return 1; 1529 /* Manual failover requested from slaves. Initialize the state 1530 * accordingly. */ 1531 resetManualFailover(); 1532 server.cluster->mf_end = mstime() + REDIS_CLUSTER_MF_TIMEOUT; 1533 server.cluster->mf_slave = sender; 1534 pauseClients(mstime()+(REDIS_CLUSTER_MF_TIMEOUT*2)); 1535 redisLog(REDIS_WARNING,"Manual failover requested by slave %.40s.", 1536 sender->name); 1537 } else if (type == CLUSTERMSG_TYPE_UPDATE) { 1538 clusterNode *n; /* The node the update is about. */ 1539 uint64_t reportedConfigEpoch = ntohu64(hdr->data.update.nodecfg.configEpoch); 1540 1541 if (!sender) return 1; /* We don't know the sender. */ 1542 n = clusterLookupNode(hdr->data.update.nodecfg.nodename); 1543 if (!n) return 1; /* We don't know the reported node. */ 1544 if (n->configEpoch >= reportedConfigEpoch) return 1; /* Nothing new. */ 1545 1546 /* If in our current config the node is a slave, set it as a master. */ 1547 if (nodeIsSlave(n)) clusterSetNodeAsMaster(n); 1548 1549 /* Update the node's configEpoch. */ 1550 n->configEpoch = reportedConfigEpoch; 1551 clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG|CLUSTER_TODO_FSYNC_CONFIG); 1552 1553 /* Check the bitmap of served slots and udpate our 1554 * config accordingly. */ 1555 clusterUpdateSlotsConfigWith(n,reportedConfigEpoch, 1556 hdr->data.update.nodecfg.slots); 1557 } else { 1558 redisLog(REDIS_WARNING,"Received unknown packet type: %d", type); 1559 } 1560 return 1; 1561 } 1562 1563 /* This function is called when we detect the link with this node is lost. 1564 We set the node as no longer connected. The Cluster Cron will detect 1565 this connection and will try to get it connected again. 1566 1567 Instead if the node is a temporary node used to accept a query, we 1568 completely free the node on error. */ 1569 void handleLinkIOError(clusterLink *link) { 1570 freeClusterLink(link); 1571 } 1572 1573 /* Send data. This is handled using a trivial send buffer that gets 1574 * consumed by write(). We don't try to optimize this for speed too much 1575 * as this is a very low traffic channel. */ 1576 void clusterWriteHandler(aeEventLoop *el, int fd, void *privdata, int mask) { 1577 clusterLink *link = (clusterLink*) privdata; 1578 ssize_t nwritten; 1579 REDIS_NOTUSED(el); 1580 REDIS_NOTUSED(mask); 1581 1582 nwritten = write(fd, link->sndbuf, sdslen(link->sndbuf)); 1583 if (nwritten <= 0) { 1584 redisLog(REDIS_DEBUG,"I/O error writing to node link: %s", 1585 strerror(errno)); 1586 handleLinkIOError(link); 1587 return; 1588 } 1589 sdsrange(link->sndbuf,nwritten,-1); 1590 if (sdslen(link->sndbuf) == 0) 1591 aeDeleteFileEvent(server.el, link->fd, AE_WRITABLE); 1592 } 1593 1594 /* Read data. Try to read the first field of the header first to check the 1595 * full length of the packet. When a whole packet is in memory this function 1596 * will call the function to process the packet. And so forth. */ 1597 void clusterReadHandler(aeEventLoop *el, int fd, void *privdata, int mask) { 1598 char buf[sizeof(clusterMsg)]; 1599 ssize_t nread; 1600 clusterMsg *hdr; 1601 clusterLink *link = (clusterLink*) privdata; 1602 int readlen, rcvbuflen; 1603 REDIS_NOTUSED(el); 1604 REDIS_NOTUSED(mask); 1605 1606 while(1) { /* Read as long as there is data to read. */ 1607 rcvbuflen = sdslen(link->rcvbuf); 1608 if (rcvbuflen < 8) { 1609 /* First, obtain the first 8 bytes to get the full message 1610 * length. */ 1611 readlen = 8 - rcvbuflen; 1612 } else { 1613 /* Finally read the full message. */ 1614 hdr = (clusterMsg*) link->rcvbuf; 1615 if (rcvbuflen == 8) { 1616 /* Perform some sanity check on the message signature 1617 * and length. */ 1618 if (memcmp(hdr->sig,"RCmb",4) != 0 || 1619 ntohl(hdr->totlen) < CLUSTERMSG_MIN_LEN) 1620 { 1621 redisLog(REDIS_WARNING, 1622 "Bad message length or signature received " 1623 "from Cluster bus."); 1624 handleLinkIOError(link); 1625 return; 1626 } 1627 } 1628 readlen = ntohl(hdr->totlen) - rcvbuflen; 1629 if (readlen > sizeof(buf)) readlen = sizeof(buf); 1630 } 1631 1632 nread = read(fd,buf,readlen); 1633 if (nread == -1 && errno == EAGAIN) return; /* No more data ready. */ 1634 1635 if (nread <= 0) { 1636 /* I/O error... */ 1637 redisLog(REDIS_DEBUG,"I/O error reading from node link: %s", 1638 (nread == 0) ? "connection closed" : strerror(errno)); 1639 handleLinkIOError(link); 1640 return; 1641 } else { 1642 /* Read data and recast the pointer to the new buffer. */ 1643 link->rcvbuf = sdscatlen(link->rcvbuf,buf,nread); 1644 hdr = (clusterMsg*) link->rcvbuf; 1645 rcvbuflen += nread; 1646 } 1647 1648 /* Total length obtained? Process this packet. */ 1649 if (rcvbuflen >= 8 && rcvbuflen == ntohl(hdr->totlen)) { 1650 if (clusterProcessPacket(link)) { 1651 sdsfree(link->rcvbuf); 1652 link->rcvbuf = sdsempty(); 1653 } else { 1654 return; /* Link no longer valid. */ 1655 } 1656 } 1657 } 1658 } 1659 1660 /* Put stuff into the send buffer. 1661 * 1662 * It is guaranteed that this function will never have as a side effect 1663 * the link to be invalidated, so it is safe to call this function 1664 * from event handlers that will do stuff with the same link later. */ 1665 void clusterSendMessage(clusterLink *link, unsigned char *msg, size_t msglen) { 1666 if (sdslen(link->sndbuf) == 0 && msglen != 0) 1667 aeCreateFileEvent(server.el,link->fd,AE_WRITABLE, 1668 clusterWriteHandler,link); 1669 1670 link->sndbuf = sdscatlen(link->sndbuf, msg, msglen); 1671 server.cluster->stats_bus_messages_sent++; 1672 } 1673 1674 /* Send a message to all the nodes that are part of the cluster having 1675 * a connected link. 1676 * 1677 * It is guaranteed that this function will never have as a side effect 1678 * some node->link to be invalidated, so it is safe to call this function 1679 * from event handlers that will do stuff with node links later. */ 1680 void clusterBroadcastMessage(void *buf, size_t len) { 1681 dictIterator *di; 1682 dictEntry *de; 1683 1684 di = dictGetSafeIterator(server.cluster->nodes); 1685 while((de = dictNext(di)) != NULL) { 1686 clusterNode *node = dictGetVal(de); 1687 1688 if (!node->link) continue; 1689 if (node->flags & (REDIS_NODE_MYSELF|REDIS_NODE_HANDSHAKE)) 1690 continue; 1691 clusterSendMessage(node->link,buf,len); 1692 } 1693 dictReleaseIterator(di); 1694 } 1695 1696 /* Build the message header */ 1697 void clusterBuildMessageHdr(clusterMsg *hdr, int type) { 1698 int totlen = 0; 1699 uint64_t offset; 1700 clusterNode *master; 1701 1702 /* If this node is a master, we send its slots bitmap and configEpoch. 1703 * If this node is a slave we send the master's information instead (the 1704 * node is flagged as slave so the receiver knows that it is NOT really 1705 * in charge for this slots. */ 1706 master = (nodeIsSlave(myself) && myself->slaveof) ? 1707 myself->slaveof : myself; 1708 1709 memset(hdr,0,sizeof(*hdr)); 1710 hdr->sig[0] = 'R'; 1711 hdr->sig[1] = 'C'; 1712 hdr->sig[2] = 'm'; 1713 hdr->sig[3] = 'b'; 1714 hdr->type = htons(type); 1715 memcpy(hdr->sender,myself->name,REDIS_CLUSTER_NAMELEN); 1716 1717 memcpy(hdr->myslots,master->slots,sizeof(hdr->myslots)); 1718 memset(hdr->slaveof,0,REDIS_CLUSTER_NAMELEN); 1719 if (myself->slaveof != NULL) 1720 memcpy(hdr->slaveof,myself->slaveof->name, REDIS_CLUSTER_NAMELEN); 1721 hdr->port = htons(server.port); 1722 hdr->flags = htons(myself->flags); 1723 hdr->state = server.cluster->state; 1724 1725 /* Set the currentEpoch and configEpochs. */ 1726 hdr->currentEpoch = htonu64(server.cluster->currentEpoch); 1727 hdr->configEpoch = htonu64(master->configEpoch); 1728 1729 /* Set the replication offset. */ 1730 if (nodeIsSlave(myself)) 1731 offset = replicationGetSlaveOffset(); 1732 else 1733 offset = server.master_repl_offset; 1734 hdr->offset = htonu64(offset); 1735 1736 /* Set the message flags. */ 1737 if (nodeIsMaster(myself) && server.cluster->mf_end) 1738 hdr->mflags[0] |= CLUSTERMSG_FLAG0_PAUSED; 1739 1740 /* Compute the message length for certain messages. For other messages 1741 * this is up to the caller. */ 1742 if (type == CLUSTERMSG_TYPE_FAIL) { 1743 totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData); 1744 totlen += sizeof(clusterMsgDataFail); 1745 } else if (type == CLUSTERMSG_TYPE_UPDATE) { 1746 totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData); 1747 totlen += sizeof(clusterMsgDataUpdate); 1748 } 1749 hdr->totlen = htonl(totlen); 1750 /* For PING, PONG, and MEET, fixing the totlen field is up to the caller. */ 1751 } 1752 1753 /* Send a PING or PONG packet to the specified node, making sure to add enough 1754 * gossip informations. */ 1755 void clusterSendPing(clusterLink *link, int type) { 1756 unsigned char buf[sizeof(clusterMsg)]; 1757 clusterMsg *hdr = (clusterMsg*) buf; 1758 int gossipcount = 0, totlen; 1759 /* freshnodes is the number of nodes we can still use to populate the 1760 * gossip section of the ping packet. Basically we start with the nodes 1761 * we have in memory minus two (ourself and the node we are sending the 1762 * message to). Every time we add a node we decrement the counter, so when 1763 * it will drop to <= zero we know there is no more gossip info we can 1764 * send. */ 1765 int freshnodes = dictSize(server.cluster->nodes)-2; 1766 1767 if (link->node && type == CLUSTERMSG_TYPE_PING) 1768 link->node->ping_sent = mstime(); 1769 clusterBuildMessageHdr(hdr,type); 1770 1771 /* Populate the gossip fields */ 1772 while(freshnodes > 0 && gossipcount < 3) { 1773 dictEntry *de = dictGetRandomKey(server.cluster->nodes); 1774 clusterNode *this = dictGetVal(de); 1775 clusterMsgDataGossip *gossip; 1776 int j; 1777 1778 /* In the gossip section don't include: 1779 * 1) Myself. 1780 * 2) Nodes in HANDSHAKE state. 1781 * 3) Nodes with the NOADDR flag set. 1782 * 4) Disconnected nodes if they don't have configured slots. 1783 */ 1784 if (this == myself || 1785 this->flags & (REDIS_NODE_HANDSHAKE|REDIS_NODE_NOADDR) || 1786 (this->link == NULL && this->numslots == 0)) 1787 { 1788 freshnodes--; /* otherwise we may loop forever. */ 1789 continue; 1790 } 1791 1792 /* Check if we already added this node */ 1793 for (j = 0; j < gossipcount; j++) { 1794 if (memcmp(hdr->data.ping.gossip[j].nodename,this->name, 1795 REDIS_CLUSTER_NAMELEN) == 0) break; 1796 } 1797 if (j != gossipcount) continue; 1798 1799 /* Add it */ 1800 freshnodes--; 1801 gossip = &(hdr->data.ping.gossip[gossipcount]); 1802 memcpy(gossip->nodename,this->name,REDIS_CLUSTER_NAMELEN); 1803 gossip->ping_sent = htonl(this->ping_sent); 1804 gossip->pong_received = htonl(this->pong_received); 1805 memcpy(gossip->ip,this->ip,sizeof(this->ip)); 1806 gossip->port = htons(this->port); 1807 gossip->flags = htons(this->flags); 1808 gossipcount++; 1809 } 1810 totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData); 1811 totlen += (sizeof(clusterMsgDataGossip)*gossipcount); 1812 hdr->count = htons(gossipcount); 1813 hdr->totlen = htonl(totlen); 1814 clusterSendMessage(link,buf,totlen); 1815 } 1816 1817 /* Send a PONG packet to every connected node that's not in handshake state 1818 * and for which we have a valid link. 1819 * 1820 * In Redis Cluster pongs are not used just for failure detection, but also 1821 * to carry important configuration information. So broadcasting a pong is 1822 * useful when something changes in the configuration and we want to make 1823 * the cluster aware ASAP (for instance after a slave promotion). 1824 * 1825 * The 'target' argument specifies the receiving instances using the 1826 * defines below: 1827 * 1828 * CLUSTER_BROADCAST_ALL -> All known instances. 1829 * CLUSTER_BROADCAST_LOCAL_SLAVES -> All slaves in my master-slaves ring. 1830 */ 1831 #define CLUSTER_BROADCAST_ALL 0 1832 #define CLUSTER_BROADCAST_LOCAL_SLAVES 1 1833 void clusterBroadcastPong(int target) { 1834 dictIterator *di; 1835 dictEntry *de; 1836 1837 di = dictGetSafeIterator(server.cluster->nodes); 1838 while((de = dictNext(di)) != NULL) { 1839 clusterNode *node = dictGetVal(de); 1840 1841 if (!node->link) continue; 1842 if (node == myself || nodeInHandshake(node)) continue; 1843 if (target == CLUSTER_BROADCAST_LOCAL_SLAVES) { 1844 int local_slave = 1845 nodeIsSlave(node) && node->slaveof && 1846 (node->slaveof == myself || node->slaveof == myself->slaveof); 1847 if (!local_slave) continue; 1848 } 1849 clusterSendPing(node->link,CLUSTERMSG_TYPE_PONG); 1850 } 1851 dictReleaseIterator(di); 1852 } 1853 1854 /* Send a PUBLISH message. 1855 * 1856 * If link is NULL, then the message is broadcasted to the whole cluster. */ 1857 void clusterSendPublish(clusterLink *link, robj *channel, robj *message) { 1858 unsigned char buf[sizeof(clusterMsg)], *payload; 1859 clusterMsg *hdr = (clusterMsg*) buf; 1860 uint32_t totlen; 1861 uint32_t channel_len, message_len; 1862 1863 channel = getDecodedObject(channel); 1864 message = getDecodedObject(message); 1865 channel_len = sdslen(channel->ptr); 1866 message_len = sdslen(message->ptr); 1867 1868 clusterBuildMessageHdr(hdr,CLUSTERMSG_TYPE_PUBLISH); 1869 totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData); 1870 totlen += sizeof(clusterMsgDataPublish) + channel_len + message_len; 1871 1872 hdr->data.publish.msg.channel_len = htonl(channel_len); 1873 hdr->data.publish.msg.message_len = htonl(message_len); 1874 hdr->totlen = htonl(totlen); 1875 1876 /* Try to use the local buffer if possible */ 1877 if (totlen < sizeof(buf)) { 1878 payload = buf; 1879 } else { 1880 payload = zmalloc(totlen); 1881 memcpy(payload,hdr,sizeof(*hdr)); 1882 hdr = (clusterMsg*) payload; 1883 } 1884 memcpy(hdr->data.publish.msg.bulk_data,channel->ptr,sdslen(channel->ptr)); 1885 memcpy(hdr->data.publish.msg.bulk_data+sdslen(channel->ptr), 1886 message->ptr,sdslen(message->ptr)); 1887 1888 if (link) 1889 clusterSendMessage(link,payload,totlen); 1890 else 1891 clusterBroadcastMessage(payload,totlen); 1892 1893 decrRefCount(channel); 1894 decrRefCount(message); 1895 if (payload != buf) zfree(payload); 1896 } 1897 1898 /* Send a FAIL message to all the nodes we are able to contact. 1899 * The FAIL message is sent when we detect that a node is failing 1900 * (REDIS_NODE_PFAIL) and we also receive a gossip confirmation of this: 1901 * we switch the node state to REDIS_NODE_FAIL and ask all the other 1902 * nodes to do the same ASAP. */ 1903 void clusterSendFail(char *nodename) { 1904 unsigned char buf[sizeof(clusterMsg)]; 1905 clusterMsg *hdr = (clusterMsg*) buf; 1906 1907 clusterBuildMessageHdr(hdr,CLUSTERMSG_TYPE_FAIL); 1908 memcpy(hdr->data.fail.about.nodename,nodename,REDIS_CLUSTER_NAMELEN); 1909 clusterBroadcastMessage(buf,ntohl(hdr->totlen)); 1910 } 1911 1912 /* Send an UPDATE message to the specified link carrying the specified 'node' 1913 * slots configuration. The node name, slots bitmap, and configEpoch info 1914 * are included. */ 1915 void clusterSendUpdate(clusterLink *link, clusterNode *node) { 1916 unsigned char buf[sizeof(clusterMsg)]; 1917 clusterMsg *hdr = (clusterMsg*) buf; 1918 1919 if (link == NULL) return; 1920 clusterBuildMessageHdr(hdr,CLUSTERMSG_TYPE_UPDATE); 1921 memcpy(hdr->data.update.nodecfg.nodename,node->name,REDIS_CLUSTER_NAMELEN); 1922 hdr->data.update.nodecfg.configEpoch = htonu64(node->configEpoch); 1923 memcpy(hdr->data.update.nodecfg.slots,node->slots,sizeof(node->slots)); 1924 clusterSendMessage(link,buf,ntohl(hdr->totlen)); 1925 } 1926 1927 /* ----------------------------------------------------------------------------- 1928 * CLUSTER Pub/Sub support 1929 * 1930 * For now we do very little, just propagating PUBLISH messages across the whole 1931 * cluster. In the future we'll try to get smarter and avoiding propagating those 1932 * messages to hosts without receives for a given channel. 1933 * -------------------------------------------------------------------------- */ 1934 void clusterPropagatePublish(robj *channel, robj *message) { 1935 clusterSendPublish(NULL, channel, message); 1936 } 1937 1938 /* ----------------------------------------------------------------------------- 1939 * SLAVE node specific functions 1940 * -------------------------------------------------------------------------- */ 1941 1942 /* This function sends a FAILOVE_AUTH_REQUEST message to every node in order to 1943 * see if there is the quorum for this slave instance to failover its failing 1944 * master. 1945 * 1946 * Note that we send the failover request to everybody, master and slave nodes, 1947 * but only the masters are supposed to reply to our query. */ 1948 void clusterRequestFailoverAuth(void) { 1949 unsigned char buf[sizeof(clusterMsg)]; 1950 clusterMsg *hdr = (clusterMsg*) buf; 1951 uint32_t totlen; 1952 1953 clusterBuildMessageHdr(hdr,CLUSTERMSG_TYPE_FAILOVER_AUTH_REQUEST); 1954 /* If this is a manual failover, set the CLUSTERMSG_FLAG0_FORCEACK bit 1955 * in the header to communicate the nodes receiving the message that 1956 * they should authorized the failover even if the master is working. */ 1957 if (server.cluster->mf_end) hdr->mflags[0] |= CLUSTERMSG_FLAG0_FORCEACK; 1958 totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData); 1959 hdr->totlen = htonl(totlen); 1960 clusterBroadcastMessage(buf,totlen); 1961 } 1962 1963 /* Send a FAILOVER_AUTH_ACK message to the specified node. */ 1964 void clusterSendFailoverAuth(clusterNode *node) { 1965 unsigned char buf[sizeof(clusterMsg)]; 1966 clusterMsg *hdr = (clusterMsg*) buf; 1967 uint32_t totlen; 1968 1969 if (!node->link) return; 1970 clusterBuildMessageHdr(hdr,CLUSTERMSG_TYPE_FAILOVER_AUTH_ACK); 1971 totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData); 1972 hdr->totlen = htonl(totlen); 1973 clusterSendMessage(node->link,buf,totlen); 1974 } 1975 1976 /* Send a MFSTART message to the specified node. */ 1977 void clusterSendMFStart(clusterNode *node) { 1978 unsigned char buf[sizeof(clusterMsg)]; 1979 clusterMsg *hdr = (clusterMsg*) buf; 1980 uint32_t totlen; 1981 1982 if (!node->link) return; 1983 clusterBuildMessageHdr(hdr,CLUSTERMSG_TYPE_MFSTART); 1984 totlen = sizeof(clusterMsg)-sizeof(union clusterMsgData); 1985 hdr->totlen = htonl(totlen); 1986 clusterSendMessage(node->link,buf,totlen); 1987 } 1988 1989 /* Vote for the node asking for our vote if there are the conditions. */ 1990 void clusterSendFailoverAuthIfNeeded(clusterNode *node, clusterMsg *request) { 1991 clusterNode *master = node->slaveof; 1992 uint64_t requestCurrentEpoch = ntohu64(request->currentEpoch); 1993 uint64_t requestConfigEpoch = ntohu64(request->configEpoch); 1994 unsigned char *claimed_slots = request->myslots; 1995 int force_ack = request->mflags[0] & CLUSTERMSG_FLAG0_FORCEACK; 1996 int j; 1997 1998 /* IF we are not a master serving at least 1 slot, we don't have the 1999 * right to vote, as the cluster size in Redis Cluster is the number 2000 * of masters serving at least one slot, and quorum is the cluster 2001 * size + 1 */ 2002 if (nodeIsSlave(myself) || myself->numslots == 0) return; 2003 2004 /* Request epoch must be >= our currentEpoch. */ 2005 if (requestCurrentEpoch < server.cluster->currentEpoch) return; 2006 2007 /* I already voted for this epoch? Return ASAP. */ 2008 if (server.cluster->last_vote_epoch == server.cluster->currentEpoch) return; 2009 2010 /* Node must be a slave and its master down. 2011 * The master can be non failing if the request is flagged 2012 * with CLUSTERMSG_FLAG0_FORCEACK (manual failover). */ 2013 if (nodeIsMaster(node) || master == NULL || 2014 (!nodeFailed(master) && !force_ack)) return; 2015 2016 /* We did not voted for a slave about this master for two 2017 * times the node timeout. This is not strictly needed for correctness 2018 * of the algorithm but makes the base case more linear. */ 2019 if (mstime() - node->slaveof->voted_time < server.cluster_node_timeout * 2) 2020 return; 2021 2022 /* The slave requesting the vote must have a configEpoch for the claimed 2023 * slots that is >= the one of the masters currently serving the same 2024 * slots in the current configuration. */ 2025 for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) { 2026 if (bitmapTestBit(claimed_slots, j) == 0) continue; 2027 if (server.cluster->slots[j] == NULL || 2028 server.cluster->slots[j]->configEpoch <= requestConfigEpoch) continue; 2029 /* If we reached this point we found a slot that in our current slots 2030 * is served by a master with a greater configEpoch than the one claimed 2031 * by the slave requesting our vote. Refuse to vote for this slave. */ 2032 return; 2033 } 2034 2035 /* We can vote for this slave. */ 2036 clusterSendFailoverAuth(node); 2037 server.cluster->last_vote_epoch = server.cluster->currentEpoch; 2038 node->slaveof->voted_time = mstime(); 2039 } 2040 2041 /* This function returns the "rank" of this instance, a slave, in the context 2042 * of its master-slaves ring. The rank of the slave is given by the number of 2043 * other slaves for the same master that have a better replication offset 2044 * compared to the local one (better means, greater, so they claim more data). 2045 * 2046 * A slave with rank 0 is the one with the greatest (most up to date) 2047 * replication offset, and so forth. Note that because how the rank is computed 2048 * multiple slaves may have the same rank, in case they have the same offset. 2049 * 2050 * The slave rank is used to add a delay to start an election in order to 2051 * get voted and replace a failing master. Slaves with better replication 2052 * offsets are more likely to win. */ 2053 int clusterGetSlaveRank(void) { 2054 long long myoffset; 2055 int j, rank = 0; 2056 clusterNode *master; 2057 2058 redisAssert(nodeIsSlave(myself)); 2059 master = myself->slaveof; 2060 if (master == NULL) return 0; /* Never called by slaves without master. */ 2061 2062 myoffset = replicationGetSlaveOffset(); 2063 for (j = 0; j < master->numslaves; j++) 2064 if (master->slaves[j] != myself && 2065 master->slaves[j]->repl_offset > myoffset) rank++; 2066 return rank; 2067 } 2068 2069 /* This function is called if we are a slave node and our master serving 2070 * a non-zero amount of hash slots is in FAIL state. 2071 * 2072 * The gaol of this function is: 2073 * 1) To check if we are able to perform a failover, is our data updated? 2074 * 2) Try to get elected by masters. 2075 * 3) Perform the failover informing all the other nodes. 2076 */ 2077 void clusterHandleSlaveFailover(void) { 2078 mstime_t data_age; 2079 mstime_t auth_age = mstime() - server.cluster->failover_auth_time; 2080 int needed_quorum = (server.cluster->size / 2) + 1; 2081 int manual_failover = server.cluster->mf_end != 0 && 2082 server.cluster->mf_can_start; 2083 int j; 2084 mstime_t auth_timeout, auth_retry_time; 2085 2086 /* Compute the failover timeout (the max time we have to send votes 2087 * and wait for replies), and the failover retry time (the time to wait 2088 * before waiting again. 2089 * 2090 * Timeout is MIN(NODE_TIMEOUT*2,2000) milliseconds. 2091 * Retry is two times the Timeout. 2092 */ 2093 auth_timeout = server.cluster_node_timeout*2; 2094 if (auth_timeout < 2000) auth_timeout = 2000; 2095 auth_retry_time = auth_timeout*2; 2096 2097 /* Pre conditions to run the function: 2098 * 1) We are a slave. 2099 * 2) Our master is flagged as FAIL, or this is a manual failover. 2100 * 3) It is serving slots. */ 2101 if (nodeIsMaster(myself) || 2102 myself->slaveof == NULL || 2103 (!nodeFailed(myself->slaveof) && !manual_failover) || 2104 myself->slaveof->numslots == 0) return; 2105 2106 /* Set data_age to the number of seconds we are disconnected from 2107 * the master. */ 2108 if (server.repl_state == REDIS_REPL_CONNECTED) { 2109 data_age = (server.unixtime - server.master->lastinteraction) * 1000; 2110 } else { 2111 data_age = (server.unixtime - server.repl_down_since) * 1000; 2112 } 2113 2114 /* Remove the node timeout from the data age as it is fine that we are 2115 * disconnected from our master at least for the time it was down to be 2116 * flagged as FAIL, that's the baseline. */ 2117 if (data_age > server.cluster_node_timeout) 2118 data_age -= server.cluster_node_timeout; 2119 2120 /* Check if our data is recent enough. For now we just use a fixed 2121 * constant of ten times the node timeout since the cluster should 2122 * react much faster to a master down. */ 2123 if (data_age > 2124 (server.repl_ping_slave_period * 1000) + 2125 (server.cluster_node_timeout * REDIS_CLUSTER_SLAVE_VALIDITY_MULT)) 2126 return; 2127 2128 /* If the previous failover attempt timedout and the retry time has 2129 * elapsed, we can setup a new one. */ 2130 if (auth_age > auth_retry_time) { 2131 server.cluster->failover_auth_time = mstime() + 2132 500 + /* Fixed delay of 500 milliseconds, let FAIL msg propagate. */ 2133 random() % 500; /* Random delay between 0 and 500 milliseconds. */ 2134 server.cluster->failover_auth_count = 0; 2135 server.cluster->failover_auth_sent = 0; 2136 server.cluster->failover_auth_rank = clusterGetSlaveRank(); 2137 /* We add another delay that is proportional to the slave rank. 2138 * Specifically 1 second * rank. This way slaves that have a probably 2139 * less updated replication offset, are penalized. */ 2140 server.cluster->failover_auth_time += 2141 server.cluster->failover_auth_rank * 1000; 2142 /* However if this is a manual failover, no delay is needed. */ 2143 if (server.cluster->mf_end) { 2144 server.cluster->failover_auth_time = mstime(); 2145 server.cluster->failover_auth_rank = 0; 2146 } 2147 redisLog(REDIS_WARNING, 2148 "Start of election delayed for %lld milliseconds " 2149 "(rank #%d, offset %lld).", 2150 server.cluster->failover_auth_time - mstime(), 2151 server.cluster->failover_auth_rank, 2152 replicationGetSlaveOffset()); 2153 /* Now that we have a scheduled election, broadcast our offset 2154 * to all the other slaves so that they'll updated their offsets 2155 * if our offset is better. */ 2156 clusterBroadcastPong(CLUSTER_BROADCAST_LOCAL_SLAVES); 2157 return; 2158 } 2159 2160 /* It is possible that we received more updated offsets from other 2161 * slaves for the same master since we computed our election delay. 2162 * Update the delay if our rank changed. */ 2163 if (server.cluster->failover_auth_sent == 0 && 2164 server.cluster->mf_end == 0) 2165 { 2166 int newrank = clusterGetSlaveRank(); 2167 if (newrank > server.cluster->failover_auth_rank) { 2168 long long added_delay = 2169 (newrank - server.cluster->failover_auth_rank) * 1000; 2170 server.cluster->failover_auth_time += added_delay; 2171 server.cluster->failover_auth_rank = newrank; 2172 redisLog(REDIS_WARNING, 2173 "Slave rank updated to #%d, added %lld milliseconds of delay.", 2174 newrank, added_delay); 2175 } 2176 } 2177 2178 /* Return ASAP if we can't still start the election. */ 2179 if (mstime() < server.cluster->failover_auth_time) return; 2180 2181 /* Return ASAP if the election is too old to be valid. */ 2182 if (auth_age > auth_timeout) return; 2183 2184 /* Ask for votes if needed. */ 2185 if (server.cluster->failover_auth_sent == 0) { 2186 server.cluster->currentEpoch++; 2187 server.cluster->failover_auth_epoch = server.cluster->currentEpoch; 2188 redisLog(REDIS_WARNING,"Starting a failover election for epoch %llu.", 2189 (unsigned long long) server.cluster->currentEpoch); 2190 clusterRequestFailoverAuth(); 2191 server.cluster->failover_auth_sent = 1; 2192 clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG| 2193 CLUSTER_TODO_UPDATE_STATE| 2194 CLUSTER_TODO_FSYNC_CONFIG); 2195 return; /* Wait for replies. */ 2196 } 2197 2198 /* Check if we reached the quorum. */ 2199 if (server.cluster->failover_auth_count >= needed_quorum) { 2200 clusterNode *oldmaster = myself->slaveof; 2201 2202 redisLog(REDIS_WARNING, 2203 "Failover election won: I'm the new master."); 2204 /* We have the quorum, perform all the steps to correctly promote 2205 * this slave to a master. 2206 * 2207 * 1) Turn this node into a master. */ 2208 clusterNodeRemoveSlave(myself->slaveof, myself); 2209 myself->flags &= ~REDIS_NODE_SLAVE; 2210 myself->flags |= REDIS_NODE_MASTER; 2211 myself->slaveof = NULL; 2212 replicationUnsetMaster(); 2213 2214 /* 2) Claim all the slots assigned to our master. */ 2215 for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) { 2216 if (clusterNodeGetSlotBit(oldmaster,j)) { 2217 clusterDelSlot(j); 2218 clusterAddSlot(myself,j); 2219 } 2220 } 2221 2222 /* 3) Update my configEpoch to the epoch of the election. */ 2223 myself->configEpoch = server.cluster->failover_auth_epoch; 2224 2225 /* 4) Update state and save config. */ 2226 clusterUpdateState(); 2227 clusterSaveConfigOrDie(1); 2228 2229 /* 5) Pong all the other nodes so that they can update the state 2230 * accordingly and detect that we switched to master role. */ 2231 clusterBroadcastPong(CLUSTER_BROADCAST_ALL); 2232 2233 /* 6) If there was a manual failover in progress, clear the state. */ 2234 resetManualFailover(); 2235 } 2236 } 2237 2238 /* ----------------------------------------------------------------------------- 2239 * CLUSTER slave migration 2240 * 2241 * Slave migration is the process that allows a slave of a master that is 2242 * already covered by at least another slave, to "migrate" to a master that 2243 * is orpaned, that is, left with no working slaves. 2244 * -------------------------------------------------------------------------- */ 2245 2246 /* This function is responsible to decide if this replica should be migrated 2247 * to a different (orphaned) master. It is called by the clusterCron() function 2248 * only if: 2249 * 2250 * 1) We are a slave node. 2251 * 2) It was detected that there is at least one orphaned master in 2252 * the cluster. 2253 * 3) We are a slave of one of the masters with the greatest number of 2254 * slaves. 2255 * 2256 * This checks are performed by the caller since it requires to iterate 2257 * the nodes anyway, so we spend time into clusterHandleSlaveMigration() 2258 * if definitely needed. 2259 * 2260 * The fuction is called with a pre-computed max_slaves, that is the max 2261 * number of working (not in FAIL state) slaves for a single master. 2262 * 2263 * Additional conditions for migration are examined inside the function. 2264 */ 2265 void clusterHandleSlaveMigration(int max_slaves) { 2266 int j, okslaves = 0; 2267 clusterNode *mymaster = myself->slaveof, *target = NULL, *candidate = NULL; 2268 dictIterator *di; 2269 dictEntry *de; 2270 2271 /* Step 1: Don't migrate if the cluster state is not ok. */ 2272 if (server.cluster->state != REDIS_CLUSTER_OK) return; 2273 2274 /* Step 2: Don't migrate if my master will not be left with at least 2275 * 'migration-barrier' slaves after my migration. */ 2276 if (mymaster == NULL) return; 2277 for (j = 0; j < mymaster->numslaves; j++) 2278 if (!nodeFailed(mymaster->slaves[j]) && 2279 !nodeTimedOut(mymaster->slaves[j])) okslaves++; 2280 if (okslaves <= server.cluster_migration_barrier) return; 2281 2282 /* Step 3: Idenitfy a candidate for migration, and check if among the 2283 * masters with the greatest number of ok slaves, I'm the one with the 2284 * smaller node ID. 2285 * 2286 * Note that this means that eventually a replica migration will occurr 2287 * since slaves that are reachable again always have their FAIL flag 2288 * cleared. At the same time this does not mean that there are no 2289 * race conditions possible (two slaves migrating at the same time), but 2290 * this is extremely unlikely to happen, and harmless. */ 2291 candidate = myself; 2292 di = dictGetSafeIterator(server.cluster->nodes); 2293 while((de = dictNext(di)) != NULL) { 2294 clusterNode *node = dictGetVal(de); 2295 int okslaves; 2296 2297 /* Only iterate over working masters. */ 2298 if (nodeIsSlave(node) || nodeFailed(node)) continue; 2299 okslaves = clusterCountNonFailingSlaves(node); 2300 2301 if (okslaves == 0 && target == NULL && node->numslots > 0) target = node; 2302 if (okslaves == max_slaves) { 2303 for (j = 0; j < node->numslaves; j++) { 2304 if (memcmp(node->slaves[j]->name, 2305 candidate->name, 2306 REDIS_CLUSTER_NAMELEN) < 0) 2307 { 2308 candidate = node->slaves[j]; 2309 } 2310 } 2311 } 2312 } 2313 2314 /* Step 4: perform the migration if there is a target, and if I'm the 2315 * candidate. */ 2316 if (target && candidate == myself) { 2317 redisLog(REDIS_WARNING,"Migrating to orphaned master %.40s", 2318 target->name); 2319 clusterSetMaster(target); 2320 } 2321 } 2322 2323 /* ----------------------------------------------------------------------------- 2324 * CLUSTER manual failover 2325 * 2326 * This are the important steps performed by slaves during a manual failover: 2327 * 1) User send CLUSTER FAILOVER command. The failover state is initialized 2328 * setting mf_end to the millisecond unix time at which we'll abort the 2329 * attempt. 2330 * 2) Slave sends a MFSTART message to the master requesting to pause clients 2331 * for two times the manual failover timeout REDIS_CLUSTER_MF_TIMEOUT. 2332 * When master is paused for manual failover, it also starts to flag 2333 * packets with CLUSTERMSG_FLAG0_PAUSED. 2334 * 3) Slave waits for master to send its replication offset flagged as PAUSED. 2335 * 4) If slave received the offset from the master, and its offset matches, 2336 * mf_can_start is set to 1, and clusterHandleSlaveFailover() will perform 2337 * the failover as usually, with the difference that the vote request 2338 * will be modified to force masters to vote for a slave that has a 2339 * working master. 2340 * 2341 * From the point of view of the master things are simpler: when a 2342 * PAUSE_CLIENTS packet is received the master sets mf_end as well and 2343 * the sender in mf_slave. During the time limit for the manual failover 2344 * the master will just send PINGs more often to this slave, flagged with 2345 * the PAUSED flag, so that the slave will set mf_master_offset when receiving 2346 * a packet from the master with this flag set. 2347 * 2348 * The gaol of the manual failover is to perform a fast failover without 2349 * data loss due to the asynchronous master-slave replication. 2350 * -------------------------------------------------------------------------- */ 2351 2352 /* Reset the manual failover state. This works for both masters and slavesa 2353 * as all the state about manual failover is cleared. 2354 * 2355 * The function can be used both to initialize the manual failover state at 2356 * startup or to abort a manual failover in progress. */ 2357 void resetManualFailover(void) { 2358 if (server.cluster->mf_end && clientsArePaused()) { 2359 server.clients_pause_end_time = 0; 2360 clientsArePaused(); /* Just use the side effect of the function. */ 2361 } 2362 server.cluster->mf_end = 0; /* No manual failover in progress. */ 2363 server.cluster->mf_can_start = 0; 2364 server.cluster->mf_slave = NULL; 2365 server.cluster->mf_master_offset = 0; 2366 } 2367 2368 /* If a manual failover timed out, abort it. */ 2369 void manualFailoverCheckTimeout(void) { 2370 if (server.cluster->mf_end && server.cluster->mf_end < mstime()) { 2371 redisLog(REDIS_WARNING,"Manual failover timed out."); 2372 resetManualFailover(); 2373 } 2374 } 2375 2376 /* This function is called from the cluster cron function in order to go 2377 * forward with a manual failover state machine. */ 2378 void clusterHandleManualFailover(void) { 2379 /* Return ASAP if no manual failover is in progress. */ 2380 if (server.cluster->mf_end == 0) return; 2381 2382 /* If mf_can_start is non-zero, the failover was alrady triggered so the 2383 * next steps are performed by clusterHandleSlaveFailover(). */ 2384 if (server.cluster->mf_can_start) return; 2385 2386 if (server.cluster->mf_master_offset == 0) return; /* Wait for offset... */ 2387 2388 if (server.cluster->mf_master_offset == replicationGetSlaveOffset()) { 2389 /* Our replication offset matches the master replication offset 2390 * announced after clients were paused. We can start the failover. */ 2391 server.cluster->mf_can_start = 1; 2392 redisLog(REDIS_WARNING,"All master replication stream processed, manual failover can start."); 2393 } 2394 } 2395 2396 /* ----------------------------------------------------------------------------- 2397 * CLUSTER cron job 2398 * -------------------------------------------------------------------------- */ 2399 2400 /* This is executed 10 times every second */ 2401 void clusterCron(void) { 2402 dictIterator *di; 2403 dictEntry *de; 2404 int update_state = 0; 2405 int orphaned_masters; /* How many masters there are without ok slaves. */ 2406 int max_slaves; /* Max number of ok slaves for a single master. */ 2407 int this_slaves; /* Number of ok slaves for our master (if we are slave). */ 2408 mstime_t min_pong = 0, now = mstime(); 2409 clusterNode *min_pong_node = NULL; 2410 static unsigned long long iteration = 0; 2411 mstime_t handshake_timeout; 2412 2413 iteration++; /* Number of times this function was called so far. */ 2414 2415 /* The handshake timeout is the time after which a handshake node that was 2416 * not turned into a normal node is removed from the nodes. Usually it is 2417 * just the NODE_TIMEOUT value, but when NODE_TIMEOUT is too small we use 2418 * the value of 1 second. */ 2419 handshake_timeout = server.cluster_node_timeout; 2420 if (handshake_timeout < 1000) handshake_timeout = 1000; 2421 2422 /* Check if we have disconnected nodes and re-establish the connection. */ 2423 di = dictGetSafeIterator(server.cluster->nodes); 2424 while((de = dictNext(di)) != NULL) { 2425 clusterNode *node = dictGetVal(de); 2426 2427 if (node->flags & (REDIS_NODE_MYSELF|REDIS_NODE_NOADDR)) continue; 2428 2429 /* A Node in HANDSHAKE state has a limited lifespan equal to the 2430 * configured node timeout. */ 2431 if (nodeInHandshake(node) && now - node->ctime > handshake_timeout) { 2432 freeClusterNode(node); 2433 continue; 2434 } 2435 2436 if (node->link == NULL) { 2437 int fd; 2438 mstime_t old_ping_sent; 2439 clusterLink *link; 2440 2441 fd = anetTcpNonBlockBindConnect(server.neterr, node->ip, 2442 node->port+REDIS_CLUSTER_PORT_INCR, 2443 server.bindaddr_count ? server.bindaddr[0] : NULL); 2444 if (fd == -1) { 2445 redisLog(REDIS_DEBUG, "Unable to connect to " 2446 "Cluster Node [%s]:%d -> %s", node->ip, 2447 node->port+REDIS_CLUSTER_PORT_INCR, 2448 server.neterr); 2449 continue; 2450 } 2451 link = createClusterLink(node); 2452 link->fd = fd; 2453 node->link = link; 2454 aeCreateFileEvent(server.el,link->fd,AE_READABLE,clusterReadHandler,link); 2455 /* Queue a PING in the new connection ASAP: this is crucial 2456 * to avoid false positives in failure detection. 2457 * 2458 * If the node is flagged as MEET, we send a MEET message instead 2459 * of a PING one, to force the receiver to add us in its node 2460 * table. */ 2461 old_ping_sent = node->ping_sent; 2462 clusterSendPing(link, node->flags & REDIS_NODE_MEET ? 2463 CLUSTERMSG_TYPE_MEET : CLUSTERMSG_TYPE_PING); 2464 if (old_ping_sent) { 2465 /* If there was an active ping before the link was 2466 * disconnected, we want to restore the ping time, otherwise 2467 * replaced by the clusterSendPing() call. */ 2468 node->ping_sent = old_ping_sent; 2469 } 2470 /* We can clear the flag after the first packet is sent. 2471 * If we'll never receive a PONG, we'll never send new packets 2472 * to this node. Instead after the PONG is received and we 2473 * are no longer in meet/handshake status, we want to send 2474 * normal PING packets. */ 2475 node->flags &= ~REDIS_NODE_MEET; 2476 2477 redisLog(REDIS_DEBUG,"Connecting with Node %.40s at %s:%d", node->name, node->ip, node->port+REDIS_CLUSTER_PORT_INCR); 2478 } 2479 } 2480 dictReleaseIterator(di); 2481 2482 /* Ping some random node 1 time every 10 iterations, so that we usually ping 2483 * one random node every second. */ 2484 if (!(iteration % 10)) { 2485 int j; 2486 2487 /* Check a few random nodes and ping the one with the oldest 2488 * pong_received time. */ 2489 for (j = 0; j < 5; j++) { 2490 de = dictGetRandomKey(server.cluster->nodes); 2491 clusterNode *this = dictGetVal(de); 2492 2493 /* Don't ping nodes disconnected or with a ping currently active. */ 2494 if (this->link == NULL || this->ping_sent != 0) continue; 2495 if (this->flags & (REDIS_NODE_MYSELF|REDIS_NODE_HANDSHAKE)) continue; 2496 if (min_pong_node == NULL || min_pong > this->pong_received) { 2497 min_pong_node = this; 2498 min_pong = this->pong_received; 2499 } 2500 } 2501 if (min_pong_node) { 2502 redisLog(REDIS_DEBUG,"Pinging node %.40s", min_pong_node->name); 2503 clusterSendPing(min_pong_node->link, CLUSTERMSG_TYPE_PING); 2504 } 2505 } 2506 2507 /* Iterate nodes to check if we need to flag something as failing. 2508 * This loop is also responsible to: 2509 * 1) Check if there are orphaned masters (masters without non failing 2510 * slaves). 2511 * 2) Count the max number of non failing slaves for a single master. 2512 * 3) Count the number of slaves for our master, if we are a slave. */ 2513 orphaned_masters = 0; 2514 max_slaves = 0; 2515 this_slaves = 0; 2516 di = dictGetSafeIterator(server.cluster->nodes); 2517 while((de = dictNext(di)) != NULL) { 2518 clusterNode *node = dictGetVal(de); 2519 now = mstime(); /* Use an updated time at every iteration. */ 2520 mstime_t delay; 2521 2522 if (node->flags & 2523 (REDIS_NODE_MYSELF|REDIS_NODE_NOADDR|REDIS_NODE_HANDSHAKE)) 2524 continue; 2525 2526 /* Orphaned master check, useful only if the current instance 2527 * is a slave that may migrate to another master. */ 2528 if (nodeIsSlave(myself) && nodeIsMaster(node) && !nodeFailed(node)) { 2529 int okslaves = clusterCountNonFailingSlaves(node); 2530 2531 if (okslaves == 0 && node->numslots > 0) orphaned_masters++; 2532 if (okslaves > max_slaves) max_slaves = okslaves; 2533 if (nodeIsSlave(myself) && myself->slaveof == node) 2534 this_slaves = okslaves; 2535 } 2536 2537 /* If we are waiting for the PONG more than half the cluster 2538 * timeout, reconnect the link: maybe there is a connection 2539 * issue even if the node is alive. */ 2540 if (node->link && /* is connected */ 2541 now - node->link->ctime > 2542 server.cluster_node_timeout && /* was not already reconnected */ 2543 node->ping_sent && /* we already sent a ping */ 2544 node->pong_received < node->ping_sent && /* still waiting pong */ 2545 /* and we are waiting for the pong more than timeout/2 */ 2546 now - node->ping_sent > server.cluster_node_timeout/2) 2547 { 2548 /* Disconnect the link, it will be reconnected automatically. */ 2549 freeClusterLink(node->link); 2550 } 2551 2552 /* If we have currently no active ping in this instance, and the 2553 * received PONG is older than half the cluster timeout, send 2554 * a new ping now, to ensure all the nodes are pinged without 2555 * a too big delay. */ 2556 if (node->link && 2557 node->ping_sent == 0 && 2558 (now - node->pong_received) > server.cluster_node_timeout/2) 2559 { 2560 clusterSendPing(node->link, CLUSTERMSG_TYPE_PING); 2561 continue; 2562 } 2563 2564 /* If we are a master and one of the slaves requested a manual 2565 * failover, ping it continuously. */ 2566 if (server.cluster->mf_end && 2567 nodeIsMaster(myself) && 2568 server.cluster->mf_slave == node && 2569 node->link) 2570 { 2571 clusterSendPing(node->link, CLUSTERMSG_TYPE_PING); 2572 continue; 2573 } 2574 2575 /* Check only if we have an active ping for this instance. */ 2576 if (node->ping_sent == 0) continue; 2577 2578 /* Compute the delay of the PONG. Note that if we already received 2579 * the PONG, then node->ping_sent is zero, so can't reach this 2580 * code at all. */ 2581 delay = now - node->ping_sent; 2582 2583 if (delay > server.cluster_node_timeout) { 2584 /* Timeout reached. Set the node as possibly failing if it is 2585 * not already in this state. */ 2586 if (!(node->flags & (REDIS_NODE_PFAIL|REDIS_NODE_FAIL))) { 2587 redisLog(REDIS_DEBUG,"*** NODE %.40s possibly failing", 2588 node->name); 2589 node->flags |= REDIS_NODE_PFAIL; 2590 update_state = 1; 2591 } 2592 } 2593 } 2594 dictReleaseIterator(di); 2595 2596 /* If we are a slave node but the replication is still turned off, 2597 * enable it if we know the address of our master and it appears to 2598 * be up. */ 2599 if (nodeIsSlave(myself) && 2600 server.masterhost == NULL && 2601 myself->slaveof && 2602 nodeHasAddr(myself->slaveof)) 2603 { 2604 replicationSetMaster(myself->slaveof->ip, myself->slaveof->port); 2605 } 2606 2607 /* Abourt a manual failover if the timeout is reached. */ 2608 manualFailoverCheckTimeout(); 2609 2610 if (nodeIsSlave(myself)) { 2611 clusterHandleManualFailover(); 2612 clusterHandleSlaveFailover(); 2613 /* If there are orphaned slaves, and we are a slave among the masters 2614 * with the max number of non-failing slaves, consider migrating to 2615 * the orphaned masters. Note that it does not make sense to try 2616 * a migration if there is no master with at least *two* working 2617 * slaves. */ 2618 if (orphaned_masters && max_slaves >= 2 && this_slaves == max_slaves) 2619 clusterHandleSlaveMigration(max_slaves); 2620 } 2621 2622 if (update_state || server.cluster->state == REDIS_CLUSTER_FAIL) 2623 clusterUpdateState(); 2624 } 2625 2626 /* This function is called before the event handler returns to sleep for 2627 * events. It is useful to perform operations that must be done ASAP in 2628 * reaction to events fired but that are not safe to perform inside event 2629 * handlers, or to perform potentially expansive tasks that we need to do 2630 * a single time before replying to clients. */ 2631 void clusterBeforeSleep(void) { 2632 /* Handle failover, this is needed when it is likely that there is already 2633 * the quorum from masters in order to react fast. */ 2634 if (server.cluster->todo_before_sleep & CLUSTER_TODO_HANDLE_FAILOVER) 2635 clusterHandleSlaveFailover(); 2636 2637 /* Update the cluster state. */ 2638 if (server.cluster->todo_before_sleep & CLUSTER_TODO_UPDATE_STATE) 2639 clusterUpdateState(); 2640 2641 /* Save the config, possibly using fsync. */ 2642 if (server.cluster->todo_before_sleep & CLUSTER_TODO_SAVE_CONFIG) { 2643 int fsync = server.cluster->todo_before_sleep & CLUSTER_TODO_FSYNC_CONFIG; 2644 clusterSaveConfigOrDie(fsync); 2645 } 2646 2647 /* Reset our flags. */ 2648 server.cluster->todo_before_sleep = 0; 2649 } 2650 2651 void clusterDoBeforeSleep(int flags) { 2652 server.cluster->todo_before_sleep |= flags; 2653 } 2654 2655 /* ----------------------------------------------------------------------------- 2656 * Slots management 2657 * -------------------------------------------------------------------------- */ 2658 2659 /* Test bit 'pos' in a generic bitmap. Return 1 if the bit is set, 2660 * otherwise 0. */ 2661 int bitmapTestBit(unsigned char *bitmap, int pos) { 2662 off_t byte = pos/8; 2663 int bit = pos&7; 2664 return (bitmap[byte] & (1<<bit)) != 0; 2665 } 2666 2667 /* Set the bit at position 'pos' in a bitmap. */ 2668 void bitmapSetBit(unsigned char *bitmap, int pos) { 2669 off_t byte = pos/8; 2670 int bit = pos&7; 2671 bitmap[byte] |= 1<<bit; 2672 } 2673 2674 /* Clear the bit at position 'pos' in a bitmap. */ 2675 void bitmapClearBit(unsigned char *bitmap, int pos) { 2676 off_t byte = pos/8; 2677 int bit = pos&7; 2678 bitmap[byte] &= ~(1<<bit); 2679 } 2680 2681 /* Set the slot bit and return the old value. */ 2682 int clusterNodeSetSlotBit(clusterNode *n, int slot) { 2683 int old = bitmapTestBit(n->slots,slot); 2684 bitmapSetBit(n->slots,slot); 2685 if (!old) n->numslots++; 2686 return old; 2687 } 2688 2689 /* Clear the slot bit and return the old value. */ 2690 int clusterNodeClearSlotBit(clusterNode *n, int slot) { 2691 int old = bitmapTestBit(n->slots,slot); 2692 bitmapClearBit(n->slots,slot); 2693 if (old) n->numslots--; 2694 return old; 2695 } 2696 2697 /* Return the slot bit from the cluster node structure. */ 2698 int clusterNodeGetSlotBit(clusterNode *n, int slot) { 2699 return bitmapTestBit(n->slots,slot); 2700 } 2701 2702 /* Add the specified slot to the list of slots that node 'n' will 2703 * serve. Return REDIS_OK if the operation ended with success. 2704 * If the slot is already assigned to another instance this is considered 2705 * an error and REDIS_ERR is returned. */ 2706 int clusterAddSlot(clusterNode *n, int slot) { 2707 if (server.cluster->slots[slot]) return REDIS_ERR; 2708 clusterNodeSetSlotBit(n,slot); 2709 server.cluster->slots[slot] = n; 2710 return REDIS_OK; 2711 } 2712 2713 /* Delete the specified slot marking it as unassigned. 2714 * Returns REDIS_OK if the slot was assigned, otherwise if the slot was 2715 * already unassigned REDIS_ERR is returned. */ 2716 int clusterDelSlot(int slot) { 2717 clusterNode *n = server.cluster->slots[slot]; 2718 2719 if (!n) return REDIS_ERR; 2720 redisAssert(clusterNodeClearSlotBit(n,slot) == 1); 2721 server.cluster->slots[slot] = NULL; 2722 return REDIS_OK; 2723 } 2724 2725 /* Delete all the slots associated with the specified node. 2726 * The number of deleted slots is returned. */ 2727 int clusterDelNodeSlots(clusterNode *node) { 2728 int deleted = 0, j; 2729 2730 for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) { 2731 if (clusterNodeGetSlotBit(node,j)) clusterDelSlot(j); 2732 deleted++; 2733 } 2734 return deleted; 2735 } 2736 2737 /* Clear the migrating / importing state for all the slots. 2738 * This is useful at initialization and when turning a master into slave. */ 2739 void clusterCloseAllSlots(void) { 2740 memset(server.cluster->migrating_slots_to,0, 2741 sizeof(server.cluster->migrating_slots_to)); 2742 memset(server.cluster->importing_slots_from,0, 2743 sizeof(server.cluster->importing_slots_from)); 2744 } 2745 2746 /* ----------------------------------------------------------------------------- 2747 * Cluster state evaluation function 2748 * -------------------------------------------------------------------------- */ 2749 2750 /* The following are defines that are only used in the evaluation function 2751 * and are based on heuristics. Actaully the main point about the rejoin and 2752 * writable delay is that they should be a few orders of magnitude larger 2753 * than the network latency. */ 2754 #define REDIS_CLUSTER_MAX_REJOIN_DELAY 5000 2755 #define REDIS_CLUSTER_MIN_REJOIN_DELAY 500 2756 #define REDIS_CLUSTER_WRITABLE_DELAY 2000 2757 2758 void clusterUpdateState(void) { 2759 int j, new_state; 2760 int unreachable_masters = 0; 2761 static mstime_t among_minority_time; 2762 static mstime_t first_call_time = 0; 2763 2764 /* If this is a master node, wait some time before turning the state 2765 * into OK, since it is not a good idea to rejoin the cluster as a writable 2766 * master, after a reboot, without giving the cluster a chance to 2767 * reconfigure this node. Note that the delay is calculated starting from 2768 * the first call to this function and not since the server start, in order 2769 * to don't count the DB loading time. */ 2770 if (first_call_time == 0) first_call_time = mstime(); 2771 if (nodeIsMaster(myself) && 2772 mstime() - first_call_time < REDIS_CLUSTER_WRITABLE_DELAY) return; 2773 2774 /* Start assuming the state is OK. We'll turn it into FAIL if there 2775 * are the right conditions. */ 2776 new_state = REDIS_CLUSTER_OK; 2777 2778 /* Check if all the slots are covered. */ 2779 for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) { 2780 if (server.cluster->slots[j] == NULL || 2781 server.cluster->slots[j]->flags & (REDIS_NODE_FAIL)) 2782 { 2783 new_state = REDIS_CLUSTER_FAIL; 2784 break; 2785 } 2786 } 2787 2788 /* Compute the cluster size, that is the number of master nodes 2789 * serving at least a single slot. 2790 * 2791 * At the same time count the number of unreachable masters with 2792 * at least one node. */ 2793 { 2794 dictIterator *di; 2795 dictEntry *de; 2796 2797 server.cluster->size = 0; 2798 di = dictGetSafeIterator(server.cluster->nodes); 2799 while((de = dictNext(di)) != NULL) { 2800 clusterNode *node = dictGetVal(de); 2801 2802 if (nodeIsMaster(node) && node->numslots) { 2803 server.cluster->size++; 2804 if (node->flags & (REDIS_NODE_FAIL|REDIS_NODE_PFAIL)) 2805 unreachable_masters++; 2806 } 2807 } 2808 dictReleaseIterator(di); 2809 } 2810 2811 /* If we can't reach at least half the masters, change the cluster state 2812 * to FAIL, as we are not even able to mark nodes as FAIL in this side 2813 * of the netsplit because of lack of majority. */ 2814 { 2815 int needed_quorum = (server.cluster->size / 2) + 1; 2816 2817 if (unreachable_masters >= needed_quorum) { 2818 new_state = REDIS_CLUSTER_FAIL; 2819 among_minority_time = mstime(); 2820 } 2821 } 2822 2823 /* Log a state change */ 2824 if (new_state != server.cluster->state) { 2825 mstime_t rejoin_delay = server.cluster_node_timeout; 2826 2827 /* If the instance is a master and was partitioned away with the 2828 * minority, don't let it accept queries for some time after the 2829 * partition heals, to make sure there is enough time to receive 2830 * a configuration update. */ 2831 if (rejoin_delay > REDIS_CLUSTER_MAX_REJOIN_DELAY) 2832 rejoin_delay = REDIS_CLUSTER_MAX_REJOIN_DELAY; 2833 if (rejoin_delay < REDIS_CLUSTER_MIN_REJOIN_DELAY) 2834 rejoin_delay = REDIS_CLUSTER_MIN_REJOIN_DELAY; 2835 2836 if (new_state == REDIS_CLUSTER_OK && 2837 nodeIsMaster(myself) && 2838 mstime() - among_minority_time < rejoin_delay) 2839 { 2840 return; 2841 } 2842 2843 /* Change the state and log the event. */ 2844 redisLog(REDIS_WARNING,"Cluster state changed: %s", 2845 new_state == REDIS_CLUSTER_OK ? "ok" : "fail"); 2846 server.cluster->state = new_state; 2847 } 2848 } 2849 2850 /* This function is called after the node startup in order to verify that data 2851 * loaded from disk is in agreement with the cluster configuration: 2852 * 2853 * 1) If we find keys about hash slots we have no responsibility for, the 2854 * following happens: 2855 * A) If no other node is in charge according to the current cluster 2856 * configuration, we add these slots to our node. 2857 * B) If according to our config other nodes are already in charge for 2858 * this lots, we set the slots as IMPORTING from our point of view 2859 * in order to justify we have those slots, and in order to make 2860 * redis-trib aware of the issue, so that it can try to fix it. 2861 * 2) If we find data in a DB different than DB0 we return REDIS_ERR to 2862 * signal the caller it should quit the server with an error message 2863 * or take other actions. 2864 * 2865 * The function always returns REDIS_OK even if it will try to correct 2866 * the error described in "1". However if data is found in DB different 2867 * from DB0, REDIS_ERR is returned. 2868 * 2869 * The function also uses the logging facility in order to warn the user 2870 * about desynchronizations between the data we have in memory and the 2871 * cluster configuration. */ 2872 int verifyClusterConfigWithData(void) { 2873 int j; 2874 int update_config = 0; 2875 2876 /* If this node is a slave, don't perform the check at all as we 2877 * completely depend on the replication stream. */ 2878 if (nodeIsSlave(myself)) return REDIS_OK; 2879 2880 /* Make sure we only have keys in DB0. */ 2881 for (j = 1; j < server.dbnum; j++) { 2882 if (dictSize(server.db[j].dict)) return REDIS_ERR; 2883 } 2884 2885 /* Check that all the slots we see populated memory have a corresponding 2886 * entry in the cluster table. Otherwise fix the table. */ 2887 for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) { 2888 if (!countKeysInSlot(j)) continue; /* No keys in this slot. */ 2889 /* Check if we are assigned to this slot or if we are importing it. 2890 * In both cases check the next slot as the configuration makes 2891 * sense. */ 2892 if (server.cluster->slots[j] == myself || 2893 server.cluster->importing_slots_from[j] != NULL) continue; 2894 2895 /* If we are here data and cluster config don't agree, and we have 2896 * slot 'j' populated even if we are not importing it, nor we are 2897 * assigned to this slot. Fix this condition. */ 2898 2899 update_config++; 2900 /* Case A: slot is unassigned. Take responsability for it. */ 2901 if (server.cluster->slots[j] == NULL) { 2902 redisLog(REDIS_WARNING, "I've keys about slot %d that is " 2903 "unassigned. Taking responsability " 2904 "for it.",j); 2905 clusterAddSlot(myself,j); 2906 } else { 2907 redisLog(REDIS_WARNING, "I've keys about slot %d that is " 2908 "already assigned to a different node. " 2909 "Setting it in importing state.",j); 2910 server.cluster->importing_slots_from[j] = server.cluster->slots[j]; 2911 } 2912 } 2913 if (update_config) clusterSaveConfigOrDie(1); 2914 return REDIS_OK; 2915 } 2916 2917 /* ----------------------------------------------------------------------------- 2918 * SLAVE nodes handling 2919 * -------------------------------------------------------------------------- */ 2920 2921 /* Set the specified node 'n' as master for this node. 2922 * If this node is currently a master, it is turned into a slave. */ 2923 void clusterSetMaster(clusterNode *n) { 2924 redisAssert(n != myself); 2925 redisAssert(myself->numslots == 0); 2926 2927 if (nodeIsMaster(myself)) { 2928 myself->flags &= ~REDIS_NODE_MASTER; 2929 myself->flags |= REDIS_NODE_SLAVE; 2930 clusterCloseAllSlots(); 2931 } else { 2932 if (myself->slaveof) 2933 clusterNodeRemoveSlave(myself->slaveof,myself); 2934 } 2935 myself->slaveof = n; 2936 clusterNodeAddSlave(n,myself); 2937 replicationSetMaster(n->ip, n->port); 2938 resetManualFailover(); 2939 } 2940 2941 /* ----------------------------------------------------------------------------- 2942 * CLUSTER command 2943 * -------------------------------------------------------------------------- */ 2944 2945 /* Generate a csv-alike representation of the specified cluster node. 2946 * See clusterGenNodesDescription() top comment for more information. 2947 * 2948 * The function returns the string representation as an SDS string. */ 2949 sds clusterGenNodeDescription(clusterNode *node) { 2950 int j, start; 2951 sds ci; 2952 2953 /* Node coordinates */ 2954 ci = sdscatprintf(sdsempty(),"%.40s %s:%d ", 2955 node->name, 2956 node->ip, 2957 node->port); 2958 2959 /* Flags */ 2960 if (node->flags == 0) ci = sdscat(ci,"noflags,"); 2961 if (node->flags & REDIS_NODE_MYSELF) ci = sdscat(ci,"myself,"); 2962 if (node->flags & REDIS_NODE_MASTER) ci = sdscat(ci,"master,"); 2963 if (node->flags & REDIS_NODE_SLAVE) ci = sdscat(ci,"slave,"); 2964 if (node->flags & REDIS_NODE_PFAIL) ci = sdscat(ci,"fail?,"); 2965 if (node->flags & REDIS_NODE_FAIL) ci = sdscat(ci,"fail,"); 2966 if (node->flags & REDIS_NODE_HANDSHAKE) ci =sdscat(ci,"handshake,"); 2967 if (node->flags & REDIS_NODE_NOADDR) ci = sdscat(ci,"noaddr,"); 2968 if (ci[sdslen(ci)-1] == ',') ci[sdslen(ci)-1] = ' '; 2969 2970 /* Slave of... or just "-" */ 2971 if (node->slaveof) 2972 ci = sdscatprintf(ci,"%.40s ",node->slaveof->name); 2973 else 2974 ci = sdscatprintf(ci,"- "); 2975 2976 /* Latency from the POV of this node, link status */ 2977 ci = sdscatprintf(ci,"%lld %lld %llu %s", 2978 (long long) node->ping_sent, 2979 (long long) node->pong_received, 2980 (unsigned long long) node->configEpoch, 2981 (node->link || node->flags & REDIS_NODE_MYSELF) ? 2982 "connected" : "disconnected"); 2983 2984 /* Slots served by this instance */ 2985 start = -1; 2986 for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) { 2987 int bit; 2988 2989 if ((bit = clusterNodeGetSlotBit(node,j)) != 0) { 2990 if (start == -1) start = j; 2991 } 2992 if (start != -1 && (!bit || j == REDIS_CLUSTER_SLOTS-1)) { 2993 if (j == REDIS_CLUSTER_SLOTS-1) j++; 2994 2995 if (start == j-1) { 2996 ci = sdscatprintf(ci," %d",start); 2997 } else { 2998 ci = sdscatprintf(ci," %d-%d",start,j-1); 2999 } 3000 start = -1; 3001 } 3002 } 3003 3004 /* Just for MYSELF node we also dump info about slots that 3005 * we are migrating to other instances or importing from other 3006 * instances. */ 3007 if (node->flags & REDIS_NODE_MYSELF) { 3008 for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) { 3009 if (server.cluster->migrating_slots_to[j]) { 3010 ci = sdscatprintf(ci," [%d->-%.40s]",j, 3011 server.cluster->migrating_slots_to[j]->name); 3012 } else if (server.cluster->importing_slots_from[j]) { 3013 ci = sdscatprintf(ci," [%d-<-%.40s]",j, 3014 server.cluster->importing_slots_from[j]->name); 3015 } 3016 } 3017 } 3018 return ci; 3019 } 3020 3021 /* Generate a csv-alike representation of the nodes we are aware of, 3022 * including the "myself" node, and return an SDS string containing the 3023 * representation (it is up to the caller to free it). 3024 * 3025 * All the nodes matching at least one of the node flags specified in 3026 * "filter" are excluded from the output, so using zero as a filter will 3027 * include all the known nodes in the representation, including nodes in 3028 * the HANDSHAKE state. 3029 * 3030 * The representation obtained using this function is used for the output 3031 * of the CLUSTER NODES function, and as format for the cluster 3032 * configuration file (nodes.conf) for a given node. */ 3033 sds clusterGenNodesDescription(int filter) { 3034 sds ci = sdsempty(), ni; 3035 dictIterator *di; 3036 dictEntry *de; 3037 3038 di = dictGetSafeIterator(server.cluster->nodes); 3039 while((de = dictNext(di)) != NULL) { 3040 clusterNode *node = dictGetVal(de); 3041 3042 if (node->flags & filter) continue; 3043 ni = clusterGenNodeDescription(node); 3044 ci = sdscatsds(ci,ni); 3045 sdsfree(ni); 3046 ci = sdscatlen(ci,"\n",1); 3047 } 3048 dictReleaseIterator(di); 3049 return ci; 3050 } 3051 3052 int getSlotOrReply(redisClient *c, robj *o) { 3053 long long slot; 3054 3055 if (getLongLongFromObject(o,&slot) != REDIS_OK || 3056 slot < 0 || slot >= REDIS_CLUSTER_SLOTS) 3057 { 3058 addReplyError(c,"Invalid or out of range slot"); 3059 return -1; 3060 } 3061 return (int) slot; 3062 } 3063 3064 void clusterCommand(redisClient *c) { 3065 if (server.cluster_enabled == 0) { 3066 addReplyError(c,"This instance has cluster support disabled"); 3067 return; 3068 } 3069 3070 if (!strcasecmp(c->argv[1]->ptr,"meet") && c->argc == 4) { 3071 long port; 3072 3073 if (getLongFromObjectOrReply(c, c->argv[3], &port, NULL) != REDIS_OK) { 3074 addReplyError(c,"Invalid TCP port specified"); 3075 return; 3076 } 3077 3078 if (clusterStartHandshake(c->argv[2]->ptr,port) == 0 && 3079 errno == EINVAL) 3080 { 3081 addReplyError(c,"Invalid node address specified"); 3082 } else { 3083 addReply(c,shared.ok); 3084 } 3085 } else if (!strcasecmp(c->argv[1]->ptr,"nodes") && c->argc == 2) { 3086 /* CLUSTER NODES */ 3087 robj *o; 3088 sds ci = clusterGenNodesDescription(0); 3089 3090 o = createObject(REDIS_STRING,ci); 3091 addReplyBulk(c,o); 3092 decrRefCount(o); 3093 } else if (!strcasecmp(c->argv[1]->ptr,"flushslots") && c->argc == 2) { 3094 /* CLUSTER FLUSHSLOTS */ 3095 if (dictSize(server.db[0].dict) != 0) { 3096 addReplyError(c,"DB must be empty to perform CLUSTER FLUSHSLOTS."); 3097 return; 3098 } 3099 clusterDelNodeSlots(myself); 3100 clusterDoBeforeSleep(CLUSTER_TODO_UPDATE_STATE|CLUSTER_TODO_SAVE_CONFIG); 3101 addReply(c,shared.ok); 3102 } else if ((!strcasecmp(c->argv[1]->ptr,"addslots") || 3103 !strcasecmp(c->argv[1]->ptr,"delslots")) && c->argc >= 3) 3104 { 3105 /* CLUSTER ADDSLOTS <slot> [slot] ... */ 3106 /* CLUSTER DELSLOTS <slot> [slot] ... */ 3107 int j, slot; 3108 unsigned char *slots = zmalloc(REDIS_CLUSTER_SLOTS); 3109 int del = !strcasecmp(c->argv[1]->ptr,"delslots"); 3110 3111 memset(slots,0,REDIS_CLUSTER_SLOTS); 3112 /* Check that all the arguments are parsable and that all the 3113 * slots are not already busy. */ 3114 for (j = 2; j < c->argc; j++) { 3115 if ((slot = getSlotOrReply(c,c->argv[j])) == -1) { 3116 zfree(slots); 3117 return; 3118 } 3119 if (del && server.cluster->slots[slot] == NULL) { 3120 addReplyErrorFormat(c,"Slot %d is already unassigned", slot); 3121 zfree(slots); 3122 return; 3123 } else if (!del && server.cluster->slots[slot]) { 3124 addReplyErrorFormat(c,"Slot %d is already busy", slot); 3125 zfree(slots); 3126 return; 3127 } 3128 if (slots[slot]++ == 1) { 3129 addReplyErrorFormat(c,"Slot %d specified multiple times", 3130 (int)slot); 3131 zfree(slots); 3132 return; 3133 } 3134 } 3135 for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) { 3136 if (slots[j]) { 3137 int retval; 3138 3139 /* If this slot was set as importing we can clear this 3140 * state as now we are the real owner of the slot. */ 3141 if (server.cluster->importing_slots_from[j]) 3142 server.cluster->importing_slots_from[j] = NULL; 3143 3144 retval = del ? clusterDelSlot(j) : 3145 clusterAddSlot(myself,j); 3146 redisAssertWithInfo(c,NULL,retval == REDIS_OK); 3147 } 3148 } 3149 zfree(slots); 3150 clusterDoBeforeSleep(CLUSTER_TODO_UPDATE_STATE|CLUSTER_TODO_SAVE_CONFIG); 3151 addReply(c,shared.ok); 3152 } else if (!strcasecmp(c->argv[1]->ptr,"setslot") && c->argc >= 4) { 3153 /* SETSLOT 10 MIGRATING <node ID> */ 3154 /* SETSLOT 10 IMPORTING <node ID> */ 3155 /* SETSLOT 10 STABLE */ 3156 /* SETSLOT 10 NODE <node ID> */ 3157 int slot; 3158 clusterNode *n; 3159 3160 if ((slot = getSlotOrReply(c,c->argv[2])) == -1) return; 3161 3162 if (!strcasecmp(c->argv[3]->ptr,"migrating") && c->argc == 5) { 3163 if (server.cluster->slots[slot] != myself) { 3164 addReplyErrorFormat(c,"I'm not the owner of hash slot %u",slot); 3165 return; 3166 } 3167 if ((n = clusterLookupNode(c->argv[4]->ptr)) == NULL) { 3168 addReplyErrorFormat(c,"I don't know about node %s", 3169 (char*)c->argv[4]->ptr); 3170 return; 3171 } 3172 server.cluster->migrating_slots_to[slot] = n; 3173 } else if (!strcasecmp(c->argv[3]->ptr,"importing") && c->argc == 5) { 3174 if (server.cluster->slots[slot] == myself) { 3175 addReplyErrorFormat(c, 3176 "I'm already the owner of hash slot %u",slot); 3177 return; 3178 } 3179 if ((n = clusterLookupNode(c->argv[4]->ptr)) == NULL) { 3180 addReplyErrorFormat(c,"I don't know about node %s", 3181 (char*)c->argv[3]->ptr); 3182 return; 3183 } 3184 server.cluster->importing_slots_from[slot] = n; 3185 } else if (!strcasecmp(c->argv[3]->ptr,"stable") && c->argc == 4) { 3186 /* CLUSTER SETSLOT <SLOT> STABLE */ 3187 server.cluster->importing_slots_from[slot] = NULL; 3188 server.cluster->migrating_slots_to[slot] = NULL; 3189 } else if (!strcasecmp(c->argv[3]->ptr,"node") && c->argc == 5) { 3190 /* CLUSTER SETSLOT <SLOT> NODE <NODE ID> */ 3191 clusterNode *n = clusterLookupNode(c->argv[4]->ptr); 3192 3193 if (!n) { 3194 addReplyErrorFormat(c,"Unknown node %s", 3195 (char*)c->argv[4]->ptr); 3196 return; 3197 } 3198 /* If this hash slot was served by 'myself' before to switch 3199 * make sure there are no longer local keys for this hash slot. */ 3200 if (server.cluster->slots[slot] == myself && n != myself) { 3201 if (countKeysInSlot(slot) != 0) { 3202 addReplyErrorFormat(c, "Can't assign hashslot %d to a different node while I still hold keys for this hash slot.", slot); 3203 return; 3204 } 3205 } 3206 /* If this slot is in migrating status but we have no keys 3207 * for it assigning the slot to another node will clear 3208 * the migratig status. */ 3209 if (countKeysInSlot(slot) == 0 && 3210 server.cluster->migrating_slots_to[slot]) 3211 server.cluster->migrating_slots_to[slot] = NULL; 3212 3213 /* If this node was importing this slot, assigning the slot to 3214 * itself also clears the importing status. */ 3215 if (n == myself && 3216 server.cluster->importing_slots_from[slot]) 3217 { 3218 /* This slot was manually migrated, set this node configEpoch 3219 * to a new epoch so that the new version can be propagated 3220 * by the cluster. 3221 * 3222 * FIXME: the new version should be agreed otherwise a race 3223 * is possible if while a manual resharding is in progress 3224 * the master is failed over by a slave. */ 3225 uint64_t maxEpoch = clusterGetMaxEpoch(); 3226 3227 if (myself->configEpoch == 0 || 3228 myself->configEpoch != maxEpoch) 3229 { 3230 server.cluster->currentEpoch++; 3231 myself->configEpoch = server.cluster->currentEpoch; 3232 clusterDoBeforeSleep(CLUSTER_TODO_FSYNC_CONFIG); 3233 } 3234 server.cluster->importing_slots_from[slot] = NULL; 3235 } 3236 clusterDelSlot(slot); 3237 clusterAddSlot(n,slot); 3238 } else { 3239 addReplyError(c,"Invalid CLUSTER SETSLOT action or number of arguments"); 3240 return; 3241 } 3242 clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG|CLUSTER_TODO_UPDATE_STATE); 3243 addReply(c,shared.ok); 3244 } else if (!strcasecmp(c->argv[1]->ptr,"info") && c->argc == 2) { 3245 /* CLUSTER INFO */ 3246 char *statestr[] = {"ok","fail","needhelp"}; 3247 int slots_assigned = 0, slots_ok = 0, slots_pfail = 0, slots_fail = 0; 3248 int j; 3249 3250 for (j = 0; j < REDIS_CLUSTER_SLOTS; j++) { 3251 clusterNode *n = server.cluster->slots[j]; 3252 3253 if (n == NULL) continue; 3254 slots_assigned++; 3255 if (nodeFailed(n)) { 3256 slots_fail++; 3257 } else if (nodeTimedOut(n)) { 3258 slots_pfail++; 3259 } else { 3260 slots_ok++; 3261 } 3262 } 3263 3264 sds info = sdscatprintf(sdsempty(), 3265 "cluster_state:%s\r\n" 3266 "cluster_slots_assigned:%d\r\n" 3267 "cluster_slots_ok:%d\r\n" 3268 "cluster_slots_pfail:%d\r\n" 3269 "cluster_slots_fail:%d\r\n" 3270 "cluster_known_nodes:%lu\r\n" 3271 "cluster_size:%d\r\n" 3272 "cluster_current_epoch:%llu\r\n" 3273 "cluster_stats_messages_sent:%lld\r\n" 3274 "cluster_stats_messages_received:%lld\r\n" 3275 , statestr[server.cluster->state], 3276 slots_assigned, 3277 slots_ok, 3278 slots_pfail, 3279 slots_fail, 3280 dictSize(server.cluster->nodes), 3281 server.cluster->size, 3282 (unsigned long long) server.cluster->currentEpoch, 3283 server.cluster->stats_bus_messages_sent, 3284 server.cluster->stats_bus_messages_received 3285 ); 3286 addReplySds(c,sdscatprintf(sdsempty(),"$%lu\r\n", 3287 (unsigned long)sdslen(info))); 3288 addReplySds(c,info); 3289 addReply(c,shared.crlf); 3290 } else if (!strcasecmp(c->argv[1]->ptr,"saveconfig") && c->argc == 2) { 3291 int retval = clusterSaveConfig(1); 3292 3293 if (retval == 0) 3294 addReply(c,shared.ok); 3295 else 3296 addReplyErrorFormat(c,"error saving the cluster node config: %s", 3297 strerror(errno)); 3298 } else if (!strcasecmp(c->argv[1]->ptr,"keyslot") && c->argc == 3) { 3299 /* CLUSTER KEYSLOT <key> */ 3300 sds key = c->argv[2]->ptr; 3301 3302 addReplyLongLong(c,keyHashSlot(key,sdslen(key))); 3303 } else if (!strcasecmp(c->argv[1]->ptr,"countkeysinslot") && c->argc == 3) { 3304 /* CLUSTER COUNTKEYSINSLOT <slot> */ 3305 long long slot; 3306 3307 if (getLongLongFromObjectOrReply(c,c->argv[2],&slot,NULL) != REDIS_OK) 3308 return; 3309 if (slot < 0 || slot >= REDIS_CLUSTER_SLOTS) { 3310 addReplyError(c,"Invalid slot"); 3311 return; 3312 } 3313 addReplyLongLong(c,countKeysInSlot(slot)); 3314 } else if (!strcasecmp(c->argv[1]->ptr,"getkeysinslot") && c->argc == 4) { 3315 /* CLUSTER GETKEYSINSLOT <slot> <count> */ 3316 long long maxkeys, slot; 3317 unsigned int numkeys, j; 3318 robj **keys; 3319 3320 if (getLongLongFromObjectOrReply(c,c->argv[2],&slot,NULL) != REDIS_OK) 3321 return; 3322 if (getLongLongFromObjectOrReply(c,c->argv[3],&maxkeys,NULL) != REDIS_OK) 3323 return; 3324 if (slot < 0 || slot >= REDIS_CLUSTER_SLOTS || maxkeys < 0) { 3325 addReplyError(c,"Invalid slot or number of keys"); 3326 return; 3327 } 3328 3329 keys = zmalloc(sizeof(robj*)*maxkeys); 3330 numkeys = getKeysInSlot(slot, keys, maxkeys); 3331 addReplyMultiBulkLen(c,numkeys); 3332 for (j = 0; j < numkeys; j++) addReplyBulk(c,keys[j]); 3333 zfree(keys); 3334 } else if (!strcasecmp(c->argv[1]->ptr,"forget") && c->argc == 3) { 3335 /* CLUSTER FORGET <NODE ID> */ 3336 clusterNode *n = clusterLookupNode(c->argv[2]->ptr); 3337 3338 if (!n) { 3339 addReplyErrorFormat(c,"Unknown node %s", (char*)c->argv[2]->ptr); 3340 return; 3341 } else if (n == myself) { 3342 addReplyError(c,"I tried hard but I can't forget myself..."); 3343 return; 3344 } else if (nodeIsSlave(myself) && myself->slaveof == n) { 3345 addReplyError(c,"Can't forget my master!"); 3346 return; 3347 } 3348 clusterBlacklistAddNode(n); 3349 clusterDelNode(n); 3350 clusterDoBeforeSleep(CLUSTER_TODO_UPDATE_STATE|CLUSTER_TODO_SAVE_CONFIG); 3351 addReply(c,shared.ok); 3352 } else if (!strcasecmp(c->argv[1]->ptr,"replicate") && c->argc == 3) { 3353 /* CLUSTER REPLICATE <NODE ID> */ 3354 clusterNode *n = clusterLookupNode(c->argv[2]->ptr); 3355 3356 /* Lookup the specified node in our table. */ 3357 if (!n) { 3358 addReplyErrorFormat(c,"Unknown node %s", (char*)c->argv[2]->ptr); 3359 return; 3360 } 3361 3362 /* I can't replicate myself. */ 3363 if (n == myself) { 3364 addReplyError(c,"Can't replicate myself"); 3365 return; 3366 } 3367 3368 /* Can't replicate a slave. */ 3369 if (n->slaveof != NULL) { 3370 addReplyError(c,"I can only replicate a master, not a slave."); 3371 return; 3372 } 3373 3374 /* If the instance is currently a master, it should have no assigned 3375 * slots nor keys to accept to replicate some other node. 3376 * Slaves can switch to another master without issues. */ 3377 if (nodeIsMaster(myself) && 3378 (myself->numslots != 0 || dictSize(server.db[0].dict) != 0)) { 3379 addReplyError(c,"To set a master the node must be empty and without assigned slots."); 3380 return; 3381 } 3382 3383 /* Set the master. */ 3384 clusterSetMaster(n); 3385 clusterDoBeforeSleep(CLUSTER_TODO_UPDATE_STATE|CLUSTER_TODO_SAVE_CONFIG); 3386 addReply(c,shared.ok); 3387 } else if (!strcasecmp(c->argv[1]->ptr,"slaves") && c->argc == 3) { 3388 /* CLUSTER SLAVES <NODE ID> */ 3389 clusterNode *n = clusterLookupNode(c->argv[2]->ptr); 3390 int j; 3391 3392 /* Lookup the specified node in our table. */ 3393 if (!n) { 3394 addReplyErrorFormat(c,"Unknown node %s", (char*)c->argv[2]->ptr); 3395 return; 3396 } 3397 3398 if (nodeIsSlave(n)) { 3399 addReplyError(c,"The specified node is not a master"); 3400 return; 3401 } 3402 3403 addReplyMultiBulkLen(c,n->numslaves); 3404 for (j = 0; j < n->numslaves; j++) { 3405 sds ni = clusterGenNodeDescription(n->slaves[j]); 3406 addReplyBulkCString(c,ni); 3407 sdsfree(ni); 3408 } 3409 } else if (!strcasecmp(c->argv[1]->ptr,"failover") && c->argc == 2) { 3410 if (nodeIsMaster(myself)) { 3411 addReplyError(c,"You should send CLUSTER FAILOVER to a slave"); 3412 return; 3413 } else if (myself->slaveof == NULL || nodeFailed(myself->slaveof) || 3414 myself->slaveof->link == NULL) 3415 { 3416 addReplyError(c,"Master is down or failed, " 3417 "please use CLUSTER FAILOVER FORCE"); 3418 return; 3419 } 3420 resetManualFailover(); 3421 server.cluster->mf_end = mstime() + REDIS_CLUSTER_MF_TIMEOUT; 3422 clusterSendMFStart(myself->slaveof); 3423 redisLog(REDIS_WARNING,"Manual failover user request accepted."); 3424 addReply(c,shared.ok); 3425 } else { 3426 addReplyError(c,"Wrong CLUSTER subcommand or number of arguments"); 3427 } 3428 } 3429 3430 /* ----------------------------------------------------------------------------- 3431 * DUMP, RESTORE and MIGRATE commands 3432 * -------------------------------------------------------------------------- */ 3433 3434 /* Generates a DUMP-format representation of the object 'o', adding it to the 3435 * io stream pointed by 'rio'. This function can't fail. */ 3436 void createDumpPayload(rio *payload, robj *o) { 3437 unsigned char buf[2]; 3438 uint64_t crc; 3439 3440 /* Serialize the object in a RDB-like format. It consist of an object type 3441 * byte followed by the serialized object. This is understood by RESTORE. */ 3442 rioInitWithBuffer(payload,sdsempty()); 3443 redisAssert(rdbSaveObjectType(payload,o)); 3444 redisAssert(rdbSaveObject(payload,o)); 3445 3446 /* Write the footer, this is how it looks like: 3447 * ----------------+---------------------+---------------+ 3448 * ... RDB payload | 2 bytes RDB version | 8 bytes CRC64 | 3449 * ----------------+---------------------+---------------+ 3450 * RDB version and CRC are both in little endian. 3451 */ 3452 3453 /* RDB version */ 3454 buf[0] = REDIS_RDB_VERSION & 0xff; 3455 buf[1] = (REDIS_RDB_VERSION >> 8) & 0xff; 3456 payload->io.buffer.ptr = sdscatlen(payload->io.buffer.ptr,buf,2); 3457 3458 /* CRC64 */ 3459 crc = crc64(0,(unsigned char*)payload->io.buffer.ptr, 3460 sdslen(payload->io.buffer.ptr)); 3461 memrev64ifbe(&crc); 3462 payload->io.buffer.ptr = sdscatlen(payload->io.buffer.ptr,&crc,8); 3463 } 3464 3465 /* Verify that the RDB version of the dump payload matches the one of this Redis 3466 * instance and that the checksum is ok. 3467 * If the DUMP payload looks valid REDIS_OK is returned, otherwise REDIS_ERR 3468 * is returned. */ 3469 int verifyDumpPayload(unsigned char *p, size_t len) { 3470 unsigned char *footer; 3471 uint16_t rdbver; 3472 uint64_t crc; 3473 3474 /* At least 2 bytes of RDB version and 8 of CRC64 should be present. */ 3475 if (len < 10) return REDIS_ERR; 3476 footer = p+(len-10); 3477 3478 /* Verify RDB version */ 3479 rdbver = (footer[1] << 8) | footer[0]; 3480 if (rdbver != REDIS_RDB_VERSION) return REDIS_ERR; 3481 3482 /* Verify CRC64 */ 3483 crc = crc64(0,p,len-8); 3484 memrev64ifbe(&crc); 3485 return (memcmp(&crc,footer+2,8) == 0) ? REDIS_OK : REDIS_ERR; 3486 } 3487 3488 /* DUMP keyname 3489 * DUMP is actually not used by Redis Cluster but it is the obvious 3490 * complement of RESTORE and can be useful for different applications. */ 3491 void dumpCommand(redisClient *c) { 3492 robj *o, *dumpobj; 3493 rio payload; 3494 3495 /* Check if the key is here. */ 3496 if ((o = lookupKeyRead(c->db,c->argv[1])) == NULL) { 3497 addReply(c,shared.nullbulk); 3498 return; 3499 } 3500 3501 /* Create the DUMP encoded representation. */ 3502 createDumpPayload(&payload,o); 3503 3504 /* Transfer to the client */ 3505 dumpobj = createObject(REDIS_STRING,payload.io.buffer.ptr); 3506 addReplyBulk(c,dumpobj); 3507 decrRefCount(dumpobj); 3508 return; 3509 } 3510 3511 /* RESTORE key ttl serialized-value [REPLACE] */ 3512 void restoreCommand(redisClient *c) { 3513 long long ttl; 3514 rio payload; 3515 int j, type, replace = 0; 3516 robj *obj; 3517 3518 /* Parse additional options */ 3519 for (j = 4; j < c->argc; j++) { 3520 if (!strcasecmp(c->argv[j]->ptr,"replace")) { 3521 replace = 1; 3522 } else { 3523 addReply(c,shared.syntaxerr); 3524 return; 3525 } 3526 } 3527 3528 /* Make sure this key does not already exist here... */ 3529 if (!replace && lookupKeyWrite(c->db,c->argv[1]) != NULL) { 3530 addReplyError(c,"Target key name is busy."); 3531 return; 3532 } 3533 3534 /* Check if the TTL value makes sense */ 3535 if (getLongLongFromObjectOrReply(c,c->argv[2],&ttl,NULL) != REDIS_OK) { 3536 return; 3537 } else if (ttl < 0) { 3538 addReplyError(c,"Invalid TTL value, must be >= 0"); 3539 return; 3540 } 3541 3542 /* Verify RDB version and data checksum. */ 3543 if (verifyDumpPayload(c->argv[3]->ptr,sdslen(c->argv[3]->ptr)) == REDIS_ERR) 3544 { 3545 addReplyError(c,"DUMP payload version or checksum are wrong"); 3546 return; 3547 } 3548 3549 rioInitWithBuffer(&payload,c->argv[3]->ptr); 3550 if (((type = rdbLoadObjectType(&payload)) == -1) || 3551 ((obj = rdbLoadObject(type,&payload)) == NULL)) 3552 { 3553 addReplyError(c,"Bad data format"); 3554 return; 3555 } 3556 3557 /* Remove the old key if needed. */ 3558 if (replace) dbDelete(c->db,c->argv[1]); 3559 3560 /* Create the key and set the TTL if any */ 3561 dbAdd(c->db,c->argv[1],obj); 3562 if (ttl) setExpire(c->db,c->argv[1],mstime()+ttl); 3563 signalModifiedKey(c->db,c->argv[1]); 3564 addReply(c,shared.ok); 3565 server.dirty++; 3566 } 3567 3568 /* MIGRATE socket cache implementation. 3569 * 3570 * We take a map between host:ip and a TCP socket that we used to connect 3571 * to this instance in recent time. 3572 * This sockets are closed when the max number we cache is reached, and also 3573 * in serverCron() when they are around for more than a few seconds. */ 3574 #define MIGRATE_SOCKET_CACHE_ITEMS 64 /* max num of items in the cache. */ 3575 #define MIGRATE_SOCKET_CACHE_TTL 10 /* close cached socekts after 10 sec. */ 3576 3577 typedef struct migrateCachedSocket { 3578 int fd; 3579 time_t last_use_time; 3580 } migrateCachedSocket; 3581 3582 /* Return a TCP scoket connected with the target instance, possibly returning 3583 * a cached one. 3584 * 3585 * This function is responsible of sending errors to the client if a 3586 * connection can't be established. In this case -1 is returned. 3587 * Otherwise on success the socket is returned, and the caller should not 3588 * attempt to free it after usage. 3589 * 3590 * If the caller detects an error while using the socket, migrateCloseSocket() 3591 * should be called so that the connection will be craeted from scratch 3592 * the next time. */ 3593 int migrateGetSocket(redisClient *c, robj *host, robj *port, long timeout) { 3594 int fd; 3595 sds name = sdsempty(); 3596 migrateCachedSocket *cs; 3597 3598 /* Check if we have an already cached socket for this ip:port pair. */ 3599 name = sdscatlen(name,host->ptr,sdslen(host->ptr)); 3600 name = sdscatlen(name,":",1); 3601 name = sdscatlen(name,port->ptr,sdslen(port->ptr)); 3602 cs = dictFetchValue(server.migrate_cached_sockets,name); 3603 if (cs) { 3604 sdsfree(name); 3605 cs->last_use_time = server.unixtime; 3606 return cs->fd; 3607 } 3608 3609 /* No cached socket, create one. */ 3610 if (dictSize(server.migrate_cached_sockets) == MIGRATE_SOCKET_CACHE_ITEMS) { 3611 /* Too many items, drop one at random. */ 3612 dictEntry *de = dictGetRandomKey(server.migrate_cached_sockets); 3613 cs = dictGetVal(de); 3614 close(cs->fd); 3615 zfree(cs); 3616 dictDelete(server.migrate_cached_sockets,dictGetKey(de)); 3617 } 3618 3619 /* Create the socket */ 3620 fd = anetTcpNonBlockConnect(server.neterr,c->argv[1]->ptr, 3621 atoi(c->argv[2]->ptr)); 3622 if (fd == -1) { 3623 sdsfree(name); 3624 addReplyErrorFormat(c,"Can't connect to target node: %s", 3625 server.neterr); 3626 return -1; 3627 } 3628 anetEnableTcpNoDelay(server.neterr,fd); 3629 3630 /* Check if it connects within the specified timeout. */ 3631 if ((aeWait(fd,AE_WRITABLE,timeout) & AE_WRITABLE) == 0) { 3632 sdsfree(name); 3633 addReplySds(c,sdsnew("-IOERR error or timeout connecting to the client\r\n")); 3634 close(fd); 3635 return -1; 3636 } 3637 3638 /* Add to the cache and return it to the caller. */ 3639 cs = zmalloc(sizeof(*cs)); 3640 cs->fd = fd; 3641 cs->last_use_time = server.unixtime; 3642 dictAdd(server.migrate_cached_sockets,name,cs); 3643 return fd; 3644 } 3645 3646 /* Free a migrate cached connection. */ 3647 void migrateCloseSocket(robj *host, robj *port) { 3648 sds name = sdsempty(); 3649 migrateCachedSocket *cs; 3650 3651 name = sdscatlen(name,host->ptr,sdslen(host->ptr)); 3652 name = sdscatlen(name,":",1); 3653 name = sdscatlen(name,port->ptr,sdslen(port->ptr)); 3654 cs = dictFetchValue(server.migrate_cached_sockets,name); 3655 if (!cs) { 3656 sdsfree(name); 3657 return; 3658 } 3659 3660 close(cs->fd); 3661 zfree(cs); 3662 dictDelete(server.migrate_cached_sockets,name); 3663 sdsfree(name); 3664 } 3665 3666 void migrateCloseTimedoutSockets(void) { 3667 dictIterator *di = dictGetSafeIterator(server.migrate_cached_sockets); 3668 dictEntry *de; 3669 3670 while((de = dictNext(di)) != NULL) { 3671 migrateCachedSocket *cs = dictGetVal(de); 3672 3673 if ((server.unixtime - cs->last_use_time) > MIGRATE_SOCKET_CACHE_TTL) { 3674 close(cs->fd); 3675 zfree(cs); 3676 dictDelete(server.migrate_cached_sockets,dictGetKey(de)); 3677 } 3678 } 3679 dictReleaseIterator(di); 3680 } 3681 3682 /* MIGRATE host port key dbid timeout [COPY | REPLACE] */ 3683 void migrateCommand(redisClient *c) { 3684 int fd, copy, replace, j; 3685 long timeout; 3686 long dbid; 3687 long long ttl, expireat; 3688 robj *o; 3689 rio cmd, payload; 3690 int retry_num = 0; 3691 3692 try_again: 3693 /* Initialization */ 3694 copy = 0; 3695 replace = 0; 3696 ttl = 0; 3697 3698 /* Parse additional options */ 3699 for (j = 6; j < c->argc; j++) { 3700 if (!strcasecmp(c->argv[j]->ptr,"copy")) { 3701 copy = 1; 3702 } else if (!strcasecmp(c->argv[j]->ptr,"replace")) { 3703 replace = 1; 3704 } else { 3705 addReply(c,shared.syntaxerr); 3706 return; 3707 } 3708 } 3709 3710 /* Sanity check */ 3711 if (getLongFromObjectOrReply(c,c->argv[5],&timeout,NULL) != REDIS_OK) 3712 return; 3713 if (getLongFromObjectOrReply(c,c->argv[4],&dbid,NULL) != REDIS_OK) 3714 return; 3715 if (timeout <= 0) timeout = 1000; 3716 3717 /* Check if the key is here. If not we reply with success as there is 3718 * nothing to migrate (for instance the key expired in the meantime), but 3719 * we include such information in the reply string. */ 3720 if ((o = lookupKeyRead(c->db,c->argv[3])) == NULL) { 3721 addReplySds(c,sdsnew("+NOKEY\r\n")); 3722 return; 3723 } 3724 3725 /* Connect */ 3726 fd = migrateGetSocket(c,c->argv[1],c->argv[2],timeout); 3727 if (fd == -1) return; /* error sent to the client by migrateGetSocket() */ 3728 3729 /* Create RESTORE payload and generate the protocol to call the command. */ 3730 rioInitWithBuffer(&cmd,sdsempty()); 3731 redisAssertWithInfo(c,NULL,rioWriteBulkCount(&cmd,'*',2)); 3732 redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,"SELECT",6)); 3733 redisAssertWithInfo(c,NULL,rioWriteBulkLongLong(&cmd,dbid)); 3734 3735 expireat = getExpire(c->db,c->argv[3]); 3736 if (expireat != -1) { 3737 ttl = expireat-mstime(); 3738 if (ttl < 1) ttl = 1; 3739 } 3740 redisAssertWithInfo(c,NULL,rioWriteBulkCount(&cmd,'*',replace ? 5 : 4)); 3741 if (server.cluster_enabled) 3742 redisAssertWithInfo(c,NULL, 3743 rioWriteBulkString(&cmd,"RESTORE-ASKING",14)); 3744 else 3745 redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,"RESTORE",7)); 3746 redisAssertWithInfo(c,NULL,sdsEncodedObject(c->argv[3])); 3747 redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,c->argv[3]->ptr,sdslen(c->argv[3]->ptr))); 3748 redisAssertWithInfo(c,NULL,rioWriteBulkLongLong(&cmd,ttl)); 3749 3750 /* Emit the payload argument, that is the serialized object using 3751 * the DUMP format. */ 3752 createDumpPayload(&payload,o); 3753 redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,payload.io.buffer.ptr, 3754 sdslen(payload.io.buffer.ptr))); 3755 sdsfree(payload.io.buffer.ptr); 3756 3757 /* Add the REPLACE option to the RESTORE command if it was specified 3758 * as a MIGRATE option. */ 3759 if (replace) 3760 redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,"REPLACE",7)); 3761 3762 /* Transfer the query to the other node in 64K chunks. */ 3763 errno = 0; 3764 { 3765 sds buf = cmd.io.buffer.ptr; 3766 size_t pos = 0, towrite; 3767 int nwritten = 0; 3768 3769 while ((towrite = sdslen(buf)-pos) > 0) { 3770 towrite = (towrite > (64*1024) ? (64*1024) : towrite); 3771 nwritten = syncWrite(fd,buf+pos,towrite,timeout); 3772 if (nwritten != (signed)towrite) goto socket_wr_err; 3773 pos += nwritten; 3774 } 3775 } 3776 3777 /* Read back the reply. */ 3778 { 3779 char buf1[1024]; 3780 char buf2[1024]; 3781 3782 /* Read the two replies */ 3783 if (syncReadLine(fd, buf1, sizeof(buf1), timeout) <= 0) 3784 goto socket_rd_err; 3785 if (syncReadLine(fd, buf2, sizeof(buf2), timeout) <= 0) 3786 goto socket_rd_err; 3787 if (buf1[0] == '-' || buf2[0] == '-') { 3788 addReplyErrorFormat(c,"Target instance replied with error: %s", 3789 (buf1[0] == '-') ? buf1+1 : buf2+1); 3790 } else { 3791 robj *aux; 3792 3793 if (!copy) { 3794 /* No COPY option: remove the local key, signal the change. */ 3795 dbDelete(c->db,c->argv[3]); 3796 signalModifiedKey(c->db,c->argv[3]); 3797 } 3798 addReply(c,shared.ok); 3799 server.dirty++; 3800 3801 /* Translate MIGRATE as DEL for replication/AOF. */ 3802 aux = createStringObject("DEL",3); 3803 rewriteClientCommandVector(c,2,aux,c->argv[3]); 3804 decrRefCount(aux); 3805 } 3806 } 3807 3808 sdsfree(cmd.io.buffer.ptr); 3809 return; 3810 3811 socket_wr_err: 3812 sdsfree(cmd.io.buffer.ptr); 3813 migrateCloseSocket(c->argv[1],c->argv[2]); 3814 if (errno != ETIMEDOUT && retry_num++ == 0) goto try_again; 3815 addReplySds(c, 3816 sdsnew("-IOERR error or timeout writing to target instance\r\n")); 3817 return; 3818 3819 socket_rd_err: 3820 sdsfree(cmd.io.buffer.ptr); 3821 migrateCloseSocket(c->argv[1],c->argv[2]); 3822 if (errno != ETIMEDOUT && retry_num++ == 0) goto try_again; 3823 addReplySds(c, 3824 sdsnew("-IOERR error or timeout reading from target node\r\n")); 3825 return; 3826 } 3827 3828 /* ----------------------------------------------------------------------------- 3829 * Cluster functions related to serving / redirecting clients 3830 * -------------------------------------------------------------------------- */ 3831 3832 /* The ASKING command is required after a -ASK redirection. 3833 * The client should issue ASKING before to actually send the command to 3834 * the target instance. See the Redis Cluster specification for more 3835 * information. */ 3836 void askingCommand(redisClient *c) { 3837 if (server.cluster_enabled == 0) { 3838 addReplyError(c,"This instance has cluster support disabled"); 3839 return; 3840 } 3841 c->flags |= REDIS_ASKING; 3842 addReply(c,shared.ok); 3843 } 3844 3845 /* The READONLY command is uesd by clients to enter the read-only mode. 3846 * In this mode slaves will not redirect clients as long as clients access 3847 * with read-only commands to keys that are served by the slave's master. */ 3848 void readonlyCommand(redisClient *c) { 3849 if (server.cluster_enabled == 0) { 3850 addReplyError(c,"This instance has cluster support disabled"); 3851 return; 3852 } 3853 c->flags |= REDIS_READONLY; 3854 addReply(c,shared.ok); 3855 } 3856 3857 /* The READWRITE command just clears the READONLY command state. */ 3858 void readwriteCommand(redisClient *c) { 3859 c->flags &= ~REDIS_READONLY; 3860 addReply(c,shared.ok); 3861 } 3862 3863 /* Return the pointer to the cluster node that is able to serve the command. 3864 * For the function to succeed the command should only target either: 3865 * 3866 * 1) A single key (even multiple times like LPOPRPUSH mylist mylist). 3867 * 2) Multiple keys in the same hash slot, while the slot is stable (no 3868 * resharding in progress). 3869 * 3870 * On success the function returns the node that is able to serve the request. 3871 * If the node is not 'myself' a redirection must be perfomed. The kind of 3872 * redirection is specified setting the integer passed by reference 3873 * 'error_code', which will be set to REDIS_CLUSTER_REDIR_ASK or 3874 * REDIS_CLUSTER_REDIR_MOVED. 3875 * 3876 * When the node is 'myself' 'error_code' is set to REDIS_CLUSTER_REDIR_NONE. 3877 * 3878 * If the command fails NULL is returned, and the reason of the failure is 3879 * provided via 'error_code', which will be set to: 3880 * 3881 * REDIS_CLUSTER_REDIR_CROSS_SLOT if the request contains multiple keys that 3882 * don't belong to the same hash slot. 3883 * 3884 * REDIS_CLUSTER_REDIR_UNSTABLE if the request contains mutliple keys 3885 * belonging to the same slot, but the slot is not stable (in migration or 3886 * importing state, likely because a resharding is in progress). */ 3887 clusterNode *getNodeByQuery(redisClient *c, struct redisCommand *cmd, robj **argv, int argc, int *hashslot, int *error_code) { 3888 clusterNode *n = NULL; 3889 robj *firstkey = NULL; 3890 int multiple_keys = 0; 3891 multiState *ms, _ms; 3892 multiCmd mc; 3893 int i, slot = 0, migrating_slot = 0, importing_slot = 0, missing_keys = 0; 3894 3895 /* Set error code optimistically for the base case. */ 3896 if (error_code) *error_code = REDIS_CLUSTER_REDIR_NONE; 3897 3898 /* We handle all the cases as if they were EXEC commands, so we have 3899 * a common code path for everything */ 3900 if (cmd->proc == execCommand) { 3901 /* If REDIS_MULTI flag is not set EXEC is just going to return an 3902 * error. */ 3903 if (!(c->flags & REDIS_MULTI)) return myself; 3904 ms = &c->mstate; 3905 } else { 3906 /* In order to have a single codepath create a fake Multi State 3907 * structure if the client is not in MULTI/EXEC state, this way 3908 * we have a single codepath below. */ 3909 ms = &_ms; 3910 _ms.commands = &mc; 3911 _ms.count = 1; 3912 mc.argv = argv; 3913 mc.argc = argc; 3914 mc.cmd = cmd; 3915 } 3916 3917 /* Check that all the keys are in the same hash slot, and obtain this 3918 * slot and the node associated. */ 3919 for (i = 0; i < ms->count; i++) { 3920 struct redisCommand *mcmd; 3921 robj **margv; 3922 int margc, *keyindex, numkeys, j; 3923 3924 mcmd = ms->commands[i].cmd; 3925 margc = ms->commands[i].argc; 3926 margv = ms->commands[i].argv; 3927 3928 keyindex = getKeysFromCommand(mcmd,margv,margc,&numkeys); 3929 for (j = 0; j < numkeys; j++) { 3930 robj *thiskey = margv[keyindex[j]]; 3931 int thisslot = keyHashSlot((char*)thiskey->ptr, 3932 sdslen(thiskey->ptr)); 3933 3934 if (firstkey == NULL) { 3935 /* This is the first key we see. Check what is the slot 3936 * and node. */ 3937 firstkey = thiskey; 3938 slot = thisslot; 3939 n = server.cluster->slots[slot]; 3940 redisAssertWithInfo(c,firstkey,n != NULL); 3941 /* If we are migrating or importing this slot, we need to check 3942 * if we have all the keys in the request (the only way we 3943 * can safely serve the request, otherwise we return a TRYAGAIN 3944 * error). To do so we set the importing/migrating state and 3945 * increment a counter for every missing key. */ 3946 if (n == myself && 3947 server.cluster->migrating_slots_to[slot] != NULL) 3948 { 3949 migrating_slot = 1; 3950 } else if (server.cluster->importing_slots_from[slot] != NULL) { 3951 importing_slot = 1; 3952 } 3953 } else { 3954 /* If it is not the first key, make sure it is exactly 3955 * the same key as the first we saw. */ 3956 if (!equalStringObjects(firstkey,thiskey)) { 3957 if (slot != thisslot) { 3958 /* Error: multiple keys from different slots. */ 3959 getKeysFreeResult(keyindex); 3960 if (error_code) 3961 *error_code = REDIS_CLUSTER_REDIR_CROSS_SLOT; 3962 return NULL; 3963 } else { 3964 /* Flag this request as one with multiple different 3965 * keys. */ 3966 multiple_keys = 1; 3967 } 3968 } 3969 } 3970 3971 /* Migarting / Improrting slot? Count keys we don't have. */ 3972 if ((migrating_slot || importing_slot) && 3973 lookupKeyRead(&server.db[0],thiskey) == NULL) 3974 { 3975 missing_keys++; 3976 } 3977 } 3978 getKeysFreeResult(keyindex); 3979 } 3980 3981 /* No key at all in command? then we can serve the request 3982 * without redirections or errors. */ 3983 if (n == NULL) return myself; 3984 3985 /* Return the hashslot by reference. */ 3986 if (hashslot) *hashslot = slot; 3987 3988 /* This request is about a slot we are migrating into another instance? 3989 * Then if we have all the keys. */ 3990 3991 /* If we don't have all the keys and we are migrating the slot, send 3992 * an ASK redirection. */ 3993 if (migrating_slot && missing_keys) { 3994 if (error_code) *error_code = REDIS_CLUSTER_REDIR_ASK; 3995 return server.cluster->migrating_slots_to[slot]; 3996 } 3997 3998 /* If we are receiving the slot, and the client correctly flagged the 3999 * request as "ASKING", we can serve the request. However if the request 4000 * involves multiple keys and we don't have them all, the only option is 4001 * to send a TRYAGAIN error. */ 4002 if (importing_slot && 4003 (c->flags & REDIS_ASKING || cmd->flags & REDIS_CMD_ASKING)) 4004 { 4005 if (multiple_keys && missing_keys) { 4006 if (error_code) *error_code = REDIS_CLUSTER_REDIR_UNSTABLE; 4007 return NULL; 4008 } else { 4009 return myself; 4010 } 4011 } 4012 4013 /* Handle the read-only client case reading from a slave: if this 4014 * node is a slave and the request is about an hash slot our master 4015 * is serving, we can reply without redirection. */ 4016 if (c->flags & REDIS_READONLY && 4017 cmd->flags & REDIS_CMD_READONLY && 4018 nodeIsSlave(myself) && 4019 myself->slaveof == n) 4020 { 4021 return myself; 4022 } 4023 4024 /* Base case: just return the right node. However if this node is not 4025 * myself, set error_code to MOVED since we need to issue a rediretion. */ 4026 if (n != myself && error_code) *error_code = REDIS_CLUSTER_REDIR_MOVED; 4027 return n; 4028 } 4029