libuv vs Python AsyncIO: 완전 가이드

import asyncio
 
# 기본 asyncio 사용
async def main():
    print("Hello AsyncIO")
 
asyncio.run(main())

import asyncio
import uvloop
 
# uvloop으로 성능 향상
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
 
async def main():
    print("Hello uvloop")
 
asyncio.run(main())

# Python selectors 모듈의 EpollSelector 구현
class EpollSelector(BaseSelector):
    def select(self, timeout=None):
        # 직접적인 epoll_wait() 시스템 호출
        events = self._selector.poll(timeout, max_events)
        ready = []
        for fd, event_mask in events:
            key = self._fd_to_key.get(fd)
            if key:
                ready.append((key, event_mask & key.events))
        return ready  # Python 리스트 반환

// libuv 내부 구현 (uv-unix.c)
static void uv__io_poll(uv_loop_t* loop, int timeout) {
    // 일괄 처리를 위한 이벤트 배치
    struct epoll_event events[1024];
    int nfds = epoll_wait(loop->backend_fd, events, 1024, timeout);
    
    // C 레벨에서 직접 콜백 디스패치 (오버헤드 최소화)
    for (int i = 0; i < nfds; i++) {
        struct uv__io_s* w = events[i].data.ptr;
        w->cb(w, events[i].events);  // 즉시 콜백 실행
    }
}

# Python EpollSelector - 매 이벤트마다 Python 함수 호출
for key, mask in events:
    callback = key.data  # Python 객체 접근
    callback(key.fileobj, mask)  # Python 함수 호출 오버헤드
 
# uvloop - C 레벨 직접 디스패치
w->cb(w, events[i].events);  // C 함수 포인터 호출 (10-50배 빠름)

# 시스템 호출 추적 결과 (strace)
Python EpollSelector: 평균 15,000 syscalls/sec
libuv: 평균 8,000 syscalls/sec  # 40% 감소

// 마이크로벤치마크 결과 (ns/operation)
Python 함수 호출: ~250ns
C 함수 포인터 호출: ~5ns  // 50배 차이

┌─────────────────┐
│   Application   │
├─────────────────┤
│   AsyncIO API   │
├─────────────────┤
│  EpollSelector  │
├─────────────────┤
│   epoll/kqueue  │
└─────────────────┘

┌─────────────────┐
│   Application   │
├─────────────────┤
│   libuv API     │
├─────────────────┤
│  Event Loop     │ ← Timer, Poll, Check, Close 단계
├─────────────────┤
│  Thread Pool    │ ← 파일 I/O, DNS, 암호화
├─────────────────┤
│ OS APIs (epoll) │
└─────────────────┘

// libuv - 여러 파일 디스크립터를 한 번에 등록
struct epoll_event events[BATCH_SIZE];  // 최대 1024개
int nfds = epoll_wait(loop->backend_fd, events, BATCH_SIZE, timeout);

// libuv는 EPOLLET 플래그를 사용하여 고부하에서 성능 향상
event.events = EPOLLIN | EPOLLOUT | EPOLLET;
epoll_ctl(loop->backend_fd, EPOLL_CTL_ADD, fd, &event);

// 미리 할당된 버퍼 풀 사용
static char buffer_pool[POOL_SIZE];
static int pool_index = 0;
 
// 할당 오버헤드 없이 버퍼 재사용
char* get_buffer() {
    return &buffer_pool[pool_index++ % POOL_SIZE];
}

pip install uvloop

import asyncio
import uvloop
 
# 방법 1: 글로벌 정책 설정
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
 
# 방법 2: 직접 실행
uvloop.run(main())
 
# 방법 3: Python 3.11+ 권장
if sys.version_info >= (3, 11):
    with asyncio.Runner(loop_factory=uvloop.new_event_loop) as runner:
        runner.run(main())

import selectors
import socket
 
def echo_server():
    sel = selectors.DefaultSelector()
    sock = socket.socket()
    sock.bind(('localhost', 1234))
    sock.listen(100)
    sock.setblocking(False)
    sel.register(sock, selectors.EVENT_READ, accept)
    
    while True:
        events = sel.select()
        for key, mask in events:
            callback = key.data
            callback(key.fileobj, mask)

import asyncio
import uvloop
import aiohttp
 
async def fetch_many_urls():
    async with aiohttp.ClientSession() as session:
        tasks = []
        for i in range(1000):
            task = fetch_url(session, f'http://api.example.com/data/{i}')
            tasks.append(task)
        
        results = await asyncio.gather(*tasks)
        return results
 
# uvloop으로 2-4배 빠른 성능
uvloop.run(fetch_many_urls())

높은 성능 필요? ──→ uvloop 선택
      ↓ 아니오
순수 Python 필수? ──→ EpollSelector
      ↓ 아니오
크로스 플랫폼? ──→ uvloop (libuv)
      ↓ 아니오
     asyncio

# 프로덕션 권장 패턴
import asyncio
import uvloop
import sys
 
if sys.platform != 'win32':  # Windows는 uvloop 미지원
    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

# 개발 환경에서는 표준 asyncio로 디버깅 용이
import asyncio
 
# DEBUG 모드에서 더 나은 에러 메시지
asyncio.run(main(), debug=True)