httpdl: drop work-stealing segment pool for one worker per segment

The pool let an idle worker steal the back half of a slower in-flight
segment, behind a mutex that also serialized every advance. At the default
-x 1 it never fired, and it carried the trickiest invariant in the tree
(minSplit >= readBuf keeps an owner's in-flight write out of the stolen tail).

makeSegments already caps the segment count at the connection count, so the
segments map one-to-one onto workers: give each its own goroutine, advance
written with a plain atomic add, and snapshot the fixed slice with no lock.
pool/newPool/acquire/steal/advance and the seg.owned field all go away.

Net -156 lines. Behaviour at the user's flags is unchanged, except a slow
mirror's tail segment is no longer rebalanced onto idle connections.

httpdl/segment_test.go

@@ -215,101 +215,58 @@ func TestParseContentRange(t *testing.T) {
 	}
 }
 
-// assertTiles checks that the pool's segments still cover [0,total) exactly:
-// sorted by start they must be contiguous with no gap and no overlap.
-func assertTiles(t *testing.T, p *pool, total int64) {
+// tilesCover checks that segs cover [0,total) exactly: sorted by start they must
+// be contiguous with no gap and no overlap.
+func tilesCover(t *testing.T, segs []seg, total int64) {
 	t.Helper()
-	p.mu.Lock()
-	type rng struct{ start, end int64 }
-	rs := make([]rng, len(p.segs))
-	for i, s := range p.segs {
-		rs[i] = rng{s.start, s.endOff()}
-	}
-	p.mu.Unlock()
-	sort.Slice(rs, func(i, j int) bool { return rs[i].start < rs[j].start })
+	sorted := append([]seg(nil), segs...)
+	sort.Slice(sorted, func(i, j int) bool { return sorted[i].start < sorted[j].start })
 	var next int64
-	for _, r := range rs {
-		if r.start != next {
-			t.Fatalf("segment gap/overlap: next byte %d, got start %d (ranges %v)", next, r.start, rs)
+	for _, s := range sorted {
+		if s.start != next {
+			t.Fatalf("segment gap/overlap: next byte %d, got start %d", next, s.start)
 		}
-		next = r.end + 1
+		next = s.endOff() + 1
 	}
 	if next != total {
 		t.Fatalf("segments cover %d bytes, want %d", next, total)
 	}
 }
 
-func TestPoolAcquireThenSteal(t *testing.T) {
-	// Sizes are in units of readBuf because that is what newPool floors minSplit
-	// to and what the no-overlap invariant is stated against.
-	const total = 8 * readBuf
-	p := newPool(makeSegments(total, readBuf, 1), readBuf) // one segment [0,total)
-
-	first := p.acquire()
-	if first == nil || first.start != 0 || first.endOff() != total-1 {
-		t.Fatalf("first acquire = %+v, want the whole [0,%d] segment", first, total-1)
-	}
-	if p.acquire() == nil {
-		// nothing fresh left, so this must steal first's back half
-		t.Fatal("second acquire returned nil, want a stolen tail")
-	}
-	// first kept the front half, a tail took the back half; together they still tile.
-	if first.endOff() != total/2-1 {
-		t.Errorf("victim end = %d, want %d after midpoint split", first.endOff(), total/2-1)
-	}
-	assertTiles(t, p, total)
-}
-
-func TestPoolNoStealBelowThreshold(t *testing.T) {
-	// remaining is below 2*minSplit, so the lone segment is not worth splitting
-	// and an idle worker is told there is nothing to do.
-	p := newPool(makeSegments(readBuf+readBuf/2, readBuf, 1), readBuf)
-	if s := p.acquire(); s == nil {
-		t.Fatal("first acquire returned nil, want the only segment")
-	}
-	if s := p.acquire(); s != nil {
-		t.Fatalf("second acquire = %+v, want nil (tail would be below min-split)", s)
-	}
-}
-
-// TestPoolConcurrentCoverage drives the pool the way real workers do — acquire a
-// segment, copy it in small chunks, commit each with p.advance, repeat — across
-// more workers than initial segments so stealing is forced. The coverage array
-// proves the core invariant: every byte is written exactly once, so a steal
-// never overlaps the owner's writes and never leaves a gap.
-func TestPoolConcurrentCoverage(t *testing.T) {
+// TestSegmentsConcurrentCoverage drives the segments the way real workers do —
+// one worker per segment, each copying its range in small chunks and committing
+// with addWritten. The coverage array proves the core invariant: every byte is
+// written exactly once, with no gap or overlap between adjacent segments.
+func TestSegmentsConcurrentCoverage(t *testing.T) {
 	const (
 		total    = int64(64 * readBuf) // 2 MiB
-		minSplit = int64(readBuf)      // steal threshold is 2*this
-		chunk    = int64(readBuf / 4)  // one "read", well below minSplit
-		workers  = 8
+		minSplit = int64(readBuf)
+		chunk    = int64(readBuf / 4) // one "read"
+		conns    = 8
 	)
-	p := newPool(makeSegments(total, minSplit, 1), minSplit) // start with a single segment
+	segs := makeSegments(total, minSplit, conns)
+	if len(segs) < 2 {
+		t.Fatalf("expected several segments, got %d", len(segs))
+	}
 	cover := make([]uint32, total)
 
 	var wg sync.WaitGroup
-	for w := 0; w < workers; w++ {
+	for i := range segs {
 		wg.Add(1)
-		go func() {
+		go func(s *seg) {
 			defer wg.Done()
-			for {
-				s := p.acquire()
-				if s == nil {
-					return
+			for s.remaining() > 0 {
+				n := chunk
+				if s.remaining() < n {
+					n = s.remaining()
 				}
-				for s.remaining() > 0 {
-					n := chunk
-					if s.remaining() < n {
-						n = s.remaining()
-					}
-					off := s.offset()
-					for j := off; j < off+n; j++ {
-						atomic.AddUint32(&cover[j], 1)
-					}
-					p.advance(s, n)
+				off := s.offset()
+				for j := off; j < off+n; j++ {
+					atomic.AddUint32(&cover[j], 1)
 				}
+				s.addWritten(n)
 			}
-		}()
+		}(&segs[i])
 	}
 	wg.Wait()
 
@@ -318,47 +275,46 @@ func TestPoolConcurrentCoverage(t *testing.T) {
 			t.Fatalf("byte %d written %d times, want exactly 1", i, c)
 		}
 	}
-	assertTiles(t, p, total)
-	if len(p.segs) == 1 {
-		t.Error("no stealing happened: still one segment after 8 workers drained it")
-	}
+	tilesCover(t, segs, total)
 }
 
-// TestPoolSnapshotAfterStealRoundTrips checks that a steal which happens before a
-// crash survives the control file: the snapshot records the stolen tail, and a
-// resume rebuilds a segment set that still tiles the file and keeps the bytes
-// already written. This is the persistence path an interrupt-and-resume relies on
-// but cannot deterministically trigger from the outside.
-func TestPoolSnapshotAfterStealRoundTrips(t *testing.T) {
-	const total = 8 * readBuf
-	p := newPool(makeSegments(total, readBuf, 1), readBuf)
-	a := p.acquire()      // the whole file
-	p.advance(a, readBuf) // owner makes some progress, then idles out
-	b := p.acquire()      // steals a's back half
-	if b == nil {
-		t.Fatal("expected a stolen tail")
+// TestSnapshotRoundTrips checks the persistence path an interrupt-and-resume
+// relies on: a snapshot of partially-written segments records each frontier, and
+// segsFromControl rebuilds a set that still tiles the file and keeps the bytes
+// already written.
+func TestSnapshotRoundTrips(t *testing.T) {
+	const (
+		total    = int64(8 * readBuf)
+		minSplit = int64(readBuf)
+	)
+	segs := makeSegments(total, minSplit, 4)
+	if len(segs) < 2 {
+		t.Fatalf("expected several segments, got %d", len(segs))
+	}
+	// Each segment makes some progress, so the snapshot has a real frontier to
+	// carry across the round trip.
+	var want int64
+	for i := range segs {
+		n := int64(readBuf)
+		if n > segs[i].length() {
+			n = segs[i].length()
+		}
+		segs[i].addWritten(n)
+		want += n
 	}
-	p.advance(b, readBuf) // the tail's worker makes progress too
 
-	c := p.snapshot("http://x", total, "", "")
-	if len(c.Segs) != 2 {
-		t.Fatalf("snapshot has %d segments, want 2 (original + stolen tail)", len(c.Segs))
+	c := snapshot("http://x", total, "", "", segs)
+	if len(c.Segs) != len(segs) {
+		t.Fatalf("snapshot has %d segments, want %d", len(c.Segs), len(segs))
 	}
 
 	rebuilt := segsFromControl(&c)
-	sort.Slice(rebuilt, func(i, j int) bool { return rebuilt[i].start < rebuilt[j].start })
-	var next, written int64
+	tilesCover(t, rebuilt, total)
+	var written int64
 	for _, s := range rebuilt {
-		if s.start != next {
-			t.Fatalf("rebuilt gap/overlap: next byte %d, got start %d", next, s.start)
-		}
-		next = s.endOff() + 1
 		written += s.progress()
 	}
-	if next != total {
-		t.Fatalf("rebuilt covers %d bytes, want %d", next, total)
-	}
-	if written != 2*readBuf {
-		t.Errorf("rebuilt written = %d, want %d (bytes must survive the round trip)", written, 2*readBuf)
+	if written != want {
+		t.Errorf("rebuilt written = %d, want %d (bytes must survive the round trip)", written, want)
 	}
 }