Loading...

If this text doesn't disappear after a few seconds, then Remark is probably broken (it's supposed to load and turn this page into a presentation).

In that case, you can view the raw markdown source of the presentation slides by inspecting the source or by expanding the textarea below.

# State of Concurrency in Rust

<sub>(Press P for slide notes.)</sub>

???

Notes on this presentation:

* Use arrow keys or scroll wheel to move through the slides
* Press ? for a full list of shortcuts
* Created using [Remark](https://github.com/gnab/remark/)

---

# hi!

???

I'm Caspar, like the friendly ghost

---

# You've probably heard...

### "Fearless concurrency with Rust"

???

E.g. https://blog.rust-lang.org/2015/04/10/Fearless-Concurrency.html published a year ago

But why is that?

I hope to give you an intuitive understanding of Rust's concurrency options

---

# Agenda

1. Refresh the basics

* ownership, borrowing

2. Standard library concurrency

* threads, data races, channels, locks, `Send` & `Sync`

3. Third party concurrency

* parallel iterators, futures, streams, tokio

???

# Assumed knowledge

* Basic Rust syntax - how to define variables, functions and structs

* `unsafe {}` exists

* What traits are & roughly how they work

---

# Ownership

* Each variable (pointing at data) is owned by another variable (or stack frame)

* Each owner triggers deallocation of owned resources (e.g. memory, file handles, etc) when its lifetime ends as it goes out of scope

* Ownership forms a directed acyclic graph rooted in `main()`'s stack frame

* We can only mutate data via a variable if that variable is the owner

???

Hard to try Rust without coming across this so let's recap ownership & borrowing very quickly

Need to be on same page to understand what follows

---

# Ownership example

```rust
struct Starscape {
    stars: Vec<Star>,
    cat: NyanCat,
}

struct NyanCat {
    pos: (i32, i32),
    volume: u64, // more bits = louder
}

fn main() {
    let universe: Starscape = /*...*/; // somehow make a Starscape
}
```

Just before `main()` completes, ownership looks like:

For example, the `main()` scope owns the `Starscape` via the `universe` let binding.

???

Notice this has very little to do with stack or heap allocation

e.g. Vec contents are on the heap but doesn't affect ownership thereof

---

# Borrowing

* You can take a reference to memory without taking ownership

* Called "borrowing" (an "immutable borrow" in this case):

```rust
let cat: NyanCat = unimplemented!(); // somehow make a cat

let cat2: &NyanCat = &cat;
```

* We can also take a "mutable borrow":

```rust
let mut cat: NyanCat = unimplemented!(); // somehow make a cat
// `cat` is `mut`, so we can mutate it

let cat2: &mut NyanCat = &mut cat;
// `cat2` is `&mut`, so we can mutate it
```

---

# Borrow checker (borrowck)

Borrowing must always follow 5 simple rules:

1. Can only have one mutable borrow at a time

2. Can have as many immutable borrows as you want

3. Cannot borrow mutatably if there is another active borrow

4. Cannot mutate without ownership or an active mutable borrow

5. A borrow may not outlive the underlying variable's lifetime

If you don't follow these rules, code will not compile.

---

# On to concurrency!

---

# Threads

Native threads, not green threads:

```rust
use std::thread;

for i in 0..10 {
  let tx = tx.clone();
  thread::spawn(|| {
    println!("Hi from thread {}", i)
  });
}
```

Threads can panic; panics are usually detected at thread boundaries (i.e. by code in another thread)

Can also catch panics within a thread but generally only done for FFI (use `Result` enum ordinarily)

???

fairly well known that Rust used to have green threads but ripped them out

green threads requiring a runtime conflicted with Rust's systems programming goal

* https://doc.rust-lang.org/std/panic/fn.catch_unwind.html
* https://doc.rust-lang.org/book/error-handling.html
* https://doc.rust-lang.org/nomicon/unwinding.html
* https://doc.rust-lang.org/nomicon/exception-safety.html
* https://doc.rust-lang.org/nomicon/poisoning.html
* http://huonw.github.io/blog/2016/04/myths-and-legends-about-integer-overflow-in-rust/

---

# Data races

With concurrency comes data races...

### Modern languages generally solve by either:

* no *shared* memory: force threads to send messages to each other

* no *mutation* of memory: force threads to take copies of data

???

Channel based approachs like Go and actor systems like Erlang and Akka force threads to send memory to each other

Other languages like Haskell, Scala force your data to be immutable and copied

---

# Rust says: "no *shared mutation* of memory"

Borrowing rules we just covered statically guarantee at compile time that **no thread can mutate data while another thread can access that same data**.

For example,

1. Allocate a `Vec` owned by the current stack frame

2. Set up 2 threads to write to the `Vec` (each runs a closure, as seen earlier)

3. Each thread's closure will try to mutably borrow* the `Vec`

4. => Run afoul of the "1 mutable borrow at a time" rule (compile time error)

<sub>
(* We could force the closure to take ownership with `move|| { ... }` syntax but then we run into the "all data must have 1 owner" rule so it's still sound.)
</sub>

???

Rust identifies that neither mutation nor sharing is inherently bad and each approach has its upsides and downsides, so allows using them.. but not together.

Put another way: shared memory XOR mutation of aforementioned memory.

More: https://doc.rust-lang.org/beta/nomicon/races.html

## Aside - move closure

A `move` closure means any referenced variables should have their ownership moved if they can't be copied (rather than take a reference).

But Rust won't let you use that approach to subvert anything either.

---

# What about data structure thread safety?

Introducing Send & Sync.

???

Sometimes you have data structures that aren't even safe to share across threads, even if the sharing happens very separately from mutation.

---

# The `Send` trait

### Q: Can I send a variable between threads?

* A: Only if it implements the **`Send`** trait

* ... which is almost all primitive types and types composed thereof

* example non-`Send` type: thread that creates an SDL Window must be the thread that polls for events, so an SDL Window can't be passed across threads

---

# The `Sync` trait

### Q: Can I access a shared variable from another thread?

A: Only if it implements the **`Sync`** trait, which is any type `T` where `&T` is `Send` (safe to pass between threads without undefined behaviour)

* intuitively, a type is `Sync` if it uses locks and atomics to avoid undefined behaviour or data races

* example non-`Sync` type: the `Rc` smart pointer (which implements reference counting) does not use atomics to count references

* You would use an `Arc` (Atomic Rc) to share a reference count between threads.

???

Only structs which are so-called `Send` can be moved to another thread (i.e. another thread can take ownership). Useful when you're wrapping a library which itself isn't thread safe.

And only structs which implement `Sync` can be accessed from another thread.

Compiler auto implements these traits for all eligible structs, which is most structs.

And if you know better than the compiler, you can opt a struct in or out (requires unsafe code, because it's an escape hatch)

* https://doc.rust-lang.org/book/concurrency.html
* https://doc.rust-lang.org/std/marker/trait.Send.html
* https://doc.rust-lang.org/std/marker/trait.Sync.html
* https://doc.rust-lang.org/nomicon/send-and-sync.html
* https://doc.rust-lang.org/std/sync/atomic/
* http://huonw.github.io/blog/2015/02/some-notes-on-send-and-sync/ (just before 1.0 release)

---

The type system will save you!
]

And if you can't make the borrow checker happy,

then fall back to using message passing and locks*.

<sub>(Or use `Cell`/`RefCell` or atomics, which we don't have time to cover)</sub>

---

# Message passing via channels

```rust
use std::thread;
use std::sync::mpsc::channel;

// Create a shared channel that can be sent along from many threads
// where tx is the sending half (tx for transmission), and rx is the receiving
// half (rx for receiving).
let (tx, rx) = channel();
for i in 0..10 {
  let tx = tx.clone();
  thread::spawn(move|| { // spawn a thread then send the thread number
    tx.send(i).unwrap(); // returns a Result; unwrap() is quick & dirty handling
  }); // spawn() returns a handle we could (but don't) use to check for panic
}

for _ in 0..10 { // wait for all threads to send their numbers
  let j = rx.recv().unwrap(); // again, dirty dirty error handling
  assert!(0 <= j && j < 10);
}
```

Since channels are not language built-ins, we also have:

* `chan` crate which provides multi-producer multi-consumer channels
* `chan-signal` for *nix signal handling

???

As an aside, we could be checking whether or not each thread panics here, but we don't because I have limited space

`chan-signal` is super useful because normally you can't do much while handling signals

* https://static.rust-lang.org/doc/master/std/sync/mpsc/index.html
* https://github.com/BurntSushi/chan
* https://github.com/BurntSushi/chan-signal

---

# "Impossible to use incorrectly" locks

```rust
// (simplification of stdlib mutex - stdlib is more ergonomic)

// create a new mutex
fn mutex<T: Send>(t: T) -> Mutex<T>;

// acquire the lock
fn lock<T: Send>(mutex: &Mutex<T>) -> MutexGuard<T>;

// access the data protected by the lock
fn access<T: Send>(guard: &mut MutexGuard<T>) -> &mut T;

// example usage
fn use_lock(mutex: &Mutex<Vec<i32>>) {
  // acquire the lock, taking ownership of a guard;
  // the lock is held for the rest of the scope
  let mut guard = lock(mutex);

// access the data by mutably borrowing the guard
  let vec = access(&mut guard);

// vec has type `&mut Vec<i32>`
  vec.push(3);

// lock automatically released here, when `guard` is destroyed
}
```

???

In most languages it's easy to hold the wrong lock (or not hold a lock when you should be holding one)

Usually only documentation guards against holding no lock or the wrong lock

In Rust, a mutex is conceptually like a container which holds the data that it protects, so it's generic over the type it's protecting

Successfully acquiring the lock gets you a MutexGuard

From the MutexGuard, you can access the data protected by the mutex

But crucially, the resulting T cannot live longer than the MutexGuard (eachg borrow has a lifetime, and the output borrow is assumed to live at most as long as the input-ted data could), and the lock is only released when the guard is destroyed.

So it's not possible to use the lock incorrectly without getting a compiler error, and it's not possible to access the data without holding the lock.

## Mention

* RwLock
* Lock poisoning

More:
* https://blog.rust-lang.org/2015/04/10/Fearless-Concurrency.html
* http://softwareengineering.stackexchange.com/questions/317873/how-does-rust-diverge-from-the-concurrency-facilities-of-c

---

## Locking and message passing is table stakes.

# What else do we have?

???

Let's dive outside the stdlib

---

# Optionally parallel iterators via Rayon crate

Regular stdlib lazy iterator (compiles to similar code as a for loop):

```rust
let dot_product = vec1.iter() // get an iterator over the 1st vector
    .zip(vec2.iter()) // make pairs of elems out of 1st & 2nd vectors
    .map(|(i, j)| i * j) // make pairs of multiply pairs together
    .sum(); // sum element 
```

Magic *optionally* parallel version from Rayon library, with no loss of safety:

```rust
let dot_product = vec1.par_iter() // now the same thing again, but in parallel!
    .zip(vec2.par_iter())
    .map(|(i, j)| i * j) // .par_iter()s above means this may run in parallel
    .sum(); // this may also run in parallel
```

If the contents of the vectors weren't safe to send to another thread (implemented `Send`), Rust would pick up that mistake!

???

Send & Sync traits keep giving

Let's say you have 2 huge vectors and you want to calculate the dot product of them

You could write a for loop for that, but using the standard library's iterator APIs is less error prone

Once you've written that you can pull in the Rayon library and have the same thing run in parallel via a work stealing thread pool

Here this is obviously thread safe, but the really cool is that even if you were mutating some state already on the stack or the heap, the code wouldn't compile if

* closures passed to `map()` touch non-threadsafe data structures (because those structures would be non-Sync)
* OR if the vectors contained data that were not safe to pass between threads (the data was non-Send)

---

# Rayon crate under the hood

`.par_iter()` is a high level wrapper around one basic & simple API:

```rust
// `do_something` and `do_something_else` *may* run in parallel via work stealing
let (result_a, result_b) = join(|| calc_result_a(), || calc_result_b())
```

1. Initial thread in pool puts `calc_result_b` onto a queue

2. Initial thread executes `calc_result_a()`

3. If another thread in the pool is not doing work, it will steal and run `calc_result_b()`

4. Initial thread finishes `calc_result_a()` and checks the queue for remaining work (e.g. `calc_result_b`), and executes it if need be

5. `join()` combines and results all results when the queue is empty

1 year ago, a Rayon-`join`-parallel quick sort saw linear speed up in run time for 1MB arrays & broke even at a 1KB array (may be even better now?)

**Note:** thread pool has 1 thread per core, so do not use for IO bound tasks!

???

Another really neat detail is that this functionality is built on a really simple primitive, which just lets 2 pieces of code optionally run in parallel.

1. Before `calc_result_a()` starts running, `calc_result_b` will be put onto a queue
2. If another thread is free, it will steal `calc_result_b` and work on it
3. If `calc_result_a()` finishes and result b isn't done yet, the same thread will calc result b

Naturally avoids splitting work into too many threads when there's not enough CPU capacity to make use of those threads

Can use this API to implement quick sort in parallel, and rust's type system will actually catch the case where you're operating on overlapping parts of the vector you're sorting

For 1024kb arrays, 4x speedup on 4 core CPU via parallel quicksort, slightly slower than normal for 1k arrays. Most expensive bit of algorithm is putting work onto the queue, but rust's focus on zero cost abstractions keeps this efficient enough.

Possibly faster now because these numbers are from a year ago.

Num of threads in the thread pool is usual equal to num logical cores, so obviously not a good idea to use this for non-CPU bound things.

More info:

* http://smallcultfollowing.com/babysteps/blog/2015/12/18/rayon-data-parallelism-in-rust/
* http://smallcultfollowing.com/babysteps/blog/2016/02/19/parallel-iterators-part-1-foundations/
* http://smallcultfollowing.com/babysteps/blog/2016/02/25/parallel-iterators-part-2-producers/
* http://smallcultfollowing.com/babysteps/blog/2016/11/14/parallel-iterators-part-3-consumers/

---

# Futures crate for non-CPU bound work

* A `Future` (from `futures` crate) is a promise of a value, which may or may not be realised yet

```rust
trait Future {
    type Item; // we get this on successful completion
    type Error; // we get this if the future couldn't complete

// The only required method, which attempts to complete the future.
    fn poll(&mut self) -> Poll<Self::Item, Self::Error>;

// Wait for this future to complete (blocks current thread)
    fn wait(self) -> Result<Self::Item, Self::Error> { ... }

// Transforms the result of the future using the given closure.
    fn map<F, U>(self, f: F) -> Map<Self, F>
        where F: FnOnce(Self::Item) -> U { ... }

// ... and many, many more provided methods
}
```

* Kind of like the old green threads (except `Future`s don't have separate stacks) - feels like NodeJS's promises, Scala's futures, etc

???

only half a year or so old, based on `mio` which was too low level for most people

futures have a successful return type and an error

implementors building futures only need to implement the `poll()` function, and typically you wouldn't be implementing that as a library user

as a user, you can call `wait` to turn an async computation into a sync computation

and has combinators like map so you can queue other computations to happen, etc

https://github.com/alexcrichton/futures-rs
https://aturon.github.io/blog/2016/08/11/futures/
https://aturon.github.io/blog/2016/09/07/futures-design/
https://tokio.rs/docs/getting-started/futures/
https://tokio.rs/docs/going-deeper/futures-model/
https://tokio.rs/docs/going-deeper/tasks/

---

# Using futures to time out a computation

```rust
fn check_primality_or_time_out() {
    let pool = CpuPool::new_num_cpus();
    let timer = Timer::default();

// a future that resolves to Err after a timeout
    let timeout = timer.sleep(Duration::from_millis(750))
        .then(|_| Err(()));

// a future that resolves to Ok with the primality result
    let prime = pool.spawn_fn(|| {
        Ok(is_prime(BIG_PRIME)) // prime check which might take a while
    });

// a future that resolves to one of the above values -- whichever
    // completes first!
    let winner = timeout.select(prime).map(|(win, _)| win);

// now block until we have a winner, then print what happened
    match winner.wait() {
        Ok(true) => println!("Prime"),
        Ok(false) => println!("Not prime"),
        Err(_) => println!("Timed out"),
    }
}
```

???

Example stolen from https://tokio.rs/docs/getting-started/futures/

---

# Futures under the hood

Conceptually, a magical event loop exists under the covers that checks whether the next stage of a future is ready, and runs your code if it is.

In more detail, futures are turned into "tasks" and run by "executors" (single thread, thread pool, etc).

In practice, epoll/kqueue is used via the `mio` crate to allow each `Future` implementation to indicate if it is ready:

```rust
trait Future { // remember this?
    // what's this Poll<> business?
    fn poll(&mut self) -> Poll<Self::Item, Self::Error>;
}

type Poll<T, E> = Result<Async<T>, E>;

enum Async<T> {
    Ready(T), // Represents that a value is immediately ready.
    NotReady, // Represents that a value is not ready yet, but may be later.
}
```

That is, Futures are readiness/pull-based (not completion-based/using callbacks as in e.g. Scala), which gives free cancellation and backpressure.

---

# Sources and Sinks for multiple Futures

A `Source` (also from `futures` crate) is to `Future` like `Iterator` is to `Result`.

Or, roughly, "1 future per HTTP response" is not enough; we want an abstraction which can handle multiple future responses:

```rust
use futures::stream::Stream;
use tokio_core::reactor::Core;
use tokio_core::net::TcpListener;

fn main() {
    let mut core = Core::new().unwrap();
    let address = "0.0.0.0:12345".parse().unwrap();
    let listener = TcpListener::bind(&address, &core.handle()).unwrap();

let connections = listener.incoming();
    let welcomes = connections.and_then(|(socket, _peer_addr)| {
        tokio_core::io::write_all(socket, b"Hello, world!\n")
    });
    let server = welcomes.for_each(|(_socket, _welcome)| {
        Ok(())
    });

core.run(server).unwrap();
}
```

`Sink`s (not shown) let you combine streams into one output future/stream.

???

Haven't shown `Sink`s but they let you combine multiple streams into one output future or stream

https://tokio.rs/docs/getting-started/streams-and-sinks/

---

# Tokio provides async programming helpers

We just saw `tokio-core` provide a `Core`, which is a better way to run `Future`s (a better Executor) and TCP support. It also provides codec/framing, timeout and UDP support.

In terms of underlying technologies...

1. epoll/kqueue/whatever-Windows-uses sits at the bottom

2. `mio` crate uses those to provide low level async IO primitives

3. `futures` crate builds on that to provide futures, streams, sinks, etc

4. `tokio-core` adds basic networking support and a nicer executor to `futures`

5. `tokio-proto` and `tokio-service` are higher levels of abstraction...

???

The name Core comes from reactor core, like nuclear reactor core.

[Reactor is a design pattern](https://en.wikipedia.org/wiki/Reactor_pattern) enabled by an event loop.

Sample from https://docs.rs/tokio-core/0.1.4/tokio_core/

---

# `tokio-proto` and `tokio-service`

`tokio-proto` aims to describe a standard way of implementing network protocols. Provides support for streaming, multiplexing, pipelining, etc.

`tokio-service` aims to provide a type for functions from some *request type* to some *response type*, e.g. HTTP request -> HTTP response:

```rust
impl Service for HelloWorld {
    type Request = http::Request;
    type Response = http::Response;
    type Error = http::Error;
    type Future = Box<Future<Item = Self::Response, Error = http::Error>>;

fn call(&self, req: http::Request) -> Self::Future {
        // Create the HTTP response
        let resp = http::Response::ok()
            .with_body(b"hello world\n");

// Return the response as an immediate future
        futures::finished(resp).boxed()
    }
}
```

Idea is that `Service` trait allows building reusable middleware which can be shared between `Service`s easily... but it's too early to tell if that's so.

???

`tokio-proto` examples are too big to fit onto a slide :)

`tokio-service`'s benefits are still mostly hand wavy because most of the effort has been put in lower down the stack.

for example tokio-service hasn't got full documentation on the tokio.rs website yet

Code example https://docs.rs/tokio-service/0.1.0/tokio_service/trait.Service.html

---

# Some available Tokio-based libraries

From https://tokio.rs/docs/going-deeper/third-party/:

* HTTP client based on libcurl

* TLS support

* Signal handling

* Process management, including their std-in/out/err

* File change notifications via inotify

* Async postgres driver

It's early days still:

* Tokio is not 1.0 yet and its API is unstable

* Hard to return `Future`s from functions without the `impl Trait` language feature being stablised (just like returning `Iterator`s is hard)

???

Had a quick look and they mostly seem to be built on tokio-core (the most mature part?)

---

.center.middle[
## thanks!

# Questions?
]