运行for循环中的goroutine时，出现“在关闭的通道上发送”恐慌。

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

Panic: Send on a closed channel when running go routine in foor loop

问题

我正在尝试制作一个并发版本的grep程序。该程序遍历目录/子目录，并返回与提供的模式匹配的任何字符串。

我正在尝试并发地运行文件搜索，一旦我获得所有要搜索的文件（参见searchPaths函数）。最初我遇到了以下错误：

fatal error: all goroutines are asleep - deadlock

直到我在searchPaths的末尾添加了close(out)，现在它返回：

Panic: Send on a closed channel when running go routine in foor loop

我试图实现类似于以下链接中的内容：

https://go.dev/blog/pipelines#fan-out-fan-in

是否是我在错误的位置关闭了通道？

  1. package main
  3. import (
  4. 	"fmt"
  5. 	"io/fs"
  6. 	"io/ioutil"
  7. 	"log"
  8. 	"os"
  9. 	"path/filepath"
  10. 	"strings"
  11. 	"sync"
  12. )
  14. type SearchResult struct {
  15. 	line       string
  16. 	lineNumber int
  17. }
  19. type Display struct {
  20. 	filePath string
  21. 	SearchResult
  22. }
  24. var wg sync.WaitGroup
  26. func (d Display) PrettyPrint() {
  27. 	fmt.Printf("Line Number: %v\nFilePath: %v\nLine: %v\n\n", d.lineNumber, d.filePath, d.line)
  28. }
  30. func searchLine(pattern string, line string, lineNumber int) (SearchResult, bool) {
  31. 	if strings.Contains(line, pattern) {
  32. 		return SearchResult{lineNumber: lineNumber + 1, line: line}, true
  33. 	}
  34. 	return SearchResult{}, false
  35. }
  37. func splitIntoLines(file string) []string {
  38. 	lines := strings.Split(file, "\n")
  39. 	return lines
  40. }
  42. func fileFromPath(path string) string {
  43. 	fileContent, err := ioutil.ReadFile(path)
  45. 	if err != nil {
  46. 		log.Fatal(err)
  47. 	}
  49. 	return string(fileContent)
  50. }
  52. func getRecursiveFilePaths(inputDir string) []string {
  53. 	var paths []string
  54. 	err := filepath.Walk(inputDir, func(path string, info fs.FileInfo, err error) error {
  55. 		if err != nil {
  56. 			fmt.Printf("prevent panic by handling failure accessing a path %q: %v\n", path, err)
  57. 			return err
  58. 		}
  59. 		if !info.IsDir() {
  60. 			paths = append(paths, path)
  61. 		}
  62. 		return nil
  63. 	})
  64. 	if err != nil {
  65. 		fmt.Printf("Error walking the path %q: %v\n", inputDir, err)
  66. 	}
  67. 	return paths
  68. }
  70. func searchPaths(paths []string, pattern string) <-chan Display {
  71. 	out := make(chan Display)
  73. 	for _, path := range paths {
  74. 		wg.Add(1)
  75. 		go func() {
  76. 			defer wg.Done()
  77. 			for _, display := range searchFile(path, pattern) {
  78. 				out <- display
  79. 			}
  80. 		}()
  81. 	}
  82. 	close(out)
  83. 	return out
  84. }
  86. func searchFile(path string, pattern string) []Display {
  87. 	var out []Display
  88. 	input := fileFromPath(path)
  89. 	lines := splitIntoLines(input)
  90. 	for index, line := range lines {
  91. 		if searchResult, ok := searchLine(pattern, line, index); ok {
  92. 			out = append(out, Display{path, searchResult})
  93. 		}
  94. 	}
  95. 	return out
  96. }
  98. func main() {
  99. 	pattern := os.Args[1]
  100. 	dirPath := os.Args[2]
  102. 	paths := getRecursiveFilePaths(dirPath)
  104. 	out := searchPaths(paths, pattern)
  105. 	wg.Wait()
  106. 	for d := range out {
  107. 		d.PrettyPrint()
  108. 	}
  110. }
英文:

I'm attempting to make a concurrent version of grep. The program walks directories/subdirectories and returns back any matching strings to a provided pattern.

I am attempting to run the file searching concurrently, once I have all the files to search (see searchPaths function). Originally I was getting:

fatal error: all goroutines are asleep - deadlock

Until I added the close(out) at the end of searchPaths, to which it now returns:

Panic: Send on a closed channel when running go routine in foor loop

I am attempting to implement something similar to:

https://go.dev/blog/pipelines#fan-out-fan-in

Is it the case that I am closing the channel at the wrong point?

  1. package main
  2. import (
  3. &quot;fmt&quot;
  4. &quot;io/fs&quot;
  5. &quot;io/ioutil&quot;
  6. &quot;log&quot;
  7. &quot;os&quot;
  8. &quot;path/filepath&quot;
  9. &quot;strings&quot;
  10. &quot;sync&quot;
  11. )
  12. type SearchResult struct {
  13. line       string
  14. lineNumber int
  15. }
  16. type Display struct {
  17. filePath string
  18. SearchResult
  19. }
  20. var wg sync.WaitGroup
  21. func (d Display) PrettyPrint() {
  22. fmt.Printf(&quot;Line Number: %v\nFilePath: %v\nLine: %v\n\n&quot;, d.lineNumber, d.filePath, d.line)
  23. }
  24. func searchLine(pattern string, line string, lineNumber int) (SearchResult, bool) {
  25. if strings.Contains(line, pattern) {
  26. return SearchResult{lineNumber: lineNumber + 1, line: line}, true
  27. }
  28. return SearchResult{}, false
  29. }
  30. func splitIntoLines(file string) []string {
  31. lines := strings.Split(file, &quot;\n&quot;)
  32. return lines
  33. }
  34. func fileFromPath(path string) string {
  35. fileContent, err := ioutil.ReadFile(path)
  36. if err != nil {
  37. log.Fatal(err)
  38. }
  39. return string(fileContent)
  40. }
  41. func getRecursiveFilePaths(inputDir string) []string {
  42. var paths []string
  43. err := filepath.Walk(inputDir, func(path string, info fs.FileInfo, err error) error {
  44. if err != nil {
  45. fmt.Printf(&quot;prevent panic by handling failure accessing a path %q: %v\n&quot;, path, err)
  46. return err
  47. }
  48. if !info.IsDir() {
  49. paths = append(paths, path)
  50. }
  51. return nil
  52. })
  53. if err != nil {
  54. fmt.Printf(&quot;Error walking the path %q: %v\n&quot;, inputDir, err)
  55. }
  56. return paths
  57. }
  58. func searchPaths(paths []string, pattern string) &lt;-chan Display {
  59. out := make(chan Display)
  60. for _, path := range paths {
  61. wg.Add(1)
  62. go func() {
  63. defer wg.Done()
  64. for _, display := range searchFile(path, pattern) {
  65. out &lt;- display
  66. }
  67. }()
  68. }
  69. close(out)
  70. return out
  71. }
  72. func searchFile(path string, pattern string) []Display {
  73. var out []Display
  74. input := fileFromPath(path)
  75. lines := splitIntoLines(input)
  76. for index, line := range lines {
  77. if searchResult, ok := searchLine(pattern, line, index); ok {
  78. out = append(out, Display{path, searchResult})
  79. }
  80. }
  81. return out
  82. }
  83. func main() {
  84. pattern := os.Args[1]
  85. dirPath := os.Args[2]
  86. paths := getRecursiveFilePaths(dirPath)
  87. out := searchPaths(paths, pattern)
  88. wg.Wait()
  89. for d := range out {
  90. d.PrettyPrint()
  91. }
  92. }

答案1

得分: 0

这段代码的两个主要问题是：

  1. 你需要在wg.Wait()完成后才关闭通道。你可以在一个单独的goroutine中完成这个操作，如下所示：
  1. go func() {
  2.     wg.Wait()
  3.     close(out)
  4. }()
  1. searchPaths函数中的path变量在for循环逻辑中被多次重新赋值，直接在goroutine中使用它是不好的做法。更好的方法是将其作为参数传递。
  1. go func(p string, w *sync.WaitGroup) {
  2.     defer w.Done()
  3.     for _, display := range searchFile(p, pattern) {
  4.         out <- display
  5.     }
  6. }(path, &wg)

修改后的代码如下所示：

  1. package main
  3. import (
  4.     "fmt"
  5.     "io/fs"
  6.     "io/ioutil"
  7.     "log"
  8.     "os"
  9.     "path/filepath"
  10.     "strings"
  11.     "sync"
  12. )
  14. type SearchResult struct {
  15.     line       string
  16.     lineNumber int
  17. }
  19. type Display struct {
  20.     filePath string
  21.     SearchResult
  22. }
  24. var wg sync.WaitGroup
  26. func (d Display) PrettyPrint() {
  27.     fmt.Printf("Line Number: %v\nFilePath: %v\nLine: %v\n\n", d.lineNumber, d.filePath, d.line)
  28. }
  30. func searchLine(pattern string, line string, lineNumber int) (SearchResult, bool) {
  31.     if strings.Contains(line, pattern) {
  32.         return SearchResult{lineNumber: lineNumber + 1, line: line}, true
  33.     }
  34.     return SearchResult{}, false
  35. }
  37. func splitIntoLines(file string) []string {
  38.     lines := strings.Split(file, "\n")
  39.     return lines
  40. }
  42. func fileFromPath(path string) string {
  43.     fileContent, err := ioutil.ReadFile(path)
  45.     if err != nil {
  46.         log.Fatal(err)
  47.     }
  49.     return string(fileContent)
  50. }
  52. func getRecursiveFilePaths(inputDir string) []string {
  53.     var paths []string
  54.     err := filepath.Walk(inputDir, func(path string, info fs.FileInfo, err error) error {
  55.         if err != nil {
  56.             fmt.Printf("prevent panic by handling failure accessing a path %q: %v\n", path, err)
  57.             return err
  58.         }
  59.         if !info.IsDir() {
  60.             paths = append(paths, path)
  61.         }
  62.         return nil
  63.     })
  64.     if err != nil {
  65.         fmt.Printf("Error walking the path %q: %v\n", inputDir, err)
  66.     }
  67.     return paths
  68. }
  70. func searchPaths(paths []string, pattern string) chan Display {
  71.     out := make(chan Display)
  72.     for _, path := range paths {
  73.         wg.Add(1)
  74.         go func(p string, w *sync.WaitGroup) {
  75.             defer w.Done()
  76.             for _, display := range searchFile(p, pattern) {
  77.                 out <- display
  78.             }
  79.         }(path, &wg)
  80.     }
  81.     return out
  82. }
  84. func searchFile(path string, pattern string) []Display {
  85.     var out []Display
  86.     input := fileFromPath(path)
  87.     lines := splitIntoLines(input)
  88.     for index, line := range lines {
  89.         if searchResult, ok := searchLine(pattern, line, index); ok {
  90.             out = append(out, Display{path, searchResult})
  91.         }
  92.     }
  93.     return out
  94. }
  96. func main() {
  97.     pattern := os.Args[1]
  98.     dirPath := os.Args[2]
  100.     paths := getRecursiveFilePaths(dirPath)
  102.     out := searchPaths(paths, pattern)
  104.     go func() {
  105.         wg.Wait()
  106.         close(out)
  107.     }()
  109.     count := 0
  110.     for d := range out {
  111.         fmt.Println(count)
  112.         d.PrettyPrint()
  113.         count += 1
  114.     }
  115. }
英文:

2 main issues with this code were

  1. you need to close the channel only after wg.Wait() completes. you can do this in a seperate goroutine as shown below
  2. as the path var in searchPaths func is reassigned multiple times as part of the for loop logic, it is not a good practice to use that var directly in the goroutines, a better approach will be to pass it as an argument.
  1. package main
  2. import (
  3. &quot;fmt&quot;
  4. &quot;io/fs&quot;
  5. &quot;io/ioutil&quot;
  6. &quot;log&quot;
  7. &quot;os&quot;
  8. &quot;path/filepath&quot;
  9. &quot;strings&quot;
  10. &quot;sync&quot;
  11. )
  12. type SearchResult struct {
  13. line       string
  14. lineNumber int
  15. }
  16. type Display struct {
  17. filePath string
  18. SearchResult
  19. }
  20. var wg sync.WaitGroup
  21. func (d Display) PrettyPrint() {
  22. fmt.Printf(&quot;Line Number: %v\nFilePath: %v\nLine: %v\n\n&quot;, d.lineNumber, d.filePath, d.line)
  23. }
  24. func searchLine(pattern string, line string, lineNumber int) (SearchResult, bool) {
  25. if strings.Contains(line, pattern) {
  26. return SearchResult{lineNumber: lineNumber + 1, line: line}, true
  27. }
  28. return SearchResult{}, false
  29. }
  30. func splitIntoLines(file string) []string {
  31. lines := strings.Split(file, &quot;\n&quot;)
  32. return lines
  33. }
  34. func fileFromPath(path string) string {
  35. fileContent, err := ioutil.ReadFile(path)
  36. if err != nil {
  37. log.Fatal(err)
  38. }
  39. return string(fileContent)
  40. }
  41. func getRecursiveFilePaths(inputDir string) []string {
  42. var paths []string
  43. err := filepath.Walk(inputDir, func(path string, info fs.FileInfo, err error) error {
  44. if err != nil {
  45. fmt.Printf(&quot;prevent panic by handling failure accessing a path %q: %v\n&quot;, path, err)
  46. return err
  47. }
  48. if !info.IsDir() {
  49. paths = append(paths, path)
  50. }
  51. return nil
  52. })
  53. if err != nil {
  54. fmt.Printf(&quot;Error walking the path %q: %v\n&quot;, inputDir, err)
  55. }
  56. return paths
  57. }
  58. func searchPaths(paths []string, pattern string) chan Display {
  59. out := make(chan Display)
  60. for _, path := range paths {
  61. wg.Add(1)
  62. go func(p string, w *sync.WaitGroup) { // as path var is changing value in the loop, it&#39;s better to provide it as a argument in goroutine
  63. defer w.Done()
  64. for _, display := range searchFile(p, pattern) {
  65. out &lt;- display
  66. }
  67. }(path, &amp;wg)
  68. }
  69. return out
  70. }
  71. func searchFile(path string, pattern string) []Display {
  72. var out []Display
  73. input := fileFromPath(path)
  74. lines := splitIntoLines(input)
  75. for index, line := range lines {
  76. if searchResult, ok := searchLine(pattern, line, index); ok {
  77. out = append(out, Display{path, searchResult})
  78. }
  79. }
  80. return out
  81. }
  82. func main() {
  83. pattern := os.Args[1]
  84. dirPath := os.Args[2]
  85. paths := getRecursiveFilePaths(dirPath)
  86. out := searchPaths(paths, pattern)
  87. go func(){
  88. wg.Wait() // waiting before closing the channel
  89. close(out)
  90. }()
  91. count := 0
  92. for d := range out {
  93. fmt.Println(count)
  94. d.PrettyPrint()
  95. count += 1
  96. }
  97. }

huangapple
  • 本文由 发表于 2022年11月10日 19:41:29
  • 转载请务必保留本文链接：https://go.coder-hub.com/74388429.html
  • concurrency
  • go
  • goroutine
匿名

发表评论

匿名网友

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

确定