Continuando a série sobre o módulo asyncio do Python 3.4, vou apresentar as corotinas e como elas simplificam a escrita de nossos programas com o loop de eventos.

Modificamos também a chamada de execução da corotina, pois agora utilizamos loop.run_until_complete para iniciar nossa corotina principal. Aproveitamos para colocar tudo entre um try-finally para terminar a execução do loop corretamente (mesmo em caso de exceção). Perceba que no exemplo anterior, com call_soon, passamos a função e seus parâmetros, mas não executamos a função em si. No caso de run_until_complete, estamos passando o retorno da chamada de print_and_repeat que é uma corotina, uma vez que a marcamos com o decorador @asyncio.coroutine.

No post anterior, comparamos a velocidade de execução entre as várias formas de se executar código em paralelo com Python. Agora veremos como usar o asyncio para criar uma aplicação prática, como um cliente e um servidor TCP/IP, indo além dos exemplos da documentação do Python.

Vamos começar pelo servidor. Um servidor TCP/IP é um exemplo clássico de programa chato a escrever. Normalmente, você pode escolher utilizar threads ou se aventurar com select e chamadas não bloqueantes para gerenciar várias conexões. Este problema se agrava em aplicações mais complexas, onde algum processamento precisa ser realizado antes de se gerar a resposta, por exemplo, a um comando do usuário.

Usando o módulo asyncio, esta tarefa fica bem mais fácil. Primeiro, porque o tratamento de dados é gerenciado por uma classe, responsável pelo protocolo. Esta classe traz métodos que são chamados em situações comuns ao programarmos um servidor TCP/IP, como chegada de uma nova conexão, desconexão, chegada de dados para leitura entre outras.

O servidor é controlado por uma classe chamada EchoServer, pois o desenvolvi a partir do servidor de Echo dado como exemplo na documentação, mas com alguns detalhes que observei no código do módulo asyncio. O protocolo implementado é bem simples, a cada linha, a data e hora atuais são enviadas. Se o cliente enviar sair a conexão é terminada.

Vamos ver o programa completo e discutir parte por parte.

A classe EchoServer herda de uma classe fornecida em asyncio.streams, chamada FlowControlMixin. A classe FlowControlMixin é por sua vez derivada de Protocols, também fornecida pelo módulo asyncio. A ideia desta classe é implementar um protocol factory, ou seja, um construtor de instâncias responsáveis pela implementação da gestão de cada nova conexão.

Aproveitamos a nova classe para contar o número de conexões ativas. Cada nova conexão recebida por nosso servidor TCP/IP chama o construtor de nossa classe e o método connection_made, passando o transporte (entenda como o socket já conectado) como parâmetro.

Utilizando o parâmetro transporte, chamamos o método get_extra_info(‘peername’) para obter o endereço do cliente que acabou de se conectar ao servidor. Logo em seguida, incrementamos o número de conexões. Veja que como o código que roda no loop de eventos não é multi-threaded, não precisamos de locks ou de outros mecanismos de controle, já que apenas uma função roda a cada vez.

O resto do método connection_made prepara as instâncias do leitor e do escritor, objetos das classes StreamReader e StreamWriter respectivamente. Estes objetos vão fornecer corotinas úteis para ler e escrever os dados de forma não bloqueante. Veja que passamos transporte tanto para o escritor quanto para o leitor e que uma série de parâmetros são necessários para a inicialização do escritor.

No fim de connection_made, usamos a função asyncio.async para iniciar o processamento da corotina self.gerencia, dentro do loop de eventos. Veja que o método gerencia foi marcado com o decorador @asyncio.coroutine.

O método gerencia contém uma estrutura de repetição while que espera uma linha do cliente. Veja que utilizamos yield from para suspender a execução de gerencia enquanto self.leitor.readline() não terminar. Neste ponto, a execução volta para o loop de eventos e retorna apenas quando self.leitor.readline() contém uma linha enviada pelo cliente ou caso uma exceção tenha ocorrido.

O uso do yield from é fundamental, pois caso o readline() esperasse o cliente enviar a linha para continuar a execução, todo o loop de eventos seria bloqueado. Como o uso do yield from, a execução volta para o loop que é livre para executar outros métodos e outras corotinas. O objetivo é não fazer o computador esperar por dados ou resultados que demoram muito tempo (ou um tempo desconhecido, possivelmente longo, para retornar).

Outra característica de yield from é que o resultado do self.leitor.readline() é retornado e no caso, armazenado na variável dados.

A execução segue normalmente e nosso servidor envia a hora e a data atual, veja que uma linha foi acrescentada ao final da string. Este fim de linha é importante, pois como nosso protocolo é em formato texto e orientado a linhas, esperamos o enter (CR) para processar o comando ou a resposta.

Depois de escrever a resposta, usamos self.escritor.drain() que é uma outra corotina. Esta nova corotina não completa até o que o buffer de escrita seja enviado. Desta forma, podemos garantir que os dados foram enviados (ainda que não possamos ter certeza se estes foram recebidos pelo cliente) antes de continuarmos.

Como usamos Python 3.4, os dados são do tipo byte e não string. A função byteToStr converte de bytes para string, usando a codificação UTF-8. Esta função será apresentada no programa common.py, compartilhando rotinas úteis tanto para nosso cliente quanto para nosso servidor.

Para facilitar o processamento de comandos, retiramos os espaços em branco do início e fim do comando, inclusive enter (CR) e LF, e convertemos o resultado para minúsculas com lower. Se o comando for igual a “sair”, chamamos o método close de self.transporte para encerarmos a conexão. Veja que ao fecharmos a conexão, finalmente retornamos como em uma função normal, utilizando return e terminando assim a execução de nossa corotina gerencia.

O método connection_lost é chamado quando a desconexão do cliente é detectada. O número de conexões ativas é decrementado e o método connection_lost da superclasse é chamado. O parâmetro exp contém None caso seja uma desconexão normal ou a exceção em caso de erro.

Já o método data_received é chamado sempre que dados forem recebidos pelo transporte. Os dados recebidos são passados como parâmetro (dados). Aqui, incluí algumas funções de debug que exibem os dados em formato hexadecimal, string e UTF-8. Estas funções são necessárias para verificarmos se os dados estão chegando no formato esperado.

Um detalhe muito importante de data_received é a chamada do método self.leitor.feed_data, que envia os dados para o leitor, responsável por quebrar os dados em linhas.

Em nosso programa principal, obtemos o loop de eventos com get_event_loop() e criamos uma corotina que inicializa nosso servidor com create_server. Em create_server, informamos o endereço que nosso servidor irá escutar (ip e porta). Veja que a classe EchoServer foi passada como protocol factory.

Ao chamarmos loop.run_until_complete(coro), o loop de eventos roda até que a corotina criada pelo create_server termine, retornando um objeto servidor, utilizado para parar o servidor e para ter acesso a todas as conexões, mas isso fica para outro post.

Chamamos loop.run_forever() para ativar nosso servidor. Para desativá-lo, digite CTRL+C.

Neste ponto, o endereço 127.0.0.1, porta 8888 estará recebendo conexões. Quanto uma conexão for recebida, uma nova instância de EchoServer será criada. Ao se estabelecer a conexão o método connection_made será chamado e ativará uma corotina gerencia para gerenciar a recepção e o envio de linhas de comandos.

Vejamos o código fonte do common.py:

import string

def byteToHex(data, sep=" "):
    return sep.join("{0:02X}".format(x) for x in data)

def strPrintable(data, sep=" "):
    return sep.join("{0:2s}".format(chr(s) if chr(s) in string.printable and s>30 else ".") for s in data)

def strToByte(s, encoding ="utf-8"):
    return s.encode(encoding)

def byteToStr(data, encoding ="utf-8"):
    return data.decode(encoding, errors="replace")

E o código fonte do cliente.py:

import asyncio
import time
from common import *

class EchoClient(asyncio.streams.FlowControlMixin):
    def connection_made(self, transporte):
        peername = transporte.get_extra_info('peername')
        print('Conectado à {}'.format(peername))
        self.transporte = transporte
        self.leitor = asyncio.StreamReader()
        self.leitor.set_transport(self.transporte)
        self.escritor = asyncio.StreamWriter(transport=self.transporte, protocol=self, reader=self.leitor, loop=asyncio.get_event_loop())
        asyncio.async(self.gerencia())
        self.feito = asyncio.Future()

    @asyncio.coroutine
    def gerencia(self):
        for x in range(10):
            self.escritor.write(strToByte("Alô\r\n"))
            yield from self.escritor.drain()

            dados = yield from self.leitor.readline()
        self.escritor.write(strToByte("sair\r\n"))
        yield from self.escritor.drain()
        self.transporte.close()
        self.feito.set_result(True)

    def connection_lost(self, exp):
        print("Conexão perdida")
        super().connection_lost(exp)

    def data_received(self, dados):
        print('dados recebidos: {0}'.format(byteToHex(dados)))
        print('      recebidos: {0}'.format(strPrintable(dados)))
        print('         string: {0}'.format(byteToStr(dados)))
        self.leitor.feed_data(dados)


loop = asyncio.get_event_loop()
coro = loop.create_connection(EchoClient, '127.0.0.1', 8888)
transporte, protocolo = loop.run_until_complete(coro)

try:
    loop.run_until_complete(protocolo.feito)
except KeyboardInterrupt:
    pass
finally:
    loop.close()

Execute o servidor e depois o cliente, cada em uma janela ou terminal diferente. Veja que o cliente termina sua execução após enviar 10 vezes o comando Alô e sair.

No próximo artigo, vamos apresentar como usar as corotinas para criar um servidor TCP/IP que possa lidar com vários clientes simultaneamente.


+++ title = “Asyncio e Coroutines” slug = “2014-08-11-asyncio-e-corotinas-multiplos-clientes” published = 2014-08-11T12:43:00.001000-04:00 author = “Nilo Menezes” tags = [ “python”, “asyncio”, “python 3” ] +++

No último post, apresentamos um servidor TCP/IP que utiliza o módulo asyncio para lidar com a execução de corotinas. Neste post, vamos mostrar como usar as corotinas para criar um servidor TCP/IP que possa lidar com vários clientes simultaneamente.

Para isso, precisaremos utilizar a classe asyncio.start_server para criar um novo loop de eventos e conectar-se ao servidor criado anteriormente.

Vamos começar por criar uma nova classe chamada EchoServerMulti, que será responsável por gerenciar as conexões com vários clientes simultaneamente.

import asyncio

class EchoServerMulti:
    def __init__(self, loop):
        self.loop = loop

    async def handle_connection(self, reader, writer):
        peername = writer.get_extra_info('peername')
        print(f'Conectado a {peername}')

        try:
            while True:
                data = await reader.read(1024)
                if not data:
                    break
                writer.write(data)
                await writer.drain()
        finally:
            writer.close()

    async def start(self):
        server = await asyncio.start_server(
            self.handle_connection,
            '127.0.0.1',
            8888,
        )

        print(f'Servidor iniciado em {server.sockets[0].getsockname()}')

        async with server:
            await server.serve_forever()

Nesta classe, a função handle_connection é chamada para cada nova conexão que chega ao servidor. Essa função lê os dados do cliente e os envia de volta.

Agora vamos criar uma função main para iniciar o servidor:

async def main():
    server = EchoServerMulti(asyncio.get_event_loop())
    await server.start()

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    try:
        loop.run_until_complete(main())
    finally:
        loop.close()

Nesta função, é criado um novo objeto EchoServerMulti e iniciado o servidor.

Para testar o servidor com vários clientes simultaneamente, vamos criar uma nova classe Client:

class Client:
    def __init__(self):
        self.reader = None
        self.writer = None

    async def start(self):
        writer = await asyncio.open_connection('127.0.0.1', 8888)
        print(f'Conectado a {writer.get_extra_info("peername")}')
        self.reader, self.writer = writer

    async def send_message(self, message):
        self.writer.write(message.encode())
        await self.writer.drain()

    async def receive_message(self):
        return await self.reader.read(1024)

client = Client()
asyncio.run(client.start())
while True:
    message = input('Digite uma mensagem: ')
    if message.lower() == 'sair':
        break
    asyncio.run(client.send_message(message))
    print(await client.receive_message().decode())

Nesta classe, a função start é chamada para iniciar a conexão com o servidor. A função send_message envia uma mensagem ao servidor e a função receive_message recebe uma resposta do servidor.

Agora vamos executar o cliente e enviar mensagens ao servidor:

python client.py

O cliente vai conectar-se ao servidor e enviar mensagens até que você digite ‘sair’.

Essa é uma forma simples de criar um servidor TCP/IP que pode lidar com vários clientes simultaneamente usando as corotinas.


+++