Rust 2026 经验谈 - async/await 底层机制

1
use std::future::Future;
2
use std::pin::Pin;
3
use std::task::{Context, Poll};
4

5
struct TimerFuture {
6
    shared_state: Arc<Mutex<TimerSharedState>>,
7
}
8

9
struct TimerSharedState {
10
    completed: bool,
11
    waker: Option<Waker>,
12
}
13

14
impl Future for TimerFuture {
15
    type Output = ();
16

17
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
18
        let mut shared_state = self.shared_state.lock().unwrap();
19

20
        if shared_state.completed {
21
            Poll::Ready(())
22
        } else {
23
            // 注册 waker：定时器触发时会调用 waker.wake()
24
            shared_state.waker = Some(cx.waker().clone());
25
            Poll::Pending
26
        }
27
    }
28
}

1
// 实际案例：把 C 的 IO 完成回调包装成 Future
2
struct IoFuture {
3
    fd: RawFd,
4
    registered: bool,
5
}
6

7
impl Future for IoFuture {
8
    type Output = io::Result<Vec<u8>>;
9

10
    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
11
        if !self.registered {
12
            // 把 cx.waker() 注册到 epoll/io_uring
13
            EPOLL.register(self.fd, cx.waker().clone());
14
            self.registered = true;
15
            return Poll::Pending;
16
        }
17

18
        match EPOLL.try_read(self.fd) {
19
            Ok(data) => Poll::Ready(Ok(data)),
20
            Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
21
                // 还没就绪，更新 waker（可能 Context 变了）
22
                EPOLL.update_waker(self.fd, cx.waker().clone());
23
                Poll::Pending
24
            }
25
            Err(e) => Poll::Ready(Err(e)),
26
        }
27
    }
28
}

1
enum ReadAndProcessFuture {
2
    State0,                          // 初始，准备调用 read_socket
3
    State1(Data),                    // read_socket 完成，准备 process
4
    State2(Result),                  // process 完成，准备 write
5
    State3,                          // 完成（或用 Ready 终结）
6
}
7

8
impl Future for ReadAndProcessFuture {
9
    type Output = Result<()>;
10

11
    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
12
        loop {
13
            match self.as_mut().get_mut() {
14
                State0 => {
15
                    match read_socket().as_mut().poll(cx) {
16
                        Poll::Ready(data) => *self = State1(data),
17
                        Poll::Pending => return Poll::Pending,
18
                    }
19
                }
20
                State1(data) => {
21
                    match process(data).as_mut().poll(cx) {
22
                        Poll::Ready(result) => *self = State2(result),
23
                        Poll::Pending => return Poll::Pending,
24
                    }
25
                }
26
                State2(result) => {
27
                    match write_response(result).as_mut().poll(cx) {
28
                        Poll::Ready(Ok(())) => return Poll::Ready(Ok(())),
29
                        Poll::Ready(Err(e)) => return Poll::Ready(Err(e)),
30
                        Poll::Pending => return Poll::Pending,
31
                    }
32
                }
33
                _ => unreachable!(),
34
            }
35
        }
36
    }
37
}

1
async fn simple() {
2
    let a = 1;
3
    let b = 2;
4
    println!("{}", a + b);
5
}
6

7
async fn with_await() {
8
    let a = expensive_computation();  // a 必须跨 .await 保存
9
    sleep(Duration::from_secs(1)).await;
10
    println!("{}", a);  // 使用保存的 a
11
}
12

13
fn main() {
14
    let f1 = simple();
15
    println!("simple: {} bytes", std::mem::size_of_val(&f1));  // 0 或很小
16
    let f2 = with_await();
17
    println!("with_await: {} bytes", std::mem::size_of_val(&f2));  // 包含 a + 状态 + waker
18
}

1
// 反模式：CPU 密集型工作不 yield，阻塞整个线程
2
async fn cpu_intensive() {
3
    let mut sum = 0u64;
4
    for i in 0..1_000_000_000 {
5
        sum += i;  // 没有 .await，不会让出！
6
    }
7
}
8

9
// 正确：用 spawn_blocking 或手动 yield
10
async fn cpu_intensive() {
11
    tokio::task::spawn_blocking(move || {
12
        let mut sum = 0u64;
13
        for i in 0..1_000_000_000 {
14
            sum += i;
15
        }
16
        sum
17
    }).await
18
}