主页 / coding / 问题 / 79341391
Accepted
SYKO
Asked: 2025-01-09 12:35:52 +0800 CST

实现管道并发模式时出现死锁

  • 772

关于程序：我正在尝试通过我自己的一流类型实现管道intJob模式。聚合管道的主要功能正是ExecutePipeline2它导致了问题。

为什么我总是遇到死锁？对我来说，我似乎在每个 goroutine 之后都关闭了所有读者使用的通道。另外，创建带缓冲区的通道也无济于事，所以我完全没有主意了，非常感谢你的帮助。

重要提示main：当基础（提到的功能）保持不变时，我无法更改功能并仅从其他功能实现这个想法。

type intJob func(a, b chan int)

func ExecutePipeline2(jobs ...intJob) {
    outs := make([]chan int, len(jobs)+1)
    wg := sync.WaitGroup{}

    for i := 0; i < len(outs); i++ {
        outs[i] = make(chan int)
    }

    for i, job := range jobs {
        job := job
        in, out := outs[i], outs[i+1]
        i := i
        wg.Add(1)
        go func() {
            job(in, out)
            fmt.Printf("job %d closed\n", i)
            close(out)
            wg.Done()
        }()
    }

    wg.Wait()
}

func pipe(_, b chan int) {
    for i := 0; i < 5; i++ {
        b <- i
    }
}

func main() {
    inputData := []int{0, 1, 1, 2, 3, 5, 8}

    hashSignJobs := []intJob{
        intJob(func(in, out chan int) {
            for _, fibNum := range inputData {
                out <- fibNum
            }
        }),
        intJob(pipe),
        intJob(func(in, out chan int) {
            for val := range in {
                fmt.Println(val)
            }
        }),
    }

    ExecutePipeline2(hashSignJobs...)
}
go

2 个回答

  1. Best Answer

    我认为关键问题在于你的第二条管道pipe()，它不是从前一条管道的输出中读取，而是开始循环生成数字。应该按如下方式编写，以便从inas中读取

    func pipe(in, out chan int) {
    for i := range in { 
        out <- i
    }
}

    您也可以将执行函数重写为如下所示的内容。完整示例位于https://go.dev/play/p/K_-UFzz0zt5

    func ExecutePipeline2(jobs ...intJob) {
    outs := make([]chan int, len(jobs)+1)
    for i := 0; i < len(outs); i++ {
        outs[i] = make(chan int)
    }

    var wg sync.WaitGroup

    for i, job := range jobs {
        wg.Add(1)
        go func(i int, job intJob) {
            defer wg.Done()
            // Close the output channel when done
            defer close(outs[i+1]) 
            job(outs[i], outs[i+1])
        }(i, job)
    }

    // Close the first channel after all input is written
    close(outs[0])

    wg.Wait()
}
    • 4

  2. 此代码片段中创建的第一个“输出”（1 索引）通道

       for i, job := range jobs {
        job := job
        in, out := outs[i], outs[i+1] // here 
        i := i
        wg.Add(1)
        go func() {
            job(in, out)
//...
        }()
    }

    不使用作为“生产者”，并且在第一次迭代之后intJob和在第二次迭代中第一次intJob出现死锁（因为chan已满）：

        hashSignJobs := []intJob{
        intJob(func(in, out chan int) {
            for _, fibNum := range inputData {
                fmt.Printf("1 - %v\n", fibNum)
                out <- fibNum // fatal error: all goroutines are asleep - deadlock!
            }
        }),
// ....

    在此处输入图片描述

    PLAYGROUND

    # ...

fatal error: all goroutines are asleep - deadlock!

goroutine 1 [semacquire]:
sync.runtime_Semacquire(0xc0000161b0?)
    /usr/local/go-faketime/src/runtime/sema.go:71 +0x25
sync.(*WaitGroup).Wait(0x10100000010?)
    /usr/local/go-faketime/src/sync/waitgroup.go:118 +0x48
main.ExecutePipeline2({0xc000084f20, 0x3, 0x0?})
    /tmp/sandbox346714581/prog.go:31 +0x1d2
main.main()
    /tmp/sandbox346714581/prog.go:59 +0xd5

goroutine 6 [chan send]:
main.main.func1(0x101000000000000?, 0xc00007a0e0)
    /tmp/sandbox346714581/prog.go:48 +0xb6
main.ExecutePipeline2.func1()
    /tmp/sandbox346714581/prog.go:24 +0x3a
created by main.ExecutePipeline2 in goroutine 1
    /tmp/sandbox346714581/prog.go:23 +0xdb
    • 0

相关问题