asyncio 네트워킹 저수준 완전 정리, open_connection과 start_server부터 프로토콜·트랜스포트 구조, pause_reading 흐름제어까지

🌐 asyncio.open_connection과 start_server로 만드는 기본 TCP 스트림 구조

asyncio에서는 TCP 클라이언트와 서버를 만드는 가장 쉬운 입구로 asyncio.open_connection() 과 asyncio.start_server() 를 제공합니다.
이 두 함수는 고수준(high-level) 스트림 API라고 부르며, 소켓을 직접 다루지 않아도 되고, 읽기/쓰기 작업을 모두 async/await 문법으로 처리할 수 있게 해줍니다.
즉, “데이터를 한 줄 읽고, 응답을 보내고, 연결을 닫는다” 같은 로직을 자연스럽게 절차형 코드로 작성할 수 있습니다.
이 방식은 채팅 서버, 간단한 프록시, 내부 마이크로서비스 간 TCP 통신 같은 곳에서 특히 많이 쓰입니다.

먼저 클라이언트 측인 asyncio.open_connection(host, port) 은 두 개의 객체를 돌려줍니다.
하나는 reader (StreamReader) 이고 다른 하나는 writer (StreamWriter) 입니다.
reader는 서버에서 온 바이트 데이터를 비동기로 읽어오는 역할을 하고, writer는 서버로 데이터를 비동기로 밀어 넣는 역할입니다.
둘 다 코루틴 친화적으로 설계되어 있어서 data = await reader.readline() 처럼 대기하거나, writer.write(b”ping”) 이후 await writer.drain() 으로 송신 버퍼가 비워질 때까지 기다리는 패턴을 그대로 쓸 수 있습니다.

반대로 서버 쪽에서는 asyncio.start_server(client_handler, host, port) 를 사용합니다.
여기서 client_handler 는 async 함수이고, 서버에 새로운 TCP 연결이 들어올 때마다 asyncio가 자동으로 실행해줍니다.
그리고 그 handler는 매번 reader, writer 를 인자로 받습니다.
즉 “한 클라이언트 세션”이 하나의 코루틴으로 캡슐화되는 구조입니다.
이 모델은 직관적이고 디버깅이 편합니다.
특정 연결과 관련된 로직(인증, 메시지 파싱, 응답 작성 등)이 하나의 함수 안에서 순차적으로 보이므로 흐름 추적이 쉽습니다.

이 스트림 API의 장점은 다음과 같이 정리할 수 있습니다.

• 🧠코루틴 기반 구조라서 코드가 동기식 로직처럼 읽힌다, 즉 유지보수가 쉽다
• 🔌reader / writer가 TCP 소켓 처리를 추상화하여, 저수준 소켓 옵션을 몰라도 빠르게 서버·클라이언트를 만들 수 있다
• 📏writer.drain() 으로 자연스럽게 백프레셔(송신 버퍼가 꽉 찼는지 여부)를 확인한 뒤 다음 데이터를 보낼 수 있다
• 🧩SSL/TLS 같은 보안 계층도 asyncio가 감싸주므로, 고수준 코드 관점에서는 크게 다르지 않게 다룰 수 있다

이 구조를 실제 코드로 보면 더 감이 옵니다.
아래 예시는 아주 전형적인 에코 서버/클라이언트 패턴입니다.
클라이언트가 보낸 데이터를 다시 돌려주는 형태라서 테스트에 자주 쓰입니다.

import asyncio

async def handle_client(reader, writer):
    addr = writer.get_extra_info("peername")
    print(f"[connect] {addr}")

    while True:
        data = await reader.readline()
        if not data:
            break  # 클라이언트가 연결 종료
        print(f"[recv] {data!r} from {addr}")

        writer.write(data)
        await writer.drain()  # 버퍼가 너무 차지 않도록 백프레셔 확인
    writer.close()
    await writer.wait_closed()
    print(f"[close] {addr}")

async def main():
    server = await asyncio.start_server(handle_client, "127.0.0.1", 8888)

    async with server:
        await server.serve_forever()

asyncio.run(main())

위에서 핵심적으로 볼 점은 await reader.readline() 과 await writer.drain() 입니다.
reader.readline() 은 클라이언트가 한 줄을 보낼 때까지 비동기적으로 대기합니다.
이 기다림 동안 현재 코루틴만 멈출 뿐, 이벤트 루프 자체는 멈추지 않기 때문에 다른 연결도 동시에 처리할 수 있습니다.
그리고 writer.drain() 은 “내가 방금 write()로 넣은 데이터를 소켓 버퍼로 충분히 흘려보냈는지”를 확인하는 역할입니다.
이게 없다면 순간적으로 엄청난 양의 데이터를 메모리에만 쌓아둘 수 있고, 그건 곧 서버 전체 메모리 압박으로 이어질 수 있습니다.
즉 drain()은 스트림 API 관점에서의 흐름제어(백프레셔)라고 이해해도 됩니다.

이 방식은 빠르게 동작하는 프로토타입이나 I/O 중심 서비스에 특히 적합합니다.
예를 들어 내부 사내 시스템끼리 상태를 주고받는 헬스 체크 서버, 단순 메시지 브로커, 로그 수집기 등에서는 start_server 기반 코드만으로도 충분히 안정적으로 운영되는 경우가 많습니다.
그리고 중요한 점 하나.
이 스트림 기반 접근은 asyncio의 더 저수준 세계인 “프로토콜/트랜스포트” 위에 구축된 일종의 편의 래퍼입니다.
즉, 우리가 reader / writer만 보고 코딩하더라도 내부적으로는 여전히 이벤트 루프가 트랜스포트 객체를 통해 소켓을 관리하고 있습니다.
이 말은 곧 “더 깊이 내려가서 직접 제어할 수도 있다”는 가능성을 열어둡니다.

💬 open_connection / start_server는 초보자 친화적인 고수준 API이고, 내부적으로는 프로토콜/트랜스포트 레이어를 활용해 동일한 이벤트 루프 위에서 동작합니다.
즉 스트림 API를 이해하면 당장 서비스를 만들 수 있고, 트랜스포트 레이어를 이해하면 서비스가 무너지지 않게 최적화할 수 있습니다.

🧩 프로토콜과 트랜스포트 구조 이해하기, 콜백으로 움직이는 저수준 네트워킹

asyncio의 저수준 네트워킹은 Protocol / Transport 모델을 중심으로 돌아갑니다.
이건 이벤트 루프의 가장 핵심적인 구성요소 중 하나로, 고수준 스트림 API(open_connection, start_server)의 기반이기도 합니다.
이 두 가지를 이해하면 asyncio 네트워크의 내부 동작을 더 깊이 있게 제어할 수 있습니다.
예를 들어 데이터 수신 시점, 전송 버퍼 상태, 연결 종료 이벤트를 직접 처리하고 싶다면 이 레벨로 내려오는 게 맞습니다.

먼저 Protocol 은 “데이터를 어떻게 처리할지”를 정의하는 콜백 객체입니다.
데이터가 들어오면 asyncio가 data_received() 메서드를 호출하고, 연결이 끊기면 connection_lost() 를 불러줍니다.
즉, 이벤트 루프가 데이터 흐름의 ‘시점’을 감지해 그에 맞는 메서드를 호출하는 구조입니다.
반면 Transport 는 실제 네트워크 I/O를 담당하는 계층입니다.
소켓을 열고 닫거나, 데이터를 송신(write)하거나, 흐름제어(pause_reading, resume_reading)를 수행하는 실질적인 통로 역할을 합니다.
이 두 객체는 항상 함께 생성되며, loop.create_server() 또는 loop.create_connection() 같은 저수준 API를 통해 직접 다룰 수 있습니다.

import asyncio

class EchoProtocol(asyncio.Protocol):
    def connection_made(self, transport):
        peer = transport.get_extra_info("peername")
        print(f"[connect] {peer}")
        self.transport = transport

    def data_received(self, data):
        print(f"[recv] {data!r}")
        self.transport.write(data)  # 받은 데이터를 그대로 전송

    def connection_lost(self, exc):
        print("[close] 연결 종료")

async def main():
    loop = asyncio.get_running_loop()
    server = await loop.create_server(
        lambda: EchoProtocol(), "127.0.0.1", 8888
    )

    async with server:
        await server.serve_forever()

asyncio.run(main())

위 코드에서 볼 수 있듯이, Protocol은 일종의 “콜백 클래스”입니다.
새로운 연결이 만들어지면 connection_made() 가 호출되어 transport 객체를 전달받습니다.
이 transport가 실제 TCP 소켓에 해당하며, 데이터를 보낼 때는 transport.write(data) 를 호출하면 됩니다.
데이터가 들어오면 asyncio가 알아서 data_received() 를 호출해주고, 여기서 원하는 처리를 수행합니다.
그리고 연결이 끊기면 connection_lost() 가 호출되어 정리 로직을 넣을 수 있습니다.

이 방식의 가장 큰 특징은 모든 것이 이벤트 기반이라는 점입니다.
즉, “언제 읽을지”를 기다리는 게 아니라 “읽혔을 때” 호출되는 구조이죠.
그렇기 때문에 data_received() 는 매우 빠르게 실행되어야 하며, I/O 블로킹이나 긴 연산을 절대 포함해서는 안 됩니다.
만약 무거운 작업을 해야 한다면, asyncio.create_task() 로 별도 코루틴을 띄우거나 run_in_executor() 로 스레드풀에 넘겨야 합니다.
그렇지 않으면 이벤트 루프가 막혀서 다른 연결을 처리하지 못하게 됩니다.

즉, Protocol/Transport 모델은 스트림보다 훨씬 저수준이지만,
그만큼 성능과 제어권을 모두 가져올 수 있는 강력한 구조입니다.
대규모 실시간 통신 서버(예: 게임 서버, 실시간 데이터 피드, IoT 메시지 허브 등)는 대부분 이 패턴으로 작성됩니다.
왜냐하면 각 연결마다 reader, writer 객체를 만들 필요가 없고, 단순한 콜백 호출만으로 입출력을 처리할 수 있기 때문입니다.

이 두 레벨의 차이를 이해하면 asyncio의 고성능 구조가 보입니다.
스트림 API는 편하지만, 내부적으로는 결국 transport.write()와 data_received()로 구성된 이 저수준 레이어 위에서 돌아갑니다.
즉, 스트림이 안정성과 편의성을 제공한다면, 프로토콜/트랜스포트는 효율성과 세밀한 제어권을 제공합니다.
고성능 서버를 튜닝하거나, 자체 프로토콜을 설계해야 하는 경우라면 이 구조를 알아두는 것이 필수입니다.

⚡ 흐름제어와 pause_reading 개념, 백프레셔는 왜 필요한가

asyncio의 네트워킹 구조에서 가장 자주 오해받는 부분이 바로 흐름제어(Flow Control)입니다.
흐름제어란, 송수신 양쪽의 데이터 처리 속도가 맞지 않을 때 시스템이 안정적으로 균형을 맞추기 위한 장치입니다.
즉, 한쪽에서 너무 빨리 데이터를 보내면 다른 쪽 버퍼가 넘쳐 메모리가 폭주할 수 있고, 반대로 너무 느리면 전송 지연이 발생합니다.
이때 사용하는 것이 바로 pause_reading() 과 resume_reading() 메서드입니다.

이 두 메서드는 트랜스포트 객체(Transport)에 존재하며, 이름 그대로 “읽기 일시중지 / 재개”를 수행합니다.
예를 들어 서버 쪽에서 받은 데이터를 처리하는 동안 애플리케이션 내부 버퍼가 꽉 찼다면, transport.pause_reading() 을 호출하여 더 이상 소켓에서 데이터를 읽지 않게 합니다.
그러면 asyncio 이벤트 루프는 내부적으로 소켓의 읽기 이벤트를 비활성화하여, 새 데이터가 들어와도 당장 읽지 않습니다.
그리고 버퍼가 다시 비워지면 transport.resume_reading() 으로 재개할 수 있습니다.

💬 pause_reading은 단순히 “잠깐 멈춰라”가 아니라, 이벤트 루프 차원에서 소켓의 읽기 알림을 끊는 기능입니다.
즉, 데이터는 커널 버퍼에 남아있지만 사용자 공간으로는 더 이상 전달되지 않습니다.

이 메커니즘을 이해하려면 백프레셔(Backpressure) 개념이 필요합니다.
백프레셔란 데이터 흐름이 빠른 쪽에서 느린 쪽으로 “압력”이 전해지는 현상입니다.
예를 들어 클라이언트가 초당 1GB 데이터를 보낼 수 있는데 서버는 초당 200MB밖에 처리하지 못하면, 결국 송신 측의 버퍼가 차오르며 압력이 거꾸로 걸리게 됩니다.
이때 pause_reading() 을 적절히 호출하여 입력을 잠시 멈추면, 서버가 처리할 시간을 벌 수 있습니다.
이것이 안정적인 I/O 시스템을 만드는 핵심 원리입니다.

class ControlledProtocol(asyncio.Protocol):
    def connection_made(self, transport):
        self.transport = transport
        self.buffer = bytearray()
        print("[connect] 클라이언트 연결")

    def data_received(self, data):
        self.buffer.extend(data)
        print(f"[recv] {len(data)} bytes, 현재 버퍼 {len(self.buffer)} bytes")

        if len(self.buffer) > 100_000:
            print("[pause] 버퍼 초과, 읽기 일시중지")
            self.transport.pause_reading()

        # 데이터 처리 후 버퍼가 비워지면 resume
        if len(self.buffer) < 20_000:
            print("[resume] 버퍼 정상화, 읽기 재개")
            self.transport.resume_reading()

위 코드처럼 일정 임계치 이상으로 버퍼가 커지면 읽기를 중단하고, 충분히 비워지면 다시 읽기를 재개합니다.
이런 구조를 사용하면 갑작스러운 트래픽 폭주 상황에서도 서버가 안정적으로 동작할 수 있습니다.
반대로 이 로직이 없으면 커널 소켓 버퍼와 사용자 버퍼가 동시에 꽉 차면서 메모리 사용량이 급증하고, 결국 OOM(Out of Memory) 오류로 서버가 다운될 위험이 있습니다.

결국 flow control 은 고성능 네트워크 프로그래밍의 생명줄과 같습니다.
프로토콜/트랜스포트 계층을 직접 사용할 때 이 흐름제어를 적절히 구현하지 않으면, 높은 처리량을 자랑하던 서버라도 갑작스런 데이터 폭주에 쉽게 무너질 수 있습니다.
그래서 asyncio는 pause_reading()을 통해 안전장치를 제공하며, 개발자는 이를 기반으로 버퍼 크기나 처리속도에 맞춘 조절 로직을 짜는 것이 좋습니다.

🔁 writer.drain과 Transport.write의 차이, 어느 쪽이 더 빠를까

asyncio 네트워킹을 공부하다 보면 비슷해 보이지만 실제 동작이 다른 두 메서드가 자주 등장합니다.
바로 writer.drain() 과 transport.write() 입니다.
두 방식 모두 데이터를 전송하지만, 내부 작동 방식과 백프레셔 처리 구조에서 큰 차이가 있습니다.
이 차이를 이해하면 고성능 네트워크 애플리케이션을 설계할 때 어떤 방식을 선택해야 할지 판단이 쉬워집니다.

먼저 writer.write() 와 writer.drain() 은 고수준 스트림 API에서 사용됩니다.
writer.write()는 데이터를 소켓으로 비동기 전송 요청을 보내고, drain()은 “이전 write가 전송 완료될 때까지 기다리는” 역할을 합니다.
즉, await writer.drain() 은 송신 버퍼가 꽉 찰 경우 이벤트 루프가 잠시 대기해주는 일종의 비동기 백프레셔라고 볼 수 있습니다.
이 덕분에 개발자는 버퍼 관리나 흐름제어를 직접 구현하지 않아도 됩니다.

반면 transport.write(data) 는 저수준 API에서 동작합니다.
이건 코루틴이 아니며, 호출 즉시 데이터를 커널 송신 버퍼에 전달하도록 요청합니다.
즉, 비동기 대기를 하지 않고 곧바로 반환됩니다.
따라서 엄청난 양의 데이터를 빠르게 전송할 수 있지만, 송신 버퍼가 가득 차도 자동으로 멈추지 않기 때문에 pause_reading() 같은 흐름제어를 직접 구현해야 합니다.
이 점이 고수준 스트림과 가장 큰 차이입니다.

실제 성능 테스트를 해보면 작은 데이터(1~10KB 단위)에서는 두 방식의 차이가 거의 없습니다.
하지만 대용량 전송(수백 MB 이상)을 반복하거나 초당 수천 연결을 처리하는 서버에서는 transport.write() 쪽이 더 빠릅니다.
다만 이 경우 pause_reading() 기반의 흐름제어를 직접 구현하지 않으면 버퍼 오버플로우가 쉽게 발생합니다.
즉, 성능을 얻는 대신 안정성을 직접 관리해야 하는 셈입니다.

💬 writer.drain()은 초보자에게 안전한 선택이고, transport.write()는 전문가에게 빠른 도구입니다.
결국 핵심은 ‘이벤트 루프를 막지 않으면서 얼마나 안정적으로 전송 흐름을 제어할 수 있느냐’입니다.

정리하자면, 스트림 API(writer.drain)는 자동 흐름제어와 안정성을 제공하는 대신 약간의 오버헤드가 있습니다.
프로토콜/트랜스포트(transport.write)는 속도는 빠르지만, 흐름제어를 수동으로 처리해야 하는 부담이 있습니다.
프로젝트의 규모와 트래픽 특성에 따라 두 방식을 선택적으로 조합하면 됩니다.
예를 들어 내부 통신은 transport.write()로, 외부 클라이언트 통신은 writer.drain() 기반으로 구성하는 식으로 병용하는 경우도 많습니다.

🧪 실전 예시로 보는 echo 서버와 대용량 송신 패턴

이제까지 asyncio의 스트림과 프로토콜, 트랜스포트 구조, 그리고 흐름제어까지 살펴봤습니다.
이제 이를 실제 코드에 녹여볼 차례입니다.
대표적인 예시로 “에코 서버”를 확장하여 대용량 데이터를 안정적으로 송신하는 패턴을 구현해보겠습니다.
이 예시는 실제 서비스 환경에서도 유용하며, 초보자에게는 asyncio 구조의 핵심을 한눈에 보여줍니다.

아래 코드는 asyncio.start_server() 기반의 에코 서버를 확장한 버전입니다.
일정 크기 이상 데이터가 쌓이면 pause_reading()으로 수신을 잠시 중단하고, 처리 후 다시 resume_reading()으로 재개하는 방식입니다.
또한 송신 시 writer.drain()을 통해 백프레셔를 관리하여 송신 버퍼가 꽉 차지 않도록 합니다.

import asyncio

class ControlledEcho(asyncio.Protocol):
    def __init__(self):
        self.buffer = bytearray()

    def connection_made(self, transport):
        self.transport = transport
        peer = transport.get_extra_info('peername')
        print(f"[connect] {peer}")

    def data_received(self, data):
        self.buffer.extend(data)
        print(f"[recv] {len(data)} bytes, buffer={len(self.buffer)}")

        if len(self.buffer) > 200_000:
            print("[pause] reading paused")
            self.transport.pause_reading()

        # 에코 응답
        self.transport.write(data)

        # 버퍼가 비워졌으면 다시 재개
        if len(self.buffer) < 50_000:
            print("[resume] reading resumed")
            self.transport.resume_reading()

    def connection_lost(self, exc):
        print("[close] 연결 종료")

async def main():
    loop = asyncio.get_running_loop()
    server = await loop.create_server(
        lambda: ControlledEcho(),
        "127.0.0.1",
        8888
    )

    async with server:
        await server.serve_forever()

asyncio.run(main())

위 예시의 핵심은 “읽기 중단과 재개를 직접 제어한다”는 점입니다.
이는 대규모 데이터 송신 또는 폭주 트래픽 환경에서 매우 중요합니다.
예를 들어, 한 번에 수천 명의 클라이언트가 데이터를 밀어넣는 상황이라면 pause_reading()이 없다면 메모리가 순식간에 소진될 수 있습니다.
이 구조를 활용하면 일정한 메모리 한도 내에서 안정적으로 처리가 가능합니다.

또한 송신 쪽에서 writer를 사용하는 경우에도 동일한 개념이 적용됩니다.
고수준 스트림 API 기반 예시를 살펴보면 다음과 같습니다.

import asyncio

async def send_large_data(writer):
    data = b"x" * 1024 * 512  # 512KB
    for _ in range(200):  # 총 100MB 전송
        writer.write(data)
        await writer.drain()  # 백프레셔 관리
    writer.close()
    await writer.wait_closed()

async def main():
    reader, writer = await asyncio.open_connection("127.0.0.1", 8888)
    await send_large_data(writer)

asyncio.run(main())

여기서 writer.drain()은 내부적으로 트랜스포트의 버퍼 상태를 모니터링하여,
데이터가 실제 커널 버퍼로 흘러가기 전에는 다음 반복으로 넘어가지 않도록 합니다.
이 덕분에 개발자는 pause_reading()을 직접 호출할 필요 없이 안전하게 송신량을 제어할 수 있습니다.
즉, 고수준 스트림에서는 drain()이 “자동 백프레셔” 역할을 하는 셈입니다.

실전 환경에서는 이 두 가지 접근을 혼합해 사용하는 경우가 많습니다.
예를 들어, 내부 모듈 간 데이터 교환은 transport 기반으로 빠르게 처리하고,
외부로 노출되는 API 서버는 writer 기반으로 안정성을 확보하는 식입니다.
이렇게 구조를 나누면 asyncio 네트워킹의 장점을 극대화할 수 있습니다.

❓ 자주 묻는 질문 (FAQ)