diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index cfb62a8..8fc30f9 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -2,8 +2,8 @@ name: build on: push: - branches: - tags: + branches: ["**"] + tags: ["**"] pull_request: jobs: @@ -12,10 +12,10 @@ jobs: steps: - name: checkout - uses: actions/checkout@v4 + uses: actions/checkout@v7 - name: set up go - uses: actions/setup-go@v5 + uses: actions/setup-go@v7 with: go-version: "1.22" id: go @@ -30,9 +30,9 @@ jobs: TZ: "America/Chicago" - name: golangci-lint - uses: golangci/golangci-lint-action@v7 + uses: golangci/golangci-lint-action@v9 with: - version: v2.6 + version: latest - name: install goveralls run: go install github.com/mattn/goveralls@latest diff --git a/README.md b/README.md index 1ce406f..f896c2d 100644 --- a/README.md +++ b/README.md @@ -72,7 +72,7 @@ It can work as a regular errgrp.Group or with early termination. It is thread-sa ```go ewg := syncs.NewErrSizedGroup(5, syncs.Preemptive) // error wait group with max size=5, don't try to start more if any error happened for i :=0; i<10; i++ { - ewg.Go(func(ctx context.Context) error { // Go here could be blocked if trying to run >5 at the same time + ewg.Go(func() error { // Go here could be blocked if trying to run >5 at the same time err := doThings(ctx) // only 5 of these will run in parallel return err }) @@ -80,3 +80,11 @@ It can work as a regular errgrp.Group or with early termination. It is thread-sa err := ewg.Wait() ``` +`Wait` returns all the collected errors as `*MultiError`, which implements `Unwrap() []error`, so `errors.Is` and `errors.As` match any of them: + +```go + if err := ewg.Wait(); errors.Is(err, context.Canceled) { + // at least one of the goroutines was canceled + } +``` + diff --git a/errsizedgroup.go b/errsizedgroup.go index d044293..27b1d7b 100644 --- a/errsizedgroup.go +++ b/errsizedgroup.go @@ -2,6 +2,7 @@ package syncs import ( "fmt" + "slices" "strings" "sync" ) @@ -160,5 +161,12 @@ func (m *MultiError) Error() string { func (m *MultiError) Errors() []error { m.lock.Lock() defer m.lock.Unlock() - return m.errors + return slices.Clone(m.errors) +} + +// Unwrap returns all errors collected, allows errors.Is and errors.As to match any of them +func (m *MultiError) Unwrap() []error { + m.lock.Lock() + defer m.lock.Unlock() + return slices.Clone(m.errors) } diff --git a/errsizedgroup_test.go b/errsizedgroup_test.go index 6b5fa13..f8f757d 100644 --- a/errsizedgroup_test.go +++ b/errsizedgroup_test.go @@ -19,8 +19,7 @@ func TestErrorSizedGroup(t *testing.T) { ewg := NewErrSizedGroup(10) var c uint32 - for i := 0; i < 1000; i++ { - i := i + for i := range 1000 { ewg.Go(func() error { time.Sleep(time.Millisecond * 10) atomic.AddUint32(&c, 1) @@ -41,14 +40,26 @@ func TestErrorSizedGroup(t *testing.T) { assert.Equal(t, uint32(1000), c, fmt.Sprintf("%d, not all routines have been executed.", c)) } +// recordMax sets m to v if v is greater than the value m holds already +func recordMax(m *atomic.Int32, v int32) { + for { + cur := m.Load() + if v <= cur || m.CompareAndSwap(cur, v) { + return + } + } +} + func TestErrorSizedGroup_Preemptive(t *testing.T) { ewg := NewErrSizedGroup(10, Preemptive) var c uint32 + base := runtime.NumGoroutine() // count of goroutines not related to the group + var running, maxRunning atomic.Int32 - for i := 0; i < 100; i++ { - i := i + for i := range 100 { ewg.Go(func() error { - assert.True(t, runtime.NumGoroutine() < 20, "goroutines %d", runtime.NumGoroutine()) + defer running.Add(-1) + recordMax(&maxRunning, running.Add(1)) atomic.AddUint32(&c, 1) if i == 10 { return errors.New("err1") @@ -61,9 +72,10 @@ func TestErrorSizedGroup_Preemptive(t *testing.T) { }) } - assert.True(t, runtime.NumGoroutine() <= 20, "goroutines %d", runtime.NumGoroutine()) + assert.LessOrEqual(t, runtime.NumGoroutine(), base+50, "no goroutine spawned per submitted function") err := ewg.Wait() require.NotNil(t, err) + assert.LessOrEqual(t, maxRunning.Load(), int32(10), "no more than the group size running at once") assert.True(t, strings.HasPrefix(err.Error(), "2 error(s) occurred:")) assert.Equal(t, uint32(100), c, fmt.Sprintf("%d, not all routines have been executed.", c)) } @@ -71,19 +83,23 @@ func TestErrorSizedGroup_Preemptive(t *testing.T) { func TestErrorSizedGroup_Discard(t *testing.T) { ewg := NewErrSizedGroup(10, Discard) var c uint32 + base := runtime.NumGoroutine() // count of goroutines not related to the group + var running, maxRunning atomic.Int32 - for i := 0; i < 1000; i++ { + for range 1000 { ewg.Go(func() error { - assert.True(t, runtime.NumGoroutine() < 20, "goroutines %d", runtime.NumGoroutine()) + defer running.Add(-1) + recordMax(&maxRunning, running.Add(1)) atomic.AddUint32(&c, 1) time.Sleep(10 * time.Millisecond) return nil }) } - assert.True(t, runtime.NumGoroutine() <= 20, "goroutines %d", runtime.NumGoroutine()) + assert.LessOrEqual(t, runtime.NumGoroutine(), base+50, "no goroutine spawned per submitted function") err := ewg.Wait() assert.NoError(t, err) + assert.LessOrEqual(t, maxRunning.Load(), int32(10), "no more than the group size running at once") assert.Equal(t, uint32(10), c) } @@ -91,7 +107,7 @@ func TestErrorSizedGroup_NoError(t *testing.T) { ewg := NewErrSizedGroup(10) var c uint32 - for i := 0; i < 1000; i++ { + for range 1000 { ewg.Go(func() error { atomic.AddUint32(&c, 1) return nil @@ -107,8 +123,7 @@ func TestErrorSizedGroup_Term(t *testing.T) { ewg := NewErrSizedGroup(10, TermOnErr) var c uint32 - for i := 0; i < 1000; i++ { - i := i + for i := range 1000 { ewg.Go(func() error { atomic.AddUint32(&c, 1) if i == 100 { @@ -131,8 +146,7 @@ func TestErrorSizedGroup_TermOnErr(t *testing.T) { const N = 1000 const errIndex = 100 // index of a function that will return an error - for i := 0; i < N; i++ { - i := i + for i := range N { ewg.Go(func() error { val := atomic.AddUint32(&c, 1) if i == errIndex || val > uint32(errIndex+1) { @@ -162,8 +176,7 @@ func TestErrorSizedGroup_TermAndPreemptive(t *testing.T) { done := make(chan struct{}) go func() { - for i := 0; i < 1000; i++ { - i := i + for i := range 1000 { ewg.Go(func() error { time.Sleep(10 * time.Millisecond) atomic.AddUint32(&c, 1) @@ -194,7 +207,7 @@ func TestErrorSizedGroup_ConcurrencyLimit(t *testing.T) { maxConcurrentGoroutines := int32(0) ewg := NewErrSizedGroup(5) // Limit of concurrent goroutines set to 5 - for i := 0; i < 100; i++ { + for range 100 { ewg.Go(func() error { atomic.AddInt32(&concurrentGoroutines, 1) defer atomic.AddInt32(&concurrentGoroutines, -1) @@ -216,8 +229,7 @@ func TestErrorSizedGroup_ConcurrencyLimit(t *testing.T) { func TestErrorSizedGroup_MultiError(t *testing.T) { ewg := NewErrSizedGroup(10) - for i := 0; i < 10; i++ { - i := i + for i := range 10 { ewg.Go(func() error { return fmt.Errorf("error from goroutine %d", i) }) @@ -226,7 +238,7 @@ func TestErrorSizedGroup_MultiError(t *testing.T) { err := ewg.Wait() assert.NotNil(t, err) - for i := 0; i < 10; i++ { + for i := range 10 { assert.Contains(t, err.Error(), fmt.Sprintf("error from goroutine %d", i)) } @@ -235,6 +247,27 @@ func TestErrorSizedGroup_MultiError(t *testing.T) { assert.Len(t, merr.Errors(), 10) } +func TestErrorSizedGroup_MultiErrorUnwrap(t *testing.T) { + errFirst := errors.New("first") + errSecond := errors.New("second") + + ewg := NewErrSizedGroup(2) + ewg.Go(func() error { return fmt.Errorf("wrapped: %w", errFirst) }) + ewg.Go(func() error { return errSecond }) + + err := ewg.Wait() + require.Error(t, err) + assert.ErrorIs(t, err, errFirst, "matches the wrapped error of one of the goroutines") + assert.ErrorIs(t, err, errSecond) + assert.NotErrorIs(t, err, errors.New("something else")) + + var merr *MultiError + require.ErrorAs(t, err, &merr) + merr.Errors()[0] = nil // the caller can't affect the collected errors + assert.Len(t, merr.Errors(), 2) + assert.NotNil(t, merr.Errors()[0]) +} + func TestErrorSizedGroup_Cancel(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -243,8 +276,7 @@ func TestErrorSizedGroup_Cancel(t *testing.T) { var c uint32 const N = 1000 - for i := 0; i < N; i++ { - i := i + for i := range N { time.Sleep(1 * time.Millisecond) // prevent all the goroutines to be started at once ewg.Go(func() error { atomic.AddUint32(&c, 1) @@ -272,8 +304,7 @@ func TestErrorSizedGroup_CancelWithPreemptive(t *testing.T) { var c uint32 const N = 1000 - for i := 0; i < N; i++ { - i := i + for i := range N { ewg.Go(func() error { atomic.AddUint32(&c, 1) if i == 100 { @@ -300,7 +331,7 @@ func TestErrorSizedGroup_CancelWithActiveErrors(t *testing.T) { release := make(chan struct{}) var returned atomic.Int32 const N = 100 - for i := 0; i < N; i++ { + for range N { ewg.Go(func() error { <-release returned.Add(1) @@ -329,7 +360,7 @@ func ExampleErrSizedGroup_go() { grp := NewErrSizedGroup(10) var c uint32 - for i := 0; i < 1000; i++ { + for range 1000 { // Go call is non-blocking, like regular go statement grp.Go(func() error { // do some work in 10 goroutines in parallel diff --git a/go.mod b/go.mod index cb24b0d..5e1c4da 100644 --- a/go.mod +++ b/go.mod @@ -2,10 +2,6 @@ module github.com/go-pkgz/syncs go 1.22 -require github.com/stretchr/testify v1.10.0 +require github.com/stretchr/testify v1.12.0 -require ( - github.com/davecgh/go-spew v1.1.1 // indirect - github.com/pmezard/go-difflib v1.0.0 // indirect - gopkg.in/yaml.v3 v3.0.1 // indirect -) +require gopkg.in/yaml.v3 v3.0.1 // indirect diff --git a/go.sum b/go.sum index 713a0b4..054a196 100644 --- a/go.sum +++ b/go.sum @@ -1,9 +1,5 @@ -github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= -github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= -github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= -github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= -github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= -github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/stretchr/testify v1.12.0 h1:K6Mr6jO9JICuend/5xzTM03ydSV3vdNRYAdPSukj8uI= +github.com/stretchr/testify v1.12.0/go.mod h1:bOYBZb5qJ00vPzWfIqBUZPaxK8jWiXc6d3ErP4Ca9Gw= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= diff --git a/sizedgroup_test.go b/sizedgroup_test.go index 581d7e0..b034266 100644 --- a/sizedgroup_test.go +++ b/sizedgroup_test.go @@ -16,7 +16,7 @@ func TestSizedGroup(t *testing.T) { swg := NewSizedGroup(10) var c uint32 - for i := 0; i < 1000; i++ { + for range 1000 { swg.Go(func(ctx context.Context) { time.Sleep(5 * time.Millisecond) atomic.AddUint32(&c, 1) @@ -30,14 +30,15 @@ func TestSizedGroup(t *testing.T) { func TestSizedGroup_Discard(t *testing.T) { swg := NewSizedGroup(10, Preemptive, Discard) var c uint32 + base := runtime.NumGoroutine() // count of goroutines not related to the group - for i := 0; i < 100; i++ { + for range 100 { swg.Go(func(ctx context.Context) { time.Sleep(5 * time.Millisecond) atomic.AddUint32(&c, 1) }) } - assert.True(t, runtime.NumGoroutine() < 15, "goroutines %d", runtime.NumGoroutine()) + assert.LessOrEqual(t, runtime.NumGoroutine(), base+50, "no goroutine spawned per submitted function") swg.Wait() assert.Equal(t, uint32(10), c, fmt.Sprintf("%d, not all routines have been executed", c)) } @@ -45,14 +46,15 @@ func TestSizedGroup_Discard(t *testing.T) { func TestSizedGroup_Preemptive(t *testing.T) { swg := NewSizedGroup(10, Preemptive) var c uint32 + base := runtime.NumGoroutine() // count of goroutines not related to the group - for i := 0; i < 100; i++ { + for range 100 { swg.Go(func(ctx context.Context) { time.Sleep(5 * time.Millisecond) atomic.AddUint32(&c, 1) }) } - assert.True(t, runtime.NumGoroutine() < 15, "goroutines %d", runtime.NumGoroutine()) + assert.LessOrEqual(t, runtime.NumGoroutine(), base+50, "no goroutine spawned per submitted function") swg.Wait() assert.Equal(t, uint32(100), c, fmt.Sprintf("%d, not all routines have been executed", c)) } @@ -63,7 +65,7 @@ func TestSizedGroup_Canceled(t *testing.T) { swg := NewSizedGroup(10, Preemptive, Context(ctx)) var c uint32 - for i := 0; i < 100; i++ { + for range 100 { swg.Go(func(ctx context.Context) { select { case <-ctx.Done(): @@ -83,7 +85,7 @@ func ExampleSizedGroup_go() { grp := NewSizedGroup(10) // create sized waiting group allowing maximum 10 goroutines var c uint32 - for i := 0; i < 1000; i++ { + for range 1000 { grp.Go(func(ctx context.Context) { // Go call is non-blocking, like regular go statement // do some work in 10 goroutines in parallel atomic.AddUint32(&c, 1)