asyncio 모듈 (2): 소켓 통신

소켓 통신

asyncio.run(main_async())

async def main_async():
    address = ('127.0.0.1', 4321)

    server = run_async_server(address) # 코루틴 인스턴스
    asyncio.create_task(server)

    results = await run_async_client(address)
    for number, outcome in results:
        print(f'클라이언트: {number}는 {outcome}')
        
async def run_async_server(address):
    server = await asyncio.start_server(
        handle_async_connection, *address)
    async with server:
        await server.serve_forever()

async def handle_async_connection(reader, writer):
    session = AsyncSession(reader, writer)
    try:
        await session.loop()
    except EOFError:
        pass

server

coroutine asyncio.start_server(client_connected_cb, host=None, port=None, *, limit=None, family=socket.AF_UNSPEC, flags=socket.AI_PASSIVE, sock=None, backlog=100, ssl=None, reuse_address=None, reuse_port=None, ssl_handshake_timeout=None, start_serving=True)

• 소켓 서버를 시작합니다.
• 반환 값인 server은 코루틴 인스턴스 이면서, server 객체 입니다.
• server 객체
  • 비동기 context manager 입니다.
  • async with 문에서 사용될 때, async with 문이 완료되면 서버 객체가 닫혀 있고 새 연결을 받아들이지 않는다는 것이 보장됩니다:
• 새 클라이언트 연결이 만들어질 때마다 client_connected_cb 콜백이 호출됩니다.
  • 이 콜백은 두 개의 인자로 (reader, writer) 쌍을 받는데, StreamReader 와 StreamWriter 클래스의 인스턴스입니다.
  • client_connected_cb는 일반 콜러블이나 코루틴 함수 일 수 있습니다; 코루틴 함수면, 자동으로 Task로 예약됩니다.
• limit는 반환된 StreamReader 인스턴스가 사용하는 버퍼 크기 한계를 결정합니다. 기본적으로 limit는 64KiB로 설정됩니다.
• 나머지 인자는 loop.create_server()로 직접 전달됩니다.
  - 예시
  - host: '127.0.0.1'
  - port: 4321

coroutine loop.create_server(protocol_factory, host=None, port=None, *, family=socket.AF_UNSPEC, flags=socket.AI_PASSIVE, sock=None, backlog=100, ssl=None, reuse_address=None, reuse_port=None, ssl_handshake_timeout=None, ssl_shutdown_timeout=None, start_serving=True)

• host 주소의 port 에서 리스닝하는 TCP 서버(소켓 유형 SOCK_STREAM)를 만듭니다.
• Server 객체를 반환합니다.

with 문

with open('data.txt') as f:
  for line in f:
    print(line)

• with 문 왜 써?

  • 자원을 획득하고 사용 후 반납해야 하는 경우 주로 사용합니다.
  • 예를들어 파일을 여는 경우, 다른 프로세스를 위해 사용한 뒤에 닫아주어야 한다.
  • 파이썬의 컨텍스트 매니저는 이러한 리소스를 with문법을 통해 with 절 내에서만 액세스를 가능하게 하고, 블록을 나가는 경우 어떤 이유든간에 리소스를 해제하게 된다.

• 파이썬의 컨텍스트 매니저

  • with 블럭을 적용할 수 있는 객체를 말한다.
  • 이러한 객체들은 with 절에서 마치 블럭에 대한 데코레이터처럼 동작한다.
  • 가장 흔한 예가 open() 함수로 생성하는 파일 입출력스트림으로, with 구문 내에서 쓰이면 블럭을 빠져나갈 때 파일을 닫는 동작을 자동으로 수행하게 된다.
  • 컨텍스트 매니저 객체는 enter(), exit() 두 개의 내장 메소드를 가지고 있는 것으로 간주된다.
    • 위 코드에서는 with 다음에 나오는 open('data.txt') 라는 코드는 파일에 대한 입출력 스트림을 반환한다.
    • 그리고 with 문을 빠져나갈 때, 파일에 대해 exit()가 호출되고 여기서 파일이 닫힐 것이다.

async with 문

async with 클래스() as 변수:
    코드

• async with 문 왜 써?
  • 컨텍스트 매니저에 대해서 with를 적용하면 코드 블럭을 진입하는 시점과 빠져나오는 시점에 약속된 동작을 수행하게 된다.
  • 그런데 이 약속된 동작들이 만약 처리 시간이 많이 걸리는 IO 작업이라면?
  • asyncio의 세계에서는 이러한 작업을 비동기처리해서 await 하고, 이렇게 기다리는 동안에는 다른 코루틴들이 진행될 수 있도록 한다.
  • 즉 이러한 전환이 with의 앞/뒤에서도 똑같이 일어날 수 있게 하려는 것이고 이것이 비동기 컨텍스트 매니저가 등장한 배경이다.
  • 즉, 클래스나 함수를 비동기로 처리한 뒤 결과를 반환하는 문법

• async with로 동작하는 클래스를 만드려면, aenter와 aexit 메서드를 구현해야 합니다.
• 그리고 메서드를 만들 때는 반드시 async def를 사용합니다.

class 클래스이름:
    async def __aenter__(self):
    	"""
        이 함수에서 값을 반환하면, as에 지정한 변수에 들어갑니다.
        """
        코드
 
    async def __aexit__(self, exc_type, exc_value, traceback):
        """
        async with as를 완전히 벗어나면 호출됩니다.
        """
        코드

• server.serve_forever()
  • 코루틴이 취소될 때까지 연결을 받아들이기 시작합니다.
  • server_forever 태스크를 취소하면 서버가 닫힙니다.
  • 이 메서드는 서버가 이미 연결을 받아들이고 있어도 호출 할 수 있습니다.
  • 하나의 server 객체 당 하나의 serve_forever 태스크만 존재할 수 있습니다.

Client

asyncio.run(main_async())

async def main_async():
    address = ('127.0.0.1', 4321)

    server = run_async_server(address) # 코루틴 인스턴스
    asyncio.create_task(server)

    results = await run_async_client(address)
    for number, outcome in results:
        print(f'클라이언트: {number}는 {outcome}')

async def run_async_client(address):
    # 서버가 시작될 수 있게 기다려주기
    await asyncio.sleep(0.1)

    streams = await asyncio.open_connection(*address)   # New
    client = AsyncClient(*streams)                      # New

    async with client.session(1, 5, 3):
        results = [(x, await client.report_outcome(x))
                   async for x in client.request_numbers(5)]

    async with client.session(10, 15, 12):
        async for number in client.request_numbers(5):
            outcome = await client.report_outcome(number)
            results.append((number, outcome))

    _, writer = streams                                # 새 기능
    writer.close()                                     # 새 기능
    await writer.wait_closed()                         # 새 기능

    return results        

class AsyncClient(AsyncConnectionBase):
    def __init__(self, *args):
        super().__init__(*args)
        self._clear_state()

    def _clear_state(self):
        self.secret = None
        self.last_distance = None

    @contextlib.asynccontextmanager                         # 변경됨
    async def session(self, lower, upper, secret):          # 변경됨
        print(f'\n{lower}와 {upper} 사이의 숫자를 맞춰보세요!'
              f' 쉿! 그 숫자는 {secret} 입니다.')
        self.secret = secret
        await self.send(f'PARAMS {lower} {upper}')          # 변경됨
        try:
            yield
        finally:
            self._clear_state()
            await self.send('PARAMS 0 -1')                   # 변경됨
            
    async def request_numbers(self, count):            # 변경됨
        for _ in range(count):
            await self.send('NUMBER')                  # 변경됨
            data = await self.receive()                # 변경됨
            yield int(data)
            if self.last_distance == 0:
                return

asyncio.open_connection(host=None, port=None, *, limit=None, ssl=None, family=0, proto=0, flags=0, sock=None, local_addr=None, server_hostname=None, ssl_handshake_timeout=None, happy_eyeballs_delay=None, interleave=None)

• 스트림을 만들고 작업합니다.
• 네트워크 연결을 만들고, (reader, writer) 객체 쌍을 반환합니다.
  • StreamReader 와 StreamWriter 클래스의 인스턴스 입니다.
• 나머지 인자는 loop.create_connection()으로 직접 전달됩니다.

coroutine loop.create_connection(protocol_factory, host=None, port=None, *, ssl=None, family=0, proto=0, flags=0, sock=None, local_addr=None, server_hostname=None, ssl_handshake_timeout=None, ssl_shutdown_timeout=None, happy_eyeballs_delay=None, interleave=None)

• 주어진 host 와 port로 지정된 주소로의 스트리밍 트랜스포트 연결을 엽니다.

contextlib.contextmanager

• 이 함수는 클래스나 별도의 enter() 와 exit() 메서드를 작성할 필요 없이
• with 문 컨텍스트 관리자를 위한 팩토리 함수를 정의하는데 사용할 수 있는 데코레이터
• yield 문이 반환된다고 보면 된다.
• with 문이 끝날 때 , finally 구문이 실행된다.

async for

• 비동기로 반복하는 문법이다.
• async for 로 동작하는 클래스를 만들려면 aiter와 anext 메서드를 구현해야 한다.
  • aiter : asynchronous iter
  • anext : asynchronous next, 하나씩 꺼내는 값을 만들 때 쓰인다.
• 그리고 메서드를 만들 때는 반드시 async def를 사용해야 한다.