Architectures distribuées et traitements asynchrones

Cela fait quelque temps que je réfléchis à l’idée d’écrire un article sur les architectures distribuées, et plus particulièrement sur les traitements asynchrones. J’en parlais déjà au Forum PHP en 2013, dans de ma conférence intitulée «De 0 à 10 millions de visiteurs uniques avec les moyens d’une startup» mais je vais entrer plus en profondeur dans le sujet.

Pour quoi faire ?

On peut vouloir faire des traitements asynchrones pour différentes raisons. Si on s’en tient au développement web, la principale raison est de répondre plus vite aux requêtes des visiteurs, en exécutant certaines actions de manière différée. L’exemple classique est lorsqu’une action sur votre site doit déclencher un appel sur l’API d’un service tiers : Cet appel d’API peut prendre plusieurs secondes, et il peut même échouer en cas de problème momentané du service en question ; dans ce cas, plutôt que de faire attendre notre internaute, il vaut mieux lui répondre immédiatement et traiter cet appel d’API de manière asynchrone. Pareil pour des envois d’emails (même si les serveurs SMTP opèrent déjà de cette manière), des traitements sur des fichiers audio/vidéo, des calculs lourds, etc.

Si on élargit le débat au-delà du web, c’est un bon moyen pour découpler les différentes briques qui composent un logiciel afin de mettre en place une architecture distribuée. C’est souvent un élément essentiel lorsqu’on a besoin de scalabilité et même de rapidité. Un système monolithique atteindra tôt ou tard ses limites, là où un système composé de plusieurs entités qui communiquent ensemble sera plus facile à faire évoluer − au prix d’une complexité initiale plus grande.
En éclatant le code applicatif en plusieurs morceaux, on peut distribuer son exécution sur plusieurs machines assez facilement, facilitant ainsi l’évolutivité de la plate-forme.

Concrètement, l’idée est d’avoir des entités qui s’envoient des messages au lieu de discuter entre elles directement. Quand un message est envoyé, on n’attend pas une réponse immédiate ; une fois que le message aura été traité, un autre message sera éventuellement envoyé en retour pour le notifier, à moins que le résultat ne soit par exemple stocké dans une base de données où il sera disponible pour de futurs besoins.

Le sujet des architectures distribuées n’est pas nouveau. L’idée de RPC (appel à procédure distante) a été décrite dès 1976. Dans les années 90, quand les langages orientés objet étaient à la mode, beaucoup d’énergie a été déployée dans les systèmes servant à exécuter des objets à distance. C’est dans ce contexte que sont apparus des ORB (object request brokers) comme CORBA dès 1991 et RMI (Remote Method Invocation) intégré à Java dès la version 1.1 en 1997.
D’ailleurs, la complexité de CORBA a facilité l’émergence de normes plus simples et basées sur des technos web, comme XML-RPC et SOAP (Simple Object Access Protocol) en 1998, elles-mêmes simplifiées avec l’apparition du REST en 2000 (qui s’est développé d’autant plus avec l’usage du JSON, sérialisation plus simple à manipuler que le XML).

Mais comme on va le voir, il ne suffit pas de considérer qu’il y a des procédures/méthodes/objets à appeler pour que le système soit robuste en toutes circonstances. On peut d’ailleurs remarquer que les illusions de l’informatique distribuée sont bien connues et listées depuis les années 90 :

  1. Le réseau est fiable.
  2. Le temps de latence est nul.
  3. La bande passante est infinie.
  4. Le réseau est sûr.
  5. La topologie du réseau ne change pas.
  6. Il y a un et un seul administrateur réseau.
  7. Le coût de transport est nul.
  8. Le réseau est homogène.

Effectivement, rien de tout cela n’est vrai. Découpler une application apporte beaucoup d’avantages, mais comporte aussi des contraintes et des risques qu’il faut connaître et évaluer.

Comment faire ?

Pour mettre en place ce type d’architecture, il existe un certain nombre de moyens techniques, depuis les plus simples jusqu’aux plus complexes. Je vais passer en revue quelques notions et solutions.

Quand on parle de traitements asynchrones, on a habituellement trois composants à mettre en place :

  • Une partie qui permet de donner des ordres de traitements, avec éventuellement des informations complémentaires (de quel type d’ordre il s’agit, avec parfois des données additionnelles, etc.). Cela prend habituellement la forme d’une simple bibliothèque logicielle, mais certaines technos imposent plus de contraintes − comment les ORB qui obligeaient à créer des stubs.
  • La pièce centrale du système, qui gère les messages qui sont envoyés, en les classant éventuellement par priorité ou en les rangeant dans différentes files. Cette partie peut être basée sur des technologies variées (bases de données, files de messages, tables de hachage, ou autre).
    Les technos utilisées, que ce soit pour gérer les messages ou pour communiquer entre les différentes briques de l’architecture, ont une influence directe sur les performances du système.
  • La dernière partie est constituée par des «workers», c’est-à-dire du code qui traite les messages et effectue le travail demandé. Là encore, il existe plusieurs manières de faire : soit les workers sont des programmes indépendants (qu’il faut alors gérer correctement), soit ils sont lancés par la brique centrale (au prix d’un coût d’instanciation qui ajoute de la latence dans les traitements).

Dans les notions qui me semblent importantes à avoir en tête, il y en a principalement deux :

  • La scalabilité : Est-ce que la solution retenue peut soutenir un très grand nombre de messages échangés, et est-il possible de répartir leur traitement sur plusieurs ordinateurs ?
  • La persistance : Si la partie centrale de l’architecture venait à tomber en panne, est-ce que les messages qui étaient en attente de traitement sont perdus ou bien sont-ils conservés pour être traités au redémarrage du système ?

Pour la mise en œuvre, je vais explorer 2 manières de faire, chacune avec 2 technos différentes :

  • L’utilisation d’une base de données dans laquelle on écrit les messages, avec un système de polling (“attente active” ou “scrutation” en français) pour instancier leurs traitements.
    1. De manière artisanale, avec l’utilisation de la crontab.
    2. Avec l’outil Resque.
  • L’utilisation d’un système intégrant une file de messages.
    1. Avec l’outil Gearman.
    2. Avec l’outil Beanstalkd.

Base de données ou file de messages ?

Parmi les quatre méthodes que je vais détailler, trois d’entre elles reposent d’une manière ou d’une autre sur l’utilisation d’une base de données. Continuer la lecture de « Architectures distribuées et traitements asynchrones »

FineDB : Architecture interne

J’ai lancé le projet FineDB il y a quelques semaines. Je vais expliquer comment fonctionnent les mécanismes internes du serveur.

Pour info, le projet a un site dédié : finedb.org

Threads

Il existe 3 types de threads dans le serveur FineDB :

  • Le thread principal, qui crée les autres threads puis écoute les nouvelles connexions venant des clients.
  • Le thread d’écriture, dont le rôle est d’écrire dans le moteur de stockage les données qui doivent être écrites de manière asynchrone.
  • Les threads de communication. Ils gèrent les échanges avec les clients, en interprétant leurs requêtes. Les lectures et les écritures synchrones sont faites directement ; les écritures asynchrones sont déléguées au thread d’écriture.

Communication entre les threads

Il y a deux flux de communication nanomsg à l’intérieur du serveur.

Le premier est un flux de type “fanout” PUSH/PULL load-balancé. Le thread principal y envoie les files descriptors correspondant aux sockets des connexions entrantes. Nanomsg distribue automatiquement ces messages aux threads de connexion, qui peuvent ainsi entamer la communication avec les clients.

Le second est un flux “fanin” SOURCE/SINK. Les threads de communication l’utilisent pour envoyer des ordres au thread d’écriture.

Schéma

  1. Une socket TCP classique attend de nouvelles connexions entrantes.
  2. Connexion nanomsg utilisée pour transmettre les connexions entrantes aux threads de connexion.
  3. Les opérations de lecture sont faites directement sur le moteur de stockage.
  4. Les opérations d’écritures synchrones sont faites directement sur le moteur de stockage.
  5. Connexion nanomsg, utilisée pour transmettre les ordres d’écriture asynchrone au thread d’écriture.
  6. Le thread d’écriture accède directement au moteur de stockage.

État des lieux

Tout ce que j’ai décrit ci-dessus est déjà implémenté sous cette exacte forme.

Cette architecture est le fruit d’un choix. Le fait de passer par des threads qui prennent en charge les connexions entrantes est différent de la mode actuelle qui se base plutôt sur de la programmation événementielle mono-processus (à base de libevent ou libev).
Le grand avantages est une simplification importante du code, grâce à son découplage. Le thread principal écoute les nouvelles connexions, et une fois qu’elle sont établies il les envoie aux threads de connexion. Nanomsg facilite grandement cette communication, en jouant au passage les rôles de load-balancer et de file de message. Les threads de connexion ont pour seul tâche de lire les requêtes, les comprendre et les exécuter.
Il faut voir aussi que la programmation réseau mono-processus fonctionne bien tant qu’aucune opération bloquante ne peut ralentir l’ensemble des traitements. Dans le cas de FineDB, certains traitements (compression/décompression des données, accès au moteur de stockage) peuvent s’exécuter rapidement mais quand même trop lentement pour vouloir prendre le risque d’ajouter de la latence sur toutes les autres connexions.

L’inconvénient principal est que le nombre de requêtes traitées simultanément est égal au nombre de threads de connexion. Les connexions supplémentaires doivent attendre qu’une connexion se termine et que le thread devienne libre pour en prendre une nouvelle en charge.
Cet inconvénient est connu et assumé. Les threads sont très légers et un grand nombre peut être créé.

Évolutions futures

L’architecture actuelle va évoluer avec deux étapes.

La première, assez simple, consistera à gérer les connexions longues, pour les couper au bout d’un certain délai. Cela nécessitera un thread supplémentaire, qui parcourra à intervalle régulier la liste des connexions ouvertes et fermera celles qui n’ont pas montré signe d’activité depuis un certain temps.

La seconde sera plus complexe. Elle va servir à mettre en place les capacités de réplication (maître-esclave et maître-maître) de FineDB. J’en reparlerai dans un autre article, mais cela nécessitera d’ajouter encore un thread supplémentaire, qui gérera les interactions avec les autres serveurs.

Une autre piste d’amélioration serait de mixer programmation réseau événementielle et threads de traitement. Ainsi, le thread principal gérerait à lui seul les connexions ouvertes, mais confierait aux threads les opérations bloquantes. À voir si cela est nécessaire, en fonction des benchmarks que je vais mener ; cela demanderait une grosse réécriture du code.