Início / coding / Perguntas / 79341391
Accepted
SYKO
Asked: 2025-01-09 12:35:52 +0800 CST

Deadlock ao implementar um padrão de simultaneidade de pipeline

  • 772

Sobre o programa: Estou tentando implementar o padrão pipeline por meio do meu próprio tipo de primeira classe intJob. A função main, que agrega os pipelines, é ExecutePipeline2e, como posso dizer, exatamente ela está causando problemas.

Por que continuo tendo deadlocks? Para mim, parece que fecho todos os canais que são usados ​​pelos leitores após cada goroutine. Além disso, criar canais com buffer não ajuda, então fiquei completamente sem ideias e ficarei muito grato pela sua ajuda.

IMPORTANTE : Não posso alterar a mainfunção e implementar essa ideia somente a partir de outras funções, quando a base (função mencionada) permanece constante e inalterada.

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 respostas

  1. Best Answer

    Acho que o problema principal está no seu segundo pipeline pipe(), que em vez de ler a saída do pipeline anterior, apenas começou a gerar números em um loop. Deveria ter sido escrito como abaixo para ler incomo

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

    Você também pode reescrever a função execute para algo como abaixo. Exemplo completo em 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. O primeiro canal "out" (1 índice) criado neste trecho de código

       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)
//...
        }()
    }

    não usa como "produtor" e após a primeira iteração na primeira intJobe na segunda iteração do primeiro intJobdeadlock acumulado (porque chanestá cheio):

        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!
            }
        }),
// ....

    insira a descrição da imagem aqui

    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

relate perguntas