Skip to content
GitLab
Menu
Projects
Groups
Snippets
Loading...
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
Menu
Open sidebar
ruanhaishen
redis
Commits
3325a9b1
Commit
3325a9b1
authored
Jul 27, 2015
by
antirez
Browse files
RDMF: more names updated.
parent
32f80e2f
Changes
8
Hide whitespace changes
Inline
Side-by-side
src/aof.c
View file @
3325a9b1
...
@@ -198,7 +198,7 @@ ssize_t aofRewriteBufferWrite(int fd) {
...
@@ -198,7 +198,7 @@ ssize_t aofRewriteBufferWrite(int fd) {
/* Starts a background task that performs fsync() against the specified
/* Starts a background task that performs fsync() against the specified
* file descriptor (the one of the AOF file) in another thread. */
* file descriptor (the one of the AOF file) in another thread. */
void
aof_background_fsync
(
int
fd
)
{
void
aof_background_fsync
(
int
fd
)
{
bioCreateBackgroundJob
(
REDIS_
BIO_AOF_FSYNC
,(
void
*
)(
long
)
fd
,
NULL
,
NULL
);
bioCreateBackgroundJob
(
BIO_AOF_FSYNC
,(
void
*
)(
long
)
fd
,
NULL
,
NULL
);
}
}
/* Called when the user switches from "appendonly yes" to "appendonly no"
/* Called when the user switches from "appendonly yes" to "appendonly no"
...
@@ -278,7 +278,7 @@ void flushAppendOnlyFile(int force) {
...
@@ -278,7 +278,7 @@ void flushAppendOnlyFile(int force) {
if
(
sdslen
(
server
.
aof_buf
)
==
0
)
return
;
if
(
sdslen
(
server
.
aof_buf
)
==
0
)
return
;
if
(
server
.
aof_fsync
==
AOF_FSYNC_EVERYSEC
)
if
(
server
.
aof_fsync
==
AOF_FSYNC_EVERYSEC
)
sync_in_progress
=
bioPendingJobsOfType
(
REDIS_
BIO_AOF_FSYNC
)
!=
0
;
sync_in_progress
=
bioPendingJobsOfType
(
BIO_AOF_FSYNC
)
!=
0
;
if
(
server
.
aof_fsync
==
AOF_FSYNC_EVERYSEC
&&
!
force
)
{
if
(
server
.
aof_fsync
==
AOF_FSYNC_EVERYSEC
&&
!
force
)
{
/* With this append fsync policy we do background fsyncing.
/* With this append fsync policy we do background fsyncing.
...
@@ -1454,7 +1454,7 @@ void backgroundRewriteDoneHandler(int exitcode, int bysignal) {
...
@@ -1454,7 +1454,7 @@ void backgroundRewriteDoneHandler(int exitcode, int bysignal) {
server
.
aof_state
=
AOF_ON
;
server
.
aof_state
=
AOF_ON
;
/* Asynchronously close the overwritten AOF. */
/* Asynchronously close the overwritten AOF. */
if
(
oldfd
!=
-
1
)
bioCreateBackgroundJob
(
REDIS_
BIO_CLOSE_FILE
,(
void
*
)(
long
)
oldfd
,
NULL
,
NULL
);
if
(
oldfd
!=
-
1
)
bioCreateBackgroundJob
(
BIO_CLOSE_FILE
,(
void
*
)(
long
)
oldfd
,
NULL
,
NULL
);
serverLog
(
LL_VERBOSE
,
serverLog
(
LL_VERBOSE
,
"Background AOF rewrite signal handler took %lldus"
,
ustime
()
-
now
);
"Background AOF rewrite signal handler took %lldus"
,
ustime
()
-
now
);
...
...
src/bio.c
View file @
3325a9b1
...
@@ -61,17 +61,17 @@
...
@@ -61,17 +61,17 @@
#include "server.h"
#include "server.h"
#include "bio.h"
#include "bio.h"
static
pthread_t
bio_threads
[
REDIS_
BIO_NUM_OPS
];
static
pthread_t
bio_threads
[
BIO_NUM_OPS
];
static
pthread_mutex_t
bio_mutex
[
REDIS_
BIO_NUM_OPS
];
static
pthread_mutex_t
bio_mutex
[
BIO_NUM_OPS
];
static
pthread_cond_t
bio_condvar
[
REDIS_
BIO_NUM_OPS
];
static
pthread_cond_t
bio_condvar
[
BIO_NUM_OPS
];
static
list
*
bio_jobs
[
REDIS_
BIO_NUM_OPS
];
static
list
*
bio_jobs
[
BIO_NUM_OPS
];
/* The following array is used to hold the number of pending jobs for every
/* The following array is used to hold the number of pending jobs for every
* OP type. This allows us to export the bioPendingJobsOfType() API that is
* OP type. This allows us to export the bioPendingJobsOfType() API that is
* useful when the main thread wants to perform some operation that may involve
* useful when the main thread wants to perform some operation that may involve
* objects shared with the background thread. The main thread will just wait
* objects shared with the background thread. The main thread will just wait
* that there are no longer jobs of this type to be executed before performing
* that there are no longer jobs of this type to be executed before performing
* the sensible operation. This data is also useful for reporting. */
* the sensible operation. This data is also useful for reporting. */
static
unsigned
long
long
bio_pending
[
REDIS_
BIO_NUM_OPS
];
static
unsigned
long
long
bio_pending
[
BIO_NUM_OPS
];
/* This structure represents a background Job. It is only used locally to this
/* This structure represents a background Job. It is only used locally to this
* file as the API does not expose the internals at all. */
* file as the API does not expose the internals at all. */
...
@@ -96,7 +96,7 @@ void bioInit(void) {
...
@@ -96,7 +96,7 @@ void bioInit(void) {
int
j
;
int
j
;
/* Initialization of state vars and objects */
/* Initialization of state vars and objects */
for
(
j
=
0
;
j
<
REDIS_
BIO_NUM_OPS
;
j
++
)
{
for
(
j
=
0
;
j
<
BIO_NUM_OPS
;
j
++
)
{
pthread_mutex_init
(
&
bio_mutex
[
j
],
NULL
);
pthread_mutex_init
(
&
bio_mutex
[
j
],
NULL
);
pthread_cond_init
(
&
bio_condvar
[
j
],
NULL
);
pthread_cond_init
(
&
bio_condvar
[
j
],
NULL
);
bio_jobs
[
j
]
=
listCreate
();
bio_jobs
[
j
]
=
listCreate
();
...
@@ -113,7 +113,7 @@ void bioInit(void) {
...
@@ -113,7 +113,7 @@ void bioInit(void) {
/* Ready to spawn our threads. We use the single argument the thread
/* Ready to spawn our threads. We use the single argument the thread
* function accepts in order to pass the job ID the thread is
* function accepts in order to pass the job ID the thread is
* responsible of. */
* responsible of. */
for
(
j
=
0
;
j
<
REDIS_
BIO_NUM_OPS
;
j
++
)
{
for
(
j
=
0
;
j
<
BIO_NUM_OPS
;
j
++
)
{
void
*
arg
=
(
void
*
)(
unsigned
long
)
j
;
void
*
arg
=
(
void
*
)(
unsigned
long
)
j
;
if
(
pthread_create
(
&
thread
,
&
attr
,
bioProcessBackgroundJobs
,
arg
)
!=
0
)
{
if
(
pthread_create
(
&
thread
,
&
attr
,
bioProcessBackgroundJobs
,
arg
)
!=
0
)
{
serverLog
(
LL_WARNING
,
"Fatal: Can't initialize Background Jobs."
);
serverLog
(
LL_WARNING
,
"Fatal: Can't initialize Background Jobs."
);
...
@@ -143,7 +143,7 @@ void *bioProcessBackgroundJobs(void *arg) {
...
@@ -143,7 +143,7 @@ void *bioProcessBackgroundJobs(void *arg) {
sigset_t
sigset
;
sigset_t
sigset
;
/* Check that the type is within the right interval. */
/* Check that the type is within the right interval. */
if
(
type
>=
REDIS_
BIO_NUM_OPS
)
{
if
(
type
>=
BIO_NUM_OPS
)
{
serverLog
(
LL_WARNING
,
serverLog
(
LL_WARNING
,
"Warning: bio thread started with wrong type %lu"
,
type
);
"Warning: bio thread started with wrong type %lu"
,
type
);
return
NULL
;
return
NULL
;
...
@@ -179,9 +179,9 @@ void *bioProcessBackgroundJobs(void *arg) {
...
@@ -179,9 +179,9 @@ void *bioProcessBackgroundJobs(void *arg) {
pthread_mutex_unlock
(
&
bio_mutex
[
type
]);
pthread_mutex_unlock
(
&
bio_mutex
[
type
]);
/* Process the job accordingly to its type. */
/* Process the job accordingly to its type. */
if
(
type
==
REDIS_
BIO_CLOSE_FILE
)
{
if
(
type
==
BIO_CLOSE_FILE
)
{
close
((
long
)
job
->
arg1
);
close
((
long
)
job
->
arg1
);
}
else
if
(
type
==
REDIS_
BIO_AOF_FSYNC
)
{
}
else
if
(
type
==
BIO_AOF_FSYNC
)
{
aof_fsync
((
long
)
job
->
arg1
);
aof_fsync
((
long
)
job
->
arg1
);
}
else
{
}
else
{
serverPanic
(
"Wrong job type in bioProcessBackgroundJobs()."
);
serverPanic
(
"Wrong job type in bioProcessBackgroundJobs()."
);
...
@@ -212,7 +212,7 @@ unsigned long long bioPendingJobsOfType(int type) {
...
@@ -212,7 +212,7 @@ unsigned long long bioPendingJobsOfType(int type) {
void
bioKillThreads
(
void
)
{
void
bioKillThreads
(
void
)
{
int
err
,
j
;
int
err
,
j
;
for
(
j
=
0
;
j
<
REDIS_
BIO_NUM_OPS
;
j
++
)
{
for
(
j
=
0
;
j
<
BIO_NUM_OPS
;
j
++
)
{
if
(
pthread_cancel
(
bio_threads
[
j
])
==
0
)
{
if
(
pthread_cancel
(
bio_threads
[
j
])
==
0
)
{
if
((
err
=
pthread_join
(
bio_threads
[
j
],
NULL
))
!=
0
)
{
if
((
err
=
pthread_join
(
bio_threads
[
j
],
NULL
))
!=
0
)
{
serverLog
(
LL_WARNING
,
serverLog
(
LL_WARNING
,
...
...
src/bio.h
View file @
3325a9b1
...
@@ -36,6 +36,6 @@ time_t bioOlderJobOfType(int type);
...
@@ -36,6 +36,6 @@ time_t bioOlderJobOfType(int type);
void
bioKillThreads
(
void
);
void
bioKillThreads
(
void
);
/* Background job opcodes */
/* Background job opcodes */
#define
REDIS_
BIO_CLOSE_FILE 0
/* Deferred close(2) syscall. */
#define BIO_CLOSE_FILE 0
/* Deferred close(2) syscall. */
#define
REDIS_
BIO_AOF_FSYNC 1
/* Deferred AOF fsync. */
#define BIO_AOF_FSYNC 1
/* Deferred AOF fsync. */
#define
REDIS_
BIO_NUM_OPS 2
#define BIO_NUM_OPS 2
src/cluster.c
View file @
3325a9b1
...
@@ -117,8 +117,8 @@ int clusterLoadConfig(char *filename) {
...
@@ -117,8 +117,8 @@ int clusterLoadConfig(char *filename) {
* present in a single line, possibly in importing or migrating state, so
* present in a single line, possibly in importing or migrating state, so
* together with the node ID of the sender/receiver.
* together with the node ID of the sender/receiver.
*
*
* To simplify we allocate 1024+
REDIS_
CLUSTER_SLOTS*128 bytes per line. */
* To simplify we allocate 1024+CLUSTER_SLOTS*128 bytes per line. */
maxline
=
1024
+
REDIS_
CLUSTER_SLOTS
*
128
;
maxline
=
1024
+
CLUSTER_SLOTS
*
128
;
line
=
zmalloc
(
maxline
);
line
=
zmalloc
(
maxline
);
while
(
fgets
(
line
,
maxline
,
fp
)
!=
NULL
)
{
while
(
fgets
(
line
,
maxline
,
fp
)
!=
NULL
)
{
int
argc
;
int
argc
;
...
@@ -178,20 +178,20 @@ int clusterLoadConfig(char *filename) {
...
@@ -178,20 +178,20 @@ int clusterLoadConfig(char *filename) {
if
(
!
strcasecmp
(
s
,
"myself"
))
{
if
(
!
strcasecmp
(
s
,
"myself"
))
{
serverAssert
(
server
.
cluster
->
myself
==
NULL
);
serverAssert
(
server
.
cluster
->
myself
==
NULL
);
myself
=
server
.
cluster
->
myself
=
n
;
myself
=
server
.
cluster
->
myself
=
n
;
n
->
flags
|=
REDIS
_NODE_MYSELF
;
n
->
flags
|=
CLUSTER
_NODE_MYSELF
;
}
else
if
(
!
strcasecmp
(
s
,
"master"
))
{
}
else
if
(
!
strcasecmp
(
s
,
"master"
))
{
n
->
flags
|=
REDIS
_NODE_MASTER
;
n
->
flags
|=
CLUSTER
_NODE_MASTER
;
}
else
if
(
!
strcasecmp
(
s
,
"slave"
))
{
}
else
if
(
!
strcasecmp
(
s
,
"slave"
))
{
n
->
flags
|=
REDIS
_NODE_SLAVE
;
n
->
flags
|=
CLUSTER
_NODE_SLAVE
;
}
else
if
(
!
strcasecmp
(
s
,
"fail?"
))
{
}
else
if
(
!
strcasecmp
(
s
,
"fail?"
))
{
n
->
flags
|=
REDIS
_NODE_PFAIL
;
n
->
flags
|=
CLUSTER
_NODE_PFAIL
;
}
else
if
(
!
strcasecmp
(
s
,
"fail"
))
{
}
else
if
(
!
strcasecmp
(
s
,
"fail"
))
{
n
->
flags
|=
REDIS
_NODE_FAIL
;
n
->
flags
|=
CLUSTER
_NODE_FAIL
;
n
->
fail_time
=
mstime
();
n
->
fail_time
=
mstime
();
}
else
if
(
!
strcasecmp
(
s
,
"handshake"
))
{
}
else
if
(
!
strcasecmp
(
s
,
"handshake"
))
{
n
->
flags
|=
REDIS
_NODE_HANDSHAKE
;
n
->
flags
|=
CLUSTER
_NODE_HANDSHAKE
;
}
else
if
(
!
strcasecmp
(
s
,
"noaddr"
))
{
}
else
if
(
!
strcasecmp
(
s
,
"noaddr"
))
{
n
->
flags
|=
REDIS
_NODE_NOADDR
;
n
->
flags
|=
CLUSTER
_NODE_NOADDR
;
}
else
if
(
!
strcasecmp
(
s
,
"noflags"
))
{
}
else
if
(
!
strcasecmp
(
s
,
"noflags"
))
{
/* nothing to do */
/* nothing to do */
}
else
{
}
else
{
...
@@ -304,7 +304,7 @@ int clusterSaveConfig(int do_fsync) {
...
@@ -304,7 +304,7 @@ int clusterSaveConfig(int do_fsync) {
/* Get the nodes description and concatenate our "vars" directive to
/* Get the nodes description and concatenate our "vars" directive to
* save currentEpoch and lastVoteEpoch. */
* save currentEpoch and lastVoteEpoch. */
ci
=
clusterGenNodesDescription
(
REDIS
_NODE_HANDSHAKE
);
ci
=
clusterGenNodesDescription
(
CLUSTER
_NODE_HANDSHAKE
);
ci
=
sdscatprintf
(
ci
,
"vars currentEpoch %llu lastVoteEpoch %llu
\n
"
,
ci
=
sdscatprintf
(
ci
,
"vars currentEpoch %llu lastVoteEpoch %llu
\n
"
,
(
unsigned
long
long
)
server
.
cluster
->
currentEpoch
,
(
unsigned
long
long
)
server
.
cluster
->
currentEpoch
,
(
unsigned
long
long
)
server
.
cluster
->
lastVoteEpoch
);
(
unsigned
long
long
)
server
.
cluster
->
lastVoteEpoch
);
...
@@ -401,7 +401,7 @@ void clusterInit(void) {
...
@@ -401,7 +401,7 @@ void clusterInit(void) {
server
.
cluster
=
zmalloc
(
sizeof
(
clusterState
));
server
.
cluster
=
zmalloc
(
sizeof
(
clusterState
));
server
.
cluster
->
myself
=
NULL
;
server
.
cluster
->
myself
=
NULL
;
server
.
cluster
->
currentEpoch
=
0
;
server
.
cluster
->
currentEpoch
=
0
;
server
.
cluster
->
state
=
REDIS_
CLUSTER_FAIL
;
server
.
cluster
->
state
=
CLUSTER_FAIL
;
server
.
cluster
->
size
=
1
;
server
.
cluster
->
size
=
1
;
server
.
cluster
->
todo_before_sleep
=
0
;
server
.
cluster
->
todo_before_sleep
=
0
;
server
.
cluster
->
nodes
=
dictCreate
(
&
clusterNodesDictType
,
NULL
);
server
.
cluster
->
nodes
=
dictCreate
(
&
clusterNodesDictType
,
NULL
);
...
@@ -411,7 +411,7 @@ void clusterInit(void) {
...
@@ -411,7 +411,7 @@ void clusterInit(void) {
server
.
cluster
->
failover_auth_count
=
0
;
server
.
cluster
->
failover_auth_count
=
0
;
server
.
cluster
->
failover_auth_rank
=
0
;
server
.
cluster
->
failover_auth_rank
=
0
;
server
.
cluster
->
failover_auth_epoch
=
0
;
server
.
cluster
->
failover_auth_epoch
=
0
;
server
.
cluster
->
cant_failover_reason
=
REDIS_
CLUSTER_CANT_FAILOVER_NONE
;
server
.
cluster
->
cant_failover_reason
=
CLUSTER_CANT_FAILOVER_NONE
;
server
.
cluster
->
lastVoteEpoch
=
0
;
server
.
cluster
->
lastVoteEpoch
=
0
;
server
.
cluster
->
stats_bus_messages_sent
=
0
;
server
.
cluster
->
stats_bus_messages_sent
=
0
;
server
.
cluster
->
stats_bus_messages_received
=
0
;
server
.
cluster
->
stats_bus_messages_received
=
0
;
...
@@ -428,7 +428,7 @@ void clusterInit(void) {
...
@@ -428,7 +428,7 @@ void clusterInit(void) {
/* No configuration found. We will just use the random name provided
/* No configuration found. We will just use the random name provided
* by the createClusterNode() function. */
* by the createClusterNode() function. */
myself
=
server
.
cluster
->
myself
=
myself
=
server
.
cluster
->
myself
=
createClusterNode
(
NULL
,
REDIS
_NODE_MYSELF
|
REDIS
_NODE_MASTER
);
createClusterNode
(
NULL
,
CLUSTER
_NODE_MYSELF
|
CLUSTER
_NODE_MASTER
);
serverLog
(
LL_NOTICE
,
"No cluster configuration found, I'm %.40s"
,
serverLog
(
LL_NOTICE
,
"No cluster configuration found, I'm %.40s"
,
myself
->
name
);
myself
->
name
);
clusterAddNode
(
myself
);
clusterAddNode
(
myself
);
...
@@ -442,7 +442,7 @@ void clusterInit(void) {
...
@@ -442,7 +442,7 @@ void clusterInit(void) {
/* Port sanity check II
/* Port sanity check II
* The other handshake port check is triggered too late to stop
* The other handshake port check is triggered too late to stop
* us from trying to use a too-high cluster port number. */
* us from trying to use a too-high cluster port number. */
if
(
server
.
port
>
(
65535
-
REDIS_
CLUSTER_PORT_INCR
))
{
if
(
server
.
port
>
(
65535
-
CLUSTER_PORT_INCR
))
{
serverLog
(
LL_WARNING
,
"Redis port number too high. "
serverLog
(
LL_WARNING
,
"Redis port number too high. "
"Cluster communication port is 10,000 port "
"Cluster communication port is 10,000 port "
"numbers higher than your Redis port. "
"numbers higher than your Redis port. "
...
@@ -451,7 +451,7 @@ void clusterInit(void) {
...
@@ -451,7 +451,7 @@ void clusterInit(void) {
exit
(
1
);
exit
(
1
);
}
}
if
(
listenToPort
(
server
.
port
+
REDIS_
CLUSTER_PORT_INCR
,
if
(
listenToPort
(
server
.
port
+
CLUSTER_PORT_INCR
,
server
.
cfd
,
&
server
.
cfd_count
)
==
C_ERR
)
server
.
cfd
,
&
server
.
cfd_count
)
==
C_ERR
)
{
{
exit
(
1
);
exit
(
1
);
...
@@ -503,7 +503,7 @@ void clusterReset(int hard) {
...
@@ -503,7 +503,7 @@ void clusterReset(int hard) {
resetManualFailover
();
resetManualFailover
();
/* Unassign all the slots. */
/* Unassign all the slots. */
for
(
j
=
0
;
j
<
REDIS_
CLUSTER_SLOTS
;
j
++
)
clusterDelSlot
(
j
);
for
(
j
=
0
;
j
<
CLUSTER_SLOTS
;
j
++
)
clusterDelSlot
(
j
);
/* Forget all the nodes, but myself. */
/* Forget all the nodes, but myself. */
di
=
dictGetSafeIterator
(
server
.
cluster
->
nodes
);
di
=
dictGetSafeIterator
(
server
.
cluster
->
nodes
);
...
@@ -526,10 +526,10 @@ void clusterReset(int hard) {
...
@@ -526,10 +526,10 @@ void clusterReset(int hard) {
/* To change the Node ID we need to remove the old name from the
/* To change the Node ID we need to remove the old name from the
* nodes table, change the ID, and re-add back with new name. */
* nodes table, change the ID, and re-add back with new name. */
oldname
=
sdsnewlen
(
myself
->
name
,
REDIS_
CLUSTER_NAMELEN
);
oldname
=
sdsnewlen
(
myself
->
name
,
CLUSTER_NAMELEN
);
dictDelete
(
server
.
cluster
->
nodes
,
oldname
);
dictDelete
(
server
.
cluster
->
nodes
,
oldname
);
sdsfree
(
oldname
);
sdsfree
(
oldname
);
getRandomHexChars
(
myself
->
name
,
REDIS_
CLUSTER_NAMELEN
);
getRandomHexChars
(
myself
->
name
,
CLUSTER_NAMELEN
);
clusterAddNode
(
myself
);
clusterAddNode
(
myself
);
}
}
...
@@ -653,9 +653,9 @@ clusterNode *createClusterNode(char *nodename, int flags) {
...
@@ -653,9 +653,9 @@ clusterNode *createClusterNode(char *nodename, int flags) {
clusterNode
*
node
=
zmalloc
(
sizeof
(
*
node
));
clusterNode
*
node
=
zmalloc
(
sizeof
(
*
node
));
if
(
nodename
)
if
(
nodename
)
memcpy
(
node
->
name
,
nodename
,
REDIS_
CLUSTER_NAMELEN
);
memcpy
(
node
->
name
,
nodename
,
CLUSTER_NAMELEN
);
else
else
getRandomHexChars
(
node
->
name
,
REDIS_
CLUSTER_NAMELEN
);
getRandomHexChars
(
node
->
name
,
CLUSTER_NAMELEN
);
node
->
ctime
=
mstime
();
node
->
ctime
=
mstime
();
node
->
configEpoch
=
0
;
node
->
configEpoch
=
0
;
node
->
flags
=
flags
;
node
->
flags
=
flags
;
...
@@ -723,7 +723,7 @@ void clusterNodeCleanupFailureReports(clusterNode *node) {
...
@@ -723,7 +723,7 @@ void clusterNodeCleanupFailureReports(clusterNode *node) {
listIter
li
;
listIter
li
;
clusterNodeFailReport
*
fr
;
clusterNodeFailReport
*
fr
;
mstime_t
maxtime
=
server
.
cluster_node_timeout
*
mstime_t
maxtime
=
server
.
cluster_node_timeout
*
REDIS_
CLUSTER_FAIL_REPORT_VALIDITY_MULT
;
CLUSTER_FAIL_REPORT_VALIDITY_MULT
;
mstime_t
now
=
mstime
();
mstime_t
now
=
mstime
();
listRewind
(
l
,
&
li
);
listRewind
(
l
,
&
li
);
...
@@ -832,7 +832,7 @@ void freeClusterNode(clusterNode *n) {
...
@@ -832,7 +832,7 @@ void freeClusterNode(clusterNode *n) {
if
(
nodeIsSlave
(
n
)
&&
n
->
slaveof
)
clusterNodeRemoveSlave
(
n
->
slaveof
,
n
);
if
(
nodeIsSlave
(
n
)
&&
n
->
slaveof
)
clusterNodeRemoveSlave
(
n
->
slaveof
,
n
);
/* Unlink from the set of nodes. */
/* Unlink from the set of nodes. */
nodename
=
sdsnewlen
(
n
->
name
,
REDIS_
CLUSTER_NAMELEN
);
nodename
=
sdsnewlen
(
n
->
name
,
CLUSTER_NAMELEN
);
serverAssert
(
dictDelete
(
server
.
cluster
->
nodes
,
nodename
)
==
DICT_OK
);
serverAssert
(
dictDelete
(
server
.
cluster
->
nodes
,
nodename
)
==
DICT_OK
);
sdsfree
(
nodename
);
sdsfree
(
nodename
);
...
@@ -848,7 +848,7 @@ int clusterAddNode(clusterNode *node) {
...
@@ -848,7 +848,7 @@ int clusterAddNode(clusterNode *node) {
int
retval
;
int
retval
;
retval
=
dictAdd
(
server
.
cluster
->
nodes
,
retval
=
dictAdd
(
server
.
cluster
->
nodes
,
sdsnewlen
(
node
->
name
,
REDIS_
CLUSTER_NAMELEN
),
node
);
sdsnewlen
(
node
->
name
,
CLUSTER_NAMELEN
),
node
);
return
(
retval
==
DICT_OK
)
?
C_OK
:
C_ERR
;
return
(
retval
==
DICT_OK
)
?
C_OK
:
C_ERR
;
}
}
...
@@ -869,7 +869,7 @@ void clusterDelNode(clusterNode *delnode) {
...
@@ -869,7 +869,7 @@ void clusterDelNode(clusterNode *delnode) {
dictEntry
*
de
;
dictEntry
*
de
;
/* 1) Mark slots as unassigned. */
/* 1) Mark slots as unassigned. */
for
(
j
=
0
;
j
<
REDIS_
CLUSTER_SLOTS
;
j
++
)
{
for
(
j
=
0
;
j
<
CLUSTER_SLOTS
;
j
++
)
{
if
(
server
.
cluster
->
importing_slots_from
[
j
]
==
delnode
)
if
(
server
.
cluster
->
importing_slots_from
[
j
]
==
delnode
)
server
.
cluster
->
importing_slots_from
[
j
]
=
NULL
;
server
.
cluster
->
importing_slots_from
[
j
]
=
NULL
;
if
(
server
.
cluster
->
migrating_slots_to
[
j
]
==
delnode
)
if
(
server
.
cluster
->
migrating_slots_to
[
j
]
==
delnode
)
...
@@ -894,7 +894,7 @@ void clusterDelNode(clusterNode *delnode) {
...
@@ -894,7 +894,7 @@ void clusterDelNode(clusterNode *delnode) {
/* Node lookup by name */
/* Node lookup by name */
clusterNode
*
clusterLookupNode
(
char
*
name
)
{
clusterNode
*
clusterLookupNode
(
char
*
name
)
{
sds
s
=
sdsnewlen
(
name
,
REDIS_
CLUSTER_NAMELEN
);
sds
s
=
sdsnewlen
(
name
,
CLUSTER_NAMELEN
);
dictEntry
*
de
;
dictEntry
*
de
;
de
=
dictFind
(
server
.
cluster
->
nodes
,
s
);
de
=
dictFind
(
server
.
cluster
->
nodes
,
s
);
...
@@ -909,14 +909,14 @@ clusterNode *clusterLookupNode(char *name) {
...
@@ -909,14 +909,14 @@ clusterNode *clusterLookupNode(char *name) {
* this function. */
* this function. */
void
clusterRenameNode
(
clusterNode
*
node
,
char
*
newname
)
{
void
clusterRenameNode
(
clusterNode
*
node
,
char
*
newname
)
{
int
retval
;
int
retval
;
sds
s
=
sdsnewlen
(
node
->
name
,
REDIS_
CLUSTER_NAMELEN
);
sds
s
=
sdsnewlen
(
node
->
name
,
CLUSTER_NAMELEN
);
serverLog
(
LL_DEBUG
,
"Renaming node %.40s into %.40s"
,
serverLog
(
LL_DEBUG
,
"Renaming node %.40s into %.40s"
,
node
->
name
,
newname
);
node
->
name
,
newname
);
retval
=
dictDelete
(
server
.
cluster
->
nodes
,
s
);
retval
=
dictDelete
(
server
.
cluster
->
nodes
,
s
);
sdsfree
(
s
);
sdsfree
(
s
);
serverAssert
(
retval
==
DICT_OK
);
serverAssert
(
retval
==
DICT_OK
);
memcpy
(
node
->
name
,
newname
,
REDIS_
CLUSTER_NAMELEN
);
memcpy
(
node
->
name
,
newname
,
CLUSTER_NAMELEN
);
clusterAddNode
(
node
);
clusterAddNode
(
node
);
}
}
...
@@ -1040,7 +1040,7 @@ void clusterHandleConfigEpochCollision(clusterNode *sender) {
...
@@ -1040,7 +1040,7 @@ void clusterHandleConfigEpochCollision(clusterNode *sender) {
if
(
sender
->
configEpoch
!=
myself
->
configEpoch
||
if
(
sender
->
configEpoch
!=
myself
->
configEpoch
||
!
nodeIsMaster
(
sender
)
||
!
nodeIsMaster
(
myself
))
return
;
!
nodeIsMaster
(
sender
)
||
!
nodeIsMaster
(
myself
))
return
;
/* Don't act if the colliding node has a smaller Node ID. */
/* Don't act if the colliding node has a smaller Node ID. */
if
(
memcmp
(
sender
->
name
,
myself
->
name
,
REDIS_
CLUSTER_NAMELEN
)
<=
0
)
return
;
if
(
memcmp
(
sender
->
name
,
myself
->
name
,
CLUSTER_NAMELEN
)
<=
0
)
return
;
/* Get the next ID available at the best of this node knowledge. */
/* Get the next ID available at the best of this node knowledge. */
server
.
cluster
->
currentEpoch
++
;
server
.
cluster
->
currentEpoch
++
;
myself
->
configEpoch
=
server
.
cluster
->
currentEpoch
;
myself
->
configEpoch
=
server
.
cluster
->
currentEpoch
;
...
@@ -1057,14 +1057,14 @@ void clusterHandleConfigEpochCollision(clusterNode *sender) {
...
@@ -1057,14 +1057,14 @@ void clusterHandleConfigEpochCollision(clusterNode *sender) {
*
*
* The nodes blacklist is just a way to ensure that a given node with a given
* The nodes blacklist is just a way to ensure that a given node with a given
* Node ID is not readded before some time elapsed (this time is specified
* Node ID is not readded before some time elapsed (this time is specified
* in seconds in
REDIS_
CLUSTER_BLACKLIST_TTL).
* in seconds in CLUSTER_BLACKLIST_TTL).
*
*
* This is useful when we want to remove a node from the cluster completely:
* This is useful when we want to remove a node from the cluster completely:
* when CLUSTER FORGET is called, it also puts the node into the blacklist so
* when CLUSTER FORGET is called, it also puts the node into the blacklist so
* that even if we receive gossip messages from other nodes that still remember
* that even if we receive gossip messages from other nodes that still remember
* about the node we want to remove, we don't re-add it before some time.
* about the node we want to remove, we don't re-add it before some time.
*
*
* Currently the
REDIS_
CLUSTER_BLACKLIST_TTL is set to 1 minute, this means
* Currently the CLUSTER_BLACKLIST_TTL is set to 1 minute, this means
* that redis-trib has 60 seconds to send CLUSTER FORGET messages to nodes
* that redis-trib has 60 seconds to send CLUSTER FORGET messages to nodes
* in the cluster without dealing with the problem of other nodes re-adding
* in the cluster without dealing with the problem of other nodes re-adding
* back the node to nodes we already sent the FORGET command to.
* back the node to nodes we already sent the FORGET command to.
...
@@ -1074,7 +1074,7 @@ void clusterHandleConfigEpochCollision(clusterNode *sender) {
...
@@ -1074,7 +1074,7 @@ void clusterHandleConfigEpochCollision(clusterNode *sender) {
* value.
* value.
* -------------------------------------------------------------------------- */
* -------------------------------------------------------------------------- */
#define
REDIS_
CLUSTER_BLACKLIST_TTL 60
/* 1 minute. */
#define CLUSTER_BLACKLIST_TTL 60
/* 1 minute. */
/* Before of the addNode() or Exists() operations we always remove expired
/* Before of the addNode() or Exists() operations we always remove expired
...
@@ -1100,7 +1100,7 @@ void clusterBlacklistCleanup(void) {
...
@@ -1100,7 +1100,7 @@ void clusterBlacklistCleanup(void) {
/* Cleanup the blacklist and add a new node ID to the black list. */
/* Cleanup the blacklist and add a new node ID to the black list. */
void
clusterBlacklistAddNode
(
clusterNode
*
node
)
{
void
clusterBlacklistAddNode
(
clusterNode
*
node
)
{
dictEntry
*
de
;
dictEntry
*
de
;
sds
id
=
sdsnewlen
(
node
->
name
,
REDIS_
CLUSTER_NAMELEN
);
sds
id
=
sdsnewlen
(
node
->
name
,
CLUSTER_NAMELEN
);
clusterBlacklistCleanup
();
clusterBlacklistCleanup
();
if
(
dictAdd
(
server
.
cluster
->
nodes_black_list
,
id
,
NULL
)
==
DICT_OK
)
{
if
(
dictAdd
(
server
.
cluster
->
nodes_black_list
,
id
,
NULL
)
==
DICT_OK
)
{
...
@@ -1109,7 +1109,7 @@ void clusterBlacklistAddNode(clusterNode *node) {
...
@@ -1109,7 +1109,7 @@ void clusterBlacklistAddNode(clusterNode *node) {
id
=
sdsdup
(
id
);
id
=
sdsdup
(
id
);
}
}
de
=
dictFind
(
server
.
cluster
->
nodes_black_list
,
id
);
de
=
dictFind
(
server
.
cluster
->
nodes_black_list
,
id
);
dictSetUnsignedIntegerVal
(
de
,
time
(
NULL
)
+
REDIS_
CLUSTER_BLACKLIST_TTL
);
dictSetUnsignedIntegerVal
(
de
,
time
(
NULL
)
+
CLUSTER_BLACKLIST_TTL
);
sdsfree
(
id
);
sdsfree
(
id
);
}
}
...
@@ -1117,7 +1117,7 @@ void clusterBlacklistAddNode(clusterNode *node) {
...
@@ -1117,7 +1117,7 @@ void clusterBlacklistAddNode(clusterNode *node) {
* You don't need to pass an sds string here, any pointer to 40 bytes
* You don't need to pass an sds string here, any pointer to 40 bytes
* will work. */
* will work. */
int
clusterBlacklistExists
(
char
*
nodeid
)
{
int
clusterBlacklistExists
(
char
*
nodeid
)
{
sds
id
=
sdsnewlen
(
nodeid
,
REDIS_
CLUSTER_NAMELEN
);
sds
id
=
sdsnewlen
(
nodeid
,
CLUSTER_NAMELEN
);
int
retval
;
int
retval
;
clusterBlacklistCleanup
();
clusterBlacklistCleanup
();
...
@@ -1167,8 +1167,8 @@ void markNodeAsFailingIfNeeded(clusterNode *node) {
...
@@ -1167,8 +1167,8 @@ void markNodeAsFailingIfNeeded(clusterNode *node) {
"Marking node %.40s as failing (quorum reached)."
,
node
->
name
);
"Marking node %.40s as failing (quorum reached)."
,
node
->
name
);
/* Mark the node as failing. */
/* Mark the node as failing. */
node
->
flags
&=
~
REDIS
_NODE_PFAIL
;
node
->
flags
&=
~
CLUSTER
_NODE_PFAIL
;
node
->
flags
|=
REDIS
_NODE_FAIL
;
node
->
flags
|=
CLUSTER
_NODE_FAIL
;
node
->
fail_time
=
mstime
();
node
->
fail_time
=
mstime
();
/* Broadcast the failing node name to everybody, forcing all the other
/* Broadcast the failing node name to everybody, forcing all the other
...
@@ -1192,7 +1192,7 @@ void clearNodeFailureIfNeeded(clusterNode *node) {
...
@@ -1192,7 +1192,7 @@ void clearNodeFailureIfNeeded(clusterNode *node) {
"Clear FAIL state for node %.40s: %s is reachable again."
,
"Clear FAIL state for node %.40s: %s is reachable again."
,
node
->
name
,
node
->
name
,
nodeIsSlave
(
node
)
?
"slave"
:
"master without slots"
);
nodeIsSlave
(
node
)
?
"slave"
:
"master without slots"
);
node
->
flags
&=
~
REDIS
_NODE_FAIL
;
node
->
flags
&=
~
CLUSTER
_NODE_FAIL
;
clusterDoBeforeSleep
(
CLUSTER_TODO_UPDATE_STATE
|
CLUSTER_TODO_SAVE_CONFIG
);
clusterDoBeforeSleep
(
CLUSTER_TODO_UPDATE_STATE
|
CLUSTER_TODO_SAVE_CONFIG
);
}
}
...
@@ -1202,12 +1202,12 @@ void clearNodeFailureIfNeeded(clusterNode *node) {
...
@@ -1202,12 +1202,12 @@ void clearNodeFailureIfNeeded(clusterNode *node) {
* Apparently no one is going to fix these slots, clear the FAIL flag. */
* Apparently no one is going to fix these slots, clear the FAIL flag. */
if
(
nodeIsMaster
(
node
)
&&
node
->
numslots
>
0
&&
if
(
nodeIsMaster
(
node
)
&&
node
->
numslots
>
0
&&
(
now
-
node
->
fail_time
)
>
(
now
-
node
->
fail_time
)
>
(
server
.
cluster_node_timeout
*
REDIS_
CLUSTER_FAIL_UNDO_TIME_MULT
))
(
server
.
cluster_node_timeout
*
CLUSTER_FAIL_UNDO_TIME_MULT
))
{
{
serverLog
(
LL_NOTICE
,
serverLog
(
LL_NOTICE
,
"Clear FAIL state for node %.40s: is reachable again and nobody is serving its slots after some time."
,
"Clear FAIL state for node %.40s: is reachable again and nobody is serving its slots after some time."
,
node
->
name
);
node
->
name
);
node
->
flags
&=
~
REDIS
_NODE_FAIL
;
node
->
flags
&=
~
CLUSTER
_NODE_FAIL
;
clusterDoBeforeSleep
(
CLUSTER_TODO_UPDATE_STATE
|
CLUSTER_TODO_SAVE_CONFIG
);
clusterDoBeforeSleep
(
CLUSTER_TODO_UPDATE_STATE
|
CLUSTER_TODO_SAVE_CONFIG
);
}
}
}
}
...
@@ -1257,7 +1257,7 @@ int clusterStartHandshake(char *ip, int port) {
...
@@ -1257,7 +1257,7 @@ int clusterStartHandshake(char *ip, int port) {
}
}
/* Port sanity check */
/* Port sanity check */
if
(
port
<=
0
||
port
>
(
65535
-
REDIS_
CLUSTER_PORT_INCR
))
{
if
(
port
<=
0
||
port
>
(
65535
-
CLUSTER_PORT_INCR
))
{
errno
=
EINVAL
;
errno
=
EINVAL
;
return
0
;
return
0
;
}
}
...
@@ -1282,7 +1282,7 @@ int clusterStartHandshake(char *ip, int port) {
...
@@ -1282,7 +1282,7 @@ int clusterStartHandshake(char *ip, int port) {
/* Add the node with a random address (NULL as first argument to
/* Add the node with a random address (NULL as first argument to
* createClusterNode()). Everything will be fixed during the
* createClusterNode()). Everything will be fixed during the
* handshake. */
* handshake. */
n
=
createClusterNode
(
NULL
,
REDIS
_NODE_HANDSHAKE
|
REDIS
_NODE_MEET
);
n
=
createClusterNode
(
NULL
,
CLUSTER
_NODE_HANDSHAKE
|
CLUSTER
_NODE_MEET
);
memcpy
(
n
->
ip
,
norm_ip
,
sizeof
(
n
->
ip
));
memcpy
(
n
->
ip
,
norm_ip
,
sizeof
(
n
->
ip
));
n
->
port
=
port
;
n
->
port
=
port
;
clusterAddNode
(
n
);
clusterAddNode
(
n
);
...
@@ -1317,7 +1317,7 @@ void clusterProcessGossipSection(clusterMsg *hdr, clusterLink *link) {
...
@@ -1317,7 +1317,7 @@ void clusterProcessGossipSection(clusterMsg *hdr, clusterLink *link) {
/* We already know this node.
/* We already know this node.
Handle failure reports, only when the sender is a master. */
Handle failure reports, only when the sender is a master. */
if
(
sender
&&
nodeIsMaster
(
sender
)
&&
node
!=
myself
)
{
if
(
sender
&&
nodeIsMaster
(
sender
)
&&
node
!=
myself
)
{
if
(
flags
&
(
REDIS
_NODE_FAIL
|
REDIS
_NODE_PFAIL
))
{
if
(
flags
&
(
CLUSTER
_NODE_FAIL
|
CLUSTER
_NODE_PFAIL
))
{
if
(
clusterNodeAddFailureReport
(
node
,
sender
))
{
if
(
clusterNodeAddFailureReport
(
node
,
sender
))
{
serverLog
(
LL_VERBOSE
,
serverLog
(
LL_VERBOSE
,
"Node %.40s reported node %.40s as not reachable."
,
"Node %.40s reported node %.40s as not reachable."
,
...
@@ -1338,7 +1338,7 @@ void clusterProcessGossipSection(clusterMsg *hdr, clusterLink *link) {
...
@@ -1338,7 +1338,7 @@ void clusterProcessGossipSection(clusterMsg *hdr, clusterLink *link) {
* handshake with the (possibly) new address: this will result
* handshake with the (possibly) new address: this will result
* into a node address update if the handshake will be
* into a node address update if the handshake will be
* successful. */
* successful. */
if
(
node
->
flags
&
(
REDIS
_NODE_FAIL
|
REDIS
_NODE_PFAIL
)
&&
if
(
node
->
flags
&
(
CLUSTER
_NODE_FAIL
|
CLUSTER
_NODE_PFAIL
)
&&
(
strcasecmp
(
node
->
ip
,
g
->
ip
)
||
node
->
port
!=
ntohs
(
g
->
port
)))
(
strcasecmp
(
node
->
ip
,
g
->
ip
)
||
node
->
port
!=
ntohs
(
g
->
port
)))
{
{
clusterStartHandshake
(
g
->
ip
,
ntohs
(
g
->
port
));
clusterStartHandshake
(
g
->
ip
,
ntohs
(
g
->
port
));
...
@@ -1351,7 +1351,7 @@ void clusterProcessGossipSection(clusterMsg *hdr, clusterLink *link) {
...
@@ -1351,7 +1351,7 @@ void clusterProcessGossipSection(clusterMsg *hdr, clusterLink *link) {
* is a well known node in our cluster, otherwise we risk
* is a well known node in our cluster, otherwise we risk
* joining another cluster. */
* joining another cluster. */
if
(
sender
&&
if
(
sender
&&
!
(
flags
&
REDIS
_NODE_NOADDR
)
&&
!
(
flags
&
CLUSTER
_NODE_NOADDR
)
&&
!
clusterBlacklistExists
(
g
->
nodename
))
!
clusterBlacklistExists
(
g
->
nodename
))
{
{
clusterStartHandshake
(
g
->
ip
,
ntohs
(
g
->
port
));
clusterStartHandshake
(
g
->
ip
,
ntohs
(
g
->
port
));
...
@@ -1396,7 +1396,7 @@ int nodeUpdateAddressIfNeeded(clusterNode *node, clusterLink *link, int port) {
...
@@ -1396,7 +1396,7 @@ int nodeUpdateAddressIfNeeded(clusterNode *node, clusterLink *link, int port) {
memcpy
(
node
->
ip
,
ip
,
sizeof
(
ip
));
memcpy
(
node
->
ip
,
ip
,
sizeof
(
ip
));
node
->
port
=
port
;
node
->
port
=
port
;
if
(
node
->
link
)
freeClusterLink
(
node
->
link
);
if
(
node
->
link
)
freeClusterLink
(
node
->
link
);
node
->
flags
&=
~
REDIS
_NODE_NOADDR
;
node
->
flags
&=
~
CLUSTER
_NODE_NOADDR
;
serverLog
(
LL_WARNING
,
"Address updated for node %.40s, now %s:%d"
,
serverLog
(
LL_WARNING
,
"Address updated for node %.40s, now %s:%d"
,
node
->
name
,
node
->
ip
,
node
->
port
);
node
->
name
,
node
->
ip
,
node
->
port
);
...
@@ -1414,8 +1414,8 @@ void clusterSetNodeAsMaster(clusterNode *n) {
...
@@ -1414,8 +1414,8 @@ void clusterSetNodeAsMaster(clusterNode *n) {
if
(
nodeIsMaster
(
n
))
return
;
if
(
nodeIsMaster
(
n
))
return
;
if
(
n
->
slaveof
)
clusterNodeRemoveSlave
(
n
->
slaveof
,
n
);
if
(
n
->
slaveof
)
clusterNodeRemoveSlave
(
n
->
slaveof
,
n
);
n
->
flags
&=
~
REDIS
_NODE_SLAVE
;
n
->
flags
&=
~
CLUSTER
_NODE_SLAVE
;
n
->
flags
|=
REDIS
_NODE_MASTER
;
n
->
flags
|=
CLUSTER
_NODE_MASTER
;
n
->
slaveof
=
NULL
;
n
->
slaveof
=
NULL
;
/* Update config and state. */
/* Update config and state. */
...
@@ -1444,7 +1444,7 @@ void clusterUpdateSlotsConfigWith(clusterNode *sender, uint64_t senderConfigEpoc
...
@@ -1444,7 +1444,7 @@ void clusterUpdateSlotsConfigWith(clusterNode *sender, uint64_t senderConfigEpoc
* If the update message is not able to demote a master to slave (in this
* If the update message is not able to demote a master to slave (in this
* case we'll resync with the master updating the whole key space), we
* case we'll resync with the master updating the whole key space), we
* need to delete all the keys in the slots we lost ownership. */
* need to delete all the keys in the slots we lost ownership. */
uint16_t
dirty_slots
[
REDIS_
CLUSTER_SLOTS
];
uint16_t
dirty_slots
[
CLUSTER_SLOTS
];
int
dirty_slots_count
=
0
;
int
dirty_slots_count
=
0
;
/* Here we set curmaster to this node or the node this node
/* Here we set curmaster to this node or the node this node
...
@@ -1457,7 +1457,7 @@ void clusterUpdateSlotsConfigWith(clusterNode *sender, uint64_t senderConfigEpoc
...
@@ -1457,7 +1457,7 @@ void clusterUpdateSlotsConfigWith(clusterNode *sender, uint64_t senderConfigEpoc
return
;
return
;
}
}
for
(
j
=
0
;
j
<
REDIS_
CLUSTER_SLOTS
;
j
++
)
{
for
(
j
=
0
;
j
<
CLUSTER_SLOTS
;
j
++
)
{
if
(
bitmapTestBit
(
slots
,
j
))
{
if
(
bitmapTestBit
(
slots
,
j
))
{
/* The slot is already bound to the sender of this message. */
/* The slot is already bound to the sender of this message. */
if
(
server
.
cluster
->
slots
[
j
]
==
sender
)
continue
;
if
(
server
.
cluster
->
slots
[
j
]
==
sender
)
continue
;
...
@@ -1654,7 +1654,7 @@ int clusterProcessPacket(clusterLink *link) {
...
@@ -1654,7 +1654,7 @@ int clusterProcessPacket(clusterLink *link) {
if
(
!
sender
&&
type
==
CLUSTERMSG_TYPE_MEET
)
{
if
(
!
sender
&&
type
==
CLUSTERMSG_TYPE_MEET
)
{
clusterNode
*
node
;
clusterNode
*
node
;
node
=
createClusterNode
(
NULL
,
REDIS
_NODE_HANDSHAKE
);
node
=
createClusterNode
(
NULL
,
CLUSTER
_NODE_HANDSHAKE
);
nodeIp2String
(
node
->
ip
,
link
);
nodeIp2String
(
node
->
ip
,
link
);
node
->
port
=
ntohs
(
hdr
->
port
);
node
->
port
=
ntohs
(
hdr
->
port
);
clusterAddNode
(
node
);
clusterAddNode
(
node
);
...
@@ -1702,17 +1702,17 @@ int clusterProcessPacket(clusterLink *link) {
...
@@ -1702,17 +1702,17 @@ int clusterProcessPacket(clusterLink *link) {
clusterRenameNode
(
link
->
node
,
hdr
->
sender
);
clusterRenameNode
(
link
->
node
,
hdr
->
sender
);
serverLog
(
LL_DEBUG
,
"Handshake with node %.40s completed."
,
serverLog
(
LL_DEBUG
,
"Handshake with node %.40s completed."
,
link
->
node
->
name
);
link
->
node
->
name
);
link
->
node
->
flags
&=
~
REDIS
_NODE_HANDSHAKE
;
link
->
node
->
flags
&=
~
CLUSTER
_NODE_HANDSHAKE
;
link
->
node
->
flags
|=
flags
&
(
REDIS
_NODE_MASTER
|
REDIS
_NODE_SLAVE
);
link
->
node
->
flags
|=
flags
&
(
CLUSTER
_NODE_MASTER
|
CLUSTER
_NODE_SLAVE
);
clusterDoBeforeSleep
(
CLUSTER_TODO_SAVE_CONFIG
);
clusterDoBeforeSleep
(
CLUSTER_TODO_SAVE_CONFIG
);
}
else
if
(
memcmp
(
link
->
node
->
name
,
hdr
->
sender
,
}
else
if
(
memcmp
(
link
->
node
->
name
,
hdr
->
sender
,
REDIS_
CLUSTER_NAMELEN
)
!=
0
)
CLUSTER_NAMELEN
)
!=
0
)
{
{
/* If the reply has a non matching node ID we
/* If the reply has a non matching node ID we
* disconnect this node and set it as not having an associated
* disconnect this node and set it as not having an associated
* address. */
* address. */
serverLog
(
LL_DEBUG
,
"PONG contains mismatching sender ID"
);
serverLog
(
LL_DEBUG
,
"PONG contains mismatching sender ID"
);
link
->
node
->
flags
|=
REDIS
_NODE_NOADDR
;
link
->
node
->
flags
|=
CLUSTER
_NODE_NOADDR
;
link
->
node
->
ip
[
0
]
=
'\0'
;
link
->
node
->
ip
[
0
]
=
'\0'
;
link
->
node
->
port
=
0
;
link
->
node
->
port
=
0
;
freeClusterLink
(
link
);
freeClusterLink
(
link
);
...
@@ -1742,7 +1742,7 @@ int clusterProcessPacket(clusterLink *link) {
...
@@ -1742,7 +1742,7 @@ int clusterProcessPacket(clusterLink *link) {
* The FAIL condition is also reversible under specific
* The FAIL condition is also reversible under specific
* conditions detected by clearNodeFailureIfNeeded(). */
* conditions detected by clearNodeFailureIfNeeded(). */
if
(
nodeTimedOut
(
link
->
node
))
{
if
(
nodeTimedOut
(
link
->
node
))
{
link
->
node
->
flags
&=
~
REDIS
_NODE_PFAIL
;
link
->
node
->
flags
&=
~
CLUSTER
_NODE_PFAIL
;
clusterDoBeforeSleep
(
CLUSTER_TODO_SAVE_CONFIG
|
clusterDoBeforeSleep
(
CLUSTER_TODO_SAVE_CONFIG
|
CLUSTER_TODO_UPDATE_STATE
);
CLUSTER_TODO_UPDATE_STATE
);
}
else
if
(
nodeFailed
(
link
->
node
))
{
}
else
if
(
nodeFailed
(
link
->
node
))
{
...
@@ -1752,7 +1752,7 @@ int clusterProcessPacket(clusterLink *link) {
...
@@ -1752,7 +1752,7 @@ int clusterProcessPacket(clusterLink *link) {
/* Check for role switch: slave -> master or master -> slave. */
/* Check for role switch: slave -> master or master -> slave. */
if
(
sender
)
{
if
(
sender
)
{
if
(
!
memcmp
(
hdr
->
slaveof
,
REDIS
_NODE_NULL_NAME
,
if
(
!
memcmp
(
hdr
->
slaveof
,
CLUSTER
_NODE_NULL_NAME
,
sizeof
(
hdr
->
slaveof
)))
sizeof
(
hdr
->
slaveof
)))
{
{
/* Node is a master. */
/* Node is a master. */
...
@@ -1764,8 +1764,8 @@ int clusterProcessPacket(clusterLink *link) {
...
@@ -1764,8 +1764,8 @@ int clusterProcessPacket(clusterLink *link) {
if
(
nodeIsMaster
(
sender
))
{
if
(
nodeIsMaster
(
sender
))
{
/* Master turned into a slave! Reconfigure the node. */
/* Master turned into a slave! Reconfigure the node. */
clusterDelNodeSlots
(
sender
);
clusterDelNodeSlots
(
sender
);
sender
->
flags
&=
~
REDIS
_NODE_MASTER
;
sender
->
flags
&=
~
CLUSTER
_NODE_MASTER
;
sender
->
flags
|=
REDIS
_NODE_SLAVE
;
sender
->
flags
|=
CLUSTER
_NODE_SLAVE
;
/* Remove the list of slaves from the node. */
/* Remove the list of slaves from the node. */
if
(
sender
->
numslaves
)
clusterNodeResetSlaves
(
sender
);
if
(
sender
->
numslaves
)
clusterNodeResetSlaves
(
sender
);
...
@@ -1791,7 +1791,7 @@ int clusterProcessPacket(clusterLink *link) {
...
@@ -1791,7 +1791,7 @@ int clusterProcessPacket(clusterLink *link) {
/* Update our info about served slots.
/* Update our info about served slots.
*
*
* Note: this MUST happen after we update the master/slave state
* Note: this MUST happen after we update the master/slave state
* so that
REDIS
_NODE_MASTER flag will be set. */
* so that
CLUSTER
_NODE_MASTER flag will be set. */
/* Many checks are only needed if the set of served slots this
/* Many checks are only needed if the set of served slots this
* instance claims is different compared to the set of slots we have
* instance claims is different compared to the set of slots we have
...
@@ -1835,7 +1835,7 @@ int clusterProcessPacket(clusterLink *link) {
...
@@ -1835,7 +1835,7 @@ int clusterProcessPacket(clusterLink *link) {
if
(
sender
&&
dirty_slots
)
{
if
(
sender
&&
dirty_slots
)
{
int
j
;
int
j
;
for
(
j
=
0
;
j
<
REDIS_
CLUSTER_SLOTS
;
j
++
)
{
for
(
j
=
0
;
j
<
CLUSTER_SLOTS
;
j
++
)
{
if
(
bitmapTestBit
(
hdr
->
myslots
,
j
))
{
if
(
bitmapTestBit
(
hdr
->
myslots
,
j
))
{
if
(
server
.
cluster
->
slots
[
j
]
==
sender
||
if
(
server
.
cluster
->
slots
[
j
]
==
sender
||
server
.
cluster
->
slots
[
j
]
==
NULL
)
continue
;
server
.
cluster
->
slots
[
j
]
==
NULL
)
continue
;
...
@@ -1875,14 +1875,14 @@ int clusterProcessPacket(clusterLink *link) {
...
@@ -1875,14 +1875,14 @@ int clusterProcessPacket(clusterLink *link) {
if
(
sender
)
{
if
(
sender
)
{
failing
=
clusterLookupNode
(
hdr
->
data
.
fail
.
about
.
nodename
);
failing
=
clusterLookupNode
(
hdr
->
data
.
fail
.
about
.
nodename
);
if
(
failing
&&
if
(
failing
&&
!
(
failing
->
flags
&
(
REDIS
_NODE_FAIL
|
REDIS
_NODE_MYSELF
)))
!
(
failing
->
flags
&
(
CLUSTER
_NODE_FAIL
|
CLUSTER
_NODE_MYSELF
)))
{
{
serverLog
(
LL_NOTICE
,
serverLog
(
LL_NOTICE
,
"FAIL message received from %.40s about %.40s"
,
"FAIL message received from %.40s about %.40s"
,
hdr
->
sender
,
hdr
->
data
.
fail
.
about
.
nodename
);
hdr
->
sender
,
hdr
->
data
.
fail
.
about
.
nodename
);
failing
->
flags
|=
REDIS
_NODE_FAIL
;
failing
->
flags
|=
CLUSTER
_NODE_FAIL
;
failing
->
fail_time
=
mstime
();
failing
->
fail_time
=
mstime
();
failing
->
flags
&=
~
REDIS
_NODE_PFAIL
;
failing
->
flags
&=
~
CLUSTER
_NODE_PFAIL
;
clusterDoBeforeSleep
(
CLUSTER_TODO_SAVE_CONFIG
|
clusterDoBeforeSleep
(
CLUSTER_TODO_SAVE_CONFIG
|
CLUSTER_TODO_UPDATE_STATE
);
CLUSTER_TODO_UPDATE_STATE
);
}
}
...
@@ -1934,9 +1934,9 @@ int clusterProcessPacket(clusterLink *link) {
...
@@ -1934,9 +1934,9 @@ int clusterProcessPacket(clusterLink *link) {
/* Manual failover requested from slaves. Initialize the state
/* Manual failover requested from slaves. Initialize the state
* accordingly. */
* accordingly. */
resetManualFailover
();
resetManualFailover
();
server
.
cluster
->
mf_end
=
mstime
()
+
REDIS_
CLUSTER_MF_TIMEOUT
;
server
.
cluster
->
mf_end
=
mstime
()
+
CLUSTER_MF_TIMEOUT
;
server
.
cluster
->
mf_slave
=
sender
;
server
.
cluster
->
mf_slave
=
sender
;
pauseClients
(
mstime
()
+
(
REDIS_
CLUSTER_MF_TIMEOUT
*
2
));
pauseClients
(
mstime
()
+
(
CLUSTER_MF_TIMEOUT
*
2
));
serverLog
(
LL_WARNING
,
"Manual failover requested by slave %.40s."
,
serverLog
(
LL_WARNING
,
"Manual failover requested by slave %.40s."
,
sender
->
name
);
sender
->
name
);
}
else
if
(
type
==
CLUSTERMSG_TYPE_UPDATE
)
{
}
else
if
(
type
==
CLUSTERMSG_TYPE_UPDATE
)
{
...
@@ -2093,7 +2093,7 @@ void clusterBroadcastMessage(void *buf, size_t len) {
...
@@ -2093,7 +2093,7 @@ void clusterBroadcastMessage(void *buf, size_t len) {
clusterNode
*
node
=
dictGetVal
(
de
);
clusterNode
*
node
=
dictGetVal
(
de
);
if
(
!
node
->
link
)
continue
;
if
(
!
node
->
link
)
continue
;
if
(
node
->
flags
&
(
REDIS
_NODE_MYSELF
|
REDIS
_NODE_HANDSHAKE
))
if
(
node
->
flags
&
(
CLUSTER
_NODE_MYSELF
|
CLUSTER
_NODE_HANDSHAKE
))
continue
;
continue
;
clusterSendMessage
(
node
->
link
,
buf
,
len
);
clusterSendMessage
(
node
->
link
,
buf
,
len
);
}
}
...
@@ -2121,12 +2121,12 @@ void clusterBuildMessageHdr(clusterMsg *hdr, int type) {
...
@@ -2121,12 +2121,12 @@ void clusterBuildMessageHdr(clusterMsg *hdr, int type) {
hdr
->
sig
[
2
]
=
'm'
;
hdr
->
sig
[
2
]
=
'm'
;
hdr
->
sig
[
3
]
=
'b'
;
hdr
->
sig
[
3
]
=
'b'
;
hdr
->
type
=
htons
(
type
);
hdr
->
type
=
htons
(
type
);
memcpy
(
hdr
->
sender
,
myself
->
name
,
REDIS_
CLUSTER_NAMELEN
);
memcpy
(
hdr
->
sender
,
myself
->
name
,
CLUSTER_NAMELEN
);
memcpy
(
hdr
->
myslots
,
master
->
slots
,
sizeof
(
hdr
->
myslots
));
memcpy
(
hdr
->
myslots
,
master
->
slots
,
sizeof
(
hdr
->
myslots
));
memset
(
hdr
->
slaveof
,
0
,
REDIS_
CLUSTER_NAMELEN
);
memset
(
hdr
->
slaveof
,
0
,
CLUSTER_NAMELEN
);
if
(
myself
->
slaveof
!=
NULL
)
if
(
myself
->
slaveof
!=
NULL
)
memcpy
(
hdr
->
slaveof
,
myself
->
slaveof
->
name
,
REDIS_
CLUSTER_NAMELEN
);
memcpy
(
hdr
->
slaveof
,
myself
->
slaveof
->
name
,
CLUSTER_NAMELEN
);
hdr
->
port
=
htons
(
server
.
port
);
hdr
->
port
=
htons
(
server
.
port
);
hdr
->
flags
=
htons
(
myself
->
flags
);
hdr
->
flags
=
htons
(
myself
->
flags
);
hdr
->
state
=
server
.
cluster
->
state
;
hdr
->
state
=
server
.
cluster
->
state
;
...
@@ -2233,7 +2233,7 @@ void clusterSendPing(clusterLink *link, int type) {
...
@@ -2233,7 +2233,7 @@ void clusterSendPing(clusterLink *link, int type) {
/* Give a bias to FAIL/PFAIL nodes. */
/* Give a bias to FAIL/PFAIL nodes. */
if
(
maxiterations
>
wanted
*
2
&&
if
(
maxiterations
>
wanted
*
2
&&
!
(
this
->
flags
&
(
REDIS
_NODE_PFAIL
|
REDIS
_NODE_FAIL
)))
!
(
this
->
flags
&
(
CLUSTER
_NODE_PFAIL
|
CLUSTER
_NODE_FAIL
)))
continue
;
continue
;
/* In the gossip section don't include:
/* In the gossip section don't include:
...
@@ -2241,7 +2241,7 @@ void clusterSendPing(clusterLink *link, int type) {
...
@@ -2241,7 +2241,7 @@ void clusterSendPing(clusterLink *link, int type) {
* 3) Nodes with the NOADDR flag set.
* 3) Nodes with the NOADDR flag set.
* 4) Disconnected nodes if they don't have configured slots.
* 4) Disconnected nodes if they don't have configured slots.
*/
*/
if
(
this
->
flags
&
(
REDIS
_NODE_HANDSHAKE
|
REDIS
_NODE_NOADDR
)
||
if
(
this
->
flags
&
(
CLUSTER
_NODE_HANDSHAKE
|
CLUSTER
_NODE_NOADDR
)
||
(
this
->
link
==
NULL
&&
this
->
numslots
==
0
))
(
this
->
link
==
NULL
&&
this
->
numslots
==
0
))
{
{
freshnodes
--
;
/* Tecnically not correct, but saves CPU. */
freshnodes
--
;
/* Tecnically not correct, but saves CPU. */
...
@@ -2251,14 +2251,14 @@ void clusterSendPing(clusterLink *link, int type) {
...
@@ -2251,14 +2251,14 @@ void clusterSendPing(clusterLink *link, int type) {
/* Check if we already added this node */
/* Check if we already added this node */
for
(
j
=
0
;
j
<
gossipcount
;
j
++
)
{
for
(
j
=
0
;
j
<
gossipcount
;
j
++
)
{
if
(
memcmp
(
hdr
->
data
.
ping
.
gossip
[
j
].
nodename
,
this
->
name
,
if
(
memcmp
(
hdr
->
data
.
ping
.
gossip
[
j
].
nodename
,
this
->
name
,
REDIS_
CLUSTER_NAMELEN
)
==
0
)
break
;
CLUSTER_NAMELEN
)
==
0
)
break
;
}
}
if
(
j
!=
gossipcount
)
continue
;
if
(
j
!=
gossipcount
)
continue
;
/* Add it */
/* Add it */
freshnodes
--
;
freshnodes
--
;
gossip
=
&
(
hdr
->
data
.
ping
.
gossip
[
gossipcount
]);
gossip
=
&
(
hdr
->
data
.
ping
.
gossip
[
gossipcount
]);
memcpy
(
gossip
->
nodename
,
this
->
name
,
REDIS_
CLUSTER_NAMELEN
);
memcpy
(
gossip
->
nodename
,
this
->
name
,
CLUSTER_NAMELEN
);
gossip
->
ping_sent
=
htonl
(
this
->
ping_sent
);
gossip
->
ping_sent
=
htonl
(
this
->
ping_sent
);
gossip
->
pong_received
=
htonl
(
this
->
pong_received
);
gossip
->
pong_received
=
htonl
(
this
->
pong_received
);
memcpy
(
gossip
->
ip
,
this
->
ip
,
sizeof
(
this
->
ip
));
memcpy
(
gossip
->
ip
,
this
->
ip
,
sizeof
(
this
->
ip
));
...
@@ -2362,15 +2362,15 @@ void clusterSendPublish(clusterLink *link, robj *channel, robj *message) {
...
@@ -2362,15 +2362,15 @@ void clusterSendPublish(clusterLink *link, robj *channel, robj *message) {
/* Send a FAIL message to all the nodes we are able to contact.
/* Send a FAIL message to all the nodes we are able to contact.
* The FAIL message is sent when we detect that a node is failing
* The FAIL message is sent when we detect that a node is failing
* (
REDIS
_NODE_PFAIL) and we also receive a gossip confirmation of this:
* (
CLUSTER
_NODE_PFAIL) and we also receive a gossip confirmation of this:
* we switch the node state to
REDIS
_NODE_FAIL and ask all the other
* we switch the node state to
CLUSTER
_NODE_FAIL and ask all the other
* nodes to do the same ASAP. */
* nodes to do the same ASAP. */
void
clusterSendFail
(
char
*
nodename
)
{
void
clusterSendFail
(
char
*
nodename
)
{
unsigned
char
buf
[
sizeof
(
clusterMsg
)];
unsigned
char
buf
[
sizeof
(
clusterMsg
)];
clusterMsg
*
hdr
=
(
clusterMsg
*
)
buf
;
clusterMsg
*
hdr
=
(
clusterMsg
*
)
buf
;
clusterBuildMessageHdr
(
hdr
,
CLUSTERMSG_TYPE_FAIL
);
clusterBuildMessageHdr
(
hdr
,
CLUSTERMSG_TYPE_FAIL
);
memcpy
(
hdr
->
data
.
fail
.
about
.
nodename
,
nodename
,
REDIS_
CLUSTER_NAMELEN
);
memcpy
(
hdr
->
data
.
fail
.
about
.
nodename
,
nodename
,
CLUSTER_NAMELEN
);
clusterBroadcastMessage
(
buf
,
ntohl
(
hdr
->
totlen
));
clusterBroadcastMessage
(
buf
,
ntohl
(
hdr
->
totlen
));
}
}
...
@@ -2383,7 +2383,7 @@ void clusterSendUpdate(clusterLink *link, clusterNode *node) {
...
@@ -2383,7 +2383,7 @@ void clusterSendUpdate(clusterLink *link, clusterNode *node) {
if
(
link
==
NULL
)
return
;
if
(
link
==
NULL
)
return
;
clusterBuildMessageHdr
(
hdr
,
CLUSTERMSG_TYPE_UPDATE
);
clusterBuildMessageHdr
(
hdr
,
CLUSTERMSG_TYPE_UPDATE
);
memcpy
(
hdr
->
data
.
update
.
nodecfg
.
nodename
,
node
->
name
,
REDIS_
CLUSTER_NAMELEN
);
memcpy
(
hdr
->
data
.
update
.
nodecfg
.
nodename
,
node
->
name
,
CLUSTER_NAMELEN
);
hdr
->
data
.
update
.
nodecfg
.
configEpoch
=
htonu64
(
node
->
configEpoch
);
hdr
->
data
.
update
.
nodecfg
.
configEpoch
=
htonu64
(
node
->
configEpoch
);
memcpy
(
hdr
->
data
.
update
.
nodecfg
.
slots
,
node
->
slots
,
sizeof
(
node
->
slots
));
memcpy
(
hdr
->
data
.
update
.
nodecfg
.
slots
,
node
->
slots
,
sizeof
(
node
->
slots
));
clusterSendMessage
(
link
,
buf
,
ntohl
(
hdr
->
totlen
));
clusterSendMessage
(
link
,
buf
,
ntohl
(
hdr
->
totlen
));
...
@@ -2527,7 +2527,7 @@ void clusterSendFailoverAuthIfNeeded(clusterNode *node, clusterMsg *request) {
...
@@ -2527,7 +2527,7 @@ void clusterSendFailoverAuthIfNeeded(clusterNode *node, clusterMsg *request) {
/* The slave requesting the vote must have a configEpoch for the claimed
/* The slave requesting the vote must have a configEpoch for the claimed
* slots that is >= the one of the masters currently serving the same
* slots that is >= the one of the masters currently serving the same
* slots in the current configuration. */
* slots in the current configuration. */
for
(
j
=
0
;
j
<
REDIS_
CLUSTER_SLOTS
;
j
++
)
{
for
(
j
=
0
;
j
<
CLUSTER_SLOTS
;
j
++
)
{
if
(
bitmapTestBit
(
claimed_slots
,
j
)
==
0
)
continue
;
if
(
bitmapTestBit
(
claimed_slots
,
j
)
==
0
)
continue
;
if
(
server
.
cluster
->
slots
[
j
]
==
NULL
||
if
(
server
.
cluster
->
slots
[
j
]
==
NULL
||
server
.
cluster
->
slots
[
j
]
->
configEpoch
<=
requestConfigEpoch
)
server
.
cluster
->
slots
[
j
]
->
configEpoch
<=
requestConfigEpoch
)
...
@@ -2595,13 +2595,13 @@ int clusterGetSlaveRank(void) {
...
@@ -2595,13 +2595,13 @@ int clusterGetSlaveRank(void) {
* when the slave finds that its master is fine (no FAIL flag).
* when the slave finds that its master is fine (no FAIL flag).
* 2) Also, the log is emitted again if the master is still down and
* 2) Also, the log is emitted again if the master is still down and
* the reason for not failing over is still the same, but more than
* the reason for not failing over is still the same, but more than
*
REDIS_
CLUSTER_CANT_FAILOVER_RELOG_PERIOD seconds elapsed.
* CLUSTER_CANT_FAILOVER_RELOG_PERIOD seconds elapsed.
* 3) Finally, the function only logs if the slave is down for more than
* 3) Finally, the function only logs if the slave is down for more than
* five seconds + NODE_TIMEOUT. This way nothing is logged when a
* five seconds + NODE_TIMEOUT. This way nothing is logged when a
* failover starts in a reasonable time.
* failover starts in a reasonable time.
*
*
* The function is called with the reason why the slave can't failover
* The function is called with the reason why the slave can't failover
* which is one of the integer macros
REDIS_
CLUSTER_CANT_FAILOVER_*.
* which is one of the integer macros CLUSTER_CANT_FAILOVER_*.
*
*
* The function is guaranteed to be called only if 'myself' is a slave. */
* The function is guaranteed to be called only if 'myself' is a slave. */
void
clusterLogCantFailover
(
int
reason
)
{
void
clusterLogCantFailover
(
int
reason
)
{
...
@@ -2611,7 +2611,7 @@ void clusterLogCantFailover(int reason) {
...
@@ -2611,7 +2611,7 @@ void clusterLogCantFailover(int reason) {
/* Don't log if we have the same reason for some time. */
/* Don't log if we have the same reason for some time. */
if
(
reason
==
server
.
cluster
->
cant_failover_reason
&&
if
(
reason
==
server
.
cluster
->
cant_failover_reason
&&
time
(
NULL
)
-
lastlog_time
<
REDIS_
CLUSTER_CANT_FAILOVER_RELOG_PERIOD
)
time
(
NULL
)
-
lastlog_time
<
CLUSTER_CANT_FAILOVER_RELOG_PERIOD
)
return
;
return
;
server
.
cluster
->
cant_failover_reason
=
reason
;
server
.
cluster
->
cant_failover_reason
=
reason
;
...
@@ -2624,16 +2624,16 @@ void clusterLogCantFailover(int reason) {
...
@@ -2624,16 +2624,16 @@ void clusterLogCantFailover(int reason) {
(
mstime
()
-
myself
->
slaveof
->
fail_time
)
<
nolog_fail_time
)
return
;
(
mstime
()
-
myself
->
slaveof
->
fail_time
)
<
nolog_fail_time
)
return
;
switch
(
reason
)
{
switch
(
reason
)
{
case
REDIS_
CLUSTER_CANT_FAILOVER_DATA_AGE
:
case
CLUSTER_CANT_FAILOVER_DATA_AGE
:
msg
=
"Disconnected from master for longer than allowed."
;
msg
=
"Disconnected from master for longer than allowed."
;
break
;
break
;
case
REDIS_
CLUSTER_CANT_FAILOVER_WAITING_DELAY
:
case
CLUSTER_CANT_FAILOVER_WAITING_DELAY
:
msg
=
"Waiting the delay before I can start a new failover."
;
msg
=
"Waiting the delay before I can start a new failover."
;
break
;
break
;
case
REDIS_
CLUSTER_CANT_FAILOVER_EXPIRED
:
case
CLUSTER_CANT_FAILOVER_EXPIRED
:
msg
=
"Failover attempt expired."
;
msg
=
"Failover attempt expired."
;
break
;
break
;
case
REDIS_
CLUSTER_CANT_FAILOVER_WAITING_VOTES
:
case
CLUSTER_CANT_FAILOVER_WAITING_VOTES
:
msg
=
"Waiting for votes, but majority still not reached."
;
msg
=
"Waiting for votes, but majority still not reached."
;
break
;
break
;
default:
default:
...
@@ -2661,7 +2661,7 @@ void clusterFailoverReplaceYourMaster(void) {
...
@@ -2661,7 +2661,7 @@ void clusterFailoverReplaceYourMaster(void) {
replicationUnsetMaster
();
replicationUnsetMaster
();
/* 2) Claim all the slots assigned to our master. */
/* 2) Claim all the slots assigned to our master. */
for
(
j
=
0
;
j
<
REDIS_
CLUSTER_SLOTS
;
j
++
)
{
for
(
j
=
0
;
j
<
CLUSTER_SLOTS
;
j
++
)
{
if
(
clusterNodeGetSlotBit
(
oldmaster
,
j
))
{
if
(
clusterNodeGetSlotBit
(
oldmaster
,
j
))
{
clusterDelSlot
(
j
);
clusterDelSlot
(
j
);
clusterAddSlot
(
myself
,
j
);
clusterAddSlot
(
myself
,
j
);
...
@@ -2721,7 +2721,7 @@ void clusterHandleSlaveFailover(void) {
...
@@ -2721,7 +2721,7 @@ void clusterHandleSlaveFailover(void) {
{
{
/* There are no reasons to failover, so we set the reason why we
/* There are no reasons to failover, so we set the reason why we
* are returning without failing over to NONE. */
* are returning without failing over to NONE. */
server
.
cluster
->
cant_failover_reason
=
REDIS_
CLUSTER_CANT_FAILOVER_NONE
;
server
.
cluster
->
cant_failover_reason
=
CLUSTER_CANT_FAILOVER_NONE
;
return
;
return
;
}
}
...
@@ -2750,7 +2750,7 @@ void clusterHandleSlaveFailover(void) {
...
@@ -2750,7 +2750,7 @@ void clusterHandleSlaveFailover(void) {
(
server
.
cluster_node_timeout
*
server
.
cluster_slave_validity_factor
)))
(
server
.
cluster_node_timeout
*
server
.
cluster_slave_validity_factor
)))
{
{
if
(
!
manual_failover
)
{
if
(
!
manual_failover
)
{
clusterLogCantFailover
(
REDIS_
CLUSTER_CANT_FAILOVER_DATA_AGE
);
clusterLogCantFailover
(
CLUSTER_CANT_FAILOVER_DATA_AGE
);
return
;
return
;
}
}
}
}
...
@@ -2809,13 +2809,13 @@ void clusterHandleSlaveFailover(void) {
...
@@ -2809,13 +2809,13 @@ void clusterHandleSlaveFailover(void) {
/* Return ASAP if we can't still start the election. */
/* Return ASAP if we can't still start the election. */
if
(
mstime
()
<
server
.
cluster
->
failover_auth_time
)
{
if
(
mstime
()
<
server
.
cluster
->
failover_auth_time
)
{
clusterLogCantFailover
(
REDIS_
CLUSTER_CANT_FAILOVER_WAITING_DELAY
);
clusterLogCantFailover
(
CLUSTER_CANT_FAILOVER_WAITING_DELAY
);
return
;
return
;
}
}
/* Return ASAP if the election is too old to be valid. */
/* Return ASAP if the election is too old to be valid. */
if
(
auth_age
>
auth_timeout
)
{
if
(
auth_age
>
auth_timeout
)
{
clusterLogCantFailover
(
REDIS_
CLUSTER_CANT_FAILOVER_EXPIRED
);
clusterLogCantFailover
(
CLUSTER_CANT_FAILOVER_EXPIRED
);
return
;
return
;
}
}
...
@@ -2851,7 +2851,7 @@ void clusterHandleSlaveFailover(void) {
...
@@ -2851,7 +2851,7 @@ void clusterHandleSlaveFailover(void) {
/* Take responsability for the cluster slots. */
/* Take responsability for the cluster slots. */
clusterFailoverReplaceYourMaster
();
clusterFailoverReplaceYourMaster
();
}
else
{
}
else
{
clusterLogCantFailover
(
REDIS_
CLUSTER_CANT_FAILOVER_WAITING_VOTES
);
clusterLogCantFailover
(
CLUSTER_CANT_FAILOVER_WAITING_VOTES
);
}
}
}
}
...
@@ -2889,7 +2889,7 @@ void clusterHandleSlaveMigration(int max_slaves) {
...
@@ -2889,7 +2889,7 @@ void clusterHandleSlaveMigration(int max_slaves) {
dictEntry
*
de
;
dictEntry
*
de
;
/* Step 1: Don't migrate if the cluster state is not ok. */
/* Step 1: Don't migrate if the cluster state is not ok. */
if
(
server
.
cluster
->
state
!=
REDIS_
CLUSTER_OK
)
return
;
if
(
server
.
cluster
->
state
!=
CLUSTER_OK
)
return
;
/* Step 2: Don't migrate if my master will not be left with at least
/* Step 2: Don't migrate if my master will not be left with at least
* 'migration-barrier' slaves after my migration. */
* 'migration-barrier' slaves after my migration. */
...
@@ -2929,7 +2929,7 @@ void clusterHandleSlaveMigration(int max_slaves) {
...
@@ -2929,7 +2929,7 @@ void clusterHandleSlaveMigration(int max_slaves) {
for
(
j
=
0
;
j
<
node
->
numslaves
;
j
++
)
{
for
(
j
=
0
;
j
<
node
->
numslaves
;
j
++
)
{
if
(
memcmp
(
node
->
slaves
[
j
]
->
name
,
if
(
memcmp
(
node
->
slaves
[
j
]
->
name
,
candidate
->
name
,
candidate
->
name
,
REDIS_
CLUSTER_NAMELEN
)
<
0
)
CLUSTER_NAMELEN
)
<
0
)
{
{
candidate
=
node
->
slaves
[
j
];
candidate
=
node
->
slaves
[
j
];
}
}
...
@@ -2955,7 +2955,7 @@ void clusterHandleSlaveMigration(int max_slaves) {
...
@@ -2955,7 +2955,7 @@ void clusterHandleSlaveMigration(int max_slaves) {
* setting mf_end to the millisecond unix time at which we'll abort the
* setting mf_end to the millisecond unix time at which we'll abort the
* attempt.
* attempt.
* 2) Slave sends a MFSTART message to the master requesting to pause clients
* 2) Slave sends a MFSTART message to the master requesting to pause clients
* for two times the manual failover timeout
REDIS_
CLUSTER_MF_TIMEOUT.
* for two times the manual failover timeout CLUSTER_MF_TIMEOUT.
* When master is paused for manual failover, it also starts to flag
* When master is paused for manual failover, it also starts to flag
* packets with CLUSTERMSG_FLAG0_PAUSED.
* packets with CLUSTERMSG_FLAG0_PAUSED.
* 3) Slave waits for master to send its replication offset flagged as PAUSED.
* 3) Slave waits for master to send its replication offset flagged as PAUSED.
...
@@ -3053,7 +3053,7 @@ void clusterCron(void) {
...
@@ -3053,7 +3053,7 @@ void clusterCron(void) {
while
((
de
=
dictNext
(
di
))
!=
NULL
)
{
while
((
de
=
dictNext
(
di
))
!=
NULL
)
{
clusterNode
*
node
=
dictGetVal
(
de
);
clusterNode
*
node
=
dictGetVal
(
de
);
if
(
node
->
flags
&
(
REDIS
_NODE_MYSELF
|
REDIS
_NODE_NOADDR
))
continue
;
if
(
node
->
flags
&
(
CLUSTER
_NODE_MYSELF
|
CLUSTER
_NODE_NOADDR
))
continue
;
/* A Node in HANDSHAKE state has a limited lifespan equal to the
/* A Node in HANDSHAKE state has a limited lifespan equal to the
* configured node timeout. */
* configured node timeout. */
...
@@ -3068,7 +3068,7 @@ void clusterCron(void) {
...
@@ -3068,7 +3068,7 @@ void clusterCron(void) {
clusterLink
*
link
;
clusterLink
*
link
;
fd
=
anetTcpNonBlockBindConnect
(
server
.
neterr
,
node
->
ip
,
fd
=
anetTcpNonBlockBindConnect
(
server
.
neterr
,
node
->
ip
,
node
->
port
+
REDIS_
CLUSTER_PORT_INCR
,
NET_FIRST_BIND_ADDR
);
node
->
port
+
CLUSTER_PORT_INCR
,
NET_FIRST_BIND_ADDR
);
if
(
fd
==
-
1
)
{
if
(
fd
==
-
1
)
{
/* We got a synchronous error from connect before
/* We got a synchronous error from connect before
* clusterSendPing() had a chance to be called.
* clusterSendPing() had a chance to be called.
...
@@ -3078,7 +3078,7 @@ void clusterCron(void) {
...
@@ -3078,7 +3078,7 @@ void clusterCron(void) {
if
(
node
->
ping_sent
==
0
)
node
->
ping_sent
=
mstime
();
if
(
node
->
ping_sent
==
0
)
node
->
ping_sent
=
mstime
();
serverLog
(
LL_DEBUG
,
"Unable to connect to "
serverLog
(
LL_DEBUG
,
"Unable to connect to "
"Cluster Node [%s]:%d -> %s"
,
node
->
ip
,
"Cluster Node [%s]:%d -> %s"
,
node
->
ip
,
node
->
port
+
REDIS_
CLUSTER_PORT_INCR
,
node
->
port
+
CLUSTER_PORT_INCR
,
server
.
neterr
);
server
.
neterr
);
continue
;
continue
;
}
}
...
@@ -3094,7 +3094,7 @@ void clusterCron(void) {
...
@@ -3094,7 +3094,7 @@ void clusterCron(void) {
* of a PING one, to force the receiver to add us in its node
* of a PING one, to force the receiver to add us in its node
* table. */
* table. */
old_ping_sent
=
node
->
ping_sent
;
old_ping_sent
=
node
->
ping_sent
;
clusterSendPing
(
link
,
node
->
flags
&
REDIS
_NODE_MEET
?
clusterSendPing
(
link
,
node
->
flags
&
CLUSTER
_NODE_MEET
?
CLUSTERMSG_TYPE_MEET
:
CLUSTERMSG_TYPE_PING
);
CLUSTERMSG_TYPE_MEET
:
CLUSTERMSG_TYPE_PING
);
if
(
old_ping_sent
)
{
if
(
old_ping_sent
)
{
/* If there was an active ping before the link was
/* If there was an active ping before the link was
...
@@ -3107,10 +3107,10 @@ void clusterCron(void) {
...
@@ -3107,10 +3107,10 @@ void clusterCron(void) {
* to this node. Instead after the PONG is received and we
* to this node. Instead after the PONG is received and we
* are no longer in meet/handshake status, we want to send
* are no longer in meet/handshake status, we want to send
* normal PING packets. */
* normal PING packets. */
node
->
flags
&=
~
REDIS
_NODE_MEET
;
node
->
flags
&=
~
CLUSTER
_NODE_MEET
;
serverLog
(
LL_DEBUG
,
"Connecting with Node %.40s at %s:%d"
,
serverLog
(
LL_DEBUG
,
"Connecting with Node %.40s at %s:%d"
,
node
->
name
,
node
->
ip
,
node
->
port
+
REDIS_
CLUSTER_PORT_INCR
);
node
->
name
,
node
->
ip
,
node
->
port
+
CLUSTER_PORT_INCR
);
}
}
}
}
dictReleaseIterator
(
di
);
dictReleaseIterator
(
di
);
...
@@ -3128,7 +3128,7 @@ void clusterCron(void) {
...
@@ -3128,7 +3128,7 @@ void clusterCron(void) {
/* Don't ping nodes disconnected or with a ping currently active. */
/* Don't ping nodes disconnected or with a ping currently active. */
if
(
this
->
link
==
NULL
||
this
->
ping_sent
!=
0
)
continue
;
if
(
this
->
link
==
NULL
||
this
->
ping_sent
!=
0
)
continue
;
if
(
this
->
flags
&
(
REDIS
_NODE_MYSELF
|
REDIS
_NODE_HANDSHAKE
))
if
(
this
->
flags
&
(
CLUSTER
_NODE_MYSELF
|
CLUSTER
_NODE_HANDSHAKE
))
continue
;
continue
;
if
(
min_pong_node
==
NULL
||
min_pong
>
this
->
pong_received
)
{
if
(
min_pong_node
==
NULL
||
min_pong
>
this
->
pong_received
)
{
min_pong_node
=
this
;
min_pong_node
=
this
;
...
@@ -3157,7 +3157,7 @@ void clusterCron(void) {
...
@@ -3157,7 +3157,7 @@ void clusterCron(void) {
mstime_t
delay
;
mstime_t
delay
;
if
(
node
->
flags
&
if
(
node
->
flags
&
(
REDIS
_NODE_MYSELF
|
REDIS
_NODE_NOADDR
|
REDIS
_NODE_HANDSHAKE
))
(
CLUSTER
_NODE_MYSELF
|
CLUSTER
_NODE_NOADDR
|
CLUSTER
_NODE_HANDSHAKE
))
continue
;
continue
;
/* Orphaned master check, useful only if the current instance
/* Orphaned master check, useful only if the current instance
...
@@ -3224,10 +3224,10 @@ void clusterCron(void) {
...
@@ -3224,10 +3224,10 @@ void clusterCron(void) {
if
(
delay
>
server
.
cluster_node_timeout
)
{
if
(
delay
>
server
.
cluster_node_timeout
)
{
/* Timeout reached. Set the node as possibly failing if it is
/* Timeout reached. Set the node as possibly failing if it is
* not already in this state. */
* not already in this state. */
if
(
!
(
node
->
flags
&
(
REDIS
_NODE_PFAIL
|
REDIS
_NODE_FAIL
)))
{
if
(
!
(
node
->
flags
&
(
CLUSTER
_NODE_PFAIL
|
CLUSTER
_NODE_FAIL
)))
{
serverLog
(
LL_DEBUG
,
"*** NODE %.40s possibly failing"
,
serverLog
(
LL_DEBUG
,
"*** NODE %.40s possibly failing"
,
node
->
name
);
node
->
name
);
node
->
flags
|=
REDIS
_NODE_PFAIL
;
node
->
flags
|=
CLUSTER
_NODE_PFAIL
;
update_state
=
1
;
update_state
=
1
;
}
}
}
}
...
@@ -3260,7 +3260,7 @@ void clusterCron(void) {
...
@@ -3260,7 +3260,7 @@ void clusterCron(void) {
clusterHandleSlaveMigration
(
max_slaves
);
clusterHandleSlaveMigration
(
max_slaves
);
}
}
if
(
update_state
||
server
.
cluster
->
state
==
REDIS_
CLUSTER_FAIL
)
if
(
update_state
||
server
.
cluster
->
state
==
CLUSTER_FAIL
)
clusterUpdateState
();
clusterUpdateState
();
}
}
...
@@ -3370,7 +3370,7 @@ int clusterDelSlot(int slot) {
...
@@ -3370,7 +3370,7 @@ int clusterDelSlot(int slot) {
int
clusterDelNodeSlots
(
clusterNode
*
node
)
{
int
clusterDelNodeSlots
(
clusterNode
*
node
)
{
int
deleted
=
0
,
j
;
int
deleted
=
0
,
j
;
for
(
j
=
0
;
j
<
REDIS_
CLUSTER_SLOTS
;
j
++
)
{
for
(
j
=
0
;
j
<
CLUSTER_SLOTS
;
j
++
)
{
if
(
clusterNodeGetSlotBit
(
node
,
j
))
clusterDelSlot
(
j
);
if
(
clusterNodeGetSlotBit
(
node
,
j
))
clusterDelSlot
(
j
);
deleted
++
;
deleted
++
;
}
}
...
@@ -3394,9 +3394,9 @@ void clusterCloseAllSlots(void) {
...
@@ -3394,9 +3394,9 @@ void clusterCloseAllSlots(void) {
* and are based on heuristics. Actaully the main point about the rejoin and
* and are based on heuristics. Actaully the main point about the rejoin and
* writable delay is that they should be a few orders of magnitude larger
* writable delay is that they should be a few orders of magnitude larger
* than the network latency. */
* than the network latency. */
#define
REDIS_
CLUSTER_MAX_REJOIN_DELAY 5000
#define CLUSTER_MAX_REJOIN_DELAY 5000
#define
REDIS_
CLUSTER_MIN_REJOIN_DELAY 500
#define CLUSTER_MIN_REJOIN_DELAY 500
#define
REDIS_
CLUSTER_WRITABLE_DELAY 2000
#define CLUSTER_WRITABLE_DELAY 2000
void
clusterUpdateState
(
void
)
{
void
clusterUpdateState
(
void
)
{
int
j
,
new_state
;
int
j
,
new_state
;
...
@@ -3414,20 +3414,20 @@ void clusterUpdateState(void) {
...
@@ -3414,20 +3414,20 @@ void clusterUpdateState(void) {
* to don't count the DB loading time. */
* to don't count the DB loading time. */
if
(
first_call_time
==
0
)
first_call_time
=
mstime
();
if
(
first_call_time
==
0
)
first_call_time
=
mstime
();
if
(
nodeIsMaster
(
myself
)
&&
if
(
nodeIsMaster
(
myself
)
&&
server
.
cluster
->
state
==
REDIS_
CLUSTER_FAIL
&&
server
.
cluster
->
state
==
CLUSTER_FAIL
&&
mstime
()
-
first_call_time
<
REDIS_
CLUSTER_WRITABLE_DELAY
)
return
;
mstime
()
-
first_call_time
<
CLUSTER_WRITABLE_DELAY
)
return
;
/* Start assuming the state is OK. We'll turn it into FAIL if there
/* Start assuming the state is OK. We'll turn it into FAIL if there
* are the right conditions. */
* are the right conditions. */
new_state
=
REDIS_
CLUSTER_OK
;
new_state
=
CLUSTER_OK
;
/* Check if all the slots are covered. */
/* Check if all the slots are covered. */
if
(
server
.
cluster_require_full_coverage
)
{
if
(
server
.
cluster_require_full_coverage
)
{
for
(
j
=
0
;
j
<
REDIS_
CLUSTER_SLOTS
;
j
++
)
{
for
(
j
=
0
;
j
<
CLUSTER_SLOTS
;
j
++
)
{
if
(
server
.
cluster
->
slots
[
j
]
==
NULL
||
if
(
server
.
cluster
->
slots
[
j
]
==
NULL
||
server
.
cluster
->
slots
[
j
]
->
flags
&
(
REDIS
_NODE_FAIL
))
server
.
cluster
->
slots
[
j
]
->
flags
&
(
CLUSTER
_NODE_FAIL
))
{
{
new_state
=
REDIS_
CLUSTER_FAIL
;
new_state
=
CLUSTER_FAIL
;
break
;
break
;
}
}
}
}
...
@@ -3449,7 +3449,7 @@ void clusterUpdateState(void) {
...
@@ -3449,7 +3449,7 @@ void clusterUpdateState(void) {
if
(
nodeIsMaster
(
node
)
&&
node
->
numslots
)
{
if
(
nodeIsMaster
(
node
)
&&
node
->
numslots
)
{
server
.
cluster
->
size
++
;
server
.
cluster
->
size
++
;
if
((
node
->
flags
&
(
REDIS
_NODE_FAIL
|
REDIS
_NODE_PFAIL
))
==
0
)
if
((
node
->
flags
&
(
CLUSTER
_NODE_FAIL
|
CLUSTER
_NODE_PFAIL
))
==
0
)
reachable_masters
++
;
reachable_masters
++
;
}
}
}
}
...
@@ -3462,7 +3462,7 @@ void clusterUpdateState(void) {
...
@@ -3462,7 +3462,7 @@ void clusterUpdateState(void) {
int
needed_quorum
=
(
server
.
cluster
->
size
/
2
)
+
1
;
int
needed_quorum
=
(
server
.
cluster
->
size
/
2
)
+
1
;
if
(
reachable_masters
<
needed_quorum
)
{
if
(
reachable_masters
<
needed_quorum
)
{
new_state
=
REDIS_
CLUSTER_FAIL
;
new_state
=
CLUSTER_FAIL
;
among_minority_time
=
mstime
();
among_minority_time
=
mstime
();
}
}
}
}
...
@@ -3475,12 +3475,12 @@ void clusterUpdateState(void) {
...
@@ -3475,12 +3475,12 @@ void clusterUpdateState(void) {
* minority, don't let it accept queries for some time after the
* minority, don't let it accept queries for some time after the
* partition heals, to make sure there is enough time to receive
* partition heals, to make sure there is enough time to receive
* a configuration update. */
* a configuration update. */
if
(
rejoin_delay
>
REDIS_
CLUSTER_MAX_REJOIN_DELAY
)
if
(
rejoin_delay
>
CLUSTER_MAX_REJOIN_DELAY
)
rejoin_delay
=
REDIS_
CLUSTER_MAX_REJOIN_DELAY
;
rejoin_delay
=
CLUSTER_MAX_REJOIN_DELAY
;
if
(
rejoin_delay
<
REDIS_
CLUSTER_MIN_REJOIN_DELAY
)
if
(
rejoin_delay
<
CLUSTER_MIN_REJOIN_DELAY
)
rejoin_delay
=
REDIS_
CLUSTER_MIN_REJOIN_DELAY
;
rejoin_delay
=
CLUSTER_MIN_REJOIN_DELAY
;
if
(
new_state
==
REDIS_
CLUSTER_OK
&&
if
(
new_state
==
CLUSTER_OK
&&
nodeIsMaster
(
myself
)
&&
nodeIsMaster
(
myself
)
&&
mstime
()
-
among_minority_time
<
rejoin_delay
)
mstime
()
-
among_minority_time
<
rejoin_delay
)
{
{
...
@@ -3489,7 +3489,7 @@ void clusterUpdateState(void) {
...
@@ -3489,7 +3489,7 @@ void clusterUpdateState(void) {
/* Change the state and log the event. */
/* Change the state and log the event. */
serverLog
(
LL_WARNING
,
"Cluster state changed: %s"
,
serverLog
(
LL_WARNING
,
"Cluster state changed: %s"
,
new_state
==
REDIS_
CLUSTER_OK
?
"ok"
:
"fail"
);
new_state
==
CLUSTER_OK
?
"ok"
:
"fail"
);
server
.
cluster
->
state
=
new_state
;
server
.
cluster
->
state
=
new_state
;
}
}
}
}
...
@@ -3531,7 +3531,7 @@ int verifyClusterConfigWithData(void) {
...
@@ -3531,7 +3531,7 @@ int verifyClusterConfigWithData(void) {
/* Check that all the slots we see populated memory have a corresponding
/* Check that all the slots we see populated memory have a corresponding
* entry in the cluster table. Otherwise fix the table. */
* entry in the cluster table. Otherwise fix the table. */
for
(
j
=
0
;
j
<
REDIS_
CLUSTER_SLOTS
;
j
++
)
{
for
(
j
=
0
;
j
<
CLUSTER_SLOTS
;
j
++
)
{
if
(
!
countKeysInSlot
(
j
))
continue
;
/* No keys in this slot. */
if
(
!
countKeysInSlot
(
j
))
continue
;
/* No keys in this slot. */
/* Check if we are assigned to this slot or if we are importing it.
/* Check if we are assigned to this slot or if we are importing it.
* In both cases check the next slot as the configuration makes
* In both cases check the next slot as the configuration makes
...
@@ -3571,8 +3571,8 @@ void clusterSetMaster(clusterNode *n) {
...
@@ -3571,8 +3571,8 @@ void clusterSetMaster(clusterNode *n) {
serverAssert
(
myself
->
numslots
==
0
);
serverAssert
(
myself
->
numslots
==
0
);
if
(
nodeIsMaster
(
myself
))
{
if
(
nodeIsMaster
(
myself
))
{
myself
->
flags
&=
~
REDIS
_NODE_MASTER
;
myself
->
flags
&=
~
CLUSTER
_NODE_MASTER
;
myself
->
flags
|=
REDIS
_NODE_SLAVE
;
myself
->
flags
|=
CLUSTER
_NODE_SLAVE
;
clusterCloseAllSlots
();
clusterCloseAllSlots
();
}
else
{
}
else
{
if
(
myself
->
slaveof
)
if
(
myself
->
slaveof
)
...
@@ -3594,13 +3594,13 @@ struct redisNodeFlags {
...
@@ -3594,13 +3594,13 @@ struct redisNodeFlags {
};
};
static
struct
redisNodeFlags
redisNodeFlagsTable
[]
=
{
static
struct
redisNodeFlags
redisNodeFlagsTable
[]
=
{
{
REDIS
_NODE_MYSELF
,
"myself,"
},
{
CLUSTER
_NODE_MYSELF
,
"myself,"
},
{
REDIS
_NODE_MASTER
,
"master,"
},
{
CLUSTER
_NODE_MASTER
,
"master,"
},
{
REDIS
_NODE_SLAVE
,
"slave,"
},
{
CLUSTER
_NODE_SLAVE
,
"slave,"
},
{
REDIS
_NODE_PFAIL
,
"fail?,"
},
{
CLUSTER
_NODE_PFAIL
,
"fail?,"
},
{
REDIS
_NODE_FAIL
,
"fail,"
},
{
CLUSTER
_NODE_FAIL
,
"fail,"
},
{
REDIS
_NODE_HANDSHAKE
,
"handshake,"
},
{
CLUSTER
_NODE_HANDSHAKE
,
"handshake,"
},
{
REDIS
_NODE_NOADDR
,
"noaddr,"
}
{
CLUSTER
_NODE_NOADDR
,
"noaddr,"
}
};
};
/* Concatenate the comma separated list of node flags to the given SDS
/* Concatenate the comma separated list of node flags to the given SDS
...
@@ -3647,19 +3647,19 @@ sds clusterGenNodeDescription(clusterNode *node) {
...
@@ -3647,19 +3647,19 @@ sds clusterGenNodeDescription(clusterNode *node) {
(
long
long
)
node
->
ping_sent
,
(
long
long
)
node
->
ping_sent
,
(
long
long
)
node
->
pong_received
,
(
long
long
)
node
->
pong_received
,
(
unsigned
long
long
)
node
->
configEpoch
,
(
unsigned
long
long
)
node
->
configEpoch
,
(
node
->
link
||
node
->
flags
&
REDIS
_NODE_MYSELF
)
?
(
node
->
link
||
node
->
flags
&
CLUSTER
_NODE_MYSELF
)
?
"connected"
:
"disconnected"
);
"connected"
:
"disconnected"
);
/* Slots served by this instance */
/* Slots served by this instance */
start
=
-
1
;
start
=
-
1
;
for
(
j
=
0
;
j
<
REDIS_
CLUSTER_SLOTS
;
j
++
)
{
for
(
j
=
0
;
j
<
CLUSTER_SLOTS
;
j
++
)
{
int
bit
;
int
bit
;
if
((
bit
=
clusterNodeGetSlotBit
(
node
,
j
))
!=
0
)
{
if
((
bit
=
clusterNodeGetSlotBit
(
node
,
j
))
!=
0
)
{
if
(
start
==
-
1
)
start
=
j
;
if
(
start
==
-
1
)
start
=
j
;
}
}
if
(
start
!=
-
1
&&
(
!
bit
||
j
==
REDIS_
CLUSTER_SLOTS
-
1
))
{
if
(
start
!=
-
1
&&
(
!
bit
||
j
==
CLUSTER_SLOTS
-
1
))
{
if
(
bit
&&
j
==
REDIS_
CLUSTER_SLOTS
-
1
)
j
++
;
if
(
bit
&&
j
==
CLUSTER_SLOTS
-
1
)
j
++
;
if
(
start
==
j
-
1
)
{
if
(
start
==
j
-
1
)
{
ci
=
sdscatprintf
(
ci
,
" %d"
,
start
);
ci
=
sdscatprintf
(
ci
,
" %d"
,
start
);
...
@@ -3673,8 +3673,8 @@ sds clusterGenNodeDescription(clusterNode *node) {
...
@@ -3673,8 +3673,8 @@ sds clusterGenNodeDescription(clusterNode *node) {
/* Just for MYSELF node we also dump info about slots that
/* Just for MYSELF node we also dump info about slots that
* we are migrating to other instances or importing from other
* we are migrating to other instances or importing from other
* instances. */
* instances. */
if
(
node
->
flags
&
REDIS
_NODE_MYSELF
)
{
if
(
node
->
flags
&
CLUSTER
_NODE_MYSELF
)
{
for
(
j
=
0
;
j
<
REDIS_
CLUSTER_SLOTS
;
j
++
)
{
for
(
j
=
0
;
j
<
CLUSTER_SLOTS
;
j
++
)
{
if
(
server
.
cluster
->
migrating_slots_to
[
j
])
{
if
(
server
.
cluster
->
migrating_slots_to
[
j
])
{
ci
=
sdscatprintf
(
ci
,
" [%d->-%.40s]"
,
j
,
ci
=
sdscatprintf
(
ci
,
" [%d->-%.40s]"
,
j
,
server
.
cluster
->
migrating_slots_to
[
j
]
->
name
);
server
.
cluster
->
migrating_slots_to
[
j
]
->
name
);
...
@@ -3726,7 +3726,7 @@ int getSlotOrReply(client *c, robj *o) {
...
@@ -3726,7 +3726,7 @@ int getSlotOrReply(client *c, robj *o) {
long
long
slot
;
long
long
slot
;
if
(
getLongLongFromObject
(
o
,
&
slot
)
!=
C_OK
||
if
(
getLongLongFromObject
(
o
,
&
slot
)
!=
C_OK
||
slot
<
0
||
slot
>=
REDIS_
CLUSTER_SLOTS
)
slot
<
0
||
slot
>=
CLUSTER_SLOTS
)
{
{
addReplyError
(
c
,
"Invalid or out of range slot"
);
addReplyError
(
c
,
"Invalid or out of range slot"
);
return
-
1
;
return
-
1
;
...
@@ -3757,17 +3757,17 @@ void clusterReplyMultiBulkSlots(client *c) {
...
@@ -3757,17 +3757,17 @@ void clusterReplyMultiBulkSlots(client *c) {
* master) and masters not serving any slot. */
* master) and masters not serving any slot. */
if
(
!
nodeIsMaster
(
node
)
||
node
->
numslots
==
0
)
continue
;
if
(
!
nodeIsMaster
(
node
)
||
node
->
numslots
==
0
)
continue
;
for
(
j
=
0
;
j
<
REDIS_
CLUSTER_SLOTS
;
j
++
)
{
for
(
j
=
0
;
j
<
CLUSTER_SLOTS
;
j
++
)
{
int
bit
,
i
;
int
bit
,
i
;
if
((
bit
=
clusterNodeGetSlotBit
(
node
,
j
))
!=
0
)
{
if
((
bit
=
clusterNodeGetSlotBit
(
node
,
j
))
!=
0
)
{
if
(
start
==
-
1
)
start
=
j
;
if
(
start
==
-
1
)
start
=
j
;
}
}
if
(
start
!=
-
1
&&
(
!
bit
||
j
==
REDIS_
CLUSTER_SLOTS
-
1
))
{
if
(
start
!=
-
1
&&
(
!
bit
||
j
==
CLUSTER_SLOTS
-
1
))
{
int
nested_elements
=
3
;
/* slots (2) + master addr (1). */
int
nested_elements
=
3
;
/* slots (2) + master addr (1). */
void
*
nested_replylen
=
addDeferredMultiBulkLength
(
c
);
void
*
nested_replylen
=
addDeferredMultiBulkLength
(
c
);
if
(
bit
&&
j
==
REDIS_
CLUSTER_SLOTS
-
1
)
j
++
;
if
(
bit
&&
j
==
CLUSTER_SLOTS
-
1
)
j
++
;
/* If slot exists in output map, add to it's list.
/* If slot exists in output map, add to it's list.
* else, create a new output map for this slot */
* else, create a new output map for this slot */
...
@@ -3837,7 +3837,7 @@ void clusterCommand(client *c) {
...
@@ -3837,7 +3837,7 @@ void clusterCommand(client *c) {
decrRefCount
(
o
);
decrRefCount
(
o
);
}
else
if
(
!
strcasecmp
(
c
->
argv
[
1
]
->
ptr
,
"myid"
)
&&
c
->
argc
==
2
)
{
}
else
if
(
!
strcasecmp
(
c
->
argv
[
1
]
->
ptr
,
"myid"
)
&&
c
->
argc
==
2
)
{
/* CLUSTER MYID */
/* CLUSTER MYID */
addReplyBulkCBuffer
(
c
,
myself
->
name
,
REDIS_
CLUSTER_NAMELEN
);
addReplyBulkCBuffer
(
c
,
myself
->
name
,
CLUSTER_NAMELEN
);
}
else
if
(
!
strcasecmp
(
c
->
argv
[
1
]
->
ptr
,
"slots"
)
&&
c
->
argc
==
2
)
{
}
else
if
(
!
strcasecmp
(
c
->
argv
[
1
]
->
ptr
,
"slots"
)
&&
c
->
argc
==
2
)
{
/* CLUSTER SLOTS */
/* CLUSTER SLOTS */
clusterReplyMultiBulkSlots
(
c
);
clusterReplyMultiBulkSlots
(
c
);
...
@@ -3856,10 +3856,10 @@ void clusterCommand(client *c) {
...
@@ -3856,10 +3856,10 @@ void clusterCommand(client *c) {
/* CLUSTER ADDSLOTS <slot> [slot] ... */
/* CLUSTER ADDSLOTS <slot> [slot] ... */
/* CLUSTER DELSLOTS <slot> [slot] ... */
/* CLUSTER DELSLOTS <slot> [slot] ... */
int
j
,
slot
;
int
j
,
slot
;
unsigned
char
*
slots
=
zmalloc
(
REDIS_
CLUSTER_SLOTS
);
unsigned
char
*
slots
=
zmalloc
(
CLUSTER_SLOTS
);
int
del
=
!
strcasecmp
(
c
->
argv
[
1
]
->
ptr
,
"delslots"
);
int
del
=
!
strcasecmp
(
c
->
argv
[
1
]
->
ptr
,
"delslots"
);
memset
(
slots
,
0
,
REDIS_
CLUSTER_SLOTS
);
memset
(
slots
,
0
,
CLUSTER_SLOTS
);
/* Check that all the arguments are parseable and that all the
/* Check that all the arguments are parseable and that all the
* slots are not already busy. */
* slots are not already busy. */
for
(
j
=
2
;
j
<
c
->
argc
;
j
++
)
{
for
(
j
=
2
;
j
<
c
->
argc
;
j
++
)
{
...
@@ -3883,7 +3883,7 @@ void clusterCommand(client *c) {
...
@@ -3883,7 +3883,7 @@ void clusterCommand(client *c) {
return
;
return
;
}
}
}
}
for
(
j
=
0
;
j
<
REDIS_
CLUSTER_SLOTS
;
j
++
)
{
for
(
j
=
0
;
j
<
CLUSTER_SLOTS
;
j
++
)
{
if
(
slots
[
j
])
{
if
(
slots
[
j
])
{
int
retval
;
int
retval
;
...
@@ -3999,7 +3999,7 @@ void clusterCommand(client *c) {
...
@@ -3999,7 +3999,7 @@ void clusterCommand(client *c) {
uint64_t
myepoch
;
uint64_t
myepoch
;
int
j
;
int
j
;
for
(
j
=
0
;
j
<
REDIS_
CLUSTER_SLOTS
;
j
++
)
{
for
(
j
=
0
;
j
<
CLUSTER_SLOTS
;
j
++
)
{
clusterNode
*
n
=
server
.
cluster
->
slots
[
j
];
clusterNode
*
n
=
server
.
cluster
->
slots
[
j
];
if
(
n
==
NULL
)
continue
;
if
(
n
==
NULL
)
continue
;
...
@@ -4063,7 +4063,7 @@ void clusterCommand(client *c) {
...
@@ -4063,7 +4063,7 @@ void clusterCommand(client *c) {
if
(
getLongLongFromObjectOrReply
(
c
,
c
->
argv
[
2
],
&
slot
,
NULL
)
!=
C_OK
)
if
(
getLongLongFromObjectOrReply
(
c
,
c
->
argv
[
2
],
&
slot
,
NULL
)
!=
C_OK
)
return
;
return
;
if
(
slot
<
0
||
slot
>=
REDIS_
CLUSTER_SLOTS
)
{
if
(
slot
<
0
||
slot
>=
CLUSTER_SLOTS
)
{
addReplyError
(
c
,
"Invalid slot"
);
addReplyError
(
c
,
"Invalid slot"
);
return
;
return
;
}
}
...
@@ -4079,7 +4079,7 @@ void clusterCommand(client *c) {
...
@@ -4079,7 +4079,7 @@ void clusterCommand(client *c) {
if
(
getLongLongFromObjectOrReply
(
c
,
c
->
argv
[
3
],
&
maxkeys
,
NULL
)
if
(
getLongLongFromObjectOrReply
(
c
,
c
->
argv
[
3
],
&
maxkeys
,
NULL
)
!=
C_OK
)
!=
C_OK
)
return
;
return
;
if
(
slot
<
0
||
slot
>=
REDIS_
CLUSTER_SLOTS
||
maxkeys
<
0
)
{
if
(
slot
<
0
||
slot
>=
CLUSTER_SLOTS
||
maxkeys
<
0
)
{
addReplyError
(
c
,
"Invalid slot or number of keys"
);
addReplyError
(
c
,
"Invalid slot or number of keys"
);
return
;
return
;
}
}
...
@@ -4213,7 +4213,7 @@ void clusterCommand(client *c) {
...
@@ -4213,7 +4213,7 @@ void clusterCommand(client *c) {
return
;
return
;
}
}
resetManualFailover
();
resetManualFailover
();
server
.
cluster
->
mf_end
=
mstime
()
+
REDIS_
CLUSTER_MF_TIMEOUT
;
server
.
cluster
->
mf_end
=
mstime
()
+
CLUSTER_MF_TIMEOUT
;
if
(
takeover
)
{
if
(
takeover
)
{
/* A takeover does not perform any initial check. It just
/* A takeover does not perform any initial check. It just
...
@@ -4760,22 +4760,22 @@ void readwriteCommand(client *c) {
...
@@ -4760,22 +4760,22 @@ void readwriteCommand(client *c) {
* On success the function returns the node that is able to serve the request.
* On success the function returns the node that is able to serve the request.
* If the node is not 'myself' a redirection must be perfomed. The kind of
* If the node is not 'myself' a redirection must be perfomed. The kind of
* redirection is specified setting the integer passed by reference
* redirection is specified setting the integer passed by reference
* 'error_code', which will be set to
REDIS_
CLUSTER_REDIR_ASK or
* 'error_code', which will be set to CLUSTER_REDIR_ASK or
*
REDIS_
CLUSTER_REDIR_MOVED.
* CLUSTER_REDIR_MOVED.
*
*
* When the node is 'myself' 'error_code' is set to
REDIS_
CLUSTER_REDIR_NONE.
* When the node is 'myself' 'error_code' is set to CLUSTER_REDIR_NONE.
*
*
* If the command fails NULL is returned, and the reason of the failure is
* If the command fails NULL is returned, and the reason of the failure is
* provided via 'error_code', which will be set to:
* provided via 'error_code', which will be set to:
*
*
*
REDIS_
CLUSTER_REDIR_CROSS_SLOT if the request contains multiple keys that
* CLUSTER_REDIR_CROSS_SLOT if the request contains multiple keys that
* don't belong to the same hash slot.
* don't belong to the same hash slot.
*
*
*
REDIS_
CLUSTER_REDIR_UNSTABLE if the request contains mutliple keys
* CLUSTER_REDIR_UNSTABLE if the request contains mutliple keys
* belonging to the same slot, but the slot is not stable (in migration or
* belonging to the same slot, but the slot is not stable (in migration or
* importing state, likely because a resharding is in progress).
* importing state, likely because a resharding is in progress).
*
*
*
REDIS_
CLUSTER_REDIR_DOWN_UNBOUND if the request addresses a slot which is
* CLUSTER_REDIR_DOWN_UNBOUND if the request addresses a slot which is
* not bound to any node. In this case the cluster global state should be
* not bound to any node. In this case the cluster global state should be
* already "down" but it is fragile to rely on the update of the global state,
* already "down" but it is fragile to rely on the update of the global state,
* so we also handle it here. */
* so we also handle it here. */
...
@@ -4788,7 +4788,7 @@ clusterNode *getNodeByQuery(client *c, struct redisCommand *cmd, robj **argv, in
...
@@ -4788,7 +4788,7 @@ clusterNode *getNodeByQuery(client *c, struct redisCommand *cmd, robj **argv, in
int
i
,
slot
=
0
,
migrating_slot
=
0
,
importing_slot
=
0
,
missing_keys
=
0
;
int
i
,
slot
=
0
,
migrating_slot
=
0
,
importing_slot
=
0
,
missing_keys
=
0
;
/* Set error code optimistically for the base case. */
/* Set error code optimistically for the base case. */
if
(
error_code
)
*
error_code
=
REDIS_
CLUSTER_REDIR_NONE
;
if
(
error_code
)
*
error_code
=
CLUSTER_REDIR_NONE
;
/* We handle all the cases as if they were EXEC commands, so we have
/* We handle all the cases as if they were EXEC commands, so we have
* a common code path for everything */
* a common code path for everything */
...
@@ -4840,7 +4840,7 @@ clusterNode *getNodeByQuery(client *c, struct redisCommand *cmd, robj **argv, in
...
@@ -4840,7 +4840,7 @@ clusterNode *getNodeByQuery(client *c, struct redisCommand *cmd, robj **argv, in
if
(
n
==
NULL
)
{
if
(
n
==
NULL
)
{
getKeysFreeResult
(
keyindex
);
getKeysFreeResult
(
keyindex
);
if
(
error_code
)
if
(
error_code
)
*
error_code
=
REDIS_
CLUSTER_REDIR_DOWN_UNBOUND
;
*
error_code
=
CLUSTER_REDIR_DOWN_UNBOUND
;
return
NULL
;
return
NULL
;
}
}
...
@@ -4864,7 +4864,7 @@ clusterNode *getNodeByQuery(client *c, struct redisCommand *cmd, robj **argv, in
...
@@ -4864,7 +4864,7 @@ clusterNode *getNodeByQuery(client *c, struct redisCommand *cmd, robj **argv, in
/* Error: multiple keys from different slots. */
/* Error: multiple keys from different slots. */
getKeysFreeResult
(
keyindex
);
getKeysFreeResult
(
keyindex
);
if
(
error_code
)
if
(
error_code
)
*
error_code
=
REDIS_
CLUSTER_REDIR_CROSS_SLOT
;
*
error_code
=
CLUSTER_REDIR_CROSS_SLOT
;
return
NULL
;
return
NULL
;
}
else
{
}
else
{
/* Flag this request as one with multiple different
/* Flag this request as one with multiple different
...
@@ -4897,7 +4897,7 @@ clusterNode *getNodeByQuery(client *c, struct redisCommand *cmd, robj **argv, in
...
@@ -4897,7 +4897,7 @@ clusterNode *getNodeByQuery(client *c, struct redisCommand *cmd, robj **argv, in
/* If we don't have all the keys and we are migrating the slot, send
/* If we don't have all the keys and we are migrating the slot, send
* an ASK redirection. */
* an ASK redirection. */
if
(
migrating_slot
&&
missing_keys
)
{
if
(
migrating_slot
&&
missing_keys
)
{
if
(
error_code
)
*
error_code
=
REDIS_
CLUSTER_REDIR_ASK
;
if
(
error_code
)
*
error_code
=
CLUSTER_REDIR_ASK
;
return
server
.
cluster
->
migrating_slots_to
[
slot
];
return
server
.
cluster
->
migrating_slots_to
[
slot
];
}
}
...
@@ -4909,7 +4909,7 @@ clusterNode *getNodeByQuery(client *c, struct redisCommand *cmd, robj **argv, in
...
@@ -4909,7 +4909,7 @@ clusterNode *getNodeByQuery(client *c, struct redisCommand *cmd, robj **argv, in
(
c
->
flags
&
CLIENT_ASKING
||
cmd
->
flags
&
CMD_ASKING
))
(
c
->
flags
&
CLIENT_ASKING
||
cmd
->
flags
&
CMD_ASKING
))
{
{
if
(
multiple_keys
&&
missing_keys
)
{
if
(
multiple_keys
&&
missing_keys
)
{
if
(
error_code
)
*
error_code
=
REDIS_
CLUSTER_REDIR_UNSTABLE
;
if
(
error_code
)
*
error_code
=
CLUSTER_REDIR_UNSTABLE
;
return
NULL
;
return
NULL
;
}
else
{
}
else
{
return
myself
;
return
myself
;
...
@@ -4929,35 +4929,35 @@ clusterNode *getNodeByQuery(client *c, struct redisCommand *cmd, robj **argv, in
...
@@ -4929,35 +4929,35 @@ clusterNode *getNodeByQuery(client *c, struct redisCommand *cmd, robj **argv, in
/* Base case: just return the right node. However if this node is not
/* Base case: just return the right node. However if this node is not
* myself, set error_code to MOVED since we need to issue a rediretion. */
* myself, set error_code to MOVED since we need to issue a rediretion. */
if
(
n
!=
myself
&&
error_code
)
*
error_code
=
REDIS_
CLUSTER_REDIR_MOVED
;
if
(
n
!=
myself
&&
error_code
)
*
error_code
=
CLUSTER_REDIR_MOVED
;
return
n
;
return
n
;
}
}
/* Send the client the right redirection code, according to error_code
/* Send the client the right redirection code, according to error_code
* that should be set to one of
REDIS_
CLUSTER_REDIR_* macros.
* that should be set to one of CLUSTER_REDIR_* macros.
*
*
* If
REDIS_
CLUSTER_REDIR_ASK or
REDIS_
CLUSTER_REDIR_MOVED error codes
* If CLUSTER_REDIR_ASK or CLUSTER_REDIR_MOVED error codes
* are used, then the node 'n' should not be NULL, but should be the
* are used, then the node 'n' should not be NULL, but should be the
* node we want to mention in the redirection. Moreover hashslot should
* node we want to mention in the redirection. Moreover hashslot should
* be set to the hash slot that caused the redirection. */
* be set to the hash slot that caused the redirection. */
void
clusterRedirectClient
(
client
*
c
,
clusterNode
*
n
,
int
hashslot
,
int
error_code
)
{
void
clusterRedirectClient
(
client
*
c
,
clusterNode
*
n
,
int
hashslot
,
int
error_code
)
{
if
(
error_code
==
REDIS_
CLUSTER_REDIR_CROSS_SLOT
)
{
if
(
error_code
==
CLUSTER_REDIR_CROSS_SLOT
)
{
addReplySds
(
c
,
sdsnew
(
"-CROSSSLOT Keys in request don't hash to the same slot
\r\n
"
));
addReplySds
(
c
,
sdsnew
(
"-CROSSSLOT Keys in request don't hash to the same slot
\r\n
"
));
}
else
if
(
error_code
==
REDIS_
CLUSTER_REDIR_UNSTABLE
)
{
}
else
if
(
error_code
==
CLUSTER_REDIR_UNSTABLE
)
{
/* The request spawns mutliple keys in the same slot,
/* The request spawns mutliple keys in the same slot,
* but the slot is not "stable" currently as there is
* but the slot is not "stable" currently as there is
* a migration or import in progress. */
* a migration or import in progress. */
addReplySds
(
c
,
sdsnew
(
"-TRYAGAIN Multiple keys request during rehashing of slot
\r\n
"
));
addReplySds
(
c
,
sdsnew
(
"-TRYAGAIN Multiple keys request during rehashing of slot
\r\n
"
));
}
else
if
(
error_code
==
REDIS_
CLUSTER_REDIR_DOWN_STATE
)
{
}
else
if
(
error_code
==
CLUSTER_REDIR_DOWN_STATE
)
{
addReplySds
(
c
,
sdsnew
(
"-CLUSTERDOWN The cluster is down
\r\n
"
));
addReplySds
(
c
,
sdsnew
(
"-CLUSTERDOWN The cluster is down
\r\n
"
));
}
else
if
(
error_code
==
REDIS_
CLUSTER_REDIR_DOWN_UNBOUND
)
{
}
else
if
(
error_code
==
CLUSTER_REDIR_DOWN_UNBOUND
)
{
addReplySds
(
c
,
sdsnew
(
"-CLUSTERDOWN Hash slot not served
\r\n
"
));
addReplySds
(
c
,
sdsnew
(
"-CLUSTERDOWN Hash slot not served
\r\n
"
));
}
else
if
(
error_code
==
REDIS_
CLUSTER_REDIR_MOVED
||
}
else
if
(
error_code
==
CLUSTER_REDIR_MOVED
||
error_code
==
REDIS_
CLUSTER_REDIR_ASK
)
error_code
==
CLUSTER_REDIR_ASK
)
{
{
addReplySds
(
c
,
sdscatprintf
(
sdsempty
(),
addReplySds
(
c
,
sdscatprintf
(
sdsempty
(),
"-%s %d %s:%d
\r\n
"
,
"-%s %d %s:%d
\r\n
"
,
(
error_code
==
REDIS_
CLUSTER_REDIR_ASK
)
?
"ASK"
:
"MOVED"
,
(
error_code
==
CLUSTER_REDIR_ASK
)
?
"ASK"
:
"MOVED"
,
hashslot
,
n
->
ip
,
n
->
port
));
hashslot
,
n
->
ip
,
n
->
port
));
}
else
{
}
else
{
serverPanic
(
"getNodeByQuery() unknown error."
);
serverPanic
(
"getNodeByQuery() unknown error."
);
...
@@ -4981,8 +4981,8 @@ int clusterRedirectBlockedClientIfNeeded(client *c) {
...
@@ -4981,8 +4981,8 @@ int clusterRedirectBlockedClientIfNeeded(client *c) {
dictIterator
*
di
;
dictIterator
*
di
;
/* If the cluster is down, unblock the client with the right error. */
/* If the cluster is down, unblock the client with the right error. */
if
(
server
.
cluster
->
state
==
REDIS_
CLUSTER_FAIL
)
{
if
(
server
.
cluster
->
state
==
CLUSTER_FAIL
)
{
clusterRedirectClient
(
c
,
NULL
,
0
,
REDIS_
CLUSTER_REDIR_DOWN_STATE
);
clusterRedirectClient
(
c
,
NULL
,
0
,
CLUSTER_REDIR_DOWN_STATE
);
return
1
;
return
1
;
}
}
...
@@ -5000,10 +5000,10 @@ int clusterRedirectBlockedClientIfNeeded(client *c) {
...
@@ -5000,10 +5000,10 @@ int clusterRedirectBlockedClientIfNeeded(client *c) {
{
{
if
(
node
==
NULL
)
{
if
(
node
==
NULL
)
{
clusterRedirectClient
(
c
,
NULL
,
0
,
clusterRedirectClient
(
c
,
NULL
,
0
,
REDIS_
CLUSTER_REDIR_DOWN_UNBOUND
);
CLUSTER_REDIR_DOWN_UNBOUND
);
}
else
{
}
else
{
clusterRedirectClient
(
c
,
node
,
slot
,
clusterRedirectClient
(
c
,
node
,
slot
,
REDIS_
CLUSTER_REDIR_MOVED
);
CLUSTER_REDIR_MOVED
);
}
}
return
1
;
return
1
;
}
}
...
...
src/cluster.h
View file @
3325a9b1
#ifndef __
REDIS_
CLUSTER_H
#ifndef __CLUSTER_H
#define __
REDIS_
CLUSTER_H
#define __CLUSTER_H
/*-----------------------------------------------------------------------------
/*-----------------------------------------------------------------------------
* Redis cluster data structures, defines, exported API.
* Redis cluster data structures, defines, exported API.
*----------------------------------------------------------------------------*/
*----------------------------------------------------------------------------*/
#define
REDIS_
CLUSTER_SLOTS 16384
#define CLUSTER_SLOTS 16384
#define
REDIS_
CLUSTER_OK 0
/* Everything looks ok */
#define CLUSTER_OK 0
/* Everything looks ok */
#define
REDIS_
CLUSTER_FAIL 1
/* The cluster can't work */
#define CLUSTER_FAIL 1
/* The cluster can't work */
#define
REDIS_
CLUSTER_NAMELEN 40
/* sha1 hex length */
#define CLUSTER_NAMELEN 40
/* sha1 hex length */
#define
REDIS_
CLUSTER_PORT_INCR 10000
/* Cluster port = baseport + PORT_INCR */
#define CLUSTER_PORT_INCR 10000
/* Cluster port = baseport + PORT_INCR */
/* The following defines are amount of time, sometimes expressed as
/* The following defines are amount of time, sometimes expressed as
* multiplicators of the node timeout value (when ending with MULT). */
* multiplicators of the node timeout value (when ending with MULT). */
#define
REDIS_
CLUSTER_DEFAULT_NODE_TIMEOUT 15000
#define CLUSTER_DEFAULT_NODE_TIMEOUT 15000
#define
REDIS_
CLUSTER_DEFAULT_SLAVE_VALIDITY 10
/* Slave max data age factor. */
#define CLUSTER_DEFAULT_SLAVE_VALIDITY 10
/* Slave max data age factor. */
#define
REDIS_
CLUSTER_DEFAULT_REQUIRE_FULL_COVERAGE 1
#define CLUSTER_DEFAULT_REQUIRE_FULL_COVERAGE 1
#define
REDIS_
CLUSTER_FAIL_REPORT_VALIDITY_MULT 2
/* Fail report validity. */
#define CLUSTER_FAIL_REPORT_VALIDITY_MULT 2
/* Fail report validity. */
#define
REDIS_
CLUSTER_FAIL_UNDO_TIME_MULT 2
/* Undo fail if master is back. */
#define CLUSTER_FAIL_UNDO_TIME_MULT 2
/* Undo fail if master is back. */
#define
REDIS_
CLUSTER_FAIL_UNDO_TIME_ADD 10
/* Some additional time. */
#define CLUSTER_FAIL_UNDO_TIME_ADD 10
/* Some additional time. */
#define
REDIS_
CLUSTER_FAILOVER_DELAY 5
/* Seconds */
#define CLUSTER_FAILOVER_DELAY 5
/* Seconds */
#define
REDIS_
CLUSTER_DEFAULT_MIGRATION_BARRIER 1
#define CLUSTER_DEFAULT_MIGRATION_BARRIER 1
#define
REDIS_
CLUSTER_MF_TIMEOUT 5000
/* Milliseconds to do a manual failover. */
#define CLUSTER_MF_TIMEOUT 5000
/* Milliseconds to do a manual failover. */
#define
REDIS_
CLUSTER_MF_PAUSE_MULT 2
/* Master pause manual failover mult. */
#define CLUSTER_MF_PAUSE_MULT 2
/* Master pause manual failover mult. */
/* Redirection errors returned by getNodeByQuery(). */
/* Redirection errors returned by getNodeByQuery(). */
#define
REDIS_
CLUSTER_REDIR_NONE 0
/* Node can serve the request. */
#define CLUSTER_REDIR_NONE 0
/* Node can serve the request. */
#define
REDIS_
CLUSTER_REDIR_CROSS_SLOT 1
/* -CROSSSLOT request. */
#define CLUSTER_REDIR_CROSS_SLOT 1
/* -CROSSSLOT request. */
#define
REDIS_
CLUSTER_REDIR_UNSTABLE 2
/* -TRYAGAIN redirection required */
#define CLUSTER_REDIR_UNSTABLE 2
/* -TRYAGAIN redirection required */
#define
REDIS_
CLUSTER_REDIR_ASK 3
/* -ASK redirection required. */
#define CLUSTER_REDIR_ASK 3
/* -ASK redirection required. */
#define
REDIS_
CLUSTER_REDIR_MOVED 4
/* -MOVED redirection required. */
#define CLUSTER_REDIR_MOVED 4
/* -MOVED redirection required. */
#define
REDIS_
CLUSTER_REDIR_DOWN_STATE 5
/* -CLUSTERDOWN, global state. */
#define CLUSTER_REDIR_DOWN_STATE 5
/* -CLUSTERDOWN, global state. */
#define
REDIS_
CLUSTER_REDIR_DOWN_UNBOUND 6
/* -CLUSTERDOWN, unbound slot. */
#define CLUSTER_REDIR_DOWN_UNBOUND 6
/* -CLUSTERDOWN, unbound slot. */
struct
clusterNode
;
struct
clusterNode
;
...
@@ -45,32 +45,32 @@ typedef struct clusterLink {
...
@@ -45,32 +45,32 @@ typedef struct clusterLink {
}
clusterLink
;
}
clusterLink
;
/* Cluster node flags and macros. */
/* Cluster node flags and macros. */
#define
REDIS
_NODE_MASTER 1
/* The node is a master */
#define
CLUSTER
_NODE_MASTER 1
/* The node is a master */
#define
REDIS
_NODE_SLAVE 2
/* The node is a slave */
#define
CLUSTER
_NODE_SLAVE 2
/* The node is a slave */
#define
REDIS
_NODE_PFAIL 4
/* Failure? Need acknowledge */
#define
CLUSTER
_NODE_PFAIL 4
/* Failure? Need acknowledge */
#define
REDIS
_NODE_FAIL 8
/* The node is believed to be malfunctioning */
#define
CLUSTER
_NODE_FAIL 8
/* The node is believed to be malfunctioning */
#define
REDIS
_NODE_MYSELF 16
/* This node is myself */
#define
CLUSTER
_NODE_MYSELF 16
/* This node is myself */
#define
REDIS
_NODE_HANDSHAKE 32
/* We have still to exchange the first ping */
#define
CLUSTER
_NODE_HANDSHAKE 32
/* We have still to exchange the first ping */
#define
REDIS
_NODE_NOADDR 64
/* We don't know the address of this node */
#define
CLUSTER
_NODE_NOADDR 64
/* We don't know the address of this node */
#define
REDIS
_NODE_MEET 128
/* Send a MEET message to this node */
#define
CLUSTER
_NODE_MEET 128
/* Send a MEET message to this node */
#define
REDIS
_NODE_PROMOTED 256
/* Master was a slave promoted by failover */
#define
CLUSTER
_NODE_PROMOTED 256
/* Master was a slave promoted by failover */
#define
REDIS
_NODE_NULL_NAME "\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000"
#define
CLUSTER
_NODE_NULL_NAME "\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000"
#define nodeIsMaster(n) ((n)->flags &
REDIS
_NODE_MASTER)
#define nodeIsMaster(n) ((n)->flags &
CLUSTER
_NODE_MASTER)
#define nodeIsSlave(n) ((n)->flags &
REDIS
_NODE_SLAVE)
#define nodeIsSlave(n) ((n)->flags &
CLUSTER
_NODE_SLAVE)
#define nodeInHandshake(n) ((n)->flags &
REDIS
_NODE_HANDSHAKE)
#define nodeInHandshake(n) ((n)->flags &
CLUSTER
_NODE_HANDSHAKE)
#define nodeHasAddr(n) (!((n)->flags &
REDIS
_NODE_NOADDR))
#define nodeHasAddr(n) (!((n)->flags &
CLUSTER
_NODE_NOADDR))
#define nodeWithoutAddr(n) ((n)->flags &
REDIS
_NODE_NOADDR)
#define nodeWithoutAddr(n) ((n)->flags &
CLUSTER
_NODE_NOADDR)
#define nodeTimedOut(n) ((n)->flags &
REDIS
_NODE_PFAIL)
#define nodeTimedOut(n) ((n)->flags &
CLUSTER
_NODE_PFAIL)
#define nodeFailed(n) ((n)->flags &
REDIS
_NODE_FAIL)
#define nodeFailed(n) ((n)->flags &
CLUSTER
_NODE_FAIL)
/* Reasons why a slave is not able to failover. */
/* Reasons why a slave is not able to failover. */
#define
REDIS_
CLUSTER_CANT_FAILOVER_NONE 0
#define CLUSTER_CANT_FAILOVER_NONE 0
#define
REDIS_
CLUSTER_CANT_FAILOVER_DATA_AGE 1
#define CLUSTER_CANT_FAILOVER_DATA_AGE 1
#define
REDIS_
CLUSTER_CANT_FAILOVER_WAITING_DELAY 2
#define CLUSTER_CANT_FAILOVER_WAITING_DELAY 2
#define
REDIS_
CLUSTER_CANT_FAILOVER_EXPIRED 3
#define CLUSTER_CANT_FAILOVER_EXPIRED 3
#define
REDIS_
CLUSTER_CANT_FAILOVER_WAITING_VOTES 4
#define CLUSTER_CANT_FAILOVER_WAITING_VOTES 4
#define
REDIS_
CLUSTER_CANT_FAILOVER_RELOG_PERIOD (60*5)
/* seconds. */
#define CLUSTER_CANT_FAILOVER_RELOG_PERIOD (60*5)
/* seconds. */
/* This structure represent elements of node->fail_reports. */
/* This structure represent elements of node->fail_reports. */
typedef
struct
clusterNodeFailReport
{
typedef
struct
clusterNodeFailReport
{
...
@@ -80,10 +80,10 @@ typedef struct clusterNodeFailReport {
...
@@ -80,10 +80,10 @@ typedef struct clusterNodeFailReport {
typedef
struct
clusterNode
{
typedef
struct
clusterNode
{
mstime_t
ctime
;
/* Node object creation time. */
mstime_t
ctime
;
/* Node object creation time. */
char
name
[
REDIS_
CLUSTER_NAMELEN
];
/* Node name, hex string, sha1-size */
char
name
[
CLUSTER_NAMELEN
];
/* Node name, hex string, sha1-size */
int
flags
;
/*
REDIS
_NODE_... */
int
flags
;
/*
CLUSTER
_NODE_... */
uint64_t
configEpoch
;
/* Last configEpoch observed for this node */
uint64_t
configEpoch
;
/* Last configEpoch observed for this node */
unsigned
char
slots
[
REDIS_
CLUSTER_SLOTS
/
8
];
/* slots handled by this node */
unsigned
char
slots
[
CLUSTER_SLOTS
/
8
];
/* slots handled by this node */
int
numslots
;
/* Number of slots handled by this node */
int
numslots
;
/* Number of slots handled by this node */
int
numslaves
;
/* Number of slave nodes, if this is a master */
int
numslaves
;
/* Number of slave nodes, if this is a master */
struct
clusterNode
**
slaves
;
/* pointers to slave nodes */
struct
clusterNode
**
slaves
;
/* pointers to slave nodes */
...
@@ -103,13 +103,13 @@ typedef struct clusterNode {
...
@@ -103,13 +103,13 @@ typedef struct clusterNode {
typedef
struct
clusterState
{
typedef
struct
clusterState
{
clusterNode
*
myself
;
/* This node */
clusterNode
*
myself
;
/* This node */
uint64_t
currentEpoch
;
uint64_t
currentEpoch
;
int
state
;
/*
REDIS_
CLUSTER_OK,
REDIS_
CLUSTER_FAIL, ... */
int
state
;
/* CLUSTER_OK, CLUSTER_FAIL, ... */
int
size
;
/* Num of master nodes with at least one slot */
int
size
;
/* Num of master nodes with at least one slot */
dict
*
nodes
;
/* Hash table of name -> clusterNode structures */
dict
*
nodes
;
/* Hash table of name -> clusterNode structures */
dict
*
nodes_black_list
;
/* Nodes we don't re-add for a few seconds. */
dict
*
nodes_black_list
;
/* Nodes we don't re-add for a few seconds. */
clusterNode
*
migrating_slots_to
[
REDIS_
CLUSTER_SLOTS
];
clusterNode
*
migrating_slots_to
[
CLUSTER_SLOTS
];
clusterNode
*
importing_slots_from
[
REDIS_
CLUSTER_SLOTS
];
clusterNode
*
importing_slots_from
[
CLUSTER_SLOTS
];
clusterNode
*
slots
[
REDIS_
CLUSTER_SLOTS
];
clusterNode
*
slots
[
CLUSTER_SLOTS
];
zskiplist
*
slots_to_keys
;
zskiplist
*
slots_to_keys
;
/* The following fields are used to take the slave state on elections. */
/* The following fields are used to take the slave state on elections. */
mstime_t
failover_auth_time
;
/* Time of previous or next election. */
mstime_t
failover_auth_time
;
/* Time of previous or next election. */
...
@@ -162,7 +162,7 @@ typedef struct clusterState {
...
@@ -162,7 +162,7 @@ typedef struct clusterState {
* to the first node, using the getsockname() function. Then we'll use this
* to the first node, using the getsockname() function. Then we'll use this
* address for all the next messages. */
* address for all the next messages. */
typedef
struct
{
typedef
struct
{
char
nodename
[
REDIS_
CLUSTER_NAMELEN
];
char
nodename
[
CLUSTER_NAMELEN
];
uint32_t
ping_sent
;
uint32_t
ping_sent
;
uint32_t
pong_received
;
uint32_t
pong_received
;
char
ip
[
NET_IP_STR_LEN
];
/* IP address last time it was seen */
char
ip
[
NET_IP_STR_LEN
];
/* IP address last time it was seen */
...
@@ -173,7 +173,7 @@ typedef struct {
...
@@ -173,7 +173,7 @@ typedef struct {
}
clusterMsgDataGossip
;
}
clusterMsgDataGossip
;
typedef
struct
{
typedef
struct
{
char
nodename
[
REDIS_
CLUSTER_NAMELEN
];
char
nodename
[
CLUSTER_NAMELEN
];
}
clusterMsgDataFail
;
}
clusterMsgDataFail
;
typedef
struct
{
typedef
struct
{
...
@@ -187,8 +187,8 @@ typedef struct {
...
@@ -187,8 +187,8 @@ typedef struct {
typedef
struct
{
typedef
struct
{
uint64_t
configEpoch
;
/* Config epoch of the specified instance. */
uint64_t
configEpoch
;
/* Config epoch of the specified instance. */
char
nodename
[
REDIS_
CLUSTER_NAMELEN
];
/* Name of the slots owner. */
char
nodename
[
CLUSTER_NAMELEN
];
/* Name of the slots owner. */
unsigned
char
slots
[
REDIS_
CLUSTER_SLOTS
/
8
];
/* Slots bitmap. */
unsigned
char
slots
[
CLUSTER_SLOTS
/
8
];
/* Slots bitmap. */
}
clusterMsgDataUpdate
;
}
clusterMsgDataUpdate
;
union
clusterMsgData
{
union
clusterMsgData
{
...
@@ -229,9 +229,9 @@ typedef struct {
...
@@ -229,9 +229,9 @@ typedef struct {
slave. */
slave. */
uint64_t
offset
;
/* Master replication offset if node is a master or
uint64_t
offset
;
/* Master replication offset if node is a master or
processed replication offset if node is a slave. */
processed replication offset if node is a slave. */
char
sender
[
REDIS_
CLUSTER_NAMELEN
];
/* Name of the sender node */
char
sender
[
CLUSTER_NAMELEN
];
/* Name of the sender node */
unsigned
char
myslots
[
REDIS_
CLUSTER_SLOTS
/
8
];
unsigned
char
myslots
[
CLUSTER_SLOTS
/
8
];
char
slaveof
[
REDIS_
CLUSTER_NAMELEN
];
char
slaveof
[
CLUSTER_NAMELEN
];
char
notused1
[
32
];
/* 32 bytes reserved for future usage. */
char
notused1
[
32
];
/* 32 bytes reserved for future usage. */
uint16_t
port
;
/* Sender TCP base port */
uint16_t
port
;
/* Sender TCP base port */
uint16_t
flags
;
/* Sender node flags */
uint16_t
flags
;
/* Sender node flags */
...
@@ -253,4 +253,4 @@ clusterNode *getNodeByQuery(client *c, struct redisCommand *cmd, robj **argv, in
...
@@ -253,4 +253,4 @@ clusterNode *getNodeByQuery(client *c, struct redisCommand *cmd, robj **argv, in
int
clusterRedirectBlockedClientIfNeeded
(
client
*
c
);
int
clusterRedirectBlockedClientIfNeeded
(
client
*
c
);
void
clusterRedirectClient
(
client
*
c
,
clusterNode
*
n
,
int
hashslot
,
int
error_code
);
void
clusterRedirectClient
(
client
*
c
,
clusterNode
*
n
,
int
hashslot
,
int
error_code
);
#endif
/* __
REDIS_
CLUSTER_H */
#endif
/* __CLUSTER_H */
src/config.c
View file @
3325a9b1
...
@@ -1798,10 +1798,10 @@ int rewriteConfig(char *path) {
...
@@ -1798,10 +1798,10 @@ int rewriteConfig(char *path) {
rewriteConfigNumericalOption
(
state
,
"lua-time-limit"
,
server
.
lua_time_limit
,
LUA_SCRIPT_TIME_LIMIT
);
rewriteConfigNumericalOption
(
state
,
"lua-time-limit"
,
server
.
lua_time_limit
,
LUA_SCRIPT_TIME_LIMIT
);
rewriteConfigYesNoOption
(
state
,
"cluster-enabled"
,
server
.
cluster_enabled
,
0
);
rewriteConfigYesNoOption
(
state
,
"cluster-enabled"
,
server
.
cluster_enabled
,
0
);
rewriteConfigStringOption
(
state
,
"cluster-config-file"
,
server
.
cluster_configfile
,
CONFIG_DEFAULT_CLUSTER_CONFIG_FILE
);
rewriteConfigStringOption
(
state
,
"cluster-config-file"
,
server
.
cluster_configfile
,
CONFIG_DEFAULT_CLUSTER_CONFIG_FILE
);
rewriteConfigYesNoOption
(
state
,
"cluster-require-full-coverage"
,
server
.
cluster_require_full_coverage
,
REDIS_
CLUSTER_DEFAULT_REQUIRE_FULL_COVERAGE
);
rewriteConfigYesNoOption
(
state
,
"cluster-require-full-coverage"
,
server
.
cluster_require_full_coverage
,
CLUSTER_DEFAULT_REQUIRE_FULL_COVERAGE
);
rewriteConfigNumericalOption
(
state
,
"cluster-node-timeout"
,
server
.
cluster_node_timeout
,
REDIS_
CLUSTER_DEFAULT_NODE_TIMEOUT
);
rewriteConfigNumericalOption
(
state
,
"cluster-node-timeout"
,
server
.
cluster_node_timeout
,
CLUSTER_DEFAULT_NODE_TIMEOUT
);
rewriteConfigNumericalOption
(
state
,
"cluster-migration-barrier"
,
server
.
cluster_migration_barrier
,
REDIS_
CLUSTER_DEFAULT_MIGRATION_BARRIER
);
rewriteConfigNumericalOption
(
state
,
"cluster-migration-barrier"
,
server
.
cluster_migration_barrier
,
CLUSTER_DEFAULT_MIGRATION_BARRIER
);
rewriteConfigNumericalOption
(
state
,
"cluster-slave-validity-factor"
,
server
.
cluster_slave_validity_factor
,
REDIS_
CLUSTER_DEFAULT_SLAVE_VALIDITY
);
rewriteConfigNumericalOption
(
state
,
"cluster-slave-validity-factor"
,
server
.
cluster_slave_validity_factor
,
CLUSTER_DEFAULT_SLAVE_VALIDITY
);
rewriteConfigNumericalOption
(
state
,
"slowlog-log-slower-than"
,
server
.
slowlog_log_slower_than
,
CONFIG_DEFAULT_SLOWLOG_LOG_SLOWER_THAN
);
rewriteConfigNumericalOption
(
state
,
"slowlog-log-slower-than"
,
server
.
slowlog_log_slower_than
,
CONFIG_DEFAULT_SLOWLOG_LOG_SLOWER_THAN
);
rewriteConfigNumericalOption
(
state
,
"latency-monitor-threshold"
,
server
.
latency_monitor_threshold
,
CONFIG_DEFAULT_LATENCY_MONITOR_THRESHOLD
);
rewriteConfigNumericalOption
(
state
,
"latency-monitor-threshold"
,
server
.
latency_monitor_threshold
,
CONFIG_DEFAULT_LATENCY_MONITOR_THRESHOLD
);
rewriteConfigNumericalOption
(
state
,
"slowlog-max-len"
,
server
.
slowlog_max_len
,
CONFIG_DEFAULT_SLOWLOG_MAX_LEN
);
rewriteConfigNumericalOption
(
state
,
"slowlog-max-len"
,
server
.
slowlog_max_len
,
CONFIG_DEFAULT_SLOWLOG_MAX_LEN
);
...
...
src/server.c
View file @
3325a9b1
...
@@ -1486,10 +1486,10 @@ void initServerConfig(void) {
...
@@ -1486,10 +1486,10 @@ void initServerConfig(void) {
server
.
repl_min_slaves_to_write
=
CONFIG_DEFAULT_MIN_SLAVES_TO_WRITE
;
server
.
repl_min_slaves_to_write
=
CONFIG_DEFAULT_MIN_SLAVES_TO_WRITE
;
server
.
repl_min_slaves_max_lag
=
CONFIG_DEFAULT_MIN_SLAVES_MAX_LAG
;
server
.
repl_min_slaves_max_lag
=
CONFIG_DEFAULT_MIN_SLAVES_MAX_LAG
;
server
.
cluster_enabled
=
0
;
server
.
cluster_enabled
=
0
;
server
.
cluster_node_timeout
=
REDIS_
CLUSTER_DEFAULT_NODE_TIMEOUT
;
server
.
cluster_node_timeout
=
CLUSTER_DEFAULT_NODE_TIMEOUT
;
server
.
cluster_migration_barrier
=
REDIS_
CLUSTER_DEFAULT_MIGRATION_BARRIER
;
server
.
cluster_migration_barrier
=
CLUSTER_DEFAULT_MIGRATION_BARRIER
;
server
.
cluster_slave_validity_factor
=
REDIS_
CLUSTER_DEFAULT_SLAVE_VALIDITY
;
server
.
cluster_slave_validity_factor
=
CLUSTER_DEFAULT_SLAVE_VALIDITY
;
server
.
cluster_require_full_coverage
=
REDIS_
CLUSTER_DEFAULT_REQUIRE_FULL_COVERAGE
;
server
.
cluster_require_full_coverage
=
CLUSTER_DEFAULT_REQUIRE_FULL_COVERAGE
;
server
.
cluster_configfile
=
zstrdup
(
CONFIG_DEFAULT_CLUSTER_CONFIG_FILE
);
server
.
cluster_configfile
=
zstrdup
(
CONFIG_DEFAULT_CLUSTER_CONFIG_FILE
);
server
.
lua_caller
=
NULL
;
server
.
lua_caller
=
NULL
;
server
.
lua_time_limit
=
LUA_SCRIPT_TIME_LIMIT
;
server
.
lua_time_limit
=
LUA_SCRIPT_TIME_LIMIT
;
...
@@ -2226,9 +2226,9 @@ int processCommand(client *c) {
...
@@ -2226,9 +2226,9 @@ int processCommand(client *c) {
{
{
int
hashslot
;
int
hashslot
;
if
(
server
.
cluster
->
state
!=
REDIS_
CLUSTER_OK
)
{
if
(
server
.
cluster
->
state
!=
CLUSTER_OK
)
{
flagTransaction
(
c
);
flagTransaction
(
c
);
clusterRedirectClient
(
c
,
NULL
,
0
,
REDIS_
CLUSTER_REDIR_DOWN_STATE
);
clusterRedirectClient
(
c
,
NULL
,
0
,
CLUSTER_REDIR_DOWN_STATE
);
return
C_OK
;
return
C_OK
;
}
else
{
}
else
{
int
error_code
;
int
error_code
;
...
@@ -2855,7 +2855,7 @@ sds genRedisInfoString(char *section) {
...
@@ -2855,7 +2855,7 @@ sds genRedisInfoString(char *section) {
server
.
aof_rewrite_scheduled
,
server
.
aof_rewrite_scheduled
,
sdslen
(
server
.
aof_buf
),
sdslen
(
server
.
aof_buf
),
aofRewriteBufferSize
(),
aofRewriteBufferSize
(),
bioPendingJobsOfType
(
REDIS_
BIO_AOF_FSYNC
),
bioPendingJobsOfType
(
BIO_AOF_FSYNC
),
server
.
aof_delayed_fsync
);
server
.
aof_delayed_fsync
);
}
}
...
...
src/syncio.c
View file @
3325a9b1
...
@@ -40,7 +40,7 @@
...
@@ -40,7 +40,7 @@
*
*
* All the functions take the timeout in milliseconds. */
* All the functions take the timeout in milliseconds. */
#define
REDIS_
SYNCIO_RESOLUTION 10
/* Resolution in milliseconds */
#define SYNCIO_
_
RESOLUTION 10
/* Resolution in milliseconds */
/* Write the specified payload to 'fd'. If writing the whole payload will be
/* Write the specified payload to 'fd'. If writing the whole payload will be
* done within 'timeout' milliseconds the operation succeeds and 'size' is
* done within 'timeout' milliseconds the operation succeeds and 'size' is
...
@@ -52,8 +52,8 @@ ssize_t syncWrite(int fd, char *ptr, ssize_t size, long long timeout) {
...
@@ -52,8 +52,8 @@ ssize_t syncWrite(int fd, char *ptr, ssize_t size, long long timeout) {
long
long
remaining
=
timeout
;
long
long
remaining
=
timeout
;
while
(
1
)
{
while
(
1
)
{
long
long
wait
=
(
remaining
>
REDIS_
SYNCIO_RESOLUTION
)
?
long
long
wait
=
(
remaining
>
SYNCIO_
_
RESOLUTION
)
?
remaining
:
REDIS_
SYNCIO_RESOLUTION
;
remaining
:
SYNCIO_
_
RESOLUTION
;
long
long
elapsed
;
long
long
elapsed
;
/* Optimistically try to write before checking if the file descriptor
/* Optimistically try to write before checking if the file descriptor
...
@@ -89,8 +89,8 @@ ssize_t syncRead(int fd, char *ptr, ssize_t size, long long timeout) {
...
@@ -89,8 +89,8 @@ ssize_t syncRead(int fd, char *ptr, ssize_t size, long long timeout) {
if
(
size
==
0
)
return
0
;
if
(
size
==
0
)
return
0
;
while
(
1
)
{
while
(
1
)
{
long
long
wait
=
(
remaining
>
REDIS_
SYNCIO_RESOLUTION
)
?
long
long
wait
=
(
remaining
>
SYNCIO_
_
RESOLUTION
)
?
remaining
:
REDIS_
SYNCIO_RESOLUTION
;
remaining
:
SYNCIO_
_
RESOLUTION
;
long
long
elapsed
;
long
long
elapsed
;
/* Optimistically try to read before checking if the file descriptor
/* Optimistically try to read before checking if the file descriptor
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
.
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment