Using Google Go, I'm trying to sync multiple threads performing an iterative filter on an image. My code basically works like outlined here:
func filter(src *image.Image, dest *image.Image, start, end, runs int, barrier ??) {
for i:= 0; i < runs; i++ {
// ... do image manipulation ...
// barrier.Await() would work here
if start == 1 {
// the first thread switches the images for the next iteration step
switchImgs(src, dest)
}
// barrier.Await() again
}
}
func main() {
//...
barrier := sync.BarrierNew(numberOfThreads)
for i := 0; i < numberOfThreads; i++ {
go filter(..., barrier)
}
The problem is that I would need a reusable barrier quite like Java's CyclicBarrier, setting the number of threads as its counter value. Unfortunately, the only implementation similar to a barrier I have found is sync.WaitGroup. The WaitGroup however cannot be reset atomically to it's previous counter value. It only offers a normal Wait() function that does not reset the counter value.
Is there any "Go idiomatic" way of achieving what I want or should I rather implement my own CyclicBarrier? Thanks a lot for your help!
You can use sync.Cond to implement CyclicBarrier, see source code of java's CyclicBarrier
Here is a minimized go version of CyclicBarrier (no timeout, no thread interrupts): http://play.golang.org/p/5JSNTm0BLe
type CyclicBarrier struct {
generation int
count int
parties int
trip *sync.Cond
}
func (b *CyclicBarrier) nextGeneration() {
// signal completion of last generation
b.trip.Broadcast()
b.count = b.parties
// set up next generation
b.generation++
}
func (b *CyclicBarrier) Await() {
b.trip.L.Lock()
defer b.trip.L.Unlock()
generation := b.generation
b.count--
index := b.count
//println(index)
if index == 0 {
b.nextGeneration()
} else {
for generation == b.generation {
//wait for current generation complete
b.trip.Wait()
}
}
}
func NewCyclicBarrier(num int) *CyclicBarrier {
b := CyclicBarrier{}
b.count = num
b.parties = num
b.trip = sync.NewCond(&sync.Mutex{})
return &b
}
I don't fully understand how CyclicBarrier works, so excuse me if I'm way off.
A very simple wrapper around SyncGroup should do the job, for example:
type Barrier struct {
NumOfThreads int
wg sync.WaitGroup
}
func NewBarrier(num int) (b *Barrier) {
b = &Barrier{NumOfThreads: num}
b.wg.Add(num)
return
}
func (b *Barrier) Await() {
b.wg.Wait()
b.wg.Add(b.NumOfThreads)
}
func (b *Barrier) Done() {
b.wg.Done()
}
func filter(src *image.Image, dest *image.Image, start, end, runs int, barrier *Barrier) {
for i := 0; i < runs; i++ {
// ... do image manipulation ...
//this filter is done, say so by using b.Done()
b.Done()
b.Await()
if start == 1 {
// the first thread switches the images for the next iteration step
//switchImgs(src, dest)
}
b.Done()
b.Await()
}
}
func main() {
barrier := NewBarrier(5)
for i := 0; i < barrier.NumOfThreads; i++ {
go filter(1, barrier)
}
}