Echanges E/S asynchrones avec Python 3

French (Français) translation by Stéphane Esteve (you can also view the original English article)

Dans ce tutoriel, vous graviterez autour des fonctionnalités de communication asynchrone, développées à partir de la version 3.4 de Python, et augmentées depuis les versions 3.5 et 3.6.

Auparavant, Python contenait déjà quelques outils pour ce type de programmation. Le support des échanges asynchrones projette ce langage vers une exploitation optimisée, incluant des API de haut niveau et une base standard, permettant désormais d'unifier l'ensemble des solutions pré-existantes (Twisted, Gevent, Tornado, asyncore, etc.).

Notez qu'il est essentiel de comprendre que l'apprentissage des concepts asynchrones en Python sont loin d'être aisés, dû à son développement constant, son étendue davantage élargie et la prise en compte des évolutions des frameworks asynchrones déjà existants. Je me concentrerai sur les dernières innovations pour en faciliter légèrement sa compréhension.

Il y a tellement d'évolutions passionnantes sur le terrain des threads, des processus et des machines distantes. Et aussi de différences et de limitations liées aux spécificités d'une plateforme. Allons donc à sa découverte.

Les boucles d'événements raccordables

La boucle d'événement est le concept central des échanges asynchrones. Il y en a plusieurs au sein d'un programme. Chaque thread contient au moins une "event loop" active et offre les fonctions suivantes :

Déclarer, exécuter et annuler des appels différé (avec des clauses de timeouts).
Créer des passerelles client et serveur pour différentes sortes de communication.
Déclencher des sous-processus et les associer à des passerelles pour communiquer avec un programme externe.
Déléguer des exécutions de fonctions coûteuses vers un groupe de threads.

Un court exemple

Voici ici un petit exemple qui démarrent deux co-routines et exécute une fonction en différé. Il s'agit d'une démonstration de l'utilisation d'une event loop pour améliorer votre programme :

import asyncio


async def foo(delay):
    for i in range(10):
        print(i)
        await asyncio.sleep(delay)


def stopper(loop):
    loop.stop()


loop = asyncio.get_event_loop()

# Schedule a call to foo()
loop.create_task(foo(0.5))
loop.create_task(foo(1))
loop.call_later(12, stopper, loop)

# Block until loop.stop() is called()
loop.run_forever()
loop.close()

La classe AbstractEventLoop délivre le même principe à toute event loop. Plusieurs choses lui sont nécessaires pour fonctionner :

Hiérarchiser les fonctions et les co-routines pour toute exécution
Créer des fonctions anticipatrices et les tâches
Gérer des serveurs TCP
Manipuler des signaux (sur Unix)
Manipuler des pipes et des sous-processus

Voici les méthodes relatives à l'exécution et à l'arrêt des événements, ainsi qu'à la coordination des fonctions et des co-routines :

class AbstractEventLoop:
    """Abstract event loop."""

    # Running and stopping the event loop.

    def run_forever(self):
        """Run the event loop until stop() is called."""
        raise NotImplementedError

    def run_until_complete(self, future):
        """Run the event loop until a Future is done.

        Return the Future's result, or raise its exception.
        """
        raise NotImplementedError

    def stop(self):
        """Stop the event loop as soon as reasonable.

        Exactly how soon that is may depend on the implementation, but
        no more I/O callbacks should be scheduled.
        """
        raise NotImplementedError

    def is_running(self):
        """Return whether the event loop is currently running."""
        raise NotImplementedError

    def is_closed(self):
        """Returns True if the event loop was closed."""
        raise NotImplementedError

    def close(self):
        """Close the loop.

        The loop should not be running.

        This is idempotent and irreversible.

        No other methods should be called after this one.
        """
        raise NotImplementedError

    def shutdown_asyncgens(self):
        """Shutdown all active asynchronous generators."""
        raise NotImplementedError

    # Methods scheduling callbacks.  All these return Handles.

    def _timer_handle_cancelled(self, handle):
        """Notification that a TimerHandle has been cancelled."""
        raise NotImplementedError

    def call_soon(self, callback, *args):
        return self.call_later(0, callback, *args)

    def call_later(self, delay, callback, *args):
        raise NotImplementedError

    def call_at(self, when, callback, *args):
        raise NotImplementedError

    def time(self):
        raise NotImplementedError

    def create_future(self):
        raise NotImplementedError

    # Method scheduling a coroutine object: create a task.

    def create_task(self, coro):
        raise NotImplementedError

    # Methods for interacting with threads.

    def call_soon_threadsafe(self, callback, *args):
        raise NotImplementedError

    def run_in_executor(self, executor, func, *args):
        raise NotImplementedError

    def set_default_executor(self, executor):
        raise NotImplementedError

Raccorder à une nouvelle Event Loop

Asyncio est conçu pour supporter diverses implémentations d'event loop qui respectent le corps de son API. A la base, se trouve la classe EventLoopPolicy qui configure l'API asyncio et facilite le contrôle de chaque aspect de l'event loop. Voici ici un exemple d'une event loop personalisée appelée uvloop et basée sur libuv, qui est sensée être bien plus rapide que d'autres solutions (je ne l'ai pas évalué moi-même) :

1	import asyncio
2	import uvloop
3	asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

C'est tout. Désormais, quelle que soit la fonction asyncio utilisée, c'est uvloop qui s'en chargera discrètement.

Co-routines, Futures et Tâches

Une co-routine est un terme chargé de sens. Tantôt, c'est une fonction qui s'exécute de manière asynchrone ; tantôt un objet qui a besoin d'être hiérarchisé. Vous la chargez en y ajoutant le mot-clé async devant sa définition.

import asyncio


async def cool_coroutine():
    return "So cool..."


Si vous l'exécutez directement, rien ne se passe. A la place un objet co-routine est renvoyé, et si aucune hiérarchisation n'est faite au cours de son exécution, voici l'avertissement obtenu :

c = cool_coroutine()
print(c)

Output:

<coroutine object cool_coroutine at 0x108a862b0>
sys:1: RuntimeWarning: coroutine 'cool_coroutine' was never awaited

Process finished with exit code 0

Pour déclencher son exécution, vous avez besoin d'une event loop :

r = loop.run_until_complete(c)
loop.close()

print(r)

Output:

So cool...

Il s'agit d'un séquençage direct. Vous pouvez aussi les lier les unes aux autres. Remarquez que vous devez inscrire await dès lors que vous invoquez les co-routines :

import asyncio

async def compute(x, y):
    print("Compute %s + %s ..." % (x, y))
    await asyncio.sleep(1.0)
    return x + y

async def print_sum(x, y):
    result = await compute(x, y)
    print("%s + %s = %s" % (x, y, result))

loop = asyncio.get_event_loop()
loop.run_until_complete(print_sum(1, 2))
loop.close()

La class Future d'asyncio est similaire à la classe concurrent.future.Future. Ce n'est pas garanti au niveau thread et prend en charge les fonctionnalités suivantes :

ajouter ou retirer des callbacks terminés
annuler
formater les résultats et les exceptions

Voici la façon dont on peut utiliser un Future dans une event loop. La co-routine take_your_time() reçoit un Future et ajuste son résultat après avoir attendu une seconde.

La fonction ensure_future() planifie la co-routine et wait_until_complete() patiente jusqu'à la fin du Future. En fait, sous le capot, il ajoute un callback terminé au Future.

import asyncio

async def take_your_time(future):
    await asyncio.sleep(1)
    future.set_result(42)

loop = asyncio.get_event_loop()
future = asyncio.Future()
asyncio.ensure_future(take_your_time(future))
loop.run_until_complete(future)
print(future.result())
loop.close()

Tout ceci reste assez lourd. Asyncio délivre des tâches adaptées aux Futures et aux co-routines. Une Tâche est une sous-classe de Future, englobant une co-routine et que vous pouvez arrêter à tout moment.

La co-routine n'a plus qu'à adopter une fonction Future explicite et ajuste son résultat ou son exception. Même exemple qu'auparavant, mais cette fois, avec une tâche :

import asyncio

async def take_your_time():
    await asyncio.sleep(1)
    return 42

loop = asyncio.get_event_loop()
task = loop.create_task(take_your_time())
loop.run_until_complete(task)
print(task.result())
loop.close()

Passerelles, protocoles et flux

Une passerelle est une abstraction d'un canal de communication. Elle s'appuie toujours sur un protocole particulier. Asyncio contient un certain nombre d'implémentations clé-en-main pour le TCP, l'UDP, le SSL et aussi des sous-processus reliés.

Si vous connaissez déjà la programmation de réseaux basés sur les sockets, alors les passerelles et les protocoles n'ont plus aucun secret pour vous. Avec Asyncio, vous avez accès à la programmation standard de réseaux asynchrones. Reprenons l'infâme exemple du serveur d'écho et de son client (le "hello world" de la programmation réseau).

Premièrement, l'application client définit une classe appelée EchoClient , déclinaison de asyncio.Protocol. Elle conserve une event loop et un message qu'elle transmettra au serveur au cours de sa connexion.

Dans le callback connection_made(), elle inscrit son message à la passerelle. Dans la méthode data_received(), elle affiche la réponse du serveur, et par la méthode connection_lost() elle stoppe l'event loop. Dès qu'une instance d'EchoClient est définie dans la méthode create_connection(), le résultat obtenu est une co-routine que la boucle exécute tant qu'elle n'est pas terminée.

import asyncio

class EchoClient(asyncio.Protocol):
    def __init__(self, message, loop):
        self.message = message
        self.loop = loop

    def connection_made(self, transport):
        transport.write(self.message.encode())
        print('Data sent: {!r}'.format(self.message))

    def data_received(self, data):
        print('Data received: {!r}'.format(data.decode()))

    def connection_lost(self, exc):
        print('The server closed the connection')
        print('Stop the event loop')
        self.loop.stop()

loop = asyncio.get_event_loop()
message = 'Hello World!'
coro = loop.create_connection(lambda: EchoClient(message, loop),
                              '127.0.0.1', 8888)
loop.run_until_complete(coro)
loop.run_forever()
loop.close()  

Le serveur reprend le même code, à l'exception qu'il s'exécute perpétuellement, attendant que des clients s'y connectent. Après avoir envoyé un signal contenant sa réponse, il coupe la connexion vers le client et il peut répondre au client suivant.

Une nouvelle instance de l'EchoServer est créée à chaque connexion. Ainsi, même si plusieurs clients se connectent en même temps, il n'y aura absolument aucun conflit grâce à l'attribut transport.

import asyncio

class EchoServer(asyncio.Protocol):
    def connection_made(self, transport):
        peername = transport.get_extra_info('peername')
        print('Connection from {}'.format(peername))
        self.transport = transport

    def data_received(self, data):
        message = data.decode()
        print('Data received: {!r}'.format(message))

        print('Send: {!r}'.format(message))
        self.transport.write(data)

        print('Close the client socket')
        self.transport.close()

loop = asyncio.get_event_loop()
# Each client connection will create a new protocol instance
coro = loop.create_server(EchoServer, '127.0.0.1', 8888)
server = loop.run_until_complete(coro)
print('Serving on {}'.format(server.sockets[0].getsockname()))
loop.run_forever()

Voici ce qu'il affiche, suite à deux connexions clients :

Serving on ('127.0.0.1', 8888)
Connection from ('127.0.0.1', 53248)
Data received: 'Hello World!'
Send: 'Hello World!'
Close the client socket
Connection from ('127.0.0.1', 53351)
Data received: 'Hello World!'
Send: 'Hello World!'
Close the client socket

Les flux nécessitent une API de haut-niveau, issues de co-routines, et produisent des abstraction de type Reader et Writer. Les protocoles et les passerelles sont transparents, ne nécessitant aucune déclaration de classe, ni aucun callbacks. Seuls sont attendus des événements de type connexion ou donnée.

Le client fait appel à la fonction open_connection() qui renvoit un objet reader et un objet writer, naturellement hérités. Pour fermer la connexion, elle détruit l'objet writer.

import asyncio


async def tcp_echo_client(message, loop):
    reader, writer = await asyncio.open_connection(
        '127.0.0.1', 
        8888, 
        loop=loop)

    print('Send: %r' % message)
    writer.write(message.encode())

    data = await reader.read(100)
    print('Received: %r' % data.decode())

    print('Close the socket')
    writer.close()


message = 'Hello World!'
loop = asyncio.get_event_loop()
loop.run_until_complete(tcp_echo_client(message, loop))
loop.close()

Le serveur est aussi davantage épuré.

import asyncio

async def handle_echo(reader, writer):
    data = await reader.read(100)
    message = data.decode()
    addr = writer.get_extra_info('peername')
    print("Received %r from %r" % (message, addr))

    print("Send: %r" % message)
    writer.write(data)
    await writer.drain()

    print("Close the client socket")
    writer.close()

loop = asyncio.get_event_loop()
coro = asyncio.start_server(handle_echo, 
                            '127.0.0.1', 
                            8888, 
                            loop=loop)
server = loop.run_until_complete(coro)
print('Serving on {}'.format(server.sockets[0].getsockname()))
loop.run_forever()

Travailler avec des sous-processus

Asyncio renferme également des interactions possibles avec les sous-processus. Le programme ci-dessous déclenche un autre processus Python et exécute le code "import this". C'est une des pochette-surprises connu de Python, qui affiche le "Zen of Python". Constatez vous-même.

Le processus Python est déclenché par la co-routine zen() en utilisant la fonction create_subprocess_exec() et renvoie sa sortie standard à un autre conduit. Ainsi, il parcoure ce contenu ligne-à-ligne via await pour laisser aux autres processus ou co-routines la possibilité de s'exécuter si cette sortie n'est pas totalement conclue.

Remarquez que sous Windows, vous devez inscrire l'event loop au ProactorEventLoop , car la fonction SelectorEventLoop ne prend pas en charge les pipes.

import asyncio.subprocess
import sys


async def zen():
    code = 'import this'
    create = asyncio.create_subprocess_exec(
        sys.executable, 
        '-c', 
        code,
        stdout=asyncio.subprocess.PIPE)
    proc = await create

    data = await proc.stdout.readline()
    while data:
        line = data.decode('ascii').rstrip()
        print(line)
        data = await proc.stdout.readline()

    await proc.wait()

if sys.platform == "win32":
    loop = asyncio.ProactorEventLoop()
    asyncio.set_event_loop(loop)
else:
    loop = asyncio.get_event_loop()

loop.run_until_complete(zen())

Output:

The Zen of Python, by Tim Peters

Beautiful is better than ugly.
Explicit is better than implicit.
Simple is better than complex.
Complex is better than complicated.
Flat is better than nested.
Sparse is better than dense.
Readability counts.
Special cases aren't special enough to break the rules.
Although practicality beats purity.
Errors should never pass silently.
Unless explicitly silenced.
In the face of ambiguity, refuse the temptation to guess.
There should be one-- and preferably only one --obvious way to
do it.
Although that way may not be obvious at first unless you're
Dutch.
Now is better than never.
Although never is often better than *right* now.
If the implementation is hard to explain, it's a bad idea.
If the implementation is easy to explain, it may be a good idea.
Namespaces are one honking great idea -- let's do more of those!

Conclusion

N'hésitez pas à consulter ou acheter ce qu'il y a de disponible sur le sujet sur notre boutique, et contactez-nous si vous avez la moindre question. Nous sommes aussi attentifs à vos remarques, que vous pouvez laisser sur le formulaire ci-dessous.

Python Asyncio est un framework complet pour toute programmation asynchrone. Ses possibilités sont immenses et il pourvoit à la fois des API de bas et haut niveaux. Il est certes encore jeune et mal compris par la communauté.

Je suis persuadé qu'avec le temps, de bonnes conduites émergeront, et davantage d'exemples feront surface et rendront son utilisation plus poussée.