Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c7ec1a367a | ||
|
|
ab1f8ea508 | ||
|
|
1fa63a78bc | ||
|
|
568c83dda7 | ||
|
|
892b1c3c58 | ||
|
|
cbf6614c1a | ||
|
|
7cb8481053 | ||
|
|
8242d069f1 | ||
|
|
6ef80f4ed2 | ||
|
|
dc03e4c51b | ||
|
|
30978004b3 | ||
|
|
652e662d1a | ||
|
|
a0ff29bcf2 | ||
|
|
e2c0d89662 | ||
|
|
5b6c764711 | ||
|
|
a74ef35f07 | ||
|
|
4698284b41 | ||
|
|
718a826c51 | ||
|
|
3915e1c71a | ||
|
|
ce4c17308e | ||
|
|
490847c681 | ||
|
|
c20218eb57 | ||
|
|
34d87be519 | ||
|
|
0513de624c | ||
|
|
49f2f691cb | ||
|
|
c2ff9de31b | ||
|
|
c5a8c8e907 | ||
|
|
4363fa1d76 | ||
|
|
5a9bc7cc10 | ||
|
|
0b76524983 | ||
|
|
421582f845 | ||
|
|
3cfca5afdc | ||
|
|
0643ba066e | ||
|
|
809e2f4da8 | ||
|
|
ba3237604f | ||
|
|
ade9bf7cb3 | ||
|
|
64b28d1097 | ||
|
|
dcc5ee156f | ||
|
|
58a8c0626c | ||
|
|
27c0ab238b | ||
|
|
547ccc4b5e | ||
|
|
99e4cf4d84 | ||
|
|
1dccb6cffb | ||
|
|
f6c65d3f45 | ||
|
|
9d2972183e | ||
|
|
32e979801b |
@@ -10,6 +10,78 @@ HIGH: There is a critical bug that may affect a subset of users. Upgrade!
|
||||
CRITICAL: There is a critical bug affecting MOST USERS. Upgrade ASAP.
|
||||
--------------------------------------------------------------------------------
|
||||
|
||||
--[ Redis 3.0.5 ] Release date: 15 Oct 2015
|
||||
|
||||
Upgrade urgency: MODERATE, the most important thing is a fix in the replication
|
||||
code that may make the slave hanging forever if the master
|
||||
remains with an open socket even if it is no longer able to
|
||||
reply.
|
||||
|
||||
* [FIX] MOVE now moves the TTL as well. A bug lasting forever... finally
|
||||
fixed thanks to Andy Grunwald that reported it.
|
||||
(reported by Andy Grunwald, fixed by Salvatore Sanfilippo)
|
||||
* [FIX] Fix a false positive in HSTRLEN test.
|
||||
* [FIX] Fix a bug in redis-cli --pipe mode that was not able to read back
|
||||
replies from the server incrementally. Now a mass import will use
|
||||
a lot less memory, and you can use --pipe to do incremental streaming.
|
||||
(reported by Twitter user @fsaintjacques, fixed by Salvatore
|
||||
Sanfilippo)
|
||||
* [FIX] Slave detection of master timeout. (fixed by Kevin McGehee, refactoring
|
||||
and regression test by Salvatore Sanfilippo)
|
||||
|
||||
* [NEW] Cluster: redis-trib fix can fix an additional case for opens lots.
|
||||
(Salvatore Sanfilippo)
|
||||
* [NEW] Cluster: redis-trib import support for --copy and --replace options
|
||||
(David Thomson)
|
||||
|
||||
--[ Redis 3.0.4 ] Release date: 8 Sep 2015
|
||||
|
||||
Upgrade urgency: HIGH for Redis and Sentinel. However note that in order to
|
||||
fix certain replication bugs, the replication internals were
|
||||
modified in a very heavy way. So while this release is
|
||||
conceptually saner, it may contain regressions. For this
|
||||
reason, before the release, QA activities were performed by
|
||||
me (antirez) and Redis Labs and no evident bug was found.
|
||||
|
||||
* [FIX] A number of bugs related to replication PSYNC and the (yet experimental)
|
||||
diskless replication feature were fixed. The bugs could lead to
|
||||
inconsistency between masters and slaves. (Salvatore Sanfilippo, Oran
|
||||
Agra fixed the issue found by Yuval Inbar)
|
||||
* [FIX] A replication bug in the context of PSYNC partial resynchonization was
|
||||
found and fixed. This bug happens even when diskless replication is off
|
||||
in the case different slaves connect at different times while the master
|
||||
is creating an RDB file, and later a partial resynchronization is
|
||||
attempted by a slave that connected not as the first one. (Salvatore
|
||||
Sanfilippo, Oran Agra)
|
||||
* [FIX] Chained replication and PSYNC interactions leading to potential stale
|
||||
chained slaves data set, see issue #2694. (Salvatore Sanfilippo fixed
|
||||
an issue reported by "GeorgeBJ" user at Github)
|
||||
* [FIX] redis-cli --scan iteration fixed when returned cursor overflows
|
||||
32 bit signed integer. (Ofir Luzon, Yuval Inbar)
|
||||
* [FIX] Sentinel: fixed a bug during the master switch process, where for a
|
||||
failed conditional check, the new configuration is rewritten, during
|
||||
a small window of time, in a corrupted way where the master is
|
||||
also reported to be one of the slaves. This bug is rare to trigger
|
||||
but apparently it happens in the wild, and the effect is to see
|
||||
a replication loop where the master will try to replicate with itself.
|
||||
A detailed explanation of the bug and its effects can be found in
|
||||
the commit message here: https://github.com/antirez/redis/commit/c20218eb5770b2cafb12bc7092313b8358fedc0a.
|
||||
The bug was found by Jan-Erik Rediger using a static analyzer and
|
||||
fixed by Salvatore Sanfilippo.
|
||||
* [FIX] Sentinel lack of arity checks for certain commands.
|
||||
(Rogerio Goncalves, Salvatore Sanfilippo)
|
||||
|
||||
* [NEW] Replication internals rewritten in order to be more resistant to bugs.
|
||||
The replication handshake in the slave side was rewritten as a non
|
||||
blocking state machine. (Salvatore Sanfilippo, Oran Agra)
|
||||
* [NEW] New "replication capabilities" feature introduced in order to signal
|
||||
from the master to the slave what are the features supported, so that
|
||||
the master can choose the kind of replication to start (diskless or
|
||||
not) when master and slave are of different versions. (Oran Agra,
|
||||
Salvatore Sanfilippo)
|
||||
* [NEW] Log clients details when SLAVEOF command is received. (Salvatore
|
||||
Sanfilippo with inputs from Nick Craver and Marc Gravell).
|
||||
|
||||
--[ Redis 3.0.3 ] Release date: 17 Jul 2015
|
||||
|
||||
Upgrade urgency: LOW for Redis and Sentinel.
|
||||
|
||||
+6
-1
@@ -1,4 +1,9 @@
|
||||
# Redis configuration file example
|
||||
# Redis configuration file example.
|
||||
#
|
||||
# Note that in order to read the configuration file, Redis must be
|
||||
# started with the file path as first argument:
|
||||
#
|
||||
# ./redis-server /path/to/redis.conf
|
||||
|
||||
# Note on units: when memory size is needed, it is possible to specify
|
||||
# it in the usual form of 1k 5GB 4M and so forth:
|
||||
|
||||
@@ -714,7 +714,7 @@ void moveCommand(redisClient *c) {
|
||||
robj *o;
|
||||
redisDb *src, *dst;
|
||||
int srcid;
|
||||
long long dbid;
|
||||
long long dbid, expire;
|
||||
|
||||
if (server.cluster_enabled) {
|
||||
addReplyError(c,"MOVE is not allowed in cluster mode");
|
||||
@@ -748,6 +748,7 @@ void moveCommand(redisClient *c) {
|
||||
addReply(c,shared.czero);
|
||||
return;
|
||||
}
|
||||
expire = getExpire(c->db,c->argv[1]);
|
||||
|
||||
/* Return zero if the key already exists in the target DB */
|
||||
if (lookupKeyWrite(dst,c->argv[1]) != NULL) {
|
||||
@@ -755,6 +756,7 @@ void moveCommand(redisClient *c) {
|
||||
return;
|
||||
}
|
||||
dbAdd(dst,c->argv[1],o);
|
||||
if (expire != -1) setExpire(dst,c->argv[1],expire);
|
||||
incrRefCount(o);
|
||||
|
||||
/* OK! key moved, free the entry in the source DB */
|
||||
|
||||
+7
-14
@@ -105,6 +105,7 @@ redisClient *createClient(int fd) {
|
||||
c->repl_ack_off = 0;
|
||||
c->repl_ack_time = 0;
|
||||
c->slave_listening_port = 0;
|
||||
c->slave_capa = SLAVE_CAPA_NONE;
|
||||
c->reply = listCreate();
|
||||
c->reply_bytes = 0;
|
||||
c->obuf_soft_limit_reached_time = 0;
|
||||
@@ -666,20 +667,6 @@ void disconnectSlaves(void) {
|
||||
}
|
||||
}
|
||||
|
||||
/* This function is called when the slave lose the connection with the
|
||||
* master into an unexpected way. */
|
||||
void replicationHandleMasterDisconnection(void) {
|
||||
server.master = NULL;
|
||||
server.repl_state = REDIS_REPL_CONNECT;
|
||||
server.repl_down_since = server.unixtime;
|
||||
/* We lost connection with our master, force our slaves to resync
|
||||
* with us as well to load the new data set.
|
||||
*
|
||||
* If server.masterhost is NULL the user called SLAVEOF NO ONE so
|
||||
* slave resync is not needed. */
|
||||
if (server.masterhost != NULL) disconnectSlaves();
|
||||
}
|
||||
|
||||
void freeClient(redisClient *c) {
|
||||
listNode *ln;
|
||||
|
||||
@@ -1679,6 +1666,12 @@ void flushSlavesOutputBuffers(void) {
|
||||
redisClient *slave = listNodeValue(ln);
|
||||
int events;
|
||||
|
||||
/* Note that the following will not flush output buffers of slaves
|
||||
* in STATE_ONLINE but having put_online_on_ack set to true: in this
|
||||
* case the writable event is never installed, since the purpose
|
||||
* of put_online_on_ack is to postpone the moment it is installed.
|
||||
* This is what we want since slaves in this state should not receive
|
||||
* writes before the first ACK. */
|
||||
events = aeGetFileEvents(server.el,slave->fd);
|
||||
if (events & AE_WRITABLE &&
|
||||
slave->replstate == REDIS_REPL_ONLINE &&
|
||||
|
||||
@@ -1420,7 +1420,7 @@ int rdbSaveToSlavesSockets(void) {
|
||||
if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) {
|
||||
clientids[numfds] = slave->id;
|
||||
fds[numfds++] = slave->fd;
|
||||
slave->replstate = REDIS_REPL_WAIT_BGSAVE_END;
|
||||
replicationSetupSlaveForFullResync(slave,getPsyncInitialOffset());
|
||||
/* Put the socket in non-blocking mode to simplify RDB transfer.
|
||||
* We'll restore it when the children returns (since duped socket
|
||||
* will share the O_NONBLOCK attribute with the parent). */
|
||||
@@ -1498,27 +1498,43 @@ int rdbSaveToSlavesSockets(void) {
|
||||
exitFromChild((retval == REDIS_OK) ? 0 : 1);
|
||||
} else {
|
||||
/* Parent */
|
||||
zfree(clientids); /* Not used by parent. Free ASAP. */
|
||||
server.stat_fork_time = ustime()-start;
|
||||
server.stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / server.stat_fork_time / (1024*1024*1024); /* GB per second. */
|
||||
latencyAddSampleIfNeeded("fork",server.stat_fork_time/1000);
|
||||
if (childpid == -1) {
|
||||
redisLog(REDIS_WARNING,"Can't save in background: fork: %s",
|
||||
strerror(errno));
|
||||
zfree(fds);
|
||||
|
||||
/* Undo the state change. The caller will perform cleanup on
|
||||
* all the slaves in BGSAVE_START state, but an early call to
|
||||
* replicationSetupSlaveForFullResync() turned it into BGSAVE_END */
|
||||
listRewind(server.slaves,&li);
|
||||
while((ln = listNext(&li))) {
|
||||
redisClient *slave = ln->value;
|
||||
int j;
|
||||
|
||||
for (j = 0; j < numfds; j++) {
|
||||
if (slave->id == clientids[j]) {
|
||||
slave->replstate = REDIS_REPL_WAIT_BGSAVE_START;
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
close(pipefds[0]);
|
||||
close(pipefds[1]);
|
||||
return REDIS_ERR;
|
||||
} else {
|
||||
redisLog(REDIS_NOTICE,"Background RDB transfer started by pid %d",
|
||||
childpid);
|
||||
server.rdb_save_time_start = time(NULL);
|
||||
server.rdb_child_pid = childpid;
|
||||
server.rdb_child_type = REDIS_RDB_CHILD_TYPE_SOCKET;
|
||||
updateDictResizePolicy();
|
||||
}
|
||||
redisLog(REDIS_NOTICE,"Background RDB transfer started by pid %d",childpid);
|
||||
server.rdb_save_time_start = time(NULL);
|
||||
server.rdb_child_pid = childpid;
|
||||
server.rdb_child_type = REDIS_RDB_CHILD_TYPE_SOCKET;
|
||||
updateDictResizePolicy();
|
||||
zfree(clientids);
|
||||
zfree(fds);
|
||||
return REDIS_OK;
|
||||
return (childpid == -1) ? REDIS_ERR : REDIS_OK;
|
||||
}
|
||||
return REDIS_OK; /* unreached */
|
||||
return REDIS_OK; /* Unreached. */
|
||||
}
|
||||
|
||||
void saveCommand(redisClient *c) {
|
||||
|
||||
+7
-2
@@ -1397,6 +1397,7 @@ static void getRDB(void) {
|
||||
* Bulk import (pipe) mode
|
||||
*--------------------------------------------------------------------------- */
|
||||
|
||||
#define PIPEMODE_WRITE_LOOP_MAX_BYTES (128*1024)
|
||||
static void pipeMode(void) {
|
||||
int fd = context->fd;
|
||||
long long errors = 0, replies = 0, obuf_len = 0, obuf_pos = 0;
|
||||
@@ -1473,6 +1474,8 @@ static void pipeMode(void) {
|
||||
|
||||
/* Handle the writable state: we can send protocol to the server. */
|
||||
if (mask & AE_WRITABLE) {
|
||||
ssize_t loop_nwritten = 0;
|
||||
|
||||
while(1) {
|
||||
/* Transfer current buffer to server. */
|
||||
if (obuf_len != 0) {
|
||||
@@ -1489,6 +1492,7 @@ static void pipeMode(void) {
|
||||
}
|
||||
obuf_len -= nwritten;
|
||||
obuf_pos += nwritten;
|
||||
loop_nwritten += nwritten;
|
||||
if (obuf_len != 0) break; /* Can't accept more data. */
|
||||
}
|
||||
/* If buffer is empty, load from stdin. */
|
||||
@@ -1524,7 +1528,8 @@ static void pipeMode(void) {
|
||||
obuf_pos = 0;
|
||||
}
|
||||
}
|
||||
if (obuf_len == 0 && eof) break;
|
||||
if ((obuf_len == 0 && eof) ||
|
||||
loop_nwritten > PIPEMODE_WRITE_LOOP_MAX_BYTES) break;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1582,7 +1587,7 @@ static redisReply *sendScan(unsigned long long *it) {
|
||||
assert(reply->element[1]->type == REDIS_REPLY_ARRAY);
|
||||
|
||||
/* Update iterator */
|
||||
*it = atoi(reply->element[0]->str);
|
||||
*it = strtoull(reply->element[0]->str, NULL, 10);
|
||||
|
||||
return reply;
|
||||
}
|
||||
|
||||
+20
-5
@@ -496,6 +496,10 @@ class RedisTrib
|
||||
# importing state in 1 slot. That's trivial to address.
|
||||
if migrating.length == 1 && importing.length == 1
|
||||
move_slot(migrating[0],importing[0],slot,:verbose=>true,:fix=>true)
|
||||
# Case 2: There are multiple nodes that claim the slot as importing,
|
||||
# they probably got keys about the slot after a restart so opened
|
||||
# the slot. In this case we just move all the keys to the owner
|
||||
# according to the configuration.
|
||||
elsif migrating.length == 0 && importing.length > 0
|
||||
xputs ">>> Moving all the #{slot} slot keys to its owner #{owner}"
|
||||
importing.each {|node|
|
||||
@@ -504,8 +508,14 @@ class RedisTrib
|
||||
xputs ">>> Setting #{slot} as STABLE in #{node}"
|
||||
node.r.cluster("setslot",slot,"stable")
|
||||
}
|
||||
# Case 3: There are no slots claiming to be in importing state, but
|
||||
# there is a migrating node that actually don't have any key. We
|
||||
# can just close the slot, probably a reshard interrupted in the middle.
|
||||
elsif importing.length == 0 && migrating.length == 1 &&
|
||||
migrating[0].r.cluster("getkeysinslot",slot,10).length == 0
|
||||
migrating[0].r.cluster("setslot",slot,"stable")
|
||||
else
|
||||
xputs "[ERR] Sorry, Redis-trib can't fix this slot yet (work in progress)"
|
||||
xputs "[ERR] Sorry, Redis-trib can't fix this slot yet (work in progress). Slot is set as migrating in #{migrating.join(",")}, as importing in #{importing.join(",")}, owner is #{owner}"
|
||||
end
|
||||
end
|
||||
|
||||
@@ -812,7 +822,7 @@ class RedisTrib
|
||||
source.r.client.call(["migrate",target.info[:host],target.info[:port],key,0,15000])
|
||||
rescue => e
|
||||
if o[:fix] && e.to_s =~ /BUSYKEY/
|
||||
xputs "*** Target key #{key} exists. Replace it for FIX."
|
||||
xputs "*** Target key #{key} exists. Replacing it for FIX."
|
||||
source.r.client.call(["migrate",target.info[:host],target.info[:port],key,0,15000,:replace])
|
||||
else
|
||||
puts ""
|
||||
@@ -1139,7 +1149,9 @@ class RedisTrib
|
||||
def import_cluster_cmd(argv,opt)
|
||||
source_addr = opt['from']
|
||||
xputs ">>> Importing data from #{source_addr} to cluster #{argv[1]}"
|
||||
|
||||
use_copy = opt['copy']
|
||||
use_replace = opt['replace']
|
||||
|
||||
# Check the existing cluster.
|
||||
load_cluster_info_from_node(argv[0])
|
||||
check_cluster
|
||||
@@ -1174,7 +1186,10 @@ class RedisTrib
|
||||
print "Migrating #{k} to #{target}: "
|
||||
STDOUT.flush
|
||||
begin
|
||||
source.client.call(["migrate",target.info[:host],target.info[:port],k,0,15000])
|
||||
cmd = ["migrate",target.info[:host],target.info[:port],k,0,15000]
|
||||
cmd << :copy if use_copy
|
||||
cmd << :replace if use_replace
|
||||
source.client.call(cmd)
|
||||
rescue => e
|
||||
puts e
|
||||
else
|
||||
@@ -1334,7 +1349,7 @@ COMMANDS={
|
||||
ALLOWED_OPTIONS={
|
||||
"create" => {"replicas" => true},
|
||||
"add-node" => {"slave" => false, "master-id" => true},
|
||||
"import" => {"from" => :required},
|
||||
"import" => {"from" => :required, "copy" => false, "replace" => false},
|
||||
"reshard" => {"from" => true, "to" => true, "slots" => true, "yes" => false}
|
||||
}
|
||||
|
||||
|
||||
+1
-1
@@ -3660,6 +3660,7 @@ int main(int argc, char **argv) {
|
||||
if (server.daemonize) createPidFile();
|
||||
redisSetProcTitle(argv[0]);
|
||||
redisAsciiArt();
|
||||
checkTcpBacklogSettings();
|
||||
|
||||
if (!server.sentinel_mode) {
|
||||
/* Things not needed when running in Sentinel mode. */
|
||||
@@ -3667,7 +3668,6 @@ int main(int argc, char **argv) {
|
||||
#ifdef __linux__
|
||||
linuxMemoryWarnings();
|
||||
#endif
|
||||
checkTcpBacklogSettings();
|
||||
loadDataFromDisk();
|
||||
if (server.cluster_enabled) {
|
||||
if (verifyClusterConfigWithData() == REDIS_ERR) {
|
||||
|
||||
+29
-8
@@ -267,22 +267,37 @@ typedef long long mstime_t; /* millisecond time type. */
|
||||
#define REDIS_CLIENT_TYPE_PUBSUB 2 /* Clients subscribed to PubSub channels. */
|
||||
#define REDIS_CLIENT_TYPE_COUNT 3
|
||||
|
||||
/* Slave replication state - from the point of view of the slave. */
|
||||
/* Slave replication state. Used in server.repl_state for slaves to remember
|
||||
* what to do next. */
|
||||
#define REDIS_REPL_NONE 0 /* No active replication */
|
||||
#define REDIS_REPL_CONNECT 1 /* Must connect to master */
|
||||
#define REDIS_REPL_CONNECTING 2 /* Connecting to master */
|
||||
/* --- Handshake states, must be ordered --- */
|
||||
#define REDIS_REPL_RECEIVE_PONG 3 /* Wait for PING reply */
|
||||
#define REDIS_REPL_TRANSFER 4 /* Receiving .rdb from master */
|
||||
#define REDIS_REPL_CONNECTED 5 /* Connected to master */
|
||||
#define REDIS_REPL_SEND_AUTH 4 /* Send AUTH to master */
|
||||
#define REDIS_REPL_RECEIVE_AUTH 5 /* Wait for AUTH reply */
|
||||
#define REDIS_REPL_SEND_PORT 6 /* Send REPLCONF listening-port */
|
||||
#define REDIS_REPL_RECEIVE_PORT 7 /* Wait for REPLCONF reply */
|
||||
#define REDIS_REPL_SEND_CAPA 8 /* Send REPLCONF capa */
|
||||
#define REDIS_REPL_RECEIVE_CAPA 9 /* Wait for REPLCONF reply */
|
||||
#define REDIS_REPL_SEND_PSYNC 10 /* Send PSYNC */
|
||||
#define REDIS_REPL_RECEIVE_PSYNC 11 /* Wait for PSYNC reply */
|
||||
/* --- End of handshake states --- */
|
||||
#define REDIS_REPL_TRANSFER 12 /* Receiving .rdb from master */
|
||||
#define REDIS_REPL_CONNECTED 13 /* Connected to master */
|
||||
|
||||
/* Slave replication state - from the point of view of the master.
|
||||
/* State of slaves from the POV of the master. Used in client->replstate.
|
||||
* In SEND_BULK and ONLINE state the slave receives new updates
|
||||
* in its output queue. In the WAIT_BGSAVE state instead the server is waiting
|
||||
* to start the next background saving in order to send updates to it. */
|
||||
#define REDIS_REPL_WAIT_BGSAVE_START 6 /* We need to produce a new RDB file. */
|
||||
#define REDIS_REPL_WAIT_BGSAVE_END 7 /* Waiting RDB file creation to finish. */
|
||||
#define REDIS_REPL_SEND_BULK 8 /* Sending RDB file to slave. */
|
||||
#define REDIS_REPL_ONLINE 9 /* RDB file transmitted, sending just updates. */
|
||||
#define REDIS_REPL_WAIT_BGSAVE_START 14 /* We need to produce a new RDB file. */
|
||||
#define REDIS_REPL_WAIT_BGSAVE_END 15 /* Waiting RDB file creation to finish. */
|
||||
#define REDIS_REPL_SEND_BULK 16 /* Sending RDB file to slave. */
|
||||
#define REDIS_REPL_ONLINE 17 /* RDB file transmitted, sending just updates. */
|
||||
|
||||
/* Slave capabilities. */
|
||||
#define SLAVE_CAPA_NONE 0
|
||||
#define SLAVE_CAPA_EOF (1<<0) /* Can parse the RDB EOF streaming format. */
|
||||
|
||||
/* Synchronous read timeout - slave side */
|
||||
#define REDIS_REPL_SYNCIO_TIMEOUT 5
|
||||
@@ -543,8 +558,12 @@ typedef struct redisClient {
|
||||
long long reploff; /* replication offset if this is our master */
|
||||
long long repl_ack_off; /* replication ack offset, if this is a slave */
|
||||
long long repl_ack_time;/* replication ack time, if this is a slave */
|
||||
long long psync_initial_offset; /* FULLRESYNC reply offset other slaves
|
||||
copying this slave output buffer
|
||||
should use. */
|
||||
char replrunid[REDIS_RUN_ID_SIZE+1]; /* master run id if this is a master */
|
||||
int slave_listening_port; /* As configured with: SLAVECONF listening-port */
|
||||
int slave_capa; /* Slave capabilities: SLAVE_CAPA_* bitwise OR. */
|
||||
multiState mstate; /* MULTI/EXEC state */
|
||||
int btype; /* Type of blocking op if REDIS_BLOCKED. */
|
||||
blockingState bpop; /* blocking state */
|
||||
@@ -1171,6 +1190,8 @@ int replicationCountAcksByOffset(long long offset);
|
||||
void replicationSendNewlineToMaster(void);
|
||||
long long replicationGetSlaveOffset(void);
|
||||
char *replicationGetSlaveName(redisClient *c);
|
||||
long long getPsyncInitialOffset(void);
|
||||
int replicationSetupSlaveForFullResync(redisClient *slave, long long offset);
|
||||
|
||||
/* Generic persistence functions */
|
||||
void startLoading(FILE *fp);
|
||||
|
||||
+421
-176
@@ -201,6 +201,7 @@ void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) {
|
||||
listRewind(slaves,&li);
|
||||
while((ln = listNext(&li))) {
|
||||
redisClient *slave = ln->value;
|
||||
if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) continue;
|
||||
addReply(slave,selectcmd);
|
||||
}
|
||||
|
||||
@@ -349,6 +350,58 @@ long long addReplyReplicationBacklog(redisClient *c, long long 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) {
|
||||
long long psync_offset = server.master_repl_offset;
|
||||
/* Add 1 to psync_offset if it the replication backlog does not exists
|
||||
* as when it will be created later we'll increment the offset by one. */
|
||||
if (server.repl_backlog == NULL) psync_offset++;
|
||||
return psync_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 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(redisClient *slave, long long offset) {
|
||||
char buf[128];
|
||||
int buflen;
|
||||
|
||||
slave->psync_initial_offset = offset;
|
||||
slave->replstate = REDIS_REPL_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 SLEECT 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 & REDIS_PRE_PSYNC)) {
|
||||
buflen = snprintf(buf,sizeof(buf),"+FULLRESYNC %s %lld\r\n",
|
||||
server.runid,offset);
|
||||
if (write(slave->fd,buf,buflen) != buflen) {
|
||||
freeClientAsync(slave);
|
||||
return REDIS_ERR;
|
||||
}
|
||||
}
|
||||
return REDIS_OK;
|
||||
}
|
||||
|
||||
/* This function handles the PSYNC command from the point of view of a
|
||||
* master receiving a request for partial resynchronization.
|
||||
*
|
||||
@@ -422,18 +475,10 @@ int masterTryPartialResynchronization(redisClient *c) {
|
||||
return REDIS_OK; /* The caller can return, no full resync needed. */
|
||||
|
||||
need_full_resync:
|
||||
/* We need a full resync for some reason... notify the client. */
|
||||
psync_offset = server.master_repl_offset;
|
||||
/* Add 1 to psync_offset if it the replication backlog does not exists
|
||||
* as when it will be created later we'll increment the offset by one. */
|
||||
if (server.repl_backlog == NULL) psync_offset++;
|
||||
/* Again, we can't use the connection buffers (see above). */
|
||||
buflen = snprintf(buf,sizeof(buf),"+FULLRESYNC %s %lld\r\n",
|
||||
server.runid,psync_offset);
|
||||
if (write(c->fd,buf,buflen) != buflen) {
|
||||
freeClientAsync(c);
|
||||
return REDIS_OK;
|
||||
}
|
||||
/* 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 REDIS_ERR;
|
||||
}
|
||||
|
||||
@@ -441,18 +486,68 @@ need_full_resync:
|
||||
* 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 succesfully 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 REDIS_OK on success or REDIS_ERR otherwise. */
|
||||
int startBgsaveForReplication(void) {
|
||||
int startBgsaveForReplication(int mincapa) {
|
||||
int retval;
|
||||
int socket_target = server.repl_diskless_sync && (mincapa & SLAVE_CAPA_EOF);
|
||||
listIter li;
|
||||
listNode *ln;
|
||||
|
||||
redisLog(REDIS_NOTICE,"Starting BGSAVE for SYNC with target: %s",
|
||||
server.repl_diskless_sync ? "slaves sockets" : "disk");
|
||||
socket_target ? "slaves sockets" : "disk");
|
||||
|
||||
if (server.repl_diskless_sync)
|
||||
if (socket_target)
|
||||
retval = rdbSaveToSlavesSockets();
|
||||
else
|
||||
retval = rdbSaveBackground(server.rdb_filename);
|
||||
|
||||
/* If we failed to BGSAVE, remove the slaves waiting for a full
|
||||
* resynchorinization from the list of salves, inform them with
|
||||
* an error about what happened, close the connection ASAP. */
|
||||
if (retval == REDIS_ERR) {
|
||||
redisLog(REDIS_WARNING,"BGSAVE for replication failed");
|
||||
listRewind(server.slaves,&li);
|
||||
while((ln = listNext(&li))) {
|
||||
redisClient *slave = ln->value;
|
||||
|
||||
if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) {
|
||||
slave->flags &= ~REDIS_SLAVE;
|
||||
listDelNode(server.slaves,ln);
|
||||
addReplyError(slave,
|
||||
"BGSAVE failed, replication can't continue");
|
||||
slave->flags |= REDIS_CLOSE_AFTER_REPLY;
|
||||
}
|
||||
}
|
||||
return retval;
|
||||
}
|
||||
|
||||
/* If the target is socket, rdbSaveToSlavesSockets() already setup
|
||||
* the salves for a full resync. Otherwise for disk target do it now.*/
|
||||
if (!socket_target) {
|
||||
listRewind(server.slaves,&li);
|
||||
while((ln = listNext(&li))) {
|
||||
redisClient *slave = ln->value;
|
||||
|
||||
if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) {
|
||||
replicationSetupSlaveForFullResync(slave,
|
||||
getPsyncInitialOffset());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/* Flush the script cache, since we need that slave differences are
|
||||
* accumulated without requiring slaves to match our cached scripts. */
|
||||
if (retval == REDIS_OK) replicationScriptCacheFlush();
|
||||
@@ -515,8 +610,16 @@ void syncCommand(redisClient *c) {
|
||||
/* Full resynchronization. */
|
||||
server.stat_sync_full++;
|
||||
|
||||
/* Here we need to check if there is a background saving operation
|
||||
* in progress, or if it is required to start one */
|
||||
/* 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 = REDIS_REPL_WAIT_BGSAVE_START;
|
||||
if (server.repl_disable_tcp_nodelay)
|
||||
anetDisableTcpNoDelay(NULL, c->fd); /* Non critical if it fails. */
|
||||
c->repldbfd = -1;
|
||||
c->flags |= REDIS_SLAVE;
|
||||
listAddNodeTail(server.slaves,c);
|
||||
|
||||
/* CASE 1: BGSAVE is in progress, with disk target. */
|
||||
if (server.rdb_child_pid != -1 &&
|
||||
server.rdb_child_type == REDIS_RDB_CHILD_TYPE_DISK)
|
||||
{
|
||||
@@ -532,51 +635,45 @@ void syncCommand(redisClient *c) {
|
||||
slave = ln->value;
|
||||
if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) break;
|
||||
}
|
||||
if (ln) {
|
||||
/* To attach this slave, we check that it has at least all the
|
||||
* capabilities of the slave that triggered the current BGSAVE. */
|
||||
if (ln && ((c->slave_capa & slave->slave_capa) == slave->slave_capa)) {
|
||||
/* Perfect, the server is already registering differences for
|
||||
* another slave. Set the right state, and copy the buffer. */
|
||||
copyClientOutputBuffer(c,slave);
|
||||
c->replstate = REDIS_REPL_WAIT_BGSAVE_END;
|
||||
replicationSetupSlaveForFullResync(c,slave->psync_initial_offset);
|
||||
redisLog(REDIS_NOTICE,"Waiting for end of BGSAVE for SYNC");
|
||||
} else {
|
||||
/* No way, we need to wait for the next BGSAVE in order to
|
||||
* register differences. */
|
||||
c->replstate = REDIS_REPL_WAIT_BGSAVE_START;
|
||||
redisLog(REDIS_NOTICE,"Waiting for next BGSAVE for SYNC");
|
||||
}
|
||||
|
||||
/* CASE 2: BGSAVE is in progress, with socket target. */
|
||||
} else if (server.rdb_child_pid != -1 &&
|
||||
server.rdb_child_type == REDIS_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. */
|
||||
c->replstate = REDIS_REPL_WAIT_BGSAVE_START;
|
||||
redisLog(REDIS_NOTICE,"Waiting for next BGSAVE for SYNC");
|
||||
|
||||
/* CASE 3: There is no BGSAVE is progress. */
|
||||
} else {
|
||||
if (server.repl_diskless_sync) {
|
||||
if (server.repl_diskless_sync && (c->slave_capa & SLAVE_CAPA_EOF)) {
|
||||
/* 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. */
|
||||
c->replstate = REDIS_REPL_WAIT_BGSAVE_START;
|
||||
if (server.repl_diskless_sync_delay)
|
||||
redisLog(REDIS_NOTICE,"Delay next BGSAVE for SYNC");
|
||||
} else {
|
||||
/* Ok we don't have a BGSAVE in progress, let's start one. */
|
||||
if (startBgsaveForReplication() != REDIS_OK) {
|
||||
redisLog(REDIS_NOTICE,"Replication failed, can't BGSAVE");
|
||||
addReplyError(c,"Unable to perform background save");
|
||||
return;
|
||||
}
|
||||
c->replstate = REDIS_REPL_WAIT_BGSAVE_END;
|
||||
/* Target is disk (or the slave is not capable of supporting
|
||||
* diskless replication) and we don't have a BGSAVE in progress,
|
||||
* let's start one. */
|
||||
if (startBgsaveForReplication(c->slave_capa) != REDIS_OK) return;
|
||||
}
|
||||
}
|
||||
|
||||
if (server.repl_disable_tcp_nodelay)
|
||||
anetDisableTcpNoDelay(NULL, c->fd); /* Non critical if it fails. */
|
||||
c->repldbfd = -1;
|
||||
c->flags |= REDIS_SLAVE;
|
||||
server.slaveseldb = -1; /* Force to re-emit the SELECT command. */
|
||||
listAddNodeTail(server.slaves,c);
|
||||
if (listLength(server.slaves) == 1 && server.repl_backlog == NULL)
|
||||
createReplicationBacklog();
|
||||
return;
|
||||
@@ -613,6 +710,10 @@ void replconfCommand(redisClient *c) {
|
||||
&port,NULL) != REDIS_OK))
|
||||
return;
|
||||
c->slave_listening_port = port;
|
||||
} 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]->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
|
||||
@@ -746,6 +847,7 @@ void sendBulkToSlave(aeEventLoop *el, int fd, void *privdata, int mask) {
|
||||
void updateSlavesWaitingBgsave(int bgsaveerr, int type) {
|
||||
listNode *ln;
|
||||
int startbgsave = 0;
|
||||
int mincapa = -1;
|
||||
listIter li;
|
||||
|
||||
listRewind(server.slaves,&li);
|
||||
@@ -754,7 +856,8 @@ void updateSlavesWaitingBgsave(int bgsaveerr, int type) {
|
||||
|
||||
if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) {
|
||||
startbgsave = 1;
|
||||
slave->replstate = REDIS_REPL_WAIT_BGSAVE_END;
|
||||
mincapa = (mincapa == -1) ? slave->slave_capa :
|
||||
(mincapa & slave->slave_capa);
|
||||
} else if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) {
|
||||
struct redis_stat buf;
|
||||
|
||||
@@ -801,24 +904,18 @@ void updateSlavesWaitingBgsave(int bgsaveerr, int type) {
|
||||
}
|
||||
}
|
||||
}
|
||||
if (startbgsave) {
|
||||
if (startBgsaveForReplication() != REDIS_OK) {
|
||||
listIter li;
|
||||
|
||||
listRewind(server.slaves,&li);
|
||||
redisLog(REDIS_WARNING,"SYNC failed. BGSAVE failed");
|
||||
while((ln = listNext(&li))) {
|
||||
redisClient *slave = ln->value;
|
||||
|
||||
if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START)
|
||||
freeClient(slave);
|
||||
}
|
||||
}
|
||||
}
|
||||
if (startbgsave) startBgsaveForReplication(mincapa);
|
||||
}
|
||||
|
||||
/* ----------------------------------- SLAVE -------------------------------- */
|
||||
|
||||
/* Returns 1 if the given replication state is a handshake state,
|
||||
* 0 otherwise. */
|
||||
int slaveIsInHandshakeState(void) {
|
||||
return server.repl_state >= REDIS_REPL_RECEIVE_PONG &&
|
||||
server.repl_state <= REDIS_REPL_RECEIVE_PSYNC;
|
||||
}
|
||||
|
||||
/* Abort the async download of the bulk dataset while SYNC-ing with master */
|
||||
void replicationAbortSyncTransfer(void) {
|
||||
redisAssert(server.repl_state == REDIS_REPL_TRANSFER);
|
||||
@@ -1062,38 +1159,54 @@ error:
|
||||
* The command returns an sds string representing the result of the
|
||||
* operation. On error the first byte is a "-".
|
||||
*/
|
||||
char *sendSynchronousCommand(int fd, ...) {
|
||||
va_list ap;
|
||||
sds cmd = sdsempty();
|
||||
char *arg, buf[256];
|
||||
#define SYNC_CMD_READ (1<<0)
|
||||
#define SYNC_CMD_WRITE (1<<1)
|
||||
#define SYNC_CMD_FULL (SYNC_CMD_READ|SYNC_CMD_WRITE)
|
||||
char *sendSynchronousCommand(int flags, int fd, ...) {
|
||||
|
||||
/* Create the command to send to the master, we use simple inline
|
||||
* protocol for simplicity as currently we only send simple strings. */
|
||||
va_start(ap,fd);
|
||||
while(1) {
|
||||
arg = va_arg(ap, char*);
|
||||
if (arg == NULL) break;
|
||||
if (flags & SYNC_CMD_WRITE) {
|
||||
char *arg;
|
||||
va_list ap;
|
||||
sds cmd = sdsempty();
|
||||
va_start(ap,fd);
|
||||
|
||||
if (sdslen(cmd) != 0) cmd = sdscatlen(cmd," ",1);
|
||||
cmd = sdscat(cmd,arg);
|
||||
}
|
||||
cmd = sdscatlen(cmd,"\r\n",2);
|
||||
while(1) {
|
||||
arg = va_arg(ap, char*);
|
||||
if (arg == NULL) break;
|
||||
|
||||
/* Transfer command to the server. */
|
||||
if (syncWrite(fd,cmd,sdslen(cmd),server.repl_syncio_timeout*1000) == -1) {
|
||||
if (sdslen(cmd) != 0) cmd = sdscatlen(cmd," ",1);
|
||||
cmd = sdscat(cmd,arg);
|
||||
}
|
||||
cmd = sdscatlen(cmd,"\r\n",2);
|
||||
|
||||
/* Transfer command to the server. */
|
||||
if (syncWrite(fd,cmd,sdslen(cmd),server.repl_syncio_timeout*1000)
|
||||
== -1)
|
||||
{
|
||||
sdsfree(cmd);
|
||||
return sdscatprintf(sdsempty(),"-Writing to master: %s",
|
||||
strerror(errno));
|
||||
}
|
||||
sdsfree(cmd);
|
||||
return sdscatprintf(sdsempty(),"-Writing to master: %s",
|
||||
strerror(errno));
|
||||
va_end(ap);
|
||||
}
|
||||
sdsfree(cmd);
|
||||
|
||||
/* Read the reply from the server. */
|
||||
if (syncReadLine(fd,buf,sizeof(buf),server.repl_syncio_timeout*1000) == -1)
|
||||
{
|
||||
return sdscatprintf(sdsempty(),"-Reading from master: %s",
|
||||
strerror(errno));
|
||||
if (flags & SYNC_CMD_READ) {
|
||||
char buf[256];
|
||||
|
||||
if (syncReadLine(fd,buf,sizeof(buf),server.repl_syncio_timeout*1000)
|
||||
== -1)
|
||||
{
|
||||
return sdscatprintf(sdsempty(),"-Reading from master: %s",
|
||||
strerror(errno));
|
||||
}
|
||||
server.repl_transfer_lastio = server.unixtime;
|
||||
return sdsnew(buf);
|
||||
}
|
||||
return sdsnew(buf);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
/* Try a partial resynchronization with the master if we are about to reconnect.
|
||||
@@ -1110,6 +1223,19 @@ char *sendSynchronousCommand(int fd, ...) {
|
||||
* 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 succeded and we can continue.
|
||||
@@ -1118,35 +1244,68 @@ char *sendSynchronousCommand(int fd, ...) {
|
||||
* 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_ERR: There was an error writing the command to the socket.
|
||||
* PSYNC_WAIT_REPLY: Call again the function with read_reply set to 1.
|
||||
*
|
||||
* 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.repl_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_CONTINUE 0
|
||||
#define PSYNC_FULLRESYNC 1
|
||||
#define PSYNC_NOT_SUPPORTED 2
|
||||
int slaveTryPartialResynchronization(int fd) {
|
||||
#define PSYNC_WRITE_ERROR 0
|
||||
#define PSYNC_WAIT_REPLY 1
|
||||
#define PSYNC_CONTINUE 2
|
||||
#define PSYNC_FULLRESYNC 3
|
||||
#define PSYNC_NOT_SUPPORTED 4
|
||||
int slaveTryPartialResynchronization(int fd, int read_reply) {
|
||||
char *psync_runid;
|
||||
char psync_offset[32];
|
||||
sds reply;
|
||||
|
||||
/* Initially set repl_master_initial_offset to -1 to mark the current
|
||||
* master run_id 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.repl_master_initial_offset = -1;
|
||||
/* Writing half */
|
||||
if (!read_reply) {
|
||||
/* Initially set repl_master_initial_offset to -1 to mark the current
|
||||
* master run_id 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.repl_master_initial_offset = -1;
|
||||
|
||||
if (server.cached_master) {
|
||||
psync_runid = server.cached_master->replrunid;
|
||||
snprintf(psync_offset,sizeof(psync_offset),"%lld", server.cached_master->reploff+1);
|
||||
redisLog(REDIS_NOTICE,"Trying a partial resynchronization (request %s:%s).", psync_runid, psync_offset);
|
||||
} else {
|
||||
redisLog(REDIS_NOTICE,"Partial resynchronization not possible (no cached master)");
|
||||
psync_runid = "?";
|
||||
memcpy(psync_offset,"-1",3);
|
||||
if (server.cached_master) {
|
||||
psync_runid = server.cached_master->replrunid;
|
||||
snprintf(psync_offset,sizeof(psync_offset),"%lld", server.cached_master->reploff+1);
|
||||
redisLog(REDIS_NOTICE,"Trying a partial resynchronization (request %s:%s).", psync_runid, psync_offset);
|
||||
} else {
|
||||
redisLog(REDIS_NOTICE,"Partial resynchronization not possible (no cached master)");
|
||||
psync_runid = "?";
|
||||
memcpy(psync_offset,"-1",3);
|
||||
}
|
||||
|
||||
/* Issue the PSYNC command */
|
||||
reply = sendSynchronousCommand(SYNC_CMD_WRITE,fd,"PSYNC",psync_runid,psync_offset,NULL);
|
||||
if (reply != NULL) {
|
||||
redisLog(REDIS_WARNING,"Unable to send PSYNC to master: %s",reply);
|
||||
sdsfree(reply);
|
||||
aeDeleteFileEvent(server.el,fd,AE_READABLE);
|
||||
return PSYNC_WRITE_ERROR;
|
||||
}
|
||||
return PSYNC_WAIT_REPLY;
|
||||
}
|
||||
|
||||
/* Issue the PSYNC command */
|
||||
reply = sendSynchronousCommand(fd,"PSYNC",psync_runid,psync_offset,NULL);
|
||||
/* Reading half */
|
||||
reply = sendSynchronousCommand(SYNC_CMD_READ,fd,NULL);
|
||||
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;
|
||||
}
|
||||
|
||||
aeDeleteFileEvent(server.el,fd,AE_READABLE);
|
||||
|
||||
if (!strncmp(reply,"+FULLRESYNC",11)) {
|
||||
char *runid = NULL, *offset = NULL;
|
||||
@@ -1190,7 +1349,7 @@ int slaveTryPartialResynchronization(int fd) {
|
||||
return PSYNC_CONTINUE;
|
||||
}
|
||||
|
||||
/* If we reach this point we receied either an error since the master does
|
||||
/* If we reach this point we received either an error since the master does
|
||||
* not understand PSYNC, or an unexpected reply from the master.
|
||||
* Return PSYNC_NOT_SUPPORTED to the caller in both cases. */
|
||||
|
||||
@@ -1209,7 +1368,7 @@ int slaveTryPartialResynchronization(int fd) {
|
||||
}
|
||||
|
||||
void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) {
|
||||
char tmpfile[256], *err;
|
||||
char tmpfile[256], *err = NULL;
|
||||
int dfd, maxtries = 5;
|
||||
int sockerr = 0, psync_result;
|
||||
socklen_t errlen = sizeof(sockerr);
|
||||
@@ -1228,16 +1387,12 @@ void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) {
|
||||
if (getsockopt(fd, SOL_SOCKET, SO_ERROR, &sockerr, &errlen) == -1)
|
||||
sockerr = errno;
|
||||
if (sockerr) {
|
||||
aeDeleteFileEvent(server.el,fd,AE_READABLE|AE_WRITABLE);
|
||||
redisLog(REDIS_WARNING,"Error condition on socket for SYNC: %s",
|
||||
strerror(sockerr));
|
||||
goto error;
|
||||
}
|
||||
|
||||
/* If we were connecting, it's time to send a non blocking PING, we want to
|
||||
* make sure the master is able to reply before going into the actual
|
||||
* replication process where we have long timeouts in the order of
|
||||
* seconds (in the meantime the slave would block). */
|
||||
/* Send a PING to check the master is able to reply without errors. */
|
||||
if (server.repl_state == REDIS_REPL_CONNECTING) {
|
||||
redisLog(REDIS_NOTICE,"Non blocking connect for SYNC fired the event.");
|
||||
/* Delete the writable event so that the readable event remains
|
||||
@@ -1246,70 +1401,109 @@ void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) {
|
||||
server.repl_state = REDIS_REPL_RECEIVE_PONG;
|
||||
/* Send the PING, don't check for errors at all, we have the timeout
|
||||
* that will take care about this. */
|
||||
syncWrite(fd,"PING\r\n",6,100);
|
||||
err = sendSynchronousCommand(SYNC_CMD_WRITE,fd,"PING",NULL);
|
||||
if (err) goto write_error;
|
||||
return;
|
||||
}
|
||||
|
||||
/* Receive the PONG command. */
|
||||
if (server.repl_state == REDIS_REPL_RECEIVE_PONG) {
|
||||
char buf[1024];
|
||||
|
||||
/* Delete the readable event, we no longer need it now that there is
|
||||
* the PING reply to read. */
|
||||
aeDeleteFileEvent(server.el,fd,AE_READABLE);
|
||||
|
||||
/* Read the reply with explicit timeout. */
|
||||
buf[0] = '\0';
|
||||
if (syncReadLine(fd,buf,sizeof(buf),
|
||||
server.repl_syncio_timeout*1000) == -1)
|
||||
{
|
||||
redisLog(REDIS_WARNING,
|
||||
"I/O error reading PING reply from master: %s",
|
||||
strerror(errno));
|
||||
goto error;
|
||||
}
|
||||
err = sendSynchronousCommand(SYNC_CMD_READ,fd,NULL);
|
||||
|
||||
/* 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 (buf[0] != '+' &&
|
||||
strncmp(buf,"-NOAUTH",7) != 0 &&
|
||||
strncmp(buf,"-ERR operation not permitted",28) != 0)
|
||||
if (err[0] != '+' &&
|
||||
strncmp(err,"-NOAUTH",7) != 0 &&
|
||||
strncmp(err,"-ERR operation not permitted",28) != 0)
|
||||
{
|
||||
redisLog(REDIS_WARNING,"Error reply to PING from master: '%s'",buf);
|
||||
redisLog(REDIS_WARNING,"Error reply to PING from master: '%s'",err);
|
||||
sdsfree(err);
|
||||
goto error;
|
||||
} else {
|
||||
redisLog(REDIS_NOTICE,
|
||||
"Master replied to PING, replication can continue...");
|
||||
}
|
||||
sdsfree(err);
|
||||
server.repl_state = REDIS_REPL_SEND_AUTH;
|
||||
}
|
||||
|
||||
/* AUTH with the master if required. */
|
||||
if(server.masterauth) {
|
||||
err = sendSynchronousCommand(fd,"AUTH",server.masterauth,NULL);
|
||||
if (server.repl_state == REDIS_REPL_SEND_AUTH) {
|
||||
if (server.masterauth) {
|
||||
err = sendSynchronousCommand(SYNC_CMD_WRITE,fd,"AUTH",server.masterauth,NULL);
|
||||
if (err) goto write_error;
|
||||
server.repl_state = REDIS_REPL_RECEIVE_AUTH;
|
||||
return;
|
||||
} else {
|
||||
server.repl_state = REDIS_REPL_SEND_PORT;
|
||||
}
|
||||
}
|
||||
|
||||
/* Receive AUTH reply. */
|
||||
if (server.repl_state == REDIS_REPL_RECEIVE_AUTH) {
|
||||
err = sendSynchronousCommand(SYNC_CMD_READ,fd,NULL);
|
||||
if (err[0] == '-') {
|
||||
redisLog(REDIS_WARNING,"Unable to AUTH to MASTER: %s",err);
|
||||
sdsfree(err);
|
||||
goto error;
|
||||
}
|
||||
sdsfree(err);
|
||||
server.repl_state = REDIS_REPL_SEND_PORT;
|
||||
}
|
||||
|
||||
/* Set the slave port, so that Master's INFO command can list the
|
||||
* slave listening port correctly. */
|
||||
{
|
||||
if (server.repl_state == REDIS_REPL_SEND_PORT) {
|
||||
sds port = sdsfromlonglong(server.port);
|
||||
err = sendSynchronousCommand(fd,"REPLCONF","listening-port",port,
|
||||
NULL);
|
||||
err = sendSynchronousCommand(SYNC_CMD_WRITE,fd,"REPLCONF",
|
||||
"listening-port",port, NULL);
|
||||
sdsfree(port);
|
||||
if (err) goto write_error;
|
||||
sdsfree(err);
|
||||
server.repl_state = REDIS_REPL_RECEIVE_PORT;
|
||||
return;
|
||||
}
|
||||
|
||||
/* Receive REPLCONF listening-port reply. */
|
||||
if (server.repl_state == REDIS_REPL_RECEIVE_PORT) {
|
||||
err = sendSynchronousCommand(SYNC_CMD_READ,fd,NULL);
|
||||
/* Ignore the error if any, not all the Redis versions support
|
||||
* REPLCONF listening-port. */
|
||||
if (err[0] == '-') {
|
||||
redisLog(REDIS_NOTICE,"(Non critical) Master does not understand REPLCONF listening-port: %s", err);
|
||||
redisLog(REDIS_NOTICE,"(Non critical) Master does not understand "
|
||||
"REPLCONF listening-port: %s", err);
|
||||
}
|
||||
sdsfree(err);
|
||||
server.repl_state = REDIS_REPL_SEND_CAPA;
|
||||
}
|
||||
|
||||
/* Inform the master of our capabilities. While we currently send
|
||||
* just one capability, it is possible to chain new capabilities here
|
||||
* in the form of REPLCONF capa X capa Y capa Z ...
|
||||
* The master will ignore capabilities it does not understand. */
|
||||
if (server.repl_state == REDIS_REPL_SEND_CAPA) {
|
||||
err = sendSynchronousCommand(SYNC_CMD_WRITE,fd,"REPLCONF",
|
||||
"capa","eof",NULL);
|
||||
if (err) goto write_error;
|
||||
sdsfree(err);
|
||||
server.repl_state = REDIS_REPL_RECEIVE_CAPA;
|
||||
return;
|
||||
}
|
||||
|
||||
/* Receive CAPA reply. */
|
||||
if (server.repl_state == REDIS_REPL_RECEIVE_CAPA) {
|
||||
err = sendSynchronousCommand(SYNC_CMD_READ,fd,NULL);
|
||||
/* Ignore the error if any, not all the Redis versions support
|
||||
* REPLCONF capa. */
|
||||
if (err[0] == '-') {
|
||||
redisLog(REDIS_NOTICE,"(Non critical) Master does not understand "
|
||||
"REPLCONF capa: %s", err);
|
||||
}
|
||||
sdsfree(err);
|
||||
server.repl_state = REDIS_REPL_SEND_PSYNC;
|
||||
}
|
||||
|
||||
/* Try a partial resynchonization. If we don't have a cached master
|
||||
@@ -1317,12 +1511,41 @@ void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) {
|
||||
* to start a full resynchronization so that we get the master run id
|
||||
* and the global offset, to try a partial resync at the next
|
||||
* reconnection attempt. */
|
||||
psync_result = slaveTryPartialResynchronization(fd);
|
||||
if (server.repl_state == REDIS_REPL_SEND_PSYNC) {
|
||||
if (slaveTryPartialResynchronization(fd,0) == PSYNC_WRITE_ERROR) {
|
||||
err = sdsnew("Write error sending the PSYNC command.");
|
||||
goto write_error;
|
||||
}
|
||||
server.repl_state = REDIS_REPL_RECEIVE_PSYNC;
|
||||
return;
|
||||
}
|
||||
|
||||
/* If reached this point, we should be in REDIS_REPL_RECEIVE_PSYNC. */
|
||||
if (server.repl_state != REDIS_REPL_RECEIVE_PSYNC) {
|
||||
redisLog(REDIS_WARNING,"syncWithMaster(): state machine error, "
|
||||
"state should be RECEIVE_PSYNC but is %d",
|
||||
server.repl_state);
|
||||
goto error;
|
||||
}
|
||||
|
||||
psync_result = slaveTryPartialResynchronization(fd,1);
|
||||
if (psync_result == PSYNC_WAIT_REPLY) return; /* Try again later... */
|
||||
|
||||
/* 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) {
|
||||
redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: Master accepted a Partial Resynchronization.");
|
||||
return;
|
||||
}
|
||||
|
||||
/* PSYNC failed or is not supported: we want our slaves to resync with us
|
||||
* as well, if we have any (chained replication case). The mater may
|
||||
* transfer us an entirely different data set and we have no way to
|
||||
* incrementally feed our slaves after that. */
|
||||
disconnectSlaves(); /* Force our slaves to resync with us as well. */
|
||||
freeReplicationBacklog(); /* Don't allow our chained slaves to PSYNC. */
|
||||
|
||||
/* Fall back to SYNC if needed. Otherwise psync_result == PSYNC_FULLRESYNC
|
||||
* and the server.repl_master_runid and repl_master_initial_offset are
|
||||
* already populated. */
|
||||
@@ -1368,10 +1591,16 @@ void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) {
|
||||
return;
|
||||
|
||||
error:
|
||||
aeDeleteFileEvent(server.el,fd,AE_READABLE|AE_WRITABLE);
|
||||
close(fd);
|
||||
server.repl_transfer_s = -1;
|
||||
server.repl_state = REDIS_REPL_CONNECT;
|
||||
return;
|
||||
|
||||
write_error: /* Handle sendSynchronousCommand(SYNC_CMD_WRITE) errors. */
|
||||
redisLog(REDIS_WARNING,"Sending command to master in replication handshake: %s", err);
|
||||
sdsfree(err);
|
||||
goto error;
|
||||
}
|
||||
|
||||
int connectWithMaster(void) {
|
||||
@@ -1405,7 +1634,7 @@ void undoConnectWithMaster(void) {
|
||||
int fd = server.repl_transfer_s;
|
||||
|
||||
redisAssert(server.repl_state == REDIS_REPL_CONNECTING ||
|
||||
server.repl_state == REDIS_REPL_RECEIVE_PONG);
|
||||
slaveIsInHandshakeState());
|
||||
aeDeleteFileEvent(server.el,fd,AE_READABLE|AE_WRITABLE);
|
||||
close(fd);
|
||||
server.repl_transfer_s = -1;
|
||||
@@ -1424,7 +1653,7 @@ int cancelReplicationHandshake(void) {
|
||||
if (server.repl_state == REDIS_REPL_TRANSFER) {
|
||||
replicationAbortSyncTransfer();
|
||||
} else if (server.repl_state == REDIS_REPL_CONNECTING ||
|
||||
server.repl_state == REDIS_REPL_RECEIVE_PONG)
|
||||
slaveIsInHandshakeState())
|
||||
{
|
||||
undoConnectWithMaster();
|
||||
} else {
|
||||
@@ -1470,6 +1699,17 @@ void replicationUnsetMaster(void) {
|
||||
server.repl_state = REDIS_REPL_NONE;
|
||||
}
|
||||
|
||||
/* This function is called when the slave lose the connection with the
|
||||
* master into an unexpected way. */
|
||||
void replicationHandleMasterDisconnection(void) {
|
||||
server.master = NULL;
|
||||
server.repl_state = REDIS_REPL_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. */
|
||||
}
|
||||
|
||||
void slaveofCommand(redisClient *c) {
|
||||
/* SLAVEOF is not allowed in cluster mode as replication is automatically
|
||||
* configured using the current address of the master node. */
|
||||
@@ -1484,7 +1724,10 @@ void slaveofCommand(redisClient *c) {
|
||||
!strcasecmp(c->argv[2]->ptr,"one")) {
|
||||
if (server.masterhost) {
|
||||
replicationUnsetMaster();
|
||||
redisLog(REDIS_NOTICE,"MASTER MODE enabled (user request)");
|
||||
sds client = catClientInfoString(sdsempty(),c);
|
||||
redisLog(REDIS_NOTICE,
|
||||
"MASTER MODE enabled (user request from '%s')",client);
|
||||
sdsfree(client);
|
||||
}
|
||||
} else {
|
||||
long port;
|
||||
@@ -1502,8 +1745,10 @@ void slaveofCommand(redisClient *c) {
|
||||
/* There was no previous master or the user specified a different one,
|
||||
* we can continue. */
|
||||
replicationSetMaster(c->argv[1]->ptr, port);
|
||||
redisLog(REDIS_NOTICE,"SLAVE OF %s:%d enabled (user request)",
|
||||
server.masterhost, server.masterport);
|
||||
sds client = catClientInfoString(sdsempty(),c);
|
||||
redisLog(REDIS_NOTICE,"SLAVE OF %s:%d enabled (user request from '%s')",
|
||||
server.masterhost, server.masterport, client);
|
||||
sdsfree(client);
|
||||
}
|
||||
addReply(c,shared.ok);
|
||||
}
|
||||
@@ -1543,14 +1788,17 @@ void roleCommand(redisClient *c) {
|
||||
addReplyBulkCBuffer(c,"slave",5);
|
||||
addReplyBulkCString(c,server.masterhost);
|
||||
addReplyLongLong(c,server.masterport);
|
||||
switch(server.repl_state) {
|
||||
case REDIS_REPL_NONE: slavestate = "none"; break;
|
||||
case REDIS_REPL_CONNECT: slavestate = "connect"; break;
|
||||
case REDIS_REPL_CONNECTING: slavestate = "connecting"; break;
|
||||
case REDIS_REPL_RECEIVE_PONG: /* see next */
|
||||
case REDIS_REPL_TRANSFER: slavestate = "sync"; break;
|
||||
case REDIS_REPL_CONNECTED: slavestate = "connected"; break;
|
||||
default: slavestate = "unknown"; break;
|
||||
if (slaveIsInHandshakeState()) {
|
||||
slavestate = "handshake";
|
||||
} else {
|
||||
switch(server.repl_state) {
|
||||
case REDIS_REPL_NONE: slavestate = "none"; break;
|
||||
case REDIS_REPL_CONNECT: slavestate = "connect"; break;
|
||||
case REDIS_REPL_CONNECTING: slavestate = "connecting"; break;
|
||||
case REDIS_REPL_TRANSFER: slavestate = "sync"; break;
|
||||
case REDIS_REPL_CONNECTED: slavestate = "connected"; break;
|
||||
default: slavestate = "unknown"; break;
|
||||
}
|
||||
}
|
||||
addReplyBulkCString(c,slavestate);
|
||||
addReplyLongLong(c,server.master ? server.master->reploff : -1);
|
||||
@@ -1936,11 +2184,13 @@ long long replicationGetSlaveOffset(void) {
|
||||
|
||||
/* Replication cron function, called 1 time per second. */
|
||||
void replicationCron(void) {
|
||||
static long long replication_cron_loops = 0;
|
||||
|
||||
/* Non blocking connection timeout? */
|
||||
if (server.masterhost &&
|
||||
(server.repl_state == REDIS_REPL_CONNECTING ||
|
||||
server.repl_state == REDIS_REPL_RECEIVE_PONG) &&
|
||||
(time(NULL)-server.repl_transfer_lastio) > server.repl_timeout)
|
||||
slaveIsInHandshakeState()) &&
|
||||
(time(NULL)-server.repl_transfer_lastio) > server.repl_timeout)
|
||||
{
|
||||
redisLog(REDIS_WARNING,"Timeout connecting to the MASTER...");
|
||||
undoConnectWithMaster();
|
||||
@@ -1982,31 +2232,34 @@ void replicationCron(void) {
|
||||
* 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. */
|
||||
if (!(server.cronloops % (server.repl_ping_slave_period * server.hz))) {
|
||||
listIter li;
|
||||
listNode *ln;
|
||||
robj *ping_argv[1];
|
||||
listIter li;
|
||||
listNode *ln;
|
||||
robj *ping_argv[1];
|
||||
|
||||
/* First, send PING */
|
||||
/* First, send PING according to ping_slave_period. */
|
||||
if ((replication_cron_loops % server.repl_ping_slave_period) == 0) {
|
||||
ping_argv[0] = createStringObject("PING",4);
|
||||
replicationFeedSlaves(server.slaves, server.slaveseldb, ping_argv, 1);
|
||||
replicationFeedSlaves(server.slaves, server.slaveseldb,
|
||||
ping_argv, 1);
|
||||
decrRefCount(ping_argv[0]);
|
||||
}
|
||||
|
||||
/* Second, send a newline to all the slaves in pre-synchronization
|
||||
* stage, that is, slaves waiting for the master to create the RDB file.
|
||||
* The newline will be ignored by the slave but will refresh the
|
||||
* last-io timer preventing a timeout. */
|
||||
listRewind(server.slaves,&li);
|
||||
while((ln = listNext(&li))) {
|
||||
redisClient *slave = ln->value;
|
||||
/* Second, send a newline to all the slaves in pre-synchronization
|
||||
* stage, that is, slaves waiting for the master to create the RDB file.
|
||||
* The newline will be ignored by the slave but will refresh the
|
||||
* last-io 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))) {
|
||||
redisClient *slave = ln->value;
|
||||
|
||||
if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START ||
|
||||
(slave->replstate == REDIS_REPL_WAIT_BGSAVE_END &&
|
||||
server.rdb_child_type != REDIS_RDB_CHILD_TYPE_SOCKET))
|
||||
{
|
||||
if (write(slave->fd, "\n", 1) == -1) {
|
||||
/* Don't worry, it's just a ping. */
|
||||
}
|
||||
if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START ||
|
||||
(slave->replstate == REDIS_REPL_WAIT_BGSAVE_END &&
|
||||
server.rdb_child_type != REDIS_RDB_CHILD_TYPE_SOCKET))
|
||||
{
|
||||
if (write(slave->fd, "\n", 1) == -1) {
|
||||
/* Don't worry, it's just a ping. */
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -2067,6 +2320,7 @@ void replicationCron(void) {
|
||||
if (server.rdb_child_pid == -1 && server.aof_child_pid == -1) {
|
||||
time_t idle, max_idle = 0;
|
||||
int slaves_waiting = 0;
|
||||
int mincapa = -1;
|
||||
listNode *ln;
|
||||
listIter li;
|
||||
|
||||
@@ -2077,28 +2331,19 @@ void replicationCron(void) {
|
||||
idle = server.unixtime - slave->lastinteraction;
|
||||
if (idle > max_idle) max_idle = idle;
|
||||
slaves_waiting++;
|
||||
mincapa = (mincapa == -1) ? slave->slave_capa :
|
||||
(mincapa & slave->slave_capa);
|
||||
}
|
||||
}
|
||||
|
||||
if (slaves_waiting && max_idle > server.repl_diskless_sync_delay) {
|
||||
/* Start a BGSAVE. Usually with socket target, or with disk target
|
||||
* if there was a recent socket -> disk config change. */
|
||||
if (startBgsaveForReplication() == REDIS_OK) {
|
||||
/* It started! We need to change the state of slaves
|
||||
* from WAIT_BGSAVE_START to WAIT_BGSAVE_END in case
|
||||
* the current target is disk. Otherwise it was already done
|
||||
* by rdbSaveToSlavesSockets() which is called by
|
||||
* startBgsaveForReplication(). */
|
||||
listRewind(server.slaves,&li);
|
||||
while((ln = listNext(&li))) {
|
||||
redisClient *slave = ln->value;
|
||||
if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START)
|
||||
slave->replstate = REDIS_REPL_WAIT_BGSAVE_END;
|
||||
}
|
||||
}
|
||||
startBgsaveForReplication(mincapa);
|
||||
}
|
||||
}
|
||||
|
||||
/* Refresh the number of slaves with lag <= min-slaves-max-lag. */
|
||||
refreshGoodSlavesCount();
|
||||
replication_cron_loops++; /* Incremented with frequency 1 HZ. */
|
||||
}
|
||||
|
||||
+8
-3
@@ -1157,7 +1157,9 @@ void sentinelDelFlagsToDictOfRedisInstances(dict *instances, int flags) {
|
||||
* 1) Remove all slaves.
|
||||
* 2) Remove all sentinels.
|
||||
* 3) Remove most of the flags resulting from runtime operations.
|
||||
* 4) Reset timers to their default value.
|
||||
* 4) Reset timers to their default value. For example after a reset it will be
|
||||
* possible to failover again the same master ASAP, without waiting the
|
||||
* failover timeout delay.
|
||||
* 5) In the process of doing this undo the failover if in progress.
|
||||
* 6) Disconnect the connections with the master (will reconnect automatically).
|
||||
*/
|
||||
@@ -1180,7 +1182,7 @@ void sentinelResetMaster(sentinelRedisInstance *ri, int flags) {
|
||||
}
|
||||
ri->failover_state = SENTINEL_FAILOVER_STATE_NONE;
|
||||
ri->failover_state_change_time = 0;
|
||||
ri->failover_start_time = 0;
|
||||
ri->failover_start_time = 0; /* We can failover again ASAP. */
|
||||
ri->promoted_slave = NULL;
|
||||
sdsfree(ri->runid);
|
||||
sdsfree(ri->slave_master_host);
|
||||
@@ -1544,7 +1546,7 @@ void rewriteConfigSentinelOption(struct rewriteConfigState *state) {
|
||||
slave_addr = master->addr;
|
||||
line = sdscatprintf(sdsempty(),
|
||||
"sentinel known-slave %s %s %d",
|
||||
master->name, ri->addr->ip, ri->addr->port);
|
||||
master->name, slave_addr->ip, slave_addr->port);
|
||||
rewriteConfigRewriteLine(state,"sentinel",line,1);
|
||||
}
|
||||
dictReleaseIterator(di2);
|
||||
@@ -2790,6 +2792,7 @@ void sentinelCommand(redisClient *c) {
|
||||
addReply(c,shared.ok);
|
||||
}
|
||||
} else if (!strcasecmp(c->argv[1]->ptr,"flushconfig")) {
|
||||
if (c->argc != 2) goto numargserr;
|
||||
sentinelFlushConfig();
|
||||
addReply(c,shared.ok);
|
||||
return;
|
||||
@@ -2797,6 +2800,7 @@ void sentinelCommand(redisClient *c) {
|
||||
/* SENTINEL REMOVE <name> */
|
||||
sentinelRedisInstance *ri;
|
||||
|
||||
if (c->argc != 3) goto numargserr;
|
||||
if ((ri = sentinelGetMasterByNameOrReplyError(c,c->argv[2]))
|
||||
== NULL) return;
|
||||
sentinelEvent(REDIS_WARNING,"-monitor",ri,"%@");
|
||||
@@ -2808,6 +2812,7 @@ void sentinelCommand(redisClient *c) {
|
||||
sentinelRedisInstance *ri;
|
||||
int usable;
|
||||
|
||||
if (c->argc != 3) goto numargserr;
|
||||
if ((ri = sentinelGetMasterByNameOrReplyError(c,c->argv[2]))
|
||||
== NULL) return;
|
||||
int result = sentinelIsQuorumReachable(ri,&usable);
|
||||
|
||||
+1
-1
@@ -1 +1 @@
|
||||
#define REDIS_VERSION "3.0.3"
|
||||
#define REDIS_VERSION "3.0.5"
|
||||
|
||||
+13
-13
@@ -7,19 +7,19 @@ start_server [list overrides [list "dir" $server_path "dbfilename" "encodings.rd
|
||||
test "RDB encoding loading test" {
|
||||
r select 0
|
||||
csvdump r
|
||||
} {"compressible","string","aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
|
||||
"hash","hash","a","1","aa","10","aaa","100","b","2","bb","20","bbb","200","c","3","cc","30","ccc","300","ddd","400","eee","5000000000",
|
||||
"hash_zipped","hash","a","1","b","2","c","3",
|
||||
"list","list","1","2","3","a","b","c","100000","6000000000","1","2","3","a","b","c","100000","6000000000","1","2","3","a","b","c","100000","6000000000",
|
||||
"list_zipped","list","1","2","3","a","b","c","100000","6000000000",
|
||||
"number","string","10"
|
||||
"set","set","1","100000","2","3","6000000000","a","b","c",
|
||||
"set_zipped_1","set","1","2","3","4",
|
||||
"set_zipped_2","set","100000","200000","300000","400000",
|
||||
"set_zipped_3","set","1000000000","2000000000","3000000000","4000000000","5000000000","6000000000",
|
||||
"string","string","Hello World"
|
||||
"zset","zset","a","1","b","2","c","3","aa","10","bb","20","cc","30","aaa","100","bbb","200","ccc","300","aaaa","1000","cccc","123456789","bbbb","5000000000",
|
||||
"zset_zipped","zset","a","1","b","2","c","3",
|
||||
} {"0","compressible","string","aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
|
||||
"0","hash","hash","a","1","aa","10","aaa","100","b","2","bb","20","bbb","200","c","3","cc","30","ccc","300","ddd","400","eee","5000000000",
|
||||
"0","hash_zipped","hash","a","1","b","2","c","3",
|
||||
"0","list","list","1","2","3","a","b","c","100000","6000000000","1","2","3","a","b","c","100000","6000000000","1","2","3","a","b","c","100000","6000000000",
|
||||
"0","list_zipped","list","1","2","3","a","b","c","100000","6000000000",
|
||||
"0","number","string","10"
|
||||
"0","set","set","1","100000","2","3","6000000000","a","b","c",
|
||||
"0","set_zipped_1","set","1","2","3","4",
|
||||
"0","set_zipped_2","set","100000","200000","300000","400000",
|
||||
"0","set_zipped_3","set","1000000000","2000000000","3000000000","4000000000","5000000000","6000000000",
|
||||
"0","string","string","Hello World"
|
||||
"0","zset","zset","a","1","b","2","c","3","aa","10","bb","20","cc","30","aaa","100","bbb","200","ccc","300","aaaa","1000","cccc","123456789","bbbb","5000000000",
|
||||
"0","zset_zipped","zset","a","1","b","2","c","3",
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -13,7 +13,11 @@ proc stop_bg_complex_data {handle} {
|
||||
#
|
||||
# You can specifiy backlog size, ttl, delay before reconnection, test duration
|
||||
# in seconds, and an additional condition to verify at the end.
|
||||
proc test_psync {descr duration backlog_size backlog_ttl delay cond} {
|
||||
#
|
||||
# If reconnect is > 0, the test actually try to break the connection and
|
||||
# reconnect with the master, otherwise just the initial synchronization is
|
||||
# checked for consistency.
|
||||
proc test_psync {descr duration backlog_size backlog_ttl delay cond diskless reconnect} {
|
||||
start_server {tags {"repl"}} {
|
||||
start_server {} {
|
||||
|
||||
@@ -24,6 +28,8 @@ proc test_psync {descr duration backlog_size backlog_ttl delay cond} {
|
||||
|
||||
$master config set repl-backlog-size $backlog_size
|
||||
$master config set repl-backlog-ttl $backlog_ttl
|
||||
$master config set repl-diskless-sync $diskless
|
||||
$master config set repl-diskless-sync-delay 1
|
||||
|
||||
set load_handle0 [start_bg_complex_data $master_host $master_port 9 100000]
|
||||
set load_handle1 [start_bg_complex_data $master_host $master_port 11 100000]
|
||||
@@ -48,22 +54,24 @@ proc test_psync {descr duration backlog_size backlog_ttl delay cond} {
|
||||
}
|
||||
}
|
||||
|
||||
test "Test replication partial resync: $descr" {
|
||||
test "Test replication partial resync: $descr (diskless: $diskless, reconnect: $reconnect)" {
|
||||
# Now while the clients are writing data, break the maste-slave
|
||||
# link multiple times.
|
||||
for {set j 0} {$j < $duration*10} {incr j} {
|
||||
after 100
|
||||
# catch {puts "MASTER [$master dbsize] keys, SLAVE [$slave dbsize] keys"}
|
||||
if ($reconnect) {
|
||||
for {set j 0} {$j < $duration*10} {incr j} {
|
||||
after 100
|
||||
# catch {puts "MASTER [$master dbsize] keys, SLAVE [$slave dbsize] keys"}
|
||||
|
||||
if {($j % 20) == 0} {
|
||||
catch {
|
||||
if {$delay} {
|
||||
$slave multi
|
||||
$slave client kill $master_host:$master_port
|
||||
$slave debug sleep $delay
|
||||
$slave exec
|
||||
} else {
|
||||
$slave client kill $master_host:$master_port
|
||||
if {($j % 20) == 0} {
|
||||
catch {
|
||||
if {$delay} {
|
||||
$slave multi
|
||||
$slave client kill $master_host:$master_port
|
||||
$slave debug sleep $delay
|
||||
$slave exec
|
||||
} else {
|
||||
$slave client kill $master_host:$master_port
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -98,18 +106,23 @@ proc test_psync {descr duration backlog_size backlog_ttl delay cond} {
|
||||
}
|
||||
}
|
||||
|
||||
test_psync {ok psync} 6 1000000 3600 0 {
|
||||
assert {[s -1 sync_partial_ok] > 0}
|
||||
}
|
||||
foreach diskless {no yes} {
|
||||
test_psync {no reconnection, just sync} 6 1000000 3600 0 {
|
||||
} $diskless 0
|
||||
|
||||
test_psync {no backlog} 6 100 3600 0.5 {
|
||||
assert {[s -1 sync_partial_err] > 0}
|
||||
}
|
||||
test_psync {ok psync} 6 1000000 3600 0 {
|
||||
assert {[s -1 sync_partial_ok] > 0}
|
||||
} $diskless 1
|
||||
|
||||
test_psync {ok after delay} 3 100000000 3600 3 {
|
||||
assert {[s -1 sync_partial_ok] > 0}
|
||||
}
|
||||
test_psync {no backlog} 6 100 3600 0.5 {
|
||||
assert {[s -1 sync_partial_err] > 0}
|
||||
} $diskless 1
|
||||
|
||||
test_psync {backlog expired} 3 100000000 1 3 {
|
||||
assert {[s -1 sync_partial_err] > 0}
|
||||
test_psync {ok after delay} 3 100000000 3600 3 {
|
||||
assert {[s -1 sync_partial_ok] > 0}
|
||||
} $diskless 1
|
||||
|
||||
test_psync {backlog expired} 3 100000000 1 3 {
|
||||
assert {[s -1 sync_partial_err] > 0}
|
||||
} $diskless 1
|
||||
}
|
||||
|
||||
@@ -1,3 +1,56 @@
|
||||
proc log_file_matches {log pattern} {
|
||||
set fp [open $log r]
|
||||
set content [read $fp]
|
||||
close $fp
|
||||
string match $pattern $content
|
||||
}
|
||||
|
||||
start_server {tags {"repl"}} {
|
||||
set slave [srv 0 client]
|
||||
set slave_host [srv 0 host]
|
||||
set slave_port [srv 0 port]
|
||||
set slave_log [srv 0 stdout]
|
||||
start_server {} {
|
||||
set master [srv 0 client]
|
||||
set master_host [srv 0 host]
|
||||
set master_port [srv 0 port]
|
||||
|
||||
# Configure the master in order to hang waiting for the BGSAVE
|
||||
# operation, so that the slave remains in the handshake state.
|
||||
$master config set repl-diskless-sync yes
|
||||
$master config set repl-diskless-sync-delay 1000
|
||||
|
||||
# Use a short replication timeout on the slave, so that if there
|
||||
# are no bugs the timeout is triggered in a reasonable amount
|
||||
# of time.
|
||||
$slave config set repl-timeout 5
|
||||
|
||||
# Start the replication process...
|
||||
$slave slaveof $master_host $master_port
|
||||
|
||||
test {Slave enters handshake} {
|
||||
wait_for_condition 50 1000 {
|
||||
[string match *handshake* [$slave role]]
|
||||
} else {
|
||||
fail "Slave does not enter handshake state"
|
||||
}
|
||||
}
|
||||
|
||||
# But make the master unable to send
|
||||
# the periodic newlines to refresh the connection. The slave
|
||||
# should detect the timeout.
|
||||
$master debug sleep 10
|
||||
|
||||
test {Slave is able to detect timeout during handshake} {
|
||||
wait_for_condition 50 1000 {
|
||||
[log_file_matches $slave_log "*Timeout connecting to the MASTER*"]
|
||||
} else {
|
||||
fail "Slave is not able to detect timeout"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
start_server {tags {"repl"}} {
|
||||
set A [srv 0 client]
|
||||
set A_host [srv 0 host]
|
||||
|
||||
+36
-32
@@ -262,46 +262,50 @@ proc formatCommand {args} {
|
||||
|
||||
proc csvdump r {
|
||||
set o {}
|
||||
foreach k [lsort [{*}$r keys *]] {
|
||||
set type [{*}$r type $k]
|
||||
append o [csvstring $k] , [csvstring $type] ,
|
||||
switch $type {
|
||||
string {
|
||||
append o [csvstring [{*}$r get $k]] "\n"
|
||||
}
|
||||
list {
|
||||
foreach e [{*}$r lrange $k 0 -1] {
|
||||
append o [csvstring $e] ,
|
||||
for {set db 0} {$db < 16} {incr db} {
|
||||
{*}$r select $db
|
||||
foreach k [lsort [{*}$r keys *]] {
|
||||
set type [{*}$r type $k]
|
||||
append o [csvstring $db] , [csvstring $k] , [csvstring $type] ,
|
||||
switch $type {
|
||||
string {
|
||||
append o [csvstring [{*}$r get $k]] "\n"
|
||||
}
|
||||
append o "\n"
|
||||
}
|
||||
set {
|
||||
foreach e [lsort [{*}$r smembers $k]] {
|
||||
append o [csvstring $e] ,
|
||||
list {
|
||||
foreach e [{*}$r lrange $k 0 -1] {
|
||||
append o [csvstring $e] ,
|
||||
}
|
||||
append o "\n"
|
||||
}
|
||||
append o "\n"
|
||||
}
|
||||
zset {
|
||||
foreach e [{*}$r zrange $k 0 -1 withscores] {
|
||||
append o [csvstring $e] ,
|
||||
set {
|
||||
foreach e [lsort [{*}$r smembers $k]] {
|
||||
append o [csvstring $e] ,
|
||||
}
|
||||
append o "\n"
|
||||
}
|
||||
append o "\n"
|
||||
}
|
||||
hash {
|
||||
set fields [{*}$r hgetall $k]
|
||||
set newfields {}
|
||||
foreach {k v} $fields {
|
||||
lappend newfields [list $k $v]
|
||||
zset {
|
||||
foreach e [{*}$r zrange $k 0 -1 withscores] {
|
||||
append o [csvstring $e] ,
|
||||
}
|
||||
append o "\n"
|
||||
}
|
||||
set fields [lsort -index 0 $newfields]
|
||||
foreach kv $fields {
|
||||
append o [csvstring [lindex $kv 0]] ,
|
||||
append o [csvstring [lindex $kv 1]] ,
|
||||
hash {
|
||||
set fields [{*}$r hgetall $k]
|
||||
set newfields {}
|
||||
foreach {k v} $fields {
|
||||
lappend newfields [list $k $v]
|
||||
}
|
||||
set fields [lsort -index 0 $newfields]
|
||||
foreach kv $fields {
|
||||
append o [csvstring [lindex $kv 0]] ,
|
||||
append o [csvstring [lindex $kv 1]] ,
|
||||
}
|
||||
append o "\n"
|
||||
}
|
||||
append o "\n"
|
||||
}
|
||||
}
|
||||
}
|
||||
{*}$r select 9
|
||||
return $o
|
||||
}
|
||||
|
||||
|
||||
@@ -433,6 +433,32 @@ start_server {tags {"basic"}} {
|
||||
set e
|
||||
} {*ERR*index out of range}
|
||||
|
||||
test {MOVE can move key expire metadata as well} {
|
||||
r select 10
|
||||
r flushdb
|
||||
r select 9
|
||||
r set mykey foo ex 100
|
||||
r move mykey 10
|
||||
assert {[r ttl mykey] == -2}
|
||||
r select 10
|
||||
assert {[r ttl mykey] > 0 && [r ttl mykey] <= 100}
|
||||
assert {[r get mykey] eq "foo"}
|
||||
r select 9
|
||||
}
|
||||
|
||||
test {MOVE does not create an expire if it does not exist} {
|
||||
r select 10
|
||||
r flushdb
|
||||
r select 9
|
||||
r set mykey foo
|
||||
r move mykey 10
|
||||
assert {[r ttl mykey] == -2}
|
||||
r select 10
|
||||
assert {[r ttl mykey] == -1}
|
||||
assert {[r get mykey] eq "foo"}
|
||||
r select 9
|
||||
}
|
||||
|
||||
test {SET/GET keys in different DBs} {
|
||||
r set a hello
|
||||
r set b world
|
||||
|
||||
@@ -2,8 +2,8 @@ start_server {tags {"hash"}} {
|
||||
test {HSET/HLEN - Small hash creation} {
|
||||
array set smallhash {}
|
||||
for {set i 0} {$i < 8} {incr i} {
|
||||
set key [randstring 0 8 alpha]
|
||||
set val [randstring 0 8 alpha]
|
||||
set key __avoid_collisions__[randstring 0 8 alpha]
|
||||
set val __avoid_collisions__[randstring 0 8 alpha]
|
||||
if {[info exists smallhash($key)]} {
|
||||
incr i -1
|
||||
continue
|
||||
@@ -21,8 +21,8 @@ start_server {tags {"hash"}} {
|
||||
test {HSET/HLEN - Big hash creation} {
|
||||
array set bighash {}
|
||||
for {set i 0} {$i < 1024} {incr i} {
|
||||
set key [randstring 0 8 alpha]
|
||||
set val [randstring 0 8 alpha]
|
||||
set key __avoid_collisions__[randstring 0 8 alpha]
|
||||
set val __avoid_collisions__[randstring 0 8 alpha]
|
||||
if {[info exists bighash($key)]} {
|
||||
incr i -1
|
||||
continue
|
||||
@@ -33,7 +33,7 @@ start_server {tags {"hash"}} {
|
||||
list [r hlen bighash]
|
||||
} {1024}
|
||||
|
||||
test {Is the big hash encoded with a ziplist?} {
|
||||
test {Is the big hash encoded with an hash table?} {
|
||||
assert_encoding hashtable bighash
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user