⚡ Async/await, Tokio và chia sẻ state an toàn

By Everwise Intern

Bài trước bạn đã bỏ panic! khỏi luồng tiền và để lỗi chảy qua Result. Nhưng service ngân hàng không xử lý từng request một: hàng trăm lệnh chuyển khoản cùng lúc, mỗi lệnh chờ database vài mili giây. Chặn một OS thread cho mỗi request là cách nhanh nhất để hết thread trước khi hết tải. Đây là lúc async/await và Tokio bước vào, kéo theo câu hỏi khó hơn: nhiều task cùng đọc/ghi một state mà không race condition, không deadlock.

Sau bài này bạn sẽ:

  • Phân biệt Future (kế hoạch) với runtime (người thi hành), viết #[tokio::main] và tokio::spawn.
  • Chia sẻ state bằng Arc<Mutex<T>> và Arc<RwLock<T>>, chọn đúng cái cho từng chỗ.
  • Giải thích Send, Sync và vì sao compiler chặn phần lớn lỗi concurrency trước khi code chạy.
  • Nhận diện ba cạm bẫy làm sập service tiền: giữ lock qua .await, deadlock do lock ordering, blocking I/O trong async.

Async/await: Future là kế hoạch, Tokio mới là người thi hành

Một async fn không chạy gì cả khi được gọi. Nó trả về một giá trị implement Future: một state machine ghi lại "đang chờ gì, làm tiếp ra sao". Future chỉ tiến triển khi có ai đó poll nó, và người đó là runtime. .await nghĩa là "chỗ này chưa xong, nhường thread cho task khác, khi có kết quả thì quay lại".

Tokio là runtime multi-thread: một pool worker thread, mỗi thread chạy nhiều task. Task là green thread do Tokio lập lịch, nhẹ hơn OS thread rất nhiều, nên 10.000 task chờ I/O là chuyện bình thường.

use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;

struct AppState {
    balances: RwLock<HashMap<u64, i64>>, // số dư tính bằng cent, không dùng f64
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let state = Arc::new(AppState { balances: RwLock::new(HashMap::new()) });

    let mut handles = Vec::new();
    for account_id in 1..=3u64 {
        let state = Arc::clone(&state); // mỗi task sở hữu một Arc riêng
        handles.push(tokio::spawn(async move {
            let mut guard = state.balances.write().await;
            guard.insert(account_id, 0);
        }));
    }

    for h in handles {
        h.await?; // JoinError nếu task panic, đừng unwrap
    }
    Ok(())
}

Điểm quan trọng: tokio::spawn yêu cầu future 'static + Send. 'static vì task có thể sống lâu hơn scope tạo ra nó. Send vì task có thể bị chuyển sang worker thread khác. Đó là lý do bạn phải Arc::clone rồi move vào, chứ không thể borrow &state.

Thử sức: sửa lỗi borrow trong task

let state = Arc::new(AppState::new());
tokio::spawn(async {
    process(&state).await; // không biên dịch
});

Xem gợi ý

Task cần sở hữu dữ liệu của nó. Sửa thành:

let state = Arc::clone(&state);
tokio::spawn(async move { process(&state).await });

Arc::clone tăng reference count (rẻ), move chuyển Arc đó vào future. Biến state gốc vẫn dùng được ở chỗ khác.

Chia sẻ state: Arc mở đường, Mutex và RwLock giữ cửa

Arc<T> cho nhiều nơi cùng sở hữu một giá trị, nhưng chỉ cho phép truy cập immutable. Muốn sửa, cần interior mutability: Mutex (độc quyền tuyệt đối) hoặc RwLock (nhiều reader song song, một writer độc quyền).

Với dữ liệu tài chính, RwLock thường đúng hơn: đọc số dư xảy ra liên tục, ghi chỉ khi có giao dịch. Nhưng đừng tối ưu sớm, Mutex đơn giản hơn và ít chỗ sai hơn.

Có hai họ lock, chọn sai là trả giá:

std::sync::Mutex / RwLock tokio::sync::Mutex / RwLock
Khi lock bị giữ Chặn OS thread .await, nhường task khác
Dùng khi Lock rất ngắn, không .await bên trong Cần giữ lock qua .await
Guard Không Send Send (nếu T: Send)

Giữ guard của std::sync::Mutex qua .await là lỗi biên dịch, không phải lỗi runtime: future mất Send và tokio::spawn từ chối.

Thử sức: Mutex hay RwLock?

Service có get_balance(id) gọi rất nhiều và apply_transfer(from, to, amount) gọi ít hơn nhiều lần. Bạn chọn gì cho HashMap<u64, i64> số dư, và tại sao?

Xem gợi ý

Arc<RwLock<HashMap<u64, i64>>>. Nhiều reader đọc song song không chặn nhau, writer độc quyền khi có giao dịch. Nhưng lưu ý: với một giao dịch chuyển khoản, bạn cần đọc cả hai tài khoản rồi ghi cả hai, nên thường phải lấy write lock một lần cho toàn bộ thao tác để giữ tính nhất quán, chứ không lấy hai read lock rồi hai write lock. Bài sau trong module sẽ đi sâu vào tính nhất quán này.

Send, Sync và ba cái bẫy làm sập service tiền

Send nghĩa là giá trị có thể chuyển ownership sang thread khác an toàn. Sync nghĩa là &T có thể chia sẻ giữa các thread an toàn. Hầu hết type bạn dùng đều có cả hai; những cái không có là có lý do: Rc không Send (đếm reference không atomic), RefCell không Sync (borrow flag không đồng bộ). Compiler tự suy ra, bạn hiếm khi phải viết unsafe impl.

Ba cạm bẫy thực chiến:

1. Giữ lock qua .await. Với std::sync::Mutex thì không biên dịch. Với tokio::sync::Mutex thì biên dịch nhưng giữ lock suốt một round-trip network, biến lock thành nút cổ chai. Cách sửa: copy dữ liệu cần thiết ra, drop(guard), rồi mới .await.

2. Deadlock do lock ordering. Task A chuyển tiền 1 → 2 (lock tài khoản 1, rồi 2). Task B chuyển 2 → 1 (lock 2, rồi 1). Cả hai chờ nhau mãi mãi.

// SAI: thứ tự lock phụ thuộc chiều chuyển tiền
let _a = locks[&from].lock().await;
let _b = locks[&to].lock().await;

Luật: luôn lock theo thứ tự account id tăng dần, bất kể chiều giao dịch.

3. Blocking trong async. Gọi file I/O đồng bộ, std::thread::sleep, hay hash một file lớn trong task async sẽ chặn worker thread, kéo theo hàng trăm task khác đói. Dùng tokio::task::spawn_blocking cho phần đó.

Thử sức: phá deadlock

Sửa đoạn code trên để hai chiều chuyển tiền không deadlock.

Xem gợi ý

let (first, second) = if from < to { (from, to) } else { (to, from) };
let _a = locks[&first].lock().await;
let _b = locks[&second].lock().await;
// dùng from/to gốc cho logic trừ tiền và cộng tiền

Thứ tự lock giờ chỉ phụ thuộc vào id, không phụ thuộc chiều giao dịch, nên không thể tạo vòng chờ. Một lựa chọn mạnh hơn: đẩy mọi lệnh ghi vào một mpsc channel và để một task duy nhất sở hữu ledger, khỏi cần lock.

Những luật ngầm khi viết async trong service tiền

  • Lock càng hẹp càng tốt: lấy guard, copy ra String/i64, drop(guard), rồi mới gọi network.
  • Không bao giờ lock().unwrap(). Poison chỉ xảy ra khi có panic, tức là đã có sự cố; hãy map nó thành lỗi hệ thống của service, không biến nó thành panic thứ hai.
  • Ghi lock ordering vào comment ngay cạnh struct state. Người sau thêm hàm mới sẽ không biết luật nếu bạn không ghi.
  • Bọc mọi .await ra hệ thống ngoài bằng tokio::time::timeout. Một database treo không được phép treo luôn request.
  • JoinHandle::await trả Result, đừng unwrap. Task panic không giết process, nhưng im lặng bỏ qua thì giao dịch mất tích.
  • Async không làm code tính toán nhanh hơn. Nó chỉ giúp chờ I/O rẻ hơn. Đo trước khi thêm tầng phức tạp.

Gói lại và bước tiếp theo

Bốn điều cần nhớ: Future là lazy, phải có Tokio poll; task cần 'static + Send nên hãy Arc::clone rồi move; Arc để chia sẻ quyền sở hữu, Mutex/RwLock để kiểm soát ghi; và ba cạm bẫy kinh điển là giữ lock qua .await, deadlock do lock ordering, blocking trong async.

Bài checkpoint tiếp theo sẽ ghép toàn bộ module lại: bạn sẽ viết luồng chuyển tiền async trả Result, xử lý lỗi từ database và từ task, đồng thời chứng minh code không panic và không deadlock dưới tải đồng thời.

  1. Một async fn trả về một _____, và Future chỉ tiến triển khi được _____ hoặc được spawn lên runtime.
  2. tokio::spawn yêu cầu future có lifetime _____ và implement trait _____.
  3. Để nhiều task cùng sở hữu state, ta bọc state trong _____, và để ghi an toàn thì bọc thêm _____ hoặc RwLock.
  4. Giữ guard của _____ qua .await làm future mất Send và bị từ chối biên dịch.
  5. Deadlock khi chuyển tiền hai chiều được sửa bằng cách lock theo thứ tự _____.
[
  { "front": "async fn trả về kiểu gì?", "back": "Một Future (impl Future<Output = T>), chưa chạy cho tới khi được await hoặc spawn." },
  { "front": "Future trong Rust có tự chạy nền không?", "back": "Không. Future là lazy, phải có runtime như Tokio poll thì mới tiến triển." },
  { "front": "tokio::spawn yêu cầu gì ở future?", "back": "'static + Send, vì task có thể sống lâu hơn scope tạo ra nó và chạy trên worker thread khác." },
  { "front": "Vì sao phải Arc::clone(&state) trước khi move vào task?", "back": "Để task sở hữu riêng một Arc, tránh borrow &state không sống đủ lâu ('static)." },
  { "front": "Arc<Mutex<T>> cho phép gì?", "back": "Nhiều task cùng sở hữu state (Arc) và truy cập tuần tự, ghi an toàn (Mutex)." },
  { "front": "RwLock phù hợp khi nào?", "back": "Đọc nhiều, ghi ít: nhiều reader song song, writer độc quyền." },
  { "front": "Send nghĩa là gì?", "back": "Type có thể chuyển ownership sang thread khác một cách an toàn." },
  { "front": "Sync nghĩa là gì?", "back": "&T có thể chia sẻ giữa nhiều thread một cách an toàn." },
  { "front": "Vì sao Rc và RefCell không dùng được trong async đa thread?", "back": "Rc không Send (đếm reference không atomic) và RefCell không Sync, nên không lọt qua bound Send của tokio::spawn.
  { "front": "Vì sao không giữ std::sync::MutexGuard qua .await?", "back": "Guard không Send, future mất Send và bị từ chối biên dịch; nếu buộc phải giữ lock qua await thì dùng tokio::sync::Mutex." },
  { "front": "Deadlock khi chuyển tiền hai chiều được phòng bằng cách nào?", "back": "Lock theo thứ tự cố định (ví dụ account id tăng dần) ở mọi chỗ trong codebase." },
  { "front": "Khi nào cần spawn_blocking?", "back": "Khi buộc phải gọi API blocking (file, driver DB đồng bộ, hash nặng) để không chiếm worker thread của runtime." }
]

Chốt lại trước khi vào checkpoint

Bốn điều cần mang theo từ bài này:

  • async fn chỉ tạo Future lazy. Không có runtime (Tokio) poll thì không có gì chạy.
  • tokio::spawn là ranh giới ownership: cần 'static + Send, nên phải Arc::clone state trước khi move vào task.
  • Chia sẻ state ghi được thì dùng Arc<Mutex<T>>, đọc nhiều ghi ít thì Arc<RwLock<T>>. Chọn theo tỉ lệ đọc/ghi, không chọn theo cảm tính.
  • Send/Sync không phải lý thuyết suông, chúng là thứ compiler dùng để chặn race condition ngay lúc build. Khi bị chặn, gần như luôn là do Rc, RefCell, hoặc một guard bị giữ qua .await.

Bài tiếp theo là checkpoint của module: bạn sẽ gặp lại toàn bộ Result, error modeling và async trong một bài tập tổng hợp, nơi mọi lỗi phải được mô hình hóa thay vì unwrap, và mọi điểm chia sẻ state phải chứng minh được là không deadlock.