diff --git a/weed/ec/ec_balance_spread_targets_test.go b/weed/ec/ec_balance_spread_targets_test.go new file mode 100644 index 0000000000..bf6b9debff --- /dev/null +++ b/weed/ec/ec_balance_spread_targets_test.go @@ -0,0 +1,259 @@ +package ec + +import ( + "errors" + "fmt" + "slices" + "sync" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding/ecbalancer" +) + +// lockstepMoves is what the planner emits for volumes freshly encoded on +// different nodes of a cluster with one node per rack: every volume sends its +// shard k to the same node k. +func lockstepMoves(volumes, shards int) []ecbalancer.Move { + var moves []ecbalancer.Move + for v := 0; v < volumes; v++ { + for s := 0; s < shards; s++ { + moves = append(moves, ecbalancer.Move{ + VolumeID: uint32(100 + v), + ShardID: s, + SourceNode: fmt.Sprintf("src-%d", v), + TargetNode: fmt.Sprintf("node-%d", s), + Phase: "cross_rack", + }) + } + } + return moves +} + +type moveRecorder struct { + mu sync.Mutex + hold int + held []string + gate chan struct{} + gateTimed bool + sequence []string + receiving map[string]int + moving map[uint32]int + running int + crowded int + ran map[string]int + order map[uint32][]int +} + +// newMoveRecorder holds the first hold moves until all of them have started, +// so the moves an executor starts together are known without relying on timing. +func newMoveRecorder(hold int) *moveRecorder { + return &moveRecorder{ + hold: hold, + gate: make(chan struct{}), + receiving: map[string]int{}, + moving: map[uint32]int{}, + ran: map[string]int{}, + order: map[uint32][]int{}, + } +} + +func (r *moveRecorder) run(m ecbalancer.Move) error { + r.mu.Lock() + if r.receiving[m.TargetNode] > 0 { + r.crowded++ + } + r.receiving[m.TargetNode]++ + r.moving[m.VolumeID]++ + r.running++ + r.ran[fmt.Sprintf("%d.%d->%s", m.VolumeID, m.ShardID, m.TargetNode)]++ + r.order[m.VolumeID] = append(r.order[m.VolumeID], m.ShardID) + r.sequence = append(r.sequence, fmt.Sprintf("%d.%d", m.VolumeID, m.ShardID)) + sameVolume := r.moving[m.VolumeID] + holding := len(r.held) < r.hold + if holding { + r.held = append(r.held, m.TargetNode) + if len(r.held) == r.hold { + close(r.gate) + } + } + r.mu.Unlock() + + if holding { + select { + case <-r.gate: + case <-time.After(5 * time.Second): + r.mu.Lock() + r.gateTimed = true + r.mu.Unlock() + } + } + time.Sleep(10 * time.Millisecond) + + r.mu.Lock() + r.receiving[m.TargetNode]-- + r.moving[m.VolumeID]-- + r.running-- + r.mu.Unlock() + if sameVolume > 1 { + return fmt.Errorf("volume %d had %d moves at once", m.VolumeID, sameVolume) + } + return nil +} + +func TestRunMovesSpreadsConcurrentCopiesOverTargets(t *testing.T) { + moves := lockstepMoves(4, 13) + r := newMoveRecorder(4) + + if err := runMovesSpreadingTargets(4, moves, r.run); err != nil { + t.Fatal(err) + } + + if len(r.ran) != len(moves) { + t.Fatalf("ran %d distinct moves, want %d", len(r.ran), len(moves)) + } + for k, n := range r.ran { + if n != 1 { + t.Fatalf("move %s ran %d times", k, n) + } + } + if r.gateTimed { + t.Fatalf("the first four moves %v did not run at the same time", r.held) + } + if first := map[string]bool{r.held[0]: true, r.held[1]: true, r.held[2]: true, r.held[3]: true}; len(first) != 4 { + t.Fatalf("the first four moves went to %v, want four different nodes", r.held) + } + // Plan order sends the four first copies to node-0 together, and so on: + // 39 of 52 moves start while another copy lands on the same node. Spreading + // leaves only the tail, when a volume's last pending targets are all busy + // (3 to 6 in practice). + if r.crowded > 12 { + t.Fatalf("%d of %d moves started while another copy was landing on the same node", r.crowded, len(moves)) + } +} + +func TestRunMovesKeepsPlanOrderWithoutParallelism(t *testing.T) { + for _, parallelization := range []int{1, 0} { + moves := lockstepMoves(3, 4) + r := newMoveRecorder(0) + + if err := runMovesSpreadingTargets(parallelization, moves, r.run); err != nil { + t.Fatal(err) + } + + var want []string + for _, m := range moves { + want = append(want, fmt.Sprintf("%d.%d", m.VolumeID, m.ShardID)) + } + if fmt.Sprint(r.sequence) != fmt.Sprint(want) { + t.Fatalf("maxParallelization %d ran %v, want plan order %v", parallelization, r.sequence, want) + } + } +} + +func TestRunMovesKeepsOrderOfMovesOfOneShard(t *testing.T) { + moves := []ecbalancer.Move{ + {VolumeID: 8, ShardID: 0, TargetNode: "busy"}, + {VolumeID: 7, ShardID: 3, TargetNode: "busy"}, + {VolumeID: 7, ShardID: 3, TargetNode: "idle"}, + } + var mu sync.Mutex + var seen []string + run := func(m ecbalancer.Move) error { + mu.Lock() + seen = append(seen, fmt.Sprintf("%d.%d->%s", m.VolumeID, m.ShardID, m.TargetNode)) + mu.Unlock() + time.Sleep(5 * time.Millisecond) + return nil + } + + if err := runMovesSpreadingTargets(2, moves, run); err != nil { + t.Fatal(err) + } + + first, second := -1, -1 + for i, s := range seen { + switch s { + case "7.3->busy": + first = i + case "7.3->idle": + second = i + } + } + if first < 0 || second < 0 || first > second { + t.Fatalf("moves of shard 7.3 ran as %v, want ->busy before ->idle", seen) + } +} + +func TestRunMovesStopsOnlyTheFailedVolume(t *testing.T) { + moves := lockstepMoves(2, 3) + boom := errors.New("copy failed") + var mu sync.Mutex + ran := map[uint32][]int{} + run := func(m ecbalancer.Move) error { + mu.Lock() + ran[m.VolumeID] = append(ran[m.VolumeID], m.ShardID) + mu.Unlock() + if m.VolumeID == 100 && m.ShardID == 0 { + return boom + } + return nil + } + + err := runMovesSpreadingTargets(2, moves, run) + + if !errors.Is(err, boom) { + t.Fatalf("error %v, want %v", err, boom) + } + if len(ran[100]) != 1 { + t.Fatalf("failed volume 100 ran shards %v, want only the failed one", ran[100]) + } + if len(ran[101]) != 3 { + t.Fatalf("volume 101 ran shards %v, want all three", ran[101]) + } +} + +func TestRunMovesLetsAMoveIntoANodeWaitForTheMoveLeavingIt(t *testing.T) { + moves := []ecbalancer.Move{ + {VolumeID: 8, ShardID: 0, SourceNode: "a", TargetNode: "busy"}, + {VolumeID: 7, ShardID: 0, SourceNode: "full", TargetNode: "busy"}, + {VolumeID: 7, ShardID: 5, SourceNode: "b", TargetNode: "full"}, + } + var mu sync.Mutex + var seen []string + run := func(m ecbalancer.Move) error { + mu.Lock() + seen = append(seen, fmt.Sprintf("%d.%d->%s", m.VolumeID, m.ShardID, m.TargetNode)) + mu.Unlock() + time.Sleep(5 * time.Millisecond) + return nil + } + + if err := runMovesSpreadingTargets(2, moves, run); err != nil { + t.Fatal(err) + } + + if slices.Index(seen, "7.0->busy") > slices.Index(seen, "7.5->full") { + t.Fatalf("moves ran as %v, want 7.0 to leave node full before 7.5 lands on it", seen) + } +} + +func TestRunMovesJoinsTheErrorsOfEveryFailedVolume(t *testing.T) { + moves := lockstepMoves(3, 2) + first, second := errors.New("volume 100 failed"), errors.New("volume 102 failed") + run := func(m ecbalancer.Move) error { + switch { + case m.VolumeID == 100 && m.ShardID == 1: + return first + case m.VolumeID == 102 && m.ShardID == 0: + return second + } + return nil + } + + err := runMovesSpreadingTargets(3, moves, run) + + if !errors.Is(err, first) || !errors.Is(err, second) { + t.Fatalf("error %v, want both %v and %v", err, first, second) + } +} diff --git a/weed/ec/ec_common.go b/weed/ec/ec_common.go index 8cabfe426a..e7b47dff23 100644 --- a/weed/ec/ec_common.go +++ b/weed/ec/ec_common.go @@ -2,6 +2,7 @@ package ec import ( "context" + "errors" "fmt" "os" "slices" @@ -23,7 +24,6 @@ import ( "github.com/seaweedfs/seaweedfs/weed/storage/super_block" "github.com/seaweedfs/seaweedfs/weed/storage/types" "github.com/seaweedfs/seaweedfs/weed/topology/balancer" - "github.com/seaweedfs/seaweedfs/weed/util" "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" @@ -1258,30 +1258,100 @@ func (ecb *ecBalancer) executePhase(byID map[string]*EcNode, moves []ecbalancer. } return nil } - // Apply mode: parallelize across volumes, but run one volume's moves within a - // phase sequentially. Concurrent moves of the same volume to a node can race - // on its shared .ecx/.ecj/.vif sidecar files. + return runMovesSpreadingTargets(ecb.maxParallelization, moves, func(m ecbalancer.Move) error { + return ecb.executeMove(byID, m) + }) +} + +// runMovesSpreadingTargets runs up to maxParallelization moves at once, never +// two of the same volume: concurrent moves of one volume to a node can race on +// its shared .ecx/.ecj/.vif sidecar files. A move waits for the earlier moves of +// its volume that carry the same shard or leave its target node: the planner +// counts the slot such a move frees as room for the later one, and the receiver +// may refuse a shard it has no slot for. Otherwise a free worker takes, among the +// volumes not already moving, the pending move whose target node receives the +// fewest moves right now. +// +// The planner gives every volume the same target order (each fresh volume's +// shard k goes to the same node), so running each volume's moves in plan order +// sends all concurrent copies to one node while the others sit idle. +// +// Like util.ExecuteParallelTaskGroups, a failed move stops the rest of its +// volume and the other volumes go on; the errors are joined. +func runMovesSpreadingTargets(maxParallelization int, moves []ecbalancer.Move, run func(ecbalancer.Move) error) error { + if maxParallelization < 1 { + maxParallelization = 1 + } var order []uint32 - byVol := make(map[uint32][]ecbalancer.Move) + pending := make(map[uint32][]ecbalancer.Move) for _, m := range moves { - if _, ok := byVol[m.VolumeID]; !ok { + if _, ok := pending[m.VolumeID]; !ok { order = append(order, m.VolumeID) } - byVol[m.VolumeID] = append(byVol[m.VolumeID], m) + pending[m.VolumeID] = append(pending[m.VolumeID], m) } - taskGroups := make([][]util.ErrorWaitGroupTask, 0, len(order)) - for _, vid := range order { - movesForVolume := byVol[vid] - taskGroup := make([]util.ErrorWaitGroupTask, 0, len(movesForVolume)) - for _, move := range movesForVolume { - move := move - taskGroup = append(taskGroup, func() error { - return ecb.executeMove(byID, move) - }) + + type result struct { + move ecbalancer.Move + err error + } + results := make(chan result) + moving := make(map[uint32]bool) + receiving := make(map[string]int) + running := 0 + var errs []error + for { + for running < maxParallelization { + vid, at, ok := pickSpreadMove(order, pending, moving, receiving) + if !ok { + break + } + m := pending[vid][at] + pending[vid] = slices.Delete(pending[vid], at, at+1) + moving[vid] = true + receiving[m.TargetNode]++ + running++ + go func() { + results <- result{m, run(m)} + }() + } + if running == 0 { + return errors.Join(errs...) + } + r := <-results + running-- + moving[r.move.VolumeID] = false + receiving[r.move.TargetNode]-- + if r.err != nil { + errs = append(errs, r.err) + delete(pending, r.move.VolumeID) } - taskGroups = append(taskGroups, taskGroup) } - return util.ExecuteParallelTaskGroups(ecb.maxParallelization, taskGroups) +} + +// pickSpreadMove returns the volume and index of the next move to start, ties +// going to plan order. +func pickSpreadMove(order []uint32, pending map[uint32][]ecbalancer.Move, moving map[uint32]bool, receiving map[string]int) (vid uint32, at int, ok bool) { + best := 0 + for _, v := range order { + if moving[v] { + continue + } + for i, m := range pending[v] { + if slices.ContainsFunc(pending[v][:i], func(e ecbalancer.Move) bool { + return e.ShardID == m.ShardID || e.SourceNode == m.TargetNode + }) { + continue + } + if n := receiving[m.TargetNode]; !ok || n < best { + vid, at, ok, best = v, i, true, n + if n == 0 { + return vid, at, ok + } + } + } + } + return vid, at, ok } // verifyEcShardOnKeepNode confirms the node a dedup move chose to keep actually diff --git a/weed/util/error_wait_group.go b/weed/util/error_wait_group.go index c76bc17aa0..a71186cb19 100644 --- a/weed/util/error_wait_group.go +++ b/weed/util/error_wait_group.go @@ -18,8 +18,7 @@ type ErrorWaitGroup struct { type ErrorWaitGroupTask func() error // ExecuteParallelTaskGroups runs tasks from each group sequentially and runs -// independent groups with up to maxParallelization concurrency. EC balancing -// uses one group per volume because its shard sidecar files are shared; volume +// independent groups with up to maxParallelization concurrency. Volume // balancing uses one group per already-reserved volume move. func ExecuteParallelTaskGroups(maxParallelization int, taskGroups [][]ErrorWaitGroupTask) error { ewg := NewErrorWaitGroup(maxParallelization)