Channels
Deferred values provide a convenient way to transfer a single value between coroutines. Channels provide a way to transfer a stream of values.
Channel basics
A Channel is conceptually very similar to BlockingQueue
. One key difference is that instead of a blocking put
operation it has a suspending send, and instead of a blocking take
operation it has a suspending receive.
The output of this code is:
Closing and iteration over channels
Unlike a queue, a channel can be closed to indicate that no more elements are coming. On the receiver side it is convenient to use a regular for
loop to receive elements from the channel.
Conceptually, a close is like sending a special close token to the channel. The iteration stops as soon as this close token is received, so there is a guarantee that all previously sent elements before the close are received:
Building channel producers
The pattern where a coroutine is producing a sequence of elements is quite common. This is a part of producer-consumer pattern that is often found in concurrent code. You could abstract such a producer into a function that takes channel as its parameter, but this goes contrary to common sense that results must be returned from functions.
There is a convenient coroutine builder named produce that makes it easy to do it right on producer side, and an extension function consumeEach, that replaces a for
loop on the consumer side:
Pipelines
A pipeline is a pattern where one coroutine is producing, possibly infinite, stream of values:
And another coroutine or coroutines are consuming that stream, doing some processing, and producing some other results. In the example below, the numbers are just squared:
The main code starts and connects the whole pipeline:
Prime numbers with pipeline
Let's take pipelines to the extreme with an example that generates prime numbers using a pipeline of coroutines. We start with an infinite sequence of numbers.
The following pipeline stage filters an incoming stream of numbers, removing all the numbers that are divisible by the given prime number:
Now we build our pipeline by starting a stream of numbers from 2, taking a prime number from the current channel, and launching new pipeline stage for each prime number found:
The following example prints the first ten prime numbers, running the whole pipeline in the context of the main thread. Since all the coroutines are launched in the scope of the main runBlocking coroutine we don't have to keep an explicit list of all the coroutines we have started. We use cancelChildren extension function to cancel all the children coroutines after we have printed the first ten prime numbers.
The output of this code is:
Note that you can build the same pipeline using iterator
coroutine builder from the standard library. Replace produce
with iterator
, send
with yield
, receive
with next
, ReceiveChannel
with Iterator
, and get rid of the coroutine scope. You will not need runBlocking
either. However, the benefit of a pipeline that uses channels as shown above is that it can actually use multiple CPU cores if you run it in Dispatchers.Default context.
Anyway, this is an extremely impractical way to find prime numbers. In practice, pipelines do involve some other suspending invocations (like asynchronous calls to remote services) and these pipelines cannot be built using sequence
/iterator
, because they do not allow arbitrary suspension, unlike produce
, which is fully asynchronous.
Fan-out
Multiple coroutines may receive from the same channel, distributing work between themselves. Let us start with a producer coroutine that is periodically producing integers (ten numbers per second):
Then we can have several processor coroutines. In this example, they just print their id and received number:
Now let us launch five processors and let them work for almost a second. See what happens:
The output will be similar to the the following one, albeit the processor ids that receive each specific integer may be different:
Note that cancelling a producer coroutine closes its channel, thus eventually terminating iteration over the channel that processor coroutines are doing.
Also, pay attention to how we explicitly iterate over channel with for
loop to perform fan-out in launchProcessor
code. Unlike consumeEach
, this for
loop pattern is perfectly safe to use from multiple coroutines. If one of the processor coroutines fails, then others would still be processing the channel, while a processor that is written via consumeEach
always consumes (cancels) the underlying channel on its normal or abnormal completion.
Fan-in
Multiple coroutines may send to the same channel. For example, let us have a channel of strings, and a suspending function that repeatedly sends a specified string to this channel with a specified delay:
Now, let us see what happens if we launch a couple of coroutines sending strings (in this example we launch them in the context of the main thread as main coroutine's children):
The output is:
Buffered channels
The channels shown so far had no buffer. Unbuffered channels transfer elements when sender and receiver meet each other (aka rendezvous). If send is invoked first, then it is suspended until receive is invoked, if receive is invoked first, it is suspended until send is invoked.
Both Channel() factory function and produce builder take an optional capacity
parameter to specify buffer size. Buffer allows senders to send multiple elements before suspending, similar to the BlockingQueue
with a specified capacity, which blocks when buffer is full.
Take a look at the behavior of the following code:
It prints "sending" five times using a buffered channel with capacity of four:
The first four elements are added to the buffer and the sender suspends when trying to send the fifth one.
Channels are fair
Send and receive operations to channels are fair with respect to the order of their invocation from multiple coroutines. They are served in first-in first-out order, e.g. the first coroutine to invoke receive
gets the element. In the following example two coroutines "ping" and "pong" are receiving the "ball" object from the shared "table" channel.
The "ping" coroutine is started first, so it is the first one to receive the ball. Even though "ping" coroutine immediately starts receiving the ball again after sending it back to the table, the ball gets received by the "pong" coroutine, because it was already waiting for it:
Note that sometimes channels may produce executions that look unfair due to the nature of the executor that is being used. See this issue for details.
Ticker channels
Ticker channel is a special rendezvous channel that produces Unit
every time given delay passes since last consumption from this channel. Though it may seem to be useless standalone, it is a useful building block to create complex time-based produce pipelines and operators that do windowing and other time-dependent processing. Ticker channel can be used in select to perform "on tick" action.
To create such channel use a factory method ticker. To indicate that no further elements are needed use ReceiveChannel.cancel method on it.
Now let's see how it works in practice:
It prints following lines:
Note that ticker is aware of possible consumer pauses and, by default, adjusts next produced element delay if a pause occurs, trying to maintain a fixed rate of produced elements.
Optionally, a mode
parameter equal to TickerMode.FIXED_DELAY can be specified to maintain a fixed delay between elements.