Skip to content
GitLab
Menu
Projects
Groups
Snippets
Loading...
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in
Toggle navigation
Menu
Open sidebar
ruanhaishen
redis
Commits
2280f4f7
Commit
2280f4f7
authored
Nov 27, 2018
by
artix
Committed by
antirez
Dec 11, 2018
Browse files
Cluster Manager: code cleanup.
parent
e084b8cc
Changes
1
Hide whitespace changes
Inline
Side-by-side
src/redis-cli.c
View file @
2280f4f7
...
@@ -2746,10 +2746,23 @@ cleanup:
...
@@ -2746,10 +2746,23 @@ cleanup:
return success;
return success;
}
}
static int clusterManagerDelSlot(clusterManagerNode *node, int slot) {
static int clusterManagerClearSlotStatus(clusterManagerNode *node, int slot) {
redisReply *reply = CLUSTER_MANAGER_COMMAND(node,
"CLUSTER SETSLOT %d %s", slot, "STABLE");
int success = clusterManagerCheckRedisReply(node, reply, NULL);
if (reply) freeReplyObject(reply);
return success;
}
static int clusterManagerDelSlot(clusterManagerNode *node, int slot,
int ignore_unassigned_err)
{
redisReply *reply = CLUSTER_MANAGER_COMMAND(node,
redisReply *reply = CLUSTER_MANAGER_COMMAND(node,
"CLUSTER DELSLOTS %d", slot);
"CLUSTER DELSLOTS %d", slot);
int success = clusterManagerCheckRedisReply(node, reply, NULL);
int success = clusterManagerCheckRedisReply(node, reply, NULL);
if (!success && reply && reply->type == REDIS_REPLY_ERROR &&
ignore_unassigned_err &&
strstr(reply->str, "already unassigned") != NULL) success = 1;
if (reply) freeReplyObject(reply);
if (reply) freeReplyObject(reply);
return success;
return success;
}
}
...
@@ -3658,24 +3671,18 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
...
@@ -3658,24 +3671,18 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
listRewind(none, &li);
listRewind(none, &li);
while ((ln = listNext(&li)) != NULL) {
while ((ln = listNext(&li)) != NULL) {
sds slot = ln->value;
sds slot = ln->value;
int s = atoi(slot);
clusterManagerNode *n = clusterManagerNodeMasterRandom();
clusterManagerNode *n = clusterManagerNodeMasterRandom();
clusterManagerLogInfo(">>> Covering slot %s with %s:%d\n",
clusterManagerLogInfo(">>> Covering slot %s with %s:%d\n",
slot, n->ip, n->port);
slot, n->ip, n->port);
/* Ensure the slot is not already assigned. */
/* Ensure the slot is not already assigned. */
redisReply *r = CLUSTER_MANAGER_COMMAND(n,
clusterManagerDelSlot(n, s, 1);
"CLUSTER DELSLOTS %s", slot);
if (!clusterManagerAddSlot(n, s)) fixed = -1;
if (r) freeReplyObject(r);
if (fixed >= 0 && !clusterManagerBumpEpoch(n)) fixed = -1;
r = CLUSTER_MANAGER_COMMAND(n,
"CLUSTER ADDSLOTS %s", slot);
if (!clusterManagerCheckRedisReply(n, r, NULL)) fixed = -1;
if (r) freeReplyObject(r);
r = CLUSTER_MANAGER_COMMAND(n, "CLUSTER BUMPEPOCH");
if (!clusterManagerCheckRedisReply(n, r, NULL)) fixed = -1;
if (r) freeReplyObject(r);
if (fixed < 0) goto cleanup;
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 */
* info into the node struct, in order to keep it synced */
n->slots[
atoi(slot)
] = 1;
n->slots[
s
] = 1;
fixed++;
fixed++;
}
}
}
}
...
@@ -3691,6 +3698,7 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
...
@@ -3691,6 +3698,7 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
listRewind(single, &li);
listRewind(single, &li);
while ((ln = listNext(&li)) != NULL) {
while ((ln = listNext(&li)) != NULL) {
sds slot = ln->value;
sds slot = ln->value;
int s = atoi(slot);
dictEntry *entry = dictFind(clusterManagerUncoveredSlots, slot);
dictEntry *entry = dictFind(clusterManagerUncoveredSlots, slot);
assert(entry != NULL);
assert(entry != NULL);
list *nodes = (list *) dictGetVal(entry);
list *nodes = (list *) dictGetVal(entry);
...
@@ -3700,16 +3708,9 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
...
@@ -3700,16 +3708,9 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
clusterManagerLogInfo(">>> Covering slot %s with %s:%d\n",
clusterManagerLogInfo(">>> Covering slot %s with %s:%d\n",
slot, n->ip, n->port);
slot, n->ip, n->port);
/* Ensure the slot is not already assigned. */
/* Ensure the slot is not already assigned. */
redisReply *r = CLUSTER_MANAGER_COMMAND(n,
clusterManagerDelSlot(n, s, 1);
"CLUSTER DELSLOTS %s", slot);
if (!clusterManagerAddSlot(n, s)) fixed = -1;
if (r) freeReplyObject(r);
if (fixed >= 0 && !clusterManagerBumpEpoch(n)) fixed = -1;
r = CLUSTER_MANAGER_COMMAND(n,
"CLUSTER ADDSLOTS %s", slot);
if (!clusterManagerCheckRedisReply(n, r, NULL)) fixed = -1;
if (r) freeReplyObject(r);
r = CLUSTER_MANAGER_COMMAND(n, "CLUSTER BUMPEPOCH");
if (!clusterManagerCheckRedisReply(n, r, NULL)) fixed = -1;
if (r) freeReplyObject(r);
if (fixed < 0) goto cleanup;
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 */
* info into the node struct, in order to keep it synced */
...
@@ -3744,21 +3745,12 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
...
@@ -3744,21 +3745,12 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
"to %s:%d\n", slot,
"to %s:%d\n", slot,
target->ip, target->port);
target->ip, target->port);
/* Ensure the slot is not already assigned. */
/* Ensure the slot is not already assigned. */
redisReply *r = CLUSTER_MANAGER_COMMAND(target,
clusterManagerDelSlot(target, s, 1);
"CLUSTER DELSLOTS %s", slot);
if (!clusterManagerAddSlot(target, s)) fixed = -1;
if (r) freeReplyObject(r);
r = CLUSTER_MANAGER_COMMAND(target,
"CLUSTER ADDSLOTS %s", slot);
if (!clusterManagerCheckRedisReply(target, r, NULL)) fixed = -1;
if (r) freeReplyObject(r);
if (fixed < 0) goto cleanup;
if (fixed < 0) goto cleanup;
r = CLUSTER_MANAGER_COMMAND(target,
if (!clusterManagerClearSlotStatus(target, s))
"CLUSTER SETSLOT %s %s", slot, "STABLE");
fixed = -1;
if (!clusterManagerCheckRedisReply(target, r, NULL)) fixed = -1;
if (fixed >= 0 && !clusterManagerBumpEpoch(target)) fixed = -1;
if (r) freeReplyObject(r);
r = CLUSTER_MANAGER_COMMAND(target, "CLUSTER BUMPEPOCH");
if (!clusterManagerCheckRedisReply(target, r, NULL)) fixed = -1;
if (r) freeReplyObject(r);
if (fixed < 0) goto cleanup;
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 */
* info into the node struct, in order to keep it synced */
...
@@ -3770,23 +3762,15 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
...
@@ -3770,23 +3762,15 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
clusterManagerNode *src = nln->value;
clusterManagerNode *src = nln->value;
if (src == target) continue;
if (src == target) continue;
/* Assign the slot to target node in the source node. */
/* Assign the slot to target node in the source node. */
redisReply *r = CLUSTER_MANAGER_COMMAND(src,
if (!clusterManagerSetSlot(src, target, s, "NODE", NULL))
"CLUSTER SETSLOT %s %s %s", slot,
"NODE", target->name);
if (!clusterManagerCheckRedisReply(src, r, NULL))
fixed = -1;
fixed = -1;
if (r) freeReplyObject(r);
if (fixed < 0) goto cleanup;
if (fixed < 0) goto cleanup;
/* Set the source node in 'importing' state
/* Set the source node in 'importing' state
* (even if we will actually migrate keys away)
* (even if we will actually migrate keys away)
* in order to avoid receiving redirections
* in order to avoid receiving redirections
* for MIGRATE. */
* for MIGRATE. */
r = CLUSTER_MANAGER_COMMAND(src,
if (!clusterManagerSetSlot(src, target, s,
"CLUSTER SETSLOT %s %s %s", slot,
"IMPORTING", NULL)) fixed = -1;
"IMPORTING", target->name);
if (!clusterManagerCheckRedisReply(src, r, NULL))
fixed = -1;
if (r) freeReplyObject(r);
if (fixed < 0) goto cleanup;
if (fixed < 0) goto cleanup;
int opts = CLUSTER_MANAGER_OPT_VERBOSE |
int opts = CLUSTER_MANAGER_OPT_VERBOSE |
CLUSTER_MANAGER_OPT_COLD;
CLUSTER_MANAGER_OPT_COLD;
...
@@ -3794,12 +3778,8 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
...
@@ -3794,12 +3778,8 @@ static int clusterManagerFixSlotsCoverage(char *all_slots) {
fixed = -1;
fixed = -1;
goto cleanup;
goto cleanup;
}
}
r = CLUSTER_MANAGER_COMMAND(src,
if (!clusterManagerClearSlotStatus(src, s))
"CLUSTER SETSLOT %s %s", slot,
"STABLE");
if (!clusterManagerCheckRedisReply(src, r, NULL))
fixed = -1;
fixed = -1;
if (r) freeReplyObject(r);
if (fixed < 0) goto cleanup;
if (fixed < 0) goto cleanup;
}
}
fixed++;
fixed++;
...
@@ -3923,24 +3903,13 @@ static int clusterManagerFixOpenSlot(int slot) {
...
@@ -3923,24 +3903,13 @@ static int clusterManagerFixOpenSlot(int slot) {
// Use ADDSLOTS to assign the 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);
owner->ip, owner->port);
redisReply *reply = CLUSTER_MANAGER_COMMAND(owner, "CLUSTER "
success = clusterManagerClearSlotStatus(owner, slot);
"SETSLOT %d %s",
slot, "STABLE");
success = clusterManagerCheckRedisReply(owner, reply, NULL);
if (reply) freeReplyObject(reply);
if (!success) goto cleanup;
if (!success) goto cleanup;
/* Ensure that the slot is unassigned before assigning it to the
/* Ensure that the slot is unassigned before assigning it to the
* owner. */
* owner. */
reply = CLUSTER_MANAGER_COMMAND(owner, "CLUSTER DELSLOTS %d", slot);
success = clusterManagerDelSlot(owner, slot, 1);
success = clusterManagerCheckRedisReply(owner, reply, NULL);
/* Ignore "already unassigned" error. */
if (!success && reply && reply->type == REDIS_REPLY_ERROR &&
strstr(reply->str, "already unassigned") != NULL) success = 1;
if (reply) freeReplyObject(reply);
if (!success) goto cleanup;
if (!success) goto cleanup;
reply = CLUSTER_MANAGER_COMMAND(owner, "CLUSTER ADDSLOTS %d", slot);
success = clusterManagerAddSlot(owner, slot);
success = clusterManagerCheckRedisReply(owner, reply, NULL);
if (reply) freeReplyObject(reply);
if (!success) goto cleanup;
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 */
* info into the node struct, in order to keep it synced */
...
@@ -3948,9 +3917,7 @@ static int clusterManagerFixOpenSlot(int slot) {
...
@@ -3948,9 +3917,7 @@ static int clusterManagerFixOpenSlot(int slot) {
/* Make sure this information will propagate. Not strictly needed
/* Make sure this information will propagate. Not strictly needed
* since there is no past owner, so all the other nodes will accept
* since there is no past owner, so all the other nodes will accept
* whatever epoch this node will claim the slot with. */
* whatever epoch this node will claim the slot with. */
reply = CLUSTER_MANAGER_COMMAND(owner, "CLUSTER BUMPEPOCH");
success = clusterManagerBumpEpoch(owner);
success = clusterManagerCheckRedisReply(owner, reply, NULL);
if (reply) freeReplyObject(reply);
if (!success) goto cleanup;
if (!success) goto cleanup;
/* Remove the owner from the list of migrating/importing
/* Remove the owner from the list of migrating/importing
* nodes. */
* nodes. */
...
@@ -3970,16 +3937,10 @@ static int clusterManagerFixOpenSlot(int slot) {
...
@@ -3970,16 +3937,10 @@ static int clusterManagerFixOpenSlot(int slot) {
* the owner has been set in the previous condition (owner == NULL). */
* the owner has been set in the previous condition (owner == NULL). */
assert(owner != NULL);
assert(owner != NULL);
listRewind(owners, &li);
listRewind(owners, &li);
redisReply *reply = NULL;
while ((ln = listNext(&li)) != NULL) {
while ((ln = listNext(&li)) != NULL) {
clusterManagerNode *n = ln->value;
clusterManagerNode *n = ln->value;
if (n == owner) continue;
if (n == owner) continue;
reply = CLUSTER_MANAGER_COMMAND(n, "CLUSTER DELSLOTS %d", slot);
success = clusterManagerDelSlot(n, slot, 1);
success = clusterManagerCheckRedisReply(n, reply, NULL);
/* Ignore "already unassigned" error. */
if (!success && reply && reply->type == REDIS_REPLY_ERROR &&
strstr(reply->str, "already unassigned") != NULL) success = 1;
if (reply) freeReplyObject(reply);
if (!success) goto cleanup;
if (!success) goto cleanup;
n->slots[slot] = 0;
n->slots[slot] = 0;
/* Assign the slot to the owner in the node 'n' configuration.' */
/* Assign the slot to the owner in the node 'n' configuration.' */
...
@@ -4021,11 +3982,7 @@ static int clusterManagerFixOpenSlot(int slot) {
...
@@ -4021,11 +3982,7 @@ static int clusterManagerFixOpenSlot(int slot) {
if (!success) goto cleanup;
if (!success) goto cleanup;
clusterManagerLogInfo(">>> Setting %d as STABLE in "
clusterManagerLogInfo(">>> Setting %d as STABLE in "
"%s:%d\n", slot, n->ip, n->port);
"%s:%d\n", slot, n->ip, n->port);
success = clusterManagerClearSlotStatus(n, slot);
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;
if (!success) goto cleanup;
}
}
/* Since the slot has been moved in "cold" mode, ensure that all the
/* Since the slot has been moved in "cold" mode, ensure that all the
...
@@ -4035,10 +3992,7 @@ static int clusterManagerFixOpenSlot(int slot) {
...
@@ -4035,10 +3992,7 @@ static int clusterManagerFixOpenSlot(int slot) {
clusterManagerNode *n = ln->value;
clusterManagerNode *n = ln->value;
if (n == owner) continue;
if (n == owner) continue;
if (n->flags & CLUSTER_MANAGER_FLAG_SLAVE) continue;
if (n->flags & CLUSTER_MANAGER_FLAG_SLAVE) continue;
redisReply *r = CLUSTER_MANAGER_COMMAND(n,
success = clusterManagerSetSlot(n, owner, slot, "NODE", NULL);
"CLUSTER SETSLOT %d %s %s", slot, "NODE", owner->name);
success = clusterManagerCheckRedisReply(n, r, NULL);
if (r) freeReplyObject(r);
if (!success) goto cleanup;
if (!success) goto cleanup;
}
}
} else {
} else {
...
@@ -4104,7 +4058,7 @@ static int clusterManagerFixMultipleSlotOwners(int slot, list *owners) {
...
@@ -4104,7 +4058,7 @@ static int clusterManagerFixMultipleSlotOwners(int slot, list *owners) {
* could reply with an "already unassigned" error and if it should fail
* could reply with an "already unassigned" error and if it should fail
* for other reasons, it would lead to a failure in the follwing ADDSLOTS
* for other reasons, it would lead to a failure in the follwing ADDSLOTS
* command. */
* command. */
clusterManagerDelSlot(owner, slot);
clusterManagerDelSlot(owner, slot
, 1
);
if (!clusterManagerAddSlot(owner, slot)) return 0;
if (!clusterManagerAddSlot(owner, slot)) return 0;
if (!clusterManagerBumpEpoch(owner)) return 0;
if (!clusterManagerBumpEpoch(owner)) return 0;
listIter li;
listIter li;
...
@@ -4120,7 +4074,7 @@ static int clusterManagerFixMultipleSlotOwners(int slot, list *owners) {
...
@@ -4120,7 +4074,7 @@ static int clusterManagerFixMultipleSlotOwners(int slot, list *owners) {
int count = clusterManagerCountKeysInSlot(n, slot);
int count = clusterManagerCountKeysInSlot(n, slot);
success = (count >= 0);
success = (count >= 0);
if (!success) break;
if (!success) break;
clusterManagerDelSlot(n, slot);
clusterManagerDelSlot(n, slot
, 1
);
if (!clusterManagerSetSlot(n, owner, slot, "node", NULL)) return 0;
if (!clusterManagerSetSlot(n, owner, slot, "node", NULL)) return 0;
if (count > 0) {
if (count > 0) {
int opts = CLUSTER_MANAGER_OPT_VERBOSE |
int opts = CLUSTER_MANAGER_OPT_VERBOSE |
...
...
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