ec.balance: spread concurrent shard moves over target nodes (#12283)

* ec.balance: spread concurrent shard moves over target nodes

The planner gives every freshly encoded volume the same target order, so
running each volume's moves in plan order sends all concurrent copies to the
same node while the other nodes idle. Pick, among the volumes not already
moving, the pending move whose target receives the fewest moves right now.
One volume still never has two moves at once, and moves of one shard keep
their plan order.

* ec.balance: let a move into a node wait for the move of its volume leaving it

The planner counts the slot an outgoing shard frees as room for a later
incoming shard of the same volume, and the receiver refuses a shard it has no
slot for. Make the spread tests deterministic: hold the first moves until all
have started, and check that errors of several volumes are all returned.

* ec.balance: pin that the spread moves run concurrently and in plan order without parallelism

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)
+	}
+}

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

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)