La concurrence via le modèles d'Acteurs, introduction à Akka

Dans cet article, nous allons présenter un modèle de programmation concurrente appelé « modèle d’acteurs », qui offre une sémantique de plus haut niveau que la manipulation de threads et la synchronisation par verrous que nous avons l’habitude de voir en Java. Nous présenterons ce modèle en utilisant le framework Akka, qui offre entre autres les fondations pour un système distribué implémentant le modèle d’acteurs.

Qu’est-ce que le modèle d’acteurs?

Le modèle d’acteurs est un modèle de programmation concurrente, dans lequel la charge de travail est répartie entre des entités s’exécutant en parallèle, les acteurs. C’est un modèle dans lequel il n’y a pas d’état partagé, les acteurs sont isolés et l’information ne peut circuler que sous forme de messages. Les acteurs reçoivent ces messages et ne peuvent réagir qu’en manipulant les données dans le message (effectuer un calcul ou une transformation sur la donnée), en envoyant un message à d’autres acteurs ou en créant de nouveaux acteurs.
Ce modèle est une abstraction de plus haut niveau que les verrous et threads, ce qui retire de la complexité pour le développeur. Les acteurs ont été définis dans une publication académique en 1973 et popularisés par le langage Erlang, créé à Ericsson pour construire des systèmes télécoms hautement concurrents et disponibles.
On peut se représenter les acteurs comme un ensemble de personnes auxquelles on va déléguer des tâches. Comme dans une entreprise, ces acteurs ont une hiérarchie à laquelle ils doivent rendre compte (propagation des erreurs). ET évidemment, ces personnes ne pouvant pas lire dans les pensées des autres elles doivent s’appuyer sur une communication orale explicite (le passage de messages) pour échanger l’information.

Présentation d’Akka

Akka est un framework pour la construction de systèmes distribués basés sur les acteurs. La philosophie adoptée en termes de tolérance de panne pour ces systèmes est « let it crash« . En effet, les acteurs sont des éléments légers, qu’il est facile d’instancier et de relancer en cas de crash (du fait qu’ils ne partagent pas leur état).
Pour cela, Akka impose une supervision afin de pouvoir gérer la mort prématurée d’acteurs, et ce dans une hiérarchie de supervision : tout nouvel acteur est supervisé par son parent. Cette supervision permet la création de systèmes qui s’auto-réparent.
De plus les acteurs sont purement asynchrones et peuvent être relocalisés de manière transparente, y compris dans une autre JVM (ce qui permet la création d’un véritable système distribué sur un cluster).
Akka permet de plus d’aller un peu plus loin que le modèle d’acteurs, en introduisant un hybride entre les acteurs et la mémoire transactionnelle logicielle (Software Transactional Memory, STM), qui est une autre alternative à la synchronisation par verrous dans laquelle les opérations sur les données partagées sont effectuées au sein de transactions, en isolation les unes des autres. Akka introduit donc les Transactors, qui ont des flux de messages « atomiques », avec possibilités de réessayage automatique et de rollback.
Akka a des API en Java, mais fait aussi désormais partie de la pile Typesafe pour la programmation réactive : le langage Scala, Akka, et le framework web Play!.

Notre premier acteur en Akka

En Akka, un acteur est défini par une classe qui étend UntypedActor. La seule méthode à implémenter est la méthode onReceive :

public class MyCommandActor extends UntypedActor {
    public void onReceive(Object message) throws Exception {
        if (message instanceof String) {
            String msg = (String) message;
            if (msg.startsWith("/cmd")) {
                System.out.println("Received command "+msg);
                return;
            }
        }
        unhandled(message);
    }
}

Cet acteur se contente de détecter des messages correspondant à des commandes dans notre application et de les afficher dans la sortie standard. Si le type ou le contenu du message n’est pas reconnu, on peut faire appel à la méthode héritée unhandled, ce qui va propager un message sur le flux d’événements du système d’acteurs (utile par exemple pour les logs).
Pour instancier un acteur en Akka, on utilise les Props, qui permettent des variantes dans la construction de l’acteur (via factory, par appel direct au constructeur, par passage de paramètres à un constructeur adéquat…). On peut passer ces Props, qui sont immuables et donc partageables, soit au système d’acteur lui-même (pour créer un acteur de haut niveau, sous la tutelle de « l’Acteur-Gardien », le plus haut dans la hiérarchie de supervision), soit au contexte d’un acteur existant pour lui créer un acteur fils :

//création d'un système d'acteurs
final ActorSystem systeme = ActorSystem.create("MonApplication");
//création d'un acteur de haut-niveau, constructeur par défaut
final ActorRef monActeur = systeme.actorOf(Props.create(MyCommandActor.class));
//création d'un acteur de haut-niveau avec passage de paramètre au constructeur
final ActorRef monActeur2 = systeme.actorOf(Props.create(MyCommandActor.class, "truc"));

//création d'un acteur fils, nommé "nomDActeur", au sein d'un acteur existant:

//...reste de la méthode onReceive par exemple
final ActorRef monActeurDansUnActeur = getContext().actorOf(Props.create(MyCommandActor.class), "nomDActeur");

On a vu que les acteurs étaient isolés, et relocalisables de manière transparente. Pour cela, il est important de ne pas utiliser des références directes à nos acteurs. C’est pourquoi
nous obtenons à chaque création un objet de type ActorRef. Un ActorRef est associé à un et un seul acteur, et peut être sérialisé et envoyé sur le réseau vers un autre noeud : il fera toujours correctement référence à l’acteur dans le noeud distant.
Une fois l’acteur créé via les méthodes actorOf, celui-ci est automatiquement démarré de manière asynchrone.
L’API UntypedActor expose les méthodes suivantes :
- getContext() : contexte de l’acteur, avec des méthodes pour créer un nouvel acteur (cf ci-dessus), accéder au système d’acteurs, au parent, aux acteurs enfants supervisés, etc…
- getSelf() : retourne la ActorRef de l’acteur
- getSender() : retourne la ActorRef de l’acteur ayant envoyé le dernier message reçu
- preStart(), preRestart(), postRestart(), postStop() : méthodes à surcharger pour personnaliser ces différentes étapes du cycle de vie de l’acteur.

La Communication : Envoyer et Recevoir des Messages

On l’a déjà vu, la réception de messages se fait via la méthode onReceive() de l’acteur. Pour envoyer un message à un acteur dont on possède la ActorRef, on pourra utiliser la méthode tell(Object message, ActorRef sender), qui est non-bloquante (on n’attend pas de réponse du destinataire). Le destinataire pourra alors très bien répondre au message en envoyant à son tour un tell à l’acteur d’origine en le retrouvant grâce à la méthode getSender. Note: le tell se fait généralement en passant getSelf() comme émetteur, mais on peut très bien utiliser d’autres ActorRef. Pour faire de la transmission de message on dispose cependant d’une méthode dédiée forward, à laquelle on passe le message et le contexte de l’acteur courant getContext(), ce qui conservera l’émetteur d’origine, même après plusieurs sauts.
Une autre manière d’envoyer un message à un acteur et de traiter une réponse est possible avec le pattern ask (demande) se basant sur les Futures d’Akka, son propre mécanisme pour représenter le résultat à venir d’un traitement asynchrone. Du fait qu’on utilise les Futures, l’appelant n’a même pas besoin d’être lui-même un acteur sur lequel on aurait à faire un tell. Cette méthode est disponible sous forme de méthode statique dans la classe akka.pattern.Patterns : Future<Object> ask(ActorRef actor, Object message, long millisecondTimeout).
Attention : l’acteur destinataire doit répondre au message par un tell pour remplir le Future, y compris en cas d’exception pour propager celle-ci plutôt que de tomber en timeout. Pour le cas des exceptions, on les enveloppera dans un new akka.actor.Status.Failure(e). Autrement, la Future se terminera par une AskTimeoutException.

Immutabilité des Données, Opérations Bloquantes

– Les informations échangées entre les acteurs devraient être immuables, et il est fortement conseillé que les classes de messages elles-même le soient aussi. Si on autorise le changement d’état sur les données échangées, on revient aux problèmes de concurrence et de locks basiques de Java.
– Il n’y a pas de garantie que deux messages traités à la suite par le même acteur le soient dans le même thread (l’acteur peut-être relocalisé dans un autre thread)
– Vos acteurs devraient être conçus de manière à éviter à la fois d’occuper et conserver trop de ressources et de trop communiquer avec l’extérieur sans bonne raison. Les opérations bloquantes sont à proscrire un maximum, à ne faire que lorsqu’on ne peut pas y échapper.
– Pour effectuer une opération bloquante, il n’est pas automatiquement suffisant de la wrapper dans une Future. Cela risque de poser des problèmes de contention, de limitation de mémoire ou de threads. Pour éviter cela il faudra mettre en place des pools de threads correctement dimensionnés, s’assurer que les éventuels appels bloquants wrappés dans un Future sont limités en nombre (là encore en configurant par exemple un pool de threads dédiés à cette catégorie de Futures).
– Une autre stratégie est de dédier un thread à la gestion de ressources bloquantes, qui se chargera de réagir aux événements liés à ces ressources en envoyant des messages aux acteurs ad-hoc.

Garanties de Livraison

Pour un mécanisme de distribution, les garanties peuvent se répartir en trois catégories, en ordre de complexité et de coût croissants :

• Au plus une fois : chaque message confié au système atteint son destinataire 0 ou 1 fois, c’est-à-dire que des messages pourraient être perdus.
• Au moins une fois : chaque message confié au système atteint son destinataire 1 à N fois, les messages ne sont pas perdus mais peuvent être dupliqués.
• Exactement une fois : chaque message arrive à son destinataire, sans pertes ni duplications.

Le coût des deux dernières catégories est accru car il faut gérer un mécanisme d’accusés-réception, et gérer un état des messages envoyés (dans la 2ème catégorie) mais aussi reçus (dans la 3ème catégorie).
Akka a pris le parti de ne fournir que la première catégorie fire-and-forget de garantie, au plus une fois. Cela rend les limites de fiabilité des systèmes distribués explicites. La seule manière significative de s’assurer qu’une interaction s’est faite avec succès est, au sens des créateurs du framework, par le biais d’une reconnaissance de réception au niveau métier de l’application, pas dans le système de distribution des messages.
On peut arguer que toutes les applications ne nécessitent pas le genre de garanties offertes par la 3ème catégorie. En n’offrant que les garanties les plus basiques, Akka n’induit pas le surcoût associé aux catégories supérieures aux applications qui n’ont pas ce besoin.
La deuxième garantie qu’Akka fourni en termes de remise des messages est la garantie de la conservation de l’ordre des messages entre un même émetteur et receveur. Plus précisément, pour une paire d’acteurs E1-R donnés, les messages envoyés du premier au second n’apparaîtront pas dans le désordre dans la mailbox de ce dernier. Les messages envoyés par un autre émetteur E2 à R pourront être reçus en alternance avec les messages de E1, et bien sûr il reste la possibilité qu’aucun message ne parvienne à R.

Gestion des Erreurs : Supervisor Strategy

Chaque acteur est le superviseur de ses acteurs fils, et est notifié par un message système des erreurs qui se produisent chez ses enfants. La réaction à ces messages se fait au travers d’une Supervisor Strategy. Celle-ci va permettre de déterminer :
– à quels acteurs la réaction s’applique selon la classe de la stratégie (l’acteur en panne pour une OneForOneStrategy ou tous les enfants du superviseur pour une OneForAllStrategy)
– les limites de redémarrage (maximum N restarts par M minutes par exemple)
– comment réagir à différentes classes d’exception, via une Function<Throwable, Directive>, les différentes directives pouvant être créées via les méthodes resume(), restart(), stop() ou escalate().
On pourra déclarer sa propre classe de stratégie de supervision, soit dans une classe à part, soit de manière plus pratique en classe fille / anonyme dans l’acteur (ce qui permet d’accéder de manière thread-safe aux champs de l’acteur, et aussi d’utilis
er la méthode getSender() pour trouver l’acteur en erreur par exemple). Escalate (remonter l’erreur au parent du superviseur) est utilisée pour toute exception non explicitement traitée.
Si on ne déclare pas de stratégie de supervision, la stratégie par défaut réagit aux exceptions suivantes :
- ActorInitializationException, ActorKilledException: l’acteur est stoppé.
- Exception : l’acteur est redémarré (restart).
– autres Throwable : ils sont remontés au parent (escalate)
Ces stratégies sont définies à la création de l’acteur et ne peuvent être changées ultérieurement. Les SupervisorStrategy loggent automatiquement les erreurs, sauf si elles sont remontées (ce qui suppose que l’exception sera traitée, et loggée, plus haut dans la hiérarchie).

Conclusion

Dans cet article, nous avons eu une introduction au modèle d’acteurs, tel qu’implémenté par Akka pour la création de systèmes distribués robustes et réactifs. Nous avons vu comment instancier des acteurs, comment les faire communiquer, et quelles sont les garanties de distribution offertes par la plate-forme, avec des pistes pour aller au-delà, et j’espère que cette entrée en matière vous sera utile dans la conception de votre prochaine application concurrente/distribuée.