Seven Concurrency Models in Seven Weeks (2014)

Chapter 6. Communicating Sequential Processes

If you’re a car nut like me, it’s easy to focus on the vehicle and forget about the roads it travels on. It’s fascinating to debate the relative merits of turbo-charging versus natural aspiration or a mid- versus a front-engine layout, forgetting that the most important aspect of a car has nothing to do with any of these things. Where you can go and how fast you can get there is primarily defined by the road network, not the car.

Similarly, the features and capabilities of a message-passing system are not primarily defined by the code between which messages are exchanged or their content, but by the transport over which they travel.

In this chapter we’ll look at a model that has surface similarities with actors but a very different feel—thanks to a difference in focus.

Communication Is Everything

As we saw in the last chapter, an actor program consists of independent, concurrently executing entities (called actors, or processes in Elixir) that communicate by sending each other messages. Each actor has a mailbox that stores messages until they’re handled.

A program using the communicating sequential processes model similarly consists of independent, concurrently executing entities that communicate by sending each other messages. The difference is one of emphasis—instead of focusing on the entities sending the messages, CSP focuses on the channels over which they are sent. Channels are first class—instead of each process being tightly coupled to a single mailbox, channels can be independently created, written to, read from, and passed between processes.

Like functional programming and actors, CSP is an old idea that’s experiencing a renaissance. CSP’s recent popularity is largely due to the Go language.^[32] We’re going to cover CSP by examining the core.async library,^[33] which brings Go’s concurrency model to Clojure.

In day 1 we’ll introduce the twin pillars upon which core.async is built: channels and go blocks. In day 2 we’ll construct a realistic example program with them. Finally, in day 3 we’ll see how core.async can be used within ClojureScript to make client-side programming easier.

Day 1: Channels and Go Blocks

The core.async library provides two primary facilities—channels and go blocks. Go blocks allow multiple concurrent tasks to be efficiently multiplexed across a limited pool of threads. But first, let’s look at channels.

Using core.async

The core.async library is a relatively recent addition to Clojure and is still in prerelease (so be aware that things may change). To use it, you need to make the library a dependency of your project and then import it. This is slightly complicated by the fact that it defines a few functions with names that clash with core Clojure library functions. To make it easier to experiment with, you can use the channels project in the book’s sample code, which imports core.async like this:

The :refer :all allows most core.async functions to be used directly, but a few (those with names that clash with core library functions) have to be given the async/ prefix.

You can start a REPL with these definitions loaded by changing the directory to the channels project and typing lein repl.

Channels

A channel is a thread-safe queue—any task with a reference to a channel can add messages to one end, and any task with a reference to it can remove messages from the other. Unlike actors, where messages are sent to and from specific actors, senders don’t have to know about receivers, or vice versa.

A new channel is created with chan:

We can write to a channel with >!! and read from it with <!!:

We’re using the handy thread utility macro provided by core.async, which, as its name suggests, runs its code on a separate thread. That thread prints a message containing whatever it reads from the channel. This blocks until we write to the channel with >!!, at which point we see the message.

Buffering

By default, channels are synchronous (or unbuffered)—writing to a channel blocks until something reads from it:

We can create a buffered channel by passing a buffer size to chan:

This creates a channel with a buffer large enough to contain five messages. As long as there’s space available, writing to a buffered channel completes immediately.

Closing Channels

The previous example demonstrated another feature of channels—they can be closed with close!. Reading from an empty closed channel returns nil, and writing to a closed channel silently discards the message. As you might expect, writing nil to a channel is an error:

Here’s a function that uses what we’ve seen so far to read from a channel until it’s closed and to return everything read as a vector:

This loops with coll initially bound to the empty vector []. Each iteration reads a value from ch and, if the value is not nil, it’s added to coll. If the value is nil (the channel has been closed), coll is returned.

And here’s writeall!!, which takes a channel and a sequence and writes the entirety of the sequence to the channel, closing it when the sequence is exhausted:

Let’s see these functions in action:

You won’t be surprised to hear that core.async provides utilities that perform similar tasks, saving us the trouble of writing our own:

The onto-chan function writes the entire contents of a collection onto a channel, closing it when the collection’s exhausted. And async/into takes an initial collection (the empty vector in the preceding example) and a channel and returns a channel. That channel will have a single collection written to it—the result of conjoining everything read from the channel with the initial collection.

Next we’ll use these utilities to investigate buffered channels in more depth.

Full Buffer Strategies

By default, writing to a full channel blocks, but we can choose an alternative strategy by passing a buffer to chan:

Here we create a channel with a dropping buffer large enough to hold five messages, and then we write the numbers 0 to 9 to it. This doesn’t block, even though the channel cannot hold so many messages. When we read its contents, we find that only the first five messages are returned—all subsequent messages have been dropped.

Clojure also provides sliding-buffer:

As before, we create a channel large enough to hold five messages, but this time with a sliding buffer. When we read its contents, we find that the five most recent messages are returned—writing to a full channel with a sliding buffer drops the oldest message. We’ll look into channels in much more detail later, but before then let’s look at core.async’s other headline feature—go blocks.

Joe asks:

What—No Automatically Growing Buffer?

We’ve now seen all three types of buffer provided by core.async as standard—blocking, dropping, and sliding. It would be quite possible to create one that simply grows as it needs to accommodate more messages. So why isn’t this provided as standard?

The reason is the age-old lesson that, whenever you have an “inexhaustible” resource, sooner or later you will exhaust it. This might be because over time your program is asked to work on larger problems, or it might be a bug that results in messages piling up because whatever should be handling them doesn’t do so.

If you avoid thinking about what to do when it happens, eventually this will lead to a damaging, obscure, and difficult-to-diagnose bug sometime in the future. Indeed, flooding a process’s mailbox is one of the few ways to comprehensively crash an Erlang system.^[34] Better to think about how you want to handle a full buffer today and nip the problem in the bud.

Go Blocks

Threads have both an overhead and a startup cost, which is why most modern programs avoid creating threads directly and use a thread pool instead (see Thread-Creation Redux). Indeed, the thread macro we used earlier today uses a CachedThreadPool under the hood.

Thread pools aren’t always very convenient to use, though. In particular, they are problematic if the code we want to run might block.

The Problem with Blocking

Thread pools are a great way to handle CPU-intensive tasks—those that tie a thread up for a brief period and then return it to the pool to be reused. But what if we want to do something that involves communication? Blocking a thread ties it up indefinitely, eliminating much of the value of using a thread pool.

There are ways around this, but they typically involve restructuring code to make it event-driven, a style of programming that will be familiar to anyone who’s done UI programming or worked with any of the recent breed of evented servers.

Although this works, it breaks up the natural flow of control and can make code difficult to read and reason about. Worse, it can lead to an excess of global state, with event handlers saving data for use by later handlers. And as we’ve seen, state and concurrency really don’t mix.

Go blocks provide an alternative that gives us the best of both worlds—the efficiency of event-driven code without having to compromise its structure or readability. They achieve this by transparently rewriting sequential code into event-driven code under the hood.

Inversion of Control

In common with other Lisps, Clojure provides a powerful macro system. If you’re used to macros in other languages (C/C++ pre-processor macros, for example), Lisp macros can seem like magic, enabling dramatic code transformations. The go macro is particularly magical.

Code within a go block is transformed into a state machine. Instead of blocking when it reads from or writes to a channel, the state machine parks, relinquishing control of the thread it’s executing on. When it’s next able to run, it performs a state transition and continues execution, potentially on another thread.

This represents an inversion of control, allowing the core.async runtime to efficiently multiplex many go blocks over a limited thread pool. Just how efficiently we’ll soon see, but first let’s see an example.

Parking

Here’s a simple example of go in action:

We start by creating a channel ch, followed by a go block that reads two values from it, then prints their sum. Although it looks like the go block should block when it reads from the channel, something far more interesting is going on.

Instead of using <!! to read from the channel, our go block is using <!. The single exclamation mark indicates that this is the parking version of a channel read, not the blocking version. As you might expect, >! is the parking version of the blocking >!!.

The go macro converts this sequential code into a state machine with three states:

1. The initial state immediately parks, waiting for something to be available for reading from ch. When it is, the state machine transitions to state 2.

2. Next the state machine binds the value read from ch to x and then parks, waiting for another value to be available, after which it transitions to state 3.

3. Finally, the state machine binds the value read from ch to y, prints a message, and terminates.

Joe asks:

What Happens If I Block in a Go Block?

If you call a blocking function, such as <!!, in a go block, you will simply block the thread it happens to be running on. Your code will probably execute OK (although if you block enough threads, you might deadlock because no more are available), but doing so defeats the purpose of using go blocks. Nothing will warn you if you make this mistake, however, so it’s up to you to be on your guard.

Happily, if you make the opposite mistake, you will be warned:

Go Blocks Are Cheap

The point of all the go macro’s cleverness is efficiency. Because (unlike threads) go blocks are cheap, we can create many of them without running out of resources. This may seem like a small benefit, but the ability to freely create concurrent tasks without worry is transformative.

You may have noticed that go (and thread, for that matter) returns a channel. This channel will have the result of the go block written to it when it’s complete:

We can use this fact to create a small function that creates a very large number of go blocks, allowing us to see just how inexpensive go blocks are:

This contender for the title “world’s most inefficient addition function” adds x to y by creating a pipeline of y go blocks, each one of which increments its argument by one.

To see how this works, let’s build it up in stages:

1. The anonymous function #(go (inc (<! %))) creates a go block that takes a single channel argument, reads a single value from it, and returns a channel containing that value incremented by one.

2. This function is passed to iterate with an initial value of (go x) (a channel that simply has the value x written to it). Recall that iterate returns a lazy sequence of the form (x (f x) (f (f x)) (f (f (f x))) …).

3. We read the y-th element of this sequence with nth, the value of which will be a channel containing the result of incrementing x y times.

4. Finally, we read the value of that channel with <!!.

Let’s see it in action:

So that’s 100,000 go blocks created and executed in around three-quarters of a second. That means that a go block compares very favorably to an Elixir process—a very impressive result given that Clojure runs on the JVM, whereas Elixir runs on the Erlang virtual machine, which was built with efficient concurrency in mind.

Now that we’ve seen both channels and go blocks in action, let’s look at how they can be combined to build more complex operations over channels.

Operations over Channels

If you’re thinking that channels have more than a little in common with sequences, you’re not wrong. Like sequences, channels represent ordered sets of values. Like sequences, we should be able to implement higher-level functions that operate over all of a channel’s contents—functions likemap, filter, and so on. And like sequences, we should be able to chain those functions to create composite operations.

Mapping over a Channel

Here’s a channel-oriented version of map:

This takes a function (f) and a source channel (from). It starts by creating a destination channel (to), which is returned at the end of the function. Before then, however, it creates a go block with go-loop, a utility function that’s equivalent to (go (loop …)). The body of the loop uses when-letto read from from and bind the value read to x. If x isn’t null, the body of the when-let is executed, (f x) is written to to, and the loop executed again. If x is null, to is closed.

Here it is in action:

As you might expect, core.async provides its own version of map-chan, called map<. There’s also a channel-oriented version of filter called filter<, mapcat called mapcat<, and so on. As you would expect, these can be combined to create chains of channels:

The preceding code uses to-chan, another core.async utility function, which creates and returns a channel containing the contents of a sequence, closing it when the sequence is exhausted.

We’re almost at the end of day 1, but before we’re done, let’s have a bit of fun.

A Concurrent Sieve of Eratosthenes

Just because we can, here’s a concurrent version of the sieve of Eratosthenes. The get-primes function returns a channel to which all the prime numbers up to limit will subsequently be written:

We’ll go through how this works in a minute (although I encourage you to work through it yourself first—you should have everything you need to do so). But first, let’s prove that it works as advertised. The following main function calls get-primes and then prints out what’s written to the channel it returns:

And here’s what we get when we run it:

Let’s see how get-primes works. It starts by creating a channel called primes, which is returned at the end of the function. It then enters a loop, with ch initially bound to numbers, a channel that will have all the numbers from 2 to limit written to it courtesy of to-chan.

The loop reads the first entry from ch, which we know is a prime number (we’ll see why this is true soon), so it’s written to primes. We then loop back around, except this time ch is bound to the result of (remove< (partial factor? prime) ch).

The remove< function is similar to filter<, except that it returns a channel to which only values for which the predicate returns false are written. In our case, it will be a channel with all values removed for which the prime we’ve just identified is a factor.

So, get-primes creates a pipeline of channels; the first contains all numbers from 2 to limit, the second has all numbers that are a multiple of 2 removed, the next has all multiples of 3 removed, and so on, as shown in the following diagram: