Async Rust không khó — nó chỉ khác

6 tháng mình né async Rust như né production deploy vào thứ Sáu. Mỗi lần google là đụng ngay Future, Waker, Poll::Pending — đọc nửa bài là đóng tab. Thôi sync cho lành.

Vấn đề không phải async Rust khó. Vấn đề là mình đọc implementation details của runtime trước khi hiểu cách dùng. Khi mình lùi lại, quên hết Waker, quên hết Poll, và nghĩ từ đầu — mọi thứ tự nhiên click.

Sai lầm khi học async Rust

Hầu hết tutorial bắt đầu từ đây:

pub trait Future {
    type Output;
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}

Đây là cách runtime triển khai, không phải cách anh em sẽ dùng. Học Future trait trước khi dùng async/await giống như học bytecode JVM trước khi viết Java.

Mental model đúng chỉ có một câu:

async fn = function trả về một “computation” chưa chạy.

Gọi fetch_data() không chạy gì cả. Nó tạo ra một object mô tả những gì sẽ làm. Khi outer task gặp .await, nó bắt đầu poll future con; nếu future trả Pending, executor sẽ poll lại outer task sau khi được Waker đánh thức. Nói ngắn gọn: .await nối computation đó vào task hiện tại, chứ không gửi riêng nó sang một executor khác.

async fn = factory cho state machine

Khi viết thế này:

async fn fetch_data(url: &str) -> String {
    let response = reqwest::get(url).await.unwrap();
    response.text().await.unwrap()
}

Rust compiler không generate một function thông thường. Nó generate một struct — một state machine — có dạng conceptually như sau:

// Pseudo-code, không phải actual generated code
enum FetchDataStateMachine {
    // State 0: chưa bắt đầu
    Start { url: String },
    // State 1: đang chờ reqwest::get hoàn thành
    WaitingForGet { future: ReqwestGetFuture },
    // State 2: đang chờ response.text() hoàn thành
    WaitingForText { future: ResponseTextFuture },
    // State 3: xong
    Done,
}

Mỗi await point là một state. Mỗi lần runtime poll struct này, nó hỏi “đang ở state nào, cái đang chờ xong chưa” — nếu xong thì chuyển sang state tiếp theo.

Hệ quả thực tế: một async fn 10 dòng có thể compile thành struct với vài chục fields lưu state. Gọi và chain async fn/.await không bắt buộc mỗi future phải có một heap allocation riêng. Nhưng runtime, I/O library và tokio::spawn vẫn có thể allocate task storage trên heap — “zero-cost” ở đây nói về cách compiler biến control flow thành state machine, không có nghĩa toàn bộ async program không bao giờ allocate.

await không block thread

Hai dòng này trông giống nhau, nhưng hoàn toàn khác nhau:

use std::thread;
use std::time::Duration;

// Cái này BLOCK thread trong 1 giây
thread::sleep(Duration::from_secs(1));

// Cái này YIELD control về runtime, thread vẫn làm việc khác
tokio::time::sleep(Duration::from_secs(1)).await;

Đứng sau .await là một lời gọi poll(). Executor poll future: nếu future báo Poll::Ready, lấy kết quả và chạy tiếp; nếu báo Poll::Pending (ví dụ timer chưa hết), future đăng ký một Waker với reactor rồi trả quyền điều khiển lại cho executor — task coi như “ngủ”, không giữ thread. Executor rảnh tay đi poll task khác. Khi timer hết, reactor gọi Waker để báo executor: task này poll lại được rồi. Thread không bị giam trong lúc chờ.

Xem timing thực tế:

use tokio::time::{sleep, Duration, Instant};

async fn task(id: u32, secs: u64) {
    println!("Task {id} bắt đầu");
    sleep(Duration::from_secs(secs)).await;
    println!("Task {id} xong");
}

#[tokio::main]
async fn main() {
    let start = Instant::now();

    // Sequential — tổng 3 giây
    task(1, 1).await;
    task(2, 2).await;
    println!("Sequential: {:.1}s", start.elapsed().as_secs_f32());

    let start = Instant::now();

    // Concurrent — tổng ~2 giây (task dài nhất)
    tokio::join!(task(1, 1), task(2, 2));
    println!("Concurrent: {:.1}s", start.elapsed().as_secs_f32());
}

Output:

Task 1 bắt đầu
Task 1 xong
Task 2 bắt đầu
Task 2 xong
Sequential: 3.0s
Task 1 bắt đầu
Task 2 bắt đầu
Task 1 xong
Task 2 xong
Concurrent: 2.0s

tokio::join! poll cả hai futures đó trong cùng một task hiện tại — không tạo task mới, không tự spawn OS thread nào. Trong lúc task 1 đang chờ (Pending), executor chuyển sang poll task 2. Lưu ý: “cùng task” không có nghĩa là “cố định một OS thread” — với runtime multi-thread mặc định của tokio, task vẫn có thể bị scheduler di chuyển giữa các worker thread khác nhau ở những lượt poll khác nhau (work-stealing). Muốn ép chạy trên đúng một OS thread thì cần current_thread runtime hoặc LocalSet.

async không tự chạy — cần executor

Cái này mình bị nhầm nhất hồi mới học. Tạo một Future không làm gì cả:

async fn send_email(to: &str) {
    println!("Gửi email đến {to}");
    // ... thực tế dùng SMTP client hay gì đó
}

fn main() {
    // Tạo Future — KHÔNG có gì xảy ra
    let _fut = send_email("[email protected]");

    // Email không được gửi. Không có side effect.
    // Compiler thậm chí warn: "unused implementor of Future"
    println!("Chương trình kết thúc");
}

Output:

Chương trình kết thúc

Email không đi đâu hết. Future là lazy computation — nó cần executor drive mới chạy được.

#[tokio::main] macro là cách nhanh nhất để có executor:

#[tokio::main]
async fn main() {
    // main() giờ là async, tokio runtime làm executor
    send_email("[email protected]").await; // Giờ mới thực sự chạy
}

Macro này expand ra roughly như sau:

fn main() {
    tokio::runtime::Runtime::new()
        .unwrap()
        .block_on(async {
            send_email("[email protected]").await;
        });
}

block_on là điểm mà sync world gặp async world — thread hiện tại block cho đến khi Future hoàn thành.

Send + Sync trong async context

Phần này hay khiến anh em bị lỗi compile lúc dùng tokio::spawn:

use std::rc::Rc;
use tokio::task;

async fn process(data: Rc<Vec<u8>>) {
    println!("Processing {} bytes", data.len());
}

#[tokio::main]
async fn main() {
    let data = Rc::new(vec![1, 2, 3]);

    // Compile error!
    task::spawn(process(data));
}

error[E0277]: `Rc<Vec<u8>>` cannot be sent between threads safely
   --> src/main.rs:11:18
    |
11  |     task::spawn(process(data));
    |                 ^^^^^^^^^^^^^ future is not `Send`

tokio::spawn có thể schedule task trên bất kỳ thread nào trong thread pool. Rc không thread-safe — nó không implement Send. Nếu task migrate sang thread khác đang giữ Rc, data race xảy ra. Borrow checker bắt được điều này tại compile time.

Fix đơn giản: dùng Arc thay Rc:

use std::sync::Arc;
use tokio::task;

async fn process(data: Arc<Vec<u8>>) {
    println!("Processing {} bytes", data.len());
}

#[tokio::main]
async fn main() {
    let data = Arc::new(vec![1, 2, 3]);

    // OK — Arc implement Send
    task::spawn(process(Arc::clone(&data))).await.unwrap();
}

Arc (Atomic Reference Count) dùng atomic operations thay vì plain integer, an toàn khi access từ nhiều thread — nhưng bản thân Arc<T> chỉ Send + Sync khi T cũng Send + Sync. Arc<RefCell<...>> vẫn không Sync, vì RefCell không thread-safe dù có bọc Arc bên ngoài. Đổi Rc thành Arc không tự động fix mọi lỗi Send — phải nhìn cả kiểu bên trong.

Rule ngắn gọn: Rc → dùng trong single-threaded context. Arc → dùng khi cần share giữa tasks hoặc threads.

Nếu anh em thực sự cần giữ Rc/RefCell (không muốn trả giá atomic của Arc, hoặc đang làm việc với thứ vốn !Send), tokio có LocalSet + spawn_local — chạy toàn bộ set task đó trên một thread cố định, không cần đổi sang Arc:

use std::rc::Rc;
use tokio::task::LocalSet;

#[tokio::main]
async fn main() {
    let data = Rc::new(vec![1, 2, 3]);
    let local = LocalSet::new();

    local.run_until(async move {
        tokio::task::spawn_local(async move {
            println!("Processing {} bytes", data.len());
        }).await.unwrap();
    }).await;
}

std::sync::Mutex và tokio::sync::Mutex không thay thế cho nhau

Cả hai đều gọi là Mutex, nhưng dùng sai chỗ là dính bug khó debug.

std::sync::Mutex là blocking mutex bình thường — lock nhanh, không tương tác gì với executor. Chỉ riêng việc giữ MutexGuard qua .await chưa block thread ngay; vấn đề xảy ra khi task khác trên executor gọi .lock() trong lúc guard vẫn được giữ. Lệnh .lock() đó block cả worker thread thay vì yield, và có thể dẫn tới deadlock nếu task đang giữ guard cần được poll tiếp mới nhả lock.

// Nguy hiểm: task khác gọi lock() trong lúc future này đang await
let guard = std_mutex.lock().unwrap();
some_async_fn().await; // guard vẫn sống; hãy drop trước await nếu có thể
drop(guard);

tokio::sync::Mutex được thiết kế để giữ qua .await — .lock() của nó là async, khi tranh chấp nó yield thay vì block thread. Nhưng nó chậm hơn std::sync::Mutex cho critical section ngắn thuần sync.

Rule ngắn gọn: critical section không chứa .await → dùng std::sync::Mutex, khóa càng ngắn càng tốt. Cần giữ lock qua .await → tokio::sync::Mutex, nhưng cân nhắc lại thiết kế trước — giữ lock xuyên async work thường là dấu hiệu nên tách nhỏ critical section ra.

Còn nếu anh em cần spawn background task và không cần đợi kết quả:

#[tokio::main]
async fn main() {
    // Spawn background task — không await, task chạy song song với main
    let handle = tokio::spawn(async {
        tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;
        println!("Background task xong");
    });

    println!("Main tiếp tục chạy ngay lập tức");

    // Nếu muốn đợi background task hoàn thành:
    handle.await.unwrap();
}

Chú ý: nếu không .await cái handle đó (task “detached”), task vẫn chạy song song — nhưng không có gì đảm bảo nó chạy xong. Nếu main() return hoặc runtime shutdown trước khi task detached hoàn thành, task đó bị hủy giữa chừng, không có warning hay error nào báo. Task detached chỉ an toàn khi anh em chắc chắn runtime sẽ sống đủ lâu, hoặc không quan tâm nó có chạy xong hay không.

Kết

• Bắt đầu với async/await và tokio::join! — không cần hiểu Future trait hay Poll để viết code async thực tế.
• Gặp “future is not Send” → thường là đổi Rc thành Arc, RefCell thành Mutex (cân nhắc tokio::sync::Mutex nếu phải giữ lock qua .await). Nếu không muốn trả giá atomic, LocalSet/spawn_local là lựa chọn khác.
• Mental model quan trọng hơn implementation details: async fn tạo state machine, await yield control (poll trả Pending, đăng ký Waker), executor drive tất cả. Nắm 3 điều này là đủ để debug hầu hết lỗi async Rust.