Écrire un moteur de tâches en Ruby

Le centre d’un moteur de tâches c’est une boucle qui fait quatre choses :

Les composants du système qui effectuent cette boucle sont généralement appelés des “workers”.

Un moteur de tâches passe la très grande majorité de son temps à exécuter les tâches.

Une fois que les tâches ont été correctement optimisées, augmenter la capacité de traitement nécessite de pouvoir lancer des tâches supplémentaires, et donc ajouter des workers.

Cela signifie que les phases en dehors de celle d’exécution de la tâche doivent pouvoir être parallélisées le plus possible. Cela passe principalement par éviter ou limiter les moment où les différents workers peuvent interférer les uns avec les autres.

Si les tâches à exécuter ont elles-même des problèmes de contention, cela va limiter la parallélisation de tout le système, mais c’est est en dehors du périmètre du moteur de tâches en lui-même.

Tout ce qu’il à faire pendant l’exécution c’est de ne pas être une gène.

Ajouter des workers peut se faire de deux manières : en lançant plusieurs workers dans un même processus et en lançant plusieurs processus.

La première approche s’appuie souvent sur des threads, qui est un mécanisme fournit par la plupart des systèmes d’exploitations. Elle a l’avantage de limiter la consommation mémoire car les différents threads peuvent partager des ressources, par exemple la mémoire dans laquelle le code qui s’exécute est chargé.

Utiliser plusieurs processus évite d’être limité par la puissance d’une machine, en effets les différents processus peuvent être lancés sur des serveurs différents. Par contre elle transforme le moteur de tâches en système distribué, et subit par conséquents les contraintes de ce type de programmes.

Les deux options sont complémentaires : on peut avoir plusieurs processus hébergeant chacun plusieurs threads.

L’idéal est d’avoir un système qui sache tirer partie des threads pour bénéficier de l’économie de ressources, tout en fonctionnant bien en multi-process quand ça devient nécessaire.

Les thread en Ruby ont une spécificité : l’implémentation Ruby de référence (probablement celle que vous utilisez si vous n’avez pas choisi spécifiquement d’en utiliser une autre) ne permet pas plusieurs thread de s’exécuter en même temps, à cause de limitations dans la manière dont elle est implémentée. Si vous utilisez un moteur de tâches en Ruby et que vous voulez pleinement profiter d’avoir plusieurs threads, il faut ainsi utiliser des implémentation alternatives comme JRuby.

Cela ne signifie pas qu’utiliser plusieurs threads en Ruby soit inutile dans un moteur de tâches avec l’implémentation de référence. Car même si un seul thread s’exécute à la fois, plusieurs tâches peuvent tout de même être en train de progresser. C’est le cas parce que lorsqu’un appel extérieur est fait, par exemple une requête en base de données ou un appel réseau, le thread qui fait l’appel est mis en pause et un autre thread peut alors s’exécuter, pendant que l’appel est en train de se faire. Lorsqu’il se termine, le premier thread va alors attendre son tour et continuer son traitement.

Dans un système ou la majorité du temps est passé à attendre le reste du monde, avoir plusieurs threads peut donc être tout de même assez avantageux. À l’inverse si les traitements demandent des traitements longs dans le moteur de tâches, les threads sont pratiquement inutiles et faudra alors s’appuyer uniquement sur des processus ou utiliser une autre machine virtuelle.

Je vais mettre l’ensemble du code (framework et code d’exemple) dans un projet unique et je ne le packagerai pas sous forme d’une gem.

Le faire ajouterait de la configuration et du code et donc un peu de complexité sans que cela apporte quelque chose aux sujets dont je veux discuter ici.

Je commence par définir la version de Ruby à utiliser

et ma dépendance de départ

Pour montrer comment fonctionnent le threading, la première implémentation des tâches se contentera d’attendre quelques secondes via la méthode Kernel#sleep ce qui permet aux autres threads de s’exécuter. Et pour le moment elle n’iront pas chercher quelles tâches sont disponibles, mais s’exécuteront de manière continue dans une boucle infinie.

La version simplifiée d’un worker se résume alors à :

Le thread principal du moteur de tâches va démarrer les différent workers puis se mettre en pause pour leur permettre s’exécuter.

Pour vérifier que cela fonctionne, il est nécessaire de lancer plusieurs workers, et d’ajouter des logs.

Je vais placer le code des workers dans un classe spécifique TaskEngine::Worker, et une autre classe TaskEngine::Engine sera chargée de les lancer, c’est elle qui sera exécutée au démarrage du moteur de tâches.

Chaque instance de worker reçoit un numéro de façon à pouvoir suivre ce qu’elle fait dans les logs.

Le nombre de workers est défini dans une constante, passer par un fichier de configuration ajouterait du code sans changer l’idée.

Il me reste à écrire une tâche Rake pour pouvoir démarrer le moteur :

Et je peux vérifier que tout fonctionne comme prévu :

Les tâches se bien lancent les unes après les autres, et se passent effectivement la main après avoir démarré une fois qu’elles atteignent le sleep. La boucle de traitement fonctionne donc bien, ou du moins sa version simplifiée.

Arrêter le programme stoppe immédiatement toutes les tâches en cours d’exécution.

Dans la partie suivante je vais m’intéresser au stockage des tâches pour que les workers aillent chercher ce qu’ils ont à faire.

Un moteur de tâches a besoin de stocker deux types de données : les tâches qui ne sont pas encore lancées et les logs de tâches exécutées.

Pour les tâches qui ne sont pas encore lancées, l’idéal est d’avoir un système de stockage :

Pour les logs de tâches exécutées, les contraintes sont plus faibles (même si on préfère tout de même éviter de perdre des données et que le système ne soit pas trop lent), on a surtout besoin d’un système qui permette de la recherche pour faciliter les investigations.

Il est possible d’utiliser deux systèmes différents pour les deux usages, par exemple pour les tâches pas encore lancées il est courant d’utiliser un bus de messages, car en plus du stockage il peut servir de médiateur entre un système qui créé des tâches et le moteur de tâches en fournissant une API facile à utiliser.

Il est également possible d’utiliser un seul système, par exemple une base de données SQL ou même une base de données comme Redis.

J’ai choisi d’utiliser PostgreSQL pour l’ensemble du stockage : le fonctionnement des transactions des bases de données SQL sera utile pour éviter la perte de données et gérer la parallélisation, et PostgreSQL fournit un système de notification permettant de répondre au troisième point. Pour la recherche, les capacités d’indexation SQL seront bien suffisantes.

Pour l’accès aux données je vais utiliser un ORM. Plutôt qu’ActiveRecord qui amène avec lui tout l’écosystème Rails, je vais utiliser Sequel.

Sequel n’est pas aussi complet qu’ActiveRecord mais il a trois avantages :

J’ajoute donc les dépendances au Gemfile :

Et je créé la base de donnée avec un accès pour le compte de l’application :

Pour commencer je vais m’intéresser aux tâches qui ne sont pas encore lancées.

Les tâches seront créées dans un état waiting, et passeront dans un état running lorsqu’un des worker du moteur de tâches les lancera. Les tâches les plus anciennes seront choisies d’abord. Après exécution, les tâches seront supprimées de la base (la gestion de l’historique sera faite plus tard).

Pour le moment je vais continuer d’utiliser les mêmes tâches qui se contentent de mettre le worker en attente, ce qui signifie que je n’ai pas à gérer de paramètres pour le moment.

Pour stocker l’état des tâches je vais utiliser un type énuméré (on dit aussi une énumération). Un peu comme dans les langages comme C ou Java, cela signifie que la base de données pourra vérifier que les valeurs qu’on lui fournit seront valides (en cas de valeur inconnue la base renvoie une erreur), et que si dans le code Ruby et dans les requêtes on utilisera les valeur textuelles, en interne la base le stockera de manière efficace comme un nombre.

Voici la migration permettant de créer la table tasks (si vous avez l’habitude des migrations ActiveRecord, vous ne devriez pas être dépaysé·e) :

Je peux ensuite la lancer (le -m indique à Sequel où se trouve le répertoire qui contient les migrations)  :

Et vérifier que la table a bien été créée :

Pour pouvoir manipuler les tâches il me reste à déclarer la connexion à la base et le modèle qui représente une tâche.

Lors de la configuration de l’accès à la base de donnée, j’indiquerai d’utiliser au maximum une connexion par worker plus une pour le moteur. Cela permet que chaque worker ait une connexion à lui plutôt que de risquer que les workers ne s’attendent pour pouvoir faire des requêtes.

Je peux alors définir une tâche Rake pour créer 100 tâches à exécuter :

Et la lancer :

Il reste la part du lion qui est de mettre à jour le moteur pour qu’il s’appuie sur la base de données.

La partie la plus complexe est l’acquisition de tâches à exécuter. Pour cela il faut sélectionner la tâche la plus ancienne, et changer son statut.

Mais comme plusieurs workers travaillent en parallèle, il faut faire en sorte que la même tâche ne soit pas prise en compte deux fois par deux workers.

Une chose importante dans le design des tâches : il est souhaitable — tant que cela est possible — qu’une même tâche puisse être exécutée plusieurs fois de suite sans que cela soit gênant.

Par exemple si une tâche est chargée d’envoyer un mail, si on l’exécute de nouveau il faudrait que le même mail ne soit pas envoyé une deuxième fois.

Cela vient du fait qu’il est très difficile de réunir deux conditions à la fois :

Dans les outils de messages, on appelle cela “at least once” (au moins une fois), “at most once” (au plus une fois) et “exactly once”.

Le moteur de tâches se contentera de faire en sorte que la même tâche ne soit pas exécutée deux fois.

Plutôt que d’essayer de faire en sorte d’avoir un système qui garantit du “exactly once”, on préfère que le moteur de tâches se concentre sur le fait de ne pas perdre de tâche, et si une même tâche est exécutée une deuxième fois c’est la tâche elle-même qui est chargée que cela n’ait pas d’effet gênant.

Cela permet d’obtenir le même objectif (que le traitement final se comporte comme s’il avait été exécuté une fois), tout en modifiant les contraintes : au lieu d’avoir toutes les contraintes au niveau du moteur de tâches, elles sont en partie au niveau du moteur de tâches et en partie au niveau des tâches en elles-mêmes.

L’autre avantage de cette approche est qu’en cas de gros problème, il devient possible de réinjecter dans le moteur de tâches des tâches passées et de les laisser les traiter, sans avoir à se préoccuper de celles qui avaient déjà été exécutées.

Revenons à nos moutons : sélectionner la tâche la plus ancienne et changer son statut, et empêcher qu’un autre thread ne fasse pareil.

En SQL pour cela on se sert de verrous (ou locks). Lorsqu’un thread va sélectionner la tâche T1 à exécuter il va en même temps verrouiller cette tâche dans la base, de sorte que les autres threads qui cherchent aussi la tâche la plus ancienne et qui pourraient vouloir utiliser T1 ne puissent le faire. Cela signifie que les autres workers seront bloqués en attendant que le verrou sur T1 soit relâché. Après avoir mis à jour le statut de T1, on la déverrouillera dès que possible pour débloquer les autres threads. À leur tour ils pourront chercher la tâche la plus ancienne à exécuter, qui sera donc la tâche suivante car le statut de T1 aura changé et elle ne sera donc plus sélectionnable pour eux.

Pour cela on va utiliser la commande SELECT … FOR UPDATE, qui verrouille les lignes sélectionnée en vue de pouvoir faire une mise à jour.

Le relâchement du verrou se fait à la fin de la transaction en cours, qui doit donc être terminée dès que possible pour limiter la période pendant laquelle les autres workers risquent d’être en attente.

L’acquisition de la tâche (sélection et mise à jour du statut) doivent de toutes façon se faire dans une transaction séparée de celle de l’exécution de la tâche :

Voici donc à quoi ressemble le code :

La méthode try_acquire_task retournera la tâche à exécuter ou nil si aucune tâche n’est disponible.

Lorsque l’exécution se termine, pas besoin de poser de verrou car il n’y a pas de risque qu’un autre worker veuille terminer la même tâche. Par contre il est toujours nécessaire d’isoler le code dans une transaction à part, notamment pour qu’une erreur dans le moteur de tâches ne risque pas d’avoir de conséquence sur la tâche qui vient d’être exécutée.

Reste ensuite à modifier TaskEngine::Worker#exécute pour utiliser ces deux méthodes :

Tant qu’il reste des tâches, chaque worker les traitera les unes après les autres :

À cette étape le noyau du système est là : on peut créer des tâches et quand on le lance le moteur va les exécuter jusqu’à épuisement.

Dans la partie suivante je vais m’intéresser à l’arrêt du moteur.

Après l’avoir lancé, voici ce qui se passe quand on éteint le moteur de tâches en appuyant sur CONTROL+C :

Le moteur de tâches s’arrête immédiatement et toutes les tâches s’arrêtent net.

Si on voulait utiliser ce type de fonctionnement dans un moteur de tâches en production, cela signifie que toutes les tâches doivent être développées de manière à ce qu’elles puissent être arrêté net n’importe quand.

Si les tâches n’utilisent que la base de données, cela peut être envisageable car alors la transaction est annulée, et quand on les relance elles recommencent de zéro.

Mais dès lors qu’elles communiquent avec l’extérieur les choses deviennent plus compliquées.

Par exemple si une tâches chargée d’envoyer un mail puis de mettre à jour la base de donnée pour indiquer que le mail est parti est arrêtée juste après l’envoi du mail.

Au redémarrage du moteur de tâches si la tâche est relancée elle enverrait un autre mail identique.

C’est pour cela qu’en général, lorsqu’un moteur de tâches reçoit une commande d’arrêt, au lieu de stopper les tâches en cours, il ne démarre plus de nouvelles tâches et attend pour s’éteindre que les tâches déjà lancées se terminent.

Cela signifie que dans les conditions normales d’opérations d’un arrêt/relance, les tâches peuvent s’appuyer sur l’hypothèse qu’elles ne seront pas stoppées en cours de route.

Je parle de conditions normale car un cas exceptionnel est toujours possible, de la même manière que n’importe quel programme, par exemple une machine virtuelle qui crashe. Cela signifie que le risque existe toujours, même s’il est faible.

Pour une tâche qui envoie un mail, il est peut-être acceptable qu’en cas de crash le mail soit envoyé deux fois, par contre pour une tâche qui fait un virement entre deux comptes bancaires, il vaut peut-être mieux se prémunir contre le risque que le virement soit effectué deux fois même si le risque de crash est faible.

Pour exécuter du code lorsqu’un programme s’arrête, il faut généralement indiquer au système qu’en cas de demande d’arrêt il doit exécuter un bloc de code ou une méthode spécifique plutôt que d’utiliser le comportement par défaut qui est de tout stopper d’un coup.

Sur les systèmes d’exploitations comme Linux ou macOS, la demande d’arrêt (par exemple lorsque vous appuyez sur CONTROL+C) est communiquée au programme sous forme d’un signal. Un programme peut informer le système d’exploitation que tel ou tel signal l’intéresse, et qu’il prend la responsabilité de répondre au signal, en remplacement du comportement par défaut.

Il existe différents signaux identifiés chacun par un nom et un numéro, certains sont normalisés et correspondent aux messages que le système avait besoin d’envoyer dans les systèmes UNIX classiques et d’autres sont dépendants du système d’exploitation.

Pour gérer une fin d’exécution, il faut en principe écouter deux signaux qui font partie de ceux qui sont normalisés :

Le comportement est souvent le même pour les deux signaux.

En Ruby, pour définir un bloc de code à appeler quand un signal est reçu il faut appeler la méthode Signal::trap en lui passant en paramètre le nom ou le numéro du signal :

À noter que le code appelé lors de la réception d’un signal n’est pas du code comme les autres car le système d’exploitation utilise des routines spécifiques pour appelle ce code et il s’exécute dans un contexte un peu spécial, et ce faisant certaines fonctionnalités qu’on tient normalement pour acquises sont indisponibles.

Cela signifie qu’il doit respecter certaines règles pour ne pas poser de problème au système d’exploitation et faite crasher l’application. Ainsi en théorie les allocation mémoire ou les entrées/sorties sont interdites.

Par exemple si le système d’exploitation décide d’arrêter un processus car sa mémoire est devenue insuffisante, si le code gérant l’arrêt essaie d’allouer de la mémoire supplémentaire cela ne fonctionnera pas.

En pratique il est possible que cela ne pose pas problème, même dans la très grande majorité des cas, mais vous n’avez aucune garantie.

Par exemple si vous essayer d’utiliser un log dans un code de gestion de signal :

Ruby vous indiquera que cela ne fonctionnera pas par un “log writing failed. can't be called from trap context”, car pour éviter le risque que cet appel fasse crasher le programme, la machine virtuelle Ruby le désactive dans ce contexte.

La bonne pratique est donc de limiter au maximum ce que vous mettez dans le bloc de gestion de signal, par exemple d’uniquement changer la valeur d’une variable, et de faire en sorte que le vrai traitement se fasse en dehors du bloc.

Dans le moteur de tâches, chaque worker exécute une boucle où il essaie de trouver une tâche à exécuter puis la lance :

Pour gérer l’extinction, le bloc de gestion de signal va donc changer la valeur d’une variable, et à chaque tour de boucle le worker va vérifier s’il faut s’arrêter.

Ce qui peut se traduire par :

Ce statut va être stocké dans l’instance de TaskEngine::Engine, auquel les workers vont accéder :

Lorsque le thread du worker n’a plus rien à faire, il s’arrête :

Après avoir appuyé sur sur CONTROL+C, les différents workers s’arrêtent après avoir terminé l’exécution de leurs tâches, mais le moteur de tâches lui ne s’arrête pas.

D’autres appuis sur CONTROL+C ne servent à rien.

Le problème c’est le thread principal qui est celui qui démarre du moteur.

Après avoir créé les différents worker, le thread principal se met en attente pour une durée indéfinie et ne s’arrête donc jamais, et il est de même pour le moteur de tâches.

On pourrait vouloir arrêter ce thread dans le code de traitement du signal, mais cela aurait pour effet d’arrêter immédiatement le programme, sans laisser aux workers la possibilité de terminer leurs traitements.

Il faut donc que le thread principal se termine après la fin de l’exécution de l’ensemble des tâches en cours, c’est à dire quand les threads correspondant aux différents workers se sont terminés.

Dans l’API des thread Thread#join permet justement de suspendre l’activité d’un thread en attendant qu’un autre thread se termine. L’idée est d’attendre les threads de tous les workers l’un après l’autre. Si un thread est déjà arrêté lorsqu’on appelle Thread#join sur lui, la méthode retourne immédiatement. Cela signifie qu’il n’y a pas à se préoccuper de savoir si un worker a déjà terminé ou pas avant de faire l’appel à join.

Lorsqu’on démarre les worker, il faut donc les stocker les différentes instances pour pouvoir ensuite accéder à leurs threads, et plutôt que d’appeler Kernel#sleep, le moteur attendra la fin du premier thread.

Après réception du signal, on voit que le thread de moteur attend successivement les différents worker pour ensuite arriver au bout de la méthode TaskEngine::Engine#initialize, ce qui stoppe le programme.

Cela règle le cas où le moteur de tâches s’arrête bien, je vais maintenant m’occuper de ce qui se passe après un arrêt d’urgence.

Le cas de l’arrêt dans une situation normale est donc couvert. Reste celui de l’arrêt exceptionnel.

Cela peut correspondre à une machine virtuelle qui crashe, un bug qui fait planter la machine virtuelle Ruby. Cela peut aussi être une conséquence d’un arrêt normal qui ne s’est pas bien terminé.

En effet si une tâche reste bloquée, par exemple parce qu’un appel réseau n’a pas de timeout en cas de non réponse, une tâche peut rester bloquée indéfiniment, et donc empêcher le moteur de tâches de s’arrêter.

Pour le stopper, on peut utiliser un autre signal, typiquement SIGKILL (qui lui ne peut pas être intercepté par du code applicatif: c’est le système d’exploitation qui se charge de faire le ménage lui-même). Ce signal portant le numéro 9, c’est lui qui est envoyé quand on appelle kill -9.

Dans ce cas le programme s’arrête net et avec lui les tâches en cours d’exécution.

Avec l’implémentation actuelle du moteur de tâches, ces tâches restent définies comme étant dans l’état running. Par exemple voici le résultat obtenu en stoppant brutalement le code précédent où le thread principal ne s’arrêtait pas :

Cela signifie qu’il est difficile de différencier les tâches mal arrêtées de celles qui tournent réellement.

Il est possible d’aller chercher dans les logs pour déterminer les tâches qui ont été lancées et qui n’ont pas été terminées avant l’arrêt, mais c’est une approche artisanale alors que pour la gestion d’erreur je préfère des solutions les plus fiables possibles.

Pour identifier ces tâches, il faut pourvoir déterminer qu’elles appartiennent à l’instance de moteur de tâches qui a été arrêté de manière brutale. Une manière d’y parvenir est de stocker dans les tâches quelle instance du moteur de tâches est en train de les exécuter en ajoutant un identifiant.

Lorsque l’instance est plantée, on peut alors facilement identifier ces tâches. Au redémarrage, l’instance du moteur de tâches peut détecter qu’il existe des tâches marquée running qui portent son identifiant.

Cette approche aide aussi aux investigations pendant l’exécution d’une tâche : si une tâche est lente on peut aller consulter le monitoring de la machine sur laquelle elle est lancée pour voir si on voit quelque chose d’anormal.

Le plus simple, si un moteur de tâches démarre et trouve des tâches en statut running est alors de stopper immédiatement le moteur de tâches en indiquant explicitement quel est le problème.

Cette approche retarde la relance de du moteur de tâches, mais elle a deux avantages :

Par exemple, il serait aussi possible qu’au démarrage d’une instance I1 de moteur de tâches, toutes les tâches marquées comme running soit mise à jour dans un statut spécial, par exemple crashed. Mais imaginez que suite à une erreur de configuration une autre instance I2 de moteur de tâches utilise le même identifiant. Lors du démarrage de I1, les tâches qu’I2 est en train d’exécuter passeraient en statut crashed. Cela n’aurait peut-être aucune conséquence sur l’exécution de ces tâches, mais cela pourrait porter à confusion les personnes qui consulteraient l’état des tâches.

La solution de s’arrêter immédiatement évite donc ce risque.

La mise en œuvre est moins longue que l’explication.

Tout d’abord il faut ajouter la colonne d' l’instance à la table tasks :

Pour rappel, pour lancer les migrations, la commande est

Ensuite il faut récupérer le nom de l’instance dans la tâche Rake de lancement et la passer au moteur. Pour ce faire je vais utiliser une variable d’environnement :

Je peux ensuite l’utiliser lorsqu’un worker récupère une tâche :

Si je lance le moteur de tâches (en passant le nom de l’instance en variable d’environnement) et que je regarde ce qui se passe dans la base de données, on retrouve bien le nom de l’instance :

Et pour terminer, au lancement du moteur de tâches, il faut vérifier si des tâches n’existent pas déjà avec le même nom d’instance :

On peut le tester en stoppant brutalement l’instance en train de se tourner et en la relançant :

On peut ensuite affiner les choses, par exemple afficher la liste des tâches qui sont en cours d’exécution, ou fournir des méthodes pour faciliter la reprise, mais la partie importante est là.

Dans la partie suivante je vais m’intéresser au paramétrage des tâches.

Le type de tâche sera défini par le nom de la classe de la tâche qu’il faut lancer, c’est ce que font tous les outils que je connais et je ne vois pas d’autre approche intéressante.

Les paramètres d’exécution des tâches sont sous forme d’une Hash qui est à la main des tâches.

Il y a plusieurs choix possibles pour le fonctionnement des tâches, par exemple :

Je vais ajouter les deux colonnes qui manquent à la table tasks. Je vais les définir comme non nullable pour éviter toute fausse manipulation. Si une tâche n’a aucun paramètre d’exécution, on utilisera une Hash vide plutôt d’un nil, cela évite l’ambiguïté entre aucun paramètre d’exécution et une valeur qui manque.

Cela signifie que la table doit être vide avant de lancer la migration, sinon elle échouera car la base n’acceptera pas d’ajouter des colonnes non nullable sans valeur par défaut.

Pour stocker les paramètres d’exécution en utilisant une colonne de type jsonb qui permet de stocker des données sous forme de JSON.

Par rapport à stocker le JSON dans un champ de texte, cela coûte un peu plus cher en temps d’insertion (car la base de données doit parser et valider le format des données à l’insertion), par contre cela peut simplifier les investigation car il est possible de d’accéder au contenu des valeurs depuis des requêtes.

Pour avoir une première définition de tâche je vais modifier le code la tâche existante pour utiliser la nouvelle API, et en profiter pour la déplacer dans un fichier séparé.

Pour mémoire, cette tâche se contentait d’attendre 5 secondes :

Puisqu’on a maintenant des paramètres, je vais rendre la durée d’attente paramétrable. Pour cela je vais récupérer la valeur dans la Hash de paramètres :

Il est possible d’insérer directement des tâches dans la base de données depuis le code métier, mais il est plus pratique de passer par une API Ruby qui isole un peu le stockage, notamment le fait de stocker les paramètres au format JSON. Je vais donc ajouter une méthode dans le module TaskEngine.

Dans une version packagée du code, on pourrait vouloir l’isoler dans une gem à part pour éviter d’avoir à embarquer le code du moteur de tâches dans une application qui ne ferait qu’insérer des tâches.

Et je peux m’en servir pour modifier la tâche Rake qui peuple la table :

Et après exécution on peut vérifier que l’insertion se passe bien :

Il manque la dernière pièce du puzzle : utiliser les valeurs pour créer la tâche et l’exécuter.

Pour récupérer une classe à partir d’une chaîne contenant son nom, Ruby fournit une méthode Module#const_get, il faudra ensuite en instancier un objet de la classe ainsi obtenue, puis appeler execute sur l’instance en lui passant les paramètres :

Ce qui donne :

Et voila : il est maintenant possible de sélectionner des tâches et de leur passer des paramètres.

Pour le moment c’est assez rustique, par exemple il n’y a pas encore de gestion d’erreur, mais la structure est là.

Dans la partie suivante je vais faire en sorte que le moteur réagisse quand on ajoute des tâches en cours de route.

Lorsque les workers sont en train de tourner, après avoir terminé une tâche ils vont en chercher une autre, même si cette dernière a été ajoutée après le démarrage du moteur de tâches. Dans ce cas il n’y a rien de spécial à faire.

Par contre, lorsqu’au moins un des workers s’est arrêté car il n’a pas trouvé de tâche disponible, il faut le solliciter pour qu’il se remette au travail quand une nouvelle tâche est créée.

Cela permet que la nouvelle tâche soit traitée au plus vite, plutôt que d’attendre qu’un worker qui est déjà en train d’exécuter une autre tâche ne la termine.

Lorsqu’on insère une nouvelle tâche, il faudra donc envoyer une notification pour réveiller les workers qui pourraient être disponibles.

Pour cela j’utiliserai un thread supplémentaire qui sera chargé d’écouter les notifications envoyées pour réveiller les workers.

En Ruby, la classe Queue est destinée à ce type d’usage où des thread doivent s’attendre et communiquer entre eux. Il s’agit d’une file (une sorte de liste dont les éléments sortent dans le même ordre que celui où ils sont insérés), à laquelle plusieurs threads peuvent accéder en parallèle. Le comportement le plus intéressant pour un moteur de tâches est que si du code demande à récupérer un élément de la queue alors qu’elle est vide, le thread correspondant se mettra en attente jusqu’à ce qu’un élément soit disponible.

Il serait possible de manipuler directement les thread des workers comme pour l’arrêt du moteur de tâches, en passant les thread des workers en le statut sleep puis en les réveillant avec Thread#wakeup, mais Queue fournit une API prête à l’emploi pour cela, et cela évite d’avoir le genre de bugs pénibles qui apparaît quand on décide de réinventer la roue sur un sujet qui touche au threading.

Il faut garder en tête que le fonctionnement final doit être compatible avec le fait d’avoir plusieurs threads et plusieurs instances de moteur de tâches, et qu’il faut toujours pouvoir arrêter les différentes instances.

Le fonctionnement actuel des worker est celui-ci :

La demande d’arrêt du moteur et l’absence de tâche à exécuter mènent tous les deux à l’arrêt du worker.

Il faut ajouter une nouvelle boucle dans le schéma : en l’absence de tâche, au lieu de s’arrêter les workers doivent se mettre en pause en attendant une notification.

Lorsqu’on les réveille, ils vérifient que le moteur est toujours en fonctionnement, et dans ce cas se remettent à chercher une tâche.

On pourrait vouloir directement passer de la pause au fait de chercher des tâches à exécuter :

Mais dans ce cas, si le worker ne trouve pas de tâche à exécuter après avoir réveillé il ne peut pas s’arrêter mais seulement se remettre en sommeil, à moins d’ajouter du code spécifique pour cela.

La proposition précédente est donc un peu plus générique.

Pour implémenter le workflow, on peut utiliser une double boucle :

Pour recevoir la notification, je dois créer un nouveau thread car l’écoute bloque l’exécution du thread qui attend d’être notifié.

Je vais créer ce thread dans le constructeur de l'Engine.

À la réception d’un message, il utilise la Queue pour réveiller un worker si l’un d’eux est disponible.

Le nom de la classe de la tâche est envoyé dans la notification pour être logué, mais ne sert pas dans le reste du code, en tous cas pour le moment.

Il faut penser à ajouter une connexion à la configuration de la base de donnée plus haut dans le fichier :

Dans le modèle que je décris, le moteur de tâches reçoit des notification l’informant qu’une tâche devient disponible et il doit ensuite acquérir la tâche dans la base de données, et en passant récupérer les paramètres d’exécution.

Cela vient du fait que dans le système de notification de PostgreSQL les notifications sont envoyées à toutes les instances qui sont à l’écoute.

D’autres système de messages (par exemple ActiveMQ) permettent que chaque message ne soit reçu que par une instance au plus. Dans ce cas l’acquisition de la tâche et la récupération des paramètres sont concentrés en une seule action : un moteur de tâches qui reçoit un message sait qu’il est le seul dans ce cas et peut donc directement exécuter la tâche.

Pour reprendre le vocabulaire de la partie 3, les systèmes de message comme ActiveMQ peuvent fournir plus ou moins du “exactly once”. Comme indiqué dans la partie 3, ce type de mécanisme est très difficile à mettre en œuvre de manière fiable, et en cas de crash du bus de message il y a des chances (même si elles sont faibles) que vous receviez le même message deux fois, il faut donc être préparé à cette éventualité.

Lors de la création d’une tâche, le code connaît la tâche à créer et ses paramètres.

On pourrait vouloir utiliser ces informations pour simplifier l’acquisition de tâches{nsbp}: quand réveille un worker, au lieu de chercher la prochaine tâche à exécuter avec la requête existante, il pourrais plutôt chercher la tâche par son id une requête qui cherche si la tâche avec un certain id est toujours disponible, ce qui devrait rendre la requête plus rapide.

Malheureusement cette solution n’est pas adaptée au fait d’avoir plusieurs {mdts} qui s’exécutent simultanément.

Par exemple si du code créé deux tâches T1 et T2, et que deux instances du moteur de tâches disposent de workers en sommeil. En utilisant la requête par défaut, le premier worker à se réveiller va récupérer T1, et le deuxième worker T2. Si les deux workers cherchent T1 par son id, le premier worker va la trouver mais pas le deuxième.

Il lui faudra alors utiliser la requête par défaut pour déterminer si une autre tâche est disponible. En échange d’une requête plus rapide, on aura donc exécuté une requête inutile.

Utiliser la requête générique évite donc cela, tout en gardant le code un peu plus simple.

Pour déclencher la notification, j’ajoute l’appel à la méthode notify lors de la création d’une tâche :

On peut tester le fonctionnement en lançant le moteur de tâches dans une console, et en lançant la création de tâche dans une autre :

Une des avantages d’utiliser le même outil pour stocker les données et gérer les messages et que les mêmes transactions s’appliquent aux deux.

Ainsi dans le code de TaskEngine.create_task, l’insertion de la tâche et l’envoi de la notification se font dans la même transaction que le code appelant.

Cela signifie que qu’en cas de rollback la tâche n’est pas crée et la notification envoyée.

Par exemple si une tâche a besoin d’un objet métier créé juste avant, si la tâche est créée puis que la création de l’objet est rollbackée, la tâche ne pourra pas le trouver.

Si vos traitements ont un risque significatif de rollbak, cela peut être assez pratique car cela évite d’avoir à prendre en compte cette situation dans le code de la tâche.

En effet un rollback de transaction ne signifie pas forcément qu’une erreur s’est produite, mais peut être le signe d’une tentative de modification concurrente dans la base de données, et dans ce cas arriver régulièrement.

Si on ne souhaite pas ce comportement, la création de tâche peut être isolée dans une transaction séparée dans le code applicatif, mais le comportement par défaut fournit une garantie qui peut être bien pratique.

Maintenant que le moteur sait traiter les ajouts de tâche, je vais m’intéresser au monitoring pour pouvoir commencer à suivre ce qui se passe.

Pour l’instant lors de l’exécution d’une tâche, trois informations sont disponibles :

Dans les trois, la première est présente dès le début de l’exécution de la tâche, et elle peut donc être définie comme non-nullable. Les autres seront renseignées à la fin de la tâche. Le résultat de l’exécution de tâches sera stocké dans une colonne de type jsonb, pour les même raisons que les paramètres des tâches.

Il faut aussi ajouter un nouveau statut stopped aux tâches qui indique qu’elles ont été exécutées, puisqu’elles ne seront plus supprimées.

Ensuite j’ajoute la déclaration du modèle, en modifiant la définition de Task pour ajouter le nouveau statut la relation entre les deux :

Reste à ajouter l’insertion et la mise à jour des informations d’exécution :

On peut ensuite tester le résultat :

La table task_executions contient bien les informations attendues :

La structure fonctionne donc et les données sont bien là, même si pour le moment on ne sait suivre que les tâches qui se terminent bien car la gestion d’erreur n’a pas encore été implémentée.

Et justement le dernier élément qui manque pour mettre en place cette gestion d’erreur est la capacité à planifier des tâches, qui sera traitée dans la partie suivante.

Avant de terminer, il faut s’intéresser un peu à la performance.

Avec le temps la table tasks devrait contenir de plus en plus d’occurrences de tâches terminées (avec l’hypothèse que le nombre de tâches à exécuter devrait rester raisonnable).

Cela risque de ralentir et fur et à mesure la sélection des tâches à exécuter, car la requête a besoin pour le moment de parcourir tous les enregistrements de la table.

Pour éviter cela, je vais ajouter un index à la table tasks.

La requête est la suivante :

Je pourrais simplement créer un index sur statut, ce qui ferait que la requête n’aurait qu’à parcourir les enregistrements qui ont le bon statut.

Mais pour que la recherche soit encore plus rapide, il est possible de créer un index sur les colonnes statut et created_at, de cette manière la recherche utilisera seulement l’index.

J’ajoute pour cela une nouvelle migration :

Après exécution de la migration, l’exécution de la requête utilise bien l’index pour faire la recherche :

Les tâches à exécution régulière correspondent à ce qu’il est possible de faire avec l’utilitaire cron dans les système Unix et Linux.

Cela peut par exemple servir pour programmer des traitement d’archivage ou des rappels toutes les nuits.

La solution la plus simple est d’utiliser l’outil cron directement. Même s’il est un peu passé de mode, il est fiable et bien documenté.

Lors d’un déploiement il est facile de mettre à jour la liste des tâches programmées si le fichier contenant la configuration cron fait partie du code.

Ses deux inconvénients majeurs pour cet usage sont :

Si vous n’êtes pas dans une de ces situations, cron peut donc parfaitement faire l’affaire.

Pour créer les tâches depuis cron, il y a deux manières de faire :

La première a l’avantage de la simplicité de mise en œuvre, par contre elle a l’inconvénient de lancer une machine virtuelle Ruby pour exécuter deux commandes SQL très simple (l’insertion de la tâche et la notification).

L’inconvénient principal de la seconde manière est que le format des données dans la base devient une API. En effet s’il change il faudra modifier le code Ruby mais peut-être aussi les scripts qui accèdent directement à la base.

Si c’est une approche qui est plutôt à éviter lorsqu’il s’agit d’une API exposée à d’autres systèmes, elle peut être acceptable ici vu que la programmation des tâches fait partie du même ensemble applicatif.

Je ne vais détailler l’utilisation de cron, il y a déjà bien assez de documentation à ce sujet, et je vais donc passer aux tâches à exécution unique.

Ces tâches ressemblent beaucoup aux tâches existantes, sauf qu’elles doivent s’exécuter dans le futur plutôt qu’immédiatement.

Jusqu’à présent, une tâche dans l’état waiting était forcément prête à être exécutée.

Je vais devoir changer ce comportement pour prendre en compte qu’une tâche puisse être en attente mais pas encore prête à être exécutée tout de suite.

Pour cela, je vais ajouter un nouveau champ scheduled_at qui contiendra l’heure à partir de laquelle la tâche peut être exécutée, avec pour valeur par défaut l’heure actuelle pour que la tâche puisse se lancer immédiatement.

L’implémentation va au final toucher peu de code.

Tout d’abord la migration qui ajoute le champ et qui recrée un nouvel index sur les colonnes statut et scheduled_at pour accélerer la requête de sélection de la prochaine tâche à exécute tout en supprimant l’index existant.

Je préfère la base de donnée se charge de donner la valeur par défaut de la colonne (qui correspond à l’heure en cours) car cela évite d’avoir à dépendre du fait que les différents clients soient bien à l’heure. Comme la base de données est unique, cela garantit la cohérence des valeurs.

Ensuite l’ajout du paramètre permettant de spécifier la valeur lors de la création d’une tâche :

Si le paramètre scheduled_at est spécifié, et que la tâche est programmée pour s’exécutée plus tard, il n’est pas nécessaire d’envoyer de notification car les notifications servent à réveiller des workers pour une exécution immédiate.

Le dernier élément à modifier est la requête qui cherche la prochaine tâche à exécuter, qui doit maintenant faire un tri sur scheduled_at plutôt que created_at, et qui doit également ignorer les tâches dont la valeur de scheduled_at est dans le futur.

Comme pour la valeur par défaut, je vais laisser la base définir l’heure actuelle. Sequel.lit indique à Sequel d’utiliser la valeur telle-quelle sans l’interpréter, elle est utile comme lorsqu’ici on utilise des fonctions SQL.

On peut vérifier que si on créé des tâches dans le futur, elles ne sont pas immédiatement exécutées.

Je vais créer une tâche qui doit s’exécuter 30 secondes dans le futur. Pour cela j’ajoute la fraction 30 / 86400 (c’est-à-dire le nombre standard de secondes dans un jour) à l’heure actuelle.

Tout d’abord comme prévu il ne se passe rien, car la date est dans le futur.

Mais au bout de 30 secondes il ne se passe rien non plus. Le problème c’est qu’en l’absence de notification aucun worker ne vérifie si la tâche est disponible pour être exécutée.

Si une autre tâche était créée en exécution immédiate après que les 30 secondes se soient passées, cela réveillerait un worker, et la tâche programmée serait traitée, mais tant qu’aucune tâche immédiate n’est créée, la tâche programmée restera à attendre.

Si de nouvelles tâches à exécution immédiates sont crées régulièrement, cette attente peut ne pas être un problème, mais peut-être que ça n’est pas le cas, ou que même si c’est le cas en principe, on peut préférer garantir que la tâche ne va pas attendre trop longtemps, en tous cas si des workers sont disponibles.

La manière idéale de faire serait faire en sorte de réveiller un worker dès qu’une tâche devient exécutable, mais il n’y a pas d’approche facile pour le faire avec les outils que j’ai choisis.

Une approche possible serait d’envoyer une notification lors de chaque création de ces tâches, et que les instances de moteur de tâches programment un réveil de worker pour chaque moment où au moins une des tâches doit s’exécuter.

Mais elle a cependant deux inconvénients.

Le premier est est que cela augmenterait la consommation mémoire des instances du moteur de tâches, avec un risque de dépassement mémoire si trop de tâches sont crées.

Le second est que le fonctionnement se mettrait à dépendre du fait que les instances de moteur de tâches ne soient pas redémarrées, car alors les informations de réveil seraient perdus.

Je vais plutôt partir du principe que si une tâche est programmée pour s’exécuter dans le futur, ce n’est pas grave si elle a quelques secondes de retard. Cela me permet de choisir une approche moins fine mais qui n’a pas les deux inconvénients cités plus hauts.

Ma solution sera de programmer à intervalle régulier des “réveils” qui reproduiront le même mécanisme que la réception d’une notification en poussant un élément dans la Queue où attendent les workers disponibles. Cela signifie envoyer un message pour réveiller un éventuel worker disponible, qui ira voir si une tâche est prête à être exécutée.

Pour cela je vais créer un nouveau Thread qui sera dédié à cet usage :

Le temps entre deux réveil, ici fixé à 10 seconde, correspond au temps maximum d’attente pour une tâche programmée, si des workers sont disponibles.

En effet si les workers sont déjà tous occupées à traiter un arriéré de tâches, il peut s’écouler un certain temps avant qu’une tâche programmée qui vient de devenir exécutable soit effectivement lancée.

Maintenant qu’il est possible de créer des tâches qui s’exécuteront plus tard, je vais pouvoir m’en servir pour mettre en place la gestion d’erreur.

La première chose à faire est d’avoir une tâche qui plante pour pouvoir tester ce qui va suivre.

Jusqu’ici le seul type de tâche qui existait était la WaitingTask, je vais donc ajouter une FailingTask :

Comme il n’y aucune gestion d’erreur pour le moment, la division par zéro fait planter l’instance du moteur de tâches.

Pour mettre en place la gestion d’erreur je vais commencer par enrichir le modèle de données : dans la table task_executions je vais ajouter un status pour déclarer que l’exécution a eu une erreur, et une error pour pouvoir enregistrer les informations sur l’erreur.

Le status running va correspondre aux tâches en cours d’exécution. Jusque là il était possible de déterminer qu’une exécution n’était pas terminée en regardant si colonne stopped_at était vide. Mais comme je préfère que la colonne status ne soit pas nullable pour éviter les oublis, j’ai besoin d’un status pendant l’exécution.

La première chose est d’ensuite de déclarer les status comme des constantes dans la classe TaskExecution :

Et je peux maintenant ajouter le début de la gestion d’erreur : quand l’exécution d’une tâche renvoie une exception, elle sera déclarée en erreur et l’erreur sera enregistrée sous la forme du message d’erreur, de la classe de l’exception, et de la pile d’appel.

Et voici le résultat quand je rééessaie d’exécuter une FailingTask{nsbp}:

Comme décrit plus haut, il est souvent bénéfique de relancer une tâche qui a échoué pour les cas où la source de l’erreur était transitoire.

Cela signifie que les tâches doivent être prévues pour cela, par exemple si une tâche échoue à la fin de son exécution alors qu’elle a déjà effectués certains traitements des traitements, le fait de la relancer ne doit pas avoir d’effet secondaires.

Parfois cela est facile à faire, par exemple si toute la tâche s’exécute dans une seule transaction, mais si la tâche interagit avec des systèmes extérieurs cela peut être plus compliqué.

Pour un moteur de tâches, une manière de résoudre ce problème est de rendre le réessai configurable, je vais commencer par implémenter des réessais systématiques, puis j’ajouterai la configurabilité.

Quand on veut réessayer une tâche dans l’espoir que le problème qui l’empêchait de fonctionner ne se produira plus, il faut que le réessai ne soit pas immédiat. En général on commence par réessayer un peu plus tard, puis encore un peu plus tard mais en attendant un peu plus, et ainsi de suite en augmentant à chaque fois la durée de l’attente.

Il s’agit de trouver un équilibre entre une attente plus longue, qui augmente les chances que le problème ait été résolu, et une attente trop longue qui deviendrait pénalisante.

Une approche souvent employée est d’utiliser des puissance de deux : le premier réessai se faut au bout de 2⁰ = 1 minute, le second au bout de 2¹ = 2 minutes, puis 4, 8, 16…

Il faut aussi savoir abandonner, car au bout d’un moment réessayer ne sert plus à rien. Pour cela je vais définir un nombre maximum de réessais au bout duquel la tâche ne sera pas relancée. Le choix de la valeur est un peu arbitraire et dépend de l’environnement dans lequel on se trouve. Je vais choisir 10, ainsi la dernière tentative exécution se produira 1 + 2 + 4 + 8 + 16 + 32 + 64 + 128 + 256 + 512 = 1023 minutes après l’exécution initiale, soit un peu plus de 17 heures.

Si besoin, il serait possible de la rendre paramétrable, en utilisant par exemple le contexte que je vais ajouter juste en dessous.

Une tâche a besoin de savoir à quel numéro de réessai elle correspond, pour calculer l’heure du prochain rééssai et pour déterminer qu’il est temps de s’arrêter quand elle a atteint le nombre maximum.

Pour stocker cette information je pourrais utiliser les paramètres de la tâche, mais je préfère les réserver aux paramètres applicatifs.

Je vais donc ajouter une nouvelle colonne à la table tasks. Plutôt que d’ajouter une colonne spécifique qui ne contiendrait que cette unique valeur, je vais créer une colonne de contexte au format jsonb. L’idée est de pouvoir ensuite l’utiliser pour ajouter d’autre métadonnées si le moteur de tâches est étendu.

Je vais aussi avoir besoin d’un nouveau statut failed pour les tâches qui indiquer qu’elles ont épuisé leurs réessais.

Quand une tâche aura été relancée, il faudra donc distinguer le statut de la tâche des status de ses exécutions successives. Une tâche en status waiting avec des exécutions en failed aura donc déjà été lancée et sera donc en train d’attendre une nouvelle tentative.

Je déclare le nouveau statut{nsbp}:

Puis vient l’implémentation des réessais.

Ce qui donne :

Comme je le disais plus haut, parfois il n’est pas nécessaire de réessayer une tâche qui a échoué, et parfois il serait néfaste de le faire :

Pour les types de tâches qui ne doivent jamais être réessayer, cette information est liée au type de la tâche, et l’idéal est de pouvoir le spécifier au niveau de la classe qui implémente la classe, plutôt que lorsqu’on créé la tâche.

Pour cela je vais utiliser une approche inspirée de Java, en utilisant un module sans méthode. Ce module servira à marquer que ce type de tâche n’est jamais à relancer.

Utiliser un module plutôt qu’une classe évite d’ajouter une contrainte sur les classes qui implémentent les tâches en les empêchant d’hériter d’une autre classe.

Voici le module :

Et voici une tâche qui va échouer et qui utilise ce module :

Lors d’une exécution en erreur, je vais vérifier si la tâche en question inclus ce module :

Ce qui donne, en lançant une TaskEngine::Tasks::FailingNotRetryTask :

Quand une tâche qui en principe devrait être réessayée ne devrait pas l’être dans certain cas, je vais m’appuyer sur l’exception qui est remontée en cas d’erreur. Comme pour les tâche, je vais utiliser un module pour permettre de marquer les classes d’exception dans ce cas.

Et voici une tâche qui va échouer en utilisant une exception de ce type :

La gestion de ce cas ressemble beaucoup au précédent :

Ce qui donne, en lançant une TaskEngine::Tasks::FailingNotRetryExceptionTask :

Ce qui boucle la gestion des cas d’erreurs{nsbp}: les erreurs sont enregistrées, les tâches sont par défaut relancées, mais il est possible de gérer de manière fine les situation où cela n’est pas souhaitable.