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 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.

La première méthode consiste à créer notre propre système d’exécution asynchrone en stockant les messages dans une base MySQL. Il faut savoir que cette méthode est souvent décriée. La raison invoquée, répétée comme un mantra appris par cœur, est qu’une base de données n’est pas une file de messages. Certes, c’est vrai. Mais d’un autre côté, une base de données permet de centraliser très facilement des informations, et on sait très bien les sauvegarder et les redonder.

L’un des apôtres de ce mantra est Stephen Corona (CTO de Twitpic) ; dans son ebook Scaling PHP Apps (livre très intéressant au demeurant, je vous le conseille), il explique qu’il ne faut pas utiliser une base de données comme une file de messages, mais ses explications ne sont pas du tout convaincantes. Il dit avoir déjà essayé cette technique, mais que des problèmes sont apparus lors d’accès concurrentiels : il faisait une requête SELECT pour récupérer l’identifiant de la prochaine tâche à traiter, puis une requête UPDATE pour indiquer que cette tâche est en cours de traitement ; évidemment, si deux programmes font cela en même temps, la tâche risque d’être traitée deux fois. On verra plus bas qu’il suffit d’intervertir les requêtes (réserver une tâche puis la récupérer) pour qu’il n’y ait plus ce souci.
Il n’en reste pas moins que cela nécessite de faire du polling, c’est-à-dire scruter la base de données très régulièrement pour voir s’il y a de nouvelles tâches en attente. C’est vrai que ce n’est pas idéal, car cela charge inutilement la base, mais surtout implique une latence dans les traitements. Mais toujours dans le même livre, il conseille d’utiliser Resque (dont je parle plus bas aussi) qui non seulement utilise une base de données − Redis − mais fait aussi du polling… Là où ça devient amusant, c’est qu’il suggère aussi d’utiliser Redis pour stocker du cache, en effaçant les données les plus anciennes (vous voyez le souci si vos messages se mettent à disparaître avant d’être traités ?).

En plus de cela, on verra plus loin que certains systèmes de files de messages utilisent une base de données pour assurer leur persistance. Donc restons calmes et pragmatiques.

MySQL + Crontab

C’est le moyen le plus simple pour commencer à faire des traitements asynchrones. C’est très artisanal, mais ça marche plutôt bien, avec une courbe d’apprentissage quasi nulle grâce à l’utilisation de technologies dont on a l’habitude.

Commençons par créer une table dédiée à cet usage dans une base de données MySQL :

CREATE TABLE Task (
    id INT UNSIGNED NOT NULL AUTO_INCREMENT,
    creation_date DATETIME NOT NULL,
    modification_date TIMESTAMP DEFAULT NULL,
    priority TINYINT UNSIGNED NOT NULL DEFAULT '0',
    action TINYTEXT NOT NULL,
    not_before DATETIME DEFAULT NULL,
    status ENUM('waiting','processing','error') NOT NULL DEFAULT 'waiting',
    token CHAR(40) NOT NULL,
    data MEDIUMTEXT DEFAULT NULL,
    PRIMARY KEY (id),
    INDEX creation_date (creation_date),
    INDEX priority (priority),
    INDEX action (action(8)),
    INDEX not_before (not_before),
    INDEX status (status),
    INDEX token (token(40))
) ENGINE=InnoDB DEFAULT CHARSET=UTF-8;

Il est ensuite très facile d’écrire une fonction qui ajoutera des messages dans cette table. Voici une sorte de pseudo-code PHP (sans la connexion à la base de données, ni de gestion des erreurs) :

/**
 * Ajoute une nouvelle tâche à traiter de manière asynchrone.
 * @param   string   $action    Nom du traitement à effectuer.
 * @param   mixed    $data      (optionnel) Données associées à la tâche.
 * @param   int      $priority  (optionnel) Priorité de la tâche.
 * @param   string   $notBefore (optionnel) Date à laquelle effectuer le traitement.
 */
public function addTask($action, $data=null, $priority=0, $notBefore=null) {
    $db = new PDO(/* DNS */, /* USER */, /* PWD */);
    $stmt = $db->prepare("
        INSERT INTO Task
        SET creation_date = NOW(),
            priority = :priority,
            action = :action,
            not_before = :notBefore,
            status = 'waiting',
            token = '',
            data = :data
    ");
    $stmt->execute([
        'priority' => $priority,
        'action' => $action,
        'notBefore' => $notBefore,
        'data' => json_encode($data)
    ]);
}

Ensuite, on va faire un programme PHP qui va être exécuté par crontab à intervalles réguliers. Ce programme va tenter de récupérer une tâche en attente. S’il n’en trouve pas, il s’arrête. S’il en trouve une, il la traite, puis essaye d’en récupérer une de nouveau (et ainsi de suite jusqu’à ce qu’il ne trouve plus de tâche en attente). Là encore, le code est largement incomplet, sans la connexion à la base de données ni gestion des erreurs.

#!/usr/bin/php

$db = new PDO(/* DNS */, /* USER */, /* PWD */);
while (true) {
    // réservation d'une tâche
    $token = uniqid();
    $stmt = $db->prepare("
        UPDATE Task
        SET token = :token,
            status = 'processing'
        WHERE status = 'waiting'
          AND token = ''
          AND (not_before IS NULL OR not_before < NOW())
        ORDER BY priority DESC, id ASC
        LIMIT 1
    ");
    $stmt->execute([ 'token' => $token ]);
    // récupération de la tâche
    $result = $db->query("SELECT * FROM Task WHERE token = '$token'");
    $task = $result[0] ?? null;
    if (!isset($task)) {
        // pas de tâche récupérée
        exit(0);
    }
    // traitement de la tâche
    // ...
    // effacement de la tâche
    $stmt = $db->prepare("DELETE FROM Task WHERE id = :id");
    $stmt->execute([
        'id' => $task['id']
    ]);
}

Il faut maintenant mettre ce programme en crontab. Il va être exécuté toutes les minutes ; à chaque exécution, il va traiter des tâches tant qu’il en trouvera en attentes. Voici la directive à mettre en crontab :

* * * * *    /chemin/vers/le/script.php

Si on souhaite avoir un délai d’exécution plus faible, il faut ruser un peu. Les directives de crontab ont une résolution à la minute au minimum ; il faut donc en exécuter plusieurs chaque minute, en ajoutant des temporisations. Par exemple, si on veut exécuter le script toutes les 15 secondes :

* * * * *               /chemin/vers/le/script.php
* * * * *    sleep 15 ; /chemin/vers/le/script.php
* * * * *    sleep 30 ; /chemin/vers/le/script.php
* * * * *    sleep 45 ; /chemin/vers/le/script.php

Cette manière de faire fonctionne très bien quand on commence à vouloir séparer certains traitements pour les rendre asynchrones. Le gros problème de cette méthode est que les tâches sont traitées séquentiellement ; si une tâche met beaucoup de temps à être traitée, elle bloquera le traitement des tâches suivantes (éventuellement jusqu’à ce que le cron lance un autre script qui prendra le relais en parallèle).
L’avantage de procéder ainsi, c’est − comme je l’ai dit plus haut − qu’on maîtrise déjà les technologies utilisées ; pas de serveur supplémentaire à installer, pas de courbe d’apprentissage.
Chose intéressante, on peut facilement faire de la répartition de charge en installant le script d’exécution sur plusieurs machines, qui se connectent tous à la même base MySQL pour récupérer les tâches en attente.

Donc ce n’est pas mal pour commencer. La fonction addTask() peut encapsuler les appels à n’importe quelle techno sous-jacente. Ainsi, si vous passez à Resque/Gearman/Beanstalkd (ou un autre), vous n’avez qu’à modifier cette fonction, sans toucher à votre code applicatif.

Pour : Facile et rapide à mettre en œuvre.
Contre : Traitement séquentiel. Polling (surcharge de la base et latence).

On peut imaginer plusieurs manières d’améliorer cette technique :

1. Pour traiter plusieurs tâches en parallèle, le programme lancé par crontab pourrait créer des sous-processus (en faisant un fork), chaque sous-processus traitant une tâche.
2. Plutôt que d’avoir un script lancé par crontab, on pourrait faire un démon qui tournerait en tâche de fond ; il interrogerait la base de données pour savoir si de nouvelles tâches sont en attente, mais il tournerait indéfiniment. Et pour chaque tâche, un nouveau sous-processus serait créé.
3. Pour éviter de créer des sous-processus, le démon pourrait faire des requêtes sur un serveur Apache ou Nginx, et couperait la communication sans attendre de retour. Ainsi c’est le serveur web (et/ou le serveur Fast-CGI si vous utilisez Nginx avec PHP-FPM par exemple) qui se chargerait de faire de la répartition de charge entre plusieurs processus.
4. Il est aussi possible de coder des “workers”, qui sont autant de petits démons qui attendent de recevoir des ordres de la part du démon principal, à travers un canal de communication à base de ZeroMQ ou nanomsg (voir la conférence que j’avais donnée à propos des démons et de ZeroMQ en PHP). Mais dans ce cas, on a besoin d’un système de gestion des workers, comme Supervisor, God ou Monit, pour gérer leur nombre et leur exécution.

Resque

Resque est un outil en Ruby créé en 2009 par GitHub (un post de blog très intéressant explique sa genèse). Il existe une déclinaison en PHP, php-resque.

Son principe est assez simple : Une librairie client permet d’ajouter des tâches dans une base de données Redis. L’ensemble de la gestion des tâches est géré par Redis, qui se charge de stocker les tâches dans des structures de données adaptées (Redis est une base NoSQL qui gère des paires clé-valeur mais aussi des types de données plus élaborées comme des listes). À côté de ça, un démon (serveur logiciel) tourne en tâche de fond ; à intervalles réguliers, il interroge le serveur Redis pour voir si de nouvelles tâches sont en attente d’exécution ; si oui, un nouveau processus est forké pour chaque tâche à exécuter (c’est le fonctionnement de php-resque ; pour le Resque original, le projet Sidekiq permet d’avoir des threads concurrents).
Il est possible d’avoir plusieurs machines faisant chacune tourner le démon Resque, tous connectés sur la même base Redis, permettant ainsi de faire de la répartition de charge.

L’avantage de Resque est qu’il est rapide à mettre en œuvre, et plutôt simple dans son fonctionnement (le plus gros de la complexité étant géré par Redis). On comprend vite son fonctionnement, et il n’est pas très compliqué d’écrire du code fonctionnel. En plus, une interface web permet de suivre les traitements, ce qui est très pratique.

D’un autre côté, le polling induit de la latence, sans compter la charge que cela induit sur le serveur Redis (même si ça devrait rester léger, mais tout dépend du nombre de tâches traitées). La création de nouveaux processus à chaque nouvelle tâche est certes simple et pratique mais pas très efficace (cela peut être pénalisant si vous avez beaucoup de petites tâches à traiter). Et malheureusement, les priorités ne sont pas gérées nativement : il faut utiliser plusieurs “listes de messages”, qui définissent des priorités suivant l’ordre dans lequel elles sont appelées ; c’est un fonctionnement différent, pas forcément un gros problème si on le gère correctement au niveau applicatif.

Comme je l’ai dit plus haut, l’utilisation de Redis peut être un souci si on l’utilise par ailleurs comme serveur de cache et qu’on l’a configuré de manière à ce qu’il efface les données les plus anciennes quand il manque de place. On risque alors d’avoir des tâches qui disparaissent avant d’être traitées, ça peut vite devenir un casse-tête.
On peut contourner ça en installant un autre serveur Redis, ou plus simplement en utilisant Memcached pour gérer le cache. Mais c’est bien d’avoir ce paramètre en tête.

Pour : Facile et rapide à installer. Interface web.
Contre : Polling. Fork.

Gearman

Gearman est un projet très complet et robuste, qui existe aussi depuis 2009. Écrit initialement en Perl, il a été complètement réécrit en C. Il est basé sur un démon qui reçoit les messages de clients, et qui les transmet à des workers qui attendent de les traiter.

Son principal intérêt est qu’il offre de très bonnes performances, et cela pour deux raisons :

• Il est basé sur des files de messages. Il gère en interne tout le traitement des messages sans les faire transiter par une base de données. Il ne fait pas de polling ; les messages reçus sont directement transmis au premier worker disponible.
• Il utilise un protocole de communication binaire, et non pas un protocole textuel. Le gain est habituellement minime, sauf si vous avez énormément de petits messages à traiter.

Par défaut, Gearman gère tout en mémoire, ce qui est très bien dans un monde où rien ne tombe jamais en rade. Il offre une option de persistance des données, ce qui permet de traiter les tâches en attente en cas de redémarrage du serveur. Plusieurs bases de données sont supportées pour cela : Memcache (mais ce serait une très mauvaise idée d’utiliser du cache en RAM pour persister des données…), MySQL, SQLite, PostgreSQL et TokyoCabinet ; Redis et MongoDB sont supportés dans la branche de développement.

Comme pour Resque, il existe des interfaces web permettant de suivre le déroulement des tâches, comme Gearman UI.
Autre similitude avec Resque, Gearman ne gère pas la priorité des messages ; il faut créer différentes files de messages, et l’ordre dans lequel elles sont utilisées détermine naturellement la priorité associée.

Gearman nécessitant la présence de workers pour traiter les tâches, le mieux est de les gérer avec GearmanManager, Supervisor ou un autre système similaire (comme God ou Monit, dont j’ai déjà parlé plus haut). Ça marche très bien, mais c’est encore une interface de gestion supplémentaire.
Les workers peuvent s’exécuter sur plusieurs machines, pour faire de la répartition de charge.

Pour : Performant. Grand nombre de langages de programmation supportés.
Contre : Un poil compliqué à configurer (entre Gearman lui-même et le gestionnaire de processus).

Beanstalkd

Beanstalkd est un serveur écrit en C, assez comparable à Gearman de manière générale. Il en diffère en deux points :

• Son protocole est textuel (inspiré par le protocole de Memcache). C’est pratique pour tester à la main avec telnet, mais un poil moins performant sur des cas extrêmes.
• Il ne gère pas la persistance de la même manière. Au lieu d’enregistrer les tâches dans une base de données, il tient un fichier de log binaire. Si le démon Beanstalkd est redémarré (après un plantage ou un redémarrage matériel), ce fichier est reparcouru au lancement.

Beanstalkd offre un certain nombre de fonctionnalités. Il gère différentes files de messages (appelées “tubes”), ainsi que les priorités et le délai avant exécution ; il permet aussi de mettre les tâches dans un statut particulier (“buried”) pour indiquer qu’une erreur a eu lieu durant leur traitement. Un worker peut indiquer qu’il a besoin de plus de temps que prévu pour traiter une tâche, afin d’éviter qu’elle ne  tombe en timeout et soit renvoyée à un autre worker. Un “tube” peut être mis en pause, ce qui peut être pratique en cas de problème qui mettrait toutes les tâches en erreur.

Comme avec Memcache, les clients peuvent se connecter à plusieurs serveurs Beanstalkd, ce qui permet de faire un premier niveau de répartition de charge ; et, comme avec les autres solutions, vous pouvez faire tourner des workers sur différentes machines, qui se connectent sur le même serveur.
Comme pour Resque et Gearman, il existe des interfaces web pour surveiller l’exécution de Beanstalkd (beanstalk_console, django-jack, Aurora, Beanstalker, phpBeanstalkdAdmin, …).
Comme pour Gearman, il est très fortement recommandé d’utiliser un système de gestion des processus (Supervisor, God ou Monit) pour lancer et contrôler les workers.

Beanstalkd semble particulièrement performant. Certains benchmarks montrent qu’il peut gérer plus de 15 fois plus de tâches par seconde que Resque (je n’ai malheureusement pas trouvé de comparatif entre Beanstalkd et Gearman).

Pour : Performant. Très grand nombre de langages supportés. Simplicité de la persistance (log binaire).
Contre : Gestion des workers.

Les autres systèmes distribués

Il en existe maintenant un assez grand nombre ; une liste très complète est disponible sur le site queues.io. Certains d’entre eux sont sûrement très bien et ils mériteraient que je me penche dessus (Darner, RestMQ, Faktory, Disque par le créateur de Redis, Siberite). D’autres m’ont semblé trop complexes à installer et configurer (ActiveMQ, RabbitMQ, Kafka, Zaqar), ne supportent pas les langages de programmation qui m’intéressent (Nats, Kue, NSQ, Qpid, queue_classic, RQ), sont notoirement peu efficaces en termes de rapidité ou de consommation mémoire (HornetQ, Delayed::Job, Mappedbus), sont des surcouches à une autre file de messages (Celery, QDB, MessageBus) ou sont des webservices qui induisent une trop grande latence (Amazon MQ, Amazon SQS, IronMQ).

Conclusion

Mettre le doigt dans l’informatique distribuée et les traitements asynchrones est quelque chose de finalement pas très compliqué, qui peut se faire graduellement (on commence avec la solution BDD + Crontab, puis on profite de l’encapsulation pour faire évoluer la plate-forme en douceur). Mais il ne faut pas sous-estimer l’impact que cela peut avoir sur la manière de développer − et surtout les difficultés que l’on peut rencontrer pour déboguer du code qui est éclaté.
Comme toute optimisation, il ne faut pas s’y lancer prématurément, sous peine de rencontrer plus d’inconvénients que d’avantages. Mais si votre architecture commence à prendre un peu d’ampleur, c’est l’une des solutions évidentes pour améliorer la rapidité d’un site et fluidifier les traitements.

Quant à la solution technique à utiliser, j’ai acquis la certitude que ce n’est au final pas très important. C’est un peu comme les langages de programmation… vous pouvez lire des benchmarks, de la documentation et des avis utilisateur, au final on peut coder quasiment n’importe quoi avec n’importe quel langage ; l’important est de trouver celui qui vous permettra de faire ce que vous voulez faire.
Donc essayez plusieurs solutions, et choisissez celle avec laquelle vous êtes le plus à l’aise − soit parce que la configuration vous semble simple, soit parce que la documentation vous paraît compréhensible, soit parce que vous trouvez qu’il est facile de programmer vos clients et workers. Les différences de performance sont assez minimes (on en reparlera quand vous devrez traiter autant de tâches que Facebook ou GitHub).

Personnellement, j’avais poussé assez loin l’approche “artisanale” avant d’être très intéressé par Gearman. Mais je dois dire que je trouve maintenant que Beanstalkd est très bien pensé, facile à configurer, et offre un écosystème florissant (ce qui permet d’avoir des outils de gestion très pratiques et des bibliothèques de programmation bien faites).