Unlike languages like Haskell, Erlang, and Go, Rust does not have a
runtime system providing green threads and asynchronous I/O. However,
for many real world use cases, async I/O is strongly desired, if not a
hard requirement. The de facto standard library for handling this in
Rust is tokio.
This post is part of a series based on teaching Rust at FP
Complete. If you’re reading this post outside
of the blog, you can find links to all posts in the series at the top of the
introduction
post. You
can also subscribe to the RSS
feed.
This lesson in the crash course is going to be a bit different from
others, since:
- There’s a lot of interconnected material to cover which can’t as
easily be separated out
- It’s more important to understand the motivation behind the design
of these libraries than in many other cases
- I believe the material may be useful for people who haven’t been
following the rest of the crash course
Therefore, some different rules will apply:
- I’m defining up front the knowledge you’ll need of Rust to
understand this lesson, namely:
- All the basics of syntax
- Traits and associated types
- Iterators
- Closures
- Instead of providing the exercise solutions in a later post, I’ll be
providing them immediately, since the material is so cumulative. I
still strongly recommend spending significant time and effort
trying to solve the exercises yourself before looking at the
solutions. It’s harder and more time consuming, but ultimately
worthwhile.
Also, this lesson is much longer and more involved than previous
lessons. You should plan on it taking more time to complete than
others. I considered breaking this up into multiple lessons, but
decided to keep all of the content together. Instead, I’ll be taking a
break from weekly lessons after this one for a bit.
Consider that the intro to the intro. Now the real intro!
Why async?
I’m going to assume that readers are already familiar with async I/O
and its motivations in general. If you’re not, it’s worth reading up a
bit on the C10k problem,
where many of us started thinking hard about async I/O. You may also
be interested in reading a post I wrote about green
threads,
a language runtime-based solution to the same problem.
At the end of the day, the goals of Rust’s approach to async I/O are:
- Minimize system resources for handling a large number of concurrent
I/O tasks
- Provide a zero-cost abstraction on top of the async I/O mechanisms
provided by operating systems
- Do it at a library level, instead of introducing a runtime to Rust
A sample problem
It’s difficult to dive in to the wonderful world of tokio. You need to
learn about futures and streams, tasks and executors, async I/O system
calls and the Async type, etc. To try and decouple this learning
experience, we’re going to start with a simplified problem. This
problem mostly, but not perfectly, models async I/O in the real world,
and will demonstrate many of the design concerns. It will also let us
play just a bit more with concurrent programming before diving into
futures and tokio.
We’re going to run a separate thread. This thread will have access to
two atomic values:
- An
AtomicBool to tell us whether we want this side thread to keep
running
- An
AtomicUsize counter
We haven’t covered atomic types yet, but they’re exactly what they
sound like: variables that can be safely accessed from multiple
threads. Since learning about them isn’t the point of this lesson,
I’ll defer further questions about the usage of these types to their
API
documentation.
Our separate thread is going to run in a loop. As long as the
AtomicBool is true, it will:
- Sleep for a given number of milliseconds
- Print a message to the console
- Increment the
AtomicUsize counter
The series of examples we’ll be looking at will then try different
approaches to observing the changes in the counter.
Interval
We’re going to call this thing an Interval, since it somewhat
represents a setInterval call in Javascript. Go ahead and start a
new project:
$ cargo new interval --bin
$ cd interval
We’re going to put the code for our Interval into a separate
module. First, put the following code into src/interval.rs:
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::Arc;
use std::thread::{sleep, spawn};
use std::time::Duration;
#[derive(Clone)]
pub struct Interval {
counter: Arc<AtomicUsize>,
still_running: Arc<AtomicBool>,
}
impl Drop for Interval {
fn drop(&mut self) {
println!("Interval thread shutting down");
self.still_running.store(false, Ordering::SeqCst);
}
}
impl Interval {
pub fn from_millis(millis: u64) -> Interval {
let duration = Duration::from_millis(millis);
let counter = Arc::new(AtomicUsize::new(0));
let counter_clone = counter.clone();
let still_running = Arc::new(AtomicBool::new(true));
let still_running_clone = still_running.clone();
spawn(move || {
println!("Interval thread launched");
while still_running_clone.load(Ordering::SeqCst) {
sleep(duration);
let old = counter_clone.fetch_add(1, Ordering::SeqCst);
println!("Interval thread still alive, value was: {}", old);
}
});
Interval {
counter,
still_running,
}
}
pub fn get_counter(&self) -> usize {
self.counter.load(Ordering::SeqCst)
}
}
Next, let’s provide a minimal src/main.rs which uses this Interval
data type:
mod interval;
use self::interval::Interval;
fn main() {
let interval = Interval::from_millis(500); // half a second
let duration = std::time::Duration::from_millis(2000); // 2 seconds
for i in 1..11 {
println!("Iteration number {}, counter is {}", i, interval.get_counter());
std::thread::sleep(duration);
}
}
You should see something like the following:
Iteration number 1, counter is 0
Interval thread launched
Interval thread still alive, value was: 0
Interval thread still alive, value was: 1
Interval thread still alive, value was: 2
Iteration number 2, counter is 3
Interval thread still alive, value was: 3
Interval thread still alive, value was: 4
...
Interval thread still alive, value was: 33
Interval thread still alive, value was: 34
Iteration number 10, counter is 35
Interval thread still alive, value was: 35
Interval thread still alive, value was: 36
Interval thread still alive, value was: 37
Interval thread still alive, value was: 38
Interval thread shutting down
Hurrah, we have some concurrent communication.
Problems with this approach
The first thing that jumps out as a problem is that we’re missing some updates in the main thread. Notice how the counter jumps from 0 to 3. This is obviously a problem with the interval set in the main thread: we’re delaying for 2 seconds instead of half a second. Let’s instead delay for a tenth of a second (100ms) in the main thread, and check if the value has changed since last time.
NOTE It’s still possible we’ll miss some updates this way, since
sleep guarantees a thread will sleep at least a given amount of
time, but may sleep longer. However, by having such a large
difference, we’re fairly certain we’ll catch all of the updates.
fn main() {
let interval = Interval::from_millis(500); // half a second
let duration = std::time::Duration::from_millis(100); // 0.1 seconds
let mut last = interval.get_counter();
for i in 1..51 {
let curr = interval.get_counter();
if curr != last {
last = curr;
println!("Iteration number {}, counter is {}", i, curr);
}
std::thread::sleep(duration);
}
}
I had to increase the number of iterations to 50, because so many of
our main thread iterations end up showing no change in the
counter. Here’s an example of running this on my machine:
Interval thread launched
Interval thread still alive, value was: 0
Iteration number 6, counter is 1
Interval thread still alive, value was: 1
Iteration number 11, counter is 2
Interval thread still alive, value was: 2
Iteration number 16, counter is 3
Interval thread still alive, value was: 3
Iteration number 21, counter is 4
Interval thread still alive, value was: 4
Iteration number 26, counter is 5
Interval thread still alive, value was: 5
Iteration number 31, counter is 6
Interval thread still alive, value was: 6
Iteration number 36, counter is 7
Interval thread still alive, value was: 7
Iteration number 41, counter is 8
Interval thread still alive, value was: 8
Iteration number 46, counter is 9
Interval thread still alive, value was: 9
Interval thread shutting down
We didn’t lose any counter updates here, but from the bumps in
interval thread numbers, we can see that we’re wasting a lot of time
in the main thread checking numbers that aren’t changing.
Another problem that’s less obvious is that we’re dedicating an entire
OS thread to this sleep-and-check iteration. In our simple program,
that’s not a big deal. But imagine we decided we wanted to have 50
different similar tasks going on. It would require 49 extra threads,
most of which would sit around sleeping the majority of the
time. That’s highly wasteful. We should be able to do better.
Finally, and perhaps least important for the moment, this is all
rather ad hoc. It seems like a common need to be able to abstract over
“this thing will produce a value in the future.” Even though this
seems like the least important problem, we’ll start by solving it
first.
The Future trait
Who’d have thought that our meandering would naturally lead to one of
the topics mentioned in this lesson’s title! You may have noticed a
pattern that developed in the main thread’s loop:
- Check if a new value is available
- Use it if it is available
- Skip if it isn’t available
That’s exactly what the Future trait allows us to do, with one
addition: it also allows for error handling. We’re not going to worry
about that for now, since my code doesn’t have any errors :).
We’ll start by adding the futures crate as a dependency. In
Cargo.toml:
[dependencies]
futures = "0.1"
Next, let’s add a new module to provide a struct that will provide a
Future implementation. Behold, src/future.rs:
extern crate futures;
use super::interval::Interval;
use futures::prelude::*;
pub struct IntervalFuture {
interval: Interval,
last: usize,
}
impl IntervalFuture {
pub fn new(interval: Interval) -> IntervalFuture {
let last = interval.get_counter();
IntervalFuture { interval, last }
}
}
impl Future for IntervalFuture {
type Item = usize;
type Error = ();
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
let curr = self.interval.get_counter();
if curr == self.last {
Ok(Async::NotReady)
} else {
self.last = curr;
Ok(Async::Ready(curr))
}
}
}
We’re going to own an Interval and the last value we provided, just
like our main loop used to. The new method is fairly
straightforward. For the impl Future, we need to define three
things:
- The thing which will be returned by this type when ready. In our
case, it’s the counter value, which is a
usize.
- The type of errors that can occur. We don’t have any errors, so we
use
(). (The Haskeller in me screams that we should use
Void. Soon enough, we’ll be able to use
never.)
- A function
poll, which returns a Result. In the error case, this
will be our Self::Error. In the success case, this is an Async
enum type. As we can see in our method body, this is either a
Ready variant with the value, or NotReady.
The logic in our function is the same as before, so I won’t comment on
implementation. Back in our src/main.rs, instead of playing around
with the curr/last logic, we can just pattern match on the result
of poll()ing:
extern crate futures;
mod future;
mod interval;
use self::interval::Interval;
use self::future::IntervalFuture;
use futures::prelude::*;
fn main() {
let interval = Interval::from_millis(500); // half a second
let mut interval_future = IntervalFuture::new(interval);
let duration = std::time::Duration::from_millis(100); // 0.1 seconds
for i in 1..51 {
match interval_future.poll() {
Ok(Async::Ready(curr)) => {
println!("Iteration number {}, counter is {}", i, curr);
}
Ok(Async::NotReady) => (),
Err(()) => unreachable!(),
}
std::thread::sleep(duration);
}
}
Arguably a minor improvement on the previous code, though nothing
major. But congratulations, you’re now officially using the futures
crate!
The Poll type definition
Just a minor helper to mention. The type Result<Async<Self::Item>, Self::Error> may look a bit unwieldy to you. If so, you’ll be happy
to learn about the Poll type definition, which lets you replace the
above with Poll<Self::Item, Self::Error>. Not a big deal, but
important to recognize as you’re reading other code.
The tokio executor
Right now, we’re running our own executor in our main function:
we’re manually looping, delaying, etc. Besides tedium, we’ve already
mentioned some downsides to this above:
- We need a single thread per task we wish to perform
- We need to implement some kind of guess-and-check thread sleeping
It’s time to pull out the big guns, and tokio this thing. We’ll end up
losing some functionality first, and then we’ll build it back.
First, add tokio = "0.1" to your Cargo.toml. Now, let’s try using
the tokio executor by calling tokio::run on a Future:
extern crate futures;
extern crate tokio;
mod future;
mod interval;
use self::interval::Interval;
use self::future::IntervalFuture;
use tokio::prelude::*;
fn main() {
let interval = Interval::from_millis(500); // half a second
let mut interval_future = IntervalFuture::new(interval);
tokio::run(interval_future)
}
This fails with a compilation error:
error[E0271]: type mismatch resolving `<future::IntervalFuture as futures::Future>::Item == ()`
--> src/main.rs:15:5
|
15 | tokio::run(interval_future)
| ^^^^^^^^^^ expected usize, found ()
|
= note: expected type `usize`
found type `()`
= note: required by `tokio::run`
The tokio::run function expects a Future where the Item is (),
but ours is usize. This kind of makes sense anyway: don’t we want to
write some code to actually do something when we get a value?
We’re going to fix this, first the overly painful way, and then the
pleasant way. That will also help you appreciate why lesson 5 spent so
much time on closures.
Define an adapter Future
Remember how you can define Iterators that consume other Iterators
and compose more powerful streams? Well, you can do the same thing
with Futures. Let’s define a new type that will:
- Wrap around an
IntervalFuture
- Print the new value whenever it’s ready
We’ll put this in src/main.rs for now, it won’t last long anyway.
extern crate futures;
extern crate tokio;
mod future;
mod interval;
use self::interval::Interval;
use self::future::IntervalFuture;
use tokio::prelude::*;
struct IntervalPrinter(IntervalFuture);
impl Future for IntervalPrinter {
type Item = ();
type Error = ();
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
match self.0.poll() {
Ok(Async::Ready(curr)) => {
println!("Counter is: {}", curr);
Ok(Async::Ready(()))
}
Ok(Async::NotReady) => Ok(Async::NotReady),
Err(e) => Err(e),
}
}
}
fn main() {
let interval = Interval::from_millis(500); // half a second
let interval_future = IntervalFuture::new(interval);
let interval_printer = IntervalPrinter(interval_future);
tokio::run(interval_printer)
}
Compile the code, but don’t run it yet. This is relatively
straight-forward given all of the types and traits we’ve seen, but
it’s obviously tedious. Let’s start off with a minor simplification.
The try_ready macro
The body of poll spends a lot of lines of code to:
- Pattern match
- If it’s
NotReady, return NotReady
- If it’s an
Err, return the Err
This is a repetitive pattern, and is pretty similar to the error
handling we saw previously in lesson 3. The futures crate provides a
macro, try_ready!, to deal with this annoyance. Add the following
above the extern crate futures; in src/main.rs:
#[macro_use]
And then your implementation of poll can be simplified to:
let curr = try_ready!(self.0.poll());
println!("Counter is: {}", curr);
Ok(Async::Ready(()))
Nice! Compile, and again, don’t run. (Getting curious yet why I keep
saying that? We’ll find out soon, just one more pitstop first.)
I need some closure
It’s amazing I’ve made it to lesson 7 in this crash course without
making that pun. Obviously, defining an entire struct and Future
implementation is a bit overkill to just print a line. Fortunately,
the authors of the futures crate noticed this too. There are a
number of combinators built into the Future trait that make it easy
to chain things together. We’re going to use the and_then method:
extern crate futures;
extern crate tokio;
mod future;
mod interval;
use self::interval::Interval;
use self::future::IntervalFuture;
use tokio::prelude::*;
fn main() {
let interval = Interval::from_millis(500); // half a second
let interval_future = IntervalFuture::new(interval);
let interval_printer = interval_future.and_then(|curr| {
println!("Counter is: {}", curr);
futures::future::ok(())
});
tokio::run(interval_printer)
}
That’s much nicer! If you’re anything like me though, the
futures::future::ok(()) is bothering you. What purpose does it serve
there? This is a vital part of the design of futures, which we’ll be
taking advantage of quite a bit going forward. It allows us to create
chains of actions to run as each bit of async I/O completes. For now,
we don’t want to do anything else after we print the first value from
the counter, so we just return futures::future::ok(()), which means
“don’t do anything, and return the item ()”.
Exercise 1 There’s another method, .map, which is actually a
better choice for us here than .and_then. Try rewriting the code
above to use .map. Note: no solution provided to this one.
Out of curiosity, what’s the type of interval_printer? Let’s use the
dumb trick from before of giving it the wrong type. Insert something
silly like let interval_printer: bool = ... and try compiling,
you’ll get some type like:
futures::AndThen<
future::IntervalFuture,
futures::FutureResult<(), ()>,
[closure@src/main.rs:14:59: 17:6]
>
If this is starting to look a bit like Iterator types, that’s by
design. Just like Iterators, Futures capture a large amount of
information in the types themselves about what they’ll do. This allows
Futures to compile down to highly efficient code, living up to the
Rust mantra of zero-cost abstractions.
Finally, run it!
Is the suspense killing you? Alright, your moment has arrived. What
happens when you run cargo run?
$ cargo run
Interval thread launched
Interval thread shutting down
Interval thread still alive, value was: 0
...
And that’s it. It hangs. It doesn’t print out the message we
painstakingly added with and_then. I’m a complete failure, my life
is in ruins.
After an hour to myself to contemplate my life and a few shots of
whiskey, things started to get clear. In our original implementation,
we had a really wasteful loop in the main function. It kept checking
if the counter had changed. We did that with our Future
implementation too at first, sleeping and then checking again for a
NotReady. But tokio probably isn’t doing that, right? (The answer
is yes, right.)
Instead of doing the silly wasteful thing, the futures crate is far
smarter. It has a mechanism to:
- determine which task is trying to get access to the data provided by
this future, and then
- notify that task that new data is available
The
futures::task::current()
function gives us the current task being run, as a
Task
struct. That struct has a method,
notify,
to let the task know that more data is available.
In our program, we have the logic split between Interval and
IntervalFuture. IntervalFuture will need to be responsible for
calling the current() function (give some thought as to why that’s
the case). The changes needed to make this work are:
- Add a new field to
Interval to hold an Arc<Mutex<Option<Task>>>
(yes, that’s a mouthful), and initialize correctly.
- Each time we call
fetch_add and update the counter, also call
notify() on that task, if it’s there.
- Provide a method
set_task to set the Task on an Interval
- When returning
NotReady in IntervalFuture, call set_task
Exercise 2 Take a stab at implementing these four changes before
looking at the solution below. A hint on some features of Rust we
haven’t covered yet: you’ll end up wanting to pattern match on the
Option held inside the Mutex. You’ll want to pattern match by
reference, which will require some code that looks like Some(ref task) =>. And the final output should look like:
Interval thread launched
Interval thread still alive, value was: 0
Counter is: 1
Interval thread shutting down
If you want to be sure, you can see the initial version of the code on Github.
Solution 2
You can check out the
diff
and full
solution
on Github. Here’s the diff included inline:
diff --git a/src/future.rs b/src/future.rs
index 9aaee3c..e231e7b 100644
--- a/src/future.rs
+++ b/src/future.rs
@@ -22,6 +22,8 @@ impl Future for IntervalFuture {
fn poll(&mut self) -> Result<Async<Self::Item>, Self::Error> {
let curr = self.interval.get_counter();
if curr == self.last {
+ let task = futures::task::current();
+ self.interval.set_task(task);
Ok(Async::NotReady)
} else {
self.last = curr;
diff --git a/src/interval.rs b/src/interval.rs
index 044e2ca..8013ac6 100644
--- a/src/interval.rs
+++ b/src/interval.rs
@@ -1,12 +1,14 @@
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
use std::thread::{sleep, spawn};
use std::time::Duration;
+use futures::task::Task;
#[derive(Clone)]
pub struct Interval {
counter: Arc<AtomicUsize>,
still_running: Arc<AtomicBool>,
+ task: Arc<Mutex<Option<Task>>>,
}
impl Drop for Interval {
@@ -26,22 +28,37 @@ impl Interval {
let still_running = Arc::new(AtomicBool::new(true));
let still_running_clone = still_running.clone();
+ let task: Arc<Mutex<Option<Task>>> = Arc::new(Mutex::new(None));
+ let task_clone = task.clone();
+
spawn(move || {
println!("Interval thread launched");
while still_running_clone.load(Ordering::SeqCst) {
sleep(duration);
let old = counter_clone.fetch_add(1, Ordering::SeqCst);
println!("Interval thread still alive, value was: {}", old);
+
+ let task = task_clone.lock().unwrap();
+ match *task {
+ None => (),
+ Some(ref task) => task.notify(),
+ };
}
});
Interval {
counter,
still_running,
+ task,
}
}
pub fn get_counter(&self) -> usize {
self.counter.load(Ordering::SeqCst)
}
+
+ pub fn set_task(&mut self, task: Task) {
+ let mut guard = self.task.lock().unwrap();
+ *guard = Some(task);
+ }
}
Hurrah, we finally have a working tokio program!
In case you’re worried about how complex that was, don’t
be. Notification is a vital aspect of how tokio works
internally. However, in most cases, you won’t be creating your own
primitive Futures, but instead dealing with existing ones provided
by tokio or other libraries. Those existing Futures will provide the
necessary notification logic. You’ll simply need to obey this one
rule:
Only return a NotReady from a poll function if you received a
NotReady from an underlying Future.
Just one value?
It’s a bit disappointing that our wonderful long running counter only
ends up printing a single value. Can we create some kind of a loop? A
simple approach like the following doesn’t work:
let interval_printer = interval_future.and_then(|curr| {
println!("Counter is: {}", curr);
interval_printer
});
This isn’t Haskell, we can’t recursively refer to interval_printer
we’re in the middle of defining. Go ahead and take a few other stabs
at doing something like that, and you’ll eventually get frustrated and
go back to the whiskey. Digging through the futures docs, a helper
function like
loop_fn
looks promising, but I didn’t see a simple way to make it work in this
case. (Please let me know if I missed something!) I ended up with
something wonky like this before stopping:
fn main() {
let interval = Interval::from_millis(500); // half a second
let interval_future = Arc::new(Mutex::new(IntervalFuture::new(interval)));
let interval_printer = loop_fn(interval_future, |interval_future| {
let interval_future_clone = interval_future.clone();
interval_future.lock().unwrap().and_then(|curr| {
println!("Counter: {}", curr);
futures::future::ok(Continue(interval_future_clone))
})
});
tokio::run(interval_printer)
}
Another struct!
Like before, we’re going to define another helper type to implement
this concept of looping. Then we’ll see that this problem has already
been solved better in the futures crate itself, but we’ll get there
soon.
We want to define a new struct, KeepPrinting, which is a newtype
around an IntervalFuture. It’s going to:
- Have a
Future implementation
- Have
Item = ()
- Use a
loop in its implementation
- Use the
try_ready! macro
Exercise 3 Try implementing KeepPrinting and using it in the
main function. Solution follows immediately, but try not to cheat!
#[macro_use]
extern crate futures;
extern crate tokio;
mod future;
mod interval;
use self::future::IntervalFuture;
use self::interval::Interval;
use tokio::prelude::*;
struct KeepPrinting(IntervalFuture);
impl Future for KeepPrinting {
type Item = ();
type Error = ();
fn poll(&mut self) -> Poll<(), ()> {
loop {
let curr = try_ready!(self.0.poll());
println!("Counter: {}", curr);
}
}
}
fn main() {
let interval = Interval::from_millis(500); // half a second
let interval_future = IntervalFuture::new(interval);
let keep_printing = KeepPrinting(interval_future);
tokio::run(keep_printing)
}
And, just like that, we get an infinitely looping program. This almost
looks like something that we could have done with Iterators. Which
makes me wonder… is there something like Iterators in futures?
Streams
A Future is an action with a delayed single result. A Stream is a
stream of results, like an Iterator, with a delay between each
value.
In src/main.rs, add mod stream; and then edit src/stream.rs. The
file will end up looking remarkably similar to src/future.rs,
except:
- Call the struct
IntervalStream instead of IntervalFuture
- Provide an
impl Stream for IntervalStream instead of impl Future
- Follow the compiler errors to fix it
Within the main function, instead of using KeepPrinting or
anything else, we’ll want to create an IntervalStream
value. However, tokio::run needs a Future, not a Stream, to
run. Fortunately, there’s a helper function, for_each, that runs a
given closure on each value in the stream.
Exercise 4 Try to implement src/stream.rs and
src/main.rs. Solution to follow.
The trickiest bit for me when first learning for_each was to realize
that, like and_then, it needs to end with a Future. I don’t know
if that was just my own shortcoming, or a common issue. In any event,
if you struggled to realize you needed something like future::ok(())
at the end of your closure, you’re in good company.
In addition, the poll function for a Stream is slightly different,
in returning an Option<Item>. This is similar to how Iterator
works. In our case, we have an infinite stream, so we never provide
the None case.
Anyway, here’s src/stream.rs:
extern crate futures;
use super::interval::Interval;
use futures::prelude::*;
pub struct IntervalStream {
interval: Interval,
last: usize,
}
impl IntervalStream {
pub fn new(interval: Interval) -> IntervalStream {
let last = interval.get_counter();
IntervalStream { interval, last }
}
}
impl Stream for IntervalStream {
type Item = usize;
type Error = ();
fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
let curr = self.interval.get_counter();
if curr == self.last {
let task = futures::task::current();
self.interval.set_task(task);
Ok(Async::NotReady)
} else {
self.last = curr;
Ok(Async::Ready(Some(curr)))
}
}
}
And src/main.rs:
extern crate futures;
extern crate tokio;
mod interval;
mod stream;
use self::interval::Interval;
use self::stream::IntervalStream;
use tokio::prelude::*;
fn main() {
let interval = Interval::from_millis(500); // half a second
let interval_stream = IntervalStream::new(interval);
let future = interval_stream.for_each(|curr| {
println!("Counter: {}", curr);
futures::future::ok(())
});
tokio::run(future)
}
Exercise 5 Like Iterators, Streams have helper methods that
you can use to build up more complex things. For example, try throwing
in map and take to print only the first 10 counter values, but
double them before printing. (No solution provided.)
This is all beginning to fit together nicely! While there are still
details to learn in the futures crate, you’ve got most of the big
ideas down. The next bit is to get familiar with the API in tokio, but
relatively speaking this is less mind-bending. To hammer home what
we’ve done so far, we’ll hit a few exercises, and then continue with
tokio.
Exercise 6
Define a new struct MyOk such that this main function works:
fn main() {
let name = String::from("Alice");
let future = MyOk::new(name).and_then(|name| {
println!("Name: {}", name);
MyOk::new(())
});
tokio::run(future)
}
Hint: before cheating and looking at the solution, here’s one piece of
help: you’ll want an Option inside the MyOk newtype, and it’s
invalid to call poll on it twice.
Solution 6
struct MyOk<T>(Option<T>);
impl<T> MyOk<T> {
fn new(t: T) -> MyOk<T> {
MyOk(Some(t))
}
}
impl<T> Future for MyOk<T> {
type Item = T;
type Error = ();
fn poll(&mut self) -> Poll<T, ()> {
Ok(Async::Ready(self.0.take().unwrap()))
}
}
Exercise 7
Use iter_ok to convert the range 1..11 to a Stream, and then
collect it as a Vec and print it.
Solution 7
fn main() {
tokio::run(stream::iter_ok(1..11).collect().and_then(|x| {
println!("{:?}", x);
future::ok(())
}))
}
Async I/O
We’ve played around with the futures crate by creating a fake async
I/O source of data (the Interval). We’ve built up Futures and
Streams in that world. And we’ve used tokio’s executor to run these
things. It’s now time to take it to use some real async I/O.
Most async I/O we care about will end up being network
traffic. Filesystem operations don’t always play nicely with async I/O
at an operating system level. That said, to get our feet wet, let’s
play with a filesystem based example.
You’ll want to look at the docs quite a bit, you can find them on
docs.rs.
List files in a directory
If you look through the docs above, you may find the function
read_dir. It
takes a path, and returns a ReadDirFuture. This is a standard
approach in tokio, like we had with Iterators: simple wrapper
functions providing access to the structs that do the heavy
lifting. One thing to get used to in tokio is how to read these docs.
Click through on the ReadDirFuture struct. It has a Future
implementation, where Item is ReadDir, and Error is
std::io::Error. Before we deal with that ReadDir, let’s just get
something that compiles. Since this is still a crash course, we’ll
bash our heads against a brick wall each step of the way.
First, to call read_dir, we need a directory. Let’s use "." (the
current directory). We’ll use command line arguments later. Here’s a
naive implementation:
extern crate tokio;
use tokio::prelude::*;
use tokio::fs;
fn main() {
let future = fs::read_dir(".");
tokio::run(future)
}
This gives us a somewhat intimidating error message:
error[E0271]: type mismatch resolving `<tokio_fs::read_dir::ReadDirFuture<&str> as tokio::prelude::Future>::Item == ()`
--> src/main.rs:8:5
|
8 | tokio::run(future)
| ^^^^^^^^^^ expected struct `tokio_fs::read_dir::ReadDir`, found ()
|
= note: expected type `tokio_fs::read_dir::ReadDir`
found type `()`
= note: required by `tokio::run`
error[E0271]: type mismatch resolving `<tokio_fs::read_dir::ReadDirFuture<&str> as tokio::prelude::Future>::Error == ()`
--> src/main.rs:8:5
|
8 | tokio::run(future)
| ^^^^^^^^^^ expected struct `std::io::Error`, found ()
|
= note: expected type `std::io::Error`
found type `()`
= note: required by `tokio::run`
However, let me narrow that down for you:
expected type `tokio_fs::read_dir::ReadDir`, found type `()`
expected type `std::io::Error`, found type `()`
Oh, right! tokio::run requires that we have Item and Error as
(). We can modify the Error with map_err. Let’s just print out
the error if one occurs:
let future = fs::read_dir(".")
.map_err(|e| eprintln!("Error reading directory: {}", e))
;
That knocked out the first compilation error. Let’s also throw in a
.and_then:
.and_then(|readdir| {
println!("FIXME: use this: {:?}", readdir);
})
Uh oh, we got this compilation error. Can you figure out how to solve it?
error[E0277]: the trait bound `(): tokio::prelude::Future` is not satisfied
In my experience, when you see that, it almost always means: “I forgot
to add future::ok(()). Remember, and_then needs to end with the
next Future to run. Add that line, and your code should
compile. Running produces the output:
FIXME: use this: ReadDir(ReadDir("."))
Cool! Now it’s time to look at the docs for
ReadDir. Instead
of a Future implementation, this has a Stream. Let’s shove a
for_each in there and see what happens.
Challenge try to guess the errors in the code below before you compile it.
.and_then(|readdir| {
readdir
.for_each(|entry| {
println!("{:?}", entry.path());
})
})
There are two problems with this code:
- It leaves an error type of
std::io::Error
- It doesn’t include
future::ok(()) at the end of the closure
provided to for_each
Go ahead and fix those problems. To be sure we’re on the same page,
here’s my solution which compiles and runs successfully:
extern crate tokio;
use tokio::prelude::*;
use tokio::fs;
fn main() {
let future = fs::read_dir(".")
.map_err(|e| eprintln!("Error reading directory: {}", e))
.and_then(|readdir| {
readdir
.map_err(|e| eprintln!("Error reading directory: {}", e))
.for_each(|entry| {
println!("{:?}", entry.path());
future::ok(())
})
})
;
tokio::run(future)
}
Duplicated error handling
It’s a bit irritating that we have two identical map_err calls. We
have two different sources of errors: the initial read_dir Future,
and then streaming the individual DirEntrys from it. However, the
type of the errors in both cases is the same:
std::io::Error. Therefore, we can move our error handling to the
end, and just do it once:
fn main() {
let future = fs::read_dir(".")
.and_then(|readdir| {
readdir
.for_each(|entry| {
println!("{:?}", entry.path());
future::ok(())
})
})
.map_err(|e| eprintln!("Error reading directory: {}", e))
;
tokio::run(future)
}
Flattening
It turns out that it’s common enough to have a Future that generates
another Future, and then we want to run that second Future, that
there’s a helper method for it flatten(). There’s also a
flatten_stream()
that does the same thing when a Future gives us a Stream.
Exercise 8 Rewrite the code above to use flatten_stream. You
should end up with no calls to and_then. Solution follows
immediately:
extern crate tokio;
use tokio::prelude::*;
use tokio::fs;
fn main() {
let future = fs::read_dir(".")
.flatten_stream()
.for_each(|entry| {
println!("{:?}", entry.path());
future::ok(())
})
.map_err(|e| eprintln!("Error reading directory: {}", e))
;
tokio::run(future)
}
Command line arguments
It’s somewhat boring to always print out what’s in the current
directory. Instead, let’s take all of the command line arguments
(skipping the first, which is the executable name), and list the
directory contents. We’ll use stream::iter_ok to convert the Args
Iterator into a Stream:
extern crate tokio;
use tokio::prelude::*;
use tokio::fs;
use std::env::args;
fn main() {
let future = stream::iter_ok(args())
.skip(1)
.for_each(|dir| {
fs::read_dir(dir)
.flatten_stream()
.for_each(|entry| {
println!("{:?}", entry.path());
future::ok(())
})
})
.map_err(|e| eprintln!("Error reading directory: {}", e))
;
tokio::run(future)
}
Unfortunately, this doesn’t compile. The full error message is large
(I encourage you to check it out yourself), but the first few lines
are sufficient to find the problem:
error[E0277]: `std::env::Args` cannot be sent between threads safely
--> src/main.rs:20:5
|
20 | tokio::run(future)
| ^^^^^^^^^^ `std::env::Args` cannot be sent between threads safely
Oh, darn. Args isn’t thread safe, and so cannot be converted into a
Stream. Fair enough: vectors to the rescue!
Exercise 9 Create a vector of arguments before the definition of
future and use that.
Solution:
extern crate tokio;
use tokio::prelude::*;
use tokio::fs;
use std::env::args;
fn main() {
let args: Vec<String> = args().skip(1).collect();
let future = stream::iter_ok(args)
.for_each(|dir| {
fs::read_dir(dir)
.flatten_stream()
.for_each(|entry| {
println!("{:?}", entry.path());
future::ok(())
})
})
.map_err(|e| eprintln!("Error reading directory: {}", e))
;
tokio::run(future)
}
Where’s the concurrency?
If you provide this program two different directories with a large
number of files, you may notice that it processes these directories
sequentially: it will print all of the files in the first directory,
and then all of the files in the second directory. Given that async
I/O and concurrency usually go hand-in-hand, that may be a bit
surprising.
So far, we’ve only ever had a single task at a time. Our program
streams out the value of args, and for each one provides a
Future. That Future is run to completion, and then the next value
from args is processed.
What if we want to process each directory concurrently? To do that, we
need to spawn another task, the same way we would spawn a new
thread. Like tokio::run, tokio::spawn takes a Future where both
Item and Error are (). Here’s a more concurrent version of our
program:
let future = stream::iter_ok(args)
.for_each(|dir| {
let future = fs::read_dir(dir)
.flatten_stream()
.map_err(|e| eprintln!("Error reading directory: {}", e))
.for_each(|entry| {
println!("{:?}", entry.path());
future::ok(())
})
;
tokio::spawn(future);
future::ok(())
})
;
Notice how I’ve put future::ok(()) after the
tokio::spawn(future); call. It turns out that’s not needed: spawn
returns a Spawn value, which behaves like future::ok(()) (via its
IntoFuture implementation). So just remove future::ok and the
semicolon after spawn, and your code will still work.
NOTE You may not notice the concurrency unless you have a large
number of files in each directory.
Skipping the vector
One final thing that annoyed me above is that Vec. It really seems
like we should be able to get away without it. We can’t convert Args
into a Stream, because that would require sending the value between
threads. But now we’ve got a new trick up our sleeves: spawning. What
if we never send the Args anywhere, but just spawn a bunch of tasks.
When I was first learning tokio, I’ll admit that I spent way more time
trying to figure out the trick I’m about to show you than I’m proud
of. We need to create a Future, then run a for loop inside of
it. How do we create a Future that lets us run some code without
waiting for anything else? We can use future::ok(()) to create a
dummy Future, and then chain the next action together with
and_then.
let future = future::ok(()).and_then(|()| {
for dir in args().skip(1) {
let future = fs::read_dir(dir)
.flatten_stream()
.map_err(|e| eprintln!("Error reading directory: {}", e))
.for_each(|entry| {
println!("{:?}", entry.path());
future::ok(())
})
;
tokio::spawn(future);
}
future::ok(())
});
Another approach, if you’re so inclined, is to use the future::poll_fn helper function. This takes a 0 argument function which returns a Result<Async<Item>, Error>, just like the poll method of Future does.
Exercise 10
Rewrite our program above to use future::poll_fn. Your program
should not use and_then at all.
Solution:
extern crate tokio;
use tokio::prelude::*;
use tokio::fs;
use std::env::args;
fn main() {
let future = future::poll_fn(|| {
for dir in args().skip(1) {
let future = fs::read_dir(dir)
.flatten_stream()
.map_err(|e| eprintln!("Error reading directory: {}", e))
.for_each(|entry| {
println!("{:?}", entry.path());
future::ok(())
})
;
tokio::spawn(future);
}
Ok(Async::Ready(()))
});
tokio::run(future)
}
Is all of this ceremony worth it for file system operations? Probably
not. Let’s get into something more interesting: network communications!
TCP client
Rust’s standard library already provides really nice support for TCP
communications out of the box. For example, the following code will
print the full response headers and body from an HTTP request to
http://httpbin.org/json:
use std::io::{Read, Write};
use std::net::TcpStream;
fn main() -> Result<(), Box<std::error::Error>> {
let mut stream = TcpStream::connect("httpbin.org:80")?;
stream.write_all(b"GET /json HTTP/1.1\r\nHost: httpbin.org\r\nConnection: close\r\n\r\n")?;
let mut buffer = vec![];
stream.read_to_end(&mut buffer)?;
println!("{}", std::str::from_utf8(&buffer)?);
Ok(())
}
There are some simplifying assumptions here, like using connection: close so that we can use read_to_end, and assuming the response
body is properly UTF-8 encoded. But that’s not really an indictment of
the TCP support in the standard library.
The real problem is the same one we’ve been talking about throughout
this lesson: it hogs an entire OS thread blocking on a successful
write to and subsequent read from the network. Let’s look at how tokio
can help.
It looks like there’s a TcpStream type in
tokio::net::TcpStream. That
looks like a good place to start. It takes a
SocketAddr,
which we can probably make easily enough. And it returns a
ConnectFuture. Let’s
start with some code that simple establishes a connection:
extern crate tokio;
use tokio::net::TcpStream;
use tokio::prelude::*;
use std::net::ToSocketAddrs;
fn main() {
let mut addr_iter = "httpbin.org:80".to_socket_addrs().unwrap();
let addr = match addr_iter.next() {
None => panic!("DNS resolution failed"),
Some(addr) => addr,
};
let future = TcpStream::connect(&addr)
.map_err(|e| eprintln!("Error connecting: {:?}", e))
.map(|stream| {
println!("Got a stream: {:?}", stream);
});
tokio::run(future)
}
The to_socket_addrs business isn’t our focus right now, so I’m going
to ignore it. Feel free as an exercise to improve the error handling
of that bit of code.
We’ve got all of the familiar pieces here: define a Future, handle
errors, and use map to chain together the action to take with the
open connection.
Let’s look a bit more closely at that stream value passed to the
closure. It comes from a ConnectFuture, so we need to look at the
Item associated type there. And sure enough, if you check the docs,
you’ll see that it’s TcpStream. Great.
We used write_all in our original, non-async, blocking code. If I
search for write_all in
tokio,
I find that there’s such a helper
function
which returns a
WriteAll
Future. Something interesting:
write_all takes two parameters: an AsyncWrite and an
AsRef<[u8]>. This will work out to be our TcpStream and the data
to send.- The
Item for AsyncWrite is a pair of the variables A and T,
which turns out to be exactly the same as the parameters we passed
in.
Giving us back the stream we originally provided is vital to
continuing the connection. I didn’t see any documentation clarifying
the point of returning the byte buffer, but I believe it’s returned so
that, if desired, you can reuse a mutable buffer.
Anyway, let’s put this together and make our request:
extern crate tokio;
use tokio::net::TcpStream;
use tokio::io::write_all;
use tokio::prelude::*;
use std::net::ToSocketAddrs;
const REQ_BODY: &[u8] = b"GET /json HTTP/1.1\r\nHost: httpbin.org\r\nConnection: close\r\n\r\n";
fn main() {
let mut addr_iter = "httpbin.org:80".to_socket_addrs().unwrap();
let addr = match addr_iter.next() {
None => panic!("DNS resolution failed"),
Some(addr) => addr,
};
let future = TcpStream::connect(&addr)
.and_then(|stream| {
write_all(stream, REQ_BODY)
.map(|(stream, _body)| println!("Write succeeded: {:?}", stream))
})
.map_err(|e| eprintln!("Error occured: {:?}", e))
;
tokio::run(future)
}
Notice how I replaced the previous map with an and_then call, so
that I could provide another Future to be performed after the
connection was established.
Would it be too much to ask to also get a read_to_end function in
tokio? Nope, not at
all.
Exercise 11 Use read_to_end to consume the entire response into
a Vec<u8>, and then print that out, using std::str::from_utf8 and
being as careless with error handling as you like.
Alright, solution:
extern crate tokio;
use std::net::ToSocketAddrs;
use tokio::io::{read_to_end, write_all};
use tokio::net::TcpStream;
use tokio::prelude::*;
const REQ_BODY: &[u8] = b"GET /json HTTP/1.1\r\nHost: httpbin.org\r\nConnection: close\r\n\r\n";
fn main() {
let mut addr_iter = "httpbin.org:80".to_socket_addrs().unwrap();
let addr = match addr_iter.next() {
None => panic!("DNS resolution failed"),
Some(addr) => addr,
};
let future = TcpStream::connect(&addr)
.and_then(|stream| {
write_all(stream, REQ_BODY)
}).and_then(|(stream, _body)| {
let buffer = vec![];
read_to_end(stream, buffer)
}).map(|(_stream, buffer)| {
let s = std::str::from_utf8(&buffer).unwrap();
println!("{}", s);
}).map_err(|e| eprintln!("Error occured: {:?}", e));
tokio::run(future)
}
Streaming to a file
I’ll repeat for maximum annoyance: tokio is not intended for
asynchronous file operations. That said, there is a
tokio::fs::File struct which we can use. Let’s try to write the
response contents to httpbin.json instead:
let future = TcpStream::connect(&addr)
.and_then(|stream| {
write_all(stream, REQ_BODY)
}).and_then(|(stream, _body)| {
let buffer = vec![];
read_to_end(stream, buffer)
}).and_then(|(_stream, buffer)| {
File::create("httpbin.json").and_then(|file| {
write_all(file, &buffer).map(|_| ())
})
}).map_err(|e| eprintln!("Error occured: {:?}", e));
Unfortunately, the compiler doesn’t like this too much:
error[E0277]: the trait bound `std::fs::File: tokio::io::AsyncWrite` is not satisfied
--> src/main.rs:25:17
|
25 | write_all(file, &buffer).map(|_| ())
| ^^^^^^^^^ the trait `tokio::io::AsyncWrite` is not implemented for `std::fs::File`
|
= note: required by `tokio::io::write_all
Well, I guess that makes sense: you can’t asynchronously write to a
File, so tokio::io::write_all isn’t going to work. Fortunately,
File does implement the Write trait, which provides a blocking
write_all, which is sufficient for our purposes.
Exercise 12 Rewrite the code above to successfully write
httpbin.json.
First solution, ignoring anything close to proper error handling:
let future = TcpStream::connect(&addr)
.and_then(|stream| {
write_all(stream, REQ_BODY)
}).and_then(|(stream, _body)| {
let buffer = vec![];
read_to_end(stream, buffer)
}).and_then(|(_stream, buffer)| {
File::create("httpbin.json").map(|mut file| {
file.write_all(&buffer).unwrap()
})
}).map_err(|e| eprintln!("Error occured: {:?}", e));
But ideally, we’d like to avoid that unwrap() and instead promote an
I/O error here to be handle by the map_err below. It turns out that
there’s a surprisingly trivial change to make that happen:
File::create("httpbin.json").and_then(|mut file| {
file.write_all(&buffer)
})
Instead of using map, we use and_then, which requires that we
return some value that implements Future. But fortunately, Result
itself implements Future! The Ok variant becomes the Item for
that Future, and the Err variant becomes its Error. Problem
solved!
Exercise 13
We haven’t taken advantage of tokio here at all! Let’s make this
program concurrent. Write a program that takes command line arguments
to determine HTTP requests to make and files to store them to. To
simplify the implementation, we’ll have it take input that looks like
the following:
$ cargo run httpbin.org:80 /json httpbin.json example.com:80 / homepage.html
Feel free to handle invalid command line arguments however’s easiest.
Solution
extern crate tokio;
use std::net::ToSocketAddrs;
use tokio::io::{read_to_end, write_all};
use tokio::net::TcpStream;
use tokio::prelude::*;
use std::fs::File;
fn download(host: String, path: String, filename: String) -> impl Future<Item=(), Error=()> {
let mut addr_iter = host.to_socket_addrs().unwrap();
let addr = match addr_iter.next() {
None => panic!("DNS resolution failed"),
Some(addr) => addr,
};
let req_body = format!(
"GET {} HTTP/1.1\r\nHost: {}:80\r\nConnection: close\r\n\r\n",
path,
host,
);
TcpStream::connect(&addr)
.and_then(|stream| {
write_all(stream, req_body).and_then(|(stream, _body)| {
let buffer = vec![];
read_to_end(stream, buffer).and_then(|(_stream, buffer)| {
File::create(filename).and_then(|mut file| {
file.write_all(&buffer)
})
})
})
}).map_err(|e| eprintln!("Error occured: {:?}", e))
}
fn main() {
tokio::run(future::poll_fn(|| {
let mut args = std::env::args().skip(1);
loop {
match (args.next(), args.next(), args.next()) {
(Some(host), Some(path), Some(filename)) => {
tokio::spawn(download(host, path, filename));
}
_ => return Ok(Async::Ready(())),
}
}
}))
}
Nicer error handling
We’re just panic!ing when we have a bad address. Let’s do a little
bit better. First, I’ll define a helper function to return a nice
Result. We’ll use a String for the Err variant, but we could
should ideally define an enum instead:
fn resolve_addr(host: &str) -> Result<SocketAddr, String> {
let mut addr_iter = match host.to_socket_addrs() {
Ok(addr_iter) => addr_iter,
Err(e) => return Err(format!("Invalid host name {:?}: {:?}", host, e)),
};
match addr_iter.next() {
None => Err(format!("No addresses found for host: {:?}", host)),
Some(addr) => Ok(addr),
}
}
Inside download, we could continue panic!ing with:
let addr = resolve_addr(&host).unwrap();
But let’s do better. Using ? won’t work, since we aren’t returning a
Result. One idea would be to use return to return early:
let addr = match resolve_addr(&host) {
Ok(addr) => addr,
Err(e) => {
eprintln!("Error resolving address: {}", e);
return future::err(());
}
};
However, we get an interesting error message from the compiler:
error[E0308]: mismatched types
--> src/main.rs:34:5
|
34 | / TcpStream::connect(&addr)
35 | | .and_then(|stream| {
36 | | write_all(stream, req_body).and_then(|(stream, _body)| {
37 | | let buffer = vec![];
... |
43 | | })
44 | | }).map_err(|e| eprintln!("Error occured: {:?}", e))
| |___________________________________________________________^ expected struct `tokio::prelude::future::FutureResult`, found struct `tokio::prelude::future::MapErr`
In order to make things work, we need to ensure that we always return
the same type. We’ve so far used impl Future to say “we’ll return
some type which is a Future,” but we haven’t told the compiler what
that type is. Instead, the compiler has inferred that. But now, we
have two different types.
One approach would be dynamic dispatch, such as using
Box<Future>. But there’s a better way: using the Either helper
type. This type is used in a case where we have two different types of
Futures which both have the same Item and Error. Let’s see how
we can rewrite our code above to use Either:
fn download(host: String, path: String, filename: String) -> impl Future<Item=(), Error=()> {
let addr = match resolve_addr(&host) {
Ok(addr) => addr,
Err(e) => {
eprintln!("Error resolving address: {}", e);
return future::Either::A(future::err(()));
}
};
let req_body = format!(
"GET {} HTTP/1.1\r\nHost: {}:80\r\nConnection: close\r\n\r\n",
path,
host,
);
future::Either::B(TcpStream::connect(&addr)
.and_then(|stream| {
write_all(stream, req_body).and_then(|(stream, _body)| {
let buffer = vec![];
read_to_end(stream, buffer).and_then(|(_stream, buffer)| {
File::create(filename).and_then(|mut file| {
file.write_all(&buffer)
})
})
})
}).map_err(|e| eprintln!("Error occured: {:?}", e)))
}
Exercise 14 Implement your own Either data type and use it in
the code above.
Solution:
enum Either<A, B> {
A(A),
B(B),
}
impl<A, B> Future for Either<A, B>
where A: Future<Item=B::Item, Error=B::Error>,
B: Future,
{
type Item = A::Item;
type Error = A::Error;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
match self {
Either::A(a) => a.poll(),
Either::B(b) => b.poll(),
}
}
}
TCP server
Having been so successful with our TCP client, let’s move over to the
server side. Conceptually, we want to:
- Bind a listening socket
- Accept connections from that socket
- Copy all data from the input side of the socket to the output side
of the socket
Binding a listening socket is going to be a blocking
call
to bind, taking our old friend SocketAddr. Since we’re not playing
around with DNS resolution anymore, we can be a bit lazier about how
we do this:
extern crate tokio;
use tokio::prelude::*;
use tokio::net::TcpListener;
fn main() {
let addr = "127.0.0.1:3000".parse().expect("couldn't parse address");
let listener = TcpListener::bind(&addr).expect("couldn't bind address");
println!("It worked! {:?}", listener);
}
We now have have a TcpListener. Unlike other types we’ve seen, this
doesn’t have an implementation of either Future or
Stream. However, it does have a method called incoming(), which
returns an Incoming, which has a Stream implementation, where
Item is TcpStream. That looks promising!
let future = listener
.incoming()
.for_each(|socket| {
println!("Accepted a connection! {:?}", socket);
future::ok(())
})
.map_err(|e| eprintln!("An error occurred: {:?}", e))
;
tokio::run(future)
And just like that, we’ve implemented points (1) and (2) above. We’re
just left with point 3: copying all of the data. Let’s search tokio
for something to do
copying. It
looks like
tokio::io::copy
will do. We need to provide it both a reader and writer. Since we’re
reader from and writing to the same socket, let’s just provide the
same value for both:
let future = listener
.incoming()
.for_each(|socket| {
copy(socket, socket)
.map(|_| println!("Connection closed"))
})
.map_err(|e| eprintln!("An error occurred: {:?}", e))
;
Are you already laughing at my comically silly mistake?
error[E0382]: use of moved value: `socket`
--> src/main.rs:13:26
|
13 | copy(socket, socket)
| ------ ^^^^^^ value used here after move
| |
| value moved here
Of course we can’t use the same value in both positions. Fortunately,
when designing Streams, the authors provided a method called
split
to give us a read and write end of the stream. With that in hand, our
echo server becomes trivial:
extern crate tokio;
use tokio::prelude::*;
use tokio::net::TcpListener;
use tokio::io::copy;
fn main() {
let addr = "127.0.0.1:3000".parse().expect("couldn't parse address");
let listener = TcpListener::bind(&addr).expect("couldn't bind address");
let future = listener
.incoming()
.for_each(|socket| {
let (reader, writer) = socket.split();
copy(reader, writer)
.map(|_| println!("Connection closed"))
})
.map_err(|e| eprintln!("An error occurred: {:?}", e))
;
tokio::run(future)
}
Writing directly
Using copy kind of ignores the gory details of what’s going on under
the surface. Let’s start off by writing some arbitrary message to the
writer side of things, using the write_all
function.
NOTE The tokio::io::write_all function takes a AsyncWrite and
returns a WriteAll Future. Don’t be confused by the presence of a
write_all method, which in fact is a blocking call. I wasted about
5 minutes fumbling with that while writing this tutorial.
extern crate tokio;
use tokio::prelude::*;
use tokio::net::TcpListener;
use tokio::io::{copy, write_all};
fn main() {
let addr = "127.0.0.1:3000".parse().expect("couldn't parse address");
let listener = TcpListener::bind(&addr).expect("couldn't bind address");
let future = listener
.incoming()
.for_each(|socket| {
let (reader, writer) = socket.split();
write_all(writer, b"Welcome to the echo server\r\n")
.map(|_| println!("Connection closed"))
})
.map_err(|e| eprintln!("An error occurred: {:?}", e))
;
tokio::run(future)
}
Exercise 15 Modify the code above so that, after printing “Welcome
to the echo server”, it proceeds to actually echo content sent in.
write_all(writer, b"Welcome to the echo server\r\n")
.and_then(|(writer, _)| {
copy(reader, writer)
.map(|_| println!("Connection closed"))
})
Codecs
Reading from the reader directly is slightly trickier than writing
to the writer. We could go play around with the underlying polling
reading functions, but we’re not going to here. (Feel free to read
the official tokio tutorial for more
information.)
Instead, we’re going to introduce a new concept, codecs. So far,
we’ve implicitly been working with the AsyncRead and AsyncWrite
traits, which essentially provide the raw polling functions we’d need
for building up our own Futures (as we did way long ago at the
beginning of this lesson). However, we often don’t want to work at
that level of abstraction. Instead, we’d like to deal with some kind
of framed (or chunked) data.
The new abstraction instead will be a Sink, which is “a value into
which other values can be sent, asynchronously.” We’ll continue to use
the Stream trait for the read side, which we’re already quite
familiar with.
Let’s contrive an example. Our echo server currently provides slightly
weird output:
Hello
Hello
There
There
World
World
It’s hard to tell what I typed in, and what the server
responded. Instead, I’d like each line sent back from the server to
begin with “You said: “. Doing that with the abstractions we’ve seen
so far would be fairly painful: we’d need to grab chunks of data, look
for the newline character, break up the input, splice in the “You
said: “ message. I know this is a crash course and all, but I’d rather
not crash into that.
Instead, let’s jump straight to the better solution. I want to treat
our TCP stream as a stream of lines of data. If I search for the word
“lines” (and this is actually how I learned about codecs), I end up
with
LinesCodec. It
provides a method new(), as well as new_with_max_length. We’ll use
new here, but I recommend reading the docs to see why that’s a
terrible idea in any kind of security sensitive context.
The only other method on the type is max_length, which doesn’t look
like it’s going to help us actually deal with a TCP socket as a stream
of lines. So let’s look down at the trait implementations. We’ve got
all of our usual suspects: Clone, PartialOrd, etc. But two new
ones stick out: Decoder and Encoder. Well that certainly looks
interesting.
Reading through the docs on Decoder, it provides a method called
framed,
which has a description that is great. (Please, take a second to
follow that link and read the docs.) Without further ado, let’s try
adding in our LinesCodec to our echo server:
extern crate tokio;
use tokio::prelude::*;
use tokio::net::TcpListener;
use tokio::codec::{Decoder, LinesCodec};
fn main() {
let addr = "127.0.0.1:3000".parse().expect("couldn't parse address");
let listener = TcpListener::bind(&addr).expect("couldn't bind address");
let future = listener
.incoming()
.for_each(|socket| {
let lines_codec = LinesCodec::new();
let socket = lines_codec.framed(socket);
socket
.send(String::from("Welcome to the echo server"))
.map(|_| println!("Connection closed"))
})
.map_err(|e| eprintln!("An error occurred: {:?}", e))
;
tokio::run(future)
}
You may have noticed that we no longer have a newline sequence at the
end of the “Welcome” string. That’s because our lines codec
automatically handles that. Additionally, we now need to use
String::from, since the Item for this Sink is a String.
We can also use split to isolate the Sink from the Stream:
let (sink, stream) = lines_codec.framed(socket).split();
sink
.send(String::from("Welcome to the echo server"))
.map(|_| println!("Connection closed"))
And, we can use for_each on the Stream side to get a stream of the
lines:
extern crate tokio;
use tokio::prelude::*;
use tokio::net::TcpListener;
use tokio::codec::{Decoder, LinesCodec};
fn main() {
let addr = "127.0.0.1:3000".parse().expect("couldn't parse address");
let listener = TcpListener::bind(&addr).expect("couldn't bind address");
let future = listener
.incoming()
.for_each(|socket| {
let lines_codec = LinesCodec::new();
let (sink, stream) = lines_codec.framed(socket).split();
sink
.send(String::from("Welcome to the echo server"))
.and_then(|sink| {
stream
.for_each(|line| {
println!("Received a line: {}", line);
future::ok(())
})
.map(|_| println!("Connection closed"))
})
})
.map_err(|e| eprintln!("An error occurred: {:?}", e))
;
tokio::run(future)
}
We’re almost done here: we just need to send the lines back to the
sink instead of to stdout. Unfortunately, using the send method
we’ve seen so far is going to be tricky, since we’ll end up consuming
the sink in each iteration of for_each. We could figure out a way
to make that all work, but instead, let’s just cut to the chase and
use send_all.
Exercise 16 Modify the code above so that, instead of printing the
lines to standard output, they get sent back to the client with the
message “You said: “. You’ll want to look at
send_all
Solution
extern crate tokio;
use tokio::prelude::*;
use tokio::net::TcpListener;
use tokio::codec::{Decoder, LinesCodec};
fn main() {
let addr = "127.0.0.1:3000".parse().expect("couldn't parse address");
let listener = TcpListener::bind(&addr).expect("couldn't bind address");
let future = listener
.incoming()
.for_each(|socket| {
let lines_codec = LinesCodec::new();
let (sink, stream) = lines_codec.framed(socket).split();
sink
.send(String::from("Welcome to the echo server"))
.and_then(|sink| {
let stream = stream
.map(|line| format!("You said: {}", line))
;
sink.send_all(stream)
.map(|_| println!("Connection closed"))
})
})
.map_err(|e| eprintln!("An error occurred: {:?}", e))
;
tokio::run(future)
}
Next time
Whew, that was a big lesson! At this point, you should have a very
solid footing in the ins and outs of tokio. It’s time to get lots more
experience with using the library, and related higher level libraries
for doing things like HTTP servers and clients.
Depending on reader feedback, the next lesson may either go deeper
into tokio and related libraries, or go back to more fundamental
aspects of Rust like lifetimes. From the tokio side, we’d play with:
- Message passing between tasks
- UDP communications
- Recursive directory traversal
- Parallel downloading of files
Rust at FP Complete | Introduction
![\displaystyle \begin{array}{@{}ll} \multicolumn{2}{@{}l}{\mathit{insertsort} :: [\mathit{Integer}] \rightarrow [\mathit{Integer}]} \\ \mathit{insertsort}\;[\,] &= [\,] \\ \mathit{insertsort}\;(a:x) &= \mathit{insert}\;a\;(\mathit{insertsort}\;x) \medskip \\ \multicolumn{2}{@{}l}{\mathit{insert} :: \mathit{Integer} \rightarrow [\mathit{Integer}] \rightarrow [\mathit{Integer}]} \\ \mathit{insert}\;b\;[\,] &= [b] \\ \mathit{insert}\;b\;(a:x) \\ \qquad \mid b \le a &= b : a : x \\ \qquad \mid b > a &= a : \mathit{insert}\;b\;x \end{array}](https://s0.wp.com/latex.php?latex=%5Cdisplaystyle++%5Cbegin%7Barray%7D%7B%40%7B%7Dll%7D+%5Cmulticolumn%7B2%7D%7B%40%7B%7Dl%7D%7B%5Cmathit%7Binsertsort%7D+%3A%3A+%5B%5Cmathit%7BInteger%7D%5D+%5Crightarrow+%5B%5Cmathit%7BInteger%7D%5D%7D+%5C%5C+%5Cmathit%7Binsertsort%7D%5C%3B%5B%5C%2C%5D+%26%3D+%5B%5C%2C%5D+%5C%5C+%5Cmathit%7Binsertsort%7D%5C%3B%28a%3Ax%29+%26%3D+%5Cmathit%7Binsert%7D%5C%3Ba%5C%3B%28%5Cmathit%7Binsertsort%7D%5C%3Bx%29+%5Cmedskip+%5C%5C+%5Cmulticolumn%7B2%7D%7B%40%7B%7Dl%7D%7B%5Cmathit%7Binsert%7D+%3A%3A+%5Cmathit%7BInteger%7D+%5Crightarrow+%5B%5Cmathit%7BInteger%7D%5D+%5Crightarrow+%5B%5Cmathit%7BInteger%7D%5D%7D+%5C%5C+%5Cmathit%7Binsert%7D%5C%3Bb%5C%3B%5B%5C%2C%5D+%26%3D+%5Bb%5D+%5C%5C+%5Cmathit%7Binsert%7D%5C%3Bb%5C%3B%28a%3Ax%29+%5C%5C+%5Cqquad+%5Cmid+b+%5Cle+a+%26%3D+b+%3A+a+%3A+x+%5C%5C+%5Cqquad+%5Cmid+b+%3E+a+%26%3D+a+%3A+%5Cmathit%7Binsert%7D%5C%3Bb%5C%3Bx+%5Cend%7Barray%7D+&bg=ffffff&fg=000000&s=0 "\displaystyle \begin{array}{@{}ll} \multicolumn{2}{@{}l}{\mathit{insertsort} :: [\mathit{Integer}] \rightarrow [\mathit{Integer}]} \\ \mathit{insertsort}\;[\,] &= [\,] \\ \mathit{insertsort}\;(a:x) &= \mathit{insert}\;a\;(\mathit{insertsort}\;x) \medskip \\ \multicolumn{2}{@{}l}{\mathit{insert} :: \mathit{Integer} \rightarrow [\mathit{Integer}] \rightarrow [\mathit{Integer}]} \\ \mathit{insert}\;b\;[\,] &= [b] \\ \mathit{insert}\;b\;(a:x) \\ \qquad \mid b \le a &= b : a : x \\ \qquad \mid b > a &= a : \mathit{insert}\;b\;x \end{array}")
![\displaystyle \begin{array}{@{}ll} \multicolumn{2}{@{}l}{\mathit{quicksort} :: [\mathit{Integer}] \rightarrow [\mathit{Integer}]} \\ \mathit{quicksort}\;[\,] & = [\,] \\ \mathit{quicksort}\;(a:x) &= \mathit{quicksort}\;y \mathbin{{+}\!\!\!{+}} [a] \mathbin{{+}\!\!\!{+}} \mathit{quicksort}\;z \\ & \qquad \mathbf{where}\; \begin{array}[t]{@{}ll} y &= [ b \mid b \leftarrow x, b \le a ] \\ z &= [ b \mid b \leftarrow x, b > a ] \end{array} \end{array}](https://s0.wp.com/latex.php?latex=%5Cdisplaystyle++%5Cbegin%7Barray%7D%7B%40%7B%7Dll%7D+%5Cmulticolumn%7B2%7D%7B%40%7B%7Dl%7D%7B%5Cmathit%7Bquicksort%7D+%3A%3A+%5B%5Cmathit%7BInteger%7D%5D+%5Crightarrow+%5B%5Cmathit%7BInteger%7D%5D%7D+%5C%5C+%5Cmathit%7Bquicksort%7D%5C%3B%5B%5C%2C%5D+%26+%3D+%5B%5C%2C%5D+%5C%5C+%5Cmathit%7Bquicksort%7D%5C%3B%28a%3Ax%29+%26%3D+%5Cmathit%7Bquicksort%7D%5C%3By+%5Cmathbin%7B%7B%2B%7D%5C%21%5C%21%5C%21%7B%2B%7D%7D+%5Ba%5D+%5Cmathbin%7B%7B%2B%7D%5C%21%5C%21%5C%21%7B%2B%7D%7D+%5Cmathit%7Bquicksort%7D%5C%3Bz+%5C%5C+%26+%5Cqquad+%5Cmathbf%7Bwhere%7D%5C%3B+%5Cbegin%7Barray%7D%5Bt%5D%7B%40%7B%7Dll%7D+y+%26%3D+%5B+b+%5Cmid+b+%5Cleftarrow+x%2C+b+%5Cle+a+%5D+%5C%5C+z+%26%3D+%5B+b+%5Cmid+b+%5Cleftarrow+x%2C+b+%3E+a+%5D+%5Cend%7Barray%7D+%5Cend%7Barray%7D+&bg=ffffff&fg=000000&s=0 "\displaystyle \begin{array}{@{}ll} \multicolumn{2}{@{}l}{\mathit{quicksort} :: [\mathit{Integer}] \rightarrow [\mathit{Integer}]} \\ \mathit{quicksort}\;[\,] & = [\,] \\ \mathit{quicksort}\;(a:x) &= \mathit{quicksort}\;y \mathbin{{+}\!\!\!{+}} [a] \mathbin{{+}\!\!\!{+}} \mathit{quicksort}\;z \\ & \qquad \mathbf{where}\; \begin{array}[t]{@{}ll} y &= [ b \mid b \leftarrow x, b \le a ] \\ z &= [ b \mid b \leftarrow x, b > a ] \end{array} \end{array}")
![\displaystyle \begin{array}{@{}ll} \multicolumn{2}{@{}l}{\mathit{readDate} :: \mathit{String} \rightarrow \mathit{Date}} \\ \mathit{readDate}\;s & = \mathit{Date}\;\{ \mathit{day} = d, \mathit{month} = m, \mathit{year} = y \} \\ & \qquad \mathbf{where}\; \begin{array}[t]{@{}ll} d & = ... s ... \\ m & = ... s ... \\ y & = ... s ... \end{array} \end{array}](https://s0.wp.com/latex.php?latex=%5Cdisplaystyle++%5Cbegin%7Barray%7D%7B%40%7B%7Dll%7D+%5Cmulticolumn%7B2%7D%7B%40%7B%7Dl%7D%7B%5Cmathit%7BreadDate%7D+%3A%3A+%5Cmathit%7BString%7D+%5Crightarrow+%5Cmathit%7BDate%7D%7D+%5C%5C+%5Cmathit%7BreadDate%7D%5C%3Bs+%26+%3D+%5Cmathit%7BDate%7D%5C%3B%5C%7B+%5Cmathit%7Bday%7D+%3D+d%2C+%5Cmathit%7Bmonth%7D+%3D+m%2C+%5Cmathit%7Byear%7D+%3D+y+%5C%7D+%5C%5C+%26+%5Cqquad+%5Cmathbf%7Bwhere%7D%5C%3B+%5Cbegin%7Barray%7D%5Bt%5D%7B%40%7B%7Dll%7D+d+%26+%3D+...+s+...+%5C%5C+m+%26+%3D+...+s+...+%5C%5C+y+%26+%3D+...+s+...+%5Cend%7Barray%7D+%5Cend%7Barray%7D+&bg=ffffff&fg=000000&s=0 "\displaystyle \begin{array}{@{}ll} \multicolumn{2}{@{}l}{\mathit{readDate} :: \mathit{String} \rightarrow \mathit{Date}} \\ \mathit{readDate}\;s & = \mathit{Date}\;\{ \mathit{day} = d, \mathit{month} = m, \mathit{year} = y \} \\ & \qquad \mathbf{where}\; \begin{array}[t]{@{}ll} d & = ... s ... \\ m & = ... s ... \\ y & = ... s ... \end{array} \end{array}")
![{[1,2,3]}](https://s0.wp.com/latex.php?latex=%7B%5B1%2C2%2C3%5D%7D&bg=ffffff&fg=000000&s=0 "{[1,2,3]}")
and ![{[4,5,6]}](https://s0.wp.com/latex.php?latex=%7B%5B4%2C5%2C6%5D%7D&bg=ffffff&fg=000000&s=0 "{[4,5,6]}")
to ![{[(1,4),(2,5),(3,6)]}](https://s0.wp.com/latex.php?latex=%7B%5B%281%2C4%29%2C%282%2C5%29%2C%283%2C6%29%5D%7D&bg=ffffff&fg=000000&s=0 "{[(1,4),(2,5),(3,6)]}")
, and for simplicity let’s say pruning the result to the length of the shorter input. One can again solve the problem by case analysis on the input, but the fact that there are two inputs makes that a bit awkward—whether to do case analysis on one list in favour of the other, or to analyse both in parallel. For example, here is the outcome of case analysis on the first input, followed if that is non-empty by a case analysis on the second input: ![\displaystyle \begin{array}{@{}ll} \multicolumn{2}{@{}l}{\mathit{zip} :: [\alpha] \rightarrow [\beta] \rightarrow [(\alpha,\beta)]} \\ \mathit{zip}\;x\;y &= \begin{array}[t]{@{}l} \mathbf{if}\; \mathit{null}\;x \;\mathbf{then}\; [\,] \;\mathbf{else} \\ \qquad \mathbf{if}\; \mathit{null}\;y \;\mathbf{then}\; [\,] \;\mathbf{else} \\ \qquad \qquad (\mathit{head}\;x,\mathit{head}\;y) : \mathit{zip}\;(\mathit{tail}\;x)\;(\mathit{tail}\;y) \end{array} \end{array}](https://s0.wp.com/latex.php?latex=%5Cdisplaystyle++%5Cbegin%7Barray%7D%7B%40%7B%7Dll%7D+%5Cmulticolumn%7B2%7D%7B%40%7B%7Dl%7D%7B%5Cmathit%7Bzip%7D+%3A%3A+%5B%5Calpha%5D+%5Crightarrow+%5B%5Cbeta%5D+%5Crightarrow+%5B%28%5Calpha%2C%5Cbeta%29%5D%7D+%5C%5C+%5Cmathit%7Bzip%7D%5C%3Bx%5C%3By+%26%3D+%5Cbegin%7Barray%7D%5Bt%5D%7B%40%7B%7Dl%7D+%5Cmathbf%7Bif%7D%5C%3B+%5Cmathit%7Bnull%7D%5C%3Bx+%5C%3B%5Cmathbf%7Bthen%7D%5C%3B+%5B%5C%2C%5D+%5C%3B%5Cmathbf%7Belse%7D+%5C%5C+%5Cqquad+%5Cmathbf%7Bif%7D%5C%3B+%5Cmathit%7Bnull%7D%5C%3By+%5C%3B%5Cmathbf%7Bthen%7D%5C%3B+%5B%5C%2C%5D+%5C%3B%5Cmathbf%7Belse%7D+%5C%5C+%5Cqquad+%5Cqquad+%28%5Cmathit%7Bhead%7D%5C%3Bx%2C%5Cmathit%7Bhead%7D%5C%3By%29+%3A+%5Cmathit%7Bzip%7D%5C%3B%28%5Cmathit%7Btail%7D%5C%3Bx%29%5C%3B%28%5Cmathit%7Btail%7D%5C%3By%29+%5Cend%7Barray%7D+%5Cend%7Barray%7D+&bg=ffffff&fg=000000&s=0 "\displaystyle \begin{array}{@{}ll} \multicolumn{2}{@{}l}{\mathit{zip} :: [\alpha] \rightarrow [\beta] \rightarrow [(\alpha,\beta)]} \\ \mathit{zip}\;x\;y &= \begin{array}[t]{@{}l} \mathbf{if}\; \mathit{null}\;x \;\mathbf{then}\; [\,] \;\mathbf{else} \\ \qquad \mathbf{if}\; \mathit{null}\;y \;\mathbf{then}\; [\,] \;\mathbf{else} \\ \qquad \qquad (\mathit{head}\;x,\mathit{head}\;y) : \mathit{zip}\;(\mathit{tail}\;x)\;(\mathit{tail}\;y) \end{array} \end{array}")
![\displaystyle \begin{array}{@{}ll} \multicolumn{2}{@{}l}{\mathit{zip} :: [\alpha] \rightarrow [\beta] \rightarrow [(\alpha,\beta)]} \\ \mathit{zip}\;x\;y &= \begin{array}[t]{@{}l} \mathbf{if}\; \mathit{null}\;x \lor \mathit{null}\;y \;\mathbf{then}\; [\,] \;\mathbf{else} \\ \qquad (\mathit{head}\;x,\mathit{head}\;y) : \mathit{zip}\;(\mathit{tail}\;x)\;(\mathit{tail}\;y) \end{array} \end{array}](https://s0.wp.com/latex.php?latex=%5Cdisplaystyle++%5Cbegin%7Barray%7D%7B%40%7B%7Dll%7D+%5Cmulticolumn%7B2%7D%7B%40%7B%7Dl%7D%7B%5Cmathit%7Bzip%7D+%3A%3A+%5B%5Calpha%5D+%5Crightarrow+%5B%5Cbeta%5D+%5Crightarrow+%5B%28%5Calpha%2C%5Cbeta%29%5D%7D+%5C%5C+%5Cmathit%7Bzip%7D%5C%3Bx%5C%3By+%26%3D+%5Cbegin%7Barray%7D%5Bt%5D%7B%40%7B%7Dl%7D+%5Cmathbf%7Bif%7D%5C%3B+%5Cmathit%7Bnull%7D%5C%3Bx+%5Clor+%5Cmathit%7Bnull%7D%5C%3By+%5C%3B%5Cmathbf%7Bthen%7D%5C%3B+%5B%5C%2C%5D+%5C%3B%5Cmathbf%7Belse%7D+%5C%5C+%5Cqquad+%28%5Cmathit%7Bhead%7D%5C%3Bx%2C%5Cmathit%7Bhead%7D%5C%3By%29+%3A+%5Cmathit%7Bzip%7D%5C%3B%28%5Cmathit%7Btail%7D%5C%3Bx%29%5C%3B%28%5Cmathit%7Btail%7D%5C%3By%29+%5Cend%7Barray%7D+%5Cend%7Barray%7D+&bg=ffffff&fg=000000&s=0 "\displaystyle \begin{array}{@{}ll} \multicolumn{2}{@{}l}{\mathit{zip} :: [\alpha] \rightarrow [\beta] \rightarrow [(\alpha,\beta)]} \\ \mathit{zip}\;x\;y &= \begin{array}[t]{@{}l} \mathbf{if}\; \mathit{null}\;x \lor \mathit{null}\;y \;\mathbf{then}\; [\,] \;\mathbf{else} \\ \qquad (\mathit{head}\;x,\mathit{head}\;y) : \mathit{zip}\;(\mathit{tail}\;x)\;(\mathit{tail}\;y) \end{array} \end{array}")
![\displaystyle \begin{array}{@{}ll} \multicolumn{2}{@{}l}{\mathit{selectSort} :: [\mathit{Integer}] \rightarrow [\mathit{Integer}]} \\ \mathit{selectSort}\;x & = \mathbf{if}\; \mathit{null}\;x \;\mathbf{then}\; [\,] \;\mathbf{else}\; a : \mathit{selectSort}\;(x \mathbin{\backslash\!\backslash} [a]) \\ & \qquad \mathbf{where}\; a = \mathit{minimum}\;x \end{array}](https://s0.wp.com/latex.php?latex=%5Cdisplaystyle++%5Cbegin%7Barray%7D%7B%40%7B%7Dll%7D+%5Cmulticolumn%7B2%7D%7B%40%7B%7Dl%7D%7B%5Cmathit%7BselectSort%7D+%3A%3A+%5B%5Cmathit%7BInteger%7D%5D+%5Crightarrow+%5B%5Cmathit%7BInteger%7D%5D%7D+%5C%5C+%5Cmathit%7BselectSort%7D%5C%3Bx+%26+%3D+%5Cmathbf%7Bif%7D%5C%3B+%5Cmathit%7Bnull%7D%5C%3Bx+%5C%3B%5Cmathbf%7Bthen%7D%5C%3B+%5B%5C%2C%5D+%5C%3B%5Cmathbf%7Belse%7D%5C%3B+a+%3A+%5Cmathit%7BselectSort%7D%5C%3B%28x+%5Cmathbin%7B%5Cbackslash%5C%21%5Cbackslash%7D+%5Ba%5D%29+%5C%5C+%26+%5Cqquad+%5Cmathbf%7Bwhere%7D%5C%3B+a+%3D+%5Cmathit%7Bminimum%7D%5C%3Bx+%5Cend%7Barray%7D+&bg=ffffff&fg=000000&s=0 "\displaystyle \begin{array}{@{}ll} \multicolumn{2}{@{}l}{\mathit{selectSort} :: [\mathit{Integer}] \rightarrow [\mathit{Integer}]} \\ \mathit{selectSort}\;x & = \mathbf{if}\; \mathit{null}\;x \;\mathbf{then}\; [\,] \;\mathbf{else}\; a : \mathit{selectSort}\;(x \mathbin{\backslash\!\backslash} [a]) \\ & \qquad \mathbf{where}\; a = \mathit{minimum}\;x \end{array}")
denotes list 
with the elements of list 
removed).
to construct an intermediate tree, followed by structural recursion 
over that tree to produce the resulting list: ![\displaystyle \begin{array}{@{}ll} \multicolumn{2}{@{}l}{\mathit{quicksort} :: [\mathit{Integer}] \rightarrow [\mathit{Integer}]} \\ \mathit{quicksort} & = \mathit{flatten} \cdot \mathit{build} \medskip \\ \multicolumn{2}{@{}l}{\mathbf{data}\;\mathit{Tree} = \mathit{Empty} \mid \mathit{Node}\;\mathit{Tree}\;\mathit{Integer}\;\mathit{Tree}} \medskip\\ \multicolumn{2}{@{}l}{\mathit{build} :: [\mathit{Integer}] \rightarrow \mathit{Tree}} \\ \mathit{build}\;[\,] & = \mathit{Empty} \\ \mathit{build}\;(a:x) &= \mathit{Node}\;(\mathit{build}\;y)\;a\;(\mathit{build}\;z) \\ & \qquad \mathbf{where}\; \begin{array}[t]{@{}ll} y &= [ b \mid b \leftarrow x, b \le a ] \\ z &= [ b \mid b \leftarrow x, b > a ] \end{array} \medskip \\ \multicolumn{2}{@{}l}{\mathit{flatten} :: \mathit{Tree} \rightarrow [\mathit{Integer}]} \\ \mathit{flatten}\;\mathit{Empty} & = [\,] \\ \mathit{flatten}\;(\mathit{Node}\;t\;a\;u) &= \mathit{flatten}\;t \mathbin{{+}\!\!\!{+}} [a] \mathbin{{+}\!\!\!{+}} \mathit{flatten}\;u \end{array}](https://s0.wp.com/latex.php?latex=%5Cdisplaystyle++%5Cbegin%7Barray%7D%7B%40%7B%7Dll%7D+%5Cmulticolumn%7B2%7D%7B%40%7B%7Dl%7D%7B%5Cmathit%7Bquicksort%7D+%3A%3A+%5B%5Cmathit%7BInteger%7D%5D+%5Crightarrow+%5B%5Cmathit%7BInteger%7D%5D%7D+%5C%5C+%5Cmathit%7Bquicksort%7D+%26+%3D+%5Cmathit%7Bflatten%7D+%5Ccdot+%5Cmathit%7Bbuild%7D+%5Cmedskip+%5C%5C+%5Cmulticolumn%7B2%7D%7B%40%7B%7Dl%7D%7B%5Cmathbf%7Bdata%7D%5C%3B%5Cmathit%7BTree%7D+%3D+%5Cmathit%7BEmpty%7D+%5Cmid+%5Cmathit%7BNode%7D%5C%3B%5Cmathit%7BTree%7D%5C%3B%5Cmathit%7BInteger%7D%5C%3B%5Cmathit%7BTree%7D%7D+%5Cmedskip%5C%5C+%5Cmulticolumn%7B2%7D%7B%40%7B%7Dl%7D%7B%5Cmathit%7Bbuild%7D+%3A%3A+%5B%5Cmathit%7BInteger%7D%5D+%5Crightarrow+%5Cmathit%7BTree%7D%7D+%5C%5C+%5Cmathit%7Bbuild%7D%5C%3B%5B%5C%2C%5D+%26+%3D+%5Cmathit%7BEmpty%7D+%5C%5C+%5Cmathit%7Bbuild%7D%5C%3B%28a%3Ax%29+%26%3D+%5Cmathit%7BNode%7D%5C%3B%28%5Cmathit%7Bbuild%7D%5C%3By%29%5C%3Ba%5C%3B%28%5Cmathit%7Bbuild%7D%5C%3Bz%29+%5C%5C+%26+%5Cqquad+%5Cmathbf%7Bwhere%7D%5C%3B+%5Cbegin%7Barray%7D%5Bt%5D%7B%40%7B%7Dll%7D+y+%26%3D+%5B+b+%5Cmid+b+%5Cleftarrow+x%2C+b+%5Cle+a+%5D+%5C%5C+z+%26%3D+%5B+b+%5Cmid+b+%5Cleftarrow+x%2C+b+%3E+a+%5D+%5Cend%7Barray%7D+%5Cmedskip+%5C%5C+%5Cmulticolumn%7B2%7D%7B%40%7B%7Dl%7D%7B%5Cmathit%7Bflatten%7D+%3A%3A+%5Cmathit%7BTree%7D+%5Crightarrow+%5B%5Cmathit%7BInteger%7D%5D%7D+%5C%5C+%5Cmathit%7Bflatten%7D%5C%3B%5Cmathit%7BEmpty%7D+%26+%3D+%5B%5C%2C%5D+%5C%5C+%5Cmathit%7Bflatten%7D%5C%3B%28%5Cmathit%7BNode%7D%5C%3Bt%5C%3Ba%5C%3Bu%29+%26%3D+%5Cmathit%7Bflatten%7D%5C%3Bt+%5Cmathbin%7B%7B%2B%7D%5C%21%5C%21%5C%21%7B%2B%7D%7D+%5Ba%5D+%5Cmathbin%7B%7B%2B%7D%5C%21%5C%21%5C%21%7B%2B%7D%7D+%5Cmathit%7Bflatten%7D%5C%3Bu+%5Cend%7Barray%7D+&bg=ffffff&fg=000000&s=0 "\displaystyle \begin{array}{@{}ll} \multicolumn{2}{@{}l}{\mathit{quicksort} :: [\mathit{Integer}] \rightarrow [\mathit{Integer}]} \\ \mathit{quicksort} & = \mathit{flatten} \cdot \mathit{build} \medskip \\ \multicolumn{2}{@{}l}{\mathbf{data}\;\mathit{Tree} = \mathit{Empty} \mid \mathit{Node}\;\mathit{Tree}\;\mathit{Integer}\;\mathit{Tree}} \medskip\\ \multicolumn{2}{@{}l}{\mathit{build} :: [\mathit{Integer}] \rightarrow \mathit{Tree}} \\ \mathit{build}\;[\,] & = \mathit{Empty} \\ \mathit{build}\;(a:x) &= \mathit{Node}\;(\mathit{build}\;y)\;a\;(\mathit{build}\;z) \\ & \qquad \mathbf{where}\; \begin{array}[t]{@{}ll} y &= [ b \mid b \leftarrow x, b \le a ] \\ z &= [ b \mid b \leftarrow x, b > a ] \end{array} \medskip \\ \multicolumn{2}{@{}l}{\mathit{flatten} :: \mathit{Tree} \rightarrow [\mathit{Integer}]} \\ \mathit{flatten}\;\mathit{Empty} & = [\,] \\ \mathit{flatten}\;(\mathit{Node}\;t\;a\;u) &= \mathit{flatten}\;t \mathbin{{+}\!\!\!{+}} [a] \mathbin{{+}\!\!\!{+}} \mathit{flatten}\;u \end{array}")
datatype, the first as structural corecursion and the second as structural recursion. A similar explanation applies to any divide-and-conquer algorithm; for example, MergeSort is another divide-and-conquer sorting algorithm, with the same intermediate tree shape, but this time with a simple splitting phase and all the comparisons in the recombining phase: ![\displaystyle \begin{array}{@{}ll} \multicolumn{2}{@{}l}{\mathit{mergesort} :: [\mathit{Integer}] \rightarrow [\mathit{Integer}]} \\ \mathit{mergesort} & = \mathit{mergeAll} \cdot \mathit{splitUp} \medskip \\ \multicolumn{2}{@{}l}{\mathit{splitUp} :: [\mathit{Integer}] \rightarrow \mathit{Tree}} \\ \mathit{splitUp}\;[\,] & = \mathit{Empty} \\ \mathit{splitUp}\;(a:x) &= \mathit{Node}\;(\mathit{splitUp}\;y)\;a\;(\mathit{splitUp}\;z) \\ & \qquad \mathbf{where}\; (y,z) = \mathit{halve}\;x \medskip \\ \multicolumn{2}{@{}l}{\mathit{halve} :: [\mathit{Integer}] \rightarrow ([\mathit{Integer}],[\mathit{Integer}])} \\ \mathit{halve}\;[\,] & = ( [\,], [\,]) \\ \mathit{halve}\;[a] & = ( [a], [\,]) \\ \mathit{halve}\;(a:b:x) & = (a:y,b:z) \;\mathbf{where}\; (y,z) = \mathit{halve}\;x \medskip \\ \multicolumn{2}{@{}l}{\mathit{mergeAll} :: \mathit{Tree} \rightarrow [\mathit{Integer}]} \\ \mathit{mergeAll}\;\mathit{Empty} & = [\,] \\ \mathit{mergeAll}\;(\mathit{Node}\;t\;a\;u) &= \mathit{merge}\;(\mathit{mergeAll}\;t)\;(\mathit{merge}\;[a]\;(\mathit{mergeAll}\;u)) \medskip \\ \multicolumn{2}{@{}l}{\mathit{merge} :: ([\mathit{Integer}],[\mathit{Integer}]) \rightarrow [\mathit{Integer}]} \\ \mathit{merge}\;[\,]\;y & = y \\ \mathit{merge}\;x\;[\,] & = x \\ \mathit{merge}\;(a:x)\;(b:y) & = \begin{array}[t]{@{}l@{}l@{}l} \mathbf{if}\; a \le b\; & \mathbf{then}\; & a : \mathit{merge}\;x\;(b:y) \\ & \mathbf{else}\; & b : \mathit{merge}\;(a:x)\;y \end{array} \end{array}](https://s0.wp.com/latex.php?latex=%5Cdisplaystyle++%5Cbegin%7Barray%7D%7B%40%7B%7Dll%7D+%5Cmulticolumn%7B2%7D%7B%40%7B%7Dl%7D%7B%5Cmathit%7Bmergesort%7D+%3A%3A+%5B%5Cmathit%7BInteger%7D%5D+%5Crightarrow+%5B%5Cmathit%7BInteger%7D%5D%7D+%5C%5C+%5Cmathit%7Bmergesort%7D+%26+%3D+%5Cmathit%7BmergeAll%7D+%5Ccdot+%5Cmathit%7BsplitUp%7D+%5Cmedskip+%5C%5C+%5Cmulticolumn%7B2%7D%7B%40%7B%7Dl%7D%7B%5Cmathit%7BsplitUp%7D+%3A%3A+%5B%5Cmathit%7BInteger%7D%5D+%5Crightarrow+%5Cmathit%7BTree%7D%7D+%5C%5C+%5Cmathit%7BsplitUp%7D%5C%3B%5B%5C%2C%5D+%26+%3D+%5Cmathit%7BEmpty%7D+%5C%5C+%5Cmathit%7BsplitUp%7D%5C%3B%28a%3Ax%29+%26%3D+%5Cmathit%7BNode%7D%5C%3B%28%5Cmathit%7BsplitUp%7D%5C%3By%29%5C%3Ba%5C%3B%28%5Cmathit%7BsplitUp%7D%5C%3Bz%29+%5C%5C+%26+%5Cqquad+%5Cmathbf%7Bwhere%7D%5C%3B+%28y%2Cz%29+%3D+%5Cmathit%7Bhalve%7D%5C%3Bx+%5Cmedskip+%5C%5C+%5Cmulticolumn%7B2%7D%7B%40%7B%7Dl%7D%7B%5Cmathit%7Bhalve%7D+%3A%3A+%5B%5Cmathit%7BInteger%7D%5D+%5Crightarrow+%28%5B%5Cmathit%7BInteger%7D%5D%2C%5B%5Cmathit%7BInteger%7D%5D%29%7D+%5C%5C+%5Cmathit%7Bhalve%7D%5C%3B%5B%5C%2C%5D+%26+%3D+%28+%5B%5C%2C%5D%2C+%5B%5C%2C%5D%29+%5C%5C+%5Cmathit%7Bhalve%7D%5C%3B%5Ba%5D+%26+%3D+%28+%5Ba%5D%2C+%5B%5C%2C%5D%29+%5C%5C+%5Cmathit%7Bhalve%7D%5C%3B%28a%3Ab%3Ax%29+%26+%3D+%28a%3Ay%2Cb%3Az%29+%5C%3B%5Cmathbf%7Bwhere%7D%5C%3B+%28y%2Cz%29+%3D+%5Cmathit%7Bhalve%7D%5C%3Bx+%5Cmedskip+%5C%5C+%5Cmulticolumn%7B2%7D%7B%40%7B%7Dl%7D%7B%5Cmathit%7BmergeAll%7D+%3A%3A+%5Cmathit%7BTree%7D+%5Crightarrow+%5B%5Cmathit%7BInteger%7D%5D%7D+%5C%5C+%5Cmathit%7BmergeAll%7D%5C%3B%5Cmathit%7BEmpty%7D+%26+%3D+%5B%5C%2C%5D+%5C%5C+%5Cmathit%7BmergeAll%7D%5C%3B%28%5Cmathit%7BNode%7D%5C%3Bt%5C%3Ba%5C%3Bu%29+%26%3D+%5Cmathit%7Bmerge%7D%5C%3B%28%5Cmathit%7BmergeAll%7D%5C%3Bt%29%5C%3B%28%5Cmathit%7Bmerge%7D%5C%3B%5Ba%5D%5C%3B%28%5Cmathit%7BmergeAll%7D%5C%3Bu%29%29+%5Cmedskip+%5C%5C+%5Cmulticolumn%7B2%7D%7B%40%7B%7Dl%7D%7B%5Cmathit%7Bmerge%7D+%3A%3A+%28%5B%5Cmathit%7BInteger%7D%5D%2C%5B%5Cmathit%7BInteger%7D%5D%29+%5Crightarrow+%5B%5Cmathit%7BInteger%7D%5D%7D+%5C%5C+%5Cmathit%7Bmerge%7D%5C%3B%5B%5C%2C%5D%5C%3By+%26+%3D+y+%5C%5C+%5Cmathit%7Bmerge%7D%5C%3Bx%5C%3B%5B%5C%2C%5D+%26+%3D+x+%5C%5C+%5Cmathit%7Bmerge%7D%5C%3B%28a%3Ax%29%5C%3B%28b%3Ay%29+%26+%3D+%5Cbegin%7Barray%7D%5Bt%5D%7B%40%7B%7Dl%40%7B%7Dl%40%7B%7Dl%7D+%5Cmathbf%7Bif%7D%5C%3B+a+%5Cle+b%5C%3B+%26+%5Cmathbf%7Bthen%7D%5C%3B+%26+a+%3A+%5Cmathit%7Bmerge%7D%5C%3Bx%5C%3B%28b%3Ay%29+%5C%5C+%26+%5Cmathbf%7Belse%7D%5C%3B+%26+b+%3A+%5Cmathit%7Bmerge%7D%5C%3B%28a%3Ax%29%5C%3By+%5Cend%7Barray%7D+%5Cend%7Barray%7D+&bg=ffffff&fg=000000&s=0 "\displaystyle \begin{array}{@{}ll} \multicolumn{2}{@{}l}{\mathit{mergesort} :: [\mathit{Integer}] \rightarrow [\mathit{Integer}]} \\ \mathit{mergesort} & = \mathit{mergeAll} \cdot \mathit{splitUp} \medskip \\ \multicolumn{2}{@{}l}{\mathit{splitUp} :: [\mathit{Integer}] \rightarrow \mathit{Tree}} \\ \mathit{splitUp}\;[\,] & = \mathit{Empty} \\ \mathit{splitUp}\;(a:x) &= \mathit{Node}\;(\mathit{splitUp}\;y)\;a\;(\mathit{splitUp}\;z) \\ & \qquad \mathbf{where}\; (y,z) = \mathit{halve}\;x \medskip \\ \multicolumn{2}{@{}l}{\mathit{halve} :: [\mathit{Integer}] \rightarrow ([\mathit{Integer}],[\mathit{Integer}])} \\ \mathit{halve}\;[\,] & = ( [\,], [\,]) \\ \mathit{halve}\;[a] & = ( [a], [\,]) \\ \mathit{halve}\;(a:b:x) & = (a:y,b:z) \;\mathbf{where}\; (y,z) = \mathit{halve}\;x \medskip \\ \multicolumn{2}{@{}l}{\mathit{mergeAll} :: \mathit{Tree} \rightarrow [\mathit{Integer}]} \\ \mathit{mergeAll}\;\mathit{Empty} & = [\,] \\ \mathit{mergeAll}\;(\mathit{Node}\;t\;a\;u) &= \mathit{merge}\;(\mathit{mergeAll}\;t)\;(\mathit{merge}\;[a]\;(\mathit{mergeAll}\;u)) \medskip \\ \multicolumn{2}{@{}l}{\mathit{merge} :: ([\mathit{Integer}],[\mathit{Integer}]) \rightarrow [\mathit{Integer}]} \\ \mathit{merge}\;[\,]\;y & = y \\ \mathit{merge}\;x\;[\,] & = x \\ \mathit{merge}\;(a:x)\;(b:y) & = \begin{array}[t]{@{}l@{}l@{}l} \mathbf{if}\; a \le b\; & \mathbf{then}\; & a : \mathit{merge}\;x\;(b:y) \\ & \mathbf{else}\; & b : \mathit{merge}\;(a:x)\;y \end{array} \end{array}")
required in 
. It would be neater to use non-empty externally labelled binary trees, with elements at the leaves and none at the branches: 
<figcaption>Fixpoints of a simple function (from Wikimedia Commons)</figcaption></figure>
