Fanin Pattern in Go

Let's take a look at the fan-in pattern. This is really useful in go when we have related data from multiple threads that we need to bring together.

For example, suppose you have made multiple API calls to different services and you need to combine the results.

This is a real easy pattern to implement, but you do need to pay attention to how you handle the channels. It is easy to get a deadlock condition.

// produce is used to simulate the different data sources
func produce(id int) chan int {
    ch := make(chan int)
    go func() {
        for i := 0; i < 10; i++ {
            ch <- id*10 + i
        }
        fmt.Printf("producer %d done\n", id)
        close(ch) // this is important!!!
    }()
    return ch
}

func fanin(inputs ...chan int) chan int {
    output := make(chan int)

    var wg sync.WaitGroup
    for i, input := range inputs {
        wg.Add(1)
        go func() {
            for value := range input {
                output <- value
            }
            fmt.Printf("done merging source %d\n", i)
            wg.Done()
        }()
    }
    go func() {
        wg.Wait()
        close(output) // this is important!!!
    }()

    return output
}

func main() {
    input1 := produce(0)
    input2 := produce(1)

    result := fanin(input1, input2)

    done := make(chan bool)
    go func() {
        for value := range result {
            fmt.Printf("got %d\n", value)
        }
        close(done)
    }()
    <-done
    fmt.Println("done")
}

Here we use the produce function to simulate the different sources. These source channels are sent to the fanin function which performs the combining operation.

The fanin function creates the output channel, then launches a goroutine that operates on each input. We use a WaitGroup to indicate when all the input sources have been combined into the output channel.

In this simple example, the main thread just iterates over the output. Notice that there is no guarantee on the order, values from the 2 inputs are mixed.

A key point to bring up is that we have to close the output channel when we are done combining the inputs. The range operator will wait indefinitely once the channel is empty. Comment out the close(output) line and you'll see that you get a deadlock condition.

How can we improve this? Leave me a comment below.

Thanks!

The code for this post and all posts in this series can be found here