1
\$\begingroup\$

I'm new in Go and excited with its easy-to-use concurrency implementation. However I'm not sure if I'm doing it right in Golang way.
Consider the code where I spawn process(i int, ch chan int) every loop as separate goroutine. Then the result which comes from the channel will be consumed by consume(ch chan int, wg *sync.WaitGroup).
I know that the channel is blocking, so let's implement buffered channel. And of course I don't want the process terminated before every goroutine finished the operation, so I add WaitGroup.

func main() {
    var wg sync.WaitGroup
    n := 5
    ch := make(chan int, n)
    for i := 0; i < n; i++ {
        println("Processing ", i)
        go process(i, ch)
        go consume(ch, &wg)
    }
    println("Finished the process")
    wg.Wait()
}

func consume(ch chan int, wg *sync.WaitGroup) {
    wg.Add(1)
    defer wg.Done()
    println("Result ", <-ch)
}

func process(i int, ch chan int) {
    ch <- (i * 5)
}

Am I doing it right? Or there are better way to do this?
Many thanks!

\$\endgroup\$
0
\$\begingroup\$

Every time you execute a go statement it is passed to the scheduler. What if scheduling is delayed? wg.Add(1) is not executed and wg.Wait() is true. For example, run your code in the Go Playground where GOMAXPROCS is 1.

package main

import (
    "sync"
)

func main() {
    var wg sync.WaitGroup
    n := 5
    ch := make(chan int, n)
    for i := 0; i < n; i++ {
        println("Processing ", i)
        go process(i, ch)
        go consume(ch, &wg)
    }
    println("Finished the process")
    wg.Wait()
}

func consume(ch chan int, wg *sync.WaitGroup) {
    wg.Add(1)
    defer wg.Done()
    println("Result ", <-ch)
}

func process(i int, ch chan int) {
    ch <- (i * 5)
}

Playground: https://play.golang.org/p/dQ_lFRz2Y8a

Output:

Processing  0
Processing  1
Processing  2
Processing  3
Processing  4
Finished the process

Make sure that all the wg.Adds are run before wg.Wait. Move the println("Finished the process") to the correct place after the wg.Wait. For example,

waiting.go:

package main

import (
    "runtime"
    "sync"
)

func main() {
    println("GOMAXPROCS", runtime.GOMAXPROCS(0))
    var wg sync.WaitGroup
    n := 5
    ch := make(chan int, n)
    for i := 0; i < n; i++ {
        println("Processing ", i)
        go process(i, ch)
        wg.Add(1)
        go consume(ch, &wg)
    }
    wg.Wait()
    println("Finished the process")
}

func consume(ch chan int, wg *sync.WaitGroup) {
    defer wg.Done()
    println("Result ", <-ch)
}

func process(i int, ch chan int) {
    ch <- (i * 5)
}

Playground: https://play.golang.org/p/3czBixAjxdT

Output:

GOMAXPROCS 1
Processing  0
Processing  1
Processing  2
Processing  3
Processing  4
Result  0
Result  5
Result  10
Result  15
Result  20
Finished the process

Run the Go data race detector to check for data races. It finds none.

$ go run -race waiting.go
GOMAXPROCS 4
Processing  0
Processing  1
Result  0
Processing  2
Result  5
Processing  3
Result  10
Processing  4
Result  15
Result  20
Finished the process
$
\$\endgroup\$
-1
\$\begingroup\$

Given that you seem to be experimenting, it's hard to say if the pattern you've employed is correct since there really isn't context. But this is largely a valid use of channels. There is one problem though: you have a race condition. More on that in a bit.

Using sync.WaitGroup is the correct approach here. Ideally, you should thread this through all goroutines that you spawn. In this case, because of the data dependency through ch (namely that there are exactly n processes and n consumers and each sends/receives one int), we know that if all of the consumers finish (and call the deferred wg.Done()) there can't be any processes running. But often, for larger projects such a dependency may not be as obvious, and may change if you alter other parts of your codebase. So, in any nontrivial application where all channel usages aren't limited to a small area of the code, you probably want to thread the sync.WaitGroup through all goroutines spawned.

A note on blocking: you made a vague statement about channel blocking. To clarify, make(chan int, 5) produces a buffered channel which only blocks if more than 5 ints have been sent but not received. make(chan int) produces a channel that blocks until a consumer receives the value. In your example, both of these will work.

The race condition:

Doing wg.Add(1) inside the goroutine is racy. Why? Because once you do go consume(ch, &wg) you can't say for sure that the next thing that will happen is the wg.Add(1). In fact, you can't even be sure that that will run at all (we could reach the end of main()). In a degenerate case, if you reached the end of main() and wg.Wait() was called before a wg.Add(1) from a goroutine, the counter in the Wait() wouldn't account for this goroutine. You need to do the wg.Add(1) before you spawn the goroutine:

wg.Add(1)
go consume(ch, &wg)

Since you know there will be n goroutines spawned, you could also do wg.Add(n) before the loop, but if that loop can exit prematurely, it is more wise (and clear) to do the wg.Add(1) before each go consume().

You should also probably defer wg.Wait() right after var wg sync.WaitGroup in main(). The point of defer is to keep resource allocation and cleanup close so you don't forget it!

Also, you should get in the habit of using directional channels. They give you some safety that you don't accidentally receive from producers or send from consumers. They also quickly communicate to readers of your code what each function does at a high level.

func consume(ch <-chan int, wg *sync.WaitGroup) {
    // ...
}

func process(i int, ch chan<- int) {
    // ...
}
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.