Blog

서버 여러 대의 상태를 순차적으로 점검하는 스크립트를 짜본 적이 있다면 익숙한 문제가 있다. 호스트 10대를 ping으로 확인하는 데 각각 1초가 걸린다면, for 루프로 하나씩 처리할 경우 전체 실행 시간은 10초에 가까워진다. 대부분의 시간은 네트워크 응답을 기다리는 데 쓰이는데, 그 대기 시간 동안 CPU는 사실상 놀고 있다. threading으로 해결할 수도 있지만 스레드마다 컨텍스트 스위칭 비용이 붙고, GIL 때문에 진짜 병렬 실행은 아니다. 이럴 때 Python의 asyncio를 쓰면 스레드를 늘리지 않고도 I/O 대기 구간을 겹쳐 실행할 수 있어 전체 실행 시간을 대기 시간의 최댓값 수준으로 줄일 수 있다. 이 글에서는 asyncio의 핵심 개념과 여러 호스트/리소스를 동시에 점검하는 실전 모니터링 코드를 정리한다.

asyncio 핵심 개념

asyncio는 하나의 스레드 안에서 이벤트 루프(event loop)가 여러 코루틴(coroutine)을 번갈아 실행하는 협력적 멀티태스킹(cooperative multitasking) 모델이다. 스레드처럼 OS가 강제로 실행을 끊는 게 아니라, 코루틴이 await를 만나 스스로 실행을 양보(yield)할 때만 다른 코루틴으로 제어가 넘어간다.

중요한 건 asyncio가 I/O 대기 시간을 겹쳐 쓰는 데는 강하지만, CPU를 많이 쓰는 연산 자체를 빠르게 만들어주지는 않는다는 점이다. 이 특성은 뒤에서 다시 다룬다.

실전 코드

먼저 여러 호스트에 ping을 동시에 날려 응답 여부를 확인하는 코드다. asyncio.create_subprocess_exec로 외부 명령을 비동기로 실행하면, 프로세스가 응답을 기다리는 동안 이벤트 루프는 다른 ping 프로세스를 계속 진행시킨다.

import asyncio

async def ping(host: str, timeout: float = 1.5) -> bool:
    proc = await asyncio.create_subprocess_exec(
        'ping', '-c', '1', '-W', '1', host,
        stdout=asyncio.subprocess.DEVNULL,
        stderr=asyncio.subprocess.DEVNULL,
    )
    try:
        returncode = await asyncio.wait_for(proc.wait(), timeout=timeout)
        return returncode == 0
    except asyncio.TimeoutError:
        proc.kill()
        await proc.wait()
        return False

async def check_hosts(hosts: list[str]) -> dict[str, bool]:
    async def check_one(host: str):
        return host, await ping(host)

    results = await asyncio.gather(*(check_one(h) for h in hosts))
    return dict(results)

async def main():
    hosts = ['127.0.0.1', '8.8.8.8', '192.0.2.1']  # 192.0.2.1은 테스트용 미응답 IP
    status = await check_hosts(hosts)
    for host, up in status.items():
        print(f'{host}: {"UP" if up else "DOWN"}')

if __name__ == '__main__':
    asyncio.run(main())

여기서 핵심은 check_one 코루틴들을 await로 하나씩 순서대로 부르지 않고 asyncio.gather()에 한꺼번에 넘긴다는 점이다. for 루프에서 매번 await하면 이전 코루틴이 끝나야 다음이 시작되므로 사실상 순차 실행과 다를 게 없다. gather()는 넘겨받은 코루틴들을 모두 Task로 스케줄링한 뒤 전부 완료될 때까지 기다리므로 진짜 동시 실행이 된다.

실제 시스템 모니터링에서는 psutil처럼 asyncio를 모르는 blocking 라이브러리를 함께 써야 하는 경우가 많다. 이런 함수를 코루틴 안에서 그냥 호출하면 그 함수가 끝날 때까지 이벤트 루프 전체가 멈춘다. run_in_executor()로 스레드풀에 위임하면 이 문제를 피할 수 있다.

import asyncio
import psutil

async def get_cpu_percent(interval: float = 1.0) -> float:
    loop = asyncio.get_running_loop()
    # psutil.cpu_percent(interval=1.0)은 1초간 blocking되므로 executor에 위임
    return await loop.run_in_executor(None, psutil.cpu_percent, interval)

async def get_memory_percent() -> float:
    loop = asyncio.get_running_loop()
    mem = await loop.run_in_executor(None, psutil.virtual_memory)
    return mem.percent

async def collect_metrics():
    cpu, mem = await asyncio.gather(
        get_cpu_percent(),
        get_memory_percent(),
    )
    return {'cpu_percent': cpu, 'mem_percent': mem}

마지막으로 위 함수들을 조합해 5초 간격으로 호스트 상태와 로컬 리소스를 동시에 갱신하는 모니터링 루프를 만든다.

async def monitor_loop(hosts: list[str], interval: int = 5):
    while True:
        host_status, metrics = await asyncio.gather(
            check_hosts(hosts),
            collect_metrics(),
        )
        down = [h for h, up in host_status.items() if not up]
        print(
            f'CPU {metrics["cpu_percent"]:.1f}% / '
            f'MEM {metrics["mem_percent"]:.1f}% / '
            f'DOWN: {down or "none"}'
        )
        await asyncio.sleep(interval)

if __name__ == '__main__':
    asyncio.run(monitor_loop(['127.0.0.1', '8.8.8.8']))

실행하면 다음과 같은 로그가 5초마다 출력된다.

CPU 12.3% / MEM 41.7% / DOWN: none
CPU 9.8% / MEM 42.0% / DOWN: none
CPU 34.5% / MEM 43.1% / DOWN: none

주의사항

CPU-bound 작업에는 효과가 없다: asyncio는 하나의 스레드에서 동작하므로 순수 연산(예: 큰 행렬 곱셈, 이미지 인코딩)을 여러 코루틴으로 나눠도 전체 처리 시간은 줄지 않는다. 오히려 코루틴 사이 스케줄링 오버헤드만 늘 수 있다. CPU-bound 작업을 병렬화하려면 GIL의 제약을 받지 않는 multiprocessing을 써야 한다.

blocking 호출이 섞이면 전체가 멈춘다: 코루틴 안에서 time.sleep()이나 requests.get()처럼 asyncio를 모르는 동기 함수를 직접 호출하면, 그 함수가 끝날 때까지 이벤트 루프 자체가 멈춰서 다른 코루틴도 함께 정지한다. asyncio.sleep()처럼 asyncio 대응 버전을 쓰거나, 대응 버전이 없다면 앞서처럼 run_in_executor()로 스레드풀에 넘겨야 한다.

예외 처리: asyncio.gather()는 기본적으로 넘겨받은 코루틴 중 하나가 예외를 던지면 나머지가 아직 끝나지 않았어도 그 예외를 즉시 전파한다. 일부가 실패해도 나머지 결과를 계속 받고 싶다면 gather(*coros, return_exceptions=True)를 써서 예외 객체를 결과 리스트에 포함시키고 직접 분기 처리해야 한다. 위 ping() 예제처럼 함수 내부에서 예외를 잡아 값으로 변환해두는 방식도 실전에서 자주 쓴다.

이벤트 루프는 한 곳에서만: asyncio.run()은 프로그램 최상위에서 한 번만 호출해야 한다. 이미 실행 중인 이벤트 루프 안에서(예: Jupyter 노트북, 다른 프레임워크의 콜백 안에서) 다시 asyncio.run()을 부르면 RuntimeError: asyncio.run() cannot be called from a running event loop가 발생한다. 이런 환경에서는 이미 떠 있는 루프에 코루틴을 create_task()로 등록하거나 await해야 한다.

Task 참조를 놓치지 말 것: 결과를 바로 기다리지 않고 백그라운드로 던져두려는 목적으로 asyncio.create_task(coro())만 호출하고 반환값을 변수에 저장하지 않으면, 이벤트 루프가 가비지 컬렉션 타이밍에 해당 Task를 수거해버려 코루틴이 중간에 조용히 취소될 수 있다. 파이썬은 이런 경우 Task was destroyed but it is pending 경고를 남긴다. 백그라운드 Task는 별도의 set이나 리스트에 보관해뒀다가 완료 시점에 제거하는 방식으로 참조를 유지해야 한다.

동시 실행 개수 제한: 호스트가 수백 대로 늘어나면 gather()에 모든 코루틴을 한꺼번에 넘기는 방식은 파일 디스크립터나 프로세스 수 제한에 걸릴 수 있다. asyncio.Semaphore로 동시 실행 개수를 제한해 한 번에 너무 많은 서브프로세스가 뜨지 않도록 조절하는 것이 안전하다.

async def check_hosts_limited(hosts: list[str], max_concurrency: int = 20):
    semaphore = asyncio.Semaphore(max_concurrency)

    async def check_one(host: str):
        async with semaphore:
            return host, await ping(host)

    results = await asyncio.gather(*(check_one(h) for h in hosts))
    return dict(results)

async with semaphore 블록에 들어갈 수 있는 코루틴 수는 max_concurrency개로 제한된다. 나머지는 세마포어가 풀릴 때까지 대기했다가 순서대로 진입하므로, 동시성의 이점을 유지하면서도 한 번에 열리는 서브프로세스/소켓 수를 통제할 수 있다.

마무리

여러 호스트나 리소스를 점검할 때 for 루프로 순차 처리하던 코드를 asyncio로 바꾸면, 코드량을 크게 늘리지 않고도 전체 실행 시간을 대기 시간의 병목 수준으로 줄일 수 있다. 다만 asyncio는 I/O 대기를 겹쳐 쓰는 도구이지 CPU 연산을 빠르게 만드는 도구가 아니라는 점, 그리고 blocking 호출이 섞이는 순간 이점이 사라진다는 점은 꼭 기억해야 한다. 이 글의 collect_metrics()처럼 psutil 기반 시스템 지표 수집과 ping 기반 호스트 체크를 함께 돌리는 패턴은 실제 헬스체크 스크립트나 간단한 모니터링 데몬을 만들 때 그대로 응용할 수 있다.

참고

Python 공식 문서 – asyncio

Python 공식 문서 – asyncio 서브프로세스

psutil documentation