Fan Out, Fan In Pattern in Go

Originally published on Substack.

In the previous article, we already discussed the pipeline pattern in Go. Now, we’ll look at another design pattern in concurrency: fan out, fan in.

Fan out is a process of separating big tasks into many sub-tasks. And processing the sub-tasks concurrently with many goroutines. The fan-in is a process of reading all the channels concurrently with many goroutines.

There will be a lot of real use cases to use this pattern. For example, we can use this pattern to write monitoring server tools. Where we can fan out goroutines to listen for CPU and memory usage. And fanning in goroutines to create metrics and generate alerts.

Now, let’s write a basic example to show what it looks like. Our program will multiply the input with 2, and print the result. Here’s the example input and output:

Input:

2
10
50

Output:

4
20
100

Let’s first write the program without concurrency:

func main() {
  nums := []int{2, 10, 50}

  for _, num := range nums {
    fmt.Println(num * 2)
  }
}

Go will run all the calculations one by one sequentially. We only process all the calculations with only one CPU, which is not efficient.

Let’s rewrite it to use many goroutines. First, we’ll create the multiply method that accepts the number and returns the channel. The returned channel will include the multiplied number.

func multiply(num int) <-chan int {
  out := make(chan int)

  go func() {
    out <- num * 2
    close(out)
  }()

  return out
}

Then, let’s create the print method that reads all the values from many channels one by one:

func print(cs ...<-chan int) {
  for _, c := range cs {
    for n := range c {
      fmt.Println(n)
    }
  }
}

The main function:

func main() {
  c1 := multiply(2)
  c2 := multiply(10)
  c3 := multiply(20)

  print(c1, c2, c3)
}

4
20
40

Now, we’ve already processed all the calculations with concurrency. We’re already using fan-out, but we still don’t use the fan-in. As you can see in the print method, we still read all the channels one by one. We’re not using many goroutines.

We can see the problem with the approach when there is complex logic in the reading channel. Like in our real use case example (creating monitoring tools). After we read the CPU and memory usage, we generate metrics and alerts. So, when we have 5 channels to read, our program will need to generate 5 metrics one by one, not concurrently. Which is not efficient.

Now, let’s rewrite the printer method to read all the channel values concurrently:

func merge(cs ...<-chan int) <-chan int {
  var wg sync.WaitGroup
  out := make(chan int)

  output := func(c <-chan int) {
    for n := range c {
      out <- n
    }
    wg.Done()
  }

  wg.Add(len(cs))
  for _, c := range cs {
    go output(c)
  }

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

  return out
}

We rename the function into the merge method. Because in this function, we no longer print all the values. Instead, we merge values from 3 channels into 1 channel.

We’ll use sync.WaitGroup to track the reading process for all goroutines. Also trigger “wg.Wait” and close the channel in another goroutine.

Now, the printer will be done in the main method:

func main() {
  c1 := multiply(2)
  c2 := multiply(10)
  c3 := multiply(20)

  result := print(c1, c2, c3)

  for n := range result {
    fmt.Println(n)
  }
}

4
40
20

The order will be different but that’s ok, because we read the channel concurrently, so we can’t control the order.

---

That’s all for this article. I hope you can get something from it. Happy hacking and see you in another article.