Reading two channels with unknown number of items

Trying to generate numbers and send them into a channel another function will read then do something and return two channels based on if there was an error or not. I don't know how many errors there will be so the out and errorCh have a defer close to signal no more items.

In this example I have a channel that I send numbers into and then pass that channel to a function which returns two channels, out and errorCh. Then at the end I read the values from out and errorCh.

I defer close the in channel so when there are no more values sent in the downstream run function will know there are no more numbers to process. That means the run function will complete after it gets the close channel signal from in? Which would also close the out and errorCh so the select statement would get the signal those two channels are closed. Am I not closing one of the channels properly? Why is there a deadlock?

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

package main
import (
"errors"
"fmt"
)
func run(in <-chan int) (chan int, chan error) {
out := make(chan int)
errorCh := make(chan error)
go func() {
defer close(out)
defer close(errorCh)
for i := range in {
if i%2 == 0 {
out <- i
} else {
errorCh <- errors.New("we don't like odd numbers")
}
}
}()
return out, errorCh
}
func main() {
in := make(chan int)
out := make(chan int)
errors := make(chan error)
// generate numbers to send to the "in" channel
go func(in chan int) {
defer close(in)
for i := 0; i < 10; i++ {
fmt.Println("  input", i)
in <- i
}
}(in)
// run function reads from the "in" channel, does something and returns
// the "out" channel and "errorCh"
go func(in chan int) {
out, errors = run(in)
}(in)
// read from the "out" channel and "errorCh" until all of the numbers
// in the "in" channel have been run
for {
select {
case i, ok := <-out:
if !ok { // out channel is closed
return // Done
}
fmt.Println("done", i)
case err, ok := <-errors:
if !ok {
return
}
if err != nil {
fmt.Println(err)
}
}
}
}

Deadlocks are pretty interesting and there are multiple reasons why deadlock might occur. Two of those are

1. If you are reading from a channel which is not ready to put data or where the data is not yet written to.

2. If you are writing into a channel but no one is there at the time of writing to read that data.

In you case, the out channel that you have mentioned in the main function and out channel that you have mentioned in run function are entirely different. Although you are trying to copy the value from out, errors = run(in) but needless to say that this would be successful only when out <- i gets executed in the run function.

Now later in the main function at case i, ok := <-out: you are trying to read form the out channel. Now let me remind you again that you haven't use the same out channel in both the go routines so Go runtime simply tries to read the value from <-out even before out <- i can occur as mentioned by @zerkms.

So this falls in to the category of Reading from the channel before it is written. On the other hand the writing part is also waiting for some goroutine to read as well. Which means reading part is waiting for writing and writing part is waiting for reading, a typical deadlock condition.

So to be precise, Your code may not have race but it has deadlocks

I presume you are trying to create a producer consumer pattern. Take a help form the following template to do such.

package main

import (
	"errors"
	"fmt"
)

func Consume(in, out chan int, err chan error) {
	defer close(out)
	defer close(err)
	for i := range in {
		if i%2 == 0 {
			out <- i
		} else {
			err <- errors.New("we don't like odd numbers")
		}
	}
}

func main() {
	in := make(chan int)
	out := make(chan int)
	err := make(chan error)
	go func(in chan int) {
		defer close(in)
		for i := 0; i < 10; i++ {
			fmt.Println("  input", i)
			in <- i
		}
	}(in)

	go Consume(in, out, err)

	for {
		select {
		case i, ok := <-out:
			if !ok {
				return
			}
			fmt.Printf("Processed Data %d\n", i)
		case er, ok := <-err:
			if !ok {
				return
			}
			if er != nil {
				fmt.Println(er)
			}
		}
	}
}

Notice how I am defining the channels in the main and passing those along instead of redefining the channel

There's a few things here you have to fix.

This is causing @zerkms's race condition:

    go func(in chan int) {
out, errors = run(in)
}(in)

out, errors must be set to the results of run(in) before you start selecting on those channels below, so you don't want to run it in a goroutine. You maked them at the beginning of main, and then remaked them in run.

    out, errors = run(in)

this makes main's makeing of channels out and errors wasteful, since run is going to create new ones.

	var out chan int
var errors chan error

    go func(in chan int) {
... 

Passing the channel into the function like this is a good way to make the scope explicit. For complex programs I avoid defining nontriial goroutine functions inline sos that it doesn't share scope.

Other than that, your code will run, but it's missing a key component of parallel processing: you aren't doing anything in parallel. You only have one worker goroutine, the one run creates.

As a workload, determining an integer's parity is a good conceptual stand-in for easily parallelized work. But you're processing work one item at a time - in reality, channels are useful in situations when multiple items of work can be processed at once.

If you have multiple workers, you can't close the out and error channels in each worker. No worker knows whether the others are done processing. In fact, none of your goroutines currently know whether all workers are done. You know that in is closed, but if there are multiple workers, this does not mean all of in's messages are finished processing.

This is where sync.WaitGroup comes in. With a sync.WaitGroup you can count the number of workers you've spawned, then wait for them all to be done. This allows you to close out and errors after all workers are done.

First we'll rewrite the worker to be able to be invoked many times. Our worker is essentially the same, but instead of closing out and errs, we
signal we are complete with wg.Done().

func worker(wg *sync.WaitGroup, in <-chan int, out chan<- int, errs chan<- error) {
defer wg.Done()
for i := range in {
if i%2 == 0 {
out <- i
} else {
errs <- fmt.Errorf("we don't like odd numbers (%d)", i)
}
}
}

Now we need to create some workers

	var in = make(chan int)
var out = make(chan int)
var errs = make(chan error)
var wg sync.WaitGroup
for i := 0; i < 10; i++ {
wg.Add(1)
go worker(&wg, in, out, errs)
}

And a goroutine to wait for the wait group before closing out and errs

	go func() {
wg.Wait()
close(errs)
close(out)
}()

We also need to adjust this behavior for both channels:

        case i, ok := <-out:
if !ok { // out channel is closed
return // Done
}

first, a correction:

            if !ok { // out channel is closed

!ok means that the channel is closed and empty.

With one worker, closing out or errors are equivalent signals that the program is done, and you know there will be no more sends on the channel because the one worker is exiting. But with many workers, we can't be sure all messages from all channels have been .If one of out and errors are empty and closed, the program ends before the other channel's messages are received.

Instead, use a neat feature of Go: select won't read from a nil channel. So as the channels close and become empty, we can set them to nil . When they're both nil, we're done.

	for errs != nil || out != nil {
select {
case i, ok := <-out:
if !ok { // out channel is closed
out = nil
}
fmt.Println("done", i)
case err, ok := <-errs:
if !ok {
errs = nil
}
if err != nil {
fmt.Println(err)
}
}
}

Here's the whole thing.

Using wait groups and channels together like this is a common practice. Typically I find it simpler and more useful to have a return type consisting of input item, reuslt, and error instead of having two return channels, one for results and one for errors.

type workResult struct {
err error
parity bool
input int
}

Often when parallelizing work it's useful to be able to associate the input with the error. Then you don't have to do any selecting either and collecting results looks more like a simple:

for res := range results {
if res.err != nil {  
// handle error ... 
} else {
// do whatever with results
}
}

When you're handling multiple errors you might consider using https://pkg.go.dev/github.com/hashicorp/go-multierror which is a useful way to collect and coalesce errors into one final value while still allowing successful work to continue.