Rust async 运行时拆解：tokio 的调度器到底在干什么

// 标准库的 Future trait，简化版
trait Future {
  type Output;

  // runtime 反复调用 poll，直到返回 Ready
  fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}

enum Poll<T> {
  Ready(T),  // 完成了，拿到结果
  Pending,  // 还没好，先挂起
}

// 一个简单的 async 函数
async fn fetch_data(url: &str) -> String {
  // 第一个 .await —— 状态机的第一个挂起点
  let response = http_get(url).await;
  // 第二个 .await —— 状态机的第二个挂起点
  let body = response.text().await;
  body
}

// 编译器生成的状态机（伪代码）
enum FetchDataFuture<'a> {
  // 初始状态：还没开始
  State0 { url: &'a str },
  // 第一个 await 挂起：等 http_get 完成
  State1 { url: &'a str, http_future: HttpGetFuture<'a> },
  // 第二个 await 挂起：等 text() 完成
  State2 { response: Response, text_future: TextFuture },
  // 已完成
  Done,
}

impl<'a> Future for FetchDataFuture<'a> {
  type Output = String;

  fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<String> {
  loop {
  match self.state {
  State0 { url } => {
  // 创建 http_get future，转移到 State1
  let http_future = http_get(url);
  *self = State1 { url, http_future };
  }
  State1 { http_future, .. } => {
  // poll 内部的 future
  match http_future.poll(cx) {
  Poll::Ready(response) => {
  let text_future = response.text();
  *self = State2 { response, text_future };
  }
  Poll::Pending => return Poll::Pending,
  }
  }
  State2 { text_future, .. } => {
  match text_future.poll(cx) {
  Poll::Ready(body) => {
  *self = Done;
  return Poll::Ready(body);
  }
  Poll::Pending => return Poll::Pending,
  }
  }
  Done => panic!("poll after completion"),
  }
  }
  }
}

// Rust：创建 future 但不执行！
let fut = fetch_data("https://example.com");
// 到这里，没有任何网络请求发出
// 只有 poll 或 .await 才会推进

// Go：goroutine 立即开始执行
// go fetchData("https://example.com")
// 到这里，goroutine 已经在跑了

fn main() {
  // 这个 future 永远不会执行
  // 编译器会警告：unused future that must be used
  async { println!("hello") };

  // 你需要一个 runtime
  tokio::runtime::Runtime::new()
  .unwrap()
  .block_on(async {
  println!("这次真的跑了");
  });
}

// tokio 调度循环的简化逻辑
fn worker_run(worker: &Worker) {
  let mut tick: u32 = 0;
  loop {
  tick = tick.wrapping_add(1);

  let task = if tick % 61 == 0 {
  // 每 61 次 poll，优先检查全局队列
  // 防止全局任务饿死
  worker.global_queue.pop()
  .or_else(|| worker.lifo_slot.take())
  .or_else(|| worker.local_queue.pop())
  } else {
  // 正常路径：LIFO slot → 本地 → 全局 → steal
  worker.lifo_slot.take()
  .or_else(|| worker.local_queue.pop())
  .or_else(|| worker.global_queue.pop())
  .or_else(|| worker.steal_from_others())
  };

  match task {
  Some(task) => task.poll(),  // 执行任务
  None => worker.park(),  // 没活干了，休眠
  }
  }
}

// 偷取逻辑的简化版
fn steal_from_others(&self) -> Option<Task> {
  let num_workers = self.workers.len();
  // 从随机位置开始遍历，避免所有空闲线程都去偷同一个
  let start = random::<usize>() % num_workers;

  for i in 0..num_workers {
  let target = (start + i) % num_workers;
  if target == self.index {
  continue; // 不偷自己
  }

  // 关键：偷一半！不是偷一个
  // 这样可以减少偷取次数，摊薄同步开销
  if let Some(tasks) = self.workers[target].local_queue.steal_half() {
  return Some(tasks);
  }
  }
  None
}

// 多线程模式（默认）：N 个 Worker 线程 + work-stealing
// 适合：生产环境、CPU 密集型混合 I/O 的场景
let rt = tokio::runtime::Builder::new_multi_thread()
  .worker_threads(4)  // 默认等于 CPU 核心数
  .build()
  .unwrap();

// 单线程模式：所有任务在一个线程上协作调度
// 适合：测试、简单脚本、确定性调试
let rt = tokio::runtime::Builder::new_current_thread()
  .enable_all()
  .build()
  .unwrap();

fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;

// 标准库中的 Context，简化版
pub struct Context<'a> {
  waker: &'a Waker,
  // Rust 1.82+ 还有 local_waker 等扩展
}

impl<'a> Context<'a> {
  pub fn waker(&self) -> &'a Waker {
  self.waker
  }
}

// tokio 的 waker 实现（简化）
impl Wake for TaskHandle {
  fn wake(self) {
  // 标记 task 为"需要 poll"
  self.header.state.transition_to_notified();

  // 把 task 放回调度队列
  // 优先放回创建它的 worker 的本地队列
  // 如果放不进去（满了或者跨线程），放全局队列
  if let Some(worker) = self.bound_worker() {
  worker.local_queue.push(self.task);
  } else {
  global_queue.push(self.task);
  }

  // 如果对应的 worker 正在休眠，叫醒它
  self.scheduler.notify();
  }
}

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

// 一个极简的"运行时"：就是一个任务队列
struct MiniRuntime {
  // 待执行的任务列表
  queue: Mutex<VecDeque<Arc<Task>>>,
}

struct Task {
  // 被 Pin 住的 Future，类型擦除
  future: Mutex<Pin<Box<dyn Future<Output = ()> + Send>>>,
  // 持有运行时的引用，wake 时把自己重新放回队列
  runtime: Arc<MiniRuntime>,
}

impl MiniRuntime {
  fn new() -> Arc<Self> {
  Arc::new(Self {
  queue: Mutex::new(VecDeque::new()),
  })
  }

  // 提交一个 future 到运行时
  fn spawn(self: &Arc<Self>, future: impl Future<Output = ()> + Send + 'static) {
  let task = Arc::new(Task {
  future: Mutex::new(Box::pin(future)),
  runtime: self.clone(),
  });
  self.queue.lock().unwrap().push_back(task);
  }

  // 不断 poll 队列里的任务，直到全部完成
  fn run(&self) {
  loop {
  let task = self.queue.lock().unwrap().pop_front();
  match task {
  Some(task) => {
  // 为这个 task 创建一个 waker
  let waker = task_to_waker(task.clone());
  let mut cx = Context::from_waker(&waker);
  // poll 这个 future
  let mut future = task.future.lock().unwrap();
  match future.as_mut().poll(&mut cx) {
  Poll::Ready(()) => {} // 完成了，不放回队列
  Poll::Pending => {}  // 没完成，等 wake() 放回来
  }
  }
  None => break, // 队列空了，结束
  }
  }
  }
}

// 注意：教学简化版 -- 生产代码请使用 Arc<Fn> 或 futures::task::ArcWake
// 下面手动管理 Arc 引用计数，必须保证 clone/wake/drop 的配对正确，
// 否则会导致 double-free 或内存泄漏。tokio 内部用宏生成来保证安全性。
fn task_to_waker(task: Arc<Task>) -> Waker {
  unsafe fn clone_fn(ptr: *const ()) -> RawWaker {
  // 从裸指针恢复 Arc，clone 后再 forget 原始 Arc 以维持引用计数
  let task = Arc::from_raw(ptr as *const Task);
  let cloned = Arc::clone(&task);
  std::mem::forget(task); // 不要 drop 原来的，引用计数保持不变
  RawWaker::new(Arc::into_raw(cloned) as *const (), &VTABLE)
  }

  unsafe fn wake_fn(ptr: *const ()) {
  // 取得所有权（引用计数不变），放回队列后 Arc 由队列持有
  let task = Arc::from_raw(ptr as *const Task);
  task.runtime.queue.lock().unwrap().push_back(Arc::clone(&task));
  // task 在此被 drop，引用计数 -1，与 clone 的 +1 抵消
  }

  unsafe fn drop_fn(ptr: *const ()) {
  drop(Arc::from_raw(ptr as *const Task)); // 引用计数 -1
  }

  static VTABLE: RawWakerVTable = RawWakerVTable::new(clone_fn,
  wake_fn,
  wake_fn, // wake_by_ref 简化为 wake_fn（教学用途，生产中应单独实现）
  drop_fn,);

  let raw = RawWaker::new(Arc::into_raw(task) as *const (), &VTABLE);
  unsafe { Waker::from_raw(raw) }
}

// Waker 内部结构
pub struct Waker {
  waker: RawWaker,
}

pub struct RawWaker {
  data: *const (),  // 指向具体数据（被类型擦除）
  vtable: &'static RawWakerVTable, // 函数指针表
}

pub struct RawWakerVTable {
  clone: unsafe fn(*const ()) -> RawWaker,
  wake: unsafe fn(*const ()),
  wake_by_ref: unsafe fn(*const ()),
  drop: unsafe fn(*const ()),
}

// mio 的核心 API（简化）
let mut poll = mio::Poll::new()?;  // 创建 epoll 实例
let mut events = mio::Events::with_capacity(1024);

// 注册一个 TCP socket，关注"可读"事件
poll.registry().register(&mut tcp_stream,
  mio::Token(42),  // 这个 token 就是 task 的标识
  mio::Interest::READABLE,)?;

// 阻塞等待事件（带超时）
poll.poll(&mut events, Some(Duration::from_millis(100)))?;

// 遍历就绪事件
for event in events.iter() {
  match event.token() {
  mio::Token(42) => {
  // socket 可读了！唤醒对应的 task
  }
  _ => {}
  }
}

// tokio I/O Driver 的简化流程
fn io_driver_loop(driver: &IoDriver) {
  let mut events = mio::Events::with_capacity(1024);

  loop {
  // 计算最近的 timer 超时，作为 poll 的超时时间
  let timeout = driver.timer.next_deadline();

  // 阻塞等待 I/O 事件（或超时）
  driver.mio_poll.poll(&mut events, timeout)?;

  // 处理 I/O 事件
  for event in events.iter() {
  // 根据 token 找到注册的 I/O 资源
  if let Some(io_resource) = driver.resources.get(event.token()) {
  // 更新就绪状态
  io_resource.set_readiness(event.readiness());
  // 唤醒等待这个资源的 task
  io_resource.wake();
  }
  }

  // 顺便处理到期的 timer（下一节详述）
  driver.timer.process_expired_timers();
  }
}

// tokio-uring 的用法：completion-based I/O
// 注意：需要单独的 runtime，不是标准 tokio runtime
tokio_uring::start(async {
  let file = tokio_uring::fs::File::open("data.txt").await.unwrap();

  // buffer 的所有权转移给内核，完成后还回来
  // 这是 completion model 的核心特征
  let buf = vec![0u8; 4096];
  let (result, buf) = file.read_at(buf, 0).await;

  println!("读取了 {} 字节", result.unwrap());
});

// sleep 的实质：创建一个在指定时刻被唤醒的 Future
pub async fn sleep(duration: Duration) {
  let deadline = Instant::now() + duration;
  let entry = timer::Entry::new(deadline);

  // 把 entry 注册到时间轮
  timer::register(&entry);

  // 挂起，等待时间轮在 deadline 到达时调用 wake()
  entry.await;
}

// timeout 就是把 sleep 和目标 future 组合起来
pub async fn timeout<F: Future>(duration: Duration, future: F) -> Result<F::Output, Elapsed> {
  tokio::select! {
  result = future => Ok(result),  // future 先完成
  _ = sleep(duration) => Err(Elapsed), // 超时了
  }
}

// 时间轮的简化数据结构
struct TimerWheel {
  // 6 层，每层 64 个 slot
  levels: [Level; 6],
  // 当前时间（毫秒精度）
  elapsed: u64,
}

struct Level {
  // 64 个 slot，每个 slot 是一个 timer 链表
  slots: [LinkedList<TimerEntry>; 64],
  // 这一层的时间粒度
  granularity: u64,
}

impl TimerWheel {
  // O(1) 插入：根据 deadline 计算应该放在哪一层的哪个 slot
  fn insert(&mut self, entry: TimerEntry) {
  let deadline = entry.deadline;
  let diff = deadline - self.elapsed;

  // 找到合适的层级
  let level = self.level_for(diff);
  // 计算 slot 索引
  let slot = (deadline / level.granularity) % 64;
  // 直接插入链表头部 —— O(1)
  level.slots[slot].push_front(entry);
  }

  // 推进时间，触发到期的 timer
  fn advance(&mut self, now: u64) {
  while self.elapsed < now {
  self.elapsed += 1;

  // 检查最底层的当前 slot
  let slot = (self.elapsed / 1) % 64;
  for entry in self.levels[0].slots[slot].drain() {
  // 到期了！调用 waker.wake()
  entry.waker.wake();
  }

  // 在适当时机做级联（cascade）
  // 高层 slot 的 timer 移到低层以获得更高精度
  self.maybe_cascade();
  }
  }
}

// interval 的漂移处理策略
use tokio::time::{interval, MissedTickBehavior};

let mut ticker = interval(Duration::from_millis(100));

// Burst：错过的 tick 立即连续触发（默认）
ticker.set_missed_tick_behavior(MissedTickBehavior::Burst);

// Delay：错过就错过，从当前时刻重新计时
ticker.set_missed_tick_behavior(MissedTickBehavior::Delay);

// Skip：跳过错过的 tick，对齐到下一个整数倍时刻
ticker.set_missed_tick_behavior(MissedTickBehavior::Skip);

// 否 错误：在 async 函数里调用同步阻塞 I/O
async fn bad_handler(path: &str) -> String {
  // std::fs::read_to_string 会阻塞当前 OS 线程！
  // 这个 Worker 线程在文件读取期间完全卡住
  // 其他 task 无法被调度
  let content = std::fs::read_to_string(path).unwrap();
  content
}

// 是 正确：用 spawn_blocking 把阻塞操作移到独立线程池
async fn good_handler(path: String) -> String {
  tokio::task::spawn_blocking(move || {
  std::fs::read_to_string(&path).unwrap()
  })
  .await
  .unwrap()
}

// 是 更好：直接用 tokio 的异步 fs API
async fn best_handler(path: &str) -> String {
  // tokio::fs 内部也是 spawn_blocking，但 API 更干净
  tokio::fs::read_to_string(path).await.unwrap()
}

// spawn_blocking 线程池的配置
let rt = tokio::runtime::Builder::new_multi_thread()
  .max_blocking_threads(512)  // 默认值
  .thread_keep_alive(Duration::from_secs(10)) // 空闲线程存活时间
  .build()
  .unwrap();

// 模拟调度延迟
async fn measure_latency() {
  let start = Instant::now();

  // 创建大量 task 争抢调度
  let mut handles = Vec::new();
  for i in 0..1_000_000 {
  handles.push(tokio::spawn(async move {
  // 每个 task 做一点点工作
  // 但 100 万个 task 排队等 8 个 Worker
  tokio::task::yield_now().await;
  }));
  }

  for handle in handles {
  handle.await.unwrap();
  }

  // 在 8 核机器上，这可能需要几秒钟
  // 每个 task 的平均调度延迟 = 总时间 / task 数
  println!("总耗时: {:?}", start.elapsed());
}

// 生产环境推荐配置
#[tokio::main]
async fn main() {
  // 等价于 Builder::new_multi_thread()
  //  .worker_threads(num_cpus)
  //  .enable_all()
  //  .build()
  server::run().await;
}

// 测试环境推荐配置
#[tokio::test]
async fn test_something() {
  // 等价于 Builder::new_current_thread()
  //  .enable_all()
  //  .build()
  assert_eq!(my_async_fn().await, 42);
}

// 否 灾难：阻塞整个 Worker 线程
async fn read_config() -> String {
  // std::fs::read_to_string 是同步阻塞的！
  // 文件在机械硬盘上？恭喜，这个 Worker 上的所有 task 陪你等 50ms
  std::fs::read_to_string("config.toml").unwrap()
}

// 是 正确：用 spawn_blocking 隔离
async fn read_config() -> String {
  tokio::task::spawn_blocking(|| {
  std::fs::read_to_string("config.toml").unwrap()
  }).await.unwrap()
}

// 是 更好：直接用 tokio::fs（内部就是 spawn_blocking 封装）
async fn read_config() -> String {
  tokio::fs::read_to_string("config.toml").await.unwrap()
}

// 否 危险：如果持锁期间有 .await，可能死锁
async fn bad_update(data: Arc<std::sync::Mutex<Vec<String>>>) {
  let mut guard = data.lock().unwrap();
  // 持有 std::sync::Mutex 的 guard 跨越了 .await 点
  // 如果这个 task 在 .await 时被挂起，其他 task 无法获取锁
  // 而且这个 task 可能被调度到不同的 Worker 线程上恢复——
  // std::sync::MutexGuard 不是 Send 的，编译器会报错（这是好事）
  let result = fetch_something().await;
  guard.push(result);
}

// 是 正确：用 tokio::sync::Mutex（支持跨 .await 持有）
async fn good_update(data: Arc<tokio::sync::Mutex<Vec<String>>>) {
  let mut guard = data.lock().await; // 注意这里是 .await
  let result = fetch_something().await;
  guard.push(result);
}

// 是 更好：如果不需要跨 .await 持锁，std::sync::Mutex 反而更快
async fn best_update(data: Arc<std::sync::Mutex<Vec<String>>>) {
  let result = fetch_something().await;
  // 先完成异步操作，再短暂持锁同步修改
  data.lock().unwrap().push(result);
  // guard 立即 drop，不跨 .await——没问题
}

// 否 常见误解：想"并行"但绕过了调度器
async fn handle_request() {
  // std::thread::spawn 创建的线程不受 tokio 管理
  // 里面不能用 tokio::spawn、tokio::time::sleep 等异步 API
  // 除非你手动在新线程里创建 runtime（浪费资源）
  std::thread::spawn(|| {
  // tokio::time::sleep(Duration::from_secs(1)).await;  // 编译错误！
  });
}

// 是 正确：用 tokio::spawn 做并发
async fn handle_request() {
  tokio::spawn(async {
  tokio::time::sleep(Duration::from_secs(1)).await;
  // 这个 task 由 tokio 调度器管理
  });
}

// 是 正确：真的需要 OS 线程做 CPU 密集计算，用 spawn_blocking
async fn handle_request() {
  let result = tokio::task::spawn_blocking(|| {
  // 这里跑 CPU 密集逻辑，不影响 async 调度器
  heavy_computation()
  }).await.unwrap();
}