Pipelining with Go
A few days ago, a friend of mine posted a question about the Go concurrency control. The question was, "Can we assure that doSomething(0) always runs before doSomething(2)?" The answer would be, "No." We cannot determine the order in which goroutine can run at a preemption point as in other languages.
var mtx = [2]sync.Mutex{}
func foo() {
for i := 0; i < 4; i++ {
go doSomething(i)
}
}
func doSomething(i int) {
mtx[i%2].Lock()
// do something
mtx[i%2].Unlock()
}Though using a mutex for concurrency management is a norm in other languages (e.g., in C++), it is not in Go. Mutex is a mechanism to solve concurrency issues by synchronizing memory access. The Go language FAQ says that "Do not communicate by sharing memory. Instead, share memory by communicating."
If the issue is really we have to maintain the running order of data for every other item, we can achieve this by using the WaitGroup of the sync package.
var (
wg sync.WaitGroup
)
// For a simplicity, I haven't handled the corner cases:
// It works only when the number of inputs is a multiple of `phase`.
func foo(input []int, phase int) {
for i := 0; i < len(input); i += phase {
wg.Add(phase)
for j := i; j < i + phase; j++ {
go doSomething(j)
}
wg.Wait()
}
}
func doSomething(i int) {
defer wg.Done()
// do something
fmt.Printf("I'm %d\n", i)
}However, the issue seemed to handle an infinite input stream while maintaining input dependencies (it can be order or anything else). And, we could guess that it was a data streaming issue, which is the functional programming's turf. I agree that Go lacks functional programming features. However, we can achieve it with the core of the Go language: goroutines and channels.
In "Concurrency in Go: Tools and Techniques for Developers, Katherine Cox-Buday, O'Reilly Media," the author is saying as follows:
Go’s philosophy on concurrency can be summed up like this: aim for simplicity, use channels when possible, and treat goroutines like a free resource.
So, we can think it as a problem to make a pipeline. Let's divide the problem as followings:
- A Generator to stream the input.
- Demux the input to multiple streams according to the dependency.
- Perform doSomething parallelly.
- Mux the results into a stream.
Input Stream Generator
If you're used to a functional programming language (such as Python..), you might miss the generator (me too!) in Go. But you can create a generator with a channel.
func generator(done <-chan interface{}, input []int) <-chan int {
var (
stream = make(chan int)
)
go func() {
defer close(stream)
for _, v := range input {
select {
case <-done:
return
case stream <- v:
}
}
}()
return stream
}This phase might be redundant, but it is a basic pattern to use goroutines and channels for the streaming.
Demuxing
Then, we split the input stream into multiple streams (channels). We're merely splitting it to the given number of streams, but you can bring your dependency here.
func demux(done <-chan interface{},
numOut int, inStream <-chan int) []chan int {
var (
streams = make([]chan int, numOut)
)
// Do not forget to initialize each channel.
for i := 0; i < numOut; i++ {
streams[i] = make(chan int)
}
go func() {
defer func() {
for _, st := range streams {
close(st)
}
}()
for i:= 0;;{
select {
case <-done:
return
case val, ok := <-inStream:
if !ok {
return
}
streams[i % numOut] <- val
i++
if i == numOut {
i = 0
}
}
}
}()
return streams
}Parallel Execution
We're going to execute each stream in a goroutine. The parallel execution does not guarantee the running order across the channels (though we can maintain the order in a demuxed channel). So, we're going to define a result struct to contain the result information.
type result struct {
channel int
input int
value int
}
func execute(done <-chan interface{},
inputChans []chan int, fn func(int) int) []chan result {
var (
numInputChans = len(inputChans)
streams = make([]chan result, numInputChans)
)
for i := 0; i < numInputChans; i++ {
streams[i] = make(chan result)
}
var wg sync.WaitGroup
do := func(chanID int, input <-chan int) {
defer wg.Done()
for {
select {
case <-done:
return
case val, ok := <-input:
if !ok {
return
}
streams[chanID] <- result{channel: chanID, input: val, value: fn(val)}
}
}
}
wg.Add(numInputChans)
for i, c := range inputChans {
go do(i, c)
}
go func() {
wg.Wait()
for _, s := range streams {
close(s)
}
}()
return streams
}Muxing
Finally, we multiplex the multiple results into a stream.
func mux(done <-chan interface{},
inputChans []chan result) <-chan result {
var (
numInputChans = len(inputChans)
stream = make(chan result)
)
var wg sync.WaitGroup
multiplex := func(c <-chan result) {
defer wg.Done()
for v := range c {
select {
case <-done:
return
case stream <- v:
}
}
}
wg.Add(numInputChans)
for _, c := range inputChans {
go multiplex(c)
}
go func() {
wg.Wait()
close(stream)
}()
return stream
}Final Result
The final code and a test running result will be as follows.
package main
import (
"fmt"
"sync"
)
func generator(done <-chan interface{}, input []int) <-chan int {
var (
stream = make(chan int)
)
go func() {
defer close(stream)
for _, v := range input {
select {
case <-done:
return
case stream <- v:
}
}
}()
return stream
}
func demux(done <-chan interface{},
numOut int, inStream <-chan int) []chan int {
var (
streams = make([]chan int, numOut)
)
// Do not forget to initialize each channel.
for i := 0; i < numOut; i++ {
streams[i] = make(chan int)
}
go func() {
defer func() {
for _, st := range streams {
close(st)
}
}()
for i:= 0;;{
select {
case <-done:
return
case val, ok := <-inStream:
if !ok {
return
}
streams[i % numOut] <- val
i++
if i == numOut {
i = 0
}
}
}
}()
return streams
}
type result struct {
channel int
input int
value int
}
func execute(done <-chan interface{},
inputChans []chan int, fn func(int) int) []chan result {
var (
numInputChans = len(inputChans)
streams = make([]chan result, numInputChans)
)
for i := 0; i < numInputChans; i++ {
streams[i] = make(chan result)
}
var wg sync.WaitGroup
do := func(chanID int, input <-chan int) {
defer wg.Done()
for {
select {
case <-done:
return
case val, ok := <-input:
if !ok {
return
}
streams[chanID] <- result{channel: chanID, input: val, value: fn(val)}
}
}
}
wg.Add(numInputChans)
for i, c := range inputChans {
go do(i, c)
}
go func() {
wg.Wait()
for _, s := range streams {
close(s)
}
}()
return streams
}
func mux(done <-chan interface{},
inputChans []chan result) <-chan result {
var (
numInputChans = len(inputChans)
stream = make(chan result)
)
var wg sync.WaitGroup
multiplex := func(c <-chan result) {
defer wg.Done()
for v := range c {
select {
case <-done:
return
case stream <- v:
}
}
}
wg.Add(numInputChans)
for _, c := range inputChans {
go multiplex(c)
}
go func() {
wg.Wait()
close(stream)
}()
return stream
}
func doSomething(i int) int {
// do something
return i*2
}
func main() {
done := make(chan interface{})
defer close(done)
var (
input = []int{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}
)
for v := range mux(done, execute(done, demux(done, 2, generator(done, input)), doSomething)) {
fmt.Printf(">> [%d]: in: %d, out: %d\n", v.channel, v.input, v.value)
}
fmt.Printf(">> done\n")
}
>> [0]: in: 0, out: 0
>> [0]: in: 2, out: 4
>> [1]: in: 1, out: 2
>> [1]: in: 3, out: 6
>> [0]: in: 4, out: 8
>> [1]: in: 5, out: 10
>> [0]: in: 6, out: 12
>> [1]: in: 7, out: 14
>> [1]: in: 9, out: 18
>> [0]: in: 8, out: 16
>> doneI assumed that you were already familiar with Go, goroutines, and channels. If you're not, I strongly recommend the book "Concurrency in Go." It is a great resource (though the first two chapters are too verbose and pedantry, but I believe it is worth a read.) for the Go concurrency features. Frankly, this article's implementation got much insight from the book's pipeline and fan-out/fan-in patterns.
