(Almost) ranging over multiple Go channels simultaneously

e · l · n
Oct 5, 2017

In my experiments (flowbase, scipipe) in using Flow-based programming (FBP) principles in pure Go, there is a common pattern that occuring all the time: The need to synchronously read a set of values from multiple channels at the same time. That is, if I have three in-bound channels, chA, chB, chC for an FBP process (for people new to the ideas of FBP in pure Go, that is basically a struct wrapper around a go-routine), I typically want to receive one set of values a, b, c at a time from these channel, do something, and then potentially send output on any number of outbound channels, or write results to stdout or a file.

I have tried various ways to do this, but most of them result in clunky, verbose and error prone code, because of the need to keep track of which channels have closed, and not, etc. Thus, it is not something one would like to do everytime creating a new flowbase component, for example.

Thus, optimally, one would want to use Go's very handy range keyword for looping over multiple channels, since range that takes care of closing the for-loop at the right time, when the inbound channel is closed. So something like this (N.B: non-working code!):

for a, b, c := range chA, chB, chC {
    doSomething(a, b, c)
}

Unfortunately this is not possible, and probably for good reason (how would it know whether to close the loop when the first, or all of the channels are closed? ... etc).

Anyways, I started a twitter thread to discuss this. Then, I read this post by Dave Cheney, which explains what the range keyword does under the hood when looping over channels. He pointed out that this:

for a := range chA {
    doSomething(a)
}

... is actually the same as this more detailed code:

for a, ok := <-chA; ok; a, ok = <-chA  {
    doSomething(a, b, c)
}

... which is using one variant of Go's for-syntax, which is:

for init variables; condition for continuing; what to do after each iteration  {
    doSomething(a, b, c)
}

See the relevant section in Effective Go for more info.

Together with a tweet by Inanc Gumus in the mentioned thread, I realized one can do something like this by using a little bit of helper funcs:

package main

import "fmt"

func main() {
	bufSize := 1 // Try changing this to 0, and see what happens!
	ch1, ch2, ch3 := make(chan int, bufSize), make(chan int, bufSize), make(chan int, bufSize)
	go func() {
		defer close(ch1); defer close(ch2); defer close(ch3)
		ch1 <- 1; ch2 <- 2; ch3 <- 3
		ch3 <- 6; ch2 <- 5; ch1 <- 4 // Notice the reversed order of sends(!)
	}()

	// THIS IS WHERE TO LOOK:
	for rs, oks := readOnce(ch1, ch2, ch3); allTrue(oks); rs, oks = readOnce(ch1, ch2, ch3) {
		fmt.Println(rs)
	}
}

// -------------------------------------------------------------------------------
// Helper functions
// -------------------------------------------------------------------------------

func readOnce(chs ...chan int) ([]int, []bool) {
	rs := []int{}
	oks := []bool{}
	for _, ch := range chs {
		r, ok := <-ch
		rs = append(rs, r)
		oks = append(oks, ok)
	}
	return rs, oks
}

func allTrue(vs []bool) bool {
	for _, v := range vs {
		if !v {
			return false
		}
	}
	return true
}

If we zoom in again on the interesting part:

for rs, oks := readOnce(ch1, ch2, ch3); allTrue(oks); rs, oks = readOnce(ch1, ch2, ch3) {
	fmt.Println(rs)
}

... we can see that this is pretty close to the more detailed version of the range-loop, that Dave Cheney showed above.

We can also note that this will work for any number of channels provided to readOnce(), as long as it is channels of type int, and that still, you will get one set of (int) results back, one array at a time, in the loop.

One must keep in mind that the close condition in the loop has to be thought-through carefully. In this example, I have created the function allTrue(), that just checks that all booleans in the input array are true, and otherwise returns false. This will close the for loop as soon as the first channel is closed. This might be OK if we are sure that all inbound channels will - by design - contain the same input items, and be closed at the same time. If not, one will need to adapt the close condition accordingly.

Anyways, I think this is at least some progress towards figuring out something as generic as possible, for reading from multiple channels simultanously and synchronosly.

Also, if you have further ideas on how to make this more succinct and generic, please let me know!

Wrap the range-like code up in a helper function

Edit: This section was added on Oct 6, 2017, 17:33 CEST

Naturally, we can now also create a helper function that wraps the somewhat messy for-loop above, so instead we would have something like this (see the syncRead() function):

package main

import "fmt"

func main() {
	bufSize := 1 // Try changing this to 0, and see what happens!
	ch1, ch2, ch3 := make(chan int, bufSize), make(chan int, bufSize), make(chan int, bufSize)
	go func() {
		defer close(ch1)
		defer close(ch2)
		defer close(ch3)
		ch1 <- 1
		ch2 <- 2
		ch3 <- 3
		ch3 <- 6
		ch2 <- 5
		ch1 <- 4 // Notice the reversed order of sends(!)
	}()

	// THIS GOT A BIT NICER NOW, RIGHT?:
	for rs := range syncRead(ch1, ch2, ch3) {
		fmt.Println(rs)
	}
}

// -- Helper funcs --------------------------------------------------------------------------------

func syncRead(chs ...chan int) chan []int {
	outChan := make(chan []int, 16)
	go func() {
		defer close(outChan)
		for rs, oks := recvOneEach(chs...); allTrue(oks); rs, oks = recvOneEach(chs...) {
			outChan <- rs
		}
	}()
	return outChan
}

func recvOneEach(chs ...chan int) ([]int, []bool) {
	rs := []int{}
	oks := []bool{}
	for _, ch := range chs {
		r, ok := <-ch
		rs = append(rs, r)
		oks = append(oks, ok)
	}
	return rs, oks
}

func allTrue(vs []bool) bool {
	for _, v := range vs {
		if !v {
			return false
		}
	}
	return true
}

The only limitation to this is of course that the the helper functions work only for channels of type int (and also all of them need to be of the same type), so you would need to create separate implementations for each type, or of course, fall back to use the empty interface (aka the G package) :o)

Keywords: Go, Golang, Pipelines, Patterns