在多个线程中运行一个函数。

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

Run a function in several threads

问题

我已经实现了一个名为contractGraph的函数，它使用随机收缩算法计算图的最小割。我正在运行它指定的次数，并计算最小割：

  1. minCut := 0
  2. for i := 0; i < totalCount; i++ {
  3. 	_minCut := contractGraph(graph)
  4. 	if minCut == 0 || _minCut < minCut {
  5. 		minCut = _minCut
  6. 	}
  7. }

contractGraph函数执行的计算非常耗费CPU资源，但是程序只使用了我的机器上的一个CPU核心。我想修改它，使得同时有4个contractGraph函数的并行执行，将结果放入通道中并同步读取，然后计算最小值。

我尝试了以下代码：

  1. func worker(graph Graph, i int, workerChan <- chan bool, minCutChan chan <- int) {
  2. 	defer func () { <- workerChan }()
  3. 	min_cut := contractGraph(graph)
  4. 	minCutChan <- min_cut
  5. }
  8. func workerRunner(graph Graph, minCutChan chan int, totalCount int, workerCount int) {
  9. 	workerChan := make(chan bool, workerCount)
  10. 	for i := 0; i < totalCount; i++ {
  11. 		go worker(graph, i, workerChan, minCutChan)
  12. 	}
  13. }
  15. 	minCutChan := make(chan int)
  16. 	go workerRunner(graph, minCutChan, totalCount, 4)
  18. 	// 读取最小割结果
  19. 	minCut := 0
  20. 	for _minCut := range minCutChan {
  21. 		if minCut == 0 || _minCut < minCut {
  22. 			minCut = _minCut
  23. 		}
  24. 	}

但是仍然只使用了一个核心，并且最后出现了以下错误：

  1. fatal error: all goroutines are asleep - deadlock!

而且我不喜欢使用两个通道，我认为应该只需要一个通道来存放结果。

你建议使用哪种模式？

英文:

I have implemented a function contractGraph which calculates a minimal cut of a graph using randomized contraction. I am running it a specified number of times and calculating the minimum cut:

  1. minCut := 0
  2. for i := 0; i &lt; totalCount; i++ {
  3. 	_minCut := contractGraph(graph)
  4. 	if minCut == 0 || _minCut &lt; minCut {
  5. 		minCut = _minCut
  6. 	}
  7. }

contractGraph does CPU intensive calculations, but the program uses only one CPU core on my machine. I want to modify it, so at any time 4 parallel executions of contractGraph happen, the results are put in channel and are read synchronously and the minimum is calculated.

I tried:

  1. func worker(graph Graph, i int, workerChan &lt;- chan bool, minCutChan chan &lt;- int) {
  2. 	defer func () { &lt;- workerChan }()
  3. 	min_cut := contractGraph(graph)
  4. 	minCutChan &lt;- min_cut
  5. }
  8. func workerRunner(graph Graph, minCutChan chan int, totalCount int, workerCount int) {
  9. 	workerChan := make(chan bool, workerCount)
  10. 	for i := 0; i &lt; totalCount; i++ {
  11. 		go worker(graph, i, workerChan, minCutChan)
  12. 	}
  13. }
  15. 	minCutChan := make(chan int)
  16. 	go workerRunner(graph, minCutChan, totalCount, 4)
  18. 	// read the resulting min cuts
  19. 	minCut := 0
  20. 	for _minCut := range minCutChan {
  21. 		if minCut == 0 || _minCut &lt; minCut {
  22. 			minCut = _minCut
  23. 		}
  24. 	}

But still only one core is used and I get at the end:

  1. fatal error: all goroutines are asleep - deadlock!

Also I don't like having to channels, I think it should be possible to have only one channel with the results.

What pattern would you recommend to use?

答案1

得分: 4

你忘记关闭minCutChan，所以main函数陷入了循环范围内，而且所有的goroutine都已经完成了。

为了不使用channel，你可以使用sync.WaitGroup。

编辑：为了处理totalCount，我会使用atomic.AddInt64，请参考下面更新后的示例代码：

你可以在这个编辑后的示例中看到这些修改的效果：http://play.golang.org/p/WyCQrWK5aa

  1. package main
  3. import (
  4. 	"fmt"
  5. 	"sync"
  6. 	"sync/atomic"
  7. )
  9. type Graph struct {
  10. }
  12. func contractGraph(Graph) int { return 0 }
  14. func worker(wg *sync.WaitGroup, graph Graph, i int, minCutChan chan<- int) {
  15. 	defer wg.Done()
  16. 	for {
  17. 		count := atomic.AddInt64(&totalCount, -1)
  18. 		if count < 0 {
  19. 			break
  20. 		}
  21. 		fmt.Println("Worker Iteration", count)
  22. 		min_cut := contractGraph(graph)
  23. 		minCutChan <- min_cut
  24. 	}
  25. }
  27. func workerRunner(graph Graph, minCutChan chan int, workerCount int) {
  28. 	wg := new(sync.WaitGroup)
  29. 	wg.Add(workerCount)
  30. 	for i := 0; i < workerCount; i++ {
  31. 		go worker(wg, graph, i, minCutChan)
  32. 	}
  33. 	wg.Wait()
  34. 	close(minCutChan)
  35. }
  37. var totalCount int64
  39. func main() {
  40. 	workerCount := 4
  41. 	graph := Graph{}
  42. 	totalCount = 100
  43. 	minCutChan := make(chan int, workerCount+1)
  44. 	go workerRunner(graph, minCutChan, workerCount)
  46. 	go func() {
  47. 	}()
  49. 	// 读取最小割
  50. 	minCut := 0
  51. 	for _minCut := range minCutChan {
  52. 		if minCut == 0 || _minCut < minCut {
  53. 			minCut = _minCut
  54. 		}
  55. 	}
  56. 	fmt.Println(minCut)
  57. }

更符合Go风格的做法是在匿名函数中启动worker：

http://play.golang.org/p/nT0uUutQyS

  1. package main
  3. import (
  4. 	"fmt"
  5. 	"sync"
  6. 	"sync/atomic"
  7. )
  9. type Graph struct {
  10. }
  12. func contractGraph(Graph) int { return 0 }
  14. var totalCount int64
  16. func workerRunner(graph Graph, minCutChan chan int, workerCount int) {
  17. 	var wg sync.WaitGroup
  18. 	wg.Add(workerCount)
  19. 	for i := 0; i < workerCount; i++ {
  20. 		go func() {
  21. 			defer wg.Done()
  22. 			for {
  23. 				count := atomic.AddInt64(&totalCount, -1)
  24. 				if count < 0 {
  25. 					break
  26. 				}
  27. 				fmt.Println("Worker Iteration", count)
  29. 				min_cut := contractGraph(graph)
  30. 				minCutChan <- min_cut
  31. 			}
  32. 		}()
  33. 	}
  34. 	wg.Wait()
  35. 	close(minCutChan)
  36. }
  38. func main() {
  39. 	workerCount := 4
  40. 	totalCount = 100
  41. 	graph := Graph{}
  42. 	minCutChan := make(chan int, workerCount+1)
  43. 	go workerRunner(graph, minCutChan, workerCount)
  45. 	// 读取最小割
  46. 	minCut := 0
  47. 	for _minCut := range minCutChan {
  48. 		if minCut == 0 || _minCut < minCut {
  49. 			minCut = _minCut
  50. 		}
  51. 	}
  52. 	fmt.Println(minCut)
  53. }
英文:

You forgot to close the minCutChan so main is stuck into range and all the go routines have completed.

to not use the channel you can use sync.WaitGroup

EDIT: To handle the totalCount I would use atomic.AddInt64 see the new updated examples:

see a working mock example with these edits: http://play.golang.org/p/WyCQrWK5aa

  1. package main
  2. import (
  3. &quot;fmt&quot;
  4. &quot;sync&quot;
  5. &quot;sync/atomic&quot;
  6. )
  7. type Graph struct {
  8. }
  9. func contractGraph(Graph) int { return 0 }
  10. func worker(wg *sync.WaitGroup, graph Graph, i int, minCutChan chan&lt;- int) {
  11. defer wg.Done()
  12. for {
  13. count := atomic.AddInt64(&amp;totalCount, -1) 
  14. if count &lt; 0 {
  15. break
  16. }
  17. fmt.Println(&quot;Worker Iteration&quot;, count)
  18. min_cut := contractGraph(graph)
  19. minCutChan &lt;- min_cut
  20. }
  21. }
  22. func workerRunner(graph Graph, minCutChan chan int, workerCount int) {
  23. wg := new(sync.WaitGroup)
  24. wg.Add(workerCount)
  25. for i := 0; i &lt; workerCount; i++ {
  26. go worker(wg, graph, i, minCutChan)
  27. }
  28. wg.Wait()
  29. close(minCutChan)
  30. }
  31. var totalCount int64
  32. func main() {
  33. workerCount := 4
  34. graph := Graph{}
  35. totalCount = 100
  36. minCutChan := make(chan int, workerCount+1)
  37. go workerRunner(graph, minCutChan, workerCount)
  38. go func() {
  39. }()
  40. // read the resulting min cuts
  41. minCut := 0
  42. for _minCut := range minCutChan {
  43. if minCut == 0 || _minCut &lt; minCut {
  44. minCut = _minCut
  45. }
  46. }
  47. fmt.Println(minCut)
  48. }

even more in go style is to spin the workers inside an anonymous function:

http://play.golang.org/p/nT0uUutQyS

package main

  1. import (
  2. &quot;fmt&quot;
  3. &quot;sync&quot;
  4. &quot;sync/atomic&quot;
  5. )
  6. type Graph struct {
  7. }
  8. func contractGraph(Graph) int { return 0 }
  9. var totalCount int64
  10. func workerRunner(graph Graph, minCutChan chan int, workerCount int) {
  11. var wg sync.WaitGroup
  12. wg.Add(workerCount)
  13. for i := 0; i &lt; workerCount; i++ {
  14. go func() {
  15. defer wg.Done()
  16. for {
  17. count := atomic.AddInt64(&amp;totalCount, -1)
  18. if count &lt; 0 {
  19. break
  20. }
  21. fmt.Println(&quot;Worker Iteration&quot;, count)
  22. min_cut := contractGraph(graph)
  23. minCutChan &lt;- min_cut
  24. }
  25. }()
  26. }
  27. wg.Wait()
  28. close(minCutChan)
  29. }
  30. func main() {
  31. workerCount := 4
  32. totalCount = 100
  33. graph := Graph{}
  34. minCutChan := make(chan int, workerCount+1)
  35. go workerRunner(graph, minCutChan, workerCount)
  36. // read the resulting min cuts
  37. minCut := 0
  38. for _minCut := range minCutChan {
  39. if minCut == 0 || _minCut &lt; minCut {
  40. minCut = _minCut
  41. }
  42. }
  43. fmt.Println(minCut)
  44. }

huangapple
  • 本文由 发表于 2014年6月3日 15:18:22
  • 转载请务必保留本文链接：https://go.coder-hub.com/24009208.html
  • go
匿名

发表评论

匿名网友

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

确定