CODEWITHSUNDEEP
parallel-processinggogoroutine
Bill
Bill

Reputation: 25555

Goroutines not exiting when data channel is closed

I'm trying to follow along the bounded goroutine example that is posted at http://blog.golang.org/pipelines/bounded.go. The problem that I'm having is that if there are more workers spun up then the amount of work to do, the extra workers never get cancelled. Everything else seems to work, the values get computed and logged, but when I close the groups channel, the workers just hang at the range statement.

I guess what I don't understand (in both my code and the example code) is how do the workers know when there is no more work to do and that they should exit?

Update

A working (i.e. non-working) example is posted at http://play.golang.org/p/T7zBCYLECp. It shows the deadlock on the workers since they are all asleep and there is no work to do. What I'm confused about is that I think the example code would have the same problem.

Here is the code that I'm currently using:

// Creates a pool of workers to do a bunch of computations
func computeAll() error {
    done := make(chan struct{})
    defer close(done)

    groups, errc := findGroups(done)

    // start a fixed number of goroutines to schedule with
    const numComputers = 20     
    c := make(chan result)
    var wg sync.WaitGroup
    wg.Add(numComputers)
    for i := 0; i < numComputers; i++ {
        go func() {
            compute(done, groups, c)
            wg.Done()
        }()
    }

    go func() {
        wg.Wait()
        close(c)
    }()

    // log the results of the computation
    for r := range c { // log the results }

    if err := <-errc; err != nil {
        return err
    }

    return nil
}

Here is the code that fills up the channel with data:

// Retrieves the groups of data the must be computed
func findGroups(done <-chan struct{}) (<-chan model, <-chan error) {
    groups := make(chan model)
    errc := make(chan error, 1)
    go func() {
        // close the groups channel after find returns
        defer close(groups)

        group, err := //... code to get the group ...
        if err == nil {
            // add the group to the channel
            select {
                case groups <- group:
            }
        }
    }()

    return groups, errc
}

And here is the code that reads the channel to do the computations.

// Computes the results for the groups of data
func compute(done <-chan struct{}, groups <-chan model, c chan<- result) {
    for group := range groups {
        value := compute(group)

        select {
        case c <- result{value}:
        case <-done:
            return
        }
    }
}

Upvotes: 1

Views: 1066

Answers (2)

OneOfOne
OneOfOne

Reputation: 99224

Because you're trying to read from errc and it's empty unless there's an error.

//edit

computeAll() will always block on <- errc if there are no errors, another approach is to use something like:

func computeAll() (err error) {
    .........
    select {
    case err = <-errc:
    default: //don't block
    }
    return
}

Upvotes: 2

fabmilo
fabmilo

Reputation: 48320

Try to close the errc as OneOfOne says

go func() {
    wg.Wait()
    close(c)
    close(errc)
}()

// log the results of the computation
for r := range c { // log the results }

if err := range errc {
   if err != nil {
    return err
   }
}

Upvotes: 1

Related Questions