并发与 async：线程、Send/Sync 和一个 Tokio 服务

1
#include <pthread.h>
2
#include <stdio.h>
3

4
static int counter = 0;
5

6
static void *work(void *arg) {
7
    (void)arg;
8
    for (int i = 0; i < 1000000; i++) counter++;
9
    return NULL;
10
}
11

12
int main(void) {
13
    pthread_t t[8];
14
    for (int i = 0; i < 8; i++) pthread_create(&t[i], NULL, work, NULL);
15
    for (int i = 0; i < 8; i++) pthread_join(t[i], NULL);
16
    printf("counter = %d (expected 8000000)\n", counter);
17
    return 0;
18
}

1
$ cargo build --bin ch14_spawn
2
error[E0373]: closure may outlive the current function, but it borrows `files`, which is owned by the current function
3
 --> src/bin/ch14_spawn.rs:6:32
4
  |
5
6 |     let handle = thread::spawn(|| {
6
  |                                ^^ may outlive borrowed value `files`
7
7 |         println!("checking {} files", files.len());
8
  |                                       ----- `files` is borrowed here
9
  |
10
note: function requires argument type to outlive `'static`
11
 --> src/bin/ch14_spawn.rs:6:18
12
  |
13
6 |       let handle = thread::spawn(|| {
14
  |  __________________^
15
7 | |         println!("checking {} files", files.len());
16
8 | |     });
17
  | |______^
18
help: to force the closure to take ownership of `files` (and any other referenced variables), use the `move` keyword
19
  |
20
6 |     let handle = thread::spawn(move || {
21
  |                                ++++
22

23
For more information about this error, try `rustc --explain E0373`.
24
error: could not compile `rust-lab` (bin "ch14_spawn") due to 1 previous error

1
use std::thread;
2

3
fn main() {
4
    let files = vec!["fib.pebble".to_string(), "hello.pebble".to_string()];
5

6
    let lengths: Vec<usize> = thread::scope(|scope| {
7
        let handles: Vec<_> = files
8
            .iter()
9
            .map(|name| scope.spawn(|| name.len()))
10
            .collect();
11
        handles.into_iter().map(|h| h.join().unwrap()).collect()
12
    });
13

14
    println!("{lengths:?}");
15
    println!("still own {} files", files.len());
16
}

1
$ cargo build --bin ch14_race
2
error[E0499]: cannot borrow `counter` as mutable more than once at a time
3
  --> src/bin/ch14_race.rs:8:25
4
   |
5
 6 |       thread::scope(|scope| {
6
   |                      ----- has type `&'1 Scope<'1, '_>`
7
 7 |           for _ in 0..2 {
8
 8 |               scope.spawn(|| {
9
   |               -           ^^ `counter` was mutably borrowed here in the previous iteration of the loop
10
   |  _____________|
11
   | |
12
 9 | |                 for _ in 0..1000 {
13
10 | |                     counter += 1;
14
   | |                     ------- borrows occur due to use of `counter` in closure
15
11 | |                 }
16
12 | |             });
17
   | |______________- argument requires that `counter` is borrowed for `'1`
18
   |
19
note: requirement that the value outlives `'1` introduced here
20
  --> /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/std/src/thread/scoped.rs:203:34
21

22
For more information about this error, try `rustc --explain E0499`.
23
error: could not compile `rust-lab` (bin "ch14_race") due to 1 previous error

1
$ cargo build --bin ch14_rc_send
2
error[E0277]: `Rc<i32>` cannot be sent between threads safely
3
  --> src/bin/ch14_rc_send.rs:8:32
4
   |
5
 8 |       let handle = thread::spawn(move || {
6
   |                    ------------- ^------
7
   |                    |             |
8
   |  __________________|_____________within this `{closure@src/bin/ch14_rc_send.rs:8:32: 8:39}`
9
   | |                  |
10
   | |                  required by a bound introduced by this call
11
 9 | |         println!("in the thread: {clone}");
12
10 | |     });
13
   | |_____^ `Rc<i32>` cannot be sent between threads safely
14
   |
15
   = help: within `{closure@src/bin/ch14_rc_send.rs:8:32: 8:39}`, the trait `Send` is not implemented for `Rc<i32>`
16
note: required because it's used within this closure
17
  --> src/bin/ch14_rc_send.rs:8:32
18
   |
19
 8 |       let handle = thread::spawn(move || {
20
   |                                ^^^^^^^
21
note: required by a bound in `spawn`
22
  --> /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/std/src/thread/functions.rs:125:0
23

24
For more information about this error, try `rustc --explain E0277`.
25
error: could not compile `rust-lab` (bin "ch14_rc_send") due to 1 previous error

1
$ cargo build --bin ch14_cell_sync
2
error[E0277]: `Cell<u32>` cannot be shared between threads safely
3
 --> src/bin/ch14_cell_sync.rs:8:21
4
  |
5
8 |         scope.spawn(|| hits.set(hits.get() + 1));
6
  |               ----- ^^^^^^^^^^^^^^^^^^^^^^^^^^^ `Cell<u32>` cannot be shared between threads safely
7
  |               |
8
  |               required by a bound introduced by this call
9
  |
10
  = help: the trait `Sync` is not implemented for `Cell<u32>`
11
  = note: if you want to do aliasing and mutation between multiple threads, use `std::sync::RwLock` or `std::sync::atomic::AtomicU32` instead
12
  = note: required for `&Cell<u32>` to implement `Send`
13
note: required because it's used within this closure
14
 --> src/bin/ch14_cell_sync.rs:8:21
15
  |
16
8 |         scope.spawn(|| hits.set(hits.get() + 1));
17
  |                     ^^
18
note: required by a bound in `Scope::<'scope, 'env>::spawn`
19
 --> /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/std/src/thread/scoped.rs:201:4
20

21
For more information about this error, try `rustc --explain E0277`.
22
error: could not compile `rust-lab` (bin "ch14_cell_sync") due to 1 previous error

1
$ cargo build --bin ch14_send_sync_table
2
error[E0277]: `Rc<u32>` cannot be sent between threads safely
3
  --> src/bin/ch14_send_sync_table.rs:15:15
4
   |
5
15 |     is_send::<Rc<u32>>();
6
   |               ^^^^^^^ `Rc<u32>` cannot be sent between threads safely
7
   |
8
   = help: the trait `Send` is not implemented for `Rc<u32>`
9
note: required by a bound in `is_send`
10
  --> src/bin/ch14_send_sync_table.rs:5:15
11
   |
12
 5 | fn is_send<T: Send>() {}
13
   |               ^^^^ required by this bound in `is_send`
14

15
error[E0277]: `std::sync::MutexGuard<'static, u32>` cannot be sent between threads safely
16
  --> src/bin/ch14_send_sync_table.rs:16:15
17
   |
18
16 |     is_send::<MutexGuard<'static, u32>>();
19
   |               ^^^^^^^^^^^^^^^^^^^^^^^^ `std::sync::MutexGuard<'static, u32>` cannot be sent between threads safely
20
   |
21
   = help: the trait `Send` is not implemented for `std::sync::MutexGuard<'static, u32>`
22
note: required by a bound in `is_send`
23
  --> src/bin/ch14_send_sync_table.rs:5:15
24
   |
25
 5 | fn is_send<T: Send>() {}
26
   |               ^^^^ required by this bound in `is_send`
27

28
For more information about this error, try `rustc --explain E0277`.
29
error: could not compile `rust-lab` (bin "ch14_send_sync_table") due to 2 previous errors

1
let counter = Arc::new(Mutex::new(0u32));
2

3
let handles: Vec<_> = (0..8)
4
    .map(|_| {
5
        let counter = Arc::clone(&counter);
6
        thread::spawn(move || {
7
            for _ in 0..1000 {
8
                *counter.lock().unwrap() += 1;
9
            }
10
        })
11
    })
12
    .collect();
13

14
for handle in handles {
15
    handle.join().unwrap();
16
}
17

18
println!("count = {}", *counter.lock().unwrap());
19
println!("strong count = {}", Arc::strong_count(&counter));

1
$ cargo build --bin ch14_mutex_forgot_lock
2
error[E0308]: mismatched types
3
 --> src/bin/ch14_mutex_forgot_lock.rs:5:22
4
  |
5
5 |     let guard: u32 = counter.lock().unwrap();
6
  |                ---   ^^^^^^^^^^^^^^^^^^^^^^^ expected `u32`, found `MutexGuard<'_, u32>`
7
  |                |
8
  |                expected due to this
9
  |
10
  = note: expected type `u32`
11
           found struct `std::sync::MutexGuard<'_, u32>`
12
help: consider dereferencing the type
13
  |
14
5 |     let guard: u32 = *counter.lock().unwrap();
15
  |                      +
16

17
For more information about this error, try `rustc --explain E0308`.
18
error: could not compile `rust-lab` (bin "ch14_mutex_forgot_lock") due to 1 previous error

1
let data = Arc::new(Mutex::new(vec![1, 2, 3]));
2

3
let worker = {
4
    let data = Arc::clone(&data);
5
    thread::spawn(move || {
6
        let mut guard = data.lock().unwrap();
7
        guard.push(4);
8
        panic!("worker died while holding the lock");
9
    })
10
};
11
let _ = worker.join();
12

13
println!("is_poisoned = {}", data.is_poisoned());
14
match data.lock() {
15
    Ok(guard) => println!("clean: {guard:?}"),
16
    Err(poisoned) => {
17
        let guard = poisoned.into_inner();
18
        println!("recovered: {guard:?}");
19
    }
20
}

1
let (sender, receiver) = mpsc::channel::<String>();
2

3
let workers: Vec<_> = ["fib", "hello", "errors"]
4
    .into_iter()
5
    .map(|name| {
6
        let sender = sender.clone();
7
        thread::spawn(move || {
8
            sender.send(format!("checked {name}.pebble")).unwrap();
9
        })
10
    })
11
    .collect();
12
drop(sender);
13

14
let mut reports: Vec<String> = receiver.iter().collect();
15
reports.sort();
16
for report in &reports {
17
    println!("{report}");
18
}
19

20
for worker in workers {
21
    worker.join().unwrap();
22
}
23
println!("{} reports, channel closed", reports.len());

1
error[E0277]: `Rc<FunctionProto>` cannot be sent between threads safely
2
   --> crates/pebble-cli/src/commands.rs:107:31
3
    |
4
107 |             .map(|path| scope.spawn(|| check_file(path)))
5
    |                               ^^^^^ `Rc<FunctionProto>` cannot be sent between threads safely
6
    |
7
    = help: within `Result<CheckReport, anyhow::Error>`, the trait `Send` is not implemented for `Rc<FunctionProto>`
8
note: required because it appears within the type `Compiled`
9
   --> crates/pebble-bytecode/src/lib.rs:371:12
10
    |
11
371 | pub struct Compiled {
12
    |            ^^^^^^^^
13
note: required because it appears within the type `CheckReport`
14
   --> crates/pebble-cli/src/commands.rs:90:8
15
    |
16
 90 | struct CheckReport {
17
    |        ^^^^^^^^^^^
18
note: required because it appears within the type `Result<CheckReport, anyhow::Error>`
19
   --> /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/core/src/result.rs:557:9
20
note: required by a bound in `Scope::<'scope, 'env>::spawn`
21
   --> /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/std/src/thread/scoped.rs:201:4

1
error[E0277]: `Rc<FunctionProto>` cannot be sent between threads safely
2
  = help: within `Compiled`, the trait `Send` is not implemented for `Rc<FunctionProto>`
3
error[E0277]: `Rc<RefCell<Env>>` cannot be sent between threads safely
4
  = help: within `Interpreter`, the trait `Send` is not implemented for `Rc<RefCell<Env>>`
5
error[E0277]: `(dyn pebble_value::Output + 'static)` cannot be sent between threads safely
6
  = help: the trait `Send` is not implemented for `(dyn pebble_value::Output + 'static)`
7
error[E0277]: `Rc<str>` cannot be sent between threads safely
8
  = help: within `pebble_value::Value`, the trait `Send` is not implemented for `Rc<str>`
9
error[E0277]: `Rc<RefCell<Vec<pebble_value::Value>>>` cannot be sent between threads safely
10
  = help: within `pebble_value::Value`, the trait `Send` is not implemented for `Rc<RefCell<Vec<pebble_value::Value>>>`
11
error[E0277]: `Rc<RefCell<HashMap<MapKey, pebble_value::Value>>>` cannot be sent between threads safely
12
  = help: within `pebble_value::Value`, the trait `Send` is not implemented for `Rc<RefCell<HashMap<MapKey, pebble_value::Value>>>`
13
error[E0277]: `Rc<pebble_value::Function>` cannot be sent between threads safely
14
  = help: within `pebble_value::Value`, the trait `Send` is not implemented for `Rc<pebble_value::Function>`
15
error: could not compile `pebble-cli` (test "send_probe") due to 7 previous errors

1
/// `pebble check`: lex, parse, and compile every file, each on its own thread,
2
/// then print the reports in command-line order.
3
///
4
/// # Errors
5
/// Returns the first host-level failure (such as an unreadable file), in
6
/// command-line order.
7
pub(crate) fn check(args: &CheckArgs) -> Result<u8> {
8
    let reports: Vec<Result<CheckReport>> = thread::scope(|scope| {
9
        let handles: Vec<_> = args
10
            .files
11
            .iter()
12
            .map(|path| scope.spawn(|| check_file(path)))
13
            .collect();
14
        handles
15
            .into_iter()
16
            .map(|handle| handle.join().expect("a check thread panicked"))
17
            .collect()
18
    });
19

20
    let mut failed = false;
21
    for report in reports {
22
        let report = report?;
23
        eprint!("{}", report.rendered);
24
        failed |= report.failed;
25
    }
26
    Ok(u8::from(failed))
27
}

1
#[test]
2
fn check_reports_only_the_broken_file() {
3
    let output = pebble_output(&[
4
        "check",
5
        "examples/fib.pebble",
6
        "examples/errors.pebble",
7
        "examples/hello.pebble",
8
    ]);
9
    assert_eq!(output.status.code(), Some(1));
10
    assert!(output.stdout.is_empty(), "check prints nothing to stdout");
11

12
    let stderr = String::from_utf8(output.stderr).expect("stderr is UTF-8");
13
    assert!(stderr.contains("examples/errors.pebble"), "{stderr}");
14
    assert!(!stderr.contains("examples/fib.pebble"), "{stderr}");
15
}
16

17
#[test]
18
fn check_requires_a_file() {
19
    pebble().arg("check").assert().code(2);
20
}

1
use std::time::{Duration, Instant};
2
use tokio::time::sleep;
3

4
async fn check(name: &str, millis: u64) -> String {
5
    sleep(Duration::from_millis(millis)).await;
6
    format!("checked {name}")
7
}
8

9
#[tokio::main]
10
async fn main() {
11
    let start = Instant::now();
12
    let a = check("fib", 200).await;
13
    let b = check("hello", 200).await;
14
    println!("{a}, {b}: sequential took {} ms", start.elapsed().as_millis() / 50 * 50);
15

16
    let start = Instant::now();
17
    let (a, b) = tokio::join!(check("fib", 200), check("hello", 200));
18
    println!("{a}, {b}: join! took {} ms", start.elapsed().as_millis() / 50 * 50);
19
}

1
struct CountDown {
2
    remaining: u32,
3
}
4

5
impl Future for CountDown {
6
    type Output = &'static str;
7

8
    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<&'static str> {
9
        if self.remaining == 0 {
10
            Poll::Ready("liftoff")
11
        } else {
12
            println!("  CountDown::poll, {} left", self.remaining);
13
            self.remaining -= 1;
14
            cx.waker().wake_by_ref();
15
            Poll::Pending
16
        }
17
    }
18
}

1
impl Delay {
2
    fn new(duration: Duration) -> Self {
3
        let shared = Arc::new(Mutex::new(Shared {
4
            done: false,
5
            waker: None,
6
        }));
7
        let thread_shared = Arc::clone(&shared);
8
        thread::spawn(move || {
9
            thread::sleep(duration);
10
            let mut state = thread_shared.lock().unwrap();
11
            state.done = true;
12
            if let Some(waker) = state.waker.take() {
13
                waker.wake();
14
            }
15
        });
16
        Delay { shared }
17
    }
18
}

1
$ cargo build --bin ch14_pin_new
2
error[E0277]: `{async fn body of answer()}` cannot be unpinned
3
  --> src/bin/ch14_pin_new.rs:12:28
4
   |
5
 3 | async fn answer() -> u32 {
6
   |                      --- within this `impl Future<Output = u32>`
7
...
8
12 |     let _pinned = Pin::new(&mut future);
9
   |                   -------- ^^^^^^^^^^^ within `impl Future<Output = u32>`, the trait `Unpin` is not implemented for `{async fn body of answer()}`
10
   |                   |
11
   |                   required by a bound introduced by this call
12
   |
13
   = note: consider using the `pin!` macro
14
           consider using `Box::pin` if you need to access the pinned value outside of the current scope
15
note: required because it appears within the type `impl Future<Output = u32>`
16
  --> src/bin/ch14_pin_new.rs:3:22
17
   |
18
 3 | async fn answer() -> u32 {
19
   |                      ^^^
20
note: required by a bound in `Pin::<Ptr>::new`
21
  --> /rustc/59807616e1fa2540724bfbac14d7976d7e4a3860/library/core/src/pin.rs:1181:4
22

23
For more information about this error, try `rustc --explain E0277`.
24
error: could not compile `rust-lab` (bin "ch14_pin_new") due to 1 previous error