使用Golang通道时出现不一致的结果

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

inconsistent results using golang channels

问题

我有一个用Go语言编写的任务，目标是从一堆文本文件中获取一个唯一的列表。我使用通道进行了一些并行处理，但现在结果不一致 - 每次使用相同的输入文件，输出的记录数会有5个左右的差异。

我在 Fedora x86_64 上使用 go run process.go | wc -l 进行测试，Go版本为 go1.1.2，处理器为 8 核的 AMD。

以下是代码：

  1. package main
  3. import (
  4.     "fmt"
  5.     "os"
  6.     "io"    
  7.     "encoding/csv"
  8.     "regexp"
  9.     "log"
  10. )
  12. var (
  13.     cleanRe *regexp.Regexp = regexp.MustCompile("[^0-9]+")
  14.     comma rune ='\t'
  15.     fieldsPerRecord=-1
  16. )
  18. func clean(s string) string {
  19.     clean:=cleanRe.ReplaceAllLiteralString(s,"")
  20.     if len(clean)<6 {return ""}
  21.     return clean
  22. }
  24. func uniqueChannel(inputChan chan []string, controlChan chan string) {
  25.     defer func(){controlChan<-"Input digester."}()
  26.     uniq:=make(map[string]map[string]bool)
  27.     i:=0
  28.     for record:= range inputChan {
  29.         i++
  30.         id,v:=record[0],record[1]
  31.         if uniq[id]==nil {
  32.             uniq[id]=make(map[string]bool)
  33.         } else if !uniq[id][v] {
  34.             uniq[id][v]=true
  35.             fmt.Println(id,string(comma),v)
  36.         }
  37.     }
  38.     log.Println("digest ", i)
  39. }
  41. func processFile(fileName string, outputChan chan []string, controlChan chan string) {
  42.     defer func(){controlChan<-fileName}()
  43.     f,err:=os.Open(fileName)
  44.     if err!=nil{log.Fatal(err)}
  45.     r:=csv.NewReader(f)
  46.     r.FieldsPerRecord = fieldsPerRecord
  47.     r.Comma = comma
  49.     //  Process the records
  50.     i:=0
  51.     for record,err:=r.Read();err!=io.EOF;record,err=r.Read() {
  52.         if err!=nil{continue}
  53.         id:=record[0]
  54.         for _,v:=range record[1:] {
  55.             if cleanV:=clean(v);cleanV!=""{
  56.                 i++
  57.                 outputChan<-[]string{id,cleanV}
  58.             }
  59.         }
  60.     }
  61.     log.Println(fileName,i)
  62. }
  65. func main() {
  66.     inputs:=[]string{}
  67.     recordChan:=make(chan []string,100)
  68.     processesLeft:=len(inputs)+1
  69.     controlChan:=make(chan string,processesLeft)
  71.     //  Ingest the inputs
  72.     for _,fName:=range inputs {
  73.         go processFile(fName,recordChan,controlChan)
  74.     }
  76.     //  This is the loop to ensure it's all unique
  77.     go uniqueChannel(recordChan,controlChan)
  79.     //  Make sure all the channels close up
  80.     for processesLeft>0 {
  81.         if processesLeft==1{
  82.             close(recordChan)
  83.         }
  84.         c:=<-controlChan
  85.         log.Println(c)
  86.         processesLeft--
  87.     }
  88.     close(controlChan)
  89. }

看起来它在通道为空之前就关闭了通道。没有关闭机制时，我遇到了死锁问题 - 我已经没有更多的想法了。

英文:

I a task written in Go to get a unique list from a bunch of text files. I put in some parallelization using channels and am having inconsistent results now - a variance of 5 records output/not output each time with the same input files.

The am testing it with go run process.go | wc -l on Fedora x86_64, go1.1.2, 8 core amd.

The code is:

  1. package main
  2. import (
  3. &quot;fmt&quot;
  4. &quot;os&quot;
  5. &quot;io&quot;    
  6. &quot;encoding/csv&quot;
  7. &quot;regexp&quot;
  8. &quot;log&quot;
  9. )
  10. var (
  11. cleanRe *regexp.Regexp = regexp.MustCompile(&quot;[^0-9]+&quot;)
  12. comma rune =&#39;\t&#39;
  13. fieldsPerRecord=-1
  14. )
  15. func clean(s string) string {
  16. clean:=cleanRe.ReplaceAllLiteralString(s,&quot;&quot;)
  17. if len(clean)&lt;6 {return &quot;&quot;}
  18. return clean
  19. }
  20. func uniqueChannel(inputChan chan []string, controlChan chan string) {
  21. defer func(){controlChan&lt;-&quot;Input digester.&quot;}()
  22. uniq:=make(map[string]map[string]bool)
  23. i:=0
  24. for record:= range inputChan {
  25. i++
  26. id,v:=record[0],record[1]
  27. if uniq[id]==nil {
  28. uniq[id]=make(map[string]bool)
  29. } else if !uniq[id][v] {
  30. uniq[id][v]=true
  31. fmt.Println(id,string(comma),v)
  32. }
  33. }
  34. log.Println(&quot;digest &quot;, i)
  35. }
  36. func processFile(fileName string, outputChan chan []string, controlChan chan string) {
  37. defer func(){controlChan&lt;-fileName}()
  38. f,err:=os.Open(fileName)
  39. if err!=nil{log.Fatal(err)}
  40. r:=csv.NewReader(f)
  41. r.FieldsPerRecord = fieldsPerRecord
  42. r.Comma = comma
  43. //  Process the records
  44. i:=0
  45. for record,err:=r.Read();err!=io.EOF;record,err=r.Read() {
  46. if err!=nil{continue}
  47. id:=record[0]
  48. for _,v:=range record[1:] {
  49. if cleanV:=clean(v);cleanV!=&quot;&quot;{
  50. i++
  51. outputChan&lt;-[]string{id,cleanV}
  52. }
  53. }
  54. }
  55. log.Println(fileName,i)
  56. }
  57. func main() {
  58. inputs:=[]string{}
  59. recordChan:=make(chan []string,100)
  60. processesLeft:=len(inputs)+1
  61. controlChan:=make(chan string,processesLeft)
  62. //  Ingest the inputs
  63. for _,fName:=range inputs {
  64. go processFile(fName,recordChan,controlChan)
  65. }
  66. //  This is the loop to ensure it&#39;s all unique
  67. go uniqueChannel(recordChan,controlChan)
  68. //  Make sure all the channels close up
  69. for processesLeft&gt;0 {
  70. if processesLeft==1{
  71. close(recordChan)
  72. }
  73. c:=&lt;-controlChan
  74. log.Println(c)
  75. processesLeft--
  76. }
  77. close(controlChan)
  78. }

It seems like it closes the channel before it's empty and quite. Without the closing mechanism I was getting deadlocks - I'm out of ideas.

答案1

得分: 1

你可以放弃控制通道，使用sync.WaitGroup：

  1. package main
  3. import (
  4. 	"encoding/csv"
  5. 	"fmt"
  6. 	"io"
  7. 	"log"
  8. 	"os"
  9. 	"regexp"
  10. 	"sync"
  11. )
  13. var (
  14. 	cleanRe         *regexp.Regexp = regexp.MustCompile("[^0-9]+")
  15. 	comma           rune           = '\t'
  16. 	fieldsPerRecord                = -1
  17. )
  19. func clean(s string) string {
  20. 	clean := cleanRe.ReplaceAllLiteralString(s, "")
  21. 	if len(clean) < 6 {
  22. 		return ""
  23. 	}
  24. 	return clean
  25. }
  27. func uniqueChannel(inputChan chan []string) {
  28. 	uniq := make(map[string]map[string]bool)
  29. 	i := 0
  30. 	for record := range inputChan {
  31. 		i++
  32. 		id, v := record[0], record[1]
  33. 		if uniq[id] == nil {
  34. 			uniq[id] = make(map[string]bool)
  35. 		} else if !uniq[id][v] {
  36. 			uniq[id][v] = true
  37. 			fmt.Println(id, string(comma), v)
  38. 		}
  39. 	}
  40. 	log.Println("digest ", i)
  41. }
  43. func processFile(fileName string, outputChan chan []string) {
  44. 	f, err := os.Open(fileName)
  45. 	if err != nil {
  46. 		log.Fatal(err)
  47. 	}
  48. 	r := csv.NewReader(f)
  49. 	r.FieldsPerRecord = fieldsPerRecord
  50. 	r.Comma = comma
  52. 	//  处理记录
  53. 	for record, err := r.Read(); err != io.EOF; record, err = r.Read() {
  54. 		if err != nil {
  55. 			continue
  56. 		}
  57. 		id := record[0]
  58. 		for _, v := range record[1:] {
  59. 			if cleanV := clean(v); cleanV != "" {
  60. 				outputChan <- []string{id, cleanV}
  61. 			}
  62. 		}
  63. 	}
  64. }
  66. func main() {
  67. 	inputs := []string{"ex.tsv"}
  68. 	recordChan := make(chan []string)
  70. 	var wg sync.WaitGroup
  71. 	//  输入数据
  72. 	for _, fName := range inputs {
  73. 		wg.Add(1)
  74. 		go func() {
  75. 			processFile(fName, recordChan)
  76. 			wg.Done()
  77. 		}()
  78. 	}
  79. 	go func() {
  80. 		wg.Wait()
  81. 		close(recordChan)
  82. 	}()
  84. 	//  确保数据唯一的循环
  85. 	uniqueChannel(recordChan)
  86. }
英文:

You could ditch the control channel and use a sync.WaitGroup:

  1. package main
  2. import (
  3. &quot;encoding/csv&quot;
  4. &quot;fmt&quot;
  5. &quot;io&quot;
  6. &quot;log&quot;
  7. &quot;os&quot;
  8. &quot;regexp&quot;
  9. &quot;sync&quot;
  10. )
  11. var (
  12. cleanRe         *regexp.Regexp = regexp.MustCompile(&quot;[^0-9]+&quot;)
  13. comma           rune           = &#39;\t&#39;
  14. fieldsPerRecord                = -1
  15. )
  16. func clean(s string) string {
  17. clean := cleanRe.ReplaceAllLiteralString(s, &quot;&quot;)
  18. if len(clean) &lt; 6 {
  19. return &quot;&quot;
  20. }
  21. return clean
  22. }
  23. func uniqueChannel(inputChan chan []string) {
  24. uniq := make(map[string]map[string]bool)
  25. i := 0
  26. for record := range inputChan {
  27. i++
  28. id, v := record[0], record[1]
  29. if uniq[id] == nil {
  30. uniq[id] = make(map[string]bool)
  31. } else if !uniq[id][v] {
  32. uniq[id][v] = true
  33. fmt.Println(id, string(comma), v)
  34. }
  35. }
  36. log.Println(&quot;digest &quot;, i)
  37. }
  38. func processFile(fileName string, outputChan chan []string) {
  39. f, err := os.Open(fileName)
  40. if err != nil {
  41. log.Fatal(err)
  42. }
  43. r := csv.NewReader(f)
  44. r.FieldsPerRecord = fieldsPerRecord
  45. r.Comma = comma
  46. //  Process the records
  47. for record, err := r.Read(); err != io.EOF; record, err = r.Read() {
  48. if err != nil {
  49. continue
  50. }
  51. id := record[0]
  52. for _, v := range record[1:] {
  53. if cleanV := clean(v); cleanV != &quot;&quot; {
  54. outputChan &lt;- []string{id, cleanV}
  55. }
  56. }
  57. }
  58. }
  59. func main() {
  60. inputs := []string{&quot;ex.tsv&quot;}
  61. recordChan := make(chan []string)
  62. var wg sync.WaitGroup
  63. //  Ingest the inputs
  64. for _, fName := range inputs {
  65. wg.Add(1)
  66. go func() {
  67. processFile(fName, recordChan)
  68. wg.Done()
  69. }()
  70. }
  71. go func() {
  72. wg.Wait()
  73. close(recordChan)
  74. }()
  75. //  This is the loop to ensure it&#39;s all unique
  76. uniqueChannel(recordChan)
  77. }

huangapple
  • 本文由 发表于 2013年12月18日 15:39:27
  • 转载请务必保留本文链接：https://go.coder-hub.com/20652221.html
  • channel
  • go
  • goroutine
匿名

发表评论

匿名网友

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

确定