I define a Cycle class to handle concurrency task. What I want is run two functions, each in a goroutine, wait their finish and combine their output error together. But I only get one error. The responsibility of each method list below:
Run
- run a function in goroutine, and collect its error
WaitAllDone
- combine all function error together and wait all function finish
Do1, Do2
- test function
import (
"fmt"
"go.uber.org/multierr"
"sync"
"testing"
)
type Cycle struct {
errChan chan error
wg sync.WaitGroup
}
func NewCycle() *Cycle {
return &Cycle{
errChan: make(chan error),
wg: sync.WaitGroup{},
}
}
// run fn and collect its error into error channel
func (c *Cycle) Run(fn func() error) {
c.wg.Add(1)
go func() {
defer c.wg.Done()
if err := fn(); err != nil {
c.errChan <- err
}
}()
}
// wait all fn finish and combine their error together
func (c *Cycle) WaitAllDone() error {
var err error
go func() {
for {
if tmpErr, ok := <-c.errChan; ok {
err = multierr.Append(err, tmpErr)
} else{
break
}
}
}()
c.wg.Wait()
close(c.errChan)
return err
}
func Do1() error {
return fmt.Errorf("ERR1")
}
func Do2() error {
return fmt.Errorf("ERR2")
}
func Test41(t *testing.T) {
c := NewCycle()
c.Run(Do1)
c.Run(Do2)
if err := c.WaitAllDone(); err != nil {
t.Log(err)
}
}
the final t.Log(err)
output ERR1
or ERR2
, but I want it output ERR1 ERR2
. Why it miss one error.
That's because (*Cycle).WaitAllDone
does not wait for the goroutine that collects the errors to finish. If you run the code with the -race
flag, sometimes it could report several DATA RACE errors. Here is one of them:
$ go test -race .
==================
WARNING: DATA RACE
Write at 0x00c0000a0610 by goroutine 10:
m.(*Cycle).WaitAllDone.func1()
/home/zeke/src/temp/76370962/main_test.go:40 +0xb6
Previous read at 0x00c0000a0610 by goroutine 7:
m.(*Cycle).WaitAllDone()
/home/zeke/src/temp/76370962/main_test.go:48 +0x14e
m.Test41()
/home/zeke/src/temp/76370962/main_test.go:63 +0xa4
testing.tRunner()
/snap/go/current/src/testing/testing.go:1576 +0x216
testing.(*T).Run.func1()
/snap/go/current/src/testing/testing.go:1629 +0x47
This change will fix the issue:
func (c *Cycle) WaitAllDone() error {
var err error
+ done := make(chan int)
go func() {
for {
if tmpErr, ok := <-c.errChan; ok {
err = multierr.Append(err, tmpErr)
} else {
break
}
}
+ close(done)
}()
c.wg.Wait()
close(c.errChan)
+ <-done
return err
}
And the for loop can be simplified with the range
clause:
func (c *Cycle) WaitAllDone() error {
var err error
done := make(chan int)
go func() {
for tmpErr := range c.errChan {
err = multierr.Append(err, tmpErr)
}
close(done)
}()
c.wg.Wait()
close(c.errChan)
<-done
return err
}