✍️ 필사 모드: 비동기 I/O 모델 완전 가이드 2025: epoll, io_uring, Reactor/Proactor, async/await 내부

data = sock.recv(1024)  # 데이터 올 때까지 블록
process(data)

def handle_client(sock):
    data = sock.recv(1024)
    process(data)

while True:
    client = server.accept()
    threading.Thread(target=handle_client, args=(client,)).start()

sock.setblocking(False)
try:
    data = sock.recv(1024)
except BlockingIOError:
    pass  # 데이터 없음, 다른 일 하기

ready, _, _ = select([sock1, sock2, sock3], [], [])
for sock in ready:
    data = sock.recv(1024)

fd_set readfds;
FD_ZERO(&readfds);
FD_SET(sock, &readfds);

int n = select(max_fd + 1, &readfds, NULL, NULL, &timeout);

if (FD_ISSET(sock, &readfds)) {
    // 읽기 가능
}

struct pollfd fds[1024];
fds[0].fd = sock;
fds[0].events = POLLIN;

int n = poll(fds, 1024, timeout);

if (fds[0].revents & POLLIN) {
    // 읽기 가능
}

int epfd = epoll_create1(0);

struct epoll_event ev;
ev.events = EPOLLIN;
ev.data.fd = sock;
epoll_ctl(epfd, EPOLL_CTL_ADD, sock, &ev);

struct epoll_event events[64];
int n = epoll_wait(epfd, events, 64, -1);

for (int i = 0; i < n; i++) {
    int fd = events[i].data.fd;
    // 처리
}

// 데이터 16KB 있고 8KB만 읽음 → 다음 epoll_wait도 알림

// 새 데이터 도착 시에만 알림 → 모두 읽어야 함
while (true) {
    n = recv(fd, buf, sizeof(buf), 0);
    if (n < 0 && errno == EAGAIN) break;  // 데이터 다 읽음
    process(buf, n);
}

int kq = kqueue();

struct kevent change;
EV_SET(&change, sock, EVFILT_READ, EV_ADD | EV_ENABLE, 0, 0, NULL);
kevent(kq, &change, 1, NULL, 0, NULL);

struct kevent events[64];
int n = kevent(kq, NULL, 0, events, 64, NULL);

[User Space]                    [Kernel Space]
[Submission Queue (SQ)] ────→ [Worker]
                                ↓
[Completion Queue (CQ)] ←──── [I/O 완료]

struct io_uring ring;
io_uring_queue_init(256, &ring, 0);

// 작업 제출
struct io_uring_sqe *sqe = io_uring_get_sqe(&ring);
io_uring_prep_read(sqe, fd, buf, sizeof(buf), 0);
io_uring_submit(&ring);

// 완료 대기
struct io_uring_cqe *cqe;
io_uring_wait_cqe(&ring, &cqe);
// cqe->res = 읽은 바이트 수
io_uring_cqe_seen(&ring, cqe);

[Event Loop]
   ↓ epoll_wait
[Event: fd=5 readable]
   ↓
[Handler]
   ↓
[read(5, ...)]  ← 사용자가 직접 read

[User] → "이 fd에서 1024 byte 읽어줘"
   ↓
[Kernel] → 비동기 작업
   ↓
[Completion: 데이터 준비됨]
   ↓
[Handler] → 사용자가 받음

async_read(socket, buffer, [](error_code ec, size_t bytes) {
    // 완료 시 호출
});

async def fetch_data():
    data = await http_request("https://api.example.com")
    return process(data)

class FetchDataStateMachine:
    state = 0
    
    def resume(self):
        if self.state == 0:
            self.future = http_request("https://api.example.com")
            self.future.set_callback(self.resume)
            self.state = 1
            return  # 일시 중지
        
        if self.state == 1:
            data = self.future.result()
            return process(data)

loop = asyncio.get_event_loop()

while True:
    # 1. ready 큐의 작업 실행
    while ready_queue:
        task = ready_queue.pop()
        task.run()
    
    # 2. epoll_wait로 I/O 이벤트 대기
    events = epoll.wait(timeout)
    
    # 3. 대기 중이던 작업을 ready 큐로
    for event in events:
        task = pending[event.fd]
        ready_queue.append(task)

import asyncio

async def main():
    # 동시 실행
    results = await asyncio.gather(
        fetch_user(1),
        fetch_user(2),
        fetch_user(3),
    )
    return results

asyncio.run(main())

async function main() {
    const data = await fetch("https://api.example.com")
    const json = await data.json()
    console.log(json)
}

#[tokio::main]
async fn main() {
    let data = fetch("https://api.example.com").await;
    println!("{:?}", data);
}

[1 스레드]
   ↓
[Event Loop] (epoll)
   ↓
[Callback 1] [Callback 2] [Callback 3]

[N 스레드 (CPU 코어 수)]
   ↓
[Event Loop per thread]
   ↓
[Work-stealing]

setsockopt(sock, SOL_SOCKET, SO_REUSEPORT, &one, sizeof(one));

[Port 80]
   ├─ [Worker 1] (epoll)
   ├─ [Worker 2] (epoll)
   └─ [Worker 3] (epoll)

Apache (prefork):    5,000 req/s
Apache (worker):     20,000 req/s
Nginx (epoll):       100,000 req/s
Nginx (io_uring):    150,000 req/s

import asyncio

async def handle_client(reader, writer):
    while True:
        data = await reader.read(1024)
        if not data:
            break
        writer.write(data)
        await writer.drain()
    writer.close()

async def main():
    server = await asyncio.start_server(handle_client, '0.0.0.0', 8080)
    async with server:
        await server.serve_forever()

asyncio.run(main())

use tokio::net::TcpListener;
use tokio::io::{AsyncReadExt, AsyncWriteExt};

#[tokio::main]
async fn main() {
    let listener = TcpListener::bind("0.0.0.0:8080").await.unwrap();
    
    loop {
        let (mut socket, _) = listener.accept().await.unwrap();
        
        tokio::spawn(async move {
            let mut buf = [0; 1024];
            loop {
                let n = socket.read(&mut buf).await.unwrap();
                if n == 0 { return }
                socket.write_all(&buf[..n]).await.unwrap();
            }
        });
    }
}

listener, _ := net.Listen("tcp", ":8080")

for {
    conn, _ := listener.Accept()
    go handle(conn)
}

func handle(conn net.Conn) {
    buf := make([]byte, 1024)
    for {
        n, err := conn.Read(buf)
        if err != nil { return }
        conn.Write(buf[:n])
    }
}