✍️ 필사 모드: 非同期 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])
    }
}