CODEWITHSUNDEEP
gogoroutine
rafalm
rafalm

Reputation: 51

Deadlock in goroutines pipeline

I need your help to understand why my readFromWorker func lead to deadlock. When I comment out lines like below it works correctly (thus I know the problem is here).

The whole is here https://play.golang.org/p/-0mRDAeD2tr

I would really appreciate your help

func readFromWorker(inCh <-chan *data, wg *sync.WaitGroup) {
    defer func() {
        wg.Done()
    }()

    //stageIn1 := make(chan *data)
    //stageOut1 := make(chan *data)

    for v := range inCh {
        fmt.Println("v", v)

        //stageIn1 <- v
    }

    //go stage1(stageIn1, stageOut1)
    //go stage2(stageOut1)
}

Upvotes: 0

Views: 148

Answers (1)

shmsr
shmsr

Reputation: 4204

I've commented on the relevant parts where you were doing it wrong. Also, I'd recommend thinking of a better pattern.

Do remember that for range on channels doesn't stop looping unless close is called for the same channel it's looping on. Also, the rule of thumb of closing a channel is that the sender sending to the channel must also close it because sending to a closed channel causes panic.

Also, be very careful when using unbuffered and buffered channels. For unbuffered channels, the sender and receiver must be ready otherwise there would be a deadlock which happened in your case as well.

package main

import (
    "fmt"
    "sync"
)

type data struct {
    id    int
    url   string
    field int
}

type job struct {
    id  int
    url string
}

func sendToWorker(id int, inCh <-chan job, outCh chan<- *data, wg *sync.WaitGroup) {
    // wg.Done() is itself a function call, no need to wrap it inside
    // an anonymous function just to use defer.
    defer wg.Done()

    for v := range inCh {
        // some pre process stuff and then pass to pipeline
        outCh <- &data{id: v.id, url: v.url}
    }
}

func readFromWorker(inCh <-chan *data, wg *sync.WaitGroup) {
    // wg.Done() is itself a function call, no need to wrap it inside
    // an anonymous function just to use defer.
    defer wg.Done()

    var (
        stageIn1  = make(chan *data)
        stageOut1 = make(chan *data)
    )

    // Spawn the goroutines so that there's no deadlock
    // as the sender and receiver both should be ready
    // when using unbuffered channels.
    go stage1(stageIn1, stageOut1)
    go stage2(stageOut1)

    for v := range inCh {
        fmt.Println("v", v)
        stageIn1 <- v
    }
    close(stageIn1)
}

func stage1(in <-chan *data, out chan<- *data) {
    for s := range in {
        fmt.Println("stage1 = ", s)
        out <- s
    }
    // Close the out channel
    close(out)
}

func stage2(out <-chan *data) {
    // Loop until close
    for s := range out {
        fmt.Println("stage2 = ", s)
    }
}

func main() {
    const chanBuffer = 1

    var (
        inputsCh  = make(chan job, chanBuffer)
        resultsCh = make(chan *data, chanBuffer)

        wgInput  sync.WaitGroup
        wgResult sync.WaitGroup
    )

    for i := 1; i <= 4; i++ {
        wgInput.Add(1)
        go sendToWorker(i, inputsCh, resultsCh, &wgInput)
    }

    wgResult.Add(1)
    go readFromWorker(resultsCh, &wgResult)

    for j := 1; j <= 10; j++ {
        inputsCh <- job{id: j, url: "google.com"}
    }

    close(inputsCh)
    wgInput.Wait()
    close(resultsCh)
    wgResult.Wait()
}

Upvotes: 1

Related Questions