✍️ 필사 모드: Rust Tokio 런타임 Deep Dive — Future, Waker, Work-Stealing, Pin, Zero-Cost Async 완전 정복 (2025)

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

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

async fn hello() -> u32 {
    42
}

fn hello() -> impl Future<Output = u32> {
    async move { 42 }
}

async fn example() -> u32 {
    let x = read_value().await;   // suspend point 1
    let y = compute(x).await;      // suspend point 2
    x + y
}

enum ExampleState {
    Start,
    Awaiting1 { fut: impl Future<Output = u32> },
    Awaiting2 { x: u32, fut: impl Future<Output = u32> },
    Done,
}

struct Example {
    state: ExampleState,
}

impl Future for Example {
    type Output = u32;
    fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<u32> {
        let this = unsafe { self.get_unchecked_mut() };
        loop {
            match &mut this.state {
                ExampleState::Start => {
                    let fut = read_value();
                    this.state = ExampleState::Awaiting1 { fut };
                }
                ExampleState::Awaiting1 { fut } => {
                    match Pin::new(fut).poll(cx) {
                        Poll::Ready(x) => {
                            let fut2 = compute(x);
                            this.state = ExampleState::Awaiting2 { x, fut: fut2 };
                        }
                        Poll::Pending => return Poll::Pending,
                    }
                }
                ExampleState::Awaiting2 { x, fut } => {
                    match Pin::new(fut).poll(cx) {
                        Poll::Ready(y) => {
                            let result = *x + y;
                            this.state = ExampleState::Done;
                            return Poll::Ready(result);
                        }
                        Poll::Pending => return Poll::Pending,
                    }
                }
                ExampleState::Done => panic!("polled after completion"),
            }
        }
    }
}

async fn big() {
    let buf = [0u8; 1024 * 1024];  // 1MB 스택 배열!
    do_io().await;
    println!("{}", buf[0]);
}

pub struct Waker {
    waker: RawWaker,
}

impl Waker {
    pub fn wake(self);      // self 소비
    pub fn wake_by_ref(&self); // 참조만
    pub fn clone(&self) -> Waker;
}

fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
    if self.io_ready() {
        Poll::Ready(value)
    } else {
        self.waker = Some(cx.waker().clone());
        // I/O 드라이버에게 "이 waker를 저장해두세요"
        self.register_io(cx.waker().clone());
        Poll::Pending
    }
}

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 ()),
}

async fn example() {
    let data = vec![1, 2, 3];
    let ptr = &data[0];  // &data[0]는 state machine의 필드 data를 가리킴
    some_future().await;
    println!("{}", *ptr);  // suspend 후에도 유효해야!
}

struct Example {
    data: Vec<i32>,
    ptr: *const i32,  // data[0]를 가리킴
    state: State,
}

pub struct Pin<P> {
    pointer: P,
}

let fut = Box::pin(my_async_fn());

use std::pin::pin;
let fut = pin!(my_async_fn());

// u32는 Unpin이므로 자유롭게 이동
let mut x = 5u32;
let pinned = Pin::new(&mut x);
let _ = pinned.as_mut();  // OK

┌─────────────────────────────────────────────────┐
│ 유저 async 코드 (async fn, .await)                │
├─────────────────────────────────────────────────┤
│ Scheduler (task, worker, queue, park/unpark)      │
├─────────────────────────────────────────────────┤
│ Driver (I/O driver = mio, time driver = 타이머 휠) │
├─────────────────────────────────────────────────┤
│ OS (epoll / kqueue / IOCP)                        │
└─────────────────────────────────────────────────┘

#[tokio::main]
async fn main() {
    do_something().await;
}

fn main() {
    tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .unwrap()
        .block_on(async {
            do_something().await;
        })
}

#[tokio::main]
async fn main() { ... }
// 또는
Builder::new_multi_thread().build()

#[tokio::main(flavor = "current_thread")]
async fn main() { ... }

struct Worker {
    // 로컬 런큐 (길이 256)
    local: LocalQueue<Task>,

    // LIFO slot (최근 spawn된 task)
    lifo_slot: Option<Task>,

    // 글로벌 런큐 참조
    remote: Arc<Mutex<RemoteQueue<Task>>>,

    // Park (잠재우기용)
    park: Parker,
}

// task A가 task B를 spawn
tokio::spawn(async {
    // B의 로컬 상태는 A의 컨텍스트와 연관
});
// A가 .await 즉시, B가 LIFO slot에서 꺼내져 실행 → L1 cache hit

loop {
    // 1. LIFO slot (최근 spawn)
    if let Some(task) = self.lifo_slot.take() {
        task.run();
        continue;
    }

    // 2. 로컬 큐
    if let Some(task) = self.local.pop() {
        task.run();
        continue;
    }

    // 3. 글로벌 큐 (tick마다)
    if self.tick % GLOBAL_POLL_INTERVAL == 0 {
        if let Some(task) = self.remote.pop() {
            task.run();
            continue;
        }
    }

    // 4. I/O 드라이버 polling
    self.poll_driver();

    // 5. 다른 worker에서 훔치기
    if let Some(task) = self.steal_from_other_worker() {
        task.run();
        continue;
    }

    // 6. park (잠자기)
    self.park.park();
}

fn steal_from(&self, victim: &Worker) -> Option<Task> {
    // 1. victim의 로컬 큐에서 절반을 가져옴
    let n = victim.local.len() / 2;
    if n == 0 { return None; }

    let mut stolen = Vec::with_capacity(n);
    for _ in 0..n {
        if let Some(task) = victim.local.steal() {
            stolen.push(task);
        }
    }

    // 2. 첫 번째를 실행, 나머지는 자기 로컬 큐에 넣기
    let first = stolen.pop()?;
    for task in stolen {
        self.local.push(task);
    }
    Some(first)
}

let handle = tokio::spawn(async { 42 });
let result: i32 = handle.await.unwrap();

let socket = TcpStream::connect("example.com:80").await?;
// 실제로는:
// 1. socket(), fcntl(O_NONBLOCK)
// 2. connect() → EINPROGRESS (즉시 반환)
// 3. epoll에 WRITE 이벤트 등록
// 4. poll() → Pending
// 5. epoll_wait → WRITE 가능 → waker 호출
// 6. poll() 다시 → 완료

pub struct TcpStream {
    io: PollEvented<mio::net::TcpStream>,
}

fn poll_read(self: Pin<&mut Self>, cx: &mut Context, buf: &mut [u8])
    -> Poll<io::Result<usize>>
{
    // 1. 실제 read 시도
    match self.io.read(buf) {
        Ok(n) => Poll::Ready(Ok(n)),
        Err(e) if e.kind() == WouldBlock => {
            // 2. WouldBlock이면 readiness 상태를 Pending으로
            self.register_readiness(cx, Readiness::READ);
            Poll::Pending
        }
        Err(e) => Poll::Ready(Err(e)),
    }
}

fn turn(&mut self, max_wait: Option<Duration>) {
    let mut events = self.events.borrow_mut();
    self.poll.poll(&mut events, max_wait).unwrap();

    for event in events.iter() {
        let token = event.token();
        let source = self.resources.get(token);
        if event.is_readable() {
            source.wake_readers();
        }
        if event.is_writable() {
            source.wake_writers();
        }
    }
}

Worker Loop:
  1. 로컬 큐에서 task 실행
  2. 큐가 비면 I/O driver turn (epoll_wait)
  3. 깨어난 task들을 런큐에 추가
  4. 다시 실행

Level 0: 64 slots × 1 ms    (0 ~ 64 ms)
Level 1: 64 slots × 64 ms   (64 ms ~ 4 s)
Level 2: 64 slots × 4 s     (4 s ~ 4 m)
Level 3: 64 slots × 4 m     (4 m ~ 4 h)
Level 4: 64 slots × 4 h     (4 h ~ 11 d)
Level 5: 64 slots × 11 d    ...

pub struct Driver {
    wheel: wheel::Wheel<Handle>,
    ...
}

let handle = tokio::spawn(async {
    read_file().await;
    compute()  // CPU 작업
});

tokio::spawn(async {
    for i in 0..1_000_000_000 {
        hash(i);  // 그냥 CPU 작업
    }
});

for i in 0..1_000_000_000 {
    hash(i);
    if i % 1000 == 0 {
        tokio::task::yield_now().await;
    }
}

let result = tokio::task::spawn_blocking(|| {
    // CPU/블로킹 작업
    expensive_computation()
}).await.unwrap();

fn sync_fn() { /* ... */ }
async fn async_fn() { /* ... */ }

// OK
async fn caller1() {
    sync_fn();             // OK: async에서 sync 호출
    async_fn().await;      // OK
}

// NG
fn caller2() {
    async_fn();            // ??? 이건 Future를 리턴할 뿐 실행되지 않음
    async_fn().await;      // 컴파일 에러! sync에서 await 불가
}

fn caller3() {
    let rt = tokio::runtime::Handle::current();
    rt.block_on(async_fn());  // OK, 하지만 worker 한 개를 잠재움 — 위험
}

pub fn sync_do_thing() -> Result<Output> {
    let rt = Runtime::new()?;
    rt.block_on(async_do_thing())
}

tokio::spawn(async {
    let rt = tokio::runtime::Handle::current();
    rt.block_on(another_async());  // 데드락 가능!
});

use tokio::sync::mpsc;

let (tx, mut rx) = mpsc::channel::<String>(32);  // 버퍼 32

tokio::spawn(async move {
    for i in 0..10 {
        tx.send(format!("msg {}", i)).await.unwrap();
    }
});

while let Some(msg) = rx.recv().await {
    println!("{}", msg);
}

let (tx, rx) = tokio::sync::oneshot::channel::<u32>();
tokio::spawn(async move {
    tx.send(42).unwrap();  // send는 async 아님 (값이 이미 있음)
});
let value = rx.await.unwrap();

let (tx, _) = tokio::sync::broadcast::channel::<String>(16);
let mut rx1 = tx.subscribe();
let mut rx2 = tx.subscribe();

tokio::spawn(async move {
    tx.send("hello".to_string()).unwrap();
});

// 모든 receiver가 같은 메시지 받음
println!("{}", rx1.recv().await.unwrap());
println!("{}", rx2.recv().await.unwrap());

let (tx, mut rx) = tokio::sync::watch::channel::<u32>(0);
tokio::spawn(async move {
    tx.send(1).unwrap();
    tx.send(2).unwrap();
});

rx.changed().await.unwrap();
let val = *rx.borrow();  // 최신 값만

tokio::select! {
    val = rx.recv() => { /* 채널 수신 */ }
    _ = tokio::time::sleep(Duration::from_secs(1)) => { /* 타임아웃 */ }
    _ = shutdown.recv() => { /* 종료 시그널 */ }
}

let mut buf = vec![0u8; 1024];

tokio::select! {
    n = socket.read(&mut buf) => { /* ... */ }
    _ = timeout => { /* ... */ }
}

tokio = { version = "1", features = ["full", "tracing"] }
console-subscriber = "0.4"

fn main() {
    console_subscriber::init();
    // ...
}

#[tracing::instrument]
async fn handle_request(req: Request) -> Response {
    // ...
}

cargo flamegraph --bin myapp

Builder::new_multi_thread()
    .worker_threads(8)  // 기본은 num_cpus
    .build()

Builder::new_multi_thread()
    .max_blocking_threads(1024)  // 기본 512
    .build()

// BAD: 16 KB future를 stack allocate
async fn big() { let _ = [0u8; 16384]; foo().await; }

// GOOD: 큰 데이터는 heap에
async fn big() { let _ = Box::new([0u8; 16384]); foo().await; }

use futures::future::join_all;
let results = join_all(futs).await;

let data = std::sync::Mutex::new(HashMap::new());
tokio::spawn(async move {
    let lock = data.lock().unwrap();  // await 없이 블로킹 락 → worker 막음
    some_async().await;  // lock을 계속 쥐고 있음 — 다른 worker도 접근 못 함
});

tokio::spawn(async {
    huge_compute();  // blocking CPU
});

let mut set = tokio::task::JoinSet::new();
set.spawn(task1());
set.spawn(task2());
while let Some(res) = set.join_next().await {
    // ...
}

#[async_trait]
trait MyTrait {
    async fn do_thing(&self);  // 매크로가 Box<dyn Future>로 변환
}

trait Fetcher {
    async fn fetch(&self, url: &str) -> Result<String>;
}

let x: Box<dyn Fetcher> = ...;  // ERROR

fn use_fetcher<F: Fetcher>(f: F) where F::fetch(..): Send { ... }

let iter = gen {
    yield 1;
    yield 2;
    yield 3;
};

"thread pool is shutting down" 또는 task가 영원히 대기.

┌─────────────────────────────────────────────────────┐
│            Rust Tokio Cheat Sheet                    │
├─────────────────────────────────────────────────────┤
│ Future trait:                                        │
│   fn poll(Pin<&mut Self>, &mut Context) -> Poll      │
│     Poll::Ready(value) | Poll::Pending               │
│                                                      │
│ async fn → State machine struct                      │
│   - suspend point = enum variant                     │
│   - 로컬 변수 = struct field                          │
│   - .await = match poll { Pending => return; .. }    │
│                                                      │
│ Waker:                                               │
│   - "이 task 다시 poll해달라" 콜백                    │
│   - Context::waker() → clone → 등록                  │
│   - I/O/timer/channel이 wake 호출                    │
│                                                      │
│ Pin:                                                 │
│   - "이동 불가" 보장                                  │
│   - 자체 참조 Future 안전                             │
│   - Box::pin, pin!, spawn이 자동 처리                 │
│                                                      │
│ Tokio scheduler:                                     │
│   - Worker + LIFO slot + local runq + global         │
│   - Work-stealing (절반)                              │
│   - I/O driver 주기적 폴링                            │
│   - Cooperative only (자동 preempt 없음)              │
│                                                      │
│ Spawning:                                            │
│   - tokio::spawn(fut): async task                    │
│   - spawn_blocking(f): 블로킹 풀                      │
│   - block_in_place(f): 현재 worker를 blocking 풀로    │
│                                                      │
│ Channels:                                            │
│   - mpsc: 버퍼 있는 생산자→소비자                      │
│   - oneshot: 1회성                                   │
│   - broadcast: 팬아웃                                 │
│   - watch: 최신 값 관찰                               │
│                                                      │
│ 주의:                                                 │
│   - std::sync::Mutex를 async에서 쓰지 말 것          │
│   - CPU-heavy는 spawn_blocking/yield_now             │
│   - select!는 cancel-safe 확인                       │
│   - big Future는 Box로                               │
│                                                      │
│ 디버깅:                                                │
│   - tokio-console                                    │
│   - tracing                                          │
│   - cargo flamegraph                                 │
└─────────────────────────────────────────────────────┘