diff options
Diffstat (limited to 'src/replication.c')
| -rw-r--r-- | src/replication.c | 4241 |
1 files changed, 4241 insertions, 0 deletions
diff --git a/src/replication.c b/src/replication.c new file mode 100644 index 0000000..fbd9b0a --- /dev/null +++ b/src/replication.c @@ -0,0 +1,4241 @@ +/* Asynchronous replication implementation. + * + * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com> + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are met: + * + * * Redistributions of source code must retain the above copyright notice, + * this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above copyright + * notice, this list of conditions and the following disclaimer in the + * documentation and/or other materials provided with the distribution. + * * Neither the name of Redis nor the names of its contributors may be used + * to endorse or promote products derived from this software without + * specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" + * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE + * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE + * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE + * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR + * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF + * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS + * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN + * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) + * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE + * POSSIBILITY OF SUCH DAMAGE. + */ + + +#include "server.h" +#include "cluster.h" +#include "bio.h" +#include "functions.h" +#include "connection.h" + +#include <memory.h> +#include <sys/time.h> +#include <unistd.h> +#include <fcntl.h> +#include <sys/socket.h> +#include <sys/stat.h> + +void replicationDiscardCachedMaster(void); +void replicationResurrectCachedMaster(connection *conn); +void replicationSendAck(void); +int replicaPutOnline(client *slave); +void replicaStartCommandStream(client *slave); +int cancelReplicationHandshake(int reconnect); + +/* We take a global flag to remember if this instance generated an RDB + * because of replication, so that we can remove the RDB file in case + * the instance is configured to have no persistence. */ +int RDBGeneratedByReplication = 0; + +/* --------------------------- Utility functions ---------------------------- */ +static ConnectionType *connTypeOfReplication(void) { + if (server.tls_replication) { + return connectionTypeTls(); + } + + return connectionTypeTcp(); +} + +/* Return the pointer to a string representing the slave ip:listening_port + * pair. Mostly useful for logging, since we want to log a slave using its + * IP address and its listening port which is more clear for the user, for + * example: "Closing connection with replica 10.1.2.3:6380". */ +char *replicationGetSlaveName(client *c) { + static char buf[NET_HOST_PORT_STR_LEN]; + char ip[NET_IP_STR_LEN]; + + ip[0] = '\0'; + buf[0] = '\0'; + if (c->slave_addr || + connAddrPeerName(c->conn,ip,sizeof(ip),NULL) != -1) + { + char *addr = c->slave_addr ? c->slave_addr : ip; + if (c->slave_listening_port) + formatAddr(buf,sizeof(buf),addr,c->slave_listening_port); + else + snprintf(buf,sizeof(buf),"%s:<unknown-replica-port>",addr); + } else { + snprintf(buf,sizeof(buf),"client id #%llu", + (unsigned long long) c->id); + } + return buf; +} + +/* Plain unlink() can block for quite some time in order to actually apply + * the file deletion to the filesystem. This call removes the file in a + * background thread instead. We actually just do close() in the thread, + * by using the fact that if there is another instance of the same file open, + * the foreground unlink() will only remove the fs name, and deleting the + * file's storage space will only happen once the last reference is lost. */ +int bg_unlink(const char *filename) { + int fd = open(filename,O_RDONLY|O_NONBLOCK); + if (fd == -1) { + /* Can't open the file? Fall back to unlinking in the main thread. */ + return unlink(filename); + } else { + /* The following unlink() removes the name but doesn't free the + * file contents because a process still has it open. */ + int retval = unlink(filename); + if (retval == -1) { + /* If we got an unlink error, we just return it, closing the + * new reference we have to the file. */ + int old_errno = errno; + close(fd); /* This would overwrite our errno. So we saved it. */ + errno = old_errno; + return -1; + } + bioCreateCloseJob(fd, 0, 0); + return 0; /* Success. */ + } +} + +/* ---------------------------------- MASTER -------------------------------- */ + +void createReplicationBacklog(void) { + serverAssert(server.repl_backlog == NULL); + server.repl_backlog = zmalloc(sizeof(replBacklog)); + server.repl_backlog->ref_repl_buf_node = NULL; + server.repl_backlog->unindexed_count = 0; + server.repl_backlog->blocks_index = raxNew(); + server.repl_backlog->histlen = 0; + /* We don't have any data inside our buffer, but virtually the first + * byte we have is the next byte that will be generated for the + * replication stream. */ + server.repl_backlog->offset = server.master_repl_offset+1; +} + +/* This function is called when the user modifies the replication backlog + * size at runtime. It is up to the function to resize the buffer and setup it + * so that it contains the same data as the previous one (possibly less data, + * but the most recent bytes, or the same data and more free space in case the + * buffer is enlarged). */ +void resizeReplicationBacklog(void) { + if (server.repl_backlog_size < CONFIG_REPL_BACKLOG_MIN_SIZE) + server.repl_backlog_size = CONFIG_REPL_BACKLOG_MIN_SIZE; + if (server.repl_backlog) + incrementalTrimReplicationBacklog(REPL_BACKLOG_TRIM_BLOCKS_PER_CALL); +} + +void freeReplicationBacklog(void) { + serverAssert(listLength(server.slaves) == 0); + if (server.repl_backlog == NULL) return; + + /* Decrease the start buffer node reference count. */ + if (server.repl_backlog->ref_repl_buf_node) { + replBufBlock *o = listNodeValue( + server.repl_backlog->ref_repl_buf_node); + serverAssert(o->refcount == 1); /* Last reference. */ + o->refcount--; + } + + /* Replication buffer blocks are completely released when we free the + * backlog, since the backlog is released only when there are no replicas + * and the backlog keeps the last reference of all blocks. */ + freeReplicationBacklogRefMemAsync(server.repl_buffer_blocks, + server.repl_backlog->blocks_index); + resetReplicationBuffer(); + zfree(server.repl_backlog); + server.repl_backlog = NULL; +} + +/* To make search offset from replication buffer blocks quickly + * when replicas ask partial resynchronization, we create one index + * block every REPL_BACKLOG_INDEX_PER_BLOCKS blocks. */ +void createReplicationBacklogIndex(listNode *ln) { + server.repl_backlog->unindexed_count++; + if (server.repl_backlog->unindexed_count >= REPL_BACKLOG_INDEX_PER_BLOCKS) { + replBufBlock *o = listNodeValue(ln); + uint64_t encoded_offset = htonu64(o->repl_offset); + raxInsert(server.repl_backlog->blocks_index, + (unsigned char*)&encoded_offset, sizeof(uint64_t), + ln, NULL); + server.repl_backlog->unindexed_count = 0; + } +} + +/* Rebase replication buffer blocks' offset since the initial + * setting offset starts from 0 when master restart. */ +void rebaseReplicationBuffer(long long base_repl_offset) { + raxFree(server.repl_backlog->blocks_index); + server.repl_backlog->blocks_index = raxNew(); + server.repl_backlog->unindexed_count = 0; + + listIter li; + listNode *ln; + listRewind(server.repl_buffer_blocks, &li); + while ((ln = listNext(&li))) { + replBufBlock *o = listNodeValue(ln); + o->repl_offset += base_repl_offset; + createReplicationBacklogIndex(ln); + } +} + +void resetReplicationBuffer(void) { + server.repl_buffer_mem = 0; + server.repl_buffer_blocks = listCreate(); + listSetFreeMethod(server.repl_buffer_blocks, (void (*)(void*))zfree); +} + +int canFeedReplicaReplBuffer(client *replica) { + /* Don't feed replicas that only want the RDB. */ + if (replica->flags & CLIENT_REPL_RDBONLY) return 0; + + /* Don't feed replicas that are still waiting for BGSAVE to start. */ + if (replica->replstate == SLAVE_STATE_WAIT_BGSAVE_START) return 0; + + return 1; +} + +/* Similar with 'prepareClientToWrite', note that we must call this function + * before feeding replication stream into global replication buffer, since + * clientHasPendingReplies in prepareClientToWrite will access the global + * replication buffer to make judgements. */ +int prepareReplicasToWrite(void) { + listIter li; + listNode *ln; + int prepared = 0; + + listRewind(server.slaves,&li); + while((ln = listNext(&li))) { + client *slave = ln->value; + if (!canFeedReplicaReplBuffer(slave)) continue; + if (prepareClientToWrite(slave) == C_ERR) continue; + prepared++; + } + + return prepared; +} + +/* Wrapper for feedReplicationBuffer() that takes Redis string objects + * as input. */ +void feedReplicationBufferWithObject(robj *o) { + char llstr[LONG_STR_SIZE]; + void *p; + size_t len; + + if (o->encoding == OBJ_ENCODING_INT) { + len = ll2string(llstr,sizeof(llstr),(long)o->ptr); + p = llstr; + } else { + len = sdslen(o->ptr); + p = o->ptr; + } + feedReplicationBuffer(p,len); +} + +/* Generally, we only have one replication buffer block to trim when replication + * backlog size exceeds our setting and no replica reference it. But if replica + * clients disconnect, we need to free many replication buffer blocks that are + * referenced. It would cost much time if there are a lots blocks to free, that + * will freeze server, so we trim replication backlog incrementally. */ +void incrementalTrimReplicationBacklog(size_t max_blocks) { + serverAssert(server.repl_backlog != NULL); + + size_t trimmed_blocks = 0; + while (server.repl_backlog->histlen > server.repl_backlog_size && + trimmed_blocks < max_blocks) + { + /* We never trim backlog to less than one block. */ + if (listLength(server.repl_buffer_blocks) <= 1) break; + + /* Replicas increment the refcount of the first replication buffer block + * they refer to, in that case, we don't trim the backlog even if + * backlog_histlen exceeds backlog_size. This implicitly makes backlog + * bigger than our setting, but makes the master accept partial resync as + * much as possible. So that backlog must be the last reference of + * replication buffer blocks. */ + listNode *first = listFirst(server.repl_buffer_blocks); + serverAssert(first == server.repl_backlog->ref_repl_buf_node); + replBufBlock *fo = listNodeValue(first); + if (fo->refcount != 1) break; + + /* We don't try trim backlog if backlog valid size will be lessen than + * setting backlog size once we release the first repl buffer block. */ + if (server.repl_backlog->histlen - (long long)fo->size <= + server.repl_backlog_size) break; + + /* Decr refcount and release the first block later. */ + fo->refcount--; + trimmed_blocks++; + server.repl_backlog->histlen -= fo->size; + + /* Go to use next replication buffer block node. */ + listNode *next = listNextNode(first); + server.repl_backlog->ref_repl_buf_node = next; + serverAssert(server.repl_backlog->ref_repl_buf_node != NULL); + /* Incr reference count to keep the new head node. */ + ((replBufBlock *)listNodeValue(next))->refcount++; + + /* Remove the node in recorded blocks. */ + uint64_t encoded_offset = htonu64(fo->repl_offset); + raxRemove(server.repl_backlog->blocks_index, + (unsigned char*)&encoded_offset, sizeof(uint64_t), NULL); + + /* Delete the first node from global replication buffer. */ + serverAssert(fo->refcount == 0 && fo->used == fo->size); + server.repl_buffer_mem -= (fo->size + + sizeof(listNode) + sizeof(replBufBlock)); + listDelNode(server.repl_buffer_blocks, first); + } + + /* Set the offset of the first byte we have in the backlog. */ + server.repl_backlog->offset = server.master_repl_offset - + server.repl_backlog->histlen + 1; +} + +/* Free replication buffer blocks that are referenced by this client. */ +void freeReplicaReferencedReplBuffer(client *replica) { + if (replica->ref_repl_buf_node != NULL) { + /* Decrease the start buffer node reference count. */ + replBufBlock *o = listNodeValue(replica->ref_repl_buf_node); + serverAssert(o->refcount > 0); + o->refcount--; + incrementalTrimReplicationBacklog(REPL_BACKLOG_TRIM_BLOCKS_PER_CALL); + } + replica->ref_repl_buf_node = NULL; + replica->ref_block_pos = 0; +} + +/* Append bytes into the global replication buffer list, replication backlog and + * all replica clients use replication buffers collectively, this function replace + * 'addReply*', 'feedReplicationBacklog' for replicas and replication backlog, + * First we add buffer into global replication buffer block list, and then + * update replica / replication-backlog referenced node and block position. */ +void feedReplicationBuffer(char *s, size_t len) { + static long long repl_block_id = 0; + + if (server.repl_backlog == NULL) return; + + while(len > 0) { + size_t start_pos = 0; /* The position of referenced block to start sending. */ + listNode *start_node = NULL; /* Replica/backlog starts referenced node. */ + int add_new_block = 0; /* Create new block if current block is total used. */ + listNode *ln = listLast(server.repl_buffer_blocks); + replBufBlock *tail = ln ? listNodeValue(ln) : NULL; + + /* Append to tail string when possible. */ + if (tail && tail->size > tail->used) { + start_node = listLast(server.repl_buffer_blocks); + start_pos = tail->used; + /* Copy the part we can fit into the tail, and leave the rest for a + * new node */ + size_t avail = tail->size - tail->used; + size_t copy = (avail >= len) ? len : avail; + memcpy(tail->buf + tail->used, s, copy); + tail->used += copy; + s += copy; + len -= copy; + server.master_repl_offset += copy; + server.repl_backlog->histlen += copy; + } + if (len) { + /* Create a new node, make sure it is allocated to at + * least PROTO_REPLY_CHUNK_BYTES */ + size_t usable_size; + /* Avoid creating nodes smaller than PROTO_REPLY_CHUNK_BYTES, so that we can append more data into them, + * and also avoid creating nodes bigger than repl_backlog_size / 16, so that we won't have huge nodes that can't + * trim when we only still need to hold a small portion from them. */ + size_t limit = max((size_t)server.repl_backlog_size / 16, (size_t)PROTO_REPLY_CHUNK_BYTES); + size_t size = min(max(len, (size_t)PROTO_REPLY_CHUNK_BYTES), limit); + tail = zmalloc_usable(size + sizeof(replBufBlock), &usable_size); + /* Take over the allocation's internal fragmentation */ + tail->size = usable_size - sizeof(replBufBlock); + size_t copy = (tail->size >= len) ? len : tail->size; + tail->used = copy; + tail->refcount = 0; + tail->repl_offset = server.master_repl_offset + 1; + tail->id = repl_block_id++; + memcpy(tail->buf, s, copy); + listAddNodeTail(server.repl_buffer_blocks, tail); + /* We also count the list node memory into replication buffer memory. */ + server.repl_buffer_mem += (usable_size + sizeof(listNode)); + add_new_block = 1; + if (start_node == NULL) { + start_node = listLast(server.repl_buffer_blocks); + start_pos = 0; + } + s += copy; + len -= copy; + server.master_repl_offset += copy; + server.repl_backlog->histlen += copy; + } + + /* For output buffer of replicas. */ + listIter li; + listRewind(server.slaves,&li); + while((ln = listNext(&li))) { + client *slave = ln->value; + if (!canFeedReplicaReplBuffer(slave)) continue; + + /* Update shared replication buffer start position. */ + if (slave->ref_repl_buf_node == NULL) { + slave->ref_repl_buf_node = start_node; + slave->ref_block_pos = start_pos; + /* Only increase the start block reference count. */ + ((replBufBlock *)listNodeValue(start_node))->refcount++; + } + + /* Check output buffer limit only when add new block. */ + if (add_new_block) closeClientOnOutputBufferLimitReached(slave, 1); + } + + /* For replication backlog */ + if (server.repl_backlog->ref_repl_buf_node == NULL) { + server.repl_backlog->ref_repl_buf_node = start_node; + /* Only increase the start block reference count. */ + ((replBufBlock *)listNodeValue(start_node))->refcount++; + + /* Replication buffer must be empty before adding replication stream + * into replication backlog. */ + serverAssert(add_new_block == 1 && start_pos == 0); + } + if (add_new_block) { + createReplicationBacklogIndex(listLast(server.repl_buffer_blocks)); + + /* It is important to trim after adding replication data to keep the backlog size close to + * repl_backlog_size in the common case. We wait until we add a new block to avoid repeated + * unnecessary trimming attempts when small amounts of data are added. See comments in + * freeMemoryGetNotCountedMemory() for details on replication backlog memory tracking. */ + incrementalTrimReplicationBacklog(REPL_BACKLOG_TRIM_BLOCKS_PER_CALL); + } + } +} + +/* Propagate write commands to replication stream. + * + * This function is used if the instance is a master: we use the commands + * received by our clients in order to create the replication stream. + * Instead if the instance is a replica and has sub-replicas attached, we use + * replicationFeedStreamFromMasterStream() */ +void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) { + int j, len; + char llstr[LONG_STR_SIZE]; + + /* In case we propagate a command that doesn't touch keys (PING, REPLCONF) we + * pass dbid=-1 that indicate there is no need to replicate `select` command. */ + serverAssert(dictid == -1 || (dictid >= 0 && dictid < server.dbnum)); + + /* If the instance is not a top level master, return ASAP: we'll just proxy + * the stream of data we receive from our master instead, in order to + * propagate *identical* replication stream. In this way this slave can + * advertise the same replication ID as the master (since it shares the + * master replication history and has the same backlog and offsets). */ + if (server.masterhost != NULL) return; + + /* If there aren't slaves, and there is no backlog buffer to populate, + * we can return ASAP. */ + if (server.repl_backlog == NULL && listLength(slaves) == 0) { + /* We increment the repl_offset anyway, since we use that for tracking AOF fsyncs + * even when there's no replication active. This code will not be reached if AOF + * is also disabled. */ + server.master_repl_offset += 1; + return; + } + + /* We can't have slaves attached and no backlog. */ + serverAssert(!(listLength(slaves) != 0 && server.repl_backlog == NULL)); + + /* Must install write handler for all replicas first before feeding + * replication stream. */ + prepareReplicasToWrite(); + + /* Send SELECT command to every slave if needed. */ + if (dictid != -1 && server.slaveseldb != dictid) { + robj *selectcmd; + + /* For a few DBs we have pre-computed SELECT command. */ + if (dictid >= 0 && dictid < PROTO_SHARED_SELECT_CMDS) { + selectcmd = shared.select[dictid]; + } else { + int dictid_len; + + dictid_len = ll2string(llstr,sizeof(llstr),dictid); + selectcmd = createObject(OBJ_STRING, + sdscatprintf(sdsempty(), + "*2\r\n$6\r\nSELECT\r\n$%d\r\n%s\r\n", + dictid_len, llstr)); + } + + feedReplicationBufferWithObject(selectcmd); + + if (dictid < 0 || dictid >= PROTO_SHARED_SELECT_CMDS) + decrRefCount(selectcmd); + + server.slaveseldb = dictid; + } + + /* Write the command to the replication buffer if any. */ + char aux[LONG_STR_SIZE+3]; + + /* Add the multi bulk reply length. */ + aux[0] = '*'; + len = ll2string(aux+1,sizeof(aux)-1,argc); + aux[len+1] = '\r'; + aux[len+2] = '\n'; + feedReplicationBuffer(aux,len+3); + + for (j = 0; j < argc; j++) { + long objlen = stringObjectLen(argv[j]); + + /* We need to feed the buffer with the object as a bulk reply + * not just as a plain string, so create the $..CRLF payload len + * and add the final CRLF */ + aux[0] = '$'; + len = ll2string(aux+1,sizeof(aux)-1,objlen); + aux[len+1] = '\r'; + aux[len+2] = '\n'; + feedReplicationBuffer(aux,len+3); + feedReplicationBufferWithObject(argv[j]); + feedReplicationBuffer(aux+len+1,2); + } +} + +/* This is a debugging function that gets called when we detect something + * wrong with the replication protocol: the goal is to peek into the + * replication backlog and show a few final bytes to make simpler to + * guess what kind of bug it could be. */ +void showLatestBacklog(void) { + if (server.repl_backlog == NULL) return; + if (listLength(server.repl_buffer_blocks) == 0) return; + + size_t dumplen = 256; + if (server.repl_backlog->histlen < (long long)dumplen) + dumplen = server.repl_backlog->histlen; + + sds dump = sdsempty(); + listNode *node = listLast(server.repl_buffer_blocks); + while(dumplen) { + if (node == NULL) break; + replBufBlock *o = listNodeValue(node); + size_t thislen = o->used >= dumplen ? dumplen : o->used; + sds head = sdscatrepr(sdsempty(), o->buf+o->used-thislen, thislen); + sds tmp = sdscatsds(head, dump); + sdsfree(dump); + dump = tmp; + dumplen -= thislen; + node = listPrevNode(node); + } + + /* Finally log such bytes: this is vital debugging info to + * understand what happened. */ + serverLog(LL_NOTICE,"Latest backlog is: '%s'", dump); + sdsfree(dump); +} + +/* This function is used in order to proxy what we receive from our master + * to our sub-slaves. */ +#include <ctype.h> +void replicationFeedStreamFromMasterStream(char *buf, size_t buflen) { + /* Debugging: this is handy to see the stream sent from master + * to slaves. Disabled with if(0). */ + if (0) { + printf("%zu:",buflen); + for (size_t j = 0; j < buflen; j++) { + printf("%c", isprint(buf[j]) ? buf[j] : '.'); + } + printf("\n"); + } + + /* There must be replication backlog if having attached slaves. */ + if (listLength(server.slaves)) serverAssert(server.repl_backlog != NULL); + if (server.repl_backlog) { + /* Must install write handler for all replicas first before feeding + * replication stream. */ + prepareReplicasToWrite(); + feedReplicationBuffer(buf,buflen); + } +} + +void replicationFeedMonitors(client *c, list *monitors, int dictid, robj **argv, int argc) { + /* Fast path to return if the monitors list is empty or the server is in loading. */ + if (monitors == NULL || listLength(monitors) == 0 || server.loading) return; + listNode *ln; + listIter li; + int j; + sds cmdrepr = sdsnew("+"); + robj *cmdobj; + struct timeval tv; + + gettimeofday(&tv,NULL); + cmdrepr = sdscatprintf(cmdrepr,"%ld.%06ld ",(long)tv.tv_sec,(long)tv.tv_usec); + if (c->flags & CLIENT_SCRIPT) { + cmdrepr = sdscatprintf(cmdrepr,"[%d lua] ",dictid); + } else if (c->flags & CLIENT_UNIX_SOCKET) { + cmdrepr = sdscatprintf(cmdrepr,"[%d unix:%s] ",dictid,server.unixsocket); + } else { + cmdrepr = sdscatprintf(cmdrepr,"[%d %s] ",dictid,getClientPeerId(c)); + } + + for (j = 0; j < argc; j++) { + if (argv[j]->encoding == OBJ_ENCODING_INT) { + cmdrepr = sdscatprintf(cmdrepr, "\"%ld\"", (long)argv[j]->ptr); + } else { + cmdrepr = sdscatrepr(cmdrepr,(char*)argv[j]->ptr, + sdslen(argv[j]->ptr)); + } + if (j != argc-1) + cmdrepr = sdscatlen(cmdrepr," ",1); + } + cmdrepr = sdscatlen(cmdrepr,"\r\n",2); + cmdobj = createObject(OBJ_STRING,cmdrepr); + + listRewind(monitors,&li); + while((ln = listNext(&li))) { + client *monitor = ln->value; + addReply(monitor,cmdobj); + updateClientMemUsageAndBucket(monitor); + } + decrRefCount(cmdobj); +} + +/* Feed the slave 'c' with the replication backlog starting from the + * specified 'offset' up to the end of the backlog. */ +long long addReplyReplicationBacklog(client *c, long long offset) { + long long skip; + + serverLog(LL_DEBUG, "[PSYNC] Replica request offset: %lld", offset); + + if (server.repl_backlog->histlen == 0) { + serverLog(LL_DEBUG, "[PSYNC] Backlog history len is zero"); + return 0; + } + + serverLog(LL_DEBUG, "[PSYNC] Backlog size: %lld", + server.repl_backlog_size); + serverLog(LL_DEBUG, "[PSYNC] First byte: %lld", + server.repl_backlog->offset); + serverLog(LL_DEBUG, "[PSYNC] History len: %lld", + server.repl_backlog->histlen); + + /* Compute the amount of bytes we need to discard. */ + skip = offset - server.repl_backlog->offset; + serverLog(LL_DEBUG, "[PSYNC] Skipping: %lld", skip); + + /* Iterate recorded blocks, quickly search the approximate node. */ + listNode *node = NULL; + if (raxSize(server.repl_backlog->blocks_index) > 0) { + uint64_t encoded_offset = htonu64(offset); + raxIterator ri; + raxStart(&ri, server.repl_backlog->blocks_index); + raxSeek(&ri, ">", (unsigned char*)&encoded_offset, sizeof(uint64_t)); + if (raxEOF(&ri)) { + /* No found, so search from the last recorded node. */ + raxSeek(&ri, "$", NULL, 0); + raxPrev(&ri); + node = (listNode *)ri.data; + } else { + raxPrev(&ri); /* Skip the sought node. */ + /* We should search from the prev node since the offset of current + * sought node exceeds searching offset. */ + if (raxPrev(&ri)) + node = (listNode *)ri.data; + else + node = server.repl_backlog->ref_repl_buf_node; + } + raxStop(&ri); + } else { + /* No recorded blocks, just from the start node to search. */ + node = server.repl_backlog->ref_repl_buf_node; + } + + /* Search the exact node. */ + while (node != NULL) { + replBufBlock *o = listNodeValue(node); + if (o->repl_offset + (long long)o->used >= offset) break; + node = listNextNode(node); + } + serverAssert(node != NULL); + + /* Install a writer handler first.*/ + prepareClientToWrite(c); + /* Setting output buffer of the replica. */ + replBufBlock *o = listNodeValue(node); + o->refcount++; + c->ref_repl_buf_node = node; + c->ref_block_pos = offset - o->repl_offset; + + return server.repl_backlog->histlen - skip; +} + +/* Return the offset to provide as reply to the PSYNC command received + * from the slave. The returned value is only valid immediately after + * the BGSAVE process started and before executing any other command + * from clients. */ +long long getPsyncInitialOffset(void) { + return server.master_repl_offset; +} + +/* Send a FULLRESYNC reply in the specific case of a full resynchronization, + * as a side effect setup the slave for a full sync in different ways: + * + * 1) Remember, into the slave client structure, the replication offset + * we sent here, so that if new slaves will later attach to the same + * background RDB saving process (by duplicating this client output + * buffer), we can get the right offset from this slave. + * 2) Set the replication state of the slave to WAIT_BGSAVE_END so that + * we start accumulating differences from this point. + * 3) Force the replication stream to re-emit a SELECT statement so + * the new slave incremental differences will start selecting the + * right database number. + * + * Normally this function should be called immediately after a successful + * BGSAVE for replication was started, or when there is one already in + * progress that we attached our slave to. */ +int replicationSetupSlaveForFullResync(client *slave, long long offset) { + char buf[128]; + int buflen; + + slave->psync_initial_offset = offset; + slave->replstate = SLAVE_STATE_WAIT_BGSAVE_END; + /* We are going to accumulate the incremental changes for this + * slave as well. Set slaveseldb to -1 in order to force to re-emit + * a SELECT statement in the replication stream. */ + server.slaveseldb = -1; + + /* Don't send this reply to slaves that approached us with + * the old SYNC command. */ + if (!(slave->flags & CLIENT_PRE_PSYNC)) { + buflen = snprintf(buf,sizeof(buf),"+FULLRESYNC %s %lld\r\n", + server.replid,offset); + if (connWrite(slave->conn,buf,buflen) != buflen) { + freeClientAsync(slave); + return C_ERR; + } + } + return C_OK; +} + +/* This function handles the PSYNC command from the point of view of a + * master receiving a request for partial resynchronization. + * + * On success return C_OK, otherwise C_ERR is returned and we proceed + * with the usual full resync. */ +int masterTryPartialResynchronization(client *c, long long psync_offset) { + long long psync_len; + char *master_replid = c->argv[1]->ptr; + char buf[128]; + int buflen; + + /* Is the replication ID of this master the same advertised by the wannabe + * slave via PSYNC? If the replication ID changed this master has a + * different replication history, and there is no way to continue. + * + * Note that there are two potentially valid replication IDs: the ID1 + * and the ID2. The ID2 however is only valid up to a specific offset. */ + if (strcasecmp(master_replid, server.replid) && + (strcasecmp(master_replid, server.replid2) || + psync_offset > server.second_replid_offset)) + { + /* Replid "?" is used by slaves that want to force a full resync. */ + if (master_replid[0] != '?') { + if (strcasecmp(master_replid, server.replid) && + strcasecmp(master_replid, server.replid2)) + { + serverLog(LL_NOTICE,"Partial resynchronization not accepted: " + "Replication ID mismatch (Replica asked for '%s', my " + "replication IDs are '%s' and '%s')", + master_replid, server.replid, server.replid2); + } else { + serverLog(LL_NOTICE,"Partial resynchronization not accepted: " + "Requested offset for second ID was %lld, but I can reply " + "up to %lld", psync_offset, server.second_replid_offset); + } + } else { + serverLog(LL_NOTICE,"Full resync requested by replica %s", + replicationGetSlaveName(c)); + } + goto need_full_resync; + } + + /* We still have the data our slave is asking for? */ + if (!server.repl_backlog || + psync_offset < server.repl_backlog->offset || + psync_offset > (server.repl_backlog->offset + server.repl_backlog->histlen)) + { + serverLog(LL_NOTICE, + "Unable to partial resync with replica %s for lack of backlog (Replica request was: %lld).", replicationGetSlaveName(c), psync_offset); + if (psync_offset > server.master_repl_offset) { + serverLog(LL_WARNING, + "Warning: replica %s tried to PSYNC with an offset that is greater than the master replication offset.", replicationGetSlaveName(c)); + } + goto need_full_resync; + } + + /* If we reached this point, we are able to perform a partial resync: + * 1) Set client state to make it a slave. + * 2) Inform the client we can continue with +CONTINUE + * 3) Send the backlog data (from the offset to the end) to the slave. */ + c->flags |= CLIENT_SLAVE; + c->replstate = SLAVE_STATE_ONLINE; + c->repl_ack_time = server.unixtime; + c->repl_start_cmd_stream_on_ack = 0; + listAddNodeTail(server.slaves,c); + /* We can't use the connection buffers since they are used to accumulate + * new commands at this stage. But we are sure the socket send buffer is + * empty so this write will never fail actually. */ + if (c->slave_capa & SLAVE_CAPA_PSYNC2) { + buflen = snprintf(buf,sizeof(buf),"+CONTINUE %s\r\n", server.replid); + } else { + buflen = snprintf(buf,sizeof(buf),"+CONTINUE\r\n"); + } + if (connWrite(c->conn,buf,buflen) != buflen) { + freeClientAsync(c); + return C_OK; + } + psync_len = addReplyReplicationBacklog(c,psync_offset); + serverLog(LL_NOTICE, + "Partial resynchronization request from %s accepted. Sending %lld bytes of backlog starting from offset %lld.", + replicationGetSlaveName(c), + psync_len, psync_offset); + /* Note that we don't need to set the selected DB at server.slaveseldb + * to -1 to force the master to emit SELECT, since the slave already + * has this state from the previous connection with the master. */ + + refreshGoodSlavesCount(); + + /* Fire the replica change modules event. */ + moduleFireServerEvent(REDISMODULE_EVENT_REPLICA_CHANGE, + REDISMODULE_SUBEVENT_REPLICA_CHANGE_ONLINE, + NULL); + + return C_OK; /* The caller can return, no full resync needed. */ + +need_full_resync: + /* We need a full resync for some reason... Note that we can't + * reply to PSYNC right now if a full SYNC is needed. The reply + * must include the master offset at the time the RDB file we transfer + * is generated, so we need to delay the reply to that moment. */ + return C_ERR; +} + +/* Start a BGSAVE for replication goals, which is, selecting the disk or + * socket target depending on the configuration, and making sure that + * the script cache is flushed before to start. + * + * The mincapa argument is the bitwise AND among all the slaves capabilities + * of the slaves waiting for this BGSAVE, so represents the slave capabilities + * all the slaves support. Can be tested via SLAVE_CAPA_* macros. + * + * Side effects, other than starting a BGSAVE: + * + * 1) Handle the slaves in WAIT_START state, by preparing them for a full + * sync if the BGSAVE was successfully started, or sending them an error + * and dropping them from the list of slaves. + * + * 2) Flush the Lua scripting script cache if the BGSAVE was actually + * started. + * + * Returns C_OK on success or C_ERR otherwise. */ +int startBgsaveForReplication(int mincapa, int req) { + int retval; + int socket_target = 0; + listIter li; + listNode *ln; + + /* We use a socket target if slave can handle the EOF marker and we're configured to do diskless syncs. + * Note that in case we're creating a "filtered" RDB (functions-only, for example) we also force socket replication + * to avoid overwriting the snapshot RDB file with filtered data. */ + socket_target = (server.repl_diskless_sync || req & SLAVE_REQ_RDB_MASK) && (mincapa & SLAVE_CAPA_EOF); + /* `SYNC` should have failed with error if we don't support socket and require a filter, assert this here */ + serverAssert(socket_target || !(req & SLAVE_REQ_RDB_MASK)); + + serverLog(LL_NOTICE,"Starting BGSAVE for SYNC with target: %s", + socket_target ? "replicas sockets" : "disk"); + + rdbSaveInfo rsi, *rsiptr; + rsiptr = rdbPopulateSaveInfo(&rsi); + /* Only do rdbSave* when rsiptr is not NULL, + * otherwise slave will miss repl-stream-db. */ + if (rsiptr) { + if (socket_target) + retval = rdbSaveToSlavesSockets(req,rsiptr); + else { + /* Keep the page cache since it'll get used soon */ + retval = rdbSaveBackground(req,server.rdb_filename,rsiptr,RDBFLAGS_KEEP_CACHE); + } + } else { + serverLog(LL_WARNING,"BGSAVE for replication: replication information not available, can't generate the RDB file right now. Try later."); + retval = C_ERR; + } + + /* If we succeeded to start a BGSAVE with disk target, let's remember + * this fact, so that we can later delete the file if needed. Note + * that we don't set the flag to 1 if the feature is disabled, otherwise + * it would never be cleared: the file is not deleted. This way if + * the user enables it later with CONFIG SET, we are fine. */ + if (retval == C_OK && !socket_target && server.rdb_del_sync_files) + RDBGeneratedByReplication = 1; + + /* If we failed to BGSAVE, remove the slaves waiting for a full + * resynchronization from the list of slaves, inform them with + * an error about what happened, close the connection ASAP. */ + if (retval == C_ERR) { + serverLog(LL_WARNING,"BGSAVE for replication failed"); + listRewind(server.slaves,&li); + while((ln = listNext(&li))) { + client *slave = ln->value; + + if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) { + slave->replstate = REPL_STATE_NONE; + slave->flags &= ~CLIENT_SLAVE; + listDelNode(server.slaves,ln); + addReplyError(slave, + "BGSAVE failed, replication can't continue"); + slave->flags |= CLIENT_CLOSE_AFTER_REPLY; + } + } + return retval; + } + + /* If the target is socket, rdbSaveToSlavesSockets() already setup + * the slaves for a full resync. Otherwise for disk target do it now.*/ + if (!socket_target) { + listRewind(server.slaves,&li); + while((ln = listNext(&li))) { + client *slave = ln->value; + + if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) { + /* Check slave has the exact requirements */ + if (slave->slave_req != req) + continue; + replicationSetupSlaveForFullResync(slave, getPsyncInitialOffset()); + } + } + } + + return retval; +} + +/* SYNC and PSYNC command implementation. */ +void syncCommand(client *c) { + /* ignore SYNC if already slave or in monitor mode */ + if (c->flags & CLIENT_SLAVE) return; + + /* Check if this is a failover request to a replica with the same replid and + * become a master if so. */ + if (c->argc > 3 && !strcasecmp(c->argv[0]->ptr,"psync") && + !strcasecmp(c->argv[3]->ptr,"failover")) + { + serverLog(LL_NOTICE, "Failover request received for replid %s.", + (unsigned char *)c->argv[1]->ptr); + if (!server.masterhost) { + addReplyError(c, "PSYNC FAILOVER can't be sent to a master."); + return; + } + + if (!strcasecmp(c->argv[1]->ptr,server.replid)) { + replicationUnsetMaster(); + sds client = catClientInfoString(sdsempty(),c); + serverLog(LL_NOTICE, + "MASTER MODE enabled (failover request from '%s')",client); + sdsfree(client); + } else { + addReplyError(c, "PSYNC FAILOVER replid must match my replid."); + return; + } + } + + /* Don't let replicas sync with us while we're failing over */ + if (server.failover_state != NO_FAILOVER) { + addReplyError(c,"-NOMASTERLINK Can't SYNC while failing over"); + return; + } + + /* Refuse SYNC requests if we are a slave but the link with our master + * is not ok... */ + if (server.masterhost && server.repl_state != REPL_STATE_CONNECTED) { + addReplyError(c,"-NOMASTERLINK Can't SYNC while not connected with my master"); + return; + } + + /* SYNC can't be issued when the server has pending data to send to + * the client about already issued commands. We need a fresh reply + * buffer registering the differences between the BGSAVE and the current + * dataset, so that we can copy to other slaves if needed. */ + if (clientHasPendingReplies(c)) { + addReplyError(c,"SYNC and PSYNC are invalid with pending output"); + return; + } + + /* Fail sync if slave doesn't support EOF capability but wants a filtered RDB. This is because we force filtered + * RDB's to be generated over a socket and not through a file to avoid conflicts with the snapshot files. Forcing + * use of a socket is handled, if needed, in `startBgsaveForReplication`. */ + if (c->slave_req & SLAVE_REQ_RDB_MASK && !(c->slave_capa & SLAVE_CAPA_EOF)) { + addReplyError(c,"Filtered replica requires EOF capability"); + return; + } + + serverLog(LL_NOTICE,"Replica %s asks for synchronization", + replicationGetSlaveName(c)); + + /* Try a partial resynchronization if this is a PSYNC command. + * If it fails, we continue with usual full resynchronization, however + * when this happens replicationSetupSlaveForFullResync will replied + * with: + * + * +FULLRESYNC <replid> <offset> + * + * So the slave knows the new replid and offset to try a PSYNC later + * if the connection with the master is lost. */ + if (!strcasecmp(c->argv[0]->ptr,"psync")) { + long long psync_offset; + if (getLongLongFromObjectOrReply(c, c->argv[2], &psync_offset, NULL) != C_OK) { + serverLog(LL_WARNING, "Replica %s asks for synchronization but with a wrong offset", + replicationGetSlaveName(c)); + return; + } + + if (masterTryPartialResynchronization(c, psync_offset) == C_OK) { + server.stat_sync_partial_ok++; + return; /* No full resync needed, return. */ + } else { + char *master_replid = c->argv[1]->ptr; + + /* Increment stats for failed PSYNCs, but only if the + * replid is not "?", as this is used by slaves to force a full + * resync on purpose when they are not able to partially + * resync. */ + if (master_replid[0] != '?') server.stat_sync_partial_err++; + } + } else { + /* If a slave uses SYNC, we are dealing with an old implementation + * of the replication protocol (like redis-cli --slave). Flag the client + * so that we don't expect to receive REPLCONF ACK feedbacks. */ + c->flags |= CLIENT_PRE_PSYNC; + } + + /* Full resynchronization. */ + server.stat_sync_full++; + + /* Setup the slave as one waiting for BGSAVE to start. The following code + * paths will change the state if we handle the slave differently. */ + c->replstate = SLAVE_STATE_WAIT_BGSAVE_START; + if (server.repl_disable_tcp_nodelay) + connDisableTcpNoDelay(c->conn); /* Non critical if it fails. */ + c->repldbfd = -1; + c->flags |= CLIENT_SLAVE; + listAddNodeTail(server.slaves,c); + + /* Create the replication backlog if needed. */ + if (listLength(server.slaves) == 1 && server.repl_backlog == NULL) { + /* When we create the backlog from scratch, we always use a new + * replication ID and clear the ID2, since there is no valid + * past history. */ + changeReplicationId(); + clearReplicationId2(); + createReplicationBacklog(); + serverLog(LL_NOTICE,"Replication backlog created, my new " + "replication IDs are '%s' and '%s'", + server.replid, server.replid2); + } + + /* CASE 1: BGSAVE is in progress, with disk target. */ + if (server.child_type == CHILD_TYPE_RDB && + server.rdb_child_type == RDB_CHILD_TYPE_DISK) + { + /* Ok a background save is in progress. Let's check if it is a good + * one for replication, i.e. if there is another slave that is + * registering differences since the server forked to save. */ + client *slave; + listNode *ln; + listIter li; + + listRewind(server.slaves,&li); + while((ln = listNext(&li))) { + slave = ln->value; + /* If the client needs a buffer of commands, we can't use + * a replica without replication buffer. */ + if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_END && + (!(slave->flags & CLIENT_REPL_RDBONLY) || + (c->flags & CLIENT_REPL_RDBONLY))) + break; + } + /* To attach this slave, we check that it has at least all the + * capabilities of the slave that triggered the current BGSAVE + * and its exact requirements. */ + if (ln && ((c->slave_capa & slave->slave_capa) == slave->slave_capa) && + c->slave_req == slave->slave_req) { + /* Perfect, the server is already registering differences for + * another slave. Set the right state, and copy the buffer. + * We don't copy buffer if clients don't want. */ + if (!(c->flags & CLIENT_REPL_RDBONLY)) + copyReplicaOutputBuffer(c,slave); + replicationSetupSlaveForFullResync(c,slave->psync_initial_offset); + serverLog(LL_NOTICE,"Waiting for end of BGSAVE for SYNC"); + } else { + /* No way, we need to wait for the next BGSAVE in order to + * register differences. */ + serverLog(LL_NOTICE,"Can't attach the replica to the current BGSAVE. Waiting for next BGSAVE for SYNC"); + } + + /* CASE 2: BGSAVE is in progress, with socket target. */ + } else if (server.child_type == CHILD_TYPE_RDB && + server.rdb_child_type == RDB_CHILD_TYPE_SOCKET) + { + /* There is an RDB child process but it is writing directly to + * children sockets. We need to wait for the next BGSAVE + * in order to synchronize. */ + serverLog(LL_NOTICE,"Current BGSAVE has socket target. Waiting for next BGSAVE for SYNC"); + + /* CASE 3: There is no BGSAVE is in progress. */ + } else { + if (server.repl_diskless_sync && (c->slave_capa & SLAVE_CAPA_EOF) && + server.repl_diskless_sync_delay) + { + /* Diskless replication RDB child is created inside + * replicationCron() since we want to delay its start a + * few seconds to wait for more slaves to arrive. */ + serverLog(LL_NOTICE,"Delay next BGSAVE for diskless SYNC"); + } else { + /* We don't have a BGSAVE in progress, let's start one. Diskless + * or disk-based mode is determined by replica's capacity. */ + if (!hasActiveChildProcess()) { + startBgsaveForReplication(c->slave_capa, c->slave_req); + } else { + serverLog(LL_NOTICE, + "No BGSAVE in progress, but another BG operation is active. " + "BGSAVE for replication delayed"); + } + } + } + return; +} + +/* REPLCONF <option> <value> <option> <value> ... + * This command is used by a replica in order to configure the replication + * process before starting it with the SYNC command. + * This command is also used by a master in order to get the replication + * offset from a replica. + * + * Currently we support these options: + * + * - listening-port <port> + * - ip-address <ip> + * What is the listening ip and port of the Replica redis instance, so that + * the master can accurately lists replicas and their listening ports in the + * INFO output. + * + * - capa <eof|psync2> + * What is the capabilities of this instance. + * eof: supports EOF-style RDB transfer for diskless replication. + * psync2: supports PSYNC v2, so understands +CONTINUE <new repl ID>. + * + * - ack <offset> [fack <aofofs>] + * Replica informs the master the amount of replication stream that it + * processed so far, and optionally the replication offset fsynced to the AOF file. + * This special pattern doesn't reply to the caller. + * + * - getack <dummy> + * Unlike other subcommands, this is used by master to get the replication + * offset from a replica. + * + * - rdb-only <0|1> + * Only wants RDB snapshot without replication buffer. + * + * - rdb-filter-only <include-filters> + * Define "include" filters for the RDB snapshot. Currently we only support + * a single include filter: "functions". Passing an empty string "" will + * result in an empty RDB. */ +void replconfCommand(client *c) { + int j; + + if ((c->argc % 2) == 0) { + /* Number of arguments must be odd to make sure that every + * option has a corresponding value. */ + addReplyErrorObject(c,shared.syntaxerr); + return; + } + + /* Process every option-value pair. */ + for (j = 1; j < c->argc; j+=2) { + if (!strcasecmp(c->argv[j]->ptr,"listening-port")) { + long port; + + if ((getLongFromObjectOrReply(c,c->argv[j+1], + &port,NULL) != C_OK)) + return; + c->slave_listening_port = port; + } else if (!strcasecmp(c->argv[j]->ptr,"ip-address")) { + sds addr = c->argv[j+1]->ptr; + if (sdslen(addr) < NET_HOST_STR_LEN) { + if (c->slave_addr) sdsfree(c->slave_addr); + c->slave_addr = sdsdup(addr); + } else { + addReplyErrorFormat(c,"REPLCONF ip-address provided by " + "replica instance is too long: %zd bytes", sdslen(addr)); + return; + } + } else if (!strcasecmp(c->argv[j]->ptr,"capa")) { + /* Ignore capabilities not understood by this master. */ + if (!strcasecmp(c->argv[j+1]->ptr,"eof")) + c->slave_capa |= SLAVE_CAPA_EOF; + else if (!strcasecmp(c->argv[j+1]->ptr,"psync2")) + c->slave_capa |= SLAVE_CAPA_PSYNC2; + } else if (!strcasecmp(c->argv[j]->ptr,"ack")) { + /* REPLCONF ACK is used by slave to inform the master the amount + * of replication stream that it processed so far. It is an + * internal only command that normal clients should never use. */ + long long offset; + + if (!(c->flags & CLIENT_SLAVE)) return; + if ((getLongLongFromObject(c->argv[j+1], &offset) != C_OK)) + return; + if (offset > c->repl_ack_off) + c->repl_ack_off = offset; + if (c->argc > j+3 && !strcasecmp(c->argv[j+2]->ptr,"fack")) { + if ((getLongLongFromObject(c->argv[j+3], &offset) != C_OK)) + return; + if (offset > c->repl_aof_off) + c->repl_aof_off = offset; + } + c->repl_ack_time = server.unixtime; + /* If this was a diskless replication, we need to really put + * the slave online when the first ACK is received (which + * confirms slave is online and ready to get more data). This + * allows for simpler and less CPU intensive EOF detection + * when streaming RDB files. + * There's a chance the ACK got to us before we detected that the + * bgsave is done (since that depends on cron ticks), so run a + * quick check first (instead of waiting for the next ACK. */ + if (server.child_type == CHILD_TYPE_RDB && c->replstate == SLAVE_STATE_WAIT_BGSAVE_END) + checkChildrenDone(); + if (c->repl_start_cmd_stream_on_ack && c->replstate == SLAVE_STATE_ONLINE) + replicaStartCommandStream(c); + /* Note: this command does not reply anything! */ + return; + } else if (!strcasecmp(c->argv[j]->ptr,"getack")) { + /* REPLCONF GETACK is used in order to request an ACK ASAP + * to the slave. */ + if (server.masterhost && server.master) replicationSendAck(); + return; + } else if (!strcasecmp(c->argv[j]->ptr,"rdb-only")) { + /* REPLCONF RDB-ONLY is used to identify the client only wants + * RDB snapshot without replication buffer. */ + long rdb_only = 0; + if (getRangeLongFromObjectOrReply(c,c->argv[j+1], + 0,1,&rdb_only,NULL) != C_OK) + return; + if (rdb_only == 1) c->flags |= CLIENT_REPL_RDBONLY; + else c->flags &= ~CLIENT_REPL_RDBONLY; + } else if (!strcasecmp(c->argv[j]->ptr,"rdb-filter-only")) { + /* REPLCONFG RDB-FILTER-ONLY is used to define "include" filters + * for the RDB snapshot. Currently we only support a single + * include filter: "functions". In the future we may want to add + * other filters like key patterns, key types, non-volatile, module + * aux fields, ... + * We might want to add the complementing "RDB-FILTER-EXCLUDE" to + * filter out certain data. */ + int filter_count, i; + sds *filters; + if (!(filters = sdssplitargs(c->argv[j+1]->ptr, &filter_count))) { + addReplyErrorFormat(c, "Missing rdb-filter-only values"); + return; + } + /* By default filter out all parts of the rdb */ + c->slave_req |= SLAVE_REQ_RDB_EXCLUDE_DATA; + c->slave_req |= SLAVE_REQ_RDB_EXCLUDE_FUNCTIONS; + for (i = 0; i < filter_count; i++) { + if (!strcasecmp(filters[i], "functions")) + c->slave_req &= ~SLAVE_REQ_RDB_EXCLUDE_FUNCTIONS; + else { + addReplyErrorFormat(c, "Unsupported rdb-filter-only option: %s", (char*)filters[i]); + sdsfreesplitres(filters, filter_count); + return; + } + } + sdsfreesplitres(filters, filter_count); + } else { + addReplyErrorFormat(c,"Unrecognized REPLCONF option: %s", + (char*)c->argv[j]->ptr); + return; + } + } + addReply(c,shared.ok); +} + +/* This function puts a replica in the online state, and should be called just + * after a replica received the RDB file for the initial synchronization. + * + * It does a few things: + * 1) Put the slave in ONLINE state. + * 2) Update the count of "good replicas". + * 3) Trigger the module event. + * + * the return value indicates that the replica should be disconnected. + * */ +int replicaPutOnline(client *slave) { + if (slave->flags & CLIENT_REPL_RDBONLY) { + slave->replstate = SLAVE_STATE_RDB_TRANSMITTED; + /* The client asked for RDB only so we should close it ASAP */ + serverLog(LL_NOTICE, + "RDB transfer completed, rdb only replica (%s) should be disconnected asap", + replicationGetSlaveName(slave)); + return 0; + } + slave->replstate = SLAVE_STATE_ONLINE; + slave->repl_ack_time = server.unixtime; /* Prevent false timeout. */ + + refreshGoodSlavesCount(); + /* Fire the replica change modules event. */ + moduleFireServerEvent(REDISMODULE_EVENT_REPLICA_CHANGE, + REDISMODULE_SUBEVENT_REPLICA_CHANGE_ONLINE, + NULL); + serverLog(LL_NOTICE,"Synchronization with replica %s succeeded", + replicationGetSlaveName(slave)); + return 1; +} + +/* This function should be called just after a replica received the RDB file + * for the initial synchronization, and we are finally ready to send the + * incremental stream of commands. + * + * It does a few things: + * 1) Close the replica's connection async if it doesn't need replication + * commands buffer stream, since it actually isn't a valid replica. + * 2) Make sure the writable event is re-installed, since when calling the SYNC + * command we had no replies and it was disabled, and then we could + * accumulate output buffer data without sending it to the replica so it + * won't get mixed with the RDB stream. */ +void replicaStartCommandStream(client *slave) { + serverAssert(!(slave->flags & CLIENT_REPL_RDBONLY)); + slave->repl_start_cmd_stream_on_ack = 0; + + putClientInPendingWriteQueue(slave); +} + +/* We call this function periodically to remove an RDB file that was + * generated because of replication, in an instance that is otherwise + * without any persistence. We don't want instances without persistence + * to take RDB files around, this violates certain policies in certain + * environments. */ +void removeRDBUsedToSyncReplicas(void) { + /* If the feature is disabled, return ASAP but also clear the + * RDBGeneratedByReplication flag in case it was set. Otherwise if the + * feature was enabled, but gets disabled later with CONFIG SET, the + * flag may remain set to one: then next time the feature is re-enabled + * via CONFIG SET we have it set even if no RDB was generated + * because of replication recently. */ + if (!server.rdb_del_sync_files) { + RDBGeneratedByReplication = 0; + return; + } + + if (allPersistenceDisabled() && RDBGeneratedByReplication) { + client *slave; + listNode *ln; + listIter li; + + int delrdb = 1; + listRewind(server.slaves,&li); + while((ln = listNext(&li))) { + slave = ln->value; + if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START || + slave->replstate == SLAVE_STATE_WAIT_BGSAVE_END || + slave->replstate == SLAVE_STATE_SEND_BULK) + { + delrdb = 0; + break; /* No need to check the other replicas. */ + } + } + if (delrdb) { + struct stat sb; + if (lstat(server.rdb_filename,&sb) != -1) { + RDBGeneratedByReplication = 0; + serverLog(LL_NOTICE, + "Removing the RDB file used to feed replicas " + "in a persistence-less instance"); + bg_unlink(server.rdb_filename); + } + } + } +} + +/* Close the repldbfd and reclaim the page cache if the client hold + * the last reference to replication DB */ +void closeRepldbfd(client *myself) { + listNode *ln; + listIter li; + int reclaim = 1; + listRewind(server.slaves,&li); + while((ln = listNext(&li))) { + client *slave = ln->value; + if (slave != myself && slave->replstate == SLAVE_STATE_SEND_BULK) { + reclaim = 0; + break; + } + } + + if (reclaim) { + bioCreateCloseJob(myself->repldbfd, 0, 1); + } else { + close(myself->repldbfd); + } + myself->repldbfd = -1; +} + +void sendBulkToSlave(connection *conn) { + client *slave = connGetPrivateData(conn); + char buf[PROTO_IOBUF_LEN]; + ssize_t nwritten, buflen; + + /* Before sending the RDB file, we send the preamble as configured by the + * replication process. Currently the preamble is just the bulk count of + * the file in the form "$<length>\r\n". */ + if (slave->replpreamble) { + nwritten = connWrite(conn,slave->replpreamble,sdslen(slave->replpreamble)); + if (nwritten == -1) { + serverLog(LL_WARNING, + "Write error sending RDB preamble to replica: %s", + connGetLastError(conn)); + freeClient(slave); + return; + } + atomicIncr(server.stat_net_repl_output_bytes, nwritten); + sdsrange(slave->replpreamble,nwritten,-1); + if (sdslen(slave->replpreamble) == 0) { + sdsfree(slave->replpreamble); + slave->replpreamble = NULL; + /* fall through sending data. */ + } else { + return; + } + } + + /* If the preamble was already transferred, send the RDB bulk data. */ + lseek(slave->repldbfd,slave->repldboff,SEEK_SET); + buflen = read(slave->repldbfd,buf,PROTO_IOBUF_LEN); + if (buflen <= 0) { + serverLog(LL_WARNING,"Read error sending DB to replica: %s", + (buflen == 0) ? "premature EOF" : strerror(errno)); + freeClient(slave); + return; + } + if ((nwritten = connWrite(conn,buf,buflen)) == -1) { + if (connGetState(conn) != CONN_STATE_CONNECTED) { + serverLog(LL_WARNING,"Write error sending DB to replica: %s", + connGetLastError(conn)); + freeClient(slave); + } + return; + } + slave->repldboff += nwritten; + atomicIncr(server.stat_net_repl_output_bytes, nwritten); + if (slave->repldboff == slave->repldbsize) { + closeRepldbfd(slave); + connSetWriteHandler(slave->conn,NULL); + if (!replicaPutOnline(slave)) { + freeClient(slave); + return; + } + replicaStartCommandStream(slave); + } +} + +/* Remove one write handler from the list of connections waiting to be writable + * during rdb pipe transfer. */ +void rdbPipeWriteHandlerConnRemoved(struct connection *conn) { + if (!connHasWriteHandler(conn)) + return; + connSetWriteHandler(conn, NULL); + client *slave = connGetPrivateData(conn); + slave->repl_last_partial_write = 0; + server.rdb_pipe_numconns_writing--; + /* if there are no more writes for now for this conn, or write error: */ + if (server.rdb_pipe_numconns_writing == 0) { + if (aeCreateFileEvent(server.el, server.rdb_pipe_read, AE_READABLE, rdbPipeReadHandler,NULL) == AE_ERR) { + serverPanic("Unrecoverable error creating server.rdb_pipe_read file event."); + } + } +} + +/* Called in diskless master during transfer of data from the rdb pipe, when + * the replica becomes writable again. */ +void rdbPipeWriteHandler(struct connection *conn) { + serverAssert(server.rdb_pipe_bufflen>0); + client *slave = connGetPrivateData(conn); + ssize_t nwritten; + if ((nwritten = connWrite(conn, server.rdb_pipe_buff + slave->repldboff, + server.rdb_pipe_bufflen - slave->repldboff)) == -1) + { + if (connGetState(conn) == CONN_STATE_CONNECTED) + return; /* equivalent to EAGAIN */ + serverLog(LL_WARNING,"Write error sending DB to replica: %s", + connGetLastError(conn)); + freeClient(slave); + return; + } else { + slave->repldboff += nwritten; + atomicIncr(server.stat_net_repl_output_bytes, nwritten); + if (slave->repldboff < server.rdb_pipe_bufflen) { + slave->repl_last_partial_write = server.unixtime; + return; /* more data to write.. */ + } + } + rdbPipeWriteHandlerConnRemoved(conn); +} + +/* Called in diskless master, when there's data to read from the child's rdb pipe */ +void rdbPipeReadHandler(struct aeEventLoop *eventLoop, int fd, void *clientData, int mask) { + UNUSED(mask); + UNUSED(clientData); + UNUSED(eventLoop); + int i; + if (!server.rdb_pipe_buff) + server.rdb_pipe_buff = zmalloc(PROTO_IOBUF_LEN); + serverAssert(server.rdb_pipe_numconns_writing==0); + + while (1) { + server.rdb_pipe_bufflen = read(fd, server.rdb_pipe_buff, PROTO_IOBUF_LEN); + if (server.rdb_pipe_bufflen < 0) { + if (errno == EAGAIN || errno == EWOULDBLOCK) + return; + serverLog(LL_WARNING,"Diskless rdb transfer, read error sending DB to replicas: %s", strerror(errno)); + for (i=0; i < server.rdb_pipe_numconns; i++) { + connection *conn = server.rdb_pipe_conns[i]; + if (!conn) + continue; + client *slave = connGetPrivateData(conn); + freeClient(slave); + server.rdb_pipe_conns[i] = NULL; + } + killRDBChild(); + return; + } + + if (server.rdb_pipe_bufflen == 0) { + /* EOF - write end was closed. */ + int stillUp = 0; + aeDeleteFileEvent(server.el, server.rdb_pipe_read, AE_READABLE); + for (i=0; i < server.rdb_pipe_numconns; i++) + { + connection *conn = server.rdb_pipe_conns[i]; + if (!conn) + continue; + stillUp++; + } + serverLog(LL_NOTICE,"Diskless rdb transfer, done reading from pipe, %d replicas still up.", stillUp); + /* Now that the replicas have finished reading, notify the child that it's safe to exit. + * When the server detects the child has exited, it can mark the replica as online, and + * start streaming the replication buffers. */ + close(server.rdb_child_exit_pipe); + server.rdb_child_exit_pipe = -1; + return; + } + + int stillAlive = 0; + for (i=0; i < server.rdb_pipe_numconns; i++) + { + ssize_t nwritten; + connection *conn = server.rdb_pipe_conns[i]; + if (!conn) + continue; + + client *slave = connGetPrivateData(conn); + if ((nwritten = connWrite(conn, server.rdb_pipe_buff, server.rdb_pipe_bufflen)) == -1) { + if (connGetState(conn) != CONN_STATE_CONNECTED) { + serverLog(LL_WARNING,"Diskless rdb transfer, write error sending DB to replica: %s", + connGetLastError(conn)); + freeClient(slave); + server.rdb_pipe_conns[i] = NULL; + continue; + } + /* An error and still in connected state, is equivalent to EAGAIN */ + slave->repldboff = 0; + } else { + /* Note: when use diskless replication, 'repldboff' is the offset + * of 'rdb_pipe_buff' sent rather than the offset of entire RDB. */ + slave->repldboff = nwritten; + atomicIncr(server.stat_net_repl_output_bytes, nwritten); + } + /* If we were unable to write all the data to one of the replicas, + * setup write handler (and disable pipe read handler, below) */ + if (nwritten != server.rdb_pipe_bufflen) { + slave->repl_last_partial_write = server.unixtime; + server.rdb_pipe_numconns_writing++; + connSetWriteHandler(conn, rdbPipeWriteHandler); + } + stillAlive++; + } + + if (stillAlive == 0) { + serverLog(LL_WARNING,"Diskless rdb transfer, last replica dropped, killing fork child."); + killRDBChild(); + } + /* Remove the pipe read handler if at least one write handler was set. */ + if (server.rdb_pipe_numconns_writing || stillAlive == 0) { + aeDeleteFileEvent(server.el, server.rdb_pipe_read, AE_READABLE); + break; + } + } +} + +/* This function is called at the end of every background saving. + * + * The argument bgsaveerr is C_OK if the background saving succeeded + * otherwise C_ERR is passed to the function. + * The 'type' argument is the type of the child that terminated + * (if it had a disk or socket target). */ +void updateSlavesWaitingBgsave(int bgsaveerr, int type) { + listNode *ln; + listIter li; + + /* Note: there's a chance we got here from within the REPLCONF ACK command + * so we must avoid using freeClient, otherwise we'll crash on our way up. */ + + listRewind(server.slaves,&li); + while((ln = listNext(&li))) { + client *slave = ln->value; + + if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_END) { + struct redis_stat buf; + + if (bgsaveerr != C_OK) { + freeClientAsync(slave); + serverLog(LL_WARNING,"SYNC failed. BGSAVE child returned an error"); + continue; + } + + /* If this was an RDB on disk save, we have to prepare to send + * the RDB from disk to the slave socket. Otherwise if this was + * already an RDB -> Slaves socket transfer, used in the case of + * diskless replication, our work is trivial, we can just put + * the slave online. */ + if (type == RDB_CHILD_TYPE_SOCKET) { + serverLog(LL_NOTICE, + "Streamed RDB transfer with replica %s succeeded (socket). Waiting for REPLCONF ACK from replica to enable streaming", + replicationGetSlaveName(slave)); + /* Note: we wait for a REPLCONF ACK message from the replica in + * order to really put it online (install the write handler + * so that the accumulated data can be transferred). However + * we change the replication state ASAP, since our slave + * is technically online now. + * + * So things work like that: + * + * 1. We end transferring the RDB file via socket. + * 2. The replica is put ONLINE but the write handler + * is not installed. + * 3. The replica however goes really online, and pings us + * back via REPLCONF ACK commands. + * 4. Now we finally install the write handler, and send + * the buffers accumulated so far to the replica. + * + * But why we do that? Because the replica, when we stream + * the RDB directly via the socket, must detect the RDB + * EOF (end of file), that is a special random string at the + * end of the RDB (for streamed RDBs we don't know the length + * in advance). Detecting such final EOF string is much + * simpler and less CPU intensive if no more data is sent + * after such final EOF. So we don't want to glue the end of + * the RDB transfer with the start of the other replication + * data. */ + if (!replicaPutOnline(slave)) { + freeClientAsync(slave); + continue; + } + slave->repl_start_cmd_stream_on_ack = 1; + } else { + if ((slave->repldbfd = open(server.rdb_filename,O_RDONLY)) == -1 || + redis_fstat(slave->repldbfd,&buf) == -1) { + freeClientAsync(slave); + serverLog(LL_WARNING,"SYNC failed. Can't open/stat DB after BGSAVE: %s", strerror(errno)); + continue; + } + slave->repldboff = 0; + slave->repldbsize = buf.st_size; + slave->replstate = SLAVE_STATE_SEND_BULK; + slave->replpreamble = sdscatprintf(sdsempty(),"$%lld\r\n", + (unsigned long long) slave->repldbsize); + + connSetWriteHandler(slave->conn,NULL); + if (connSetWriteHandler(slave->conn,sendBulkToSlave) == C_ERR) { + freeClientAsync(slave); + continue; + } + } + } + } +} + +/* Change the current instance replication ID with a new, random one. + * This will prevent successful PSYNCs between this master and other + * slaves, so the command should be called when something happens that + * alters the current story of the dataset. */ +void changeReplicationId(void) { + getRandomHexChars(server.replid,CONFIG_RUN_ID_SIZE); + server.replid[CONFIG_RUN_ID_SIZE] = '\0'; +} + +/* Clear (invalidate) the secondary replication ID. This happens, for + * example, after a full resynchronization, when we start a new replication + * history. */ +void clearReplicationId2(void) { + memset(server.replid2,'0',sizeof(server.replid)); + server.replid2[CONFIG_RUN_ID_SIZE] = '\0'; + server.second_replid_offset = -1; +} + +/* Use the current replication ID / offset as secondary replication + * ID, and change the current one in order to start a new history. + * This should be used when an instance is switched from slave to master + * so that it can serve PSYNC requests performed using the master + * replication ID. */ +void shiftReplicationId(void) { + memcpy(server.replid2,server.replid,sizeof(server.replid)); + /* We set the second replid offset to the master offset + 1, since + * the slave will ask for the first byte it has not yet received, so + * we need to add one to the offset: for example if, as a slave, we are + * sure we have the same history as the master for 50 bytes, after we + * are turned into a master, we can accept a PSYNC request with offset + * 51, since the slave asking has the same history up to the 50th + * byte, and is asking for the new bytes starting at offset 51. */ + server.second_replid_offset = server.master_repl_offset+1; + changeReplicationId(); + serverLog(LL_NOTICE,"Setting secondary replication ID to %s, valid up to offset: %lld. New replication ID is %s", server.replid2, server.second_replid_offset, server.replid); +} + +/* ----------------------------------- SLAVE -------------------------------- */ + +/* Returns 1 if the given replication state is a handshake state, + * 0 otherwise. */ +int slaveIsInHandshakeState(void) { + return server.repl_state >= REPL_STATE_RECEIVE_PING_REPLY && + server.repl_state <= REPL_STATE_RECEIVE_PSYNC_REPLY; +} + +/* Avoid the master to detect the slave is timing out while loading the + * RDB file in initial synchronization. We send a single newline character + * that is valid protocol but is guaranteed to either be sent entirely or + * not, since the byte is indivisible. + * + * The function is called in two contexts: while we flush the current + * data with emptyDb(), and while we load the new data received as an + * RDB file from the master. */ +void replicationSendNewlineToMaster(void) { + static time_t newline_sent; + if (time(NULL) != newline_sent) { + newline_sent = time(NULL); + /* Pinging back in this stage is best-effort. */ + if (server.repl_transfer_s) connWrite(server.repl_transfer_s, "\n", 1); + } +} + +/* Callback used by emptyDb() while flushing away old data to load + * the new dataset received by the master and by discardTempDb() + * after loading succeeded or failed. */ +void replicationEmptyDbCallback(dict *d) { + UNUSED(d); + if (server.repl_state == REPL_STATE_TRANSFER) + replicationSendNewlineToMaster(); +} + +/* Once we have a link with the master and the synchronization was + * performed, this function materializes the master client we store + * at server.master, starting from the specified file descriptor. */ +void replicationCreateMasterClient(connection *conn, int dbid) { + server.master = createClient(conn); + if (conn) + connSetReadHandler(server.master->conn, readQueryFromClient); + + /** + * Important note: + * The CLIENT_DENY_BLOCKING flag is not, and should not, be set here. + * For commands like BLPOP, it makes no sense to block the master + * connection, and such blocking attempt will probably cause deadlock and + * break the replication. We consider such a thing as a bug because + * commands as BLPOP should never be sent on the replication link. + * A possible use-case for blocking the replication link is if a module wants + * to pass the execution to a background thread and unblock after the + * execution is done. This is the reason why we allow blocking the replication + * connection. */ + server.master->flags |= CLIENT_MASTER; + + server.master->authenticated = 1; + server.master->reploff = server.master_initial_offset; + server.master->read_reploff = server.master->reploff; + server.master->user = NULL; /* This client can do everything. */ + memcpy(server.master->replid, server.master_replid, + sizeof(server.master_replid)); + /* If master offset is set to -1, this master is old and is not + * PSYNC capable, so we flag it accordingly. */ + if (server.master->reploff == -1) + server.master->flags |= CLIENT_PRE_PSYNC; + if (dbid != -1) selectDb(server.master,dbid); +} + +/* This function will try to re-enable the AOF file after the + * master-replica synchronization: if it fails after multiple attempts + * the replica cannot be considered reliable and exists with an + * error. */ +void restartAOFAfterSYNC(void) { + unsigned int tries, max_tries = 10; + for (tries = 0; tries < max_tries; ++tries) { + if (startAppendOnly() == C_OK) break; + serverLog(LL_WARNING, + "Failed enabling the AOF after successful master synchronization! " + "Trying it again in one second."); + sleep(1); + } + if (tries == max_tries) { + serverLog(LL_WARNING, + "FATAL: this replica instance finished the synchronization with " + "its master, but the AOF can't be turned on. Exiting now."); + exit(1); + } +} + +static int useDisklessLoad(void) { + /* compute boolean decision to use diskless load */ + int enabled = server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB || + (server.repl_diskless_load == REPL_DISKLESS_LOAD_WHEN_DB_EMPTY && dbTotalServerKeyCount()==0); + + if (enabled) { + /* Check all modules handle read errors, otherwise it's not safe to use diskless load. */ + if (!moduleAllDatatypesHandleErrors()) { + serverLog(LL_NOTICE, + "Skipping diskless-load because there are modules that don't handle read errors."); + enabled = 0; + } + /* Check all modules handle async replication, otherwise it's not safe to use diskless load. */ + else if (server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB && !moduleAllModulesHandleReplAsyncLoad()) { + serverLog(LL_NOTICE, + "Skipping diskless-load because there are modules that are not aware of async replication."); + enabled = 0; + } + } + return enabled; +} + +/* Helper function for readSyncBulkPayload() to initialize tempDb + * before socket-loading the new db from master. The tempDb may be populated + * by swapMainDbWithTempDb or freed by disklessLoadDiscardTempDb later. */ +redisDb *disklessLoadInitTempDb(void) { + return initTempDb(); +} + +/* Helper function for readSyncBulkPayload() to discard our tempDb + * when the loading succeeded or failed. */ +void disklessLoadDiscardTempDb(redisDb *tempDb) { + discardTempDb(tempDb, replicationEmptyDbCallback); +} + +/* If we know we got an entirely different data set from our master + * we have no way to incrementally feed our replicas after that. + * We want our replicas to resync with us as well, if we have any sub-replicas. + * This is useful on readSyncBulkPayload in places where we just finished transferring db. */ +void replicationAttachToNewMaster(void) { + /* Replica starts to apply data from new master, we must discard the cached + * master structure. */ + serverAssert(server.master == NULL); + replicationDiscardCachedMaster(); + + disconnectSlaves(); /* Force our replicas to resync with us as well. */ + freeReplicationBacklog(); /* Don't allow our chained replicas to PSYNC. */ +} + +/* Asynchronously read the SYNC payload we receive from a master */ +#define REPL_MAX_WRITTEN_BEFORE_FSYNC (1024*1024*8) /* 8 MB */ +void readSyncBulkPayload(connection *conn) { + char buf[PROTO_IOBUF_LEN]; + ssize_t nread, readlen, nwritten; + int use_diskless_load = useDisklessLoad(); + redisDb *diskless_load_tempDb = NULL; + functionsLibCtx* temp_functions_lib_ctx = NULL; + int empty_db_flags = server.repl_slave_lazy_flush ? EMPTYDB_ASYNC : + EMPTYDB_NO_FLAGS; + off_t left; + + /* Static vars used to hold the EOF mark, and the last bytes received + * from the server: when they match, we reached the end of the transfer. */ + static char eofmark[CONFIG_RUN_ID_SIZE]; + static char lastbytes[CONFIG_RUN_ID_SIZE]; + static int usemark = 0; + + /* If repl_transfer_size == -1 we still have to read the bulk length + * from the master reply. */ + if (server.repl_transfer_size == -1) { + nread = connSyncReadLine(conn,buf,1024,server.repl_syncio_timeout*1000); + if (nread == -1) { + serverLog(LL_WARNING, + "I/O error reading bulk count from MASTER: %s", + connGetLastError(conn)); + goto error; + } else { + /* nread here is returned by connSyncReadLine(), which calls syncReadLine() and + * convert "\r\n" to '\0' so 1 byte is lost. */ + atomicIncr(server.stat_net_repl_input_bytes, nread+1); + } + + if (buf[0] == '-') { + serverLog(LL_WARNING, + "MASTER aborted replication with an error: %s", + buf+1); + goto error; + } else if (buf[0] == '\0') { + /* At this stage just a newline works as a PING in order to take + * the connection live. So we refresh our last interaction + * timestamp. */ + server.repl_transfer_lastio = server.unixtime; + return; + } else if (buf[0] != '$') { + 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); + goto error; + } + + /* There are two possible forms for the bulk payload. One is the + * usual $<count> bulk format. The other is used for diskless transfers + * when the master does not know beforehand the size of the file to + * transfer. In the latter case, the following format is used: + * + * $EOF:<40 bytes delimiter> + * + * At the end of the file the announced delimiter is transmitted. The + * delimiter is long and random enough that the probability of a + * collision with the actual file content can be ignored. */ + if (strncmp(buf+1,"EOF:",4) == 0 && strlen(buf+5) >= CONFIG_RUN_ID_SIZE) { + usemark = 1; + memcpy(eofmark,buf+5,CONFIG_RUN_ID_SIZE); + memset(lastbytes,0,CONFIG_RUN_ID_SIZE); + /* Set any repl_transfer_size to avoid entering this code path + * at the next call. */ + server.repl_transfer_size = 0; + serverLog(LL_NOTICE, + "MASTER <-> REPLICA sync: receiving streamed RDB from master with EOF %s", + use_diskless_load? "to parser":"to disk"); + } else { + usemark = 0; + server.repl_transfer_size = strtol(buf+1,NULL,10); + serverLog(LL_NOTICE, + "MASTER <-> REPLICA sync: receiving %lld bytes from master %s", + (long long) server.repl_transfer_size, + use_diskless_load? "to parser":"to disk"); + } + return; + } + + if (!use_diskless_load) { + /* Read the data from the socket, store it to a file and search + * for the EOF. */ + if (usemark) { + readlen = sizeof(buf); + } else { + left = server.repl_transfer_size - server.repl_transfer_read; + readlen = (left < (signed)sizeof(buf)) ? left : (signed)sizeof(buf); + } + + nread = connRead(conn,buf,readlen); + if (nread <= 0) { + if (connGetState(conn) == CONN_STATE_CONNECTED) { + /* equivalent to EAGAIN */ + return; + } + serverLog(LL_WARNING,"I/O error trying to sync with MASTER: %s", + (nread == -1) ? connGetLastError(conn) : "connection lost"); + cancelReplicationHandshake(1); + return; + } + atomicIncr(server.stat_net_repl_input_bytes, nread); + + /* When a mark is used, we want to detect EOF asap in order to avoid + * writing the EOF mark into the file... */ + int eof_reached = 0; + + if (usemark) { + /* Update the last bytes array, and check if it matches our + * delimiter. */ + if (nread >= CONFIG_RUN_ID_SIZE) { + memcpy(lastbytes,buf+nread-CONFIG_RUN_ID_SIZE, + CONFIG_RUN_ID_SIZE); + } else { + int rem = CONFIG_RUN_ID_SIZE-nread; + memmove(lastbytes,lastbytes+nread,rem); + memcpy(lastbytes+rem,buf,nread); + } + if (memcmp(lastbytes,eofmark,CONFIG_RUN_ID_SIZE) == 0) + eof_reached = 1; + } + + /* Update the last I/O time for the replication transfer (used in + * order to detect timeouts during replication), and write what we + * got from the socket to the dump file on disk. */ + server.repl_transfer_lastio = server.unixtime; + if ((nwritten = write(server.repl_transfer_fd,buf,nread)) != nread) { + serverLog(LL_WARNING, + "Write error or short write writing to the DB dump file " + "needed for MASTER <-> REPLICA synchronization: %s", + (nwritten == -1) ? strerror(errno) : "short write"); + goto error; + } + server.repl_transfer_read += nread; + + /* Delete the last 40 bytes from the file if we reached EOF. */ + if (usemark && eof_reached) { + if (ftruncate(server.repl_transfer_fd, + server.repl_transfer_read - CONFIG_RUN_ID_SIZE) == -1) + { + serverLog(LL_WARNING, + "Error truncating the RDB file received from the master " + "for SYNC: %s", strerror(errno)); + goto error; + } + } + + /* Sync data on disk from time to time, otherwise at the end of the + * transfer we may suffer a big delay as the memory buffers are copied + * into the actual disk. */ + if (server.repl_transfer_read >= + server.repl_transfer_last_fsync_off + REPL_MAX_WRITTEN_BEFORE_FSYNC) + { + off_t sync_size = server.repl_transfer_read - + server.repl_transfer_last_fsync_off; + rdb_fsync_range(server.repl_transfer_fd, + server.repl_transfer_last_fsync_off, sync_size); + server.repl_transfer_last_fsync_off += sync_size; + } + + /* Check if the transfer is now complete */ + if (!usemark) { + if (server.repl_transfer_read == server.repl_transfer_size) + eof_reached = 1; + } + + /* If the transfer is yet not complete, we need to read more, so + * return ASAP and wait for the handler to be called again. */ + if (!eof_reached) return; + } + + /* We reach this point in one of the following cases: + * + * 1. The replica is using diskless replication, that is, it reads data + * directly from the socket to the Redis memory, without using + * a temporary RDB file on disk. In that case we just block and + * read everything from the socket. + * + * 2. Or when we are done reading from the socket to the RDB file, in + * such case we want just to read the RDB file in memory. */ + + /* We need to stop any AOF rewriting child before flushing and parsing + * the RDB, otherwise we'll create a copy-on-write disaster. */ + if (server.aof_state != AOF_OFF) stopAppendOnly(); + /* Also try to stop save RDB child before flushing and parsing the RDB: + * 1. Ensure background save doesn't overwrite synced data after being loaded. + * 2. Avoid copy-on-write disaster. */ + if (server.child_type == CHILD_TYPE_RDB) { + if (!use_diskless_load) { + serverLog(LL_NOTICE, + "Replica is about to load the RDB file received from the " + "master, but there is a pending RDB child running. " + "Killing process %ld and removing its temp file to avoid " + "any race", + (long) server.child_pid); + } + killRDBChild(); + } + + if (use_diskless_load && server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB) { + /* Initialize empty tempDb dictionaries. */ + diskless_load_tempDb = disklessLoadInitTempDb(); + temp_functions_lib_ctx = functionsLibCtxCreate(); + + moduleFireServerEvent(REDISMODULE_EVENT_REPL_ASYNC_LOAD, + REDISMODULE_SUBEVENT_REPL_ASYNC_LOAD_STARTED, + NULL); + } else { + replicationAttachToNewMaster(); + + serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Flushing old data"); + emptyData(-1,empty_db_flags,replicationEmptyDbCallback); + } + + /* Before loading the DB into memory we need to delete the readable + * handler, otherwise it will get called recursively since + * rdbLoad() will call the event loop to process events from time to + * time for non blocking loading. */ + connSetReadHandler(conn, NULL); + + serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Loading DB in memory"); + rdbSaveInfo rsi = RDB_SAVE_INFO_INIT; + if (use_diskless_load) { + rio rdb; + redisDb *dbarray; + functionsLibCtx* functions_lib_ctx; + int asyncLoading = 0; + + if (server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB) { + /* Async loading means we continue serving read commands during full resync, and + * "swap" the new db with the old db only when loading is done. + * It is enabled only on SWAPDB diskless replication when master replication ID hasn't changed, + * because in that state the old content of the db represents a different point in time of the same + * data set we're currently receiving from the master. */ + if (memcmp(server.replid, server.master_replid, CONFIG_RUN_ID_SIZE) == 0) { + asyncLoading = 1; + } + dbarray = diskless_load_tempDb; + functions_lib_ctx = temp_functions_lib_ctx; + } else { + dbarray = server.db; + functions_lib_ctx = functionsLibCtxGetCurrent(); + functionsLibCtxClear(functions_lib_ctx); + } + + rioInitWithConn(&rdb,conn,server.repl_transfer_size); + + /* Put the socket in blocking mode to simplify RDB transfer. + * We'll restore it when the RDB is received. */ + connBlock(conn); + connRecvTimeout(conn, server.repl_timeout*1000); + startLoading(server.repl_transfer_size, RDBFLAGS_REPLICATION, asyncLoading); + + int loadingFailed = 0; + rdbLoadingCtx loadingCtx = { .dbarray = dbarray, .functions_lib_ctx = functions_lib_ctx }; + if (rdbLoadRioWithLoadingCtx(&rdb,RDBFLAGS_REPLICATION,&rsi,&loadingCtx) != C_OK) { + /* RDB loading failed. */ + serverLog(LL_WARNING, + "Failed trying to load the MASTER synchronization DB " + "from socket, check server logs."); + loadingFailed = 1; + } else if (usemark) { + /* Verify the end mark is correct. */ + if (!rioRead(&rdb, buf, CONFIG_RUN_ID_SIZE) || + memcmp(buf, eofmark, CONFIG_RUN_ID_SIZE) != 0) + { + serverLog(LL_WARNING, "Replication stream EOF marker is broken"); + loadingFailed = 1; + } + } + + if (loadingFailed) { + stopLoading(0); + cancelReplicationHandshake(1); + rioFreeConn(&rdb, NULL); + + if (server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB) { + /* Discard potentially partially loaded tempDb. */ + moduleFireServerEvent(REDISMODULE_EVENT_REPL_ASYNC_LOAD, + REDISMODULE_SUBEVENT_REPL_ASYNC_LOAD_ABORTED, + NULL); + + disklessLoadDiscardTempDb(diskless_load_tempDb); + functionsLibCtxFree(temp_functions_lib_ctx); + serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Discarding temporary DB in background"); + } else { + /* Remove the half-loaded data in case we started with an empty replica. */ + emptyData(-1,empty_db_flags,replicationEmptyDbCallback); + } + + /* Note that there's no point in restarting the AOF on SYNC + * failure, it'll be restarted when sync succeeds or the replica + * gets promoted. */ + return; + } + + /* RDB loading succeeded if we reach this point. */ + if (server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB) { + /* We will soon swap main db with tempDb and replicas will start + * to apply data from new master, we must discard the cached + * master structure and force resync of sub-replicas. */ + replicationAttachToNewMaster(); + + serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Swapping active DB with loaded DB"); + swapMainDbWithTempDb(diskless_load_tempDb); + + /* swap existing functions ctx with the temporary one */ + functionsLibCtxSwapWithCurrent(temp_functions_lib_ctx); + + moduleFireServerEvent(REDISMODULE_EVENT_REPL_ASYNC_LOAD, + REDISMODULE_SUBEVENT_REPL_ASYNC_LOAD_COMPLETED, + NULL); + + /* Delete the old db as it's useless now. */ + disklessLoadDiscardTempDb(diskless_load_tempDb); + serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Discarding old DB in background"); + } + + /* Inform about db change, as replication was diskless and didn't cause a save. */ + server.dirty++; + + stopLoading(1); + + /* Cleanup and restore the socket to the original state to continue + * with the normal replication. */ + rioFreeConn(&rdb, NULL); + connNonBlock(conn); + connRecvTimeout(conn,0); + } else { + + /* Make sure the new file (also used for persistence) is fully synced + * (not covered by earlier calls to rdb_fsync_range). */ + if (fsync(server.repl_transfer_fd) == -1) { + serverLog(LL_WARNING, + "Failed trying to sync the temp DB to disk in " + "MASTER <-> REPLICA synchronization: %s", + strerror(errno)); + cancelReplicationHandshake(1); + return; + } + + /* Rename rdb like renaming rewrite aof asynchronously. */ + int old_rdb_fd = open(server.rdb_filename,O_RDONLY|O_NONBLOCK); + if (rename(server.repl_transfer_tmpfile,server.rdb_filename) == -1) { + serverLog(LL_WARNING, + "Failed trying to rename the temp DB into %s in " + "MASTER <-> REPLICA synchronization: %s", + server.rdb_filename, strerror(errno)); + cancelReplicationHandshake(1); + if (old_rdb_fd != -1) close(old_rdb_fd); + return; + } + /* Close old rdb asynchronously. */ + if (old_rdb_fd != -1) bioCreateCloseJob(old_rdb_fd, 0, 0); + + /* Sync the directory to ensure rename is persisted */ + if (fsyncFileDir(server.rdb_filename) == -1) { + serverLog(LL_WARNING, + "Failed trying to sync DB directory %s in " + "MASTER <-> REPLICA synchronization: %s", + server.rdb_filename, strerror(errno)); + cancelReplicationHandshake(1); + return; + } + + if (rdbLoad(server.rdb_filename,&rsi,RDBFLAGS_REPLICATION) != RDB_OK) { + serverLog(LL_WARNING, + "Failed trying to load the MASTER synchronization " + "DB from disk, check server logs."); + cancelReplicationHandshake(1); + if (server.rdb_del_sync_files && allPersistenceDisabled()) { + serverLog(LL_NOTICE,"Removing the RDB file obtained from " + "the master. This replica has persistence " + "disabled"); + bg_unlink(server.rdb_filename); + } + /* Note that there's no point in restarting the AOF on sync failure, + it'll be restarted when sync succeeds or replica promoted. */ + return; + } + + /* Cleanup. */ + if (server.rdb_del_sync_files && allPersistenceDisabled()) { + serverLog(LL_NOTICE,"Removing the RDB file obtained from " + "the master. This replica has persistence " + "disabled"); + bg_unlink(server.rdb_filename); + } + + zfree(server.repl_transfer_tmpfile); + close(server.repl_transfer_fd); + server.repl_transfer_fd = -1; + server.repl_transfer_tmpfile = NULL; + } + + /* Final setup of the connected slave <- master link */ + replicationCreateMasterClient(server.repl_transfer_s,rsi.repl_stream_db); + server.repl_state = REPL_STATE_CONNECTED; + server.repl_down_since = 0; + + /* Fire the master link modules event. */ + moduleFireServerEvent(REDISMODULE_EVENT_MASTER_LINK_CHANGE, + REDISMODULE_SUBEVENT_MASTER_LINK_UP, + NULL); + + /* After a full resynchronization we use the replication ID and + * offset of the master. The secondary ID / offset are cleared since + * we are starting a new history. */ + memcpy(server.replid,server.master->replid,sizeof(server.replid)); + server.master_repl_offset = server.master->reploff; + clearReplicationId2(); + + /* Let's create the replication backlog if needed. Slaves need to + * accumulate the backlog regardless of the fact they have sub-slaves + * or not, in order to behave correctly if they are promoted to + * masters after a failover. */ + if (server.repl_backlog == NULL) createReplicationBacklog(); + serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Finished with success"); + + if (server.supervised_mode == SUPERVISED_SYSTEMD) { + redisCommunicateSystemd("STATUS=MASTER <-> REPLICA sync: Finished with success. Ready to accept connections in read-write mode.\n"); + } + + /* Send the initial ACK immediately to put this replica in online state. */ + if (usemark) replicationSendAck(); + + /* Restart the AOF subsystem now that we finished the sync. This + * will trigger an AOF rewrite, and when done will start appending + * to the new file. */ + if (server.aof_enabled) restartAOFAfterSYNC(); + return; + +error: + cancelReplicationHandshake(1); + return; +} + +char *receiveSynchronousResponse(connection *conn) { + char buf[256]; + /* Read the reply from the server. */ + if (connSyncReadLine(conn,buf,sizeof(buf),server.repl_syncio_timeout*1000) == -1) + { + serverLog(LL_WARNING, "Failed to read response from the server: %s", connGetLastError(conn)); + return NULL; + } + server.repl_transfer_lastio = server.unixtime; + return sdsnew(buf); +} + +/* Send a pre-formatted multi-bulk command to the connection. */ +char* sendCommandRaw(connection *conn, sds cmd) { + if (connSyncWrite(conn,cmd,sdslen(cmd),server.repl_syncio_timeout*1000) == -1) { + return sdscatprintf(sdsempty(),"-Writing to master: %s", + connGetLastError(conn)); + } + return NULL; +} + +/* Compose a multi-bulk command and send it to the connection. + * Used to send AUTH and REPLCONF commands to the master before starting the + * replication. + * + * Takes a list of char* arguments, terminated by a NULL argument. + * + * The command returns an sds string representing the result of the + * operation. On error the first byte is a "-". + */ +char *sendCommand(connection *conn, ...) { + va_list ap; + sds cmd = sdsempty(); + sds cmdargs = sdsempty(); + size_t argslen = 0; + char *arg; + + /* Create the command to send to the master, we use redis binary + * protocol to make sure correct arguments are sent. This function + * is not safe for all binary data. */ + va_start(ap,conn); + while(1) { + arg = va_arg(ap, char*); + if (arg == NULL) break; + cmdargs = sdscatprintf(cmdargs,"$%zu\r\n%s\r\n",strlen(arg),arg); + argslen++; + } + + cmd = sdscatprintf(cmd,"*%zu\r\n",argslen); + cmd = sdscatsds(cmd,cmdargs); + sdsfree(cmdargs); + + va_end(ap); + char* err = sendCommandRaw(conn, cmd); + sdsfree(cmd); + if(err) + return err; + return NULL; +} + +/* Compose a multi-bulk command and send it to the connection. + * Used to send AUTH and REPLCONF commands to the master before starting the + * replication. + * + * argv_lens is optional, when NULL, strlen is used. + * + * The command returns an sds string representing the result of the + * operation. On error the first byte is a "-". + */ +char *sendCommandArgv(connection *conn, int argc, char **argv, size_t *argv_lens) { + sds cmd = sdsempty(); + char *arg; + int i; + + /* Create the command to send to the master. */ + cmd = sdscatfmt(cmd,"*%i\r\n",argc); + for (i=0; i<argc; i++) { + int len; + arg = argv[i]; + len = argv_lens ? argv_lens[i] : strlen(arg); + cmd = sdscatfmt(cmd,"$%i\r\n",len); + cmd = sdscatlen(cmd,arg,len); + cmd = sdscatlen(cmd,"\r\n",2); + } + char* err = sendCommandRaw(conn, cmd); + sdsfree(cmd); + if (err) + return err; + return NULL; +} + +/* Try a partial resynchronization with the master if we are about to reconnect. + * If there is no cached master structure, at least try to issue a + * "PSYNC ? -1" command in order to trigger a full resync using the PSYNC + * command in order to obtain the master replid and the master replication + * global offset. + * + * This function is designed to be called from syncWithMaster(), so the + * following assumptions are made: + * + * 1) We pass the function an already connected socket "fd". + * 2) This function does not close the file descriptor "fd". However in case + * of successful partial resynchronization, the function will reuse + * 'fd' as file descriptor of the server.master client structure. + * + * The function is split in two halves: if read_reply is 0, the function + * writes the PSYNC command on the socket, and a new function call is + * needed, with read_reply set to 1, in order to read the reply of the + * command. This is useful in order to support non blocking operations, so + * that we write, return into the event loop, and read when there are data. + * + * When read_reply is 0 the function returns PSYNC_WRITE_ERR if there + * was a write error, or PSYNC_WAIT_REPLY to signal we need another call + * with read_reply set to 1. However even when read_reply is set to 1 + * the function may return PSYNC_WAIT_REPLY again to signal there were + * insufficient data to read to complete its work. We should re-enter + * into the event loop and wait in such a case. + * + * The function returns: + * + * PSYNC_CONTINUE: If the PSYNC command succeeded and we can continue. + * PSYNC_FULLRESYNC: If PSYNC is supported but a full resync is needed. + * In this case the master replid and global replication + * offset is saved. + * PSYNC_NOT_SUPPORTED: If the server does not understand PSYNC at all and + * the caller should fall back to SYNC. + * PSYNC_WRITE_ERROR: There was an error writing the command to the socket. + * PSYNC_WAIT_REPLY: Call again the function with read_reply set to 1. + * PSYNC_TRY_LATER: Master is currently in a transient error condition. + * + * Notable side effects: + * + * 1) As a side effect of the function call the function removes the readable + * event handler from "fd", unless the return value is PSYNC_WAIT_REPLY. + * 2) server.master_initial_offset is set to the right value according + * to the master reply. This will be used to populate the 'server.master' + * structure replication offset. + */ + +#define PSYNC_WRITE_ERROR 0 +#define PSYNC_WAIT_REPLY 1 +#define PSYNC_CONTINUE 2 +#define PSYNC_FULLRESYNC 3 +#define PSYNC_NOT_SUPPORTED 4 +#define PSYNC_TRY_LATER 5 +int slaveTryPartialResynchronization(connection *conn, int read_reply) { + char *psync_replid; + char psync_offset[32]; + sds reply; + + /* Writing half */ + if (!read_reply) { + /* Initially set master_initial_offset to -1 to mark the current + * master replid and offset as not valid. Later if we'll be able to do + * a FULL resync using the PSYNC command we'll set the offset at the + * right value, so that this information will be propagated to the + * client structure representing the master into server.master. */ + server.master_initial_offset = -1; + + if (server.cached_master) { + psync_replid = server.cached_master->replid; + snprintf(psync_offset,sizeof(psync_offset),"%lld", server.cached_master->reploff+1); + serverLog(LL_NOTICE,"Trying a partial resynchronization (request %s:%s).", psync_replid, psync_offset); + } else { + serverLog(LL_NOTICE,"Partial resynchronization not possible (no cached master)"); + psync_replid = "?"; + memcpy(psync_offset,"-1",3); + } + + /* Issue the PSYNC command, if this is a master with a failover in + * progress then send the failover argument to the replica to cause it + * to become a master */ + if (server.failover_state == FAILOVER_IN_PROGRESS) { + reply = sendCommand(conn,"PSYNC",psync_replid,psync_offset,"FAILOVER",NULL); + } else { + reply = sendCommand(conn,"PSYNC",psync_replid,psync_offset,NULL); + } + + if (reply != NULL) { + serverLog(LL_WARNING,"Unable to send PSYNC to master: %s",reply); + sdsfree(reply); + connSetReadHandler(conn, NULL); + return PSYNC_WRITE_ERROR; + } + return PSYNC_WAIT_REPLY; + } + + /* Reading half */ + reply = receiveSynchronousResponse(conn); + /* Master did not reply to PSYNC */ + if (reply == NULL) { + connSetReadHandler(conn, NULL); + serverLog(LL_WARNING, "Master did not reply to PSYNC, will try later"); + return PSYNC_TRY_LATER; + } + + if (sdslen(reply) == 0) { + /* The master may send empty newlines after it receives PSYNC + * and before to reply, just to keep the connection alive. */ + sdsfree(reply); + return PSYNC_WAIT_REPLY; + } + + connSetReadHandler(conn, NULL); + + if (!strncmp(reply,"+FULLRESYNC",11)) { + char *replid = NULL, *offset = NULL; + + /* FULL RESYNC, parse the reply in order to extract the replid + * and the replication offset. */ + replid = strchr(reply,' '); + if (replid) { + replid++; + offset = strchr(replid,' '); + if (offset) offset++; + } + if (!replid || !offset || (offset-replid-1) != CONFIG_RUN_ID_SIZE) { + serverLog(LL_WARNING, + "Master replied with wrong +FULLRESYNC syntax."); + /* This is an unexpected condition, actually the +FULLRESYNC + * reply means that the master supports PSYNC, but the reply + * format seems wrong. To stay safe we blank the master + * replid to make sure next PSYNCs will fail. */ + memset(server.master_replid,0,CONFIG_RUN_ID_SIZE+1); + } else { + memcpy(server.master_replid, replid, offset-replid-1); + server.master_replid[CONFIG_RUN_ID_SIZE] = '\0'; + server.master_initial_offset = strtoll(offset,NULL,10); + serverLog(LL_NOTICE,"Full resync from master: %s:%lld", + server.master_replid, + server.master_initial_offset); + } + sdsfree(reply); + return PSYNC_FULLRESYNC; + } + + if (!strncmp(reply,"+CONTINUE",9)) { + /* Partial resync was accepted. */ + serverLog(LL_NOTICE, + "Successful partial resynchronization with master."); + + /* Check the new replication ID advertised by the master. If it + * changed, we need to set the new ID as primary ID, and set + * secondary ID as the old master ID up to the current offset, so + * that our sub-slaves will be able to PSYNC with us after a + * disconnection. */ + char *start = reply+10; + char *end = reply+9; + while(end[0] != '\r' && end[0] != '\n' && end[0] != '\0') end++; + if (end-start == CONFIG_RUN_ID_SIZE) { + char new[CONFIG_RUN_ID_SIZE+1]; + memcpy(new,start,CONFIG_RUN_ID_SIZE); + new[CONFIG_RUN_ID_SIZE] = '\0'; + + if (strcmp(new,server.cached_master->replid)) { + /* Master ID changed. */ + serverLog(LL_NOTICE,"Master replication ID changed to %s",new); + + /* Set the old ID as our ID2, up to the current offset+1. */ + memcpy(server.replid2,server.cached_master->replid, + sizeof(server.replid2)); + server.second_replid_offset = server.master_repl_offset+1; + + /* Update the cached master ID and our own primary ID to the + * new one. */ + memcpy(server.replid,new,sizeof(server.replid)); + memcpy(server.cached_master->replid,new,sizeof(server.replid)); + + /* Disconnect all the sub-slaves: they need to be notified. */ + disconnectSlaves(); + } + } + + /* Setup the replication to continue. */ + sdsfree(reply); + replicationResurrectCachedMaster(conn); + + /* If this instance was restarted and we read the metadata to + * PSYNC from the persistence file, our replication backlog could + * be still not initialized. Create it. */ + if (server.repl_backlog == NULL) createReplicationBacklog(); + return PSYNC_CONTINUE; + } + + /* If we reach this point we received either an error (since the master does + * not understand PSYNC or because it is in a special state and cannot + * serve our request), or an unexpected reply from the master. + * + * Return PSYNC_NOT_SUPPORTED on errors we don't understand, otherwise + * return PSYNC_TRY_LATER if we believe this is a transient error. */ + + if (!strncmp(reply,"-NOMASTERLINK",13) || + !strncmp(reply,"-LOADING",8)) + { + serverLog(LL_NOTICE, + "Master is currently unable to PSYNC " + "but should be in the future: %s", reply); + sdsfree(reply); + return PSYNC_TRY_LATER; + } + + if (strncmp(reply,"-ERR",4)) { + /* If it's not an error, log the unexpected event. */ + serverLog(LL_WARNING, + "Unexpected reply to PSYNC from master: %s", reply); + } else { + serverLog(LL_NOTICE, + "Master does not support PSYNC or is in " + "error state (reply: %s)", reply); + } + sdsfree(reply); + return PSYNC_NOT_SUPPORTED; +} + +/* This handler fires when the non blocking connect was able to + * establish a connection with the master. */ +void syncWithMaster(connection *conn) { + char tmpfile[256], *err = NULL; + int dfd = -1, maxtries = 5; + int psync_result; + + /* If this event fired after the user turned the instance into a master + * with SLAVEOF NO ONE we must just return ASAP. */ + if (server.repl_state == REPL_STATE_NONE) { + connClose(conn); + return; + } + + /* Check for errors in the socket: after a non blocking connect() we + * may find that the socket is in error state. */ + if (connGetState(conn) != CONN_STATE_CONNECTED) { + serverLog(LL_WARNING,"Error condition on socket for SYNC: %s", + connGetLastError(conn)); + goto error; + } + + /* Send a PING to check the master is able to reply without errors. */ + if (server.repl_state == REPL_STATE_CONNECTING) { + serverLog(LL_NOTICE,"Non blocking connect for SYNC fired the event."); + /* Delete the writable event so that the readable event remains + * registered and we can wait for the PONG reply. */ + connSetReadHandler(conn, syncWithMaster); + connSetWriteHandler(conn, NULL); + server.repl_state = REPL_STATE_RECEIVE_PING_REPLY; + /* Send the PING, don't check for errors at all, we have the timeout + * that will take care about this. */ + err = sendCommand(conn,"PING",NULL); + if (err) goto write_error; + return; + } + + /* Receive the PONG command. */ + if (server.repl_state == REPL_STATE_RECEIVE_PING_REPLY) { + err = receiveSynchronousResponse(conn); + + /* The master did not reply */ + if (err == NULL) goto no_response_error; + + /* We accept only two replies as valid, a positive +PONG reply + * (we just check for "+") or an authentication error. + * Note that older versions of Redis replied with "operation not + * permitted" instead of using a proper error code, so we test + * both. */ + if (err[0] != '+' && + strncmp(err,"-NOAUTH",7) != 0 && + strncmp(err,"-NOPERM",7) != 0 && + strncmp(err,"-ERR operation not permitted",28) != 0) + { + serverLog(LL_WARNING,"Error reply to PING from master: '%s'",err); + sdsfree(err); + goto error; + } else { + serverLog(LL_NOTICE, + "Master replied to PING, replication can continue..."); + } + sdsfree(err); + err = NULL; + server.repl_state = REPL_STATE_SEND_HANDSHAKE; + } + + if (server.repl_state == REPL_STATE_SEND_HANDSHAKE) { + /* AUTH with the master if required. */ + if (server.masterauth) { + char *args[3] = {"AUTH",NULL,NULL}; + size_t lens[3] = {4,0,0}; + int argc = 1; + if (server.masteruser) { + args[argc] = server.masteruser; + lens[argc] = strlen(server.masteruser); + argc++; + } + args[argc] = server.masterauth; + lens[argc] = sdslen(server.masterauth); + argc++; + err = sendCommandArgv(conn, argc, args, lens); + if (err) goto write_error; + } + + /* Set the slave port, so that Master's INFO command can list the + * slave listening port correctly. */ + { + int port; + if (server.slave_announce_port) + port = server.slave_announce_port; + else if (server.tls_replication && server.tls_port) + port = server.tls_port; + else + port = server.port; + sds portstr = sdsfromlonglong(port); + err = sendCommand(conn,"REPLCONF", + "listening-port",portstr, NULL); + sdsfree(portstr); + if (err) goto write_error; + } + + /* Set the slave ip, so that Master's INFO command can list the + * slave IP address port correctly in case of port forwarding or NAT. + * Skip REPLCONF ip-address if there is no slave-announce-ip option set. */ + if (server.slave_announce_ip) { + err = sendCommand(conn,"REPLCONF", + "ip-address",server.slave_announce_ip, NULL); + if (err) goto write_error; + } + + /* Inform the master of our (slave) capabilities. + * + * EOF: supports EOF-style RDB transfer for diskless replication. + * PSYNC2: supports PSYNC v2, so understands +CONTINUE <new repl ID>. + * + * The master will ignore capabilities it does not understand. */ + err = sendCommand(conn,"REPLCONF", + "capa","eof","capa","psync2",NULL); + if (err) goto write_error; + + server.repl_state = REPL_STATE_RECEIVE_AUTH_REPLY; + return; + } + + if (server.repl_state == REPL_STATE_RECEIVE_AUTH_REPLY && !server.masterauth) + server.repl_state = REPL_STATE_RECEIVE_PORT_REPLY; + + /* Receive AUTH reply. */ + if (server.repl_state == REPL_STATE_RECEIVE_AUTH_REPLY) { + err = receiveSynchronousResponse(conn); + if (err == NULL) goto no_response_error; + if (err[0] == '-') { + serverLog(LL_WARNING,"Unable to AUTH to MASTER: %s",err); + sdsfree(err); + goto error; + } + sdsfree(err); + err = NULL; + server.repl_state = REPL_STATE_RECEIVE_PORT_REPLY; + return; + } + + /* Receive REPLCONF listening-port reply. */ + if (server.repl_state == REPL_STATE_RECEIVE_PORT_REPLY) { + err = receiveSynchronousResponse(conn); + if (err == NULL) goto no_response_error; + /* Ignore the error if any, not all the Redis versions support + * REPLCONF listening-port. */ + if (err[0] == '-') { + serverLog(LL_NOTICE,"(Non critical) Master does not understand " + "REPLCONF listening-port: %s", err); + } + sdsfree(err); + server.repl_state = REPL_STATE_RECEIVE_IP_REPLY; + return; + } + + if (server.repl_state == REPL_STATE_RECEIVE_IP_REPLY && !server.slave_announce_ip) + server.repl_state = REPL_STATE_RECEIVE_CAPA_REPLY; + + /* Receive REPLCONF ip-address reply. */ + if (server.repl_state == REPL_STATE_RECEIVE_IP_REPLY) { + err = receiveSynchronousResponse(conn); + if (err == NULL) goto no_response_error; + /* Ignore the error if any, not all the Redis versions support + * REPLCONF ip-address. */ + if (err[0] == '-') { + serverLog(LL_NOTICE,"(Non critical) Master does not understand " + "REPLCONF ip-address: %s", err); + } + sdsfree(err); + server.repl_state = REPL_STATE_RECEIVE_CAPA_REPLY; + return; + } + + /* Receive CAPA reply. */ + if (server.repl_state == REPL_STATE_RECEIVE_CAPA_REPLY) { + err = receiveSynchronousResponse(conn); + if (err == NULL) goto no_response_error; + /* Ignore the error if any, not all the Redis versions support + * REPLCONF capa. */ + if (err[0] == '-') { + serverLog(LL_NOTICE,"(Non critical) Master does not understand " + "REPLCONF capa: %s", err); + } + sdsfree(err); + err = NULL; + server.repl_state = REPL_STATE_SEND_PSYNC; + } + + /* Try a partial resynchronization. If we don't have a cached master + * slaveTryPartialResynchronization() will at least try to use PSYNC + * to start a full resynchronization so that we get the master replid + * and the global offset, to try a partial resync at the next + * reconnection attempt. */ + if (server.repl_state == REPL_STATE_SEND_PSYNC) { + if (slaveTryPartialResynchronization(conn,0) == PSYNC_WRITE_ERROR) { + err = sdsnew("Write error sending the PSYNC command."); + abortFailover("Write error to failover target"); + goto write_error; + } + server.repl_state = REPL_STATE_RECEIVE_PSYNC_REPLY; + return; + } + + /* If reached this point, we should be in REPL_STATE_RECEIVE_PSYNC_REPLY. */ + if (server.repl_state != REPL_STATE_RECEIVE_PSYNC_REPLY) { + serverLog(LL_WARNING,"syncWithMaster(): state machine error, " + "state should be RECEIVE_PSYNC but is %d", + server.repl_state); + goto error; + } + + psync_result = slaveTryPartialResynchronization(conn,1); + if (psync_result == PSYNC_WAIT_REPLY) return; /* Try again later... */ + + /* Check the status of the planned failover. We expect PSYNC_CONTINUE, + * but there is nothing technically wrong with a full resync which + * could happen in edge cases. */ + if (server.failover_state == FAILOVER_IN_PROGRESS) { + if (psync_result == PSYNC_CONTINUE || psync_result == PSYNC_FULLRESYNC) { + clearFailoverState(); + } else { + abortFailover("Failover target rejected psync request"); + return; + } + } + + /* If the master is in an transient error, we should try to PSYNC + * from scratch later, so go to the error path. This happens when + * the server is loading the dataset or is not connected with its + * master and so forth. */ + if (psync_result == PSYNC_TRY_LATER) goto error; + + /* Note: if PSYNC does not return WAIT_REPLY, it will take care of + * uninstalling the read handler from the file descriptor. */ + + if (psync_result == PSYNC_CONTINUE) { + serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Master accepted a Partial Resynchronization."); + if (server.supervised_mode == SUPERVISED_SYSTEMD) { + redisCommunicateSystemd("STATUS=MASTER <-> REPLICA sync: Partial Resynchronization accepted. Ready to accept connections in read-write mode.\n"); + } + return; + } + + /* Fall back to SYNC if needed. Otherwise psync_result == PSYNC_FULLRESYNC + * and the server.master_replid and master_initial_offset are + * already populated. */ + if (psync_result == PSYNC_NOT_SUPPORTED) { + serverLog(LL_NOTICE,"Retrying with SYNC..."); + if (connSyncWrite(conn,"SYNC\r\n",6,server.repl_syncio_timeout*1000) == -1) { + serverLog(LL_WARNING,"I/O error writing to MASTER: %s", + connGetLastError(conn)); + goto error; + } + } + + /* Prepare a suitable temp file for bulk transfer */ + if (!useDisklessLoad()) { + while(maxtries--) { + snprintf(tmpfile,256, + "temp-%d.%ld.rdb",(int)server.unixtime,(long int)getpid()); + dfd = open(tmpfile,O_CREAT|O_WRONLY|O_EXCL,0644); + if (dfd != -1) break; + sleep(1); + } + if (dfd == -1) { + serverLog(LL_WARNING,"Opening the temp file needed for MASTER <-> REPLICA synchronization: %s",strerror(errno)); + goto error; + } + server.repl_transfer_tmpfile = zstrdup(tmpfile); + server.repl_transfer_fd = dfd; + } + + /* Setup the non blocking download of the bulk file. */ + if (connSetReadHandler(conn, readSyncBulkPayload) + == C_ERR) + { + char conninfo[CONN_INFO_LEN]; + serverLog(LL_WARNING, + "Can't create readable event for SYNC: %s (%s)", + strerror(errno), connGetInfo(conn, conninfo, sizeof(conninfo))); + goto error; + } + + server.repl_state = REPL_STATE_TRANSFER; + server.repl_transfer_size = -1; + server.repl_transfer_read = 0; + server.repl_transfer_last_fsync_off = 0; + server.repl_transfer_lastio = server.unixtime; + return; + +no_response_error: /* Handle receiveSynchronousResponse() error when master has no reply */ + serverLog(LL_WARNING, "Master did not respond to command during SYNC handshake"); + /* Fall through to regular error handling */ + +error: + if (dfd != -1) close(dfd); + connClose(conn); + server.repl_transfer_s = NULL; + if (server.repl_transfer_fd != -1) + close(server.repl_transfer_fd); + if (server.repl_transfer_tmpfile) + zfree(server.repl_transfer_tmpfile); + server.repl_transfer_tmpfile = NULL; + server.repl_transfer_fd = -1; + server.repl_state = REPL_STATE_CONNECT; + return; + +write_error: /* Handle sendCommand() errors. */ + serverLog(LL_WARNING,"Sending command to master in replication handshake: %s", err); + sdsfree(err); + goto error; +} + +int connectWithMaster(void) { + server.repl_transfer_s = connCreate(connTypeOfReplication()); + if (connConnect(server.repl_transfer_s, server.masterhost, server.masterport, + server.bind_source_addr, syncWithMaster) == C_ERR) { + serverLog(LL_WARNING,"Unable to connect to MASTER: %s", + connGetLastError(server.repl_transfer_s)); + connClose(server.repl_transfer_s); + server.repl_transfer_s = NULL; + return C_ERR; + } + + + server.repl_transfer_lastio = server.unixtime; + server.repl_state = REPL_STATE_CONNECTING; + serverLog(LL_NOTICE,"MASTER <-> REPLICA sync started"); + return C_OK; +} + +/* This function can be called when a non blocking connection is currently + * in progress to undo it. + * Never call this function directly, use cancelReplicationHandshake() instead. + */ +void undoConnectWithMaster(void) { + connClose(server.repl_transfer_s); + server.repl_transfer_s = NULL; +} + +/* Abort the async download of the bulk dataset while SYNC-ing with master. + * Never call this function directly, use cancelReplicationHandshake() instead. + */ +void replicationAbortSyncTransfer(void) { + serverAssert(server.repl_state == REPL_STATE_TRANSFER); + undoConnectWithMaster(); + if (server.repl_transfer_fd!=-1) { + close(server.repl_transfer_fd); + bg_unlink(server.repl_transfer_tmpfile); + zfree(server.repl_transfer_tmpfile); + server.repl_transfer_tmpfile = NULL; + server.repl_transfer_fd = -1; + } +} + +/* This function aborts a non blocking replication attempt if there is one + * in progress, by canceling the non-blocking connect attempt or + * the initial bulk transfer. + * + * If there was a replication handshake in progress 1 is returned and + * the replication state (server.repl_state) set to REPL_STATE_CONNECT. + * + * Otherwise zero is returned and no operation is performed at all. */ +int cancelReplicationHandshake(int reconnect) { + if (server.repl_state == REPL_STATE_TRANSFER) { + replicationAbortSyncTransfer(); + server.repl_state = REPL_STATE_CONNECT; + } else if (server.repl_state == REPL_STATE_CONNECTING || + slaveIsInHandshakeState()) + { + undoConnectWithMaster(); + server.repl_state = REPL_STATE_CONNECT; + } else { + return 0; + } + + if (!reconnect) + return 1; + + /* try to re-connect without waiting for replicationCron, this is needed + * for the "diskless loading short read" test. */ + serverLog(LL_NOTICE,"Reconnecting to MASTER %s:%d after failure", + server.masterhost, server.masterport); + connectWithMaster(); + + return 1; +} + +/* Set replication to the specified master address and port. */ +void replicationSetMaster(char *ip, int port) { + int was_master = server.masterhost == NULL; + + sdsfree(server.masterhost); + server.masterhost = NULL; + if (server.master) { + freeClient(server.master); + } + disconnectAllBlockedClients(); /* Clients blocked in master, now slave. */ + + /* Setting masterhost only after the call to freeClient since it calls + * replicationHandleMasterDisconnection which can trigger a re-connect + * directly from within that call. */ + server.masterhost = sdsnew(ip); + server.masterport = port; + + /* Update oom_score_adj */ + setOOMScoreAdj(-1); + + /* Here we don't disconnect with replicas, since they may hopefully be able + * to partially resync with us. We will disconnect with replicas and force + * them to resync with us when changing replid on partially resync with new + * master, or finishing transferring RDB and preparing loading DB on full + * sync with new master. */ + + cancelReplicationHandshake(0); + /* Before destroying our master state, create a cached master using + * our own parameters, to later PSYNC with the new master. */ + if (was_master) { + replicationDiscardCachedMaster(); + replicationCacheMasterUsingMyself(); + } + + /* Fire the role change modules event. */ + moduleFireServerEvent(REDISMODULE_EVENT_REPLICATION_ROLE_CHANGED, + REDISMODULE_EVENT_REPLROLECHANGED_NOW_REPLICA, + NULL); + + /* Fire the master link modules event. */ + if (server.repl_state == REPL_STATE_CONNECTED) + moduleFireServerEvent(REDISMODULE_EVENT_MASTER_LINK_CHANGE, + REDISMODULE_SUBEVENT_MASTER_LINK_DOWN, + NULL); + + server.repl_state = REPL_STATE_CONNECT; + serverLog(LL_NOTICE,"Connecting to MASTER %s:%d", + server.masterhost, server.masterport); + connectWithMaster(); +} + +/* Cancel replication, setting the instance as a master itself. */ +void replicationUnsetMaster(void) { + if (server.masterhost == NULL) return; /* Nothing to do. */ + + /* Fire the master link modules event. */ + if (server.repl_state == REPL_STATE_CONNECTED) + moduleFireServerEvent(REDISMODULE_EVENT_MASTER_LINK_CHANGE, + REDISMODULE_SUBEVENT_MASTER_LINK_DOWN, + NULL); + + /* Clear masterhost first, since the freeClient calls + * replicationHandleMasterDisconnection which can attempt to re-connect. */ + sdsfree(server.masterhost); + server.masterhost = NULL; + if (server.master) freeClient(server.master); + replicationDiscardCachedMaster(); + cancelReplicationHandshake(0); + /* When a slave is turned into a master, the current replication ID + * (that was inherited from the master at synchronization time) is + * used as secondary ID up to the current offset, and a new replication + * ID is created to continue with a new replication history. */ + shiftReplicationId(); + /* Disconnecting all the slaves is required: we need to inform slaves + * of the replication ID change (see shiftReplicationId() call). However + * the slaves will be able to partially resync with us, so it will be + * a very fast reconnection. */ + disconnectSlaves(); + server.repl_state = REPL_STATE_NONE; + + /* We need to make sure the new master will start the replication stream + * with a SELECT statement. This is forced after a full resync, but + * with PSYNC version 2, there is no need for full resync after a + * master switch. */ + server.slaveseldb = -1; + + /* Update oom_score_adj */ + setOOMScoreAdj(-1); + + /* Once we turn from slave to master, we consider the starting time without + * slaves (that is used to count the replication backlog time to live) as + * starting from now. Otherwise the backlog will be freed after a + * failover if slaves do not connect immediately. */ + server.repl_no_slaves_since = server.unixtime; + + /* Reset down time so it'll be ready for when we turn into replica again. */ + server.repl_down_since = 0; + + /* Fire the role change modules event. */ + moduleFireServerEvent(REDISMODULE_EVENT_REPLICATION_ROLE_CHANGED, + REDISMODULE_EVENT_REPLROLECHANGED_NOW_MASTER, + NULL); + + /* Restart the AOF subsystem in case we shut it down during a sync when + * we were still a slave. */ + if (server.aof_enabled && server.aof_state == AOF_OFF) restartAOFAfterSYNC(); +} + +/* This function is called when the slave lose the connection with the + * master into an unexpected way. */ +void replicationHandleMasterDisconnection(void) { + /* Fire the master link modules event. */ + if (server.repl_state == REPL_STATE_CONNECTED) + moduleFireServerEvent(REDISMODULE_EVENT_MASTER_LINK_CHANGE, + REDISMODULE_SUBEVENT_MASTER_LINK_DOWN, + NULL); + + server.master = NULL; + server.repl_state = REPL_STATE_CONNECT; + server.repl_down_since = server.unixtime; + /* We lost connection with our master, don't disconnect slaves yet, + * maybe we'll be able to PSYNC with our master later. We'll disconnect + * the slaves only if we'll have to do a full resync with our master. */ + + /* Try to re-connect immediately rather than wait for replicationCron + * waiting 1 second may risk backlog being recycled. */ + if (server.masterhost) { + serverLog(LL_NOTICE,"Reconnecting to MASTER %s:%d", + server.masterhost, server.masterport); + connectWithMaster(); + } +} + +void replicaofCommand(client *c) { + /* SLAVEOF is not allowed in cluster mode as replication is automatically + * configured using the current address of the master node. */ + if (server.cluster_enabled) { + addReplyError(c,"REPLICAOF not allowed in cluster mode."); + return; + } + + if (server.failover_state != NO_FAILOVER) { + addReplyError(c,"REPLICAOF not allowed while failing over."); + return; + } + + /* The special host/port combination "NO" "ONE" turns the instance + * into a master. Otherwise the new master address is set. */ + if (!strcasecmp(c->argv[1]->ptr,"no") && + !strcasecmp(c->argv[2]->ptr,"one")) { + if (server.masterhost) { + replicationUnsetMaster(); + sds client = catClientInfoString(sdsempty(),c); + serverLog(LL_NOTICE,"MASTER MODE enabled (user request from '%s')", + client); + sdsfree(client); + } + } else { + long port; + + if (c->flags & CLIENT_SLAVE) + { + /* If a client is already a replica they cannot run this command, + * because it involves flushing all replicas (including this + * client) */ + addReplyError(c, "Command is not valid when client is a replica."); + return; + } + + if (getRangeLongFromObjectOrReply(c, c->argv[2], 0, 65535, &port, + "Invalid master port") != C_OK) + return; + + /* Check if we are already attached to the specified master */ + if (server.masterhost && !strcasecmp(server.masterhost,c->argv[1]->ptr) + && server.masterport == port) { + serverLog(LL_NOTICE,"REPLICAOF would result into synchronization " + "with the master we are already connected " + "with. No operation performed."); + addReplySds(c,sdsnew("+OK Already connected to specified " + "master\r\n")); + return; + } + /* There was no previous master or the user specified a different one, + * we can continue. */ + replicationSetMaster(c->argv[1]->ptr, port); + sds client = catClientInfoString(sdsempty(),c); + serverLog(LL_NOTICE,"REPLICAOF %s:%d enabled (user request from '%s')", + server.masterhost, server.masterport, client); + sdsfree(client); + } + addReply(c,shared.ok); +} + +/* ROLE command: provide information about the role of the instance + * (master or slave) and additional information related to replication + * in an easy to process format. */ +void roleCommand(client *c) { + if (server.sentinel_mode) { + sentinelRoleCommand(c); + return; + } + + if (server.masterhost == NULL) { + listIter li; + listNode *ln; + void *mbcount; + int slaves = 0; + + addReplyArrayLen(c,3); + addReplyBulkCBuffer(c,"master",6); + addReplyLongLong(c,server.master_repl_offset); + mbcount = addReplyDeferredLen(c); + listRewind(server.slaves,&li); + while((ln = listNext(&li))) { + client *slave = ln->value; + char ip[NET_IP_STR_LEN], *slaveaddr = slave->slave_addr; + + if (!slaveaddr) { + if (connAddrPeerName(slave->conn,ip,sizeof(ip),NULL) == -1) + continue; + slaveaddr = ip; + } + if (slave->replstate != SLAVE_STATE_ONLINE) continue; + addReplyArrayLen(c,3); + addReplyBulkCString(c,slaveaddr); + addReplyBulkLongLong(c,slave->slave_listening_port); + addReplyBulkLongLong(c,slave->repl_ack_off); + slaves++; + } + setDeferredArrayLen(c,mbcount,slaves); + } else { + char *slavestate = NULL; + + addReplyArrayLen(c,5); + addReplyBulkCBuffer(c,"slave",5); + addReplyBulkCString(c,server.masterhost); + addReplyLongLong(c,server.masterport); + if (slaveIsInHandshakeState()) { + slavestate = "handshake"; + } else { + switch(server.repl_state) { + case REPL_STATE_NONE: slavestate = "none"; break; + case REPL_STATE_CONNECT: slavestate = "connect"; break; + case REPL_STATE_CONNECTING: slavestate = "connecting"; break; + case REPL_STATE_TRANSFER: slavestate = "sync"; break; + case REPL_STATE_CONNECTED: slavestate = "connected"; break; + default: slavestate = "unknown"; break; + } + } + addReplyBulkCString(c,slavestate); + addReplyLongLong(c,server.master ? server.master->reploff : -1); + } +} + +/* Send a REPLCONF ACK command to the master to inform it about the current + * processed offset. If we are not connected with a master, the command has + * no effects. */ +void replicationSendAck(void) { + client *c = server.master; + + if (c != NULL) { + int send_fack = server.fsynced_reploff != -1; + c->flags |= CLIENT_MASTER_FORCE_REPLY; + addReplyArrayLen(c,send_fack ? 5 : 3); + addReplyBulkCString(c,"REPLCONF"); + addReplyBulkCString(c,"ACK"); + addReplyBulkLongLong(c,c->reploff); + if (send_fack) { + addReplyBulkCString(c,"FACK"); + addReplyBulkLongLong(c,server.fsynced_reploff); + } + c->flags &= ~CLIENT_MASTER_FORCE_REPLY; + } +} + +/* ---------------------- MASTER CACHING FOR PSYNC -------------------------- */ + +/* In order to implement partial synchronization we need to be able to cache + * our master's client structure after a transient disconnection. + * It is cached into server.cached_master and flushed away using the following + * functions. */ + +/* This function is called by freeClient() in order to cache the master + * client structure instead of destroying it. freeClient() will return + * ASAP after this function returns, so every action needed to avoid problems + * with a client that is really "suspended" has to be done by this function. + * + * The other functions that will deal with the cached master are: + * + * replicationDiscardCachedMaster() that will make sure to kill the client + * as for some reason we don't want to use it in the future. + * + * replicationResurrectCachedMaster() that is used after a successful PSYNC + * handshake in order to reactivate the cached master. + */ +void replicationCacheMaster(client *c) { + serverAssert(server.master != NULL && server.cached_master == NULL); + serverLog(LL_NOTICE,"Caching the disconnected master state."); + + /* Unlink the client from the server structures. */ + unlinkClient(c); + + /* Reset the master client so that's ready to accept new commands: + * we want to discard the non processed query buffers and non processed + * offsets, including pending transactions, already populated arguments, + * pending outputs to the master. */ + sdsclear(server.master->querybuf); + server.master->qb_pos = 0; + server.master->repl_applied = 0; + server.master->read_reploff = server.master->reploff; + if (c->flags & CLIENT_MULTI) discardTransaction(c); + listEmpty(c->reply); + c->sentlen = 0; + c->reply_bytes = 0; + c->bufpos = 0; + resetClient(c); + + /* Save the master. Server.master will be set to null later by + * replicationHandleMasterDisconnection(). */ + server.cached_master = server.master; + + /* Invalidate the Peer ID cache. */ + if (c->peerid) { + sdsfree(c->peerid); + c->peerid = NULL; + } + /* Invalidate the Sock Name cache. */ + if (c->sockname) { + sdsfree(c->sockname); + c->sockname = NULL; + } + + /* Caching the master happens instead of the actual freeClient() call, + * so make sure to adjust the replication state. This function will + * also set server.master to NULL. */ + replicationHandleMasterDisconnection(); +} + +/* This function is called when a master is turned into a slave, in order to + * create from scratch a cached master for the new client, that will allow + * to PSYNC with the slave that was promoted as the new master after a + * failover. + * + * Assuming this instance was previously the master instance of the new master, + * the new master will accept its replication ID, and potential also the + * current offset if no data was lost during the failover. So we use our + * current replication ID and offset in order to synthesize a cached master. */ +void replicationCacheMasterUsingMyself(void) { + serverLog(LL_NOTICE, + "Before turning into a replica, using my own master parameters " + "to synthesize a cached master: I may be able to synchronize with " + "the new master with just a partial transfer."); + + /* This will be used to populate the field server.master->reploff + * by replicationCreateMasterClient(). We'll later set the created + * master as server.cached_master, so the replica will use such + * offset for PSYNC. */ + server.master_initial_offset = server.master_repl_offset; + + /* The master client we create can be set to any DBID, because + * the new master will start its replication stream with SELECT. */ + replicationCreateMasterClient(NULL,-1); + + /* Use our own ID / offset. */ + memcpy(server.master->replid, server.replid, sizeof(server.replid)); + + /* Set as cached master. */ + unlinkClient(server.master); + server.cached_master = server.master; + server.master = NULL; +} + +/* Free a cached master, called when there are no longer the conditions for + * a partial resync on reconnection. */ +void replicationDiscardCachedMaster(void) { + if (server.cached_master == NULL) return; + + serverLog(LL_NOTICE,"Discarding previously cached master state."); + server.cached_master->flags &= ~CLIENT_MASTER; + freeClient(server.cached_master); + server.cached_master = NULL; +} + +/* Turn the cached master into the current master, using the file descriptor + * passed as argument as the socket for the new master. + * + * This function is called when successfully setup a partial resynchronization + * so the stream of data that we'll receive will start from where this + * master left. */ +void replicationResurrectCachedMaster(connection *conn) { + server.master = server.cached_master; + server.cached_master = NULL; + server.master->conn = conn; + connSetPrivateData(server.master->conn, server.master); + server.master->flags &= ~(CLIENT_CLOSE_AFTER_REPLY|CLIENT_CLOSE_ASAP); + server.master->authenticated = 1; + server.master->lastinteraction = server.unixtime; + server.repl_state = REPL_STATE_CONNECTED; + server.repl_down_since = 0; + + /* Fire the master link modules event. */ + moduleFireServerEvent(REDISMODULE_EVENT_MASTER_LINK_CHANGE, + REDISMODULE_SUBEVENT_MASTER_LINK_UP, + NULL); + + /* Re-add to the list of clients. */ + linkClient(server.master); + if (connSetReadHandler(server.master->conn, readQueryFromClient)) { + serverLog(LL_WARNING,"Error resurrecting the cached master, impossible to add the readable handler: %s", strerror(errno)); + freeClientAsync(server.master); /* Close ASAP. */ + } + + /* We may also need to install the write handler as well if there is + * pending data in the write buffers. */ + if (clientHasPendingReplies(server.master)) { + if (connSetWriteHandler(server.master->conn, sendReplyToClient)) { + serverLog(LL_WARNING,"Error resurrecting the cached master, impossible to add the writable handler: %s", strerror(errno)); + freeClientAsync(server.master); /* Close ASAP. */ + } + } +} + +/* ------------------------- MIN-SLAVES-TO-WRITE --------------------------- */ + +/* This function counts the number of slaves with lag <= min-slaves-max-lag. + * If the option is active, the server will prevent writes if there are not + * enough connected slaves with the specified lag (or less). */ +void refreshGoodSlavesCount(void) { + listIter li; + listNode *ln; + int good = 0; + + if (!server.repl_min_slaves_to_write || + !server.repl_min_slaves_max_lag) return; + + listRewind(server.slaves,&li); + while((ln = listNext(&li))) { + client *slave = ln->value; + time_t lag = server.unixtime - slave->repl_ack_time; + + if (slave->replstate == SLAVE_STATE_ONLINE && + lag <= server.repl_min_slaves_max_lag) good++; + } + server.repl_good_slaves_count = good; +} + +/* return true if status of good replicas is OK. otherwise false */ +int checkGoodReplicasStatus(void) { + return server.masterhost || /* not a primary status should be OK */ + !server.repl_min_slaves_max_lag || /* Min slave max lag not configured */ + !server.repl_min_slaves_to_write || /* Min slave to write not configured */ + server.repl_good_slaves_count >= server.repl_min_slaves_to_write; /* check if we have enough slaves */ +} + +/* ----------------------- SYNCHRONOUS REPLICATION -------------------------- + * Redis synchronous replication design can be summarized in points: + * + * - Redis masters have a global replication offset, used by PSYNC. + * - Master increment the offset every time new commands are sent to slaves. + * - Slaves ping back masters with the offset processed so far. + * + * So synchronous replication adds a new WAIT command in the form: + * + * WAIT <num_replicas> <milliseconds_timeout> + * + * That returns the number of replicas that processed the query when + * we finally have at least num_replicas, or when the timeout was + * reached. + * + * The command is implemented in this way: + * + * - Every time a client processes a command, we remember the replication + * offset after sending that command to the slaves. + * - When WAIT is called, we ask slaves to send an acknowledgement ASAP. + * The client is blocked at the same time (see blocked.c). + * - Once we receive enough ACKs for a given offset or when the timeout + * is reached, the WAIT command is unblocked and the reply sent to the + * client. + */ + +/* This just set a flag so that we broadcast a REPLCONF GETACK command + * to all the slaves in the beforeSleep() function. Note that this way + * we "group" all the clients that want to wait for synchronous replication + * in a given event loop iteration, and send a single GETACK for them all. */ +void replicationRequestAckFromSlaves(void) { + server.get_ack_from_slaves = 1; +} + +/* Return the number of slaves that already acknowledged the specified + * replication offset. */ +int replicationCountAcksByOffset(long long offset) { + listIter li; + listNode *ln; + int count = 0; + + listRewind(server.slaves,&li); + while((ln = listNext(&li))) { + client *slave = ln->value; + + if (slave->replstate != SLAVE_STATE_ONLINE) continue; + if (slave->repl_ack_off >= offset) count++; + } + return count; +} + +/* Return the number of replicas that already acknowledged the specified + * replication offset being AOF fsynced. */ +int replicationCountAOFAcksByOffset(long long offset) { + listIter li; + listNode *ln; + int count = 0; + + listRewind(server.slaves,&li); + while((ln = listNext(&li))) { + client *slave = ln->value; + + if (slave->replstate != SLAVE_STATE_ONLINE) continue; + if (slave->repl_aof_off >= offset) count++; + } + return count; +} + +/* WAIT for N replicas to acknowledge the processing of our latest + * write command (and all the previous commands). */ +void waitCommand(client *c) { + mstime_t timeout; + long numreplicas, ackreplicas; + long long offset = c->woff; + + if (server.masterhost) { + 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."); + return; + } + + /* Argument parsing. */ + if (getLongFromObjectOrReply(c,c->argv[1],&numreplicas,NULL) != C_OK) + return; + if (getTimeoutFromObjectOrReply(c,c->argv[2],&timeout,UNIT_MILLISECONDS) + != C_OK) return; + + /* First try without blocking at all. */ + ackreplicas = replicationCountAcksByOffset(c->woff); + if (ackreplicas >= numreplicas || c->flags & CLIENT_DENY_BLOCKING) { + addReplyLongLong(c,ackreplicas); + return; + } + + /* Otherwise block the client and put it into our list of clients + * waiting for ack from slaves. */ + blockForReplication(c,timeout,offset,numreplicas); + + /* Make sure that the server will send an ACK request to all the slaves + * before returning to the event loop. */ + replicationRequestAckFromSlaves(); +} + +/* WAIT for N replicas and / or local master to acknowledge our latest + * write command got synced to the disk. */ +void waitaofCommand(client *c) { + mstime_t timeout; + long numreplicas, numlocal, ackreplicas, acklocal; + + /* Argument parsing. */ + if (getRangeLongFromObjectOrReply(c,c->argv[1],0,1,&numlocal,NULL) != C_OK) + return; + if (getPositiveLongFromObjectOrReply(c,c->argv[2],&numreplicas,NULL) != C_OK) + return; + if (getTimeoutFromObjectOrReply(c,c->argv[3],&timeout,UNIT_MILLISECONDS) != C_OK) + return; + + if (server.masterhost) { + addReplyError(c,"WAITAOF cannot be used with replica instances. Please also note that writes to replicas are just local and are not propagated."); + return; + } + if (numlocal && !server.aof_enabled) { + addReplyError(c, "WAITAOF cannot be used when numlocal is set but appendonly is disabled."); + return; + } + + /* First try without blocking at all. */ + ackreplicas = replicationCountAOFAcksByOffset(c->woff); + acklocal = server.fsynced_reploff >= c->woff; + if ((ackreplicas >= numreplicas && acklocal >= numlocal) || c->flags & CLIENT_DENY_BLOCKING) { + addReplyArrayLen(c,2); + addReplyLongLong(c,acklocal); + addReplyLongLong(c,ackreplicas); + return; + } + + /* Otherwise block the client and put it into our list of clients + * waiting for ack from slaves. */ + blockForAofFsync(c,timeout,c->woff,numlocal,numreplicas); + + /* Make sure that the server will send an ACK request to all the slaves + * before returning to the event loop. */ + replicationRequestAckFromSlaves(); +} + +/* This is called by unblockClient() to perform the blocking op type + * specific cleanup. We just remove the client from the list of clients + * waiting for replica acks. Never call it directly, call unblockClient() + * instead. */ +void unblockClientWaitingReplicas(client *c) { + listNode *ln = listSearchKey(server.clients_waiting_acks,c); + serverAssert(ln != NULL); + listDelNode(server.clients_waiting_acks,ln); + updateStatsOnUnblock(c, 0, 0, 0); +} + +/* Check if there are clients blocked in WAIT or WAITAOF that can be unblocked + * since we received enough ACKs from slaves. */ +void processClientsWaitingReplicas(void) { + long long last_offset = 0; + long long last_aof_offset = 0; + int last_numreplicas = 0; + int last_aof_numreplicas = 0; + + listIter li; + listNode *ln; + + listRewind(server.clients_waiting_acks,&li); + while((ln = listNext(&li))) { + int numlocal = 0; + int numreplicas = 0; + + client *c = ln->value; + int is_wait_aof = c->bstate.btype == BLOCKED_WAITAOF; + + if (is_wait_aof && c->bstate.numlocal && !server.aof_enabled) { + addReplyError(c, "WAITAOF cannot be used when numlocal is set but appendonly is disabled."); + unblockClient(c, 1); + continue; + } + + /* Every time we find a client that is satisfied for a given + * offset and number of replicas, we remember it so the next client + * may be unblocked without calling replicationCountAcksByOffset() + * or calling replicationCountAOFAcksByOffset() + * if the requested offset / replicas were equal or less. */ + if (!is_wait_aof && last_offset && last_offset >= c->bstate.reploffset && + last_numreplicas >= c->bstate.numreplicas) + { + numreplicas = last_numreplicas; + } else if (is_wait_aof && last_aof_offset && last_aof_offset >= c->bstate.reploffset && + last_aof_numreplicas >= c->bstate.numreplicas) + { + numreplicas = last_aof_numreplicas; + } else { + numreplicas = is_wait_aof ? + replicationCountAOFAcksByOffset(c->bstate.reploffset) : + replicationCountAcksByOffset(c->bstate.reploffset); + + /* Check if the number of replicas is satisfied. */ + if (numreplicas < c->bstate.numreplicas) continue; + + if (is_wait_aof) { + last_aof_offset = c->bstate.reploffset; + last_aof_numreplicas = numreplicas; + } else { + last_offset = c->bstate.reploffset; + last_numreplicas = numreplicas; + } + } + + /* Check if the local constraint of WAITAOF is served */ + if (is_wait_aof) { + numlocal = server.fsynced_reploff >= c->bstate.reploffset; + if (numlocal < c->bstate.numlocal) continue; + } + + /* Reply before unblocking, because unblock client calls reqresAppendResponse */ + if (is_wait_aof) { + /* WAITAOF has an array reply */ + addReplyArrayLen(c, 2); + addReplyLongLong(c, numlocal); + addReplyLongLong(c, numreplicas); + } else { + addReplyLongLong(c, numreplicas); + } + + unblockClient(c, 1); + } +} + +/* Return the slave replication offset for this instance, that is + * the offset for which we already processed the master replication stream. */ +long long replicationGetSlaveOffset(void) { + long long offset = 0; + + if (server.masterhost != NULL) { + if (server.master) { + offset = server.master->reploff; + } else if (server.cached_master) { + offset = server.cached_master->reploff; + } + } + /* offset may be -1 when the master does not support it at all, however + * this function is designed to return an offset that can express the + * amount of data processed by the master, so we return a positive + * integer. */ + if (offset < 0) offset = 0; + return offset; +} + +/* --------------------------- REPLICATION CRON ---------------------------- */ + +/* Replication cron function, called 1 time per second. */ +void replicationCron(void) { + static long long replication_cron_loops = 0; + + /* Check failover status first, to see if we need to start + * handling the failover. */ + updateFailoverStatus(); + + /* Non blocking connection timeout? */ + if (server.masterhost && + (server.repl_state == REPL_STATE_CONNECTING || + slaveIsInHandshakeState()) && + (time(NULL)-server.repl_transfer_lastio) > server.repl_timeout) + { + serverLog(LL_WARNING,"Timeout connecting to the MASTER..."); + cancelReplicationHandshake(1); + } + + /* Bulk transfer I/O timeout? */ + if (server.masterhost && server.repl_state == REPL_STATE_TRANSFER && + (time(NULL)-server.repl_transfer_lastio) > server.repl_timeout) + { + 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."); + cancelReplicationHandshake(1); + } + + /* Timed out master when we are an already connected slave? */ + if (server.masterhost && server.repl_state == REPL_STATE_CONNECTED && + (time(NULL)-server.master->lastinteraction) > server.repl_timeout) + { + serverLog(LL_WARNING,"MASTER timeout: no data nor PING received..."); + freeClient(server.master); + } + + /* Check if we should connect to a MASTER */ + if (server.repl_state == REPL_STATE_CONNECT) { + serverLog(LL_NOTICE,"Connecting to MASTER %s:%d", + server.masterhost, server.masterport); + connectWithMaster(); + } + + /* Send ACK to master from time to time. + * Note that we do not send periodic acks to masters that don't + * support PSYNC and replication offsets. */ + if (server.masterhost && server.master && + !(server.master->flags & CLIENT_PRE_PSYNC)) + replicationSendAck(); + + /* If we have attached slaves, PING them from time to time. + * So slaves can implement an explicit timeout to masters, and will + * be able to detect a link disconnection even if the TCP connection + * will not actually go down. */ + listIter li; + listNode *ln; + robj *ping_argv[1]; + + /* First, send PING according to ping_slave_period. */ + if ((replication_cron_loops % server.repl_ping_slave_period) == 0 && + listLength(server.slaves)) + { + /* Note that we don't send the PING if the clients are paused during + * a Redis Cluster manual failover: the PING we send will otherwise + * alter the replication offsets of master and slave, and will no longer + * match the one stored into 'mf_master_offset' state. */ + int manual_failover_in_progress = + ((server.cluster_enabled && + server.cluster->mf_end) || + server.failover_end_time) && + isPausedActionsWithUpdate(PAUSE_ACTION_REPLICA); + + if (!manual_failover_in_progress) { + ping_argv[0] = shared.ping; + replicationFeedSlaves(server.slaves, -1, + ping_argv, 1); + } + } + + /* Second, send a newline to all the slaves in pre-synchronization + * stage, that is, slaves waiting for the master to create the RDB file. + * + * Also send the a newline to all the chained slaves we have, if we lost + * connection from our master, to keep the slaves aware that their + * master is online. This is needed since sub-slaves only receive proxied + * data from top-level masters, so there is no explicit pinging in order + * to avoid altering the replication offsets. This special out of band + * pings (newlines) can be sent, they will have no effect in the offset. + * + * The newline will be ignored by the slave but will refresh the + * last interaction timer preventing a timeout. In this case we ignore the + * ping period and refresh the connection once per second since certain + * timeouts are set at a few seconds (example: PSYNC response). */ + listRewind(server.slaves,&li); + while((ln = listNext(&li))) { + client *slave = ln->value; + + int is_presync = + (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START || + (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_END && + server.rdb_child_type != RDB_CHILD_TYPE_SOCKET)); + + if (is_presync) { + connWrite(slave->conn, "\n", 1); + } + } + + /* Disconnect timedout slaves. */ + if (listLength(server.slaves)) { + listIter li; + listNode *ln; + + listRewind(server.slaves,&li); + while((ln = listNext(&li))) { + client *slave = ln->value; + + if (slave->replstate == SLAVE_STATE_ONLINE) { + if (slave->flags & CLIENT_PRE_PSYNC) + continue; + if ((server.unixtime - slave->repl_ack_time) > server.repl_timeout) { + serverLog(LL_WARNING, "Disconnecting timedout replica (streaming sync): %s", + replicationGetSlaveName(slave)); + freeClient(slave); + continue; + } + } + /* We consider disconnecting only diskless replicas because disk-based replicas aren't fed + * by the fork child so if a disk-based replica is stuck it doesn't prevent the fork child + * from terminating. */ + if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_END && server.rdb_child_type == RDB_CHILD_TYPE_SOCKET) { + if (slave->repl_last_partial_write != 0 && + (server.unixtime - slave->repl_last_partial_write) > server.repl_timeout) + { + serverLog(LL_WARNING, "Disconnecting timedout replica (full sync): %s", + replicationGetSlaveName(slave)); + freeClient(slave); + continue; + } + } + } + } + + /* If this is a master without attached slaves and there is a replication + * backlog active, in order to reclaim memory we can free it after some + * (configured) time. Note that this cannot be done for slaves: slaves + * without sub-slaves attached should still accumulate data into the + * backlog, in order to reply to PSYNC queries if they are turned into + * masters after a failover. */ + if (listLength(server.slaves) == 0 && server.repl_backlog_time_limit && + server.repl_backlog && server.masterhost == NULL) + { + time_t idle = server.unixtime - server.repl_no_slaves_since; + + if (idle > server.repl_backlog_time_limit) { + /* When we free the backlog, we always use a new + * replication ID and clear the ID2. This is needed + * because when there is no backlog, the master_repl_offset + * is not updated, but we would still retain our replication + * ID, leading to the following problem: + * + * 1. We are a master instance. + * 2. Our slave is promoted to master. It's repl-id-2 will + * be the same as our repl-id. + * 3. We, yet as master, receive some updates, that will not + * increment the master_repl_offset. + * 4. Later we are turned into a slave, connect to the new + * master that will accept our PSYNC request by second + * replication ID, but there will be data inconsistency + * because we received writes. */ + changeReplicationId(); + clearReplicationId2(); + freeReplicationBacklog(); + serverLog(LL_NOTICE, + "Replication backlog freed after %d seconds " + "without connected replicas.", + (int) server.repl_backlog_time_limit); + } + } + + replicationStartPendingFork(); + + /* Remove the RDB file used for replication if Redis is not running + * with any persistence. */ + removeRDBUsedToSyncReplicas(); + + /* Sanity check replication buffer, the first block of replication buffer blocks + * must be referenced by someone, since it will be freed when not referenced, + * otherwise, server will OOM. also, its refcount must not be more than + * replicas number + 1(replication backlog). */ + if (listLength(server.repl_buffer_blocks) > 0) { + replBufBlock *o = listNodeValue(listFirst(server.repl_buffer_blocks)); + serverAssert(o->refcount > 0 && + o->refcount <= (int)listLength(server.slaves)+1); + } + + /* Refresh the number of slaves with lag <= min-slaves-max-lag. */ + refreshGoodSlavesCount(); + replication_cron_loops++; /* Incremented with frequency 1 HZ. */ +} + +int shouldStartChildReplication(int *mincapa_out, int *req_out) { + /* We should start a BGSAVE good for replication if we have slaves in + * WAIT_BGSAVE_START state. + * + * In case of diskless replication, we make sure to wait the specified + * number of seconds (according to configuration) so that other slaves + * have the time to arrive before we start streaming. */ + if (!hasActiveChildProcess()) { + time_t idle, max_idle = 0; + int slaves_waiting = 0; + int mincapa; + int req; + int first = 1; + listNode *ln; + listIter li; + + listRewind(server.slaves,&li); + while((ln = listNext(&li))) { + client *slave = ln->value; + if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) { + if (first) { + /* Get first slave's requirements */ + req = slave->slave_req; + } else if (req != slave->slave_req) { + /* Skip slaves that don't match */ + continue; + } + idle = server.unixtime - slave->lastinteraction; + if (idle > max_idle) max_idle = idle; + slaves_waiting++; + mincapa = first ? slave->slave_capa : (mincapa & slave->slave_capa); + first = 0; + } + } + + if (slaves_waiting && + (!server.repl_diskless_sync || + (server.repl_diskless_sync_max_replicas > 0 && + slaves_waiting >= server.repl_diskless_sync_max_replicas) || + max_idle >= server.repl_diskless_sync_delay)) + { + if (mincapa_out) + *mincapa_out = mincapa; + if (req_out) + *req_out = req; + return 1; + } + } + + return 0; +} + +void replicationStartPendingFork(void) { + int mincapa = -1; + int req = -1; + + if (shouldStartChildReplication(&mincapa, &req)) { + /* Start the BGSAVE. The called function may start a + * BGSAVE with socket target or disk target depending on the + * configuration and slaves capabilities and requirements. */ + startBgsaveForReplication(mincapa, req); + } +} + +/* Find replica at IP:PORT from replica list */ +static client *findReplica(char *host, int port) { + listIter li; + listNode *ln; + client *replica; + + listRewind(server.slaves,&li); + while((ln = listNext(&li))) { + replica = ln->value; + char ip[NET_IP_STR_LEN], *replicaip = replica->slave_addr; + + if (!replicaip) { + if (connAddrPeerName(replica->conn, ip, sizeof(ip), NULL) == -1) + continue; + replicaip = ip; + } + + if (!strcasecmp(host, replicaip) && + (port == replica->slave_listening_port)) + return replica; + } + + return NULL; +} + +const char *getFailoverStateString(void) { + switch(server.failover_state) { + case NO_FAILOVER: return "no-failover"; + case FAILOVER_IN_PROGRESS: return "failover-in-progress"; + case FAILOVER_WAIT_FOR_SYNC: return "waiting-for-sync"; + default: return "unknown"; + } +} + +/* Resets the internal failover configuration, this needs + * to be called after a failover either succeeds or fails + * as it includes the client unpause. */ +void clearFailoverState(void) { + server.failover_end_time = 0; + server.force_failover = 0; + zfree(server.target_replica_host); + server.target_replica_host = NULL; + server.target_replica_port = 0; + server.failover_state = NO_FAILOVER; + unpauseActions(PAUSE_DURING_FAILOVER); +} + +/* Abort an ongoing failover if one is going on. */ +void abortFailover(const char *err) { + if (server.failover_state == NO_FAILOVER) return; + + if (server.target_replica_host) { + serverLog(LL_NOTICE,"FAILOVER to %s:%d aborted: %s", + server.target_replica_host,server.target_replica_port,err); + } else { + serverLog(LL_NOTICE,"FAILOVER to any replica aborted: %s",err); + } + if (server.failover_state == FAILOVER_IN_PROGRESS) { + replicationUnsetMaster(); + } + clearFailoverState(); +} + +/* + * FAILOVER [TO <HOST> <PORT> [FORCE]] [ABORT] [TIMEOUT <timeout>] + * + * This command will coordinate a failover between the master and one + * of its replicas. The happy path contains the following steps: + * 1) The master will initiate a client pause write, to stop replication + * traffic. + * 2) The master will periodically check if any of its replicas has + * consumed the entire replication stream through acks. + * 3) Once any replica has caught up, the master will itself become a replica. + * 4) The master will send a PSYNC FAILOVER request to the target replica, which + * if accepted will cause the replica to become the new master and start a sync. + * + * FAILOVER ABORT is the only way to abort a failover command, as replicaof + * will be disabled. This may be needed if the failover is unable to progress. + * + * The optional arguments [TO <HOST> <IP>] allows designating a specific replica + * to be failed over to. + * + * FORCE flag indicates that even if the target replica is not caught up, + * failover to it anyway. This must be specified with a timeout and a target + * HOST and IP. + * + * TIMEOUT <timeout> indicates how long should the primary wait for + * a replica to sync up before aborting. If not specified, the failover + * will attempt forever and must be manually aborted. + */ +void failoverCommand(client *c) { + if (server.cluster_enabled) { + addReplyError(c,"FAILOVER not allowed in cluster mode. " + "Use CLUSTER FAILOVER command instead."); + return; + } + + /* Handle special case for abort */ + if ((c->argc == 2) && !strcasecmp(c->argv[1]->ptr,"abort")) { + if (server.failover_state == NO_FAILOVER) { + addReplyError(c, "No failover in progress."); + return; + } + + abortFailover("Failover manually aborted"); + addReply(c,shared.ok); + return; + } + + long timeout_in_ms = 0; + int force_flag = 0; + long port = 0; + char *host = NULL; + + /* Parse the command for syntax and arguments. */ + for (int j = 1; j < c->argc; j++) { + if (!strcasecmp(c->argv[j]->ptr,"timeout") && (j + 1 < c->argc) && + timeout_in_ms == 0) + { + if (getLongFromObjectOrReply(c,c->argv[j + 1], + &timeout_in_ms,NULL) != C_OK) return; + if (timeout_in_ms <= 0) { + addReplyError(c,"FAILOVER timeout must be greater than 0"); + return; + } + j++; + } else if (!strcasecmp(c->argv[j]->ptr,"to") && (j + 2 < c->argc) && + !host) + { + if (getLongFromObjectOrReply(c,c->argv[j + 2],&port,NULL) != C_OK) + return; + host = c->argv[j + 1]->ptr; + j += 2; + } else if (!strcasecmp(c->argv[j]->ptr,"force") && !force_flag) { + force_flag = 1; + } else { + addReplyErrorObject(c,shared.syntaxerr); + return; + } + } + + if (server.failover_state != NO_FAILOVER) { + addReplyError(c,"FAILOVER already in progress."); + return; + } + + if (server.masterhost) { + addReplyError(c,"FAILOVER is not valid when server is a replica."); + return; + } + + if (listLength(server.slaves) == 0) { + addReplyError(c,"FAILOVER requires connected replicas."); + return; + } + + if (force_flag && (!timeout_in_ms || !host)) { + addReplyError(c,"FAILOVER with force option requires both a timeout " + "and target HOST and IP."); + return; + } + + /* If a replica address was provided, validate that it is connected. */ + if (host) { + client *replica = findReplica(host, port); + + if (replica == NULL) { + addReplyError(c,"FAILOVER target HOST and PORT is not " + "a replica."); + return; + } + + /* Check if requested replica is online */ + if (replica->replstate != SLAVE_STATE_ONLINE) { + addReplyError(c,"FAILOVER target replica is not online."); + return; + } + + server.target_replica_host = zstrdup(host); + server.target_replica_port = port; + serverLog(LL_NOTICE,"FAILOVER requested to %s:%ld.",host,port); + } else { + serverLog(LL_NOTICE,"FAILOVER requested to any replica."); + } + + mstime_t now = commandTimeSnapshot(); + if (timeout_in_ms) { + server.failover_end_time = now + timeout_in_ms; + } + + server.force_failover = force_flag; + server.failover_state = FAILOVER_WAIT_FOR_SYNC; + /* Cluster failover will unpause eventually */ + pauseActions(PAUSE_DURING_FAILOVER, + LLONG_MAX, + PAUSE_ACTIONS_CLIENT_WRITE_SET); + addReply(c,shared.ok); +} + +/* Failover cron function, checks coordinated failover state. + * + * Implementation note: The current implementation calls replicationSetMaster() + * to start the failover request, this has some unintended side effects if the + * failover doesn't work like blocked clients will be unblocked and replicas will + * be disconnected. This could be optimized further. + */ +void updateFailoverStatus(void) { + if (server.failover_state != FAILOVER_WAIT_FOR_SYNC) return; + mstime_t now = server.mstime; + + /* Check if failover operation has timed out */ + if (server.failover_end_time && server.failover_end_time <= now) { + if (server.force_failover) { + serverLog(LL_NOTICE, + "FAILOVER to %s:%d time out exceeded, failing over.", + server.target_replica_host, server.target_replica_port); + server.failover_state = FAILOVER_IN_PROGRESS; + /* If timeout has expired force a failover if requested. */ + replicationSetMaster(server.target_replica_host, + server.target_replica_port); + return; + } else { + /* Force was not requested, so timeout. */ + abortFailover("Replica never caught up before timeout"); + return; + } + } + + /* Check to see if the replica has caught up so failover can start */ + client *replica = NULL; + if (server.target_replica_host) { + replica = findReplica(server.target_replica_host, + server.target_replica_port); + } else { + listIter li; + listNode *ln; + + listRewind(server.slaves,&li); + /* Find any replica that has matched our repl_offset */ + while((ln = listNext(&li))) { + replica = ln->value; + if (replica->repl_ack_off == server.master_repl_offset) { + char ip[NET_IP_STR_LEN], *replicaaddr = replica->slave_addr; + + if (!replicaaddr) { + if (connAddrPeerName(replica->conn,ip,sizeof(ip),NULL) == -1) + continue; + replicaaddr = ip; + } + + /* We are now failing over to this specific node */ + server.target_replica_host = zstrdup(replicaaddr); + server.target_replica_port = replica->slave_listening_port; + break; + } + } + } + + /* We've found a replica that is caught up */ + if (replica && (replica->repl_ack_off == server.master_repl_offset)) { + server.failover_state = FAILOVER_IN_PROGRESS; + serverLog(LL_NOTICE, + "Failover target %s:%d is synced, failing over.", + server.target_replica_host, server.target_replica_port); + /* Designated replica is caught up, failover to it. */ + replicationSetMaster(server.target_replica_host, + server.target_replica_port); + } +} |