Bài 254: tokio::time::sleep, interval, timeout

Sau bài học, bạn sẽ:

• Dùng được tokio::time::sleep(Duration) để delay async không block worker thread.
• Hiểu vì sao std::thread::sleep trong async context là bug nghiêm trọng và mô tả được tác hại lên multi-thread runtime.
• Dựng được loop tick định kỳ bằng tokio::time::interval với interval.tick().await.
• Chọn được MissedTickBehavior phù hợp giữa Burst (mặc định), Delay và Skip tuỳ workload.
• Bọc một future bằng tokio::time::timeout(Duration, fut) và xử lý Err(Elapsed) khi vượt deadline.
• Viết được pattern retry with exponential backoff kết hợp sleep + timeout.

Primitive đơn giản nhất: tokio::time::sleep trả về một Future hoàn tất sau khoảng thời gian truyền vào. Khi task .await nó, task được park, executor được tự do chạy task khác. Đến khi timer reactor báo deadline tới, task được wake và chạy tiếp.

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

#[tokio::main]
async fn main() {
    println!("bắt đầu");
    sleep(Duration::from_secs(1)).await;     // yield 1 giây
    println!("sau 1 giây");
    sleep(Duration::from_millis(500)).await; // yield 500ms
    println!("sau 1.5 giây");
}

Vài tính chất quan trọng:

• Không block thread. Trong lúc task này ngủ, worker thread của runtime đi chạy task khác — đó là cốt lõi của async.
• Không chính xác tuyệt đối. sleep(1s) nghĩa là "ít nhất 1 giây", có thể dài hơn vài ms tuỳ load và scheduler. Đừng dùng cho timing tới microsecond.
• Future cancellable. Drop future trước khi nó hoàn tất là an toàn — đây là tiền đề cho timeout và select!.
• Cần runtime tokio. Gọi sleep ngoài runtime tokio sẽ panic vì không có timer driver — đây là biểu hiện của tách runtime ra ngoài std đã nói ở Bài 245.

Có hai biến thể đáng nhớ: tokio::time::sleep_until(Instant) ngủ tới một thời điểm tuyệt đối (hữu ích khi cần đồng bộ nhiều task tới cùng deadline), và bản tương đương trong async là tokio::task::yield_now().await khi chỉ muốn "nhường lượt" mà không thật sự chờ thời gian.

Đây là lỗi kinh điển của người mới viết async Rust: thấy std::thread::sleep(Duration::from_secs(1)) trong code sync rồi copy vào async fn. Compile được, chạy được — nhưng phá runtime.

// SAI — std::thread::sleep block worker thread
#[tokio::main(worker_threads = 4)]
async fn main() {
    for i in 0..8 {
        tokio::spawn(async move {
            std::thread::sleep(Duration::from_secs(1)); // BLOCK!
            println!("task {} done", i);
        });
    }
    tokio::time::sleep(Duration::from_secs(3)).await;
}

Lý do: std::thread::sleep không phải async — nó gọi syscall nanosleep trên OS thread đang chạy. Mà với multi-thread runtime, thread đó là worker thread của tokio. Sleep nó nghĩa là cả worker bị đóng băng, không poll được task khác trong suốt 1 giây đó.

Worker pool có 4 thread, 8 task cùng spawn:

  Worker-1: [T0 sleep 1s ........] [T4 sleep 1s ........] ...
  Worker-2: [T1 sleep 1s ........] [T5 sleep 1s ........] ...
  Worker-3: [T2 sleep 1s ........] [T6 sleep 1s ........] ...
  Worker-4: [T3 sleep 1s ........] [T7 sleep 1s ........] ...

  -> Tổng thời gian: ~2 giây cho 8 task (serial 2 đợt vì worker bị chặn).
     Đúng ra với tokio::time::sleep, cả 8 task chạy concurrent xong sau ~1s.

Nguy hiểm hơn: nếu bạn dùng #[tokio::main(flavor = "current_thread")] (1 worker duy nhất), một std::thread::sleep sẽ đóng băng toàn bộ runtime. Mọi task khác — kể cả I/O — đều dừng.

Quy tắc: trong async context luôn dùng tokio::time::sleep. Nếu thật sự phải gọi code blocking (CPU-bound hoặc API sync chưa wrap), wrap nó bằng tokio::task::spawn_blocking để chạy trên blocking thread pool riêng — sẽ học ở bài sau.

Khi cần chạy việc gì đó định kỳ — health check mỗi 5 giây, flush metric mỗi 30 giây, refresh token mỗi 10 phút — đừng dùng loop { sleep(period).await; do_work().await; }. Cách đó đúng đắn ở mức cơ bản nhưng drift: nếu do_work mất 800ms, chu kỳ thực bị giãn thành period + 800ms.

Tokio cung cấp tokio::time::interval(period) — một stream sinh ra "tick" cách đều nhau theo lịch tuyệt đối, không cộng dồn thời gian xử lý:

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

#[tokio::main]
async fn main() {
    let mut ticker = interval(Duration::from_secs(5));

    loop {
        ticker.tick().await;            // chờ tới tick tiếp theo
        println!("chạy health check");
        // do_work().await; // ví dụ 800ms; chu kỳ vẫn ~5s
    }
}

Quy tắc nhớ: tick đầu tiên không chờ — gọi ticker.tick().await lần đầu hoàn tất ngay lập tức. Đây là behavior có chủ đích để job chạy ngay khi service start, sau đó mới đợi đủ period cho lần sau. Nếu muốn đợi period trước lần đầu, dùng interval_at(Instant::now() + period, period).

So với sleep trong loop: interval nhớ thời điểm tick tiếp theo lẽ ra phải xảy ra, không phải "ngủ thêm period kể từ bây giờ". Đó là khác biệt then chốt khi job có độ trễ biến động.

Vấn đề "lỡ tick" xảy ra khi do_work lâu hơn period. Ví dụ period 1 giây nhưng một lần xử lý mất 3 giây — đã có 2 tick bị bỏ lỡ trong lúc xử lý. Tokio cho bạn chọn cách reactor xử lý tình huống này qua enum MissedTickBehavior:

• Burst (mặc định) — bù ngay các tick lỡ. Sau khi do_work 3 giây xong, ba lần tick().await kế tiếp đều ready ngay. Phù hợp khi mỗi tick là một việc không thể bỏ (vd: gom batch).
• Delay — bỏ qua các tick lỡ, lên lịch tick tiếp theo cách thời điểm hiện tại đúng period. Phù hợp khi không cần bù (vd: refresh cache, polling, log).
• Skip — bỏ qua các tick lỡ, lên lịch tick tiếp theo theo lịch tuyệt đối kế tiếp. Phù hợp cron-like job cần "rơi đúng phút chẵn".

use tokio::time::{interval, Duration, MissedTickBehavior};

#[tokio::main]
async fn main() {
    let mut ticker = interval(Duration::from_secs(1));
    ticker.set_missed_tick_behavior(MissedTickBehavior::Delay);

    loop {
        ticker.tick().await;
        println!("refresh cache");
        // Giả sử do_work mất 3s — với Delay: chỉ tick lại sau 1s nữa,
        // không bị "đuổi tick" gây spam ngay sau khi xong.
        slow_work().await;
    }
}

async fn slow_work() {
    tokio::time::sleep(Duration::from_secs(3)).await;
}

Pitfall thường gặp: mặc định là Burst. Nếu bạn viết health checker mỗi 1 giây nhưng đôi khi job kéo 10 giây, mặc định Burst sẽ phun 10 tick liên tiếp ngay sau đó, làm load tăng đột biến. Trừ khi bạn thực sự muốn behavior đó, hãy set_missed_tick_behavior(MissedTickBehavior::Delay) ngay sau khi tạo.

Async I/O có thể treo vô hạn nếu server bên kia chậm hoặc đứt mạng. Bạn cần "cấp cho task này tối đa X giây, quá hạn thì huỷ". Tokio có tokio::time::timeout(duration, future) trả về Result<T, Elapsed>:

use std::time::Duration;
use tokio::time::{timeout, error::Elapsed};

async fn fetch_user(id: u64) -> Result<String, Elapsed> {
    let result = timeout(Duration::from_secs(5), call_slow_api(id)).await?;
    Ok(result)
}

async fn call_slow_api(_id: u64) -> String {
    tokio::time::sleep(Duration::from_secs(10)).await; // mô phỏng slow API
    String::from("user data")
}

#[tokio::main]
async fn main() {
    match fetch_user(42).await {
        Ok(data) => println!("OK: {}", data),
        Err(_elapsed) => eprintln!("request timed out sau 5s"),
    }
}

Cơ chế: timeout tạo một "race" giữa future của bạn và một timer 5 giây. Cái nào hoàn tất trước thắng. Nếu future thắng — trả Ok(value). Nếu timer thắng — future bị drop, trả Err(Elapsed).

Điểm quan trọng: future bị drop là cách huỷ trong async Rust. Khi call_slow_api bị drop, tất cả tài nguyên giữ trong nó (TCP socket, file handle, lock guard) sẽ được dọn dẹp qua Drop theo đúng nguyên tắc RAII. Nhưng không phải mọi future đều an toàn khi bị drop giữa chừng (cancellation-safety) — chủ đề sẽ học sâu ở bài về select!.

Có biến thể timeout_at(Instant, future) nhận deadline tuyệt đối — hữu ích khi truyền deadline qua nhiều layer function để toàn bộ cây call cùng hết hạn ở một mốc thời gian cố định.

Ba primitive đứng riêng đã hữu ích, nhưng giá trị thực sự đến khi kết hợp. Pattern phổ biến nhất trong production: gọi API ngoài, nếu lỗi mạng/quá tải thì retry với delay tăng dần (exponential backoff) và mỗi lần đều có timeout riêng.

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

async fn fetch_with_retry<F, Fut, T, E>(
    mut op: F,
    max_attempts: u32,
    per_call_timeout: Duration,
) -> Result<T, String>
where
    F: FnMut() -> Fut,
    Fut: std::future::Future<Output = Result<T, E>>,
    E: std::fmt::Display,
{
    let mut delay = Duration::from_millis(200);

    for attempt in 1..=max_attempts {
        match timeout(per_call_timeout, op()).await {
            Ok(Ok(value)) => return Ok(value),
            Ok(Err(e)) => eprintln!("attempt {attempt} lỗi: {e}"),
            Err(_) => eprintln!("attempt {attempt} timeout"),
        }

        if attempt < max_attempts {
            sleep(delay).await;
            delay = (delay * 2).min(Duration::from_secs(10)); // cap 10s
        }
    }

    Err(format!("thất bại sau {max_attempts} lần thử"))
}

#[tokio::main]
async fn main() {
    let result = fetch_with_retry(
        || async { reqwest::get("https://api.example.com/health").await },
        5,
        Duration::from_secs(3),
    )
    .await;

    println!("kết quả: {:?}", result.is_ok());
}

Pattern này gói gọn ba bài học: (1) timeout chặn mỗi attempt không treo vô hạn; (2) sleep tạo backoff giữa các attempt — quan trọng để giảm áp lực lên server bên kia khi nó đang quá tải; (3) cap delay (ở đây 10 giây) tránh exponential tăng mãi. Delay 200ms → 400 → 800 → 1.6s → 3.2s — đủ để hệ thống "thở" giữa các lần thử.

Trong production, thường thêm jitter (cộng random vào delay) để tránh "thundering herd" khi hàng nghìn client cùng retry cùng lúc sau một outage. Crate tokio-retry hoặc backoff đóng gói sẵn các strategy này nếu bạn không muốn tự viết.

• tokio::time::sleep(Duration) yield task hiện tại, không block worker — executor đi chạy task khác trong lúc chờ.
• std::thread::sleep trong async context là bug nghiêm trọng: block toàn bộ worker thread, có thể giảm throughput dữ dội hoặc đóng băng runtime current_thread.
• tokio::time::interval(period) sinh tick định kỳ theo lịch tuyệt đối, không drift theo độ trễ xử lý như loop { sleep; }. Tick đầu tiên không chờ.
• MissedTickBehavior ba lựa chọn: Burst (mặc định, bù tick lỡ), Delay (bỏ qua, đợi period từ now), Skip (bỏ qua, theo lịch tuyệt đối). Mặc định Burst hay gây spam — đặt Delay cho job thông thường.
• tokio::time::timeout(Duration, fut) trả về Result<T, Elapsed>; future bị drop khi quá hạn, dọn tài nguyên qua RAII. Có biến thể timeout_at(Instant, fut) cho deadline tuyệt đối.
• Pattern retry with exponential backoff: timeout chặn mỗi attempt, sleep backoff giữa attempt với delay tăng gấp đôi và cap trần. Trong prod nên thêm jitter.
• Mọi primitive trong tokio::time đều cần runtime tokio active — gọi ngoài runtime sẽ panic.

Tự trả lời, đáp án ở cuối:

1. Vì sao std::thread::sleep(Duration::from_secs(1)) trong async fn trên multi-thread runtime tokio làm giảm throughput? Tình huống nào tệ nhất (đóng băng cả runtime)?
2. Bạn viết loop { sleep(Duration::from_secs(5)).await; do_work().await; } với do_work mất 2 giây. Chu kỳ thực giữa hai lần do_work bắt đầu là bao nhiêu? Nếu muốn đúng 5 giây cố định, nên dùng gì?
3. Liệt kê ba MissedTickBehavior. Cho một health-check chạy mỗi 1 giây nhưng đôi khi mất 5 giây, behavior nào phù hợp nhất và vì sao?
4. Viết signature đúng cho hàm fetch(url) trả về dữ liệu trong tối đa 3 giây, dùng tokio::time::timeout. Khi timeout xảy ra, future bên trong có được dọn dẹp không?
5. Pattern retry exponential backoff giải quyết vấn đề gì khi gọi API ngoài? Tại sao cần cap trần delay? Tại sao trong production lại thêm jitter?
6. Vì sao tokio::time::sleep đòi hỏi đang chạy trong runtime tokio active, không như std::thread::sleep chạy được mọi nơi?

Bài 255: tokio::sync::mpsc Channel — giới thiệu channel multi-producer single-consumer của tokio: phân biệt bounded (mpsc::channel(100)) vs unbounded, tx.send().await trả backpressure khi đầy, rx.recv().await phía consumer, và pattern worker pool — nền tảng cho mọi pipeline xử lý concurrent trong tokio.