动手：mini-tokio

type BoxFuture = Pin<Box<dyn Future<Output = ()> + Send>>;
type RunQueue = Arc<Mutex<VecDeque<Arc<Task>>>>;

struct Task {
  // 完成后置 None；Mutex 只是为了让 Task: Sync，单线程下从不争用
  future: Mutex<Option<BoxFuture>>,
  queue: RunQueue,
}

impl Wake for Task {
  // 唤醒 = 把自己放回运行队列尾部
  fn wake(self: Arc<Self>) {
    let queue = self.queue.clone();
    queue.lock().unwrap().push_back(self);
  }
}

fn run(main: impl Future<Output = ()> + Send + 'static) {
  spawn(main);
  let queue = QUEUE.with(|q| q.clone());
  loop {
    // 1. 把队列里的任务都跑一遍
    loop {
      let task = queue.lock().unwrap().pop_front();
      let Some(task) = task else { break };
      let mut slot = task.future.lock().unwrap();
      // 已完成的任务又被唤醒过一次：丢掉
      let Some(fut) = slot.as_mut() else { continue };
      let waker = Waker::from(task.clone());
      if fut.as_mut().poll(&mut Context::from_waker(&waker)).is_ready() {
        *slot = None;
        LIVE.with(|n| n.set(n.get() - 1));
      }
    }
    if LIVE.with(|n| n.get()) == 0 {
      return;
    }
    // 2. 没有可跑的任务：睡在 poll(2) 上
    park();
  }
}

#[derive(Default)]
struct Reactor {
  // 每个 (fd, 方向) 只存一个 waker，后来的覆盖先前的
  io: HashMap<(RawFd, Interest), Waker>,
  timers: BTreeMap<Instant, Vec<Waker>>,
}

fn park() {
  let wakers = REACTOR.with(|r| {
    let mut r = r.borrow_mut();
    // 超时 = 最近一个定时器还有多久，向上取整到毫秒；没有定时器就无限等
    let timeout = match r.timers.keys().next() {
      Some(&when) => {
        let ns = when.saturating_duration_since(Instant::now()).as_nanos();
        ns.div_ceil(1_000_000).min(i32::MAX as u128) as i32
      }
      None => -1,
    };
    let keys: Vec<(RawFd, Interest)> = r.io.keys().copied().collect();
    let mut fds: Vec<libc::pollfd> = keys
      .iter()
      .map(|&(fd, interest)| libc::pollfd {
        fd,
        events: if interest == Interest::Read { libc::POLLIN } else { libc::POLLOUT },
        revents: 0,
      })
      .collect();
    let n = unsafe { libc::poll(fds.as_mut_ptr(), fds.len() as libc::nfds_t, timeout) };
    // ...

    let mut wakers = Vec::new();
    for (pfd, key) in fds.iter().zip(&keys) {
      if pfd.revents != 0 {
        wakers.extend(r.io.remove(key));
      }
    }
    // 到期的定时器：split_off 留下 > now 的，前面那一半全部到期
    let later = r.timers.split_off(&(Instant::now() + Duration::from_nanos(1)));
    let expired = std::mem::replace(&mut r.timers, later);
    // ...
    wakers.extend(expired.into_values().flatten());
    // ...
    wakers
  });
  // 放开 RefCell 之后再 wake（tokio 也是放锁之后才叫醒）
  for w in wakers {
    w.wake();
  }
}

// 先直接做系统调用；WouldBlock 才登记 waker、返回 Pending
fn poll_io<T>(fd: RawFd, interest: Interest, cx: &Context<'_>, r: io::Result<T>) -> Poll<io::Result<T>> {
  match r {
    Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
      register(fd, interest, cx);
      Poll::Pending
    }
    r => Poll::Ready(r),
  }
}

async fn read(s: &mut TcpStream, buf: &mut [u8]) -> io::Result<usize> {
  let fd = s.as_raw_fd();
  poll_fn(|cx| poll_io(fd, Interest::Read, cx, s.read(buf))).await
}

    spawn(async move {
      log("server: waiting for a connection");
      let mut s = accept(&listener).await.unwrap();
      log("server: accepted");
      let mut buf = [0u8; 64];
      loop {
        let n = read(&mut s, &mut buf).await.unwrap();
        if n == 0 {
          log("server: client closed");
          break;
        }
        log(format!("server: echo {:?}", std::str::from_utf8(&buf[..n]).unwrap()));
        write(&mut s, &buf[..n]).await.unwrap();
      }
    });

    spawn(async move {
      sleep(Duration::from_millis(50)).await;
      // 连本机监听端口，内核 backlog 直接完成握手，这里用阻塞 connect 图省事
      let mut s = TcpStream::connect(addr).unwrap();
      s.set_nonblocking(true).unwrap();
      log("client: connected");
      let mut buf = [0u8; 64];
      for msg in ["hello", "mini-tokio"] {
        write(&mut s, msg.as_bytes()).await.unwrap();
        let n = read(&mut s, &mut buf).await.unwrap();
        log(format!("client: got {:?}", std::str::from_utf8(&buf[..n]).unwrap()));
      }
    });

    spawn(async {
      for i in 1..=3 {
        sleep(Duration::from_millis(30)).await;
        log(format!("ticker: tick {i}"));
      }
    });

[  0ms] main: spawned 3 tasks
[  0ms] server: waiting for a connection
[ 31ms] park: poll(1 fds, timeout 30ms) -> 0 io, 1 timer
[ 31ms] ticker: tick 1
[ 52ms] park: poll(1 fds, timeout 19ms) -> 0 io, 1 timer
[ 52ms] client: connected
[ 52ms] park: poll(2 fds, timeout 9ms) -> 1 io, 0 timer
[ 52ms] server: accepted
[ 52ms] server: echo "hello"
[ 52ms] park: poll(2 fds, timeout 9ms) -> 1 io, 0 timer
[ 52ms] client: got "hello"
[ 53ms] park: poll(2 fds, timeout 9ms) -> 1 io, 0 timer
[ 53ms] server: echo "mini-tokio"
[ 53ms] park: poll(2 fds, timeout 9ms) -> 1 io, 0 timer
[ 53ms] client: got "mini-tokio"
[ 53ms] park: poll(1 fds, timeout 9ms) -> 1 io, 0 timer
[ 53ms] server: client closed
[ 62ms] park: poll(0 fds, timeout 9ms) -> 0 io, 1 timer
[ 62ms] ticker: tick 2
[ 93ms] park: poll(0 fds, timeout 30ms) -> 0 io, 1 timer
[ 93ms] ticker: tick 3
[ 93ms] run: all tasks done

#[tokio::main(flavor = "current_thread")]
async fn main() {
  let start = Instant::now();
  let a = tokio::spawn(async move {
    for _ in 0..100_000 {
      yield_now().await; // 和 mini-tokio 用的是同一个手写 yield_now
      YIELDS.fetch_add(1, Relaxed);
    }
    println!("[{:>3}ms] A: done", start.elapsed().as_millis());
  });
  let b = tokio::spawn(async move {
    tokio::time::sleep(Duration::from_millis(1)).await;
    println!("[{:>3}ms] B: woke up, A has yielded {} times", start.elapsed().as_millis(), YIELDS.load(Relaxed));
  });
  a.await.unwrap();
  b.await.unwrap();
}

[  3ms] A: done
[  3ms] B: woke up, A has yielded 100000 times

[  1ms] B: woke up, A has yielded 7075 times
[ 24ms] A: done