Migrate all keys in the given slot from source to target.*/
| 3365 | |
| 3366 | /* Migrate all keys in the given slot from source to target.*/ |
| 3367 | static int clusterManagerMigrateKeysInSlot(clusterManagerNode *source, |
| 3368 | clusterManagerNode *target, |
| 3369 | int slot, int timeout, |
| 3370 | int pipeline, int verbose, |
| 3371 | char **err) |
| 3372 | { |
| 3373 | int success = 1; |
| 3374 | int do_fix = config.cluster_manager_command.flags & |
| 3375 | CLUSTER_MANAGER_CMD_FLAG_FIX; |
| 3376 | int do_replace = config.cluster_manager_command.flags & |
| 3377 | CLUSTER_MANAGER_CMD_FLAG_REPLACE; |
| 3378 | while (1) { |
| 3379 | char *dots = NULL; |
| 3380 | redisReply *reply = NULL, *migrate_reply = NULL; |
| 3381 | reply = CLUSTER_MANAGER_COMMAND(source, "CLUSTER " |
| 3382 | "GETKEYSINSLOT %d %d", slot, |
| 3383 | pipeline); |
| 3384 | success = (reply != NULL); |
| 3385 | if (!success) return 0; |
| 3386 | if (reply->type == REDIS_REPLY_ERROR) { |
| 3387 | success = 0; |
| 3388 | if (err != NULL) { |
| 3389 | *err = zmalloc((reply->len + 1) * sizeof(char), MALLOC_LOCAL); |
| 3390 | strcpy(*err, reply->str); |
| 3391 | CLUSTER_MANAGER_PRINT_REPLY_ERROR(source, *err); |
| 3392 | } |
| 3393 | goto next; |
| 3394 | } |
| 3395 | assert(reply->type == REDIS_REPLY_ARRAY); |
| 3396 | size_t count = reply->elements; |
| 3397 | if (count == 0) { |
| 3398 | freeReplyObject(reply); |
| 3399 | break; |
| 3400 | } |
| 3401 | if (verbose) dots = zmalloc((count+1) * sizeof(char), MALLOC_LOCAL); |
| 3402 | /* Calling MIGRATE command. */ |
| 3403 | migrate_reply = clusterManagerMigrateKeysInReply(source, target, |
| 3404 | reply, 0, timeout, |
| 3405 | dots); |
| 3406 | if (migrate_reply == NULL) goto next; |
| 3407 | if (migrate_reply->type == REDIS_REPLY_ERROR) { |
| 3408 | int is_busy = strstr(migrate_reply->str, "BUSYKEY") != NULL; |
| 3409 | int not_served = 0; |
| 3410 | if (!is_busy) { |
| 3411 | /* Check if the slot is unassigned (not served) in the |
| 3412 | * source node's configuration. */ |
| 3413 | char *get_owner_err = NULL; |
| 3414 | clusterManagerNode *served_by = |
| 3415 | clusterManagerGetSlotOwner(source, slot, &get_owner_err); |
| 3416 | if (!served_by) { |
| 3417 | if (get_owner_err == NULL) not_served = 1; |
| 3418 | else { |
| 3419 | CLUSTER_MANAGER_PRINT_REPLY_ERROR(source, |
| 3420 | get_owner_err); |
| 3421 | zfree(get_owner_err); |
| 3422 | } |
| 3423 | } |
| 3424 | } |
no test coverage detected