par
par is a Go library to process calls with the following characteristics:
- calls will be processed in parallel
- parallelism may be optionally bounded
- leverages Google's Context library to allow calls to be timed our or actively canceled
Example - Coin Toss
Let's start off with a Coin Toss example. Suppose we would like to flip a coin N times in parallel. Using par, we could write the following:
import (
"code.google.com/p/go.net/context"
"github.com/savaki/par"
"math/rand"
)
func flipCoin(results chan CoinFlip) par.RequestFunc {
return func(context.Context) error {
if rand.Intn(2) == 0 {
results <- Heads
} else {
results <- Tails
}
return nil
}
}
func simple() {
flips := 10
// 1. create a channel to hold your results
results := make(chan CoinFlip, flips)
// 2. create a channel of requests
requests := make(chan par.RequestFunc, flips)
for flip := 0; flip < flips; flip = flip + 1 {
requests <- flipCoin(results)
}
close(requests)
// 3. execute the flips in parallel
_ = par.Requests(requests).Do()
// 4. results channel now has your results and can be ranged over
}Find the weather in 3 cities with a concurrency of 1
Seems like a lot of work to execute 10 parallel flips. But let's take a more realistic scenario. Suppose we want to check the weather in three cities. Using openweathermap, we can query the weather in any one city. For our three cities, querying them one at a time is slow. par allows us to query these cities in parallel. As follows:
package main
import (
"code.google.com/p/go.net/context"
"github.com/savaki/par"
"github.com/savaki/openweathermap"
"log"
)
func ok(err error) {
if err != nil {
log.Fatalln(err)
}
}
func find(city string, responses chan *openweathermap.Forecast) par.RequestFunc {
return func(ctx context.Context) error {
forecast, err := openweathermap.WithContext(ctx).ByCityName(city)
ok(err)
responses <- forecast
return nil
}
}
func main() {
// create a channel to capture our results
forecasts := make(chan *openweathermap.Forecast, 3)
// create our channel of requests
requests := make(chan par.RequestFunc, 3)
requests <- find("San Francisco", forecasts)
requests <- find("Oakland", forecasts)
requests <- find("Berkeley", forecasts)
close(requests) // important to remember to close the channel
// execute the requests with a concurrency of 1
resolver := par.Requests(requests).WithConcurrency(1)
err := resolver.Do()
ok(err)
// the forecasts channel now contains all our forecasts
close(forecasts)
cities := map[string]*openweathermap.Forecast{}
for forecast := range forecasts {
cities[forecast.Name] = forecast
}
}
Here we're using a go client for openweathermap to get the weather by city. Since we want to be a well behaved netizen, we don't want to flood their service, so we'll execute 1 query at a time.
Results
You can see from the results, that we wait until we receive a query before sending the next one.
22:00:07.800 349us 349us par - request: 0
22:00:08.393 593ms 593ms openweathermap - http - ok
22:00:08.393 593ms 593ms par - received - 0
22:00:08.393 3us 3us par - request: 1
22:00:08.684 290ms 290ms openweathermap - http - ok
22:00:08.684 290ms 290ms par - received - 1
22:00:08.684 3us 3us par - request: 2
22:00:08.869 184ms 184ms openweathermap - http - ok
22:00:08.869 184ms 184ms par - received - 2
22:00:08.869 5us 5us par - finished
Example - Find the weather in 3 cities with a concurrency of 3
As we've stated before, this can be slow for the end user. So now, we've talked to openweathermap and set up an account and now our concurrent limit has been upped to 3. Wonderful. Let's first look at what code change would be required. Here we add .WithConcurrency(3) to indicate make at most three calls at a time.
resolver := par.Requests(requests).WithConcurrency(3)
err := resolver.Do()
ok(err)
Results
Looking at our results we can see what calls 0, 1, 2 go through immediately since our limit is 3. Great.
22:00:58.584 155us 155us par - request: 0
22:00:58.584 40us 40us par - request: 1
22:00:58.584 8us 8us par - request: 2
22:00:58.781 196ms 196ms openweathermap - http - ok
22:00:58.781 196ms 196ms par - received - 0
22:00:58.786 5ms 5ms openweathermap - http - ok
22:00:58.786 5ms 5ms par - received - 1
22:00:58.787 1ms 1ms openweathermap - http - ok
22:00:58.787 1ms 1ms par - received - 2
22:00:58.787 4us 4us par - finished
Example - Find the weather in 3 cities with a concurrency of 2 and timeout of 250ms
Now let's suppose we can't afford 3 concurrent calls with this service provide and we need to drop down to two. We still want to provide a responsive service to our users, so let's set an upper bound on how long the call can take via .DoWithContext(ctx) the excellent context library is courtesy of Google.
resolver := par.Requests(requests).WithConcurrency(2)
ctx, cancel := context.WithTimeout(context.Background(), 250*time.Millisecond)
defer cancel()
err := resolver.DoWithContext(ctx)
Result
So now we see that we're only making two calls at a time and that we didn't finish our third call before the timeout happened.
22:06:14.061 381us 381us par - request: 0
22:06:14.061 422us 422us par - request: 1
22:06:14.237 176ms 176ms openweathermap - http - ok
22:06:14.237 175ms 175ms par - received - 0
22:06:14.237 2us 2us par - request: 2
22:06:14.248 11ms 11ms openweathermap - http - ok
22:06:14.249 11ms 11ms par - received - 1
22:06:14.312 63ms 63ms par - timeout
Example - Find the weather in 3 cities with an unbounded pool, redundancy of 2 and timeout of 200ms
Sometimes, it's not about trying to fit within the bounds of an api agreement. Sometimes, we want to provide the best experience possible for our users. Every now and then, when you invoke a service, you get a slow response. Maybe it's one in ten times you get 500ms of latency. 90% of the time, it's fine. But 10% of the time, ugh. Now instead of a single call, if you made two redundant calls and took the fast responses, your 1 in 10 chance now drops to 1 in 100. Much better!
The more concurrent calls to other services you make, the more likely you'll want to make redundant calls to minimize the latency issues that will arise. Let's take a look at our example again. Here we'd like to use unbounded parallelism (make as many concurrent calls as you like) with a redundancy of 2 (make each city's weather request twice). We can do that via the .WithRedundancy(2) call.
Here's the core element of code.
resolver := par.Requests(requests).WithRedundancy(2)
ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
defer cancel()
err := resolver.DoWithContext(ctx)
Result
In the results, you can see we're making multiple calls for each of our cities, but in spite of that, we're still not able to finish in the 200ms response timeout window.
22:12:41.270 649us 649us par - request: 0
22:12:41.270 45us 45us par - request: 1
22:12:41.270 10us 10us par - request: 2
22:12:41.270 13us 13us par - request: 0
22:12:41.270 11us 11us par - request: 1
22:12:41.270 10us 9us par - request: 2
22:12:41.453 183ms 183ms openweathermap - http - ok
22:12:41.453 183ms 183ms par - received - 1
22:12:41.455 2ms 2ms openweathermap - http - ok
22:12:41.455 2ms 2ms par - received - 2
22:12:41.457 1ms 1ms openweathermap - http - ok
22:12:41.457 1ms 1ms par - received - 1
22:12:41.458 1ms 1ms openweathermap - http - ok
22:12:41.463 5ms 5ms openweathermap - http - ok
22:12:41.466 2ms 2ms openweathermap - http - ok
22:12:41.470 13ms 13ms par - timeout
Debugging output
To see the debug output for yourself, execute the following in your shell:
export DEBUG=\*