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
62a4a8c1
"docs/vscode:/vscode.git/clone" did not exist on "577e2ea8b23f5ab18040c371dc49494ccb725237"
Commit
62a4a8c1
authored
Jul 07, 2018
by
minkikim89
Committed by
Salvatore Sanfilippo
Jul 09, 2018
Browse files
fix whitespace in redis-cli.c
parent
0a5805d7
Changes
1
Hide whitespace changes
Inline
Side-by-side
src/redis-cli.c
View file @
62a4a8c1
...
...
@@ -41,14 +41,14 @@
#include <errno.h>
#include <sys/stat.h>
#include <sys/time.h>
#include <assert.h>
#include <assert.h>
#include <fcntl.h>
#include <limits.h>
#include <math.h>
#include <hiredis.h>
#include <sds.h>
/* use sds.h from hiredis, so that only one set of sds functions will be present in the binary */
#include "dict.h"
#include "dict.h"
#include "adlist.h"
#include "zmalloc.h"
#include "linenoise.h"
...
...
@@ -125,7 +125,7 @@
#define CLUSTER_MANAGER_OPT_QUIET 1 << 6
#define CLUSTER_MANAGER_OPT_VERBOSE 1 << 7
#define CLUSTER_MANAGER_LOG_LVL_INFO 1
#define CLUSTER_MANAGER_LOG_LVL_INFO 1
#define CLUSTER_MANAGER_LOG_LVL_WARN 2
#define CLUSTER_MANAGER_LOG_LVL_ERR 3
#define CLUSTER_MANAGER_LOG_LVL_SUCCESS 4
...
...
@@ -150,7 +150,7 @@ int spectrum_palette_size;
/* Dict Helpers */
static
uint64_t
dictSdsHash
(
const
void
*
key
);
static
int
dictSdsKeyCompare
(
void
*
privdata
,
const
void
*
key1
,
static
int
dictSdsKeyCompare
(
void
*
privdata
,
const
void
*
key1
,
const
void
*
key2
);
static
void
dictSdsDestructor
(
void
*
privdata
,
void
*
val
);
static
void
dictListDestructor
(
void
*
privdata
,
void
*
val
);
...
...
@@ -1310,13 +1310,13 @@ static int parseOptions(int argc, char **argv) {
}
else
if
(
!
strcmp
(
argv
[
i
],
"--cluster"
)
&&
lastarg
)
{
usage
();
}
else
if
(
!
strcmp
(
argv
[
i
],
"--cluster-replicas"
)
&&
!
lastarg
)
{
config
.
cluster_manager_command
.
replicas
=
atoi
(
argv
[
++
i
]);
config
.
cluster_manager_command
.
replicas
=
atoi
(
argv
[
++
i
]);
}
else
if
(
!
strcmp
(
argv
[
i
],
"--cluster-master-id"
)
&&
!
lastarg
)
{
config
.
cluster_manager_command
.
master_id
=
argv
[
++
i
];
config
.
cluster_manager_command
.
master_id
=
argv
[
++
i
];
}
else
if
(
!
strcmp
(
argv
[
i
],
"--cluster-from"
)
&&
!
lastarg
)
{
config
.
cluster_manager_command
.
from
=
argv
[
++
i
];
config
.
cluster_manager_command
.
from
=
argv
[
++
i
];
}
else
if
(
!
strcmp
(
argv
[
i
],
"--cluster-to"
)
&&
!
lastarg
)
{
config
.
cluster_manager_command
.
to
=
argv
[
++
i
];
config
.
cluster_manager_command
.
to
=
argv
[
++
i
];
}
else
if
(
!
strcmp
(
argv
[
i
],
"--cluster-weight"
)
&&
!
lastarg
)
{
int
widx
=
i
+
1
;
char
**
weight
=
argv
+
widx
;
...
...
@@ -1326,36 +1326,36 @@ static int parseOptions(int argc, char **argv) {
wargc
++
;
}
if
(
wargc
>
0
)
{
config
.
cluster_manager_command
.
weight
=
weight
;
config
.
cluster_manager_command
.
weight
=
weight
;
config
.
cluster_manager_command
.
weight_argc
=
wargc
;
i
+=
wargc
;
}
}
else
if
(
!
strcmp
(
argv
[
i
],
"--cluster-slots"
)
&&
!
lastarg
)
{
config
.
cluster_manager_command
.
slots
=
atoi
(
argv
[
++
i
]);
config
.
cluster_manager_command
.
slots
=
atoi
(
argv
[
++
i
]);
}
else
if
(
!
strcmp
(
argv
[
i
],
"--cluster-timeout"
)
&&
!
lastarg
)
{
config
.
cluster_manager_command
.
timeout
=
atoi
(
argv
[
++
i
]);
config
.
cluster_manager_command
.
timeout
=
atoi
(
argv
[
++
i
]);
}
else
if
(
!
strcmp
(
argv
[
i
],
"--cluster-pipeline"
)
&&
!
lastarg
)
{
config
.
cluster_manager_command
.
pipeline
=
atoi
(
argv
[
++
i
]);
config
.
cluster_manager_command
.
pipeline
=
atoi
(
argv
[
++
i
]);
}
else
if
(
!
strcmp
(
argv
[
i
],
"--cluster-threshold"
)
&&
!
lastarg
)
{
config
.
cluster_manager_command
.
threshold
=
atof
(
argv
[
++
i
]);
config
.
cluster_manager_command
.
threshold
=
atof
(
argv
[
++
i
]);
}
else
if
(
!
strcmp
(
argv
[
i
],
"--cluster-yes"
))
{
config
.
cluster_manager_command
.
flags
|=
CLUSTER_MANAGER_CMD_FLAG_YES
;
config
.
cluster_manager_command
.
flags
|=
CLUSTER_MANAGER_CMD_FLAG_YES
;
}
else
if
(
!
strcmp
(
argv
[
i
],
"--cluster-simulate"
))
{
config
.
cluster_manager_command
.
flags
|=
CLUSTER_MANAGER_CMD_FLAG_SIMULATE
;
config
.
cluster_manager_command
.
flags
|=
CLUSTER_MANAGER_CMD_FLAG_SIMULATE
;
}
else
if
(
!
strcmp
(
argv
[
i
],
"--cluster-replace"
))
{
config
.
cluster_manager_command
.
flags
|=
CLUSTER_MANAGER_CMD_FLAG_REPLACE
;
config
.
cluster_manager_command
.
flags
|=
CLUSTER_MANAGER_CMD_FLAG_REPLACE
;
}
else
if
(
!
strcmp
(
argv
[
i
],
"--cluster-copy"
))
{
config
.
cluster_manager_command
.
flags
|=
CLUSTER_MANAGER_CMD_FLAG_COPY
;
config
.
cluster_manager_command
.
flags
|=
CLUSTER_MANAGER_CMD_FLAG_COPY
;
}
else
if
(
!
strcmp
(
argv
[
i
],
"--cluster-slave"
))
{
config
.
cluster_manager_command
.
flags
|=
CLUSTER_MANAGER_CMD_FLAG_SLAVE
;
config
.
cluster_manager_command
.
flags
|=
CLUSTER_MANAGER_CMD_FLAG_SLAVE
;
}
else
if
(
!
strcmp
(
argv
[
i
],
"--cluster-use-empty-masters"
))
{
config
.
cluster_manager_command
.
flags
|=
CLUSTER_MANAGER_CMD_FLAG_EMPTYMASTER
;
config
.
cluster_manager_command
.
flags
|=
CLUSTER_MANAGER_CMD_FLAG_EMPTYMASTER
;
}
else
if
(
!
strcmp
(
argv
[
i
],
"-v"
)
||
!
strcmp
(
argv
[
i
],
"--version"
))
{
sds
version
=
cliVersion
();
printf
(
"redis-cli %s
\n
"
,
version
);
...
...
@@ -1905,7 +1905,7 @@ static clusterManagerNode *clusterManagerNodeByAbbreviatedName(const char *n);
static
void
clusterManagerNodeResetSlots
(
clusterManagerNode
*
node
);
static
int
clusterManagerNodeIsCluster
(
clusterManagerNode
*
node
,
char
**
err
);
static
void
clusterManagerPrintNotClusterNodeError
(
clusterManagerNode
*
node
,
char
*
err
);
char
*
err
);
static
int
clusterManagerNodeLoadInfo
(
clusterManagerNode
*
node
,
int
opts
,
char
**
err
);
static
int
clusterManagerLoadInfoFromNode
(
clusterManagerNode
*
node
,
int
opts
);
...
...
@@ -1923,12 +1923,12 @@ static int clusterManagerCheckCluster(int quiet);
static
void
clusterManagerLog
(
int
level
,
const
char
*
fmt
,
...);
static
int
clusterManagerIsConfigConsistent
(
void
);
static
void
clusterManagerOnError
(
sds
err
);
static
void
clusterManagerNodeArrayInit
(
clusterManagerNodeArray
*
array
,
static
void
clusterManagerNodeArrayInit
(
clusterManagerNodeArray
*
array
,
int
len
);
static
void
clusterManagerNodeArrayReset
(
clusterManagerNodeArray
*
array
);
static
void
clusterManagerNodeArrayShift
(
clusterManagerNodeArray
*
array
,
static
void
clusterManagerNodeArrayShift
(
clusterManagerNodeArray
*
array
,
clusterManagerNode
**
nodeptr
);
static
void
clusterManagerNodeArrayAdd
(
clusterManagerNodeArray
*
array
,
static
void
clusterManagerNodeArrayAdd
(
clusterManagerNodeArray
*
array
,
clusterManagerNode
*
node
);
/* Cluster Manager commands. */
...
...
@@ -1955,24 +1955,24 @@ typedef struct clusterManagerCommandDef {
}
clusterManagerCommandDef
;
clusterManagerCommandDef
clusterManagerCommands
[]
=
{
{
"create"
,
clusterManagerCommandCreate
,
-
2
,
"host1:port1 ... hostN:portN"
,
{
"create"
,
clusterManagerCommandCreate
,
-
2
,
"host1:port1 ... hostN:portN"
,
"replicas <arg>"
},
{
"check"
,
clusterManagerCommandCheck
,
-
1
,
"host:port"
,
NULL
},
{
"info"
,
clusterManagerCommandInfo
,
-
1
,
"host:port"
,
NULL
},
{
"fix"
,
clusterManagerCommandFix
,
-
1
,
"host:port"
,
NULL
},
{
"reshard"
,
clusterManagerCommandReshard
,
-
1
,
"host:port"
,
{
"reshard"
,
clusterManagerCommandReshard
,
-
1
,
"host:port"
,
"from <arg>,to <arg>,slots <arg>,yes,timeout <arg>,pipeline <arg>"
},
{
"rebalance"
,
clusterManagerCommandRebalance
,
-
1
,
"host:port"
,
{
"rebalance"
,
clusterManagerCommandRebalance
,
-
1
,
"host:port"
,
"weight <node1=w1...nodeN=wN>,use-empty-masters,"
"timeout <arg>,simulate,pipeline <arg>,threshold <arg>"
},
{
"add-node"
,
clusterManagerCommandAddNode
,
2
,
"timeout <arg>,simulate,pipeline <arg>,threshold <arg>"
},
{
"add-node"
,
clusterManagerCommandAddNode
,
2
,
"new_host:new_port existing_host:existing_port"
,
"slave,master-id <arg>"
},
{
"del-node"
,
clusterManagerCommandDeleteNode
,
2
,
"host:port node_id"
,
NULL
},
{
"call"
,
clusterManagerCommandCall
,
-
2
,
{
"call"
,
clusterManagerCommandCall
,
-
2
,
"host:port command arg arg .. arg"
,
NULL
},
{
"set-timeout"
,
clusterManagerCommandSetTimeout
,
2
,
{
"set-timeout"
,
clusterManagerCommandSetTimeout
,
2
,
"host:port milliseconds"
,
NULL
},
{
"import"
,
clusterManagerCommandImport
,
1
,
"host:port"
,
{
"import"
,
clusterManagerCommandImport
,
1
,
"host:port"
,
"from <arg>,copy,replace"
},
{
"help"
,
clusterManagerCommandHelp
,
0
,
NULL
,
NULL
}
};
...
...
@@ -1988,19 +1988,19 @@ static void createClusterManagerCommand(char *cmdname, int argc, char **argv) {
static
clusterManagerCommandProc
*
validateClusterManagerCommand
(
void
)
{
int
i
,
commands_count
=
sizeof
(
clusterManagerCommands
)
/
int
i
,
commands_count
=
sizeof
(
clusterManagerCommands
)
/
sizeof
(
clusterManagerCommandDef
);
clusterManagerCommandProc
*
proc
=
NULL
;
char
*
cmdname
=
config
.
cluster_manager_command
.
name
;
int
argc
=
config
.
cluster_manager_command
.
argc
;
for
(
i
=
0
;
i
<
commands_count
;
i
++
)
{
clusterManagerCommandDef
cmddef
=
clusterManagerCommands
[
i
];
clusterManagerCommandDef
cmddef
=
clusterManagerCommands
[
i
];
if
(
!
strcmp
(
cmddef
.
name
,
cmdname
))
{
if
((
cmddef
.
arity
>
0
&&
argc
!=
cmddef
.
arity
)
||
if
((
cmddef
.
arity
>
0
&&
argc
!=
cmddef
.
arity
)
||
(
cmddef
.
arity
<
0
&&
argc
<
(
cmddef
.
arity
*
-
1
)))
{
fprintf
(
stderr
,
"[ERR] Wrong number of arguments for "
"specified --cluster sub command
\n
"
);
return
NULL
;
return
NULL
;
}
proc
=
cmddef
.
proc
;
}
...
...
@@ -2009,13 +2009,13 @@ static clusterManagerCommandProc *validateClusterManagerCommand(void) {
return
proc
;
}
/* Get host ip and port from command arguments. If only one argument has
* been provided it must be in the form of 'ip:port', elsewhere
/* Get host ip and port from command arguments. If only one argument has
* been provided it must be in the form of 'ip:port', elsewhere
* the first argument must be the ip and the second one the port.
* If host and port can be detected, it returns 1 and it stores host and
* port into variables referenced by'ip_ptr' and 'port_ptr' pointers,
* If host and port can be detected, it returns 1 and it stores host and
* port into variables referenced by'ip_ptr' and 'port_ptr' pointers,
* elsewhere it returns 0. */
static
int
getClusterHostFromCmdArgs
(
int
argc
,
char
**
argv
,
static
int
getClusterHostFromCmdArgs
(
int
argc
,
char
**
argv
,
char
**
ip_ptr
,
int
*
port_ptr
)
{
int
port
=
0
;
char
*
ip
=
NULL
;
...
...
@@ -2053,13 +2053,13 @@ static void freeClusterManagerNodeFlags(list *flags) {
}
static
void
freeClusterManagerNode
(
clusterManagerNode
*
node
)
{
if
(
node
->
context
!=
NULL
)
redisFree
(
node
->
context
);
if
(
node
->
context
!=
NULL
)
redisFree
(
node
->
context
);
if
(
node
->
friends
!=
NULL
)
{
listIter
li
;
listIter
li
;
listNode
*
ln
;
listRewind
(
node
->
friends
,
&
li
);
while
((
ln
=
listNext
(
&
li
))
!=
NULL
)
{
clusterManagerNode
*
fn
=
ln
->
value
;
clusterManagerNode
*
fn
=
ln
->
value
;
freeClusterManagerNode
(
fn
);
}
listRelease
(
node
->
friends
);
...
...
@@ -2086,12 +2086,12 @@ static void freeClusterManagerNode(clusterManagerNode *node) {
}
static
void
freeClusterManager
(
void
)
{
listIter
li
;
listIter
li
;
listNode
*
ln
;
if
(
cluster_manager
.
nodes
!=
NULL
)
{
listRewind
(
cluster_manager
.
nodes
,
&
li
);
while
((
ln
=
listNext
(
&
li
))
!=
NULL
)
{
clusterManagerNode
*
n
=
ln
->
value
;
clusterManagerNode
*
n
=
ln
->
value
;
freeClusterManagerNode
(
n
);
}
listRelease
(
cluster_manager
.
nodes
);
...
...
@@ -2100,13 +2100,13 @@ static void freeClusterManager(void) {
if
(
cluster_manager
.
errors
!=
NULL
)
{
listRewind
(
cluster_manager
.
errors
,
&
li
);
while
((
ln
=
listNext
(
&
li
))
!=
NULL
)
{
sds
err
=
ln
->
value
;
sds
err
=
ln
->
value
;
sdsfree
(
err
);
}
listRelease
(
cluster_manager
.
errors
);
cluster_manager
.
errors
=
NULL
;
}
if
(
clusterManagerUncoveredSlots
!=
NULL
)
if
(
clusterManagerUncoveredSlots
!=
NULL
)
dictRelease
(
clusterManagerUncoveredSlots
);
}
...
...
@@ -2135,27 +2135,27 @@ static clusterManagerNode *clusterManagerNewNode(char *ip, int port) {
return
node
;
}
/* Check whether reply is NULL or its type is REDIS_REPLY_ERROR. In the
* latest case, if the 'err' arg is not NULL, it gets allocated with a copy
* of reply error (it's up to the caller function to free it), elsewhere
/* Check whether reply is NULL or its type is REDIS_REPLY_ERROR. In the
* latest case, if the 'err' arg is not NULL, it gets allocated with a copy
* of reply error (it's up to the caller function to free it), elsewhere
* the error is directly printed. */
static
int
clusterManagerCheckRedisReply
(
clusterManagerNode
*
n
,
redisReply
*
r
,
char
**
err
)
static
int
clusterManagerCheckRedisReply
(
clusterManagerNode
*
n
,
redisReply
*
r
,
char
**
err
)
{
int
is_err
=
0
;
if
(
!
r
||
(
is_err
=
(
r
->
type
==
REDIS_REPLY_ERROR
)))
{
if
(
is_err
)
{
if
(
err
!=
NULL
)
{
*
err
=
zmalloc
((
r
->
len
+
1
)
*
sizeof
(
char
));
strcpy
(
*
err
,
r
->
str
);
}
else
CLUSTER_MANAGER_PRINT_REPLY_ERROR
(
n
,
r
->
str
);
*
err
=
zmalloc
((
r
->
len
+
1
)
*
sizeof
(
char
));
strcpy
(
*
err
,
r
->
str
);
}
else
CLUSTER_MANAGER_PRINT_REPLY_ERROR
(
n
,
r
->
str
);
}
return
0
;
}
return
1
;
}
return
1
;
}
static
void
clusterManagerRemoveNodeFromList
(
list
*
nodelist
,
static
void
clusterManagerRemoveNodeFromList
(
list
*
nodelist
,
clusterManagerNode
*
node
)
{
listIter
li
;
listNode
*
ln
;
...
...
@@ -2179,7 +2179,7 @@ static clusterManagerNode *clusterManagerNodeByName(const char *name) {
listNode
*
ln
;
listRewind
(
cluster_manager
.
nodes
,
&
li
);
while
((
ln
=
listNext
(
&
li
))
!=
NULL
)
{
clusterManagerNode
*
n
=
ln
->
value
;
clusterManagerNode
*
n
=
ln
->
value
;
if
(
n
->
name
&&
!
sdscmp
(
n
->
name
,
lcname
))
{
found
=
n
;
break
;
...
...
@@ -2204,8 +2204,8 @@ static clusterManagerNode *clusterManagerNodeByAbbreviatedName(const char*name)
listNode
*
ln
;
listRewind
(
cluster_manager
.
nodes
,
&
li
);
while
((
ln
=
listNext
(
&
li
))
!=
NULL
)
{
clusterManagerNode
*
n
=
ln
->
value
;
if
(
n
->
name
&&
clusterManagerNode
*
n
=
ln
->
value
;
if
(
n
->
name
&&
strstr
(
n
->
name
,
lcname
)
==
n
->
name
)
{
found
=
n
;
break
;
...
...
@@ -2221,17 +2221,17 @@ static void clusterManagerNodeResetSlots(clusterManagerNode *node) {
}
/* Call "INFO" redis command on the specified node and return the reply. */
static
redisReply
*
clusterManagerGetNodeRedisInfo
(
clusterManagerNode
*
node
,
char
**
err
)
static
redisReply
*
clusterManagerGetNodeRedisInfo
(
clusterManagerNode
*
node
,
char
**
err
)
{
redisReply
*
info
=
CLUSTER_MANAGER_COMMAND
(
node
,
"INFO"
);
if
(
err
!=
NULL
)
*
err
=
NULL
;
if
(
info
==
NULL
)
return
NULL
;
if
(
info
->
type
==
REDIS_REPLY_ERROR
)
{
if
(
err
!=
NULL
)
{
*
err
=
zmalloc
((
info
->
len
+
1
)
*
sizeof
(
char
));
strcpy
(
*
err
,
info
->
str
);
}
*
err
=
zmalloc
((
info
->
len
+
1
)
*
sizeof
(
char
));
strcpy
(
*
err
,
info
->
str
);
}
freeReplyObject
(
info
);
return
NULL
;
}
...
...
@@ -2246,7 +2246,7 @@ static int clusterManagerNodeIsCluster(clusterManagerNode *node, char **err) {
return
is_cluster
;
}
/* Checks whether the node is empty. Node is considered not-empty if it has
/* Checks whether the node is empty. Node is considered not-empty if it has
* some key or if it already knows other nodes */
static
int
clusterManagerNodeIsEmpty
(
clusterManagerNode
*
node
,
char
**
err
)
{
redisReply
*
info
=
clusterManagerGetNodeRedisInfo
(
node
,
err
);
...
...
@@ -2292,12 +2292,12 @@ result:
* to this violation, so that we'll optimize for the second factor only
* if it does not impact the first one.
*
* The ipnodes argument is an array of clusterManagerNodeArray, one for
* The ipnodes argument is an array of clusterManagerNodeArray, one for
* each IP, while ip_count is the total number of IPs in the configuration.
*
* The function returns the above score, and the list of
* offending slaves can be stored into the 'offending' argument,
* so that the optimizer can try changing the configuration of the
* offending slaves can be stored into the 'offending' argument,
* so that the optimizer can try changing the configuration of the
* slaves violating the anti-affinity goals. */
static
int
clusterManagerGetAntiAffinityScore
(
clusterManagerNodeArray
*
ipnodes
,
int
ip_count
,
clusterManagerNode
***
offending
,
int
*
offending_len
)
...
...
@@ -2309,7 +2309,7 @@ static int clusterManagerGetAntiAffinityScore(clusterManagerNodeArray *ipnodes,
*
offending
=
zcalloc
(
node_len
*
sizeof
(
clusterManagerNode
*
));
offending_p
=
*
offending
;
}
/* For each set of nodes in the same host, split by
/* For each set of nodes in the same host, split by
* related nodes (masters and slaves which are involved in
* replication of each other) */
for
(
i
=
0
;
i
<
ip_count
;
i
++
)
{
...
...
@@ -2330,13 +2330,13 @@ static int clusterManagerGetAntiAffinityScore(clusterManagerNodeArray *ipnodes,
otypes
=
sdsempty
();
dictAdd
(
related
,
key
,
otypes
);
}
// Master type 'm' is always set as the first character of the
// Master type 'm' is always set as the first character of the
// types string.
if
(
!
node
->
replicate
)
types
=
sdscatprintf
(
otypes
,
"m%s"
,
otypes
);
else
types
=
sdscat
(
otypes
,
"s"
);
else
types
=
sdscat
(
otypes
,
"s"
);
if
(
types
!=
otypes
)
dictReplace
(
related
,
key
,
types
);
}
/* Now it's trivial to check, for each related group having the
/* Now it's trivial to check, for each related group having the
* same host, what is their local score. */
dictIterator
*
iter
=
dictGetIterator
(
related
);
dictEntry
*
entry
;
...
...
@@ -2356,7 +2356,7 @@ static int clusterManagerGetAntiAffinityScore(clusterManagerNodeArray *ipnodes,
clusterManagerNode
*
n
=
ln
->
value
;
if
(
n
->
replicate
==
NULL
)
continue
;
if
(
!
strcmp
(
n
->
replicate
,
name
)
&&
!
strcmp
(
n
->
ip
,
ip
))
{
*
(
offending_p
++
)
=
n
;
*
(
offending_p
++
)
=
n
;
if
(
offending_len
!=
NULL
)
(
*
offending_len
)
++
;
break
;
}
...
...
@@ -2373,7 +2373,7 @@ static void clusterManagerOptimizeAntiAffinity(clusterManagerNodeArray *ipnodes,
int
ip_count
)
{
clusterManagerNode
**
offenders
=
NULL
;
int
score
=
clusterManagerGetAntiAffinityScore
(
ipnodes
,
ip_count
,
int
score
=
clusterManagerGetAntiAffinityScore
(
ipnodes
,
ip_count
,
NULL
,
NULL
);
if
(
score
==
0
)
goto
cleanup
;
clusterManagerLogInfo
(
">>> Trying to optimize slaves allocation "
...
...
@@ -2387,18 +2387,18 @@ static void clusterManagerOptimizeAntiAffinity(clusterManagerNodeArray *ipnodes,
zfree
(
offenders
);
offenders
=
NULL
;
}
score
=
clusterManagerGetAntiAffinityScore
(
ipnodes
,
ip_count
,
&
offenders
,
score
=
clusterManagerGetAntiAffinityScore
(
ipnodes
,
ip_count
,
&
offenders
,
&
offending_len
);
if
(
score
==
0
)
break
;
// Optimal anti affinity reached
/* We'll try to randomly swap a slave's assigned master causing
* an affinity problem with another random slave, to see if we
* can improve the affinity. */
int
rand_idx
=
rand
()
%
offending_len
;
clusterManagerNode
*
first
=
offenders
[
rand_idx
],
clusterManagerNode
*
first
=
offenders
[
rand_idx
],
*
second
=
NULL
;
clusterManagerNode
**
other_replicas
=
zcalloc
((
node_len
-
1
)
*
clusterManagerNode
**
other_replicas
=
zcalloc
((
node_len
-
1
)
*
sizeof
(
*
other_replicas
));
int
other_replicas_count
=
0
;
listIter
li
;
...
...
@@ -2415,12 +2415,12 @@ static void clusterManagerOptimizeAntiAffinity(clusterManagerNodeArray *ipnodes,
}
rand_idx
=
rand
()
%
other_replicas_count
;
second
=
other_replicas
[
rand_idx
];
char
*
first_master
=
first
->
replicate
,
char
*
first_master
=
first
->
replicate
,
*
second_master
=
second
->
replicate
;
first
->
replicate
=
second_master
,
first
->
dirty
=
1
;
second
->
replicate
=
first_master
,
second
->
dirty
=
1
;
int
new_score
=
clusterManagerGetAntiAffinityScore
(
ipnodes
,
ip_count
,
int
new_score
=
clusterManagerGetAntiAffinityScore
(
ipnodes
,
ip_count
,
NULL
,
NULL
);
/* If the change actually makes thing worse, revert. Otherwise
* leave as it is because the best solution may need a few
...
...
@@ -2435,10 +2435,10 @@ static void clusterManagerOptimizeAntiAffinity(clusterManagerNodeArray *ipnodes,
score
=
clusterManagerGetAntiAffinityScore
(
ipnodes
,
ip_count
,
NULL
,
NULL
);
char
*
msg
;
int
perfect
=
(
score
==
0
);
int
log_level
=
(
perfect
?
CLUSTER_MANAGER_LOG_LVL_SUCCESS
:
int
log_level
=
(
perfect
?
CLUSTER_MANAGER_LOG_LVL_SUCCESS
:
CLUSTER_MANAGER_LOG_LVL_WARN
);
if
(
perfect
)
msg
=
"[OK] Perfect anti-affinity obtained!"
;
else
if
(
score
>=
10000
)
if
(
perfect
)
msg
=
"[OK] Perfect anti-affinity obtained!"
;
else
if
(
score
>=
10000
)
msg
=
(
"[WARNING] Some slaves are in the same host as their master"
);
else
msg
=
(
"[WARNING] Some slaves of the same master are in the same host"
);
...
...
@@ -2480,10 +2480,10 @@ static sds clusterManagerNodeSlotsString(clusterManagerNode *node) {
last_slot_idx
=
i
;
}
else
{
if
(
last_slot_idx
>=
0
)
{
if
(
first_range_idx
==
last_slot_idx
)
if
(
first_range_idx
==
last_slot_idx
)
slots
=
sdscat
(
slots
,
"]"
);
else
slots
=
sdscatfmt
(
slots
,
"-%u]"
,
last_slot_idx
);
}
}
last_slot_idx
=
-
1
;
first_range_idx
=
-
1
;
}
...
...
@@ -2543,16 +2543,16 @@ static sds clusterManagerNodeInfo(clusterManagerNode *node, int indent) {
sds
flags
=
clusterManagerNodeFlagString
(
node
);
info
=
sdscatfmt
(
info
,
"%s: %S %s:%u
\n
"
"%s slots:%S (%u slots) "
"%S"
,
role
,
node
->
name
,
node
->
ip
,
node
->
port
,
spaces
,
"%S"
,
role
,
node
->
name
,
node
->
ip
,
node
->
port
,
spaces
,
slots
,
node
->
slots_count
,
flags
);
sdsfree
(
slots
);
sdsfree
(
flags
);
}
if
(
node
->
replicate
!=
NULL
)
if
(
node
->
replicate
!=
NULL
)
info
=
sdscatfmt
(
info
,
"
\n
%s replicates %S"
,
spaces
,
node
->
replicate
);
else
if
(
node
->
replicas_count
)
info
=
sdscatfmt
(
info
,
"
\n
%s %U additional replica(s)"
,
info
=
sdscatfmt
(
info
,
"
\n
%s %U additional replica(s)"
,
spaces
,
node
->
replicas_count
);
sdsfree
(
spaces
);
return
info
;
...
...
@@ -2607,8 +2607,8 @@ static void clusterManagerShowClusterInfo(void) {
return
;
};
if
(
reply
!=
NULL
)
freeReplyObject
(
reply
);
printf
(
"%s:%d (%s...) -> %d keys | %d slots | %d slaves.
\n
"
,
node
->
ip
,
node
->
port
,
name
,
dbsize
,
printf
(
"%s:%d (%s...) -> %d keys | %d slots | %d slaves.
\n
"
,
node
->
ip
,
node
->
port
,
name
,
dbsize
,
node
->
slots_count
,
replicas
);
masters
++
;
keys
+=
dbsize
;
...
...
@@ -2626,7 +2626,7 @@ static int clusterManagerAddSlots(clusterManagerNode *node, char**err)
void
*
_reply
=
NULL
;
int
success
=
1
;
/* First two args are used for the command itself. */
int
argc
=
node
->
slots_count
+
2
;
int
argc
=
node
->
slots_count
+
2
;
sds
*
argv
=
zmalloc
(
argc
*
sizeof
(
*
argv
));
size_t
*
argvlen
=
zmalloc
(
argc
*
sizeof
(
*
argvlen
));
argv
[
0
]
=
"CLUSTER"
;
...
...
@@ -2665,12 +2665,12 @@ cleanup:
}
/* Set slot status to "importing" or "migrating" */
static
int
clusterManagerSetSlot
(
clusterManagerNode
*
node1
,
clusterManagerNode
*
node2
,
static
int
clusterManagerSetSlot
(
clusterManagerNode
*
node1
,
clusterManagerNode
*
node2
,
int
slot
,
const
char
*
status
,
char
**
err
)
{
redisReply
*
reply
=
CLUSTER_MANAGER_COMMAND
(
node1
,
"CLUSTER "
"SETSLOT %d %s %s"
,
slot
,
status
,
"SETSLOT %d %s %s"
,
slot
,
status
,
(
char
*
)
node2
->
name
);
if
(
err
!=
NULL
)
*
err
=
NULL
;
if
(
!
reply
)
return
0
;
...
...
@@ -2678,8 +2678,8 @@ static int clusterManagerSetSlot(clusterManagerNode *node1,
if
(
reply
->
type
==
REDIS_REPLY_ERROR
)
{
success
=
0
;
if
(
err
!=
NULL
)
{
*
err
=
zmalloc
((
reply
->
len
+
1
)
*
sizeof
(
char
));
strcpy
(
*
err
,
reply
->
str
);
*
err
=
zmalloc
((
reply
->
len
+
1
)
*
sizeof
(
char
));
strcpy
(
*
err
,
reply
->
str
);
CLUSTER_MANAGER_PRINT_REPLY_ERROR
(
node1
,
err
);
}
goto
cleanup
;
...
...
@@ -2689,14 +2689,14 @@ cleanup:
return
success
;
}
/* Migrate keys taken from reply->elements. It returns the reply from the
* MIGRATE command, or NULL if something goes wrong. If the argument 'dots'
/* Migrate keys taken from reply->elements. It returns the reply from the
* MIGRATE command, or NULL if something goes wrong. If the argument 'dots'
* is not NULL, a dot will be printed for every migrated key. */
static
redisReply
*
clusterManagerMigrateKeysInReply
(
clusterManagerNode
*
source
,
clusterManagerNode
*
target
,
redisReply
*
reply
,
int
replace
,
int
timeout
,
char
*
dots
)
static
redisReply
*
clusterManagerMigrateKeysInReply
(
clusterManagerNode
*
source
,
clusterManagerNode
*
target
,
redisReply
*
reply
,
int
replace
,
int
timeout
,
char
*
dots
)
{
redisReply
*
migrate_reply
=
NULL
;
char
**
argv
=
NULL
;
...
...
@@ -2753,28 +2753,28 @@ cleanup:
}
/* Migrate all keys in the given slot from source to target.*/
static
int
clusterManagerMigrateKeysInSlot
(
clusterManagerNode
*
source
,
clusterManagerNode
*
target
,
int
slot
,
int
timeout
,
static
int
clusterManagerMigrateKeysInSlot
(
clusterManagerNode
*
source
,
clusterManagerNode
*
target
,
int
slot
,
int
timeout
,
int
pipeline
,
int
verbose
,
char
**
err
)
char
**
err
)
{
int
success
=
1
;
int
do_fix
=
(
config
.
cluster_manager_command
.
flags
&
int
do_fix
=
(
config
.
cluster_manager_command
.
flags
&
CLUSTER_MANAGER_CMD_FLAG_FIX
);
while
(
1
)
{
char
*
dots
=
NULL
;
redisReply
*
reply
=
NULL
,
*
migrate_reply
=
NULL
;
reply
=
CLUSTER_MANAGER_COMMAND
(
source
,
"CLUSTER "
"GETKEYSINSLOT %d %d"
,
slot
,
"GETKEYSINSLOT %d %d"
,
slot
,
pipeline
);
success
=
(
reply
!=
NULL
);
if
(
!
success
)
return
0
;
if
(
reply
->
type
==
REDIS_REPLY_ERROR
)
{
success
=
0
;
if
(
err
!=
NULL
)
{
*
err
=
zmalloc
((
reply
->
len
+
1
)
*
sizeof
(
char
));
strcpy
(
*
err
,
reply
->
str
);
*
err
=
zmalloc
((
reply
->
len
+
1
)
*
sizeof
(
char
));
strcpy
(
*
err
,
reply
->
str
);
CLUSTER_MANAGER_PRINT_REPLY_ERROR
(
source
,
err
);
}
goto
next
;
...
...
@@ -2787,8 +2787,8 @@ static int clusterManagerMigrateKeysInSlot(clusterManagerNode *source,
}
if
(
verbose
)
dots
=
zmalloc
((
count
+
1
)
*
sizeof
(
char
));
/* Calling MIGRATE command. */
migrate_reply
=
clusterManagerMigrateKeysInReply
(
source
,
target
,
reply
,
0
,
timeout
,
migrate_reply
=
clusterManagerMigrateKeysInReply
(
source
,
target
,
reply
,
0
,
timeout
,
dots
);
if
(
migrate_reply
==
NULL
)
goto
next
;
if
(
migrate_reply
->
type
==
REDIS_REPLY_ERROR
)
{
...
...
@@ -2797,22 +2797,22 @@ static int clusterManagerMigrateKeysInSlot(clusterManagerNode *source,
"Replacing it for FIX.
\n
"
);
freeReplyObject
(
migrate_reply
);
/* Try to migrate keys adding REPLACE option. */
migrate_reply
=
clusterManagerMigrateKeysInReply
(
source
,
target
,
reply
,
migrate_reply
=
clusterManagerMigrateKeysInReply
(
source
,
target
,
reply
,
1
,
timeout
,
NULL
);
success
=
(
migrate_reply
!=
NULL
&&
success
=
(
migrate_reply
!=
NULL
&&
migrate_reply
->
type
!=
REDIS_REPLY_ERROR
);
}
else
success
=
0
;
if
(
!
success
)
{
if
(
migrate_reply
!=
NULL
)
{
if
(
err
)
{
*
err
=
zmalloc
((
migrate_reply
->
len
+
1
)
*
sizeof
(
char
));
strcpy
(
*
err
,
migrate_reply
->
str
);
strcpy
(
*
err
,
migrate_reply
->
str
);
}
printf
(
"
\n
"
);
CLUSTER_MANAGER_PRINT_REPLY_ERROR
(
source
,
CLUSTER_MANAGER_PRINT_REPLY_ERROR
(
source
,
migrate_reply
->
str
);
}
goto
next
;
...
...
@@ -2832,34 +2832,34 @@ next:
}
/* Move slots between source and target nodes using MIGRATE.
*
*
* Options:
* CLUSTER_MANAGER_OPT_VERBOSE -- Print a dot for every moved key.
* CLUSTER_MANAGER_OPT_COLD -- Move keys without opening slots /
* CLUSTER_MANAGER_OPT_COLD -- Move keys without opening slots /
* reconfiguring the nodes.
* CLUSTER_MANAGER_OPT_UPDATE -- Update node->slots for source/target nodes.
* CLUSTER_MANAGER_OPT_QUIET -- Don't print info messages.
*/
static
int
clusterManagerMoveSlot
(
clusterManagerNode
*
source
,
clusterManagerNode
*
target
,
int
slot
,
int
opts
,
char
**
err
)
static
int
clusterManagerMoveSlot
(
clusterManagerNode
*
source
,
clusterManagerNode
*
target
,
int
slot
,
int
opts
,
char
**
err
)
{
if
(
!
(
opts
&
CLUSTER_MANAGER_OPT_QUIET
))
{
printf
(
"Moving slot %d from %s:%d to %s:%d: "
,
slot
,
source
->
ip
,
printf
(
"Moving slot %d from %s:%d to %s:%d: "
,
slot
,
source
->
ip
,
source
->
port
,
target
->
ip
,
target
->
port
);
fflush
(
stdout
);
}
if
(
err
!=
NULL
)
*
err
=
NULL
;
int
pipeline
=
config
.
cluster_manager_command
.
pipeline
,
int
pipeline
=
config
.
cluster_manager_command
.
pipeline
,
timeout
=
config
.
cluster_manager_command
.
timeout
,
print_dots
=
(
opts
&
CLUSTER_MANAGER_OPT_VERBOSE
),
option_cold
=
(
opts
&
CLUSTER_MANAGER_OPT_COLD
),
success
=
1
;
if
(
!
option_cold
)
{
success
=
clusterManagerSetSlot
(
target
,
source
,
slot
,
success
=
clusterManagerSetSlot
(
target
,
source
,
slot
,
"importing"
,
err
);
if
(
!
success
)
return
0
;
success
=
clusterManagerSetSlot
(
source
,
target
,
slot
,
success
=
clusterManagerSetSlot
(
source
,
target
,
slot
,
"migrating"
,
err
);
if
(
!
success
)
return
0
;
}
...
...
@@ -2876,16 +2876,16 @@ static int clusterManagerMoveSlot(clusterManagerNode *source,
clusterManagerNode
*
n
=
ln
->
value
;
if
(
n
->
flags
&
CLUSTER_MANAGER_FLAG_SLAVE
)
continue
;
redisReply
*
r
=
CLUSTER_MANAGER_COMMAND
(
n
,
"CLUSTER "
"SETSLOT %d %s %s"
,
slot
,
"node"
,
target
->
name
);
"SETSLOT %d %s %s"
,
slot
,
"node"
,
target
->
name
);
success
=
(
r
!=
NULL
);
if
(
!
success
)
return
0
;
if
(
r
->
type
==
REDIS_REPLY_ERROR
)
{
success
=
0
;
if
(
err
!=
NULL
)
{
*
err
=
zmalloc
((
r
->
len
+
1
)
*
sizeof
(
char
));
strcpy
(
*
err
,
r
->
str
);
*
err
=
zmalloc
((
r
->
len
+
1
)
*
sizeof
(
char
));
strcpy
(
*
err
,
r
->
str
);
CLUSTER_MANAGER_PRINT_REPLY_ERROR
(
n
,
err
);
}
}
...
...
@@ -2901,7 +2901,7 @@ static int clusterManagerMoveSlot(clusterManagerNode *source,
return
1
;
}
/* Flush the dirty node configuration by calling replicate for slaves or
/* Flush the dirty node configuration by calling replicate for slaves or
* adding the slots defined in the masters. */
static
int
clusterManagerFlushNodeConfig
(
clusterManagerNode
*
node
,
char
**
err
)
{
if
(
!
node
->
dirty
)
return
0
;
...
...
@@ -2909,13 +2909,13 @@ static int clusterManagerFlushNodeConfig(clusterManagerNode *node, char **err) {
int
is_err
=
0
,
success
=
1
;
if
(
err
!=
NULL
)
*
err
=
NULL
;
if
(
node
->
replicate
!=
NULL
)
{
reply
=
CLUSTER_MANAGER_COMMAND
(
node
,
"CLUSTER REPLICATE %s"
,
node
->
replicate
);
reply
=
CLUSTER_MANAGER_COMMAND
(
node
,
"CLUSTER REPLICATE %s"
,
node
->
replicate
);
if
(
reply
==
NULL
||
(
is_err
=
(
reply
->
type
==
REDIS_REPLY_ERROR
)))
{
if
(
is_err
&&
err
!=
NULL
)
{
*
err
=
zmalloc
((
reply
->
len
+
1
)
*
sizeof
(
char
));
strcpy
(
*
err
,
reply
->
str
);
}
*
err
=
zmalloc
((
reply
->
len
+
1
)
*
sizeof
(
char
));
strcpy
(
*
err
,
reply
->
str
);
}
success
=
0
;
/* If the cluster did not already joined it is possible that
* the slave does not know the master node yet. So on errors
...
...
@@ -2947,12 +2947,12 @@ static void clusterManagerWaitForClusterJoin(void) {
/* Load node's cluster configuration by calling "CLUSTER NODES" command.
* Node's configuration (name, replicate, slots, ...) is then updated.
* If CLUSTER_MANAGER_OPT_GETFRIENDS flag is set into 'opts' argument,
* and node already knows other nodes, the node's friends list is populated
* and node already knows other nodes, the node's friends list is populated
* with the other nodes info. */
static
int
clusterManagerNodeLoadInfo
(
clusterManagerNode
*
node
,
int
opts
,
char
**
err
)
static
int
clusterManagerNodeLoadInfo
(
clusterManagerNode
*
node
,
int
opts
,
char
**
err
)
{
redisReply
*
reply
=
CLUSTER_MANAGER_COMMAND
(
node
,
"CLUSTER NODES"
);
redisReply
*
reply
=
CLUSTER_MANAGER_COMMAND
(
node
,
"CLUSTER NODES"
);
int
success
=
1
;
*
err
=
NULL
;
if
(
!
clusterManagerCheckRedisReply
(
node
,
reply
,
err
))
{
...
...
@@ -2962,16 +2962,16 @@ static int clusterManagerNodeLoadInfo(clusterManagerNode *node, int opts,
int
getfriends
=
(
opts
&
CLUSTER_MANAGER_OPT_GETFRIENDS
);
char
*
lines
=
reply
->
str
,
*
p
,
*
line
;
while
((
p
=
strstr
(
lines
,
"
\n
"
))
!=
NULL
)
{
*
p
=
'\0'
;
*
p
=
'\0'
;
line
=
lines
;
lines
=
p
+
1
;
char
*
name
=
NULL
,
*
addr
=
NULL
,
*
flags
=
NULL
,
*
master_id
=
NULL
,
*
ping_sent
=
NULL
,
*
ping_recv
=
NULL
,
*
config_epoch
=
NULL
,
char
*
name
=
NULL
,
*
addr
=
NULL
,
*
flags
=
NULL
,
*
master_id
=
NULL
,
*
ping_sent
=
NULL
,
*
ping_recv
=
NULL
,
*
config_epoch
=
NULL
,
*
link_status
=
NULL
;
UNUSED
(
link_status
);
int
i
=
0
;
while
((
p
=
strchr
(
line
,
' '
))
!=
NULL
)
{
*
p
=
'\0'
;
*
p
=
'\0'
;
char
*
token
=
line
;
line
=
p
+
1
;
switch
(
i
++
){
...
...
@@ -2985,7 +2985,7 @@ static int clusterManagerNodeLoadInfo(clusterManagerNode *node, int opts,
case
7
:
link_status
=
token
;
break
;
}
if
(
i
==
8
)
break
;
// Slots
}
}
if
(
!
flags
)
{
success
=
0
;
goto
cleanup
;
...
...
@@ -3004,7 +3004,7 @@ static int clusterManagerNodeLoadInfo(clusterManagerNode *node, int opts,
remaining
-=
(
p
-
line
);
char
*
slotsdef
=
line
;
*
p
=
'\0'
;
*
p
=
'\0'
;
if
(
remaining
)
{
line
=
p
+
1
;
remaining
--
;
...
...
@@ -3020,11 +3020,11 @@ static int clusterManagerNodeLoadInfo(clusterManagerNode *node, int opts,
sds
slot
=
sdsnew
(
slotsdef
);
sds
dst
=
sdsnew
(
p
);
node
->
migrating_count
+=
2
;
node
->
migrating
=
zrealloc
(
node
->
migrating
,
node
->
migrating
=
zrealloc
(
node
->
migrating
,
(
node
->
migrating_count
*
sizeof
(
sds
)));
node
->
migrating
[
node
->
migrating_count
-
2
]
=
node
->
migrating
[
node
->
migrating_count
-
2
]
=
slot
;
node
->
migrating
[
node
->
migrating_count
-
1
]
=
node
->
migrating
[
node
->
migrating_count
-
1
]
=
dst
;
}
else
if
((
p
=
strstr
(
slotsdef
,
"-<-"
)))
{
//Importing
*
p
=
'\0'
;
...
...
@@ -3034,13 +3034,13 @@ static int clusterManagerNodeLoadInfo(clusterManagerNode *node, int opts,
sds
slot
=
sdsnew
(
slotsdef
);
sds
src
=
sdsnew
(
p
);
node
->
importing_count
+=
2
;
node
->
importing
=
zrealloc
(
node
->
importing
,
node
->
importing
=
zrealloc
(
node
->
importing
,
(
node
->
importing_count
*
sizeof
(
sds
)));
node
->
importing
[
node
->
importing_count
-
2
]
=
node
->
importing
[
node
->
importing_count
-
2
]
=
slot
;
node
->
importing
[
node
->
importing_count
-
1
]
=
node
->
importing
[
node
->
importing_count
-
1
]
=
src
;
}
}
}
else
if
((
dash
=
strchr
(
slotsdef
,
'-'
))
!=
NULL
)
{
p
=
dash
;
int
start
,
stop
;
...
...
@@ -3092,7 +3092,7 @@ static int clusterManagerNodeLoadInfo(clusterManagerNode *node, int opts,
sds
flag
=
NULL
;
char
*
fp
=
strchr
(
flags
,
','
);
if
(
fp
)
{
*
fp
=
'\0'
;
*
fp
=
'\0'
;
flag
=
sdsnew
(
flags
);
flags
=
fp
+
1
;
}
else
{
...
...
@@ -3114,7 +3114,7 @@ static int clusterManagerNodeLoadInfo(clusterManagerNode *node, int opts,
}
listAddNodeTail
(
currentNode
->
flags_str
,
flag
);
}
if
(
config_epoch
!=
NULL
)
if
(
config_epoch
!=
NULL
)
currentNode
->
current_epoch
=
atoll
(
config_epoch
);
if
(
ping_sent
!=
NULL
)
currentNode
->
ping_sent
=
atoll
(
ping_sent
);
if
(
ping_recv
!=
NULL
)
currentNode
->
ping_recv
=
atoll
(
ping_recv
);
...
...
@@ -3125,16 +3125,16 @@ cleanup:
return
success
;
}
/* Retrieves info about the cluster using argument 'node' as the starting
/* Retrieves info about the cluster using argument 'node' as the starting
* point. All nodes will be loaded inside the cluster_manager.nodes list.
* Warning: if something goes wrong, it will free the starting node before
* Warning: if something goes wrong, it will free the starting node before
* returning 0. */
static
int
clusterManagerLoadInfoFromNode
(
clusterManagerNode
*
node
,
int
opts
)
{
if
(
node
->
context
==
NULL
)
if
(
node
->
context
==
NULL
)
CLUSTER_MANAGER_NODE_CONNECT
(
node
);
if
(
node
->
context
->
err
)
{
fprintf
(
stderr
,
"Could not connect to Redis at "
);
fprintf
(
stderr
,
"%s:%d: %s
\n
"
,
node
->
ip
,
node
->
port
,
fprintf
(
stderr
,
"%s:%d: %s
\n
"
,
node
->
ip
,
node
->
port
,
node
->
context
->
errstr
);
freeClusterManagerNode
(
node
);
return
0
;
...
...
@@ -3160,7 +3160,7 @@ static int clusterManagerLoadInfoFromNode(clusterManagerNode *node, int opts) {
listNode
*
ln
;
if
(
cluster_manager
.
nodes
!=
NULL
)
{
listRewind
(
cluster_manager
.
nodes
,
&
li
);
while
((
ln
=
listNext
(
&
li
))
!=
NULL
)
while
((
ln
=
listNext
(
&
li
))
!=
NULL
)
freeClusterManagerNode
((
clusterManagerNode
*
)
ln
->
value
);
listRelease
(
cluster_manager
.
nodes
);
}
...
...
@@ -3176,8 +3176,8 @@ static int clusterManagerLoadInfoFromNode(clusterManagerNode *node, int opts) {
if
(
friend
->
context
->
err
)
goto
invalid_friend
;
e
=
NULL
;
if
(
clusterManagerNodeLoadInfo
(
friend
,
0
,
&
e
))
{
if
(
friend
->
flags
&
(
CLUSTER_MANAGER_FLAG_NOADDR
|
CLUSTER_MANAGER_FLAG_DISCONNECT
|
if
(
friend
->
flags
&
(
CLUSTER_MANAGER_FLAG_NOADDR
|
CLUSTER_MANAGER_FLAG_DISCONNECT
|
CLUSTER_MANAGER_FLAG_FAIL
))
goto
invalid_friend
;
listAddNodeTail
(
cluster_manager
.
nodes
,
friend
);
...
...
@@ -3197,25 +3197,25 @@ invalid_friend:
// Count replicas for each node
listRewind
(
cluster_manager
.
nodes
,
&
li
);
while
((
ln
=
listNext
(
&
li
))
!=
NULL
)
{
clusterManagerNode
*
n
=
ln
->
value
;
clusterManagerNode
*
n
=
ln
->
value
;
if
(
n
->
replicate
!=
NULL
)
{
clusterManagerNode
*
master
=
clusterManagerNodeByName
(
n
->
replicate
);
if
(
master
==
NULL
)
{
clusterManagerLogWarn
(
"*** WARNING: %s:%d claims to be "
"slave of unknown node ID %s.
\n
"
,
"slave of unknown node ID %s.
\n
"
,
n
->
ip
,
n
->
port
,
n
->
replicate
);
}
else
master
->
replicas_count
++
;
}
}
return
1
;
return
1
;
}
/* Compare functions used by various sorting operations. */
int
clusterManagerSlotCompare
(
const
void
*
slot1
,
const
void
*
slot2
)
{
int
clusterManagerSlotCompare
(
const
void
*
slot1
,
const
void
*
slot2
)
{
const
char
**
i1
=
(
const
char
**
)
slot1
;
const
char
**
i2
=
(
const
char
**
)
slot2
;
return
strcmp
(
*
i1
,
*
i2
);
}
}
int
clusterManagerSlotCountCompareDesc
(
const
void
*
n1
,
const
void
*
n2
)
{
clusterManagerNode
*
node1
=
*
((
clusterManagerNode
**
)
n1
);
...
...
@@ -3233,19 +3233,19 @@ static sds clusterManagerGetConfigSignature(clusterManagerNode *node) {
sds
signature
=
NULL
;
int
node_count
=
0
,
i
=
0
,
name_len
=
0
;
char
**
node_configs
=
NULL
;
redisReply
*
reply
=
CLUSTER_MANAGER_COMMAND
(
node
,
"CLUSTER NODES"
);
redisReply
*
reply
=
CLUSTER_MANAGER_COMMAND
(
node
,
"CLUSTER NODES"
);
if
(
reply
==
NULL
||
reply
->
type
==
REDIS_REPLY_ERROR
)
goto
cleanup
;
char
*
lines
=
reply
->
str
,
*
p
,
*
line
;
while
((
p
=
strstr
(
lines
,
"
\n
"
))
!=
NULL
)
{
i
=
0
;
*
p
=
'\0'
;
*
p
=
'\0'
;
line
=
lines
;
lines
=
p
+
1
;
char
*
nodename
=
NULL
;
int
tot_size
=
0
;
while
((
p
=
strchr
(
line
,
' '
))
!=
NULL
)
{
*
p
=
'\0'
;
*
p
=
'\0'
;
char
*
token
=
line
;
line
=
p
+
1
;
if
(
i
==
0
)
{
...
...
@@ -3268,7 +3268,7 @@ static sds clusterManagerGetConfigSignature(clusterManagerNode *node) {
remaining
-=
size
;
tot_size
+=
size
;
char
*
slotsdef
=
line
;
*
p
=
'\0'
;
*
p
=
'\0'
;
if
(
remaining
)
{
line
=
p
+
1
;
remaining
--
;
...
...
@@ -3281,9 +3281,9 @@ static sds clusterManagerGetConfigSignature(clusterManagerNode *node) {
}
if
(
c
>
0
)
{
if
(
c
>
1
)
qsort
(
slots
,
c
,
sizeof
(
char
*
),
clusterManagerSlotCompare
);
qsort
(
slots
,
c
,
sizeof
(
char
*
),
clusterManagerSlotCompare
);
node_count
++
;
node_configs
=
node_configs
=
zrealloc
(
node_configs
,
(
node_count
*
sizeof
(
char
*
)));
/* Make room for '|' separators. */
tot_size
+=
(
sizeof
(
char
)
*
(
c
-
1
));
...
...
@@ -3304,12 +3304,12 @@ static sds clusterManagerGetConfigSignature(clusterManagerNode *node) {
}
if
(
node_count
>
0
)
{
if
(
node_count
>
1
)
{
qsort
(
node_configs
,
node_count
,
sizeof
(
char
*
),
qsort
(
node_configs
,
node_count
,
sizeof
(
char
*
),
clusterManagerSlotCompare
);
}
signature
=
sdsempty
();
for
(
i
=
0
;
i
<
node_count
;
i
++
)
{
if
(
i
>
0
)
signature
=
sdscatprintf
(
signature
,
"%c"
,
'|'
);
if
(
i
>
0
)
signature
=
sdscatprintf
(
signature
,
"%c"
,
'|'
);
signature
=
sdscatfmt
(
signature
,
"%s"
,
node_configs
[
i
]);
}
}
...
...
@@ -3357,8 +3357,8 @@ static void clusterManagerOnError(sds err) {
clusterManagerLogErr
(
"%s
\n
"
,
(
char
*
)
err
);
}
/* Check the slots coverage of the cluster. The 'all_slots' argument must be
* and array of 16384 bytes. Every covered slot will be set to 1 in the
/* Check the slots coverage of the cluster. The 'all_slots' argument must be
* and array of 16384 bytes. Every covered slot will be set to 1 in the
* 'all_slots' array. The function returns the total number if covered slots.*/
static
int
clusterManagerGetCoveredSlots
(
char
*
all_slots
)
{
if
(
cluster_manager
.
nodes
==
NULL
)
return
0
;
...
...
@@ -3384,7 +3384,7 @@ static void clusterManagerPrintSlotsList(list *slots) {
listRewind
(
slots
,
&
li
);
sds
first
=
NULL
;
while
((
ln
=
listNext
(
&
li
))
!=
NULL
)
{
sds
slot
=
ln
->
value
;
sds
slot
=
ln
->
value
;
if
(
!
first
)
first
=
slot
;
else
printf
(
", "
);
printf
(
"%s"
,
slot
);
...
...
@@ -3395,8 +3395,8 @@ static void clusterManagerPrintSlotsList(list *slots) {
/* Return the node, among 'nodes' with the greatest number of keys
* in the specified slot. */
static
clusterManagerNode
*
clusterManagerGetNodeWithMostKeysInSlot
(
list
*
nodes
,
int
slot
,
char
**
err
)
int
slot
,
char
**
err
)
{
clusterManagerNode
*
node
=
NULL
;
int
numkeys
=
0
;
...
...
@@ -3408,19 +3408,19 @@ static clusterManagerNode * clusterManagerGetNodeWithMostKeysInSlot(list *nodes,
clusterManagerNode
*
n
=
ln
->
value
;
if
(
n
->
flags
&
CLUSTER_MANAGER_FLAG_SLAVE
||
n
->
replicate
)
continue
;
redisReply
*
r
=
CLUSTER_MANAGER_COMMAND
(
n
,
"CLUSTER COUNTKEYSINSLOT %d"
,
slot
);
redisReply
*
r
=
CLUSTER_MANAGER_COMMAND
(
n
,
"CLUSTER COUNTKEYSINSLOT %d"
,
slot
);
int
success
=
clusterManagerCheckRedisReply
(
n
,
r
,
err
);
if
(
success
)
{
if
(
r
->
integer
>
numkeys
||
node
==
NULL
)
{
numkeys
=
r
->
integer
;
node
=
n
;
}
}
}
if
(
r
!=
NULL
)
freeReplyObject
(
r
);
/* If the reply contains errors */
if
(
!
success
)
{
if
(
err
!=
NULL
&&
*
err
!=
NULL
)
if
(
err
!=
NULL
&&
*
err
!=
NULL
)
CLUSTER_MANAGER_PRINT_REPLY_ERROR
(
n
,
err
);
node
=
NULL
;
break
;
...
...
@@ -3458,7 +3458,7 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
int
uncovered_count
=
0
;
sds
log
=
sdsempty
();
for
(
i
=
0
;
i
<
CLUSTER_MANAGER_SLOTS
;
i
++
)
{
int
covered
=
all_slots
[
i
];
int
covered
=
all_slots
[
i
];
if
(
!
covered
)
{
sds
key
=
sdsfromlonglong
((
long
long
)
i
);
if
(
uncovered_count
++
>
0
)
printf
(
","
);
...
...
@@ -3469,10 +3469,10 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
listNode
*
ln
;
listRewind
(
cluster_manager
.
nodes
,
&
li
);
while
((
ln
=
listNext
(
&
li
))
!=
NULL
)
{
clusterManagerNode
*
n
=
ln
->
value
;
clusterManagerNode
*
n
=
ln
->
value
;
if
(
n
->
flags
&
CLUSTER_MANAGER_FLAG_SLAVE
||
n
->
replicate
)
continue
;
redisReply
*
reply
=
CLUSTER_MANAGER_COMMAND
(
n
,
redisReply
*
reply
=
CLUSTER_MANAGER_COMMAND
(
n
,
"CLUSTER GETKEYSINSLOT %d %d"
,
i
,
1
);
if
(
!
clusterManagerCheckRedisReply
(
n
,
reply
,
NULL
))
{
fixed
=
-
1
;
...
...
@@ -3484,12 +3484,12 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
listAddNodeTail
(
slot_nodes
,
n
);
if
(
listLength
(
slot_nodes
)
>
1
)
slot_nodes_str
=
sdscat
(
slot_nodes_str
,
", "
);
slot_nodes_str
=
sdscatfmt
(
slot_nodes_str
,
slot_nodes_str
=
sdscatfmt
(
slot_nodes_str
,
"%s:%u"
,
n
->
ip
,
n
->
port
);
}
freeReplyObject
(
reply
);
}
log
=
sdscatfmt
(
log
,
"
\n
Slot %S has keys in %u nodes: %S"
,
log
=
sdscatfmt
(
log
,
"
\n
Slot %S has keys in %u nodes: %S"
,
key
,
listLength
(
slot_nodes
),
slot_nodes_str
);
sdsfree
(
slot_nodes_str
);
dictAdd
(
clusterManagerUncoveredSlots
,
key
,
slot_nodes
);
...
...
@@ -3512,7 +3512,7 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
case
0
:
listAddNodeTail
(
none
,
slot
);
break
;
case
1
:
listAddNodeTail
(
single
,
slot
);
break
;
default:
listAddNodeTail
(
multi
,
slot
);
break
;
}
}
}
dictReleaseIterator
(
iter
);
...
...
@@ -3527,19 +3527,19 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
listNode
*
ln
;
listRewind
(
none
,
&
li
);
while
((
ln
=
listNext
(
&
li
))
!=
NULL
)
{
sds
slot
=
ln
->
value
;
sds
slot
=
ln
->
value
;
long
idx
=
(
long
)
(
rand
()
%
listLength
(
cluster_manager
.
nodes
));
listNode
*
node_n
=
listIndex
(
cluster_manager
.
nodes
,
idx
);
assert
(
node_n
!=
NULL
);
clusterManagerNode
*
n
=
node_n
->
value
;
clusterManagerLogInfo
(
">>> Covering slot %s with %s:%d
\n
"
,
clusterManagerLogInfo
(
">>> Covering slot %s with %s:%d
\n
"
,
slot
,
n
->
ip
,
n
->
port
);
redisReply
*
r
=
CLUSTER_MANAGER_COMMAND
(
n
,
redisReply
*
r
=
CLUSTER_MANAGER_COMMAND
(
n
,
"CLUSTER ADDSLOTS %s"
,
slot
);
if
(
!
clusterManagerCheckRedisReply
(
n
,
r
,
NULL
))
fixed
=
-
1
;
if
(
r
)
freeReplyObject
(
r
);
if
(
fixed
<
0
)
goto
cleanup
;
/* Since CLUSTER ADDSLOTS succeeded, we also update the slot
/* Since CLUSTER ADDSLOTS succeeded, we also update the slot
* info into the node struct, in order to keep it synced */
n
->
slots
[
atoi
(
slot
)]
=
1
;
fixed
++
;
...
...
@@ -3556,21 +3556,21 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
listNode
*
ln
;
listRewind
(
single
,
&
li
);
while
((
ln
=
listNext
(
&
li
))
!=
NULL
)
{
sds
slot
=
ln
->
value
;
sds
slot
=
ln
->
value
;
dictEntry
*
entry
=
dictFind
(
clusterManagerUncoveredSlots
,
slot
);
assert
(
entry
!=
NULL
);
list
*
nodes
=
(
list
*
)
dictGetVal
(
entry
);
listNode
*
fn
=
listFirst
(
nodes
);
assert
(
fn
!=
NULL
);
clusterManagerNode
*
n
=
fn
->
value
;
clusterManagerLogInfo
(
">>> Covering slot %s with %s:%d
\n
"
,
clusterManagerLogInfo
(
">>> Covering slot %s with %s:%d
\n
"
,
slot
,
n
->
ip
,
n
->
port
);
redisReply
*
r
=
CLUSTER_MANAGER_COMMAND
(
n
,
redisReply
*
r
=
CLUSTER_MANAGER_COMMAND
(
n
,
"CLUSTER ADDSLOTS %s"
,
slot
);
if
(
!
clusterManagerCheckRedisReply
(
n
,
r
,
NULL
))
fixed
=
-
1
;
if
(
r
)
freeReplyObject
(
r
);
if
(
fixed
<
0
)
goto
cleanup
;
/* Since CLUSTER ADDSLOTS succeeded, we also update the slot
/* Since CLUSTER ADDSLOTS succeeded, we also update the slot
* info into the node struct, in order to keep it synced */
n
->
slots
[
atoi
(
slot
)]
=
1
;
fixed
++
;
...
...
@@ -3588,21 +3588,21 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
listNode
*
ln
;
listRewind
(
multi
,
&
li
);
while
((
ln
=
listNext
(
&
li
))
!=
NULL
)
{
sds
slot
=
ln
->
value
;
sds
slot
=
ln
->
value
;
dictEntry
*
entry
=
dictFind
(
clusterManagerUncoveredSlots
,
slot
);
assert
(
entry
!=
NULL
);
list
*
nodes
=
(
list
*
)
dictGetVal
(
entry
);
int
s
=
atoi
(
slot
);
clusterManagerNode
*
target
=
clusterManagerNode
*
target
=
clusterManagerGetNodeWithMostKeysInSlot
(
nodes
,
s
,
NULL
);
if
(
target
==
NULL
)
{
fixed
=
-
1
;
goto
cleanup
;
}
clusterManagerLogInfo
(
">>> Covering slot %s moving keys "
"to %s:%d
\n
"
,
slot
,
"to %s:%d
\n
"
,
slot
,
target
->
ip
,
target
->
port
);
redisReply
*
r
=
CLUSTER_MANAGER_COMMAND
(
target
,
redisReply
*
r
=
CLUSTER_MANAGER_COMMAND
(
target
,
"CLUSTER ADDSLOTS %s"
,
slot
);
if
(
!
clusterManagerCheckRedisReply
(
target
,
r
,
NULL
))
fixed
=
-
1
;
if
(
r
)
freeReplyObject
(
r
);
...
...
@@ -3612,7 +3612,7 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
if
(
!
clusterManagerCheckRedisReply
(
target
,
r
,
NULL
))
fixed
=
-
1
;
if
(
r
)
freeReplyObject
(
r
);
if
(
fixed
<
0
)
goto
cleanup
;
/* Since CLUSTER ADDSLOTS succeeded, we also update the slot
/* Since CLUSTER ADDSLOTS succeeded, we also update the slot
* info into the node struct, in order to keep it synced */
target
->
slots
[
atoi
(
slot
)]
=
1
;
listIter
nli
;
...
...
@@ -3621,18 +3621,18 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
while
((
nln
=
listNext
(
&
nli
))
!=
NULL
)
{
clusterManagerNode
*
src
=
nln
->
value
;
if
(
src
==
target
)
continue
;
/* Set the source node in 'importing' state
* (even if we will actually migrate keys away)
* in order to avoid receiving redirections
/* Set the source node in 'importing' state
* (even if we will actually migrate keys away)
* in order to avoid receiving redirections
* for MIGRATE. */
redisReply
*
r
=
CLUSTER_MANAGER_COMMAND
(
src
,
"CLUSTER SETSLOT %s %s %s"
,
slot
,
redisReply
*
r
=
CLUSTER_MANAGER_COMMAND
(
src
,
"CLUSTER SETSLOT %s %s %s"
,
slot
,
"IMPORTING"
,
target
->
name
);
if
(
!
clusterManagerCheckRedisReply
(
target
,
r
,
NULL
))
if
(
!
clusterManagerCheckRedisReply
(
target
,
r
,
NULL
))
fixed
=
-
1
;
if
(
r
)
freeReplyObject
(
r
);
if
(
fixed
<
0
)
goto
cleanup
;
int
opts
=
CLUSTER_MANAGER_OPT_VERBOSE
|
int
opts
=
CLUSTER_MANAGER_OPT_VERBOSE
|
CLUSTER_MANAGER_OPT_COLD
;
if
(
!
clusterManagerMoveSlot
(
src
,
target
,
s
,
opts
,
NULL
))
{
fixed
=
-
1
;
...
...
@@ -3673,7 +3673,7 @@ static int clusterManagerFixOpenSlot(int slot) {
if
(
n
->
flags
&
CLUSTER_MANAGER_FLAG_SLAVE
)
continue
;
if
(
n
->
slots
[
slot
])
{
if
(
owner
==
NULL
)
owner
=
n
;
listAddNodeTail
(
owners
,
n
);
listAddNodeTail
(
owners
,
n
);
}
}
listRewind
(
cluster_manager
.
nodes
,
&
li
);
...
...
@@ -3682,10 +3682,10 @@ static int clusterManagerFixOpenSlot(int slot) {
if
(
n
->
flags
&
CLUSTER_MANAGER_FLAG_SLAVE
)
continue
;
if
(
n
->
migrating
)
{
for
(
int
i
=
0
;
i
<
n
->
migrating_count
;
i
+=
2
)
{
sds
migrating_slot
=
n
->
migrating
[
i
];
sds
migrating_slot
=
n
->
migrating
[
i
];
if
(
atoi
(
migrating_slot
)
==
slot
)
{
char
*
sep
=
(
listLength
(
migrating
)
==
0
?
""
:
","
);
migrating_str
=
sdscatfmt
(
migrating_str
,
"%s%S:%u"
,
migrating_str
=
sdscatfmt
(
migrating_str
,
"%s%S:%u"
,
sep
,
n
->
ip
,
n
->
port
);
listAddNodeTail
(
migrating
,
n
);
break
;
...
...
@@ -3694,10 +3694,10 @@ static int clusterManagerFixOpenSlot(int slot) {
}
if
(
n
->
importing
)
{
for
(
int
i
=
0
;
i
<
n
->
importing_count
;
i
+=
2
)
{
sds
importing_slot
=
n
->
importing
[
i
];
sds
importing_slot
=
n
->
importing
[
i
];
if
(
atoi
(
importing_slot
)
==
slot
)
{
char
*
sep
=
(
listLength
(
importing
)
==
0
?
""
:
","
);
importing_str
=
sdscatfmt
(
importing_str
,
"%s%S:%u"
,
importing_str
=
sdscatfmt
(
importing_str
,
"%s%S:%u"
,
sep
,
n
->
ip
,
n
->
port
);
listAddNodeTail
(
importing
,
n
);
break
;
...
...
@@ -3712,7 +3712,7 @@ static int clusterManagerFixOpenSlot(int slot) {
if
(
owner
==
NULL
)
{
clusterManagerLogInfo
(
">>> Nobody claims ownership, "
"selecting an owner...
\n
"
);
owner
=
clusterManagerGetNodeWithMostKeysInSlot
(
cluster_manager
.
nodes
,
owner
=
clusterManagerGetNodeWithMostKeysInSlot
(
cluster_manager
.
nodes
,
slot
,
NULL
);
// If we still don't have an owner, we can't fix it.
if
(
owner
==
NULL
)
{
...
...
@@ -3723,10 +3723,10 @@ static int clusterManagerFixOpenSlot(int slot) {
}
// Use ADDSLOTS to assign the slot.
clusterManagerLogWarn
(
"*** Configuring %s:%d as the slot owner
\n
"
,
clusterManagerLogWarn
(
"*** Configuring %s:%d as the slot owner
\n
"
,
owner
->
ip
,
owner
->
port
);
redisReply
*
reply
=
CLUSTER_MANAGER_COMMAND
(
owner
,
"CLUSTER "
"SETSLOT %d %s"
,
"SETSLOT %d %s"
,
slot
,
"STABLE"
);
success
=
clusterManagerCheckRedisReply
(
owner
,
reply
,
NULL
);
if
(
reply
)
freeReplyObject
(
reply
);
...
...
@@ -3735,7 +3735,7 @@ static int clusterManagerFixOpenSlot(int slot) {
success
=
clusterManagerCheckRedisReply
(
owner
,
reply
,
NULL
);
if
(
reply
)
freeReplyObject
(
reply
);
if
(
!
success
)
goto
cleanup
;
/* Since CLUSTER ADDSLOTS succeeded, we also update the slot
/* Since CLUSTER ADDSLOTS succeeded, we also update the slot
* info into the node struct, in order to keep it synced */
owner
->
slots
[
slot
]
=
1
;
/* Make sure this information will propagate. Not strictly needed
...
...
@@ -3769,7 +3769,7 @@ static int clusterManagerFixOpenSlot(int slot) {
success
=
clusterManagerCheckRedisReply
(
n
,
reply
,
NULL
);
if
(
reply
)
freeReplyObject
(
reply
);
if
(
!
success
)
goto
cleanup
;
success
=
clusterManagerSetSlot
(
n
,
owner
,
slot
,
"importing"
,
NULL
);
success
=
clusterManagerSetSlot
(
n
,
owner
,
slot
,
"importing"
,
NULL
);
if
(
!
success
)
goto
cleanup
;
clusterManagerRemoveNodeFromList
(
importing
,
n
);
//Avoid duplicates
listAddNodeTail
(
importing
,
n
);
...
...
@@ -3786,7 +3786,7 @@ static int clusterManagerFixOpenSlot(int slot) {
clusterManagerNode
*
src
=
listFirst
(
migrating
)
->
value
;
clusterManagerNode
*
dst
=
listFirst
(
importing
)
->
value
;
success
=
clusterManagerMoveSlot
(
src
,
dst
,
slot
,
move_opts
,
NULL
);
}
}
/* Case 2: There are multiple nodes that claim the slot as importing,
* they probably got keys about the slot after a restart so opened
* the slot. In this case we just move all the keys to the owner
...
...
@@ -3804,18 +3804,18 @@ static int clusterManagerFixOpenSlot(int slot) {
clusterManagerLogInfo
(
">>> Setting %d as STABLE in "
"%s:%d
\n
"
,
slot
,
n
->
ip
,
n
->
port
);
redisReply
*
r
=
CLUSTER_MANAGER_COMMAND
(
n
,
"CLUSTER SETSLOT %d %s"
,
redisReply
*
r
=
CLUSTER_MANAGER_COMMAND
(
n
,
"CLUSTER SETSLOT %d %s"
,
slot
,
"STABLE"
);
success
=
clusterManagerCheckRedisReply
(
n
,
r
,
NULL
);
if
(
r
)
freeReplyObject
(
r
);
if
(
!
success
)
goto
cleanup
;
}
}
else
{
int
try_to_close_slot
=
(
listLength
(
importing
)
==
0
&&
listLength
(
migrating
)
==
1
);
int
try_to_close_slot
=
(
listLength
(
importing
)
==
0
&&
listLength
(
migrating
)
==
1
);
if
(
try_to_close_slot
)
{
clusterManagerNode
*
n
=
listFirst
(
migrating
)
->
value
;
redisReply
*
r
=
CLUSTER_MANAGER_COMMAND
(
n
,
redisReply
*
r
=
CLUSTER_MANAGER_COMMAND
(
n
,
"CLUSTER GETKEYSINSLOT %d %d"
,
slot
,
10
);
success
=
clusterManagerCheckRedisReply
(
n
,
r
,
NULL
);
if
(
r
)
{
...
...
@@ -3829,7 +3829,7 @@ static int clusterManagerFixOpenSlot(int slot) {
* can just close the slot, probably a reshard interrupted in the middle. */
if
(
try_to_close_slot
)
{
clusterManagerNode
*
n
=
listFirst
(
migrating
)
->
value
;
redisReply
*
r
=
CLUSTER_MANAGER_COMMAND
(
n
,
"CLUSTER SETSLOT %d %s"
,
redisReply
*
r
=
CLUSTER_MANAGER_COMMAND
(
n
,
"CLUSTER SETSLOT %d %s"
,
slot
,
"STABLE"
);
success
=
clusterManagerCheckRedisReply
(
n
,
r
,
NULL
);
if
(
r
)
freeReplyObject
(
r
);
...
...
@@ -3839,7 +3839,7 @@ static int clusterManagerFixOpenSlot(int slot) {
clusterManagerLogErr
(
"[ERR] Sorry, redis-cli can't fix this slot "
"yet (work in progress). Slot is set as "
"migrating in %s, as importing in %s, "
"owner is %s:%d
\n
"
,
migrating_str
,
"owner is %s:%d
\n
"
,
migrating_str
,
importing_str
,
owner
->
ip
,
owner
->
port
);
}
}
...
...
@@ -3856,10 +3856,10 @@ static int clusterManagerCheckCluster(int quiet) {
listNode
*
ln
=
listFirst
(
cluster_manager
.
nodes
);
if
(
!
ln
)
return
0
;
int
result
=
1
;
int
do_fix
=
config
.
cluster_manager_command
.
flags
&
int
do_fix
=
config
.
cluster_manager_command
.
flags
&
CLUSTER_MANAGER_CMD_FLAG_FIX
;
clusterManagerNode
*
node
=
ln
->
value
;
clusterManagerLogInfo
(
">>> Performing Cluster Check (using node %s:%d)
\n
"
,
clusterManagerLogInfo
(
">>> Performing Cluster Check (using node %s:%d)
\n
"
,
node
->
ip
,
node
->
port
);
if
(
!
quiet
)
clusterManagerShowNodes
();
if
(
!
clusterManagerIsConfigConsistent
())
{
...
...
@@ -3877,12 +3877,12 @@ static int clusterManagerCheckCluster(int quiet) {
int
i
;
dict
*
open_slots
=
NULL
;
while
((
ln
=
listNext
(
&
li
))
!=
NULL
)
{
clusterManagerNode
*
n
=
ln
->
value
;
clusterManagerNode
*
n
=
ln
->
value
;
if
(
n
->
migrating
!=
NULL
)
{
if
(
open_slots
==
NULL
)
if
(
open_slots
==
NULL
)
open_slots
=
dictCreate
(
&
clusterManagerDictType
,
NULL
);
sds
errstr
=
sdsempty
();
errstr
=
sdscatprintf
(
errstr
,
errstr
=
sdscatprintf
(
errstr
,
"[WARNING] Node %s:%d has slots in "
"migrating state "
,
n
->
ip
,
...
...
@@ -3892,15 +3892,15 @@ static int clusterManagerCheckCluster(int quiet) {
dictAdd
(
open_slots
,
slot
,
sdsdup
(
n
->
migrating
[
i
+
1
]));
char
*
fmt
=
(
i
>
0
?
",%S"
:
"%S"
);
errstr
=
sdscatfmt
(
errstr
,
fmt
,
slot
);
}
}
errstr
=
sdscat
(
errstr
,
"."
);
clusterManagerOnError
(
errstr
);
}
if
(
n
->
importing
!=
NULL
)
{
if
(
open_slots
==
NULL
)
if
(
open_slots
==
NULL
)
open_slots
=
dictCreate
(
&
clusterManagerDictType
,
NULL
);
sds
errstr
=
sdsempty
();
errstr
=
sdscatprintf
(
errstr
,
errstr
=
sdscatprintf
(
errstr
,
"[WARNING] Node %s:%d has slots in "
"importing state "
,
n
->
ip
,
...
...
@@ -3910,7 +3910,7 @@ static int clusterManagerCheckCluster(int quiet) {
dictAdd
(
open_slots
,
slot
,
sdsdup
(
n
->
importing
[
i
+
1
]));
char
*
fmt
=
(
i
>
0
?
",%S"
:
"%S"
);
errstr
=
sdscatfmt
(
errstr
,
fmt
,
slot
);
}
}
errstr
=
sdscat
(
errstr
,
"."
);
clusterManagerOnError
(
errstr
);
}
...
...
@@ -3946,12 +3946,12 @@ static int clusterManagerCheckCluster(int quiet) {
memset
(
slots
,
0
,
CLUSTER_MANAGER_SLOTS
);
int
coverage
=
clusterManagerGetCoveredSlots
(
slots
);
if
(
coverage
==
CLUSTER_MANAGER_SLOTS
)
{
clusterManagerLogOk
(
"[OK] All %d slots covered.
\n
"
,
clusterManagerLogOk
(
"[OK] All %d slots covered.
\n
"
,
CLUSTER_MANAGER_SLOTS
);
}
else
{
sds
err
=
sdsempty
();
err
=
sdscatprintf
(
err
,
"[ERR] Not all %d slots are "
"covered by nodes.
\n
"
,
"covered by nodes.
\n
"
,
CLUSTER_MANAGER_SLOTS
);
clusterManagerOnError
(
err
);
result
=
0
;
...
...
@@ -3966,11 +3966,11 @@ static int clusterManagerCheckCluster(int quiet) {
return
result
;
}
static
clusterManagerNode
*
clusterNodeForResharding
(
char
*
id
,
static
clusterManagerNode
*
clusterNodeForResharding
(
char
*
id
,
clusterManagerNode
*
target
,
int
*
raise_err
)
int
*
raise_err
)
{
clusterManagerNode
*
node
=
NULL
;
clusterManagerNode
*
node
=
NULL
;
const
char
*
invalid_node_msg
=
"*** The specified node (%s) is not known "
"or not a master, please retry.
\n
"
;
node
=
clusterManagerNodeByName
(
id
);
...
...
@@ -3985,7 +3985,7 @@ static clusterManagerNode *clusterNodeForResharding(char *id,
"the target node as "
"source node.
\n
"
);
return
NULL
;
}
}
}
return
node
;
}
...
...
@@ -4002,10 +4002,10 @@ static list *clusterManagerComputeReshardTable(list *sources, int numslots) {
tot_slots
+=
node
->
slots_count
;
sorted
[
i
++
]
=
node
;
}
qsort
(
sorted
,
src_count
,
sizeof
(
clusterManagerNode
*
),
qsort
(
sorted
,
src_count
,
sizeof
(
clusterManagerNode
*
),
clusterManagerSlotCountCompareDesc
);
for
(
i
=
0
;
i
<
src_count
;
i
++
)
{
clusterManagerNode
*
node
=
sorted
[
i
];
clusterManagerNode
*
node
=
sorted
[
i
];
float
n
=
((
float
)
numslots
/
tot_slots
*
node
->
slots_count
);
if
(
i
==
0
)
n
=
ceil
(
n
);
else
n
=
floor
(
n
);
...
...
@@ -4050,10 +4050,10 @@ static void clusterManagerReleaseReshardTable(list *table) {
}
static
void
clusterManagerLog
(
int
level
,
const
char
*
fmt
,
...)
{
int
use_colors
=
int
use_colors
=
(
config
.
cluster_manager_command
.
flags
&
CLUSTER_MANAGER_CMD_FLAG_COLOR
);
if
(
use_colors
)
{
printf
(
"
\033
["
);
printf
(
"
\033
["
);
switch
(
level
)
{
case
CLUSTER_MANAGER_LOG_LVL_INFO
:
printf
(
LOG_COLOR_BOLD
);
break
;
case
CLUSTER_MANAGER_LOG_LVL_WARN
:
printf
(
LOG_COLOR_YELLOW
);
break
;
...
...
@@ -4069,8 +4069,8 @@ static void clusterManagerLog(int level, const char* fmt, ...) {
if
(
use_colors
)
printf
(
"
\033
["
LOG_COLOR_RESET
);
}
static
void
clusterManagerNodeArrayInit
(
clusterManagerNodeArray
*
array
,
int
alloc_len
)
static
void
clusterManagerNodeArrayInit
(
clusterManagerNodeArray
*
array
,
int
alloc_len
)
{
array
->
nodes
=
zcalloc
(
alloc_len
*
sizeof
(
clusterManagerNode
*
));
array
->
alloc
=
array
->
nodes
;
...
...
@@ -4093,8 +4093,8 @@ static void clusterManagerNodeArrayReset(clusterManagerNodeArray *array) {
}
/* Shift array->nodes and store the shifted node into 'nodeptr'. */
static
void
clusterManagerNodeArrayShift
(
clusterManagerNodeArray
*
array
,
clusterManagerNode
**
nodeptr
)
static
void
clusterManagerNodeArrayShift
(
clusterManagerNodeArray
*
array
,
clusterManagerNode
**
nodeptr
)
{
assert
(
array
->
nodes
<
(
array
->
nodes
+
array
->
len
));
/* If the first node to be shifted is not NULL, decrement count. */
...
...
@@ -4107,7 +4107,7 @@ static void clusterManagerNodeArrayShift(clusterManagerNodeArray *array,
}
static
void
clusterManagerNodeArrayAdd
(
clusterManagerNodeArray
*
array
,
clusterManagerNode
*
node
)
clusterManagerNode
*
node
)
{
assert
(
array
->
nodes
<
(
array
->
nodes
+
array
->
len
));
assert
(
node
!=
NULL
);
...
...
@@ -4115,10 +4115,10 @@ static void clusterManagerNodeArrayAdd(clusterManagerNodeArray *array,
array
->
nodes
[
array
->
count
++
]
=
node
;
}
static
void
clusterManagerPrintNotEmptyNodeError
(
clusterManagerNode
*
node
,
char
*
err
)
static
void
clusterManagerPrintNotEmptyNodeError
(
clusterManagerNode
*
node
,
char
*
err
)
{
char
*
msg
;
char
*
msg
;
if
(
err
)
msg
=
err
;
else
{
msg
=
"is not empty. Either the node already knows other "
...
...
@@ -4128,8 +4128,8 @@ static void clusterManagerPrintNotEmptyNodeError(clusterManagerNode *node,
clusterManagerLogErr
(
"[ERR] Node %s:%d %s
\n
"
,
node
->
ip
,
node
->
port
,
msg
);
}
static
void
clusterManagerPrintNotClusterNodeError
(
clusterManagerNode
*
node
,
char
*
err
)
static
void
clusterManagerPrintNotClusterNodeError
(
clusterManagerNode
*
node
,
char
*
err
)
{
char
*
msg
=
(
err
?
err
:
"is not configured as a cluster node."
);
clusterManagerLogErr
(
"[ERR] Node %s:%d %s
\n
"
,
node
->
ip
,
node
->
port
,
msg
);
...
...
@@ -4206,17 +4206,17 @@ static int clusterManagerCommandCreate(int argc, char **argv) {
if
(
masters_count
<
3
)
{
clusterManagerLogErr
(
"*** ERROR: Invalid configuration for cluster creation.
\n
"
"*** Redis Cluster requires at least 3 master nodes.
\n
"
"*** This is not possible with %d nodes and %d replicas per node."
,
"*** Redis Cluster requires at least 3 master nodes.
\n
"
"*** This is not possible with %d nodes and %d replicas per node."
,
node_len
,
replicas
);
clusterManagerLogErr
(
"
\n
*** At least %d nodes are required.
\n
"
,
clusterManagerLogErr
(
"
\n
*** At least %d nodes are required.
\n
"
,
3
*
(
replicas
+
1
));
return
0
;
}
clusterManagerLogInfo
(
">>> Performing hash slots allocation "
"on %d nodes...
\n
"
,
node_len
);
int
interleaved_len
=
0
,
ip_count
=
0
;
clusterManagerNode
**
interleaved
=
zcalloc
(
node_len
*
sizeof
(
**
interleaved
));
clusterManagerNode
**
interleaved
=
zcalloc
(
node_len
*
sizeof
(
**
interleaved
));
char
**
ips
=
zcalloc
(
node_len
*
sizeof
(
char
*
));
clusterManagerNodeArray
*
ip_nodes
=
zcalloc
(
node_len
*
sizeof
(
*
ip_nodes
));
listIter
li
;
...
...
@@ -4248,7 +4248,7 @@ static int clusterManagerCommandCreate(int argc, char **argv) {
clusterManagerNodeArrayShift
(
node_array
,
&
n
);
interleaved
[
interleaved_len
++
]
=
n
;
}
}
}
}
clusterManagerNode
**
masters
=
interleaved
;
interleaved
+=
masters_count
;
...
...
@@ -4258,7 +4258,7 @@ static int clusterManagerCommandCreate(int argc, char **argv) {
float
cursor
=
0
.
0
f
;
for
(
i
=
0
;
i
<
masters_count
;
i
++
)
{
clusterManagerNode
*
master
=
masters
[
i
];
long
last
=
lround
(
cursor
+
slots_per_node
-
1
);
long
last
=
lround
(
cursor
+
slots_per_node
-
1
);
if
(
last
>
CLUSTER_MANAGER_SLOTS
||
i
==
(
masters_count
-
1
))
last
=
CLUSTER_MANAGER_SLOTS
-
1
;
if
(
last
<
first
)
last
=
first
;
...
...
@@ -4279,7 +4279,7 @@ assign_replicas:
clusterManagerNode
*
master
=
masters
[
i
];
int
assigned_replicas
=
0
;
while
(
assigned_replicas
<
replicas
)
{
if
(
available_count
==
0
)
break
;
if
(
available_count
==
0
)
break
;
clusterManagerNode
*
found
=
NULL
,
*
slave
=
NULL
;
int
firstNodeIdx
=
-
1
;
for
(
j
=
0
;
j
<
interleaved_len
;
j
++
)
{
...
...
@@ -4294,7 +4294,7 @@ assign_replicas:
}
if
(
found
)
slave
=
found
;
else
if
(
firstNodeIdx
>=
0
)
{
slave
=
interleaved
[
firstNodeIdx
];
slave
=
interleaved
[
firstNodeIdx
];
interleaved_len
-=
(
interleaved
-
(
interleaved
+
firstNodeIdx
));
interleaved
+=
(
firstNodeIdx
+
1
);
}
...
...
@@ -4302,10 +4302,10 @@ assign_replicas:
assigned_replicas
++
;
available_count
--
;
if
(
slave
->
replicate
)
sdsfree
(
slave
->
replicate
);
slave
->
replicate
=
sdsnew
(
master
->
name
);
slave
->
replicate
=
sdsnew
(
master
->
name
);
slave
->
dirty
=
1
;
}
else
break
;
printf
(
"Adding replica %s:%d to %s:%d
\n
"
,
slave
->
ip
,
slave
->
port
,
printf
(
"Adding replica %s:%d to %s:%d
\n
"
,
slave
->
ip
,
slave
->
port
,
master
->
ip
,
master
->
port
);
if
(
assign_unused
)
break
;
}
...
...
@@ -4344,8 +4344,8 @@ assign_replicas:
while
((
ln
=
listNext
(
&
li
))
!=
NULL
)
{
clusterManagerNode
*
node
=
ln
->
value
;
redisReply
*
reply
=
NULL
;
reply
=
CLUSTER_MANAGER_COMMAND
(
node
,
"cluster set-config-epoch %d"
,
reply
=
CLUSTER_MANAGER_COMMAND
(
node
,
"cluster set-config-epoch %d"
,
config_epoch
++
);
if
(
reply
!=
NULL
)
freeReplyObject
(
reply
);
}
...
...
@@ -4360,7 +4360,7 @@ assign_replicas:
continue
;
}
redisReply
*
reply
=
NULL
;
reply
=
CLUSTER_MANAGER_COMMAND
(
node
,
"cluster meet %s %d"
,
reply
=
CLUSTER_MANAGER_COMMAND
(
node
,
"cluster meet %s %d"
,
first
->
ip
,
first
->
port
);
int
is_err
=
0
;
if
(
reply
!=
NULL
)
{
...
...
@@ -4377,7 +4377,7 @@ assign_replicas:
}
}
/* Give one second for the join to start, in order to avoid that
* waiting for cluster join will find all the nodes agree about
* waiting for cluster join will find all the nodes agree about
* the config as they are still empty with unassigned slots. */
sleep
(
1
);
clusterManagerWaitForClusterJoin
();
...
...
@@ -4408,7 +4408,7 @@ assign_replicas:
listEmpty
(
cluster_manager
.
nodes
);
if
(
!
clusterManagerLoadInfoFromNode
(
first_node
,
0
))
{
success
=
0
;
goto
cleanup
;
goto
cleanup
;
}
clusterManagerCheckCluster
(
0
);
}
...
...
@@ -4429,11 +4429,11 @@ static int clusterManagerCommandAddNode(int argc, char **argv) {
redisReply
*
reply
=
NULL
;
char
*
ref_ip
=
NULL
,
*
ip
=
NULL
;
int
ref_port
=
0
,
port
=
0
;
if
(
!
getClusterHostFromCmdArgs
(
argc
-
1
,
argv
+
1
,
&
ref_ip
,
&
ref_port
))
if
(
!
getClusterHostFromCmdArgs
(
argc
-
1
,
argv
+
1
,
&
ref_ip
,
&
ref_port
))
goto
invalid_args
;
if
(
!
getClusterHostFromCmdArgs
(
1
,
argv
,
&
ip
,
&
port
))
goto
invalid_args
;
clusterManagerLogInfo
(
">>> Adding node %s:%d to cluster %s:%d
\n
"
,
ip
,
port
,
clusterManagerLogInfo
(
">>> Adding node %s:%d to cluster %s:%d
\n
"
,
ip
,
port
,
ref_ip
,
ref_port
);
// Check the existing cluster
clusterManagerNode
*
refnode
=
clusterManagerNewNode
(
ref_ip
,
ref_port
);
...
...
@@ -4454,17 +4454,17 @@ static int clusterManagerCommandAddNode(int argc, char **argv) {
}
else
{
master_node
=
clusterManagerNodeWithLeastReplicas
();
assert
(
master_node
!=
NULL
);
printf
(
"Automatically selected master %s:%d
\n
"
,
master_node
->
ip
,
printf
(
"Automatically selected master %s:%d
\n
"
,
master_node
->
ip
,
master_node
->
port
);
}
}
}
}
// Add the new node
clusterManagerNode
*
new_node
=
clusterManagerNewNode
(
ip
,
port
);
int
added
=
0
;
CLUSTER_MANAGER_NODE_CONNECT
(
new_node
);
if
(
new_node
->
context
->
err
)
{
clusterManagerLogErr
(
"[ERR] Sorry, can't connect to node %s:%d
\n
"
,
clusterManagerLogErr
(
"[ERR] Sorry, can't connect to node %s:%d
\n
"
,
ip
,
port
);
success
=
0
;
goto
cleanup
;
...
...
@@ -4495,23 +4495,23 @@ static int clusterManagerCommandAddNode(int argc, char **argv) {
// Send CLUSTER MEET command to the new node
clusterManagerLogInfo
(
">>> Send CLUSTER MEET to node %s:%d to make it "
"join the cluster.
\n
"
,
ip
,
port
);
reply
=
CLUSTER_MANAGER_COMMAND
(
new_node
,
"CLUSTER MEET %s %d"
,
reply
=
CLUSTER_MANAGER_COMMAND
(
new_node
,
"CLUSTER MEET %s %d"
,
first
->
ip
,
first
->
port
);
if
(
!
(
success
=
clusterManagerCheckRedisReply
(
new_node
,
reply
,
NULL
)))
goto
cleanup
;
/* Additional configuration is needed if the node is added as a slave. */
if
(
master_node
)
{
sleep
(
1
);
clusterManagerWaitForClusterJoin
();
clusterManagerLogInfo
(
">>> Configure node as replica of %s:%d.
\n
"
,
clusterManagerLogInfo
(
">>> Configure node as replica of %s:%d.
\n
"
,
master_node
->
ip
,
master_node
->
port
);
freeReplyObject
(
reply
);
reply
=
CLUSTER_MANAGER_COMMAND
(
new_node
,
"CLUSTER REPLICATE %s"
,
reply
=
CLUSTER_MANAGER_COMMAND
(
new_node
,
"CLUSTER REPLICATE %s"
,
master_node
->
name
);
if
(
!
(
success
=
clusterManagerCheckRedisReply
(
new_node
,
reply
,
NULL
)))
goto
cleanup
;
}
}
clusterManagerLogOk
(
"[OK] New node added correctly.
\n
"
);
cleanup:
if
(
!
added
&&
new_node
)
freeClusterManagerNode
(
new_node
);
...
...
@@ -4529,14 +4529,14 @@ static int clusterManagerCommandDeleteNode(int argc, char **argv) {
char
*
ip
=
NULL
;
if
(
!
getClusterHostFromCmdArgs
(
1
,
argv
,
&
ip
,
&
port
))
goto
invalid_args
;
char
*
node_id
=
argv
[
1
];
clusterManagerLogInfo
(
">>> Removing node %s from cluster %s:%d
\n
"
,
clusterManagerLogInfo
(
">>> Removing node %s from cluster %s:%d
\n
"
,
node_id
,
ip
,
port
);
clusterManagerNode
*
ref_node
=
clusterManagerNewNode
(
ip
,
port
);
clusterManagerNode
*
node
=
NULL
;
// Load cluster information
if
(
!
clusterManagerLoadInfoFromNode
(
ref_node
,
0
))
return
0
;
// Check if the node exists and is not empty
node
=
clusterManagerNodeByName
(
node_id
);
if
(
node
==
NULL
)
{
...
...
@@ -4548,7 +4548,7 @@ static int clusterManagerCommandDeleteNode(int argc, char **argv) {
"away and try again.
\n
"
,
node
->
ip
,
node
->
port
);
return
0
;
}
// Send CLUSTER FORGET to all the nodes but the node to remove
clusterManagerLogInfo
(
">>> Sending CLUSTER FORGET messages to the "
"cluster...
\n
"
);
...
...
@@ -4562,16 +4562,16 @@ static int clusterManagerCommandDeleteNode(int argc, char **argv) {
// Reconfigure the slave to replicate with some other node
clusterManagerNode
*
master
=
clusterManagerNodeWithLeastReplicas
();
assert
(
master
!=
NULL
);
clusterManagerLogInfo
(
">>> %s:%d as replica of %s:%d
\n
"
,
clusterManagerLogInfo
(
">>> %s:%d as replica of %s:%d
\n
"
,
n
->
ip
,
n
->
port
,
master
->
ip
,
master
->
port
);
redisReply
*
r
=
CLUSTER_MANAGER_COMMAND
(
n
,
"CLUSTER REPLICATE %s"
,
redisReply
*
r
=
CLUSTER_MANAGER_COMMAND
(
n
,
"CLUSTER REPLICATE %s"
,
master
->
name
);
success
=
clusterManagerCheckRedisReply
(
n
,
r
,
NULL
);
if
(
r
)
freeReplyObject
(
r
);
if
(
!
success
)
return
0
;
}
redisReply
*
r
=
CLUSTER_MANAGER_COMMAND
(
n
,
"CLUSTER FORGET %s"
,
node_id
);
redisReply
*
r
=
CLUSTER_MANAGER_COMMAND
(
n
,
"CLUSTER FORGET %s"
,
node_id
);
success
=
clusterManagerCheckRedisReply
(
n
,
r
,
NULL
);
if
(
r
)
freeReplyObject
(
r
);
if
(
!
success
)
return
0
;
...
...
@@ -4628,14 +4628,14 @@ static int clusterManagerCommandReshard(int argc, char **argv) {
clusterManagerCheckCluster
(
0
);
if
(
cluster_manager
.
errors
&&
listLength
(
cluster_manager
.
errors
)
>
0
)
{
fflush
(
stdout
);
fprintf
(
stderr
,
fprintf
(
stderr
,
"*** Please fix your cluster problems before resharding
\n
"
);
return
0
;
}
int
slots
=
config
.
cluster_manager_command
.
slots
;
if
(
!
slots
)
{
while
(
slots
<=
0
||
slots
>
CLUSTER_MANAGER_SLOTS
)
{
printf
(
"How many slots do you want to move (from 1 to %d)? "
,
printf
(
"How many slots do you want to move (from 1 to %d)? "
,
CLUSTER_MANAGER_SLOTS
);
fflush
(
stdout
);
char
buf
[
6
];
...
...
@@ -4643,7 +4643,7 @@ static int clusterManagerCommandReshard(int argc, char **argv) {
if
(
nread
<=
0
)
continue
;
int
last_idx
=
nread
-
1
;
if
(
buf
[
last_idx
]
!=
'\n'
)
{
int
ch
;
int
ch
;
while
((
ch
=
getchar
())
!=
'\n'
&&
ch
!=
EOF
)
{}
}
buf
[
last_idx
]
=
'\0'
;
...
...
@@ -4651,16 +4651,16 @@ static int clusterManagerCommandReshard(int argc, char **argv) {
}
}
char
buf
[
255
];
char
*
to
=
config
.
cluster_manager_command
.
to
,
char
*
to
=
config
.
cluster_manager_command
.
to
,
*
from
=
config
.
cluster_manager_command
.
from
;
while
(
to
==
NULL
)
{
printf
(
"What is the receiving node ID? "
);
printf
(
"What is the receiving node ID? "
);
fflush
(
stdout
);
int
nread
=
read
(
fileno
(
stdin
),
buf
,
255
);
if
(
nread
<=
0
)
continue
;
int
last_idx
=
nread
-
1
;
if
(
buf
[
last_idx
]
!=
'\n'
)
{
int
ch
;
int
ch
;
while
((
ch
=
getchar
())
!=
'\n'
&&
ch
!=
EOF
)
{}
}
buf
[
last_idx
]
=
'\0'
;
...
...
@@ -4678,13 +4678,13 @@ static int clusterManagerCommandReshard(int argc, char **argv) {
"the hash slots.
\n
"
);
printf
(
" Type 'done' once you entered all the source nodes IDs.
\n
"
);
while
(
1
)
{
printf
(
"Source node #%lu: "
,
listLength
(
sources
)
+
1
);
printf
(
"Source node #%lu: "
,
listLength
(
sources
)
+
1
);
fflush
(
stdout
);
int
nread
=
read
(
fileno
(
stdin
),
buf
,
255
);
if
(
nread
<=
0
)
continue
;
int
last_idx
=
nread
-
1
;
if
(
buf
[
last_idx
]
!=
'\n'
)
{
int
ch
;
int
ch
;
while
((
ch
=
getchar
())
!=
'\n'
&&
ch
!=
EOF
)
{}
}
buf
[
last_idx
]
=
'\0'
;
...
...
@@ -4693,7 +4693,7 @@ static int clusterManagerCommandReshard(int argc, char **argv) {
all
=
1
;
break
;
}
else
{
clusterManagerNode
*
src
=
clusterManagerNode
*
src
=
clusterNodeForResharding
(
buf
,
target
,
&
raise_err
);
if
(
src
!=
NULL
)
listAddNodeTail
(
sources
,
src
);
else
if
(
raise_err
)
{
...
...
@@ -4703,14 +4703,14 @@ static int clusterManagerCommandReshard(int argc, char **argv) {
}
}
}
else
{
char
*
p
;
char
*
p
;
while
((
p
=
strchr
(
from
,
','
))
!=
NULL
)
{
*
p
=
'\0'
;
if
(
!
strcmp
(
from
,
"all"
))
{
all
=
1
;
break
;
}
else
{
clusterManagerNode
*
src
=
clusterManagerNode
*
src
=
clusterNodeForResharding
(
from
,
target
,
&
raise_err
);
if
(
src
!=
NULL
)
listAddNodeTail
(
sources
,
src
);
else
if
(
raise_err
)
{
...
...
@@ -4724,7 +4724,7 @@ static int clusterManagerCommandReshard(int argc, char **argv) {
if
(
!
all
&&
strlen
(
from
)
>
0
)
{
if
(
!
strcmp
(
from
,
"all"
))
all
=
1
;
if
(
!
all
)
{
clusterManagerNode
*
src
=
clusterManagerNode
*
src
=
clusterNodeForResharding
(
from
,
target
,
&
raise_err
);
if
(
src
!=
NULL
)
listAddNodeTail
(
sources
,
src
);
else
if
(
raise_err
)
{
...
...
@@ -4737,7 +4737,7 @@ static int clusterManagerCommandReshard(int argc, char **argv) {
listIter
li
;
listNode
*
ln
;
if
(
all
)
{
listEmpty
(
sources
);
listEmpty
(
sources
);
listRewind
(
cluster_manager
.
nodes
,
&
li
);
while
((
ln
=
listNext
(
&
li
))
!=
NULL
)
{
clusterManagerNode
*
n
=
ln
->
value
;
...
...
@@ -4768,8 +4768,8 @@ static int clusterManagerCommandReshard(int argc, char **argv) {
table
=
clusterManagerComputeReshardTable
(
sources
,
slots
);
printf
(
" Resharding plan:
\n
"
);
clusterManagerShowReshardTable
(
table
);
if
(
!
(
config
.
cluster_manager_command
.
flags
&
CLUSTER_MANAGER_CMD_FLAG_YES
))
if
(
!
(
config
.
cluster_manager_command
.
flags
&
CLUSTER_MANAGER_CMD_FLAG_YES
))
{
printf
(
"Do you want to proceed with the proposed "
"reshard plan (yes/no)? "
);
...
...
@@ -4787,7 +4787,7 @@ static int clusterManagerCommandReshard(int argc, char **argv) {
while
((
ln
=
listNext
(
&
li
))
!=
NULL
)
{
clusterManagerReshardTableItem
*
item
=
ln
->
value
;
char
*
err
=
NULL
;
result
=
clusterManagerMoveSlot
(
item
->
source
,
target
,
item
->
slot
,
result
=
clusterManagerMoveSlot
(
item
->
source
,
target
,
item
->
slot
,
opts
,
&
err
);
if
(
!
result
)
{
if
(
err
!=
NULL
)
{
...
...
@@ -4827,18 +4827,18 @@ static int clusterManagerCommandRebalance(int argc, char **argv) {
float
w
=
atof
(
++
p
);
clusterManagerNode
*
n
=
clusterManagerNodeByAbbreviatedName
(
name
);
if
(
n
==
NULL
)
{
clusterManagerLogErr
(
"*** No such master node %s
\n
"
,
name
);
clusterManagerLogErr
(
"*** No such master node %s
\n
"
,
name
);
result
=
0
;
goto
cleanup
;
}
n
->
weight
=
w
;
}
}
}
float
total_weight
=
0
;
float
total_weight
=
0
;
int
nodes_involved
=
0
;
int
use_empty
=
config
.
cluster_manager_command
.
flags
&
int
use_empty
=
config
.
cluster_manager_command
.
flags
&
CLUSTER_MANAGER_CMD_FLAG_EMPTYMASTER
;
involved
=
listCreate
();
involved
=
listCreate
();
listIter
li
;
listNode
*
ln
;
listRewind
(
cluster_manager
.
nodes
,
&
li
);
...
...
@@ -4851,7 +4851,7 @@ static int clusterManagerCommandRebalance(int argc, char **argv) {
n
->
weight
=
0
;
continue
;
}
total_weight
+=
n
->
weight
;
total_weight
+=
n
->
weight
;
nodes_involved
++
;
listAddNodeTail
(
involved
,
n
);
}
...
...
@@ -4913,10 +4913,10 @@ static int clusterManagerCommandRebalance(int argc, char **argv) {
}
}
/* Sort nodes by their slots balance. */
qsort
(
weightedNodes
,
nodes_involved
,
sizeof
(
clusterManagerNode
*
),
qsort
(
weightedNodes
,
nodes_involved
,
sizeof
(
clusterManagerNode
*
),
clusterManagerCompareNodeBalance
);
clusterManagerLogInfo
(
">>> Rebalancing across %d nodes. "
"Total weight = %.2f
\n
"
,
"Total weight = %.2f
\n
"
,
nodes_involved
,
total_weight
);
if
(
config
.
verbose
)
{
for
(
i
=
0
;
i
<
nodes_involved
;
i
++
)
{
...
...
@@ -4931,7 +4931,7 @@ static int clusterManagerCommandRebalance(int argc, char **argv) {
* find nodes that need to get/provide slots. */
int
dst_idx
=
0
;
int
src_idx
=
nodes_involved
-
1
;
int
simulate
=
config
.
cluster_manager_command
.
flags
&
int
simulate
=
config
.
cluster_manager_command
.
flags
&
CLUSTER_MANAGER_CMD_FLAG_SIMULATE
;
while
(
dst_idx
<
src_idx
)
{
clusterManagerNode
*
dst
=
weightedNodes
[
dst_idx
];
...
...
@@ -4940,14 +4940,14 @@ static int clusterManagerCommandRebalance(int argc, char **argv) {
int
sb
=
abs
(
src
->
balance
);
int
numslots
=
(
db
<
sb
?
db
:
sb
);
if
(
numslots
>
0
)
{
printf
(
"Moving %d slots from %s:%d to %s:%d
\n
"
,
numslots
,
src
->
ip
,
src
->
port
,
dst
->
ip
,
printf
(
"Moving %d slots from %s:%d to %s:%d
\n
"
,
numslots
,
src
->
ip
,
src
->
port
,
dst
->
ip
,
dst
->
port
);
/* Actually move the slots. */
list
*
lsrc
=
listCreate
(),
*
table
=
NULL
;
listAddNodeTail
(
lsrc
,
src
);
listAddNodeTail
(
lsrc
,
src
);
table
=
clusterManagerComputeReshardTable
(
lsrc
,
numslots
);
listRelease
(
lsrc
);
int
table_len
=
(
int
)
listLength
(
table
);
...
...
@@ -4960,14 +4960,14 @@ static int clusterManagerCommandRebalance(int argc, char **argv) {
if
(
simulate
)
{
for
(
i
=
0
;
i
<
table_len
;
i
++
)
printf
(
"#"
);
}
else
{
int
opts
=
CLUSTER_MANAGER_OPT_QUIET
|
int
opts
=
CLUSTER_MANAGER_OPT_QUIET
|
CLUSTER_MANAGER_OPT_UPDATE
;
listRewind
(
table
,
&
li
);
while
((
ln
=
listNext
(
&
li
))
!=
NULL
)
{
clusterManagerReshardTableItem
*
item
=
ln
->
value
;
result
=
clusterManagerMoveSlot
(
item
->
source
,
dst
,
item
->
slot
,
clusterManagerReshardTableItem
*
item
=
ln
->
value
;
result
=
clusterManagerMoveSlot
(
item
->
source
,
dst
,
item
->
slot
,
opts
,
NULL
);
if
(
!
result
)
goto
end_move
;
printf
(
"#"
);
...
...
@@ -5019,8 +5019,8 @@ static int clusterManagerCommandSetTimeout(int argc, char **argv) {
while
((
ln
=
listNext
(
&
li
))
!=
NULL
)
{
clusterManagerNode
*
n
=
ln
->
value
;
char
*
err
=
NULL
;
redisReply
*
reply
=
CLUSTER_MANAGER_COMMAND
(
n
,
"CONFIG %s %s %d"
,
"SET"
,
redisReply
*
reply
=
CLUSTER_MANAGER_COMMAND
(
n
,
"CONFIG %s %s %d"
,
"SET"
,
"cluster-node-timeout"
,
timeout
);
if
(
reply
==
NULL
)
goto
reply_err
;
...
...
@@ -5032,17 +5032,17 @@ static int clusterManagerCommandSetTimeout(int argc, char **argv) {
ok
=
clusterManagerCheckRedisReply
(
n
,
reply
,
&
err
);
freeReplyObject
(
reply
);
if
(
!
ok
)
goto
reply_err
;
clusterManagerLogWarn
(
"*** New timeout set for %s:%d
\n
"
,
n
->
ip
,
clusterManagerLogWarn
(
"*** New timeout set for %s:%d
\n
"
,
n
->
ip
,
n
->
port
);
ok_count
++
;
continue
;
reply_err:
if
(
err
==
NULL
)
err
=
""
;
clusterManagerLogErr
(
"ERR setting node-timeot for %s:%d: %s
\n
"
,
n
->
ip
,
clusterManagerLogErr
(
"ERR setting node-timeot for %s:%d: %s
\n
"
,
n
->
ip
,
n
->
port
,
err
);
err_count
++
;
}
clusterManagerLogInfo
(
">>> New node timeout set. %d OK, %d ERR.
\n
"
,
clusterManagerLogInfo
(
">>> New node timeout set. %d OK, %d ERR.
\n
"
,
ok_count
,
err_count
);
return
1
;
invalid_args:
...
...
@@ -5070,7 +5070,7 @@ static int clusterManagerCommandImport(int argc, char **argv) {
"pass a valid address (ie. 120.0.0.1:7000).
\n
"
;
goto
invalid_args
;
}
clusterManagerLogInfo
(
">>> Importing data from %s:%d to cluster %s:%d
\n
"
,
clusterManagerLogInfo
(
">>> Importing data from %s:%d to cluster %s:%d
\n
"
,
src_ip
,
src_port
,
ip
,
port
);
clusterManagerNode
*
refnode
=
clusterManagerNewNode
(
ip
,
port
);
...
...
@@ -5082,7 +5082,7 @@ static int clusterManagerCommandImport(int argc, char **argv) {
redisContext
*
src_ctx
=
redisConnect
(
src_ip
,
src_port
);
if
(
src_ctx
->
err
)
{
success
=
0
;
fprintf
(
stderr
,
"Could not connect to Redis at %s:%d: %s.
\n
"
,
src_ip
,
fprintf
(
stderr
,
"Could not connect to Redis at %s:%d: %s.
\n
"
,
src_ip
,
src_port
,
src_ctx
->
errstr
);
goto
cleanup
;
}
...
...
@@ -5094,7 +5094,7 @@ static int clusterManagerCommandImport(int argc, char **argv) {
}
if
(
getLongInfoField
(
src_reply
->
str
,
"cluster_enabled"
))
{
clusterManagerLogErr
(
"[ERR] The source node should not be a "
"cluster node.
\n
"
);
"cluster node.
\n
"
);
success
=
0
;
goto
cleanup
;
}
...
...
@@ -5123,7 +5123,7 @@ static int clusterManagerCommandImport(int argc, char **argv) {
slots_map
[
i
]
=
n
;
break
;
}
}
}
}
char
cmdfmt
[
50
]
=
"MIGRATE %s %d %s %d %d"
;
...
...
@@ -5138,7 +5138,7 @@ static int clusterManagerCommandImport(int argc, char **argv) {
while
(
cursor
!=
0
)
{
if
(
cursor
<
0
)
cursor
=
0
;
freeReplyObject
(
src_reply
);
src_reply
=
reconnectingRedisCommand
(
src_ctx
,
"SCAN %d COUNT %d"
,
src_reply
=
reconnectingRedisCommand
(
src_ctx
,
"SCAN %d COUNT %d"
,
cursor
,
1000
);
if
(
!
src_reply
||
src_reply
->
type
==
REDIS_REPLY_ERROR
)
{
if
(
src_reply
&&
src_reply
->
str
)
reply_err
=
src_reply
->
str
;
...
...
@@ -5148,7 +5148,7 @@ static int clusterManagerCommandImport(int argc, char **argv) {
assert
(
src_reply
->
type
==
REDIS_REPLY_ARRAY
);
assert
(
src_reply
->
elements
>=
2
);
assert
(
src_reply
->
element
[
1
]
->
type
==
REDIS_REPLY_ARRAY
);
if
(
src_reply
->
element
[
0
]
->
type
==
REDIS_REPLY_STRING
)
if
(
src_reply
->
element
[
0
]
->
type
==
REDIS_REPLY_STRING
)
cursor
=
atoi
(
src_reply
->
element
[
0
]
->
str
);
else
if
(
src_reply
->
element
[
0
]
->
type
==
REDIS_REPLY_INTEGER
)
cursor
=
src_reply
->
element
[
0
]
->
integer
;
...
...
@@ -5160,14 +5160,14 @@ static int clusterManagerCommandImport(int argc, char **argv) {
uint16_t
slot
=
clusterManagerKeyHashSlot
(
key
,
kr
->
len
);
clusterManagerNode
*
target
=
slots_map
[
slot
];
printf
(
"Migrating %s to %s:%d: "
,
key
,
target
->
ip
,
target
->
port
);
redisReply
*
r
=
reconnectingRedisCommand
(
src_ctx
,
cmdfmt
,
target
->
ip
,
target
->
port
,
key
,
0
,
timeout
,
redisReply
*
r
=
reconnectingRedisCommand
(
src_ctx
,
cmdfmt
,
target
->
ip
,
target
->
port
,
key
,
0
,
timeout
,
"COPY"
,
"REPLACE"
);
if
(
!
r
||
r
->
type
==
REDIS_REPLY_ERROR
)
{
if
(
r
&&
r
->
str
)
{
clusterManagerLogErr
(
"Source %s:%d replied with "
"error:
\n
%s
\n
"
,
src_ip
,
src_port
,
"error:
\n
%s
\n
"
,
src_ip
,
src_port
,
r
->
str
);
}
success
=
0
;
...
...
@@ -5178,8 +5178,8 @@ static int clusterManagerCommandImport(int argc, char **argv) {
}
}
cleanup:
if
(
reply_err
)
clusterManagerLogErr
(
"Source %s:%d replied with error:
\n
%s
\n
"
,
if
(
reply_err
)
clusterManagerLogErr
(
"Source %s:%d replied with error:
\n
%s
\n
"
,
src_ip
,
src_port
,
reply_err
);
if
(
src_ctx
)
redisFree
(
src_ctx
);
if
(
src_reply
)
freeReplyObject
(
src_reply
);
...
...
@@ -5208,13 +5208,13 @@ static int clusterManagerCommandCall(int argc, char **argv) {
listNode
*
ln
;
listRewind
(
cluster_manager
.
nodes
,
&
li
);
while
((
ln
=
listNext
(
&
li
))
!=
NULL
)
{
clusterManagerNode
*
n
=
ln
->
value
;
clusterManagerNode
*
n
=
ln
->
value
;
if
(
!
n
->
context
)
CLUSTER_MANAGER_NODE_CONNECT
(
n
);
redisReply
*
reply
=
NULL
;
redisAppendCommandArgv
(
n
->
context
,
argc
,
(
const
char
**
)
argv
,
argvlen
);
int
status
=
redisGetReply
(
n
->
context
,
(
void
**
)(
&
reply
));
if
(
status
!=
REDIS_OK
||
reply
==
NULL
)
printf
(
"%s:%d: Failed!
\n
"
,
n
->
ip
,
n
->
port
);
if
(
status
!=
REDIS_OK
||
reply
==
NULL
)
printf
(
"%s:%d: Failed!
\n
"
,
n
->
ip
,
n
->
port
);
else
{
sds
formatted_reply
=
cliFormatReplyTTY
(
reply
,
""
);
printf
(
"%s:%d: %s
\n
"
,
n
->
ip
,
n
->
port
,
(
char
*
)
formatted_reply
);
...
...
@@ -5232,7 +5232,7 @@ invalid_args:
static
int
clusterManagerCommandHelp
(
int
argc
,
char
**
argv
)
{
UNUSED
(
argc
);
UNUSED
(
argv
);
int
commands_count
=
sizeof
(
clusterManagerCommands
)
/
int
commands_count
=
sizeof
(
clusterManagerCommands
)
/
sizeof
(
clusterManagerCommandDef
);
int
i
=
0
,
j
;
fprintf
(
stderr
,
"Cluster Manager Commands:
\n
"
);
...
...
@@ -6575,7 +6575,7 @@ int main(int argc, char **argv) {
config
.
cluster_manager_command
.
slots
=
0
;
config
.
cluster_manager_command
.
timeout
=
CLUSTER_MANAGER_MIGRATE_TIMEOUT
;
config
.
cluster_manager_command
.
pipeline
=
CLUSTER_MANAGER_MIGRATE_PIPELINE
;
config
.
cluster_manager_command
.
threshold
=
config
.
cluster_manager_command
.
threshold
=
CLUSTER_MANAGER_REBALANCE_THRESHOLD
;
pref
.
hints
=
1
;
...
...
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