file - 戈兰 : Getting "fatal error: all goroutines are asleep - deadlock" on waitGroup. 等待()

Question

我正在尝试编写一个代码，它对文件进行并发读取并将内容发布到一个 channel 。

Here是我的代码的链接，代码:

func main() {
    bufferSize := int64(10)
    f, err := os.Open("tags-c.csv")
    if err != nil {
        panic(err)
    }
    fileinfo, err := f.Stat()
    if err != nil {
        fmt.Println(err)
        return
    }
    filesize := int64(fileinfo.Size())
    fmt.Println(filesize)
    routines := filesize / bufferSize
    if remainder := filesize % bufferSize; remainder != 0 {
        routines++
    }
    fmt.Println("Total routines : ", routines)

    channel := make(chan string, 10)
    wg := &sync.WaitGroup{}

    for i := int64(0); i < int64(routines); i++ {
        wg.Add(1)
        go read(i*bufferSize, f, channel, bufferSize, filesize, wg)

    }
    fmt.Println("waiting")
    wg.Wait()
    fmt.Println("wait over")
    close(channel)

    readChannel(channel)
}

func readChannel(channel chan string) {
    for {
        data, more := <-channel
        if more == false {
            break
        }
        fmt.Print(data)
    }
}

func read(seek int64, file *os.File, channel chan string, bufferSize int64, filesize int64, wg *sync.WaitGroup) {
    defer wg.Done()
    fmt.Println("read :: ", seek)
    var buf []byte
    if filesize < bufferSize {
        buf = make([]byte, filesize)
    } else if (filesize - seek) < bufferSize {
        buf = make([]byte, filesize-seek)
    } else {
        buf = make([]byte, bufferSize)
    }

    n, err := file.ReadAt(buf, seek)
    if err != nil {
        log.Printf("loc %d err: %v", seek, err)
        return
    }
    if n > 0 {
        channel <- string(buf[:n])
        fmt.Println("ret :: ", seek)
    }
}

我尝试在线查看，但令我惊讶的是我已经解决了所提到的解决方案。任何帮助将不胜感激。

Answer 1

问题是您希望所有已启动的阅读器 goroutine 在您继续并耗尽它们提供结果的 channel 之前完成。

并且 channel 是缓冲的，最多可以容纳10个元素。一旦 10 个 goroutines 向其发送消息，其余的将被阻塞，因此它们永远不会完成(因为只有在它们全部返回后才能从该 channel 读取:这就是死锁)。

因此，您应该启动另一个 goroutine 以与 reader goroutines 同时接收结果:

done := make(chan struct{})
go readChannel(channel, done)

fmt.Println("waiting")
wg.Wait()
fmt.Println("wait over")
close(channel)

// Wait for completion of collecting the results:
<-done

读取 channel 的位置应该是 for range(当 channel 关闭并且已从 channel 接收到在 channel 关闭之前发送到 channel 上的所有值时终止):

func readChannel(channel chan string, done chan struct{}) {
    for data := range channel {
        fmt.Print(data)
    }
    close(done)
}

请注意，我使用了一个 done channel ，因此主 goroutine 也会等待接收结果的 goroutine 完成。

另请注意，由于在大多数情况下磁盘 IO 是瓶颈而不是 CPU，并且由于从多个 goroutine 传递和接收结果也有一些开销，所以很可能您不会看到任何改进阅读多个 goroutines 同时处理文件。