rust/comprehensive-rust
1
0
Fork 0
mirror of https://github.com/google/comprehensive-rust.git synced 2025-07-01 20:54:31 +02:00
Code Issues Releases Activity

Update Concurrency course with times (#2007)

Browse Source 
As I mentioned in #1536:

* Break into segments at approximately the places @fw-immunant put
breaks
 * Move all of the files into `src/concurrency`
 * Add timings and segment/session metadata so course outlines appear

There's room for more work here, including some additional feedback from
@fw-immunant after the session I observed, but let's do one step at a
time :)
This commit is contained in:
Dustin J. Mitchell
2024-04-23 09:26:41 -04:00
committed by GitHub
parent a03b7e68e5
commit face5af783
58 changed files with 385 additions and 246 deletions
Download Patch File Download Diff File Expand all files Collapse all files

83
src/concurrency/async-pitfalls/async-traits.md Normal file
View File

@ -0,0 +1,83 @@
---
minutes: 5
---
# Async Traits
Async methods in traits are were stabilized only recently, in the 1.75 release.
This required support for using return-position `impl Trait` (RPIT) in traits,
as the desugaring for `async fn` includes `-> impl Future<Output = ...>`.
However, even with the native support today there are some pitfalls around
`async fn` and RPIT in traits:
- Return-position impl Trait captures all in-scope lifetimes (so some patterns
of borrowing cannot be expressed)
- Traits whose methods use return-position `impl trait` or `async` are not `dyn`
compatible.
If we do need `dyn` support, the crate
[async_trait](https://docs.rs/async-trait/latest/async_trait/) provides a
workaround through a macro, with some caveats:
```rust,editable,compile_fail
use async_trait::async_trait;
use std::time::Instant;
use tokio::time::{sleep, Duration};
#[async_trait]
trait Sleeper {
async fn sleep(&self);
}
struct FixedSleeper {
sleep_ms: u64,
}
#[async_trait]
impl Sleeper for FixedSleeper {
async fn sleep(&self) {
sleep(Duration::from_millis(self.sleep_ms)).await;
}
}
async fn run_all_sleepers_multiple_times(
sleepers: Vec<Box<dyn Sleeper>>,
n_times: usize,
) {
for _ in 0..n_times {
println!("running all sleepers..");
for sleeper in &sleepers {
let start = Instant::now();
sleeper.sleep().await;
println!("slept for {}ms", start.elapsed().as_millis());
}
}
}
#[tokio::main]
async fn main() {
let sleepers: Vec<Box<dyn Sleeper>> = vec![
Box::new(FixedSleeper { sleep_ms: 50 }),
Box::new(FixedSleeper { sleep_ms: 100 }),
];
run_all_sleepers_multiple_times(sleepers, 5).await;
}
```
<details>
- `async_trait` is easy to use, but note that it's using heap allocations to
achieve this. This heap allocation has performance overhead.
- The challenges in language support for `async trait` are deep Rust and
probably not worth describing in-depth. Niko Matsakis did a good job of
explaining them in
[this post](https://smallcultfollowing.com/babysteps/blog/2019/10/26/async-fn-in-traits-are-hard/)
if you are interested in digging deeper.
- Try creating a new sleeper struct that will sleep for a random amount of time
and adding it to the Vec.
</details>

54
src/concurrency/async-pitfalls/blocking-executor.md Normal file
View File

@ -0,0 +1,54 @@
---
minutes: 10
---
# Blocking the executor
Most async runtimes only allow IO tasks to run concurrently. This means that CPU
blocking tasks will block the executor and prevent other tasks from being
executed. An easy workaround is to use async equivalent methods where possible.
```rust,editable,compile_fail
use futures::future::join_all;
use std::time::Instant;
async fn sleep_ms(start: &Instant, id: u64, duration_ms: u64) {
std::thread::sleep(std::time::Duration::from_millis(duration_ms));
println!(
"future {id} slept for {duration_ms}ms, finished after {}ms",
start.elapsed().as_millis()
);
}
#[tokio::main(flavor = "current_thread")]
async fn main() {
let start = Instant::now();
let sleep_futures = (1..=10).map(|t| sleep_ms(&start, t, t * 10));
join_all(sleep_futures).await;
}
```
<details>
- Run the code and see that the sleeps happen consecutively rather than
concurrently.
- The `"current_thread"` flavor puts all tasks on a single thread. This makes
the effect more obvious, but the bug is still present in the multi-threaded
flavor.
- Switch the `std::thread::sleep` to `tokio::time::sleep` and await its result.
- Another fix would be to `tokio::task::spawn_blocking` which spawns an actual
thread and transforms its handle into a future without blocking the executor.
- You should not think of tasks as OS threads. They do not map 1 to 1 and most
executors will allow many tasks to run on a single OS thread. This is
particularly problematic when interacting with other libraries via FFI, where
that library might depend on thread-local storage or map to specific OS
threads (e.g., CUDA). Prefer `tokio::task::spawn_blocking` in such situations.
- Use sync mutexes with care. Holding a mutex over an `.await` may cause another
task to block, and that task may be running on the same thread.
</details>

121
src/concurrency/async-pitfalls/cancellation.md Normal file
View File

@ -0,0 +1,121 @@
---
minutes: 18
---
# Cancellation
Dropping a future implies it can never be polled again. This is called
_cancellation_ and it can occur at any `await` point. Care is needed to ensure
the system works correctly even when futures are cancelled. For example, it
shouldn't deadlock or lose data.
```rust,editable,compile_fail
use std::io::{self, ErrorKind};
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt, DuplexStream};
struct LinesReader {
stream: DuplexStream,
}
impl LinesReader {
fn new(stream: DuplexStream) -> Self {
Self { stream }
}
async fn next(&mut self) -> io::Result<Option<String>> {
let mut bytes = Vec::new();
let mut buf = [0];
while self.stream.read(&mut buf[..]).await? != 0 {
bytes.push(buf[0]);
if buf[0] == b'\n' {
break;
}
}
if bytes.is_empty() {
return Ok(None);
}
let s = String::from_utf8(bytes)
.map_err(|_| io::Error::new(ErrorKind::InvalidData, "not UTF-8"))?;
Ok(Some(s))
}
}
async fn slow_copy(source: String, mut dest: DuplexStream) -> std::io::Result<()> {
for b in source.bytes() {
dest.write_u8(b).await?;
tokio::time::sleep(Duration::from_millis(10)).await
}
Ok(())
}
#[tokio::main]
async fn main() -> std::io::Result<()> {
let (client, server) = tokio::io::duplex(5);
let handle = tokio::spawn(slow_copy("hi\nthere\n".to_owned(), client));
let mut lines = LinesReader::new(server);
let mut interval = tokio::time::interval(Duration::from_millis(60));
loop {
tokio::select! {
_ = interval.tick() => println!("tick!"),
line = lines.next() => if let Some(l) = line? {
print!("{}", l)
} else {
break
},
}
}
handle.await.unwrap()?;
Ok(())
}
```
<details>
- The compiler doesn't help with cancellation-safety. You need to read API
documentation and consider what state your `async fn` holds.
- Unlike `panic` and `?`, cancellation is part of normal control flow (vs
error-handling).
- The example loses parts of the string.
- Whenever the `tick()` branch finishes first, `next()` and its `buf` are
dropped.
- `LinesReader` can be made cancellation-safe by making `buf` part of the
struct:
```rust,compile_fail
struct LinesReader {
stream: DuplexStream,
bytes: Vec<u8>,
buf: [u8; 1],
}
impl LinesReader {
fn new(stream: DuplexStream) -> Self {
Self { stream, bytes: Vec::new(), buf: [0] }
}
async fn next(&mut self) -> io::Result<Option<String>> {
// prefix buf and bytes with self.
// ...
let raw = std::mem::take(&mut self.bytes);
let s = String::from_utf8(raw)
// ...
}
}
```
- [`Interval::tick`](https://docs.rs/tokio/latest/tokio/time/struct.Interval.html#method.tick)
is cancellation-safe because it keeps track of whether a tick has been
'delivered'.
- [`AsyncReadExt::read`](https://docs.rs/tokio/latest/tokio/io/trait.AsyncReadExt.html#method.read)
is cancellation-safe because it either returns or doesn't read data.
- [`AsyncBufReadExt::read_line`](https://docs.rs/tokio/latest/tokio/io/trait.AsyncBufReadExt.html#method.read_line)
is similar to the example and _isn't_ cancellation-safe. See its documentation
for details and alternatives.
</details>

132
src/concurrency/async-pitfalls/pin.md Normal file
View File

@ -0,0 +1,132 @@
---
minutes: 20
---
# `Pin`
Async blocks and functions return types implementing the `Future` trait. The
type returned is the result of a compiler transformation which turns local
variables into data stored inside the future.
Some of those variables can hold pointers to other local variables. Because of
that, the future should never be moved to a different memory location, as it
would invalidate those pointers.
To prevent moving the future type in memory, it can only be polled through a
pinned pointer. `Pin` is a wrapper around a reference that disallows all
operations that would move the instance it points to into a different memory
location.
```rust,editable,compile_fail
use tokio::sync::{mpsc, oneshot};
use tokio::task::spawn;
use tokio::time::{sleep, Duration};
// A work item. In this case, just sleep for the given time and respond
// with a message on the `respond_on` channel.
#[derive(Debug)]
struct Work {
input: u32,
respond_on: oneshot::Sender<u32>,
}
// A worker which listens for work on a queue and performs it.
async fn worker(mut work_queue: mpsc::Receiver<Work>) {
let mut iterations = 0;
loop {
tokio::select! {
Some(work) = work_queue.recv() => {
sleep(Duration::from_millis(10)).await; // Pretend to work.
work.respond_on
.send(work.input * 1000)
.expect("failed to send response");
iterations += 1;
}
// TODO: report number of iterations every 100ms
}
}
}
// A requester which requests work and waits for it to complete.
async fn do_work(work_queue: &mpsc::Sender<Work>, input: u32) -> u32 {
let (tx, rx) = oneshot::channel();
work_queue
.send(Work { input, respond_on: tx })
.await
.expect("failed to send on work queue");
rx.await.expect("failed waiting for response")
}
#[tokio::main]
async fn main() {
let (tx, rx) = mpsc::channel(10);
spawn(worker(rx));
for i in 0..100 {
let resp = do_work(&tx, i).await;
println!("work result for iteration {i}: {resp}");
}
}
```
<details>
- You may recognize this as an example of the actor pattern. Actors typically
call `select!` in a loop.
- This serves as a summation of a few of the previous lessons, so take your time
with it.
- Naively add a `_ = sleep(Duration::from_millis(100)) => { println!(..) }` to
the `select!`. This will never execute. Why?
- Instead, add a `timeout_fut` containing that future outside of the `loop`:
```rust,compile_fail
let mut timeout_fut = sleep(Duration::from_millis(100));
loop {
select! {
..,
_ = timeout_fut => { println!(..); },
}
}
```
- This still doesn't work. Follow the compiler errors, adding `&mut` to the
`timeout_fut` in the `select!` to work around the move, then using
`Box::pin`:
```rust,compile_fail
let mut timeout_fut = Box::pin(sleep(Duration::from_millis(100)));
loop {
select! {
..,
_ = &mut timeout_fut => { println!(..); },
}
}
```
- This compiles, but once the timeout expires it is `Poll::Ready` on every
iteration (a fused future would help with this). Update to reset
`timeout_fut` every time it expires.
- Box allocates on the heap. In some cases, `std::pin::pin!` (only recently
stabilized, with older code often using `tokio::pin!`) is also an option, but
that is difficult to use for a future that is reassigned.
- Another alternative is to not use `pin` at all but spawn another task that
will send to a `oneshot` channel every 100ms.
- Data that contains pointers to itself is called self-referential. Normally,
the Rust borrow checker would prevent self-referential data from being moved,
as the references cannot outlive the data they point to. However, the code
transformation for async blocks and functions is not verified by the borrow
checker.
- `Pin` is a wrapper around a reference. An object cannot be moved from its
place using a pinned pointer. However, it can still be moved through an
unpinned pointer.
- The `poll` method of the `Future` trait uses `Pin<&mut Self>` instead of
`&mut Self` to refer to the instance. That's why it can only be called on a
pinned pointer.
</details>