Journal SeqTools 1.0.0: la programmation concurrente, c'est dur!

Sommaire

• Le multi-processing/multi-threading c'est difficile!
• Extension en C pour des transferts sans copies
• Pour la suite

J'ai profité des vacances pour améliorer ma librairie en python SeqTools, déjà présentée sur linuxfr.

Pour rappel, cette librairie permet le traitement "paresseux" de séquences, c'est-à-dire de tout conteneur qui permet l'accès à ses éléments par indexation, en gros des listes, des tableaux ou un objet qui implémente __getitem__.
La librairie se destine plutôt à des transformations éléments par éléments ou à la réorganisation/combinaison d'une ou de plusieurs séquences entre elles. Pour les pythonistes: c'est un peu le pendant de itertools pour les séquences.
L'intérêt du traitement paresseux, c'est que l'on peut travailler aisément sur des jeux de données qui ne rentreraient pas en mémoire ou prendraient beaucoup de temps si on devait appliquer les transformations pour tous les éléments, alors que l'on ne souhaite parfois en évaluer que quelques-uns (cf le journal précédent).
Pour avoir une meilleure idée de ce dont il retourne, vous pouvez jeter un œil aux exemples de la doc.

Au menu des nouveautés pour cette v1, c'est surtout une refonte de la partie évaluation des éléments avec des processus/threads séparés, qui permet d'utiliser plusieurs cœurs quand on veut lire beaucoup d'éléments successifs.
À noter que j'ai retiré le support pour python 2.

Le multi-processing/multi-threading c'est difficile!

Une des fonctionnalités les plus avancées de seqtools, c'est la fonction prefetch qui calcule la valeur des éléments de manière asynchrone en utilisant des processus ou des threads. Concrètement, si on demande l'élément 10, alors 11, 12, 13, etc. sont aussi envoyés dans la file des tâches et seront immédiatement/plus rapidement disponibles ensuite.

En pratique, c'est transparent pour l'utilisateur:

def f1(x):
sleep(0.005 * (1 + random.random()))
return x

x = list(range(1000))
y = seqtools.smap(f1, x) # applique f1 à tous les éléments de x
y = seqtools.prefetch(y, nworkers=4, max_buffered=16, method="thread")

# notez l'utilisation de l'index ci-dessous: on peut toujours accéder
# à des éléments aléatoirement, on perdra simplement l'accès plus rapide
# aux éléments précalculés
for i in range(1000):
    print(y[i])

La difficulté principale ne vient étonnement pas du caractère asynchrone des réponses qu'il faut réorganiser (ça se résout en une vingtaine de lignes), le problème c'est les problèmes! Ou comment gérer les plantages des workers, du processus parent, de la communication, etc.
Précédemment, les workers n'étaient pas en mode démon, donc le processus parent attendait que les workers quittent pour s'arrêter à son tour. Côté workers, il y avait un timeout pour quitter automatiquement en cas d'inactivité ce qui évite de bloquer indéfiniment à la fin du script. J'ai choisi cette approche car je ne voulais pas prendre le risque de laisser traîner des processus de worker zombies.

À l'usage, cette approche s'est révélée une fausse bonne idée:

• Il faut relancer les processus qui ont atteint leur timeout (par exemple si le programme prend du temps avant de lire une nouvelle valeur de la séquence). Ça ajoute un délai et le fork qui re-crée un worker à un endroit imprévisible.
• Si le processus parent plante (segfault, sigbus, sigterm…) les enfants peuvent rester bloqués sur la communication avec le parent (ça devrait pas être le cas, mais la bibliothèque multiprocessing n'est pas spécialement endurcie contre les plantages).
• Si le timeout est grand, ça retarde d'autant la fin du script quand on fait un ctrl-c pour l'interrompre.

La nouvelle version de SeqTools utilise donc la stratégie inverse où les workers vérifient ponctuellement si le parent est toujours vivant.
Ils finiront donc par quitter naturellement après la fin du script principal. Par ailleurs, j'utilise désormais un Pipe plutôt que multiprocessing.Manager pour communiquer les valeurs, ce qui s'avère à la fois plus rapide et plus robuste aux plantages.

Extension en C pour des transferts sans copies

Pour communiquer au parent une valeur de la séquence calculée par le worker, il faut la sérialiser, l'envoyer dans le tuyau, la réceptionner et la décoder. Ces opérations ajoutent un surcoût assez important alors que l'on manipule souvent de simples tableaux avec une taille fixe bien connue. Pour accélérer ce cas d'usage, seqtools permet désormais d'utiliser de la mémoire partagée où les valeurs sont écrites par les workers et lues par le processus principal.

Comme à l’accoutumée, quelques écueils ont séparé l'idée de l'implémentation:

• Comme la mémoire partagé ne s'alloue pas à la volée, on peut seulement créer un buffer partagé unique qui stockera à tout instant un nombre limité d'éléments de la séquence.
• Pour s’accommoder de l'espace limité, il suffit de mettre en place du recyclage: quand une entrée de la mémoire partagée n'est plus utilisée par le script principal, elle est remise en service côté workers pour recevoir une nouvelle valeur de la séquence.
• Bonne nouvelle: Python est un si bon langage qu'il dispose d'un destructeur, il suffit donc de surveiller quand une ancienne entrée est détruite pour déclencher son recyclage.
• Mauvaise nouvelle: À partir du tableau en mémoire partagée, un utilisateur peut créer une vue (ex: memoryview ou l'équivalent numpy) qui pointe directement vers la mémoire sous-jacente; le tableau que j'ai renvoyé peut donc être supprimé alors que la mémoire est toujours en service. C'est une fonctionnalité des objets implémentant l'interface buffers, laquelle est largement exploitée par les librairies de manipulation de tableaux comme numpy.
• Bonne nouvelle: on peut aussi suivre la création et la suppression des vues!
• Mauvaise nouvelle: c'est seulement au niveau de l'API des extensions en C (plus précisément ici), du coup j'ai dû coder un peu en C.

Pour la suite

J'ai décidé de passer le projet en v1 car il contient tout ce dont j'ai besoin pour mon travail mais je peux très bien ajouter des fonctionnalités. Il y a une assez une bonne couverture du code par des tests, ce qui m'a bien aidé pour chasser les bugs, mais je vous invite à me contacter si vous en trouvez encore.