Cloud Native Patterns Illustrated: Fan-in and Fan-out
Tags: go, cloudnative
Published: 2022-10-31 16:35:57

---
title: Cloud Native Patterns Illustrated: Fan-in and Fan-out
published: true
date: 2022-10-31 16:35:00 UTC
tags: Go,CloudNative,Golang
canonical_url: https://www.sazak.io/cloud-native-patterns-illustrated-fan-in-and-fan-out/
series: Cloud Native Patterns Illustrated
---
![Cloud Native Patterns Illustrated: Fan-in and Fan-out](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/er1ksscsc0zxej0lle4u.png)

I've started reading technical books after my graduation from CS to broaden my knowledge on many topics, such as system design, low-level CS, cloud native, etc. Nowadays, I'm reading [Cloud Native Go](https://www.oreilly.com/library/view/cloud-native-go/9781492076322/) by Matthew Titmus, which gives a smooth introduction to the cloud native world using the benefits of Go. The book starts with Go basics and cloud native patterns, and that's where I thought "I should write about these patterns!".

In this series of blog posts, I will write about the cloud native patterns I've learnt – also with illustrations using [Manim](https://www.manim.community).

## The Problem

Let's say we have a constant data source (such as a Go [channel](https://go.dev/tour/concurrency/2)), and we need to process the data and put the processed data into a destination channel with a latency as low as possible.

In the general case, we have a processor function which is in the middle of source and destinations, which processes the data packets (packet here is an abstraction for simplicity) one by one.

![Cloud Native Patterns Illustrated: Fan-in and Fan-out](https://www.sazak.io/content/images/2022/10/1.gif)

However, every processing function have latency. This may be because of network or CPU usage, blocking syscalls, etc. If we send enough packets per second to the processor function, voilà! We now have a bottleneck!

![Cloud Native Patterns Illustrated: Fan-in and Fan-out](https://www.sazak.io/content/images/2022/10/2.gif)

## Solution

There's a fairly simple solution to this problem: using multiple processors inside the data pipeline. This way, we can process the data stream concurrently, which drops the overall latency and decreases the pipeline congestion.

![Cloud Native Patterns Illustrated: Fan-in and Fan-out](https://www.sazak.io/content/images/2022/10/3.gif)

## That's where Fan-in and Fan-out comes in…

We can implement this solution by utilizing shared memory, such as message queues.

With this approach, we will split the incoming data packets into different input queues. Then, each queue's own processor will take the packets one by one, process the data, and put the processed data into its related output queue. Finally, the destination (another processor, queue, or some other system) will take the processed data packets from each of the output queues, and aggregate them into a single data source.

The first approach, splitting the data source (input) to multiple data sources (input queues) is called the `Fan-out` pattern. The second one, aggregating multiple data sources (output queues) into a single data source (destination) is called the `Fan-in` pattern.

> For simplicity, we've specified one input queue and one output queue for each processor in our pipeline. We could use multiple input/output queues per processor, or a shared queue between a few processors, depending on the overall system requirements.

![Cloud Native Patterns Illustrated: Fan-in and Fan-out](https://www.sazak.io/content/images/2022/10/4.gif)

## Go Implementation

Let's get our hands dirty with some concurrent Go! First, let's define our data sources:

```go
func stringGenerator(prefix string, count uint) <-chan string {
	ch := make(chan string)

	go func() {
		defer close(ch)
		for i := uint(0); i < count; i++ {
			ch <- fmt.Sprintf("%s-%d", prefix, i)
		}
	}()

	return ch
}
```

The function `stringGenerator` creates a receive-only string channel, creates a [goroutine](https://go.dev/tour/concurrency/1) that puts prefixed strings into the channel, and returns the channel. We read from this channel later in the Fan-out code.

Our processor function will be fairly simple too:

```go
func stringProcessor(id int, source <-chan string) <-chan string {
	ch := make(chan string)

	go func() {
		defer close(ch)
		for data := range source {
			ch <- fmt.Sprintf("P-%d processed message: %s", id, data)
			millisecs := rand.Intn(901) + 100
			time.Sleep(time.Duration(millisecs * int(time.Millisecond)))
		}
	}()

	return ch
}
```

Inside the processor function, we will wait for a random while to simulate the processor latency.

### Fan-out Implementation

In the Fan-out implementation, we will take a receive-only channel, and return a slice of receive-only channels:

```go
// Splitter implements the Fan-out pattern
func Splitter(source <-chan string, numDests uint) []<-chan string {
	destinations := make([]<-chan string, 0, numDests)

	for i := uint(0); i < numDests; i++ {
		dest := make(chan string)

		go func() {
			defer close(dest)
			for data := range source {
				dest <- data
			}
		}()

		destinations = append(destinations, dest)
	}

	return destinations
}
```

Goroutines created inside the `Splitter` function will handle the data routing logic. Notice that inside the goroutines, we used a single `range` statement to receive from the `source` channel:

```go
for data := range source {
	dest <- data
}
```

This means each goroutine will try to read from the channel in the loop, and the first one to do the read will receive the next item. In other words, **each goroutine will fight over the next data instance**.

We could use a centralized solution for Fan-out instead of competing goroutines. In that case, we would define a `master process` to distribute each incoming data instance (`string` in our case) among all the output channels.

### Fan-in Implementation

In Fan-in, we will basically do the reverse of Fan-out, with a few differences:

```go
// Aggregator implements the Fan-in pattern
func Aggregator(sources []<-chan string) <-chan string {
	destination := make(chan string)

	var wg sync.WaitGroup
	wg.Add(len(sources))

	for _, source := range sources {
		go func(src <-chan string) {
			defer wg.Done()
			for data := range src {
				destination <- data
			}
		}(source)
	}
	go func() {
		wg.Wait()
		close(destination)
	}()

	return destination
}
```

The `Aggregator` function takes a slice of receive-only input sources and returns a single receive-only output channel. Inside, we created a goroutine for each input source, which reads from the source continuously and populates the output channel (`destination`) with the data it received.

Notice that we've used a `sync.WaitGroup` to wait for the aggregator goroutines to finish. After an input source (`channel`) is closed, the for loop inside the respective goroutine will end, and it will finish its job.

After all input sources are closed, we're done with aggregating, and we now can close the destination channel. **This is a crucial step** , and if we don't close the channel we have created, the Go runtime will complain with a fatal error:

```
fatal error: all goroutines are asleep - deadlock!
```

### main()

Putting all of our functions together, we're ready to run our code:

```go
const (
	processorCount = 10
)

func main() {
	splitSources := Splitter(stringGenerator("test", 100), processorCount)
	procSources := make([]<-chan string, 0, processorCount)

	for i, src := range splitSources {
		procSources = append(procSources, stringProcessor(i, src))
	}

	for out := range Aggregator(procSources) {
		fmt.Println(out)
	}
}
```

## tl;dr

- Data pipelines with a single processor will eventually bump into a bottleneck
- Splitting the input source into several queues and processing concurrently is helpful
- This split-process-aggregate pattern is called Fan-in/Fan-out

Thank you for reading! For more blog posts on Go and cloud native, take a look at my blog [sazak.io](https://sazak.io).