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