Redis/redis
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

New Cluster Command: CLUSTER DELSLOTSRANGE and CLUSTER ADDSLOTSRANGE (#9445)

Browse Source
This commit is contained in:
Wen Hui
2021-10-26 21:44:33 -07:00
committed by GitHub
parent 37dc2f13b4
commit 5fb4adba65
2 changed files with 152 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cluster.c
+84 -25
View File
@@ -4440,6 +4440,42 @@ static int isReplicaAvailable(clusterNode *node) {
return (repl_offset != 0);
}
int checkSlotAssignmentsOrReply(client *c, unsigned char *slots, int del, int start_slot, int end_slot) {
int slot;
for (slot = start_slot; slot <= end_slot; slot++) {
if (del && server.cluster->slots[slot] == NULL) {
addReplyErrorFormat(c,"Slot %d is already unassigned", slot);
return C_ERR;
} else if (!del && server.cluster->slots[slot]) {
addReplyErrorFormat(c,"Slot %d is already busy", slot);
return C_ERR;
}
if (slots[slot]++ == 1) {
addReplyErrorFormat(c,"Slot %d specified multiple times",(int)slot);
return C_ERR;
}
}
return C_OK;
}
void clusterUpdateSlots(client *c, unsigned char *slots, int del) {
int j;
for (j = 0; j < CLUSTER_SLOTS; j++) {
if (slots[j]) {
int retval;
/* If this slot was set as importing we can clear this
* state as now we are the real owner of the slot. */
if (server.cluster->importing_slots_from[j])
server.cluster->importing_slots_from[j] = NULL;
retval = del ? clusterDelSlot(j) :
clusterAddSlot(myself,j);
serverAssertWithInfo(c,NULL,retval == C_OK);
}
}
}
void addNodeReplyForClusterSlot(client *c, clusterNode *node, int start_slot, int end_slot) {
int i, nested_elements = 3; /* slots (2) + master addr (1) */
void *nested_replylen = addReplyDeferredLen(c);
@@ -4517,6 +4553,8 @@ void clusterCommand(client *c) {
const char *help[] = {
"ADDSLOTS <slot> [<slot> ...]",
" Assign slots to current node.",
"ADDSLOTSRANGE <start slot> <end slot> [<start slot> <end slot> ...]",
" Assign slots which are between <start-slot> and <end-slot> to current node.",
"BUMPEPOCH",
" Advance the cluster config epoch.",
"COUNT-FAILURE-REPORTS <node-id>",
@@ -4525,6 +4563,8 @@ void clusterCommand(client *c) {
" Return the number of keys in <slot>.",
"DELSLOTS <slot> [<slot> ...]",
" Delete slots information from current node.",
"DELSLOTSRANGE <start slot> <end slot> [<start slot> <end slot> ...]",
" Delete slots information which are between <start-slot> and <end-slot> from current node.",
"FAILOVER [FORCE|TAKEOVER]",
" Promote current replica node to being a master.",
"FORGET <node-id>",
@@ -4623,44 +4663,63 @@ NULL
unsigned char *slots = zmalloc(CLUSTER_SLOTS);
int del = !strcasecmp(c->argv[1]->ptr,"delslots");
memset(slots,0,CLUSTER_SLOTS);
/* Check that all the arguments are parseable.*/
for (j = 2; j < c->argc; j++) {
if ((slot = getSlotOrReply(c,c->argv[j])) == C_ERR) {
zfree(slots);
return;
}
}
/* Check that the slots are not already busy. */
for (j = 2; j < c->argc; j++) {
slot = getSlotOrReply(c,c->argv[j]);
if (checkSlotAssignmentsOrReply(c, slots, del, slot, slot) == C_ERR) {
zfree(slots);
return;
}
}
clusterUpdateSlots(c, slots, del);
zfree(slots);
clusterDoBeforeSleep(CLUSTER_TODO_UPDATE_STATE|CLUSTER_TODO_SAVE_CONFIG);
addReply(c,shared.ok);
} else if ((!strcasecmp(c->argv[1]->ptr,"addslotsrange") ||
!strcasecmp(c->argv[1]->ptr,"delslotsrange")) && c->argc >= 4) {
if (c->argc % 2 == 1) {
addReplyErrorFormat(c,"wrong number of arguments for '%s' command",
c->cmd->name);
return;
}
/* CLUSTER ADDSLOTSRANGE <start slot> <end slot> [<start slot> <end slot> ...] */
/* CLUSTER DELSLOTSRANGE <start slot> <end slot> [<start slot> <end slot> ...] */
int j, startslot, endslot;
unsigned char *slots = zmalloc(CLUSTER_SLOTS);
int del = !strcasecmp(c->argv[1]->ptr,"delslotsrange");
memset(slots,0,CLUSTER_SLOTS);
/* Check that all the arguments are parseable and that all the
* slots are not already busy. */
for (j = 2; j < c->argc; j++) {
if ((slot = getSlotOrReply(c,c->argv[j])) == -1) {
for (j = 2; j < c->argc; j += 2) {
if ((startslot = getSlotOrReply(c,c->argv[j])) == C_ERR) {
zfree(slots);
return;
}
if (del && server.cluster->slots[slot] == NULL) {
addReplyErrorFormat(c,"Slot %d is already unassigned", slot);
zfree(slots);
return;
} else if (!del && server.cluster->slots[slot]) {
addReplyErrorFormat(c,"Slot %d is already busy", slot);
if ((endslot = getSlotOrReply(c,c->argv[j+1])) == C_ERR) {
zfree(slots);
return;
}
if (slots[slot]++ == 1) {
addReplyErrorFormat(c,"Slot %d specified multiple times",
(int)slot);
if (startslot > endslot) {
addReplyErrorFormat(c,"start slot number %d is greater than end slot number %d", startslot, endslot);
zfree(slots);
return;
}
if (checkSlotAssignmentsOrReply(c, slots, del, startslot, endslot) == C_ERR) {
zfree(slots);
return;
}
}
for (j = 0; j < CLUSTER_SLOTS; j++) {
if (slots[j]) {
int retval;
/* If this slot was set as importing we can clear this
* state as now we are the real owner of the slot. */
if (server.cluster->importing_slots_from[j])
server.cluster->importing_slots_from[j] = NULL;
retval = del ? clusterDelSlot(j) :
clusterAddSlot(myself,j);
serverAssertWithInfo(c,NULL,retval == C_OK);
}
}
clusterUpdateSlots(c, slots, del);
zfree(slots);
clusterDoBeforeSleep(CLUSTER_TODO_UPDATE_STATE|CLUSTER_TODO_SAVE_CONFIG);
addReply(c,shared.ok);
tests/cluster/tests/23-multiple-slot-operations.tcl
+68
View File
@@ -0,0 +1,68 @@
# Check the multiple slot add and remove commands
source "../tests/includes/init-tests.tcl"
proc cluster_allocate_with_continuous_slots {n} {
R 0 cluster ADDSLOTSRANGE 0 3276
R 1 cluster ADDSLOTSRANGE 3277 6552
R 2 cluster ADDSLOTSRANGE 6553 9828
R 3 cluster ADDSLOTSRANGE 9829 13104
R 4 cluster ADDSLOTSRANGE 13105 16383
}
proc cluster_create_with_continuous_slots {masters slaves} {
cluster_allocate_with_continuous_slots $masters
if {$slaves} {
cluster_allocate_slaves $masters $slaves
}
assert_cluster_state ok
}
test "Create a 5 nodes cluster" {
cluster_create_with_continuous_slots 5 5
}
test "Cluster should start ok" {
assert_cluster_state ok
}
set master1 [Rn 0]
set master2 [Rn 1]
set master3 [Rn 2]
set master4 [Rn 3]
set master5 [Rn 4]
test "Continuous slots distribution" {
assert_match "* 0-3276*" [$master1 CLUSTER NODES]
assert_match "* 3277-6552*" [$master2 CLUSTER NODES]
assert_match "* 6553-9828*" [$master3 CLUSTER NODES]
assert_match "* 9829-13104*" [$master4 CLUSTER NODES]
assert_match "* 13105-16383*" [$master5 CLUSTER NODES]
assert_match "*0 3276*" [$master1 CLUSTER SLOTS]
assert_match "*3277 6552*" [$master2 CLUSTER SLOTS]
assert_match "*6553 9828*" [$master3 CLUSTER SLOTS]
assert_match "*9829 13104*" [$master4 CLUSTER SLOTS]
assert_match "*13105 16383*" [$master5 CLUSTER SLOTS]
$master1 CLUSTER DELSLOTSRANGE 3001 3050
assert_match "* 0-3000 3051-3276*" [$master1 CLUSTER NODES]
assert_match "*0 3000*3051 3276*" [$master1 CLUSTER SLOTS]
$master2 CLUSTER DELSLOTSRANGE 5001 5500
assert_match "* 3277-5000 5501-6552*" [$master2 CLUSTER NODES]
assert_match "*3277 5000*5501 6552*" [$master2 CLUSTER SLOTS]
$master3 CLUSTER DELSLOTSRANGE 7001 7100 8001 8500
assert_match "* 6553-7000 7101-8000 8501-9828*" [$master3 CLUSTER NODES]
assert_match "*6553 7000*7101 8000*8501 9828*" [$master3 CLUSTER SLOTS]
$master4 CLUSTER DELSLOTSRANGE 11001 12000 12101 12200
assert_match "* 9829-11000 12001-12100 12201-13104*" [$master4 CLUSTER NODES]
assert_match "*9829 11000*12001 12100*12201 13104*" [$master4 CLUSTER SLOTS]
$master5 CLUSTER DELSLOTSRANGE 13501 14000 15001 16000
assert_match "* 13105-13500 14001-15000 16001-16383*" [$master5 CLUSTER NODES]
assert_match "*13105 13500*14001 15000*16001 16383*" [$master5 CLUSTER SLOTS]
}