ec placement: spread EC shards evenly across machines, not onto the lowest-id one (#9855)

* ec placement: steer shards to less-loaded machines, not the lowest id EC encode places every volume against one shared topology snapshot (it reserves the shards it assigns so later volumes see reduced capacity), but node selection ranked only by this volume's shard count and broke ties by sorted id. So the lowest-id machine won the first shard of every volume and accumulated far more total shards than the rest -- on a 6-machine cluster the first machines drifted to ~1.5x. Rank eligible nodes by the machine's shards of this volume, then the machine's free capacity, then the node's shards of this volume, then the node's free capacity. Free capacity reflects the load already placed, so ties steer toward the least-loaded machine instead of the lowest id, keeping total EC shards even across machines. * test: ec.balance converges to even per-machine load from a skew Starts machine 10.0.0.1 at 4 shards/volume and the rest at 2, then runs repeated worker-style capped passes; asserts convergence to an even per-machine total (reaches exactly even in ~13 rounds). * reduce comments on the placement fix Trim narration to the non-obvious why. * test: assert convergence and count zero-shard machines Seed the per-machine map with every host so a fully drained machine still registers, and fail explicitly if balance doesn't converge before the round cap.
2026-08-19 13:46:58 +00:00 · 2026-06-07 20:45:17 -07:00
parent 89cbb1c558
commit f9d3105e80
4 changed files with 154 additions and 6 deletions
@@ -1120,6 +1120,15 @@ func countShardsByHost(vk volKey, nodes map[string]*Node) map[string]int {
 	return m
 }

+// freeSlotsByHost sums each machine's free EC shard slots.
+func freeSlotsByHost(nodes map[string]*Node) map[string]int {
+	m := make(map[string]int)
+	for _, node := range nodes {
+		m[node.host] += node.freeSlots
+	}
+	return m
+}
+
 func countShardsByNode(vk volKey, nodes map[string]*Node) map[string]int {
 	m := make(map[string]int)
 	for id, node := range nodes {
@@ -260,6 +260,87 @@ func allBits(n int) erasure_coding.ShardBits {
 	return b
 }

+// TestPlanConvergesToBalancedLoadOverRounds starts skewed -- machine 10.0.0.1 holds
+// 4 shards of every volume, the others 2 -- and runs repeated worker-style balance
+// passes (capped moves per pass). It must converge to an even per-machine shard load.
+func TestPlanConvergesToBalancedLoadOverRounds(t *testing.T) {
+	topo := NewTopology()
+	hosts := []string{"10.0.0.1", "10.0.0.2", "10.0.0.3", "10.0.0.4", "10.0.0.5", "10.0.0.6"}
+	for _, h := range hosts {
+		for _, p := range []string{":8080", ":8081"} {
+			n := topo.AddNode(h+p, "dc1", "dc1:rack1", 4000)
+			n.SetHost(h)
+			n.AddDisk(0, "", 4000, 0)
+		}
+	}
+	add := func(id string, vid uint32, ids ...int) {
+		n := topo.nodes[id]
+		n.AddShards(vid, "c1", 0, bits(ids...))
+		n.disks[0].shardCount += len(ids)
+		n.disks[0].freeSlots -= len(ids)
+		n.freeSlots -= len(ids)
+	}
+	const volumes = 60
+	for v := 0; v < volumes; v++ {
+		vid := uint32(v)
+		add("10.0.0.1:8080", vid, 0, 1) // machine 10.0.0.1 gets 4 shards/volume
+		add("10.0.0.1:8081", vid, 2, 3)
+		add("10.0.0.2:8080", vid, 4)
+		add("10.0.0.2:8081", vid, 5)
+		add("10.0.0.3:8080", vid, 6)
+		add("10.0.0.3:8081", vid, 7)
+		add("10.0.0.4:8080", vid, 8)
+		add("10.0.0.4:8081", vid, 9)
+		add("10.0.0.5:8080", vid, 10)
+		add("10.0.0.5:8081", vid, 11)
+		add("10.0.0.6:8080", vid, 12)
+		add("10.0.0.6:8081", vid, 13)
+	}
+
+	perMachine := func() map[string]int {
+		m := map[string]int{}
+		for _, h := range hosts {
+			m[h] = 0 // seed every host so a fully drained one still counts
+		}
+		for _, n := range topo.nodes {
+			for _, info := range n.shards {
+				m[n.host] += info.shardBits.Count()
+			}
+		}
+		return m
+	}
+	opts := Options{Ratio: ratio(10, 4), GlobalUtilizationBased: true, GlobalMaxMovesPerRack: 8}
+
+	rounds, converged := 0, false
+	for ; rounds < 200; rounds++ {
+		if len(Plan(topo, opts)) == 0 {
+			converged = true
+			break
+		}
+	}
+	if !converged {
+		t.Fatalf("balance did not converge within %d rounds", rounds)
+	}
+
+	pm := perMachine()
+	min, max := 1<<30, 0
+	for _, c := range pm {
+		if c < min {
+			min = c
+		}
+		if c > max {
+			max = c
+		}
+	}
+	t.Logf("converged after %d rounds: %v", rounds, pm)
+	if rounds < 2 {
+		t.Errorf("expected convergence to take multiple rounds (capped moves), took %d", rounds)
+	}
+	if float64(max) > 1.1*float64(min) {
+		t.Errorf("did not converge to balanced load: min=%d max=%d %v", min, max, pm)
+	}
+}
+
 func TestGlobalImbalanceMovesFromFullToEmpty(t *testing.T) {
 	topo := NewTopology()
 	n1 := topo.AddNode("node1", "dc1", "dc1:rack1", 5)
@@ -389,12 +389,15 @@ func rackHasFreeDisk(r *rack, eligible func(*disk) bool) bool {
 // eligible disk. FromActiveTopology keeps all disk types/tags in the snapshot, so
 // without this a node with free volume slots but no eligible disk could be chosen.
 //
-// Among eligible nodes it prefers the one whose machine holds the fewest shards of
-// the volume (tie-broken by the node's own count), spreading shards across machines.
+// Among eligible nodes it ranks by fewest shards of the volume per machine, then per
+// node, with free capacity breaking ties. The free-capacity tie-break (not sorted id)
+// keeps the lowest-id machine from winning every volume's first shard against the
+// shared encode snapshot and piling up load.
 func pickNodeInRackEligible(r *rack, vk volKey, rp *super_block.ReplicaPlacement, eligible func(*disk) bool) *Node {
 	machineShards := countShardsByHost(vk, r.nodes)
+	machineFree := freeSlotsByHost(r.nodes)
 	var best *Node
-	bestMachineCount, bestNodeCount := -1, -1
+	var bestMCount, bestMFree, bestNCount, bestNFree int
 	for _, id := range sortedNodeKeys(r.nodes) {
 		node := r.nodes[id]
 		if node.freeSlots <= 0 {
@@ -407,9 +410,22 @@ func pickNodeInRackEligible(r *rack, vk volKey, rp *super_block.ReplicaPlacement
 		if rp != nil && rp.SameRackCount > 0 && count >= rp.SameRackCount {
 			continue
 		}
-		mCount := machineShards[node.host]
-		if best == nil || mCount < bestMachineCount || (mCount == bestMachineCount && count < bestNodeCount) {
-			best, bestMachineCount, bestNodeCount = node, mCount, count
+		mCount, mFree := machineShards[node.host], machineFree[node.host]
+		better := false
+		switch {
+		case best == nil:
+			better = true
+		case mCount != bestMCount:
+			better = mCount < bestMCount
+		case mFree != bestMFree:
+			better = mFree > bestMFree
+		case count != bestNCount:
+			better = count < bestNCount
+		default:
+			better = node.freeSlots > bestNFree
+		}
+		if better {
+			best, bestMCount, bestMFree, bestNCount, bestNFree = node, mCount, mFree, count, node.freeSlots
 		}
 	}
 	return best
@@ -73,6 +73,48 @@ func TestPlaceStrictSpreadAndCaps(t *testing.T) {
 	}
 }

+// TestPlaceDistributesEvenlyAcrossManyVolumes: placing many volumes against the shared
+// encode snapshot must not pile shards onto the sorted-first machine.
+func TestPlaceDistributesEvenlyAcrossManyVolumes(t *testing.T) {
+	topo := NewTopology()
+	hosts := []string{"10.0.0.1", "10.0.0.2", "10.0.0.3", "10.0.0.4", "10.0.0.5", "10.0.0.6"}
+	for _, h := range hosts {
+		for _, port := range []string{":8080", ":8081"} {
+			n := topo.AddNode(h+port, "dc1", "dc1:rack1", 1000)
+			n.SetHost(h)
+			n.AddDisk(0, "", 1000, 0)
+		}
+	}
+
+	const volumes = 120
+	for v := 0; v < volumes; v++ {
+		if _, err := topo.Place(uint32(v), "c1", allShards(), Constraints{}, PlaceDurabilityFirst); err != nil {
+			t.Fatalf("Place volume %d: %v", v, err)
+		}
+	}
+
+	perMachine := map[string]int{}
+	for _, n := range topo.nodes {
+		for _, d := range n.disks {
+			perMachine[n.host] += d.shardCount
+		}
+	}
+	min, max := 1<<30, 0
+	for _, c := range perMachine {
+		if c < min {
+			min = c
+		}
+		if c > max {
+			max = c
+		}
+	}
+	// Even distribution is 120*14/6 = 280 shards/machine. Catch a lopsided pileup
+	// (the production symptom was the first machine at ~1.7x the rest).
+	if float64(max) > 1.25*float64(min) {
+		t.Errorf("EC shards piled unevenly across machines: min=%d max=%d %v", min, max, perMachine)
+	}
+}
+
 // TestPlaceSpreadsAcrossMachines: one rack runs two physical machines (each with
 // four volume servers). Placing a 10+4 volume must spread its shards across both
 // machines so no single machine holds more than ceil(14/2)=7, otherwise losing one