Includes solved 02-worker-pool and 09-safe-counter (mutex, atomic, sharded). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01GbnJj9ZqEx9nUvzkjdwfMq
106 lines
2.0 KiB
Go
106 lines
2.0 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"runtime"
|
|
"sort"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
func gen(vals ...int) <-chan int {
|
|
ch := make(chan int)
|
|
go func() {
|
|
defer close(ch)
|
|
for _, v := range vals {
|
|
ch <- v
|
|
}
|
|
}()
|
|
return ch
|
|
}
|
|
|
|
func TestMergeAllValues(t *testing.T) {
|
|
out := Merge(context.Background(), gen(1, 2, 3), gen(4, 5), gen(6))
|
|
got := []int{}
|
|
done := make(chan struct{})
|
|
go func() {
|
|
for v := range out {
|
|
got = append(got, v)
|
|
}
|
|
close(done)
|
|
}()
|
|
select {
|
|
case <-done:
|
|
case <-time.After(2 * time.Second):
|
|
t.Fatal("output never closed after all inputs closed")
|
|
}
|
|
sort.Ints(got)
|
|
want := []int{1, 2, 3, 4, 5, 6}
|
|
if len(got) != len(want) {
|
|
t.Fatalf("want %v, got %v", want, got)
|
|
}
|
|
for i := range want {
|
|
if got[i] != want[i] {
|
|
t.Fatalf("want %v, got %v", want, got)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestNoInputs(t *testing.T) {
|
|
out := Merge[int](context.Background())
|
|
select {
|
|
case _, ok := <-out:
|
|
if ok {
|
|
t.Fatal("value from zero inputs")
|
|
}
|
|
case <-time.After(time.Second):
|
|
t.Fatal("output not closed with zero inputs")
|
|
}
|
|
}
|
|
|
|
func TestCancellationReleasesGoroutines(t *testing.T) {
|
|
before := runtime.NumGoroutine()
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
// Inputs that never close and keep producing.
|
|
mk := func() <-chan int {
|
|
ch := make(chan int)
|
|
go func() {
|
|
for {
|
|
select {
|
|
case ch <- 1:
|
|
case <-ctx.Done():
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
return ch
|
|
}
|
|
out := Merge(ctx, mk(), mk(), mk())
|
|
|
|
// Read a few values, then walk away AND cancel — forwarders must not
|
|
// stay blocked on `out <- v`.
|
|
for i := 0; i < 5; i++ {
|
|
select {
|
|
case <-out:
|
|
case <-time.After(2 * time.Second):
|
|
cancel()
|
|
t.Fatal("no values arriving on merged channel")
|
|
}
|
|
}
|
|
cancel()
|
|
|
|
deadline := time.After(2 * time.Second)
|
|
for {
|
|
runtime.GC()
|
|
if runtime.NumGoroutine() <= before+2 { // small slack
|
|
return
|
|
}
|
|
select {
|
|
case <-deadline:
|
|
t.Fatalf("goroutine leak after cancel: before=%d now=%d", before, runtime.NumGoroutine())
|
|
case <-time.After(20 * time.Millisecond):
|
|
}
|
|
}
|
|
}
|