Go：Java的CyclicBarrier之类的可重用屏障？

Question

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!

Answer 1

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
}