Data loss when sending to multiple slices using channel

user_15360567 0
Viewed 74

I'm recently learning concurrency in golang, and I'm thinking a program with generate a series of number and then send them to three slices, concurrently. Here's the code:

func main() {
    ch := make(chan int)
    done := make(chan bool)
    var bag1 []int
    var bag2 []int
    var bag3 []int

    go func() {
        for i := 0; i < 1000000; i++ {
            ch <- i
        }
        close(ch)
        done <- true
    }()

    go sendToBag(&bag1, ch)
    go sendToBag(&bag2, ch)
    go sendToBag(&bag3, ch)

    <-done

    len1 := (len(bag1))
    len2 := (len(bag2))
    len3 := (len(bag3))

    fmt.Println("length of bag1:", len1)
    fmt.Println("length of bag2:", len2)
    fmt.Println("length of bag3:", len3)
    fmt.Println("total length:", len1+len2+len3)
}

func sendToBag(bag *[]int, ch <-chan int) {
    for n := range ch {
        *bag = append(*bag, n)
    }
}

which gives the output like the following:

length of bag1: 327643
length of bag2: 335630
length of bag3: 336725
total length: 999998

The total length of three slices doesn't always add up to the count of numbers which were send to them. So my question is: what caused the promblem and how to improve the code

go concurrency
2 Answers

Your use of the done channel is simply not enough to guarantee that all 3 sendToBag goroutines finish their job completely.

While it's true that the channel is "fully drained" by the for n := range ch { statement before <-done is executed, there is nothing in your code that ensures that <-done is executed after the final *bag = append(*bag, n) is executed, and therefore there is no guarantee that the len() calls on the bags will be executed only after the append() calls.

Use a wait group instead of the done channel.

func main() {
    ch := make(chan int)
    wg := sync.WaitGroup{}
    var bag1 []int
    var bag2 []int
    var bag3 []int

    go func() {
        for i := 0; i < 1000000; i++ {
            ch <- i
        }
        close(ch)
    }()

    wg.Add(3)
    go sendToBag(&bag1, ch, &wg)
    go sendToBag(&bag2, ch, &wg)
    go sendToBag(&bag3, ch, &wg)

    wg.Wait()

    len1 := (len(bag1))
    len2 := (len(bag2))
    len3 := (len(bag3))

    fmt.Println("length of bag1:", len1)
    fmt.Println("length of bag2:", len2)
    fmt.Println("length of bag3:", len3)
    fmt.Println("total length:", len1+len2+len3)
}

func sendToBag(bag *[]int, ch <-chan int, wg *sync.WaitGroup) {
    for n := range ch {
        *bag = append(*bag, n)
    }
    wg.Done()
}

https://go.dev/play/p/_wDgS5aS7bI

user_9659000

@mkopriva's answer is preferred but if you want to do with channel only, here is the solution.

func main() {
    ch := make(chan int)
    done := make(chan bool)

    var bag1 []int
    var bag2 []int
    var bag3 []int

    go func() {
        for i := 0; i < 1000000; i++ {
            ch <- i
        }
        close(ch)
        done <- true

    }()
    ch1 := make(chan bool)
    ch2 := make(chan bool)
    ch3 := make(chan bool)
    go sendToBag(&bag1, ch, ch1)
    go sendToBag(&bag2, ch, ch2)
    go sendToBag(&bag3, ch, ch3)
    <-ch1
    <-ch2
    <-ch3
    <-done

    len1 := (len(bag1))
    len2 := (len(bag2))
    len3 := (len(bag3))

    fmt.Println("length of bag1:", len1)
    fmt.Println("length of bag2:", len2)
    fmt.Println("length of bag3:", len3)
    fmt.Println("total length:", len1+len2+len3)
}

func sendToBag(bag *[]int, ch <-chan int, ch1 chan bool) {

    for n := range ch {
        *bag = append(*bag, n)
    }
    ch1 <- true

}

https://go.dev/play/p/id3l_VtUq8E

Here we add more channels in the sendToBag function, so the main goroutine will be blocked until some other goroutine write on ch1/ch2/ch3, which ensures <-done is executed after the final *bag = append(*bag, n) is executed.

user_93983000
Related