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

huangapple
go评论122阅读模式
英文:

Go: Reusable barrier like Java's CyclicBarrier?

问题

使用Google Go，我正在尝试在图像上同步多个执行迭代滤波的线程。我的代码基本上按照这里的描述工作：

  1. func filter(src *image.Image, dest *image.Image, start, end, runs int, barrier ??) {
  2.     for i:= 0; i < runs; i++ {
  3.         // ... 进行图像处理 ...
  5.         // barrier.Await() 可以在这里使用
  7.         if start == 1 {
  8.             // 第一个线程切换图像以进行下一次迭代步骤
  9.             switchImgs(src, dest)
  10.         }
  11.         
  12.         // 再次使用 barrier.Await()
  13.      }
  14. }
  16. func main() {
  17.     //...
  18.     barrier := sync.BarrierNew(numberOfThreads)
  19.     for i := 0; i < numberOfThreads; i++ {
  20.         go filter(..., barrier)
  21.     }
  22. }

问题是，我需要一个可重用的屏障，类似于Java的CyclicBarrier，将线程数设置为其计数器值。不幸的是，我找到的唯一类似于屏障的实现是sync.WaitGroup。然而，WaitGroup无法原子地重置为其先前的计数器值。它只提供了一个普通的Wait()函数，不会重置计数器值。

有没有任何"Go惯用"的方法来实现我想要的，还是我应该自己实现一个CyclicBarrier？非常感谢你的帮助！

英文:

Using Google Go, I'm trying to sync multiple threads performing an iterative filter on an image. My code basically works like outlined here:

  1. func filter(src *image.Image, dest *image.Image, start, end, runs int, barrier ??) {
  2.     for i:= 0; i &lt; runs; i++ {
  3.         // ... do image manipulation ...
  5.         // barrier.Await() would work here
  7.         if start == 1 {
  8.             // the first thread switches the images for the next iteration step
  9.             switchImgs(src, dest)
  10.         }
  11.         
  12.         // barrier.Await() again
  13.      }
  14. }
  16. func main() {
  17.     //...
  18.     barrier := sync.BarrierNew(numberOfThreads)
  19.     for i := 0; i &lt; numberOfThreads; i++ {
  20.         go filter(..., barrier)
  21.     }

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!

答案1

得分: 0

我不完全理解CyclicBarrier的工作原理，如果我理解错了，请原谅。

一个非常简单的SyncGroup的包装器应该可以完成工作，例如：

  1. type Barrier struct {
  2.     NumOfThreads int
  3.     wg           sync.WaitGroup
  4. }
  6. func NewBarrier(num int) (b *Barrier) {
  7.     b = &Barrier{NumOfThreads: num}
  8.     b.wg.Add(num)
  9.     return
  10. }
  12. func (b *Barrier) Await() {
  13.     b.wg.Wait()
  14.     b.wg.Add(b.NumOfThreads)
  15. }
  17. func (b *Barrier) Done() {
  18.     b.wg.Done()
  19. }
  21. func filter(src *image.Image, dest *image.Image, start, end, runs int, barrier *Barrier) {
  22.     for i := 0; i < runs; i++ {
  23.         // ... 进行图像处理 ...
  24.         // 通过 b.Done() 表示该过滤器已完成
  25.         b.Done()
  26.         b.Await()
  27.         if start == 1 {
  28.             // 第一个线程切换图像以进行下一次迭代步骤
  29.             //switchImgs(src, dest)
  30.         }
  32.         b.Done()
  33.         b.Await()
  34.     }
  35. }
  37. func main() {
  38.     barrier := NewBarrier(5)
  39.     for i := 0; i < barrier.NumOfThreads; i++ {
  40.         go filter(1, barrier)
  41.     }
  42. }

以上是对代码的翻译。

英文:

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:

  1. type Barrier struct {
  2.     NumOfThreads int
  3. 	wg           sync.WaitGroup
  4. }
  6. func NewBarrier(num int) (b *Barrier) {
  7. 	b = &amp;Barrier{NumOfThreads: num}
  8. 	b.wg.Add(num)
  9. 	return
  10. }
  12. func (b *Barrier) Await() {
  13. 	b.wg.Wait()
  14. 	b.wg.Add(b.NumOfThreads)
  15. }
  17. func (b *Barrier) Done() {
  18. 	b.wg.Done()
  19. }
  20. func filter(src *image.Image, dest *image.Image, start, end, runs int, barrier *Barrier) {
  21. 	for i := 0; i &lt; runs; i++ {
  22. 		// ... do image manipulation ...
  23. 		//this filter is done, say so by using b.Done()
  24. 		b.Done()
  25. 		b.Await()
  26. 		if start == 1 {
  27. 			// the first thread switches the images for the next iteration step
  28. 			//switchImgs(src, dest)
  29. 		}
  31. 		b.Done()
  32. 		b.Await()
  33. 	}
  34. }
  36. func main() {
  37. 	barrier := NewBarrier(5)
  38. 	for i := 0; i &lt; barrier.NumOfThreads; i++ {
  39. 		go filter(1, barrier)
  40. 	}
  41. }

答案2

得分: 0

你可以使用sync.Cond来实现CyclicBarrier，参考Java的CyclicBarrier源代码。

以下是一个简化的Go版本的CyclicBarrier（没有超时，没有线程中断）：

  1. type CyclicBarrier struct {
  2.     generation int
  3.     count      int
  4.     parties    int
  5.     trip       *sync.Cond
  6. }
  8. func (b *CyclicBarrier) nextGeneration() {
  9.     // signal completion of last generation
  10.     b.trip.Broadcast()
  11.     b.count = b.parties
  12.     // set up next generation
  13.     b.generation++
  14. }
  16. func (b *CyclicBarrier) Await() {
  17.     b.trip.L.Lock()
  18.     defer b.trip.L.Unlock()
  19.     
  20.     generation := b.generation
  22.     b.count--
  23.     index := b.count
  24.     //println(index)
  26.     if index == 0 {
  27.         b.nextGeneration()
  28.     } else {
  29.         for generation == b.generation {
  30.             //wait for current generation complete
  31.             b.trip.Wait()
  32.         }
  33.     }
  34. }
  36. func NewCyclicBarrier(num int) *CyclicBarrier {
  37.     b := CyclicBarrier{}
  38.     b.count = num
  39.     b.parties = num
  40.     b.trip = sync.NewCond(&sync.Mutex{})
  42.     return &b
  43. }

希望对你有帮助！

英文:

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

  1. type CyclicBarrier struct {
  2. 	generation int
  3. 	count      int
  4. 	parties    int
  5. 	trip       *sync.Cond
  6. }
  8. func (b *CyclicBarrier) nextGeneration() {
  9. 	// signal completion of last generation
  10. 	b.trip.Broadcast()
  11. 	b.count = b.parties
  12. 	// set up next generation
  13. 	b.generation++
  14. }
  16. func (b *CyclicBarrier) Await() {
  17. 	b.trip.L.Lock()
  18. 	defer b.trip.L.Unlock()
  19. 	
  20. 	generation := b.generation
  22. 	b.count--
  23. 	index := b.count
  24. 	//println(index)
  26. 	if index == 0 {
  27. 		b.nextGeneration()
  28. 	} else {
  29. 		for generation == b.generation {
  30. 			//wait for current generation complete
  31. 			b.trip.Wait()
  32. 		}
  33. 	}
  34. }
  36. func NewCyclicBarrier(num int) *CyclicBarrier {
  37. 	b := CyclicBarrier{}
  38. 	b.count = num
  39. 	b.parties = num
  40. 	b.trip = sync.NewCond(&amp;sync.Mutex{})
  42. 	return &amp;b
  43. }

huangapple
  • 本文由 发表于 2014年7月21日 23:41:14
  • 转载请务必保留本文链接：https://go.coder-hub.com/24869114.html
  • go
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定