07-并发与异步安全

data race Rust Send + Sync trait Send = Sync = &T

• ****deadlock Mutex
• ****livelock
• ****race condition——Rust

tokio::spawn Future Send task Send Future LocalSet

cancellation safetytokio::select! Future tokio::sync::oneshot async fn Result

Mutex<T> Arc RwLock<T> writer starvation

:

---

Send + Sync

Trait

// auto trait
//  impl
 
// Send:  T 
pub unsafe auto trait Send {}
 
// Sync:  &T 
// : T  Send  &T  Send 
pub unsafe auto trait Sync {}

Send/Sync

	Send	Sync
i32, String, Vec<T>
Arc<T>		T: Sync
Mutex<T>
RwLock<T>
Cell<T>, RefCell<T>
Rc<T>
*const T, *mut T
dyn Future			Send
tokio::task::JoinHandle<T>

use std::rc::Rc;
use std::sync::Arc;
 
struct MyStruct {
    data: Vec<i32>,           // Send + Sync 
    pointer: *const u8,       //   MyStruct  !Send + !Sync
    cache: Rc<String>,        //  
}
 
//  !Send !Send
// MyStruct: !Send, !Sync
 
// unsafe— 
unsafe impl Send for MyStruct {}   //  unsafe
unsafe impl Sync for MyStruct {}   //  unsafe
 
//  Send  Send
struct SendWrapper<T>(T);
unsafe impl<T> Send for SendWrapper<T> {}
//  

Send

use tokio::task;
 
#[tokio::main]
async fn main() {
    // tokio::spawn  Future: Send
    let handle = task::spawn(async {
        //  async block  Send 
        let data = vec![1, 2, 3];
        task::spawn_in_place(|| {
            // 
            println!("{:?}", data);  // data  Send
        }).await.unwrap();
    });
 
    handle.await.unwrap();
 
    //  Send  Future LocalSet
    let local = task::LocalSet::new();
    local.run_until(async {
        let rc = std::rc::Rc::new(42);  // Rc  !Send
        // spawn_local  Send
        task::spawn_local(async move {
            println!("{}", rc);
        }).await.unwrap();
    }).await;
}

---

async

Cancellation Safety

use tokio::sync::mpsc;
use tokio::time::{sleep, Duration};
 
//  select! 
async fn unsafe_read(reader: &mut (impl tokio::io::AsyncRead + Unpin)) {
    use tokio::io::AsyncReadExt;
    let mut buf = [0u8; 1024];
    tokio::select! {
        result = reader.read(&mut buf) => {
            //  ctrl_c read 
            // buf 
            println!("read {} bytes", result.unwrap());
        }
        _ = tokio::signal::ctrl_c() => {
            println!("shutting down");
        }
    }
}
 
//   tokio::io::read_buf 
async fn safe_read(reader: &mut (impl tokio::io::AsyncRead + Unpin)) {
    use tokio::io::AsyncReadExt;
    let mut buf = [0u8; 1024];
    tokio::select! {
        result = tokio::io::read_buf(reader, &mut &mut buf[..]) => {
            println!("read {} bytes", result.unwrap());
        }
        _ = tokio::signal::ctrl_c() => {
            println!("shutting down");
        }
    }
}
 
//  oneshot channel 
async fn safe_with_channel(rx: &mut tokio::sync::oneshot::Receiver<String>) {
    tokio::select! {
        result = rx => {
            match result {
                Ok(data) => println!("received: {}", data),
                Err(_) => println!("channel closed"),
            }
        }
        _ = tokio::signal::ctrl_c() => {
            println!("shutting down");
        }
    }
}

tokio

use tokio::sync::{mpsc, oneshot, broadcast, watch, Semaphore};
use tokio::time::{interval, Duration};
use std::sync::Arc;
 
//  1: mpsc — 
async fn mpsc_example() {
    let (tx, mut rx) = mpsc::channel::<String>(32);
 
    // 
    for i in 0..5 {
        let tx = tx.clone();
        tokio::spawn(async move {
            tx.send(format!("msg {}", i)).await.unwrap();
        });
    }
    drop(tx);  // 
 
    // 
    while let Some(msg) = rx.recv().await {
        println!("received: {}", msg);
    }
}
 
//  2: broadcast — 
async fn broadcast_example() {
    let (tx, _) = broadcast::channel::<String>(16);
 
    // 
    for _ in 0..3 {
        let mut rx = tx.subscribe();
        tokio::spawn(async move {
            while let Ok(msg) = rx.recv().await {
                println!("consumer received: {}", msg);
            }
        });
    }
 
    // 
    tx.send("hello".into()).unwrap();
}
 
//  3: watch — 
async fn watch_example() {
    let (tx, mut rx) = watch::channel("initial".to_string());
 
    tokio::spawn(async move {
        loop {
            tokio::time::sleep(Duration::from_secs(1)).await;
            tx.send(format!("update at {:?}", std::time::Instant::now())).unwrap();
        }
    });
 
    // 
    while rx.changed().await.is_ok() {
        println!("new value: {}", *rx.borrow());
    }
}
 
//  4: Semaphore — 
async fn semaphore_example() {
    let semaphore = Arc::new(Semaphore::new(3));  //  3 
    let mut handles = vec![];
 
    for i in 0..10 {
        let sem = semaphore.clone();
        handles.push(tokio::spawn(async move {
            let _permit = sem.acquire().await.unwrap();
            println!("task {} running", i);
            tokio::time::sleep(Duration::from_secs(1)).await;
        }));
    }
 
    for h in handles {
        h.await.unwrap();
    }
}
 
//  5: spawn_blocking — CPU 
async fn spawn_blocking_example() {
    let result = tokio::task::spawn_blocking(|| {
        // CPU 
        let mut sum = 0u64;
        for i in 0..1_000_000 {
            sum += i;
        }
        sum
    }).await.unwrap();
 
    println!("result: {}", result);
}

---

use std::sync::{Arc, Mutex};
 
struct Account {
    id: u64,
    balance: u64,
}
 
//  
fn bad_transfer(
    from: &Mutex<Account>,
    to: &Mutex<Account>,
    amount: u64,
) {
    //  1: transfer(A, B) →  A B
    //  2: transfer(B, A) →  B A
    // → 
    let mut f = from.lock().unwrap();
    let mut t = to.lock().unwrap();
    f.balance -= amount;
    t.balance += amount;
}
 
//  
fn safe_transfer(
    from: &Mutex<Account>,
    to: &Mutex<Account>,
    amount: u64,
) -> Result<(), String> {
    //  id 
    let (first, second) = if from.lock().unwrap().id < to.lock().unwrap().id {
        (from, to)
    } else {
        (to, from)
    };
 
    let mut f = first.lock().unwrap();
    let mut t = second.lock().unwrap();
 
    if f.balance < amount {
        return Err("insufficient funds".into());
    }
 
    f.balance -= amount;
    t.balance += amount;
    Ok(())
}
 
//   tokio::sync::Mutex  try_lock
async fn try_lock_transfer(
    from: &tokio::sync::Mutex<Account>,
    to: &tokio::sync::Mutex<Account>,
    amount: u64,
) -> Result<(), String> {
    // try_lock 
    let (first, second) = if from.lock().await.id < to.lock().await.id {
        (from, to)
    } else {
        (to, from)
    };
 
    let mut f = first.lock().await;
    let mut t = second.lock().await;
 
    if f.balance < amount {
        return Err("insufficient funds".into());
    }
 
    f.balance -= amount;
    t.balance += amount;
    Ok(())
}

---

Rust

use std::sync::Arc;
use std::thread;
 
//   — 
fn thread_safe_example() {
    let data = Arc::new(vec![1, 2, 3, 4, 5]);
    let mut handles = vec![];
 
    for _ in 0..5 {
        let data = data.clone();  // Arc::clone 
        handles.push(thread::spawn(move || {
            //  — 
            println!("sum: {}", data.iter().sum::<i32>());
        }));
    }
 
    for h in handles {
        h.join().unwrap();
    }
}
 
//   Rust 
fn data_race_impossible() {
    let mut data = vec![1, 2, 3];
    let handle = thread::spawn(move || {
        // data  move 
        data.push(4);
    });
    // println!("{:?}", data);  //  data  move
    handle.join().unwrap();
}

---

use std::sync::atomic::{AtomicUsize, AtomicBool, Ordering};
use std::sync::Arc;
use crossbeam::queue::SegQueue;  // 
use dashmap::DashMap;            //  HashMap
use parking_lot::Mutex;          //  Mutex 
 
//  — 
struct Counter {
    count: AtomicUsize,
}
 
impl Counter {
    fn new() -> Self {
        Self { count: AtomicUsize::new(0) }
    }
 
    fn increment(&self) {
        self.count.fetch_add(1, Ordering::Relaxed);
    }
 
    fn get(&self) -> usize {
        self.count.load(Ordering::Relaxed)
    }
}
 
// DashMap —  HashMap
fn dashmap_example() {
    let map = Arc::new(DashMap::new());
 
    let mut handles = vec![];
    for i in 0..100 {
        let map = map.clone();
        handles.push(std::thread::spawn(move || {
            map.insert(i, i * 2);
        }));
    }
 
    for h in handles {
        h.join().unwrap();
    }
 
    assert_eq!(map.len(), 100);
}
 
// crossbeam 
fn crossbeam_queue_example() {
    let queue = Arc::new(SegQueue::new());
    let mut handles = vec![];
 
    for i in 0..100 {
        let queue = queue.clone();
        handles.push(std::thread::spawn(move || {
            queue.push(i);
        }));
    }
 
    for h in handles {
        h.join().unwrap();
    }
 
    let mut items: Vec<_> = queue.iter().collect();
    items.sort();
    assert_eq!(items, (0..100).collect::<Vec<_>>());
}

---

async drop

Rust Drop trait async ——drop() async fn

use tokio::sync::mpsc;
 
//  1:  async close 
struct AsyncConnection {
    tx: mpsc::Sender<()>,
    handle: tokio::task::JoinHandle<()>,
}
 
impl AsyncConnection {
    async fn close(self) {
        drop(self.tx);       //  channel
        self.handle.await.ok();  // 
    }
}
 
// Drop 
impl Drop for AsyncConnection {
    fn drop(&mut self) {
        //  close()
        //  await
        eprintln!("warning: AsyncConnection dropped without close()");
    }
}
 
//  2: 
async fn process() {
    let conn = AsyncConnection::new().await;
 
    // 
    conn.send("hello").await;
 
    // 
    conn.close().await;
}  // drop 
 
//  3: RAII  + background task
struct BackgroundTask {
    abort: Option<tokio::sync::oneshot::Sender<()>>,
}
 
impl Drop for BackgroundTask {
    fn drop(&mut self) {
        if let Some(tx) = self.abort.take() {
            let _ = tx.send(());  // 
        }
    }
}

---

use tokio::sync::{broadcast, watch};
use std::sync::Arc;
 
struct App {
    shutdown_tx: watch::Sender<bool>,
    task_handles: Vec<tokio::task::JoinHandle<()>>,
}
 
impl App {
    async fn run(&mut self) {
        loop {
            tokio::select! {
                _ = self.shutdown_tx.changed() => {
                    if *self.shutdown_tx.borrow() {
                        break;
                    }
                }
                _ = async {
                    // 
                    tokio::time::sleep(std::time::Duration::from_secs(1)).await;
                } => {}
            }
        }
 
        // 
        for handle in self.task_handles.drain(..) {
            let _ = handle.await;
        }
    }
 
    fn shutdown(&self) {
        let _ = self.shutdown_tx.send(true);
    }
}
 
//  tokio_util  Cancellation Token
use tokio_util::sync::CancellationToken;
 
async fn worker(token: CancellationToken) {
    loop {
        tokio::select! {
            _ = token.cancelled() => {
                println!("worker received shutdown signal");
                break;
            }
            _ = do_work() => {}
        }
    }
}
 
async fn run_app() {
    let token = CancellationToken::new();
 
    let mut handles = vec![];
    for i in 0..5 {
        let token = token.clone();
        handles.push(tokio::spawn(async move {
            println!("worker {} started", i);
            worker(token).await;
            println!("worker {} finished", i);
        }));
    }
 
    // 
    tokio::signal::ctrl_c().await.ok();
    token.cancel();
 
    for h in handles {
        let _ = h.await;
    }
    println!("all workers stopped");
}

---

	std::thread	tokio::task	rayon
	OS		work-stealing
	CPU	IO
	~1μs	~20ns
	~2MB/	~/task
	OS
		select!/cancel

---