CODEWITHSUNDEEP
goparallel-processingchannel
Peter Butkovic
Peter Butkovic

Reputation: 12149

Reading from golang channels in order

I'm trying to achieve parallel processing and communication over the channels in go.

What I basically try to solve is process a specifc data in parallel, and get results in order => introduced type Chunk for the purpose (see bellow).

I just make new channel for each chunk processing and keep them in slice => expect to be ordered once I iterate over them afterwards.

Simplified version of my program is (https://play.golang.org/p/RVtDGgUVCV):

package main

import (
    "fmt"
)

type Chunk struct {
    from int
    to   int
}

func main() {
    chunks := []Chunk{
        Chunk{
            from: 0,
            to:   2,
        },
        Chunk{
            from: 2,
            to:   4,
        },
    }

    outChannels := [](<-chan struct {
        string
        error
    }){}

    for _, chunk := range chunks {
        outChannels = append(outChannels, processChunk(&chunk))
    }

    for _, outChannel := range outChannels {
        for out := range outChannel {
            if out.error != nil {
                fmt.Printf("[ERROR] %s", out.error)
                return
            }

            fmt.Printf("[STDOUT] %s", out.string)
        }
    }
}

func processChunk(c *Chunk) <-chan struct {
    string
    error
} {

    outChannel := make(chan struct {
        string
        error
    })

    go func() {
        outChannel <- struct {
            string
            error
        }{fmt.Sprintf("from: %d to: %d\n", c.from, c.to), nil}

        close(outChannel)
    }()

    return outChannel
}

The output I see is:

[STDOUT] from: 2 to: 4
[STDOUT] from: 2 to: 4

What I'd however expect to see would be:

[STDOUT] from: 0 to: 2
[STDOUT] from: 2 to: 4

What am I doing wrong here? I don't see it.

Upvotes: 1

Views: 1463

Answers (1)

abhink
abhink

Reputation: 9126

The trouble is in the very first for loop of main. When you use for range loop, the loop variable (chunk here) gets created once and is assigned a copy of each slice element per iteration.

When you call processChunk(&chunk), you are passing the address of this loop variable, and the value of this variable changes with each iteration. Thus the function processChunk always ends up working on the last item in the chunks loop since that is what *chunk points to after the for loop finishes.

To fix, use slice indexing:

for i := 0; i < len(chunks); i++ {
    // pass chunk objects by indexing chunks
    outChannels = append(outChannels, processChunk(&chunks[i]))
}

Fixed code: https://play.golang.org/p/A1_DtkncY_

You can read more about range here.

Upvotes: 4

Related Questions