当前位置： > > > > 使用 waitgroup 程序陷入死锁
来源：stackoverflow
2024-04-21 23:18:36
0浏览
收藏
学习知识要善于思考，思考，再思考！今天小编就给大家带来《使用 waitgroup 程序陷入死锁》，以下内容主要包含等知识点，如果你正在学习或准备学习Golang，就都不要错过本文啦~让我们一起来看看吧，能帮助到你就更好了！
问题内容
我正在编写一个程序，该程序读取名为orders.csv 的文件中的订单号列表，并将其与文件夹中存在的其他 csv 文件进行比较。
问题是，即使使用 waitgroup，它也会陷入死锁，我不知道为什么。
由于某种原因，stackoverflow 说我的帖子主要是代码，所以我必须添加这一行，因为如果有人想帮助我调试我遇到的这个问题，整个代码是必要的。
package main
import (
"bufio"
"fmt"
"log"
"os"
"path/filepath"
"strings"
"sync"
)
type Files struct {
filenames []string
}
type Orders struct {
ID []string
}
var ordersFilename string = "orders.csv"
func main() {
var (
ordersFile *os.File
files Files
orders Orders
err error
)
mu := new(sync.Mutex)
wg := &sync.WaitGroup{}
wg.Add(1)
if ordersFile, err = os.Open(ordersFilename); err != nil {
log.Fatalln("Could not open file: " + ordersFilename)
}
orders = getOrderIDs(ordersFile)
files.filenames = getCSVsFromCurrentDir()
var filenamesSize = len(files.filenames)
var ch = make(chan map[string][]string, filenamesSize)
var done = make(chan bool)
for i, filename := range files.filenames {
go func(currentFilename string, ch chan<- map[string][]string, i int, orders Orders, wg *sync.WaitGroup, filenamesSize *int, mu *sync.Mutex, done chan<- bool) {
wg.Add(1)
defer wg.Done()
checkFile(currentFilename, orders, ch)
mu.Lock()
*filenamesSize--
mu.Unlock()
if i == *filenamesSize {
done <- true
close(done)
}
}(filename, ch, i, orders, wg, &filenamesSize, mu, done)
}
select {
case str := <-ch:
fmt.Printf("%+v\n", str)
case <-done:
wg.Done()
break
}
wg.Wait()
close(ch)
}
// getCSVsFromCurrentDir returns a string slice
// with the filenames of csv files inside the
// current directory that are not "orders.csv"
func getCSVsFromCurrentDir() []string {
var filenames []string
err := filepath.Walk(".", func(path string, info os.FileInfo, err error) error {
if path != "." && strings.HasSuffix(path, ".csv") && path != ordersFilename {
filenames = append(filenames, path)
}
return nil
})
if err != nil {
log.Fatalln("Could not read file names in current dir")
}
return filenames
}
// getOrderIDs returns an Orders struct filled
// with order IDs retrieved from the file
func getOrderIDs(file *os.File) Orders {
var (
orders Orders
err error
fileContent string
)
reader := bufio.NewReader(file)
if fileContent, err = readLine(reader); err != nil {
log.Fatalln("Could not read file: " + ordersFilename)
}
for err == nil {
orders.ID = append(orders.ID, fileContent)
fileContent, err = readLine(reader)
}
return orders
}
func checkFile(filename string, orders Orders, ch chan<- map[string][]string) {
var (
err error
file *os.File
fileContent string
orderFilesMap map[string][]string
counter int
)
orderFilesMap = make(map[string][]string)
if file, err = os.Open(filename); err != nil {
log.Fatalln("Could not read file: " + filename)
}
reader := bufio.NewReader(file)
if fileContent, err = readLine(reader); err != nil {
log.Fatalln("Could not read file: " + filename)
}
for err == nil {
if containedInSlice(fileContent, orders.ID) && !containedInSlice(fileContent, orderFilesMap[filename]) {
orderFilesMap[filename] = append(orderFilesMap[filename], fileContent)
// fmt.Println("Found: ", fileContent, " in ", filename)
} else {
// fmt.Printf("Could not find: '%s' in '%s'\n", fileContent, filename)
}
counter++
fileContent, err = readLine(reader)
}
ch <- orderFilesMap
}
// containedInSlice returns true or false
// based on whether the string is contained
// in the slice
func containedInSlice(str string, slice []string) bool {
for _, ID := range slice {
if ID == str {
return true
}
}
return false
}
// readLine returns a line from the passed reader
func readLine(r *bufio.Reader) (string, error) {
var (
isPrefix bool = true
err error = nil
line, ln []byte
)
for isPrefix && err == nil {
line, isPrefix, err = r.ReadLine()
ln = append(ln, line...)
}
return string(ln), err
}
解决方案
-
第一个问题是
wg.add始终必须位于它代表的 goroutine 之外。如果不是，则wg.wait调用可能会在 goutine 实际开始运行之前调用（并称为wg.add），因此会“思考” 没有什么可以等待的。 -
代码的第二个问题是它有多种方式等待例程完成。有
waitgroup并且有done通道。仅使用其中之一。哪一个还取决于结果如何 使用 goroutine。我们来讨论下一个问题。 -
第三个问题是收集结果。目前，代码仅打印/使用 goroutine 的单个结果。 在 select 周围放置一个
for { ... }循环，并在done通道关闭时使用return跳出循环。 （请注意，您不需要在done通道上发送任何内容，关闭它就足够了。）
改进版本0.0.1
这里是第一个版本（包括其他一些“代码清理”），其中有一个用于关闭的 done 通道，并删除了 waitgroup：
func main() {
ordersfile, err := os.open(ordersfilename)
if err != nil {
log.fatalln("could not open file: " + ordersfilename)
}
orders := getorderids(ordersfile)
files := files{
filenames: getcsvsfromcurrentdir(),
}
var (
mu = new(sync.mutex)
filenamessize = len(files.filenames)
ch = make(chan map[string][]string, filenamessize)
done = make(chan bool)
)
for i, filename := range files.filenames {
go func(currentfilename string, ch chan<- map[string][]string, i int, orders orders, filenamessize *int, mu *sync.mutex, done chan<- bool) {
checkfile(currentfilename, orders, ch)
mu.lock()
*filenamessize--
mu.unlock()
// todo: this also accesses filenamessize, so it also needs to be protected with the mutex:
if i == *filenamessize {
done <- true
close(done)
}
}(filename, ch, i, orders, &filenamessize, mu, done)
}
// note: closing a channel is not really needed, so you can omit this:
defer close(ch)
for {
select {
case str := <-ch:
fmt.printf("%+v\n", str)
case <-done:
return
}
}
}
改进版本0.0.2
- 对于您的情况，我们有一些优势。我们确切地知道我们启动了多少个 goroutine，因此也知道如何启动 我们期待许多结果。 （当然，如果每个 goroutine 返回当前代码所做的结果。）这给出了 我们还有另一种选择，因为我们可以使用另一个具有相同迭代次数的 for 循环来收集结果：
func main() {
ordersfile, err := os.open(ordersfilename)
if err != nil {
log.fatalln("could not open file: " + ordersfilename)
}
orders := getorderids(ordersfile)
files := files{
filenames: getcsvsfromcurrentdir(),
}
var (
// note: a buffered channel helps speed things up. the size does not need to match the size of the items that will
// be passed through the channel. a fixed, small size is perfect here.
ch = make(chan map[string][]string, 5)
)
for _, filename := range files.filenames {
go func(filename string) {
// orders and channel are not variables of the loop and can be used without copying
checkfile(filename, orders, ch)
}(filename)
}
for range files.filenames {
str := <-ch
fmt.printf("%+v\n", str)
}
}
简单多了，不是吗？希望有帮助！
这段代码有很多错误。
- 您使用的 waitgroup 是错误的。 add 必须在主 goroutine 中调用，否则有可能在所有 add 调用完成之前调用 wait。
- 初始化 waitgroup 后立即进行了一个无关的 add(1) 调用，该调用与 done() 调用不匹配，因此 wait 永远不会返回（假设上面的点已修复）。
- 您正在使用 waitgroup 和 done 通道来表示完成。这充其量是多余的。
- 您在未持有锁的情况下读取 filenamessize（在
if i == *filenamessize语句中）。这是一个竞争条件。 i == *filenamessize条件首先没有任何意义。 goroutine 以任意顺序执行，因此你无法确定 i == 0 的 goroutine 是最后一个减少 filenamessize 的 goroutine
这一切都可以通过摆脱大部分 if 同步原语并在所有 goroutine 完成时简单地关闭 ch 通道来简化：
func main() {
ch := make(chan map[string][]string)
var wg WaitGroup
for _, filename := range getCSVsFromCurrentDir() {
filename := filename // capture loop var
wg.Add(1)
go func() {
checkFile(filename, orders, ch)
wg.Done()
}()
}
go func() {
wg.Wait() // after all goroutines are done...
close(ch) // let range loop below exit
}()
for str := range ch {
// ...
}
}
到这里，我们也就讲完了《使用 waitgroup 程序陷入死锁》的内容了。个人认为，基础知识的学习和巩固，是为了更好的将其运用到项目中，欢迎关注公众号，带你了解更多关于的知识点！
