Code Review Bench PR #14637 - Fix adjacent slot range behavior in ASM operations #2
+125
−31
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -184,6 +184,7 @@ slotRangeArray *slotRangeArrayDup(slotRangeArray *slots); | |
| void slotRangeArraySet(slotRangeArray *slots, int idx, int start, int end); | ||
| sds slotRangeArrayToString(slotRangeArray *slots); | ||
| slotRangeArray *slotRangeArrayFromString(sds data); | ||
| void slotRangeArraySortAndMerge(slotRangeArray *slots); | ||
| int slotRangeArrayIsEqual(slotRangeArray *slots1, slotRangeArray *slots2); | ||
| slotRangeArray *slotRangeArrayAppend(slotRangeArray *slots, int slot); | ||
| int slotRangeArrayContains(slotRangeArray *slots, unsigned int slot); | ||
|
|
@@ -193,7 +194,7 @@ slotRangeArrayIter *slotRangeArrayGetIterator(slotRangeArray *slots); | |
| int slotRangeArrayNext(slotRangeArrayIter *it); | ||
| int slotRangeArrayGetCurrentSlot(slotRangeArrayIter *it); | ||
| void slotRangeArrayIteratorFree(slotRangeArrayIter *it); | ||
| int slotRangeArrayNormalizeAndValidate(slotRangeArray *slots, sds *err); | ||
| slotRangeArray *parseSlotRangesOrReply(client *c, int argc, int pos); | ||
|
|
||
| unsigned int clusterDelKeysInSlot(unsigned int hashslot, int by_command); | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -627,7 +627,6 @@ static asmTask *lookupAsmTaskBySlotRange(slotRange *req) { | |
| /* Validates the given slot ranges for a migration task: | ||
| * - Ensures the current node is a master. | ||
| * - Verifies all slots are in a STABLE state. | ||
| * - Confirms all slots belong to a single source node. | ||
| * - Confirms no ongoing import task that overlaps with the slot ranges. | ||
| * | ||
|
|
@@ -804,11 +803,11 @@ asmTask *asmCreateImportTask(const char *task_id, slotRangeArray *slots, sds *er | |
| * initiated for them. */ | ||
| source = validateImportSlotRanges(slots, err, NULL); | ||
| if (!source) | ||
| goto err; | ||
|
|
||
| if (source == getMyClusterNode()) { | ||
| *err = sdsnew("this node is already the owner of the slot range"); | ||
| goto err; | ||
| } | ||
|
|
||
| /* Only support a single task at a time now. */ | ||
|
|
@@ -820,15 +819,15 @@ asmTask *asmCreateImportTask(const char *task_id, slotRangeArray *slots, sds *er | |
| asmTaskCancel(current, "new import requested"); | ||
| } else { | ||
| *err = sdsnew("another ASM task is already in progress"); | ||
| goto err; | ||
| } | ||
| } | ||
| /* There should be no task in progress. */ | ||
| serverAssert(listLength(asmManager->tasks) == 0); | ||
|
|
||
| /* Create a slot migration task */ | ||
| asmTask *task = asmTaskCreate(task_id); | ||
| task->slots = slots; | ||
| task->state = ASM_NONE; | ||
| task->operation = ASM_IMPORT; | ||
| task->source_node = source; | ||
|
|
@@ -842,6 +841,10 @@ asmTask *asmCreateImportTask(const char *task_id, slotRangeArray *slots, sds *er | |
| sdsfree(slots_str); | ||
|
|
||
| return task; | ||
|
|
||
| err: | ||
| slotRangeArrayFree(slots); | ||
| return NULL; | ||
| } | ||
|
|
||
| /* CLUSTER MIGRATION IMPORT <start-slot end-slot [start-slot end-slot ...]> | ||
|
|
@@ -860,7 +863,6 @@ static void clusterMigrationCommandImport(client *c) { | |
|
|
||
| sds err = NULL; | ||
| asmTask *task = asmCreateImportTask(NULL, slots, &err); | ||
| if (!task) { | ||
| addReplyErrorSds(c, err); | ||
| return; | ||
|
|
@@ -1006,6 +1008,20 @@ void clusterMigrationCommand(client *c) { | |
| } | ||
| } | ||
|
|
||
| /* Return the number of keys in the specified slot ranges. */ | ||
| unsigned long long asmCountKeysInSlots(slotRangeArray *slots) { | ||
| if (!slots) return 0; | ||
|
|
||
| unsigned long long key_count = 0; | ||
| int total_ranges = slots->num_ranges; | ||
| for (int i = 0; i < slots->num_ranges; i++) { | ||
| for (int j = slots->ranges[i].start; j <= slots->ranges[i].end; j++) { | ||
| key_count += kvstoreDictSize(server.db[0].keys, j); | ||
| } | ||
| } | ||
| return key_count; | ||
| } | ||
|
|
||
| /* Log a human-readable message for ASM task lifecycle events. */ | ||
| void asmLogTaskEvent(asmTask *task, int event) { | ||
| sds str = slotRangeArrayToString(task->slots); | ||
|
|
@@ -1021,10 +1037,12 @@ void asmLogTaskEvent(asmTask *task, int event) { | |
| serverLog(LL_NOTICE, "Import task %s is ready to takeover slots: %s", task->id, str); | ||
| break; | ||
| case ASM_EVENT_IMPORT_COMPLETED: | ||
| serverLog(LL_NOTICE, "Import task %s completed for slots: %s (imported %llu keys)", | ||
| task->id, str, asmCountKeysInSlots(task->slots)); | ||
| break; | ||
| case ASM_EVENT_MIGRATE_STARTED: | ||
| serverLog(LL_NOTICE, "Migrate task %s started for slots: %s (keys at start: %llu)", | ||
| task->id, str, asmCountKeysInSlots(task->slots)); | ||
| break; | ||
| case ASM_EVENT_MIGRATE_FAILED: | ||
| serverLog(LL_NOTICE, "Migrate task %s failed for slots: %s", task->id, str); | ||
|
|
@@ -1033,7 +1051,8 @@ void asmLogTaskEvent(asmTask *task, int event) { | |
| serverLog(LL_NOTICE, "Migrate task %s preparing to handoff for slots: %s", task->id, str); | ||
| break; | ||
| case ASM_EVENT_MIGRATE_COMPLETED: | ||
| serverLog(LL_NOTICE, "Migrate task %s completed for slots: %s (migrated %llu keys)", | ||
| task->id, str, asmCountKeysInSlots(task->slots)); | ||
| break; | ||
| default: | ||
| break; | ||
|
|
@@ -2847,24 +2866,36 @@ int clusterAsmProcess(const char *task_id, int event, void *arg, char **err) { | |
| if (err) *err = NULL; | ||
|
|
||
| switch (event) { | ||
| case ASM_EVENT_IMPORT_START: { | ||
| /* Validate the slot ranges. */ | ||
| slotRangeArray *slots = slotRangeArrayDup(arg); | ||
| if (slotRangeArrayNormalizeAndValidate(slots, &errsds) != C_OK) { | ||
| slotRangeArrayFree(slots); | ||
| ret = C_ERR; | ||
| break; | ||
| } | ||
| ret = asmCreateImportTask(task_id, slots, &errsds) ? C_OK : C_ERR; | ||
| break; | ||
| } | ||
| case ASM_EVENT_CANCEL: { | ||
| num_cancelled = clusterAsmCancel(task_id, "user request"); | ||
| if (arg) *((int *)arg) = num_cancelled; | ||
| ret = C_OK; | ||
| break; | ||
| } | ||
| case ASM_EVENT_HANDOFF: { | ||
| ret = clusterAsmHandoff(task_id, &errsds); | ||
| break; | ||
| } | ||
| case ASM_EVENT_DONE: { | ||
| ret = clusterAsmDone(task_id, &errsds); | ||
| break; | ||
| } | ||
| default: { | ||
| ret = C_ERR; | ||
| errsds = sdscatprintf(sdsempty(), "Unknown operation: %d", event); | ||
| break; | ||
| } | ||
| } | ||
|
|
||
| if (ret != C_OK && errsds && err) { | ||
|
|
@@ -3273,10 +3304,7 @@ void asmActiveTrimStart(void) { | |
| asmManager->active_trim_current_job_trimmed = 0; | ||
|
|
||
| /* Count the number of keys to trim */ | ||
| asmManager->active_trim_current_job_keys = asmCountKeysInSlots(slots); | ||
|
|
||
| RedisModuleClusterSlotMigrationTrimInfoV1 fsi = { | ||
| REDISMODULE_CLUSTER_SLOT_MIGRATION_TRIMINFO_VERSION, | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -511,6 +511,42 @@ start_cluster 3 3 {tags {external:skip cluster} overrides {cluster-node-timeout | |
| wait_for_asm_done | ||
| } | ||
|
|
||
| test "Test IMPORT with unsorted and adjacent ranges" { | ||
| # Redis should sort and merge adjacent ranges | ||
| # Adjacent means: prev.end + 1 == next.start | ||
| # e.g. 7000-7001 7002-7003 7004-7005 => 7000-7005 | ||
|
|
||
| # Test with adjacent ranges | ||
| set task_id [R 0 CLUSTER MIGRATION IMPORT 7000 7001 7002 7100] | ||
| wait_for_asm_done | ||
| # verify migration is successfully completed on both nodes | ||
| assert_equal "completed" [migration_status 0 $task_id state] | ||
| assert_equal "completed" [migration_status 1 $task_id state] | ||
| # verify slot ranges are merged correctly | ||
| assert_equal "7000-7100" [migration_status 0 $task_id slots] | ||
| assert_equal "7000-7100" [migration_status 1 $task_id slots] | ||
|
|
||
| # Test with unsorted and adjacent ranges | ||
| set task_id [R 1 CLUSTER MIGRATION IMPORT 7050 7051 7010 7049 7000 7005] | ||
| wait_for_asm_done | ||
| # verify migration is successfully completed on both nodes | ||
| assert_equal "completed" [migration_status 0 $task_id state] | ||
| assert_equal "completed" [migration_status 1 $task_id state] | ||
| # verify slot ranges are merged correctly | ||
| assert_equal "7000-7005 7010-7051" [migration_status 0 $task_id slots] | ||
| assert_equal "7000-7005 7010-7051" [migration_status 1 $task_id slots] | ||
|
|
||
| # Another test with unsorted and adjacent ranges | ||
| set task_id [R 1 CLUSTER MIGRATION IMPORT 7007 7007 7008 7009 7006 7006] | ||
| wait_for_asm_done | ||
| # verify migration is successfully completed on both nodes | ||
| assert_equal "completed" [migration_status 0 $task_id state] | ||
| assert_equal "completed" [migration_status 1 $task_id state] | ||
| # verify slot ranges are merged correctly | ||
| assert_equal "7006-7009" [migration_status 0 $task_id slots] | ||
| assert_equal "7006-7009" [migration_status 1 $task_id slots] | ||
| } | ||
|
|
||
| test "Simple slot migration with write load" { | ||
| # Perform slot migration while traffic is on and verify data consistency. | ||
| # Trimming is disabled on source nodes so, we can compare the dbs after | ||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
slotRangeArrayIsEqualchecks count before mergingThe
num_rangescomparison at line 1877 happens beforeslotRangeArraySortAndMerge()is called at lines 1879-1880. This means two slot range arrays that represent the same set of slots but have different pre-mergenum_rangeswill incorrectly be reported as not equal.For example:
{7000-7001, 7002-7003}→ num_ranges = 2, merges to{7000-7003}(num_ranges = 1){7000-7003}→ num_ranges = 1slotRangeArrayIsEqual(A, B)would return 0 at line 1877 before ever merging, even though both arrays represent slots 7000-7003.The fix is to move the
num_rangescomparison after bothslotRangeArraySortAndMerge()calls.Was this helpful? React with 👍 / 👎