1 /* Asynchronous replication 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
32 #include "server.h"
33 #include "cluster.h"
34 #include "bio.h"
35
36 #include <memory.h>
37 #include <sys/time.h>
38 #include <unistd.h>
39 #include <fcntl.h>
40 #include <sys/socket.h>
41 #include <sys/stat.h>
42
43 void replicationDiscardCachedMaster(void);
44 void replicationResurrectCachedMaster(connection *conn);
45 void replicationSendAck(void);
46 void putSlaveOnline(client *slave);
47 int cancelReplicationHandshake(int reconnect);
48
49 /* We take a global flag to remember if this instance generated an RDB
50 * because of replication, so that we can remove the RDB file in case
51 * the instance is configured to have no persistence. */
52 int RDBGeneratedByReplication = 0;
53
54 /* --------------------------- Utility functions ---------------------------- */
55
56 /* Return the pointer to a string representing the slave ip:listening_port
57 * pair. Mostly useful for logging, since we want to log a slave using its
58 * IP address and its listening port which is more clear for the user, for
59 * example: "Closing connection with replica 10.1.2.3:6380". */
replicationGetSlaveName(client * c)60 char *replicationGetSlaveName(client *c) {
61 static char buf[NET_HOST_PORT_STR_LEN];
62 char ip[NET_IP_STR_LEN];
63
64 ip[0] = '\0';
65 buf[0] = '\0';
66 if (c->slave_addr ||
67 connPeerToString(c->conn,ip,sizeof(ip),NULL) != -1)
68 {
69 char *addr = c->slave_addr ? c->slave_addr : ip;
70 if (c->slave_listening_port)
71 anetFormatAddr(buf,sizeof(buf),addr,c->slave_listening_port);
72 else
73 snprintf(buf,sizeof(buf),"%s:<unknown-replica-port>",addr);
74 } else {
75 snprintf(buf,sizeof(buf),"client id #%llu",
76 (unsigned long long) c->id);
77 }
78 return buf;
79 }
80
81 /* Plain unlink() can block for quite some time in order to actually apply
82 * the file deletion to the filesystem. This call removes the file in a
83 * background thread instead. We actually just do close() in the thread,
84 * by using the fact that if there is another instance of the same file open,
85 * the foreground unlink() will only remove the fs name, and deleting the
86 * file's storage space will only happen once the last reference is lost. */
bg_unlink(const char * filename)87 int bg_unlink(const char *filename) {
88 int fd = open(filename,O_RDONLY|O_NONBLOCK);
89 if (fd == -1) {
90 /* Can't open the file? Fall back to unlinking in the main thread. */
91 return unlink(filename);
92 } else {
93 /* The following unlink() removes the name but doesn't free the
94 * file contents because a process still has it open. */
95 int retval = unlink(filename);
96 if (retval == -1) {
97 /* If we got an unlink error, we just return it, closing the
98 * new reference we have to the file. */
99 int old_errno = errno;
100 close(fd); /* This would overwrite our errno. So we saved it. */
101 errno = old_errno;
102 return -1;
103 }
104 bioCreateCloseJob(fd);
105 return 0; /* Success. */
106 }
107 }
108
109 /* ---------------------------------- MASTER -------------------------------- */
110
createReplicationBacklog(void)111 void createReplicationBacklog(void) {
112 serverAssert(server.repl_backlog == NULL);
113 server.repl_backlog = zmalloc(sizeof(replBacklog));
114 server.repl_backlog->ref_repl_buf_node = NULL;
115 server.repl_backlog->unindexed_count = 0;
116 server.repl_backlog->blocks_index = raxNew();
117 server.repl_backlog->histlen = 0;
118 /* We don't have any data inside our buffer, but virtually the first
119 * byte we have is the next byte that will be generated for the
120 * replication stream. */
121 server.repl_backlog->offset = server.master_repl_offset+1;
122 }
123
124 /* This function is called when the user modifies the replication backlog
125 * size at runtime. It is up to the function to resize the buffer and setup it
126 * so that it contains the same data as the previous one (possibly less data,
127 * but the most recent bytes, or the same data and more free space in case the
128 * buffer is enlarged). */
resizeReplicationBacklog(void)129 void resizeReplicationBacklog(void) {
130 if (server.repl_backlog_size < CONFIG_REPL_BACKLOG_MIN_SIZE)
131 server.repl_backlog_size = CONFIG_REPL_BACKLOG_MIN_SIZE;
132 if (server.repl_backlog)
133 incrementalTrimReplicationBacklog(REPL_BACKLOG_TRIM_BLOCKS_PER_CALL);
134 }
135
freeReplicationBacklog(void)136 void freeReplicationBacklog(void) {
137 serverAssert(listLength(server.slaves) == 0);
138 if (server.repl_backlog == NULL) return;
139
140 /* Decrease the start buffer node reference count. */
141 if (server.repl_backlog->ref_repl_buf_node) {
142 replBufBlock *o = listNodeValue(
143 server.repl_backlog->ref_repl_buf_node);
144 serverAssert(o->refcount == 1); /* Last reference. */
145 o->refcount--;
146 }
147
148 /* Replication buffer blocks are completely released when we free the
149 * backlog, since the backlog is released only when there are no replicas
150 * and the backlog keeps the last reference of all blocks. */
151 freeReplicationBacklogRefMemAsync(server.repl_buffer_blocks,
152 server.repl_backlog->blocks_index);
153 resetReplicationBuffer();
154 zfree(server.repl_backlog);
155 server.repl_backlog = NULL;
156 }
157
158 /* To make search offset from replication buffer blocks quickly
159 * when replicas ask partial resynchronization, we create one index
160 * block every REPL_BACKLOG_INDEX_PER_BLOCKS blocks. */
createReplicationBacklogIndex(listNode * ln)161 void createReplicationBacklogIndex(listNode *ln) {
162 server.repl_backlog->unindexed_count++;
163 if (server.repl_backlog->unindexed_count >= REPL_BACKLOG_INDEX_PER_BLOCKS) {
164 replBufBlock *o = listNodeValue(ln);
165 uint64_t encoded_offset = htonu64(o->repl_offset);
166 raxInsert(server.repl_backlog->blocks_index,
167 (unsigned char*)&encoded_offset, sizeof(uint64_t),
168 ln, NULL);
169 server.repl_backlog->unindexed_count = 0;
170 }
171 }
172
173 /* Rebase replication buffer blocks' offset since the initial
174 * setting offset starts from 0 when master restart. */
rebaseReplicationBuffer(long long base_repl_offset)175 void rebaseReplicationBuffer(long long base_repl_offset) {
176 raxFree(server.repl_backlog->blocks_index);
177 server.repl_backlog->blocks_index = raxNew();
178 server.repl_backlog->unindexed_count = 0;
179
180 listIter li;
181 listNode *ln;
182 listRewind(server.repl_buffer_blocks, &li);
183 while ((ln = listNext(&li))) {
184 replBufBlock *o = listNodeValue(ln);
185 o->repl_offset += base_repl_offset;
186 createReplicationBacklogIndex(ln);
187 }
188 }
189
resetReplicationBuffer(void)190 void resetReplicationBuffer(void) {
191 server.repl_buffer_mem = 0;
192 server.repl_buffer_blocks = listCreate();
193 listSetFreeMethod(server.repl_buffer_blocks, (void (*)(void*))zfree);
194 }
195
canFeedReplicaReplBuffer(client * replica)196 int canFeedReplicaReplBuffer(client *replica) {
197 /* Don't feed replicas that only want the RDB. */
198 if (replica->flags & CLIENT_REPL_RDBONLY) return 0;
199
200 /* Don't feed replicas that are still waiting for BGSAVE to start. */
201 if (replica->replstate == SLAVE_STATE_WAIT_BGSAVE_START) return 0;
202
203 return 1;
204 }
205
206 /* Similar with 'prepareClientToWrite', note that we must call this function
207 * before feeding replication stream into global replication buffer, since
208 * clientHasPendingReplies in prepareClientToWrite will access the global
209 * replication buffer to make judgements. */
prepareReplicasToWrite(void)210 int prepareReplicasToWrite(void) {
211 listIter li;
212 listNode *ln;
213 int prepared = 0;
214
215 listRewind(server.slaves,&li);
216 while((ln = listNext(&li))) {
217 client *slave = ln->value;
218 if (!canFeedReplicaReplBuffer(slave)) continue;
219 if (prepareClientToWrite(slave) == C_ERR) continue;
220 prepared++;
221 }
222
223 return prepared;
224 }
225
226 /* Wrapper for feedReplicationBuffer() that takes Redis string objects
227 * as input. */
feedReplicationBufferWithObject(robj * o)228 void feedReplicationBufferWithObject(robj *o) {
229 char llstr[LONG_STR_SIZE];
230 void *p;
231 size_t len;
232
233 if (o->encoding == OBJ_ENCODING_INT) {
234 len = ll2string(llstr,sizeof(llstr),(long)o->ptr);
235 p = llstr;
236 } else {
237 len = sdslen(o->ptr);
238 p = o->ptr;
239 }
240 feedReplicationBuffer(p,len);
241 }
242
243 /* Generally, we only have one replication buffer block to trim when replication
244 * backlog size exceeds our setting and no replica reference it. But if replica
245 * clients disconnect, we need to free many replication buffer blocks that are
246 * referenced. It would cost much time if there are a lots blocks to free, that
247 * will freeze server, so we trim replication backlog incrementally. */
incrementalTrimReplicationBacklog(size_t max_blocks)248 void incrementalTrimReplicationBacklog(size_t max_blocks) {
249 serverAssert(server.repl_backlog != NULL);
250
251 size_t trimmed_blocks = 0;
252 while (server.repl_backlog->histlen > server.repl_backlog_size &&
253 trimmed_blocks < max_blocks)
254 {
255 /* We never trim backlog to less than one block. */
256 if (listLength(server.repl_buffer_blocks) <= 1) break;
257
258 /* Replicas increment the refcount of the first replication buffer block
259 * they refer to, in that case, we don't trim the backlog even if
260 * backlog_histlen exceeds backlog_size. This implicitly makes backlog
261 * bigger than our setting, but makes the master accept partial resync as
262 * much as possible. So that backlog must be the last reference of
263 * replication buffer blocks. */
264 listNode *first = listFirst(server.repl_buffer_blocks);
265 serverAssert(first == server.repl_backlog->ref_repl_buf_node);
266 replBufBlock *fo = listNodeValue(first);
267 if (fo->refcount != 1) break;
268
269 /* We don't try trim backlog if backlog valid size will be lessen than
270 * setting backlog size once we release the first repl buffer block. */
271 if (server.repl_backlog->histlen - (long long)fo->size <=
272 server.repl_backlog_size) break;
273
274 /* Decr refcount and release the first block later. */
275 fo->refcount--;
276 trimmed_blocks++;
277 server.repl_backlog->histlen -= fo->size;
278
279 /* Go to use next replication buffer block node. */
280 listNode *next = listNextNode(first);
281 server.repl_backlog->ref_repl_buf_node = next;
282 serverAssert(server.repl_backlog->ref_repl_buf_node != NULL);
283 /* Incr reference count to keep the new head node. */
284 ((replBufBlock *)listNodeValue(next))->refcount++;
285
286 /* Remove the node in recorded blocks. */
287 uint64_t encoded_offset = htonu64(fo->repl_offset);
288 raxRemove(server.repl_backlog->blocks_index,
289 (unsigned char*)&encoded_offset, sizeof(uint64_t), NULL);
290
291 /* Delete the first node from global replication buffer. */
292 serverAssert(fo->refcount == 0 && fo->used == fo->size);
293 server.repl_buffer_mem -= (fo->size +
294 sizeof(listNode) + sizeof(replBufBlock));
295 listDelNode(server.repl_buffer_blocks, first);
296 }
297
298 /* Set the offset of the first byte we have in the backlog. */
299 server.repl_backlog->offset = server.master_repl_offset -
300 server.repl_backlog->histlen + 1;
301 }
302
303 /* Free replication buffer blocks that are referenced by this client. */
freeReplicaReferencedReplBuffer(client * replica)304 void freeReplicaReferencedReplBuffer(client *replica) {
305 if (replica->ref_repl_buf_node != NULL) {
306 /* Decrease the start buffer node reference count. */
307 replBufBlock *o = listNodeValue(replica->ref_repl_buf_node);
308 serverAssert(o->refcount > 0);
309 o->refcount--;
310 incrementalTrimReplicationBacklog(REPL_BACKLOG_TRIM_BLOCKS_PER_CALL);
311 }
312 replica->ref_repl_buf_node = NULL;
313 replica->ref_block_pos = 0;
314 }
315
316 /* Append bytes into the global replication buffer list, replication backlog and
317 * all replica clients use replication buffers collectively, this function replace
318 * 'addReply*', 'feedReplicationBacklog' for replicas and replication backlog,
319 * First we add buffer into global replication buffer block list, and then
320 * update replica / replication-backlog referenced node and block position. */
feedReplicationBuffer(char * s,size_t len)321 void feedReplicationBuffer(char *s, size_t len) {
322 static long long repl_block_id = 0;
323
324 if (server.repl_backlog == NULL) return;
325 server.master_repl_offset += len;
326 server.repl_backlog->histlen += len;
327
328 /* Install write handler for all replicas. */
329 prepareReplicasToWrite();
330
331 size_t start_pos = 0; /* The position of referenced blok to start sending. */
332 listNode *start_node = NULL; /* Replica/backlog starts referenced node. */
333 int add_new_block = 0; /* Create new block if current block is total used. */
334 listNode *ln = listLast(server.repl_buffer_blocks);
335 replBufBlock *tail = ln ? listNodeValue(ln) : NULL;
336
337 /* Append to tail string when possible. */
338 if (tail && tail->size > tail->used) {
339 start_node = listLast(server.repl_buffer_blocks);
340 start_pos = tail->used;
341 /* Copy the part we can fit into the tail, and leave the rest for a
342 * new node */
343 size_t avail = tail->size - tail->used;
344 size_t copy = (avail >= len) ? len : avail;
345 memcpy(tail->buf + tail->used, s, copy);
346 tail->used += copy;
347 s += copy;
348 len -= copy;
349 }
350 if (len) {
351 /* Create a new node, make sure it is allocated to at
352 * least PROTO_REPLY_CHUNK_BYTES */
353 size_t usable_size;
354 size_t size = (len < PROTO_REPLY_CHUNK_BYTES) ? PROTO_REPLY_CHUNK_BYTES : len;
355 tail = zmalloc_usable(size + sizeof(replBufBlock), &usable_size);
356 /* Take over the allocation's internal fragmentation */
357 tail->size = usable_size - sizeof(replBufBlock);
358 tail->used = len;
359 tail->refcount = 0;
360 tail->repl_offset = server.master_repl_offset - tail->used + 1;
361 tail->id = repl_block_id++;
362 memcpy(tail->buf, s, len);
363 listAddNodeTail(server.repl_buffer_blocks, tail);
364 /* We also count the list node memory into replication buffer memory. */
365 server.repl_buffer_mem += (usable_size + sizeof(listNode));
366 add_new_block = 1;
367 if (start_node == NULL) {
368 start_node = listLast(server.repl_buffer_blocks);
369 start_pos = 0;
370 }
371 }
372
373 /* For output buffer of replicas. */
374 listIter li;
375 listRewind(server.slaves,&li);
376 while((ln = listNext(&li))) {
377 client *slave = ln->value;
378 if (!canFeedReplicaReplBuffer(slave)) continue;
379
380 /* Update shared replication buffer start position. */
381 if (slave->ref_repl_buf_node == NULL) {
382 slave->ref_repl_buf_node = start_node;
383 slave->ref_block_pos = start_pos;
384 /* Only increase the start block reference count. */
385 ((replBufBlock *)listNodeValue(start_node))->refcount++;
386 }
387
388 /* Check output buffer limit only when add new block. */
389 if (add_new_block) closeClientOnOutputBufferLimitReached(slave, 1);
390 }
391
392 /* For replication backlog */
393 if (server.repl_backlog->ref_repl_buf_node == NULL) {
394 server.repl_backlog->ref_repl_buf_node = start_node;
395 /* Only increase the start block reference count. */
396 ((replBufBlock *)listNodeValue(start_node))->refcount++;
397
398 /* Replication buffer must be empty before adding replication stream
399 * into replication backlog. */
400 serverAssert(add_new_block == 1 && start_pos == 0);
401 }
402 if (add_new_block) {
403 createReplicationBacklogIndex(listLast(server.repl_buffer_blocks));
404 }
405 /* Try to trim replication backlog since replication backlog may exceed
406 * our setting when we add replication stream. Note that it is important to
407 * try to trim at least one node since in the common case this is where one
408 * new backlog node is added and one should be removed. See also comments
409 * in freeMemoryGetNotCountedMemory for details. */
410 incrementalTrimReplicationBacklog(REPL_BACKLOG_TRIM_BLOCKS_PER_CALL);
411 }
412
413 /* Propagate write commands to replication stream.
414 *
415 * This function is used if the instance is a master: we use the commands
416 * received by our clients in order to create the replication stream.
417 * Instead if the instance is a replica and has sub-replicas attached, we use
418 * replicationFeedStreamFromMasterStream() */
replicationFeedSlaves(list * slaves,int dictid,robj ** argv,int argc)419 void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) {
420 int j, len;
421 char llstr[LONG_STR_SIZE];
422
423 /* If the instance is not a top level master, return ASAP: we'll just proxy
424 * the stream of data we receive from our master instead, in order to
425 * propagate *identical* replication stream. In this way this slave can
426 * advertise the same replication ID as the master (since it shares the
427 * master replication history and has the same backlog and offsets). */
428 if (server.masterhost != NULL) return;
429
430 /* If there aren't slaves, and there is no backlog buffer to populate,
431 * we can return ASAP. */
432 if (server.repl_backlog == NULL && listLength(slaves) == 0) return;
433
434 /* We can't have slaves attached and no backlog. */
435 serverAssert(!(listLength(slaves) != 0 && server.repl_backlog == NULL));
436
437 /* Send SELECT command to every slave if needed. */
438 if (server.slaveseldb != dictid) {
439 robj *selectcmd;
440
441 /* For a few DBs we have pre-computed SELECT command. */
442 if (dictid >= 0 && dictid < PROTO_SHARED_SELECT_CMDS) {
443 selectcmd = shared.select[dictid];
444 } else {
445 int dictid_len;
446
447 dictid_len = ll2string(llstr,sizeof(llstr),dictid);
448 selectcmd = createObject(OBJ_STRING,
449 sdscatprintf(sdsempty(),
450 "*2\r\n$6\r\nSELECT\r\n$%d\r\n%s\r\n",
451 dictid_len, llstr));
452 }
453
454 feedReplicationBufferWithObject(selectcmd);
455
456 if (dictid < 0 || dictid >= PROTO_SHARED_SELECT_CMDS)
457 decrRefCount(selectcmd);
458 }
459 server.slaveseldb = dictid;
460
461 /* Write the command to the replication buffer if any. */
462 char aux[LONG_STR_SIZE+3];
463
464 /* Add the multi bulk reply length. */
465 aux[0] = '*';
466 len = ll2string(aux+1,sizeof(aux)-1,argc);
467 aux[len+1] = '\r';
468 aux[len+2] = '\n';
469 feedReplicationBuffer(aux,len+3);
470
471 for (j = 0; j < argc; j++) {
472 long objlen = stringObjectLen(argv[j]);
473
474 /* We need to feed the buffer with the object as a bulk reply
475 * not just as a plain string, so create the $..CRLF payload len
476 * and add the final CRLF */
477 aux[0] = '$';
478 len = ll2string(aux+1,sizeof(aux)-1,objlen);
479 aux[len+1] = '\r';
480 aux[len+2] = '\n';
481 feedReplicationBuffer(aux,len+3);
482 feedReplicationBufferWithObject(argv[j]);
483 feedReplicationBuffer(aux+len+1,2);
484 }
485 }
486
487 /* This is a debugging function that gets called when we detect something
488 * wrong with the replication protocol: the goal is to peek into the
489 * replication backlog and show a few final bytes to make simpler to
490 * guess what kind of bug it could be. */
showLatestBacklog(void)491 void showLatestBacklog(void) {
492 if (server.repl_backlog == NULL) return;
493 if (listLength(server.repl_buffer_blocks) == 0) return;
494
495 size_t dumplen = 256;
496 if (server.repl_backlog->histlen < (long long)dumplen)
497 dumplen = server.repl_backlog->histlen;
498
499 sds dump = sdsempty();
500 listNode *node = listLast(server.repl_buffer_blocks);
501 while(dumplen) {
502 if (node == NULL) break;
503 replBufBlock *o = listNodeValue(node);
504 size_t thislen = o->used >= dumplen ? dumplen : o->used;
505 sds head = sdscatrepr(sdsempty(), o->buf+o->used-thislen, thislen);
506 sds tmp = sdscatsds(head, dump);
507 sdsfree(dump);
508 dump = tmp;
509 dumplen -= thislen;
510 node = listPrevNode(node);
511 }
512
513 /* Finally log such bytes: this is vital debugging info to
514 * understand what happened. */
515 serverLog(LL_WARNING,"Latest backlog is: '%s'", dump);
516 sdsfree(dump);
517 }
518
519 /* This function is used in order to proxy what we receive from our master
520 * to our sub-slaves. */
521 #include <ctype.h>
replicationFeedStreamFromMasterStream(char * buf,size_t buflen)522 void replicationFeedStreamFromMasterStream(char *buf, size_t buflen) {
523 /* Debugging: this is handy to see the stream sent from master
524 * to slaves. Disabled with if(0). */
525 if (0) {
526 printf("%zu:",buflen);
527 for (size_t j = 0; j < buflen; j++) {
528 printf("%c", isprint(buf[j]) ? buf[j] : '.');
529 }
530 printf("\n");
531 }
532
533 /* There must be replication backlog if having attached slaves. */
534 if (listLength(server.slaves)) serverAssert(server.repl_backlog != NULL);
535 if (server.repl_backlog) feedReplicationBuffer(buf,buflen);
536 }
537
replicationFeedMonitors(client * c,list * monitors,int dictid,robj ** argv,int argc)538 void replicationFeedMonitors(client *c, list *monitors, int dictid, robj **argv, int argc) {
539 if (!(listLength(server.monitors) && !server.loading)) return;
540 listNode *ln;
541 listIter li;
542 int j;
543 sds cmdrepr = sdsnew("+");
544 robj *cmdobj;
545 struct timeval tv;
546
547 gettimeofday(&tv,NULL);
548 cmdrepr = sdscatprintf(cmdrepr,"%ld.%06ld ",(long)tv.tv_sec,(long)tv.tv_usec);
549 if (c->flags & CLIENT_LUA) {
550 cmdrepr = sdscatprintf(cmdrepr,"[%d lua] ",dictid);
551 } else if (c->flags & CLIENT_UNIX_SOCKET) {
552 cmdrepr = sdscatprintf(cmdrepr,"[%d unix:%s] ",dictid,server.unixsocket);
553 } else {
554 cmdrepr = sdscatprintf(cmdrepr,"[%d %s] ",dictid,getClientPeerId(c));
555 }
556
557 for (j = 0; j < argc; j++) {
558 if (argv[j]->encoding == OBJ_ENCODING_INT) {
559 cmdrepr = sdscatprintf(cmdrepr, "\"%ld\"", (long)argv[j]->ptr);
560 } else {
561 cmdrepr = sdscatrepr(cmdrepr,(char*)argv[j]->ptr,
562 sdslen(argv[j]->ptr));
563 }
564 if (j != argc-1)
565 cmdrepr = sdscatlen(cmdrepr," ",1);
566 }
567 cmdrepr = sdscatlen(cmdrepr,"\r\n",2);
568 cmdobj = createObject(OBJ_STRING,cmdrepr);
569
570 listRewind(monitors,&li);
571 while((ln = listNext(&li))) {
572 client *monitor = ln->value;
573 addReply(monitor,cmdobj);
574 updateClientMemUsage(c);
575 }
576 decrRefCount(cmdobj);
577 }
578
579 /* Feed the slave 'c' with the replication backlog starting from the
580 * specified 'offset' up to the end of the backlog. */
addReplyReplicationBacklog(client * c,long long offset)581 long long addReplyReplicationBacklog(client *c, long long offset) {
582 long long skip;
583
584 serverLog(LL_DEBUG, "[PSYNC] Replica request offset: %lld", offset);
585
586 if (server.repl_backlog->histlen == 0) {
587 serverLog(LL_DEBUG, "[PSYNC] Backlog history len is zero");
588 return 0;
589 }
590
591 serverLog(LL_DEBUG, "[PSYNC] Backlog size: %lld",
592 server.repl_backlog_size);
593 serverLog(LL_DEBUG, "[PSYNC] First byte: %lld",
594 server.repl_backlog->offset);
595 serverLog(LL_DEBUG, "[PSYNC] History len: %lld",
596 server.repl_backlog->histlen);
597
598 /* Compute the amount of bytes we need to discard. */
599 skip = offset - server.repl_backlog->offset;
600 serverLog(LL_DEBUG, "[PSYNC] Skipping: %lld", skip);
601
602 /* Iterate recorded blocks, quickly search the approximate node. */
603 listNode *node = NULL;
604 if (raxSize(server.repl_backlog->blocks_index) > 0) {
605 uint64_t encoded_offset = htonu64(offset);
606 raxIterator ri;
607 raxStart(&ri, server.repl_backlog->blocks_index);
608 raxSeek(&ri, ">", (unsigned char*)&encoded_offset, sizeof(uint64_t));
609 if (raxEOF(&ri)) {
610 /* No found, so search from the last recorded node. */
611 raxSeek(&ri, "$", NULL, 0);
612 raxPrev(&ri);
613 node = (listNode *)ri.data;
614 } else {
615 raxPrev(&ri); /* Skip the sought node. */
616 /* We should search from the prev node since the offset of current
617 * sought node exceeds searching offset. */
618 if (raxPrev(&ri))
619 node = (listNode *)ri.data;
620 else
621 node = server.repl_backlog->ref_repl_buf_node;
622 }
623 raxStop(&ri);
624 } else {
625 /* No recorded blocks, just from the start node to search. */
626 node = server.repl_backlog->ref_repl_buf_node;
627 }
628
629 /* Search the exact node. */
630 while (node != NULL) {
631 replBufBlock *o = listNodeValue(node);
632 if (o->repl_offset + (long long)o->used >= offset) break;
633 node = listNextNode(node);
634 }
635 serverAssert(node != NULL);
636
637 /* Install a writer handler first.*/
638 prepareClientToWrite(c);
639 /* Setting output buffer of the replica. */
640 replBufBlock *o = listNodeValue(node);
641 o->refcount++;
642 c->ref_repl_buf_node = node;
643 c->ref_block_pos = offset - o->repl_offset;
644
645 return server.repl_backlog->histlen - skip;
646 }
647
648 /* Return the offset to provide as reply to the PSYNC command received
649 * from the slave. The returned value is only valid immediately after
650 * the BGSAVE process started and before executing any other command
651 * from clients. */
getPsyncInitialOffset(void)652 long long getPsyncInitialOffset(void) {
653 return server.master_repl_offset;
654 }
655
656 /* Send a FULLRESYNC reply in the specific case of a full resynchronization,
657 * as a side effect setup the slave for a full sync in different ways:
658 *
659 * 1) Remember, into the slave client structure, the replication offset
660 * we sent here, so that if new slaves will later attach to the same
661 * background RDB saving process (by duplicating this client output
662 * buffer), we can get the right offset from this slave.
663 * 2) Set the replication state of the slave to WAIT_BGSAVE_END so that
664 * we start accumulating differences from this point.
665 * 3) Force the replication stream to re-emit a SELECT statement so
666 * the new slave incremental differences will start selecting the
667 * right database number.
668 *
669 * Normally this function should be called immediately after a successful
670 * BGSAVE for replication was started, or when there is one already in
671 * progress that we attached our slave to. */
replicationSetupSlaveForFullResync(client * slave,long long offset)672 int replicationSetupSlaveForFullResync(client *slave, long long offset) {
673 char buf[128];
674 int buflen;
675
676 slave->psync_initial_offset = offset;
677 slave->replstate = SLAVE_STATE_WAIT_BGSAVE_END;
678 /* We are going to accumulate the incremental changes for this
679 * slave as well. Set slaveseldb to -1 in order to force to re-emit
680 * a SELECT statement in the replication stream. */
681 server.slaveseldb = -1;
682
683 /* Don't send this reply to slaves that approached us with
684 * the old SYNC command. */
685 if (!(slave->flags & CLIENT_PRE_PSYNC)) {
686 buflen = snprintf(buf,sizeof(buf),"+FULLRESYNC %s %lld\r\n",
687 server.replid,offset);
688 if (connWrite(slave->conn,buf,buflen) != buflen) {
689 freeClientAsync(slave);
690 return C_ERR;
691 }
692 }
693 return C_OK;
694 }
695
696 /* This function handles the PSYNC command from the point of view of a
697 * master receiving a request for partial resynchronization.
698 *
699 * On success return C_OK, otherwise C_ERR is returned and we proceed
700 * with the usual full resync. */
masterTryPartialResynchronization(client * c)701 int masterTryPartialResynchronization(client *c) {
702 long long psync_offset, psync_len;
703 char *master_replid = c->argv[1]->ptr;
704 char buf[128];
705 int buflen;
706
707 /* Parse the replication offset asked by the slave. Go to full sync
708 * on parse error: this should never happen but we try to handle
709 * it in a robust way compared to aborting. */
710 if (getLongLongFromObjectOrReply(c,c->argv[2],&psync_offset,NULL) !=
711 C_OK) goto need_full_resync;
712
713 /* Is the replication ID of this master the same advertised by the wannabe
714 * slave via PSYNC? If the replication ID changed this master has a
715 * different replication history, and there is no way to continue.
716 *
717 * Note that there are two potentially valid replication IDs: the ID1
718 * and the ID2. The ID2 however is only valid up to a specific offset. */
719 if (strcasecmp(master_replid, server.replid) &&
720 (strcasecmp(master_replid, server.replid2) ||
721 psync_offset > server.second_replid_offset))
722 {
723 /* Replid "?" is used by slaves that want to force a full resync. */
724 if (master_replid[0] != '?') {
725 if (strcasecmp(master_replid, server.replid) &&
726 strcasecmp(master_replid, server.replid2))
727 {
728 serverLog(LL_NOTICE,"Partial resynchronization not accepted: "
729 "Replication ID mismatch (Replica asked for '%s', my "
730 "replication IDs are '%s' and '%s')",
731 master_replid, server.replid, server.replid2);
732 } else {
733 serverLog(LL_NOTICE,"Partial resynchronization not accepted: "
734 "Requested offset for second ID was %lld, but I can reply "
735 "up to %lld", psync_offset, server.second_replid_offset);
736 }
737 } else {
738 serverLog(LL_NOTICE,"Full resync requested by replica %s",
739 replicationGetSlaveName(c));
740 }
741 goto need_full_resync;
742 }
743
744 /* We still have the data our slave is asking for? */
745 if (!server.repl_backlog ||
746 psync_offset < server.repl_backlog->offset ||
747 psync_offset > (server.repl_backlog->offset + server.repl_backlog->histlen))
748 {
749 serverLog(LL_NOTICE,
750 "Unable to partial resync with replica %s for lack of backlog (Replica request was: %lld).", replicationGetSlaveName(c), psync_offset);
751 if (psync_offset > server.master_repl_offset) {
752 serverLog(LL_WARNING,
753 "Warning: replica %s tried to PSYNC with an offset that is greater than the master replication offset.", replicationGetSlaveName(c));
754 }
755 goto need_full_resync;
756 }
757
758 /* If we reached this point, we are able to perform a partial resync:
759 * 1) Set client state to make it a slave.
760 * 2) Inform the client we can continue with +CONTINUE
761 * 3) Send the backlog data (from the offset to the end) to the slave. */
762 c->flags |= CLIENT_SLAVE;
763 c->replstate = SLAVE_STATE_ONLINE;
764 c->repl_ack_time = server.unixtime;
765 c->repl_put_online_on_ack = 0;
766 listAddNodeTail(server.slaves,c);
767 /* We can't use the connection buffers since they are used to accumulate
768 * new commands at this stage. But we are sure the socket send buffer is
769 * empty so this write will never fail actually. */
770 if (c->slave_capa & SLAVE_CAPA_PSYNC2) {
771 buflen = snprintf(buf,sizeof(buf),"+CONTINUE %s\r\n", server.replid);
772 } else {
773 buflen = snprintf(buf,sizeof(buf),"+CONTINUE\r\n");
774 }
775 if (connWrite(c->conn,buf,buflen) != buflen) {
776 freeClientAsync(c);
777 return C_OK;
778 }
779 psync_len = addReplyReplicationBacklog(c,psync_offset);
780 serverLog(LL_NOTICE,
781 "Partial resynchronization request from %s accepted. Sending %lld bytes of backlog starting from offset %lld.",
782 replicationGetSlaveName(c),
783 psync_len, psync_offset);
784 /* Note that we don't need to set the selected DB at server.slaveseldb
785 * to -1 to force the master to emit SELECT, since the slave already
786 * has this state from the previous connection with the master. */
787
788 refreshGoodSlavesCount();
789
790 /* Fire the replica change modules event. */
791 moduleFireServerEvent(REDISMODULE_EVENT_REPLICA_CHANGE,
792 REDISMODULE_SUBEVENT_REPLICA_CHANGE_ONLINE,
793 NULL);
794
795 return C_OK; /* The caller can return, no full resync needed. */
796
797 need_full_resync:
798 /* We need a full resync for some reason... Note that we can't
799 * reply to PSYNC right now if a full SYNC is needed. The reply
800 * must include the master offset at the time the RDB file we transfer
801 * is generated, so we need to delay the reply to that moment. */
802 return C_ERR;
803 }
804
805 /* Start a BGSAVE for replication goals, which is, selecting the disk or
806 * socket target depending on the configuration, and making sure that
807 * the script cache is flushed before to start.
808 *
809 * The mincapa argument is the bitwise AND among all the slaves capabilities
810 * of the slaves waiting for this BGSAVE, so represents the slave capabilities
811 * all the slaves support. Can be tested via SLAVE_CAPA_* macros.
812 *
813 * Side effects, other than starting a BGSAVE:
814 *
815 * 1) Handle the slaves in WAIT_START state, by preparing them for a full
816 * sync if the BGSAVE was successfully started, or sending them an error
817 * and dropping them from the list of slaves.
818 *
819 * 2) Flush the Lua scripting script cache if the BGSAVE was actually
820 * started.
821 *
822 * Returns C_OK on success or C_ERR otherwise. */
startBgsaveForReplication(int mincapa)823 int startBgsaveForReplication(int mincapa) {
824 int retval;
825 int socket_target = server.repl_diskless_sync && (mincapa & SLAVE_CAPA_EOF);
826 listIter li;
827 listNode *ln;
828
829 serverLog(LL_NOTICE,"Starting BGSAVE for SYNC with target: %s",
830 socket_target ? "replicas sockets" : "disk");
831
832 rdbSaveInfo rsi, *rsiptr;
833 rsiptr = rdbPopulateSaveInfo(&rsi);
834 /* Only do rdbSave* when rsiptr is not NULL,
835 * otherwise slave will miss repl-stream-db. */
836 if (rsiptr) {
837 if (socket_target)
838 retval = rdbSaveToSlavesSockets(rsiptr);
839 else
840 retval = rdbSaveBackground(server.rdb_filename,rsiptr);
841 } else {
842 serverLog(LL_WARNING,"BGSAVE for replication: replication information not available, can't generate the RDB file right now. Try later.");
843 retval = C_ERR;
844 }
845
846 /* If we succeeded to start a BGSAVE with disk target, let's remember
847 * this fact, so that we can later delete the file if needed. Note
848 * that we don't set the flag to 1 if the feature is disabled, otherwise
849 * it would never be cleared: the file is not deleted. This way if
850 * the user enables it later with CONFIG SET, we are fine. */
851 if (retval == C_OK && !socket_target && server.rdb_del_sync_files)
852 RDBGeneratedByReplication = 1;
853
854 /* If we failed to BGSAVE, remove the slaves waiting for a full
855 * resynchronization from the list of slaves, inform them with
856 * an error about what happened, close the connection ASAP. */
857 if (retval == C_ERR) {
858 serverLog(LL_WARNING,"BGSAVE for replication failed");
859 listRewind(server.slaves,&li);
860 while((ln = listNext(&li))) {
861 client *slave = ln->value;
862
863 if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) {
864 slave->replstate = REPL_STATE_NONE;
865 slave->flags &= ~CLIENT_SLAVE;
866 listDelNode(server.slaves,ln);
867 addReplyError(slave,
868 "BGSAVE failed, replication can't continue");
869 slave->flags |= CLIENT_CLOSE_AFTER_REPLY;
870 }
871 }
872 return retval;
873 }
874
875 /* If the target is socket, rdbSaveToSlavesSockets() already setup
876 * the slaves for a full resync. Otherwise for disk target do it now.*/
877 if (!socket_target) {
878 listRewind(server.slaves,&li);
879 while((ln = listNext(&li))) {
880 client *slave = ln->value;
881
882 if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) {
883 replicationSetupSlaveForFullResync(slave,
884 getPsyncInitialOffset());
885 }
886 }
887 }
888
889 /* Flush the script cache, since we need that slave differences are
890 * accumulated without requiring slaves to match our cached scripts. */
891 if (retval == C_OK) replicationScriptCacheFlush();
892 return retval;
893 }
894
895 /* SYNC and PSYNC command implementation. */
syncCommand(client * c)896 void syncCommand(client *c) {
897 /* ignore SYNC if already slave or in monitor mode */
898 if (c->flags & CLIENT_SLAVE) return;
899
900 /* Check if this is a failover request to a replica with the same replid and
901 * become a master if so. */
902 if (c->argc > 3 && !strcasecmp(c->argv[0]->ptr,"psync") &&
903 !strcasecmp(c->argv[3]->ptr,"failover"))
904 {
905 serverLog(LL_WARNING, "Failover request received for replid %s.",
906 (unsigned char *)c->argv[1]->ptr);
907 if (!server.masterhost) {
908 addReplyError(c, "PSYNC FAILOVER can't be sent to a master.");
909 return;
910 }
911
912 if (!strcasecmp(c->argv[1]->ptr,server.replid)) {
913 replicationUnsetMaster();
914 sds client = catClientInfoString(sdsempty(),c);
915 serverLog(LL_NOTICE,
916 "MASTER MODE enabled (failover request from '%s')",client);
917 sdsfree(client);
918 } else {
919 addReplyError(c, "PSYNC FAILOVER replid must match my replid.");
920 return;
921 }
922 }
923
924 /* Don't let replicas sync with us while we're failing over */
925 if (server.failover_state != NO_FAILOVER) {
926 addReplyError(c,"-NOMASTERLINK Can't SYNC while failing over");
927 return;
928 }
929
930 /* Refuse SYNC requests if we are a slave but the link with our master
931 * is not ok... */
932 if (server.masterhost && server.repl_state != REPL_STATE_CONNECTED) {
933 addReplyError(c,"-NOMASTERLINK Can't SYNC while not connected with my master");
934 return;
935 }
936
937 /* SYNC can't be issued when the server has pending data to send to
938 * the client about already issued commands. We need a fresh reply
939 * buffer registering the differences between the BGSAVE and the current
940 * dataset, so that we can copy to other slaves if needed. */
941 if (clientHasPendingReplies(c)) {
942 addReplyError(c,"SYNC and PSYNC are invalid with pending output");
943 return;
944 }
945
946 serverLog(LL_NOTICE,"Replica %s asks for synchronization",
947 replicationGetSlaveName(c));
948
949 /* Try a partial resynchronization if this is a PSYNC command.
950 * If it fails, we continue with usual full resynchronization, however
951 * when this happens replicationSetupSlaveForFullResync will replied
952 * with:
953 *
954 * +FULLRESYNC <replid> <offset>
955 *
956 * So the slave knows the new replid and offset to try a PSYNC later
957 * if the connection with the master is lost. */
958 if (!strcasecmp(c->argv[0]->ptr,"psync")) {
959 if (masterTryPartialResynchronization(c) == C_OK) {
960 server.stat_sync_partial_ok++;
961 return; /* No full resync needed, return. */
962 } else {
963 char *master_replid = c->argv[1]->ptr;
964
965 /* Increment stats for failed PSYNCs, but only if the
966 * replid is not "?", as this is used by slaves to force a full
967 * resync on purpose when they are not able to partially
968 * resync. */
969 if (master_replid[0] != '?') server.stat_sync_partial_err++;
970 }
971 } else {
972 /* If a slave uses SYNC, we are dealing with an old implementation
973 * of the replication protocol (like redis-cli --slave). Flag the client
974 * so that we don't expect to receive REPLCONF ACK feedbacks. */
975 c->flags |= CLIENT_PRE_PSYNC;
976 }
977
978 /* Full resynchronization. */
979 server.stat_sync_full++;
980
981 /* Setup the slave as one waiting for BGSAVE to start. The following code
982 * paths will change the state if we handle the slave differently. */
983 c->replstate = SLAVE_STATE_WAIT_BGSAVE_START;
984 if (server.repl_disable_tcp_nodelay)
985 connDisableTcpNoDelay(c->conn); /* Non critical if it fails. */
986 c->repldbfd = -1;
987 c->flags |= CLIENT_SLAVE;
988 listAddNodeTail(server.slaves,c);
989
990 /* Create the replication backlog if needed. */
991 if (listLength(server.slaves) == 1 && server.repl_backlog == NULL) {
992 /* When we create the backlog from scratch, we always use a new
993 * replication ID and clear the ID2, since there is no valid
994 * past history. */
995 changeReplicationId();
996 clearReplicationId2();
997 createReplicationBacklog();
998 serverLog(LL_NOTICE,"Replication backlog created, my new "
999 "replication IDs are '%s' and '%s'",
1000 server.replid, server.replid2);
1001 }
1002
1003 /* CASE 1: BGSAVE is in progress, with disk target. */
1004 if (server.child_type == CHILD_TYPE_RDB &&
1005 server.rdb_child_type == RDB_CHILD_TYPE_DISK)
1006 {
1007 /* Ok a background save is in progress. Let's check if it is a good
1008 * one for replication, i.e. if there is another slave that is
1009 * registering differences since the server forked to save. */
1010 client *slave;
1011 listNode *ln;
1012 listIter li;
1013
1014 listRewind(server.slaves,&li);
1015 while((ln = listNext(&li))) {
1016 slave = ln->value;
1017 /* If the client needs a buffer of commands, we can't use
1018 * a replica without replication buffer. */
1019 if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_END &&
1020 (!(slave->flags & CLIENT_REPL_RDBONLY) ||
1021 (c->flags & CLIENT_REPL_RDBONLY)))
1022 break;
1023 }
1024 /* To attach this slave, we check that it has at least all the
1025 * capabilities of the slave that triggered the current BGSAVE. */
1026 if (ln && ((c->slave_capa & slave->slave_capa) == slave->slave_capa)) {
1027 /* Perfect, the server is already registering differences for
1028 * another slave. Set the right state, and copy the buffer.
1029 * We don't copy buffer if clients don't want. */
1030 if (!(c->flags & CLIENT_REPL_RDBONLY))
1031 copyReplicaOutputBuffer(c,slave);
1032 replicationSetupSlaveForFullResync(c,slave->psync_initial_offset);
1033 serverLog(LL_NOTICE,"Waiting for end of BGSAVE for SYNC");
1034 } else {
1035 /* No way, we need to wait for the next BGSAVE in order to
1036 * register differences. */
1037 serverLog(LL_NOTICE,"Can't attach the replica to the current BGSAVE. Waiting for next BGSAVE for SYNC");
1038 }
1039
1040 /* CASE 2: BGSAVE is in progress, with socket target. */
1041 } else if (server.child_type == CHILD_TYPE_RDB &&
1042 server.rdb_child_type == RDB_CHILD_TYPE_SOCKET)
1043 {
1044 /* There is an RDB child process but it is writing directly to
1045 * children sockets. We need to wait for the next BGSAVE
1046 * in order to synchronize. */
1047 serverLog(LL_NOTICE,"Current BGSAVE has socket target. Waiting for next BGSAVE for SYNC");
1048
1049 /* CASE 3: There is no BGSAVE is in progress. */
1050 } else {
1051 if (server.repl_diskless_sync && (c->slave_capa & SLAVE_CAPA_EOF) &&
1052 server.repl_diskless_sync_delay)
1053 {
1054 /* Diskless replication RDB child is created inside
1055 * replicationCron() since we want to delay its start a
1056 * few seconds to wait for more slaves to arrive. */
1057 serverLog(LL_NOTICE,"Delay next BGSAVE for diskless SYNC");
1058 } else {
1059 /* We don't have a BGSAVE in progress, let's start one. Diskless
1060 * or disk-based mode is determined by replica's capacity. */
1061 if (!hasActiveChildProcess()) {
1062 startBgsaveForReplication(c->slave_capa);
1063 } else {
1064 serverLog(LL_NOTICE,
1065 "No BGSAVE in progress, but another BG operation is active. "
1066 "BGSAVE for replication delayed");
1067 }
1068 }
1069 }
1070 return;
1071 }
1072
1073 /* REPLCONF <option> <value> <option> <value> ...
1074 * This command is used by a replica in order to configure the replication
1075 * process before starting it with the SYNC command.
1076 * This command is also used by a master in order to get the replication
1077 * offset from a replica.
1078 *
1079 * Currently we support these options:
1080 *
1081 * - listening-port <port>
1082 * - ip-address <ip>
1083 * What is the listening ip and port of the Replica redis instance, so that
1084 * the master can accurately lists replicas and their listening ports in the
1085 * INFO output.
1086 *
1087 * - capa <eof|psync2>
1088 * What is the capabilities of this instance.
1089 * eof: supports EOF-style RDB transfer for diskless replication.
1090 * psync2: supports PSYNC v2, so understands +CONTINUE <new repl ID>.
1091 *
1092 * - ack <offset>
1093 * Replica informs the master the amount of replication stream that it
1094 * processed so far.
1095 *
1096 * - getack
1097 * Unlike other subcommands, this is used by master to get the replication
1098 * offset from a replica.
1099 *
1100 * - rdb-only
1101 * Only wants RDB snapshot without replication buffer. */
replconfCommand(client * c)1102 void replconfCommand(client *c) {
1103 int j;
1104
1105 if ((c->argc % 2) == 0) {
1106 /* Number of arguments must be odd to make sure that every
1107 * option has a corresponding value. */
1108 addReplyErrorObject(c,shared.syntaxerr);
1109 return;
1110 }
1111
1112 /* Process every option-value pair. */
1113 for (j = 1; j < c->argc; j+=2) {
1114 if (!strcasecmp(c->argv[j]->ptr,"listening-port")) {
1115 long port;
1116
1117 if ((getLongFromObjectOrReply(c,c->argv[j+1],
1118 &port,NULL) != C_OK))
1119 return;
1120 c->slave_listening_port = port;
1121 } else if (!strcasecmp(c->argv[j]->ptr,"ip-address")) {
1122 sds addr = c->argv[j+1]->ptr;
1123 if (sdslen(addr) < NET_HOST_STR_LEN) {
1124 if (c->slave_addr) sdsfree(c->slave_addr);
1125 c->slave_addr = sdsdup(addr);
1126 } else {
1127 addReplyErrorFormat(c,"REPLCONF ip-address provided by "
1128 "replica instance is too long: %zd bytes", sdslen(addr));
1129 return;
1130 }
1131 } else if (!strcasecmp(c->argv[j]->ptr,"capa")) {
1132 /* Ignore capabilities not understood by this master. */
1133 if (!strcasecmp(c->argv[j+1]->ptr,"eof"))
1134 c->slave_capa |= SLAVE_CAPA_EOF;
1135 else if (!strcasecmp(c->argv[j+1]->ptr,"psync2"))
1136 c->slave_capa |= SLAVE_CAPA_PSYNC2;
1137 } else if (!strcasecmp(c->argv[j]->ptr,"ack")) {
1138 /* REPLCONF ACK is used by slave to inform the master the amount
1139 * of replication stream that it processed so far. It is an
1140 * internal only command that normal clients should never use. */
1141 long long offset;
1142
1143 if (!(c->flags & CLIENT_SLAVE)) return;
1144 if ((getLongLongFromObject(c->argv[j+1], &offset) != C_OK))
1145 return;
1146 if (offset > c->repl_ack_off)
1147 c->repl_ack_off = offset;
1148 c->repl_ack_time = server.unixtime;
1149 /* If this was a diskless replication, we need to really put
1150 * the slave online when the first ACK is received (which
1151 * confirms slave is online and ready to get more data). This
1152 * allows for simpler and less CPU intensive EOF detection
1153 * when streaming RDB files.
1154 * There's a chance the ACK got to us before we detected that the
1155 * bgsave is done (since that depends on cron ticks), so run a
1156 * quick check first (instead of waiting for the next ACK. */
1157 if (server.child_type == CHILD_TYPE_RDB && c->replstate == SLAVE_STATE_WAIT_BGSAVE_END)
1158 checkChildrenDone();
1159 if (c->repl_put_online_on_ack && c->replstate == SLAVE_STATE_ONLINE)
1160 putSlaveOnline(c);
1161 /* Note: this command does not reply anything! */
1162 return;
1163 } else if (!strcasecmp(c->argv[j]->ptr,"getack")) {
1164 /* REPLCONF GETACK is used in order to request an ACK ASAP
1165 * to the slave. */
1166 if (server.masterhost && server.master) replicationSendAck();
1167 return;
1168 } else if (!strcasecmp(c->argv[j]->ptr,"rdb-only")) {
1169 /* REPLCONF RDB-ONLY is used to identify the client only wants
1170 * RDB snapshot without replication buffer. */
1171 long rdb_only = 0;
1172 if (getRangeLongFromObjectOrReply(c,c->argv[j+1],
1173 0,1,&rdb_only,NULL) != C_OK)
1174 return;
1175 if (rdb_only == 1) c->flags |= CLIENT_REPL_RDBONLY;
1176 else c->flags &= ~CLIENT_REPL_RDBONLY;
1177 } else {
1178 addReplyErrorFormat(c,"Unrecognized REPLCONF option: %s",
1179 (char*)c->argv[j]->ptr);
1180 return;
1181 }
1182 }
1183 addReply(c,shared.ok);
1184 }
1185
1186 /* This function puts a replica in the online state, and should be called just
1187 * after a replica received the RDB file for the initial synchronization, and
1188 * we are finally ready to send the incremental stream of commands.
1189 *
1190 * It does a few things:
1191 * 1) Close the replica's connection async if it doesn't need replication
1192 * commands buffer stream, since it actually isn't a valid replica.
1193 * 2) Put the slave in ONLINE state. Note that the function may also be called
1194 * for a replicas that are already in ONLINE state, but having the flag
1195 * repl_put_online_on_ack set to true: we still have to install the write
1196 * handler in that case. This function will take care of that.
1197 * 3) Make sure the writable event is re-installed, since calling the SYNC
1198 * command disables it, so that we can accumulate output buffer without
1199 * sending it to the replica.
1200 * 4) Update the count of "good replicas". */
putSlaveOnline(client * slave)1201 void putSlaveOnline(client *slave) {
1202 slave->replstate = SLAVE_STATE_ONLINE;
1203 slave->repl_put_online_on_ack = 0;
1204 slave->repl_ack_time = server.unixtime; /* Prevent false timeout. */
1205
1206 if (slave->flags & CLIENT_REPL_RDBONLY) {
1207 serverLog(LL_NOTICE,
1208 "Close the connection with replica %s as RDB transfer is complete",
1209 replicationGetSlaveName(slave));
1210 freeClientAsync(slave);
1211 return;
1212 }
1213 if (connSetWriteHandler(slave->conn, sendReplyToClient) == C_ERR) {
1214 serverLog(LL_WARNING,"Unable to register writable event for replica bulk transfer: %s", strerror(errno));
1215 freeClient(slave);
1216 return;
1217 }
1218 refreshGoodSlavesCount();
1219 /* Fire the replica change modules event. */
1220 moduleFireServerEvent(REDISMODULE_EVENT_REPLICA_CHANGE,
1221 REDISMODULE_SUBEVENT_REPLICA_CHANGE_ONLINE,
1222 NULL);
1223 serverLog(LL_NOTICE,"Synchronization with replica %s succeeded",
1224 replicationGetSlaveName(slave));
1225 }
1226
1227 /* We call this function periodically to remove an RDB file that was
1228 * generated because of replication, in an instance that is otherwise
1229 * without any persistence. We don't want instances without persistence
1230 * to take RDB files around, this violates certain policies in certain
1231 * environments. */
removeRDBUsedToSyncReplicas(void)1232 void removeRDBUsedToSyncReplicas(void) {
1233 /* If the feature is disabled, return ASAP but also clear the
1234 * RDBGeneratedByReplication flag in case it was set. Otherwise if the
1235 * feature was enabled, but gets disabled later with CONFIG SET, the
1236 * flag may remain set to one: then next time the feature is re-enabled
1237 * via CONFIG SET we have it set even if no RDB was generated
1238 * because of replication recently. */
1239 if (!server.rdb_del_sync_files) {
1240 RDBGeneratedByReplication = 0;
1241 return;
1242 }
1243
1244 if (allPersistenceDisabled() && RDBGeneratedByReplication) {
1245 client *slave;
1246 listNode *ln;
1247 listIter li;
1248
1249 int delrdb = 1;
1250 listRewind(server.slaves,&li);
1251 while((ln = listNext(&li))) {
1252 slave = ln->value;
1253 if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START ||
1254 slave->replstate == SLAVE_STATE_WAIT_BGSAVE_END ||
1255 slave->replstate == SLAVE_STATE_SEND_BULK)
1256 {
1257 delrdb = 0;
1258 break; /* No need to check the other replicas. */
1259 }
1260 }
1261 if (delrdb) {
1262 struct stat sb;
1263 if (lstat(server.rdb_filename,&sb) != -1) {
1264 RDBGeneratedByReplication = 0;
1265 serverLog(LL_NOTICE,
1266 "Removing the RDB file used to feed replicas "
1267 "in a persistence-less instance");
1268 bg_unlink(server.rdb_filename);
1269 }
1270 }
1271 }
1272 }
1273
sendBulkToSlave(connection * conn)1274 void sendBulkToSlave(connection *conn) {
1275 client *slave = connGetPrivateData(conn);
1276 char buf[PROTO_IOBUF_LEN];
1277 ssize_t nwritten, buflen;
1278
1279 /* Before sending the RDB file, we send the preamble as configured by the
1280 * replication process. Currently the preamble is just the bulk count of
1281 * the file in the form "$<length>\r\n". */
1282 if (slave->replpreamble) {
1283 nwritten = connWrite(conn,slave->replpreamble,sdslen(slave->replpreamble));
1284 if (nwritten == -1) {
1285 serverLog(LL_WARNING,
1286 "Write error sending RDB preamble to replica: %s",
1287 connGetLastError(conn));
1288 freeClient(slave);
1289 return;
1290 }
1291 atomicIncr(server.stat_net_output_bytes, nwritten);
1292 sdsrange(slave->replpreamble,nwritten,-1);
1293 if (sdslen(slave->replpreamble) == 0) {
1294 sdsfree(slave->replpreamble);
1295 slave->replpreamble = NULL;
1296 /* fall through sending data. */
1297 } else {
1298 return;
1299 }
1300 }
1301
1302 /* If the preamble was already transferred, send the RDB bulk data. */
1303 lseek(slave->repldbfd,slave->repldboff,SEEK_SET);
1304 buflen = read(slave->repldbfd,buf,PROTO_IOBUF_LEN);
1305 if (buflen <= 0) {
1306 serverLog(LL_WARNING,"Read error sending DB to replica: %s",
1307 (buflen == 0) ? "premature EOF" : strerror(errno));
1308 freeClient(slave);
1309 return;
1310 }
1311 if ((nwritten = connWrite(conn,buf,buflen)) == -1) {
1312 if (connGetState(conn) != CONN_STATE_CONNECTED) {
1313 serverLog(LL_WARNING,"Write error sending DB to replica: %s",
1314 connGetLastError(conn));
1315 freeClient(slave);
1316 }
1317 return;
1318 }
1319 slave->repldboff += nwritten;
1320 atomicIncr(server.stat_net_output_bytes, nwritten);
1321 if (slave->repldboff == slave->repldbsize) {
1322 close(slave->repldbfd);
1323 slave->repldbfd = -1;
1324 connSetWriteHandler(slave->conn,NULL);
1325 putSlaveOnline(slave);
1326 }
1327 }
1328
1329 /* Remove one write handler from the list of connections waiting to be writable
1330 * during rdb pipe transfer. */
rdbPipeWriteHandlerConnRemoved(struct connection * conn)1331 void rdbPipeWriteHandlerConnRemoved(struct connection *conn) {
1332 if (!connHasWriteHandler(conn))
1333 return;
1334 connSetWriteHandler(conn, NULL);
1335 client *slave = connGetPrivateData(conn);
1336 slave->repl_last_partial_write = 0;
1337 server.rdb_pipe_numconns_writing--;
1338 /* if there are no more writes for now for this conn, or write error: */
1339 if (server.rdb_pipe_numconns_writing == 0) {
1340 if (aeCreateFileEvent(server.el, server.rdb_pipe_read, AE_READABLE, rdbPipeReadHandler,NULL) == AE_ERR) {
1341 serverPanic("Unrecoverable error creating server.rdb_pipe_read file event.");
1342 }
1343 }
1344 }
1345
1346 /* Called in diskless master during transfer of data from the rdb pipe, when
1347 * the replica becomes writable again. */
rdbPipeWriteHandler(struct connection * conn)1348 void rdbPipeWriteHandler(struct connection *conn) {
1349 serverAssert(server.rdb_pipe_bufflen>0);
1350 client *slave = connGetPrivateData(conn);
1351 int nwritten;
1352 if ((nwritten = connWrite(conn, server.rdb_pipe_buff + slave->repldboff,
1353 server.rdb_pipe_bufflen - slave->repldboff)) == -1)
1354 {
1355 if (connGetState(conn) == CONN_STATE_CONNECTED)
1356 return; /* equivalent to EAGAIN */
1357 serverLog(LL_WARNING,"Write error sending DB to replica: %s",
1358 connGetLastError(conn));
1359 freeClient(slave);
1360 return;
1361 } else {
1362 slave->repldboff += nwritten;
1363 atomicIncr(server.stat_net_output_bytes, nwritten);
1364 if (slave->repldboff < server.rdb_pipe_bufflen) {
1365 slave->repl_last_partial_write = server.unixtime;
1366 return; /* more data to write.. */
1367 }
1368 }
1369 rdbPipeWriteHandlerConnRemoved(conn);
1370 }
1371
1372 /* Called in diskless master, when there's data to read from the child's rdb pipe */
rdbPipeReadHandler(struct aeEventLoop * eventLoop,int fd,void * clientData,int mask)1373 void rdbPipeReadHandler(struct aeEventLoop *eventLoop, int fd, void *clientData, int mask) {
1374 UNUSED(mask);
1375 UNUSED(clientData);
1376 UNUSED(eventLoop);
1377 int i;
1378 if (!server.rdb_pipe_buff)
1379 server.rdb_pipe_buff = zmalloc(PROTO_IOBUF_LEN);
1380 serverAssert(server.rdb_pipe_numconns_writing==0);
1381
1382 while (1) {
1383 server.rdb_pipe_bufflen = read(fd, server.rdb_pipe_buff, PROTO_IOBUF_LEN);
1384 if (server.rdb_pipe_bufflen < 0) {
1385 if (errno == EAGAIN || errno == EWOULDBLOCK)
1386 return;
1387 serverLog(LL_WARNING,"Diskless rdb transfer, read error sending DB to replicas: %s", strerror(errno));
1388 for (i=0; i < server.rdb_pipe_numconns; i++) {
1389 connection *conn = server.rdb_pipe_conns[i];
1390 if (!conn)
1391 continue;
1392 client *slave = connGetPrivateData(conn);
1393 freeClient(slave);
1394 server.rdb_pipe_conns[i] = NULL;
1395 }
1396 killRDBChild();
1397 return;
1398 }
1399
1400 if (server.rdb_pipe_bufflen == 0) {
1401 /* EOF - write end was closed. */
1402 int stillUp = 0;
1403 aeDeleteFileEvent(server.el, server.rdb_pipe_read, AE_READABLE);
1404 for (i=0; i < server.rdb_pipe_numconns; i++)
1405 {
1406 connection *conn = server.rdb_pipe_conns[i];
1407 if (!conn)
1408 continue;
1409 stillUp++;
1410 }
1411 serverLog(LL_WARNING,"Diskless rdb transfer, done reading from pipe, %d replicas still up.", stillUp);
1412 /* Now that the replicas have finished reading, notify the child that it's safe to exit.
1413 * When the server detects the child has exited, it can mark the replica as online, and
1414 * start streaming the replication buffers. */
1415 close(server.rdb_child_exit_pipe);
1416 server.rdb_child_exit_pipe = -1;
1417 return;
1418 }
1419
1420 int stillAlive = 0;
1421 for (i=0; i < server.rdb_pipe_numconns; i++)
1422 {
1423 int nwritten;
1424 connection *conn = server.rdb_pipe_conns[i];
1425 if (!conn)
1426 continue;
1427
1428 client *slave = connGetPrivateData(conn);
1429 if ((nwritten = connWrite(conn, server.rdb_pipe_buff, server.rdb_pipe_bufflen)) == -1) {
1430 if (connGetState(conn) != CONN_STATE_CONNECTED) {
1431 serverLog(LL_WARNING,"Diskless rdb transfer, write error sending DB to replica: %s",
1432 connGetLastError(conn));
1433 freeClient(slave);
1434 server.rdb_pipe_conns[i] = NULL;
1435 continue;
1436 }
1437 /* An error and still in connected state, is equivalent to EAGAIN */
1438 slave->repldboff = 0;
1439 } else {
1440 /* Note: when use diskless replication, 'repldboff' is the offset
1441 * of 'rdb_pipe_buff' sent rather than the offset of entire RDB. */
1442 slave->repldboff = nwritten;
1443 atomicIncr(server.stat_net_output_bytes, nwritten);
1444 }
1445 /* If we were unable to write all the data to one of the replicas,
1446 * setup write handler (and disable pipe read handler, below) */
1447 if (nwritten != server.rdb_pipe_bufflen) {
1448 slave->repl_last_partial_write = server.unixtime;
1449 server.rdb_pipe_numconns_writing++;
1450 connSetWriteHandler(conn, rdbPipeWriteHandler);
1451 }
1452 stillAlive++;
1453 }
1454
1455 if (stillAlive == 0) {
1456 serverLog(LL_WARNING,"Diskless rdb transfer, last replica dropped, killing fork child.");
1457 killRDBChild();
1458 }
1459 /* Remove the pipe read handler if at least one write handler was set. */
1460 if (server.rdb_pipe_numconns_writing || stillAlive == 0) {
1461 aeDeleteFileEvent(server.el, server.rdb_pipe_read, AE_READABLE);
1462 break;
1463 }
1464 }
1465 }
1466
1467 /* This function is called at the end of every background saving,
1468 * or when the replication RDB transfer strategy is modified from
1469 * disk to socket or the other way around.
1470 *
1471 * The goal of this function is to handle slaves waiting for a successful
1472 * background saving in order to perform non-blocking synchronization, and
1473 * to schedule a new BGSAVE if there are slaves that attached while a
1474 * BGSAVE was in progress, but it was not a good one for replication (no
1475 * other slave was accumulating differences).
1476 *
1477 * The argument bgsaveerr is C_OK if the background saving succeeded
1478 * otherwise C_ERR is passed to the function.
1479 * The 'type' argument is the type of the child that terminated
1480 * (if it had a disk or socket target). */
updateSlavesWaitingBgsave(int bgsaveerr,int type)1481 void updateSlavesWaitingBgsave(int bgsaveerr, int type) {
1482 listNode *ln;
1483 listIter li;
1484
1485 listRewind(server.slaves,&li);
1486 while((ln = listNext(&li))) {
1487 client *slave = ln->value;
1488
1489 if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_END) {
1490 struct redis_stat buf;
1491
1492 if (bgsaveerr != C_OK) {
1493 freeClient(slave);
1494 serverLog(LL_WARNING,"SYNC failed. BGSAVE child returned an error");
1495 continue;
1496 }
1497
1498 /* If this was an RDB on disk save, we have to prepare to send
1499 * the RDB from disk to the slave socket. Otherwise if this was
1500 * already an RDB -> Slaves socket transfer, used in the case of
1501 * diskless replication, our work is trivial, we can just put
1502 * the slave online. */
1503 if (type == RDB_CHILD_TYPE_SOCKET) {
1504 serverLog(LL_NOTICE,
1505 "Streamed RDB transfer with replica %s succeeded (socket). Waiting for REPLCONF ACK from slave to enable streaming",
1506 replicationGetSlaveName(slave));
1507 /* Note: we wait for a REPLCONF ACK message from the replica in
1508 * order to really put it online (install the write handler
1509 * so that the accumulated data can be transferred). However
1510 * we change the replication state ASAP, since our slave
1511 * is technically online now.
1512 *
1513 * So things work like that:
1514 *
1515 * 1. We end transferring the RDB file via socket.
1516 * 2. The replica is put ONLINE but the write handler
1517 * is not installed.
1518 * 3. The replica however goes really online, and pings us
1519 * back via REPLCONF ACK commands.
1520 * 4. Now we finally install the write handler, and send
1521 * the buffers accumulated so far to the replica.
1522 *
1523 * But why we do that? Because the replica, when we stream
1524 * the RDB directly via the socket, must detect the RDB
1525 * EOF (end of file), that is a special random string at the
1526 * end of the RDB (for streamed RDBs we don't know the length
1527 * in advance). Detecting such final EOF string is much
1528 * simpler and less CPU intensive if no more data is sent
1529 * after such final EOF. So we don't want to glue the end of
1530 * the RDB transfer with the start of the other replication
1531 * data. */
1532 slave->replstate = SLAVE_STATE_ONLINE;
1533 slave->repl_put_online_on_ack = 1;
1534 slave->repl_ack_time = server.unixtime; /* Timeout otherwise. */
1535 } else {
1536 if ((slave->repldbfd = open(server.rdb_filename,O_RDONLY)) == -1 ||
1537 redis_fstat(slave->repldbfd,&buf) == -1) {
1538 freeClient(slave);
1539 serverLog(LL_WARNING,"SYNC failed. Can't open/stat DB after BGSAVE: %s", strerror(errno));
1540 continue;
1541 }
1542 slave->repldboff = 0;
1543 slave->repldbsize = buf.st_size;
1544 slave->replstate = SLAVE_STATE_SEND_BULK;
1545 slave->replpreamble = sdscatprintf(sdsempty(),"$%lld\r\n",
1546 (unsigned long long) slave->repldbsize);
1547
1548 connSetWriteHandler(slave->conn,NULL);
1549 if (connSetWriteHandler(slave->conn,sendBulkToSlave) == C_ERR) {
1550 freeClient(slave);
1551 continue;
1552 }
1553 }
1554 }
1555 }
1556 }
1557
1558 /* Change the current instance replication ID with a new, random one.
1559 * This will prevent successful PSYNCs between this master and other
1560 * slaves, so the command should be called when something happens that
1561 * alters the current story of the dataset. */
changeReplicationId(void)1562 void changeReplicationId(void) {
1563 getRandomHexChars(server.replid,CONFIG_RUN_ID_SIZE);
1564 server.replid[CONFIG_RUN_ID_SIZE] = '\0';
1565 }
1566
1567 /* Clear (invalidate) the secondary replication ID. This happens, for
1568 * example, after a full resynchronization, when we start a new replication
1569 * history. */
clearReplicationId2(void)1570 void clearReplicationId2(void) {
1571 memset(server.replid2,'0',sizeof(server.replid));
1572 server.replid2[CONFIG_RUN_ID_SIZE] = '\0';
1573 server.second_replid_offset = -1;
1574 }
1575
1576 /* Use the current replication ID / offset as secondary replication
1577 * ID, and change the current one in order to start a new history.
1578 * This should be used when an instance is switched from slave to master
1579 * so that it can serve PSYNC requests performed using the master
1580 * replication ID. */
shiftReplicationId(void)1581 void shiftReplicationId(void) {
1582 memcpy(server.replid2,server.replid,sizeof(server.replid));
1583 /* We set the second replid offset to the master offset + 1, since
1584 * the slave will ask for the first byte it has not yet received, so
1585 * we need to add one to the offset: for example if, as a slave, we are
1586 * sure we have the same history as the master for 50 bytes, after we
1587 * are turned into a master, we can accept a PSYNC request with offset
1588 * 51, since the slave asking has the same history up to the 50th
1589 * byte, and is asking for the new bytes starting at offset 51. */
1590 server.second_replid_offset = server.master_repl_offset+1;
1591 changeReplicationId();
1592 serverLog(LL_WARNING,"Setting secondary replication ID to %s, valid up to offset: %lld. New replication ID is %s", server.replid2, server.second_replid_offset, server.replid);
1593 }
1594
1595 /* ----------------------------------- SLAVE -------------------------------- */
1596
1597 /* Returns 1 if the given replication state is a handshake state,
1598 * 0 otherwise. */
slaveIsInHandshakeState(void)1599 int slaveIsInHandshakeState(void) {
1600 return server.repl_state >= REPL_STATE_RECEIVE_PING_REPLY &&
1601 server.repl_state <= REPL_STATE_RECEIVE_PSYNC_REPLY;
1602 }
1603
1604 /* Avoid the master to detect the slave is timing out while loading the
1605 * RDB file in initial synchronization. We send a single newline character
1606 * that is valid protocol but is guaranteed to either be sent entirely or
1607 * not, since the byte is indivisible.
1608 *
1609 * The function is called in two contexts: while we flush the current
1610 * data with emptyDb(), and while we load the new data received as an
1611 * RDB file from the master. */
replicationSendNewlineToMaster(void)1612 void replicationSendNewlineToMaster(void) {
1613 static time_t newline_sent;
1614 if (time(NULL) != newline_sent) {
1615 newline_sent = time(NULL);
1616 /* Pinging back in this stage is best-effort. */
1617 if (server.repl_transfer_s) connWrite(server.repl_transfer_s, "\n", 1);
1618 }
1619 }
1620
1621 /* Callback used by emptyDb() while flushing away old data to load
1622 * the new dataset received by the master and by discardTempDb()
1623 * after loading succeeded or failed. */
replicationEmptyDbCallback(dict * d)1624 void replicationEmptyDbCallback(dict *d) {
1625 UNUSED(d);
1626 if (server.repl_state == REPL_STATE_TRANSFER)
1627 replicationSendNewlineToMaster();
1628 }
1629
1630 /* Once we have a link with the master and the synchronization was
1631 * performed, this function materializes the master client we store
1632 * at server.master, starting from the specified file descriptor. */
replicationCreateMasterClient(connection * conn,int dbid)1633 void replicationCreateMasterClient(connection *conn, int dbid) {
1634 server.master = createClient(conn);
1635 if (conn)
1636 connSetReadHandler(server.master->conn, readQueryFromClient);
1637
1638 /**
1639 * Important note:
1640 * The CLIENT_DENY_BLOCKING flag is not, and should not, be set here.
1641 * For commands like BLPOP, it makes no sense to block the master
1642 * connection, and such blocking attempt will probably cause deadlock and
1643 * break the replication. We consider such a thing as a bug because
1644 * commands as BLPOP should never be sent on the replication link.
1645 * A possible use-case for blocking the replication link is if a module wants
1646 * to pass the execution to a background thread and unblock after the
1647 * execution is done. This is the reason why we allow blocking the replication
1648 * connection. */
1649 server.master->flags |= CLIENT_MASTER;
1650
1651 server.master->authenticated = 1;
1652 server.master->reploff = server.master_initial_offset;
1653 server.master->read_reploff = server.master->reploff;
1654 server.master->user = NULL; /* This client can do everything. */
1655 memcpy(server.master->replid, server.master_replid,
1656 sizeof(server.master_replid));
1657 /* If master offset is set to -1, this master is old and is not
1658 * PSYNC capable, so we flag it accordingly. */
1659 if (server.master->reploff == -1)
1660 server.master->flags |= CLIENT_PRE_PSYNC;
1661 if (dbid != -1) selectDb(server.master,dbid);
1662 }
1663
1664 /* This function will try to re-enable the AOF file after the
1665 * master-replica synchronization: if it fails after multiple attempts
1666 * the replica cannot be considered reliable and exists with an
1667 * error. */
restartAOFAfterSYNC()1668 void restartAOFAfterSYNC() {
1669 unsigned int tries, max_tries = 10;
1670 for (tries = 0; tries < max_tries; ++tries) {
1671 if (startAppendOnly() == C_OK) break;
1672 serverLog(LL_WARNING,
1673 "Failed enabling the AOF after successful master synchronization! "
1674 "Trying it again in one second.");
1675 sleep(1);
1676 }
1677 if (tries == max_tries) {
1678 serverLog(LL_WARNING,
1679 "FATAL: this replica instance finished the synchronization with "
1680 "its master, but the AOF can't be turned on. Exiting now.");
1681 exit(1);
1682 }
1683 }
1684
useDisklessLoad()1685 static int useDisklessLoad() {
1686 /* compute boolean decision to use diskless load */
1687 int enabled = server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB ||
1688 (server.repl_diskless_load == REPL_DISKLESS_LOAD_WHEN_DB_EMPTY && dbTotalServerKeyCount()==0);
1689
1690 if (enabled) {
1691 /* Check all modules handle read errors, otherwise it's not safe to use diskless load. */
1692 if (!moduleAllDatatypesHandleErrors()) {
1693 serverLog(LL_WARNING,
1694 "Skipping diskless-load because there are modules that don't handle read errors.");
1695 enabled = 0;
1696 }
1697 /* Check all modules handle async replication, otherwise it's not safe to use diskless load. */
1698 else if (server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB && !moduleAllModulesHandleReplAsyncLoad()) {
1699 serverLog(LL_WARNING,
1700 "Skipping diskless-load because there are modules that are not aware of async replication.");
1701 enabled = 0;
1702 }
1703 }
1704 return enabled;
1705 }
1706
1707 /* Helper function for readSyncBulkPayload() to initialize tempDb
1708 * before socket-loading the new db from master. The tempDb may be populated
1709 * by swapMainDbWithTempDb or freed by disklessLoadDiscardTempDb later. */
disklessLoadInitTempDb(void)1710 redisDb *disklessLoadInitTempDb(void) {
1711 return initTempDb();
1712 }
1713
1714 /* Helper function for readSyncBulkPayload() to discard our tempDb
1715 * when the loading succeeded or failed. */
disklessLoadDiscardTempDb(redisDb * tempDb)1716 void disklessLoadDiscardTempDb(redisDb *tempDb) {
1717 discardTempDb(tempDb, replicationEmptyDbCallback);
1718 }
1719
1720 /* If we know we got an entirely different data set from our master
1721 * we have no way to incrementally feed our replicas after that.
1722 * We want our replicas to resync with us as well, if we have any sub-replicas.
1723 * This is useful on readSyncBulkPayload in places where we just finished transferring db. */
replicationAttachToNewMaster()1724 void replicationAttachToNewMaster() {
1725 /* Replica starts to apply data from new master, we must discard the cached
1726 * master structure. */
1727 serverAssert(server.master == NULL);
1728 replicationDiscardCachedMaster();
1729
1730 disconnectSlaves(); /* Force our replicas to resync with us as well. */
1731 freeReplicationBacklog(); /* Don't allow our chained replicas to PSYNC. */
1732 }
1733
1734 /* Asynchronously read the SYNC payload we receive from a master */
1735 #define REPL_MAX_WRITTEN_BEFORE_FSYNC (1024*1024*8) /* 8 MB */
readSyncBulkPayload(connection * conn)1736 void readSyncBulkPayload(connection *conn) {
1737 char buf[PROTO_IOBUF_LEN];
1738 ssize_t nread, readlen, nwritten;
1739 int use_diskless_load = useDisklessLoad();
1740 redisDb *diskless_load_tempDb = NULL;
1741 int empty_db_flags = server.repl_slave_lazy_flush ? EMPTYDB_ASYNC :
1742 EMPTYDB_NO_FLAGS;
1743 off_t left;
1744
1745 /* Static vars used to hold the EOF mark, and the last bytes received
1746 * from the server: when they match, we reached the end of the transfer. */
1747 static char eofmark[CONFIG_RUN_ID_SIZE];
1748 static char lastbytes[CONFIG_RUN_ID_SIZE];
1749 static int usemark = 0;
1750
1751 /* If repl_transfer_size == -1 we still have to read the bulk length
1752 * from the master reply. */
1753 if (server.repl_transfer_size == -1) {
1754 if (connSyncReadLine(conn,buf,1024,server.repl_syncio_timeout*1000) == -1) {
1755 serverLog(LL_WARNING,
1756 "I/O error reading bulk count from MASTER: %s",
1757 strerror(errno));
1758 goto error;
1759 }
1760
1761 if (buf[0] == '-') {
1762 serverLog(LL_WARNING,
1763 "MASTER aborted replication with an error: %s",
1764 buf+1);
1765 goto error;
1766 } else if (buf[0] == '\0') {
1767 /* At this stage just a newline works as a PING in order to take
1768 * the connection live. So we refresh our last interaction
1769 * timestamp. */
1770 server.repl_transfer_lastio = server.unixtime;
1771 return;
1772 } else if (buf[0] != '$') {
1773 serverLog(LL_WARNING,"Bad protocol from MASTER, the first byte is not '$' (we received '%s'), are you sure the host and port are right?", buf);
1774 goto error;
1775 }
1776
1777 /* There are two possible forms for the bulk payload. One is the
1778 * usual $<count> bulk format. The other is used for diskless transfers
1779 * when the master does not know beforehand the size of the file to
1780 * transfer. In the latter case, the following format is used:
1781 *
1782 * $EOF:<40 bytes delimiter>
1783 *
1784 * At the end of the file the announced delimiter is transmitted. The
1785 * delimiter is long and random enough that the probability of a
1786 * collision with the actual file content can be ignored. */
1787 if (strncmp(buf+1,"EOF:",4) == 0 && strlen(buf+5) >= CONFIG_RUN_ID_SIZE) {
1788 usemark = 1;
1789 memcpy(eofmark,buf+5,CONFIG_RUN_ID_SIZE);
1790 memset(lastbytes,0,CONFIG_RUN_ID_SIZE);
1791 /* Set any repl_transfer_size to avoid entering this code path
1792 * at the next call. */
1793 server.repl_transfer_size = 0;
1794 serverLog(LL_NOTICE,
1795 "MASTER <-> REPLICA sync: receiving streamed RDB from master with EOF %s",
1796 use_diskless_load? "to parser":"to disk");
1797 } else {
1798 usemark = 0;
1799 server.repl_transfer_size = strtol(buf+1,NULL,10);
1800 serverLog(LL_NOTICE,
1801 "MASTER <-> REPLICA sync: receiving %lld bytes from master %s",
1802 (long long) server.repl_transfer_size,
1803 use_diskless_load? "to parser":"to disk");
1804 }
1805 return;
1806 }
1807
1808 if (!use_diskless_load) {
1809 /* Read the data from the socket, store it to a file and search
1810 * for the EOF. */
1811 if (usemark) {
1812 readlen = sizeof(buf);
1813 } else {
1814 left = server.repl_transfer_size - server.repl_transfer_read;
1815 readlen = (left < (signed)sizeof(buf)) ? left : (signed)sizeof(buf);
1816 }
1817
1818 nread = connRead(conn,buf,readlen);
1819 if (nread <= 0) {
1820 if (connGetState(conn) == CONN_STATE_CONNECTED) {
1821 /* equivalent to EAGAIN */
1822 return;
1823 }
1824 serverLog(LL_WARNING,"I/O error trying to sync with MASTER: %s",
1825 (nread == -1) ? strerror(errno) : "connection lost");
1826 cancelReplicationHandshake(1);
1827 return;
1828 }
1829 atomicIncr(server.stat_net_input_bytes, nread);
1830
1831 /* When a mark is used, we want to detect EOF asap in order to avoid
1832 * writing the EOF mark into the file... */
1833 int eof_reached = 0;
1834
1835 if (usemark) {
1836 /* Update the last bytes array, and check if it matches our
1837 * delimiter. */
1838 if (nread >= CONFIG_RUN_ID_SIZE) {
1839 memcpy(lastbytes,buf+nread-CONFIG_RUN_ID_SIZE,
1840 CONFIG_RUN_ID_SIZE);
1841 } else {
1842 int rem = CONFIG_RUN_ID_SIZE-nread;
1843 memmove(lastbytes,lastbytes+nread,rem);
1844 memcpy(lastbytes+rem,buf,nread);
1845 }
1846 if (memcmp(lastbytes,eofmark,CONFIG_RUN_ID_SIZE) == 0)
1847 eof_reached = 1;
1848 }
1849
1850 /* Update the last I/O time for the replication transfer (used in
1851 * order to detect timeouts during replication), and write what we
1852 * got from the socket to the dump file on disk. */
1853 server.repl_transfer_lastio = server.unixtime;
1854 if ((nwritten = write(server.repl_transfer_fd,buf,nread)) != nread) {
1855 serverLog(LL_WARNING,
1856 "Write error or short write writing to the DB dump file "
1857 "needed for MASTER <-> REPLICA synchronization: %s",
1858 (nwritten == -1) ? strerror(errno) : "short write");
1859 goto error;
1860 }
1861 server.repl_transfer_read += nread;
1862
1863 /* Delete the last 40 bytes from the file if we reached EOF. */
1864 if (usemark && eof_reached) {
1865 if (ftruncate(server.repl_transfer_fd,
1866 server.repl_transfer_read - CONFIG_RUN_ID_SIZE) == -1)
1867 {
1868 serverLog(LL_WARNING,
1869 "Error truncating the RDB file received from the master "
1870 "for SYNC: %s", strerror(errno));
1871 goto error;
1872 }
1873 }
1874
1875 /* Sync data on disk from time to time, otherwise at the end of the
1876 * transfer we may suffer a big delay as the memory buffers are copied
1877 * into the actual disk. */
1878 if (server.repl_transfer_read >=
1879 server.repl_transfer_last_fsync_off + REPL_MAX_WRITTEN_BEFORE_FSYNC)
1880 {
1881 off_t sync_size = server.repl_transfer_read -
1882 server.repl_transfer_last_fsync_off;
1883 rdb_fsync_range(server.repl_transfer_fd,
1884 server.repl_transfer_last_fsync_off, sync_size);
1885 server.repl_transfer_last_fsync_off += sync_size;
1886 }
1887
1888 /* Check if the transfer is now complete */
1889 if (!usemark) {
1890 if (server.repl_transfer_read == server.repl_transfer_size)
1891 eof_reached = 1;
1892 }
1893
1894 /* If the transfer is yet not complete, we need to read more, so
1895 * return ASAP and wait for the handler to be called again. */
1896 if (!eof_reached) return;
1897 }
1898
1899 /* We reach this point in one of the following cases:
1900 *
1901 * 1. The replica is using diskless replication, that is, it reads data
1902 * directly from the socket to the Redis memory, without using
1903 * a temporary RDB file on disk. In that case we just block and
1904 * read everything from the socket.
1905 *
1906 * 2. Or when we are done reading from the socket to the RDB file, in
1907 * such case we want just to read the RDB file in memory. */
1908
1909 /* We need to stop any AOF rewriting child before flushing and parsing
1910 * the RDB, otherwise we'll create a copy-on-write disaster. */
1911 if (server.aof_state != AOF_OFF) stopAppendOnly();
1912
1913 if (use_diskless_load && server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB) {
1914 /* Initialize empty tempDb dictionaries. */
1915 diskless_load_tempDb = disklessLoadInitTempDb();
1916
1917 moduleFireServerEvent(REDISMODULE_EVENT_REPL_ASYNC_LOAD,
1918 REDISMODULE_SUBEVENT_REPL_ASYNC_LOAD_STARTED,
1919 NULL);
1920 } else {
1921 replicationAttachToNewMaster();
1922
1923 serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Flushing old data");
1924 emptyDb(-1,empty_db_flags,replicationEmptyDbCallback);
1925 }
1926
1927 /* Before loading the DB into memory we need to delete the readable
1928 * handler, otherwise it will get called recursively since
1929 * rdbLoad() will call the event loop to process events from time to
1930 * time for non blocking loading. */
1931 connSetReadHandler(conn, NULL);
1932
1933 serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Loading DB in memory");
1934 rdbSaveInfo rsi = RDB_SAVE_INFO_INIT;
1935 if (use_diskless_load) {
1936 rio rdb;
1937 redisDb *dbarray;
1938 int asyncLoading = 0;
1939
1940 if (server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB) {
1941 /* Async loading means we continue serving read commands during full resync, and
1942 * "swap" the new db with the old db only when loading is done.
1943 * It is enabled only on SWAPDB diskless replication when master replication ID hasn't changed,
1944 * because in that state the old content of the db represents a different point in time of the same
1945 * data set we're currently receiving from the master. */
1946 if (memcmp(server.replid, server.master_replid, CONFIG_RUN_ID_SIZE) == 0) {
1947 asyncLoading = 1;
1948 }
1949 dbarray = diskless_load_tempDb;
1950 } else {
1951 dbarray = server.db;
1952 }
1953
1954 rioInitWithConn(&rdb,conn,server.repl_transfer_size);
1955
1956 /* Put the socket in blocking mode to simplify RDB transfer.
1957 * We'll restore it when the RDB is received. */
1958 connBlock(conn);
1959 connRecvTimeout(conn, server.repl_timeout*1000);
1960 startLoading(server.repl_transfer_size, RDBFLAGS_REPLICATION, asyncLoading);
1961
1962 int loadingFailed = 0;
1963 if (rdbLoadRio(&rdb,RDBFLAGS_REPLICATION,&rsi,dbarray) != C_OK) {
1964 /* RDB loading failed. */
1965 serverLog(LL_WARNING,
1966 "Failed trying to load the MASTER synchronization DB "
1967 "from socket: %s", strerror(errno));
1968 loadingFailed = 1;
1969 } else if (usemark) {
1970 /* Verify the end mark is correct. */
1971 if (!rioRead(&rdb, buf, CONFIG_RUN_ID_SIZE) ||
1972 memcmp(buf, eofmark, CONFIG_RUN_ID_SIZE) != 0)
1973 {
1974 serverLog(LL_WARNING, "Replication stream EOF marker is broken");
1975 loadingFailed = 1;
1976 }
1977 }
1978
1979 if (loadingFailed) {
1980 stopLoading(0);
1981 cancelReplicationHandshake(1);
1982 rioFreeConn(&rdb, NULL);
1983
1984 if (server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB) {
1985 /* Discard potentially partially loaded tempDb. */
1986 moduleFireServerEvent(REDISMODULE_EVENT_REPL_ASYNC_LOAD,
1987 REDISMODULE_SUBEVENT_REPL_ASYNC_LOAD_ABORTED,
1988 NULL);
1989
1990 disklessLoadDiscardTempDb(diskless_load_tempDb);
1991 serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Discarding temporary DB in background");
1992 } else {
1993 /* Remove the half-loaded data in case we started with an empty replica. */
1994 emptyDb(-1,empty_db_flags,replicationEmptyDbCallback);
1995 }
1996
1997 /* Note that there's no point in restarting the AOF on SYNC
1998 * failure, it'll be restarted when sync succeeds or the replica
1999 * gets promoted. */
2000 return;
2001 }
2002
2003 /* RDB loading succeeded if we reach this point. */
2004 if (server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB) {
2005 /* We will soon swap main db with tempDb and replicas will start
2006 * to apply data from new master, we must discard the cached
2007 * master structure and force resync of sub-replicas. */
2008 replicationAttachToNewMaster();
2009
2010 serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Swapping active DB with loaded DB");
2011 swapMainDbWithTempDb(diskless_load_tempDb);
2012
2013 moduleFireServerEvent(REDISMODULE_EVENT_REPL_ASYNC_LOAD,
2014 REDISMODULE_SUBEVENT_REPL_ASYNC_LOAD_COMPLETED,
2015 NULL);
2016
2017 /* Delete the old db as it's useless now. */
2018 disklessLoadDiscardTempDb(diskless_load_tempDb);
2019 serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Discarding old DB in background");
2020 }
2021
2022 /* Inform about db change, as replication was diskless and didn't cause a save. */
2023 server.dirty++;
2024
2025 stopLoading(1);
2026
2027 /* Cleanup and restore the socket to the original state to continue
2028 * with the normal replication. */
2029 rioFreeConn(&rdb, NULL);
2030 connNonBlock(conn);
2031 connRecvTimeout(conn,0);
2032 } else {
2033 /* Ensure background save doesn't overwrite synced data */
2034 if (server.child_type == CHILD_TYPE_RDB) {
2035 serverLog(LL_NOTICE,
2036 "Replica is about to load the RDB file received from the "
2037 "master, but there is a pending RDB child running. "
2038 "Killing process %ld and removing its temp file to avoid "
2039 "any race",
2040 (long) server.child_pid);
2041 killRDBChild();
2042 }
2043
2044 /* Make sure the new file (also used for persistence) is fully synced
2045 * (not covered by earlier calls to rdb_fsync_range). */
2046 if (fsync(server.repl_transfer_fd) == -1) {
2047 serverLog(LL_WARNING,
2048 "Failed trying to sync the temp DB to disk in "
2049 "MASTER <-> REPLICA synchronization: %s",
2050 strerror(errno));
2051 cancelReplicationHandshake(1);
2052 return;
2053 }
2054
2055 /* Rename rdb like renaming rewrite aof asynchronously. */
2056 int old_rdb_fd = open(server.rdb_filename,O_RDONLY|O_NONBLOCK);
2057 if (rename(server.repl_transfer_tmpfile,server.rdb_filename) == -1) {
2058 serverLog(LL_WARNING,
2059 "Failed trying to rename the temp DB into %s in "
2060 "MASTER <-> REPLICA synchronization: %s",
2061 server.rdb_filename, strerror(errno));
2062 cancelReplicationHandshake(1);
2063 if (old_rdb_fd != -1) close(old_rdb_fd);
2064 return;
2065 }
2066 /* Close old rdb asynchronously. */
2067 if (old_rdb_fd != -1) bioCreateCloseJob(old_rdb_fd);
2068
2069 if (rdbLoad(server.rdb_filename,&rsi,RDBFLAGS_REPLICATION) != C_OK) {
2070 serverLog(LL_WARNING,
2071 "Failed trying to load the MASTER synchronization "
2072 "DB from disk: %s", strerror(errno));
2073 cancelReplicationHandshake(1);
2074 if (server.rdb_del_sync_files && allPersistenceDisabled()) {
2075 serverLog(LL_NOTICE,"Removing the RDB file obtained from "
2076 "the master. This replica has persistence "
2077 "disabled");
2078 bg_unlink(server.rdb_filename);
2079 }
2080 /* Note that there's no point in restarting the AOF on sync failure,
2081 it'll be restarted when sync succeeds or replica promoted. */
2082 return;
2083 }
2084
2085 /* Cleanup. */
2086 if (server.rdb_del_sync_files && allPersistenceDisabled()) {
2087 serverLog(LL_NOTICE,"Removing the RDB file obtained from "
2088 "the master. This replica has persistence "
2089 "disabled");
2090 bg_unlink(server.rdb_filename);
2091 }
2092
2093 zfree(server.repl_transfer_tmpfile);
2094 close(server.repl_transfer_fd);
2095 server.repl_transfer_fd = -1;
2096 server.repl_transfer_tmpfile = NULL;
2097 }
2098
2099 /* Final setup of the connected slave <- master link */
2100 replicationCreateMasterClient(server.repl_transfer_s,rsi.repl_stream_db);
2101 server.repl_state = REPL_STATE_CONNECTED;
2102 server.repl_down_since = 0;
2103
2104 /* Fire the master link modules event. */
2105 moduleFireServerEvent(REDISMODULE_EVENT_MASTER_LINK_CHANGE,
2106 REDISMODULE_SUBEVENT_MASTER_LINK_UP,
2107 NULL);
2108
2109 /* After a full resynchronization we use the replication ID and
2110 * offset of the master. The secondary ID / offset are cleared since
2111 * we are starting a new history. */
2112 memcpy(server.replid,server.master->replid,sizeof(server.replid));
2113 server.master_repl_offset = server.master->reploff;
2114 clearReplicationId2();
2115
2116 /* Let's create the replication backlog if needed. Slaves need to
2117 * accumulate the backlog regardless of the fact they have sub-slaves
2118 * or not, in order to behave correctly if they are promoted to
2119 * masters after a failover. */
2120 if (server.repl_backlog == NULL) createReplicationBacklog();
2121 serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Finished with success");
2122
2123 if (server.supervised_mode == SUPERVISED_SYSTEMD) {
2124 redisCommunicateSystemd("STATUS=MASTER <-> REPLICA sync: Finished with success. Ready to accept connections in read-write mode.\n");
2125 }
2126
2127 /* Send the initial ACK immediately to put this replica in online state. */
2128 if (usemark) replicationSendAck();
2129
2130 /* Restart the AOF subsystem now that we finished the sync. This
2131 * will trigger an AOF rewrite, and when done will start appending
2132 * to the new file. */
2133 if (server.aof_enabled) restartAOFAfterSYNC();
2134 return;
2135
2136 error:
2137 cancelReplicationHandshake(1);
2138 return;
2139 }
2140
receiveSynchronousResponse(connection * conn)2141 char *receiveSynchronousResponse(connection *conn) {
2142 char buf[256];
2143 /* Read the reply from the server. */
2144 if (connSyncReadLine(conn,buf,sizeof(buf),server.repl_syncio_timeout*1000) == -1)
2145 {
2146 return sdscatprintf(sdsempty(),"-Reading from master: %s",
2147 strerror(errno));
2148 }
2149 server.repl_transfer_lastio = server.unixtime;
2150 return sdsnew(buf);
2151 }
2152
2153 /* Send a pre-formatted multi-bulk command to the connection. */
sendCommandRaw(connection * conn,sds cmd)2154 char* sendCommandRaw(connection *conn, sds cmd) {
2155 if (connSyncWrite(conn,cmd,sdslen(cmd),server.repl_syncio_timeout*1000) == -1) {
2156 return sdscatprintf(sdsempty(),"-Writing to master: %s",
2157 connGetLastError(conn));
2158 }
2159 return NULL;
2160 }
2161
2162 /* Compose a multi-bulk command and send it to the connection.
2163 * Used to send AUTH and REPLCONF commands to the master before starting the
2164 * replication.
2165 *
2166 * Takes a list of char* arguments, terminated by a NULL argument.
2167 *
2168 * The command returns an sds string representing the result of the
2169 * operation. On error the first byte is a "-".
2170 */
sendCommand(connection * conn,...)2171 char *sendCommand(connection *conn, ...) {
2172 va_list ap;
2173 sds cmd = sdsempty();
2174 sds cmdargs = sdsempty();
2175 size_t argslen = 0;
2176 char *arg;
2177
2178 /* Create the command to send to the master, we use redis binary
2179 * protocol to make sure correct arguments are sent. This function
2180 * is not safe for all binary data. */
2181 va_start(ap,conn);
2182 while(1) {
2183 arg = va_arg(ap, char*);
2184 if (arg == NULL) break;
2185 cmdargs = sdscatprintf(cmdargs,"$%zu\r\n%s\r\n",strlen(arg),arg);
2186 argslen++;
2187 }
2188
2189 cmd = sdscatprintf(cmd,"*%zu\r\n",argslen);
2190 cmd = sdscatsds(cmd,cmdargs);
2191 sdsfree(cmdargs);
2192
2193 va_end(ap);
2194 char* err = sendCommandRaw(conn, cmd);
2195 sdsfree(cmd);
2196 if(err)
2197 return err;
2198 return NULL;
2199 }
2200
2201 /* Compose a multi-bulk command and send it to the connection.
2202 * Used to send AUTH and REPLCONF commands to the master before starting the
2203 * replication.
2204 *
2205 * argv_lens is optional, when NULL, strlen is used.
2206 *
2207 * The command returns an sds string representing the result of the
2208 * operation. On error the first byte is a "-".
2209 */
sendCommandArgv(connection * conn,int argc,char ** argv,size_t * argv_lens)2210 char *sendCommandArgv(connection *conn, int argc, char **argv, size_t *argv_lens) {
2211 sds cmd = sdsempty();
2212 char *arg;
2213 int i;
2214
2215 /* Create the command to send to the master. */
2216 cmd = sdscatfmt(cmd,"*%i\r\n",argc);
2217 for (i=0; i<argc; i++) {
2218 int len;
2219 arg = argv[i];
2220 len = argv_lens ? argv_lens[i] : strlen(arg);
2221 cmd = sdscatfmt(cmd,"$%i\r\n",len);
2222 cmd = sdscatlen(cmd,arg,len);
2223 cmd = sdscatlen(cmd,"\r\n",2);
2224 }
2225 char* err = sendCommandRaw(conn, cmd);
2226 sdsfree(cmd);
2227 if (err)
2228 return err;
2229 return NULL;
2230 }
2231
2232 /* Try a partial resynchronization with the master if we are about to reconnect.
2233 * If there is no cached master structure, at least try to issue a
2234 * "PSYNC ? -1" command in order to trigger a full resync using the PSYNC
2235 * command in order to obtain the master replid and the master replication
2236 * global offset.
2237 *
2238 * This function is designed to be called from syncWithMaster(), so the
2239 * following assumptions are made:
2240 *
2241 * 1) We pass the function an already connected socket "fd".
2242 * 2) This function does not close the file descriptor "fd". However in case
2243 * of successful partial resynchronization, the function will reuse
2244 * 'fd' as file descriptor of the server.master client structure.
2245 *
2246 * The function is split in two halves: if read_reply is 0, the function
2247 * writes the PSYNC command on the socket, and a new function call is
2248 * needed, with read_reply set to 1, in order to read the reply of the
2249 * command. This is useful in order to support non blocking operations, so
2250 * that we write, return into the event loop, and read when there are data.
2251 *
2252 * When read_reply is 0 the function returns PSYNC_WRITE_ERR if there
2253 * was a write error, or PSYNC_WAIT_REPLY to signal we need another call
2254 * with read_reply set to 1. However even when read_reply is set to 1
2255 * the function may return PSYNC_WAIT_REPLY again to signal there were
2256 * insufficient data to read to complete its work. We should re-enter
2257 * into the event loop and wait in such a case.
2258 *
2259 * The function returns:
2260 *
2261 * PSYNC_CONTINUE: If the PSYNC command succeeded and we can continue.
2262 * PSYNC_FULLRESYNC: If PSYNC is supported but a full resync is needed.
2263 * In this case the master replid and global replication
2264 * offset is saved.
2265 * PSYNC_NOT_SUPPORTED: If the server does not understand PSYNC at all and
2266 * the caller should fall back to SYNC.
2267 * PSYNC_WRITE_ERROR: There was an error writing the command to the socket.
2268 * PSYNC_WAIT_REPLY: Call again the function with read_reply set to 1.
2269 * PSYNC_TRY_LATER: Master is currently in a transient error condition.
2270 *
2271 * Notable side effects:
2272 *
2273 * 1) As a side effect of the function call the function removes the readable
2274 * event handler from "fd", unless the return value is PSYNC_WAIT_REPLY.
2275 * 2) server.master_initial_offset is set to the right value according
2276 * to the master reply. This will be used to populate the 'server.master'
2277 * structure replication offset.
2278 */
2279
2280 #define PSYNC_WRITE_ERROR 0
2281 #define PSYNC_WAIT_REPLY 1
2282 #define PSYNC_CONTINUE 2
2283 #define PSYNC_FULLRESYNC 3
2284 #define PSYNC_NOT_SUPPORTED 4
2285 #define PSYNC_TRY_LATER 5
slaveTryPartialResynchronization(connection * conn,int read_reply)2286 int slaveTryPartialResynchronization(connection *conn, int read_reply) {
2287 char *psync_replid;
2288 char psync_offset[32];
2289 sds reply;
2290
2291 /* Writing half */
2292 if (!read_reply) {
2293 /* Initially set master_initial_offset to -1 to mark the current
2294 * master replid and offset as not valid. Later if we'll be able to do
2295 * a FULL resync using the PSYNC command we'll set the offset at the
2296 * right value, so that this information will be propagated to the
2297 * client structure representing the master into server.master. */
2298 server.master_initial_offset = -1;
2299
2300 if (server.cached_master) {
2301 psync_replid = server.cached_master->replid;
2302 snprintf(psync_offset,sizeof(psync_offset),"%lld", server.cached_master->reploff+1);
2303 serverLog(LL_NOTICE,"Trying a partial resynchronization (request %s:%s).", psync_replid, psync_offset);
2304 } else {
2305 serverLog(LL_NOTICE,"Partial resynchronization not possible (no cached master)");
2306 psync_replid = "?";
2307 memcpy(psync_offset,"-1",3);
2308 }
2309
2310 /* Issue the PSYNC command, if this is a master with a failover in
2311 * progress then send the failover argument to the replica to cause it
2312 * to become a master */
2313 if (server.failover_state == FAILOVER_IN_PROGRESS) {
2314 reply = sendCommand(conn,"PSYNC",psync_replid,psync_offset,"FAILOVER",NULL);
2315 } else {
2316 reply = sendCommand(conn,"PSYNC",psync_replid,psync_offset,NULL);
2317 }
2318
2319 if (reply != NULL) {
2320 serverLog(LL_WARNING,"Unable to send PSYNC to master: %s",reply);
2321 sdsfree(reply);
2322 connSetReadHandler(conn, NULL);
2323 return PSYNC_WRITE_ERROR;
2324 }
2325 return PSYNC_WAIT_REPLY;
2326 }
2327
2328 /* Reading half */
2329 reply = receiveSynchronousResponse(conn);
2330 if (sdslen(reply) == 0) {
2331 /* The master may send empty newlines after it receives PSYNC
2332 * and before to reply, just to keep the connection alive. */
2333 sdsfree(reply);
2334 return PSYNC_WAIT_REPLY;
2335 }
2336
2337 connSetReadHandler(conn, NULL);
2338
2339 if (!strncmp(reply,"+FULLRESYNC",11)) {
2340 char *replid = NULL, *offset = NULL;
2341
2342 /* FULL RESYNC, parse the reply in order to extract the replid
2343 * and the replication offset. */
2344 replid = strchr(reply,' ');
2345 if (replid) {
2346 replid++;
2347 offset = strchr(replid,' ');
2348 if (offset) offset++;
2349 }
2350 if (!replid || !offset || (offset-replid-1) != CONFIG_RUN_ID_SIZE) {
2351 serverLog(LL_WARNING,
2352 "Master replied with wrong +FULLRESYNC syntax.");
2353 /* This is an unexpected condition, actually the +FULLRESYNC
2354 * reply means that the master supports PSYNC, but the reply
2355 * format seems wrong. To stay safe we blank the master
2356 * replid to make sure next PSYNCs will fail. */
2357 memset(server.master_replid,0,CONFIG_RUN_ID_SIZE+1);
2358 } else {
2359 memcpy(server.master_replid, replid, offset-replid-1);
2360 server.master_replid[CONFIG_RUN_ID_SIZE] = '\0';
2361 server.master_initial_offset = strtoll(offset,NULL,10);
2362 serverLog(LL_NOTICE,"Full resync from master: %s:%lld",
2363 server.master_replid,
2364 server.master_initial_offset);
2365 }
2366 sdsfree(reply);
2367 return PSYNC_FULLRESYNC;
2368 }
2369
2370 if (!strncmp(reply,"+CONTINUE",9)) {
2371 /* Partial resync was accepted. */
2372 serverLog(LL_NOTICE,
2373 "Successful partial resynchronization with master.");
2374
2375 /* Check the new replication ID advertised by the master. If it
2376 * changed, we need to set the new ID as primary ID, and set
2377 * secondary ID as the old master ID up to the current offset, so
2378 * that our sub-slaves will be able to PSYNC with us after a
2379 * disconnection. */
2380 char *start = reply+10;
2381 char *end = reply+9;
2382 while(end[0] != '\r' && end[0] != '\n' && end[0] != '\0') end++;
2383 if (end-start == CONFIG_RUN_ID_SIZE) {
2384 char new[CONFIG_RUN_ID_SIZE+1];
2385 memcpy(new,start,CONFIG_RUN_ID_SIZE);
2386 new[CONFIG_RUN_ID_SIZE] = '\0';
2387
2388 if (strcmp(new,server.cached_master->replid)) {
2389 /* Master ID changed. */
2390 serverLog(LL_WARNING,"Master replication ID changed to %s",new);
2391
2392 /* Set the old ID as our ID2, up to the current offset+1. */
2393 memcpy(server.replid2,server.cached_master->replid,
2394 sizeof(server.replid2));
2395 server.second_replid_offset = server.master_repl_offset+1;
2396
2397 /* Update the cached master ID and our own primary ID to the
2398 * new one. */
2399 memcpy(server.replid,new,sizeof(server.replid));
2400 memcpy(server.cached_master->replid,new,sizeof(server.replid));
2401
2402 /* Disconnect all the sub-slaves: they need to be notified. */
2403 disconnectSlaves();
2404 }
2405 }
2406
2407 /* Setup the replication to continue. */
2408 sdsfree(reply);
2409 replicationResurrectCachedMaster(conn);
2410
2411 /* If this instance was restarted and we read the metadata to
2412 * PSYNC from the persistence file, our replication backlog could
2413 * be still not initialized. Create it. */
2414 if (server.repl_backlog == NULL) createReplicationBacklog();
2415 return PSYNC_CONTINUE;
2416 }
2417
2418 /* If we reach this point we received either an error (since the master does
2419 * not understand PSYNC or because it is in a special state and cannot
2420 * serve our request), or an unexpected reply from the master.
2421 *
2422 * Return PSYNC_NOT_SUPPORTED on errors we don't understand, otherwise
2423 * return PSYNC_TRY_LATER if we believe this is a transient error. */
2424
2425 if (!strncmp(reply,"-NOMASTERLINK",13) ||
2426 !strncmp(reply,"-LOADING",8))
2427 {
2428 serverLog(LL_NOTICE,
2429 "Master is currently unable to PSYNC "
2430 "but should be in the future: %s", reply);
2431 sdsfree(reply);
2432 return PSYNC_TRY_LATER;
2433 }
2434
2435 if (strncmp(reply,"-ERR",4)) {
2436 /* If it's not an error, log the unexpected event. */
2437 serverLog(LL_WARNING,
2438 "Unexpected reply to PSYNC from master: %s", reply);
2439 } else {
2440 serverLog(LL_NOTICE,
2441 "Master does not support PSYNC or is in "
2442 "error state (reply: %s)", reply);
2443 }
2444 sdsfree(reply);
2445 return PSYNC_NOT_SUPPORTED;
2446 }
2447
2448 /* This handler fires when the non blocking connect was able to
2449 * establish a connection with the master. */
syncWithMaster(connection * conn)2450 void syncWithMaster(connection *conn) {
2451 char tmpfile[256], *err = NULL;
2452 int dfd = -1, maxtries = 5;
2453 int psync_result;
2454
2455 /* If this event fired after the user turned the instance into a master
2456 * with SLAVEOF NO ONE we must just return ASAP. */
2457 if (server.repl_state == REPL_STATE_NONE) {
2458 connClose(conn);
2459 return;
2460 }
2461
2462 /* Check for errors in the socket: after a non blocking connect() we
2463 * may find that the socket is in error state. */
2464 if (connGetState(conn) != CONN_STATE_CONNECTED) {
2465 serverLog(LL_WARNING,"Error condition on socket for SYNC: %s",
2466 connGetLastError(conn));
2467 goto error;
2468 }
2469
2470 /* Send a PING to check the master is able to reply without errors. */
2471 if (server.repl_state == REPL_STATE_CONNECTING) {
2472 serverLog(LL_NOTICE,"Non blocking connect for SYNC fired the event.");
2473 /* Delete the writable event so that the readable event remains
2474 * registered and we can wait for the PONG reply. */
2475 connSetReadHandler(conn, syncWithMaster);
2476 connSetWriteHandler(conn, NULL);
2477 server.repl_state = REPL_STATE_RECEIVE_PING_REPLY;
2478 /* Send the PING, don't check for errors at all, we have the timeout
2479 * that will take care about this. */
2480 err = sendCommand(conn,"PING",NULL);
2481 if (err) goto write_error;
2482 return;
2483 }
2484
2485 /* Receive the PONG command. */
2486 if (server.repl_state == REPL_STATE_RECEIVE_PING_REPLY) {
2487 err = receiveSynchronousResponse(conn);
2488
2489 /* We accept only two replies as valid, a positive +PONG reply
2490 * (we just check for "+") or an authentication error.
2491 * Note that older versions of Redis replied with "operation not
2492 * permitted" instead of using a proper error code, so we test
2493 * both. */
2494 if (err[0] != '+' &&
2495 strncmp(err,"-NOAUTH",7) != 0 &&
2496 strncmp(err,"-NOPERM",7) != 0 &&
2497 strncmp(err,"-ERR operation not permitted",28) != 0)
2498 {
2499 serverLog(LL_WARNING,"Error reply to PING from master: '%s'",err);
2500 sdsfree(err);
2501 goto error;
2502 } else {
2503 serverLog(LL_NOTICE,
2504 "Master replied to PING, replication can continue...");
2505 }
2506 sdsfree(err);
2507 err = NULL;
2508 server.repl_state = REPL_STATE_SEND_HANDSHAKE;
2509 }
2510
2511 if (server.repl_state == REPL_STATE_SEND_HANDSHAKE) {
2512 /* AUTH with the master if required. */
2513 if (server.masterauth) {
2514 char *args[3] = {"AUTH",NULL,NULL};
2515 size_t lens[3] = {4,0,0};
2516 int argc = 1;
2517 if (server.masteruser) {
2518 args[argc] = server.masteruser;
2519 lens[argc] = strlen(server.masteruser);
2520 argc++;
2521 }
2522 args[argc] = server.masterauth;
2523 lens[argc] = sdslen(server.masterauth);
2524 argc++;
2525 err = sendCommandArgv(conn, argc, args, lens);
2526 if (err) goto write_error;
2527 }
2528
2529 /* Set the slave port, so that Master's INFO command can list the
2530 * slave listening port correctly. */
2531 {
2532 int port;
2533 if (server.slave_announce_port)
2534 port = server.slave_announce_port;
2535 else if (server.tls_replication && server.tls_port)
2536 port = server.tls_port;
2537 else
2538 port = server.port;
2539 sds portstr = sdsfromlonglong(port);
2540 err = sendCommand(conn,"REPLCONF",
2541 "listening-port",portstr, NULL);
2542 sdsfree(portstr);
2543 if (err) goto write_error;
2544 }
2545
2546 /* Set the slave ip, so that Master's INFO command can list the
2547 * slave IP address port correctly in case of port forwarding or NAT.
2548 * Skip REPLCONF ip-address if there is no slave-announce-ip option set. */
2549 if (server.slave_announce_ip) {
2550 err = sendCommand(conn,"REPLCONF",
2551 "ip-address",server.slave_announce_ip, NULL);
2552 if (err) goto write_error;
2553 }
2554
2555 /* Inform the master of our (slave) capabilities.
2556 *
2557 * EOF: supports EOF-style RDB transfer for diskless replication.
2558 * PSYNC2: supports PSYNC v2, so understands +CONTINUE <new repl ID>.
2559 *
2560 * The master will ignore capabilities it does not understand. */
2561 err = sendCommand(conn,"REPLCONF",
2562 "capa","eof","capa","psync2",NULL);
2563 if (err) goto write_error;
2564
2565 server.repl_state = REPL_STATE_RECEIVE_AUTH_REPLY;
2566 return;
2567 }
2568
2569 if (server.repl_state == REPL_STATE_RECEIVE_AUTH_REPLY && !server.masterauth)
2570 server.repl_state = REPL_STATE_RECEIVE_PORT_REPLY;
2571
2572 /* Receive AUTH reply. */
2573 if (server.repl_state == REPL_STATE_RECEIVE_AUTH_REPLY) {
2574 err = receiveSynchronousResponse(conn);
2575 if (err[0] == '-') {
2576 serverLog(LL_WARNING,"Unable to AUTH to MASTER: %s",err);
2577 sdsfree(err);
2578 goto error;
2579 }
2580 sdsfree(err);
2581 err = NULL;
2582 server.repl_state = REPL_STATE_RECEIVE_PORT_REPLY;
2583 return;
2584 }
2585
2586 /* Receive REPLCONF listening-port reply. */
2587 if (server.repl_state == REPL_STATE_RECEIVE_PORT_REPLY) {
2588 err = receiveSynchronousResponse(conn);
2589 /* Ignore the error if any, not all the Redis versions support
2590 * REPLCONF listening-port. */
2591 if (err[0] == '-') {
2592 serverLog(LL_NOTICE,"(Non critical) Master does not understand "
2593 "REPLCONF listening-port: %s", err);
2594 }
2595 sdsfree(err);
2596 server.repl_state = REPL_STATE_RECEIVE_IP_REPLY;
2597 return;
2598 }
2599
2600 if (server.repl_state == REPL_STATE_RECEIVE_IP_REPLY && !server.slave_announce_ip)
2601 server.repl_state = REPL_STATE_RECEIVE_CAPA_REPLY;
2602
2603 /* Receive REPLCONF ip-address reply. */
2604 if (server.repl_state == REPL_STATE_RECEIVE_IP_REPLY) {
2605 err = receiveSynchronousResponse(conn);
2606 /* Ignore the error if any, not all the Redis versions support
2607 * REPLCONF ip-address. */
2608 if (err[0] == '-') {
2609 serverLog(LL_NOTICE,"(Non critical) Master does not understand "
2610 "REPLCONF ip-address: %s", err);
2611 }
2612 sdsfree(err);
2613 server.repl_state = REPL_STATE_RECEIVE_CAPA_REPLY;
2614 return;
2615 }
2616
2617 /* Receive CAPA reply. */
2618 if (server.repl_state == REPL_STATE_RECEIVE_CAPA_REPLY) {
2619 err = receiveSynchronousResponse(conn);
2620 /* Ignore the error if any, not all the Redis versions support
2621 * REPLCONF capa. */
2622 if (err[0] == '-') {
2623 serverLog(LL_NOTICE,"(Non critical) Master does not understand "
2624 "REPLCONF capa: %s", err);
2625 }
2626 sdsfree(err);
2627 err = NULL;
2628 server.repl_state = REPL_STATE_SEND_PSYNC;
2629 }
2630
2631 /* Try a partial resynchronization. If we don't have a cached master
2632 * slaveTryPartialResynchronization() will at least try to use PSYNC
2633 * to start a full resynchronization so that we get the master replid
2634 * and the global offset, to try a partial resync at the next
2635 * reconnection attempt. */
2636 if (server.repl_state == REPL_STATE_SEND_PSYNC) {
2637 if (slaveTryPartialResynchronization(conn,0) == PSYNC_WRITE_ERROR) {
2638 err = sdsnew("Write error sending the PSYNC command.");
2639 abortFailover("Write error to failover target");
2640 goto write_error;
2641 }
2642 server.repl_state = REPL_STATE_RECEIVE_PSYNC_REPLY;
2643 return;
2644 }
2645
2646 /* If reached this point, we should be in REPL_STATE_RECEIVE_PSYNC. */
2647 if (server.repl_state != REPL_STATE_RECEIVE_PSYNC_REPLY) {
2648 serverLog(LL_WARNING,"syncWithMaster(): state machine error, "
2649 "state should be RECEIVE_PSYNC but is %d",
2650 server.repl_state);
2651 goto error;
2652 }
2653
2654 psync_result = slaveTryPartialResynchronization(conn,1);
2655 if (psync_result == PSYNC_WAIT_REPLY) return; /* Try again later... */
2656
2657 /* Check the status of the planned failover. We expect PSYNC_CONTINUE,
2658 * but there is nothing technically wrong with a full resync which
2659 * could happen in edge cases. */
2660 if (server.failover_state == FAILOVER_IN_PROGRESS) {
2661 if (psync_result == PSYNC_CONTINUE || psync_result == PSYNC_FULLRESYNC) {
2662 clearFailoverState();
2663 } else {
2664 abortFailover("Failover target rejected psync request");
2665 return;
2666 }
2667 }
2668
2669 /* If the master is in an transient error, we should try to PSYNC
2670 * from scratch later, so go to the error path. This happens when
2671 * the server is loading the dataset or is not connected with its
2672 * master and so forth. */
2673 if (psync_result == PSYNC_TRY_LATER) goto error;
2674
2675 /* Note: if PSYNC does not return WAIT_REPLY, it will take care of
2676 * uninstalling the read handler from the file descriptor. */
2677
2678 if (psync_result == PSYNC_CONTINUE) {
2679 serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Master accepted a Partial Resynchronization.");
2680 if (server.supervised_mode == SUPERVISED_SYSTEMD) {
2681 redisCommunicateSystemd("STATUS=MASTER <-> REPLICA sync: Partial Resynchronization accepted. Ready to accept connections in read-write mode.\n");
2682 }
2683 return;
2684 }
2685
2686 /* Fall back to SYNC if needed. Otherwise psync_result == PSYNC_FULLRESYNC
2687 * and the server.master_replid and master_initial_offset are
2688 * already populated. */
2689 if (psync_result == PSYNC_NOT_SUPPORTED) {
2690 serverLog(LL_NOTICE,"Retrying with SYNC...");
2691 if (connSyncWrite(conn,"SYNC\r\n",6,server.repl_syncio_timeout*1000) == -1) {
2692 serverLog(LL_WARNING,"I/O error writing to MASTER: %s",
2693 strerror(errno));
2694 goto error;
2695 }
2696 }
2697
2698 /* Prepare a suitable temp file for bulk transfer */
2699 if (!useDisklessLoad()) {
2700 while(maxtries--) {
2701 snprintf(tmpfile,256,
2702 "temp-%d.%ld.rdb",(int)server.unixtime,(long int)getpid());
2703 dfd = open(tmpfile,O_CREAT|O_WRONLY|O_EXCL,0644);
2704 if (dfd != -1) break;
2705 sleep(1);
2706 }
2707 if (dfd == -1) {
2708 serverLog(LL_WARNING,"Opening the temp file needed for MASTER <-> REPLICA synchronization: %s",strerror(errno));
2709 goto error;
2710 }
2711 server.repl_transfer_tmpfile = zstrdup(tmpfile);
2712 server.repl_transfer_fd = dfd;
2713 }
2714
2715 /* Setup the non blocking download of the bulk file. */
2716 if (connSetReadHandler(conn, readSyncBulkPayload)
2717 == C_ERR)
2718 {
2719 char conninfo[CONN_INFO_LEN];
2720 serverLog(LL_WARNING,
2721 "Can't create readable event for SYNC: %s (%s)",
2722 strerror(errno), connGetInfo(conn, conninfo, sizeof(conninfo)));
2723 goto error;
2724 }
2725
2726 server.repl_state = REPL_STATE_TRANSFER;
2727 server.repl_transfer_size = -1;
2728 server.repl_transfer_read = 0;
2729 server.repl_transfer_last_fsync_off = 0;
2730 server.repl_transfer_lastio = server.unixtime;
2731 return;
2732
2733 error:
2734 if (dfd != -1) close(dfd);
2735 connClose(conn);
2736 server.repl_transfer_s = NULL;
2737 if (server.repl_transfer_fd != -1)
2738 close(server.repl_transfer_fd);
2739 if (server.repl_transfer_tmpfile)
2740 zfree(server.repl_transfer_tmpfile);
2741 server.repl_transfer_tmpfile = NULL;
2742 server.repl_transfer_fd = -1;
2743 server.repl_state = REPL_STATE_CONNECT;
2744 return;
2745
2746 write_error: /* Handle sendCommand() errors. */
2747 serverLog(LL_WARNING,"Sending command to master in replication handshake: %s", err);
2748 sdsfree(err);
2749 goto error;
2750 }
2751
connectWithMaster(void)2752 int connectWithMaster(void) {
2753 server.repl_transfer_s = server.tls_replication ? connCreateTLS() : connCreateSocket();
2754 if (connConnect(server.repl_transfer_s, server.masterhost, server.masterport,
2755 server.bind_source_addr, syncWithMaster) == C_ERR) {
2756 serverLog(LL_WARNING,"Unable to connect to MASTER: %s",
2757 connGetLastError(server.repl_transfer_s));
2758 connClose(server.repl_transfer_s);
2759 server.repl_transfer_s = NULL;
2760 return C_ERR;
2761 }
2762
2763
2764 server.repl_transfer_lastio = server.unixtime;
2765 server.repl_state = REPL_STATE_CONNECTING;
2766 serverLog(LL_NOTICE,"MASTER <-> REPLICA sync started");
2767 return C_OK;
2768 }
2769
2770 /* This function can be called when a non blocking connection is currently
2771 * in progress to undo it.
2772 * Never call this function directly, use cancelReplicationHandshake() instead.
2773 */
undoConnectWithMaster(void)2774 void undoConnectWithMaster(void) {
2775 connClose(server.repl_transfer_s);
2776 server.repl_transfer_s = NULL;
2777 }
2778
2779 /* Abort the async download of the bulk dataset while SYNC-ing with master.
2780 * Never call this function directly, use cancelReplicationHandshake() instead.
2781 */
replicationAbortSyncTransfer(void)2782 void replicationAbortSyncTransfer(void) {
2783 serverAssert(server.repl_state == REPL_STATE_TRANSFER);
2784 undoConnectWithMaster();
2785 if (server.repl_transfer_fd!=-1) {
2786 close(server.repl_transfer_fd);
2787 bg_unlink(server.repl_transfer_tmpfile);
2788 zfree(server.repl_transfer_tmpfile);
2789 server.repl_transfer_tmpfile = NULL;
2790 server.repl_transfer_fd = -1;
2791 }
2792 }
2793
2794 /* This function aborts a non blocking replication attempt if there is one
2795 * in progress, by canceling the non-blocking connect attempt or
2796 * the initial bulk transfer.
2797 *
2798 * If there was a replication handshake in progress 1 is returned and
2799 * the replication state (server.repl_state) set to REPL_STATE_CONNECT.
2800 *
2801 * Otherwise zero is returned and no operation is performed at all. */
cancelReplicationHandshake(int reconnect)2802 int cancelReplicationHandshake(int reconnect) {
2803 if (server.repl_state == REPL_STATE_TRANSFER) {
2804 replicationAbortSyncTransfer();
2805 server.repl_state = REPL_STATE_CONNECT;
2806 } else if (server.repl_state == REPL_STATE_CONNECTING ||
2807 slaveIsInHandshakeState())
2808 {
2809 undoConnectWithMaster();
2810 server.repl_state = REPL_STATE_CONNECT;
2811 } else {
2812 return 0;
2813 }
2814
2815 if (!reconnect)
2816 return 1;
2817
2818 /* try to re-connect without waiting for replicationCron, this is needed
2819 * for the "diskless loading short read" test. */
2820 serverLog(LL_NOTICE,"Reconnecting to MASTER %s:%d after failure",
2821 server.masterhost, server.masterport);
2822 connectWithMaster();
2823
2824 return 1;
2825 }
2826
2827 /* Set replication to the specified master address and port. */
replicationSetMaster(char * ip,int port)2828 void replicationSetMaster(char *ip, int port) {
2829 int was_master = server.masterhost == NULL;
2830
2831 sdsfree(server.masterhost);
2832 server.masterhost = NULL;
2833 if (server.master) {
2834 freeClient(server.master);
2835 }
2836 disconnectAllBlockedClients(); /* Clients blocked in master, now slave. */
2837
2838 /* Setting masterhost only after the call to freeClient since it calls
2839 * replicationHandleMasterDisconnection which can trigger a re-connect
2840 * directly from within that call. */
2841 server.masterhost = sdsnew(ip);
2842 server.masterport = port;
2843
2844 /* Update oom_score_adj */
2845 setOOMScoreAdj(-1);
2846
2847 /* Here we don't disconnect with replicas, since they may hopefully be able
2848 * to partially resync with us. We will disconnect with replicas and force
2849 * them to resync with us when changing replid on partially resync with new
2850 * master, or finishing transferring RDB and preparing loading DB on full
2851 * sync with new master. */
2852
2853 cancelReplicationHandshake(0);
2854 /* Before destroying our master state, create a cached master using
2855 * our own parameters, to later PSYNC with the new master. */
2856 if (was_master) {
2857 replicationDiscardCachedMaster();
2858 replicationCacheMasterUsingMyself();
2859 }
2860
2861 /* Fire the role change modules event. */
2862 moduleFireServerEvent(REDISMODULE_EVENT_REPLICATION_ROLE_CHANGED,
2863 REDISMODULE_EVENT_REPLROLECHANGED_NOW_REPLICA,
2864 NULL);
2865
2866 /* Fire the master link modules event. */
2867 if (server.repl_state == REPL_STATE_CONNECTED)
2868 moduleFireServerEvent(REDISMODULE_EVENT_MASTER_LINK_CHANGE,
2869 REDISMODULE_SUBEVENT_MASTER_LINK_DOWN,
2870 NULL);
2871
2872 server.repl_state = REPL_STATE_CONNECT;
2873 serverLog(LL_NOTICE,"Connecting to MASTER %s:%d",
2874 server.masterhost, server.masterport);
2875 connectWithMaster();
2876 }
2877
2878 /* Cancel replication, setting the instance as a master itself. */
replicationUnsetMaster(void)2879 void replicationUnsetMaster(void) {
2880 if (server.masterhost == NULL) return; /* Nothing to do. */
2881
2882 /* Fire the master link modules event. */
2883 if (server.repl_state == REPL_STATE_CONNECTED)
2884 moduleFireServerEvent(REDISMODULE_EVENT_MASTER_LINK_CHANGE,
2885 REDISMODULE_SUBEVENT_MASTER_LINK_DOWN,
2886 NULL);
2887
2888 /* Clear masterhost first, since the freeClient calls
2889 * replicationHandleMasterDisconnection which can attempt to re-connect. */
2890 sdsfree(server.masterhost);
2891 server.masterhost = NULL;
2892 if (server.master) freeClient(server.master);
2893 replicationDiscardCachedMaster();
2894 cancelReplicationHandshake(0);
2895 /* When a slave is turned into a master, the current replication ID
2896 * (that was inherited from the master at synchronization time) is
2897 * used as secondary ID up to the current offset, and a new replication
2898 * ID is created to continue with a new replication history. */
2899 shiftReplicationId();
2900 /* Disconnecting all the slaves is required: we need to inform slaves
2901 * of the replication ID change (see shiftReplicationId() call). However
2902 * the slaves will be able to partially resync with us, so it will be
2903 * a very fast reconnection. */
2904 disconnectSlaves();
2905 server.repl_state = REPL_STATE_NONE;
2906
2907 /* We need to make sure the new master will start the replication stream
2908 * with a SELECT statement. This is forced after a full resync, but
2909 * with PSYNC version 2, there is no need for full resync after a
2910 * master switch. */
2911 server.slaveseldb = -1;
2912
2913 /* Update oom_score_adj */
2914 setOOMScoreAdj(-1);
2915
2916 /* Once we turn from slave to master, we consider the starting time without
2917 * slaves (that is used to count the replication backlog time to live) as
2918 * starting from now. Otherwise the backlog will be freed after a
2919 * failover if slaves do not connect immediately. */
2920 server.repl_no_slaves_since = server.unixtime;
2921
2922 /* Reset down time so it'll be ready for when we turn into replica again. */
2923 server.repl_down_since = 0;
2924
2925 /* Fire the role change modules event. */
2926 moduleFireServerEvent(REDISMODULE_EVENT_REPLICATION_ROLE_CHANGED,
2927 REDISMODULE_EVENT_REPLROLECHANGED_NOW_MASTER,
2928 NULL);
2929
2930 /* Restart the AOF subsystem in case we shut it down during a sync when
2931 * we were still a slave. */
2932 if (server.aof_enabled && server.aof_state == AOF_OFF) restartAOFAfterSYNC();
2933 }
2934
2935 /* This function is called when the slave lose the connection with the
2936 * master into an unexpected way. */
replicationHandleMasterDisconnection(void)2937 void replicationHandleMasterDisconnection(void) {
2938 /* Fire the master link modules event. */
2939 if (server.repl_state == REPL_STATE_CONNECTED)
2940 moduleFireServerEvent(REDISMODULE_EVENT_MASTER_LINK_CHANGE,
2941 REDISMODULE_SUBEVENT_MASTER_LINK_DOWN,
2942 NULL);
2943
2944 server.master = NULL;
2945 server.repl_state = REPL_STATE_CONNECT;
2946 server.repl_down_since = server.unixtime;
2947 /* We lost connection with our master, don't disconnect slaves yet,
2948 * maybe we'll be able to PSYNC with our master later. We'll disconnect
2949 * the slaves only if we'll have to do a full resync with our master. */
2950
2951 /* Try to re-connect immediately rather than wait for replicationCron
2952 * waiting 1 second may risk backlog being recycled. */
2953 if (server.masterhost) {
2954 serverLog(LL_NOTICE,"Reconnecting to MASTER %s:%d",
2955 server.masterhost, server.masterport);
2956 connectWithMaster();
2957 }
2958 }
2959
replicaofCommand(client * c)2960 void replicaofCommand(client *c) {
2961 /* SLAVEOF is not allowed in cluster mode as replication is automatically
2962 * configured using the current address of the master node. */
2963 if (server.cluster_enabled) {
2964 addReplyError(c,"REPLICAOF not allowed in cluster mode.");
2965 return;
2966 }
2967
2968 if (server.failover_state != NO_FAILOVER) {
2969 addReplyError(c,"REPLICAOF not allowed while failing over.");
2970 return;
2971 }
2972
2973 /* The special host/port combination "NO" "ONE" turns the instance
2974 * into a master. Otherwise the new master address is set. */
2975 if (!strcasecmp(c->argv[1]->ptr,"no") &&
2976 !strcasecmp(c->argv[2]->ptr,"one")) {
2977 if (server.masterhost) {
2978 replicationUnsetMaster();
2979 sds client = catClientInfoString(sdsempty(),c);
2980 serverLog(LL_NOTICE,"MASTER MODE enabled (user request from '%s')",
2981 client);
2982 sdsfree(client);
2983 }
2984 } else {
2985 long port;
2986
2987 if (c->flags & CLIENT_SLAVE)
2988 {
2989 /* If a client is already a replica they cannot run this command,
2990 * because it involves flushing all replicas (including this
2991 * client) */
2992 addReplyError(c, "Command is not valid when client is a replica.");
2993 return;
2994 }
2995
2996 if (getRangeLongFromObjectOrReply(c, c->argv[2], 0, 65535, &port,
2997 "Invalid master port") != C_OK)
2998 return;
2999
3000 /* Check if we are already attached to the specified master */
3001 if (server.masterhost && !strcasecmp(server.masterhost,c->argv[1]->ptr)
3002 && server.masterport == port) {
3003 serverLog(LL_NOTICE,"REPLICAOF would result into synchronization "
3004 "with the master we are already connected "
3005 "with. No operation performed.");
3006 addReplySds(c,sdsnew("+OK Already connected to specified "
3007 "master\r\n"));
3008 return;
3009 }
3010 /* There was no previous master or the user specified a different one,
3011 * we can continue. */
3012 replicationSetMaster(c->argv[1]->ptr, port);
3013 sds client = catClientInfoString(sdsempty(),c);
3014 serverLog(LL_NOTICE,"REPLICAOF %s:%d enabled (user request from '%s')",
3015 server.masterhost, server.masterport, client);
3016 sdsfree(client);
3017 }
3018 addReply(c,shared.ok);
3019 }
3020
3021 /* ROLE command: provide information about the role of the instance
3022 * (master or slave) and additional information related to replication
3023 * in an easy to process format. */
roleCommand(client * c)3024 void roleCommand(client *c) {
3025 if (server.sentinel_mode) {
3026 sentinelRoleCommand(c);
3027 return;
3028 }
3029
3030 if (server.masterhost == NULL) {
3031 listIter li;
3032 listNode *ln;
3033 void *mbcount;
3034 int slaves = 0;
3035
3036 addReplyArrayLen(c,3);
3037 addReplyBulkCBuffer(c,"master",6);
3038 addReplyLongLong(c,server.master_repl_offset);
3039 mbcount = addReplyDeferredLen(c);
3040 listRewind(server.slaves,&li);
3041 while((ln = listNext(&li))) {
3042 client *slave = ln->value;
3043 char ip[NET_IP_STR_LEN], *slaveaddr = slave->slave_addr;
3044
3045 if (!slaveaddr) {
3046 if (connPeerToString(slave->conn,ip,sizeof(ip),NULL) == -1)
3047 continue;
3048 slaveaddr = ip;
3049 }
3050 if (slave->replstate != SLAVE_STATE_ONLINE) continue;
3051 addReplyArrayLen(c,3);
3052 addReplyBulkCString(c,slaveaddr);
3053 addReplyBulkLongLong(c,slave->slave_listening_port);
3054 addReplyBulkLongLong(c,slave->repl_ack_off);
3055 slaves++;
3056 }
3057 setDeferredArrayLen(c,mbcount,slaves);
3058 } else {
3059 char *slavestate = NULL;
3060
3061 addReplyArrayLen(c,5);
3062 addReplyBulkCBuffer(c,"slave",5);
3063 addReplyBulkCString(c,server.masterhost);
3064 addReplyLongLong(c,server.masterport);
3065 if (slaveIsInHandshakeState()) {
3066 slavestate = "handshake";
3067 } else {
3068 switch(server.repl_state) {
3069 case REPL_STATE_NONE: slavestate = "none"; break;
3070 case REPL_STATE_CONNECT: slavestate = "connect"; break;
3071 case REPL_STATE_CONNECTING: slavestate = "connecting"; break;
3072 case REPL_STATE_TRANSFER: slavestate = "sync"; break;
3073 case REPL_STATE_CONNECTED: slavestate = "connected"; break;
3074 default: slavestate = "unknown"; break;
3075 }
3076 }
3077 addReplyBulkCString(c,slavestate);
3078 addReplyLongLong(c,server.master ? server.master->reploff : -1);
3079 }
3080 }
3081
3082 /* Send a REPLCONF ACK command to the master to inform it about the current
3083 * processed offset. If we are not connected with a master, the command has
3084 * no effects. */
replicationSendAck(void)3085 void replicationSendAck(void) {
3086 client *c = server.master;
3087
3088 if (c != NULL) {
3089 c->flags |= CLIENT_MASTER_FORCE_REPLY;
3090 addReplyArrayLen(c,3);
3091 addReplyBulkCString(c,"REPLCONF");
3092 addReplyBulkCString(c,"ACK");
3093 addReplyBulkLongLong(c,c->reploff);
3094 c->flags &= ~CLIENT_MASTER_FORCE_REPLY;
3095 }
3096 }
3097
3098 /* ---------------------- MASTER CACHING FOR PSYNC -------------------------- */
3099
3100 /* In order to implement partial synchronization we need to be able to cache
3101 * our master's client structure after a transient disconnection.
3102 * It is cached into server.cached_master and flushed away using the following
3103 * functions. */
3104
3105 /* This function is called by freeClient() in order to cache the master
3106 * client structure instead of destroying it. freeClient() will return
3107 * ASAP after this function returns, so every action needed to avoid problems
3108 * with a client that is really "suspended" has to be done by this function.
3109 *
3110 * The other functions that will deal with the cached master are:
3111 *
3112 * replicationDiscardCachedMaster() that will make sure to kill the client
3113 * as for some reason we don't want to use it in the future.
3114 *
3115 * replicationResurrectCachedMaster() that is used after a successful PSYNC
3116 * handshake in order to reactivate the cached master.
3117 */
replicationCacheMaster(client * c)3118 void replicationCacheMaster(client *c) {
3119 serverAssert(server.master != NULL && server.cached_master == NULL);
3120 serverLog(LL_NOTICE,"Caching the disconnected master state.");
3121
3122 /* Unlink the client from the server structures. */
3123 unlinkClient(c);
3124
3125 /* Reset the master client so that's ready to accept new commands:
3126 * we want to discard the non processed query buffers and non processed
3127 * offsets, including pending transactions, already populated arguments,
3128 * pending outputs to the master. */
3129 sdsclear(server.master->querybuf);
3130 sdsclear(server.master->pending_querybuf);
3131 server.master->read_reploff = server.master->reploff;
3132 if (c->flags & CLIENT_MULTI) discardTransaction(c);
3133 listEmpty(c->reply);
3134 c->sentlen = 0;
3135 c->reply_bytes = 0;
3136 c->bufpos = 0;
3137 resetClient(c);
3138
3139 /* Save the master. Server.master will be set to null later by
3140 * replicationHandleMasterDisconnection(). */
3141 server.cached_master = server.master;
3142
3143 /* Invalidate the Peer ID cache. */
3144 if (c->peerid) {
3145 sdsfree(c->peerid);
3146 c->peerid = NULL;
3147 }
3148 /* Invalidate the Sock Name cache. */
3149 if (c->sockname) {
3150 sdsfree(c->sockname);
3151 c->sockname = NULL;
3152 }
3153
3154 /* Caching the master happens instead of the actual freeClient() call,
3155 * so make sure to adjust the replication state. This function will
3156 * also set server.master to NULL. */
3157 replicationHandleMasterDisconnection();
3158 }
3159
3160 /* This function is called when a master is turned into a slave, in order to
3161 * create from scratch a cached master for the new client, that will allow
3162 * to PSYNC with the slave that was promoted as the new master after a
3163 * failover.
3164 *
3165 * Assuming this instance was previously the master instance of the new master,
3166 * the new master will accept its replication ID, and potential also the
3167 * current offset if no data was lost during the failover. So we use our
3168 * current replication ID and offset in order to synthesize a cached master. */
replicationCacheMasterUsingMyself(void)3169 void replicationCacheMasterUsingMyself(void) {
3170 serverLog(LL_NOTICE,
3171 "Before turning into a replica, using my own master parameters "
3172 "to synthesize a cached master: I may be able to synchronize with "
3173 "the new master with just a partial transfer.");
3174
3175 /* This will be used to populate the field server.master->reploff
3176 * by replicationCreateMasterClient(). We'll later set the created
3177 * master as server.cached_master, so the replica will use such
3178 * offset for PSYNC. */
3179 server.master_initial_offset = server.master_repl_offset;
3180
3181 /* The master client we create can be set to any DBID, because
3182 * the new master will start its replication stream with SELECT. */
3183 replicationCreateMasterClient(NULL,-1);
3184
3185 /* Use our own ID / offset. */
3186 memcpy(server.master->replid, server.replid, sizeof(server.replid));
3187
3188 /* Set as cached master. */
3189 unlinkClient(server.master);
3190 server.cached_master = server.master;
3191 server.master = NULL;
3192 }
3193
3194 /* Free a cached master, called when there are no longer the conditions for
3195 * a partial resync on reconnection. */
replicationDiscardCachedMaster(void)3196 void replicationDiscardCachedMaster(void) {
3197 if (server.cached_master == NULL) return;
3198
3199 serverLog(LL_NOTICE,"Discarding previously cached master state.");
3200 server.cached_master->flags &= ~CLIENT_MASTER;
3201 freeClient(server.cached_master);
3202 server.cached_master = NULL;
3203 }
3204
3205 /* Turn the cached master into the current master, using the file descriptor
3206 * passed as argument as the socket for the new master.
3207 *
3208 * This function is called when successfully setup a partial resynchronization
3209 * so the stream of data that we'll receive will start from were this
3210 * master left. */
replicationResurrectCachedMaster(connection * conn)3211 void replicationResurrectCachedMaster(connection *conn) {
3212 server.master = server.cached_master;
3213 server.cached_master = NULL;
3214 server.master->conn = conn;
3215 connSetPrivateData(server.master->conn, server.master);
3216 server.master->flags &= ~(CLIENT_CLOSE_AFTER_REPLY|CLIENT_CLOSE_ASAP);
3217 server.master->authenticated = 1;
3218 server.master->lastinteraction = server.unixtime;
3219 server.repl_state = REPL_STATE_CONNECTED;
3220 server.repl_down_since = 0;
3221
3222 /* Fire the master link modules event. */
3223 moduleFireServerEvent(REDISMODULE_EVENT_MASTER_LINK_CHANGE,
3224 REDISMODULE_SUBEVENT_MASTER_LINK_UP,
3225 NULL);
3226
3227 /* Re-add to the list of clients. */
3228 linkClient(server.master);
3229 if (connSetReadHandler(server.master->conn, readQueryFromClient)) {
3230 serverLog(LL_WARNING,"Error resurrecting the cached master, impossible to add the readable handler: %s", strerror(errno));
3231 freeClientAsync(server.master); /* Close ASAP. */
3232 }
3233
3234 /* We may also need to install the write handler as well if there is
3235 * pending data in the write buffers. */
3236 if (clientHasPendingReplies(server.master)) {
3237 if (connSetWriteHandler(server.master->conn, sendReplyToClient)) {
3238 serverLog(LL_WARNING,"Error resurrecting the cached master, impossible to add the writable handler: %s", strerror(errno));
3239 freeClientAsync(server.master); /* Close ASAP. */
3240 }
3241 }
3242 }
3243
3244 /* ------------------------- MIN-SLAVES-TO-WRITE --------------------------- */
3245
3246 /* This function counts the number of slaves with lag <= min-slaves-max-lag.
3247 * If the option is active, the server will prevent writes if there are not
3248 * enough connected slaves with the specified lag (or less). */
refreshGoodSlavesCount(void)3249 void refreshGoodSlavesCount(void) {
3250 listIter li;
3251 listNode *ln;
3252 int good = 0;
3253
3254 if (!server.repl_min_slaves_to_write ||
3255 !server.repl_min_slaves_max_lag) return;
3256
3257 listRewind(server.slaves,&li);
3258 while((ln = listNext(&li))) {
3259 client *slave = ln->value;
3260 time_t lag = server.unixtime - slave->repl_ack_time;
3261
3262 if (slave->replstate == SLAVE_STATE_ONLINE &&
3263 lag <= server.repl_min_slaves_max_lag) good++;
3264 }
3265 server.repl_good_slaves_count = good;
3266 }
3267
3268 /* ----------------------- REPLICATION SCRIPT CACHE --------------------------
3269 * The goal of this code is to keep track of scripts already sent to every
3270 * connected slave, in order to be able to replicate EVALSHA as it is without
3271 * translating it to EVAL every time it is possible.
3272 *
3273 * We use a capped collection implemented by a hash table for fast lookup
3274 * of scripts we can send as EVALSHA, plus a linked list that is used for
3275 * eviction of the oldest entry when the max number of items is reached.
3276 *
3277 * We don't care about taking a different cache for every different slave
3278 * since to fill the cache again is not very costly, the goal of this code
3279 * is to avoid that the same big script is transmitted a big number of times
3280 * per second wasting bandwidth and processor speed, but it is not a problem
3281 * if we need to rebuild the cache from scratch from time to time, every used
3282 * script will need to be transmitted a single time to reappear in the cache.
3283 *
3284 * This is how the system works:
3285 *
3286 * 1) Every time a new slave connects, we flush the whole script cache.
3287 * 2) We only send as EVALSHA what was sent to the master as EVALSHA, without
3288 * trying to convert EVAL into EVALSHA specifically for slaves.
3289 * 3) Every time we transmit a script as EVAL to the slaves, we also add the
3290 * corresponding SHA1 of the script into the cache as we are sure every
3291 * slave knows about the script starting from now.
3292 * 4) On SCRIPT FLUSH command, we replicate the command to all the slaves
3293 * and at the same time flush the script cache.
3294 * 5) When the last slave disconnects, flush the cache.
3295 * 6) We handle SCRIPT LOAD as well since that's how scripts are loaded
3296 * in the master sometimes.
3297 */
3298
3299 /* Initialize the script cache, only called at startup. */
replicationScriptCacheInit(void)3300 void replicationScriptCacheInit(void) {
3301 server.repl_scriptcache_size = 10000;
3302 server.repl_scriptcache_dict = dictCreate(&replScriptCacheDictType);
3303 server.repl_scriptcache_fifo = listCreate();
3304 }
3305
3306 /* Empty the script cache. Should be called every time we are no longer sure
3307 * that every slave knows about all the scripts in our set, or when the
3308 * current AOF "context" is no longer aware of the script. In general we
3309 * should flush the cache:
3310 *
3311 * 1) Every time a new slave reconnects to this master and performs a
3312 * full SYNC (PSYNC does not require flushing).
3313 * 2) Every time an AOF rewrite is performed.
3314 * 3) Every time we are left without slaves at all, and AOF is off, in order
3315 * to reclaim otherwise unused memory.
3316 */
replicationScriptCacheFlush(void)3317 void replicationScriptCacheFlush(void) {
3318 dictEmpty(server.repl_scriptcache_dict,NULL);
3319 listRelease(server.repl_scriptcache_fifo);
3320 server.repl_scriptcache_fifo = listCreate();
3321 }
3322
3323 /* Add an entry into the script cache, if we reach max number of entries the
3324 * oldest is removed from the list. */
replicationScriptCacheAdd(sds sha1)3325 void replicationScriptCacheAdd(sds sha1) {
3326 int retval;
3327 sds key = sdsdup(sha1);
3328
3329 /* Evict oldest. */
3330 if (listLength(server.repl_scriptcache_fifo) == server.repl_scriptcache_size)
3331 {
3332 listNode *ln = listLast(server.repl_scriptcache_fifo);
3333 sds oldest = listNodeValue(ln);
3334
3335 retval = dictDelete(server.repl_scriptcache_dict,oldest);
3336 serverAssert(retval == DICT_OK);
3337 listDelNode(server.repl_scriptcache_fifo,ln);
3338 }
3339
3340 /* Add current. */
3341 retval = dictAdd(server.repl_scriptcache_dict,key,NULL);
3342 listAddNodeHead(server.repl_scriptcache_fifo,key);
3343 serverAssert(retval == DICT_OK);
3344 }
3345
3346 /* Returns non-zero if the specified entry exists inside the cache, that is,
3347 * if all the slaves are aware of this script SHA1. */
replicationScriptCacheExists(sds sha1)3348 int replicationScriptCacheExists(sds sha1) {
3349 return dictFind(server.repl_scriptcache_dict,sha1) != NULL;
3350 }
3351
3352 /* ----------------------- SYNCHRONOUS REPLICATION --------------------------
3353 * Redis synchronous replication design can be summarized in points:
3354 *
3355 * - Redis masters have a global replication offset, used by PSYNC.
3356 * - Master increment the offset every time new commands are sent to slaves.
3357 * - Slaves ping back masters with the offset processed so far.
3358 *
3359 * So synchronous replication adds a new WAIT command in the form:
3360 *
3361 * WAIT <num_replicas> <milliseconds_timeout>
3362 *
3363 * That returns the number of replicas that processed the query when
3364 * we finally have at least num_replicas, or when the timeout was
3365 * reached.
3366 *
3367 * The command is implemented in this way:
3368 *
3369 * - Every time a client processes a command, we remember the replication
3370 * offset after sending that command to the slaves.
3371 * - When WAIT is called, we ask slaves to send an acknowledgement ASAP.
3372 * The client is blocked at the same time (see blocked.c).
3373 * - Once we receive enough ACKs for a given offset or when the timeout
3374 * is reached, the WAIT command is unblocked and the reply sent to the
3375 * client.
3376 */
3377
3378 /* This just set a flag so that we broadcast a REPLCONF GETACK command
3379 * to all the slaves in the beforeSleep() function. Note that this way
3380 * we "group" all the clients that want to wait for synchronous replication
3381 * in a given event loop iteration, and send a single GETACK for them all. */
replicationRequestAckFromSlaves(void)3382 void replicationRequestAckFromSlaves(void) {
3383 server.get_ack_from_slaves = 1;
3384 }
3385
3386 /* Return the number of slaves that already acknowledged the specified
3387 * replication offset. */
replicationCountAcksByOffset(long long offset)3388 int replicationCountAcksByOffset(long long offset) {
3389 listIter li;
3390 listNode *ln;
3391 int count = 0;
3392
3393 listRewind(server.slaves,&li);
3394 while((ln = listNext(&li))) {
3395 client *slave = ln->value;
3396
3397 if (slave->replstate != SLAVE_STATE_ONLINE) continue;
3398 if (slave->repl_ack_off >= offset) count++;
3399 }
3400 return count;
3401 }
3402
3403 /* WAIT for N replicas to acknowledge the processing of our latest
3404 * write command (and all the previous commands). */
waitCommand(client * c)3405 void waitCommand(client *c) {
3406 mstime_t timeout;
3407 long numreplicas, ackreplicas;
3408 long long offset = c->woff;
3409
3410 if (server.masterhost) {
3411 addReplyError(c,"WAIT cannot be used with replica instances. Please also note that since Redis 4.0 if a replica is configured to be writable (which is not the default) writes to replicas are just local and are not propagated.");
3412 return;
3413 }
3414
3415 /* Argument parsing. */
3416 if (getLongFromObjectOrReply(c,c->argv[1],&numreplicas,NULL) != C_OK)
3417 return;
3418 if (getTimeoutFromObjectOrReply(c,c->argv[2],&timeout,UNIT_MILLISECONDS)
3419 != C_OK) return;
3420
3421 /* First try without blocking at all. */
3422 ackreplicas = replicationCountAcksByOffset(c->woff);
3423 if (ackreplicas >= numreplicas || c->flags & CLIENT_MULTI) {
3424 addReplyLongLong(c,ackreplicas);
3425 return;
3426 }
3427
3428 /* Otherwise block the client and put it into our list of clients
3429 * waiting for ack from slaves. */
3430 c->bpop.timeout = timeout;
3431 c->bpop.reploffset = offset;
3432 c->bpop.numreplicas = numreplicas;
3433 listAddNodeHead(server.clients_waiting_acks,c);
3434 blockClient(c,BLOCKED_WAIT);
3435
3436 /* Make sure that the server will send an ACK request to all the slaves
3437 * before returning to the event loop. */
3438 replicationRequestAckFromSlaves();
3439 }
3440
3441 /* This is called by unblockClient() to perform the blocking op type
3442 * specific cleanup. We just remove the client from the list of clients
3443 * waiting for replica acks. Never call it directly, call unblockClient()
3444 * instead. */
unblockClientWaitingReplicas(client * c)3445 void unblockClientWaitingReplicas(client *c) {
3446 listNode *ln = listSearchKey(server.clients_waiting_acks,c);
3447 serverAssert(ln != NULL);
3448 listDelNode(server.clients_waiting_acks,ln);
3449 }
3450
3451 /* Check if there are clients blocked in WAIT that can be unblocked since
3452 * we received enough ACKs from slaves. */
processClientsWaitingReplicas(void)3453 void processClientsWaitingReplicas(void) {
3454 long long last_offset = 0;
3455 int last_numreplicas = 0;
3456
3457 listIter li;
3458 listNode *ln;
3459
3460 listRewind(server.clients_waiting_acks,&li);
3461 while((ln = listNext(&li))) {
3462 client *c = ln->value;
3463
3464 /* Every time we find a client that is satisfied for a given
3465 * offset and number of replicas, we remember it so the next client
3466 * may be unblocked without calling replicationCountAcksByOffset()
3467 * if the requested offset / replicas were equal or less. */
3468 if (last_offset && last_offset >= c->bpop.reploffset &&
3469 last_numreplicas >= c->bpop.numreplicas)
3470 {
3471 unblockClient(c);
3472 addReplyLongLong(c,last_numreplicas);
3473 } else {
3474 int numreplicas = replicationCountAcksByOffset(c->bpop.reploffset);
3475
3476 if (numreplicas >= c->bpop.numreplicas) {
3477 last_offset = c->bpop.reploffset;
3478 last_numreplicas = numreplicas;
3479 unblockClient(c);
3480 addReplyLongLong(c,numreplicas);
3481 }
3482 }
3483 }
3484 }
3485
3486 /* Return the slave replication offset for this instance, that is
3487 * the offset for which we already processed the master replication stream. */
replicationGetSlaveOffset(void)3488 long long replicationGetSlaveOffset(void) {
3489 long long offset = 0;
3490
3491 if (server.masterhost != NULL) {
3492 if (server.master) {
3493 offset = server.master->reploff;
3494 } else if (server.cached_master) {
3495 offset = server.cached_master->reploff;
3496 }
3497 }
3498 /* offset may be -1 when the master does not support it at all, however
3499 * this function is designed to return an offset that can express the
3500 * amount of data processed by the master, so we return a positive
3501 * integer. */
3502 if (offset < 0) offset = 0;
3503 return offset;
3504 }
3505
3506 /* --------------------------- REPLICATION CRON ---------------------------- */
3507
3508 /* Replication cron function, called 1 time per second. */
replicationCron(void)3509 void replicationCron(void) {
3510 static long long replication_cron_loops = 0;
3511
3512 /* Check failover status first, to see if we need to start
3513 * handling the failover. */
3514 updateFailoverStatus();
3515
3516 /* Non blocking connection timeout? */
3517 if (server.masterhost &&
3518 (server.repl_state == REPL_STATE_CONNECTING ||
3519 slaveIsInHandshakeState()) &&
3520 (time(NULL)-server.repl_transfer_lastio) > server.repl_timeout)
3521 {
3522 serverLog(LL_WARNING,"Timeout connecting to the MASTER...");
3523 cancelReplicationHandshake(1);
3524 }
3525
3526 /* Bulk transfer I/O timeout? */
3527 if (server.masterhost && server.repl_state == REPL_STATE_TRANSFER &&
3528 (time(NULL)-server.repl_transfer_lastio) > server.repl_timeout)
3529 {
3530 serverLog(LL_WARNING,"Timeout receiving bulk data from MASTER... If the problem persists try to set the 'repl-timeout' parameter in redis.conf to a larger value.");
3531 cancelReplicationHandshake(1);
3532 }
3533
3534 /* Timed out master when we are an already connected slave? */
3535 if (server.masterhost && server.repl_state == REPL_STATE_CONNECTED &&
3536 (time(NULL)-server.master->lastinteraction) > server.repl_timeout)
3537 {
3538 serverLog(LL_WARNING,"MASTER timeout: no data nor PING received...");
3539 freeClient(server.master);
3540 }
3541
3542 /* Check if we should connect to a MASTER */
3543 if (server.repl_state == REPL_STATE_CONNECT) {
3544 serverLog(LL_NOTICE,"Connecting to MASTER %s:%d",
3545 server.masterhost, server.masterport);
3546 connectWithMaster();
3547 }
3548
3549 /* Send ACK to master from time to time.
3550 * Note that we do not send periodic acks to masters that don't
3551 * support PSYNC and replication offsets. */
3552 if (server.masterhost && server.master &&
3553 !(server.master->flags & CLIENT_PRE_PSYNC))
3554 replicationSendAck();
3555
3556 /* If we have attached slaves, PING them from time to time.
3557 * So slaves can implement an explicit timeout to masters, and will
3558 * be able to detect a link disconnection even if the TCP connection
3559 * will not actually go down. */
3560 listIter li;
3561 listNode *ln;
3562 robj *ping_argv[1];
3563
3564 /* First, send PING according to ping_slave_period. */
3565 if ((replication_cron_loops % server.repl_ping_slave_period) == 0 &&
3566 listLength(server.slaves))
3567 {
3568 /* Note that we don't send the PING if the clients are paused during
3569 * a Redis Cluster manual failover: the PING we send will otherwise
3570 * alter the replication offsets of master and slave, and will no longer
3571 * match the one stored into 'mf_master_offset' state. */
3572 int manual_failover_in_progress =
3573 ((server.cluster_enabled &&
3574 server.cluster->mf_end) ||
3575 server.failover_end_time) &&
3576 checkClientPauseTimeoutAndReturnIfPaused();
3577
3578 if (!manual_failover_in_progress) {
3579 ping_argv[0] = shared.ping;
3580 replicationFeedSlaves(server.slaves, server.slaveseldb,
3581 ping_argv, 1);
3582 }
3583 }
3584
3585 /* Second, send a newline to all the slaves in pre-synchronization
3586 * stage, that is, slaves waiting for the master to create the RDB file.
3587 *
3588 * Also send the a newline to all the chained slaves we have, if we lost
3589 * connection from our master, to keep the slaves aware that their
3590 * master is online. This is needed since sub-slaves only receive proxied
3591 * data from top-level masters, so there is no explicit pinging in order
3592 * to avoid altering the replication offsets. This special out of band
3593 * pings (newlines) can be sent, they will have no effect in the offset.
3594 *
3595 * The newline will be ignored by the slave but will refresh the
3596 * last interaction timer preventing a timeout. In this case we ignore the
3597 * ping period and refresh the connection once per second since certain
3598 * timeouts are set at a few seconds (example: PSYNC response). */
3599 listRewind(server.slaves,&li);
3600 while((ln = listNext(&li))) {
3601 client *slave = ln->value;
3602
3603 int is_presync =
3604 (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START ||
3605 (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_END &&
3606 server.rdb_child_type != RDB_CHILD_TYPE_SOCKET));
3607
3608 if (is_presync) {
3609 connWrite(slave->conn, "\n", 1);
3610 }
3611 }
3612
3613 /* Disconnect timedout slaves. */
3614 if (listLength(server.slaves)) {
3615 listIter li;
3616 listNode *ln;
3617
3618 listRewind(server.slaves,&li);
3619 while((ln = listNext(&li))) {
3620 client *slave = ln->value;
3621
3622 if (slave->replstate == SLAVE_STATE_ONLINE) {
3623 if (slave->flags & CLIENT_PRE_PSYNC)
3624 continue;
3625 if ((server.unixtime - slave->repl_ack_time) > server.repl_timeout) {
3626 serverLog(LL_WARNING, "Disconnecting timedout replica (streaming sync): %s",
3627 replicationGetSlaveName(slave));
3628 freeClient(slave);
3629 continue;
3630 }
3631 }
3632 /* We consider disconnecting only diskless replicas because disk-based replicas aren't fed
3633 * by the fork child so if a disk-based replica is stuck it doesn't prevent the fork child
3634 * from terminating. */
3635 if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_END && server.rdb_child_type == RDB_CHILD_TYPE_SOCKET) {
3636 if (slave->repl_last_partial_write != 0 &&
3637 (server.unixtime - slave->repl_last_partial_write) > server.repl_timeout)
3638 {
3639 serverLog(LL_WARNING, "Disconnecting timedout replica (full sync): %s",
3640 replicationGetSlaveName(slave));
3641 freeClient(slave);
3642 continue;
3643 }
3644 }
3645 }
3646 }
3647
3648 /* If this is a master without attached slaves and there is a replication
3649 * backlog active, in order to reclaim memory we can free it after some
3650 * (configured) time. Note that this cannot be done for slaves: slaves
3651 * without sub-slaves attached should still accumulate data into the
3652 * backlog, in order to reply to PSYNC queries if they are turned into
3653 * masters after a failover. */
3654 if (listLength(server.slaves) == 0 && server.repl_backlog_time_limit &&
3655 server.repl_backlog && server.masterhost == NULL)
3656 {
3657 time_t idle = server.unixtime - server.repl_no_slaves_since;
3658
3659 if (idle > server.repl_backlog_time_limit) {
3660 /* When we free the backlog, we always use a new
3661 * replication ID and clear the ID2. This is needed
3662 * because when there is no backlog, the master_repl_offset
3663 * is not updated, but we would still retain our replication
3664 * ID, leading to the following problem:
3665 *
3666 * 1. We are a master instance.
3667 * 2. Our slave is promoted to master. It's repl-id-2 will
3668 * be the same as our repl-id.
3669 * 3. We, yet as master, receive some updates, that will not
3670 * increment the master_repl_offset.
3671 * 4. Later we are turned into a slave, connect to the new
3672 * master that will accept our PSYNC request by second
3673 * replication ID, but there will be data inconsistency
3674 * because we received writes. */
3675 changeReplicationId();
3676 clearReplicationId2();
3677 freeReplicationBacklog();
3678 serverLog(LL_NOTICE,
3679 "Replication backlog freed after %d seconds "
3680 "without connected replicas.",
3681 (int) server.repl_backlog_time_limit);
3682 }
3683 }
3684
3685 /* If AOF is disabled and we no longer have attached slaves, we can
3686 * free our Replication Script Cache as there is no need to propagate
3687 * EVALSHA at all. */
3688 if (listLength(server.slaves) == 0 &&
3689 server.aof_state == AOF_OFF &&
3690 listLength(server.repl_scriptcache_fifo) != 0)
3691 {
3692 replicationScriptCacheFlush();
3693 }
3694
3695 replicationStartPendingFork();
3696
3697 /* Remove the RDB file used for replication if Redis is not running
3698 * with any persistence. */
3699 removeRDBUsedToSyncReplicas();
3700
3701 /* Sanity check replication buffer, the first block of replication buffer blocks
3702 * must be referenced by someone, since it will be freed when not referenced,
3703 * otherwise, server will OOM. also, its refcount must not be more than
3704 * replicas number + 1(replication backlog). */
3705 if (listLength(server.repl_buffer_blocks) > 0) {
3706 replBufBlock *o = listNodeValue(listFirst(server.repl_buffer_blocks));
3707 serverAssert(o->refcount > 0 &&
3708 o->refcount <= (int)listLength(server.slaves)+1);
3709 }
3710
3711 /* Refresh the number of slaves with lag <= min-slaves-max-lag. */
3712 refreshGoodSlavesCount();
3713 replication_cron_loops++; /* Incremented with frequency 1 HZ. */
3714 }
3715
replicationStartPendingFork(void)3716 void replicationStartPendingFork(void) {
3717 /* Start a BGSAVE good for replication if we have slaves in
3718 * WAIT_BGSAVE_START state.
3719 *
3720 * In case of diskless replication, we make sure to wait the specified
3721 * number of seconds (according to configuration) so that other slaves
3722 * have the time to arrive before we start streaming. */
3723 if (!hasActiveChildProcess()) {
3724 time_t idle, max_idle = 0;
3725 int slaves_waiting = 0;
3726 int mincapa = -1;
3727 listNode *ln;
3728 listIter li;
3729
3730 listRewind(server.slaves,&li);
3731 while((ln = listNext(&li))) {
3732 client *slave = ln->value;
3733 if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) {
3734 idle = server.unixtime - slave->lastinteraction;
3735 if (idle > max_idle) max_idle = idle;
3736 slaves_waiting++;
3737 mincapa = (mincapa == -1) ? slave->slave_capa :
3738 (mincapa & slave->slave_capa);
3739 }
3740 }
3741
3742 if (slaves_waiting &&
3743 (!server.repl_diskless_sync ||
3744 max_idle >= server.repl_diskless_sync_delay))
3745 {
3746 /* Start the BGSAVE. The called function may start a
3747 * BGSAVE with socket target or disk target depending on the
3748 * configuration and slaves capabilities. */
3749 startBgsaveForReplication(mincapa);
3750 }
3751 }
3752 }
3753
3754 /* Find replica at IP:PORT from replica list */
findReplica(char * host,int port)3755 static client *findReplica(char *host, int port) {
3756 listIter li;
3757 listNode *ln;
3758 client *replica;
3759
3760 listRewind(server.slaves,&li);
3761 while((ln = listNext(&li))) {
3762 replica = ln->value;
3763 char ip[NET_IP_STR_LEN], *replicaip = replica->slave_addr;
3764
3765 if (!replicaip) {
3766 if (connPeerToString(replica->conn, ip, sizeof(ip), NULL) == -1)
3767 continue;
3768 replicaip = ip;
3769 }
3770
3771 if (!strcasecmp(host, replicaip) &&
3772 (port == replica->slave_listening_port))
3773 return replica;
3774 }
3775
3776 return NULL;
3777 }
3778
getFailoverStateString()3779 const char *getFailoverStateString() {
3780 switch(server.failover_state) {
3781 case NO_FAILOVER: return "no-failover";
3782 case FAILOVER_IN_PROGRESS: return "failover-in-progress";
3783 case FAILOVER_WAIT_FOR_SYNC: return "waiting-for-sync";
3784 default: return "unknown";
3785 }
3786 }
3787
3788 /* Resets the internal failover configuration, this needs
3789 * to be called after a failover either succeeds or fails
3790 * as it includes the client unpause. */
clearFailoverState()3791 void clearFailoverState() {
3792 server.failover_end_time = 0;
3793 server.force_failover = 0;
3794 zfree(server.target_replica_host);
3795 server.target_replica_host = NULL;
3796 server.target_replica_port = 0;
3797 server.failover_state = NO_FAILOVER;
3798 unpauseClients();
3799 }
3800
3801 /* Abort an ongoing failover if one is going on. */
abortFailover(const char * err)3802 void abortFailover(const char *err) {
3803 if (server.failover_state == NO_FAILOVER) return;
3804
3805 if (server.target_replica_host) {
3806 serverLog(LL_NOTICE,"FAILOVER to %s:%d aborted: %s",
3807 server.target_replica_host,server.target_replica_port,err);
3808 } else {
3809 serverLog(LL_NOTICE,"FAILOVER to any replica aborted: %s",err);
3810 }
3811 if (server.failover_state == FAILOVER_IN_PROGRESS) {
3812 replicationUnsetMaster();
3813 }
3814 clearFailoverState();
3815 }
3816
3817 /*
3818 * FAILOVER [TO <HOST> <PORT> [FORCE]] [ABORT] [TIMEOUT <timeout>]
3819 *
3820 * This command will coordinate a failover between the master and one
3821 * of its replicas. The happy path contains the following steps:
3822 * 1) The master will initiate a client pause write, to stop replication
3823 * traffic.
3824 * 2) The master will periodically check if any of its replicas has
3825 * consumed the entire replication stream through acks.
3826 * 3) Once any replica has caught up, the master will itself become a replica.
3827 * 4) The master will send a PSYNC FAILOVER request to the target replica, which
3828 * if accepted will cause the replica to become the new master and start a sync.
3829 *
3830 * FAILOVER ABORT is the only way to abort a failover command, as replicaof
3831 * will be disabled. This may be needed if the failover is unable to progress.
3832 *
3833 * The optional arguments [TO <HOST> <IP>] allows designating a specific replica
3834 * to be failed over to.
3835 *
3836 * FORCE flag indicates that even if the target replica is not caught up,
3837 * failover to it anyway. This must be specified with a timeout and a target
3838 * HOST and IP.
3839 *
3840 * TIMEOUT <timeout> indicates how long should the primary wait for
3841 * a replica to sync up before aborting. If not specified, the failover
3842 * will attempt forever and must be manually aborted.
3843 */
failoverCommand(client * c)3844 void failoverCommand(client *c) {
3845 if (server.cluster_enabled) {
3846 addReplyError(c,"FAILOVER not allowed in cluster mode. "
3847 "Use CLUSTER FAILOVER command instead.");
3848 return;
3849 }
3850
3851 /* Handle special case for abort */
3852 if ((c->argc == 2) && !strcasecmp(c->argv[1]->ptr,"abort")) {
3853 if (server.failover_state == NO_FAILOVER) {
3854 addReplyError(c, "No failover in progress.");
3855 return;
3856 }
3857
3858 abortFailover("Failover manually aborted");
3859 addReply(c,shared.ok);
3860 return;
3861 }
3862
3863 long timeout_in_ms = 0;
3864 int force_flag = 0;
3865 long port = 0;
3866 char *host = NULL;
3867
3868 /* Parse the command for syntax and arguments. */
3869 for (int j = 1; j < c->argc; j++) {
3870 if (!strcasecmp(c->argv[j]->ptr,"timeout") && (j + 1 < c->argc) &&
3871 timeout_in_ms == 0)
3872 {
3873 if (getLongFromObjectOrReply(c,c->argv[j + 1],
3874 &timeout_in_ms,NULL) != C_OK) return;
3875 if (timeout_in_ms <= 0) {
3876 addReplyError(c,"FAILOVER timeout must be greater than 0");
3877 return;
3878 }
3879 j++;
3880 } else if (!strcasecmp(c->argv[j]->ptr,"to") && (j + 2 < c->argc) &&
3881 !host)
3882 {
3883 if (getLongFromObjectOrReply(c,c->argv[j + 2],&port,NULL) != C_OK)
3884 return;
3885 host = c->argv[j + 1]->ptr;
3886 j += 2;
3887 } else if (!strcasecmp(c->argv[j]->ptr,"force") && !force_flag) {
3888 force_flag = 1;
3889 } else {
3890 addReplyErrorObject(c,shared.syntaxerr);
3891 return;
3892 }
3893 }
3894
3895 if (server.failover_state != NO_FAILOVER) {
3896 addReplyError(c,"FAILOVER already in progress.");
3897 return;
3898 }
3899
3900 if (server.masterhost) {
3901 addReplyError(c,"FAILOVER is not valid when server is a replica.");
3902 return;
3903 }
3904
3905 if (listLength(server.slaves) == 0) {
3906 addReplyError(c,"FAILOVER requires connected replicas.");
3907 return;
3908 }
3909
3910 if (force_flag && (!timeout_in_ms || !host)) {
3911 addReplyError(c,"FAILOVER with force option requires both a timeout "
3912 "and target HOST and IP.");
3913 return;
3914 }
3915
3916 /* If a replica address was provided, validate that it is connected. */
3917 if (host) {
3918 client *replica = findReplica(host, port);
3919
3920 if (replica == NULL) {
3921 addReplyError(c,"FAILOVER target HOST and PORT is not "
3922 "a replica.");
3923 return;
3924 }
3925
3926 /* Check if requested replica is online */
3927 if (replica->replstate != SLAVE_STATE_ONLINE) {
3928 addReplyError(c,"FAILOVER target replica is not online.");
3929 return;
3930 }
3931
3932 server.target_replica_host = zstrdup(host);
3933 server.target_replica_port = port;
3934 serverLog(LL_NOTICE,"FAILOVER requested to %s:%ld.",host,port);
3935 } else {
3936 serverLog(LL_NOTICE,"FAILOVER requested to any replica.");
3937 }
3938
3939 mstime_t now = mstime();
3940 if (timeout_in_ms) {
3941 server.failover_end_time = now + timeout_in_ms;
3942 }
3943
3944 server.force_failover = force_flag;
3945 server.failover_state = FAILOVER_WAIT_FOR_SYNC;
3946 /* Cluster failover will unpause eventually */
3947 pauseClients(LLONG_MAX,CLIENT_PAUSE_WRITE);
3948 addReply(c,shared.ok);
3949 }
3950
3951 /* Failover cron function, checks coordinated failover state.
3952 *
3953 * Implementation note: The current implementation calls replicationSetMaster()
3954 * to start the failover request, this has some unintended side effects if the
3955 * failover doesn't work like blocked clients will be unblocked and replicas will
3956 * be disconnected. This could be optimized further.
3957 */
updateFailoverStatus(void)3958 void updateFailoverStatus(void) {
3959 if (server.failover_state != FAILOVER_WAIT_FOR_SYNC) return;
3960 mstime_t now = server.mstime;
3961
3962 /* Check if failover operation has timed out */
3963 if (server.failover_end_time && server.failover_end_time <= now) {
3964 if (server.force_failover) {
3965 serverLog(LL_NOTICE,
3966 "FAILOVER to %s:%d time out exceeded, failing over.",
3967 server.target_replica_host, server.target_replica_port);
3968 server.failover_state = FAILOVER_IN_PROGRESS;
3969 /* If timeout has expired force a failover if requested. */
3970 replicationSetMaster(server.target_replica_host,
3971 server.target_replica_port);
3972 return;
3973 } else {
3974 /* Force was not requested, so timeout. */
3975 abortFailover("Replica never caught up before timeout");
3976 return;
3977 }
3978 }
3979
3980 /* Check to see if the replica has caught up so failover can start */
3981 client *replica = NULL;
3982 if (server.target_replica_host) {
3983 replica = findReplica(server.target_replica_host,
3984 server.target_replica_port);
3985 } else {
3986 listIter li;
3987 listNode *ln;
3988
3989 listRewind(server.slaves,&li);
3990 /* Find any replica that has matched our repl_offset */
3991 while((ln = listNext(&li))) {
3992 replica = ln->value;
3993 if (replica->repl_ack_off == server.master_repl_offset) {
3994 char ip[NET_IP_STR_LEN], *replicaaddr = replica->slave_addr;
3995
3996 if (!replicaaddr) {
3997 if (connPeerToString(replica->conn,ip,sizeof(ip),NULL) == -1)
3998 continue;
3999 replicaaddr = ip;
4000 }
4001
4002 /* We are now failing over to this specific node */
4003 server.target_replica_host = zstrdup(replicaaddr);
4004 server.target_replica_port = replica->slave_listening_port;
4005 break;
4006 }
4007 }
4008 }
4009
4010 /* We've found a replica that is caught up */
4011 if (replica && (replica->repl_ack_off == server.master_repl_offset)) {
4012 server.failover_state = FAILOVER_IN_PROGRESS;
4013 serverLog(LL_NOTICE,
4014 "Failover target %s:%d is synced, failing over.",
4015 server.target_replica_host, server.target_replica_port);
4016 /* Designated replica is caught up, failover to it. */
4017 replicationSetMaster(server.target_replica_host,
4018 server.target_replica_port);
4019 }
4020 }
4021