In the Not Too Distant Future…
Building a Futures Library with Channels, Goroutines, and Selects
In previous blog posts, we’ve looked at how to use channels to build unbounded queues, pools, and manage multiple parallel requests. We’re able to do all these things with channels, and in these cases the abstraction made sense, but sometimes, we want to frame concurrency problems in a different way. For example, Node.js developers use the concept of futures (sometimes called promises) to organize callbacks and background tasks. How can we build futures in Go to achieve the same workflow?
If you aren’t familiar with futures, here’s the basic idea. They are a programming abstraction that lets you run a bit of code in the background, and get back a reference to the result immediately. When you eventually try to read the result from the reference, the future checks to see if the result has been calculated yet. If it has, it returns immediately. If not, you wait until the result is populated.
Futures have two important properties. First, you don’t have to wait for the result of a background process until you actually need the result. Second, using futures hides a lot of the messy boilerplate that is often involved in concurrent programming, making your source code much easier to follow.
Design the API
So, let’s figure out what an API for a basic future would look like. We want to:
- pass in the code to run
- return a reference that will eventually contain the value calculated by the code
Let’s look at the return type first. Since this is Go, we have a couple of language-specific details to attend to.
First, once again, there are no generics in Go, so we need to write things in the most general terms possible and have our client code cast the responses to the right data types. That means using interface{} as the returned data type.
Second, the idiomatic way to handle errors in Go is by returning an error as the final parameter, so we need to return an error also, in case our future errors out.
We can’t just return a struct with the values populated, because we return the reference immediately and populate the data later. That means that we need to invoke a method that may potentially wait for data to show up. Rather than use a concrete type, exposing an interface makes the most sense and gives us the most flexibility. So, our return type is going to look like:
Now, what about the function we are passing in? Since Go has closures, we can use one to wrap the code that we really want to run. That means that we don’t need to worry about passing in parameters to our function; we will capture them from the environment. We do need to return values from our closure so that we can populate our future.Interface implementation, so our function must return (interface{}, error).
Now that we have defined the input and the output types, here’s what the first pass at our API looks like:
We will fill in the implementation soon. Here’s what the client code looks like when creating a future:
First Implementation
Now that we know what the code should look like to the outside world, we can start thinking about how it will do what it needs to do. Let’s see if we can implement our future with only the standard Go concurrency mechanisms: select, channels, and goroutines.
First, we need a structure to hold the data and implement the interface. We don’t want to expose it to our clients, as they should only use the interface and the factory function. This means it should have a lower-case letter for the first character of its name.
You might be surprised by that done channel of type struct{}. We're going to take advantage of a pattern in Go that's used to signal when work is done. When a channel is unbuffered, trying to read from it will pause the reading goroutine until one of two things happens: a value is written to the channel or the channel is closed. Our done channel will never have a value written to it; it exists only to be closed when the work is complete. This allows you to use a closed channel as a permanent signal from one goroutine to another that it's OK to go on.
Now we need to implement New:
In our code, New creates a futureImpl, and then launches a goroutine that runs the incoming function, assigns the value and error returned by the function to the fields in our futureImpl, and closes the done channel. Meanwhile, the Get method waits until the done channel is closed and then returns the value and the error. If you call Get again, the closed done channel will return immediately, and we will get back the value and error without waiting.
We’ve got the basic structure of a future working in about 20 lines of code; not too bad at all.
If you want to see it working, check out this link on the Go Playground.
Don’t Wait Forever
Now that the basics are done, it’s time to add some more functionality to our future. The first thing that might occur to you is that it’d be good to not have to wait forever for the future to complete. After all, the future might run for a longer time than we want to wait. We should add a second method to our interface so we can limit how long we are willing to wait:
Calling GetUntil instead of Get will only wait for the specified Duration. If the result comes back before the timeout, we want to return at that point. If the result comes back after that point, we want to stop waiting. In order to know whether or not the request timed out, we return back a Boolean flag in addition to the value and the error. If the flag is set to true, then the request timed out. If it's false, the value and error that came back was the value and error calculated by our function.
In order to support GetUntil, we don't need to modify New or Get. We just need to add the following method implementation:
The select statement waits on both the done channel in f and the channel returned by time.After. Whichever returns first determines which result is returned.
Also notice that if GetUntil times out, you haven't lost your work. You can call Get or GetUntil again and get the result of the future. All that GetUntil does is prevent you from waiting until the result is calculated.
Here’s sample client code using GetUntil:
If you want to see it working, check out this link on the Go Playground.
Build a Chain
Great, our future implementation is moving along. Let’s add something else.
Another feature that futures usually have is the ability to be chained together. If you have several long-running processes that depend on each other, one after another, it’d be nice to have the future automatically call from one to the other. If one call in the chain fails, we want to stop processing the work in the chain and return back the error that caused the failure. This means a new methods needs to be added to our interface:
Our new method is called Then. It takes in a function and returns a new future interface. The function that's passed into Then is slightly different from the one that's passed in to New. This one has a single input parameter, which is the output from the previous step in the chain, hence the name.
Once again, we can add the new method implementation without modifying any of the existing code. We’re going to call the existing New function to make a new future. Since the New function expects a function with no input parameters, we're going to pass it a closure that does three things:
1. Call the Get method on the previous future in the chain.
2. Check to see if the previous future returned an error. If it did, return it.
3. Otherwise, call the next function with the result from the previous future.
Adding a Then step to the future is fairly straightforward:
With this addition, we have a fairly complete future implementation. We can run jobs in the background, choose how long we want to wait for them to complete, and chain together jobs with a very concise syntax. But adding future chaining means that there’s one more thing that would be nice to have: cancellation.
You can see this sample code executing at this link on the Go Playground.
Fight the Future
The difference between cancellation and timeout is subtle, but important. A timeout means that I don’t want to wait right now for the work to complete, but the work continues to process in the background, just in case you want to come back for the results later.
A cancellation means that I don’t care about the results, and if there are any additional steps after the current one finishes, don’t run them.
Unlike our other additions, adding cancel support requires modifying our existing code. This makes sense; we are adding a way to interrupt something that wasn’t interruptible before. Luckily, Go’s concurrency features make it easy to add this support.
First, we need to add the ability to cancel to our future’s interface:
One thing to be clear about is that Cancel can't stop the currently running function from completing; once a function starts executing, there's nothing you can do from outside the function to stop it. All we can do is immediately stop waiting for its results and prevent any additional items in the Then chain from running.
Next, we are going to modify futureImpl and add our Cancel and IsCancelled implementations:
We need to add a field to futureImpl so we can track the cancellation status. The implementations of Cancel and IsCancelled are straightforward. In Cancel, we only close the cancel channel if it's not already closed and if done wasn't closed first. Both methods take advantage of the default clause in a select to make sure that there is an immediate response.
You might notice that there’s another new field added to futureImpl. One of Go’s rules for channels is that closing a channel more than once causes a panic. Since a panic will kill a running application, we need to make sure that our cancel channel is only closed once. If multiple goroutines all call Cancel on the same future at the same time, it’s possible that more than one will make it to the default case in the select.
The Go standard library has a solution to this problem. It includes a type called sync.Once. Wrapping the cancel channel close in a closure invoked by the sync.Once.Do method ensures that only one goroutine will actually close the channel, no matter no many goroutines attempt to do so at the same time.
In order to populate and use the cancel channel, we need to modify New, Then, Get, and GetUntil:
The changes to Get and GetUntil simply add a check for the cancel channel. The changes to New are slightly more involved. We want to cancel once and affect the entire future chain. The easiest way to do so is for all of the futures in the chain to listen to the same cancel channel. The initial entry in the chain won't have a cancel channel, so we need to create it. All subsequent Steps added in Then will use the cancel channel created for the first future.
Sharing a reference to the same channel broadcasts the cancel channel’s state to all of the steps in the future chain. Since a closed channel always returns immediately, all of the goroutines are able to read a value without conflicting.
Here’s some sample code that demonstrates cancellation:
You can run this code at this link on the Go Playground.
Put the Future in Context
What if we want something outside of our immediate function to trigger cancellation? Maybe we have a request that should only run for a certain amount of time. In that case, we should use a Context. Go 1.7 added Context into the standard library in 1.7. But what is it?
Context is the solution for two overlapping problems in Go:
- How do I support goroutine-local variables?
2. How do I signal to a goroutine that it should stop working?
Context is an interface that provides access to:
- Key/value pairs that are passed from earlier stages in a request to later stages (like getting the user in an authentication handler and using it as part of a database query).
- A done channel that closes when it’s time to stop work on the current request or when the context was explicitly cancelled.
- A deadline that tells you when the done channel will close.
- An error that tells you why the done channel was closed The context package contains functions to create contexts with deadlines and values, and to get access to the cancel function for a context.
We can piggy-back on the context’s cancellation support to create timed cancelations for our futures. All it takes is adding a new factory function for futures, NewWithContext:
If we are supplied a Context with its cancellation channel set, we need to set up a goroutine that waits to see if the Context is cancelled before the future finishes its work. If it is, we trigger the future’s cancellation.
The code to use it looks like this:
You can run this code at this link on the Go Playground.
Future Extensions
There are additional features that are found in other future implementations. There are many options, but some of them include:
- Getting the value and error and returning immediately whether or not it is done.
- Stages that only run when there’s an error.
- Specifying multiple functions for a future to run and waiting until any one of them completes.
- Specifying multiple functions for a future to run and waiting until all of them complete.
- Merging values from multiple futures into a single future.
Adding these features is beyond the scope of this post, but all of them can be built by using the building blocks that I’ve already outlined. Channels, goroutines, and selects allow us to build these complicated structures using straightforward code that’s far easier to understand than the usual collection of threads and mutexes provided by most other languages.
If anyone is interested in using this future library, Capital One has open sourced a more advanced version of it that can be found at https://github.com/capitalone/go-future-context. Please feel free to fork it and submit pull requests with additional features and bug fixes.
DISCLOSURE STATEMENT: These opinions are those of the author. Unless noted otherwise in this post, Capital One is not affiliated with, nor is it endorsed by, any of the companies mentioned. All trademarks and other intellectual property used or displayed are th ownership of their respective owners. This article is © 2017 Capital One.
Additional Links
