尝试编写工作方法池时出现死锁

Question

在下面的代码中，我不明白为什么“Worker”方法似乎退出而不是从“in”输入通道中提取值并处理它们。

我曾假设它们只会在消耗来自输入通道“in”的所有输入并处理它们之后才会返回

package main

import (
    "fmt"
    "sync"
)

type ParallelCallback func(chan int, chan Result, int, *sync.WaitGroup)

type Result struct {
    i   int
    val int
}

func Worker(in chan int, out chan Result, id int, wg *sync.WaitGroup) {
    for item := range in {
        item *= item // returns the square of the input value
        fmt.Printf("=> %d: %d\n", id, item)
        out <- Result{item, id}
    }
    wg.Done()
    fmt.Printf("%d exiting ", id)
}

func Run_parallel(n_workers int, in chan int, out chan Result, Worker ParallelCallback) {
    wg := sync.WaitGroup{}
    for id := 0; id < n_workers; id++ {
        fmt.Printf("Starting : %d\n", id)
        wg.Add(1)
        go Worker(in, out, id, &wg)
    }
    wg.Wait()  // wait for all workers to complete their tasks
    close(out) // close the output channel when all tasks are completed
}

const (
    NW = 4
)

func main() {
    in := make(chan int)
    out := make(chan Result)

    go func() {
        for i := 0; i < 100; i++ {
            in <- i
        }
        close(in)
    }()
    Run_parallel(NW, in, out, Worker)

    for item := range out {
        fmt.Printf("From out : %d: %d", item.i, item.val)
    }
}

输出是

Starting : 0
Starting : 1
Starting : 2
Starting : 3
=> 3: 0
=> 0: 1
=> 1: 4
=> 2: 9
fatal error: all goroutines are asleep - deadlock!

Answer 1

致命错误：所有 goroutine 都处于休眠状态 - 死锁！

完整的错误显示每个 goroutine “卡住”的位置。 If you run this in the playground，它甚至会显示行号。这让我很容易诊断。

您的Run_parallel 在main groutine 中运行，所以在main 可以从out 读取之前，Run_parallel 必须返回。在Run_parallel 可以返回之前，它必须是wg.Wait()。但是在工人打电话给wg.Done()之前，他们必须写信给out。这就是导致死锁的原因。

一个解决方案很简单：只需在自己的 Goroutine 中同时运行 Run_parallel。

    go Run_parallel(NW, in, out, Worker)

现在，main 的范围超过 out，等待 outs 关闭以表示完成。 Run_parallel 用wg.Wait() 等待workers，workers 的范围将超过in。所有的工作都会完成，并且在完成之前程序不会结束。 (https://go.dev/play/p/oMrgH2U09tQ)

Answer 2

解决方案：

Run_parallel 必须在它自己的 goroutine 中运行：

package main

import (
    "fmt"
    "sync"
)

type ParallelCallback func(chan int, chan Result, int, *sync.WaitGroup)

type Result struct {
    id  int
    val int
}

func Worker(in chan int, out chan Result, id int, wg *sync.WaitGroup) {
    defer wg.Done()
    for item := range in {
        item *= 2 // returns the double of the input value (Bogus handling of data)
        out <- Result{id, item}
    }
}

func Run_parallel(n_workers int, in chan int, out chan Result, Worker ParallelCallback) {
    wg := sync.WaitGroup{}
    for id := 0; id < n_workers; id++ {
        wg.Add(1)
        go Worker(in, out, id, &wg)
    }
    wg.Wait()  // wait for all workers to complete their tasks
    close(out) // close the output channel when all tasks are completed
}

const (
    NW = 8
)

func main() {

    in := make(chan int)
    out := make(chan Result)

    go func() {
        for i := 0; i < 10; i++ {
            in <- i
        }
        close(in)
    }()

    go Run_parallel(NW, in, out, Worker)

    for item := range out {
        fmt.Printf("From out [%d]: %d\n", item.id, item.val)
    }

    println("- - - All done - - -")

}

Answer 3

解决方案的替代配方：

在那个替代公式中，没有必要将 Run_parallel 作为 goroutine 启动（它会触发自己的 goroutine）。 我更喜欢第二种解决方案，因为它自动化了 Run_parallel() 必须与主函数并行运行的事实。此外，出于同样的原因，它更安全，更不容易出错（无需记住使用 go 关键字运行 Run_parallel）。

package main

import (
    "fmt"
    "sync"
)

type ParallelCallback func(chan int, chan Result, int, *sync.WaitGroup)

type Result struct {
    id  int
    val int
}

func Worker(in chan int, out chan Result, id int, wg *sync.WaitGroup) {
    defer wg.Done()
    for item := range in {
        item *= 2 // returns the double of the input value (Bogus handling of data)
        out <- Result{id, item}
    }
}

func Run_parallel(n_workers int, in chan int, out chan Result, Worker ParallelCallback) {
    go func() {
        wg := sync.WaitGroup{}
        defer close(out) // close the output channel when all tasks are completed
        for id := 0; id < n_workers; id++ {
            wg.Add(1)
            go Worker(in, out, id, &wg)
        }
        wg.Wait() // wait for all workers to complete their tasks *and* trigger the -differed- close(out)
    }()
}

const (
    NW = 8
)

func main() {

    in := make(chan int)
    out := make(chan Result)

    go func() {
        defer close(in)
        for i := 0; i < 10; i++ {
            in <- i
        }
    }()

    Run_parallel(NW, in, out, Worker)

    for item := range out {
        fmt.Printf("From out [%d]: %d\n", item.id, item.val)
    }

    println("- - - All done - - -")
}