| 196 | } |
| 197 | |
| 198 | func (cluster *Cluster) MigrateSlot(ctx context.Context, slot SlotRange, targetShardIdx int, slotOnly bool) error { |
| 199 | if targetShardIdx < 0 || targetShardIdx >= len(cluster.Shards) { |
| 200 | return consts.ErrIndexOutOfRange |
| 201 | } |
| 202 | sourceShardIdx, err := cluster.findShardIndexBySlot(slot) |
| 203 | if err != nil { |
| 204 | return err |
| 205 | } |
| 206 | if sourceShardIdx == targetShardIdx { |
| 207 | return consts.ErrShardIsSame |
| 208 | } |
| 209 | if slotOnly { |
| 210 | // clear source migrating info to avoid mismatch migrating slot error |
| 211 | cluster.Shards[sourceShardIdx].ClearMigrateState() |
| 212 | cluster.Shards[sourceShardIdx].SlotRanges = RemoveSlotFromSlotRanges(cluster.Shards[sourceShardIdx].SlotRanges, slot) |
| 213 | cluster.Shards[targetShardIdx].SlotRanges = AddSlotToSlotRanges(cluster.Shards[targetShardIdx].SlotRanges, slot) |
| 214 | return nil |
| 215 | } |
| 216 | |
| 217 | if cluster.Shards[sourceShardIdx].IsMigrating() || cluster.Shards[targetShardIdx].IsMigrating() { |
| 218 | return consts.ErrShardSlotIsMigrating |
| 219 | } |
| 220 | // Send the migration command to the source node |
| 221 | sourceMasterNode := cluster.Shards[sourceShardIdx].GetMasterNode() |
| 222 | if sourceMasterNode == nil { |
| 223 | return consts.ErrNotFound |
| 224 | } |
| 225 | targetNodeID := cluster.Shards[targetShardIdx].GetMasterNode().ID() |
| 226 | if err := sourceMasterNode.MigrateSlot(ctx, slot, targetNodeID); err != nil { |
| 227 | return err |
| 228 | } |
| 229 | |
| 230 | // Will start the data migration in the background |
| 231 | cluster.Shards[sourceShardIdx].MigratingSlot = FromSlotRange(slot) |
| 232 | cluster.Shards[sourceShardIdx].TargetShardIndex = targetShardIdx |
| 233 | return nil |
| 234 | } |
| 235 | |
| 236 | func (cluster *Cluster) SetSlot(ctx context.Context, slot int, targetNodeID string) error { |
| 237 | version := cluster.Version.Add(1) |