Hacking on the KEDA HTTP Addon, I found myself having to do something familiar:
Split some work into N pieces, do them all concurrently, wait for them all to be done, and then merge all the results together.
I've done this a bunch of times before, but this time I forgot how to do it. I took a few minutes away from the computer to gather my thoughts and came back to it. So I don't have to forget how to do it again, I want to write the algorithm down here!
First thing's first - we need a problem we can break down into a bunch of pieces. Sometimes it's called an "embarassingly parallel" problem.
Note that concurrency and parallelism aren't equivalent, but I'm going to be using the word "parallel" hereafter because I'm hoping the machine you run this algorithm on will be able to execute the work on different cores simultaneously.
The primary goal here is to run each different piece of the work in a goroutine. That's pretty easy in Go - just put go before the function call that does the work. The tougher part is to get the results of the work, check for errors, and wait for them all to be done -- not necessarrily in that order 🤣.
Even though the concept is simple, there is a big-ish gotcha when the rubber hits the road. Below is some code that does "fake" work, annotated with comments to explain it:
workToDo := []int{"do", "some", "work"}
for idx, work := range workToDo {
// make sure you pass the index and work into the
// function that runs in the goroutine.
// this mechanism makes sure that the goroutine
// gets a (stack-allocated) _copy_ of the data.
// if you don't do this, idx and work will change out
// from under it as the loop progresses.
go func(idx int, work string) {
fmt.Println(idx, work)
}(idx, work)
}
The biggest gotcha is in that comment inside the for loop.
Now that we've got goroutines running with the right parameters, let's add a sync.WaitGroup to the mix. This mechanism will let us wait for all these goroutines to finish.
var wg sync.WaitGroup
workToDo := []int{"do", "some", "work"}
for idx, work := range workToDo {
// add 1 to the waitgroup _before_ you start the goroutine.
// you want to do this in the same goroutine as where
// you call wg.Wait() so that you're sure that, even if
// none of the goroutines started yet, you have the
// right number of pending work.
wg.Add(1)
// make sure you pass the index and work into the
// function that runs in the goroutine.
// this mechanism makes sure that the goroutine
// gets a (stack-allocated) _copy_ of the data.
// if you don't do this, idx and work will change out
// from under it as the loop progresses.
go func(idx int, work string) {
// wg.Done() tells the WaitGroup that we're done in
// this goroutine. In other words, it decrements
// the internal WaitGroup counter, whereas wg.Add(1)
// above increments it.
// Most commonly, we just do this in a defer statement.
defer wg.Done()
// this is the "work". in the next section, we'll be
// changing this to return a value, because we'll
// need to send that value somewhere
fmt.Println(idx, work)
}(idx, work)
}
// wait for all the goroutines to finish. this call
// blocks until the WaitGroup's internal count goes
// to zero
wg.Wait()
So, now we know when all the work is done, we need to get the results. There are two kinds of results we need to get - the actual values of the work we're doing -- we'll call this the "success value" -- and the errors that it might have returned.
Let's focus on the success values first. We're going to use one group of channels, one "final" channel, and a clever way of shuttling data between the former and the latter:
Read this code on the Go Playground, if you prefer
// this is the channel that will hold the results of the work
resultCh := make(chan string)
var wg sync.WaitGroup
workToDo := []string{"do", "some", "work"}
for idx, work := range workToDo {
// add 1 to the waitgroup _before_ you start the goroutine.
// you want to do this in the same goroutine as where
// you call wg.Wait() so that you're sure that, even if
// none of the goroutines started yet, you have the
// right number of pending work.
wg.Add(1)
// this is the loop-local channel that our first goroutine
// will send its results to. we'll start up a second
// goroutine to forward its results to the final channel.
ch := make(chan string)
// make sure you pass the index and work into the
// function that runs in the goroutine.
// this mechanism makes sure that the goroutine
// gets a (stack-allocated) _copy_ of the data.
// if you don't do this, idx and work will change out
// from under it as the loop progresses.
go func(idx int, work string) {
// this is the "work". right now, it just returns an
// int. in the next section, it will return both an int
// and an error
res := doSomeWork(idx, work)
ch <- res
}(idx, work)
// start up another goroutine to forward the results from
// ch to resultCh
go func() {
// we want to indicate that we're done after we forward
// the result to the final channel, _not_ just when we're
// done with the actual computation. this arrangement
// will be useful below, in our final goroutine that
// runs after the for loop is done
defer wg.Done()
res := <-ch
resultCh <- res
}()
}
// start up a final goroutine that is going to watch for
// the moment when all of the loop goroutines are both
//
// 1. done with their work
// 2. done sending their results to the final channel
//
// after that, we can close the resultCh. this closure is
// important for the following for loop, since ranging over
// a channel will only stop after that channel is closed
go func() {
wg.Wait()
close(resultCh)
}()
// now that we have that final goroutine running, we can
// be sure that this for loop will end after:
//
// 1. all goroutines are done with their work
// 2. all goroutines are done sending their work to resultCh
// 3. we have processed each result
// (in this case, we just print it out)
for result := range resultCh {
fmt.Println("result:", result)
}
The code is extensively commented, but notice a few more "meta" things about it:
2N goroutines rather than N (where N is the number of work items to do).
wg.Wait() in the main goroutine. Instead, we range over resultCh to both get the results and determine when all the work items are done.Now that you (hopefully) have a decent grasp over the code in the previous section, consider that, for most workloads, you'll also have to deal with error handling. It doesn't take a lot of additional code to do it, but it does add a bit more complexity. Let's see how it works:
Read this code on the Go Playground, if you prefer
Note that you can reduce complexity by using the
errgrouppackage. The code herein implements functionality similar to that oferrgroup.
// this is the channel that will hold the results of the work
resultCh := make(chan string)
// this channel receives all the errors that occur.
// for each work item, either resultCh or errCh will receive
// precisely once. both channels will be closed immediately
// after all receives happen
errCh := make(chan error)
var wg sync.WaitGroup
workToDo := []string{"do", "some", "work"}
for idx, work := range workToDo {
// add 1 to the waitgroup _before_ you start the goroutine.
// you want to do this in the same goroutine as where
// you call wg.Wait() so that you're sure that, even if
// none of the goroutines started yet, you have the
// right number of pending work.
wg.Add(1)
// this is the loop-local channel that our first goroutine
// will send its results to. we'll start up a second
// goroutine to forward its results to the final channel.
ch := make(chan string)
// this is the loop-local channel that our first goroutine
// will send errors on. for each loop iteration, exactly
// one of ch or errCh will receive
eCh := make(chan error)
// make sure you pass the index and work into the
// function that runs in the goroutine.
// this mechanism makes sure that the goroutine
// gets a (stack-allocated) _copy_ of the data.
// if you don't do this, idx and work will change out
// from under it as the loop progresses.
go func(idx int, work string) {
// this is the "work". right now, it just returns an
// int. in the next section, it will return both an int
// and an error
res, err := doSomeWork(idx, work)
if err != nil {
eCh <- err
} else {
ch <- res
}
}(idx, work)
// start up another goroutine to forward the results from:
//
// - ch to resultCh
// - eCh to errCh
go func() {
// we want to indicate that we're done after we do the
// forward operation, similar to the code in the
// previous section
defer wg.Done()
// only one forward operation will happen per loop
// iteration, so we use a select to choose exactly
// one of the channels - either the success or error
// one.
select {
case res := <-ch:
resultCh <- res
case err := <-eCh:
errCh <- err
}
}()
}
// start up a final goroutine that is going to watch for
// the moment when all of the loop goroutines are both
//
// 1. done with their work
// 2. done sending their results to the appropriate channel
//
// after that, we can close both resultCh and errCh.
go func() {
wg.Wait()
close(resultCh)
close(errCh)
}()
// we're now at a point where we have two "final" channels:
//
// - one for the successful results
// - one for the errors
//
// we have a few choices on how to handle them, and it's
// largely up to your use case how you handle errors or success
// results. In our case, we'll loop through both channels,
// print out the result either way, and then exit when all
// receives happen.
// these two booleans are going to keep track of when
// each channel is closed and done receiving
resultsDone := false
errsDone := false
// we're going to use an infinite loop and break out of it
// when both channels are done receiving
for {
if resultsDone && errsDone {
break
}
select {
case res, valid := <-resultCh:
if !valid {
resultsDone = true
} else {
fmt.Println("result:", res)
}
case err, valid := <-errCh:
if !valid {
errsDone = true
} else {
fmt.Println("error:", err)
}
}
}
A few more things to note here:
errCh and resultCh (respectively), or tell the remaining goroutines to exit.
contexterrgroup package.