fluencelabs/redis
1
0
Fork 0
mirror of https://github.com/fluencelabs/redis synced 2025-03-30 22:31:03 +00:00
Code Issues Projects Releases Wiki Activity

CLUSTER BUMPEPOCH initial implementation fixed.

Browse Source
This commit is contained in:
antirez 2016-01-11 15:39:09 +01:00
parent e4eb6c7a5d
commit 02c40c9dc2
2 changed files with 25 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
src/cluster.c
View File

@ -4022,13 +4022,6 @@ void clusterCommand(client *c) {
} }
clusterDelSlot(slot); clusterDelSlot(slot);
clusterAddSlot(n,slot); clusterAddSlot(n,slot);
} else if (!strcasecmp(c->argv[3]->ptr,"bumpepoch") && c->argc == 2) {
/* CLUSTER BUMPEPOCH */
int retval = clusterBumpConfigEpochWithoutConsensus();
sds reply = sdscatprintf(sdsempty(),"%s %llu\r\n",
(retval == C_OK) ? "BUMPED" : "STILL",
(unsigned long long) myself->configEpoch);
addReplySds(c,reply);
} else { } else {
addReplyError(c, addReplyError(c,
"Invalid CLUSTER SETSLOT action or number of arguments"); "Invalid CLUSTER SETSLOT action or number of arguments");
@ -4036,6 +4029,13 @@ void clusterCommand(client *c) {
} }
clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG|CLUSTER_TODO_UPDATE_STATE); clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG|CLUSTER_TODO_UPDATE_STATE);
addReply(c,shared.ok); addReply(c,shared.ok);
} else if (!strcasecmp(c->argv[1]->ptr,"bumpepoch") && c->argc == 2) {
/* CLUSTER BUMPEPOCH */
int retval = clusterBumpConfigEpochWithoutConsensus();
sds reply = sdscatprintf(sdsempty(),"+%s %llu\r\n",
(retval == C_OK) ? "BUMPED" : "STILL",
(unsigned long long) myself->configEpoch);
addReplySds(c,reply);
} else if (!strcasecmp(c->argv[1]->ptr,"info") && c->argc == 2) { } else if (!strcasecmp(c->argv[1]->ptr,"info") && c->argc == 2) {
/* CLUSTER INFO */ /* CLUSTER INFO */
char *statestr[] = {"ok","fail","needhelp"}; char *statestr[] = {"ok","fail","needhelp"};

25
src/redis-trib.rb
View File

@ -499,13 +499,15 @@ class RedisTrib
end end
# Return the owner of the specified slot # Return the owner of the specified slot
def get_slot_owner(slot) def get_slot_owners(slot)
owners = []
@nodes.each{|n| @nodes.each{|n|
next if n.has_flag?("slave")
n.slots.each{|s,_| n.slots.each{|s,_|
return n if s == slot owners << n if s == slot
} }
} }
nil owners
end end
# Return the node, among 'nodes' with the greatest number of keys # Return the node, among 'nodes' with the greatest number of keys
@ -532,7 +534,8 @@ class RedisTrib
# Try to obtain the current slot owner, according to the current # Try to obtain the current slot owner, according to the current
# nodes configuration. # nodes configuration.
owner = get_slot_owner(slot) owners = get_slot_owners(slot)
owner = owners[0] if owners.length == 1
migrating = [] migrating = []
importing = [] importing = []
@ -585,9 +588,17 @@ class RedisTrib
# Note that this case also covers multiple nodes having the slot # Note that this case also covers multiple nodes having the slot
# in migrating state, since migrating is a valid state only for # in migrating state, since migrating is a valid state only for
# slot owners. # slot owners.
# if owners.length > 1
# TODO: Use CLUSTER BUMPEPOCH in order to make the owner = get_node_with_most_keys_in_slot(owners,slot)
# winner able to claim the slot over all the other nodes. owners.each{|n|
next if n == owner
n.r.cluster('delslots',slot)
n.r.cluster('setslot',slot,'importing',owner.info[:name])
importing.delete(n) # Avoid duplciates
importing << n
}
owner.r.cluster('bumpepoch')
end
# Case 1: The slot is in migrating state in one slot, and in # Case 1: The slot is in migrating state in one slot, and in
# importing state in 1 slot. That's trivial to address. # importing state in 1 slot. That's trivial to address.