diff --git a/errsizedgroup.go b/errsizedgroup.go index 2c31417..d044293 100644 --- a/errsizedgroup.go +++ b/errsizedgroup.go @@ -15,7 +15,6 @@ type ErrSizedGroup struct { sema Locker err *MultiError - errLock sync.RWMutex errOnce sync.Once } @@ -88,8 +87,6 @@ func (g *ErrSizedGroup) Go(f func() error) { if !g.termOnError { return false } - g.errLock.RLock() - defer g.errLock.RUnlock() return g.err.ErrorOrNil() != nil } @@ -109,9 +106,7 @@ func (g *ErrSizedGroup) Go(f func() error) { } if err := f(); err != nil { - g.errLock.Lock() - g.err = g.err.append(err) - g.errLock.Unlock() + g.err.append(err) } }() } @@ -129,11 +124,10 @@ type MultiError struct { lock sync.Mutex } -func (m *MultiError) append(err error) *MultiError { +func (m *MultiError) append(err error) { m.lock.Lock() m.errors = append(m.errors, err) m.lock.Unlock() - return m } // ErrorOrNil returns nil if no errors or multierror if errors occurred diff --git a/errsizedgroup_test.go b/errsizedgroup_test.go index 89f8988..6b5fa13 100644 --- a/errsizedgroup_test.go +++ b/errsizedgroup_test.go @@ -260,7 +260,8 @@ func TestErrorSizedGroup_Cancel(t *testing.T) { require.EqualError(t, err, "1 error(s) occurred: [0] {context canceled}") assert.ErrorIs(t, ctx.Err(), context.Canceled, ctx.Err()) t.Logf("completed: %d", c) - require.LessOrEqual(t, c, uint32(110), "some of goroutines has to be terminated early") + // 100 submitted before the cancellation, up to 10 more running and one waiting for the semaphore + require.LessOrEqual(t, c, uint32(120), "some of goroutines has to be terminated early") } func TestErrorSizedGroup_CancelWithPreemptive(t *testing.T) { @@ -287,7 +288,38 @@ func TestErrorSizedGroup_CancelWithPreemptive(t *testing.T) { require.EqualError(t, err, "1 error(s) occurred: [0] {context canceled}") assert.ErrorIs(t, ctx.Err(), context.Canceled, ctx.Err()) t.Logf("completed: %d", c) - require.LessOrEqual(t, c, uint32(110), "some of goroutines has to be terminated early") + // 100 submitted before the cancellation, up to 10 more running and one waiting for the semaphore + require.LessOrEqual(t, c, uint32(120), "some of goroutines has to be terminated early") +} + +func TestErrorSizedGroup_CancelWithActiveErrors(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + ewg := NewErrSizedGroup(4, Context(ctx)) + + release := make(chan struct{}) + var returned atomic.Int32 + const N = 100 + for i := 0; i < N; i++ { + ewg.Go(func() error { + <-release + returned.Add(1) + return errors.New("failed") + }) + } + + cancel() + close(release) + for returned.Load() < N/2 { // make sure workers record their errors while the canceled call records ctx.Err() + runtime.Gosched() + } + ewg.Go(func() error { return nil }) + + err := ewg.Wait() + require.Error(t, err) + var merr *MultiError + require.True(t, errors.As(err, &merr)) + assert.Len(t, merr.Errors(), N+1) } // illustrates the use of a SizedGroup for concurrent, limited execution of goroutines.