Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

test: fix TestTaskError and TestMultiError flaky tests (#2779) #2820

Merged
merged 7 commits into from
Sep 22, 2021
Merged

test: fix TestTaskError and TestMultiError flaky tests (#2779) #2820

Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
a5fe2f0
test: fix TestMultiError flaky test
Rustin170506 Sep 9, 2021
f27622a
test: Return after context canceled
Rustin170506 Sep 9, 2021
5c5bb57
test: respect context canceled error
Rustin170506 Sep 9, 2021
19da87a
test: better way to fix it
Rustin170506 Sep 13, 2021
fb8ab04
test: remove useless code
Rustin170506 Sep 13, 2021
ec005e2
Merge branch 'release-5.2' into cherry-pick-2779-to-release-5.2
ti-chi-bot Sep 18, 2021
abb5917
Merge branch 'release-5.2' into cherry-pick-2779-to-release-5.2
ti-chi-bot Sep 22, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 16 additions & 9 deletions pkg/workerpool/pool_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,6 @@ var _ = check.Suite(&workerPoolSuite{})
func (s *workerPoolSuite) TestTaskError(c *check.C) {
defer testleak.AfterTest(c)()
ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
defer cancel()

pool := newDefaultPoolImpl(&defaultHasher{}, 4)
errg, ctx := errgroup.WithContext(ctx)
Expand All @@ -55,23 +54,27 @@ func (s *workerPoolSuite) TestTaskError(c *check.C) {
c.Assert(err, check.ErrorMatches, "test error")
})

errg.Go(func() error {
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
for i := 0; i < 10; i++ {
err := handle.AddEvent(ctx, i)
if err != nil {
c.Assert(err, check.ErrorMatches, ".*ErrWorkerPoolHandleCancelled.*")
return nil
}
}
return nil
})
}()

select {
case <-ctx.Done():
c.FailNow()
case err := <-handle.ErrCh():
c.Assert(err, check.ErrorMatches, "test error")
}
// Only cancel the context after all events have been sent,
// otherwise the event delivery may fail due to context cancellation.
wg.Wait()
cancel()

err := errg.Wait()
Expand Down Expand Up @@ -115,7 +118,6 @@ func (s *workerPoolSuite) TestTimerError(c *check.C) {
func (s *workerPoolSuite) TestMultiError(c *check.C) {
defer testleak.AfterTest(c)()
ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
defer cancel()

pool := newDefaultPoolImpl(&defaultHasher{}, 4)
errg, ctx := errgroup.WithContext(ctx)
Expand All @@ -130,22 +132,27 @@ func (s *workerPoolSuite) TestMultiError(c *check.C) {
return nil
})

errg.Go(func() error {
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
for i := 0; i < 10; i++ {
err := handle.AddEvent(ctx, i)
if err != nil {
c.Assert(err, check.ErrorMatches, ".*ErrWorkerPoolHandleCancelled.*")
}
}
return nil
})
}()

select {
case <-ctx.Done():
c.FailNow()
case err := <-handle.ErrCh():
c.Assert(err, check.ErrorMatches, "test error")
}
// Only cancel the context after all events have been sent,
// otherwise the event delivery may fail due to context cancellation.
wg.Wait()
cancel()

err := errg.Wait()
Expand Down