Rust Future 运行原理：从 IO Model 到 epoll 的完整链路

① 阻塞 I/O (BIO)
   线程调用 read() ──> 内核挂起线程 ──> 硬件数据到达 ──> 唤醒线程
   代价：1 个连接 = 1 个 OS 线程（几 MB 栈内存 + 上下文切换开销），无法支撑百万并发。

② 非阻塞轮询 (NIO)
   线程循环 read() ──> 没数据立刻返回 EWOULDBLOCK ──> 疯狂占用 100% CPU 空转。

③ 多路复用 (I/O Multiplexing: epoll/kqueue) 【就绪通知模型（Readiness）】
   线程向内核注册 10 万个 fd ──> 线程调用 epoll_wait 睡眠 
   ──> 哪个 fd 就绪内核就唤醒线程 ──> 线程主动调用 read(fd) 提取数据。
   ★ 特点：由应用程序自己主动“拉取（Pull）”数据。

④ 异步完成 I/O (AIO: io_uring / Windows IOCP) 【完成通知模型（Completion）】
   应用程序提前准备一块内存 buffer 借给内核 ──> 内核在后台把数据填满 
   ──> 完成后通知应用程序“已经读好了”。
   ★ 特点：内核完成后主动“推送（Push）”数据。

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

pub enum Poll<T> {
    Ready(T),
    Pending,
}

传统阻塞 I/O：   线程调用 read()  ──> 内核挂起线程 ──> 一个线程同时只能死等一个 I/O
Future 拉式模型： 任务调用 poll() ──> 没数据返 Pending ──> 线程立刻去跑别的任务
                                                         └── 数据到了通过 Waker 把任务重新放回就绪队列

async fn foo() -> i32 {
    let a = bar().await;       // 断点 1
    let b = baz(a).await;      // 断点 2
    a + b
}

enum FooFuture {
    Start,                                          // 初始状态
    State1 { fut: BarFuture },                      // 停在 await bar()
    State2 { a: i32, fut: BazFuture },              // 停在 await baz(a)，注意 a 被内嵌了
    Done,
}

impl Future for FooFuture {
    type Output = i32;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<i32> {
        let this = unsafe { self.get_unchecked_mut() };
        loop {
            match this {
                FooFuture::Start => {
                    *this = FooFuture::State1 { fut: bar() };
                    // 不 return，继续 loop，立即推进下一阶段
                }
                FooFuture::State1 { fut } => {
                    match Pin::new_unchecked(fut).poll(cx) {
                        Poll::Pending => return Poll::Pending,
                        Poll::Ready(a) => {
                            *this = FooFuture::State2 { a, fut: baz(a) };
                            // 不 return，继续 loop，直接进入 State2
                        }
                    }
                }
                FooFuture::State2 { a, fut } => {
                    match Pin::new_unchecked(fut).poll(cx) {
                        Poll::Pending => return Poll::Pending,
                        Poll::Ready(b) => {
                            let res = *a + b;
                            *this = FooFuture::Done;
                            return Poll::Ready(res);
                        }
                    }
                }
                FooFuture::Done => panic!("poll after completion"),
            }
        }
    }
}

let mut x = 1;
let r = &x;
println!("{r}");

async fn f() {
    let mut x = 1;
    let r = &x;            // r 引用了局部变量 x
    something().await;     // 挂起点
    println!("{r}");       // 恢复后继续使用 r
}

struct FFuture {
    x: i32,
    r: *const i32,   // 物理上指向本结构体内部的 x 字段！
}

移动前：
[ 地址 0x1000: x = 1 ] <─── 指向 0x1000 ─── [ 地址 0x1008: r ]

移动后（搬到了新地址 0x2000）：
[ 地址 0x2000: x = 1 ]                       [ 地址 0x2008: r = 0x1000 (悬挂野指针！) ]

pub struct Pin<P> {
    pointer: P,   // 内部就包着一个指针（如 &mut T 或 Box<T>）
}

┌─────────────────────────────────────────────────────────┐
│  Executor (调度环)                                      │
│                                                         │
│   loop {                                                │
│     task = ready_queue.pop_front()  ◄─────────────┐     │
│     waker = Waker::from(task)                     │     │
│     match task.future.poll(waker) {               │     │
│       Ready   => 任务结束，释放资源                 │     │
│       Pending => 丢弃（绝不自动重新入队，等 wake 通知）│     │
│     }                                             │     │
│   }                                               │     │
└───────────────────────────────────────────────────┼─────┘
                                                    │
                                          wake() 时将 Task push_back
                                                    │
┌───────────────────────────────────────────────────┴─────┐
│  Waker 内部指针 (Arc<Task>)                             │
│                                                         │
│  impl Wake for Task {                                   │
│    fn wake(self: Arc<Self>) {                           │
│        self.queue.push_back(self.clone());              │
│    }                                                    │
│  }                                                      │
└─────────────────────────────────────────────────────────┘

use std::collections::VecDeque;
use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll, Wake, Waker};
use std::thread;
use std::time::{Duration, Instant};

// ========== 1. 防御虚假唤醒的安全 Delay Future ==========
struct SafeDelay {
    deadline: Instant,
    started: bool,
}

impl SafeDelay {
    fn new(dur: Duration) -> Self {
        Self {
            deadline: Instant::now() + dur,
            started: false,
        }
    }
}

impl Future for SafeDelay {
    type Output = &'static str;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        // ★ 核心检查：时间真的到了吗？严格防范协作任务触发的虚假唤醒
        if Instant::now() >= self.deadline {
            return Poll::Ready("pong");
        }

        // 仅在首次 poll 时起后台线程模拟硬件中断
        if !self.started {
            self.started = true;
            let waker = cx.waker().clone();
            let sleep_time = self.deadline.saturating_duration_since(Instant::now());
            thread::spawn(move || {
                thread::sleep(sleep_time);
                println!("  [硬件中断/内核通知] 定时器时间到，调用 waker.wake()");
                waker.wake();
            });
        }

        println!("  [poll] 硬件未就绪，返回 Pending");
        Poll::Pending
    }
}

// ========== 2. Task：将 Future 绑定调度与唤醒能力 ==========
struct Task {
    future: Mutex<Pin<Box<dyn Future<Output = ()> + Send>>>,
    queue: Arc<Mutex<VecDeque<Arc<Task>>>>,
}

impl Wake for Task {
    fn wake(self: Arc<Self>) {
        println!("  [wake] Task 被唤醒，放回 Executor 就绪队列");
        self.queue.lock().unwrap().push_back(self.clone());
    }
}

// ========== 3. Executor：纯粹的就绪队列拉取环 ==========
fn run_executor(queue: Arc<Mutex<VecDeque<Arc<Task>>>>) {
    loop {
        let task = queue.lock().unwrap().pop_front();
        let Some(task) = task else { break };

        let waker = Waker::from(task.clone());
        let mut cx = Context::from_waker(&waker);

        let mut fut = task.future.lock().unwrap();
        match fut.as_mut().poll(&mut cx) {
            Poll::Ready(()) => println!("  [executor] Task 彻底完成"),
            Poll::Pending => println!("  [executor] Task 挂起，出让线程跑其他任务"),
        }
    }
}

// ========== 4. 驱动主流程 ==========
fn main() {
    let queue = Arc::new(Mutex::new(VecDeque::new()));

    let future = Box::pin(async {
        println!("  [future] 开始执行");
        let res = SafeDelay::new(Duration::from_millis(100)).await;
        println!("  [future] 收到结果: {res}");
    });

    let task = Arc::new(Task {
        future: Mutex::new(future),
        queue: queue.clone(),
    });
    queue.lock().unwrap().push_back(task);

    println!("执行器启动");
    run_executor(queue);
    println!("执行器退出");
}

执行器启动
  [future] 开始执行
  [poll] 硬件未就绪，返回 Pending
  [executor] Task 挂起，出让线程跑其他任务
  [硬件中断/内核通知] 定时器时间到，调用 waker.wake()
  [wake] Task 被唤醒，放回 Executor 就绪队列
  [future] 收到结果: pong
  [executor] Task 彻底完成
执行器退出

【环 1：Executor 调度环（Worker 线程池）】       【环 2：Reactor 驱动环（Driver 线程）】
     从就绪队列取 Task ──> 调用 poll()                 循环调用 epoll_wait(timeout) 
            ▲                                                  │
            │                                                  ▼
            └──────── Waker::wake() ◄── epoll 就绪唤醒触发对应的 ScheduledIo 槽位

// Tokio 内部的读轮询骨架
fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
    loop {
        // 1. 检查就绪标记（如果未就绪，存入当前 cx.waker() 并返回 Pending）
        ready!(self.io.poll_read_ready(cx))?;

        // 2. 尝试读取
        match self.io.get_ref().try_read(buf) {
            Ok(n) => return Poll::Ready(Ok(())),
            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
                // 3. 抹除旧就绪标记！重新回循环头，下一次迭代将安全挂起
                self.io.clear_readiness(Ready::READABLE);
            }
            Err(e) => return Poll::Ready(Err(e)),
        }
    }
}

tokio::select! {
    res = socket.read_exact(&mut buf) => { ... } // ❌ 非取消安全！
    _ = timeout => { ... }
}

// ❌ 编译报错：Future 不能跨线程发送
tokio::spawn(async {
    let rc = Rc::new(100);
    sleep(Duration::from_millis(10)).await;
    println!("{rc}");
});

① 用户代码            stream.read(&mut buf).await
   │
② 乐观尝试            直接非阻塞 recv(fd) ──> 若命中缓冲区，直接返回 Ready(n)（零开销退出）
   │ (WouldBlock)
③ 挂号登记            将当前 Task 的 Waker 存入 ScheduledIo，通过 epoll_ctl 注册监听
   │
④ 出让线程            返回 Poll::Pending，Worker 线程去执行队列中的其他就绪任务
   │
⑤ 硬件就绪            网卡接收数据，DMA 写入内存并触发硬件中断
   │
⑥ 内核置位            Linux TCP 协议栈将数据放入 socket 接收缓冲区，标记 fd 可读
   │
⑦ Reactor 唤醒        Driver 线程从 epoll_wait 醒来，解析 Token 找到对应 ScheduledIo
   │
⑧ 唤醒通知            Reactor 执行 waker.wake()，将挂起的 Task 重新推入 Executor 就绪队列
   │
⑨ 调度重入            空闲 Worker 线程取出 Task，再次调用 poll(Pin<&mut State>, cx)
   │
⑩ 真正读取            恢复到断点，再次调用 recv(fd) 成功获取数据，返回 Poll::Ready(n)