learning/go-patterns/stages/07-cancel-join/bounded_test.go

package main

import (
	"context"
	"errors"
	"sync/atomic"
	"testing"
)

func TestRunFailFastKeepsConcurrencyLimit(t *testing.T) {
	entered := make(chan struct{}, 8)
	release := make(chan struct{})
	var active, peak atomic.Int32
	tasks := make([]Task, 8)
	for i := range tasks {
		tasks[i] = func(context.Context) (string, error) {
			n := active.Add(1)
			for old := peak.Load(); n > old; old = peak.Load() {
				if peak.CompareAndSwap(old, n) {
					break
				}
			}
			entered <- struct{}{}
			<-release
			active.Add(-1)
			return "ok", nil
		}
	}
	done := make(chan struct{})
	var result []string
	var err error
	go func() { result, err = RunFailFast(context.Background(), tasks, 2); close(done) }()
	waitFor(t, entered, "first worker")
	waitFor(t, entered, "second worker")
	close(release)
	waitFor(t, done, "batch")
	if err != nil || peak.Load() > 2 || len(result) != 8 {
		t.Fatalf("err=%v peak=%d result=%v", err, peak.Load(), result)
	}
	for _, value := range result {
		if value != "ok" {
			t.Fatal(result)
		}
	}
}
func TestRunFailFastPreCancelledAndInvalidLimit(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	_, err := RunFailFast(ctx, []Task{func(context.Context) (string, error) { t.Error("task started"); return "", nil }}, 1)
	if !errors.Is(err, context.Canceled) {
		t.Fatal(err)
	}
	if _, err := RunFailFast(context.Background(), nil, 0); err == nil {
		t.Fatal("zero limit accepted")
	}
}