Async chuyên sâu trong Rust: select!, timeout, stream và Pin

Ở trang Async/Await, bạn đã học những điều cơ bản: một async fn trả về một future lười, .await chạy nó, #[tokio::main] khởi động runtime, còn tokio::join! và tokio::spawn chạy nhiều thứ cùng lúc. Nếu những điều đó còn mới với bạn, hãy đọc trang đó trước. Trang này xây tiếp trên nền đó.

Các chương trình async thực tế cần thêm vài công cụ. Hãy nghĩ lại về một gian bếp đông khách. Đôi khi bạn chờ mọi món (join!). Đôi khi bạn chỉ quan tâm tới điều đầu tiên xảy ra: món ăn đã xong, hoặc khách bỏ về (select!). Đôi khi bạn bỏ cuộc sau năm phút (timeout). Và đôi khi đơn hàng cứ tới, từng đơn một, suốt cả buổi tối (stream). Trang này nói về tất cả những thứ đó, cùng với Pin, từ ngữ kỳ lạ mà sớm muộn bạn sẽ gặp trong thông báo lỗi, và giải thích nó bằng ngôn ngữ dễ hiểu.

Mọi ví dụ đều dùng tokio với features = ["full"], giống trang cơ bản. Các nút “Tự chạy thử” đều dùng được, vì Rust Playground có sẵn tokio.

Chạy nhiều thứ cùng lúc

Bạn đã biết tokio::join!: nó chờ mọi future của nó và trả kết quả dưới dạng một tuple. Người anh em tokio::try_join! dành cho các future trả về Result. Nó cho bạn Ok với tất cả giá trị, hoặc Err đầu tiên, và ngừng chờ các future khác ngay khi có một cái thất bại:

use std::time::Duration;
use tokio::time::sleep;

async fn load_profile(id: u32) -> Result<String, String> {
    sleep(Duration::from_millis(20)).await;
    Ok(format!("hồ sơ #{id}"))
}

async fn load_settings(id: u32) -> Result<String, String> {
    sleep(Duration::from_millis(10)).await;
    if id == 2 {
        return Err(format!("không có cài đặt cho người dùng {id}"));
    }
    Ok(String::from("chế độ tối"))
}

#[tokio::main]
async fn main() {
    for id in [1, 2] {
        match tokio::try_join!(load_profile(id), load_settings(id)) {
            Ok((profile, settings)) => println!("người dùng {id}: {profile}, {settings}"),
            Err(e) => println!("người dùng {id} thất bại: {e}"),
        }
    }
}

Kết quả:

người dùng 1: hồ sơ #1, chế độ tối
người dùng 2 thất bại: không có cài đặt cho người dùng 2

▶ Tự chạy thử

join! và try_join! chạy các future bên trong task hiện tại, thay phiên nhau. Điều đó hoàn hảo cho một vài future bạn biết trước. Khi bạn có một danh sách công việc mà chỉ lúc chạy mới biết số lượng, hãy spawn chúng thành các task riêng. Bạn có thể đẩy từng JoinHandle vào một Vec, như ở trang cơ bản. Nhưng tokio có một công cụ đẹp hơn cho việc này: JoinSet, một túi chứa các task, trả cho bạn từng kết quả ngay khi nó xong:

use std::time::Duration;
use tokio::task::JoinSet;
use tokio::time::sleep;

async fn check_site(site: &'static str, ms: u64) -> (&'static str, u64) {
    sleep(Duration::from_millis(ms)).await; // giả vờ đây là một lời gọi mạng
    (site, ms)
}

#[tokio::main]
async fn main() {
    let mut set = JoinSet::new();
    for (site, ms) in [("rust-lang.org", 30), ("docs.rs", 10), ("crates.io", 20)] {
        set.spawn(check_site(site, ms));
    }

    let mut results = Vec::new();
    while let Some(res) = set.join_next().await {
        results.push(res.unwrap()); // chỉ là Err nếu task đã panic
    }

    results.sort(); // thứ tự xong có thể khác nhau giữa các lần chạy
    for (site, ms) in results {
        println!("{site} trả lời sau {ms}ms");
    }
}

Kết quả:

crates.io trả lời sau 20ms
docs.rs trả lời sau 10ms
rust-lang.org trả lời sau 30ms

▶ Tự chạy thử

join_next() trả về None khi túi đã rỗng, nên vòng while let tự kết thúc. Kết quả đến theo thứ tự xong, không phải thứ tự spawn. Khi các task mất thời gian gần bằng nhau, thứ tự đó có thể thay đổi giữa các lần chạy, nên chúng ta sắp xếp trước khi in. Thêm một điểm hay nữa: nếu bạn drop một JoinSet, mọi task còn bên trong sẽ bị hủy. Không có gì vô tình bị bỏ lại chạy ngầm.

Chạy đua các future với select!

join! chờ mọi thứ. tokio::select! chờ cái đầu tiên. Mỗi nhánh có một mẫu (pattern), một future và một khối code: pattern = future => code. Future nào xong trước thì code của nó chạy, và các nhánh khác bị hủy.

Bị hủy trong Rust nghĩa là gì? Future thua cuộc đơn giản là bị drop, giống như bất kỳ giá trị nào khác ra khỏi phạm vi. Nó không bao giờ chạy lại nữa. Các biến cục bộ của nó cũng bị drop, nên code dọn dẹp của chúng sẽ chạy. Ví dụ này dùng một kiểu Drop nhỏ để thấy rõ điều đó:

use std::time::Duration;
use tokio::time::sleep;

struct Download(&'static str);

impl Drop for Download {
    fn drop(&mut self) {
        println!("dọn dẹp lượt tải từ {}", self.0);
    }
}

async fn download(mirror: &'static str, ms: u64) -> &'static str {
    let _download = Download(mirror);
    sleep(Duration::from_millis(ms)).await;
    println!("tải từ {mirror} đã xong");
    mirror
}

#[tokio::main]
async fn main() {
    tokio::select! {
        winner = download("châu Âu", 50) => println!("thắng cuộc: {winner}"),
        winner = download("châu Á", 1000) => println!("thắng cuộc: {winner}"),
    }
    println!("main tiếp tục");
}

Kết quả:

tải từ châu Âu đã xong
dọn dẹp lượt tải từ châu Âu
dọn dẹp lượt tải từ châu Á
thắng cuộc: châu Âu
main tiếp tục

▶ Tự chạy thử

Hãy nhìn thứ bị thiếu: “tải từ châu Á đã xong” không bao giờ được in. Future châu Á bị dừng ở .await của nó và bị drop, và giá trị Download của nó được dọn dẹp ngay lập tức. Đây là cách việc hủy hoạt động ở khắp nơi trong async Rust: ngừng poll một future và drop nó. Không cần lời gọi “hủy” đặc biệt nào.

Một cách dùng kinh điển của select! là timeout: cho công việc của bạn chạy đua với một bộ hẹn giờ.

tokio::select! {
    report = build_report() => println!("{report}"),
    _ = tokio::time::sleep(Duration::from_secs(5)) => println!("quá chậm, bỏ cuộc"),
}

Cách này chạy tốt, nhưng có một lối tắt cho nó, tokio::time::timeout, được giới thiệu ở phần tiếp theo.

Làm việc cho tới khi có tín hiệu tắt

select! thật sự tỏa sáng bên trong một loop. Ở đây một worker lấy việc từ một channel, nhưng cũng lắng nghe tin nhắn “làm ơn dừng lại” trên một channel thứ hai. Một channel oneshot mang đúng một tin nhắn, nên nó là tín hiệu tắt hoàn hảo:

use std::time::Duration;
use tokio::sync::{mpsc, oneshot};
use tokio::time::sleep;

#[tokio::main]
async fn main() {
    let (job_tx, mut job_rx) = mpsc::channel::<String>(10);
    let (stop_tx, mut stop_rx) = oneshot::channel::<()>();

    let worker = tokio::spawn(async move {
        let mut done = 0;
        loop {
            tokio::select! {
                Some(job) = job_rx.recv() => {
                    println!("đang làm: {job}");
                    done += 1;
                }
                _ = &mut stop_rx => {
                    println!("đã nhận tín hiệu tắt");
                    break;
                }
            }
        }
        done
    });

    for job in ["thay đổi kích thước ảnh", "gửi email", "cập nhật chỉ mục tìm kiếm"] {
        job_tx.send(job.to_string()).await.unwrap();
    }
    sleep(Duration::from_millis(100)).await; // cho worker thời gian làm xong
    stop_tx.send(()).unwrap();

    let done = worker.await.unwrap();
    println!("worker đã xong {done} việc");
}

Kết quả:

đang làm: thay đổi kích thước ảnh
đang làm: gửi email
đang làm: cập nhật chỉ mục tìm kiếm
đã nhận tín hiệu tắt
worker đã xong 3 việc

▶ Tự chạy thử

Vài chi tiết đáng biết:

• Chúng ta viết &mut stop_rx, không phải stop_rx. Mỗi vòng lặp, select! cần lại cùng một đầu nhận, nên ta cho mượn nó thay vì move nó vào vòng đầu tiên.
• Mỗi vòng lặp, job_rx.recv() tạo ra một future mới, và future thua cuộc bị drop. Không sao cả, vì recv() là cancel-safe (an toàn khi hủy): drop nó không bao giờ làm mất tin nhắn.
• Nếu mẫu không khớp (ở đây, recv() trả về None vì mọi đầu gửi đã không còn), select! tắt nhánh đó và tiếp tục chờ các nhánh khác.
• Nếu nhiều nhánh sẵn sàng cùng một lúc, select! chọn ngẫu nhiên một nhánh, để không nhánh nào chiếm hết vòng lặp. Hãy viết biased; ở dòng đầu tiên bên trong select! nếu muốn nó kiểm tra các nhánh từ trên xuống dưới.

An toàn khi hủy, gói gọn trong một đoạn. Vì nhánh thua cuộc bị drop, mọi việc nó đang làm dở đều bị bỏ đi. Với recv() trên một channel hay sleep(), chuyện đó vô hại: chưa có gì bị lấy ra cả. Nhưng hãy tưởng tượng một future đọc 100 byte từ một socket mạng và đã đọc được 60 byte thì bị drop. 60 byte đó mất luôn. Future như vậy không cancel-safe. Tài liệu của tokio ghi rõ với từng phương thức là nó có cancel-safe hay không. Khi bạn dùng select! trong một vòng lặp, hãy kiểm tra các phương thức bạn gọi, hoặc tạo future một lần bên ngoài vòng lặp rồi tiếp tục poll chính future đó (bạn sẽ thấy cách làm ở phần Pin).

Timeout và bộ hẹn giờ

tokio::time::timeout(duration, future) bọc quanh bất kỳ future nào. Nếu future xong kịp giờ, bạn nhận được Ok(value). Nếu hết giờ trước, future bị drop (bị hủy, giống hệt một nhánh select! thua cuộc) và bạn nhận được Err(Elapsed):

use std::time::Duration;
use tokio::time::{sleep, timeout};

async fn fetch_price(server_delay_ms: u64) -> u32 {
    sleep(Duration::from_millis(server_delay_ms)).await;
    42
}

#[tokio::main]
async fn main() {
    let limit = Duration::from_millis(200);

    for (server, delay) in [("máy chủ nhanh", 20), ("máy chủ chậm", 2000)] {
        match timeout(limit, fetch_price(delay)).await {
            Ok(price) => println!("{server}: giá là {price}"),
            Err(e) => println!("{server}: {e}"),
        }
    }
}

Kết quả:

máy chủ nhanh: giá là 42
máy chủ chậm: deadline has elapsed

▶ Tự chạy thử

“deadline has elapsed” là thông báo lỗi có sẵn của tokio (bằng tiếng Anh), nghĩa là “đã hết thời hạn”. Hãy đặt timeout cho mọi thứ giao tiếp với thế giới bên ngoài: lời gọi HTTP, truy vấn cơ sở dữ liệu, đọc từ socket. Một máy chủ không bao giờ trả lời không nên làm chương trình của bạn đứng im mãi mãi.

sleep và interval. sleep chờ một lần. interval cho bạn một nhịp đều đặn, giống như máy đếm nhịp. Bạn gọi .tick().await trong một vòng lặp, và mỗi tick đến đúng lịch. Tick đầu tiên kích hoạt ngay lập tức:

use std::time::Duration;
use tokio::time::interval;

#[tokio::main]
async fn main() {
    let mut heartbeat = interval(Duration::from_millis(50));

    for beat in 1..=3 {
        heartbeat.tick().await; // tick đầu tiên hoàn tất ngay lập tức
        println!("nhịp tim {beat}");
    }
}

Kết quả:

nhịp tim 1
nhịp tim 2
nhịp tim 3

▶ Tự chạy thử

Sao không chỉ sleep 50ms trong một vòng lặp? Vì khi đó mỗi vòng mất “thời gian làm việc + 50ms”, và lịch trình sẽ trôi dần. Một interval đo từ thời điểm bắt đầu dự kiến của mỗi tick, nên nếu việc của bạn mất 10ms, nó chỉ chờ 40ms còn lại. Dùng interval cho heartbeat, polling và các việc kiểu “lưu mỗi phút”. Dùng sleep cho một lần tạm dừng.

Stream: iterator bất đồng bộ

Một iterator đưa cho bạn từng giá trị một bằng next(). Một stream là phiên bản async: next().await cho bạn giá trị tiếp theo khi nó sẵn sàng, và None khi đã hết. Hãy nghĩ tới bảng giá cổ phiếu chạy liên tục, các hàng dữ liệu đến từ cơ sở dữ liệu, hay tin nhắn từ một máy chủ chat.

Thật ra bạn đã dùng một thứ giống stream rồi: đầu nhận của channel tokio. rx.recv().await chờ tin nhắn tiếp theo và trả về None khi mọi đầu gửi đã không còn:

use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::sleep;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(8);

    tokio::spawn(async move {
        for temp in [21, 22, 24, 23] {
            tx.send(temp).await.unwrap();
            sleep(Duration::from_millis(10)).await; // mỗi 10ms một lần đo
        }
        // tx bị drop ở đây, việc này đóng channel
    });

    while let Some(temp) = rx.recv().await {
        println!("nhiệt độ: {temp}°C");
    }
    println!("cảm biến đã dừng");
}

Kết quả:

nhiệt độ: 21°C
nhiệt độ: 22°C
nhiệt độ: 24°C
nhiệt độ: 23°C
cảm biến đã dừng

▶ Tự chạy thử

Thư viện chuẩn chưa có trait Stream (nó vẫn đang được thiết kế). Hệ sinh thái dùng trait từ crate futures, và crate hỗ trợ của tokio là tokio-stream thêm trait StreamExt với các phương thức bạn đã quen từ iterator: next(), map(), filter(), take() và nhiều hơn nữa. Thêm nó vào dự án của bạn:

[dependencies]
tokio = { version = "1", features = ["full"] }
tokio-stream = "0.1"

use tokio_stream::{self as stream, StreamExt};

#[tokio::main]
async fn main() {
    let prices = stream::iter(vec![100, 102, 98, 105, 110]);

    let mut doubled = prices.map(|p| p * 2).take(3);

    while let Some(p) = doubled.next().await {
        println!("{p}");
    }
}

Chương trình này in ra 200, 204 và 196. Nó đọc giống hệt code iterator, chỉ khác .await sau next(). Để biến đầu nhận của channel tokio thành một stream thật sự, hãy bọc nó: tokio_stream::wrappers::ReceiverStream::new(rx). (Crate futures có StreamExt riêng với cùng ý tưởng. Mỗi tệp chỉ nên chọn một, vì cả hai trait đều có phương thức next() và chúng sẽ xung đột.)

Pin, nói cho dễ hiểu

Sớm hay muộn bạn sẽ thấy Pin trong một kiểu hoặc một thông báo lỗi. Đây là ý tưởng của nó, không cần lý thuyết sâu.

Khi Rust biên dịch một async fn, nó biến hàm đó thành một struct ẩn, một máy trạng thái (state machine). Mỗi biến cục bộ vẫn còn cần dùng sau một .await trở thành một trường của struct đó. Giờ hãy tưởng tượng một biến cục bộ mượn một biến khác, như let name = String::from("Ana"); let r = &name; với một .await ở giữa. Khi đó struct chứa một con trỏ trỏ tới chính trường của nó. Nếu bạn di chuyển struct sang một chỗ khác trong bộ nhớ, con trỏ đó sẽ trỏ vào chỗ cũ, giờ đã trống.

Pin là cách sửa. Một giá trị đã được pin là một lời hứa: “giá trị này sẽ không di chuyển trong bộ nhớ nữa.” Khi một future đã được pin, poll nó là an toàn. Phần lớn thời gian bạn không hề để ý, vì .await tự pin future giúp bạn.

Bạn sẽ gặp Pin trong ba tình huống:

1. Poll cùng một future nhiều lần, ví dụ một thời hạn trong vòng lặp select!.
2. Lưu trữ future trong một Vec hoặc một trường của struct, thường ở dạng Pin<Box<dyn Future<Output = T>>>.
3. Đệ quy trong một async fn.

Với tình huống đầu tiên, đây là điều xảy ra nếu bạn cố dùng lại một sleep trong vòng lặp mà không pin nó:

let deadline = sleep(Duration::from_millis(300));

loop {
    tokio::select! {
        _ = &mut deadline => break,
        Some(n) = rx.recv() => println!("{n}"),
    }
}

error[E0277]: `PhantomPinned` cannot be unpinned
   |
   = note: consider using the `pin!` macro
           consider using `Box::pin` if you need to access the pinned value outside of the current scope
   = note: required for `Sleep` to implement `Unpin`

Trình biên dịch (bằng tiếng Anh) thậm chí còn chỉ cho bạn cách sửa: dùng macro pin!, hoặc Box::pin. std::pin::pin! (hoặc tokio::pin!) pin future ngay tại chỗ trên stack. Giờ cùng một thời hạn được dùng chung cho mọi vòng của vòng lặp:

use std::pin::pin;
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::sleep;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(8);
    tokio::spawn(async move {
        for order in 1..=3 {
            tx.send(order).await.unwrap();
            sleep(Duration::from_millis(10)).await;
        }
        sleep(Duration::from_secs(5)).await; // rồi cửa hàng trở nên vắng lặng
    });

    let mut deadline = pin!(sleep(Duration::from_millis(300)));

    loop {
        tokio::select! {
            _ = &mut deadline => {
                println!("đến giờ đóng cửa");
                break;
            }
            Some(order) = rx.recv() => println!("nhận đơn #{order}"),
        }
    }
}

Kết quả:

nhận đơn #1
nhận đơn #2
nhận đơn #3
đến giờ đóng cửa

▶ Tự chạy thử

Nếu chúng ta viết sleep(...) bên trong select!, mỗi đơn hàng mới sẽ khởi động một bộ hẹn giờ 300ms mới, và một cửa hàng đông khách sẽ không bao giờ đóng cửa. Pin một thời hạn duy nhất bên ngoài vòng lặp sẽ sửa được điều đó.

Với đệ quy, trình biên dịch cần biết future lớn cỡ nào, mà một future chứa chính nó sẽ lớn vô hạn. Gọi countdown(n - 1).await bên trong async fn countdown sẽ cho ra:

error[E0733]: recursion in an async fn requires boxing
  |
  = note: a recursive `async fn` call must introduce indirection such as `Box::pin` to avoid an infinitely sized future

Thông báo nói rằng đệ quy trong một async fn cần được “đóng hộp” (boxing). Box::pin chuyển future bên trong lên heap và pin nó ở đó. Giờ future bên ngoài chỉ giữ một con trỏ, thứ có kích thước cố định:

async fn countdown(n: u32) {
    if n == 0 {
        println!("cất cánh!");
        return;
    }
    println!("{n}...");
    Box::pin(countdown(n - 1)).await;
}

#[tokio::main]
async fn main() {
    countdown(3).await;
}

Kết quả:

3...
2...
1...
cất cánh!

▶ Tự chạy thử

Thật lòng mà nói, đó là tất cả những gì hầu hết mọi người cần: pin! cho một future bạn poll nhiều lần, Box::pin để lưu trữ future hoặc để đệ quy. Nếu bạn muốn câu chuyện đầy đủ, chương sách The Rust Book trong danh sách Đọc thêm giải thích nó từng bước một.

async fn trong trait và async closure

Từ Rust 1.75, bạn có thể viết async fn trực tiếp trong một trait. Từ Rust 1.85, bạn còn có async closure, viết là async |x| ..., với các trait tương ứng AsyncFn, AsyncFnMut và AsyncFnOnce (phiên bản async của các trait closure). Đây là cả hai trong một ví dụ:

use std::time::Duration;
use tokio::time::sleep;

trait Notifier {
    async fn notify(&self, user: &str) -> String;
}

struct Email;

impl Notifier for Email {
    async fn notify(&self, user: &str) -> String {
        sleep(Duration::from_millis(10)).await; // giả vờ nói chuyện với máy chủ mail
        format!("đã gửi email cho {user}")
    }
}

async fn for_each_user(users: &[&str], job: impl AsyncFn(&str) -> String) {
    for user in users {
        println!("{}", job(user).await);
    }
}

#[tokio::main]
async fn main() {
    let email = Email;
    let users = ["ana", "ben"];

    for_each_user(&users, async |user: &str| email.notify(user).await).await;
}

Kết quả:

đã gửi email cho ana
đã gửi email cho ben

▶ Tự chạy thử

Có một điểm vướng với async fn trong trait. Trait không hứa rằng future trả về là Send, trong khi tokio::spawn cần future Send (chúng có thể di chuyển giữa các luồng). Vì vậy spawn một lời gọi qua một kiểu generic N: Notifier sẽ thất bại:

fn send_in_background<N: Notifier + Send + Sync + 'static>(notifier: N) {
    tokio::spawn(async move {
        let msg = notifier.notify("ana").await;
        println!("{msg}");
    });
}

error: future cannot be sent between threads safely
  |
7 |         let msg = notifier.notify("ana").await;
  |                   ^^^^^^^^^^^^^^^^^^^^^^ await occurs here on type `impl Future<Output = String>`, which is not `Send`
  |
help: `Send` can be made part of the associated future's guarantees for all implementations of `Notifier::notify`
  |
2 -     async fn notify(&self, user: &str) -> String;
2 +     fn notify(&self, user: &str) -> impl std::future::Future<Output = String> + Send;

Thông báo nói rằng future không thể gửi an toàn giữa các luồng. Gợi ý của trình biên dịch cũng là cách sửa thường dùng: viết phương thức của trait thành fn ... -> impl Future<Output = String> + Send. Bên triển khai vẫn có thể viết một async fn bình thường. Cũng lưu ý rằng trait có async fn hiện chưa dùng được làm trait object dyn Notifier. Nếu bạn cần điều đó, crate phổ biến async-trait sẽ đóng hộp các future giúp bạn.

Lỗi thường gặp

Chặn runtime. std::thread::sleep, tính toán nặng, hoặc đọc tệp lớn bằng std::fs bên trong một task async sẽ đóng băng cả luồng worker, và mọi task khác trên luồng đó phải chờ. Trang cơ bản đã cho thấy phiên bản với sleep. Với việc nặng về CPU hoặc mang tính chặn, hãy giao nó cho tokio::task::spawn_blocking, hàm này chạy nó trên một nhóm luồng riêng dành cho việc đó:

fn count_primes(limit: u64) -> usize {
    (2..limit)
        .filter(|&n| (2..).take_while(|d| d * d <= n).all(|d| n % d != 0))
        .count()
}

#[tokio::main]
async fn main() {
    let handle = tokio::task::spawn_blocking(|| count_primes(50_000));

    println!("vẫn trả lời các yêu cầu trong lúc đếm...");

    let primes = handle.await.unwrap();
    println!("có {primes} số nguyên tố nhỏ hơn 50000");
}

Kết quả:

vẫn trả lời các yêu cầu trong lúc đếm...
có 5133 số nguyên tố nhỏ hơn 50000

▶ Tự chạy thử

Giữ guard của std::sync::Mutex qua .await. Đoạn này trông vô hại:

let handle = tokio::spawn(async move {
    let mut count = visits.lock().unwrap();
    *count += 1;
    save_to_disk(*count).await;
});

error: future cannot be sent between threads safely
   |
   = help: within `{async block@src/main.rs:13:31: 13:41}`, the trait `Send` is not implemented for `std::sync::MutexGuard<'_, u32>`
note: future is not `Send` as this value is used across an await
   |
14 |         let mut count = visits.lock().unwrap();
   |             --------- has type `std::sync::MutexGuard<'_, u32>` which is not `Send`
15 |         *count += 1;
16 |         save_to_disk(*count).await;
   |                              ^^^^^ await occurs here, with `mut count` maybe used later

Thông báo nói rằng MutexGuard không phải Send và lại được dùng qua một .await. Trình biên dịch đang bảo vệ bạn. Trong lúc task này ngủ ở .await, nó vẫn giữ khóa, và bất kỳ task nào khác muốn lấy khóa sẽ chặn cả luồng của nó. Cách sửa là drop guard trước .await, bằng cách đặt phần khóa vào một khối riêng:

let count = {
    let mut count = visits.lock().unwrap();
    *count += 1;
    *count
}; // guard bị drop ở đây
save_to_disk(count).await;

Nếu bạn thật sự cần giữ khóa trong lúc await, hãy dùng tokio::sync::Mutex, với lock().await chờ mà không chặn luồng. (Để biết thêm về khóa, xem Trạng thái dùng chung.)

Quên .await. Future là lười, nên save_file(); mà không có .await thì không làm gì cả. Trình biên dịch cảnh báo bạn bằng câu “futures do nothing unless you .await or poll them”. Xem Async/Await để biết đầy đủ câu chuyện, và đừng bỏ qua cảnh báo đó.

Spawn một task rồi không bao giờ chờ nó. Khi main trả về, runtime tắt và mọi task chưa xong đều bị drop:

use std::time::Duration;
use tokio::time::sleep;

#[tokio::main]
async fn main() {
    tokio::spawn(async {
        sleep(Duration::from_millis(100)).await;
        println!("đã lưu báo cáo"); // không bao giờ được in!
    });

    println!("main đã xong");
}

Kết quả:

main đã xong

▶ Tự chạy thử

Báo cáo không bao giờ được lưu. Nếu công việc quan trọng, hãy giữ JoinHandle và .await nó (hoặc đặt các task vào một JoinSet và chờ hết) trước khi main kết thúc.

Luyện tập

1. Hỏi giá có thời hạn

Bốn cửa hàng trả lời yêu cầu hỏi giá sau 20, 40, 300 và 60 mili giây. Hỏi cả bốn cùng lúc bằng một JoinSet, nhưng chỉ cho mỗi cửa hàng 150ms. In ra các giá đã đến (sắp xếp theo tên cửa hàng), rồi in số cửa hàng bị quá giờ. Gợi ý: spawn timeout(limit, ask_shop(name, delay)) vào set. Mỗi task xong sẽ cho bạn một Ok hoặc một Err(Elapsed), nên hãy trả về tên cửa hàng cùng với kết quả đó. (Tên cửa hàng được giữ bằng tiếng Anh vì kết quả được sắp xếp theo chữ cái.)

Xem lời giải

use std::time::Duration;
use tokio::task::JoinSet;
use tokio::time::{sleep, timeout};

async fn ask_shop(delay_ms: u64) -> u32 {
    sleep(Duration::from_millis(delay_ms)).await;
    100 + delay_ms as u32 // một mức giá bịa ra
}

#[tokio::main]
async fn main() {
    let shops = [("Corner Shop", 20), ("MegaMart", 40), ("SlowCo", 300), ("Bits & Bobs", 60)];
    let limit = Duration::from_millis(150);

    let mut set = JoinSet::new();
    for (name, delay) in shops {
        set.spawn(async move { (name, timeout(limit, ask_shop(delay)).await) });
    }

    let mut prices = Vec::new();
    let mut timed_out = 0;
    while let Some(res) = set.join_next().await {
        match res.unwrap() {
            (name, Ok(price)) => prices.push((name, price)),
            (_, Err(_)) => timed_out += 1,
        }
    }

    prices.sort();
    for (name, price) in prices {
        println!("{name}: {price}");
    }
    println!("{timed_out} cửa hàng bị quá giờ");
}

Kết quả:

Bits & Bobs: 160
Corner Shop: 120
MegaMart: 140
1 cửa hàng bị quá giờ

▶ Tự chạy thử

Mọi cửa hàng cùng chờ một lúc, nên toàn bộ việc kiểm tra mất khoảng 150ms, bất kể SlowCo chậm tới đâu. Khi timeout của nó kích hoạt, future của nó bị drop, nên nó không thể tiếp tục chạy ngầm.

2. Kích thước thư mục với đệ quy async

Một Folder có tên, danh sách kích thước tệp tính bằng KB, và danh sách thư mục con. Viết async fn total_size(folder: &Folder) -> u64 để cộng các tệp của chính nó với kích thước của mọi thư mục con, bằng cách gọi lại chính nó cho từng thư mục con. Giả vờ rằng mỗi thư mục cần một lần “đọc đĩa” async ngắn trước, bằng tokio::task::yield_now().await. Gợi ý: lời gọi đệ quy cần được bọc trong Box::pin(...), nếu không bạn sẽ gặp lỗi E0733.

Xem lời giải

struct Folder {
    name: &'static str,
    files_kb: Vec<u64>,
    subfolders: Vec<Folder>,
}

async fn total_size(folder: &Folder) -> u64 {
    tokio::task::yield_now().await; // giả vờ đọc đĩa

    let mut total: u64 = folder.files_kb.iter().sum();
    for sub in &folder.subfolders {
        total += Box::pin(total_size(sub)).await;
    }
    println!("{}: {} KB", folder.name, total);
    total
}

#[tokio::main]
async fn main() {
    let home = Folder {
        name: "home",
        files_kb: vec![4],
        subfolders: vec![
            Folder { name: "photos", files_kb: vec![2048, 1024], subfolders: vec![] },
            Folder {
                name: "code",
                files_kb: vec![12, 8],
                subfolders: vec![Folder { name: "target", files_kb: vec![500], subfolders: vec![] }],
            },
        ],
    };

    let total = total_size(&home).await;
    println!("tổng: {total} KB");
}

Kết quả:

photos: 3072 KB
target: 500 KB
code: 520 KB
home: 3596 KB
tổng: 3596 KB

▶ Tự chạy thử

Mỗi thư mục được in sau tất cả thư mục con của nó, vì nó phải chờ kích thước của chúng trước. Box::pin đặt future của mỗi lời gọi lồng nhau lên heap, nên future bên ngoài giữ được kích thước cố định.

3. Làm hết hàng đợi rồi mới dừng

Sửa ví dụ tín hiệu tắt sao cho không cần sleep(100ms) trước khi gửi tín hiệu dừng. Gửi ba việc và tín hiệu dừng ngay lập tức, nhưng đảm bảo worker vẫn làm xong mọi việc trong hàng đợi trước khi dừng. Gợi ý: nếu không có gì giúp, select! chọn ngẫu nhiên một nhánh khi cả hai cùng sẵn sàng. Hãy thêm biased; và đặt nhánh công việc lên đầu.

Xem lời giải

use tokio::sync::{mpsc, oneshot};

#[tokio::main]
async fn main() {
    let (job_tx, mut job_rx) = mpsc::channel::<String>(10);
    let (stop_tx, mut stop_rx) = oneshot::channel::<()>();

    for job in ["thay đổi kích thước ảnh", "gửi email", "cập nhật chỉ mục tìm kiếm"] {
        job_tx.send(job.to_string()).await.unwrap();
    }
    stop_tx.send(()).unwrap();

    let worker = tokio::spawn(async move {
        let mut done = 0;
        loop {
            tokio::select! {
                biased;
                Some(job) = job_rx.recv() => {
                    println!("đang làm: {job}");
                    done += 1;
                }
                _ = &mut stop_rx => {
                    println!("hàng đợi đã trống, dừng lại");
                    break;
                }
            }
        }
        done
    });

    let done = worker.await.unwrap();
    println!("worker đã xong {done} việc");
}

Kết quả:

đang làm: thay đổi kích thước ảnh
đang làm: gửi email
đang làm: cập nhật chỉ mục tìm kiếm
hàng đợi đã trống, dừng lại
worker đã xong 3 việc

▶ Tự chạy thử

Với biased;, select! luôn thử nhánh công việc trước, nên nó chỉ nhìn tới tín hiệu dừng khi recv() không còn gì sẵn sàng. Không có nó, worker có thể dừng sau một hoặc hai việc, một lỗi chỉ thỉnh thoảng mới xuất hiện.

Tóm tắt

• join! chờ mọi future, try_join! dừng ở Err đầu tiên, và JoinSet chạy nhiều task đã spawn và trả lại kết quả khi chúng xong (hãy sắp xếp nếu thứ tự quan trọng).
• select! chờ future đầu tiên xong. Các future khác bị drop, đó là cách việc hủy hoạt động trong async Rust. Bên trong một vòng lặp, hãy dùng future cancel-safe và biased; khi thứ tự quan trọng.
• timeout bỏ cuộc với một future chậm. interval tick theo lịch đều đặn, sleep chờ một lần.
• Stream là một iterator async. Đầu nhận của channel hoạt động giống một stream, và tokio-stream thêm map, take cùng nhiều phương thức khác.
• Pin hứa rằng một future sẽ không di chuyển trong bộ nhớ. Dùng pin! để poll một future nhiều lần và Box::pin cho future được lưu trữ và đệ quy async.
• Đừng chặn runtime (spawn_blocking), đừng giữ khóa của std qua .await, và hãy await những task bạn đã spawn.

Kiểm tra nhanh

Kiểm tra nhanh

1. Trong tokio::select!, điều gì xảy ra với các nhánh không xong trước?

Chúng tiếp tục chạy ngầm Chúng bị drop, nên dừng lại và dọn dẹp Chúng chạy lại ở vòng lặp tiếp theo

2. tokio::time::timeout trả về gì khi hết giờ?

Nó panic Ok với một giá trị mặc định Err(Elapsed), và future bên trong bị drop

3. Vì sao bạn cần Box::pin khi một async fn gọi lại chính nó?

Một future chứa chính nó sẽ có kích thước vô hạn Đệ quy hoàn toàn không được phép trong code async Để hàm chạy trên một luồng khác

4. Bạn phải làm 2 giây việc nặng về CPU bên trong một chương trình tokio. Công cụ tốt nhất là gì?

Bọc std::thread::sleep quanh nó tokio::select! Một JoinSet với tokio::spawn tokio::task::spawn_blocking

Điểm: 0 / 4

Đọc thêm

• Select — the Tokio tutorial (tiếng Anh): chi tiết về select!: các nhánh hoạt động ra sao, hủy nghĩa là gì, và dùng nó trong vòng lặp.
• Streams — the Tokio tutorial (tiếng Anh): cách dùng, biến đổi và xây dựng stream với tokio-stream.
• A Closer Look at the Traits for Async — The Rust Book (tiếng Anh): cái nhìn nhẹ nhàng, từng bước về Future, Pin, Unpin và Stream.
• std::pin — thư viện chuẩn Rust (tiếng Anh): lời giải thích chính thức về pinning, Pin, Unpin và macro pin!.
• tokio::time trên docs.rs (tiếng Anh): sleep, timeout, interval và cách xử lý các tick bị lỡ.