Keeping a steady pace with Go
When starting with Go, one of the best resources to check is the concurrency section of the Tour Of Go to learn the basics of channels and goroutines.
At the end you will find an excellent talk about Go Concurrency Patterns by Rob Pike with its Slides.
I’d like to expand on one of the patterns presented there “Restoring Sequencing”, proposing a more scalable approach.
But first, what is this pattern and why would us need that?
Suppose we have two goroutines generating some output and we would like to fan-in into a common output as the next slide shows.
One way to solve this is to use a fan-in function to let whosoever is ready talk.
If you run the above code snippet you will find that the output of those two goroutines may not be in sync, this means that it is possible the same goroutine prints twice breaking the order.
So what if we would like to prevent that? What if we want no matter the speed on which each goroutine produces its output make it wait its turn?
What we need is to restore sequencing, how do we do it? The talk show us two slides.
Each producer has a ‘waitForIt’ channel which sends on the Message channel to allow the synchronization. Each producer will be blocked until a boolean is sended through that private channel. Let’s look at the full source code.
This works but it has a clear drawback, what if we were to have more than two producers? We will have to send a wait signal to each of them. How can we make this scale? Ideally we would have a single wait signal so no matter how many producers there are, we just send one wait per for loop cycle.
In order to do this we will need to have a centralized wait signal that each producer will receive. We could make each boring function receive the message string plus this mentioned wait signal, but we are going to take a more funcional approach and take advantage of Go closures and functions as first class citizens. Bear with me.
Now our not so boring function returns another function that receives the wait signal. Each time we send a Time in the wait signal channel, it will produce a Message.
You may ask: Nico, why do you need that extra level of function wrapping?
Well, remember I want to have N amount of producers so passing the wait signal manually to each of them doesn’t seem like a good idea. It would be great if we could somehow receive a wait signal and spread it across all the producers, let’s do that now.
The syncAll function receives the centralized wait signal plus a collection of producers that need such a wait signal. This is why it was a good idea to make the boring function return a function that was expecting the wait signal. The responsability of the syncAll function is to clone the wait signal as many times as inputs are received and to give each input a clone of the wait signal. To do this it uses a helper function called demuxSignal that is defined as follows.
The demuxSignal function receives a wait signal and the amount of child signals it must create. Every time a wait signal is emitted, it will be propagated to all its children. Think of it as a one to many mapping.
Let’s put it all together!!!
We have made good progress, now we only need just one wait signal instead of one for each producer. But it also contains a fatal flaw, we need as many channel reads as producers. See what happens if you comment out any of the three msg. Proposed exercise, think why that happens.
How can we read out the channel “c” without taking into account the amount of producers?
The trick is to have a separated goroutine that will work as a metronome and a separate channel “nextBatchSignal” that basically does the same as “waitSignal” the difference is that “waitSignal” is consumed by the producers so if we try to read from it we will never get a value. Try it locally, because the error displayed by the playground is not what one would expect.
As a bonus, lets leverage on demuxSignal to create a single signal to manage everything for us.
Here we are exploiting a very important property of demuxSignal which is that the order of the channels returned is the same order of how the signal is propagated.
And that’s all, we now have a single wait signal and we do not need to know the amount of producers we may have in advance :)
